GodopuandClaude Opus 5
964abc6965
policy.decide is a pure function so the phase-0 scenario fixtures can be fed to it verbatim. Those seven decisions survived three-reviewer scrutiny, so they are the contract rather than a fresh set of rules, and the test asserts against expected_policy.decision directly. Order inside the policy matters. Transport doubt and staleness invalidate the evidence itself, so they are answered before anything is inferred from it; a blocked agent is waiting for a human and can never be completed; attribution including the execution anchor comes before quiescence, because idle is not evidence of finishing when a session is reused; and a report must be finalized, never merely paused, since a debounce is a candidate filter and not an authority. The shadow runner is never handed the engine, so it has no route to a write. That is the structural half of P2A-03. The measured half hashes every file under the registry directory around a full run and asserts it is unchanged, which is a stronger claim than checking a status field. Adds the five scenarios P2A-04 requires: a raced double dispatch, a reused attempt with no anchor, a mid-write report, an unknown status, and a pane restarted mid-flight. They pass phase 0's own fixture validator, which is what keeps them honest. The phase-0 test that pinned exactly seven scenarios now requires the seven to still exist rather than forbidding growth. Also pins the two phase 1 handoffs as dormant: nothing here spawns a daemon, so preexec_fn stays unreachable, and nothing calls a frozen shell helper. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UrDNJkHmm18uu7n7JwV5Qd
896 lines
45 KiB
Python
896 lines
45 KiB
Python
#!/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 '<pane_id>:<pid>:<start>'; 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)
|