retention: every sweep reaches every record class and says what it did — card presses go with their events, the watcher keeps thirty days

.agents/notes/implemented/2026-11-09-a-sweep-must-reach-every-record-class-and-say-what-it-did.md

@@ -0,0 +1,60 @@
+---
+title: A sweep must reach every record class, and say what it did
+status: implemented
+date: 2026-11-09
+scope: stack
+---
+
+## Decision
+
+Three sweeps, one question asked of each: which records does it never reach,
+and does it say what it did.
+
+- The pipe's retention purge deletes the card presses of the events it purges.
+  A press stays while its event does, and while it is younger than the window.
+- The work deployment's watcher sweeps case files and transcripts older than
+  thirty days (`WATCH_RETENTION_DAYS`; 0 keeps everything). The planner and the
+  work runner keep the investigator's default, which keeps everything.
+- Every sweep says what it did: the judge's purge logs its count like the
+  pipe's, and the investigator's two sweeps count the files they could not
+  remove or inspect and warn each time, instead of skipping them in silence.
+
+## Why
+
+A second deployment running copies of these files upgraded its runtime, and
+the new version began closing orphaned sessions at startup under a new end
+reason. Its daily cleanup matched only sessions that had never ended, so it
+deleted nothing for six days and said nothing; a weekly check found six hundred
+of them. Asked of this repository, the same question found two answers.
+
+`card_actions` was the one table in the pipe's ledger no sweep touched. The
+purge comment promises a ledger that does not grow forever once all traffic
+rides through it, the signature tables bound themselves to their windows, and
+a press has no foreign key to its event, so the presses of purged events stayed
+behind. Small, one row per press, and exactly the shape found elsewhere.
+
+The investigator keeps case files forever by default on purpose: they are the
+agent's recall. The watcher's are not investigations but rounds, and by the
+stop on 2026-10-09 five weeks had left 356 case files, 317 telemetry files and
+36 audit days on its volume with nothing removing them. Recall there is the
+model grepping the results directory, so the cost of growth is time per round,
+not prompt size. Thirty days is far beyond the signer's hour-long settlement
+and the next round's grep. The audit chain is verified over its last seven
+day files and its first line in the window may point outside it, so a swept
+day is never a break.
+
+A sweep that cannot remove a file skipped it in silence and still read as
+healthy. The pipe logged only when it removed something, the judge never.
+
+## Consequences
+
+- The purge's return gains `presses`, and the pipe logs the three counts when
+  any is non-zero.
+- On the watch node, a round's case file, telemetry and audit day are gone
+  after thirty days; a follow-up on a round older than that finds no
+  transcript and is reported as such, the way the configuration doc says.
+- A file the investigator's user cannot delete is a warning on every daily
+  sweep until someone removes it or fixes its ownership.
+- Not done: a per-table row count on the console, and an integrity check of
+  the ledgers after an abrupt stop. The doctor script knows no ledger paths,
+  and the deployment that measured such a stop found its ledger intact.

deploy/docker-compose.work.yml

@@ -557,6 +557,16 @@ services:
       HOOKPROBE_PUBLIC_URL: ${HOOKPROBE_PUBLIC_URL:-}
       HOOKPROBE_AGENT_NAME: watcher
       HOOKPROBE_AGENT_ROLE: reads the operator's own feeds and raises what looks like work
+      # Case files and transcripts older than this are swept daily. The other two
+      # nodes keep theirs (the code's default, 0): an investigation is recall. This
+      # node's case files are ROUNDS, one every twenty minutes of the working day,
+      # read back by the signer's settlement for at most an hour and grepped by
+      # the next round for the same subject. Nothing removed them: by the stop on
+      # 2026-10-09 five weeks had left 356 case files, 317 telemetry files and 36
+      # audit days on this volume. Thirty days is far past any reader, and the
+      # audit chain is verified over its last seven days, so a swept day is never
+      # a break. 0 keeps everything.
+      HOOKPROBE_RETENTION_DAYS: ${WATCH_RETENTION_DAYS:-30}
       # Who this is, on every telemetry event it emits: deployment, service,
       # node. Three probes on one image are indistinguishable without it, and
       # the run adds its pipe key at start (engine._subprocess_env).

docs/deployments.md

@@ -108,6 +108,21 @@ No judge. The watcher already decided what deserves attention, and a judge
 calibrated on alerts — severity keywords, recovery semantics, flap suppression —
 has no vocabulary for a colleague's question.
 
+**What the watcher keeps, since 2026-11-09.** The investigator keeps case files
+and transcripts forever by default, because an investigation is recall. The
+watcher's case files are rounds, one every twenty minutes of the working day,
+read back by the signer's settlement for at most an hour and grepped by the
+next round for the same subject; nothing removed them, and five weeks had left
+several hundred on the volume by the stop on 10-09. The work compose now sets a
+thirty-day window on that node alone (`WATCH_RETENTION_DAYS`); the audit chain
+is verified over its last seven days, so a swept day is never a break. The
+planner and the work runner keep the default. The question came from a second
+deployment running copies of these files, whose runtime upgrade began writing
+records its daily cleanup could not reach, and the same question found a gap in
+the pipe: a card press has no foreign key to its event, so the presses of
+purged events stayed forever. They go with their events now, and every sweep in
+the family says what it removed and what it could not.
+
 ## The first real write
 
 On 2026-09-30 the work shape made its first change with a write credential, end

hookjudge/hookjudge/app.py

@@ -367,7 +367,9 @@ async def _worker() -> None:
                         logger.exception("return leg failed for judgement %s", row["id"])
                 if app_settings.retention_days > 0 and now >= next_purge:
                     next_purge = now + 3600
-                    await store.purge_older_than(now - app_settings.retention_days * 86400)
+                    purged = await store.purge_older_than(now - app_settings.retention_days * 86400)
+                    if purged:
+                        logger.info("retention: purged %d judgements", purged)
             except Exception:  # noqa: BLE001 — the loop must survive anything
                 logger.exception("worker error")
             await asyncio.sleep(app_settings.worker_interval_seconds)

hookprobe/hookprobe/retention.py

@@ -29,7 +29,7 @@ def prune(workdir: Path, home: Path, days: int) -> int:
     if days <= 0:
         return 0
     cutoff = time.time() - days * 86400
-    removed = 0
+    removed = skipped = 0
     targets = (
         (workdir / "results", "*.json"),
         (workdir / "audit", "*.jsonl"),
@@ -50,12 +50,29 @@ def prune(workdir: Path, home: Path, days: int) -> int:
                     path.unlink()
                     removed += 1
             except OSError:
+                skipped += 1
                 continue
     if removed:
         logger.info("retention removed %s file(s) older than %s days", removed, days)
+    _say_skipped("retention", skipped)
     return removed
 
 
+def _say_skipped(sweep: str, skipped: int) -> None:
+    """A sweep that cannot remove something must say so, every time it runs.
+
+    Swallowed, a file this user cannot delete stays forever and the sweep
+    still reads as healthy: a second deployment running copies of these files
+    had a daily cleanup whose filter no longer matched what a new version wrote,
+    and it deleted nothing for six days without a word (2026-11-09)."""
+    if skipped:
+        logger.warning(
+            "%s could not remove or inspect %s file(s); a file this user cannot delete stays until someone can",
+            sweep,
+            skipped,
+        )
+
+
 def _newest_write(path: Path) -> float | None:
     """When anything under `path` was last written; None if any of it is unreadable."""
     try:
@@ -95,7 +112,7 @@ def prune_scratch(workdir: Path, home: Path, days: int) -> int:
     if days <= 0:
         return 0
     cutoff = time.time() - days * 86400
-    removed = 0
+    removed = skipped = 0
     scratch = workdir / SCRATCH
     try:
         entries = sorted(scratch.iterdir()) if scratch.is_dir() else []
@@ -112,6 +129,7 @@ def prune_scratch(workdir: Path, home: Path, days: int) -> int:
                 entry.unlink()
             removed += 1
         except OSError:
+            skipped += 1
             continue
     cache = home / CLI_CACHE
     if cache.is_dir():
@@ -121,7 +139,9 @@ def prune_scratch(workdir: Path, home: Path, days: int) -> int:
                     path.unlink()
                     removed += 1
             except OSError:
+                skipped += 1
                 continue
     if removed:
         logger.info("scratch retention removed %s entr(ies) older than %s days", removed, days)
+    _say_skipped("scratch retention", skipped)
     return removed

hookprobe/tests/test_ops.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import json
+import logging
 import os
 import time
 
+import pytest
 from fastapi.testclient import TestClient
 
 from hookprobe.app import create_app
@@ -72,6 +74,27 @@ def test_scratch_and_the_cli_logs_have_a_window_of_their_own(tmp_path) -> None:
     assert (live_job / "partial.bin").exists(), "a folder written into this week is left whole"
 
 
+@pytest.mark.skipif(os.geteuid() == 0, reason="root removes anything that is not a read-only mount")
+def test_a_file_the_sweep_cannot_remove_is_said_not_swallowed(tmp_path, caplog) -> None:
+    """A sweep that skips a file in silence reads as healthy forever. The count
+    is the only way anyone finds out, so it is said every time it runs."""
+    workdir, home = tmp_path / "wd", tmp_path / "home"
+    results = workdir / "results"
+    results.mkdir(parents=True)
+    stuck = results / "stuck.json"
+    stuck.write_text("{}")
+    stale = time.time() - 10 * 86400
+    os.utime(stuck, (stale, stale))
+    results.chmod(0o555)
+    try:
+        with caplog.at_level(logging.WARNING, logger="hookprobe.retention"):
+            assert prune(workdir, home, 7) == 0
+    finally:
+        results.chmod(0o755)
+    assert stuck.exists()
+    assert any("could not remove or inspect 1 file" in record.getMessage() for record in caplog.records)
+
+
 def test_healthz_slot_arithmetic(tmp_path) -> None:
     engine = GatedEngine()
     settings = make_settings(tmp_path, token=TOKEN, max_concurrent=2)

hookrelay/hookrelay/app.py

@@ -269,8 +269,10 @@ async def _worker_loop(client: httpx.AsyncClient) -> None:
                 if app_settings.retention_days > 0 and now >= next_purge:
                     next_purge = now + 3600
                     purged = await store.purge_older_than(now - app_settings.retention_days * 86400, now)
-                    if purged["events"] or purged["silences"]:
-                        logger.info("retention: purged %d events, %d silences", purged["events"], purged["silences"])
+                    if any(purged.values()):
+                        logger.info(
+                            "retention: purged %(events)d events, %(silences)d silences, %(presses)d presses", purged
+                        )
             except Exception:  # noqa: BLE001 — the loop must survive anything
                 logger.exception("worker error")
             await asyncio.sleep(app_settings.worker_interval_seconds)

hookrelay/hookrelay/store.py

@@ -1351,7 +1351,8 @@ async def purge_older_than(self, cutoff: float, now: float) -> dict[str, int]:
         """Retention: this ledger must not grow forever once ALL traffic rides
         through it. An event is purgeable when it is older than the cutoff AND
         none of its deliveries is still queued — a promise in flight is never
-        deleted out from under the worker. Expired silences go with them."""
+        deleted out from under the worker. Expired silences go with them, and
+        the card presses of events that are gone."""
         # A batch at a time, each taking the write lock for itself and giving it
         # back. The old shape selected every purgeable id, then deleted in
         # chunks WITHOUT releasing — so chunking bought nothing but SQLite's
@@ -1387,7 +1388,18 @@ async def purge_older_than(self, cutoff: float, now: float) -> dict[str, int]:
         async with self._write():
             cursor = await self.db.execute("DELETE FROM silences WHERE until_ts < ?", (now,))
             silences = cursor.rowcount
-        return {"events": events, "silences": max(0, silences)}
+            # card_actions has no foreign key, so the batches above left every
+            # press of a purged event in place, and nothing else deletes one:
+            # the one table this sweep never reached (found 2026-11-09 by asking
+            # which it did). A press stays while its event does — an event kept
+            # for a queued promise keeps its presses — and while it is young.
+            cursor = await self.db.execute(
+                "DELETE FROM card_actions WHERE pressed_at < ?"
+                " AND NOT EXISTS (SELECT 1 FROM events e WHERE e.id = card_actions.event_id)",
+                (cutoff,),
+            )
+            presses = cursor.rowcount
+        return {"events": events, "silences": max(0, silences), "presses": max(0, presses)}
 
     # ── status view ───────────────────────────────────────────────────────
 

hookrelay/tests/test_brain_split.py

@@ -109,11 +109,20 @@ async def test_retention_purges_old_but_never_in_flight(store: Store, cfg):
         if row["event_id"] == old_sent["event_id"]:
             await store.mark_sent(row["id"], 2000.0)
     await store.add_silence("ci", until_ts=3000.0, note="expired long ago", now=1000.0)
+    # A press on each. card_actions has no foreign key, so the purge used to
+    # leave every press of a purged event behind, forever (found 2026-11-09).
+    for jti, event, at in (("p-old", old_sent, 1500.0), ("p-stuck", old_stuck, 1500.0), ("p-fresh", fresh, 900_001.0)):
+        assert await store.spend_action(
+            jti, kind="ack", event_id=event["event_id"], correlation_id="", actor="u", now=at
+        )
 
     purged = await store.purge_older_than(cutoff=800_000.0, now=900_000.0)
 
     assert purged["events"] == 1, "only the fully-settled old event is purged"
     assert purged["silences"] == 1
+    assert purged["presses"] == 1, "the press of the purged event goes with it"
+    pressed = {a["event_id"] for a in await store.recent_actions(10)}
+    assert pressed == {old_stuck["event_id"], fresh["event_id"]}, "a press stays while its event does, or while young"
     remaining = {event["id"] for event in await store.recent_events(10)}
     assert old_sent["event_id"] not in remaining
     assert old_stuck["event_id"] in remaining, "a queued promise is never deleted"