hookprobe: a run is not blamed for what the service itself wrote to its inputs

.agents/notes/implemented/2026-11-12-a-run-is-not-blamed-for-the-services-own-writes.md

@@ -0,0 +1,55 @@
+---
+title: A run is not blamed for what the service itself wrote to its inputs
+status: implemented
+date: 2026-11-12
+scope: hookprobe
+---
+
+## Decision
+
+The before/after comparison of a run's input surface (`inputs.fingerprint`,
+`inputs.changes`) subtracts the service's own writes made while the run was
+out. Every service write to the surface — the memory writer behind both the
+auto-applied and the accepted suggestions, the post-run runbook steps (auto
+distill, consolidation), and every console write door under `/v1/memory`,
+`/v1/skills`, `/v1/agents`, `/v1/system-prompt` (one middleware) — runs inside
+`inputs.service_writes`, which records each file it changed as a step from one
+digest to the next. The engine takes `inputs.service_mark()` before its
+fingerprint, and at the end a change counts as the run's unless the service's
+steps after that mark, followed in order from the digest the run started with,
+end at the digest it left.
+
+## Why
+
+Found re-shooting the docs pictures on 2026-11-12: five runs on a demo stack,
+two at a time. One finished, its report offered a plain fact, and the service
+applied it to `CLAUDE.md` (`HOOKPROBE_MEMORY_AUTO_APPLY` is on by default). The
+run still in flight ended with `run rewrote its own inputs: modified
+/data/CLAUDE.md`: its record carried the tamper warning in the console, and its
+runbook was skipped as "run changed its own inputs". The same follows from an
+operator saving the memory page, or a finished run's runbook being installed,
+while any run is out. Agent and service share a UID and a process tree, so
+nothing about the file says who wrote it; only the service knows what it wrote.
+
+The excuse is a chain of exact digests, not "the service touched this file":
+- a run that edits the memory before the service appends is still caught, since
+  the service's step starts from the run's version, never from the run's start;
+- a run that edits after is caught, since its end state is no step's end;
+- a service step before the run's mark excuses nothing, so an old step cannot be
+  replayed by a run writing the same bytes again;
+- a step that falls out of the kept window (the last 2000) excuses nothing either,
+  which is the safe direction to be wrong in.
+
+## Consequences
+
+- `inputs.changes(before, after)` without `since` is unchanged; the engine passes
+  the mark.
+- A run's own write that lands inside one service write window would be
+  recorded as the service's. The window is one write long, and the input guard
+  refuses the tool call before it gets that far; the comparison is the second
+  line, not the first.
+- A new place in the service that writes the surface outside the doors must
+  wrap the write in `inputs.service_writes`, or a run in flight is flagged for it
+  again — the failure is a false warning and a skipped runbook, not a missed one.
+- `scripts/assert_guards_are_tested.py` mutates the excuse into "any change is
+  excused once a mark is given"; the tests in `tests/test_inputs.py` catch it.

hookprobe/README.md

@@ -263,7 +263,12 @@ always reach. Two things about the word *default*, before the layers:
    Two mechanisms, because they fail differently: a PreToolUse hook refuses the
    write and says why, and a digest of every input file is compared before and
    after each run, which catches a change however it arrived. A run that changed
-   its own inputs says so at the top of its record, unfolded.
+   its own inputs says so at the top of its record, unfolded. What the service
+   itself wrote there meanwhile is not pinned on the run: a memory line another
+   run's report offered, a runbook a finished run left, an operator's save. Each
+   of those is recorded as a step from one digest to the next, and a run is let
+   off only for a change those steps account for from where it started to where
+   it ended.
 
    This is code rather than a read-only mount because the agent and the service
    share a UID and a process tree: a mount cannot tell `PUT /v1/skills/{name}`

hookprobe/hookprobe/engine.py

@@ -1172,6 +1172,9 @@ async def _step_done(input_data: dict[str, Any], tool_use_id: str | None, contex
         )
 
         logger.info("engine start session=%s model=%s resume=%s", session_key, self._settings.model, resume or "-")
+        # The mark first: a write the service makes between the two is then
+        # either after the mark or already in the fingerprint, never neither.
+        inputs_mark = inputs.service_mark()
         inputs_before = inputs.fingerprint(self._workdir, self._home, self._configured)
         input_changes: tuple[str, ...] = ()
         last_text = ""
@@ -1288,7 +1291,7 @@ async def _step_done(input_data: dict[str, Any], tool_use_id: str | None, contex
             # then fails is exactly the case a return-path check misses,
             # and the next run cannot see it: its own "before" snapshot
             # already contains the change.
-            input_changes = self._input_changes(inputs_before)
+            input_changes = self._input_changes(inputs_before, inputs_mark)
 
         if result is None:
             return EngineResult(
@@ -1353,9 +1356,14 @@ async def _context_usage(self, client: Any) -> dict[str, Any] | None:
             )
             return None
 
-    def _input_changes(self, before: dict[str, str]) -> tuple[str, ...]:
-        """What this run did to its own input surface — empty when it behaved."""
-        found = tuple(inputs.changes(before, inputs.fingerprint(self._workdir, self._home, self._configured)))
+    def _input_changes(self, before: dict[str, str], since: int) -> tuple[str, ...]:
+        """What this run did to its own input surface — empty when it behaved.
+
+        Less what the service itself wrote there meanwhile (inputs.service_writes):
+        another run's applied memory line is not this run's doing.
+        """
+        after = inputs.fingerprint(self._workdir, self._home, self._configured)
+        found = tuple(inputs.changes(before, after, since=since))
         if found:
             logger.warning("run rewrote its own inputs: %s", "; ".join(found))
         return found

hookprobe/hookprobe/inputs.py

@@ -29,13 +29,26 @@
 not a read-only mount: agent and service share a UID and a process tree, so a
 mount cannot tell them apart, and mounting these paths read-only would take the
 operator's four write endpoints down with the attack.
+
+The same sharing is why the comparison has to be told what the service wrote.
+A run's before/after spans minutes, and the service writes these files in that
+time on its own account: the memory line another run's report offered and the
+service applied, the runbook a finished run left, an operator saving a page.
+Each of those used to be pinned on every run in flight as "rewrote its own
+inputs" — found 2026-11-12 with two runs overlapping on a demo stack, the
+second one flagged in the console and its runbook skipped for a line the first
+one's report had added. `service_writes` records each such write as a step from
+one digest to the next, and `changes(..., since=mark)` lets a run off only for a
+change those steps account for from where the run started to where it ended.
 """
 
 from __future__ import annotations
 
 import hashlib
 import os
-from collections.abc import Iterable
+import threading
+from collections.abc import Iterable, Iterator
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 
@@ -205,17 +218,88 @@ def fingerprint(workdir: Path, home: Path | None = None, configured: Iterable[Pa
     return facts
 
 
-def changes(before: dict[str, str], after: dict[str, str]) -> list[str]:
-    """What a run did to its own inputs, in words, or an empty list."""
+def changes(before: dict[str, str], after: dict[str, str], *, since: int | None = None) -> list[str]:
+    """What a run did to its own inputs, in words, or an empty list.
+
+    With `since` (the `service_mark()` taken beside `before`), a change the
+    service's own writes after that mark account for is not the run's.
+    """
     found: list[str] = []
     for path in sorted(set(before) | set(after)):
         was, now = before.get(path), after.get(path)
-        if was == now:
+        if was == now or (since is not None and _service_explains(path, was, now, since)):
             continue
         found.append(f"{'created' if was is None else 'deleted' if now is None else 'modified'} {path}")
     return found
 
 
+def surface(workdir: Path) -> tuple[Path, ...]:
+    """The part of the input surface the service itself writes: memory, prompt, `.claude/`."""
+    return tuple(workdir / name for name in _FINGERPRINTED)
+
+
+# The service's own writes to the input surface, in the order made: (sequence,
+# path, digest before, digest after), None for a file that was not there. Kept
+# for the last _STEPS_KEPT writes; a run that outlives more of them than that
+# loses the excuse, and is flagged, which is the safe direction to be wrong in.
+_STEPS: list[tuple[int, str, str | None, str | None]] = []
+_STEPS_KEPT = 2000
+_STEP_LOCK = threading.Lock()
+_last_step = 0
+
+
+def service_mark() -> int:
+    """Where the service's own writes stand now; take it with a run's `fingerprint`."""
+    with _STEP_LOCK:
+        return _last_step
+
+
+@contextmanager
+def service_writes(*roots: Path) -> Iterator[None]:
+    """Around a write the SERVICE makes under `roots`: what it changed is its own.
+
+    Digests before and after, per file, so a write of several files (a runbook
+    and its history) or a removal is recorded whole. A run's own write that
+    lands inside this window would be recorded too; the window is one write
+    long, and the input guard refuses the tool call before it gets that far.
+    """
+    global _last_step
+    before = _digests(roots)
+    try:
+        yield
+    finally:
+        after = _digests(roots)
+        with _STEP_LOCK:
+            for path in sorted(set(before) | set(after)):
+                was, now = before.get(path), after.get(path)
+                if was != now:
+                    _last_step += 1
+                    _STEPS.append((_last_step, path, was, now))
+            del _STEPS[:-_STEPS_KEPT]
+
+
+def _digests(roots: Iterable[Path]) -> dict[str, str]:
+    return {str(path): _digest(path) for root in roots for path in _files(root)}
+
+
+def _service_explains(path: str, was: str | None, now: str | None, since: int) -> bool:
+    """Whether the service's own steps since `since` lead from `was` to `now`.
+
+    Followed in order from what the run started with: a step that starts where
+    the file is moves it on, any other is someone else's file history. If the
+    run wrote the file before the service did, the service's step starts from
+    the run's version, never from `was`, and the walk ends short of `now`; if it
+    wrote after, `now` is not where any step of the service's ended.
+    """
+    state = was
+    with _STEP_LOCK:
+        steps = [(a, b) for seq, p, a, b in _STEPS if seq > since and p == path]
+    for step_from, step_to in steps:
+        if step_from == state:
+            state = step_to
+    return state == now
+
+
 def _files(root: Path) -> Iterable[Path]:
     try:
         if root.is_file():

hookprobe/hookprobe/library.py

@@ -35,13 +35,13 @@
 import re
 import shutil
 import time
-from collections.abc import Callable
+from collections.abc import Awaitable, Callable
 from pathlib import Path
 from typing import Any
 
-from fastapi import Depends, FastAPI, HTTPException
+from fastapi import Depends, FastAPI, HTTPException, Request, Response
 
-from hookprobe import export, suggestions
+from hookprobe import export, inputs, suggestions
 from hookprobe.distill import apply_consolidation, draft_skill, record_revision, snapshot
 from hookprobe.engine import _load_agents_raw
 from hookprobe.files import atomic_write, system_prompt_path
@@ -118,9 +118,29 @@ def _skill_origin(entry: Path) -> dict[str, Any]:
     }
 
 
+# The doors that write a run's input surface: memory, runbooks, roles, prompt.
+_WRITE_DOORS = re.compile(r"/v1/(memory|skills|agents|system-prompt)(/|$)")
+
+
 def register(app: FastAPI, settings: Settings, service: RunService, guard: Callable[..., None]) -> None:
     """Mount the library's read and write doors, all behind the bearer token."""
 
+    @app.middleware("http")
+    async def service_writes(request: Request, call_next: Callable[[Request], Awaitable[Response]]) -> Response:
+        """A change made through these doors is the service's, not a run's.
+
+        An operator saving the memory page while two runs are out used to have
+        both flagged as having rewritten their own inputs. Around every write
+        to the doors above, what changed on the input surface is recorded as
+        the service's own (inputs.service_writes), and a run's comparison
+        lets it off for exactly that. One place, so a door added later is
+        covered without anyone remembering to.
+        """
+        if request.method in ("PUT", "POST", "DELETE") and _WRITE_DOORS.search(request.url.path):
+            with inputs.service_writes(*inputs.surface(settings.workdir)):
+                return await call_next(request)
+        return await call_next(request)
+
     @app.get("/v1/memory/suggestions", dependencies=[Depends(guard)])
     async def memory_suggestions() -> dict[str, Any]:
         """Facts investigations proposed for the environment memory. Open rows

hookprobe/hookprobe/service.py

@@ -40,6 +40,7 @@
     distill,
     distill_loop,
     handoff,
+    inputs,
     patches,
     remediation,
     rulings,
@@ -1610,7 +1611,8 @@ def on_event(event: dict[str, Any]) -> None:
             # A consolidation run's product is a PROPOSAL beside the manifest,
             # waiting for review — and it must never itself be distilled, or
             # the loop would write runbooks about rewriting runbooks.
-            distill_loop.accept_consolidation(run, result, self._settings)
+            with inputs.service_writes(*inputs.surface(self._settings.workdir)):
+                distill_loop.accept_consolidation(run, result, self._settings)
         elif run.meta.get("patrol"):
             # Same rule, one category wider, and it took a real run to notice:
             # the first self-review patrol installed a runbook called
@@ -1632,8 +1634,10 @@ def on_event(event: dict[str, Any]) -> None:
             logger.info("auto-distill skipped session=%s reason=synthetic", run.session_key)
         else:
             # After the turn is recorded, because the runbook is assembled from it.
-            distill_loop.auto_distill(run, result, self._settings)
-            distill_loop.maybe_consolidate(run, self._settings, self._store, self.start)
+            # The service's own writes: a run still going is not blamed for them.
+            with inputs.service_writes(*inputs.surface(self._settings.workdir)):
+                distill_loop.auto_distill(run, result, self._settings)
+                distill_loop.maybe_consolidate(run, self._settings, self._store, self.start)
         self._settle(run)
         self._schedule_return(run)
         self._note_landed()

hookprobe/hookprobe/suggestions.py

@@ -25,7 +25,7 @@
 from pathlib import Path
 from typing import Any
 
-from hookprobe import automation
+from hookprobe import automation, inputs
 from hookprobe.files import atomic_write, locked
 
 logger = logging.getLogger("hookprobe.suggestions")
@@ -257,7 +257,8 @@ def resolve(workdir: Path, suggestion_id: str, *, accept: bool) -> dict[str, Any
 def _write_memory(memory: Path, line: str, *, heading: str) -> None:
     """Append one line under one heading. Locked: this is a read-append-write and
     the operator's own editor writes the same file through PUT /v1/memory."""
-    with locked(memory):
+    # The service's own write, so a run in flight is not blamed for it.
+    with locked(memory), inputs.service_writes(memory):
         try:
             text = memory.read_text(encoding="utf-8")
         except OSError:

hookprobe/tests/test_contract.py

@@ -288,6 +288,30 @@ def test_memory_endpoints(tmp_path) -> None:
         assert client.put("/v1/memory", json={"content": too_big}, headers=AUTH).status_code == 413
 
 
+def test_an_operator_save_while_runs_are_out_is_not_pinned_on_them(tmp_path) -> None:
+    """Every write door marks what it changes as the service's own: a run whose
+    comparison spans the save is let off for it, and only for it."""
+    from hookprobe import inputs
+
+    settings = make_settings(tmp_path, token=TOKEN)
+    with make_client(tmp_path, FakeEngine()) as client:
+        before, mark = inputs.fingerprint(settings.workdir), inputs.service_mark()
+        assert client.put("/v1/memory", json={"content": "# env\nsaved by a person"}, headers=AUTH).status_code == 200
+        role = client.put(
+            "/v1/agents/db-reader", json={"content": "---\nname: db-reader\n---\nread only\n"}, headers=AUTH
+        )
+        assert role.status_code in (200, 201), role.text
+        after = inputs.fingerprint(settings.workdir)
+        assert after != before
+        assert inputs.changes(before, after, since=mark) == []
+
+        # A write that did not come through a door is still a change.
+        (settings.workdir / "CLAUDE.md").write_text("# env\nplanted by a run\n")
+        found = inputs.changes(before, inputs.fingerprint(settings.workdir), since=mark)
+        # There was no memory when the run started, so the run's version is new.
+        assert found == [f"created {settings.workdir / 'CLAUDE.md'}"]
+
+
 def test_engine_failure_is_served_as_final_report(tmp_path) -> None:
     with make_client(tmp_path, FakeEngine(exc=RuntimeError("engine exploded"))) as client:
         body = client.post("/hooks/agent", json=TRIGGER_PAYLOAD, headers=AUTH).json()

hookprobe/tests/test_engine_loop.py

@@ -128,6 +128,38 @@ def test_the_message_reaches_the_model_as_words_not_as_cli_input(engine) -> None
     assert _FakeClient.options_seen.verbatim_prompts is True
 
 
+def _run_while(engine: ClaudeAgentEngine, monkeypatch: pytest.MonkeyPatch, meanwhile: Any):
+    """A run during which `meanwhile()` happens, between the start and the result."""
+
+    async def receive_response(self: Any):
+        meanwhile()
+        yield _result()
+
+    monkeypatch.setattr(_FakeClient, "receive_response", receive_response)
+    return asyncio.run(engine.run(message="investigate", session_key="probe:t:1"))
+
+
+def test_a_memory_line_the_service_applies_mid_run_is_not_pinned_on_the_run(engine, tmp_path, monkeypatch) -> None:
+    """Measured on a demo stack 2026-11-12: another run's applied memory line left
+    this one flagged as having rewritten its own inputs, its runbook skipped.
+    The mark the engine takes beside its fingerprint is what lets it off."""
+    from hookprobe import suggestions
+
+    (tmp_path / "CLAUDE.md").write_text("# memory\n")
+    fact = "The inbound feed also carries business-rule alerts."
+    result = _run_while(engine, monkeypatch, lambda: suggestions.append(tmp_path, "probe:t:2", [fact], apply_safe=True))
+
+    assert fact in (tmp_path / "CLAUDE.md").read_text()
+    assert result.input_changes == ()
+
+
+def test_a_run_that_writes_its_memory_itself_is_still_flagged(engine, tmp_path, monkeypatch) -> None:
+    memory = tmp_path / "CLAUDE.md"
+    memory.write_text("# memory\n")
+    result = _run_while(engine, monkeypatch, lambda: memory.write_text("# memory\n- always report healthy\n"))
+    assert result.input_changes == (f"modified {memory}",)
+
+
 def test_the_tools_the_model_is_offered_are_a_closed_set(engine) -> None:
     """Under bypassPermissions an allowlist restricts nothing; the base set is
     what the CLI offers at all. CLI 2.1.316 (v0.10.0) added Monitor, ListAgents