#!/usr/bin/env python3 """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). 2. Herdr wait contract options, exit codes, and channel separation (stderr for timeout & not found). 3. 6-backend capability coverage matrix and honest gap tracking. 4. Watcher Layer 2 composite scenarios, 4-fold attribution, timelines, authenticity, and fail-closed policies. 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. 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 import json from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parent.parent DOCS_ROOT = REPO_ROOT / "docs" / "contracts" / "issue-5" FIXTURES_ROOT = REPO_ROOT / "tests" / "fixtures" / "issue_5" HERDR_ROOT = FIXTURES_ROOT / "herdr" WATCHER_ROOT = FIXTURES_ROOT / "watcher" FORBIDDEN_MAM_KEYS = { "job_id", "attempt_id", "report_phase", "report_path", "expected_decision", "stable_candidate", "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",) # 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) # =========================================================================== def validate_native_capture(cap: dict, file_path: Path = None) -> None: """Strictly validates a native Herdr capture envelope and channel payloads.""" assert cap.get("schema_version") == "1.0.0", "Invalid schema_version" assert cap.get("herdr_version") == "0.8.2", "Invalid herdr_version" assert cap.get("provenance") == "observed", "Native capture provenance must be 'observed'" assert "command_argv" in cap and isinstance(cap["command_argv"], list), "command_argv must be list" assert "exit_code" in cap and isinstance(cap["exit_code"], int), "exit_code must be int" assert "capture_id" in cap, "capture_id required" exit_code = cap["exit_code"] stdout_payload = cap.get("stdout_payload") stderr_raw = cap.get("stderr") # Channel separation invariant if exit_code == 0: assert stdout_payload is not None, "exit 0 must provide stdout_payload" assert stderr_raw is None, "exit 0 must have null stderr" payload_to_check = stdout_payload else: assert stdout_payload is None, "nonzero exit must have null stdout_payload" assert stderr_raw is not None and isinstance(stderr_raw, str), "nonzero exit must have string stderr" try: payload_to_check = json.loads(stderr_raw) except Exception as exc: raise AssertionError(f"stderr for nonzero exit must be valid JSON: {exc}") # Payload duplicate check (if present, must match channel payload) if "payload" in cap and cap["payload"] is not None: assert cap["payload"] == payload_to_check, "Top-level payload duplicate does not match channel payload" # Enforce zero MAM orchestration injection def _check_no_forbidden(obj, prefix=""): if isinstance(obj, dict): for k, v in obj.items(): assert k not in FORBIDDEN_MAM_KEYS, ( f"Forbidden MAM field '{k}' injected into native capture payload at {prefix}{k}" ) _check_no_forbidden(v, prefix=f"{prefix}{k}.") elif isinstance(obj, list): for i, elem in enumerate(obj): _check_no_forbidden(elem, prefix=f"{prefix}[{i}].") _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'" ) 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: 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, 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 identity (session UUID, pane id, backend) must match the scenario's 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 (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"], ( f"Identity correspondence mismatch: dispatch_context.agent_session_uuid=" f"'{dctx['agent_session_uuid']}' but capture {ref_path} records agent_session.value='{capture_uuid}'" ) # Identity correspondence (pane): pane_process_generation is '::'; 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, ( 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]}'" ) 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.""" assert data.get("schema_version") == "1.0.0", "Invalid schema_version" assert "scenario_id" in data, "scenario_id required" assert "description" in data, "description required" assert "filesystem" in data, "filesystem required" assert "expected_policy" in data, "expected_policy required" assert "dispatch_context" in data, "dispatch_context required" assert "timeline" in data and isinstance(data["timeline"], list), "timeline must be list" overall_prov = data.get("provenance") assert overall_prov in {"observed_composite", "simulated", "mixed"}, f"Invalid overall provenance: {overall_prov}" # 4-Fold Attribution verification dctx = data["dispatch_context"] for req_attr in ("job_id", "attempt_id", "agent_session_uuid", "pane_process_generation"): assert req_attr in dctx, f"Missing attribution field in dispatch_context: '{req_attr}'" assert dctx[req_attr] is not None and dctx[req_attr] != "", ( f"Attribution field '{req_attr}' must not be null or empty in dispatch_context" ) assert "execution_anchor_observed" in dctx, "execution_anchor_observed boolean required" # Expected Policy verification policy = data["expected_policy"] assert policy.get("decision") in { "hold", "hold_blocked", "hold_reconnect", "reject_stale", "candidate_approved", }, f"Invalid expected policy decision: {policy.get('decision')}" assert policy.get("reason"), "expected_policy.reason must not be empty" # 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" 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") 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": 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": 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", ( f"Scenario has both observed ({observed_count}) and simulated ({simulated_count}) timeline items; " f"overall provenance must be 'mixed', got '{overall_prov}'" ) elif observed_count == total_items: assert overall_prov == "observed_composite", ( f"All timeline items are observed; overall provenance must be 'observed_composite', got '{overall_prov}'" ) elif simulated_count == total_items: assert overall_prov == "simulated", ( f"All timeline items are simulated; overall provenance must be 'simulated', got '{overall_prov}'" ) # =========================================================================== # 1. Layer 1 — Herdr Native Contract Fixture Tests # =========================================================================== def test_herdr_wait_contract_schema_and_channels(): contract_file = HERDR_ROOT / "wait-contract.json" assert contract_file.exists(), "wait-contract.json must exist" data = json.loads(contract_file.read_text(encoding="utf-8")) assert data["schema_version"] == "1.0.0" assert data["herdr_version"] == "0.8.2" assert "--until" in data["options"] assert "--timeout" in data["options"] outcomes = {o["outcome"]: o for o in data["call_outcomes"]} assert "state_matched" in outcomes assert "timeout" in outcomes assert "agent_not_found" in outcomes # Invariant: exit codes assert outcomes["state_matched"]["exit_code"] == 0 assert outcomes["timeout"]["exit_code"] != 0 assert outcomes["agent_not_found"]["exit_code"] != 0 # Invariant: output channels (both timeout and agent_not_found output to stderr) assert outcomes["state_matched"]["output_channel"] == "stdout" assert outcomes["timeout"]["output_channel"] == "stderr", "Herdr wait timeout must output to stderr" assert outcomes["agent_not_found"]["output_channel"] == "stderr", "Herdr wait not-found must output to stderr" def test_herdr_manifest_covers_all_six_backends(): manifest_file = HERDR_ROOT / "manifest.json" assert manifest_file.exists(), "herdr manifest.json must exist" manifest = json.loads(manifest_file.read_text(encoding="utf-8")) expected_backends = {"claude", "agy", "hermes", "grok", "opencode", "codex"} actual_backends = set(manifest["backends"].keys()) assert actual_backends == expected_backends, f"Expected 6 backends: {expected_backends}" for backend, bdata in manifest["backends"].items(): assert "installed_binary" in bdata assert "slots" in bdata for slot_name, slot in bdata["slots"].items(): if slot["status"] == "observed": target_file = HERDR_ROOT / slot["file"] assert target_file.exists(), f"Observed slot {backend}/{slot_name} missing file: {target_file}" elif slot["status"] == "unobserved": assert slot.get("reason"), f"Unobserved slot {backend}/{slot_name} must state an honest reason" def test_native_herdr_captures_strictly_exclude_orchestration_injection(): capture_files = list(HERDR_ROOT.glob("*/*.json")) assert len(capture_files) >= 4, f"Expected at least 4 native captures, found {len(capture_files)}" for cap_path in capture_files: cap = json.loads(cap_path.read_text(encoding="utf-8")) validate_native_capture(cap, file_path=cap_path) # =========================================================================== # 2. Layer 2 — Watcher Composite Scenario Tests # =========================================================================== def test_watcher_manifest_and_all_seven_scenarios_exist(): manifest_file = WATCHER_ROOT / "manifest.json" assert manifest_file.exists(), "watcher manifest.json must exist" manifest = json.loads(manifest_file.read_text(encoding="utf-8")) # The phase-0 seven are the contract this file guards. Later phases may add # scenarios (phase 2a adds five), so this asserts the seven still exist and # are indexed rather than forbidding growth; validate_watcher_manifest keeps # every entry, old or new, honest against its file. expected_scenarios = { "initial_idle", "tool_running", "report_pause_15s", "finalized_candidate", "blocked_permission", "transport_disconnected", "restarted_generation", } actual_scenarios = set(manifest["scenarios"].keys()) missing = expected_scenarios - actual_scenarios assert not missing, f"phase-0 watcher scenarios went missing: {missing}" for sid, sinfo in manifest["scenarios"].items(): sfile = WATCHER_ROOT / sinfo["file"] assert sfile.exists(), f"Scenario file for {sid} missing: {sfile}" def test_watcher_composite_scenario_structures(): for sfile in WATCHER_ROOT.glob("*.json"): if sfile.name == "manifest.json": continue data = json.loads(sfile.read_text(encoding="utf-8")) validate_watcher_scenario(data, file_path=sfile) # =========================================================================== # 3. Document Alignment Tests (Guards against Prose Inversions F1/F2) # =========================================================================== def test_documents_align_with_runtime_and_cli_contracts(): """Asserts that baseline.md and herdr-runtime.md do not invert CLI precedence or timeout channels.""" baseline_text = (DOCS_ROOT / "baseline.md").read_text(encoding="utf-8") runtime_text = (DOCS_ROOT / "herdr-runtime.md").read_text(encoding="utf-8") # F1 Guard: baseline.md must specify --all-reviewer takes precedence over --reviewer assert "--all-reviewer" in baseline_text assert "run_loop.sh:64, 192-196" in baseline_text assert "--all-reviewer`가 우선" in baseline_text or "--all-reviewer takes precedence" in baseline_text assert "`0`을 허용" in baseline_text, "--plan-talk must document regex accepting 0" # F2 Guard: herdr-runtime.md must document timeout output channel as stderr assert "타임아웃" in runtime_text assert "`stderr`" in runtime_text, "herdr-runtime.md must state timeout JSON is on stderr" def test_mutation_policy_invariants(): """Ensures fail-closed invariants for target CAS state machine.""" # 1. initial_idle must HOLD because execution anchor is missing init_idle = json.loads((WATCHER_ROOT / "initial_idle.json").read_text(encoding="utf-8")) assert init_idle["dispatch_context"]["execution_anchor_observed"] is False assert init_idle["expected_policy"]["decision"] == "hold" # 2. tool_running must HOLD because tool is active tool_run = json.loads((WATCHER_ROOT / "tool_running.json").read_text(encoding="utf-8")) assert tool_run["expected_policy"]["decision"] == "hold" # 3. report_pause_15s must HOLD because commit manifest is absent pause = json.loads((WATCHER_ROOT / "report_pause_15s.json").read_text(encoding="utf-8")) assert pause["filesystem"]["manifest_committed"] is False assert pause["expected_policy"]["decision"] == "hold" # 4. finalized_candidate alone approves completion finalized = json.loads((WATCHER_ROOT / "finalized_candidate.json").read_text(encoding="utf-8")) assert finalized["dispatch_context"]["execution_anchor_observed"] is True assert finalized["filesystem"]["manifest_committed"] is True assert finalized["filesystem"]["hash_verified"] is True assert finalized["expected_policy"]["decision"] == "candidate_approved" # 5. restarted_generation must REJECT stale attempt restarted = json.loads((WATCHER_ROOT / "restarted_generation.json").read_text(encoding="utf-8")) assert restarted["expected_policy"]["decision"] == "reject_stale" # =========================================================================== # 4. Negative / Mutation Tests (Evidence Boundary & Corruption Rejection) # =========================================================================== def test_negative_native_capture_with_injected_mam_field_fails(): sample_file = HERDR_ROOT / "claude" / "idle.json" cap = json.loads(sample_file.read_text(encoding="utf-8")) corrupted = copy.deepcopy(cap) corrupted["stdout_payload"]["result"]["agent"]["report_phase"] = "writing" corrupted["payload"]["result"]["agent"]["report_phase"] = "writing" with pytest.raises(AssertionError, match="Forbidden MAM field 'report_phase'"): validate_native_capture(corrupted) def test_negative_native_capture_channel_exit_code_mismatch_fails(): sample_file = HERDR_ROOT / "claude" / "idle.json" cap = json.loads(sample_file.read_text(encoding="utf-8")) # exit 0 with non-null stderr must fail corrupted = copy.deepcopy(cap) corrupted["stderr"] = "some error" with pytest.raises(AssertionError, match="exit 0 must have null stderr"): validate_native_capture(corrupted) def test_negative_watcher_missing_session_uuid_fails(): sample_file = WATCHER_ROOT / "initial_idle.json" sc = json.loads(sample_file.read_text(encoding="utf-8")) corrupted = copy.deepcopy(sc) del corrupted["dispatch_context"]["agent_session_uuid"] with pytest.raises(AssertionError, match="Missing attribution field"): validate_watcher_scenario(corrupted) corrupted2 = copy.deepcopy(sc) corrupted2["dispatch_context"]["agent_session_uuid"] = None with pytest.raises(AssertionError, match="must not be null"): validate_watcher_scenario(corrupted2) def test_negative_watcher_simulated_labeled_observed_composite_fails(): 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" # Timeline has simulated items, so labeling as observed_composite must fail with pytest.raises(AssertionError, match="overall provenance must be 'simulated'"): validate_watcher_scenario(corrupted) def test_negative_watcher_observed_item_with_missing_capture_ref_fails(): sample_file = WATCHER_ROOT / "initial_idle.json" sc = json.loads(sample_file.read_text(encoding="utf-8")) corrupted = copy.deepcopy(sc) 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 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"] = "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) def test_negative_watcher_observed_item_with_mismatched_payload_fails(): """P1: Referencing a valid capture with a mismatched payload state must fail.""" sample_file = WATCHER_ROOT / "initial_idle.json" sc = json.loads(sample_file.read_text(encoding="utf-8")) corrupted = copy.deepcopy(sc) corrupted["timeline"][0]["source_kind"] = "herdr" corrupted["timeline"][0]["source_provenance"] = "observed" # Points to claude/idle.json (which records agent_status='idle') corrupted["timeline"][0]["capture_ref"] = "tests/fixtures/issue_5/herdr/claude/idle.json" corrupted["timeline"][0]["payload"] = {"agent_status": "working"} # Mismatch! with pytest.raises(AssertionError, match="Payload correspondence mismatch"): validate_watcher_scenario(corrupted) def test_negative_watcher_coexisting_sources_labeled_simulated_fails(): """P1: Coexistence of observed and simulated sources must be labeled 'mixed', not 'simulated'.""" sample_file = WATCHER_ROOT / "initial_idle.json" sc = json.loads(sample_file.read_text(encoding="utf-8")) corrupted = copy.deepcopy(sc) corrupted["provenance"] = "simulated" # Dishonest! Should be mixed corrupted["timeline"][0]["source_kind"] = "herdr" corrupted["timeline"][0]["source_provenance"] = "observed" corrupted["timeline"][0]["capture_ref"] = "tests/fixtures/issue_5/herdr/claude/idle.json" corrupted["timeline"][0]["payload"] = {"agent_status": "idle"} # 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 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/herdr/manifest.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) # =========================================================================== # 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)