#!/usr/bin/env python3 """Hostile tests for the PStack causal-evidence contract.""" from __future__ import annotations import copy import datetime as dt import importlib.util import json import tempfile import unittest from pathlib import Path ROOT = Path(__file__).resolve().parents[2] MODULE_PATH = ROOT / "scripts" / "ops" / "pstack" / "pstack_evidence.py" CONTRACT_PATH = ROOT / "config" / "pstack-evidence-contract.json" WORKGRAPH_PATH = ROOT / "config" / "pstack-program-workgraph.json" SPEC = importlib.util.spec_from_file_location("pstack_evidence", MODULE_PATH) assert SPEC and SPEC.loader pstack = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(pstack) NOW = dt.datetime(2026, 9, 4, 5, 5, tzinfo=dt.timezone.utc) OBSERVED = "2026-09-04T05:00:00Z" EXPIRES = "2026-09-04T05:20:00Z" def selector() -> dict: return { "repository": "Biji-Biji-Initiative/bbi-infrastructure", "pr": 11573, "head": "a" * 40, "surfaces": ["Policy Guards", "policy-admission (shadow)"], } def valid_bundle() -> dict: selection = selector() return { "schema": "bbi.process-truth/v1", "bundle_id": "mode-s-pr-11573-generation-33772003659.1", "program": "mode-s", "subject": { "kind": "workflow_transaction", "identity": { "repository": "Biji-Biji-Initiative/bbi-infrastructure", "pr": 11573, "head": "a" * 40, "base": "b" * 40, "run_id": 33772003659, "run_attempt": 1, }, }, "collector": {"name": "github-admission-adapter", "version": "1.0.0", "git_sha": "c" * 40}, "clock": {"observed_at": OBSERVED, "monotonic_ns": 123456789, "deadline_ms": 30000}, "universe": { "selector": selection, "selector_sha256": pstack._selector_digest(selection), "expected_count": 3, "observed_count": 3, "terminated_by": "provider_pagination_end", "coverage_evidence": ["control.known-shadow-positive"], }, "completeness": "bounded", "bounds": { "surface_cap": 16, "claim_cap": 16, "unknown_cap": 8, "stdout_byte_cap": 65536, "stderr_byte_cap": 65536, }, "surfaces": [ { "id": "provider.policy-guards", "source": "provider.api.check-runs", "authority": "provider", "status": "ok", "duration_ms": 142, "deadline_ms": 5000, "artifact": "github://check-runs/exact-head", "digest": "1" * 64, "bytes": 2048, "truncated": False, "facts": {"same_name_count": 1, "trusted_count": 1, "conclusion": "success", "generation": "33772003659.1"}, }, { "id": "provider.shadow-row", "source": "provider.api.check-runs", "authority": "provider", "status": "ok", "duration_ms": 113, "deadline_ms": 5000, "artifact": "github://check-runs/exact-head", "digest": "2" * 64, "bytes": 1024, "truncated": False, "facts": {"same_name_count": 0, "trusted_count": 0}, }, { "id": "control.known-shadow-positive", "source": "provider.api.check-runs", "authority": "provider", "status": "ok", "duration_ms": 101, "deadline_ms": 5000, "artifact": "github://check-runs/known-positive-control", "digest": "3" * 64, "bytes": 1024, "truncated": False, "facts": {"same_name_count": 1, "trusted_count": 1}, }, ], "claims": [ { "id": "fact.policy-guards-success", "kind": "fact", "state": "observed", "method": "direct_observation", "polarity": "presence", "scope": "exact", "statement": "The exact PR head has one canonical successful Policy Guards generation.", "evidence": ["provider.policy-guards"], "negative_control_evidence": [], "confidence": None, "falsifier": None, "expires_at": EXPIRES, "supersedes": [], }, { "id": "fact.shadow-absent", "kind": "fact", "state": "observed", "method": "direct_observation", "polarity": "absence", "scope": "bounded", "statement": "No trusted shadow row was present in the bounded exact-head census.", "evidence": ["provider.shadow-row"], "negative_control_evidence": ["control.known-shadow-positive"], "confidence": None, "falsifier": None, "expires_at": EXPIRES, "supersedes": [], }, { "id": "hypothesis.publisher-race", "kind": "hypothesis", "state": "open", "method": "causal_inference", "polarity": "state", "scope": "bounded", "statement": "The publisher lost the open-PR race after Policy Guards completed.", "evidence": ["provider.policy-guards", "provider.shadow-row"], "negative_control_evidence": [], "confidence": 0.7, "falsifier": "A provider-bound publisher run reached its write step while the PR was open and emitted another typed terminal.", "expires_at": EXPIRES, "supersedes": [], }, ], "unknowns": [ { "id": "unknown.first-missing-transition", "state": "open", "question": "Which publisher or reconciler transition was first absent for this generation?", "blocks": ["PUBLISH_SHADOW", "PUBLISH_APP_ONLY_ADMISSION"], "owner": "conveyor-02", "probe": { "adapter": "github-workflow-generation-trace", "action": "Enumerate trusted publisher and reconciler jobs for this exact head and generation.", "deadline_ms": 30000, "expected_discriminator": "One edge among wake, bind, decide, write, or read-back is absent or held.", "abort_conditions": ["provider census incomplete", "head identity changed"], }, "resolution_evidence": [], } ], "next_transition": { "type": "DISCRIMINATE", "owner": "conveyor-02", "preconditions": ["exact head remains resolvable", "provider census is complete within cap"], "success_receipt": "FIRST_MISSING_TRANSITION_CLASSIFIED", "abort_conditions": ["provider ambiguity", "identity race", "census truncation"], }, "bundle_digest": "", } class ContractTest(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.contract = pstack.load_contract(CONTRACT_PATH) def sealed(self, bundle: dict | None = None) -> dict: return pstack.seal_bundle(bundle or valid_bundle(), self.contract, now=NOW) def rejected(self, bundle: dict, fragment: str) -> None: with self.assertRaises(pstack.ContractError) as caught: pstack.seal_bundle(bundle, self.contract, now=NOW) self.assertIn(fragment, str(caught.exception)) def test_valid_bundle_seals_and_verifies(self) -> None: sealed = self.sealed() summary = pstack.validate_bundle(sealed, self.contract, now=NOW) self.assertEqual(summary["verdict"], "PASS") self.assertEqual(summary["open_unknowns"], 1) self.assertRegex(sealed["bundle_digest"], r"^[0-9a-f]{64}$") def test_digest_mutation_is_detected(self) -> None: sealed = self.sealed() sealed["claims"][0]["statement"] += " mutated" with self.assertRaisesRegex(pstack.ContractError, "bundle_digest:mismatch"): pstack.validate_bundle(sealed, self.contract, now=NOW) def test_process_identity_requires_start_ticks(self) -> None: bundle = valid_bundle() bundle["subject"] = { "kind": "process", "identity": { "host": "wk-01", "boot_id": "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee", "pid": 123, "uid": 1000, }, } self.rejected(bundle, "bundle.subject.identity:shape") def test_selector_digest_is_load_bearing(self) -> None: bundle = valid_bundle() bundle["universe"]["selector"]["pr"] = 99999 self.rejected(bundle, "selector_sha256:mismatch") def test_complete_bundle_requires_count_and_coverage(self) -> None: bundle = valid_bundle() bundle["completeness"] = "complete" bundle["universe"]["expected_count"] = 4 self.rejected(bundle, "count_mismatch") def test_blocking_surface_requires_deadline(self) -> None: bundle = valid_bundle() bundle["surfaces"][0]["source"] = "procfs.stack" bundle["surfaces"][0]["deadline_ms"] = None self.rejected(bundle, "required_for_blocking_surface") def test_complete_bundle_rejects_timeout(self) -> None: bundle = valid_bundle() bundle["completeness"] = "complete" bundle["surfaces"][0]["status"] = "timeout" self.rejected(bundle, "complete_with_incomplete_surfaces") def test_universal_claim_requires_complete_bundle(self) -> None: bundle = valid_bundle() bundle["claims"][0]["scope"] = "universal" self.rejected(bundle, "universal_requires_complete_bundle") def test_absence_requires_negative_control(self) -> None: bundle = valid_bundle() bundle["claims"][1]["negative_control_evidence"] = [] self.rejected(bundle, "absence_requires_negative_control") def test_hypothesis_requires_falsifier(self) -> None: bundle = valid_bundle() bundle["claims"][2]["falsifier"] = None self.rejected(bundle, "invalid_string") def test_fact_cannot_smuggle_confidence(self) -> None: bundle = valid_bundle() bundle["claims"][0]["confidence"] = 0.99 self.rejected(bundle, "fact_must_be_null") def test_evidence_reference_must_resolve(self) -> None: bundle = valid_bundle() bundle["claims"][0]["evidence"] = ["missing.surface"] self.rejected(bundle, "evidence:unresolved") def test_duplicate_surface_ids_fail_closed(self) -> None: bundle = valid_bundle() bundle["surfaces"].append(copy.deepcopy(bundle["surfaces"][0])) self.rejected(bundle, "duplicate_id") def test_timeout_unreaped_helper_is_preserved(self) -> None: bundle = valid_bundle() surface = bundle["surfaces"][0] surface["source"] = "procfs.stack" surface["status"] = "timeout_unreaped_helper" surface["facts"] = {"helper_pid": 9876, "signal_sent": "SIGKILL", "reaped": False} bundle["claims"][0]["statement"] = "The procfs helper exceeded its deadline and remained unreaped." self.assertEqual(self.sealed(bundle)["surfaces"][0]["status"], "timeout_unreaped_helper") def test_open_unknown_requires_probe(self) -> None: bundle = valid_bundle() bundle["unknowns"][0]["probe"] = None self.rejected(bundle, "probe:not_object") def test_stale_bundle_is_rejected(self) -> None: with self.assertRaisesRegex(pstack.ContractError, "observed_at:stale"): pstack.seal_bundle( valid_bundle(), self.contract, now=dt.datetime(2026, 9, 4, 6, 0, tzinfo=dt.timezone.utc), ) def test_diff_refuses_identity_change(self) -> None: before = self.sealed() raw = valid_bundle() raw["subject"]["identity"]["head"] = "d" * 40 after = self.sealed(raw) with self.assertRaisesRegex(pstack.ContractError, "subject_identity_changed"): pstack.diff_bundles(before, after) def test_handoff_is_frontier_not_raw_dump(self) -> None: handoff = pstack.handoff_view(self.sealed()) self.assertEqual(handoff["schema"], "bbi.agent-handoff/v1") self.assertEqual(len(handoff["facts"]), 2) self.assertNotIn("surfaces", handoff) def test_duplicate_json_key_rejected(self) -> None: with tempfile.TemporaryDirectory() as directory: path = Path(directory) / "duplicate.json" path.write_text('{"schema":"a","schema":"b"}', encoding="utf-8") with self.assertRaisesRegex(pstack.ContractError, "duplicate_json_key:schema"): pstack._load_json(path) def test_workgraph_has_one_legal_frontier(self) -> None: graph, _ = pstack._load_json(WORKGRAPH_PATH) summary = pstack.validate_workgraph(graph) self.assertEqual(summary["frontier"], ["PSTACK-000"]) def test_workgraph_rejects_dependency_cycle(self) -> None: graph, _ = pstack._load_json(WORKGRAPH_PATH) graph["tranches"][0]["dependencies"] = ["PSTACK-110"] with self.assertRaisesRegex(pstack.ContractError, "dependency_cycle"): pstack.validate_workgraph(graph) def test_workgraph_rejects_ready_task_with_blocked_dependency(self) -> None: graph, _ = pstack._load_json(WORKGRAPH_PATH) graph["tranches"][1]["status"] = "READY" with self.assertRaisesRegex(pstack.ContractError, "ready_with_unsatisfied_dependencies"): pstack.validate_workgraph(graph) if __name__ == "__main__": unittest.main(verbosity=2)