feat(issue-5): shadow watcher that logs would-complete and cannot write

policy.decide is a pure function so the phase-0 scenario fixtures can be fed to
it verbatim. Those seven decisions survived three-reviewer scrutiny, so they are
the contract rather than a fresh set of rules, and the test asserts against
expected_policy.decision directly.

Order inside the policy matters. Transport doubt and staleness invalidate the
evidence itself, so they are answered before anything is inferred from it; a
blocked agent is waiting for a human and can never be completed; attribution
including the execution anchor comes before quiescence, because idle is not
evidence of finishing when a session is reused; and a report must be finalized,
never merely paused, since a debounce is a candidate filter and not an authority.

The shadow runner is never handed the engine, so it has no route to a write.
That is the structural half of P2A-03. The measured half hashes every file under
the registry directory around a full run and asserts it is unchanged, which is a
stronger claim than checking a status field.

Adds the five scenarios P2A-04 requires: a raced double dispatch, a reused
attempt with no anchor, a mid-write report, an unknown status, and a pane
restarted mid-flight. They pass phase 0's own fixture validator, which is what
keeps them honest. The phase-0 test that pinned exactly seven scenarios now
requires the seven to still exist rather than forbidding growth.

Also pins the two phase 1 handoffs as dormant: nothing here spawns a daemon, so
preexec_fn stays unreachable, and nothing calls a frozen shell helper.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd

tests/test_issue5_phase2a_loop.py

@@ -537,3 +537,161 @@ def test_attribution_pane_generation_prefix_is_not_a_substring_match():
     assert attribution.pane_matches_generation("w2X:p1", "w2X:p1:1234:t")
     assert not attribution.pane_matches_generation("w2X:p1", "w2X:p10:1234:t")
     assert not attribution.pane_matches_generation("", "w2X:p1:1:t")
+
+
+# ==============================================================================
+# 6. policy — checked against the phase-0 scenario contracts, not fresh rules
+# ==============================================================================
+
+def _scenarios():
+    out = {}
+    for path in sorted(WATCHER_FIXTURES.glob("*.json")):
+        if path.name == "manifest.json":
+            continue
+        data = json.loads(path.read_text(encoding="utf-8"))
+        out[data["scenario_id"]] = data
+    return out
+
+
+@pytest.mark.parametrize("scenario_id", [
+    # the phase-0 seven ...
+    "initial_idle", "tool_running", "report_pause_15s", "finalized_candidate",
+    "blocked_permission", "transport_disconnected", "restarted_generation",
+    # ... and the five P2A-04 requires, held to the same contract
+    "race_double_dispatch", "attempt_reuse", "partial_report",
+    "unknown_status", "restart_midflight",
+])
+def test_g7_policy_matches_the_phase0_expected_decision(scenario_id):
+    """G7: the seven scenarios that survived three-reviewer scrutiny are the
+    contract. The policy must reproduce their decisions exactly."""
+    from lib_py.loop import policy
+    scenario = _scenarios()[scenario_id]
+    expected = scenario["expected_policy"]["decision"]
+    got = policy.decide(scenario)
+    assert got.decision == expected, (
+        "{}: expected {!r}, got {!r} ({})".format(scenario_id, expected, got.decision, got.reason)
+    )
+
+
+def test_policy_is_pure_no_io_no_clock(monkeypatch):
+    """Purity is what makes the fixtures usable as vectors; guard it."""
+    import socket
+    from lib_py.loop import policy
+
+    scenarios = list(_scenarios().values())   # load before the guards go up
+
+    def boom(*_a, **_k):
+        raise AssertionError("policy.decide must not perform I/O")
+
+    monkeypatch.setattr(subprocess, "run", boom)
+    monkeypatch.setattr(socket, "socket", boom)
+    monkeypatch.setattr(Path, "read_text", boom)
+    monkeypatch.setattr(time, "time", boom)
+    for scenario in scenarios:
+        policy.decide(scenario)
+
+
+def test_policy_debounce_alone_never_authorises_completion():
+    """A stable mtime is a candidate filter, not an authority to complete."""
+    from lib_py.loop import policy
+    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
+    scenario["filesystem"]["manifest_committed"] = False
+    assert policy.decide(scenario).decision == "hold"
+    scenario["filesystem"]["manifest_committed"] = True
+    scenario["filesystem"]["hash_verified"] = False
+    assert policy.decide(scenario).decision == "hold"
+
+
+def test_policy_holds_when_the_execution_anchor_is_missing():
+    """Idle is not evidence of finishing; a reused session looks identical."""
+    from lib_py.loop import policy
+    scenario = json.loads((WATCHER_FIXTURES / "finalized_candidate.json").read_text(encoding="utf-8"))
+    scenario["dispatch_context"]["execution_anchor_observed"] = False
+    decision = policy.decide(scenario)
+    assert decision.decision == "hold" and "anchor" in decision.reason
+
+
+# ==============================================================================
+# 7. shadow — cannot write, proven structurally and by hashing the registry
+# ==============================================================================
+
+def test_g4_shadow_run_leaves_the_registry_byte_identical(mam_sandbox, tmp_path, monkeypatch):
+    """G4: stronger than 'status was not changed' -- nothing under the registry
+    directory changes at all."""
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.loop import shadow
+
+    reg = Path(_reg(tmp_path))
+    job_ids = [_make_job(str(reg)) for _ in range(3)]
+    from lib_py.delegation import engine
+    engine.transition(job_ids[0], "claim", registry_dir=str(reg))
+    before = _dir_fingerprint(reg)
+    assert before, "fixture precondition: the registry has content to protect"
+
+    runner = shadow.ShadowRunner(workspace=str(mam_sandbox))
+    results = runner.evaluate_all(list(_scenarios().values()))
+
+    assert _dir_fingerprint(reg) == before, "shadow mode wrote to the registry"
+    assert any(r["would_complete"] for r in results), "the approved scenario must be reported"
+    assert all(r["shadow"] is True for r in results)
+
+
+def test_g4_shadow_runner_holds_no_writer_at_all():
+    """Structural, not promised: the runner has no route to a write."""
+    import ast
+
+    src = (SKILLS / "lib_py" / "loop" / "shadow.py").read_text(encoding="utf-8")
+    tree = ast.parse(src)
+    imported = set()
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and node.module:
+            for alias in node.names:
+                imported.add("{}.{}".format(node.module, alias.name))
+        elif isinstance(node, ast.Import):
+            for alias in node.names:
+                imported.add(alias.name)
+    assert not any("engine" in name for name in imported), imported
+    for forbidden in ("update_job_status", "_atomic_write_record", "transition("):
+        assert forbidden not in src, "shadow must not reference {}".format(forbidden)
+
+
+def test_shadow_records_its_would_complete_decisions(mam_sandbox, monkeypatch):
+    monkeypatch.chdir(mam_sandbox)
+    from lib_py.loop import shadow, store
+
+    scenario = _scenarios()["finalized_candidate"]
+    runner = shadow.ShadowRunner(workspace=str(mam_sandbox))
+    entry = runner.evaluate(scenario)
+    assert entry["would_complete"] is True
+    logged = store.read_decisions(entry["job_id"], workspace=str(mam_sandbox))
+    assert logged and logged[-1]["decision"] == "candidate_approved"
+
+
+def test_g9_loop_package_imports_only_delegation_among_siblings():
+    """G9: the rule that caught the engine's original placement."""
+    import ast
+
+    offenders = []
+    for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")):
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        for node in ast.walk(tree):
+            names = []
+            if isinstance(node, ast.Import):
+                names = [a.name for a in node.names]
+            elif isinstance(node, ast.ImportFrom) and node.module:
+                names = [node.module]
+            for name in names:
+                if name.startswith("lib_py.") and \
+                        not name.startswith("lib_py.loop") and \
+                        not name.startswith("lib_py.delegation"):
+                    offenders.append("{}:{} {}".format(path.name, node.lineno, name))
+    assert not offenders, offenders
+
+
+def test_g9_phase2a_spawns_no_daemon_and_calls_no_frozen_helper():
+    """The two phase 1 handoffs stay dormant: preexec_fn is unreachable because
+    nothing here spawns, and no frozen lib.sh helper is invoked."""
+    for path in sorted((SKILLS / "lib_py" / "loop").glob("*.py")):
+        src = path.read_text(encoding="utf-8")
+        for forbidden in ("ensure_subscriber_daemon", "preexec_fn", "lib.sh"):
+            assert forbidden not in src, "{} references {}".format(path.name, forbidden)