feat(guard): record-as-you-go recency gate — hard block on deferred bookkeeping

The advisory contract rule was not enough: flash-class agents defer hypothesis/coverage/PTT writes to closeout, then reconstruct results from conversation memory after compression — the false-positive factory. check_hypothesis_freshness now hard-blocks further target commands when the newest execution evidence is newer than the hypothesis board's last update (15 min grace for burst timing). The agent MUST record the batch result via violin_record_hypothesis before proceeding. Bursts preflight all commands before any executes, so no mid-burst false blocks. Also fixes a silent test-suite gap: tests/__init__.py and the four subpackage __init__.py files were missing, so 171 integration tests (incl. test_plugin_guard.py which pins freshness behavior) NEVER ran. Suite went from 248 passed / 7 collection errors to 419 passed, 0 errors.
2026-08-14 12:33:37 +02:00 · 2026-08-11 13:46:53 +01:00
parent 119e0c1600
commit b8262d585d
4 changed files with 147 additions and 0 deletions
@@ -42,6 +42,12 @@ __all__ = [
 # Argument / Result dataclasses
 # ---------------------------------------------------------------------------

+# Grace window for the record-as-you-go recency gate: evidence newer than the
+# hypothesis board's last update by more than this many seconds blocks further
+# target commands. 15 minutes is generous enough for burst timing/clock skew
+# while still catching run-long bookkeeping deferral.
+_RECORD_AS_YOU_GO_GRACE = 15 * 60
+

@dataclass
 class CheckCommandArgs:
@@ -508,6 +514,49 @@ def check_hypothesis_freshness(
    if stale:
        result.add_warning(f"hypothesis guard: {stale} hypothesis(es) not updated in 48h")

+    # Recency gate — record-as-you-go enforcement. If the newest execution
+    # evidence is NEWER than the hypothesis board's last update, the agent is
+    # deferring bookkeeping and will reconstruct results from conversation
+    # memory later (the false-positive factory). Block further commands until
+    # the result is recorded on the board via violin_record_hypothesis.
+    # Bursts preflight every command before any executes, so this never
+    # false-fires mid-burst; a grace window absorbs same-burst timing skew.
+    exec_dir = eng_dir / "evidence" / "executions"
+    newest_evidence = 0.0
+    if exec_dir.is_dir():
+        for path in exec_dir.iterdir():
+            if path.suffix == ".json" and not path.name.endswith((".lock", ".tmp")):
+                try:
+                    newest_evidence = max(newest_evidence, path.stat().st_mtime)
+                except OSError:
+                    continue
+    if newest_evidence:
+        for h in relevant:
+            if not h.updated:
+                continue
+            raw = h.updated.strip()
+            candidate = raw.removesuffix(" UTC").removesuffix("Z").strip()
+            updated_ts = None
+            for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%dT%H:%M", "%Y-%m-%d %H:%M:%S"):
+                try:
+                    updated_ts = datetime.strptime(candidate, fmt)
+                    break
+                except ValueError:
+                    continue
+            if updated_ts is None:
+                continue
+            updated_ts = updated_ts.replace(tzinfo=UTC)
+            # evidence mtime is naive epoch — treat as UTC for comparison
+            if newest_evidence > updated_ts.timestamp() + _RECORD_AS_YOU_GO_GRACE:
+                result.add_error(
+                    f"hypothesis H-{h.id} has not been updated since the latest execution "
+                    "evidence — record the batch result on the hypothesis board NOW via "
+                    "violin_record_hypothesis (status, Test Response, Runtime Evidence, "
+                    "Updated) before running further commands. Deferred bookkeeping forces "
+                    "memory-based reconstruction and is how false positives are born."
+                )
+                break  # one clear blocker per check is enough
+
    return result


@@ -0,0 +1,98 @@
+"""Tests for the record-as-you-go recency gate (deferred bookkeeping blocker)."""
+
+from __future__ import annotations
+
+import os
+import time
+from pathlib import Path
+
+from plugins.violin_guard import command
+from plugins.violin_guard.command import Phase
+
+_STALE_HYP = """### H-001: Queue service validation
+- **Target:** 10.129.47.140:1515
+- **Status:** Validated
+- **Phase:** EXPLOITATION
+- **CVE Research:** web_search queue 1515 CVE; NVD; no results
+- **Exploit Research:** web_search queue 1515 exploit; GitHub; no results
+- **Updated:** 2026-08-01 10:00
+"""
+
+_FRESH_HYP = """### H-001: Queue service validation
+- **Target:** 10.129.47.140:1515
+- **Status:** Validated
+- **Phase:** EXPLOITATION
+- **CVE Research:** web_search queue 1515 CVE; NVD; no results
+- **Exploit Research:** web_search queue 1515 exploit; GitHub; no results
+- **Updated:** {updated}
+"""
+
+
+def _make_engagement(tmp_path: Path, hyp_text: str, evidence_age: float) -> Path:
+    eng = tmp_path / "eng"
+    eng.mkdir(parents=True, exist_ok=True)
+    (eng / "hypotheses.md").write_text(hyp_text, encoding="utf-8")
+    exec_dir = eng / "evidence" / "executions"
+    exec_dir.mkdir(parents=True)
+    receipt = exec_dir / "2026-08-10T120000-deadbeef-exec.json"
+    receipt.write_text('{"command": "test"}', encoding="utf-8")
+    # age the evidence file: now - evidence_age seconds
+    stamp = time.time() - evidence_age
+    os.utime(receipt, (stamp, stamp))
+    return eng
+
+
+def test_recency_gate_blocks_when_board_stale(tmp_path: Path) -> None:
+    """Evidence 2h old, board updated a month ago -> block further commands."""
+    eng = _make_engagement(tmp_path, _STALE_HYP, evidence_age=2 * 3600)
+    result = command.check_hypothesis_freshness(
+        eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515"
+    )
+    assert any("not been updated since" in err for err in result.errors)
+    assert "violin_record_hypothesis" in " ".join(result.errors)
+
+
+def test_recency_gate_passes_when_board_fresh(tmp_path: Path) -> None:
+    """Board updated after the latest evidence -> gate silent."""
+    now = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime())
+    eng = _make_engagement(
+        tmp_path, _FRESH_HYP.format(updated=now), evidence_age=60
+    )
+    result = command.check_hypothesis_freshness(
+        eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515"
+    )
+    assert not any("not been updated since" in err for err in result.errors)
+
+
+def test_recency_gate_grace_window(tmp_path: Path) -> None:
+    """Evidence slightly newer than board (within grace) must not block."""
+    # board updated 5 min ago, evidence 10 min ago -> evidence is OLDER
+    now = time.time()
+    updated_str = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(now - 300))
+    eng = _make_engagement(
+        tmp_path, _FRESH_HYP.format(updated=updated_str), evidence_age=600
+    )
+    result = command.check_hypothesis_freshness(
+        eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515"
+    )
+    assert not any("not been updated since" in err for err in result.errors)
+
+
+def test_recency_gate_noop_without_evidence(tmp_path: Path) -> None:
+    """No execution evidence -> gate never fires (recon/early phases)."""
+    eng = tmp_path / "eng"
+    eng.mkdir(parents=True, exist_ok=True)
+    (eng / "hypotheses.md").write_text(_STALE_HYP, encoding="utf-8")
+    result = command.check_hypothesis_freshness(
+        eng, Phase.EXPLOITATION, "python3 exploit.py 10.129.47.140 1515"
+    )
+    assert not any("not been updated since" in err for err in result.errors)
+
+
+def test_recency_gate_recon_phases_untouched(tmp_path: Path) -> None:
+    """Recon does not require hypotheses at all — gate must stay silent."""
+    eng = _make_engagement(tmp_path, _STALE_HYP, evidence_age=2 * 3600)
+    result = command.check_hypothesis_freshness(
+        eng, Phase.RECON, "nmap -p- 10.129.47.140"
+    )
+    assert not result.errors