test(issue-5): harden watcher evidence validation and add codex pass review report

- Enforce evidence-root confinement for capture_ref and 1:1 capture citation uniqueness
- Add pane identity and backend identity cross-validation
- Add bidirectional agreement checks between watcher/herdr manifests and scenario fixtures
- Expand contract test suite to 41 tests (100% pass)
- Document evidence confinement rules in herdr-runtime.md and update task.md
- Archive Codex 4th review report (Job 1bd561e6: [VERDICT: PASS])

tests/test_issue5_phase0_contracts.py

@@ -13,6 +13,11 @@ Validates:
    accepted as 'observed' when an evidence schema exists for that source kind (currently Herdr only).
    Event / filesystem / harness items have no independent evidence schema yet and are fail-closed when
    labeled 'observed'; they must remain 'simulated' until such schemas exist.
+8. Evidence identity (Rev.5 self-review hardening): observed capture_refs must be repo-relative paths confined
+   to the native Herdr evidence root; one capture may substantiate at most one timeline observation; the
+   capture's session UUID and pane id must match the scenario's 4-fold attribution; a backend-specific
+   scenario may only cite captures of that backend; and both fixture manifests must agree with the files
+   they index (no drift, no orphan captures).
 """
 
 import copy
@@ -48,6 +53,12 @@ OBSERVED_EVIDENCE_SCHEMAS = {"herdr"}
 # against the native capture cannot be skipped.
 HERDR_MANDATORY_ITEM_FIELDS = ("agent_status",)
 
+# Native agent_status values a Herdr manifest slot may be named after (slot name == recorded agent_status).
+HERDR_AGENT_STATUS_SLOTS = {"idle", "working", "blocked", "done", "unknown"}
+
+# Scenario backend marker meaning "any backend's capture may be cited".
+BACKEND_AGNOSTIC = "backend_agnostic"
+
 
 # ===========================================================================
 # Validation Engine Functions (Used for both real fixtures & negative tests)
@@ -103,10 +114,24 @@ def _load_observed_capture_ref(index: int, item: dict) -> "tuple[Path, dict]":
     assert "capture_ref" in item and item["capture_ref"], (
         f"Timeline item [{index}] is marked observed but lacks non-empty 'capture_ref'"
     )
-    ref_path = REPO_ROOT / item["capture_ref"]
+    raw_ref = str(item["capture_ref"])
+    assert not Path(raw_ref).is_absolute(), (
+        f"Observed timeline item [{index}] capture_ref must be a repo-relative path, got absolute '{raw_ref}'"
+    )
+    ref_path = REPO_ROOT / raw_ref
     assert ref_path.exists(), f"Observed timeline item [{index}] points to nonexistent capture_ref: {ref_path}"
     assert ref_path.suffix == ".json", f"capture_ref must point to a JSON evidence record, got {ref_path}"
 
+    # Confinement: evidence may only come from the native Herdr evidence root (no traversal, no foreign files).
+    resolved = ref_path.resolve()
+    try:
+        resolved.relative_to(HERDR_ROOT.resolve())
+    except ValueError:
+        raise AssertionError(
+            f"Observed timeline item [{index}] capture_ref '{raw_ref}' resolves outside the native Herdr "
+            f"evidence root {HERDR_ROOT}; foreign or traversed paths are not admissible evidence"
+        )
+
     try:
         ref_data = json.loads(ref_path.read_text(encoding="utf-8"))
     except Exception as exc:
@@ -133,12 +158,12 @@ def _extract_native_agent_object(ref_data: dict):
     return agent
 
 
-def _validate_observed_herdr_item(index: int, item: dict, dctx: dict) -> None:
-    """Strict evidence correspondence for an observed Herdr timeline item.
+def _validate_observed_herdr_item(index: int, item: dict, dctx: dict, scenario_backend: str = None) -> str:
+    """Strict evidence correspondence for an observed Herdr timeline item. Returns the capture_id.
 
     Every declared payload field must be a native field of the referenced capture's agent object
     and must match its value exactly. Mandatory fields cannot be omitted to dodge the comparison,
-    and the capture's agent session identity must match the scenario's dispatch attribution.
+    and the capture's identity (session UUID, pane id, backend) must match the scenario's attribution.
     """
     ref_path, ref_data = _load_observed_capture_ref(index, item)
 
@@ -164,7 +189,7 @@ def _validate_observed_herdr_item(index: int, item: dict, dctx: dict) -> None:
         f"it cannot substantiate observed herdr timeline item [{index}]"
     )
 
-    # Identity correspondence: the capture must be about the session this scenario attributes.
+    # Identity correspondence (session): the capture must be about the session this scenario attributes.
     session_obj = agent_obj.get("agent_session")
     capture_uuid = session_obj.get("value") if isinstance(session_obj, dict) else None
     assert capture_uuid == dctx["agent_session_uuid"], (
@@ -172,6 +197,25 @@ def _validate_observed_herdr_item(index: int, item: dict, dctx: dict) -> None:
         f"'{dctx['agent_session_uuid']}' but capture {ref_path} records agent_session.value='{capture_uuid}'"
     )
 
+    # Identity correspondence (pane): pane_process_generation is '<pane_id>:<pid>:<start>'; the capture's
+    # native pane_id must be its leading component, otherwise the capture is about a different pane/generation.
+    capture_pane = agent_obj.get("pane_id")
+    assert isinstance(capture_pane, str) and capture_pane, (
+        f"capture {ref_path} lacks a native pane_id; pane identity for timeline item [{index}] cannot be established"
+    )
+    ppg = str(dctx["pane_process_generation"])
+    assert ppg.startswith(capture_pane + ":"), (
+        f"Identity correspondence mismatch: dispatch_context.pane_process_generation='{ppg}' does not belong to "
+        f"pane_id='{capture_pane}' recorded by capture {ref_path}"
+    )
+
+    # Backend correspondence: a backend-specific scenario may only cite captures of that backend.
+    if scenario_backend and scenario_backend != BACKEND_AGNOSTIC:
+        assert ref_data.get("backend") == scenario_backend, (
+            f"Backend correspondence mismatch: scenario declares backend='{scenario_backend}' but capture "
+            f"{ref_path} records backend='{ref_data.get('backend')}'"
+        )
+
     # Payload correspondence: every declared field must exist natively and match exactly.
     for key, declared in item_payload.items():
         assert key in agent_obj, (
@@ -183,6 +227,55 @@ def _validate_observed_herdr_item(index: int, item: dict, dctx: dict) -> None:
             f"but capture {ref_path} records {key}='{agent_obj[key]}'"
         )
 
+    return str(ref_data["capture_id"])
+
+
+def validate_watcher_manifest(manifest: dict) -> None:
+    """The watcher manifest must not drift from the scenario files it indexes."""
+    for sid, sinfo in manifest["scenarios"].items():
+        data = json.loads((WATCHER_ROOT / sinfo["file"]).read_text(encoding="utf-8"))
+        assert data["scenario_id"] == sid, f"{sinfo['file']} scenario_id '{data['scenario_id']}' != manifest key '{sid}'"
+        assert data["provenance"] == sinfo["provenance"], (
+            f"{sid}: manifest provenance '{sinfo['provenance']}' != file provenance '{data['provenance']}'"
+        )
+        assert data["expected_policy"]["decision"] == sinfo["expected_decision"], (
+            f"{sid}: manifest expected_decision '{sinfo['expected_decision']}' != "
+            f"file decision '{data['expected_policy']['decision']}'"
+        )
+
+
+def validate_herdr_manifest(manifest: dict) -> None:
+    """Every observed slot must be backed by a capture of the same backend recording the slot's state,
+    and every capture on disk must be indexed by exactly one observed slot (no drift, no orphans)."""
+    referenced = {}
+    for backend, bdata in manifest["backends"].items():
+        for slot_name, slot in bdata["slots"].items():
+            if slot["status"] != "observed":
+                continue
+            rel = slot["file"]
+            assert rel not in referenced, f"capture {rel} indexed by two slots: {referenced[rel]} and {backend}/{slot_name}"
+            referenced[rel] = f"{backend}/{slot_name}"
+            cap_path = HERDR_ROOT / rel
+            assert cap_path.exists(), f"Observed slot {backend}/{slot_name} references missing capture {rel}"
+            cap = json.loads(cap_path.read_text(encoding="utf-8"))
+            assert cap["backend"] == backend, f"{rel}: capture backend '{cap['backend']}' != manifest backend '{backend}'"
+            agent_obj = _extract_native_agent_object(cap)
+            if slot_name in HERDR_AGENT_STATUS_SLOTS:
+                assert agent_obj is not None, f"{rel}: status slot '{slot_name}' requires an agent object"
+                assert agent_obj.get("agent_status") == slot_name, (
+                    f"{rel}: manifest slot '{slot_name}' but capture records agent_status='{agent_obj.get('agent_status')}'"
+                )
+            elif slot_name == "wait_timeout":
+                assert cap["exit_code"] != 0 and agent_obj is None, f"{rel}: wait_timeout slot must be a nonzero-exit capture"
+                err = json.loads(cap["stderr"]).get("error", {})
+                assert err.get("code") == "timeout", f"{rel}: wait_timeout slot must record error.code='timeout'"
+            else:
+                raise AssertionError(f"Unknown observed slot kind '{slot_name}' for backend '{backend}'")
+
+    on_disk = {str(pth.relative_to(HERDR_ROOT)) for pth in HERDR_ROOT.glob("*/*.json")}
+    orphans = on_disk - set(referenced)
+    assert not orphans, f"Captures on disk not indexed by any observed manifest slot: {sorted(orphans)}"
+
 
 def validate_watcher_scenario(data: dict, file_path: Path = None) -> None:
     """Strictly validates a Watcher composite scenario, attribution, timeline authenticity, and correspondence."""
@@ -222,6 +315,8 @@ def validate_watcher_scenario(data: dict, file_path: Path = None) -> None:
     simulated_count = 0
     total_items = len(data["timeline"])
     assert total_items > 0, "timeline must contain at least one item"
+    scenario_backend = data.get("backend")
+    cited_capture_ids: dict = {}  # capture_id -> first timeline index that cited it
 
     for i, item in enumerate(data["timeline"]):
         source_kind = item.get("source_kind")
@@ -247,10 +342,17 @@ def validate_watcher_scenario(data: dict, file_path: Path = None) -> None:
         )
 
         if source_kind == "herdr":
-            _validate_observed_herdr_item(i, item, dctx)
+            capture_id = _validate_observed_herdr_item(i, item, dctx, scenario_backend)
         else:  # pragma: no cover - guarded by the assertion above
             raise AssertionError(f"Unhandled observed source_kind '{source_kind}' at timeline item [{i}]")
 
+        # One capture is one point-in-time observation; it cannot substantiate two timeline points.
+        assert capture_id not in cited_capture_ids, (
+            f"Timeline item [{i}] re-cites capture_id '{capture_id}' already used by item "
+            f"[{cited_capture_ids[capture_id]}]; a single capture cannot substantiate multiple observations"
+        )
+        cited_capture_ids[capture_id] = i
+
     # Provenance honesty invariant
     if observed_count > 0 and simulated_count > 0:
         assert overall_prov == "mixed", (
@@ -636,11 +738,11 @@ def test_negative_observed_herdr_item_session_identity_mismatch_fails():
 
 
 def test_negative_observed_herdr_item_referencing_non_native_json_record_fails():
-    """A JSON file that is not a native Herdr capture (e.g. a Watcher scenario) is not Herdr evidence."""
+    """A JSON file inside the evidence root that is not a native capture (e.g. the manifest) is not evidence."""
     sc = _load_scenario("initial_idle.json")
     sc["provenance"] = "mixed"
     sc["timeline"][1]["source_provenance"] = "observed"
-    sc["timeline"][1]["capture_ref"] = "tests/fixtures/issue_5/watcher/finalized_candidate.json"
+    sc["timeline"][1]["capture_ref"] = "tests/fixtures/issue_5/herdr/manifest.json"
     sc["timeline"][1]["payload"] = {"agent_status": "idle"}
     with pytest.raises(AssertionError, match="must have provenance 'observed'"):
         validate_watcher_scenario(sc)
@@ -665,3 +767,124 @@ def test_positive_observed_herdr_item_with_full_native_correspondence_passes():
     sc["provenance"] = "observed_composite"
     with pytest.raises(AssertionError, match="overall provenance must be 'mixed'"):
         validate_watcher_scenario(sc)
+
+
+# ===========================================================================
+# 6. Evidence Identity & Manifest Drift Tests (Rev.5 self-review hardening)
+# ===========================================================================
+
+def _observed_herdr_scenario(payload: dict, capture_ref: str = HERDR_CLAUDE_IDLE) -> dict:
+    """initial_idle with its herdr item (index 1) relabeled observed against `capture_ref`."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "mixed"
+    sc["timeline"][1]["source_provenance"] = "observed"
+    sc["timeline"][1]["capture_ref"] = capture_ref
+    sc["timeline"][1]["payload"] = payload
+    return sc
+
+
+@pytest.mark.parametrize("bad_ref, expected", [
+    # Escapes the evidence root via traversal even though the target is a well-formed JSON file.
+    ("tests/fixtures/issue_5/herdr/../watcher/finalized_candidate.json", "resolves outside the native Herdr evidence root"),
+    # Well-formed JSON in the repo but not under the evidence root.
+    ("tests/fixtures/issue_5/watcher/finalized_candidate.json", "resolves outside the native Herdr evidence root"),
+    # Absolute paths are never admissible, even when they point at a real capture.
+    (str(HERDR_ROOT / "claude" / "idle.json"), "must be a repo-relative path"),
+])
+def test_negative_observed_capture_ref_outside_evidence_root_fails(bad_ref, expected):
+    sc = _observed_herdr_scenario({"agent_status": "idle"}, capture_ref=bad_ref)
+    with pytest.raises(AssertionError, match=expected):
+        validate_watcher_scenario(sc)
+
+
+def test_negative_single_capture_cited_by_two_observed_items_fails():
+    """One point-in-time capture cannot substantiate two distinct timeline observations."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "observed_composite"
+    # Turn the event item into a second herdr observation citing the very same capture.
+    for item in sc["timeline"]:
+        item["source_kind"] = "herdr"
+        item["source_provenance"] = "observed"
+        item["capture_ref"] = HERDR_CLAUDE_IDLE
+        item["payload"] = {"agent_status": "idle"}
+    with pytest.raises(AssertionError, match="re-cites capture_id 'cap-claude-idle-01'"):
+        validate_watcher_scenario(sc)
+
+
+def test_negative_observed_herdr_item_pane_identity_mismatch_fails():
+    """Session UUID may match while the pane generation belongs to a different pane; must be rejected."""
+    sc = _observed_herdr_scenario({"agent_status": "idle"})
+    sc["dispatch_context"]["pane_process_generation"] = "w2X:p9:1234:2026-09-06T00:00:00Z"  # capture says w2X:p1
+    with pytest.raises(AssertionError, match="does not belong to pane_id='w2X:p1'"):
+        validate_watcher_scenario(sc)
+
+    # A pane id that merely shares a prefix (w2X:p1 vs w2X:p10) must not be accepted either.
+    sc["dispatch_context"]["pane_process_generation"] = "w2X:p10:1234:2026-09-06T00:00:00Z"
+    with pytest.raises(AssertionError, match="does not belong to pane_id='w2X:p1'"):
+        validate_watcher_scenario(sc)
+
+
+def test_negative_backend_specific_scenario_citing_other_backend_capture_fails():
+    """A scenario pinned to one backend may not borrow another backend's capture as evidence."""
+    sc = _observed_herdr_scenario({"agent_status": "idle"})
+    sc["backend"] = "grok"  # capture is claude
+    with pytest.raises(AssertionError, match="Backend correspondence mismatch"):
+        validate_watcher_scenario(sc)
+
+    sc["backend"] = "claude"
+    validate_watcher_scenario(sc)  # matching backend is accepted
+
+    sc["backend"] = BACKEND_AGNOSTIC
+    validate_watcher_scenario(sc)  # agnostic scenarios may cite any backend
+
+
+def test_watcher_manifest_agrees_with_scenario_files():
+    validate_watcher_manifest(json.loads((WATCHER_ROOT / "manifest.json").read_text(encoding="utf-8")))
+
+
+def test_negative_watcher_manifest_drift_is_detected():
+    manifest = json.loads((WATCHER_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    manifest["scenarios"]["initial_idle"]["expected_decision"] = "candidate_approved"  # file says hold
+    with pytest.raises(AssertionError, match="manifest expected_decision 'candidate_approved' != file decision 'hold'"):
+        validate_watcher_manifest(manifest)
+
+    manifest = json.loads((WATCHER_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    manifest["scenarios"]["finalized_candidate"]["provenance"] = "observed_composite"  # file says simulated
+    with pytest.raises(AssertionError, match="manifest provenance 'observed_composite' != file provenance 'simulated'"):
+        validate_watcher_manifest(manifest)
+
+
+def test_herdr_manifest_observed_slots_agree_with_captures_and_no_orphan_captures():
+    validate_herdr_manifest(json.loads((HERDR_ROOT / "manifest.json").read_text(encoding="utf-8")))
+
+
+def test_negative_herdr_manifest_slot_status_drift_is_detected():
+    """Re-indexing claude/idle.json (records idle) under the 'done' slot must be rejected."""
+    manifest = json.loads((HERDR_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    slots = manifest["backends"]["claude"]["slots"]
+    slots["done"] = {"status": "observed", "file": "claude/idle.json"}
+    slots["idle"] = {"status": "unobserved", "reason": "mutated"}
+    with pytest.raises(AssertionError, match="manifest slot 'done' but capture records agent_status='idle'"):
+        validate_herdr_manifest(manifest)
+
+
+def test_negative_herdr_manifest_backend_or_orphan_drift_is_detected():
+    manifest = json.loads((HERDR_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    # Borrow another backend's capture: grok's idle slot pointed at claude's capture (claude's own slot
+    # de-indexed so the duplicate-index guard does not fire first and the backend check is what rejects).
+    manifest["backends"]["claude"]["slots"]["idle"] = {"status": "unobserved", "reason": "mutated"}
+    manifest["backends"]["grok"]["slots"]["idle"]["file"] = "claude/idle.json"
+    with pytest.raises(AssertionError, match="capture backend 'claude' != manifest backend 'grok'"):
+        validate_herdr_manifest(manifest)
+
+    manifest = json.loads((HERDR_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    # Same capture indexed by two slots must be rejected.
+    manifest["backends"]["grok"]["slots"]["idle"]["file"] = "claude/idle.json"
+    with pytest.raises(AssertionError, match="indexed by two slots"):
+        validate_herdr_manifest(manifest)
+
+    manifest = json.loads((HERDR_ROOT / "manifest.json").read_text(encoding="utf-8"))
+    # De-index a real capture: it becomes an orphan on disk and must be reported.
+    manifest["backends"]["codex"]["slots"]["done"] = {"status": "unobserved", "reason": "mutated"}
+    with pytest.raises(AssertionError, match=r"not indexed by any observed manifest slot: \['codex/done.json'\]"):
+        validate_herdr_manifest(manifest)