test(issue-5): implement fail-closed evidence dispatch and fix VALID_STATUSES citation

- Fix Codex [P1]: add SUPPORTED_SOURCE_KINDS dispatch and enforce fail-closed rejection for non-herdr observed items without independent schema
- Add strict correspondence checks for native Herdr observed items (agent_status, native schema, session UUID)
- Add negative test reproducing Codex in-memory mutation of finalized_candidate
- Fix Claude Finding 2: update VALID_STATUSES citation in baseline.md to registry.py:38
- Update Phase 0 collected test counts to 499 in contract docs

docs/contracts/issue-5/baseline.md

@@ -4,7 +4,7 @@
 - **Issue**: [Gitea Issue #5 — `[Refactor] lib.sh 작업 위임 로직 및 run_loop.sh 오케스트레이터의 Python 모듈화 (lib_py 승격)`](https://git.godopu.com/tmpl/multi-agent-mux/issues/5)
 - **Baseline Commit**: `47640e2fd7b108be29446a9384a343727e87edf8`
 - **Environment Baseline**: Python 3.9.6, Herdr 0.8.2, macOS Darwin 24.3.0
-- **Test Suite Baseline**: 469 tests collected initially; 484 tests collected with Phase 0 contract suite (+15 contract & mutation tests)
+- **Test Suite Baseline**: 469 tests collected initially; 499 tests collected with Phase 0 contract suite (+30 contract & mutation tests)
 - **Status**: Phase 0 Baseline Contract (Observed vs. Proposed vs. Gap)
 
 ---
@@ -90,7 +90,7 @@ stateDiagram-v2
 | `cancelled` | `cancelled` | `true` | 세션 중단 등에 의한 취소. Terminal 상태 | `[Observed]` |
 
 > ⚠️ **이벤트-상태 간극 Gap (`[Gap]`)**:
-> 1. `progress`와 `permission_required`는 이벤트로 발행되지만 레지스트리 `status` enum(`VALID_STATUSES = ('pending', 'running', 'completed', 'error', 'cancelled')`)에 포함되지 않아, 레지스트리 파일만으로는 권한 승인 대기 여부를 직접 알 수 없습니다 (`mqtt_common.py:44-48`).
+> 1. `progress`와 `permission_required`는 이벤트로 발행되지만 레지스트리 `status` enum(`VALID_STATUSES = ('pending', 'running', 'completed', 'error', 'cancelled')`)에 포함되지 않아, 레지스트리 파일만으로는 권한 승인 대기 여부를 직접 알 수 없습니다 (`registry.py:38`).
 > 2. `publish_event.py`의 Terminal 이벤트는 `("completed", "error", "cancelled")`이지만, `job_subscriber.py:46`의 Terminal 이벤트는 `("completed", "error")`로 정의되어 있습니다. 가입자가 디스크 폴백(`_check_disk_fallback`)을 수행할 때 비로소 `cancelled` 상태를 감지하여 가상 `error` 이벤트(`synth_event="error"`)로 변환하여 처리합니다 (`job_subscriber.py:80`).
 > 3. **Terminal Non-reentry (`[Proposed]` vs `[Gap]`)**: 한 번 Terminal 상태에 도달한 후 중복/상충 이벤트에 의해 상태가 역행하지 않는 불변식은 Phase 2a의 목표 계약(`[Proposed]`)이며, 현행 `update_job_status`는 락 내에서 단순 필드 덮어쓰기(`record.update(fields)`)를 수행하므로 완벽한 비역행이 보장되지 않는 간극(`[Gap]`)이 있습니다.
 

docs/contracts/issue-5/implementation_plan.md

@@ -21,7 +21,7 @@
 
 | 단계 | 작업 목표 | 세부 구현 내용 | 필수 완료 게이트 (Gate Criteria) | 롤백 절차 (Rollback Path) |
 |---|---|---|---|---|
-| **Phase 0**<br>*(현재)* | **사전 계약 고정 및 Fixture 수집** | • CLI, 이벤트, 리포트, 락 계약 기준선 문서화<br>• Herdr 0.8.2 네이티브 캡처 및 복합 Watcher 시나리오 2계층 분리<br>• 6개 백엔드 실측 증거 행렬 수집 | • 484개 전체 테스트 수집 및 회귀 0-Fail (100% 통과)<br>• Fixture 스키마 및 증거 경계 검증 통과<br>• 프로덕션 코드 무변경 diff 검증 | 미추적 산출물 즉시 삭제 (`rm -rf docs/contracts/issue-5 tests/fixtures/issue_5 tests/test_issue5_phase0_contracts.py`) |
+| **Phase 0**<br>*(현재)* | **사전 계약 고정 및 Fixture 수집** | • CLI, 이벤트, 리포트, 락 계약 기준선 문서화<br>• Herdr 0.8.2 네이티브 캡처 및 복합 Watcher 시나리오 2계층 분리<br>• 6개 백엔드 실측 증거 행렬 수집 | • 499개 전체 테스트 수집 및 회귀 0-Fail (100% 통과)<br>• Fixture 스키마 및 증거 경계 검증 통과<br>• 프로덕션 코드 무변경 diff 검증 | 미추적 산출물 즉시 삭제 (`rm -rf docs/contracts/issue-5 tests/fixtures/issue_5 tests/test_issue5_phase0_contracts.py`) |
 | **Phase 1** | **`lib_py/delegation/` 추출** | • `lib.sh` L1833 이하 위임 헬퍼를 파이썬 서비스 클래스로 추출<br>• `start_watchdog`을 내부 명칭 `ensure_subscriber_daemon`으로 명명(기존 함수 별칭 보존)<br>• 장수 프로세스 단일 MQTT 연결 재사용 프로토타입 | • CLI/non-fatal/HMAC/seq/디스크 폴백 회귀 100% 통과<br>• 데몬 수명 및 PID 계약 일치 확인<br>• **자동완료 정책 변경 없음** | `lib.sh` 위임 헬퍼 호출을 이전 인라인/셸 방식으로 즉시 복원 |
 | **Phase 2a** | **`lib_py/loop/` 코어 & Shadow Watcher** | • `engine.py` (단일 전이 주체), `watcher.py` (`herdr agent wait` 래핑) 구현<br>• 모든 Writer에 대한 단일 CAS 트랜잭션 경계 구현<br>• 자동완료를 수행하지 않고 "Would-Complete" 로깅만 수행하는 Shadow 모드 가동 | • 락 내 비동기/I/O 배제 검증<br>• 소켓 단절/스냅샷 재연결 백오프 검증<br>• Shadow 모드가 레지스트리 상태를 쓰지 않음을 검증 | 1) Shadow 모니터 프로세스 종료 및 인메모리 큐 폐기<br>2) CAS 서비스는 하위 호환 모드로 유지하거나 `MAM_ENGINE_CAS=0` 플래그로 비활성화하여 레거시 무조건 필드 병합(`update_job_status`)으로 복귀<br>3) 기존 셸 루프 상태 및 레코드 불변 유지 |
 | **Phase 2b** | **이중 감지 자동완료 활성화 (Flag-Gated)** | • 4중 실행 귀속 + 리포트 확정 계약 + 단일 CAS 전이 게이트 연동<br>• Pending Notification 지연 알림 및 크래시 복구 구현<br>• 검증된 백엔드/작업에 한해 opt-in 플래그로 활성화 (기본값 OFF) | • 부수효과 작업 예외 및 무판정 fail-closed 테스트 완료<br>• 경쟁/재시작 시나리오 통과<br>• 플래그 OFF 시 기존 explicit event 경로 100% 보존 | 환경변수/플래그 `MAM_AUTO_COMPLETE=0` 설정; 이미 확정된 terminal 레코드는 보존하고 신규 자동완료만 중단 |

docs/contracts/issue-5/task.md

@@ -9,7 +9,7 @@
 ## Phase 0: 사전 계약 기준선 및 Fixture 수집 (이번 실행 범위)
 
 - [x] **P0-01** 기준 commit, dirty diff, Herdr/Python 버전, pytest collection 수 기록.
-  - *증거*: Commit `47640e2`, Herdr `0.8.2`, Python `3.9.6`, Pytest `469 tests collected initially` ➔ `484 tests collected with Phase 0 suite` (+15 contract & mutation tests).
+  - *증거*: Commit `47640e2`, Herdr `0.8.2`, Python `3.9.6`, Pytest `469 tests collected initially` ➔ `499 tests collected with Phase 0 suite` (+30 contract & mutation tests).
 - [x] **P0-02** `docs/contracts/issue-5/baseline.md` 작성: 공개 CLI/default/invalid input/stdout-stderr/exit/interpreter/path/signal 계약과 소스·기존 테스트 매핑.
   - *증거*: `docs/contracts/issue-5/baseline.md` §2 (`--all-reviewer` 우선순위 및 L192–196 소스 매핑 반영).
 - [x] **P0-03** event schema/HMAC/seq/retain/audit/디스크 fallback 및 publisher 실패 후 로컬 commit 동작 기록.

tests/test_issue5_phase0_contracts.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Phase 0 contract and fixture verification tests for Issue #5 (Rev.4).
+"""Phase 0 contract and fixture verification tests for Issue #5 (Rev.5).
 
 Validates:
 1. Herdr Layer 1 native contract captures (pure native observation, no injected MAM fields).
@@ -9,6 +9,10 @@ Validates:
 5. Contract cross-checks against baseline.md and herdr-runtime.md.
 6. Mutation and negative tests demonstrating that corruption, unrelated references, payload mismatches,
    and dishonest provenance labels are rejected.
+7. Source-kind dispatch: every timeline item must declare a supported source kind, and an item may only be
+   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.
 """
 
 import copy
@@ -32,6 +36,18 @@ FORBIDDEN_MAM_KEYS = {
     "manifest_hash",
 }
 
+# Explicit set of source kinds a Watcher timeline item may declare. Anything else is rejected.
+SUPPORTED_SOURCE_KINDS = {"herdr", "event", "filesystem", "harness"}
+
+# Source kinds for which an independent, machine-checkable evidence schema is implemented.
+# Only these may carry source_provenance == "observed". All other supported kinds are
+# fail-closed for observed claims until their evidence schema exists.
+OBSERVED_EVIDENCE_SCHEMAS = {"herdr"}
+
+# For observed Herdr items, these payload fields MUST be declared so that the comparison
+# against the native capture cannot be skipped.
+HERDR_MANDATORY_ITEM_FIELDS = ("agent_status",)
+
 
 # ===========================================================================
 # Validation Engine Functions (Used for both real fixtures & negative tests)
@@ -82,6 +98,92 @@ def validate_native_capture(cap: dict, file_path: Path = None) -> None:
     _check_no_forbidden(payload_to_check)
 
 
+def _load_observed_capture_ref(index: int, item: dict) -> "tuple[Path, dict]":
+    """Common envelope checks for any observed timeline item's capture_ref. Returns (path, record)."""
+    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"]
+    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}"
+
+    try:
+        ref_data = json.loads(ref_path.read_text(encoding="utf-8"))
+    except Exception as exc:
+        raise AssertionError(f"capture_ref {ref_path} must be valid JSON: {exc}")
+
+    assert isinstance(ref_data, dict), f"capture_ref {ref_path} must be a JSON object evidence record"
+    assert ref_data.get("schema_version") == "1.0.0", f"capture_ref {ref_path} missing schema_version 1.0.0"
+    assert ref_data.get("provenance") == "observed", f"capture_ref {ref_path} must have provenance 'observed'"
+    assert "capture_id" in ref_data, f"capture_ref {ref_path} missing capture_id"
+    return ref_path, ref_data
+
+
+def _extract_native_agent_object(ref_data: dict):
+    """Returns the native `result.agent` object from a Herdr `agent get`-style capture, or None."""
+    payload_obj = ref_data.get("stdout_payload")
+    if not isinstance(payload_obj, dict):
+        return None
+    result = payload_obj.get("result")
+    if not isinstance(result, dict):
+        return None
+    agent = result.get("agent")
+    if not isinstance(agent, dict):
+        return None
+    return agent
+
+
+def _validate_observed_herdr_item(index: int, item: dict, dctx: dict) -> None:
+    """Strict evidence correspondence for an observed Herdr timeline item.
+
+    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.
+    """
+    ref_path, ref_data = _load_observed_capture_ref(index, item)
+
+    # Must be a native Herdr capture, and must itself pass the Layer 1 native contract.
+    assert "herdr_version" in ref_data and "backend" in ref_data, (
+        f"capture_ref {ref_path} for herdr source must be a native Herdr capture"
+    )
+    validate_native_capture(ref_data, file_path=ref_path)
+
+    item_payload = item.get("payload")
+    assert isinstance(item_payload, dict) and item_payload, (
+        f"Observed herdr timeline item [{index}] must declare a non-empty payload dict"
+    )
+    for mandatory in HERDR_MANDATORY_ITEM_FIELDS:
+        assert mandatory in item_payload, (
+            f"Observed herdr timeline item [{index}] is missing mandatory native field '{mandatory}'; "
+            f"comparison against {ref_path} cannot be skipped"
+        )
+
+    agent_obj = _extract_native_agent_object(ref_data)
+    assert agent_obj is not None, (
+        f"capture_ref {ref_path} does not contain a native agent object (exit_code={ref_data.get('exit_code')}); "
+        f"it cannot substantiate observed herdr timeline item [{index}]"
+    )
+
+    # Identity correspondence: 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"], (
+        f"Identity correspondence mismatch: dispatch_context.agent_session_uuid="
+        f"'{dctx['agent_session_uuid']}' but capture {ref_path} records agent_session.value='{capture_uuid}'"
+    )
+
+    # Payload correspondence: every declared field must exist natively and match exactly.
+    for key, declared in item_payload.items():
+        assert key in agent_obj, (
+            f"Payload correspondence mismatch: timeline item [{index}] declares field '{key}' which is not a "
+            f"native field of capture {ref_path}; observed herdr items may only declare native fields"
+        )
+        assert agent_obj[key] == declared, (
+            f"Payload correspondence mismatch: timeline specifies {key}='{declared}' "
+            f"but capture {ref_path} records {key}='{agent_obj[key]}'"
+        )
+
+
 def validate_watcher_scenario(data: dict, file_path: Path = None) -> None:
     """Strictly validates a Watcher composite scenario, attribution, timeline authenticity, and correspondence."""
     assert data.get("schema_version") == "1.0.0", "Invalid schema_version"
@@ -115,58 +217,39 @@ def validate_watcher_scenario(data: dict, file_path: Path = None) -> None:
     }, f"Invalid expected policy decision: {policy.get('decision')}"
     assert policy.get("reason"), "expected_policy.reason must not be empty"
 
-    # Timeline authenticity & correspondence
+    # Timeline authenticity & correspondence (dispatched by explicit source kind)
     observed_count = 0
     simulated_count = 0
     total_items = len(data["timeline"])
     assert total_items > 0, "timeline must contain at least one item"
 
     for i, item in enumerate(data["timeline"]):
+        source_kind = item.get("source_kind")
+        assert source_kind in SUPPORTED_SOURCE_KINDS, (
+            f"Timeline item [{i}] declares unsupported source_kind '{source_kind}'; "
+            f"supported: {sorted(SUPPORTED_SOURCE_KINDS)}"
+        )
+
         item_prov = item.get("source_provenance")
         assert item_prov in {"observed", "simulated", "synthetic"}, f"Invalid source_provenance: {item_prov}"
 
-        if item_prov == "observed":
+        if item_prov != "observed":
-            observed_count += 1
-            assert "capture_ref" in item and item["capture_ref"], (
-                f"Timeline item [{i}] is marked observed but lacks non-empty 'capture_ref'"
-            )
-            ref_path = REPO_ROOT / item["capture_ref"]
-            assert ref_path.exists(), f"Observed timeline item [{i}] 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}"
-
-            # Authenticity check: load and verify evidence record
-            try:
-                ref_data = json.loads(ref_path.read_text(encoding="utf-8"))
-            except Exception as exc:
-                raise AssertionError(f"capture_ref {ref_path} must be valid JSON: {exc}")
-
-            assert ref_data.get("schema_version") == "1.0.0", f"capture_ref {ref_path} missing schema_version 1.0.0"
-            assert ref_data.get("provenance") == "observed", f"capture_ref {ref_path} must have provenance 'observed'"
-            assert "capture_id" in ref_data, f"capture_ref {ref_path} missing capture_id"
-
-            # Source correspondence check
-            source_kind = item.get("source_kind")
-            if source_kind == "herdr":
-                assert "backend" in ref_data or "herdr_version" in ref_data, (
-                    f"capture_ref {ref_path} for herdr source must be a native Herdr capture"
-                )
-                item_payload = item.get("payload", {})
-                if "agent_status" in item_payload:
-                    ref_status = None
-                    payload_obj = ref_data.get("stdout_payload") or ref_data.get("payload") or {}
-                    if isinstance(payload_obj, dict):
-                        if "result" in payload_obj and isinstance(payload_obj["result"], dict):
-                            ag = payload_obj["result"].get("agent", {})
-                            if isinstance(ag, dict):
-                                ref_status = ag.get("agent_status")
-                        if not ref_status and "agent_status" in payload_obj:
-                            ref_status = payload_obj["agent_status"]
-                    assert ref_status == item_payload["agent_status"], (
-                        f"Payload correspondence mismatch: timeline specifies agent_status='{item_payload['agent_status']}' "
-                        f"but capture {ref_path} records agent_status='{ref_status}'"
-                    )
-        else:
             simulated_count += 1
+            continue
+
+        observed_count += 1
+
+        # Fail-closed: an observed claim is only acceptable for source kinds with an evidence schema.
+        assert source_kind in OBSERVED_EVIDENCE_SCHEMAS, (
+            f"Timeline item [{i}] (source_kind='{source_kind}') is marked observed but no independent "
+            f"evidence schema is implemented for this source kind; it must remain 'simulated' "
+            f"(observed schemas: {sorted(OBSERVED_EVIDENCE_SCHEMAS)})"
+        )
+
+        if source_kind == "herdr":
+            _validate_observed_herdr_item(i, item, dctx)
+        else:  # pragma: no cover - guarded by the assertion above
+            raise AssertionError(f"Unhandled observed source_kind '{source_kind}' at timeline item [{i}]")
 
     # Provenance honesty invariant
     if observed_count > 0 and simulated_count > 0:
@@ -382,22 +465,24 @@ def test_negative_watcher_observed_item_with_missing_capture_ref_fails():
     sc = json.loads(sample_file.read_text(encoding="utf-8"))
 
     corrupted = copy.deepcopy(sc)
-    corrupted["timeline"][0]["source_provenance"] = "observed"
+    corrupted["provenance"] = "mixed"
-    corrupted["timeline"][0]["capture_ref"] = "nonexistent/capture/path.json"
+    # timeline[1] is the herdr item in initial_idle (the only source kind with an observed evidence schema).
+    corrupted["timeline"][1]["source_provenance"] = "observed"
+    corrupted["timeline"][1]["capture_ref"] = "nonexistent/capture/path.json"
     with pytest.raises(AssertionError, match="points to nonexistent capture_ref"):
         validate_watcher_scenario(corrupted)
 
 
 def test_negative_watcher_observed_item_pointing_to_unrelated_file_fails():
-    """P1: Pointing an observed timeline reference to an existing unrelated file (e.g. README.md) must fail."""
+    """P1: Pointing an observed herdr timeline reference to an existing unrelated file (e.g. README.md) must fail."""
     sample_file = WATCHER_ROOT / "initial_idle.json"
     sc = json.loads(sample_file.read_text(encoding="utf-8"))
 
     corrupted = copy.deepcopy(sc)
-    corrupted["provenance"] = "observed_composite"
+    corrupted["provenance"] = "mixed"
-    for item in corrupted["timeline"]:
+    # timeline[1] is the herdr item in initial_idle; timeline[0] (event) stays simulated.
-        item["source_provenance"] = "observed"
+    corrupted["timeline"][1]["source_provenance"] = "observed"
-        item["capture_ref"] = "README.md"
+    corrupted["timeline"][1]["capture_ref"] = "README.md"
     with pytest.raises(AssertionError, match="JSON evidence record"):
         validate_watcher_scenario(corrupted)
 
@@ -431,3 +516,152 @@ def test_negative_watcher_coexisting_sources_labeled_simulated_fails():
     # timeline[1] remains simulated
     with pytest.raises(AssertionError, match="overall provenance must be 'mixed'"):
         validate_watcher_scenario(corrupted)
+
+
+# ===========================================================================
+# 5. Source-Kind Dispatch & Evidence Boundary Tests (Codex [P1], Rev.5)
+# ===========================================================================
+
+HERDR_CLAUDE_IDLE = "tests/fixtures/issue_5/herdr/claude/idle.json"
+HERDR_CODEX_DONE = "tests/fixtures/issue_5/herdr/codex/done.json"
+HERDR_AGY_TIMEOUT = "tests/fixtures/issue_5/herdr/agy/wait_timeout.json"
+
+
+def _load_scenario(name: str) -> dict:
+    return json.loads((WATCHER_ROOT / name).read_text(encoding="utf-8"))
+
+
+def test_negative_codex_reproduction_finalized_candidate_all_observed_via_idle_capture_fails():
+    """Codex [P1] reproduction: relabeling every finalized_candidate item as observed and pointing them all
+    at an unrelated native idle snapshot must be rejected. A terminal idle capture cannot prove an
+    external started event or a finalized report manifest."""
+    x = _load_scenario("finalized_candidate.json")
+    x["provenance"] = "observed_composite"
+    for item in x["timeline"]:
+        item["source_provenance"] = "observed"
+        item["capture_ref"] = HERDR_CLAUDE_IDLE
+    with pytest.raises(AssertionError, match="no independent evidence schema"):
+        validate_watcher_scenario(x)
+
+
+@pytest.mark.parametrize("source_kind", sorted(SUPPORTED_SOURCE_KINDS - OBSERVED_EVIDENCE_SCHEMAS))
+def test_negative_non_herdr_source_kinds_are_fail_closed_when_observed(source_kind):
+    """event / filesystem / harness have no evidence schema: an observed claim must be rejected even
+    when capture_ref points at a perfectly valid native Herdr capture."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "mixed"
+    sc["timeline"][0]["source_kind"] = source_kind
+    sc["timeline"][0]["source_provenance"] = "observed"
+    sc["timeline"][0]["capture_ref"] = HERDR_CLAUDE_IDLE
+    with pytest.raises(AssertionError, match=f"source_kind='{source_kind}'.*no independent evidence schema"):
+        validate_watcher_scenario(sc)
+
+
+@pytest.mark.parametrize("source_kind", sorted(SUPPORTED_SOURCE_KINDS - OBSERVED_EVIDENCE_SCHEMAS))
+def test_non_herdr_source_kinds_remain_accepted_when_simulated(source_kind):
+    """The existing positive simulated fixtures must keep working: non-herdr kinds are fine as simulated."""
+    sc = _load_scenario("initial_idle.json")
+    sc["timeline"][0]["source_kind"] = source_kind
+    sc["timeline"][0]["source_provenance"] = "simulated"
+    validate_watcher_scenario(sc)
+
+
+def test_negative_unsupported_source_kind_rejected_regardless_of_provenance():
+    for prov in ("simulated", "observed"):
+        sc = _load_scenario("initial_idle.json")
+        sc["timeline"][0]["source_kind"] = "mqtt_broker_log"
+        sc["timeline"][0]["source_provenance"] = prov
+        sc["timeline"][0]["capture_ref"] = HERDR_CLAUDE_IDLE
+        with pytest.raises(AssertionError, match="unsupported source_kind 'mqtt_broker_log'"):
+            validate_watcher_scenario(sc)
+
+
+def test_negative_observed_herdr_item_missing_mandatory_agent_status_fails():
+    """Omitting agent_status must not let the comparison be skipped (fail-closed on missing mandatory field)."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "mixed"
+    sc["timeline"][1]["source_provenance"] = "observed"
+    sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
+    sc["timeline"][1]["payload"] = {"interactive_ready": True}  # agent_status deliberately absent
+    with pytest.raises(AssertionError, match="missing mandatory native field 'agent_status'"):
+        validate_watcher_scenario(sc)
+
+    sc["timeline"][1]["payload"] = {}
+    with pytest.raises(AssertionError, match="non-empty payload dict"):
+        validate_watcher_scenario(sc)
+
+
+def test_negative_observed_herdr_item_declaring_non_native_field_fails():
+    """An observed herdr item may only declare fields the native capture can substantiate."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "mixed"
+    sc["timeline"][1]["source_provenance"] = "observed"
+    sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
+    sc["timeline"][1]["payload"] = {"agent_status": "idle", "matched_rule": "permission_prompt"}
+    with pytest.raises(AssertionError, match="declares field 'matched_rule' which is not a native field"):
+        validate_watcher_scenario(sc)
+
+
+def test_negative_observed_herdr_item_secondary_field_mismatch_fails():
+    """Every declared field is compared, not only agent_status."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "mixed"
+    sc["timeline"][1]["source_provenance"] = "observed"
+    sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
+    sc["timeline"][1]["payload"] = {"agent_status": "idle", "interactive_ready": False}  # capture says True
+    with pytest.raises(AssertionError, match="timeline specifies interactive_ready='False'"):
+        validate_watcher_scenario(sc)
+
+
+def test_negative_observed_herdr_item_referencing_capture_without_agent_object_fails():
+    """A nonzero-exit wait/timeout capture carries no agent object and cannot prove an agent_status."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "mixed"
+    sc["timeline"][1]["source_provenance"] = "observed"
+    sc["timeline"][1]["capture_ref"] = HERDR_AGY_TIMEOUT
+    sc["timeline"][1]["payload"] = {"agent_status": "idle"}
+    with pytest.raises(AssertionError, match="does not contain a native agent object"):
+        validate_watcher_scenario(sc)
+
+
+def test_negative_observed_herdr_item_session_identity_mismatch_fails():
+    """The capture must be about the session the scenario attributes (4-fold attribution identity)."""
+    sc = _load_scenario("initial_idle.json")  # attributes 7d5700d0... (claude)
+    sc["provenance"] = "mixed"
+    sc["timeline"][1]["source_provenance"] = "observed"
+    sc["timeline"][1]["capture_ref"] = HERDR_CODEX_DONE  # records 01a071d1... (codex)
+    sc["timeline"][1]["payload"] = {"agent_status": "done"}
+    with pytest.raises(AssertionError, match="Identity correspondence mismatch"):
+        validate_watcher_scenario(sc)
+
+
+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."""
+    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]["payload"] = {"agent_status": "idle"}
+    with pytest.raises(AssertionError, match="must have provenance 'observed'"):
+        validate_watcher_scenario(sc)
+
+
+def test_positive_observed_herdr_item_with_full_native_correspondence_passes():
+    """Sanity check that the strict herdr path is not vacuously rejecting: a fully corresponding
+    observed herdr item (matching identity and every declared native field) is accepted."""
+    sc = _load_scenario("initial_idle.json")
+    sc["provenance"] = "mixed"
+    sc["timeline"][1]["source_provenance"] = "observed"
+    sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
+    sc["timeline"][1]["payload"] = {
+        "agent_status": "idle",
+        "interactive_ready": True,
+        "pane_id": "w2X:p1",
+        "name": "planner-reviewer-claude-01",
+    }
+    validate_watcher_scenario(sc)
+
+    # Relabeling this mixed scenario as observed_composite while an event item stays simulated must fail.
+    sc["provenance"] = "observed_composite"
+    with pytest.raises(AssertionError, match="overall provenance must be 'mixed'"):
+        validate_watcher_scenario(sc)