tmpl/multi-agent-mux
1
0 Fork 0
Code Issues 9 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

test(issue-5): implement fail-closed evidence dispatch and fix VALID_STATUSES citation

Browse Source 
- Fix Codex [P1]: add SUPPORTED_SOURCE_KINDS dispatch and enforce fail-closed rejection for non-herdr observed items without independent schema
- Add strict correspondence checks for native Herdr observed items (agent_status, native schema, session UUID)
- Add negative test reproducing Codex in-memory mutation of finalized_candidate
- Fix Claude Finding 2: update VALID_STATUSES citation in baseline.md to registry.py:38
- Update Phase 0 collected test counts to 499 in contract docs
This commit is contained in:
Godopu committed 2026-09-06 13:37:07 +09:00
1 parent dd774b1a9f
commit 19d798bf8f
4 files changed
+288 -54
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/test_issue5_phase0_contracts.py
+284 -50
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Phase 0 contract and fixture verification tests for Issue #5 (Rev.4).
"""Phase 0 contract and fixture verification tests for Issue #5 (Rev.5).
Validates:
1. Herdr Layer 1 native contract captures (pure native observation, no injected MAM fields).
@@ -9,6 +9,10 @@ Validates:
5. Contract cross-checks against baseline.md and herdr-runtime.md.
6. Mutation and negative tests demonstrating that corruption, unrelated references, payload mismatches,
and dishonest provenance labels are rejected.
7. Source-kind dispatch: every timeline item must declare a supported source kind, and an item may only be
accepted as 'observed' when an evidence schema exists for that source kind (currently Herdr only).
Event / filesystem / harness items have no independent evidence schema yet and are fail-closed when
labeled 'observed'; they must remain 'simulated' until such schemas exist.
"""
import copy
@@ -32,6 +36,18 @@ FORBIDDEN_MAM_KEYS = {
"manifest_hash",
}
# Explicit set of source kinds a Watcher timeline item may declare. Anything else is rejected.
SUPPORTED_SOURCE_KINDS = {"herdr", "event", "filesystem", "harness"}
# Source kinds for which an independent, machine-checkable evidence schema is implemented.
# Only these may carry source_provenance == "observed". All other supported kinds are
# fail-closed for observed claims until their evidence schema exists.
OBSERVED_EVIDENCE_SCHEMAS = {"herdr"}
# For observed Herdr items, these payload fields MUST be declared so that the comparison
# against the native capture cannot be skipped.
HERDR_MANDATORY_ITEM_FIELDS = ("agent_status",)
# ===========================================================================
# Validation Engine Functions (Used for both real fixtures & negative tests)
@@ -82,6 +98,92 @@ def validate_native_capture(cap: dict, file_path: Path = None) -> None:
_check_no_forbidden(payload_to_check)
def _load_observed_capture_ref(index: int, item: dict) -> "tuple[Path, dict]":
"""Common envelope checks for any observed timeline item's capture_ref. Returns (path, record)."""
assert "capture_ref" in item and item["capture_ref"], (
f"Timeline item [{index}] is marked observed but lacks non-empty 'capture_ref'"
)
ref_path = REPO_ROOT / item["capture_ref"]
assert ref_path.exists(), f"Observed timeline item [{index}] points to nonexistent capture_ref: {ref_path}"
assert ref_path.suffix == ".json", f"capture_ref must point to a JSON evidence record, got {ref_path}"
try:
ref_data = json.loads(ref_path.read_text(encoding="utf-8"))
except Exception as exc:
raise AssertionError(f"capture_ref {ref_path} must be valid JSON: {exc}")
assert isinstance(ref_data, dict), f"capture_ref {ref_path} must be a JSON object evidence record"
assert ref_data.get("schema_version") == "1.0.0", f"capture_ref {ref_path} missing schema_version 1.0.0"
assert ref_data.get("provenance") == "observed", f"capture_ref {ref_path} must have provenance 'observed'"
assert "capture_id" in ref_data, f"capture_ref {ref_path} missing capture_id"
return ref_path, ref_data
def _extract_native_agent_object(ref_data: dict):
"""Returns the native `result.agent` object from a Herdr `agent get`-style capture, or None."""
payload_obj = ref_data.get("stdout_payload")
if not isinstance(payload_obj, dict):
return None
result = payload_obj.get("result")
if not isinstance(result, dict):
return None
agent = result.get("agent")
if not isinstance(agent, dict):
return None
return agent
def _validate_observed_herdr_item(index: int, item: dict, dctx: dict) -> None:
"""Strict evidence correspondence for an observed Herdr timeline item.
Every declared payload field must be a native field of the referenced capture's agent object
and must match its value exactly. Mandatory fields cannot be omitted to dodge the comparison,
and the capture's agent session identity must match the scenario's dispatch attribution.
"""
ref_path, ref_data = _load_observed_capture_ref(index, item)
# Must be a native Herdr capture, and must itself pass the Layer 1 native contract.
assert "herdr_version" in ref_data and "backend" in ref_data, (
f"capture_ref {ref_path} for herdr source must be a native Herdr capture"
)
validate_native_capture(ref_data, file_path=ref_path)
item_payload = item.get("payload")
assert isinstance(item_payload, dict) and item_payload, (
f"Observed herdr timeline item [{index}] must declare a non-empty payload dict"
)
for mandatory in HERDR_MANDATORY_ITEM_FIELDS:
assert mandatory in item_payload, (
f"Observed herdr timeline item [{index}] is missing mandatory native field '{mandatory}'; "
f"comparison against {ref_path} cannot be skipped"
)
agent_obj = _extract_native_agent_object(ref_data)
assert agent_obj is not None, (
f"capture_ref {ref_path} does not contain a native agent object (exit_code={ref_data.get('exit_code')}); "
f"it cannot substantiate observed herdr timeline item [{index}]"
)
# Identity correspondence: the capture must be about the session this scenario attributes.
session_obj = agent_obj.get("agent_session")
capture_uuid = session_obj.get("value") if isinstance(session_obj, dict) else None
assert capture_uuid == dctx["agent_session_uuid"], (
f"Identity correspondence mismatch: dispatch_context.agent_session_uuid="
f"'{dctx['agent_session_uuid']}' but capture {ref_path} records agent_session.value='{capture_uuid}'"
)
# Payload correspondence: every declared field must exist natively and match exactly.
for key, declared in item_payload.items():
assert key in agent_obj, (
f"Payload correspondence mismatch: timeline item [{index}] declares field '{key}' which is not a "
f"native field of capture {ref_path}; observed herdr items may only declare native fields"
)
assert agent_obj[key] == declared, (
f"Payload correspondence mismatch: timeline specifies {key}='{declared}' "
f"but capture {ref_path} records {key}='{agent_obj[key]}'"
)
def validate_watcher_scenario(data: dict, file_path: Path = None) -> None:
"""Strictly validates a Watcher composite scenario, attribution, timeline authenticity, and correspondence."""
assert data.get("schema_version") == "1.0.0", "Invalid schema_version"
@@ -115,58 +217,39 @@ def validate_watcher_scenario(data: dict, file_path: Path = None) -> None:
}, f"Invalid expected policy decision: {policy.get('decision')}"
assert policy.get("reason"), "expected_policy.reason must not be empty"
# Timeline authenticity & correspondence
# Timeline authenticity & correspondence (dispatched by explicit source kind)
observed_count = 0
simulated_count = 0
total_items = len(data["timeline"])
assert total_items > 0, "timeline must contain at least one item"
for i, item in enumerate(data["timeline"]):
source_kind = item.get("source_kind")
assert source_kind in SUPPORTED_SOURCE_KINDS, (
f"Timeline item [{i}] declares unsupported source_kind '{source_kind}'; "
f"supported: {sorted(SUPPORTED_SOURCE_KINDS)}"
)
item_prov = item.get("source_provenance")
assert item_prov in {"observed", "simulated", "synthetic"}, f"Invalid source_provenance: {item_prov}"
if item_prov == "observed":
observed_count += 1
assert "capture_ref" in item and item["capture_ref"], (
f"Timeline item [{i}] is marked observed but lacks non-empty 'capture_ref'"
)
ref_path = REPO_ROOT / item["capture_ref"]
assert ref_path.exists(), f"Observed timeline item [{i}] points to nonexistent capture_ref: {ref_path}"
assert ref_path.suffix == ".json", f"capture_ref must point to a JSON evidence record, got {ref_path}"
# Authenticity check: load and verify evidence record
try:
ref_data = json.loads(ref_path.read_text(encoding="utf-8"))
except Exception as exc:
raise AssertionError(f"capture_ref {ref_path} must be valid JSON: {exc}")
assert ref_data.get("schema_version") == "1.0.0", f"capture_ref {ref_path} missing schema_version 1.0.0"
assert ref_data.get("provenance") == "observed", f"capture_ref {ref_path} must have provenance 'observed'"
assert "capture_id" in ref_data, f"capture_ref {ref_path} missing capture_id"
# Source correspondence check
source_kind = item.get("source_kind")
if source_kind == "herdr":
assert "backend" in ref_data or "herdr_version" in ref_data, (
f"capture_ref {ref_path} for herdr source must be a native Herdr capture"
)
item_payload = item.get("payload", {})
if "agent_status" in item_payload:
ref_status = None
payload_obj = ref_data.get("stdout_payload") or ref_data.get("payload") or {}
if isinstance(payload_obj, dict):
if "result" in payload_obj and isinstance(payload_obj["result"], dict):
ag = payload_obj["result"].get("agent", {})
if isinstance(ag, dict):
ref_status = ag.get("agent_status")
if not ref_status and "agent_status" in payload_obj:
ref_status = payload_obj["agent_status"]
assert ref_status == item_payload["agent_status"], (
f"Payload correspondence mismatch: timeline specifies agent_status='{item_payload['agent_status']}' "
f"but capture {ref_path} records agent_status='{ref_status}'"
)
else:
if item_prov != "observed":
simulated_count += 1
continue
observed_count += 1
# Fail-closed: an observed claim is only acceptable for source kinds with an evidence schema.
assert source_kind in OBSERVED_EVIDENCE_SCHEMAS, (
f"Timeline item [{i}] (source_kind='{source_kind}') is marked observed but no independent "
f"evidence schema is implemented for this source kind; it must remain 'simulated' "
f"(observed schemas: {sorted(OBSERVED_EVIDENCE_SCHEMAS)})"
)
if source_kind == "herdr":
_validate_observed_herdr_item(i, item, dctx)
else: # pragma: no cover - guarded by the assertion above
raise AssertionError(f"Unhandled observed source_kind '{source_kind}' at timeline item [{i}]")
# Provenance honesty invariant
if observed_count > 0 and simulated_count > 0:
@@ -382,22 +465,24 @@ def test_negative_watcher_observed_item_with_missing_capture_ref_fails():
sc = json.loads(sample_file.read_text(encoding="utf-8"))
corrupted = copy.deepcopy(sc)
corrupted["timeline"][0]["source_provenance"] = "observed"
corrupted["timeline"][0]["capture_ref"] = "nonexistent/capture/path.json"
corrupted["provenance"] = "mixed"
# timeline[1] is the herdr item in initial_idle (the only source kind with an observed evidence schema).
corrupted["timeline"][1]["source_provenance"] = "observed"
corrupted["timeline"][1]["capture_ref"] = "nonexistent/capture/path.json"
with pytest.raises(AssertionError, match="points to nonexistent capture_ref"):
validate_watcher_scenario(corrupted)
def test_negative_watcher_observed_item_pointing_to_unrelated_file_fails():
"""P1: Pointing an observed timeline reference to an existing unrelated file (e.g. README.md) must fail."""
"""P1: Pointing an observed herdr timeline reference to an existing unrelated file (e.g. README.md) must fail."""
sample_file = WATCHER_ROOT / "initial_idle.json"
sc = json.loads(sample_file.read_text(encoding="utf-8"))
corrupted = copy.deepcopy(sc)
corrupted["provenance"] = "observed_composite"
for item in corrupted["timeline"]:
item["source_provenance"] = "observed"
item["capture_ref"] = "README.md"
corrupted["provenance"] = "mixed"
# timeline[1] is the herdr item in initial_idle; timeline[0] (event) stays simulated.
corrupted["timeline"][1]["source_provenance"] = "observed"
corrupted["timeline"][1]["capture_ref"] = "README.md"
with pytest.raises(AssertionError, match="JSON evidence record"):
validate_watcher_scenario(corrupted)
@@ -431,3 +516,152 @@ def test_negative_watcher_coexisting_sources_labeled_simulated_fails():
# timeline[1] remains simulated
with pytest.raises(AssertionError, match="overall provenance must be 'mixed'"):
validate_watcher_scenario(corrupted)
# ===========================================================================
# 5. Source-Kind Dispatch & Evidence Boundary Tests (Codex [P1], Rev.5)
# ===========================================================================
HERDR_CLAUDE_IDLE = "tests/fixtures/issue_5/herdr/claude/idle.json"
HERDR_CODEX_DONE = "tests/fixtures/issue_5/herdr/codex/done.json"
HERDR_AGY_TIMEOUT = "tests/fixtures/issue_5/herdr/agy/wait_timeout.json"
def _load_scenario(name: str) -> dict:
return json.loads((WATCHER_ROOT / name).read_text(encoding="utf-8"))
def test_negative_codex_reproduction_finalized_candidate_all_observed_via_idle_capture_fails():
"""Codex [P1] reproduction: relabeling every finalized_candidate item as observed and pointing them all
at an unrelated native idle snapshot must be rejected. A terminal idle capture cannot prove an
external started event or a finalized report manifest."""
x = _load_scenario("finalized_candidate.json")
x["provenance"] = "observed_composite"
for item in x["timeline"]:
item["source_provenance"] = "observed"
item["capture_ref"] = HERDR_CLAUDE_IDLE
with pytest.raises(AssertionError, match="no independent evidence schema"):
validate_watcher_scenario(x)
@pytest.mark.parametrize("source_kind", sorted(SUPPORTED_SOURCE_KINDS - OBSERVED_EVIDENCE_SCHEMAS))
def test_negative_non_herdr_source_kinds_are_fail_closed_when_observed(source_kind):
"""event / filesystem / harness have no evidence schema: an observed claim must be rejected even
when capture_ref points at a perfectly valid native Herdr capture."""
sc = _load_scenario("initial_idle.json")
sc["provenance"] = "mixed"
sc["timeline"][0]["source_kind"] = source_kind
sc["timeline"][0]["source_provenance"] = "observed"
sc["timeline"][0]["capture_ref"] = HERDR_CLAUDE_IDLE
with pytest.raises(AssertionError, match=f"source_kind='{source_kind}'.*no independent evidence schema"):
validate_watcher_scenario(sc)
@pytest.mark.parametrize("source_kind", sorted(SUPPORTED_SOURCE_KINDS - OBSERVED_EVIDENCE_SCHEMAS))
def test_non_herdr_source_kinds_remain_accepted_when_simulated(source_kind):
"""The existing positive simulated fixtures must keep working: non-herdr kinds are fine as simulated."""
sc = _load_scenario("initial_idle.json")
sc["timeline"][0]["source_kind"] = source_kind
sc["timeline"][0]["source_provenance"] = "simulated"
validate_watcher_scenario(sc)
def test_negative_unsupported_source_kind_rejected_regardless_of_provenance():
for prov in ("simulated", "observed"):
sc = _load_scenario("initial_idle.json")
sc["timeline"][0]["source_kind"] = "mqtt_broker_log"
sc["timeline"][0]["source_provenance"] = prov
sc["timeline"][0]["capture_ref"] = HERDR_CLAUDE_IDLE
with pytest.raises(AssertionError, match="unsupported source_kind 'mqtt_broker_log'"):
validate_watcher_scenario(sc)
def test_negative_observed_herdr_item_missing_mandatory_agent_status_fails():
"""Omitting agent_status must not let the comparison be skipped (fail-closed on missing mandatory field)."""
sc = _load_scenario("initial_idle.json")
sc["provenance"] = "mixed"
sc["timeline"][1]["source_provenance"] = "observed"
sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
sc["timeline"][1]["payload"] = {"interactive_ready": True} # agent_status deliberately absent
with pytest.raises(AssertionError, match="missing mandatory native field 'agent_status'"):
validate_watcher_scenario(sc)
sc["timeline"][1]["payload"] = {}
with pytest.raises(AssertionError, match="non-empty payload dict"):
validate_watcher_scenario(sc)
def test_negative_observed_herdr_item_declaring_non_native_field_fails():
"""An observed herdr item may only declare fields the native capture can substantiate."""
sc = _load_scenario("initial_idle.json")
sc["provenance"] = "mixed"
sc["timeline"][1]["source_provenance"] = "observed"
sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
sc["timeline"][1]["payload"] = {"agent_status": "idle", "matched_rule": "permission_prompt"}
with pytest.raises(AssertionError, match="declares field 'matched_rule' which is not a native field"):
validate_watcher_scenario(sc)
def test_negative_observed_herdr_item_secondary_field_mismatch_fails():
"""Every declared field is compared, not only agent_status."""
sc = _load_scenario("initial_idle.json")
sc["provenance"] = "mixed"
sc["timeline"][1]["source_provenance"] = "observed"
sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
sc["timeline"][1]["payload"] = {"agent_status": "idle", "interactive_ready": False} # capture says True
with pytest.raises(AssertionError, match="timeline specifies interactive_ready='False'"):
validate_watcher_scenario(sc)
def test_negative_observed_herdr_item_referencing_capture_without_agent_object_fails():
"""A nonzero-exit wait/timeout capture carries no agent object and cannot prove an agent_status."""
sc = _load_scenario("initial_idle.json")
sc["provenance"] = "mixed"
sc["timeline"][1]["source_provenance"] = "observed"
sc["timeline"][1]["capture_ref"] = HERDR_AGY_TIMEOUT
sc["timeline"][1]["payload"] = {"agent_status": "idle"}
with pytest.raises(AssertionError, match="does not contain a native agent object"):
validate_watcher_scenario(sc)
def test_negative_observed_herdr_item_session_identity_mismatch_fails():
"""The capture must be about the session the scenario attributes (4-fold attribution identity)."""
sc = _load_scenario("initial_idle.json") # attributes 7d5700d0... (claude)
sc["provenance"] = "mixed"
sc["timeline"][1]["source_provenance"] = "observed"
sc["timeline"][1]["capture_ref"] = HERDR_CODEX_DONE # records 01a071d1... (codex)
sc["timeline"][1]["payload"] = {"agent_status": "done"}
with pytest.raises(AssertionError, match="Identity correspondence mismatch"):
validate_watcher_scenario(sc)
def test_negative_observed_herdr_item_referencing_non_native_json_record_fails():
"""A JSON file that is not a native Herdr capture (e.g. a Watcher scenario) is not Herdr evidence."""
sc = _load_scenario("initial_idle.json")
sc["provenance"] = "mixed"
sc["timeline"][1]["source_provenance"] = "observed"
sc["timeline"][1]["capture_ref"] = "tests/fixtures/issue_5/watcher/finalized_candidate.json"
sc["timeline"][1]["payload"] = {"agent_status": "idle"}
with pytest.raises(AssertionError, match="must have provenance 'observed'"):
validate_watcher_scenario(sc)
def test_positive_observed_herdr_item_with_full_native_correspondence_passes():
"""Sanity check that the strict herdr path is not vacuously rejecting: a fully corresponding
observed herdr item (matching identity and every declared native field) is accepted."""
sc = _load_scenario("initial_idle.json")
sc["provenance"] = "mixed"
sc["timeline"][1]["source_provenance"] = "observed"
sc["timeline"][1]["capture_ref"] = HERDR_CLAUDE_IDLE
sc["timeline"][1]["payload"] = {
"agent_status": "idle",
"interactive_ready": True,
"pane_id": "w2X:p1",
"name": "planner-reviewer-claude-01",
}
validate_watcher_scenario(sc)
# Relabeling this mixed scenario as observed_composite while an event item stays simulated must fail.
sc["provenance"] = "observed_composite"
with pytest.raises(AssertionError, match="overall provenance must be 'mixed'"):
validate_watcher_scenario(sc)