Wave D: local-mode rewrite — file transport default, WS hardening, protocol version

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]:

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]:

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]:

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]:

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]:

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]:

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]:

README.md

@@ -112,6 +112,7 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 | `max_batch_size` | `int` | `100` | Early-flush threshold (number of events). |
 | `max_buckets` | `int` | `2000` | Maximum unique (provider, endpoint, method) triplets per window. Crossing this triggers an early flush. |
 | `local_port` | `int` | `9847` | WebSocket port for the VS Code extension. |
+| `local_transport` | `Literal["file", "ws"]` | `"file"` | Which local-mode transport to use. `"file"` (default) writes NDJSON to `~/.recost/local-telemetry/{project_id}.jsonl`. `"ws"` opts into a WebSocket to `localhost:{local_port}` (no server hosts this by default — see [extension#91](https://github.com/recost-dev/extension/issues/91)). |
 | `debug` | `bool` | `False` | Log telemetry activity to stderr. |
 | `enabled` | `bool` | `True` | Master kill switch — set `False` to disable entirely. |
 | `custom_providers` | `list[ProviderDef]` | `[]` | Extra provider rules with higher priority than built-ins. |
@@ -125,6 +126,39 @@ All fields are optional. Pass them as keyword arguments or via a `RecostConfig`
 
 > **Note on exclusions:** `exclude_patterns` performs substring matching against both `event.url` and `event.host`; patterns containing `*` raise `ValueError` at init time (substring matching is not glob). For unambiguous host-level exclusion without substring false-positives (e.g., excluding `api.example.com` without also dropping `myapi.example.com`), use `exclude_hosts` instead. Both are applied additively — events matching either are dropped before reaching the aggregator.
 
+### Local-mode transports
+
+When no `api_key` is set, the SDK runs in **local mode**. Two transports are available:
+
+#### File (default — recommended)
+
+`local_transport="file"`: each `WindowSummary` is appended as one NDJSON line to:
+
+```
+$RECOST_LOCAL_DIR/{project_id}.jsonl # if RECOST_LOCAL_DIR is set
+~/.recost/local-telemetry/{project_id}.jsonl # otherwise (POSIX & macOS)
+```
+
+If `project_id` is empty, the file is named `default.jsonl`.
+
+On POSIX systems the file is `chmod`'d to `0o600` (owner read/write only). On Windows, the ACL is not adjusted — Python's `chmod` is mostly a no-op there.
+
+Multi-process writes from different processes targeting the same `project_id` are safe for typical telemetry frames (POSIX `O_APPEND` is atomic for writes ≤ `PIPE_BUF`, ~4 KB on Linux). Very large frames may interleave across processes.
+
+If the directory can't be created or the file can't be opened (`PermissionError`, disk full), `on_error` fires once per failure-episode and subsequent writes are silently dropped until the next successful write.
+
+#### WebSocket (opt-in)
+
+`local_transport="ws"`: opens `ws://127.0.0.1:{local_port}` (default port 9847). The VS Code extension does not currently host a WS server, so this is only useful if you've stood up your own listener.
+
+Hardening for opt-in WS users:
+- Outbound queue is capped at 1000 frames with drop-oldest semantics. The first dropped frame fires `on_error` once per overflow episode (cleared on reconnect).
+- After 10 consecutive failed reconnect attempts, the transport gives up and fires `on_error` once with a message pointing back to `local_transport="file"`.
+
+#### Wire format
+
+Every frame on every transport carries a top-level `protocolVersion: "1.0"` field. Consumers must reject frames with an unknown MAJOR version; MINOR bumps are forward-compatible.
+
 ### Custom providers
 
 ```python

recost/__init__.py

@@ -7,10 +7,11 @@
 
 from ._types import (
  FlushStatus,
-RecostConfig,
+LocalTransportMode,
  MetricEntry,
  ProviderDef,
  RawEvent,
+ RecostConfig,
  RecostError,
  RecostAuthError,
  RecostFatalAuthError,
@@ -36,6 +37,7 @@
  "RecostFatalAuthError",
  "RecostRateLimitError",
  "TransportMode",
+ "LocalTransportMode",
  "FlushStatus",
  "ProviderRegistry",
  "BUILTIN_PROVIDERS",

recost/_aggregator.py

@@ -178,6 +178,7 @@ def flush(self) -> Optional[WindowSummary]:
  window_start=window_start,
  window_end=window_end,
  metrics=metrics,
+ protocol_version="1.0",
  )
 
  @property

recost/_transport.py

@@ -12,6 +12,8 @@
 import asyncio
 import json
 import logging
+import os
+import pathlib
 import random
 import threading
 import time
@@ -20,7 +22,7 @@
 from dataclasses import dataclass, replace
 from datetime import datetime, timezone
 from email.utils import parsedate_to_datetime
-from typing import Callable, Optional
+from typing import IO, Callable, Optional, Union
 
 from ._types import FlushStatus, RecostConfig, TransportMode, WindowSummary
 
@@ -138,10 +140,17 @@ def __init__(
  port: int,
  debug: bool = False,
  shutdown_timeout_s: float = 3.0,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ max_queue_size: int = 1000,
+ max_reconnect_attempts: int = 10,
  ) -> None:
  self._port = port
  self._debug = debug
  self._shutdown_timeout_s = shutdown_timeout_s
+ self._on_error = on_error
+ self._max_queue_size = max_queue_size
+ self._max_reconnect_attempts = max_reconnect_attempts
+ self._drop_announced = False
  self._thread: Optional[threading.Thread] = None
  self._running = False
  self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -170,7 +179,7 @@ def _run(self) -> None:
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  self._loop = loop
- self._queue = asyncio.Queue()
+ self._queue = asyncio.Queue(maxsize=self._max_queue_size)
  self._ready.set()
  try:
  loop.run_until_complete(self._ws_loop())
@@ -205,6 +214,7 @@ async def _ws_loop(self) -> None:
  try:
  async with websockets.connect(url) as ws:
  reconnect_attempts = 0
+ self._drop_announced = False # New episode after reconnect.
  while self._running:
  try:
  msg = await asyncio.wait_for(self._queue.get(), timeout=0.5)
@@ -225,9 +235,28 @@ async def _ws_loop(self) -> None:
  except Exception:
  if not self._running:
  return
- base = min(0.5 * (2 ** reconnect_attempts), 30.0)
- delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  reconnect_attempts += 1
+ if reconnect_attempts > self._max_reconnect_attempts:
+ # Give up. Announce via on_error and stop the loop.
+ self._running = False
+ if self._on_error is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ port = self._port
+ attempts = reconnect_attempts - 1
+ try:
+ cb(RecostError(
+ f"recost: local WS transport gave up after "
+ f"{attempts} failed connects to "
+ f"127.0.0.1:{port}. Set local_transport='file' "
+ f"to write telemetry to disk instead."
+ ))
+ except Exception:
+ # User callback raised — never crash the loop.
+ pass
+ return
+ base = min(0.5 * (2 ** (reconnect_attempts - 1)), 30.0)
+ delay = base * (1 + (random.random() - 0.5) * 0.5) # 0.75x..1.25x
  await asyncio.sleep(delay)
 
  def send(self, payload: str) -> None:
@@ -237,8 +266,45 @@ def send(self, payload: str) -> None:
  queue_ = self._queue
  if loop is None or queue_ is None or loop.is_closed():
  return
+ # Bounded queue with drop-oldest semantics. Schedule a coroutine
+ # that, on QueueFull, drains one element and retries put_nowait.
+ async def _put_with_overflow() -> None:
+ assert queue_ is not None # local for mypy after capture
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Drop the oldest queued frame and put the new one.
+ try:
+ queue_.get_nowait()
+ except asyncio.QueueEmpty:
+ pass
+ try:
+ queue_.put_nowait(payload)
+ except asyncio.QueueFull:
+ # Should be unreachable after the get_nowait; treat as
+ # silent drop rather than crashing the loop.
+ return
+ # Announce once per overflow episode; cleared on reconnect.
+ if not self._drop_announced:
+ self._drop_announced = True
+ if self._on_error is not None:
+ # Capture for the cross-thread schedule below
+ loop_local = self._loop
+ if loop_local is not None:
+ from ._types import RecostError
+ cb = self._on_error
+ # Schedule the user callback via call_soon to
+ # avoid running it inside the asyncio loop's
+ # body (matches the rest of the file's pattern).
+ loop_local.call_soon(
+ lambda: cb(RecostError(
+ "recost: local WS queue full; dropping "
+ f"oldest frame (cap={self._max_queue_size})"
+ ))
+ )
+
  try:
- asyncio.run_coroutine_threadsafe(queue_.put(payload), loop)
+ asyncio.run_coroutine_threadsafe(_put_with_overflow(), loop)
  except RuntimeError:
  # Loop was closed between the is_closed() check and the schedule.
  pass
@@ -277,6 +343,111 @@ def dispose(self) -> None:
  self._thread = None
 
 
+# ---------------------------------------------------------------------------
+# Local file transport
+# ---------------------------------------------------------------------------
+
+
+class _LocalFileTransport:
+ """Append-only NDJSON transport for local mode (#38).
+
+ Writes one ``json.dumps(window_summary.to_dict()) + "\\n"`` per send()
+ to ``~/.recost/local-telemetry/{project_id}.jsonl``. Tooling can read
+ the file later; the SDK never reads back.
+
+ Defaults to the new local mode (was WebSocket — see _LocalTransport).
+ The WS target does not exist (recost-dev/extension#91), so writing
+ to disk gives users a working "no cloud account" path.
+
+ POSIX: file is chmod'd to 0o600 on creation.
+ Windows: file ACLs are not adjusted (Python's chmod is mostly a no-op
+ on Windows). Document this in the README.
+
+ Multi-process: POSIX O_APPEND is atomic for writes <= PIPE_BUF (~4 KB
+ on Linux). Larger frames may interleave across processes writing the
+ same file. Documented limitation; sufficient for typical telemetry
+ window sizes.
+
+ Failure modes: PermissionError / OSError (disk full, ENOSPC) fires
+ on_error ONCE per failure-episode (cleared on next successful write)
+ and drops the line. Never raises.
+ """
+
+ def __init__(
+ self,
+ project_id: str,
+ on_error: Optional[Callable[[Exception], None]] = None,
+ debug: bool = False,
+ ) -> None:
+ self._project_id = project_id or "default"
+ self._on_error = on_error
+ self._debug = debug
+ self._path = self._resolve_path()
+ self._fh: Optional[IO[str]] = None
+ self._error_announced = False
+ self._open_or_warn()
+
+ @staticmethod
+ def _resolve_dir() -> pathlib.Path:
+ """``$RECOST_LOCAL_DIR`` if set, else ``~/.recost/local-telemetry/``."""
+ env = os.environ.get("RECOST_LOCAL_DIR")
+ if env:
+ return pathlib.Path(env)
+ return pathlib.Path.home() / ".recost" / "local-telemetry"
+
+ def _resolve_path(self) -> pathlib.Path:
+ return self._resolve_dir() / f"{self._project_id}.jsonl"
+
+ def _open_or_warn(self) -> None:
+ """Open the append handle; announce on_error once per episode if
+ directory creation or file open fails. ``line buffering`` so each
+ ``json.dumps(...) + "\\n"`` write is flushed to disk on newline."""
+ try:
+ self._path.parent.mkdir(parents=True, exist_ok=True)
+ # Open in append text mode with line buffering (newline = flush).
+ self._fh = open(self._path, "a", encoding="utf-8", buffering=1)
+ # POSIX-only: tighten permissions. Best effort — chmod is a
+ # no-op on Windows.
+ try:
+ os.chmod(self._path, 0o600)
+ except OSError:
+ pass
+ self._error_announced = False
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def _announce_once(self, exc: Exception) -> None:
+ if not self._error_announced:
+ self._error_announced = True
+ if self._on_error is not None:
+ self._on_error(exc)
+
+ def send(self, payload: str) -> None:
+ """Append one NDJSON line. ``payload`` is the already-serialized
+ body (the same string the cloud path POSTs). Append a newline."""
+ if self._fh is None:
+ # Failed open at construction; try once more, silently.
+ self._open_or_warn()
+ if self._fh is None:
+ return
+ try:
+ self._fh.write(payload + "\n")
+ # Line buffering flushes on the newline; an explicit flush()
+ # would be belt-and-suspenders but adds a syscall per send.
+ except OSError as exc:
+ self._announce_once(exc)
+
+ def dispose(self) -> None:
+ """Flush and close the file handle. Safe to call multiple times."""
+ if self._fh is not None:
+ try:
+ self._fh.flush()
+ self._fh.close()
+ except OSError:
+ pass
+ self._fh = None
+
+
 # ---------------------------------------------------------------------------
 # Transport class
 # ---------------------------------------------------------------------------
@@ -302,13 +473,25 @@ def __init__(self, config: RecostConfig) -> None:
  self._auth_warned_stderr: bool = False
  self._defer_callback: Optional[Callable[[int], None]] = None
 
- self._local: Optional[_LocalTransport] = None
+ # Local-mode transport: file (default) or ws (opt-in, see #38).
+ # The attribute is typed as the union of both — both expose
+ # send(payload: str) and dispose() so the rest of Transport
+ # doesn't care which one is wired.
+ self._local: Optional[Union[_LocalFileTransport, _LocalTransport]] = None
  if self.mode == "local":
- self._local = _LocalTransport(
- config.local_port,
- config.debug,
- shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
- )
+ if config.local_transport == "file":
+ self._local = _LocalFileTransport(
+ project_id=config.project_id or "",
+ on_error=config.on_error,
+ debug=config.debug,
+ )
+ else: # "ws" — opt-in
+ self._local = _LocalTransport(
+ config.local_port,
+ config.debug,
+ shutdown_timeout_s=config.shutdown_flush_timeout_ms / 1000.0,
+ on_error=config.on_error,
+ )
 
  @property
  def last_flush_status(self) -> Optional[FlushStatus]: