diff --git a/tests/test_issue5_phase0_contracts.py b/tests/test_issue5_phase0_contracts.py new file mode 100644 index 0000000..5cede2a --- /dev/null +++ b/tests/test_issue5_phase0_contracts.py @@ -0,0 +1,433 @@ +#!/usr/bin/env python3 +"""Phase 0 contract and fixture verification tests for Issue #5 (Rev.4). + +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. +""" + +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", +} + + +# =========================================================================== +# 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 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 + 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"]): + 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: + simulated_count += 1 + + # 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")) + + expected_scenarios = { + "initial_idle", + "tool_running", + "report_pause_15s", + "finalized_candidate", + "blocked_permission", + "transport_disconnected", + "restarted_generation", + } + actual_scenarios = set(manifest["scenarios"].keys()) + assert actual_scenarios == expected_scenarios, f"Mismatch in watcher scenarios: {actual_scenarios}" + + 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["timeline"][0]["source_provenance"] = "observed" + corrupted["timeline"][0]["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.""" + 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" + 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)