diff --git a/DESIGN.md b/DESIGN.md index 661c739..6151928 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -40,7 +40,7 @@ The AI session talks **only** to the local database. Scripts perform every actio | Runtime | This public client. Team-specific rules live elsewhere and must not ship a second store binary. | | Session mail | Addressed to a **session id**. Delivery does not require a subscription. | | TUI knock | Script wakes the session with only `da ist Post id `. The agent reads that row from local Postgres. | -| Device daemon | Always-on user service on this device. `agent init` installs and starts it with knock (`LISTEN` plus usage / pending / github pending / mail pending / `pr.merged` polls) and the local dashboard; daemon `sync --follow` starts only after `agent pair`, once `device.json` has token and hub URL. | +| Device daemon | Always-on user service on this device. `agent init` installs and starts it with knock (`LISTEN` plus usage / pending / github pending / mail pending / `pr.merged` polls, plus a hub sync — push then pull — each cycle once paired) and the local dashboard; daemon `sync --follow` starts only after `agent pair`, once `device.json` has token and hub URL. | | Outside facts | Scripts notice GitHub (and other outside) state. The agent is not told by a human and does not poll GitHub. Example: a recorded PR merges → script writes `pr.merged` on that session and knocks. | | AI vs scripts | The AI inserts local intent. Scripts perform every side effect that leaves the machine. Model text is never a state transition. | | Checks and gates | A **check** records a fact (`agent check record`). A **gate** is a policy verdict over evidence (`agent gate record`). A model claim is neither. Confidence is not proof. | @@ -412,7 +412,7 @@ agent sync [--follow] agent restore agent ping send|list|ack agent daemon [--install|--uninstall] # always-on supervisor; init installs the user service -agent knock [--once] # --once drains; without --once is foreground; user service is the supported always-on path +agent knock [--once] # --once drains; without --once is foreground, syncing (push + pull) with the hub after each cycle once paired; user service is the supported always-on path agent watch pr-merged # one scan; device daemon covers the loop agent watch pending # one scan; LISTEN agent_work / execute subscription.set and query.request agent watch grok-usage # one scan; knock child (under the device daemon) polls every 60s @@ -469,7 +469,7 @@ These are not silent defaults in code; they are human steps after merge: 2. Create a GitHub OAuth App whose callback is `{public-url}/auth/github/callback`. 3. Deploy `agent-core` with every `AGENT_CORE_*` variable set. 4. Add GitHub logins to `teams.yaml` via pull request. -5. On each laptop: PostgreSQL 15+ (`initdb`/`pg_ctl` on `PATH`, or `AGENT_PG_BIN` / `AGENT_PG_DSN`), `pip install -e .`, `agent init` (installs and starts the user-service daemon for knock, usage, pending, github pending, mail pending, `pr.merged`, and the local dashboard; daemon `sync --follow` starts only after pair, once `device.json` has token and hub URL), `agent pair --hub …`. Do not leave a separate `agent knock` or `agent sync --follow` as the always-on path; one-shot `agent sync` remains fine after pairing. +5. On each laptop: PostgreSQL 15+ (`initdb`/`pg_ctl` on `PATH`, or `AGENT_PG_BIN` / `AGENT_PG_DSN`), `pip install -e .`, `agent init` (installs and starts the user-service daemon for knock, usage, pending, github pending, mail pending, `pr.merged`, a per-cycle hub sync once paired, and the local dashboard; daemon `sync --follow` starts only after pair, once `device.json` has token and hub URL), `agent pair --hub …`. Do not leave a separate `agent knock` or `agent sync --follow` as the always-on path; one-shot `agent sync` remains fine after pairing. Later product work (not required to operate v1 after merge): @@ -593,7 +593,7 @@ The script: 1. Authenticates with credentials that never enter the store or `evidence`. 2. Pulls new lines since the last cursor (persisted next to the config). 3. Filters to incident lines only: HTTP access-log lines (`METHOD path status`) are dropped; lines with a logger level token `ERROR` / `FATAL` / `PANIC` / `CRITICAL` are kept; lines with an `*Error` / `*Exception` class are kept; other lines (including ones that merely mention the word "error") are dropped. Optional config strings `line_must_match` / `line_must_not_match` further filter (non-empty regexes; invalid values are rejected at load). Filtered lines advance the cursor but do not insert `error.seen`. -4. Redacts secrets and obvious personal data **before** any row is written. +4. Redacts secrets and obvious personal data **before** any row is written. Separately from the `$AGENT_HOME`-configured redaction, an OTel `trace_id`/`span_id`/`traceparent` value (logfmt, colon, JSON, or quoted-value form) is always stripped before hashing — those are per-occurrence random ids, not secrets, but leaving them in would make every occurrence of the same recurring error hash to a different stack signature and never dedupe. 5. Computes a fingerprint: service + error class + normalized stack signature + environment. 6. Inserts `error.seen` or **enriches** an existing **open** row with that fingerprint on this session (`count`, `last_seen`, optional extra excerpt, optional `line_fingerprint`). First insert knocks `da ist Post id `. Enrichment never knocks. After skip or a terminal implement task, the next match is a new `error.seen` (new id, knocks). 7. Payload holds a **sanitized** excerpt plus an optional pointer to raw evidence on this disk. It does not hold the full log dump. diff --git a/README.md b/README.md index dbcf15e..9e59283 100644 --- a/README.md +++ b/README.md @@ -97,14 +97,14 @@ agent watch assigned [--follow] # allowlisted assignments; needs `gh` and `$AGE agent watch errors # one scan; $AGENT_HOME/error-fix.json; no log host in this package agent watch error-fix # one scan; find-or-create implement task + isolated worktree agent supervise --session ID [--repo OWNER/REPO --number N] [--once|--follow] -# agent knock (daemon, no --once) polls grok-usage, pending, pr.merged, github pending, mail pending, errors, and error-fix every 60s +# agent knock (daemon, no --once) polls grok-usage, pending, pr.merged, github pending, mail pending, errors, and error-fix every 60s, then syncs (push + pull) with the hub when paired ``` `agent supervise` posts a short status line to Telegram when both `TELEGRAM_BOT_TOKEN` and `TELEGRAM_CHAT_ID` are set in the environment. The follow CLI does not ask closed questions. Working vs not working for paging is whether the Grok tmux session exists: it posts `not working` only when that session is gone, not when the prompt is idle between turns. The TUI working probe (`Thinking…`, `Waiting for response`, `Preparing …`, `[stop]`, `Esc:cancel`, `command still running`, queued `Enter to send now`) is for the follow loop, not for Telegram. A send failure is printed to stderr and does not stop the loop. Credentials stay out of git. The error-fix executor find-or-creates the implement task and isolated worktree; `agent github pending` still opens draft pull requests. -`agent watch grok-usage` uses the existing Grok login token from the Grok auth file, does not start a Grok session, and does not knock the TUI. Each `usage.snapshot` includes the account email, provider, and subscription tier. Under the device daemon, the knock child records those snapshots (and scans pending, `pr.merged`, github pending, mail pending, errors when `$AGENT_HOME/error-fix.json` exists, and pending `error.fix`) on the same interval. `agent daemon --install` / `--uninstall` manage the user service; `agent init` already installs and starts it. +`agent watch grok-usage` uses the existing Grok login token from the Grok auth file, does not start a Grok session, and does not knock the TUI. Each `usage.snapshot` includes the account email, provider, and subscription tier. Under the device daemon, the knock child records those snapshots (and scans pending, `pr.merged`, github pending, mail pending, errors when `$AGENT_HOME/error-fix.json` exists, and pending `error.fix`) on the same interval, then syncs (push + pull) with the hub when the device is paired. `agent daemon --install` / `--uninstall` manage the user service; `agent init` already installs and starts it. `agent watch assigned` reads `$AGENT_HOME/watch.json`: diff --git a/src/agent_cli/errors.py b/src/agent_cli/errors.py index 85d1147..72cf592 100644 --- a/src/agent_cli/errors.py +++ b/src/agent_cli/errors.py @@ -28,6 +28,11 @@ r'(?i)("[^"]*(?:password|secret|token|api[_-]?key|access[_-]?token|client[_-]?secret|authorization|passwd|access_key)[^"]*"\s*:\s*")[^"]*(")' ) _HEX = re.compile(r"\b[a-fA-F0-9]{20,}\b") +_OTEL_TRACE_ID = re.compile(r'(?i)(\btrace_id(?:\\?["\'])?\s*[:=]\s*(?:\\?["\'])?)[0-9a-fA-F]+\b') +_OTEL_SPAN_ID = re.compile(r'(?i)(\bspan_id(?:\\?["\'])?\s*[:=]\s*(?:\\?["\'])?)[0-9a-fA-F]+\b') +_OTEL_TRACEPARENT = re.compile( + r'(?i)(\btraceparent(?:\\?["\'])?\s*[:=]\s*(?:\\?["\'])?)[0-9a-fA-F]+-[0-9a-fA-F]+-[0-9a-fA-F]+-[0-9a-fA-F]+\b' +) _SECRET = re.compile( r"(?i)(? str: out = _AKIA.sub("[redacted]", out) out = _JWT.sub("[redacted]", out) out = _EMAIL.sub("[redacted]", out) + out = _OTEL_TRACE_ID.sub(r"\1[redacted]", out) + out = _OTEL_SPAN_ID.sub(r"\1[redacted]", out) + out = _OTEL_TRACEPARENT.sub(r"\1[redacted]", out) out = _HEX.sub("[redacted]", out) return out diff --git a/src/agent_cli/hub.py b/src/agent_cli/hub.py index 43b2282..fc4ce38 100644 --- a/src/agent_cli/hub.py +++ b/src/agent_cli/hub.py @@ -2,6 +2,7 @@ from __future__ import annotations +import json from typing import Any from urllib.parse import urljoin @@ -40,7 +41,17 @@ def request(self, method: str, path: str, **kwargs: Any) -> Any: detail = _detail(response) raise HubError(f"hub {method} {path} → HTTP {response.status_code}: {detail}") if response.content: - return response.json() + try: + return response.json() + except (json.JSONDecodeError, UnicodeDecodeError, ValueError, RecursionError) as exc: + # RecursionError: CPython's C-accelerated json decoder still + # bounds recursion by C stack depth (Py_EnterRecursiveCall), + # not just sys.getrecursionlimit() - a pathologically nested + # body (adversarial or buggy hub) hits it well before running + # out of memory. Confirmed empirically: json.loads('[' * n + + # ']' * n) raises RecursionError around n=1_000_000, not + # JSONDecodeError, so it needs its own arm in this tuple. + raise HubError(f"hub {method} {path} → invalid JSON response") from exc return None def prepare(self, device_id: str, challenge: str, device_name: str) -> dict[str, Any]: @@ -105,7 +116,7 @@ def connect_sync_ws(self) -> Any: def _detail(response: httpx.Response) -> str: try: body = response.json() - except ValueError: + except (ValueError, RecursionError): return response.text if isinstance(body, dict) and "detail" in body: return str(body["detail"]) diff --git a/src/agent_cli/main.py b/src/agent_cli/main.py index f85449b..54c7b2f 100644 --- a/src/agent_cli/main.py +++ b/src/agent_cli/main.py @@ -13,6 +13,8 @@ import time import uuid import webbrowser +from collections.abc import Callable +from datetime import datetime from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any @@ -36,6 +38,7 @@ from .lane import LANE_ROLES, LANE_VENDORS, LaneResult, launch from .pg import PgError, cluster_exists, cluster_running, ensure_cluster, require_loopback_dsn, stop_cluster from .runtime import ( + Completed, Runtime, grok_model, grok_new_session_id, @@ -44,7 +47,7 @@ tmux_name, ) from .skills import SKILL_NAMES, has_skill, skill_for_agent_role -from .store import Store, StoreConnectionError, StoreError, utcnow +from .store import OWNED_TABLES, Store, StoreConnectionError, StoreError, utcnow from .usage import AuthStale, scan_usage, usage_poll_due from .watch import ( assigned_session_id, @@ -1451,7 +1454,7 @@ def _run_sync_ws_session( for raw in ws: try: message = json.loads(raw) - except (TypeError, ValueError): + except (TypeError, ValueError, RecursionError): continue if not isinstance(message, dict): continue @@ -1462,7 +1465,9 @@ def _run_sync_ws_session( rows = message.get("rows") if isinstance(rows, list): for row in rows: - if not isinstance(row, dict) or not row.get("table"): + try: + row = _check_pull_row(row) + except _PullShapeError: continue try: store.apply_replica_row(row) @@ -1489,6 +1494,8 @@ def cmd_restore(_: list[str]) -> None: body = hub.restore() finally: hub.close() + if not isinstance(body, dict): + die("restore response is not an object") if body.get("device_id") != store.device_id(): die("restore device_id does not match this device") if "own_events" in body: @@ -1497,14 +1504,47 @@ def cmd_restore(_: list[str]) -> None: events = body.get("events") if not isinstance(events, list): die("restore response missing own_events") + coerced_events: list[dict[str, Any]] = [] for event in events: - store.apply_remote(event, wake=False) - store.mark_origin(event["origin_device_id"], int(event["origin_seq"])) - snapshots = list(body.get("inbox") or []) + list(body.get("pings") or []) + try: + coerced_events.append(_coerce_pull_event(event, store.device_id())) + except _PullShapeError as exc: + die(f"restore {exc}") + snapshots: list[dict[str, Any]] = [] + for key in ("inbox", "pings"): + value = body.get(key) + if value is None: + continue + if not isinstance(value, list): + die(f"restore response {key} is not a list") + snapshots.extend(value) + for row in snapshots: + try: + _check_pull_row(row) + except _PullShapeError as exc: + die(f"restore {exc}") + # Every event and snapshot's SHAPE is validated above before either + # is applied here - a malformed snapshot must not be discovered only + # after well-formed events ahead of it in the response are already + # durably committed and their origin cursor advanced. This does not + # cover semantic conflicts (an origin_seq gap, a foreign row- + # ownership conflict): those can only be checked against live store + # state at apply time, inside each item's own transaction, so a + # batch that's shape-valid throughout can still partially commit + # before a later semantic conflict is discovered. Closing that would + # need one transaction spanning the whole apply loop, a bigger + # change than this fix - tracked separately. + for event in coerced_events: + try: + store.apply_remote(event, wake=False) + store.mark_origin(event["origin_device_id"], event["origin_seq"]) + except Exception as exc: + die(f"restore event could not be applied: {exc}") for row in snapshots: - if not isinstance(row, dict): - die("restore snapshot is not an object") - store.apply_replica_row(row, wake=False) + try: + store.apply_replica_row(row, wake=False) + except Exception as exc: + die(f"restore snapshot could not be applied: {exc}") print(f"restored events={len(events)} snapshots={len(snapshots)}") finally: store.close() @@ -1725,7 +1765,112 @@ def lookup_session() -> dict | None: store.close() +_PULL_EVENT_FIELDS = ("origin_device_id", "origin_seq", "table", "op", "row_id", "payload", "occurred_at") +_PULL_ROW_FIELDS = ("table", "origin_device_id", "row_id", "payload", "updated_at") + + +class _PullShapeError(ValueError): + """A hub-supplied event/row/response doesn't have the shape callers need. + Internal only: every caller catches this and converts it to whatever error + convention fits that call site (HubError for _sync_once, die() for the + one-shot cmd_restore CLI) - it must never itself propagate out of this + module.""" + + +def _coerce_pull_event(event: object, own_device_id: str) -> dict[str, Any]: + """Validate a pulled/restored event has every field _insert_event_idempotent + indexes, and that table/op are shaped the way Store._write_in_txn already + requires for this device's own local writes (op/table checked there; the + hub-pull/restore path must not be laxer). payload is additionally + required to be an object here even though _write_in_txn doesn't check + that for local writes either - a payload that isn't one is otherwise + stored as-is and only fails later, on every future read of that whole + table, not at write time - a risk specific to externally-supplied hub + data, not to this codebase's own trusted call sites. Normalize + origin_seq to an int (rejecting bool, a fractional float, and anything + int() can't convert, including an out-of-range float that would + otherwise raise OverflowError). occurred_at must actually parse as a + timestamp, same reasoning and check as _check_pull_row's updated_at (a + bogus value would otherwise be stored as-is - apply_remote writes it + into row_data.updated_at too, via _materialize). Also validates origin_device_id equals + own_device_id: DESIGN.md's sync contract is "own events, gapless" - + foreign-origin data arrives as row snapshots, never as an event + (apply_replica_row already enforces the row-side half of this, ignoring + a same-device non-ping snapshot rather than applying it) - so a + pulled/restored event claiming a foreign origin_device_id is a + malformed hub response, not a normal case apply_remote/mark_origin + should accept. row_id must be a non-empty string too, same reasoning as + every other field here - an unchecked value (e.g. a list) would only + fail later, as a raw type error from whatever stores it. Returns a new + dict; the caller's own copy of the raw event is left untouched.""" + if not isinstance(event, dict) or any(field not in event for field in _PULL_EVENT_FIELDS): + raise _PullShapeError("event is missing required fields") + if event["origin_device_id"] != own_device_id: + raise _PullShapeError("event origin_device_id is not this device's own") + if not isinstance(event["table"], str) or event["table"] not in OWNED_TABLES: + raise _PullShapeError("event has an unknown table") + if not isinstance(event["row_id"], str) or event["row_id"] == "": + raise _PullShapeError("event row_id is not a valid id") + if not isinstance(event["payload"], dict): + raise _PullShapeError("event payload is not an object") + if event["op"] not in ("insert", "update", "delete"): + raise _PullShapeError("event has an unknown op") + if not isinstance(event["occurred_at"], str): + raise _PullShapeError("event occurred_at is not a valid timestamp") + try: + datetime.fromisoformat(re.sub(r"[Zz]$", "+00:00", event["occurred_at"])) + except ValueError as exc: + raise _PullShapeError("event occurred_at is not a valid timestamp") from exc + raw_seq = event["origin_seq"] + try: + if isinstance(raw_seq, bool): + raise ValueError("origin_seq must not be a boolean") + if isinstance(raw_seq, float) and not raw_seq.is_integer(): + raise ValueError("origin_seq must be a whole number") + origin_seq = int(raw_seq) + except (TypeError, ValueError, OverflowError) as exc: + raise _PullShapeError("event has a non-numeric origin_seq") from exc + return {**event, "origin_seq": origin_seq} + + +def _check_pull_row(row: object) -> dict[str, Any]: + """Validate a pulled/restored snapshot row has every field + apply_replica_row indexes directly, and that payload is an object - same + reasoning as _coerce_pull_event: an unvalidated non-object payload would + otherwise be stored as-is and only fail later, on every future read of + that whole table. updated_at additionally must actually parse as a + timestamp: it's stored as-is on first insert (the row_data upsert only + compares updated_at against an existing row on conflict), so a bogus + value isn't rejected until some later write to that same row fails its + ::timestamptz cast - by which point the row is already stuck with a + value no legitimate update can pass the "newer than" check against. + row_id and origin_device_id must be non-empty strings too, same + reasoning - an unchecked value would only fail later, as a raw type + error from whatever stores it.""" + if not isinstance(row, dict) or any(field not in row for field in _PULL_ROW_FIELDS): + raise _PullShapeError("snapshot is missing required fields") + if not isinstance(row["table"], str) or row["table"] not in OWNED_TABLES: + raise _PullShapeError("snapshot has an unknown table") + if not isinstance(row["row_id"], str) or row["row_id"] == "": + raise _PullShapeError("snapshot row_id is not a valid id") + if not isinstance(row["origin_device_id"], str) or row["origin_device_id"] == "": + raise _PullShapeError("snapshot origin_device_id is not a valid id") + if not isinstance(row["payload"], dict): + raise _PullShapeError("snapshot payload is not an object") + if not isinstance(row["updated_at"], str): + raise _PullShapeError("snapshot updated_at is not a valid timestamp") + try: + datetime.fromisoformat(re.sub(r"[Zz]$", "+00:00", row["updated_at"])) + except ValueError as exc: + raise _PullShapeError("snapshot updated_at is not a valid timestamp") from exc + return row + + def _sync_once(store: Store) -> None: + # Every malformed-hub-response check below raises HubError (not die()'s bare + # SystemExit): both cmd_knock's _knock_scan_cycle and cmd_sync --follow's + # reconnect loop already catch HubError, so a bad response logs/retries in + # whichever loop is calling instead of killing that process. hub = _hub_from_store(store) try: pending = store.pending_events() @@ -1733,24 +1878,59 @@ def _sync_once(store: Store) -> None: hub.push(pending) store.mark_pushed(pending[-1]["origin_seq"]) pulled = hub.pull(store.all_cursors()) + if not isinstance(pulled, dict): + raise HubError("pull response is not an object") events = pulled.get("events") if not isinstance(events, list): - die("pull response missing events") + raise HubError("pull response missing events") + coerced_events: list[dict[str, Any]] = [] for event in events: - store.apply_remote(event) - store.mark_origin(event["origin_device_id"], int(event["origin_seq"])) - snapshots = ( - list(pulled.get("inbox") or []) - + list(pulled.get("pings") or []) - + list(pulled.get("subscriptions") or []) - ) + try: + coerced_events.append(_coerce_pull_event(event, store.device_id())) + except _PullShapeError as exc: + raise HubError(f"pull {exc}") from exc + snapshots: list[dict[str, Any]] = [] + for key in ("inbox", "pings", "subscriptions"): + value = pulled.get(key) + if value is None: + continue + if not isinstance(value, list): + raise HubError(f"pull response {key} is not a list") + snapshots.extend(value) + for row in snapshots: + try: + _check_pull_row(row) + except _PullShapeError as exc: + raise HubError(f"pull {exc}") from exc + # Every event and snapshot's SHAPE is validated above before either + # is applied here - a malformed snapshot must not be discovered only + # after well-formed events ahead of it in the response are already + # durably committed and their origin cursor advanced. This does not + # cover semantic conflicts (an origin_seq gap, a foreign row- + # ownership conflict): those can only be checked against live store + # state at apply time, inside each item's own transaction, so a + # batch that's shape-valid throughout can still partially commit + # before a later semantic conflict is discovered. Closing that would + # need one transaction spanning the whole apply loop, a bigger + # change than this fix - tracked separately. + for event in coerced_events: + # Field presence and origin_seq are validated above, but not the + # shape of nested values (e.g. payload["type"]) - apply_remote/ + # mark_origin can still hit a genuinely unanticipated shape deep + # inside store.py. Convert any such failure to a HubError rather + # than let it crash whichever loop called _sync_once; HubError/ + # StoreError themselves pass through unchanged (SystemExit is not + # an Exception subclass). + try: + store.apply_remote(event) + store.mark_origin(event["origin_device_id"], event["origin_seq"]) + except Exception as exc: + raise HubError(f"pull event could not be applied: {exc}") from exc for row in snapshots: - if not isinstance(row, dict): - die("pull snapshot is not an object") - sessions = [r for r in snapshots if isinstance(r, dict) and r.get("table") == "session"] - rest = [r for r in snapshots if not (isinstance(r, dict) and r.get("table") == "session")] - for row in sessions + rest: - store.apply_replica_row(row) + try: + store.apply_replica_row(row) + except Exception as exc: + raise HubError(f"pull snapshot could not be applied: {exc}") from exc print(f"sync pushed={len(pending)} pulled={len(events)} snapshots={len(snapshots)}") finally: hub.close() @@ -2904,6 +3084,75 @@ def cmd_lane(args: list[str]) -> None: raise SystemExit(2) +def _knock_scan_cycle(store: Store, run_argv: Callable[[list[str]], Completed]) -> None: + from .pending import scan_pending + + try: + usage_id = scan_usage(store) + if usage_id: + print(f"usage.snapshot {usage_id}") + except AuthStale: + pass + except StoreError as exc: + print(f"usage.snapshot error: {exc}", file=sys.stderr) + try: + created, skipped = scan_merged(store, run_argv) + for activity_id in created: + print(f"pr.merged {activity_id}") + if skipped: + print(f"watch skipped {skipped} pr.open rows", file=sys.stderr) + except StoreError as exc: + print(f"pr.merged error: {exc}", file=sys.stderr) + hub_url = store.meta("hub_url") + hub_token = store.meta("device_token") + if hub_url and hub_token: + hub = Hub(hub_url, hub_token) + try: + lines = scan_pending(store, hub) + for line in lines: + print(line) + except (HubError, StoreError) as exc: + print(f"pending error: {exc}", file=sys.stderr) + finally: + hub.close() + from .github_act import scan_github + from .mail_act import scan_mail + + try: + for line in scan_github(store, run_argv): + print(line) + except StoreError as exc: + print(f"github pending error: {exc}", file=sys.stderr) + try: + for line in scan_mail(store, run_argv): + print(line) + except StoreError as exc: + print(f"mail pending error: {exc}", file=sys.stderr) + from .errors import config_path, default_fetch, scan_errors + + if config_path(store.home).is_file(): + try: + created, enriched = scan_errors(store, default_fetch) + for activity_id in created: + print(f"error.seen {activity_id}") + for activity_id in enriched: + print(f"error.seen enrich {activity_id}") + except StoreError as exc: + print(f"error.seen error: {exc}", file=sys.stderr) + from .error_fix_act import scan_error_fix + + try: + for line in scan_error_fix(store, run_argv): + print(line) + except StoreError as exc: + print(f"error.fix error: {exc}", file=sys.stderr) + if hub_url and hub_token: + try: + _sync_once(store) + except (HubError, StoreError) as exc: + print(f"sync error: {exc}", file=sys.stderr) + + def cmd_knock(args: list[str]) -> None: once = "--once" in args store = open_store() @@ -2913,71 +3162,12 @@ def cmd_knock(args: list[str]) -> None: for activity_id, status in knock_drain(store, runtime): print(f"knock {activity_id} {status}") return - from .pending import scan_pending from .runtime import run_argv last_poll: float | None = None while True: if usage_poll_due(last_poll, time.monotonic()): - try: - usage_id = scan_usage(store) - if usage_id: - print(f"usage.snapshot {usage_id}") - except AuthStale: - pass - except StoreError as exc: - print(f"usage.snapshot error: {exc}", file=sys.stderr) - try: - created, skipped = scan_merged(store, run_argv) - for activity_id in created: - print(f"pr.merged {activity_id}") - if skipped: - print(f"watch skipped {skipped} pr.open rows", file=sys.stderr) - except StoreError as exc: - print(f"pr.merged error: {exc}", file=sys.stderr) - hub_url = store.meta("hub_url") - hub_token = store.meta("device_token") - if hub_url and hub_token: - hub = Hub(hub_url, hub_token) - try: - lines = scan_pending(store, hub) - for line in lines: - print(line) - except (HubError, StoreError) as exc: - print(f"pending error: {exc}", file=sys.stderr) - finally: - hub.close() - from .github_act import scan_github - from .mail_act import scan_mail - - try: - for line in scan_github(store, run_argv): - print(line) - except StoreError as exc: - print(f"github pending error: {exc}", file=sys.stderr) - try: - for line in scan_mail(store, run_argv): - print(line) - except StoreError as exc: - print(f"mail pending error: {exc}", file=sys.stderr) - from .errors import config_path, default_fetch, scan_errors - - if config_path(store.home).is_file(): - try: - created, enriched = scan_errors(store, default_fetch) - for activity_id in created: - print(f"error.seen {activity_id}") - for activity_id in enriched: - print(f"error.seen enrich {activity_id}") - except StoreError as exc: - print(f"error.seen error: {exc}", file=sys.stderr) - from .error_fix_act import scan_error_fix - - try: - for line in scan_error_fix(store, run_argv): - print(line) - except StoreError as exc: - print(f"error.fix error: {exc}", file=sys.stderr) + _knock_scan_cycle(store, run_argv) last_poll = time.monotonic() activity_id = knock_listen(store, runtime, timeout=30.0) if activity_id: diff --git a/src/agent_cli/store.py b/src/agent_cli/store.py index 3c5030a..e63c08d 100644 --- a/src/agent_cli/store.py +++ b/src/agent_cli/store.py @@ -355,7 +355,8 @@ def apply_remote(self, event: dict[str, Any], *, wake: bool = True) -> None: self._maybe_wake(event) elif inserted: payload = event.get("payload") - if isinstance(payload, dict) and payload.get("type") in WAKE_ACTIVITY_TYPES: + typ = payload.get("type") if isinstance(payload, dict) else None + if isinstance(typ, str) and typ in WAKE_ACTIVITY_TYPES: target = self._inbox_target(payload) if target is not None and self._owns_session(target): self.enqueue_wake(event["row_id"], target) @@ -494,7 +495,8 @@ def apply_replica_row(self, row: dict[str, Any], *, wake: bool = True) -> None: self._maybe_wake(event) else: payload = event.get("payload") - if isinstance(payload, dict) and payload.get("type") in WAKE_ACTIVITY_TYPES: + typ = payload.get("type") if isinstance(payload, dict) else None + if isinstance(typ, str) and typ in WAKE_ACTIVITY_TYPES: target = self._inbox_target(payload) if target is not None and self._owns_session(target): self.enqueue_wake(event["row_id"], target) @@ -522,7 +524,14 @@ def pending_events(self) -> list[dict[str, Any]]: @_wrap_pg_errors def mark_pushed(self, seq: int) -> None: - self.sync_set("pushed_origin_seq", str(seq)) + # Read-then-write is not atomic across the separate cmd_knock/cmd_sync + # --follow processes; a fully concurrent interleaving can still lose an + # update. Accepted: ledger_event is append-only and the hub treats the + # same event id as idempotent, so the worst case is a redundant re-push, + # not data loss. mark_origin below has the identical shape. + current = int(self.sync_get("pushed_origin_seq", "0") or "0") + if seq > current: + self.sync_set("pushed_origin_seq", str(seq)) def origin_cursor(self, origin: str) -> int: raw = self.sync_get(f"origin:{origin}", "0") @@ -645,7 +654,8 @@ def pending_work(self) -> list[dict[str, Any]]: continue if payload.get("execution_status") != "pending": continue - if payload.get("type") not in EXECUTABLE_ACTIVITY_TYPES: + typ = payload.get("type") + if not isinstance(typ, str) or typ not in EXECUTABLE_ACTIVITY_TYPES: continue payload["_origin_device_id"] = row["origin_device_id"] out.append(payload) @@ -707,14 +717,12 @@ def _maybe_wake(self, event: dict[str, Any]) -> None: payload = event.get("payload") if not isinstance(payload, dict): return - if payload.get("type") not in WAKE_ACTIVITY_TYPES: + typ = payload.get("type") + if not isinstance(typ, str) or typ not in WAKE_ACTIVITY_TYPES: return - if event.get("op") == "update" and payload.get("type") == "error.seen": + if event.get("op") == "update" and typ == "error.seen": return - if ( - payload.get("type") in DONE_WAKE_ACTIVITY_TYPES - and payload.get("execution_status") != "done" - ): + if typ in DONE_WAKE_ACTIVITY_TYPES and payload.get("execution_status") != "done": return target = self._inbox_target(payload) if target is None: @@ -733,7 +741,8 @@ def _maybe_work(self, event: dict[str, Any]) -> None: return if payload.get("execution_status") != "pending": return - if payload.get("type") not in EXECUTABLE_ACTIVITY_TYPES: + typ = payload.get("type") + if not isinstance(typ, str) or typ not in EXECUTABLE_ACTIVITY_TYPES: return sid = payload.get("session_id") if not isinstance(sid, str) or sid == "": diff --git a/tests/test_errors.py b/tests/test_errors.py index 04b0b6c..53ba2fb 100644 --- a/tests/test_errors.py +++ b/tests/test_errors.py @@ -168,6 +168,91 @@ def test_redact_and_fingerprint() -> None: fp = fingerprint(service="api", error_class="TimeoutError", stack_sig=sig, environment="prod") assert fp.startswith("api|TimeoutError|") assert fp.endswith("|prod") + trace_id = "a" * 32 + span_id = "b" * 16 + otel = redact(f"TimeoutError boom trace_id={trace_id} span_id={span_id}") + assert trace_id not in otel + assert span_id not in otel + assert "trace_id=[redacted]" in otel + assert "span_id=[redacted]" in otel + otel_upper = redact(f"TimeoutError boom TRACE_ID={trace_id.upper()} SPAN_ID={span_id.upper()}") + assert trace_id.upper() not in otel_upper + assert span_id.upper() not in otel_upper + traceparent = redact(f"TimeoutError boom traceparent: 00-{trace_id}-{span_id}-01") + assert trace_id not in traceparent + assert span_id not in traceparent + assert "traceparent: [redacted]" in traceparent + traceparent_eq = redact(f"TimeoutError boom traceparent=00-{trace_id}-{span_id}-01") + assert trace_id not in traceparent_eq + assert span_id not in traceparent_eq + assert "traceparent=[redacted]" in traceparent_eq + otel_colon = redact(f"TimeoutError boom trace_id: {trace_id} span_id: {span_id}") + assert trace_id not in otel_colon + assert span_id not in otel_colon + assert "trace_id: [redacted]" in otel_colon + assert "span_id: [redacted]" in otel_colon + otel_json = redact(f'TimeoutError boom "trace_id":"{trace_id}","span_id":"{span_id}"') + assert trace_id not in otel_json + assert span_id not in otel_json + assert '"trace_id":"[redacted]"' in otel_json + assert '"span_id":"[redacted]"' in otel_json + otel_json_spaced = redact(f'TimeoutError boom "trace_id": "{trace_id}", "span_id": "{span_id}"') + assert trace_id not in otel_json_spaced + assert span_id not in otel_json_spaced + assert '"trace_id": "[redacted]"' in otel_json_spaced + assert '"span_id": "[redacted]"' in otel_json_spaced + otel_quoted_value = redact(f'TimeoutError boom trace_id="{trace_id}" span_id="{span_id}"') + assert trace_id not in otel_quoted_value + assert span_id not in otel_quoted_value + assert 'trace_id="[redacted]"' in otel_quoted_value + assert 'span_id="[redacted]"' in otel_quoted_value + # Docker json-file log driver wraps an application's JSON log line inside its + # own "log" field, escaping the inner quotes - a very plausible real shape + # for a containerized service's logs, not a hypothetical one. + otel_escaped_json = redact( + '{"log":"...{\\"trace_id\\":\\"' + trace_id + '\\",\\"span_id\\":\\"' + span_id + '\\"}..."}' + ) + assert trace_id not in otel_escaped_json + assert span_id not in otel_escaped_json + assert '\\"trace_id\\":\\"[redacted]\\"' in otel_escaped_json + assert '\\"span_id\\":\\"[redacted]\\"' in otel_escaped_json + traceparent_value = f"00-{trace_id}-{span_id}-01" + traceparent_json = redact(f'TimeoutError boom "traceparent":"{traceparent_value}"') + assert trace_id not in traceparent_json + assert span_id not in traceparent_json + assert '"traceparent":"[redacted]"' in traceparent_json + traceparent_json_spaced = redact(f'TimeoutError boom "traceparent": "{traceparent_value}"') + assert trace_id not in traceparent_json_spaced + assert span_id not in traceparent_json_spaced + assert '"traceparent": "[redacted]"' in traceparent_json_spaced + traceparent_quoted_value = redact(f'TimeoutError boom traceparent="{traceparent_value}"') + assert trace_id not in traceparent_quoted_value + assert span_id not in traceparent_quoted_value + assert 'traceparent="[redacted]"' in traceparent_quoted_value + traceparent_escaped_json = redact('{"log":"...{\\"traceparent\\":\\"' + traceparent_value + '\\"}..."}') + assert trace_id not in traceparent_escaped_json + assert span_id not in traceparent_escaped_json + assert '\\"traceparent\\":\\"[redacted]\\"' in traceparent_escaped_json + + +def test_redact_strips_otel_ids_of_non_standard_length() -> None: + """Regression test: _OTEL_TRACE_ID/_OTEL_SPAN_ID required an exact + 32/16-char hex value. A malformed or non-conformant id (e.g. a span_id + of 17-19 hex chars) falls between that exact match and the generic + _HEX fallback's 20-char floor, so it used to pass through unredacted - + undermining the "always stripped before hashing" dedup guarantee for + exactly the malformed values most likely to vary occurrence to + occurrence. The label match already does the real specificity work, so + matching the value's actual length instead of requiring the standard + one only helps.""" + odd_span_id = "b" * 18 + otel = redact(f"TimeoutError boom trace_id={'a' * 32} span_id={odd_span_id}") + assert odd_span_id not in otel + assert "span_id=[redacted]" in otel + short_trace_id = "a" * 10 + otel_short_trace = redact(f"TimeoutError boom trace_id={short_trace_id} span_id={'b' * 16}") + assert short_trace_id not in otel_short_trace + assert "trace_id=[redacted]" in otel_short_trace def test_scan_inserts_once_then_enriches(tmp_path: Path) -> None: @@ -208,6 +293,40 @@ def fetch(_cfg: dict, _cursor: str | None) -> tuple[list[dict], str | None]: assert "line_fingerprint" not in payload +def test_scan_dedups_across_different_otel_ids(tmp_path: Path) -> None: + store = Store(tmp_path) + _runner_session(store) + _write_config(tmp_path) + line1 = ( + "TimeoutError boom " + "trace_id=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa " + "span_id=bbbbbbbbbbbbbbbb" + ) + line2 = ( + "TimeoutError boom " + "trace_id=cccccccccccccccccccccccccccccccc " + "span_id=dddddddddddddddd" + ) + calls = {"n": 0} + + def fetch(_cfg: dict, _cursor: str | None) -> tuple[list[dict], str | None]: + calls["n"] += 1 + if calls["n"] == 1: + return ([{"ts": "2026-08-23T16:00:00Z", "line": line1}], "2026-08-23T16:00:00Z") + return ([{"ts": "2026-08-23T16:00:01Z", "line": line2}], "2026-08-23T16:00:01Z") + + created, enriched = scan_errors(store, fetch) + assert enriched == [] + assert len(created) == 1 + + created2, enriched2 = scan_errors(store, fetch) + assert created2 == [] + assert enriched2 == created + again = store.row("activity", created[0]) + assert again is not None + assert again["payload"]["count"] == 2 + + def test_line_fingerprint_is_sha256_of_server_container_line() -> None: import hashlib diff --git a/tests/test_hub.py b/tests/test_hub.py new file mode 100644 index 0000000..7712e98 --- /dev/null +++ b/tests/test_hub.py @@ -0,0 +1,70 @@ +from __future__ import annotations + +import httpx +import pytest + +from agent_cli.hub import Hub, HubError + + +def test_request_invalid_json_response_raises_hub_error_not_json_decode_error() -> None: + """Regression test: a 2xx response with a non-JSON body used to leak a raw + json.JSONDecodeError out of Hub.request, which callers like _sync_once (via + hub.pull) do not catch, killing whatever loop called it instead of logging a + HubError like every other hub-communication failure.""" + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, text="not-json") + + with httpx.Client(transport=httpx.MockTransport(handler)) as client: + hub = Hub("https://hub.example", "tok", client=client) + with pytest.raises(HubError, match="invalid JSON"): + hub.pull({}) + + +def test_request_recursion_error_response_raises_hub_error_not_recursion_error( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression test: CPython's C-accelerated json decoder still bounds + recursion by C stack depth, not just sys.getrecursionlimit() - a + pathologically deeply-nested response body (adversarial or buggy hub) + can raise RecursionError, a RuntimeError subclass the previous except + tuple (JSONDecodeError/UnicodeDecodeError/ValueError) didn't catch. That + would have escaped every caller's HubError/StoreError guard the same way + a raw json.JSONDecodeError used to before this method existed.""" + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, text="[1]") + + def raise_recursion_error(self: httpx.Response) -> None: + raise RecursionError("Stack overflow while decoding a JSON array") + + monkeypatch.setattr(httpx.Response, "json", raise_recursion_error) + with httpx.Client(transport=httpx.MockTransport(handler)) as client: + hub = Hub("https://hub.example", "tok", client=client) + with pytest.raises(HubError, match="invalid JSON"): + hub.pull({}) + + +def test_request_recursion_error_on_error_response_body_falls_back_to_text( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression test: the row-side sibling of the test above, for _detail() + - called only for an HTTP >= 400 response to build the error message. + _detail() had its own separate response.json() call with its own except + tuple (bare ValueError), never updated when the success-path tuple above + gained RecursionError. A >= 400 response with a pathologically nested + error body used to crash Hub.request with an uncaught RecursionError + instead of falling back to the raw response text like every other + unparseable error body already does.""" + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(400, text="bad request, raw text") + + def raise_recursion_error(self: httpx.Response) -> None: + raise RecursionError("Stack overflow while decoding a JSON array") + + monkeypatch.setattr(httpx.Response, "json", raise_recursion_error) + with httpx.Client(transport=httpx.MockTransport(handler)) as client: + hub = Hub("https://hub.example", "tok", client=client) + with pytest.raises(HubError, match="bad request, raw text"): + hub.pull({}) diff --git a/tests/test_knock_scan_cycle.py b/tests/test_knock_scan_cycle.py new file mode 100644 index 0000000..5ee548d --- /dev/null +++ b/tests/test_knock_scan_cycle.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from agent_cli import main as main_mod +from agent_cli.hub import HubError +from agent_cli.main import open_store +from agent_cli.runtime import Completed + + +def _init_paired_store(tmp_path: Path) -> None: + os.environ["AGENT_HOME"] = str(tmp_path) + main_mod.main(["init"]) + store = open_store() + try: + store.set_meta("hub_url", "https://hub.example") + store.set_meta("device_token", "fake-token") + finally: + store.close() + + +def _stub_scans(monkeypatch: pytest.MonkeyPatch, order: list[str] | None = None) -> None: + log = order if order is not None else [] + + def scan_usage(store: object) -> None: + log.append("scan_usage") + + def scan_merged(store: object, run_argv: object) -> tuple[list[str], int]: + log.append("scan_merged") + return ([], 0) + + def scan_pending(store: object, hub: object) -> list[str]: + log.append("scan_pending") + return [] + + def scan_github(store: object, run_argv: object) -> list[str]: + log.append("scan_github") + return [] + + def scan_mail(store: object, run_argv: object) -> list[str]: + log.append("scan_mail") + return [] + + def scan_error_fix(store: object, run_argv: object) -> list[str]: + log.append("scan_error_fix") + return [] + + monkeypatch.setattr(main_mod, "scan_usage", scan_usage) + monkeypatch.setattr(main_mod, "scan_merged", scan_merged) + monkeypatch.setattr("agent_cli.pending.scan_pending", scan_pending) + monkeypatch.setattr("agent_cli.github_act.scan_github", scan_github) + monkeypatch.setattr("agent_cli.mail_act.scan_mail", scan_mail) + monkeypatch.setattr("agent_cli.error_fix_act.scan_error_fix", scan_error_fix) + + +def test_knock_scan_cycle_syncs_when_paired( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Sync must run once, and after every scan - not just alongside them - since + it is meant to push whatever those scans just created.""" + order: list[str] = [] + monkeypatch.setattr(main_mod, "_sync_once", lambda store: order.append("sync")) + _stub_scans(monkeypatch, order) + _init_paired_store(tmp_path) + + store = open_store() + try: + main_mod._knock_scan_cycle(store, lambda _argv: Completed(0, "", "")) + finally: + store.close() + + assert order[-1] == "sync" + assert order.count("sync") == 1 + assert set(order[:-1]) == { + "scan_usage", + "scan_merged", + "scan_pending", + "scan_github", + "scan_mail", + "scan_error_fix", + } + + +def test_knock_scan_cycle_skips_sync_when_unpaired( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + calls: list[int] = [] + monkeypatch.setattr(main_mod, "_sync_once", lambda store: calls.append(1)) + _stub_scans(monkeypatch) + + os.environ["AGENT_HOME"] = str(tmp_path) + main_mod.main(["init"]) + + store = open_store() + try: + main_mod._knock_scan_cycle(store, lambda _argv: Completed(0, "", "")) + finally: + store.close() + + assert calls == [] + + +def test_knock_scan_cycle_logs_and_continues_on_malformed_pull_response( + tmp_path: Path, capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: _sync_once() raises HubError when the hub returns a + malformed pull payload. A narrow except that only caught (HubError, StoreError) + still covers this - HubError is exactly what the malformed-response checks + raise - so the daemon logs and moves on instead of dying. _sync_once's + _hub_from_store() call can still raise a bare SystemExit ("device is not + paired") if pairing is ever revoked concurrently with this call, but no + code path in this repo clears hub_url/device_token once set (verified: + they're write-once, only ever set by cmd_pair) - not reachable today, only + a future concern if an unpair command is ever added.""" + + def _raise(_store: object) -> None: + raise HubError("pull response missing events") + + monkeypatch.setattr(main_mod, "_sync_once", _raise) + _stub_scans(monkeypatch) + _init_paired_store(tmp_path) + capsys.readouterr() + + store = open_store() + try: + main_mod._knock_scan_cycle(store, lambda _argv: Completed(0, "", "")) + finally: + store.close() + + captured = capsys.readouterr() + assert "sync error: pull response missing events" in captured.err diff --git a/tests/test_pending.py b/tests/test_pending.py index 6d08c30..755f598 100644 --- a/tests/test_pending.py +++ b/tests/test_pending.py @@ -247,6 +247,532 @@ def test_sync_once_applies_subscription_snapshots(tmp_path: Path, monkeypatch: p assert row["_origin_device_id"] == "other-device" +def _paired_store(tmp_path: Path) -> Store: + store = Store(tmp_path) + store.set_meta("hub_url", "http://hub.example") + store.set_meta("device_token", "tok") + return store + + +def test_sync_once_raises_hub_error_on_non_dict_pull_response(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test: Hub.request returns None for a 2xx response with an empty + body. _sync_once used to call pulled.get("events") straight on that, raising a + raw AttributeError instead of a catchable HubError - which would have escaped + both _knock_scan_cycle's and cmd_sync --follow's guards and killed whichever + process called it.""" + hub = FakeHub() + hub.pull_body = None # type: ignore[assignment] + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull response is not an object"): + _sync_once(_paired_store(tmp_path)) + + +def _valid_pull_event(device_id: str) -> dict[str, Any]: + return { + "origin_device_id": device_id, + "origin_seq": 1, + "table": "activity", + "op": "insert", + "row_id": "x", + "payload": {}, + "occurred_at": "2026-08-13T12:00:00Z", + } + + +@pytest.mark.parametrize( + "field", + ["origin_device_id", "origin_seq", "table", "op", "row_id", "payload", "occurred_at"], +) +def test_sync_once_raises_hub_error_on_event_missing_one_required_field( + field: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: a pull event missing any single one of the fields + _insert_event_idempotent indexes directly used to raise a raw KeyError deep + inside store.apply_remote instead of a catchable HubError raised before that + call. Parametrized per field so a future accidental narrowing of + _PULL_EVENT_FIELDS to any one of them is still caught.""" + store = _paired_store(tmp_path) + event = _valid_pull_event(store.device_id()) + del event[field] + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event is missing required fields"): + _sync_once(store) + + +def test_sync_once_raises_hub_error_on_event_with_foreign_origin_device_id( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: DESIGN.md's sync contract is "own events, gapless" - + foreign-origin data arrives as row snapshots, never as an event. An event + whose origin_device_id doesn't match this device's own used to pass every + check and reach apply_remote/mark_origin unchanged, letting a malformed + hub response poison this device's own ledger under a foreign device's + identity.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "origin_device_id": "other"} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="event origin_device_id is not this device's own"): + _sync_once(store) + assert store.rows("activity") == [] + + +def test_sync_once_raises_hub_error_on_event_with_non_dict_payload( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: field presence was checked but not payload's type. + A non-dict payload (e.g. a JSON list) used to pass validation untouched, + get committed by apply_remote (Store._maybe_wake/_maybe_work each already + return early on a non-dict payload, so nothing rolls back the write), and + only fail later - on every future store.rows()/store.row() call for that + whole table, not at write time. Store._write_in_txn doesn't check this + shape either for local writes, but that's this codebase's own trusted + code constructing payloads, not externally-supplied hub data - the risk + this test guards against is specific to the pull path.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "payload": ["not", "an", "object"]} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event payload is not an object"): + _sync_once(store) + assert store.origin_cursor(store.device_id()) == 0 + assert store.rows("activity") == [] + + +def test_sync_once_raises_hub_error_on_event_with_unknown_op( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: sibling of the non-dict-payload test above, for op. + Store._write_in_txn rejects any op outside insert/update/delete for a + local write; the hub-pull path must reject it too, not silently accept + and materialize a row under an op nothing else in the codebase expects.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "op": "bogus"} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event has an unknown op"): + _sync_once(store) + assert store.origin_cursor(store.device_id()) == 0 + assert store.rows("activity") == [] + + +def test_sync_once_raises_hub_error_on_event_with_unparseable_occurred_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the event-side sibling of + test_sync_once_raises_hub_error_on_snapshot_with_unparseable_updated_at. + occurred_at was presence-checked only, like updated_at used to be; a + bogus non-timestamp string used to pass validation and be stored as-is + by apply_remote (which writes it into row_data.updated_at too, via + _materialize).""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "occurred_at": "not-a-timestamp"} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event occurred_at is not a valid timestamp"): + _sync_once(store) + assert store.origin_cursor(store.device_id()) == 0 + assert store.rows("activity") == [] + + +def test_sync_once_raises_hub_error_on_event_with_unknown_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: sibling of the unknown-op test above, for table. + Store._write_in_txn rejects any table outside OWNED_TABLES for a local + write; the hub-pull path must reject it too, instead of durably + committing an orphaned row under a table name no application code ever + reads back.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "table": "not_a_real_table"} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event has an unknown table"): + _sync_once(store) + + +def test_sync_once_raises_hub_error_on_event_with_unhashable_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: `table not in OWNED_TABLES` (a frozenset) requires + table to be hashable - the exact bug class already fixed for + payload["type"] elsewhere in this PR (isinstance(typ, str) guards in + store.py). A JSON-decoded list/dict for table used to raise a raw + TypeError instead of a catchable HubError.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "table": ["activity"]} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event has an unknown table"): + _sync_once(store) + + +@pytest.mark.parametrize("bad_row_id", ["", ["not", "a", "string"]]) +def test_sync_once_raises_hub_error_on_event_with_invalid_row_id( + bad_row_id: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: row_id was presence-checked only, like every other + field here before it got its own validation - an empty string or a + non-string value used to pass validation untouched and only fail later, + as a raw type/constraint error from whatever eventually stores it.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "row_id": bad_row_id} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event row_id is not a valid id"): + _sync_once(store) + + +@pytest.mark.parametrize( + "bad_seq", + [ + "not-a-number", + 1e309, # float('inf') once parsed - int() raises OverflowError, not ValueError + True, # bool is an int subclass in Python - int(True) == 1 would pass silently + 1.5, # non-integral float - int(1.5) == 1 would silently truncate + ], +) +def test_sync_once_raises_hub_error_on_non_numeric_origin_seq( + bad_seq: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: int(event["origin_seq"]) used to run unguarded (raising a + raw ValueError on garbage, OverflowError on an out-of-range float) and also + silently accepted a bool or a fractional float as a valid sequence number.""" + store = _paired_store(tmp_path) + hub = FakeHub() + hub.pull_body = {"events": [{**_valid_pull_event(store.device_id()), "origin_seq": bad_seq}]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="non-numeric origin_seq"): + _sync_once(store) + + +def test_sync_once_accepts_an_activity_payload_whose_type_is_not_a_wake_type_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: a payload whose "type" is an unhashable value (e.g. a + list) used to make Store._maybe_wake's `payload.get("type") not in + WAKE_ACTIVITY_TYPES` raise a raw TypeError from deep inside + store.apply_remote - a shape neither _PULL_EVENT_FIELDS nor the broad + except-Exception safety net's introduction actually fixed, just quietly + converted into a permanent HubError retry loop (the cursor never advances, + so the hub keeps re-serving the same event forever). _maybe_wake now treats + a non-string type as simply "not a wake type" and lets the event apply + normally instead of failing the whole transaction.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "payload": {"type": ["not", "hashable"]}} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + _sync_once(store) + assert store.origin_cursor(store.device_id()) == 1 + row = store.row("activity", "x") + assert row is not None + + +def test_sync_once_raises_hub_error_when_apply_remote_hits_an_unanticipated_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test for the general safety net itself: field presence is + validated, but not every possible shape of every nested value can be + anticipated and fixed at its root (unlike the non-string-type case above). + A payload containing a value json.dumps cannot serialize (impossible from a + real JSON hub response, but a stand-in for "something genuinely + unanticipated") still reaches the broad except-Exception net around + apply_remote/mark_origin and becomes a catchable HubError instead of an + uncaught crash.""" + store = _paired_store(tmp_path) + event = {**_valid_pull_event(store.device_id()), "payload": {"type": "message", "body": {1, 2, 3}}} + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event could not be applied"): + _sync_once(store) + + +def test_sync_once_accepts_a_numeric_string_origin_seq( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: int(event["origin_seq"]) validated the value but the + original (still-string) event dict was what reached store.apply_remote(). A + numeric string like "1" passes int() cleanly, but _insert_event_idempotent's + `event["origin_seq"] != last_seq + 1` is a plain Python != - "1" != 1 is + always True - so a genuinely valid next sequence number raised a false + "origin_seq gap" StoreError.""" + store = _paired_store(tmp_path) + event = { + **_valid_pull_event(store.device_id()), + "origin_seq": "1", + "table": "task", + "row_id": "t1", + "payload": {"id": "t1"}, + } + hub = FakeHub() + hub.pull_body = {"events": [event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + _sync_once(store) + assert store.origin_cursor(store.device_id()) == 1 + row = store.row("task", "t1") + assert row is not None + + +def test_sync_once_raises_hub_error_on_non_list_snapshot_field(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test: list(pulled.get("inbox") or []) used to run unguarded; a + truthy non-iterable value (e.g. a malformed hub response sending an object + instead of a list) raised a raw TypeError instead of a catchable HubError.""" + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": {"not": "a list"}} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="inbox is not a list"): + _sync_once(_paired_store(tmp_path)) + + +def _valid_pull_row() -> dict[str, Any]: + return { + "table": "activity", + "origin_device_id": "other", + "row_id": "x", + "payload": {}, + "updated_at": "2026-08-13T12:00:00Z", + } + + +@pytest.mark.parametrize("field", ["table", "origin_device_id", "row_id", "payload", "updated_at"]) +def test_sync_once_raises_hub_error_on_snapshot_missing_one_required_field( + field: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: apply_replica_row indexes row["table"], + row["origin_device_id"], row["row_id"], row["payload"], row["updated_at"] + directly; a snapshot row missing any single one of those used to raise a raw + KeyError instead of a catchable HubError raised before that call. + Parametrized per field so a future accidental narrowing of _PULL_ROW_FIELDS + to any one of them is still caught.""" + row = _valid_pull_row() + del row[field] + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot is missing required fields"): + _sync_once(_paired_store(tmp_path)) + + +def test_sync_once_raises_hub_error_on_snapshot_with_unknown_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_sync_once_raises_hub_error_on_event_with_unknown_table.""" + row = {**_valid_pull_row(), "table": "not_a_real_table"} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot has an unknown table"): + _sync_once(_paired_store(tmp_path)) + + +def test_sync_once_raises_hub_error_on_snapshot_with_unhashable_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_sync_once_raises_hub_error_on_event_with_unhashable_table.""" + row = {**_valid_pull_row(), "table": ["activity"]} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot has an unknown table"): + _sync_once(_paired_store(tmp_path)) + + +@pytest.mark.parametrize("bad_row_id", ["", ["not", "a", "string"]]) +def test_sync_once_raises_hub_error_on_snapshot_with_invalid_row_id( + bad_row_id: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_sync_once_raises_hub_error_on_event_with_invalid_row_id.""" + row = {**_valid_pull_row(), "row_id": bad_row_id} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot row_id is not a valid id"): + _sync_once(_paired_store(tmp_path)) + + +@pytest.mark.parametrize("bad_origin_device_id", ["", ["not", "a", "string"]]) +def test_sync_once_raises_hub_error_on_snapshot_with_invalid_origin_device_id( + bad_origin_device_id: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: unlike an event's origin_device_id (checked for + ownership), a snapshot row's origin_device_id is legitimately foreign - + but it was still only presence-checked, never type/emptiness-checked, + before apply_replica_row compares it against this device's own id and + stores it as the row's recorded owner.""" + row = {**_valid_pull_row(), "origin_device_id": bad_origin_device_id} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot origin_device_id is not a valid id"): + _sync_once(_paired_store(tmp_path)) + + +def test_sync_once_raises_hub_error_on_snapshot_with_empty_updated_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: updated_at was presence-checked but not validated as a + genuine timestamp. An empty string used to pass validation, get stored + as-is on first insert (the row_data upsert only compares updated_at + against an existing row on conflict, not on a fresh insert), and only + fail later - on the next legitimate update to that same row, which would + raise a raw ::timestamptz cast error trying to compare against it.""" + row = {**_valid_pull_row(), "updated_at": ""} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot updated_at is not a valid timestamp"): + _sync_once(_paired_store(tmp_path)) + + +def test_sync_once_raises_hub_error_on_snapshot_with_unparseable_updated_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: sibling of the empty-string test above, for a + non-empty but still bogus value. A non-blank string that isn't a real + timestamp (e.g. "not-a-timestamp") used to pass a mere non-empty check, + hitting the identical stuck-row failure mode later on the first + conflicting update.""" + row = {**_valid_pull_row(), "updated_at": "not-a-timestamp"} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot updated_at is not a valid timestamp"): + _sync_once(_paired_store(tmp_path)) + + +def test_sync_once_accepts_a_lowercase_z_updated_at(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test: str.replace("Z", "+00:00") is case-sensitive, but + RFC 3339 (SS5.6) permits a lowercase "z" as the UTC designator just as + validly as an uppercase one. A genuinely valid, standards-conformant + timestamp ending in a lowercase "z" used to be falsely rejected as an + invalid timestamp instead of being accepted.""" + row = {**_valid_pull_row(), "updated_at": "2026-08-13T12:00:00z"} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + store = _paired_store(tmp_path) + _sync_once(store) + row_stored = store.row("activity", "x") + assert row_stored is not None + + +def test_sync_once_rejects_whole_batch_when_one_of_two_snapshots_is_invalid( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test for the validate-all-then-apply-all split itself: the + row-side sibling of the equivalent tests/test_restore.py test. An earlier + version of this PR validated and applied snapshot rows in the same loop, + so a batch with one valid row before an invalid one would durably commit + the valid row before raising HubError on the invalid one - a partial + apply of an atomically-intended batch.""" + valid_row = {**_valid_pull_row(), "row_id": "valid-1"} + invalid_row = {**_valid_pull_row(), "row_id": "invalid-1", "table": "not_a_real_table"} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [valid_row, invalid_row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + store = _paired_store(tmp_path) + with pytest.raises(HubError, match="pull snapshot has an unknown table"): + _sync_once(store) + assert store.rows("activity") == [] + + +def test_sync_once_rejects_whole_event_batch_when_one_of_two_events_is_invalid( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the event-side sibling of + test_sync_once_rejects_whole_batch_when_one_of_two_snapshots_is_invalid. + The events loop kept validating and applying each event in the same + iteration even after the snapshots loop was split into a + validate-all-then-apply-all pattern to fix exactly this partial-apply + shape - a batch with one valid event before an invalid one used to + durably commit the valid event (and advance its origin cursor) before + raising HubError on the invalid one.""" + store = _paired_store(tmp_path) + valid_event = {**_valid_pull_event(store.device_id()), "row_id": "valid-1"} + invalid_event = { + **_valid_pull_event(store.device_id()), + "origin_seq": 2, + "row_id": "invalid-1", + "table": "not_a_real_table", + } + hub = FakeHub() + hub.pull_body = {"events": [valid_event, invalid_event]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull event has an unknown table"): + _sync_once(store) + assert store.origin_cursor(store.device_id()) == 0 + assert store.rows("activity") == [] + + +def test_sync_once_rejects_whole_response_when_a_valid_event_is_paired_with_an_invalid_snapshot( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: events were fully validated AND applied (a + complete, separate phase, including the store write and cursor advance) + before snapshot validation even began - so a well-formed event paired + with a malformed snapshot in the same pull response used to durably + commit the event before dying on the snapshot, the identical + partial-apply bug this PR already fixed twice, once within the events + list and once within the snapshots list, just one level up between the + two lists themselves.""" + store = _paired_store(tmp_path) + valid_event = {**_valid_pull_event(store.device_id())} + invalid_row = {**_valid_pull_row(), "table": "not_a_real_table"} + hub = FakeHub() + hub.pull_body = {"events": [valid_event], "inbox": [invalid_row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot has an unknown table"): + _sync_once(store) + assert store.origin_cursor(store.device_id()) == 0 + assert store.rows("activity") == [] + + +def test_sync_once_accepts_a_snapshot_payload_whose_type_is_not_a_wake_type_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_sync_once_accepts_an_activity_payload_whose_type_is_not_a_wake_type_shape. + apply_replica_row also calls _maybe_wake, so the same non-string-type fix + must let the row apply normally instead of raising.""" + row = {**_valid_pull_row(), "payload": {"type": ["not", "hashable"]}} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + store = _paired_store(tmp_path) + _sync_once(store) + stored = store.row("activity", "x") + assert stored is not None + + +def test_sync_once_raises_hub_error_when_apply_replica_row_hits_an_unanticipated_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_sync_once_raises_hub_error_when_apply_remote_hits_an_unanticipated_shape, + proving the broad safety net around store.apply_replica_row catches a + genuinely unanticipated shape too, not just the event-side one.""" + row = {**_valid_pull_row(), "payload": {"type": "message", "body": {1, 2, 3}}} + hub = FakeHub() + hub.pull_body = {"events": [], "inbox": [row]} + monkeypatch.setattr("agent_cli.main._hub_from_store", lambda _s: hub) + with pytest.raises(HubError, match="pull snapshot could not be applied"): + _sync_once(_paired_store(tmp_path)) + + def test_watch_pending_skips_other_executable_types(tmp_path: Path) -> None: store = Store(tmp_path) _owned_session(store) diff --git a/tests/test_restore.py b/tests/test_restore.py new file mode 100644 index 0000000..ed7f08c --- /dev/null +++ b/tests/test_restore.py @@ -0,0 +1,603 @@ +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any + +import pytest + +from agent_cli import main as main_mod +from agent_cli.main import open_store + + +class _FakeRestoreHub: + def __init__(self, body: Any) -> None: + self.body = body + + def restore(self) -> Any: + return self.body + + def close(self) -> None: + return None + + +def _init_store(tmp_path: Path) -> None: + os.environ["AGENT_HOME"] = str(tmp_path) + main_mod.main(["init"]) + + +def _run_restore(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, body: Any) -> None: + _init_store(tmp_path) + store = open_store() + device_id = store.device_id() + store.close() + if isinstance(body, dict) and "device_id" not in body: + body = {**body, "device_id": device_id} + monkeypatch.setattr(main_mod, "_hub_from_store", lambda _s: _FakeRestoreHub(body)) + main_mod.cmd_restore([]) + + +def test_cmd_restore_dies_on_non_dict_body(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test: Hub.request (and so Hub.restore) returns None for a 2xx + response with an empty body - cmd_restore used to call body.get("device_id") + straight on that, raising a raw AttributeError instead of a clean die().""" + with pytest.raises(SystemExit, match="restore response is not an object"): + _run_restore(tmp_path, monkeypatch, None) + + +def _valid_restore_event(device_id: str) -> dict[str, Any]: + return { + "origin_device_id": device_id, + "origin_seq": 1, + "table": "activity", + "op": "insert", + "row_id": "x", + "payload": {}, + "occurred_at": "2026-08-13T12:00:00Z", + } + + +def _own_device_id(tmp_path: Path) -> str: + _init_store(tmp_path) + store = open_store() + try: + return store.device_id() + finally: + store.close() + + +@pytest.mark.parametrize( + "field", + ["origin_device_id", "origin_seq", "table", "op", "row_id", "payload", "occurred_at"], +) +def test_cmd_restore_dies_on_event_missing_one_required_field( + field: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: a restored event missing any single one of the fields + _insert_event_idempotent indexes used to raise a raw KeyError deep inside + store.apply_remote instead of a clean die(). Parametrized per field (like + the equivalent tests/test_pending.py _sync_once tests) so a future + accidental narrowing of _PULL_EVENT_FIELDS to any one of them is still + caught, not just the "several fields missing at once" case.""" + event = _valid_restore_event(_own_device_id(tmp_path)) + del event[field] + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event is missing required fields"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_event_with_foreign_origin_device_id( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: own_events replays this device's own history + (DESIGN.md: "own events, gapless" - foreign data only ever arrives as row + snapshots). An event whose origin_device_id doesn't match this device's + own used to pass every check and reach apply_remote/mark_origin + unchanged, letting a malformed restore response poison this device's own + ledger under a foreign device's identity.""" + event = {**_valid_restore_event(_own_device_id(tmp_path)), "origin_device_id": "other"} + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event origin_device_id is not this device's own"): + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.rows("activity") == [] + finally: + store.close() + + +def _valid_restore_row() -> dict[str, Any]: + return { + "table": "activity", + "origin_device_id": "other", + "row_id": "x", + "payload": {}, + "updated_at": "2026-08-13T12:00:00Z", + } + + +@pytest.mark.parametrize("field", ["table", "origin_device_id", "row_id", "payload", "updated_at"]) +def test_cmd_restore_dies_on_snapshot_missing_one_required_field( + field: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: a restored snapshot row missing any single one of the + fields apply_replica_row indexes used to raise a raw KeyError instead of a + clean die(). Parametrized per field, same reasoning as the event test + above.""" + row = _valid_restore_row() + del row[field] + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot is missing required fields"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_event_with_non_dict_payload( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_event_with_non_dict_payload. Field + presence was checked but not payload's type - a non-dict payload used to + pass validation, get committed by apply_remote, and only fail later on + every future read of that whole table.""" + event = {**_valid_restore_event(_own_device_id(tmp_path)), "payload": ["not", "an", "object"]} + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event payload is not an object"): + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.rows("activity") == [] + finally: + store.close() + + +def test_cmd_restore_dies_on_event_with_unknown_op(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_event_with_unknown_op.""" + event = {**_valid_restore_event(_own_device_id(tmp_path)), "op": "bogus"} + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event has an unknown op"): + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.rows("activity") == [] + finally: + store.close() + + +def test_cmd_restore_dies_on_event_with_unparseable_occurred_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_event_with_unparseable_occurred_at.""" + event = {**_valid_restore_event(_own_device_id(tmp_path)), "occurred_at": "not-a-timestamp"} + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event occurred_at is not a valid timestamp"): + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.rows("activity") == [] + finally: + store.close() + + +def test_cmd_restore_dies_on_event_with_unknown_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_event_with_unknown_table.""" + event = {**_valid_restore_event(_own_device_id(tmp_path)), "table": "not_a_real_table"} + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event has an unknown table"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_event_with_unhashable_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_event_with_unhashable_table.""" + event = {**_valid_restore_event(_own_device_id(tmp_path)), "table": ["activity"]} + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event has an unknown table"): + _run_restore(tmp_path, monkeypatch, body) + + +@pytest.mark.parametrize("bad_row_id", ["", ["not", "a", "string"]]) +def test_cmd_restore_dies_on_event_with_invalid_row_id( + bad_row_id: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_event_with_invalid_row_id.""" + event = {**_valid_restore_event(_own_device_id(tmp_path)), "row_id": bad_row_id} + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event row_id is not a valid id"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_snapshot_with_non_dict_payload( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_cmd_restore_dies_on_event_with_non_dict_payload.""" + row = {**_valid_restore_row(), "payload": ["not", "an", "object"]} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot payload is not an object"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_snapshot_with_unknown_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_cmd_restore_dies_on_event_with_unknown_table.""" + row = {**_valid_restore_row(), "table": "not_a_real_table"} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot has an unknown table"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_snapshot_with_unhashable_table( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_cmd_restore_dies_on_event_with_unhashable_table.""" + row = {**_valid_restore_row(), "table": ["activity"]} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot has an unknown table"): + _run_restore(tmp_path, monkeypatch, body) + + +@pytest.mark.parametrize("bad_row_id", ["", ["not", "a", "string"]]) +def test_cmd_restore_dies_on_snapshot_with_invalid_row_id( + bad_row_id: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_snapshot_with_invalid_row_id.""" + row = {**_valid_restore_row(), "row_id": bad_row_id} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot row_id is not a valid id"): + _run_restore(tmp_path, monkeypatch, body) + + +@pytest.mark.parametrize("bad_origin_device_id", ["", ["not", "a", "string"]]) +def test_cmd_restore_dies_on_snapshot_with_invalid_origin_device_id( + bad_origin_device_id: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_snapshot_with_invalid_origin_device_id.""" + row = {**_valid_restore_row(), "origin_device_id": bad_origin_device_id} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot origin_device_id is not a valid id"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_snapshot_with_empty_updated_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of the equivalent + tests/test_pending.py _sync_once test. updated_at was presence-checked + but not validated as a genuine timestamp; an empty string used to pass + validation, get stored as-is on first insert, and only fail later on the + next legitimate update to that same row via a raw ::timestamptz cast + error.""" + row = {**_valid_restore_row(), "updated_at": ""} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot updated_at is not a valid timestamp"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_on_snapshot_with_unparseable_updated_at( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of the equivalent + tests/test_pending.py _sync_once test, for a non-empty but still bogus + value. A non-blank string that isn't a real timestamp (e.g. + "not-a-timestamp") used to pass a mere non-empty check, hitting the + identical stuck-row failure mode later on the first conflicting + update.""" + row = {**_valid_restore_row(), "updated_at": "not-a-timestamp"} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot updated_at is not a valid timestamp"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_accepts_a_lowercase_z_updated_at(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test: the row-side sibling of the equivalent + tests/test_pending.py _sync_once test. str.replace("Z", "+00:00") is + case-sensitive, but RFC 3339 (SS5.6) permits a lowercase "z" as the UTC + designator just as validly as an uppercase one. A genuinely valid, + standards-conformant timestamp ending in a lowercase "z" used to be + falsely rejected as an invalid timestamp instead of being accepted.""" + row = {**_valid_restore_row(), "updated_at": "2026-08-13T12:00:00z"} + body = {"own_events": [], "inbox": [row]} + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.row("activity", "x") is not None + finally: + store.close() + + +def test_cmd_restore_rejects_whole_batch_when_one_of_two_snapshots_is_invalid( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test for the validate-all-then-apply-all split itself: an + earlier version of this PR validated and applied snapshot rows in the + same loop, so a batch with one valid row before an invalid one would + durably commit the valid row before dying on the invalid one - a partial + apply of an atomically-intended batch. Proves the fix holds: with a + valid row before the invalid one in the list, cmd_restore must still die + without committing the valid row at all.""" + valid_row = {**_valid_restore_row(), "row_id": "valid-1"} + invalid_row = {**_valid_restore_row(), "row_id": "invalid-1", "table": "not_a_real_table"} + body = {"own_events": [], "inbox": [valid_row, invalid_row]} + with pytest.raises(SystemExit, match="restore snapshot has an unknown table"): + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.rows("activity") == [] + finally: + store.close() + + +def test_cmd_restore_rejects_whole_event_batch_when_one_of_two_events_is_invalid( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the event-side sibling of + test_cmd_restore_rejects_whole_batch_when_one_of_two_snapshots_is_invalid. + cmd_restore's events loop kept validating and applying each event in the + same iteration even after its snapshots loop was split into a + validate-all-then-apply-all pattern to fix exactly this partial-apply + shape - a batch with one valid event before an invalid one used to + durably commit the valid event (and advance its origin cursor) before + dying on the invalid one.""" + device_id = _own_device_id(tmp_path) + valid_event = {**_valid_restore_event(device_id), "row_id": "valid-1"} + invalid_event = { + **_valid_restore_event(device_id), + "origin_seq": 2, + "row_id": "invalid-1", + "table": "not_a_real_table", + } + body = {"own_events": [valid_event, invalid_event]} + with pytest.raises(SystemExit, match="restore event has an unknown table"): + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.origin_cursor(device_id) == 0 + assert store.rows("activity") == [] + finally: + store.close() + + +def test_cmd_restore_rejects_whole_response_when_a_valid_event_is_paired_with_an_invalid_snapshot( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_rejects_whole_response_when_a_valid_event_is_paired_with_an_invalid_snapshot. + cmd_restore's events were fully validated AND applied (a complete, + separate phase, including the store write and cursor advance) before + snapshot validation even began - so a well-formed event paired with a + malformed snapshot in the same restore response used to durably commit + the event before dying on the snapshot.""" + device_id = _own_device_id(tmp_path) + valid_event = _valid_restore_event(device_id) + invalid_row = {**_valid_restore_row(), "table": "not_a_real_table"} + body = {"own_events": [valid_event], "inbox": [invalid_row]} + with pytest.raises(SystemExit, match="restore snapshot has an unknown table"): + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.origin_cursor(device_id) == 0 + assert store.rows("activity") == [] + finally: + store.close() + + +@pytest.mark.parametrize( + "bad_seq", + [ + "not-a-number", + 1e309, # float('inf') once parsed - int() raises OverflowError, not ValueError + True, # bool is an int subclass in Python - int(True) == 1 would pass silently + 1.5, # non-integral float - int(1.5) == 1 would silently truncate + ], +) +def test_cmd_restore_dies_on_non_numeric_origin_seq( + bad_seq: object, tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_non_numeric_origin_seq. cmd_restore + shares _sync_once's origin_seq coercion (_coerce_pull_event), but had no + test proving the restore side actually rejects a bad origin_seq rather + than just accepting a good one.""" + event = _valid_restore_event(_own_device_id(tmp_path)) + event["origin_seq"] = bad_seq + body = {"own_events": [event]} + with pytest.raises(SystemExit, match="restore event has a non-numeric origin_seq"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_accepts_a_numeric_string_origin_seq( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: cmd_restore passed event["origin_seq"] to + apply_remote uncoerced, only int()-converting it afterward for + mark_origin. A numeric-string origin_seq (e.g. "1") then hit + _insert_event_idempotent's plain Python `event["origin_seq"] != last_seq + + 1` as a string, deterministically raising a false "origin_seq gap" for the + very first restored event. cmd_restore now shares _sync_once's coercion, + applied before apply_remote sees the event. own_events replays this + device's own history, so origin_device_id must genuinely be this + device's own id, not a foreign one - same reasoning as + test_cmd_restore_applies_events_and_snapshots below.""" + _init_store(tmp_path) + store = open_store() + device_id = store.device_id() + store.close() + body = { + "own_events": [ + { + "origin_device_id": device_id, + "origin_seq": "1", + "table": "task", + "op": "insert", + "row_id": "t1", + "payload": {"id": "t1"}, + "occurred_at": "2026-08-13T12:00:00Z", + } + ] + } + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.origin_cursor(device_id) == 1 + assert store.row("task", "t1") is not None + finally: + store.close() + + +def test_cmd_restore_dies_on_non_list_snapshot_field(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_on_non_list_snapshot_field. A truthy + non-list "inbox" (e.g. a malformed hub response sending an object instead + of a list) used to raise a raw TypeError from list(...) instead of a + clean die().""" + body = {"own_events": [], "inbox": {"not": "a list"}} + with pytest.raises(SystemExit, match="restore response inbox is not a list"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_when_apply_remote_hits_an_unanticipated_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_when_apply_remote_hits_an_unanticipated_shape. + A payload containing a value json.dumps cannot serialize (a stand-in for + "something genuinely unanticipated") reaches the broad except-Exception net + around apply_remote/mark_origin and becomes a clean die() instead of an + uncaught crash.""" + body = { + "own_events": [ + { + "origin_device_id": _own_device_id(tmp_path), + "origin_seq": 1, + "table": "activity", + "op": "insert", + "row_id": "x", + "payload": {"type": "message", "body": {1, 2, 3}}, + "occurred_at": "2026-08-13T12:00:00Z", + } + ] + } + with pytest.raises(SystemExit, match="restore event could not be applied"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_dies_when_apply_replica_row_hits_an_unanticipated_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the restore-side sibling of + test_sync_once_raises_hub_error_when_apply_replica_row_hits_an_unanticipated_shape.""" + row = {**_valid_restore_row(), "payload": {"type": "message", "body": {1, 2, 3}}} + body = {"own_events": [], "inbox": [row]} + with pytest.raises(SystemExit, match="restore snapshot could not be applied"): + _run_restore(tmp_path, monkeypatch, body) + + +def test_cmd_restore_accepts_an_event_whose_type_is_not_a_wake_type_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: apply_remote's wake=False branch (the only one + cmd_restore ever uses) has its own inline isinstance(typ, str) guard + before the WAKE_ACTIVITY_TYPES membership check (store.py's + apply_remote, elif inserted: branch). Unlike the wake=True path + (_maybe_wake, covered by tests/test_pending.py), nothing exercised this + wake=False guard specifically - a regression that reintroduced the + unguarded check only there would stay green on the sync side while + breaking every restore whose event has a non-string type.""" + _init_store(tmp_path) + store = open_store() + device_id = store.device_id() + store.close() + body = { + "own_events": [ + { + "origin_device_id": device_id, + "origin_seq": 1, + "table": "activity", + "op": "insert", + "row_id": "x", + "payload": {"type": ["not", "hashable"]}, + "occurred_at": "2026-08-13T12:00:00Z", + } + ] + } + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.row("activity", "x") is not None + finally: + store.close() + + +def test_cmd_restore_accepts_a_snapshot_whose_type_is_not_a_wake_type_shape( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: the row-side sibling of + test_cmd_restore_accepts_an_event_whose_type_is_not_a_wake_type_shape. + apply_replica_row's wake=False branch has the same inline guard.""" + row = {**_valid_restore_row(), "payload": {"type": ["not", "hashable"]}} + body = {"own_events": [], "inbox": [row]} + _run_restore(tmp_path, monkeypatch, body) + store = open_store() + try: + assert store.row("activity", "x") is not None + finally: + store.close() + + +def test_cmd_restore_applies_events_and_snapshots( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str] +) -> None: + """Happy-path coverage: cmd_restore had none before this change. Replays + one own event (own_events replays this device's own history, so its + origin_device_id must genuinely be this device's own id, not a foreign + one - foreign data only ever arrives via inbox/pings) and one inbox + snapshot row (from another device, correctly foreign).""" + _init_store(tmp_path) + store = open_store() + device_id = store.device_id() + store.close() + body = { + "own_events": [ + { + "origin_device_id": device_id, + "origin_seq": 1, + "table": "task", + "op": "insert", + "row_id": "t1", + "payload": {"id": "t1"}, + "occurred_at": "2026-08-13T12:00:00Z", + } + ], + "inbox": [ + { + "table": "activity", + "origin_device_id": "other", + "row_id": "a1", + "payload": {"id": "a1", "type": "message"}, + "updated_at": "2026-08-13T12:00:00Z", + } + ], + } + capsys.readouterr() + _run_restore(tmp_path, monkeypatch, body) + captured = capsys.readouterr() + assert "restored events=1 snapshots=1" in captured.out + store = open_store() + try: + assert store.row("task", "t1") is not None + assert store.row("activity", "a1") is not None + finally: + store.close() diff --git a/tests/test_store.py b/tests/test_store.py index 279dfee..74c3cf2 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -45,6 +45,36 @@ def test_write_emits_seq_and_blocks_foreign(tmp_path: Path) -> None: assert pending[0]["origin_seq"] == 1 +def test_mark_pushed_does_not_regress_the_cursor(tmp_path: Path) -> None: + store = Store(tmp_path) + store.mark_pushed(10) + store.mark_pushed(3) + assert store.sync_get("pushed_origin_seq", "0") == "10" + + +def test_pending_work_skips_a_non_string_type_instead_of_raising(tmp_path: Path) -> None: + """Regression test: pending_work's `payload.get("type") not in + EXECUTABLE_ACTIVITY_TYPES` used to run unguarded, same defect as the + hub-pull path this PR otherwise hardened. Own-origin rows aren't only + ever written by this device's own trusted code - cmd_restore applies + hub-returned "own_events" via apply_remote(wake=False), so a malformed + type can still reach here via a corrupted restore response. A non-string + type must be treated as simply not executable, not raise TypeError.""" + store = Store(tmp_path) + store.write( + "activity", + "insert", + "a1", + { + "id": "a1", + "session_id": "s1", + "type": ["not", "hashable"], + "execution_status": "pending", + }, + ) + assert store.pending_work() == [] + + def test_remote_gap_fail_closed(tmp_path: Path) -> None: store = Store(tmp_path) with pytest.raises(StoreError, match="gap"): diff --git a/tests/test_sync_follow_reconnect.py b/tests/test_sync_follow_reconnect.py index 4099029..af85612 100644 --- a/tests/test_sync_follow_reconnect.py +++ b/tests/test_sync_follow_reconnect.py @@ -378,6 +378,130 @@ def fake_sync_once(store: object) -> None: assert calls == [1], "a genuine data-integrity StoreError must not be retried" +def test_subscription_row_with_non_dict_payload_is_skipped_not_committed(tmp_path: Path) -> None: + """Regression test: incoming websocket subscription rows went straight to + store.apply_replica_row(row) guarded only by isinstance(row, dict) and + row.get("table") - never through _check_pull_row. A non-dict payload + doesn't raise anywhere in apply_replica_row/_upsert_row (dumps([]) is + valid JSON), so it used to get durably committed and only fail later, on + every future store.rows()/row() call for that whole table - the same bug + class this PR closed for _sync_once/cmd_restore, reachable a third way + through the live subscription push path.""" + _init_paired_store(tmp_path) + store = open_store() + try: + row = { + "table": "session", + "origin_device_id": "other", + "row_id": "s1", + "payload": ["not", "an", "object"], + "updated_at": "2026-08-13T12:00:00Z", + } + + class _SubscriptionWs: + def send(self, data: str) -> None: + pass + + def __iter__(self): + return iter([json.dumps({"type": "subscription", "rows": [row]})]) + + def close(self) -> None: + pass + + class _FakeHub: + def connect_sync_ws(self) -> _SubscriptionWs: + return _SubscriptionWs() + + with pytest.raises(HubError, match="websocket closed"): + main_mod._run_sync_ws_session(store, _FakeHub(), _FakeRuntime(), {}, {}, {}) + assert store.rows("session") == [] + finally: + store.close() + + +def test_subscription_row_with_unhashable_table_is_skipped_not_crashed(tmp_path: Path) -> None: + """Regression test: `table not in OWNED_TABLES` inside _check_pull_row + requires table to be hashable. A JSON-decoded list/dict for table used to + raise a raw TypeError that neither the _PullShapeError catch around + _check_pull_row nor the (StoreError, KeyError, TypeError) catch around + apply_replica_row (which only wraps the *other* try block) would catch - + crashing the whole agent sync --follow daemon on a malformed WS frame.""" + _init_paired_store(tmp_path) + store = open_store() + try: + row = { + "table": ["session"], + "origin_device_id": "other", + "row_id": "s1", + "payload": {"id": "s1"}, + "updated_at": "2026-08-13T12:00:00Z", + } + + class _SubscriptionWs: + def send(self, data: str) -> None: + pass + + def __iter__(self): + return iter([json.dumps({"type": "subscription", "rows": [row]})]) + + def close(self) -> None: + pass + + class _FakeHub: + def connect_sync_ws(self) -> _SubscriptionWs: + return _SubscriptionWs() + + with pytest.raises(HubError, match="websocket closed"): + main_mod._run_sync_ws_session(store, _FakeHub(), _FakeRuntime(), {}, {}, {}) + assert store.rows("session") == [] + finally: + store.close() + + +def test_recursion_error_from_deeply_nested_frame_is_skipped_not_crashed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Regression test: json.loads(raw) on an incoming websocket frame was + only guarded by except (TypeError, ValueError) - CPython's C-accelerated + json decoder still bounds recursion by C stack depth, not just + sys.getrecursionlimit(), so a pathologically deeply-nested frame + (adversarial or buggy hub) raises RecursionError, a RuntimeError + subclass neither of those catches - the same gap just fixed in + Hub.request, reachable a third way through the live websocket loop. + Monkeypatches json.loads directly (matching tests/test_hub.py's + equivalent test) rather than constructing a real, actually-deeply-nested + frame string: relying on CPython's C decoder genuinely overflowing at a + specific depth would make this test depend on interpreter/platform + internals rather than deterministically exercising the new except arm.""" + _init_paired_store(tmp_path) + store = open_store() + try: + + def raise_recursion_error(_raw: str) -> None: + raise RecursionError("Stack overflow while decoding a JSON array") + + monkeypatch.setattr(main_mod.json, "loads", raise_recursion_error) + + class _SubscriptionWs: + def send(self, data: str) -> None: + pass + + def __iter__(self): + return iter(["[1]"]) + + def close(self) -> None: + pass + + class _FakeHub: + def connect_sync_ws(self) -> _SubscriptionWs: + return _SubscriptionWs() + + with pytest.raises(HubError, match="websocket closed"): + main_mod._run_sync_ws_session(store, _FakeHub(), _FakeRuntime(), {}, {}, {}) + finally: + store.close() + + def test_subscription_row_store_connection_error_reaches_the_reconnect_loop( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -402,9 +526,14 @@ def send(self, data: str) -> None: pass def __iter__(self): - return iter( - [json.dumps({"type": "subscription", "rows": [{"table": "session", "id": "s1"}]})] - ) + row = { + "table": "session", + "origin_device_id": "other", + "row_id": "s1", + "payload": {"id": "s1"}, + "updated_at": "2026-08-13T12:00:00Z", + } + return iter([json.dumps({"type": "subscription", "rows": [row]})]) def close(self) -> None: pass