test(issue-5): add phase 0 contract validation and negative test suite

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)