From 19d798bf8f7c1a4dc1fb531f4e58a85734172547 Mon Sep 17 00:00:00 2001 From: Godopu Date: Sun, 6 Sep 2026 13:37:07 +0900 Subject: [PATCH] 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 +- docs/contracts/issue-5/implementation_plan.md | 2 +- docs/contracts/issue-5/task.md | 2 +- tests/test_issue5_phase0_contracts.py | 334 +++++++++++++++--- 4 files changed, 288 insertions(+), 54 deletions(-) diff --git a/docs/contracts/issue-5/baseline.md b/docs/contracts/issue-5/baseline.md index 4d192f4..b0d601a 100644 --- a/docs/contracts/issue-5/baseline.md +++ b/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]`)이 있습니다. diff --git a/docs/contracts/issue-5/implementation_plan.md b/docs/contracts/issue-5/implementation_plan.md index 21e2a69..6dd281a 100644 --- a/docs/contracts/issue-5/implementation_plan.md +++ b/docs/contracts/issue-5/implementation_plan.md @@ -21,7 +21,7 @@ | 단계 | 작업 목표 | 세부 구현 내용 | 필수 완료 게이트 (Gate Criteria) | 롤백 절차 (Rollback Path) | |---|---|---|---|---| -| **Phase 0**
*(현재)* | **사전 계약 고정 및 Fixture 수집** | • CLI, 이벤트, 리포트, 락 계약 기준선 문서화
• Herdr 0.8.2 네이티브 캡처 및 복합 Watcher 시나리오 2계층 분리
• 6개 백엔드 실측 증거 행렬 수집 | • 484개 전체 테스트 수집 및 회귀 0-Fail (100% 통과)
• Fixture 스키마 및 증거 경계 검증 통과
• 프로덕션 코드 무변경 diff 검증 | 미추적 산출물 즉시 삭제 (`rm -rf docs/contracts/issue-5 tests/fixtures/issue_5 tests/test_issue5_phase0_contracts.py`) | +| **Phase 0**
*(현재)* | **사전 계약 고정 및 Fixture 수집** | • CLI, 이벤트, 리포트, 락 계약 기준선 문서화
• Herdr 0.8.2 네이티브 캡처 및 복합 Watcher 시나리오 2계층 분리
• 6개 백엔드 실측 증거 행렬 수집 | • 499개 전체 테스트 수집 및 회귀 0-Fail (100% 통과)
• Fixture 스키마 및 증거 경계 검증 통과
• 프로덕션 코드 무변경 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 이하 위임 헬퍼를 파이썬 서비스 클래스로 추출
• `start_watchdog`을 내부 명칭 `ensure_subscriber_daemon`으로 명명(기존 함수 별칭 보존)
• 장수 프로세스 단일 MQTT 연결 재사용 프로토타입 | • CLI/non-fatal/HMAC/seq/디스크 폴백 회귀 100% 통과
• 데몬 수명 및 PID 계약 일치 확인
• **자동완료 정책 변경 없음** | `lib.sh` 위임 헬퍼 호출을 이전 인라인/셸 방식으로 즉시 복원 | | **Phase 2a** | **`lib_py/loop/` 코어 & Shadow Watcher** | • `engine.py` (단일 전이 주체), `watcher.py` (`herdr agent wait` 래핑) 구현
• 모든 Writer에 대한 단일 CAS 트랜잭션 경계 구현
• 자동완료를 수행하지 않고 "Would-Complete" 로깅만 수행하는 Shadow 모드 가동 | • 락 내 비동기/I/O 배제 검증
• 소켓 단절/스냅샷 재연결 백오프 검증
• Shadow 모드가 레지스트리 상태를 쓰지 않음을 검증 | 1) Shadow 모니터 프로세스 종료 및 인메모리 큐 폐기
2) CAS 서비스는 하위 호환 모드로 유지하거나 `MAM_ENGINE_CAS=0` 플래그로 비활성화하여 레거시 무조건 필드 병합(`update_job_status`)으로 복귀
3) 기존 셸 루프 상태 및 레코드 불변 유지 | | **Phase 2b** | **이중 감지 자동완료 활성화 (Flag-Gated)** | • 4중 실행 귀속 + 리포트 확정 계약 + 단일 CAS 전이 게이트 연동
• Pending Notification 지연 알림 및 크래시 복구 구현
• 검증된 백엔드/작업에 한해 opt-in 플래그로 활성화 (기본값 OFF) | • 부수효과 작업 예외 및 무판정 fail-closed 테스트 완료
• 경쟁/재시작 시나리오 통과
• 플래그 OFF 시 기존 explicit event 경로 100% 보존 | 환경변수/플래그 `MAM_AUTO_COMPLETE=0` 설정; 이미 확정된 terminal 레코드는 보존하고 신규 자동완료만 중단 | diff --git a/docs/contracts/issue-5/task.md b/docs/contracts/issue-5/task.md index 809c709..0703dd0 100644 --- a/docs/contracts/issue-5/task.md +++ b/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 동작 기록. diff --git a/tests/test_issue5_phase0_contracts.py b/tests/test_issue5_phase0_contracts.py index 5cede2a..89f2b5b 100644 --- a/tests/test_issue5_phase0_contracts.py +++ b/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": - 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: + if item_prov != "observed": 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["timeline"][0]["capture_ref"] = "nonexistent/capture/path.json" + corrupted["provenance"] = "mixed" + # 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" - for item in corrupted["timeline"]: - item["source_provenance"] = "observed" - item["capture_ref"] = "README.md" + corrupted["provenance"] = "mixed" + # timeline[1] is the herdr item in initial_idle; timeline[0] (event) stays simulated. + corrupted["timeline"][1]["source_provenance"] = "observed" + 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)