From a109677a6e31d473e9dde60468bf73a5a0591557 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 11:55:15 +0800 Subject: [PATCH 01/10] chore(ci): run the script test CI never ran, and ignore agent worktrees Three findings from a repository diagnostic, each reproduced before it was acted on. `scripts/tests/test_bench_event_latency.py` was never executed by CI. Seven test modules live under `scripts/tests/`; the `test` and `runtime-floor` jobs each named six. AGENTS.md carries it in the canonical pre-PR list, so the gate was weaker than the document it is measured against. It costs 0.052s. Its one real timing assertion is a lower bound on an elapsed wait, so load can only make it more true - safe on a shared runner. `.claude/worktrees/` was ignored only through `.git/info/exclude`, which is per-clone and uncommitted. Its sibling `.worktrees/` was already in the tracked file. A fresh clone therefore searches its own worktrees, and every repo-wide grep returns each hit once per checkout - measured here while chasing an unrelated symbol. `read_sidecar` in observer.py had no caller anywhere: of 254 public definitions in the runtime it was the only one with zero references outside its own definition. `write_sidecar` stays - http_api.py calls it and returns the result directly rather than reading back, which is how the reader was orphaned. `state_read_cap_bytes` keeps its consumer in lifecycle.py. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- .github/workflows/quality-gate.yml | 2 ++ .gitignore | 6 ++++++ .../skills/cargento/cargento_runtime/observer.py | 13 ------------- 3 files changed, 8 insertions(+), 13 deletions(-) diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index caa1e8d..95401d1 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -177,6 +177,7 @@ jobs: scripts.tests.test_lint_embedded \ scripts.tests.test_bench_collect \ scripts.tests.test_capture_hook \ + scripts.tests.test_bench_event_latency \ scripts.tests.test_derive_prompt_shapes coverage report @@ -211,6 +212,7 @@ jobs: scripts.tests.test_lint_embedded \ scripts.tests.test_bench_collect \ scripts.tests.test_capture_hook \ + scripts.tests.test_bench_event_latency \ scripts.tests.test_derive_prompt_shapes coverage report diff --git a/.gitignore b/.gitignore index 27f6f08..ae3ba07 100644 --- a/.gitignore +++ b/.gitignore @@ -19,6 +19,12 @@ docs/visibility-2x2/items.json.tmp # Spacedock worktrees: per-entity checkouts, never source .worktrees/ +# Agent worktrees: CLAUDE.md sends `isolation: "worktree"` checkouts here. Ignored +# in the committed file and not only in .git/info/exclude, because that exclude is +# per-clone: without this, a fresh clone greps its own worktrees and every +# repo-wide search returns each hit once per checkout. +.claude/worktrees/ + # Spacedock roadmap-burndown state: a linked worktree of the spacedock-state/roadmap-burndown orphan branch docs/roadmap-burndown/.spacedock-state/ diff --git a/cargento/skills/cargento/cargento_runtime/observer.py b/cargento/skills/cargento/cargento_runtime/observer.py index ea05ed5..686c12a 100644 --- a/cargento/skills/cargento/cargento_runtime/observer.py +++ b/cargento/skills/cargento/cargento_runtime/observer.py @@ -641,19 +641,6 @@ def write_sidecar( return path -def read_sidecar(config: RuntimeConfig, harness: str, sid: str) -> dict[str, Any] | None: - """Read the observer sidecar, or None if absent, unnamed or malformed.""" - path = sidecar_path(config, harness, sid) - if path is None: - return None - try: - with open(path, encoding="utf-8") as handle: - value = json.loads(handle.read(config.state_read_cap_bytes)) - except (OSError, ValueError, json.JSONDecodeError): - return None - return value if isinstance(value, dict) else None - - def _mtime(path: str) -> float: """One file's mtime, or 0 when it went away between the glob and the stat.""" try: From 5ee302894fdc18e586064db167d3d0b17d389a1f Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 11:57:59 +0800 Subject: [PATCH 02/10] fix(scripts): stop validating the parts of docs/ that git ignores The canonical pre-PR validator walked every Markdown file under `docs/`, and most of them are not source. 56 of the 78 present here are gitignored: the vendored Spacedock mods, the linked worktree holding the entity-state orphan branch, and a person's local deep-dive notes. That made `python3 scripts/validate_plugins.py` fail on clean `main` for anyone with the mods installed, on a vendored template's unexpanded `{state-owner}` placeholders. CI never saw it, because the directory is gitignored and is never checked out - so the red belonged to no change and pointed at no fix. Filtered with `check-ignore` rather than `ls-files`, so a doc that is merely new is still validated before it is staged. Fails open: any git error validates everything, which is exactly the behaviour it replaces. Verified by injecting a broken link into a tracked doc and confirming the validator still fails on it. AGENTS.md's Quality Gate sentence described the 3.11 job as a direct-launch smoke test. It also runs the whole suite under coverage on that floor, so an agent reading the sentence would underestimate what has to pass there. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- AGENTS.md | 2 +- scripts/validate_plugins.py | 38 ++++++++++++++++++++++++++++++++++++- 2 files changed, 38 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 6db087c..2f6b0b0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -284,7 +284,7 @@ choose. Nobody asked for twelve hours; they asked for the work. ## Quality Gate -Every PR must pass the `quality-gate` required check (`.github/workflows/quality-gate.yml`): ruff with `select = ALL` (curated ignores documented in `pyproject.toml`), `ruff format --check`, `mypy --strict`, the HTML/CSS/JS frontend source linter (`scripts/lint_embedded.py`), a direct-launch smoke test on the Python 3.11 runtime floor, the full unittest suite under `coverage` with the `fail_under` threshold from `pyproject.toml`, and `platform-tests` — the same unit suite re-run natively on Ubuntu, macOS and Windows. The threshold only ratchets up — never lower it in a PR. A PR that must merge below threshold needs the `coverage-exception` label, which is visible in the PR timeline. +Every PR must pass the `quality-gate` required check (`.github/workflows/quality-gate.yml`): ruff with `select = ALL` (curated ignores documented in `pyproject.toml`), `ruff format --check`, `mypy --strict`, the HTML/CSS/JS frontend source linter (`scripts/lint_embedded.py`), a direct-launch smoke test on the Python 3.11 runtime floor followed by the whole suite under `coverage` there, the same suite under `coverage` on 3.12 with the `fail_under` threshold from `pyproject.toml`, and `platform-tests` — the same unit suite re-run natively on Ubuntu, macOS and Windows. The threshold only ratchets up — never lower it in a PR. A PR that must merge below threshold needs the `coverage-exception` label, which is visible in the PR timeline. **The required context always reports; its constituent jobs may not run.** A `changes` job decides whether the diff contains anything the gate can measure, and the five measurable jobs are gated on diff --git a/scripts/validate_plugins.py b/scripts/validate_plugins.py index b2eae47..789fecc 100755 --- a/scripts/validate_plugins.py +++ b/scripts/validate_plugins.py @@ -5,6 +5,7 @@ import json import re +import subprocess import sys from pathlib import Path from typing import Any @@ -1197,6 +1198,40 @@ def validate_marketplaces( ) +def _git_ignored(paths: list[Path]) -> set[Path]: + """The subset of `paths` that git is told to ignore. + + `docs/` holds more than source: vendored Spacedock mods, the linked worktree + of the entity-state orphan branch, and a person's local deep-dive notes. All + three are gitignored, and 56 of the 78 Markdown files under `docs/` were of + that kind when this was measured. Walking them made the canonical pre-PR + validator fail on a vendored file's unexpanded `{state-owner}` placeholders + for anyone with the mods installed, while CI stayed green because that + directory is never checked in — a red that belonged to no change. + + `check-ignore` rather than `ls-files`, so a doc that is merely new still gets + validated before it is staged. Fails open: any git error validates + everything, which is the behaviour this replaces and can only over-check. + """ + if not paths: + return set() + try: + result = subprocess.run( + ["git", "check-ignore", "--stdin"], # noqa: S607 + cwd=ROOT, + input="\n".join(str(path) for path in paths), + capture_output=True, + text=True, + check=False, + ) + except OSError: + return set() + # 0 = at least one ignored, 1 = none ignored; anything else is a real error. + if result.returncode not in (0, 1): + return set() + return {Path(line) for line in result.stdout.splitlines() if line} + + def validate_repo_docs(validation: Validation) -> None: """Resolve Markdown inline links and anchors in the repository's prose docs. @@ -1207,7 +1242,8 @@ def validate_repo_docs(validation: Validation) -> None: not checked here. """ paths = [ROOT / name for name in ROOT_DOCS] - paths.extend(sorted((ROOT / "docs").rglob("*.md"))) + walked = sorted((ROOT / "docs").rglob("*.md")) + paths.extend(path for path in walked if path not in _git_ignored(walked)) # Repository development skills. Not shipped, so the portability markers # they document are legal there — but their links still have to resolve. paths.extend(sorted(ROOT.glob(".claude/skills/*/SKILL.md"))) From a554fea6311b63c23c4247e249df8d9e6f361395 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:31:19 +0800 Subject: [PATCH 03/10] test(support): give the suite a home of its own `support.py` already seeds CARGENTO_HOME into the process environment, and the comment there explains why: unset, the suite reads the developer's real ~/.cargento and its verdict depends on what they happen to have dismissed. The same argument applies one level up and was never made. Harness store roots resolve from the user's home, not from CARGENTO_HOME (config.py:403 reads USERPROFILE on Windows and HOME everywhere else), so every test that collects without patching a store walked the developer's real ~/.claude, ~/.codex and ~/.cursor. On this machine that is 24,981 files under ~/.claude/projects alone. Measured on the same command, `unittest discover` over the dashboard tests: 59.5s before, 38.1s after. 2,005 tests, green both ways. Per module the effect tracks how much collecting each one does: test_quota 4.63s -> 1.69s, test_lifecycle 8.04s -> 4.97s, test_http_api 11.34s -> 8.96s. CI will not see that gain - a fresh runner's home holds no agent stores - so this is a local-developer and local-agent saving. The hermeticity half applies everywhere: a suite whose runtime and whose result both depend on what the developer's other agents were writing while it ran is not measuring what it claims to. A second temp directory rather than reusing STATE_HOME, because the two answer different questions: a test that redirects CARGENTO_HOME itself must still see an empty harness home rather than the state it just pointed elsewhere. This also subsumes the separate `_clean_env` fix in test_lifecycle: that helper copies os.environ, which now carries the seeded HOME before it is read. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- cargento/skills/cargento/tests/support.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/cargento/skills/cargento/tests/support.py b/cargento/skills/cargento/tests/support.py index bcc27a2..0c2782a 100644 --- a/cargento/skills/cargento/tests/support.py +++ b/cargento/skills/cargento/tests/support.py @@ -67,6 +67,27 @@ atexit.register(_STATE_HOME.cleanup) os.environ[CARGENTO_HOME_ENV] = STATE_HOME +# The same leak one level up. CARGENTO_HOME redirects only the dashboard's own +# state; every harness store root is resolved from the user's home instead +# (config.py:403 reads USERPROFILE on Windows and HOME everywhere else), so any +# test that collects without patching a store reads the developer's real +# ~/.claude, ~/.codex, ~/.cursor and the rest. Measured here: 24,981 files under +# ~/.claude/projects alone, and redirecting HOME took test_quota from 4.63s to +# 1.69s, test_lifecycle from 8.04s to 4.97s and test_http_api from 11.34s to +# 8.96s. Speed is the smaller half — otherwise the suite's verdict depends on +# what the developer's own agents happened to be writing while it ran. +# +# A second directory rather than STATE_HOME, because the two answer different +# questions: a test that points CARGENTO_HOME at a directory of its own must +# still see an empty harness home, not the state it just redirected. USERPROFILE +# is seeded too — unset on this platform, but config.py prefers it on Windows, +# and the suite runs natively there in platform-tests. +_USER_HOME = tempfile.TemporaryDirectory(prefix="cargento-test-user-") +USER_HOME = _USER_HOME.name +atexit.register(_USER_HOME.cleanup) +os.environ["HOME"] = USER_HOME +os.environ["USERPROFILE"] = USER_HOME + # Store key -> path. Patch with mock.patch.dict; runtime() folds it into config. STORE_OVERRIDES: dict[str, Any] = {} # str, or a tuple/list of candidates # RuntimeConfig field -> value, applied by runtime() after the build: how a test From c3fec840479035f114515058bf938b093029e25f Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:34:39 +0800 Subject: [PATCH 04/10] chore(ci): bound the unbounded job, and stop the canary chasing workflow edits `validate` was the only job in the repository with no `timeout-minutes`, so it inherited GitHub's 360-minute default. Its longest step is the same `unittest discover` that AGENTS.md records hanging on subprocess and socket waits under load, and it is a required check - so a hang held the PR pending for six hours instead of failing in ten minutes. Its checkout also lacked the `persist-credentials: false` that every other checkout in the repository sets. `plugin-compatibility` ignores prose but not workflows, so editing any workflow file installed four upstream CLIs over the network to prove that cargento's package layout had not changed. A workflow file cannot change that layout. The trade, which the filter now states: a change to the canary itself no longer runs on its own PR. The push-to-main and Monday schedule runs cover it, and it is not a required check, so nothing is gated on the gap. mypy's `files` listed three of the six entry-point scripts. The other three - the hook adapters the shipped manifests actually register - were checked only because tests/test_events_ingress.py imports them and `tests` is on the list, a guarantee lasting exactly as long as those imports. Now stated: 110 source files before, 113 after, clean either way. AGENTS.md's architecture tree named one hook, `notify_hook.py`, which no manifest registers. The three that ship - `event_hook.py` for Claude and Codex, `agy_hook.py` and `statusline_hook.py` for Antigravity - were absent, so an agent orienting on the tree would not find the adapter code that receives real harness events. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- .github/workflows/plugin-compatibility.yml | 7 +++++++ .github/workflows/validate.yml | 8 ++++++++ AGENTS.md | 3 +++ pyproject.toml | 7 +++++++ 4 files changed, 25 insertions(+) diff --git a/.github/workflows/plugin-compatibility.yml b/.github/workflows/plugin-compatibility.yml index ea54a73..85468c7 100644 --- a/.github/workflows/plugin-compatibility.yml +++ b/.github/workflows/plugin-compatibility.yml @@ -28,6 +28,13 @@ on: - 'CODE_OF_CONDUCT.md' - 'docs/**' - '.github/PULL_REQUEST_TEMPLATE.md' + - '.github/ISSUE_TEMPLATE/**' + # A workflow file cannot change the package LAYOUT this canary measures, + # and editing one used to install four upstream CLIs over the network to + # prove it. The trade is that a change to THIS file no longer runs on its + # own PR; the push-to-main and Monday schedule runs cover it, and it is + # not a required check, so nothing is gated on the gap. + - '.github/workflows/**' push: branches: - main diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index e23deba..8bf64a5 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -22,10 +22,18 @@ concurrency: jobs: validate: runs-on: ubuntu-latest + # The only job in the repository that had no timeout, so it inherited + # GitHub's 360-minute default. Its longest step is the same `unittest + # discover` that AGENTS.md records hanging on subprocess and socket waits + # under load, and this is a required check — a hang would hold the PR + # pending for six hours rather than failing in ten minutes. + timeout-minutes: 10 steps: # actions/checkout@v7.0.1 — pinned to immutable SHA (this is the # required check; mutable major tags don't belong here). - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 + with: + persist-credentials: false # actions/setup-python@v7.0.0 — pinned to immutable SHA. - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 with: diff --git a/AGENTS.md b/AGENTS.md index 2f6b0b0..8856ec5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,6 +31,9 @@ cargento/ # plugin root: Claude Code, Codex, Antigravi ├── SKILL.md # shared skill body (all harnesses) ├── server.py # the stable launcher: calls cargento_runtime.cli.main ├── notify_hook.py # loopback POST forwarder for the user-installed Claude hooks + ├── event_hook.py # posts Claude and Codex command-hook lifecycle events + ├── agy_hook.py # posts Antigravity's hook events + ├── statusline_hook.py # posts Antigravity's status-line state ├── mcp_server.py # stdio MCP server: the one tool a session calls to ask the reader ├── cargento_runtime/ # importable dashboard runtime package │ ├── aggregate.py # harness registry, failure boundary, and the application diff --git a/pyproject.toml b/pyproject.toml index 4a25163..6875e40 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -140,9 +140,16 @@ mypy_path = ["cargento/skills/cargento", "scripts"] explicit_package_bases = true files = [ "cargento/skills/cargento/cargento_runtime", + # The hook adapters the shipped manifests register. They were already checked, + # but only because tests/test_events_ingress.py imports all three and `tests` + # is on this list — a guarantee that lasts exactly as long as those imports do, + # and that nothing stated. Listed so it is stated. + "cargento/skills/cargento/agy_hook.py", + "cargento/skills/cargento/event_hook.py", "cargento/skills/cargento/mcp_server.py", "cargento/skills/cargento/notify_hook.py", "cargento/skills/cargento/server.py", + "cargento/skills/cargento/statusline_hook.py", "cargento/skills/cargento/tests", "scripts", ] From 7fc312ea05482d72261e9b7063e981623d7c83c4 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:34:40 +0800 Subject: [PATCH 05/10] refactor: delete the wrapper nothing called, and three plans that shipped `spacedock.stage_names` had no caller in the runtime, the hooks, the MCP server or the scripts - only two test assertions. The one production site that needs exactly its value, spacedock.py:676, open-codes the comprehension instead. The result was a public function tested in isolation from the path it describes: a change to either could not fail the other's tests. Deleted rather than called from production, which was the other proposal. That site already holds `entries` and uses it twice, for `stages` and for `resting`; routing it through `stage_names(config, lines)` would re-read and re-parse the frontmatter to rebuild a list it is already holding. The tests now assert against `stage_entries`, which both of those methods already use a few lines away. AGENTS.md: "docs/plans/*.md | Transient plans for unshipped work. Delete a plan once its work ships." All three event-driven phase plans shipped - `scripts/bench_collect.py` for Phase 0, `cargento_runtime/snapshot.py` for 1a, `cargento_runtime/stream.py` for 1b - and 2,592 lines of them stayed. They open by instructing an agent to execute their 114 unchecked task boxes task-by-task, which is the wrong instruction to leave lying in a repository agents read first. The design owner, `event-driven-session-observation.md`, stays: it is linked from docs/design-runtime-architecture.md and docs/captures/README.md, and it already records what each phase delivered. Its one link to a deleted plan is rewritten. `native-notifications.md` stays too - still linked from two design docs and from notifications.py, and its decision is still open. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- .../cargento/cargento_runtime/spacedock.py | 5 - .../skills/cargento/tests/test_spacedock.py | 4 +- docs/plans/event-driven-phase-0.md | 1090 ----------------- docs/plans/event-driven-phase-1a.md | 672 ---------- docs/plans/event-driven-phase-1b.md | 830 ------------- .../plans/event-driven-session-observation.md | 2 +- 6 files changed, 3 insertions(+), 2600 deletions(-) delete mode 100644 docs/plans/event-driven-phase-0.md delete mode 100644 docs/plans/event-driven-phase-1a.md delete mode 100644 docs/plans/event-driven-phase-1b.md diff --git a/cargento/skills/cargento/cargento_runtime/spacedock.py b/cargento/skills/cargento/cargento_runtime/spacedock.py index 8025f46..5bcca27 100644 --- a/cargento/skills/cargento/cargento_runtime/spacedock.py +++ b/cargento/skills/cargento/cargento_runtime/spacedock.py @@ -172,11 +172,6 @@ def stage_entries(config: RuntimeConfig, lines: list[str]) -> list[dict[str, Any return entries -def stage_names(config: RuntimeConfig, lines: list[str]) -> list[str]: - """The ordered stage names, or [] if the states block is unrecognised.""" - return [entry["name"] for entry in stage_entries(config, lines)] - - def _codex_tool_output(record: dict[str, Any]) -> list[str] | None: """A Codex rollout's tool-output text, or None when the record is not one. diff --git a/cargento/skills/cargento/tests/test_spacedock.py b/cargento/skills/cargento/tests/test_spacedock.py index d29f2be..83b0af5 100644 --- a/cargento/skills/cargento/tests/test_spacedock.py +++ b/cargento/skills/cargento/tests/test_spacedock.py @@ -78,7 +78,7 @@ def test_stage_names_read_document_order_past_sibling_blocks(self) -> None: self.assertEqual("spacedock@0.22.0", spacedock.scalar(lines, "commissioned-by")) self.assertEqual( ["intake", "review", "fix-and-harden", "escalated", "posted"], - spacedock.stage_names(config, lines), + [entry["name"] for entry in spacedock.stage_entries(config, lines)], ) # The initial and terminal flags belong to the item they are nested # under, and `gate:`/`worktree:`/the decision options are not flags. @@ -138,7 +138,7 @@ def test_stage_names_refuse_shapes_the_scanner_cannot_model(self) -> None: for label, block in cases.items(): with self.subTest(case=label): lines = ("---\n" + block + "---\n").split("\n")[1:-2] - self.assertEqual([], spacedock.stage_names(config, lines)) + self.assertEqual([], spacedock.stage_entries(config, lines)) def test_workers_are_attributed_to_a_known_slug(self) -> None: """Cycle markers appear on either side of the stage, and a slug may end diff --git a/docs/plans/event-driven-phase-0.md b/docs/plans/event-driven-phase-0.md deleted file mode 100644 index 7c54cf5..0000000 --- a/docs/plans/event-driven-phase-0.md +++ /dev/null @@ -1,1090 +0,0 @@ -# Event-driven session observation, Phase 0 Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Produce the measurements that gate DRC-4080, make the Claude collector cheap enough that those measurements mean something, and close the two quota opt-out defects that must be fixed before any later phase can claim the security contract is preserved. - -**Architecture:** No new subsystem. Phase 0 is a repeatable benchmark script under `scripts/`, three behaviour-preserving changes inside `collectors/claude.py` plus one new bounded cache on `RuntimeState`, and two small correctness fixes in `lifecycle.py` and `quota.py`. Nothing in this phase adds an event path, a snapshot, or an HTTP endpoint. - -**Tech Stack:** Python 3.11 standard library only. `unittest` for tests, `coverage` for the ratchet, `ruff` and `mypy --strict` for the gate. - -**Scope note.** This plan covers Phase 0 only. Phase 1 (materialized snapshot and SSE) gets its own plan, written after Task 7 lands, because two of its inputs are Phase 0 outputs: the direct-GET freshness threshold the design states as "2.5-second-or-better" has to be chosen against measured post-fix collection time, and the selective-reuse gate decides whether Phase 1's publish protocol is built for per-harness merges or for one full aggregate per floor. Writing those tasks now would mean guessing both. - -**Design owner:** [`event-driven-session-observation.md`](event-driven-session-observation.md). This plan implements its Phase 0 section and must not restate its rationale. - -## Global Constraints - -- Standard library only. No dependency may be added, ever. Python floor is 3.11, owned by `COMPATIBILITY.md`. -- `ruff check .` with `select = ALL` and `ruff format --check` must pass. Curated ignores live in `pyproject.toml`. -- `mypy` must pass under `--strict`. -- `coverage report` must meet `fail_under = 73` in `pyproject.toml`. The threshold only ratchets up. Never lower it. -- Tests run on Ubuntu, macOS and Windows. Any path, process or filesystem assumption needs an OS guard or a skip. -- Never edit a `version` field in any manifest. The tag-driven Release workflow owns all five, and `version-guard` fails a PR that touches them. -- Every commit uses `git commit -s` for DCO. Commit subject format is `(): `. -- `docs/plans/*.md` is inside the sync-docs tone gate: no em dashes, en dashes, or curly quotes in this file or any doc you edit. -- New runtime files must not invert R-2, the inward-only dependency rule. `test_runtime_import_graph_matches_the_reviewed_allowlist` enforces it. Phase 0 adds no runtime module, so this constraint should stay untriggered. If you find yourself adding one, stop and revisit the design doc. -- Behaviour-preserving means byte-identical `/api/data` output for the same store contents. Tasks 2 through 4 each prove that with an equivalence test, not by inspection. -- Run the pre-PR suite from `AGENTS.md` before opening the PR. Do not rely on CI to surface failures. - ---- - -### Task 1: Reproducible collection benchmark - -The Phase 0 gate needs numbers anyone can regenerate on any machine. Today the only figures are from one developer's laptop, recorded in the design doc as provisional. This task makes them reproducible before any code changes them. - -**Files:** -- Create: `scripts/bench_collect.py` -- Create: `scripts/tests/test_bench_collect.py` - -**Interfaces:** -- Consumes: `cargento_runtime.cli.build_application(config, state, clock=...)`, `cargento_runtime.config.build_runtime_config`, `cargento_runtime.state.build_runtime_state`. -- Produces: `measure(app, *, repeat: int) -> dict[str, Any]` returning `{"total_ms": float, "per_harness_ms": dict[str, float], "discovery_ms": float, "repeat": int}`, and `format_report(result: dict[str, Any]) -> str`. Task 7 calls `measure` and pastes `format_report` output into the design doc. - -**Two signatures to get right, both verified against the source:** - -- `Application.__init__` is `(config, state, harnesses, *, native_notifier, popup_notifier, diagnostic_sink, clock=time.time)`. Do not construct it directly. Use `cli.build_application(config, state, clock=time.time)`, which is how both the CLI and `tests/support.build_app` build one. -- `Application.collect` is `collect(self, *, show_all: bool)`. `show_all` is a required keyword and there is no `window_hours` parameter; the window comes from `config.window_hours`. Every call in this task passes `show_all=False`. - -- [ ] **Step 1: Write the failing test** - -Create `scripts/tests/test_bench_collect.py`: - -```python -from __future__ import annotations - -import unittest - -from scripts import bench_collect - - -class FakeSpec: - """HarnessSpec's identity field is `key`, not `name`. - - Verified against aggregate.HarnessSpec, whose fields are: collect, discover, - key, label, usage, usage_is_fetch. - """ - - def __init__(self, key: str) -> None: - self.key = key - - -class FakeApp: - """Stands in for Application: two harnesses, one slow and one fast.""" - - def __init__(self) -> None: - self.collect_calls = 0 - - def collect(self, *, show_all: bool) -> dict[str, object]: - # Mirrors Application.collect exactly: show_all is keyword-only and - # required, and there is no window_hours parameter. - self.collect_calls += 1 - return {"sessions": [], "generated": 0.0} - - -class MeasureTest(unittest.TestCase): - def test_measure_reports_total_and_repeat_count(self) -> None: - app = FakeApp() - result = bench_collect.measure(app, repeat=3) - self.assertEqual(result["repeat"], 3) - self.assertEqual(app.collect_calls, 3) - self.assertIsInstance(result["total_ms"], float) - self.assertGreaterEqual(result["total_ms"], 0.0) - - def test_measure_rejects_a_non_positive_repeat(self) -> None: - with self.assertRaises(ValueError): - bench_collect.measure(FakeApp(), repeat=0) - - def test_format_report_names_every_measured_harness(self) -> None: - report = bench_collect.format_report( - { - "total_ms": 285.0, - "per_harness_ms": {"claude": 270.0, "codex": 15.0}, - "discovery_ms": 0.4, - "repeat": 5, - } - ) - self.assertIn("claude", report) - self.assertIn("codex", report) - self.assertIn("285.0", report) - - -if __name__ == "__main__": - unittest.main() -``` - -- [ ] **Step 2: Run test to verify it fails** - -Run: `cd /Users/jaredmscott/repos/recce/cargento && python3 -m unittest scripts.tests.test_bench_collect -v` - -Expected: FAIL with `ModuleNotFoundError: No module named 'scripts.bench_collect'`. - -- [ ] **Step 3: Write minimal implementation** - -Create `scripts/bench_collect.py`: - -```python -#!/usr/bin/env python3 -"""Measure Cargento collection cost, per harness and in total. - -Phase 0 of the event-driven observation plan gates on these numbers, so they -have to be reproducible on a reviewer's machine rather than quoted from one -laptop. Timing uses perf_counter and reports a median, because a cold page -cache makes a single sample useless. -""" - -from __future__ import annotations - -import argparse -import cProfile -import pstats -import statistics -import sys -import time -from typing import Any - - -def measure(app: Any, *, repeat: int) -> dict[str, Any]: - """Median wall-clock cost of a full collect, and of each harness inside it. - - Every sample is a real collect against the caller's stores. The per-harness - figures come from wrapping each registry entry's collector, so they sum to - the total minus serialization rather than being estimated. - """ - if repeat < 1: - raise ValueError("repeat must be at least 1") - totals: list[float] = [] - per_harness: dict[str, list[float]] = {} - discovery: list[float] = [] - for _ in range(repeat): - started = time.perf_counter() - app.collect(show_all=False) - totals.append((time.perf_counter() - started) * 1000.0) - return { - "total_ms": statistics.median(totals), - "per_harness_ms": {name: statistics.median(v) for name, v in per_harness.items()}, - "discovery_ms": statistics.median(discovery) if discovery else 0.0, - "repeat": repeat, - } - - -def format_report(result: dict[str, Any]) -> str: - lines = [ - f"repeat: {result['repeat']}", - f"total_ms: {result['total_ms']}", - f"discovery_ms: {result['discovery_ms']}", - ] - for name, value in sorted( - result["per_harness_ms"].items(), key=lambda kv: -float(kv[1]) - ): - lines.append(f" {name}: {value} ms") - return "\n".join(lines) - - -def main(argv: list[str]) -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--repeat", type=int, default=5) - parser.add_argument("--window-hours", type=float, default=24.0) - parser.add_argument("--profile", action="store_true", help="cProfile one collect by function") - args = parser.parse_args(argv[1:]) - - from cargento_runtime import cli - - # build_runtime freezes config and state with no side effects, and takes the - # same namespace shape the CLI parses. usage_fetch_enabled is off here on - # purpose: a benchmark must never make an outbound vendor quota request, and - # a fetch would pollute the timing besides. - runtime_args = argparse.Namespace( - port=4553, - window_hours=args.window_hours, - no_spacedock=False, - no_usage=True, - ) - config, state = cli.build_runtime(runtime_args, started=time.time()) - # build_application, not Application(...): the constructor also requires the - # harness registry and three injected sinks, and the CLI is what assembles - # them. The no-op sink keeps store diagnostics out of the report. - app = cli.build_application( - config, - state, - clock=time.time, - diagnostic_sink=lambda _message: None, - ) - if args.profile: - profiler = cProfile.Profile() - profiler.enable() - app.collect(show_all=False) - profiler.disable() - pstats.Stats(profiler).sort_stats("cumulative").print_stats(25) - return 0 - print(format_report(measure(app, repeat=args.repeat))) - return 0 - - -if __name__ == "__main__": - sys.exit(main(sys.argv)) -``` - -- [ ] **Step 4: Run test to verify it passes** - -Run: `python3 -m unittest scripts.tests.test_bench_collect -v` - -Expected: PASS, 3 tests. - -- [ ] **Step 5: Fill in the per-harness and discovery timing** - -`measure` currently returns empty `per_harness_ms` and a zero `discovery_ms`, so the report is not yet useful. Wrap the registry so each harness is timed separately. `HarnessSpec` identifies a harness by `key`, not `name`. Its full field set, verified against `aggregate.HarnessSpec`, is `collect`, `discover`, `key`, `label`, `usage`, `usage_is_fetch`. A real registry has ten entries, and `app.harnesses` is a tuple. - -`HarnessSpec` is declared `@dataclass(frozen=True)` at `aggregate.py:31`, so assigning over `spec.collect` raises `FrozenInstanceError`. Wrap at the registry level instead: build a new tuple with `dataclasses.replace(spec, collect=wrapped, discover=wrapped_discover)` for each entry, assign it to `app.harnesses` for the duration of the measurement, and restore the original tuple in a `finally`. - -Add to `measure`, before the sample loop: - -```python - specs = tuple(getattr(app, "harnesses", ()) or ()) - for spec in specs: - per_harness.setdefault(spec.key, []) -``` - -Each wrapper records its own duration in milliseconds into `per_harness[spec.key]`, or into `discovery` for `discover`, then delegates to the original callable and returns its result unchanged. A wrapper must not swallow an exception: the per-harness failure boundary lives in `Application.collect` and the benchmark must not change which harnesses fail. - -Add two tests: a two-harness fake yields two entries in `per_harness_ms`, and a raising collector still leaves `app.harnesses` identical to the original tuple afterwards. - -- [ ] **Step 6: Run the suite and the gate** - -Run: -```bash -python3 -m unittest scripts.tests.test_bench_collect -v -ruff check scripts/bench_collect.py scripts/tests/test_bench_collect.py -ruff format --check scripts/bench_collect.py scripts/tests/test_bench_collect.py -mypy -``` -Expected: all pass. - -- [ ] **Step 7: Record the pre-fix baseline** - -Run and keep the output for Task 7: -```bash -python3 scripts/bench_collect.py --repeat 7 | tee /tmp/cargento-bench-pre.txt -python3 scripts/bench_collect.py --profile | head -40 | tee /tmp/cargento-profile-pre.txt -``` - -This is the "before" side of the comparison Tasks 2 through 4 have to beat. If `total_ms` is already low on this machine because its Claude store is small, say so in Task 7 rather than treating the fix as unnecessary: the cost scales with historical transcript count, not with active sessions. - -- [ ] **Step 8: Commit** - -```bash -git add scripts/bench_collect.py scripts/tests/test_bench_collect.py -git commit -s -m "test(bench): add a reproducible collection benchmark (DRC-4080)" -``` - ---- - -### Task 2: Stop globbing every subagent directory twice - -`claude.collect` calls `agent_transcripts(transcript)` at line 236 and then `load_subagents(config, transcript, now)` at line 237, and `load_subagents` calls `agent_transcripts` again internally. Every session prefix therefore pays the subagent glob twice. `agent_files` is genuinely used later, at lines 254 and 392, so the fix is to pass the list in rather than to delete the call. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/collectors/claude.py:96-120` (`load_subagents`) and `:236-237` (the call site) -- Test: `cargento/skills/cargento/tests/test_claude.py` - -**Interfaces:** -- Consumes: nothing from earlier tasks. -- Produces: `load_subagents(config, transcript, now, *, found: list[tuple[str, float]] | None = None) -> list[dict[str, Any]]`. When `found` is given it is used verbatim and no globbing happens. Task 4 caches what gets passed as `found`. - -Every subagent fixture below writes `agent-*.jsonl` directly into ``, which no -SUBAGENT_GLOBS pattern matches, so none of them can pass as written: the assertion that the fixture -produced at least one transcript fails first. Whether they were run and rewritten, or never run, is -not recorded here. A working fixture writes into -`/subagents/` (flat) or `/subagents/workflows//` (workflow). See -`SubagentListingCacheTest` in `cargento/skills/cargento/tests/test_claude.py` for the fixtures that -shipped. - -- [ ] **Step 1: Write the failing test** - -Add to `cargento/skills/cargento/tests/test_claude.py`: - -```python -class SubagentGlobCostTest(RuntimeTestCase): - """One subagent scan per session, not two. - - The parent transcript and one fresh subagent transcript are both real files, - so the test fails if the optimisation changes which subagents are found. - """ - - def test_load_subagents_accepts_a_precomputed_listing(self) -> None: - config, state = runtime() - now = time.time() - with tempfile.TemporaryDirectory() as root: - parent = os.path.join(root, "abcd1234-session.jsonl") - Path(parent).write_text("{}\n", encoding="utf-8") - sess_dir = os.path.join(root, "abcd1234-session") - os.makedirs(sess_dir) - child = os.path.join(sess_dir, "agent-worker.jsonl") - Path(child).write_text("{}\n", encoding="utf-8") - - found = claude_collector.agent_transcripts(parent) - self.assertTrue(found, "fixture must produce at least one subagent transcript") - - with mock.patch.object( - claude_collector, "agent_transcripts", side_effect=AssertionError("globbed again") - ): - agents = claude_collector.load_subagents(config, parent, now, found=found) - - self.assertEqual([a["label"] for a in agents], ["subagent"]) - - def test_precomputed_and_self_scanned_results_are_identical(self) -> None: - config, state = runtime() - now = time.time() - with tempfile.TemporaryDirectory() as root: - parent = os.path.join(root, "abcd1234-session.jsonl") - Path(parent).write_text("{}\n", encoding="utf-8") - sess_dir = os.path.join(root, "abcd1234-session") - os.makedirs(sess_dir) - for name in ("agent-a.jsonl", "agent-b.jsonl"): - Path(os.path.join(sess_dir, name)).write_text("{}\n", encoding="utf-8") - - self.assertEqual( - claude_collector.load_subagents(config, parent, now), - claude_collector.load_subagents( - config, parent, now, found=claude_collector.agent_transcripts(parent) - ), - ) -``` - -- [ ] **Step 2: Run test to verify it fails** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_claude.SubagentGlobCostTest -v` - -Expected: FAIL with `TypeError: load_subagents() got an unexpected keyword argument 'found'`. - -- [ ] **Step 3: Write minimal implementation** - -In `collectors/claude.py`, change the `load_subagents` signature and its first loop line: - -```python -def load_subagents( - config: RuntimeConfig, - transcript: str | None, - now: float, - *, - found: list[tuple[str, float]] | None = None, -) -> list[dict[str, Any]]: - """Running Claude subagents beneath the session directory; fresh mtime = - running. Covers both layouts in ``SUBAGENT_GLOBS``. - - ``found`` lets a caller that has already listed the directory hand the - listing over, so one session costs one scan. The collector needs the full - listing anyway for its parked-parent activity check. - """ - agents: list[dict[str, Any]] = [] - for fp, mtime in (agent_transcripts(transcript) if found is None else found): -``` - -Leave the rest of the body unchanged. - -- [ ] **Step 4: Update the call site** - -At `collectors/claude.py:236-237`, pass the listing through: - -```python - agent_files = agent_transcripts(transcript) - subagents = load_subagents(config, transcript, now, found=agent_files) -``` - -- [ ] **Step 5: Run tests to verify they pass** - -Run: -```bash -python3 -m unittest cargento.skills.cargento.tests.test_claude -v -``` -Expected: PASS, including the pre-existing Claude tests. Those are the equivalence check: if any of them change behaviour, the optimisation is wrong. - -- [ ] **Step 6: Prove the collector output is unchanged** - -Run the whole suite, not just the Claude module, because `sessions`, `spacedock` and page tests all consume collector output: -```bash -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` -Expected: PASS with no change in count. - -- [ ] **Step 7: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/collectors/claude.py \ - cargento/skills/cargento/tests/test_claude.py -git commit -s -m "perf(claude): scan each session's subagents once, not twice (DRC-4080)" -``` - ---- - -### Task 3: Skip the subagent glob when the session directory is absent - -Most historical prefixes are expected to have no session directory at all, and `agent_transcripts` -still runs every pattern in `SUBAGENT_GLOBS` against a path that does not exist. An `isdir` check -replaces the glob of every pattern with one stat. Neither the prefix ratio nor the syscall saving is -measured; the timing delta in Step 5 is what settles whether the check is worth having. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/collectors/claude.py:77-92` (`agent_transcripts`) -- Test: `cargento/skills/cargento/tests/test_claude.py` - -**Interfaces:** -- Consumes: `load_subagents(..., found=...)` from Task 2 (unchanged by this task). -- Produces: no signature change. `agent_transcripts` returns `[]` without globbing when the session directory does not exist. - -- [ ] **Step 1: Write the failing test** - -Add to `SubagentGlobCostTest` in `test_claude.py`: - -```python - def test_absent_session_directory_is_not_globbed(self) -> None: - with tempfile.TemporaryDirectory() as root: - parent = os.path.join(root, "abcd1234-session.jsonl") - Path(parent).write_text("{}\n", encoding="utf-8") - # No sibling "abcd1234-session/" directory: the common case for a - # historical session that never ran a subagent. - with mock.patch.object( - claude_collector.runtime_io, - "glob_under", - side_effect=AssertionError("globbed a directory that does not exist"), - ): - self.assertEqual(claude_collector.agent_transcripts(parent), []) - - def test_present_session_directory_is_still_globbed(self) -> None: - with tempfile.TemporaryDirectory() as root: - parent = os.path.join(root, "abcd1234-session.jsonl") - Path(parent).write_text("{}\n", encoding="utf-8") - sess_dir = os.path.join(root, "abcd1234-session") - os.makedirs(sess_dir) - Path(os.path.join(sess_dir, "agent-a.jsonl")).write_text("{}\n", encoding="utf-8") - found = claude_collector.agent_transcripts(parent) - self.assertEqual([os.path.basename(p) for p, _ in found], ["agent-a.jsonl"]) -``` - -- [ ] **Step 2: Run test to verify it fails** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_claude.SubagentGlobCostTest -v` - -Expected: `test_absent_session_directory_is_not_globbed` FAILS with `AssertionError: globbed a directory that does not exist`. - -- [ ] **Step 3: Write minimal implementation** - -In `agent_transcripts`, after computing `sess_dir` and before the pattern loop: - -```python - sess_dir = os.path.join( - os.path.dirname(transcript), os.path.basename(transcript)[: -len(".jsonl")] - ) - # Most historical prefixes never ran a subagent, so the directory is absent. - # One stat is cheaper than running every SUBAGENT_GLOBS pattern against a - # path that cannot match. - if not os.path.isdir(sess_dir): - return [] - found: list[tuple[str, float]] = [] -``` - -- [ ] **Step 4: Run tests to verify they pass** - -Run: -```bash -python3 -m unittest cargento.skills.cargento.tests.test_claude -v -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` -Expected: PASS. - -- [ ] **Step 5: Confirm the improvement is real** - -Run: `python3 scripts/bench_collect.py --repeat 7` - -Compare `total_ms` against `/tmp/cargento-bench-pre.txt`. Record the delta for Task 7. If there is no measurable change, the machine's Claude store has few historical prefixes; note that rather than reverting, and say so in Task 7. - -- [ ] **Step 6: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/collectors/claude.py \ - cargento/skills/cargento/tests/test_claude.py -git commit -s -m "perf(claude): skip the subagent glob when no session dir exists (DRC-4080)" -``` - ---- - -### Task 4: Cache subagent listings on session-directory mtime - -Task 4 did not ship as written, and its premise is false. A subagent transcript never lives in the -session directory itself: SUBAGENT_GLOBS matches `/subagents/agent-*.jsonl` and -`/subagents/workflows/*/agent-*.jsonl`. Keying on the session directory's own mtime misses -every flat agent after the first, every workflow agent, and the create-then-write race where a run -directory exists a beat before its first transcript. The cached value must be paths only, because -appending to a transcript moves no directory and a cached mtime would go stale. What shipped -fingerprints every directory a pattern can reach and stamps it before the listing: see -`subagent_tree_stamp` in `collectors/claude.py`, and `event-driven-session-observation.md` lines -174-184. The task below is kept as the prototype it was, with its false claims marked in place. - -Tasks 2 and 3 cut the constant factor. This one cuts the work itself: a session directory whose mtime has not moved cannot have gained or lost a transcript, so its listing can be reused across collections. The parked-parent case makes this delicate, so the cache key is the directory mtime and nothing else. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/state.py:41` area (add one cache field) -- Modify: `cargento/skills/cargento/cargento_runtime/collectors/claude.py:77-92` (`agent_transcripts`) -- Test: `cargento/skills/cargento/tests/test_claude.py` - -**Interfaces:** -- Consumes: `agent_transcripts` behaviour from Task 3. -- Produces: `agent_transcripts(transcript, *, config: RuntimeConfig | None = None, state: RuntimeState | None = None) -> list[tuple[str, float]]`. With both `config` and `state` supplied it consults and fills `state.claude_subagent_cache`. With either omitted it behaves exactly as in Task 3, so existing callers and tests keep working. - -- [ ] **Step 1: Write the failing test** - -Add to `test_claude.py`: - -```python -class SubagentListingCacheTest(RuntimeTestCase): - """A session directory with an unchanged mtime is listed once. - - Keyed on directory mtime rather than on a freshness window, because a - workflow that runs for hours parks its parent transcript and keeps writing - only to subagent files. Dropping old prefixes would lose those sessions. - """ - - def _fixture(self, root: str, *names: str) -> str: - parent = os.path.join(root, "abcd1234-session.jsonl") - Path(parent).write_text("{}\n", encoding="utf-8") - sess_dir = os.path.join(root, "abcd1234-session") - os.makedirs(sess_dir, exist_ok=True) - for name in names: - Path(os.path.join(sess_dir, name)).write_text("{}\n", encoding="utf-8") - return parent - - def test_second_call_with_unchanged_mtime_does_not_glob(self) -> None: - config, state = runtime() - with tempfile.TemporaryDirectory() as root: - parent = self._fixture(root, "agent-a.jsonl") - first = claude_collector.agent_transcripts(parent, config=config, state=state) - self.assertTrue(first) - with mock.patch.object( - claude_collector.runtime_io, - "glob_under", - side_effect=AssertionError("re-globbed an unchanged directory"), - ): - second = claude_collector.agent_transcripts(parent, config=config, state=state) - self.assertEqual(first, second) - - def test_a_new_subagent_file_invalidates_the_entry(self) -> None: - config, state = runtime() - with tempfile.TemporaryDirectory() as root: - parent = self._fixture(root, "agent-a.jsonl") - first = claude_collector.agent_transcripts(parent, config=config, state=state) - sess_dir = os.path.join(root, "abcd1234-session") - # Force a distinct directory mtime: a coarse filesystem timestamp - # would otherwise make this test pass or fail on timing alone. - Path(os.path.join(sess_dir, "agent-b.jsonl")).write_text("{}\n", encoding="utf-8") - os.utime(sess_dir, (time.time() + 5, time.time() + 5)) - second = claude_collector.agent_transcripts(parent, config=config, state=state) - self.assertEqual(len(first), 1) - self.assertEqual(len(second), 2) - - def test_without_state_the_behaviour_is_uncached(self) -> None: - with tempfile.TemporaryDirectory() as root: - parent = self._fixture(root, "agent-a.jsonl") - self.assertEqual( - claude_collector.agent_transcripts(parent), - claude_collector.agent_transcripts(parent), - ) - - def test_a_parked_parent_keeps_its_subagent_activity(self) -> None: - """The regression this cache design exists to avoid. - - The parent transcript is hours old; only the subagent file is fresh. The - session must still report its subagent activity. - """ - config, state = runtime() - with tempfile.TemporaryDirectory() as root: - parent = self._fixture(root, "agent-a.jsonl") - stale = time.time() - 6 * 3600 - os.utime(parent, (stale, stale)) - found = claude_collector.agent_transcripts(parent, config=config, state=state) - self.assertEqual(len(found), 1) - self.assertGreater(found[0][1], stale) -``` - -- [ ] **Step 2: Run test to verify it fails** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_claude.SubagentListingCacheTest -v` - -Expected: FAIL with `TypeError: agent_transcripts() got an unexpected keyword argument 'config'`. - -- [ ] **Step 3: Add the cache field** - -In `cargento_runtime/state.py`, beside the other Claude caches (near `claude_user_event_cache`): - -```python - # This shape did not ship, and the invariant below is false: the session - # directory's own mtime does not move when a workflow run gains its first - # agent, and a cached mtime goes stale because appending to a transcript - # moves no directory. What shipped is - # dict[str, tuple[tuple[float, ...], list[str]]] - a fingerprint over every - # directory a pattern can reach, and paths only. See state.py in the repo. - # sess_dir -> (directory mtime, listing). A directory whose mtime has not - # moved cannot have gained or lost a transcript, and keying on mtime rather - # than on a freshness window keeps a parked parent's subagents visible. - claude_subagent_cache: dict[str, tuple[float, list[tuple[str, float]]]] = field( - default_factory=dict - ) -``` - -- [ ] **Step 4: Write the implementation** - -Rewrite `agent_transcripts` in `collectors/claude.py`: - -```python -def agent_transcripts( - transcript: str | None, - *, - config: RuntimeConfig | None = None, - state: RuntimeState | None = None, -) -> list[tuple[str, float]]: - """(path, mtime) for every subagent transcript belonging to a session. - - With ``config`` and ``state`` the listing is memoised on the session - directory's mtime. That key is not sufficient and did not ship; see the - correction at the head of this task. Without them the scan is unmemoised, - so callers outside a runtime keep working. - """ - if not transcript: - return [] - sess_dir = os.path.join( - os.path.dirname(transcript), os.path.basename(transcript)[: -len(".jsonl")] - ) - # Most historical prefixes never ran a subagent, so the directory is absent. - # One stat is cheaper than running every SUBAGENT_GLOBS pattern against a - # path that cannot match. - try: - dir_mtime = os.stat(sess_dir).st_mtime - except OSError: - return [] # absent, or a file where a directory was expected - if not os.path.isdir(sess_dir): - return [] - cache = None if state is None or config is None else state.claude_subagent_cache - if cache is not None: - hit = cache.get(sess_dir) - if hit is not None and hit[0] == dir_mtime: - return list(hit[1]) - found: list[tuple[str, float]] = [] - for pattern in SUBAGENT_GLOBS: - for fp in runtime_io.glob_under(sess_dir, *pattern): - try: - found.append((fp, os.path.getmtime(fp))) - except OSError: - continue # transcript rotated/deleted between glob and stat - if cache is not None and config is not None: - runtime_state.bounded_put( - cache, sess_dir, (dir_mtime, list(found)), limit=config.max_cache_entries - ) - return found -``` - -Add the imports this needs if they are not already present: `from cargento_runtime import state as runtime_state` and the `RuntimeState` type under `TYPE_CHECKING`. Check the existing import block first and match its style. - -The returned list is copied on both the hit and the store path so a caller cannot mutate the cached listing. - -- [ ] **Step 5: Thread config and state through the call site** - -At `collectors/claude.py:236`: - -```python - agent_files = agent_transcripts(transcript, config=config, state=state) -``` - -Confirm `state` is in scope in `collect`. If it is not, read the function signature and thread it from the collector's arguments rather than reaching for a global. - -- [ ] **Step 6: Run tests to verify they pass** - -Run: -```bash -python3 -m unittest cargento.skills.cargento.tests.test_claude -v -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` -Expected: PASS. - -- [ ] **Step 7: Verify the cache is bounded and typed** - -Run: -```bash -mypy -ruff check cargento/skills/cargento/cargento_runtime/ -``` -Expected: PASS. `mypy --strict` is what catches a wrong cache value shape here. - -- [ ] **Step 8: Measure** - -Run: `python3 scripts/bench_collect.py --repeat 7 | tee /tmp/cargento-bench-post.txt` - -Expected: `claude` per-harness time materially below the pre-fix figure on a machine with real history. Keep the file for Task 7. - -- [ ] **Step 9: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/state.py \ - cargento/skills/cargento/cargento_runtime/collectors/claude.py \ - cargento/skills/cargento/tests/test_claude.py -git commit -s -m "perf(claude): memoise subagent listings on session-dir mtime (DRC-4080)" -``` - ---- - -### Task 5: Carry `--no-usage` through the Windows respawn - -Read from the source rather than run: `spawn_argv` rebuilds the child's argv from the parsed namespace and forwards `--port`, `--window-hours` and `--no-spacedock` but not `--no-usage`, so on Windows, which has no fork and always respawns, the child loses the opt-out. Whether the child then performs an outbound fetch has not been observed on Windows; the argv is what the test pins. `SECURITY.md` states that with the feature off nothing is fetched, so a child that fetched would violate a published contract. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/lifecycle.py:515-539` (`spawn_argv`) -- Test: `cargento/skills/cargento/tests/test_lifecycle.py` - -**Interfaces:** -- Consumes: nothing from earlier tasks. -- Produces: no signature change. `spawn_argv` output now contains `--no-usage` when `args.no_usage` is true. - -- [ ] **Step 1: Write the failing test** - -Add to `cargento/skills/cargento/tests/test_lifecycle.py`: - -```python -class SpawnArgvOptOutTest(unittest.TestCase): - """Every opt-out the parent was given has to reach the respawned child. - - Windows has no fork, so the daemon is always a respawn. A flag dropped here - is a flag silently ignored for every Windows daemon user. - """ - - def _args(self, **overrides: object) -> argparse.Namespace: - base = { - "port": 4553, - "window_hours": 24.0, - "no_spacedock": False, - "no_usage": False, - } - base.update(overrides) - return argparse.Namespace(**base) - - def test_no_usage_is_forwarded(self) -> None: - config = support.cfg() - argv = lifecycle.spawn_argv(config, self._args(no_usage=True)) - self.assertIn("--no-usage", argv) - - def test_no_usage_is_absent_when_not_requested(self) -> None: - config = support.cfg() - argv = lifecycle.spawn_argv(config, self._args(no_usage=False)) - self.assertNotIn("--no-usage", argv) - - def test_daemon_is_never_forwarded(self) -> None: - """Forwarding --daemon would respawn forever.""" - config = support.cfg() - argv = lifecycle.spawn_argv(config, self._args(no_usage=True)) - self.assertNotIn("--daemon", argv) -``` - -Import `argparse` and `lifecycle` at the top of the module if they are not already imported, and match how the existing tests obtain a config. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_lifecycle.SpawnArgvOptOutTest -v` - -Expected: `test_no_usage_is_forwarded` FAILS with `AssertionError: '--no-usage' not found in [...]`. - -- [ ] **Step 3: Write minimal implementation** - -In `spawn_argv`, beside the existing `no_spacedock` branch: - -```python - if args.no_spacedock: - argv.append("--no-spacedock") - if args.no_usage: - argv.append("--no-usage") - return argv -``` - -Update the docstring's promise so the next reader knows the rule: every opt-out is forwarded, `--daemon` deliberately is not. - -- [ ] **Step 4: Run tests to verify they pass** - -Run: -```bash -python3 -m unittest cargento.skills.cargento.tests.test_lifecycle -v -``` -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/lifecycle.py \ - cargento/skills/cargento/tests/test_lifecycle.py -git commit -s -m "fix(lifecycle): forward --no-usage to the respawned daemon (DRC-4080)" -``` - ---- - -### Task 6: Discard pushed quota when server-side usage is disabled - -`quota.receive_statusline` shapes and stores whatever `POST /api/usage` delivers, without consulting `config.usage_fetch_enabled`. So `--no-usage` suppresses outbound fetching but not pushed retention, and a user who turned usage off still gets a quota band from Antigravity's status line. The lifecycle fields in that payload stay useful as a dirty signal, so the fix drops quota before storage rather than rejecting the request. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/quota.py` (`receive_statusline`) -- Test: `cargento/skills/cargento/tests/test_quota.py` - -**Interfaces:** -- Consumes: nothing from earlier tasks. -- Produces: `receive_statusline(state, payload, *, now, config: RuntimeConfig | None = None) -> dict[str, Any]`. When `config` is supplied and `config.usage_fetch_enabled` is false, the stored entry list is empty and the response reports `usage: 0`. The endpoint keeps returning 200 so the harness's status line never sees an error. - -- [ ] **Step 1: Write the failing test** - -Add to `cargento/skills/cargento/tests/test_quota.py`: - -```python -class PushedReceiptOptOutTest(RuntimeTestCase): - """--no-usage means no quota is retained, pushed or fetched. - - SECURITY.md publishes that with the feature off nothing is fetched or - retained. A pushed receipt is a second way in, so it needs the same gate. - """ - - def _payload(self) -> dict[str, Any]: - return { - "quota": { - "gemini-2.5-pro": { - "remaining_fraction": 0.42, - "reset_in_seconds": 3600, - } - } - } - - def test_a_receipt_is_stored_when_usage_is_enabled(self) -> None: - config = support.make_config(usage_fetch_enabled=True) - state = support.state_of() - response = quota.receive_statusline( - state, self._payload(), now=1000.0, config=config - ) - self.assertGreater(response["usage"], 0) - self.assertTrue(state.usage_receipts["antigravity"]["entries"]) - - def test_quota_is_dropped_before_storage_when_usage_is_disabled(self) -> None: - config = support.make_config(usage_fetch_enabled=False) - state = support.state_of() - response = quota.receive_statusline( - state, self._payload(), now=1000.0, config=config - ) - self.assertEqual(response["usage"], 0) - self.assertEqual(state.usage_receipts["antigravity"]["entries"], []) - - def test_the_endpoint_still_reports_success_when_disabled(self) -> None: - """A status-line command must never see an error from Cargento.""" - config = support.make_config(usage_fetch_enabled=False) - state = support.state_of() - response = quota.receive_statusline( - state, self._payload(), now=1000.0, config=config - ) - self.assertTrue(response["ok"]) -``` - -Read the top of `test_quota.py` first and match how it builds a config and state. If `support.make_config` does not accept `usage_fetch_enabled`, use `support.config_patch` instead. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_quota.PushedReceiptOptOutTest -v` - -Expected: `test_quota_is_dropped_before_storage_when_usage_is_disabled` FAILS, because the entries list is populated. - -- [ ] **Step 3: Write minimal implementation** - -In `quota.py`: - -```python -def receive_statusline( - state: RuntimeState, - payload: dict[str, Any], - *, - now: float, - config: RuntimeConfig | None = None, -) -> dict[str, Any]: - """Store a pushed status-line receipt. Returns the endpoint's wire response. - - Storing an empty entry list on an unusable payload is deliberate: it stamps - the arrival, so a harness that stops reporting quota goes stale and drops - out of the band rather than showing whatever it last said forever. - - With server-side usage disabled the quota fields are dropped before storage, - not rejected at the door: SECURITY.md promises nothing is retained with the - feature off, and the response still reports success so a harness's status - line never surfaces a Cargento error. - """ - enabled = True if config is None else config.usage_fetch_enabled - entries = shape_statusline(payload, now) if enabled else [] - with state.usage_fetch_lock: - state.usage_receipts["antigravity"] = {"ts": now, "entries": entries} - return {"ok": True, "usage": len(entries)} -``` - -- [ ] **Step 4: Pass the config from the endpoint** - -In `http_api.py`, in `_usage_receipt`, add the argument: - -```python - response = quota.receive_statusline( - application.state, - payload, - now=application.clock(), - config=application.config, - ) -``` - -- [ ] **Step 5: Run tests to verify they pass** - -Run: -```bash -python3 -m unittest cargento.skills.cargento.tests.test_quota cargento.skills.cargento.tests.test_http_api -v -``` -Expected: PASS. - -- [ ] **Step 6: Add the endpoint-level test** - -A unit test on `receive_statusline` does not prove the endpoint wires the config through. Add one to `test_http_api.py` that posts to `/api/usage` on an application built with `usage_fetch_enabled=False` and asserts the response body reports `usage: 0` and that `state.usage_receipts["antigravity"]["entries"]` is empty. Use the existing `support.make_server` and `support.serve_until_closed` helpers, matching how the other `/api/usage` tests in that module are written. - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_http_api -v` -Expected: PASS. - -- [ ] **Step 7: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/quota.py \ - cargento/skills/cargento/cargento_runtime/http_api.py \ - cargento/skills/cargento/tests/test_quota.py \ - cargento/skills/cargento/tests/test_http_api.py -git commit -s -m "fix(quota): drop pushed quota when server-side usage is off (DRC-4080)" -``` - ---- - -### Task 7: Evaluate the gates and record the decision - -Phase 0 exists to produce a decision, not just faster code. This task writes the measured numbers into the design doc, replacing the provisional single-machine table, and states which gates passed. - -**Files:** -- Modify: `docs/plans/event-driven-session-observation.md` (the Recommendation measurement table and the Phase 0 gate list) -- Modify: `SECURITY.md` (record that the two opt-out defects are fixed, if it describes them as exposures) - -**Interfaces:** -- Consumes: `bench_collect.measure` and `format_report` from Task 1; `/tmp/cargento-bench-pre.txt` and `/tmp/cargento-bench-post.txt`. -- Produces: a filled measurement table and an explicit gate verdict that Phase 1's plan depends on. - -- [ ] **Step 1: Collect the post-fix numbers on every OS you can reach** - -Run on each available platform: -```bash -python3 scripts/bench_collect.py --repeat 7 -python3 scripts/bench_collect.py --profile | head -40 -``` - -Record the machine's Claude history size alongside each figure, because the cost scales with it: -```bash -python3 -c "import glob,os;p=os.path.expanduser('~/.claude/projects');print(len(glob.glob(p+'/*/*.jsonl')))" -``` - -If you can only reach one OS, say so explicitly in the doc rather than presenting one platform's numbers as the matrix. - -- [ ] **Step 2: Replace the provisional table** - -In `docs/plans/event-driven-session-observation.md`, update the Recommendation measurement table with before and after columns and the transcript count each was taken against. Keep the sentence that these must be reproduced elsewhere only if they still have not been. - -Remember the tone gate: no em dashes, en dashes, or curly quotes. - -- [ ] **Step 3: Evaluate the selective-reuse gate** - -The design states: if per-harness reuse saves less than 25% of post-fix collection time, keep the coordinator but run one full aggregate collection per floor, and do not build the dirty queue. - -Compute it from the post-fix per-harness figures: the saving available to per-harness reuse is the total minus the largest single harness, as a fraction of the total. Write the arithmetic into the doc so a reviewer can check it, then state the verdict as one sentence. - -Before writing the verdict, state the store profile it came from: file count per harness, not just -total collection time. If one harness's store is more than a few times larger than the others, the -arithmetic measures that skew and not the product, and the verdict is NOT DECIDABLE rather than -failed. A gate may be recorded as failed only from a profile that resembles a user running several -harnesses. - -- [ ] **Step 4: State the remaining gate verdicts** - -For each of the other three gates in the Phase 0 section, write either the measured verdict or an explicit "not yet measured, blocks the phase it gates": - -- Coarse probe: mutation corpus false negatives and CPU/IO budget on three OSes. This plan does not build the probe, so this is expected to be unmeasured. Say so. -- Adapter semantics: contract or real-CLI fixtures per transition. Unmeasured until Phase 2. -- Operational rollout: CPU duty, memory, thread ceilings, p95 render latency, missed-event repair rate. Unmeasured until Phase 1 delivers a render path. - -Do not mark a gate passed because it was not reached. - -- [ ] **Step 5: Run the docs gate** - -Run: -```bash -python3 scripts/validate_plugins.py -``` -Expected: exit 0. - -Then run the tone check. Do not retype its pattern here: copy check (e) verbatim out of -`.claude/skills/sync-docs/SKILL.md`, because the pattern is a list of the characters it bans and -reproducing it inside this file would make this file fail its own check. Expected: `tone clean`. - -- [ ] **Step 6: Run the full pre-PR suite** - -Run the canonical block from `AGENTS.md` § Pre-PR Checks in full. At minimum: -```bash -ruff check . -ruff format --check . -mypy -python3 scripts/lint_embedded.py --allow-missing-node -python3 scripts/validate_plugins.py -python3 scripts/bump_version.py --current -coverage erase -coverage run -m unittest discover -s cargento/skills/cargento/tests -t . -coverage run -a -m unittest scripts.tests.test_validate_plugins scripts.tests.test_bump_version scripts.tests.test_lint_embedded scripts.tests.test_bench_collect -coverage report -``` -Expected: all pass, coverage at or above `fail_under`. - -If coverage dropped below the floor, add tests rather than lowering the threshold. The threshold only ratchets up. - -- [ ] **Step 7: Reconcile the docs** - -Run `/sync-docs` and let it commit any doc updates onto this branch. `scripts/bench_collect.py` is a new script, so check whether it belongs in any inventory the validator owns. - -- [ ] **Step 8: Commit and open the PR** - -```bash -git add -A -git commit -s -m "docs(plans): record Phase 0 measurements and gate verdicts (DRC-4080)" -git push -u origin HEAD -gh pr create -``` - -The PR body should state the gate verdicts up front, because they decide what Phase 1's plan is allowed to build. Include `Closes DRC-4080` only if the ticket is scoped to Phase 0; it is not, so reference it instead. - ---- - -## Self-Review - -**Spec coverage against the design doc's Phase 0 section:** - -| Phase 0 requirement | Task | -|---|---| -| Cold and memo-hit `/api/data` duration | 1 (partial: the script times `collect`; a memo-hit variant needs `collect_json`, add in Task 1 Step 5) | -| Discovery and collection duration per harness | 1 | -| `cProfile` of the slowest collector by function | 1 (`--profile`) | -| Files or bytes consulted | Not covered. Left out deliberately: the design says "where cheaply measurable", and no counter exists without instrumenting `io.py`. Task 7 Step 4 records it as unmeasured. | -| Number of collections with one and several tabs | Not covered by this plan. Needs the Phase 1 render path to be meaningful. Recorded as unmeasured in Task 7. | -| Forwarder p50/p95/p99 per OS | Not covered. Depends on authenticated discovery, which Phase 2 designs. Recorded as unmeasured. | -| Native hook event count per turn | Not covered. Recorded as unmeasured. | -| Probe dependency table and mutation corpus | Not covered. This plan does not build the probe. Task 7 Step 4 states the gate is unreached. | -| Collector fixes from "Make collection cheaper" | 2, 3, 4 | -| Parked-parent and nested-workflow equivalence tests | 4 Step 1 (`test_a_parked_parent_keeps_its_subagent_activity`), plus the full-suite equivalence runs in 2 Step 6 and 3 Step 4 | -| `--no-usage` Windows respawn defect | 5 | -| Pushed-receipt discard defect | 6 | -| The four independent gates | 7 | - -Four measurement items are deliberately out of scope because they depend on later phases. That is recorded in Task 7 Step 4 rather than silently dropped, so no gate can be reported as passed when it was merely unreached. - -**Placeholder scan:** Task 1 Step 5 and Task 6 Step 6 describe work without a full code block. Both are cases where the correct code depends on names the implementer must read first (`HarnessSpec` field names; the existing `/api/usage` test style), and both name the exact file and the exact assertion required. Every other step carries runnable content. - -**Type consistency:** `agent_transcripts` gains keyword-only `config` and `state` in Task 4 and is called with both at `claude.py:236`. `load_subagents` gains keyword-only `found` in Task 2 and is called with it at the same site, so Task 4's change to `agent_transcripts` feeds Task 2's `found` parameter without a signature clash. `receive_statusline` gains keyword-only `config` in Task 6, defaulted to `None` so the existing call in `http_api.py` keeps type-checking until Step 4 updates it. `state.claude_subagent_cache` is `dict[str, tuple[float, list[tuple[str, float]]]]`, matching what `agent_transcripts` returns. diff --git a/docs/plans/event-driven-phase-1a.md b/docs/plans/event-driven-phase-1a.md deleted file mode 100644 index 459d2ce..0000000 --- a/docs/plans/event-driven-phase-1a.md +++ /dev/null @@ -1,672 +0,0 @@ -# Event-driven session observation, Phase 1a Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Replace the response memo with a versioned in-memory snapshot whose revision survives a server restart, without changing what any client sees. - -**Architecture:** One new runtime module, `snapshot.py`, holding the published `(revision, bytes)` value and its lock. `Application.collect_json` becomes a thin reader over it. `/api/data` gains a revision header and keeps its freshness guarantee. No background thread, no new endpoint, no client change. - -**Tech Stack:** Python 3.11 standard library only. `unittest`, `coverage`, `ruff`, `mypy --strict`. - -## Why this is its own PR - -Phase 1 in the design doc is snapshot plus SSE. Those split cleanly and this is the first half: - -- **1a, this plan.** Snapshot plumbing. Invisible to users: the JSON body, its freshness, and the five-second page poll are all unchanged. Reviewable as a pure refactor with one new observable, a response header. -- **1b, next.** `GET /api/stream`, leader-tab election, browser reconnect, the demand-scoped producer, the time-derived-field tick, and the quota consent lease. That is where behaviour changes. - -Deliberately no background producer here. A producer exists to keep the snapshot warm for a connected stream, and there is no stream until 1b. Building one now would mean a timer with no consumer, which is exactly the "no work while nobody is looking" regression the design warns about. - -**Ticket:** DRC-4084. **Design owner:** [`event-driven-session-observation.md`](event-driven-session-observation.md) § Phase 1. **Stacked on:** the Phase 0 branch, PR #83. Do not retarget this branch at `main` until #83 merges. - -## Global Constraints - -- Standard library only. No dependency, ever. Python floor 3.11, owned by `COMPATIBILITY.md`. -- `ruff check .` (`select = ALL`) and `ruff format --check .` must pass. -- `mypy` must pass under `--strict`. -- `coverage report` must meet `fail_under` in `pyproject.toml`. It only ratchets up. -- Tests run on Ubuntu, macOS and Windows. -- Never edit a `version` field. `version-guard` fails the PR. -- `git commit -s` for DCO. Subject format `(): `. -- `docs/plans/*.md` is inside the sync-docs tone gate: no em dashes, en dashes, or curly quotes. -- **R-2, inward-only imports.** `snapshot.py` must import no runtime module except `config` for types. Adding it requires a reviewed entry in `RuntimeImportGraphTest.EXPECTED` in `cargento/skills/cargento/tests/test_contracts.py`. That entry is the ownership decision, not a formality: if you find yourself adding `aggregate` to snapshot's set, the design is wrong. -- **Behaviour-preserving.** For the same store contents, `/api/data` must return byte-identical JSON - with the same worst-case staleness as today. Task 5 checks that, but it does not prove it: the - in-suite check in Step 1 collects twice through the changed code and never compares against the - Phase 0 branch, so it can catch a payload that drifts between two calls and nothing else. The only - comparison against the old behaviour is the hand diff in Step 2, run once, outside the suite. Read - byte-identity as confirmed by hand and not guarded by the suite. The shipped version of the Step 1 - check also drifted from what Step 1 asks for, read the developer's real store, and was muted with - `@unittest.skip`. DRC-4088 repaired it: a seeded fixture store, a pinned clock, and an assertion on - the collected session ids so an override that misses fails loudly instead of comparing two empty - payloads. The check is back in the suite, and still only checks what Step 1 asks of it. - ---- - -### Task 1: The snapshot container - -**Files:** -- Create: `cargento/skills/cargento/cargento_runtime/snapshot.py` -- Modify: `cargento/skills/cargento/tests/test_contracts.py` (the `EXPECTED` allowlist) -- Test: `cargento/skills/cargento/tests/test_snapshot.py` - -**Interfaces:** -- Consumes: `cargento_runtime.config.RuntimeConfig` for typing only, under `TYPE_CHECKING`. -- Produces: - - `Revision = tuple[float, int]`, the pair `(server_started, counter)`. - - `class Snapshot` with `publish(key, body) -> Revision`, `current(key) -> tuple[Revision, bytes] | None`, and `age(key, now) -> float | None`. - - `format_revision(rev: Revision) -> str` rendering `"."`. - -`key` is the existing memo key shape, `(window_hours, show_all)`, so the two response variants stay separate exactly as they are today. - -- [ ] **Step 1: Write the failing test** - -Create `cargento/skills/cargento/tests/test_snapshot.py`: - -```python -from __future__ import annotations - -import unittest - -from cargento_runtime import snapshot as runtime_snapshot - - -class SnapshotTest(unittest.TestCase): - def _snap(self, started: float = 1000.0) -> runtime_snapshot.Snapshot: - return runtime_snapshot.Snapshot(server_started=started) - - def test_publishing_returns_a_monotonic_counter(self) -> None: - snap = self._snap() - first = snap.publish((24.0, False), b'{"a":1}') - second = snap.publish((24.0, False), b'{"a":2}') - self.assertEqual(first[1] + 1, second[1]) - - def test_the_counter_is_shared_across_keys_so_it_orders_the_whole_process(self) -> None: - # A client holds one cursor, not one per variant, so a per-key counter - # would let ?all=1 and the default view report the same number for - # different states. - snap = self._snap() - a = snap.publish((24.0, False), b"{}") - b = snap.publish((24.0, True), b"{}") - self.assertNotEqual(a[1], b[1]) - - def test_the_revision_carries_the_server_start_stamp(self) -> None: - snap = self._snap(started=1234.5) - rev, _body = snap.current((24.0, False)) or (None, None) - self.assertIsNone(rev) - published = snap.publish((24.0, False), b"{}") - self.assertEqual(published[0], 1234.5) - - def test_current_returns_the_published_bytes_and_its_revision(self) -> None: - snap = self._snap() - rev = snap.publish((24.0, False), b'{"x":1}') - got = snap.current((24.0, False)) - self.assertIsNotNone(got) - assert got is not None - self.assertEqual(got, (rev, b'{"x":1}')) - - def test_an_unpublished_key_is_absent_rather_than_empty(self) -> None: - self.assertIsNone(self._snap().current((24.0, True))) - - def test_age_measures_from_the_publish_clock(self) -> None: - snap = self._snap() - snap.publish((24.0, False), b"{}", now=500.0) - self.assertAlmostEqual(snap.age((24.0, False), now=502.5), 2.5) - - def test_age_of_an_unpublished_key_is_none_not_zero(self) -> None: - # Zero would read as "fresh" and skip the collection a cold GET needs. - self.assertIsNone(self._snap().age((24.0, False), now=1.0)) - - def test_format_revision_is_stable_and_restart_qualified(self) -> None: - self.assertEqual(runtime_snapshot.format_revision((1700000000.0, 7)), "1700000000.7") - - def test_two_snapshots_with_different_starts_never_collide(self) -> None: - # The whole point of the pair: a tab holding revision 512 from a previous - # process must not treat the new process's revision 3 as older. - a = self._snap(started=1000.0).publish((24.0, False), b"{}") - b = self._snap(started=2000.0).publish((24.0, False), b"{}") - self.assertEqual(a[1], b[1]) - self.assertNotEqual(a, b) - - -if __name__ == "__main__": - unittest.main() -``` - -- [ ] **Step 2: Run test to verify it fails** - -Run: `cd /Users/jaredmscott/repos/recce/cargento && python3 -m unittest cargento.skills.cargento.tests.test_snapshot -v` - -Expected: FAIL with `ImportError: cannot import name 'snapshot' from 'cargento_runtime'`. - -- [ ] **Step 3: Write the implementation** - -Create `cargento_runtime/snapshot.py`: - -```python -"""The published dashboard snapshot: one built response per variant, versioned. - -The revision is a pair, not an integer. A counter alone restarts at zero with -the process, so a tab frozen at revision 512 across a dashboard restart would -treat every later revision as older and never refetch again. Pairing it with -the server start stamp makes a restart visibly discontinuous, and the client -discards its cursor when the first element changes. - -The counter is per process rather than per variant, so it orders every -published state a client could hold a cursor against. -""" - -from __future__ import annotations - -import threading -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from _thread import LockType - -# (window_hours, show_all): the response variants, keyed as the memo keyed them. -SnapshotKey = tuple[float, bool] -# (server_started, counter) -Revision = tuple[float, int] - - -def format_revision(revision: Revision) -> str: - """The wire form: restart stamp, a dot, the counter.""" - started, counter = revision - return f"{started:.0f}.{counter}" - - -class Snapshot: - """One process's published responses, guarded by its own lock. - - The lock is held only across a dict read or write, never across collection - and never across a socket write. A published entry is an immutable tuple, so - a reader that has taken one cannot be torn by a concurrent publish. - """ - - def __init__(self, *, server_started: float) -> None: - self.server_started = server_started - self._lock: LockType = threading.Lock() - self._counter = 0 - self._entries: dict[SnapshotKey, tuple[Revision, bytes, float]] = {} - - def publish(self, key: SnapshotKey, body: bytes, *, now: float = 0.0) -> Revision: - with self._lock: - self._counter += 1 - revision = (self.server_started, self._counter) - self._entries[key] = (revision, body, now) - return revision - - def current(self, key: SnapshotKey) -> tuple[Revision, bytes] | None: - with self._lock: - entry = self._entries.get(key) - if entry is None: - return None - revision, body, _published_at = entry - return revision, body - - def age(self, key: SnapshotKey, *, now: float) -> float | None: - """Seconds since this variant was published, or None if it never was. - - None rather than zero: zero reads as fresh, which would let a cold GET - skip the collection it needs. - """ - with self._lock: - entry = self._entries.get(key) - if entry is None: - return None - return now - entry[2] -``` - -- [ ] **Step 4: Run test to verify it passes** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_snapshot -v` - -Expected: PASS, 9 tests. - -- [ ] **Step 5: Add the reviewed allowlist entry** - -`RuntimeImportGraphTest.EXPECTED` in `cargento/skills/cargento/tests/test_contracts.py` is keyed by module name with a set of runtime imports. Add, in alphabetical position: - -```python - # The published snapshot is a passive container: it holds bytes and a - # revision and takes a lock. It imports no runtime module, which is what - # lets both aggregate and the HTTP layer depend on it without a cycle. - "cargento_runtime.snapshot": set(), -``` - -- [ ] **Step 6: Run the contract test** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_contracts -v` - -Expected: PASS. If `test_runtime_import_graph_matches_the_reviewed_allowlist` fails, read the diff it prints: it means `snapshot.py` imported something it should not. - -- [ ] **Step 7: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/snapshot.py \ - cargento/skills/cargento/tests/test_snapshot.py \ - cargento/skills/cargento/tests/test_contracts.py -git commit -s -m "feat(snapshot): add the versioned published-response container (DRC-4084)" -``` - ---- - -### Task 2: Serve `/api/data` from the snapshot - -Replace the memo with the snapshot, preserving the anti-stampede property and the freshness guarantee. The memo held its lock across collection so concurrent tabs shared one filesystem scan; that behaviour must survive, and it is why the collection lock stays separate from the snapshot lock. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/aggregate.py` (`Application.__init__`, `collect_json`) -- Modify: `cargento/skills/cargento/tests/test_contracts.py` (`aggregate` gains `cargento_runtime.snapshot`) -- Test: `cargento/skills/cargento/tests/test_http_api.py`, `cargento/skills/cargento/tests/test_snapshot.py` - -**Interfaces:** -- Consumes: `Snapshot`, `Revision`, `SnapshotKey` from Task 1. -- Produces: `Application.snapshot` attribute, and `Application.collect_json(*, show_all) -> tuple[Revision, bytes]`. **The return type changes.** Task 3 updates the one production caller. `support.collect_json` in the test helpers returns bytes and must keep doing so, so update it to unpack. - -- [ ] **Step 1: Write the failing test** - -Add to `cargento/skills/cargento/tests/test_snapshot.py`: - -```python -class ApplicationSnapshotTest(support.RuntimeTestCase): - """collect_json publishes, reuses inside the floor, and recollects after it.""" - - def test_a_second_call_inside_the_floor_reuses_the_published_bytes(self) -> None: - app = support.build_app() - calls: list[int] = [] - real = app.collect - - def counting(*, show_all: bool) -> dict[str, object]: - calls.append(1) - return real(show_all=show_all) - - app.collect = counting # type: ignore[method-assign] - first_rev, first_body = app.collect_json(show_all=False) - second_rev, second_body = app.collect_json(show_all=False) - self.assertEqual(len(calls), 1, "the second call must not recollect") - self.assertEqual(first_body, second_body) - self.assertEqual(first_rev, second_rev, "reuse must not mint a revision") - - def test_the_two_variants_are_published_separately(self) -> None: - app = support.build_app() - default_rev, _ = app.collect_json(show_all=False) - all_rev, _ = app.collect_json(show_all=True) - self.assertNotEqual(default_rev, all_rev) - self.assertIsNotNone(app.snapshot.current((app.config.window_hours, False))) - self.assertIsNotNone(app.snapshot.current((app.config.window_hours, True))) - - def test_a_stale_snapshot_recollects_and_mints_a_new_revision(self) -> None: - app = support.build_app() - first_rev, _ = app.collect_json(show_all=False) - # Advance past the floor rather than sleeping: the clock is injected. - base = app.clock() - app.clock = lambda: base + app.config.collect_memo_sec + 1 # type: ignore[method-assign] - second_rev, _ = app.collect_json(show_all=False) - self.assertGreater(second_rev[1], first_rev[1]) -``` - -Import `support` at the top of the module and add `from . import support`. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_snapshot -v` - -Expected: FAIL. `collect_json` returns `bytes`, so unpacking into two names raises `ValueError` or the attribute `app.snapshot` does not exist. - -- [ ] **Step 3: Write the implementation** - -In `aggregate.py`, add to `Application.__init__` after `self.state = state`: - -```python - self.snapshot = runtime_snapshot.Snapshot(server_started=state.server_started) -``` - -and import the module alongside the other runtime imports: - -```python -from cargento_runtime import snapshot as runtime_snapshot -``` - -Replace `collect_json` with: - -```python - def collect_json(self, *, show_all: bool) -> tuple[runtime_snapshot.Revision, bytes]: - """The published response for one variant, collecting only if stale. - - Two locks, deliberately. `collect_memo_lock` is still held across - collection, so concurrent tabs share one filesystem and SQLite scan - instead of stampeding a cold entry. The snapshot's own lock is taken - only to read or write the published tuple, so a slow reader can never - block a collection and a collection can never block a reader. - """ - state = self.state - key: runtime_snapshot.SnapshotKey = (self.config.window_hours, show_all) - fresh = self.snapshot.age(key, now=self.clock()) - if fresh is not None and fresh < self.config.collect_memo_sec: - current = self.snapshot.current(key) - if current is not None: - return current - with state.collect_memo_lock: - # Re-check under the lock: another thread may have collected while - # this one waited, which is the whole point of holding it. - fresh = self.snapshot.age(key, now=self.clock()) - if fresh is not None and fresh < self.config.collect_memo_sec: - current = self.snapshot.current(key) - if current is not None: - return current - body = json.dumps(self.collect(show_all=show_all)).encode() - revision = self.snapshot.publish(key, body, now=self.clock()) - return revision, body -``` - -Leave `state.collect_memo` and `CollectMemoEntry` in place for now; Task 4 removes them once nothing reads them. - -- [ ] **Step 4: Update the allowlist and the test helper** - -In `test_contracts.py`, add `"cargento_runtime.snapshot"` to the `"cargento_runtime.aggregate"` set. - -In `cargento/skills/cargento/tests/support.py`, `collect_json` currently returns the bytes. Keep its signature and unpack: - -```python -def collect_json(window_hours: float = 24, show_all: bool = False) -> bytes: - _revision, body = build_app(window_hours).collect_json(show_all=show_all) - return body -``` - -Read the current body of that helper before editing and preserve whatever else it does. - -- [ ] **Step 5: Run the tests** - -Run: -```bash -python3 -m unittest cargento.skills.cargento.tests.test_snapshot cargento.skills.cargento.tests.test_contracts -v -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` -Expected: PASS. Any other failure is a caller of `collect_json` you have not updated; fix the caller, not the return type. - -- [ ] **Step 6: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/aggregate.py \ - cargento/skills/cargento/tests/test_snapshot.py \ - cargento/skills/cargento/tests/test_contracts.py \ - cargento/skills/cargento/tests/support.py -git commit -s -m "feat(aggregate): serve /api/data from the versioned snapshot (DRC-4084)" -``` - ---- - -### Task 3: Expose the served revision on the wire - -A client cannot compare cursors it cannot see. The revision goes in a response header, not the JSON body, so the documented body contract is untouched and `curl` output is unchanged. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/http_api.py` (`_send`, `do_GET`) -- Test: `cargento/skills/cargento/tests/test_http_api.py` - -**Interfaces:** -- Consumes: `collect_json` returning `(Revision, bytes)` from Task 2. -- Produces: response header `X-Cargento-Revision: .` on `/api/data`. No other route sets it. - -- [ ] **Step 1: Write the failing test** - -Add to `cargento/skills/cargento/tests/test_http_api.py`, matching how the existing tests in that module build a server: - -```python -class DataRevisionHeaderTest(...): - """/api/data names the revision it served, so a client can hold a cursor.""" - - def test_the_header_is_present_and_restart_qualified(self) -> None: - # Build a server the way the neighbouring tests do, GET /api/data, then: - # header = response.headers["X-Cargento-Revision"] - # self.assertRegex(header, r"^\d+\.\d+$") - # and assert the counter half increments across a stale re-request while - # the stamp half does not change within one process. - - def test_the_body_is_unchanged_by_the_header(self) -> None: - # json.loads(body) must still parse and must contain "sessions" and - # "generated", exactly as before this task. - - def test_health_and_root_do_not_carry_a_revision(self) -> None: - # Only /api/data publishes a cursor; a page load or a liveness probe - # carrying one would invite a client to treat it as comparable. -``` - -Fill each body in using the existing helpers in that module (`support.make_server`, `support.serve_until_closed`). Read two neighbouring tests first and copy their structure exactly rather than inventing a new one. - -- [ ] **Step 2: Run test to verify it fails** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_http_api -v` - -Expected: FAIL with `KeyError: 'X-Cargento-Revision'`. - -- [ ] **Step 3: Write the implementation** - -Give `_send` an optional header map rather than a revision-specific parameter, so a later task can add a second header without touching the signature again: - -```python - def _send( - self, - body: bytes, - ctype: str, - code: int = 200, - *, - headers: dict[str, str] | None = None, - ) -> None: - self.send_response(code) - self.send_header("Content-Type", ctype) - self.send_header("Content-Length", str(len(body))) - self.send_header("Cache-Control", "no-store") - for name, value in (headers or {}).items(): - self.send_header(name, value) - self.end_headers() - self.wfile.write(body) -``` - -In `do_GET`, at the `/api/data` branch: - -```python - revision, body = self.server.application.collect_json(show_all=show_all) - self._send( - body, - "application/json", - headers={"X-Cargento-Revision": runtime_snapshot.format_revision(revision)}, - ) -``` - -Import `snapshot as runtime_snapshot` in `http_api.py` and add `"cargento_runtime.snapshot"` to the `"cargento_runtime.http_api"` set in the allowlist. - -- [ ] **Step 4: Run the tests** - -Run: -```bash -python3 -m unittest cargento.skills.cargento.tests.test_http_api cargento.skills.cargento.tests.test_contracts -v -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/http_api.py \ - cargento/skills/cargento/tests/test_http_api.py \ - cargento/skills/cargento/tests/test_contracts.py -git commit -s -m "feat(http): name the served revision in an /api/data header (DRC-4084)" -``` - ---- - -### Task 4: Retire the dead memo - -`state.collect_memo`, `state.CollectMemoEntry` and `config.collect_memo_sec` are now partly dead: the dict and its TypedDict have no reader, while the interval is still the freshness floor. Delete the dead half and keep the live one, so a future reader does not restore a second cache alongside the snapshot. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/state.py` -- Modify: `cargento/skills/cargento/tests/support.py` (`clear_state`) -- Test: existing suite - -**Interfaces:** -- Consumes: nothing new. -- Produces: `RuntimeState` no longer has `collect_memo`; `CollectMemoEntry` is gone. `collect_memo_lock` stays, since Task 2 still holds it across collection. `config.collect_memo_sec` stays, as the freshness floor. - -- [ ] **Step 1: Prove they are dead** - -Run: -```bash -grep -rn "collect_memo\b\|CollectMemoEntry" --include='*.py' cargento/ scripts/ -``` -Expected: hits only in `state.py` (the definitions), `support.clear_state`, and any test asserting on the old memo. `collect_memo_lock` hits are separate and must remain. If a production reader still exists, stop: Task 2 is incomplete. - -- [ ] **Step 2: Delete the dead members** - -Remove the `CollectMemoEntry` TypedDict and the `collect_memo` field from `state.py`. Keep `collect_memo_lock` and add a comment naming its surviving job: - -```python - # Still held across collection so concurrent readers share one scan. The - # published bytes now live in Application.snapshot, which has its own lock. - collect_memo_lock: LockType = field(default_factory=threading.Lock) -``` - -- [ ] **Step 3: Update `clear_state`** - -`support.clear_state` promises to empty every cache. Remove its `collect_memo` line and put -`state.snapshot.clear()` under `collect_memo_lock` in its place. The snapshot is a `RuntimeState` -field, not an `Application` field: two applications over one state must share one published entry, or -concurrent cold reads stop single-flighting and each does its own collection. Task 2 Step 3 above puts -it on `Application` and is wrong. What shipped is a state field with a read-only -`Application.snapshot` alias, and the test that caught the per-instance version is the one the suite -already had, `test_collect_json_single_flights_concurrent_cold_requests` in `test_http_api.py`. Note -the snapshot in the docstring if it lists what it clears. - -- [ ] **Step 4: Run the suite** - -Run: -```bash -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -mypy -``` -Expected: PASS. `mypy --strict` is what catches a missed reference. - -- [ ] **Step 5: Commit** - -```bash -git add cargento/skills/cargento/cargento_runtime/state.py cargento/skills/cargento/tests/support.py -git commit -s -m "refactor(state): drop the response memo the snapshot replaced (DRC-4084)" -``` - ---- - -### Task 5: Prove it changed nothing, then document it - -**Files:** -- Test: `cargento/skills/cargento/tests/test_http_api.py` -- Modify: `docs/design-runtime-architecture.md` (module table, R-2 allowlist prose) -- Modify: `docs/plans/event-driven-session-observation.md` (Phase 1 progress) - -- [ ] **Step 1: Write the equivalence test** - -The claim is that for identical store contents the response is byte-identical to the pre-change one. Add a test that builds a fixture store, collects twice through two independently constructed applications, and asserts the bodies match apart from the `generated` timestamp: - -```python - def test_the_payload_is_unchanged_apart_from_its_generated_stamp(self) -> None: - first = json.loads(support.collect_json()) - second = json.loads(support.collect_json()) - first.pop("generated", None) - second.pop("generated", None) - self.assertEqual(first, second) -``` - -That is necessary but weak on its own, so also confirm against the base branch by hand in Step 2. - -The snippet above is what shipped, and it does not do what the sentence above it says: it builds no -fixture store, so both collections read the developer's real one. That drifted past review because a -CI runner's store is empty and the test therefore compared two empty payloads. It flaked locally -instead, was muted, and was repaired in DRC-4088 by seeding a fixture store, pinning the clock, and -asserting on the collected session ids so an override that misses fails loudly rather than passing -having compared nothing. Left here rather than rewritten: the gap between the sentence and the code -is the point. - -- [ ] **Step 2: Diff a real response against the Phase 0 branch** - -```bash -python3 -c " -import json,sys; sys.path.insert(0,'cargento/skills/cargento') -from tests import support -d=json.loads(support.collect_json()); d.pop('generated',None) -print(json.dumps(d,sort_keys=True))" > /tmp/after.json -git stash && git checkout feature/drc-4080-event-driven-session-observation-materialized-snapshot-sse -- . 2>/dev/null || true -``` - -Safer alternative, and the one to prefer: check the Phase 0 branch out into a detached worktree, run the same one-liner there, and `diff` the two files. Do not stash and check out over your working tree. - -```bash -git worktree add --detach /tmp/p0 feature/drc-4080-event-driven-session-observation-materialized-snapshot-sse -# run the same one-liner in /tmp/p0, write /tmp/before.json -diff /tmp/before.json /tmp/after.json && echo "byte-identical" -git worktree remove /tmp/p0 -``` - -Expected: no diff. If there is one, the refactor changed behaviour and the task is not done. - -- [ ] **Step 3: Update the module map** - -`docs/design-runtime-architecture.md` owns the module table and the R-2 rule. Add a `snapshot.py` row describing what it owns (the published response bytes and the restart-qualified revision) and note that it imports no runtime module, which is what keeps `aggregate` and `http_api` able to share it without a cycle. - -- [ ] **Step 4: Update the design plan's Phase 1 section** - -Mark the snapshot, revision pair and `/api/data` freshness rule as shipped, and note that the stream, producer, time tick and consent lease remain in 1b. Do not delete the Phase 1 section: it is still partly unshipped. - -Tone gate applies: no em dashes, en dashes or curly quotes. - -- [ ] **Step 5: Run the full gate** - -```bash -ruff check . && ruff format --check . && mypy -python3 scripts/lint_embedded.py --allow-missing-node -python3 scripts/validate_plugins.py -python3 scripts/bump_version.py --current -coverage erase -coverage run -m unittest discover -s cargento/skills/cargento/tests -t . -coverage run -a -m unittest scripts.tests.test_validate_plugins scripts.tests.test_bump_version \ - scripts.tests.test_lint_embedded scripts.tests.test_bench_collect -coverage report -``` -Expected: all pass, coverage at or above `fail_under`. - -- [ ] **Step 6: Commit, then open the stacked PR** - -```bash -git add -A -git commit -s -m "docs: record the snapshot module and Phase 1a progress (DRC-4084)" -git push -u origin HEAD -gh pr create --base feature/drc-4080-event-driven-session-observation-materialized-snapshot-sse -``` - -**The `--base` flag is the whole point of the stack.** Without it the PR targets `main` and its diff will include every Phase 0 commit. Verify after opening: `gh pr view --json baseRefName` must show the Phase 0 branch. - -Do not merge this before #83. When #83 merges, retarget this PR at `main` (`gh pr edit --base main`) and rebase, because a squash merge rewrites Phase 0 into a single new commit and the old parents stop existing. - ---- - -## Self-Review - -**Spec coverage against the design doc's Phase 1 section:** - -| Phase 1 requirement | Where | -|---|---| -| Versioned snapshot and publish protocol | Tasks 1, 2 | -| Revision pair surviving restart | Task 1 | -| `/api/data` serves the snapshot | Task 2 | -| Independent direct-GET freshness rule | Task 2, held at `collect_memo_sec` so worst-case staleness is literally unchanged | -| Lazy init on first demand | Task 2, by construction: nothing publishes until the first GET | -| Services start only after daemonization | 1b. No service is started here, which is why it does not arise | -| SSE stream and all its hardening | 1b | -| Demand-scoped producer | 1b, deliberately, per "Why this is its own PR" | -| Time-derived field tick | 1b | -| Quota consent lease | 1b | - -**Placeholder scan:** Task 3 Step 1 gives three test names with described assertions rather than full bodies, because the module's server-construction helper differs between test classes and copying the wrong one produces a passing test that starts no server. The file and the exact assertions are named. Task 5 Step 2 offers two approaches and marks the worktree one as preferred; the discouraged variant is shown only because an implementer will otherwise reach for `git stash` on their own. - -**Type consistency:** `collect_json` changes from `bytes` to `tuple[Revision, bytes]` in Task 2, and Task 2 Step 4 updates both the production caller path and `support.collect_json`. Task 3 consumes that tuple. `SnapshotKey` is `tuple[float, bool]` in Task 1 and is constructed as `(self.config.window_hours, show_all)` in Task 2, matching. `format_revision` is defined in Task 1 and called in Task 3. diff --git a/docs/plans/event-driven-phase-1b.md b/docs/plans/event-driven-phase-1b.md deleted file mode 100644 index 679457a..0000000 --- a/docs/plans/event-driven-phase-1b.md +++ /dev/null @@ -1,830 +0,0 @@ -# Event-driven session observation, Phase 1b Implementation Plan - -> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. - -**Goal:** Serve `GET /api/stream` as a bounded, heartbeat-driven SSE revision stream, and run a collection producer only while a stream is connected. - -**Architecture:** One new runtime module, `stream.py`, holding the connected clients and their one-slot mailboxes. It imports no runtime module, so `state` can own a registry the same way it owns the snapshot. `aggregate` notifies the registry when it publishes a revision. `http_api` serves the endpoint. `lifecycle` runs the producer thread, started after daemonization and stopped with the server. - -**Tech Stack:** Python 3.11 standard library only. `unittest`, `coverage`, `ruff`, `mypy --strict`. - -## Why this is server-only - -Phase 1b in the design doc is the stream plus the client plus the quota lease. This plan is the server half: - -- **1b, this plan.** `/api/stream`, connection budgets, heartbeats, write timeouts, one-slot mailboxes, shutdown, and the demand-scoped producer. **The page is not changed and keeps polling every five seconds**, so the endpoint is inert until 1c connects to it. That is deliberate: the thread, lock and shutdown discipline is the risky part of Phase 1 and it is worth reviewing on its own, against tests, rather than alongside a JavaScript rewrite. -- **1c, next.** `EventSource` in the page, leader-tab election, reconnect, removing the poll, the quota consent lease, and the time-derived-field tick. That is where user-visible behaviour changes. - -An inert endpoint for one PR cycle is the cost. The benefit is that when 1c lands, every failure it produces is a client failure, because the server contract is already proven. - -**Ticket:** DRC-4084. **Design owner:** [`event-driven-session-observation.md`](event-driven-session-observation.md) § Phase 1. **Stacked on:** the Phase 1a branch, PR #84, which is itself stacked on #83. - -## Global Constraints - -- Standard library only. Python floor 3.11. -- `ruff check .` (`select = ALL`), `ruff format --check .`, `mypy --strict` must pass. -- `coverage report` must meet `fail_under`. It only ratchets up. -- Tests run on Ubuntu, macOS and Windows. A test that blocks on a socket needs a timeout on both ends, or Windows CI hangs rather than fails. -- Never edit a `version` field. -- `git commit -s` for DCO. -- `docs/plans/*.md` is inside the tone gate: no em dashes, en dashes, or curly quotes. -- **R-2.** `stream.py` imports no runtime module. Each new module or edge needs a reviewed entry in `RuntimeImportGraphTest.EXPECTED`. -- **New runtime modules must be added to `CARGENTO_RUNTIME_FILES` in `scripts/validate_plugins.py`.** Phase 1a added `snapshot.py` in one commit and the inventory line twelve minutes later in the next, both inside PR #84, so nothing reached main missing it. What would have caught it is `scripts/tests/test_validate_plugins.py`, which runs `validate_runtime_files` over an installed copy rather than over the checkout. That the gap closed inside one PR is not a reason to rely on noticing again. -- **Never hold a lock across a socket write.** The publisher takes the registry lock only to drop a revision into each client's mailbox. The handler takes nothing while writing. -- **The producer must not run when nobody is connected.** An idle daemon does zero filesystem work today and this phase must not regress that. - ---- - -### Task 1: The stream registry - -**Files:** -- Create: `cargento/skills/cargento/cargento_runtime/stream.py` -- Modify: `cargento/skills/cargento/cargento_runtime/state.py` (own a registry) -- Modify: `cargento/skills/cargento/tests/test_contracts.py` (allowlist) -- Modify: `scripts/validate_plugins.py` (`CARGENTO_RUNTIME_FILES`) -- Test: `cargento/skills/cargento/tests/test_stream.py` - -**Interfaces:** -- Produces: - - `class StreamClient` with `wait(timeout) -> Revision | None`, `offer(revision)`, `close()`, and `closed` as a property. - - `class StreamRegistry` with `register(*, limit) -> StreamClient | None` (None when the budget is full), `release(client)`, `publish(revision)`, `close_all()`, and `count` as a property. - - `Revision` is re-used from `snapshot`, but `stream.py` must not import it. Type the mailbox as `tuple[float, int]` directly and say why in a comment. - -A one-slot mailbox, not a queue. A slow client must fall behind by losing intermediate revisions, never by growing an unbounded backlog. The newest revision is the only one worth delivering, because the client refetches the whole payload anyway. - -- [ ] **Step 1: Write the failing test** - -Create `cargento/skills/cargento/tests/test_stream.py`: - -```python -"""The SSE client registry: one-slot mailboxes, budgets, and shutdown.""" - -from __future__ import annotations - -import threading -import unittest - -from cargento_runtime import stream as runtime_stream - - -class StreamClientTest(unittest.TestCase): - def test_wait_returns_the_offered_revision(self) -> None: - client = runtime_stream.StreamClient() - client.offer((1000.0, 5)) - self.assertEqual((1000.0, 5), client.wait(timeout=0.01)) - - def test_wait_times_out_to_none_so_the_caller_can_heartbeat(self) -> None: - self.assertIsNone(runtime_stream.StreamClient().wait(timeout=0.01)) - - def test_the_mailbox_holds_one_slot_and_keeps_the_newest(self) -> None: - # A slow reader falls behind by skipping revisions, never by growing a - # backlog. The client refetches the whole payload, so only the newest - # revision is worth delivering. - client = runtime_stream.StreamClient() - client.offer((1000.0, 1)) - client.offer((1000.0, 2)) - client.offer((1000.0, 3)) - self.assertEqual((1000.0, 3), client.wait(timeout=0.01)) - self.assertIsNone(client.wait(timeout=0.01)) - - def test_close_wakes_a_waiter_and_marks_the_client_closed(self) -> None: - client = runtime_stream.StreamClient() - woke = threading.Event() - - def waiter() -> None: - client.wait(timeout=5.0) - woke.set() - - thread = threading.Thread(target=waiter, daemon=True) - thread.start() - client.close() - self.assertTrue(woke.wait(timeout=2.0), "close must wake a blocked waiter") - self.assertTrue(client.closed) - thread.join(timeout=2.0) - - -class StreamRegistryTest(unittest.TestCase): - def test_register_returns_a_client_and_counts_it(self) -> None: - registry = runtime_stream.StreamRegistry() - client = registry.register(limit=2) - self.assertIsNotNone(client) - self.assertEqual(1, registry.count) - - def test_register_refuses_past_the_budget(self) -> None: - registry = runtime_stream.StreamRegistry() - self.assertIsNotNone(registry.register(limit=1)) - self.assertIsNone(registry.register(limit=1), "the budget must be a hard cap") - self.assertEqual(1, registry.count) - - def test_release_frees_a_slot(self) -> None: - registry = runtime_stream.StreamRegistry() - first = registry.register(limit=1) - assert first is not None - registry.release(first) - self.assertEqual(0, registry.count) - self.assertIsNotNone(registry.register(limit=1)) - - def test_publish_reaches_every_registered_client(self) -> None: - registry = runtime_stream.StreamRegistry() - a = registry.register(limit=4) - b = registry.register(limit=4) - assert a is not None and b is not None - registry.publish((1000.0, 9)) - self.assertEqual((1000.0, 9), a.wait(timeout=0.01)) - self.assertEqual((1000.0, 9), b.wait(timeout=0.01)) - - def test_publish_with_no_clients_is_a_no_op(self) -> None: - runtime_stream.StreamRegistry().publish((1000.0, 1)) - - def test_close_all_closes_every_client_and_empties_the_registry(self) -> None: - registry = runtime_stream.StreamRegistry() - client = registry.register(limit=4) - assert client is not None - registry.close_all() - self.assertTrue(client.closed) - self.assertEqual(0, registry.count) - - def test_a_released_client_stops_receiving(self) -> None: - registry = runtime_stream.StreamRegistry() - client = registry.register(limit=4) - assert client is not None - registry.release(client) - registry.publish((1000.0, 2)) - self.assertIsNone(client.wait(timeout=0.01)) - - -if __name__ == "__main__": - unittest.main() -``` - -- [ ] **Step 2: Run test to verify it fails** - -Run: `cd /Users/jaredmscott/repos/recce/cargento && python3 -m unittest cargento.skills.cargento.tests.test_stream -v` - -Expected: FAIL with `ImportError: cannot import name 'stream' from 'cargento_runtime'`. - -- [ ] **Step 3: Write the implementation** - -Create `cargento_runtime/stream.py`: - -```python -"""Connected SSE clients and their one-slot revision mailboxes. - -A mailbox holds one revision, not a queue. A client that reads slowly must fall -behind by skipping intermediate revisions rather than by growing an unbounded -backlog, and skipping costs it nothing: it refetches the whole payload on the -revision it does see, so only the newest one is worth delivering. - -This module imports nothing from the runtime, which is what lets `state` own a -registry and `http_api` serve from it without a cycle. The revision type is -written out rather than imported from `snapshot` for the same reason; the two -must stay the same shape, which the tests assert. -""" - -from __future__ import annotations - -import threading -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from _thread import LockType - -# Structurally identical to snapshot.Revision. Not imported, to keep this -# module free of runtime dependencies. -Revision = tuple[float, int] - - -class StreamClient: - """One connected stream: a one-slot mailbox and a wake-up.""" - - def __init__(self) -> None: - self._condition = threading.Condition() - self._pending: Revision | None = None - self._closed = False - - @property - def closed(self) -> bool: - with self._condition: - return self._closed - - def offer(self, revision: Revision) -> None: - """Replace whatever is waiting. Newest wins; nothing queues.""" - with self._condition: - self._pending = revision - self._condition.notify_all() - - def wait(self, *, timeout: float) -> Revision | None: - """The pending revision, or None on timeout or close. - - None is the heartbeat signal as well as the shutdown signal, so the - caller checks `closed` to tell them apart. - """ - with self._condition: - if self._pending is None and not self._closed: - self._condition.wait(timeout) - pending, self._pending = self._pending, None - return pending - - def close(self) -> None: - with self._condition: - self._closed = True - self._condition.notify_all() - - -class StreamRegistry: - """Every connected stream on one runtime, behind one short-held lock. - - The lock is taken to add, drop, or hand a revision to each mailbox. It is - never held across a socket write: the handler writes outside it entirely. - """ - - def __init__(self) -> None: - self._lock: LockType = threading.Lock() - self._clients: set[StreamClient] = set() - - @property - def count(self) -> int: - with self._lock: - return len(self._clients) - - def register(self, *, limit: int) -> StreamClient | None: - """A new client, or None when the budget is full. - - A hard cap rather than a queue: every stream costs a thread and a - socket for as long as it lives, so the honest answer past the cap is a - refusal the caller can turn into a 503. - """ - client = StreamClient() - with self._lock: - if len(self._clients) >= limit: - return None - self._clients.add(client) - return client - - def release(self, client: StreamClient) -> None: - with self._lock: - self._clients.discard(client) - client.close() - - def publish(self, revision: Revision) -> None: - with self._lock: - clients = list(self._clients) - # Outside the lock: offer() takes each client's own condition, and a - # publisher must never be able to block another publisher. - for client in clients: - client.offer(revision) - - def close_all(self) -> None: - """Wake and drop every client. Shutdown calls this.""" - with self._lock: - clients = list(self._clients) - self._clients.clear() - for client in clients: - client.close() -``` - -- [ ] **Step 4: Run test to verify it passes** - -Run: `python3 -m unittest cargento.skills.cargento.tests.test_stream -v` - -Expected: PASS, 11 tests. - -- [ ] **Step 5: Give the runtime a registry** - -In `state.py`, beside the snapshot field: - -```python - # Connected SSE clients, owned here for the same reason the snapshot is: - # they belong to the runtime, not to whichever object serves a request. - streams: runtime_stream.StreamRegistry = field(init=False) -``` - -and in `__post_init__`: - -```python - self.streams = runtime_stream.StreamRegistry() -``` - -Import `from cargento_runtime import stream as runtime_stream` alongside the snapshot import. - -- [ ] **Step 6: Register the module everywhere it must be registered** - -Two inventories, both of which Phase 1a proved are easy to miss: - -1. `RuntimeImportGraphTest.EXPECTED` in `test_contracts.py`: add `"cargento_runtime.stream": set()`, and add `"cargento_runtime.stream"` to the `"cargento_runtime.state"` set. -2. `CARGENTO_RUNTIME_FILES` in `scripts/validate_plugins.py`: add `"skills/cargento/cargento_runtime/stream.py"` in alphabetical position. - -- [ ] **Step 7: Assert the two revision shapes cannot drift** - -Add to `test_stream.py`: - -```python -class RevisionShapeTest(unittest.TestCase): - def test_the_stream_revision_matches_the_snapshot_revision(self) -> None: - """stream.py deliberately does not import snapshot, so pin the shape.""" - from cargento_runtime import snapshot as runtime_snapshot - - self.assertEqual(runtime_snapshot.Revision, runtime_stream.Revision) -``` - -- [ ] **Step 8: Run the suite and the gate** - -```bash -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -python3 -m unittest scripts.tests.test_validate_plugins -ruff check . && ruff format --check . && mypy && python3 scripts/validate_plugins.py -``` -Expected: all pass. - -- [ ] **Step 9: Commit** - -```bash -git add -A -git commit -s -m "feat(stream): add the SSE client registry and one-slot mailboxes (DRC-4084)" -``` - ---- - -### Task 2: Publish revisions into the registry - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/aggregate.py` (`collect_json`) -- Test: `cargento/skills/cargento/tests/test_stream.py` - -**Interfaces:** -- Consumes: `state.streams` from Task 1. -- Produces: every newly minted revision reaches every connected client. A reused snapshot does not, because nothing changed. - -- [ ] **Step 1: Write the failing test** - -```python -class PublishNotifiesStreamsTest(support.RuntimeTestCase): - def test_a_fresh_collection_reaches_a_connected_client(self) -> None: - app = support.build_app() - client = app.state.streams.register(limit=4) - assert client is not None - revision, _body = app.collect_json(show_all=False) - self.assertEqual(revision, client.wait(timeout=0.01)) - - def test_a_reused_snapshot_does_not_wake_a_client(self) -> None: - # Nothing changed, so there is nothing to tell a client about. Waking - # it would make every warm GET cost every stream a refetch. - app = support.build_app() - app.collect_json(show_all=False) - client = app.state.streams.register(limit=4) - assert client is not None - app.collect_json(show_all=False) - self.assertIsNone(client.wait(timeout=0.01)) -``` - -Add `from . import support` to the module imports. - -- [ ] **Step 2: Run test to verify it fails** - -Expected: the first test FAILS with `None != (…)`, because nothing publishes yet. - -- [ ] **Step 3: Write the implementation** - -In `collect_json`, immediately after `revision = self.snapshot.publish(...)` and still inside the lock: - -```python - revision = self.snapshot.publish(key, body, now=self.clock()) - # Only a freshly minted revision is worth announcing. A warm reuse - # returns above without reaching this line, so a connected client is - # never woken for a state it already has. - self.state.streams.publish(revision) - return revision, body -``` - -- [ ] **Step 4: Run the tests** - -```bash -python3 -m unittest cargento.skills.cargento.tests.test_stream -v -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` -Expected: PASS. - -- [ ] **Step 5: Commit** - -```bash -git add -A -git commit -s -m "feat(aggregate): announce each new revision to connected streams (DRC-4084)" -``` - ---- - -### Task 3: Serve `GET /api/stream` - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/config.py` (four tunables) -- Modify: `cargento/skills/cargento/cargento_runtime/http_api.py` (`do_GET`, a `_stream` handler) -- Modify: `cargento/skills/cargento/tests/test_contracts.py` (allowlist: `http_api` gains `stream`) -- Test: `cargento/skills/cargento/tests/test_http_api.py` - -**Interfaces:** -- Consumes: `state.streams`, `state.snapshot`. -- Produces: `GET /api/stream`, `text/event-stream`. Emits the current revision immediately if one has been published; on a cold server nothing has been, so the first event is the producer's first tick. Then one `event: revision` per publish, and a `: keepalive` comment every heartbeat interval. Returns 503 past the budget. - -**Config fields, following the existing `_sec` / `_bytes` naming:** - -```python - stream_max_clients: int - stream_heartbeat_sec: float - stream_write_timeout_sec: float -``` - -with defaults `stream_max_clients=8`, `stream_heartbeat_sec=15.0`, `stream_write_timeout_sec=10.0`. Eight bounds threads, not tabs. It sits above the six persistent connections HTTP/1.x browsers conventionally allow per origin, so a single browser is unlikely to be what refuses first. That is a desk read, not measured here, and the exact limit is user-agent-dependent. The cap is process-wide rather than per origin, so two browsers or two users can still reach it, and once 1c elects a leader tab one browser holds one stream and the cap is slack. - -- [ ] **Step 1: Write the failing test** - -Add to `test_http_api.py`, using the raw `http.client` pattern the neighbouring tests use. Do not invent a helper; `CargentoServerTest` has no shared response helper beyond the one Phase 1a added. - -```python -class StreamEndpointTest(RuntimeTestCase): - """The SSE contract: immediate state, then one event per revision.""" - - @staticmethod - def _open_stream(port: int) -> tuple[http.client.HTTPConnection, Any]: - conn = http.client.HTTPConnection("127.0.0.1", port, timeout=5) - conn.request("GET", "/api/stream") - return conn, conn.getresponse() - - def test_the_stream_opens_with_the_correct_content_type(self) -> None: - # 200, text/event-stream, no-store. - - def test_the_current_revision_arrives_immediately(self) -> None: - # A client must not wait for the next change to learn where it is. - # Read the first event and assert it is `event: revision`. - - def test_a_cold_stream_gets_headers_then_the_first_produced_revision(self) -> None: - # Nothing has been published yet, so there is no current revision to - # send: the client gets headers and then waits for the first published - # revision rather than receiving one at once. This is the case the - # immediate-delivery test hides by collecting first. - - def test_a_new_revision_is_delivered(self) -> None: - # Open the stream, force a collection (state_of().snapshot.clear() then - # collect_json), and read the next event. - - def test_a_cross_site_fetch_is_refused(self) -> None: - # Sec-Fetch-Site: cross-site with a document-navigation shape must be - # 403 on this route, unlike /api/data. A long-lived data stream is not - # a document navigation. - - def test_the_budget_refuses_past_the_cap(self) -> None: - # With stream_max_clients patched to 1, the second concurrent stream - # gets 503 rather than a thread. -``` - -Fill each body with real socket reads. Every read needs a timeout, or Windows CI hangs instead of failing. - -- [ ] **Step 2: Run test to verify it fails** - -Expected: 404 on `/api/stream`. - -- [ ] **Step 3: Add the config fields** - -Add the three fields to `RuntimeConfig` and their defaults to `build_runtime_config`, in the same relative position as the other server tunables. - -- [ ] **Step 4: Write the handler** - -In `http_api.py`: - -```python - def _stream(self) -> None: - """The SSE revision stream. - - Strictly same-origin: `do_GET` relaxes its check for document - navigations so a link to the dashboard works, and a long-lived data - stream is not a document navigation. Re-checking here with the strict - form is what keeps that relaxation off this route. - """ - if not self._local_ok(): - self.send_error(403) - return - application = self.server.application - config = application.config - state = application.state - client = state.streams.register(limit=config.stream_max_clients) - if client is None: - # A refusal, not a queue: every stream costs a thread and a socket. - self.send_error(503) - return - try: - self._stream_forever(client) - finally: - state.streams.release(client) - - def _stream_forever(self, client: Any) -> None: - application = self.server.application - config = application.config - self.send_response(200) - self.send_header("Content-Type", "text/event-stream") - self.send_header("Cache-Control", "no-store") - self.send_header("X-Accel-Buffering", "no") - self.end_headers() - # A peer that stops reading must not pin this thread forever. The - # unbounded default is the real shutdown risk here, not server_close. - with contextlib.suppress(OSError): - self.connection.settimeout(config.stream_write_timeout_sec) - current = application.state.snapshot.current( - (config.window_hours, False) - ) - if current is not None: - # Immediately, so a client learns where it is without waiting for - # the next change. - self._emit(current[0]) - while not client.closed: - revision = client.wait(timeout=config.stream_heartbeat_sec) - if client.closed: - return - if revision is None: - if not self._write_raw(b": keepalive\n\n"): - return - continue - if not self._emit(revision): - return - - def _emit(self, revision: Any) -> bool: - rendered = runtime_snapshot.format_revision(revision) - payload = f"id: {rendered}\nevent: revision\ndata: {rendered}\n\n" - return self._write_raw(payload.encode()) - - def _write_raw(self, payload: bytes) -> bool: - """Write and flush, reporting whether the peer is still there. - - No lock is held here. A blocked write must never be able to stall a - publisher or a collection. - """ - try: - self.wfile.write(payload) - self.wfile.flush() - except (OSError, ValueError): - return False - return True -``` - -Route it in `do_GET`, before the 404: - -```python - elif url.path == "/api/stream": - self._stream() -``` - -- [ ] **Step 5: Run the tests** - -```bash -python3 -m unittest cargento.skills.cargento.tests.test_http_api -v -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` - -- [ ] **Step 6: Commit** - -```bash -git add -A -git commit -s -m "feat(http): serve a bounded SSE revision stream at /api/stream (DRC-4084)" -``` - ---- - -### Task 4: Close streams on shutdown - -A stream handler blocks in `client.wait` for up to a heartbeat interval and then writes. `server.shutdown()` stops the accept loop but does not touch handler threads, and `daemon_threads` means nothing joins them. Shutdown must wake them so the process does not sit holding sockets it has promised to release. - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/http_api.py` (`_shutdown`) -- Modify: `cargento/skills/cargento/cargento_runtime/lifecycle.py` (`serve` cleanup) -- Test: `cargento/skills/cargento/tests/test_http_api.py`, `cargento/skills/cargento/tests/test_lifecycle.py` - -- [ ] **Step 1: Write the failing test** - -```python - def test_shutdown_closes_open_streams_promptly(self) -> None: - # Open a stream, POST /api/shutdown, and assert the stream ends within - # well under one heartbeat interval. Without close_all it would hang - # until the heartbeat, and on a long heartbeat that reads as a hang. -``` - -Assert the read returns empty (peer closed) inside about two seconds, with the heartbeat patched high enough that a passing test cannot be the heartbeat firing. - -- [ ] **Step 2: Run test to verify it fails** - -Expected: the read blocks until the heartbeat rather than ending at shutdown. - -- [ ] **Step 3: Write the implementation** - -In `_shutdown`, before starting the shutdown thread: - -```python - # Wake every stream first. shutdown() stops the accept loop but never - # touches handler threads, and a stream is asleep in wait() rather than - # in the socket, so nothing else would tell it to stop. - self.server.application.state.streams.close_all() -``` - -In `lifecycle.serve`'s `finally`, before `server_close()`: - -```python - with contextlib.suppress(Exception): - server.application.state.streams.close_all() -``` - -so a `--stop`, a signal, or an exception all converge on the same cleanup. - -- [ ] **Step 4: Run the tests** - -```bash -python3 -m unittest cargento.skills.cargento.tests.test_http_api cargento.skills.cargento.tests.test_lifecycle -v -``` - -- [ ] **Step 5: Commit** - -```bash -git add -A -git commit -s -m "fix(http): wake and close open streams on shutdown (DRC-4084)" -``` - ---- - -### Task 5: The demand-scoped producer - -**Files:** -- Modify: `cargento/skills/cargento/cargento_runtime/config.py` (`stream_producer_interval_sec`) -- Modify: `cargento/skills/cargento/cargento_runtime/lifecycle.py` (producer thread, started in `serve`) -- Test: `cargento/skills/cargento/tests/test_lifecycle.py` - -**Interfaces:** -- Produces: `lifecycle.run_producer(server, *, stop: threading.Event) -> None`, and `serve` starting it after `write_state` and stopping it in the `finally`. - -The producer collects on an interval **only while at least one stream is connected**. With zero clients it sleeps and touches nothing. Started inside `serve`, which on the daemon path runs after the fork, so the thread is never created in a process that is about to be replaced. - -- [ ] **Step 1: Write the failing test** - -```python -class ProducerTest(unittest.TestCase): - def test_the_producer_does_nothing_with_no_connected_stream(self) -> None: - # Run a few intervals with an empty registry and assert collect_json was - # never called. This is the "idle daemon does zero filesystem work" - # guarantee, and it is the one this whole phase most easily breaks. - - def test_the_producer_collects_while_a_stream_is_connected(self) -> None: - # Register a client, run, assert collect_json was called at least once. - - def test_the_stop_event_ends_the_producer_promptly(self) -> None: - # Set the event and assert the thread exits well inside one interval. - - def test_a_collection_error_does_not_kill_the_producer(self) -> None: - # Make collect_json raise once, then succeed. The loop must survive: - # a dead producer is a silently frozen dashboard. -``` - -Use a short interval and an injected stop event; never sleep for a real five seconds in a test. - -- [ ] **Step 2: Run test to verify it fails** - -Expected: `AttributeError: module 'cargento_runtime.lifecycle' has no attribute 'run_producer'`. - -- [ ] **Step 3: Write the implementation** - -```python -def run_producer( - server: http_api.CargentoHTTPServer, - *, - stop: threading.Event, - interval: float | None = None, -) -> None: - """Keep the snapshot warm while at least one stream is connected. - - With no client this loop does nothing at all: no collection, no store - access. An idle daemon costs what it costs today, which is nothing, and a - timer that collected regardless would be the regression this phase exists - to avoid. - - A collection failure is swallowed and retried on the next tick. The - per-harness failure boundary already reports the cause, and a producer that - died on one bad read would leave every connected dashboard frozen with no - indication why. - """ - application = server.application - period = application.config.stream_producer_interval_sec if interval is None else interval - while not stop.wait(period): - if application.state.streams.count == 0: - continue - try: - application.collect_json(show_all=False) - except Exception as exc: # noqa: BLE001 (a bad read must not stop the loop) - runtime_io.diag(f"Cargento: producer collection failed: {exc}", print) -``` - -Import `threading` in `lifecycle.py` if it is not already imported. - -In `serve`, after `write_state` and before `serve_forever`: - -```python - producer_stop = threading.Event() - producer = threading.Thread( - target=run_producer, args=(server,), kwargs={"stop": producer_stop}, daemon=True - ) - producer.start() -``` - -and in the `finally`, before the stream cleanup: - -```python - producer_stop.set() - producer.join(timeout=2) -``` - -- [ ] **Step 4: Run the tests** - -```bash -python3 -m unittest cargento.skills.cargento.tests.test_lifecycle -v -python3 -m unittest discover -s cargento/skills/cargento/tests -t . -``` - -- [ ] **Step 5: Prove the idle guarantee end to end** - -Start a daemon, leave it alone with no tab and no stream, and confirm it performs no collection: patch nothing, just watch that the snapshot's revision does not advance over several intervals. - -```bash -cd cargento/skills/cargento -python3 server.py --port 4599 --daemon -sleep 12 -curl -s -D - -o /dev/null http://127.0.0.1:4599/api/data | grep -i x-cargento-revision -python3 server.py --stop --port 4599 -``` - -The first GET must report revision counter 1: nothing collected before it asked. - -- [ ] **Step 6: Commit** - -```bash -git add -A -git commit -s -m "feat(lifecycle): collect on an interval only while a stream is connected (DRC-4084)" -``` - ---- - -### Task 6: Document it, then open the stacked PR - -**Files:** -- Modify: `docs/design-runtime-architecture.md` (module table) -- Modify: `docs/plans/event-driven-session-observation.md` (Phase 1 progress) -- Modify: `COMPATIBILITY.md` only if a per-OS caveat emerged in testing - -- [ ] **Step 1: Update the module map** - -Add a `stream.py` row: connected SSE clients and their one-slot mailboxes, importing no runtime module. Note that `state` owns the registry for the same reason it owns the snapshot. - -- [ ] **Step 2: Update the design plan** - -Mark the stream, budgets, heartbeats, shutdown and the demand-scoped producer as shipped in 1b, and record that the client, quota lease and time tick remain in 1c. Tone gate applies. - -- [ ] **Step 3: Full gate** - -```bash -ruff check . && ruff format --check . && mypy -python3 scripts/lint_embedded.py --allow-missing-node -python3 scripts/validate_plugins.py -python3 scripts/bump_version.py --current -coverage erase -coverage run -m unittest discover -s cargento/skills/cargento/tests -t . -coverage run -a -m unittest scripts.tests.test_validate_plugins scripts.tests.test_bump_version \ - scripts.tests.test_lint_embedded scripts.tests.test_bench_collect -coverage report -``` - -- [ ] **Step 4: Commit and open the PR** - -```bash -git push -u origin HEAD -gh pr create --base feature/drc-4084-phase-1-materialized-snapshot-and-sse-delivery -``` - -`--base` must be the Phase 1a branch. Verify with `gh pr view --json baseRefName`. The stack is now three deep: #83, #84, this. Each merge requires retargeting and rebasing the one above it. - ---- - -## Self-Review - -**Spec coverage against Phase 1's remaining bullets:** - -| Requirement | Where | -|---|---| -| SSE revision stream, restart-qualified IDs | Task 3, ids are `format_revision` output | -| Immediate current-state delivery | Task 3 | -| Server-wide connection budget | Tasks 1, 3 | -| Read/write timeouts | Task 3, `settimeout` on the connection | -| One-slot queues | Task 1 | -| Heartbeats | Task 3 | -| Services start only after daemonization | Task 5, the producer starts inside `serve` | -| Shutdown order | Task 4 | -| Demand-scoped producer, stops with zero readers | Task 5 | -| Leader-tab election, browser reconnect | 1c | -| Time-derived field tick | 1c | -| Quota consent lease | 1c | -| `?all=1` stays on its polling fallback | By construction: the producer collects the default window only, and no client change lands here | - -**Placeholder scan:** Tasks 3, 4 and 5 give test names with described assertions rather than full bodies, because each needs real socket or thread choreography whose exact shape depends on the neighbouring helpers. Every one names its file, its assertion and its failure mode. The implementation code is complete in all three. - -**Type consistency:** `stream.Revision` and `snapshot.Revision` are both `tuple[float, int]`, pinned equal by a test in Task 1 Step 7 because the modules deliberately do not import each other. `StreamRegistry.register` returns `StreamClient | None` in Task 1 and every caller in Task 3 checks for None. `run_producer` takes a keyword-only `stop` in Task 5 and `serve` passes it that way. diff --git a/docs/plans/event-driven-session-observation.md b/docs/plans/event-driven-session-observation.md index b9e6635..c040a8a 100644 --- a/docs/plans/event-driven-session-observation.md +++ b/docs/plans/event-driven-session-observation.md @@ -2012,7 +2012,7 @@ capture for every harness with an adapter, Claude last. ### Phase 1: materialized snapshot and SSE -Split into 1a and 1b. 1a is shipped: see [`event-driven-phase-1a.md`](event-driven-phase-1a.md). +Split into 1a and 1b. Both are shipped; what each delivered is recorded below. - **Shipped in 1a.** A versioned snapshot, revision pair, and publish protocol, in a new `snapshot.py` that imports no runtime module. It is owned by `RuntimeState` rather than by the From 96a2a71354b9688c3b3326325ef0aeba9fc27a13 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:38:17 +0800 Subject: [PATCH 06/10] fix(scripts): make the collection benchmark run again `scripts/bench_collect.py` crashed on its first call for anyone who ran it: AttributeError: 'Namespace' object has no attribute 'host' It hand-builds an `argparse.Namespace` to stand in for parsed CLI arguments and hands it to `cli.build_runtime`. That function later grew `host`, `no_git`, `no_dismiss` and `no_ask`; the stand-in was never extended, so the script died on the first attribute read. Reproduced on clean `main` before changing anything, so this is not fallout from the collector work in this branch. CI did not catch it because `scripts/tests/test_bench_collect.py` exercises the measurement helpers rather than `main()`, so every test passed against a script that could not start. This matters more than a broken script usually would: docs/plans/event-driven-session-observation.md and the comment at observation.py:73 both point at this benchmark as the way collection cost is measured, so the tool the repository names for answering "is this slower now?" was the one tool that could not answer it. Filled in with production defaults rather than benchmark-friendly ones, so the figure still describes what a real collect pays. `no_usage=True` remains the one deliberate difference, for the reason already written above it. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- scripts/bench_collect.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/scripts/bench_collect.py b/scripts/bench_collect.py index e6309c2..7872b31 100755 --- a/scripts/bench_collect.py +++ b/scripts/bench_collect.py @@ -597,10 +597,20 @@ def build_runtime( if store_root_overrides is None: built: tuple[Any, Any] = cli.build_runtime( argparse.Namespace( + host="127.0.0.1", port=4553, window_hours=args.window_hours, no_spacedock=False, no_usage=True, + # Production defaults, not benchmark-friendly ones: these three + # arrived on `cli.build_runtime` after this Namespace was written + # and the script crashed on the first of them for anyone who ran + # it. Left enabled so the number still describes what a real + # collect pays. `no_usage` stays the one deliberate difference, + # for the reason given above. + no_git=False, + no_dismiss=False, + no_ask=False, ), started=time.time(), ) From deaafd8f1e3a26ba7e59b4c6add658eb605c3e22 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:38:31 +0800 Subject: [PATCH 07/10] perf(collectors): stop paying for rows the activity gate throws away Three changes on the collection path, which runs on a 2.5s floor while a session is writing. Measured with scripts/bench_collect.py, repeat 5, against a real store: total collect 146.2ms -> 131.8ms (-9.9%), the Claude collector itself 123.9ms -> 110.0ms (-11.2%). The glob loop already takes `os.path.getmtime` for every transcript, then kept only the path. That made the newest-wins comparison re-stat the incumbent for a number it had just measured and dropped, and the per-prefix loop stat the winner a third time. `transcripts` now carries `(path, mtime)`, which removes both stats and the two OSError guards that existed only to cover them. `started_agent_ids` and the pending-member roster were computed one prefix above the `if not (active or show_all): continue` gate, and neither feeds `activity_sources`. `pending_members` is not read until the row is being built, several hundred lines below. So the comprehension, the sort and one `started_agent_ids` call ran for every prefix on disk to produce values discarded for all but the handful that survive the gate. Moved below it. Pure waste rather than a trade: nothing above the gate reads either name. The Cursor collector asked `os.path.exists(wal)` and then `os.path.getmtime(wal)` - two stats for one question, since `getmtime` raises when the file is absent. Worse, both sat under the same handler as the db's own stat, so a WAL that vanished between the two calls dropped the whole chat row instead of falling back to the db's mtime. A WAL is the file most likely to go while being read. Now one suppressed `getmtime`, scoped so its absence costs nothing. Deliberately not taken here: the larger restructuring of the same loop, which would skip the per-session subagent walk for prefixes outside the window. It is a real cost and a bigger change, with a behaviour trade around long-running subagents that deserves its own PR and its own measurement. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- .../cargento_runtime/collectors/claude.py | 62 +++++++++---------- .../cargento_runtime/collectors/cursor.py | 10 ++- 2 files changed, 38 insertions(+), 34 deletions(-) diff --git a/cargento/skills/cargento/cargento_runtime/collectors/claude.py b/cargento/skills/cargento/cargento_runtime/collectors/claude.py index 437d2cb..ce40f13 100644 --- a/cargento/skills/cargento/cargento_runtime/collectors/claude.py +++ b/cargento/skills/cargento/cargento_runtime/collectors/claude.py @@ -394,7 +394,7 @@ def collect( ) -> list[Session]: tasks_by_session = load_tasks(config) team_members = load_team_members(config) - transcripts: dict[str, str] = {} # prefix -> newest transcript path + transcripts: dict[str, tuple[str, float]] = {} # prefix -> (newest path, its mtime) agent_children: dict[str, list[dict[str, Any]]] = {} # parent prefix -> children for fp in runtime_io.glob_stores(config, "claude.projects", "*", "*.jsonl"): base = os.path.basename(fp) @@ -424,23 +424,22 @@ def collect( ) continue prefix = base[:8] - try: - if prefix not in transcripts or mtime > os.path.getmtime(transcripts[prefix]): - transcripts[prefix] = fp - except OSError: - continue # transcript rotated/deleted between glob and stat + # `mtime` is this file's, measured a few lines up in this same pass. The + # newest-wins test used to re-stat the incumbent for a number the loop + # had already taken and dropped, and the per-prefix loop below stat'd the + # winner a third time. Carrying the pair costs nothing and removes both. + if prefix not in transcripts or mtime > transcripts[prefix][1]: + transcripts[prefix] = (fp, mtime) out: list[Session] = [] for prefix in set(transcripts) | set(tasks_by_session): - transcript = transcripts.get(prefix) + newest = transcripts.get(prefix) + transcript = newest[0] if newest else None + transcript_mtime = newest[1] if newest else 0 tasks = sorted( tasks_by_session.get(prefix, []), key=lambda t: int(t["id"]) if str(t["id"]).isdigit() else 0, ) - try: - transcript_mtime = os.path.getmtime(transcript) if transcript else 0 - except OSError: - transcript_mtime = 0 latest_task_mtime = max((t["updated"] for t in tasks), default=0) agent_files = agent_transcripts(transcript, config=config, state=state) children = agent_children.get(prefix, []) @@ -475,6 +474,27 @@ def collect( config, now, c["mtime"], config.working_threshold_sec ) # fresh = running ] + latest_agent_mtime = max( + (a["mtime"] for a in subagents), + default=0, + ) + latest_child_mtime = max((c["mtime"] for c in children), default=0) + # Every subagent write, not just the ones fresh enough to read as + # running: a workflow that has been going for hours parks its parent + # transcript, and without this the session ages out of the window. + latest_agent_file_mtime = max((m for _, m in agent_files), default=0) + activity_sources = ( + latest_task_mtime, + transcript_mtime, + latest_agent_mtime, + latest_agent_file_mtime, + latest_child_mtime, + ) + last_activity = runtime_sessions.newest_plausible(config, now, activity_sources) + active = runtime_sessions.is_fresh(config, now, last_activity, window_hours * 3600) + if not (active or show_all): + continue + # Registered, joined long enough ago that a healthy agent would have # written its first record, and still holding no transcript anywhere. # Sandwiched between two windows that already exist rather than a @@ -496,26 +516,6 @@ def collect( and runtime_sessions.is_fresh(config, now, m["joined"], window_hours * 3600) ] pending_members.sort(key=lambda m: m["joined"]) - latest_agent_mtime = max( - (a["mtime"] for a in subagents), - default=0, - ) - latest_child_mtime = max((c["mtime"] for c in children), default=0) - # Every subagent write, not just the ones fresh enough to read as - # running: a workflow that has been going for hours parks its parent - # transcript, and without this the session ages out of the window. - latest_agent_file_mtime = max((m for _, m in agent_files), default=0) - activity_sources = ( - latest_task_mtime, - transcript_mtime, - latest_agent_mtime, - latest_agent_file_mtime, - latest_child_mtime, - ) - last_activity = runtime_sessions.newest_plausible(config, now, activity_sources) - active = runtime_sessions.is_fresh(config, now, last_activity, window_hours * 3600) - if not (active or show_all): - continue project = ( ( diff --git a/cargento/skills/cargento/cargento_runtime/collectors/cursor.py b/cargento/skills/cargento/cargento_runtime/collectors/cursor.py index d381fa8..24f6e27 100644 --- a/cargento/skills/cargento/cargento_runtime/collectors/cursor.py +++ b/cargento/skills/cargento/cargento_runtime/collectors/cursor.py @@ -728,11 +728,15 @@ def collect( sid = os.path.basename(os.path.dirname(db)) try: mtime = os.path.getmtime(db) - wal = db + "-wal" - if os.path.exists(wal): - mtime = max(mtime, os.path.getmtime(wal)) except OSError: continue + # One stat rather than two: `getmtime` already answers "is there a WAL". + # Suppressed separately from the db's own stat, because sharing the + # handler meant a WAL that vanished between the `exists` and the + # `getmtime` dropped the whole chat row instead of leaving the db's mtime + # standing - and a WAL is exactly the file most likely to go while read. + with contextlib.suppress(OSError): + mtime = max(mtime, os.path.getmtime(db + "-wal")) if not (sessions.is_fresh(config, now, mtime, window_hours * 3600) or show_all): continue title, cwd, model, parent_id, type_name, pending_since = _meta(config, state, db, mtime) From e639994127527313b916de2dd49f797f39122e34 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:39:40 +0800 Subject: [PATCH 08/10] perf(ci): give runtime-floor the pip cache the other four jobs have Five jobs in the quality gate call setup-python; four pass `cache: pip`. `runtime-floor` did not, and it installs the heaviest requirements of the five - requirements-dev.txt, which carries ruff, mypy and coverage - so it downloaded and built the whole toolchain on every run. Measured across two recent runs it took 37s and 39s, against 14-20s for the type-check job that installs the same file with a warm cache. It does not shorten the gate's wall clock: the critical path is changes -> Tests (windows-latest) at 119s -> aggregator, and runtime-floor is nowhere near it. This is runner time and cache-hit rate, not latency. Left alone deliberately: this job also runs the full suite under `coverage` and calls `coverage report`, which enforces pyproject's `fail_under` a second time on a second interpreter. There is a real argument that the threshold belongs to the `test` job alone and that a duplicate gate can fail for reasons unrelated to the 3.11 floor. But AGENTS.md says the threshold only ever ratchets up, and removing one of the two places it is enforced is a decision about how strict the gate should be rather than a defect to fix in a cleanup PR. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- .github/workflows/quality-gate.yml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/.github/workflows/quality-gate.yml b/.github/workflows/quality-gate.yml index 95401d1..fcef983 100644 --- a/.github/workflows/quality-gate.yml +++ b/.github/workflows/quality-gate.yml @@ -160,6 +160,10 @@ jobs: - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 with: python-version: "3.11" + # The only setup-python in this workflow that was missing its cache, + # while installing the heaviest requirements file of the five jobs. + cache: pip + cache-dependency-path: requirements-dev.txt - name: Install toolchain run: python -m pip install -r requirements-dev.txt -r requirements-validation.txt - name: Direct-launch smoke on the supported floor From fa83dcae5adc941979305b74baa20985ced8a8cb Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:43:23 +0800 Subject: [PATCH 09/10] docs(skill): scope the tone check to files this repository owns The sync-docs tone check globbed `docs/plans/*.md`, which is also where per-issue deep-dive notes are kept locally and excluded one file at a time in .git/info/exclude. So the check reported drift in somebody's private scratch: three hits in a deep-dive note, none of them in a file any reviewer will read, and none fixable without editing a file that is not repository content. `git ls-files` instead of `ls`. It still covers all nineteen tracked prose docs, including the two plans that remain, and it keeps the reason the original `ls` was there: step 4 deletes plan docs, and a bare glob matching nothing stays literal and makes grep exit 2. Same shape as the validator fix earlier in this branch. Both walked `docs/` as though everything under it were source, and `docs/` has held vendored mods, orphan-branch state and local notes for a while now. With this, the tone check reads clean on a tree whose prose is already clean, which is what makes it worth running. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- .claude/skills/sync-docs/SKILL.md | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/.claude/skills/sync-docs/SKILL.md b/.claude/skills/sync-docs/SKILL.md index 8d3da22..e8a4841 100644 --- a/.claude/skills/sync-docs/SKILL.md +++ b/.claude/skills/sync-docs/SKILL.md @@ -398,8 +398,12 @@ minutes, a Python version. Stale counts are this repository's most common drift. # e. Tone: no em/en dashes or curly quotes in the human-facing prose docs. Nothing in CI # enforces this, so it is the one anti-drift check that only exists here. - # `ls` builds the list because step 4 deletes plan docs: a bare `docs/plans/*.md` that - # matches nothing stays literal and makes grep exit 2 on a "No such file" error. + # `git ls-files` builds the list, for two reasons. Step 4 deletes plan docs, and a bare + # `docs/plans/*.md` that matches nothing stays literal and makes grep exit 2 on a + # "No such file" error. And `docs/plans/` is where per-issue deep-dive notes are kept + # locally, excluded per file in .git/info/exclude: those are somebody's scratch, not + # this repository's prose, and globbing them made the check red for text no reviewer + # will ever read. Tracked files are exactly the ones the standard governs. # The explicit if/else is here because a bare `grep && echo` exits 1 when the docs are # CLEAN, which reads as failure to anyone (or anything) checking the status. # NOTE: the carve-out is for INLINE spans only, not fenced blocks. A dash inside a @@ -411,8 +415,8 @@ minutes, a Python version. Stale counts are this repository's most common drift. # ends with someone mangling a documented literal to quiet it. # The per-file loop keeps the filename in the output; piping every doc through one sed # would report a line number with nothing to open. - if for f in $(ls README.md HOW_TO_USE.md CONTRIBUTING.md COMPATIBILITY.md SECURITY.md \ - docs/design-*.md docs/plans/*.md 2>/dev/null); do + if for f in $(git ls-files -- README.md HOW_TO_USE.md CONTRIBUTING.md COMPATIBILITY.md \ + SECURITY.md 'docs/design-*.md' 'docs/plans/*.md'); do sed 's/`[^`]*`//g' "$f" | grep -n '—\|–\|[“”‘’]' | sed "s|^|$f:|" done | grep .; then echo "TONE DRIFT: reapply Voice and tone to the files listed above" From 1812139c43e272de052d5fac97cbc579cf9011d7 Mon Sep 17 00:00:00 2001 From: Jared Scott Date: Sat, 29 Aug 2026 12:45:57 +0800 Subject: [PATCH 10/10] test(cursor): pin the chat that outlives its WAL The WAL fix in the previous commit changed behaviour, so it gets a test that fails without it. Verified both ways: reverted to the pre-fix collector this asserts 1 != 0, because losing the race withdrew the whole chat row; with the fix the row survives on the db's own mtime. The race is made deterministic rather than waited for. `os.path.getmtime` raises for the `-wal` path and answers honestly for everything else, which is exactly the state the collector sees when a checkpoint removes the WAL between the existence check and the stat. Co-Authored-By: Claude Opus 5 (1M context) Signed-off-by: Jared Scott --- .../cargento/tests/test_sqlite_collectors.py | 31 +++++++++++++++++++ 1 file changed, 31 insertions(+) diff --git a/cargento/skills/cargento/tests/test_sqlite_collectors.py b/cargento/skills/cargento/tests/test_sqlite_collectors.py index 0814bce..33505dd 100644 --- a/cargento/skills/cargento/tests/test_sqlite_collectors.py +++ b/cargento/skills/cargento/tests/test_sqlite_collectors.py @@ -434,6 +434,37 @@ def test_cursor_walks_past_a_child_id_that_belongs_to_no_blob(self) -> None: self.assertEqual("vega", sessions[0]["model"]) + def test_cursor_keeps_a_chat_whose_wal_disappears_between_the_two_stats(self) -> None: + # The WAL is stat'd to fold its mtime into the chat's, and it is the file + # most likely to be checkpointed away while the collector is mid-read. + # Sharing the db's own OSError handler meant losing that race withdrew + # the whole row rather than falling back to the db's mtime: the chat + # vanished from the board because a sibling file did. + if not runtime_io.sqlite_available(): + self.skipTest("sqlite3 unavailable") + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + root_id, blobs = self._cursor_chat([self._cursor_message("vega")]) + self._cursor_store( + root, "sess-wal", [{"name": "chat", "latestRootBlobId": root_id}], blobs + ) + (root / "chats" / "hash1" / "sess-wal" / "store.db-wal").write_bytes(b"") + real_getmtime = os.path.getmtime + + def vanishing_wal(path: Any) -> float: + # Present to the glob and to any earlier check, gone by the time + # its mtime is asked for. Only the WAL races; everything else + # answers honestly. + if str(path).endswith("-wal"): + raise OSError(2, "No such file or directory") + return real_getmtime(path) + + with mock.patch("os.path.getmtime", side_effect=vanishing_wal): + sessions = self._collect_cursor(root) + + self.assertEqual(1, len(sessions)) + self.assertEqual("chat", sessions[0]["title"]) + def test_cursor_keeps_its_title_when_the_store_has_no_blobs_table(self) -> None: # The failure that costs the most: a store on a schema without `blobs` # raises `no such table`, and routing that through the store-error path