diff --git a/config/pstack-evidence-contract.json b/config/pstack-evidence-contract.json new file mode 100644 index 0000000000000000000000000000000000000000..c990399e4a342311a1b80ce5fe0caf098e233581 --- /dev/null +++ b/config/pstack-evidence-contract.json @@ -0,0 +1,140 @@ +{ + "apiVersion": "bbi.dev/v1", + "kind": "PStackEvidenceContract", + "metadata": { + "name": "program-causal-truth-plane", + "version": 1 + }, + "spec": { + "schema": "bbi.process-truth/v1", + "maxBundleBytes": 1048576, + "maxSurfaces": 128, + "maxClaims": 64, + "maxUnknowns": 32, + "maxTextChars": 4096, + "maxClockSkewSeconds": 30, + "subjectKinds": [ + "process", + "thread", + "workflow_transaction", + "network_stream", + "storage_transfer", + "alert_delivery" + ], + "requiredIdentity": { + "process": { + "host": "string", + "boot_id": "uuid", + "pid": "positive_int", + "start_ticks": "positive_int", + "uid": "int" + }, + "thread": { + "host": "string", + "boot_id": "uuid", + "pid": "positive_int", + "tid": "positive_int", + "process_start_ticks": "positive_int", + "uid": "int" + }, + "workflow_transaction": { + "repository": "repository", + "pr": "positive_int", + "head": "sha40", + "base": "sha40", + "run_id": "positive_int", + "run_attempt": "positive_int" + }, + "network_stream": { + "source": "string", + "destination": "string", + "protocol": "string", + "flow_id": "string" + }, + "storage_transfer": { + "volume": "string", + "source_node": "string", + "target_node": "string", + "engine": "string", + "transfer_id": "string" + }, + "alert_delivery": { + "alertname": "string", + "fingerprint": "string", + "receiver": "string", + "route_generation": "string" + } + }, + "surfaceStatuses": [ + "ok", + "timeout", + "timeout_unreaped_helper", + "permission_denied", + "not_found", + "io_error", + "unsupported", + "truncated", + "identity_race", + "provider_ambiguous", + "self_error" + ], + "surfaceAuthorities": [ + "kernel", + "provider", + "runtime", + "filesystem", + "network", + "database", + "operator" + ], + "blockingSurfacePatterns": [ + "procfs.stack", + "procfs.wchan", + "procfs.syscall", + "ptrace.*", + "provider.api.*", + "network.stream.*", + "database.scan.*" + ], + "claimKinds": ["fact", "inference", "hypothesis"], + "claimStates": ["observed", "open", "confirmed", "falsified", "superseded"], + "claimMethods": [ + "direct_observation", + "derived_computation", + "causal_inference", + "operator_attestation" + ], + "claimPolarities": ["presence", "absence", "state"], + "claimScopes": ["exact", "bounded", "universal"], + "completenessStates": ["complete", "bounded", "partial"], + "unknownStates": ["open", "answered", "superseded"], + "transitionTypes": [ + "OBSERVE", + "DISCRIMINATE", + "SOURCE_FIXED", + "MERGED", + "REALIZED", + "VERIFIED", + "SOAKED", + "SUPERSEDED", + "CLOSED", + "HYPOTHESIS_FALSIFIED", + "BLOCKER_RECLASSIFIED", + "PUBLISH_SHADOW", + "PUBLISH_APP_ONLY_ADMISSION", + "SHA_FENCED_MERGE", + "RESTORE_REDUNDANCY", + "TRACE_STREAM_FAILURE", + "TRACE_ALERT_DELIVERY", + "OWNER_DECISION" + ], + "maxEvidenceAgeSeconds": { + "process": 300, + "thread": 300, + "workflow_transaction": 1800, + "network_stream": 900, + "storage_transfer": 900, + "alert_delivery": 1800 + } + } +} diff --git a/config/pstack-program-workgraph.json b/config/pstack-program-workgraph.json new file mode 100644 index 0000000000000000000000000000000000000000..3fff616d4b9e4193c7b1dd9cf3b8e1000c7ba4aa --- /dev/null +++ b/config/pstack-program-workgraph.json @@ -0,0 +1,188 @@ +{ + "schema": "bbi.pstack-program-workgraph/v1", + "programme": "authentik-rke2-reliability", + "authority": "https://github.com/Biji-Biji-Initiative/bbi-infrastructure/issues/11515", + "wip": { + "mutation_slots": 2, + "classes": ["ci-control", "runtime-remediation"] + }, + "status_vocabulary": [ + "BLOCKED", + "READY", + "ACTIVE", + "SOURCE_FIXED", + "MERGED", + "REALIZED", + "VERIFIED", + "SOAKED", + "SUPERSEDED", + "CLOSED" + ], + "tranches": [ + { + "id": "PSTACK-000", + "title": "Admit the causal-truth foundation", + "priority": 0, + "owner_role": "independent-reviewer", + "mutation_class": "ci-control", + "status": "READY", + "dependencies": [], + "coordination_key": "pstack-foundation", + "owned_paths": [ + "config/pstack-evidence-contract.json", + "config/pstack-program-workgraph.json", + "docs/architecture/pstack-causal-truth-plane.md", + "docs/runbooks/pstack-agent-loop.md", + "scripts/ops/pstack/pstack_evidence.py", + "scripts/qa/test-pstack-evidence-contract.py" + ], + "first_legal_action": "Run an independent exact-head review and hostile suite.", + "success_receipt": "PSTACK_FOUNDATION_MERGED", + "abort_conditions": [ + "exact head moved", + "hostile test fails", + "contract permits a universal claim on partial evidence" + ], + "non_goals": [ + "No production mutation", + "No new collector", + "No replacement of canonical domain authorities" + ] + }, + { + "id": "PSTACK-110", + "title": "Shadow-adopt bounded Linux process truth", + "priority": 1, + "owner_role": "mode-s-runtime", + "mutation_class": "runtime-remediation", + "status": "BLOCKED", + "dependencies": ["PSTACK-000"], + "coordination_key": "mode-s-process-truth", + "owned_paths": ["internal/ci/mode_s/execution/**"], + "first_legal_action": "Compare existing #11478 envelopes against the shared contract without changing consumers.", + "success_receipt": "PROCESS_TRUTH_SHADOW_AGREEMENT", + "abort_conditions": ["unbounded procfs read", "PID identity race", "helper leak budget exhausted"], + "non_goals": ["No signal or reap behavior change"] + }, + { + "id": "PSTACK-111", + "title": "Prove scheduler-state classification", + "priority": 1, + "owner_role": "mode-s-runtime", + "mutation_class": "runtime-remediation", + "status": "BLOCKED", + "dependencies": ["PSTACK-110"], + "coordination_key": "mode-s-process-truth", + "owned_paths": ["scripts/ops/fastlane/protected-candidate/**", "scripts/qa/test-mode-s-*.py"], + "first_legal_action": "Plant R/S/D/Z fixtures and kill the all-R mutation.", + "success_receipt": "PROCESS_STATE_CLASSIFIER_MUTATION_KILLED", + "abort_conditions": ["fixture does not exercise production parser"], + "non_goals": ["No candidate attribution from a bare PID"] + }, + { + "id": "PSTACK-112", + "title": "Bound every potentially blocking procfs read", + "priority": 0, + "owner_role": "mode-s-runtime", + "mutation_class": "runtime-remediation", + "status": "BLOCKED", + "dependencies": ["PSTACK-110"], + "coordination_key": "mode-s-process-truth", + "owned_paths": ["internal/ci/mode_s/execution/**"], + "first_legal_action": "Run stack/wchan/syscall reads in isolated helpers with per-surface deadlines and a locked leak budget.", + "success_receipt": "PROCFS_BLOCKING_READ_BOUNDED", + "abort_conditions": ["unreaped helper budget exceeded", "collector blocks caller"], + "non_goals": ["No always-on process monitor"] + }, + { + "id": "PSTACK-210", + "title": "Add executable Mode S attribution-hold resolution", + "priority": 1, + "owner_role": "mode-s-lifecycle", + "mutation_class": "ci-control", + "status": "BLOCKED", + "dependencies": ["PSTACK-000", "PSTACK-111", "PSTACK-112"], + "coordination_key": "mode-s-attribution", + "owned_paths": ["internal/ci/mode_s/candidate_lifecycle/**", ".github/workflows/mode-s-*.yml"], + "first_legal_action": "Separate ATTRIBUTION_HOLD from permanent typed-red quarantine.", + "success_receipt": "ATTRIBUTION_RESOLUTION_STATE_MACHINE_VERIFIED", + "abort_conditions": ["typed-red quarantine can be exonerated", "resolution is not bound to original hold digest"], + "non_goals": ["No retrospective rewrite of historical evidence"] + }, + { + "id": "PSTACK-310", + "title": "Emit the admission-to-merge process vector", + "priority": 0, + "owner_role": "ci-control", + "mutation_class": "ci-control", + "status": "BLOCKED", + "dependencies": ["PSTACK-000"], + "coordination_key": "conveyor-02", + "owned_paths": [".github/workflows/policy-admission-*.yml", "scripts/ci/app-only-merge-*.py", "scripts/ops/butler/**"], + "first_legal_action": "Observe one exact PR generation and report the first missing edge.", + "success_receipt": "ADMISSION_PROCESS_VECTOR_EMITTED", + "abort_conditions": ["head moved", "check census incomplete", "trusted-row cardinality ambiguous"], + "non_goals": ["No merge authority in the observer"] + }, + { + "id": "PSTACK-311", + "title": "Repair the first missing admission edge", + "priority": 0, + "owner_role": "ci-control", + "mutation_class": "ci-control", + "status": "BLOCKED", + "dependencies": ["PSTACK-310"], + "coordination_key": "conveyor-02", + "owned_paths": [".github/workflows/policy-admission-*.yml", "scripts/ci/app-only-merge-*.py", "scripts/ops/butler/**"], + "first_legal_action": "Modify only the publisher, reconciler, or merger edge identified by PSTACK-310.", + "success_receipt": "NATURAL_BUTLER_CONVERSION_VERIFIED", + "abort_conditions": ["retrospective pre-merge receipt", "operator performs canary merge", "receipt not exact-head bound"], + "non_goals": ["No new merge controller"] + }, + { + "id": "PSTACK-410", + "title": "Trace sustained stream survival", + "priority": 0, + "owner_role": "underlay-truth", + "mutation_class": "read-only", + "status": "BLOCKED", + "dependencies": ["PSTACK-000"], + "coordination_key": "issue-10080", + "owned_paths": ["scripts/ops/network/**", "scripts/qa/*egress*", "docs/evidence/network/**"], + "first_legal_action": "Capture natural Longhorn, Git, and GHCR streams with route/MTU/TCP/RST identity.", + "success_receipt": "STREAM_FAILURE_EDGE_IDENTIFIED", + "abort_conditions": ["synthetic load risks a crown jewel", "stream identity incomplete"], + "non_goals": ["No timeout or concurrency increase"] + }, + { + "id": "PSTACK-420", + "title": "Trace one critical alert to acknowledgement", + "priority": 0, + "owner_role": "incident-routing", + "mutation_class": "read-only", + "status": "BLOCKED", + "dependencies": ["PSTACK-000"], + "coordination_key": "alert-lifecycle", + "owned_paths": ["platform/monitoring/**", "docs/runbooks/**"], + "first_legal_action": "Follow one firing Longhorn critical through Alertmanager, receiver, external message ID, and acknowledgement.", + "success_receipt": "FIRST_MISSING_ALERT_EDGE_IDENTIFIED", + "abort_conditions": ["alert fingerprint changed", "route census incomplete"], + "non_goals": ["No duplicate alert"] + }, + { + "id": "PSTACK-510", + "title": "Publish evidence-addressed agent handoffs", + "priority": 1, + "owner_role": "agent-control-plane", + "mutation_class": "ci-control", + "status": "BLOCKED", + "dependencies": ["PSTACK-000", "PSTACK-310", "PSTACK-410", "PSTACK-420"], + "coordination_key": "agent-mission-control", + "owned_paths": ["CONTROL.md", "specs/agent-control-plane.md", "bin/gitops-ops"], + "first_legal_action": "Attach a compact frontier handoff and bundle digest to one live programme transition.", + "success_receipt": "FRESH_AGENT_TAKES_CORRECT_FIRST_MOVE", + "abort_conditions": ["handoff contains uncited facts", "evidence expired", "raw secret-bearing payload embedded"], + "non_goals": ["No second work tracker"] + } + ] +} diff --git a/docs/architecture/pstack-causal-truth-plane.md b/docs/architecture/pstack-causal-truth-plane.md new file mode 100644 index 0000000000000000000000000000000000000000..032a85baf9cbee5dea5e0730b386b3e5f50e000c --- /dev/null +++ b/docs/architecture/pstack-causal-truth-plane.md @@ -0,0 +1,248 @@ +# PStack Causal Truth Plane + +**Status:** foundation extension +**Scope:** Mode S, CI admission/merge, Fastlane runners, Longhorn recovery, alert delivery, and agent handoffs +**Principle:** inspect the live mechanism at the layer where the claim lives + +## Why this exists + +`pstack` replaces a story about a process with the process's actual thread state. The reliability programme repeatedly needs the same move at larger scales: + +- a green workflow is not proof that its intended transition happened; +- a firing alert is not proof that an owner received and acknowledged it; +- a Longhorn volume marked healthy is not proof every database page is readable; +- low node load is not proof a multi-minute TCP stream can survive; +- a clean PR is not proof its finalizer receipts exist or the writer is legally admitted; +- a PID is not a process identity unless start time, UID, boot ID, namespace, and cgroup still match. + +The common defect is **proxy substitution**: observing an adjacent signal and promoting it into a claim about the mechanism. + +> A process is any live state machine. Read its state directly, bind the observation to an immutable identity, bound the read, and keep inference separate from evidence. + +The Causal Truth Plane is not another dashboard or database. It is a shared evidence protocol for existing collectors and authorities. + +## Five laws + +### 1. State before story + +Capture the mechanism before interpreting it. Logs and summaries are direct evidence only when the claim is about logs or summaries. + +| Claim | Direct surface | +|---|---| +| process is blocked | thread state, `wchan`, stack/syscall, scheduler state | +| PR is merge-ready | exact gate generation, finalizer receipts, review/hold state, lane, writer admission | +| rebuild fails in transport | source/target identity, TCP state, engine terminal | +| alert reached an owner | route, receiver result, external message ID, acknowledgement | +| database is readable | logical dump or page-integrity proof, not replica robustness alone | + +### 2. Identity before state + +Minimum subject identities: + +- process: host, boot ID, PID, start ticks, UID; +- thread: process identity plus TID; +- workflow transaction: repository, PR, exact head/base, run ID, attempt; +- network stream: source, destination, protocol, flow ID; +- storage transfer: volume, engine, source node, target node, transfer ID; +- alert delivery: alert fingerprint, route generation, receiver. + +Collectors re-read identity around slow surfaces. A mismatch is `identity_race`, never a best-effort attachment to a newer subject. + +### 3. Bounded observation before completeness + +Every potentially blocking surface has a per-surface deadline and every bundle has a whole-collection deadline. Timeout is evidence. + +For procfs, `/proc//stack`, `wchan`, and `syscall` can block on the very D-state task being diagnosed. A timer around synchronous `open().read()` is not a deadline. Use an isolated helper. If the helper cannot be reaped, emit `timeout_unreaped_helper`, journal its identity, consume a bounded leak budget, and continue with partial evidence. + +A `complete` bundle contains only successful, untruncated surfaces and a complete declared universe. A `bounded` or `partial` bundle cannot support universal claims. + +### 4. Falsifiers before fixes + +Every hypothesis states what observation would kill it before mutation. Prefer the cheapest safe probe that separates important branches. + +### 5. Receipts before closure + +`SOURCE_FIXED`, `MERGED`, `REALIZED`, `VERIFIED`, `SOAKED`, and `CLOSED` are distinct transitions. One layer's receipt cannot substitute for another. + +## Architecture + +```text +existing collectors and provider readers + │ typed surfaces, never prose + ▼ +PStack Evidence Envelope + identity · clock · universe · bounds · surfaces · claims · unknowns · next transition + │ + ├── canonical seal + SHA-256 digest + ├── closed-world validator + ├── deterministic before/after diff + └── compact agent-frontier handoff + ▼ +Mode S · Butler/finalizers · watchdogs · storage recovery · alert lifecycle · agents +``` + +`scripts/ops/pstack/pstack_evidence.py` does not collect data. Existing domain readers remain adapters. The shared layer governs identity, freshness, completeness, claims, unknowns, transitions, sealing, and handoff. + +## Evidence envelope + +Every envelope contains: + +1. **Subject** — immutable identity of the observed process or transaction. +2. **Collector** — name, version, and exact source commit. +3. **Clock** — wall time, monotonic observation point, and bundle deadline. +4. **Universe** — selector, expected/observed count, termination reason, and selector digest. +5. **Bounds** — explicit caps. +6. **Surfaces** — authority, typed status, deadline, digest, bytes, truncation, and structured facts. +7. **Claims** — facts, inferences, or hypotheses with evidence bindings. +8. **Unknowns** — answerable questions, blocked transitions, owners, and discriminating probes. +9. **Next transition** — one legal frontier action, success receipt, and abort conditions. +10. **Bundle digest** — canonical SHA-256 over every field except the digest itself. + +Large or sensitive payloads remain in artifacts referenced by URI and digest. + +## Existing foundations to strengthen + +### Linux process truth + +Converge the Mode S process collector and shard-executor survivor classifier around shared identity and typed surfaces. The classifier requires planted `R`, `S`, `D`, and `Z` fixtures. Mapping every state to `R` must fail. + +Capture identity twice: + +```text +before slow reads: boot_id, pid, start_ticks, uid, namespace, cgroup +slow bounded surfaces +before emit: boot_id, pid, start_ticks, uid, namespace, cgroup +``` + +Any change invalidates attribution. + +### Mode S lifecycle + +A candidate is a distributed process: + +```text +claim -> build -> execute -> terminal classify -> hold/quarantine/release + -> journal -> provider receipt -> consumer acceptance +``` + +Every edge is `not_started`, `running`, `success`, a typed hold/error, or `unknown`. The observer reports the first missing edge. + +`ambiguous_failure` must not reuse permanent typed-red quarantine. It is an attribution hold with an executable resolution: + +```text +ATTRIBUTION_HOLD + ├── CONFIRMED_RED -> TYPED_RED_QUARANTINE + └── EXONERATED -> RELEASED exactly once +``` + +### Admission to merge + +For each PR, capture the full readiness vector: + +```text +exact head/base +canonical Policy Guards generation +policy-admission shadow cardinality and generation +app-only admission cardinality and expiry binding +review-thread and hold-label state +Butler lane classification +writer credential eligibility +merge commitment +B/H/C/tree/actor receipt +post-merge sentinel +``` + +Example frontier: + +```text +policy_guards success generation X +shadow missing +app_only_admission blocked_on_shadow +writer_admission not_legal +next_transition PUBLISH_SHADOW +``` + +The merger re-reads the same vector immediately before the SHA-fenced write. Observation explains authority; it never grants it. + +### Harness liveness + +Measure service rate by lane: + +- routine eligible vs routine converted; +- controlled eligible vs controlled dispatched/merged; +- L3 awaiting owner decision. + +Twenty controlled PRs and no routine PRs must not trigger a routine-throughput alarm. + +### Storage and transport + +Model a Longhorn rebuild as a storage transfer linked to a network stream: + +```text +volume/replica and instance-manager identity +route, interface, MTU and PMTU +TCP bytes, duration, retransmits, RTOs, zero-window and resets +RST endpoint +NIC/veth drops, conntrack pressure, Cilium verdict +engine progress and terminal +``` + +CPU load is a hypothesis surface, not the stream state. + +### Alert delivery + +Trace the entire process: + +```text +Prometheus firing -> Alertmanager received -> silence/inhibition -> route +-> receiver response -> external message ID -> acknowledgement -> repair -> recovery +``` + +A firing alert proves only the first edge. + +## Agent loop + +```text +CAPTURE -> BIND IDENTITY -> CLASSIFY UNKNOWN -> CHOOSE FALSIFIER + -> MUTATE -> CAPTURE AGAIN -> DIFF -> ISSUE RECEIPT +``` + +Before mutation require one sealed bundle, typed completeness, facts separated from hypotheses, blocking unknowns with probes, and one next transition. After mutation produce a second bundle for the same subject, deterministic diff, success or typed-failure receipt, updated hypothesis state, and compact handoff. + +## Rollout + +### P0-A — envelope and validator + +Land the contract, CLI, hostile tests, architecture, runbook, and work graph. Use an existing CI shard; add no required context. + +### P0-B — procfs adapter + +Extract identity/deadline primitives and typed procfs surfaces, then kill all-`R`, stale-PID, missing-deadline, and helper-leak mutations. + +### P0-C — Mode S and admission adapters + +Emit first-missing-edge vectors for one real candidate and one real PR. Require zero unexplained disagreement with provider behavior before consumers switch. + +### P0-D — transport and alert adapters + +Instrument natural Longhorn/Git/GHCR streams and trace one live critical alert end-to-end. Do not change timeouts, concurrency, routing, or production workloads in this phase. + +### P1 — evidence-addressed handoffs + +Agent Mail carries a compact handoff plus bundle digest. Beads records canonical URI, observation time, expiry, and next transition. + +### P2 — consumer enforcement + +Only after shadow agreement should Mode S reject stale evidence, Butler consume readiness vectors, watchdogs consume lane-specific service rates, and closure tooling require named receipts. + +## Acceptance metrics + +- time to first discriminating observation decreases; +- proxy-substitution retractions decrease; +- repeated investigation of settled questions decreases; +- open unknowns without an owner and probe reach zero; +- stale or unbound handoffs reach zero; +- transitions per agent-hour increase; +- tokens spent narrating status decrease. + +The adversarial acceptance test gives a new agent only the compact handoff and artifact access. It should identify the first missing transition, challenge the highest-impact hypothesis, and take the next legal action without rereading the programme history. diff --git a/docs/runbooks/pstack-agent-loop.md b/docs/runbooks/pstack-agent-loop.md new file mode 100644 index 0000000000000000000000000000000000000000..1a2a6f1120468112fcec704f3c9771fd9ba6af3e --- /dev/null +++ b/docs/runbooks/pstack-agent-loop.md @@ -0,0 +1,248 @@ +# PStack Agent Loop + +Use this runbook before an agent diagnoses, mutates, hands off, or closes a Mode S, CI, runner, storage, identity, or reliability incident. + +## Fast path + +```bash +TOOL=scripts/ops/pstack/pstack_evidence.py +CONTRACT=config/pstack-evidence-contract.json + +python3 "$TOOL" seal \ + --contract "$CONTRACT" \ + --input var/process-truth/current.raw.json \ + --output var/process-truth/current.json + +python3 "$TOOL" verify \ + --contract "$CONTRACT" \ + --input var/process-truth/current.json + +python3 "$TOOL" handoff \ + --contract "$CONTRACT" \ + --input var/process-truth/current.json \ + --output var/process-truth/handoff.json +``` + +After a mutation: + +```bash +python3 "$TOOL" seal \ + --contract "$CONTRACT" \ + --input var/process-truth/after.raw.json \ + --output var/process-truth/after.json + +python3 "$TOOL" diff \ + --contract "$CONTRACT" \ + --before var/process-truth/current.json \ + --after var/process-truth/after.json \ + --output var/process-truth/delta.json +``` + +Never edit a sealed bundle. Change raw input and seal again. + +## 1. Choose one stable subject + +| Work | Subject kind | Required identity | +|---|---|---| +| process | `process` | host, boot ID, PID, start ticks, UID | +| thread | `thread` | process identity plus TID | +| PR admission/merge | `workflow_transaction` | repo, PR, exact head/base, run ID/attempt | +| Git/GHCR/TCP transfer | `network_stream` | source, destination, protocol, flow ID | +| Longhorn rebuild | `storage_transfer` | volume, nodes, engine, transfer ID | +| alert routing | `alert_delivery` | alert fingerprint, receiver, route generation | + +A changed head, start time, boot ID, transfer ID, or alert fingerprint means a new subject. Do not compare it as one continuing process. + +## 2. Declare the observation universe + +Before collecting, record: + +```text +selector +selector SHA-256 +expected count or null with reason +observed count +termination reason +coverage-evidence surfaces +``` + +A complete bundle cannot self-declare completeness without a complete universe. + +## 3. Collect direct surfaces + +Every surface records: + +```text +stable ID +source +owning authority +typed status +duration +per-surface deadline +artifact URI +digest +bytes +truncation +structured facts +``` + +Typed statuses: + +```text +ok + timeout + timeout_unreaped_helper + permission_denied + not_found + io_error + unsupported + truncated + identity_race + provider_ambiguous + self_error +``` + +Missing output is never clean output. + +## 4. Process-collection safety + +Read cheap identity surfaces first: + +```text +/proc/sys/kernel/random/boot_id +/proc//stat +/proc//status +/proc//ns/* +/proc//cgroup +``` + +Run potentially blocking surfaces in isolated helpers: + +```text +/proc//task//stack +/proc//task//wchan +/proc//task//syscall +ptrace-based collectors +``` + +If a helper times out and cannot be reaped: + +1. emit `timeout_unreaped_helper`; +2. record helper PID/start ticks, target identity, signal, and timestamp; +3. journal the helper under a lock-protected bounded leak budget; +4. refuse unbounded replacement helpers; +5. continue with partial evidence; +6. never report the target as clean. + +Re-read target identity before sealing. Any mismatch is `identity_race`. + +## 5. Write claims honestly + +### Fact + +A fact is direct or deterministically derived. Confidence must be `null`. + +### Absence fact + +An absence requires negative-control evidence proving the observer can detect a known-positive or planted signal. + +Bad: + +```text +No EIO entries were found; therefore the disk is healthy. +``` + +Better: + +```text +The bounded kernel-log census found no matching EIO signatures, and the same parser detected the planted control signature. This does not prove all database pages are readable. +``` + +### Inference or hypothesis + +It requires confidence and a falsifier. Preserve it as `falsified` when killed; do not erase it. + +## 6. Name unknowns + +One unknown contains: + +- one answerable question; +- blocked transitions; +- one owner; +- the cheapest safe probe; +- expected discriminator; +- deadline; +- abort conditions. + +## 7. Choose one next transition + +A bundle has exactly one frontier transition, such as: + +```text +DISCRIMINATE +PUBLISH_SHADOW +PUBLISH_APP_ONLY_ADMISSION +SHA_FENCED_MERGE +TRACE_STREAM_FAILURE +TRACE_ALERT_DELIVERY +RESTORE_REDUNDANCY +SOURCE_FIXED +REALIZED +VERIFIED +CLOSED +``` + +Define the success receipt before work starts. Without the named receipt, the transition did not complete. + +## Programme adapters + +### Mode S candidate + +```text +claim -> build -> execute -> terminal classify -> hold/quarantine/release + -> journal -> provider receipt -> consumer acceptance +``` + +Report the first missing edge. Attribution hold and permanent red quarantine are distinct states. + +### PR admission and merge + +Capture exact head/base, Policy Guards generation, shadow row, app-only admission, review threads/holds, classifier lane, writer eligibility, merge receipt, and post-merge sentinel. + +“Checks green” is never terminal. + +### Longhorn and network + +Link storage-transfer and network-stream bundles. Capture natural transfers while crown jewels are exposed; do not generate synthetic multi-gigabyte load. CPU/load is a hypothesis surface. + +### Alert delivery + +Trace Prometheus through acknowledgement. A firing alert proves only one edge. + +## Handoff discipline + +Send the compact handoff plus sealed bundle URI and digest. A receiver should not have to trust prose. + +A valid handoff answers: + +1. What exact subject was observed? +2. How fresh and complete is the evidence? +3. Which facts are direct? +4. Which hypotheses remain open, and what kills them? +5. Which unknowns block action, and who owns each probe? +6. What is the one next legal transition? +7. What receipt proves it happened? + +## Closure discipline + +| Claim | Required receipt | +|---|---| +| source fixed | tests/mutation proof at exact source head | +| carrier delivered | merged commit on protected main | +| runtime realized | live identity at intended revision | +| data safe | store-appropriate restore or integrity proof | +| conveyor healthy | natural finalizer + writer conversion and sentinel | +| alerting effective | external delivery plus acknowledgement | +| incident closed | causal condition absent and recurrence control realized | + +A stale green check, merged PR, replica count, or process heartbeat cannot stand in for another layer’s receipt. diff --git a/scripts/ops/pstack/pstack_evidence.py b/scripts/ops/pstack/pstack_evidence.py new file mode 100755 index 0000000000000000000000000000000000000000..9a03c63cdcd55264bf05e89a1d79d5ab4dc0b012 --- /dev/null +++ b/scripts/ops/pstack/pstack_evidence.py @@ -0,0 +1,865 @@ +#!/usr/bin/env python3 +"""Seal, verify, diff, and compact BBI PStack evidence envelopes. + +Collectors remain domain-owned. This module enforces a shared evidence boundary: +identity, freshness, observation-universe completeness, typed surfaces, explicit +facts/hypotheses/unknowns, canonical digests, and one next legal transition. +""" + +from __future__ import annotations + +import argparse +import copy +import datetime as dt +import fnmatch +import hashlib +import json +import math +import re +import sys +from pathlib import Path +from typing import Any, Iterable + + +class ContractError(ValueError): + """A fail-closed contract violation.""" + + +def _no_duplicate_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise ContractError(f"duplicate_json_key:{key}") + result[key] = value + return result + + +def _load_json(path: Path, *, max_bytes: int | None = None) -> tuple[dict[str, Any], bytes]: + try: + raw = path.read_bytes() + except OSError as exc: + raise ContractError(f"unreadable_json:{path}") from exc + if not raw: + raise ContractError(f"empty_json:{path}") + if max_bytes is not None and len(raw) > max_bytes: + raise ContractError(f"json_size_exceeded:{len(raw)}>{max_bytes}") + try: + value = json.loads(raw, object_pairs_hook=_no_duplicate_object) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + raise ContractError(f"invalid_json:{path}") from exc + if not isinstance(value, dict): + raise ContractError(f"json_root_not_object:{path}") + return value, raw + + +def _exact_keys(value: Any, expected: Iterable[str], field: str) -> dict[str, Any]: + if not isinstance(value, dict): + raise ContractError(f"{field}:not_object") + expected_set = set(expected) + actual = set(value) + if actual != expected_set: + raise ContractError( + f"{field}:shape:missing={sorted(expected_set - actual)}:extra={sorted(actual - expected_set)}" + ) + return value + + +def _list(value: Any, field: str, *, max_items: int | None = None) -> list[Any]: + if not isinstance(value, list): + raise ContractError(f"{field}:not_list") + if max_items is not None and len(value) > max_items: + raise ContractError(f"{field}:too_many:{len(value)}>{max_items}") + return value + + +def _string(value: Any, field: str, *, max_chars: int = 4096) -> str: + if ( + not isinstance(value, str) + or not value.strip() + or len(value) > max_chars + or "\x00" in value + ): + raise ContractError(f"{field}:invalid_string") + return value + + +def _optional_string(value: Any, field: str, *, max_chars: int = 4096) -> str | None: + if value is None: + return None + return _string(value, field, max_chars=max_chars) + + +def _unique_strings(value: Any, field: str, *, max_items: int | None = None) -> list[str]: + rows = _list(value, field, max_items=max_items) + if any(not isinstance(row, str) or not row.strip() or "\x00" in row for row in rows): + raise ContractError(f"{field}:invalid_string") + if len(rows) != len(set(rows)): + raise ContractError(f"{field}:duplicates") + return rows + + +def _int(value: Any, field: str, *, minimum: int = 0, maximum: int | None = None) -> int: + if isinstance(value, bool) or not isinstance(value, int) or value < minimum: + raise ContractError(f"{field}:invalid_int") + if maximum is not None and value > maximum: + raise ContractError(f"{field}:too_large:{value}>{maximum}") + return value + + +def _number(value: Any, field: str, *, minimum: float, maximum: float) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ContractError(f"{field}:not_number") + number = float(value) + if not math.isfinite(number) or not minimum <= number <= maximum: + raise ContractError(f"{field}:out_of_range") + return number + + +def _bool(value: Any, field: str) -> bool: + if not isinstance(value, bool): + raise ContractError(f"{field}:not_bool") + return value + + +def _one_of(value: Any, allowed: set[str], field: str) -> str: + candidate = _string(value, field, max_chars=128) + if candidate not in allowed: + raise ContractError(f"{field}:unsupported:{candidate}") + return candidate + + +def _hex(value: Any, length: int, field: str) -> str: + candidate = _string(value, field, max_chars=length) + if len(candidate) != length or re.fullmatch(r"[0-9a-f]+", candidate) is None: + raise ContractError(f"{field}:invalid_hex{length}") + return candidate + + +def _time(value: Any, field: str) -> dt.datetime: + candidate = _string(value, field, max_chars=64) + if not candidate.endswith("Z"): + raise ContractError(f"{field}:must_be_utc_z") + try: + parsed = dt.datetime.fromisoformat(candidate[:-1] + "+00:00") + except ValueError as exc: + raise ContractError(f"{field}:invalid_timestamp") from exc + if parsed.tzinfo is None or parsed.utcoffset() != dt.timedelta(0): + raise ContractError(f"{field}:must_be_utc") + return parsed + + +def _parse_now(value: str | None) -> dt.datetime: + return dt.datetime.now(dt.timezone.utc) if value is None else _time(value, "now") + + +def _validate_identity_value(value: Any, kind: str, field: str) -> None: + if kind == "string": + _string(value, field, max_chars=512) + elif kind == "int": + _int(value, field) + elif kind == "positive_int": + _int(value, field, minimum=1) + elif kind == "sha40": + _hex(value, 40, field) + elif kind == "sha64": + _hex(value, 64, field) + elif kind == "repository": + candidate = _string(value, field, max_chars=256) + if re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", candidate) is None: + raise ContractError(f"{field}:invalid_repository") + elif kind == "uuid": + candidate = _string(value, field, max_chars=64) + if re.fullmatch( + r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}", + candidate, + ) is None: + raise ContractError(f"{field}:invalid_uuid") + else: + raise ContractError(f"contract:unknown_identity_type:{kind}") + + +def _canonical_without_digest(bundle: dict[str, Any]) -> bytes: + payload = copy.deepcopy(bundle) + payload.pop("bundle_digest", None) + try: + text = json.dumps( + payload, + ensure_ascii=False, + allow_nan=False, + sort_keys=True, + separators=(",", ":"), + ) + except (TypeError, ValueError) as exc: + raise ContractError("bundle:not_canonicalizable") from exc + return text.encode("utf-8") + + +def bundle_digest(bundle: dict[str, Any]) -> str: + return hashlib.sha256(_canonical_without_digest(bundle)).hexdigest() + + +def load_contract(path: Path) -> dict[str, Any]: + contract, _ = _load_json(path, max_bytes=262_144) + _exact_keys(contract, {"apiVersion", "kind", "metadata", "spec"}, "contract") + if contract["apiVersion"] != "bbi.dev/v1" or contract["kind"] != "PStackEvidenceContract": + raise ContractError("contract:identity") + metadata = _exact_keys(contract["metadata"], {"name", "version"}, "contract.metadata") + _string(metadata["name"], "contract.metadata.name", max_chars=128) + _int(metadata["version"], "contract.metadata.version", minimum=1) + + spec = _exact_keys( + contract["spec"], + { + "schema", + "maxBundleBytes", + "maxSurfaces", + "maxClaims", + "maxUnknowns", + "maxTextChars", + "maxClockSkewSeconds", + "subjectKinds", + "requiredIdentity", + "surfaceStatuses", + "surfaceAuthorities", + "blockingSurfacePatterns", + "claimKinds", + "claimStates", + "claimMethods", + "claimPolarities", + "claimScopes", + "completenessStates", + "unknownStates", + "transitionTypes", + "maxEvidenceAgeSeconds", + }, + "contract.spec", + ) + _string(spec["schema"], "contract.spec.schema", max_chars=128) + for key in ( + "maxBundleBytes", + "maxSurfaces", + "maxClaims", + "maxUnknowns", + "maxTextChars", + "maxClockSkewSeconds", + ): + _int(spec[key], f"contract.spec.{key}", minimum=1) + + subject_kinds = set(_unique_strings(spec["subjectKinds"], "contract.spec.subjectKinds")) + identity_contract = spec["requiredIdentity"] + if not isinstance(identity_contract, dict) or set(identity_contract) != subject_kinds: + raise ContractError("contract.spec.requiredIdentity:shape") + valid_identity_types = {"string", "int", "positive_int", "sha40", "sha64", "repository", "uuid"} + for subject_kind, fields in identity_contract.items(): + if not isinstance(fields, dict) or not fields: + raise ContractError(f"contract.spec.requiredIdentity.{subject_kind}:invalid") + for name, type_name in fields.items(): + _string(name, f"contract.identity.{subject_kind}.field", max_chars=128) + _one_of(type_name, valid_identity_types, f"contract.identity.{subject_kind}.{name}") + + for key in ( + "surfaceStatuses", + "surfaceAuthorities", + "blockingSurfacePatterns", + "claimKinds", + "claimStates", + "claimMethods", + "claimPolarities", + "claimScopes", + "completenessStates", + "unknownStates", + "transitionTypes", + ): + if not _unique_strings(spec[key], f"contract.spec.{key}"): + raise ContractError(f"contract.spec.{key}:empty") + + ages = spec["maxEvidenceAgeSeconds"] + if not isinstance(ages, dict) or set(ages) != subject_kinds: + raise ContractError("contract.spec.maxEvidenceAgeSeconds:shape") + for subject_kind, seconds in ages.items(): + _int(seconds, f"contract.spec.maxEvidenceAgeSeconds.{subject_kind}", minimum=1) + return contract + + +def _matches_blocking(source: str, patterns: list[str]) -> bool: + return any(fnmatch.fnmatchcase(source, pattern) for pattern in patterns) + + +def _selector_digest(selector: dict[str, Any]) -> str: + return hashlib.sha256( + json.dumps(selector, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + ).hexdigest() + + +def validate_bundle( + bundle: dict[str, Any], + contract: dict[str, Any], + *, + now: dt.datetime | None = None, + require_digest: bool = True, + raw_size: int | None = None, +) -> dict[str, Any]: + spec = contract["spec"] + if raw_size is not None and raw_size > spec["maxBundleBytes"]: + raise ContractError(f"bundle:size:{raw_size}>{spec['maxBundleBytes']}") + + _exact_keys( + bundle, + { + "schema", + "bundle_id", + "program", + "subject", + "collector", + "clock", + "universe", + "completeness", + "bounds", + "surfaces", + "claims", + "unknowns", + "next_transition", + "bundle_digest", + }, + "bundle", + ) + if bundle["schema"] != spec["schema"]: + raise ContractError("bundle:schema") + _string(bundle["bundle_id"], "bundle.bundle_id", max_chars=128) + _string(bundle["program"], "bundle.program", max_chars=128) + + subject = _exact_keys(bundle["subject"], {"kind", "identity"}, "bundle.subject") + subject_kind = _one_of(subject["kind"], set(spec["subjectKinds"]), "bundle.subject.kind") + identity_contract = spec["requiredIdentity"][subject_kind] + identity = _exact_keys(subject["identity"], set(identity_contract), "bundle.subject.identity") + for field, type_name in identity_contract.items(): + _validate_identity_value(identity[field], type_name, f"bundle.subject.identity.{field}") + + collector = _exact_keys(bundle["collector"], {"name", "version", "git_sha"}, "bundle.collector") + _string(collector["name"], "bundle.collector.name", max_chars=128) + _string(collector["version"], "bundle.collector.version", max_chars=64) + _hex(collector["git_sha"], 40, "bundle.collector.git_sha") + + clock = _exact_keys(bundle["clock"], {"observed_at", "monotonic_ns", "deadline_ms"}, "bundle.clock") + observed_at = _time(clock["observed_at"], "bundle.clock.observed_at") + _int(clock["monotonic_ns"], "bundle.clock.monotonic_ns") + _int(clock["deadline_ms"], "bundle.clock.deadline_ms", minimum=1) + effective_now = now or dt.datetime.now(dt.timezone.utc) + if effective_now.tzinfo is None: + effective_now = effective_now.replace(tzinfo=dt.timezone.utc) + if observed_at > effective_now + dt.timedelta(seconds=spec["maxClockSkewSeconds"]): + raise ContractError("bundle.clock.observed_at:future") + max_age = dt.timedelta(seconds=spec["maxEvidenceAgeSeconds"][subject_kind]) + if effective_now > observed_at + max_age: + raise ContractError("bundle.clock.observed_at:stale") + + universe = _exact_keys( + bundle["universe"], + {"selector", "selector_sha256", "expected_count", "observed_count", "terminated_by", "coverage_evidence"}, + "bundle.universe", + ) + selector = universe["selector"] + if not isinstance(selector, dict) or not selector: + raise ContractError("bundle.universe.selector:invalid") + _hex(universe["selector_sha256"], 64, "bundle.universe.selector_sha256") + if universe["selector_sha256"] != _selector_digest(selector): + raise ContractError("bundle.universe.selector_sha256:mismatch") + if universe["expected_count"] is not None: + _int(universe["expected_count"], "bundle.universe.expected_count") + _int(universe["observed_count"], "bundle.universe.observed_count") + _one_of( + universe["terminated_by"], + {"natural_end", "declared_cap", "deadline", "provider_pagination_end", "identity_race", "self_error"}, + "bundle.universe.terminated_by", + ) + coverage_evidence = _unique_strings( + universe["coverage_evidence"], "bundle.universe.coverage_evidence", max_items=spec["maxSurfaces"] + ) + + completeness = _one_of(bundle["completeness"], set(spec["completenessStates"]), "bundle.completeness") + bounds = _exact_keys( + bundle["bounds"], + {"surface_cap", "claim_cap", "unknown_cap", "stdout_byte_cap", "stderr_byte_cap"}, + "bundle.bounds", + ) + for bundle_key, contract_key in { + "surface_cap": "maxSurfaces", + "claim_cap": "maxClaims", + "unknown_cap": "maxUnknowns", + }.items(): + _int(bounds[bundle_key], f"bundle.bounds.{bundle_key}", minimum=1, maximum=spec[contract_key]) + for key in ("stdout_byte_cap", "stderr_byte_cap"): + _int(bounds[key], f"bundle.bounds.{key}", minimum=1, maximum=spec["maxBundleBytes"]) + + surfaces = _list(bundle["surfaces"], "bundle.surfaces", max_items=spec["maxSurfaces"]) + if len(surfaces) > bounds["surface_cap"]: + raise ContractError("bundle.surfaces:declared_cap_exceeded") + surface_by_id: dict[str, dict[str, Any]] = {} + for index, row in enumerate(surfaces): + field = f"bundle.surfaces[{index}]" + surface = _exact_keys( + row, + {"id", "source", "authority", "status", "duration_ms", "deadline_ms", "artifact", "digest", "bytes", "truncated", "facts"}, + field, + ) + surface_id = _string(surface["id"], f"{field}.id", max_chars=128) + if surface_id in surface_by_id: + raise ContractError(f"bundle.surfaces:duplicate_id:{surface_id}") + source = _string(surface["source"], f"{field}.source", max_chars=256) + _one_of(surface["authority"], set(spec["surfaceAuthorities"]), f"{field}.authority") + status = _one_of(surface["status"], set(spec["surfaceStatuses"]), f"{field}.status") + _int(surface["duration_ms"], f"{field}.duration_ms") + deadline = surface["deadline_ms"] + if deadline is not None: + _int(deadline, f"{field}.deadline_ms", minimum=1) + if _matches_blocking(source, spec["blockingSurfacePatterns"]) and deadline is None: + raise ContractError(f"{field}.deadline_ms:required_for_blocking_surface") + _optional_string(surface["artifact"], f"{field}.artifact", max_chars=2048) + _hex(surface["digest"], 64, f"{field}.digest") + _int(surface["bytes"], f"{field}.bytes") + truncated = _bool(surface["truncated"], f"{field}.truncated") + if status == "ok" and truncated: + raise ContractError(f"{field}:ok_cannot_be_truncated") + if status == "truncated" and not truncated: + raise ContractError(f"{field}:truncated_status_requires_flag") + if not isinstance(surface["facts"], dict): + raise ContractError(f"{field}.facts:not_object") + surface_by_id[surface_id] = surface + + unresolved_coverage = sorted(set(coverage_evidence) - set(surface_by_id)) + if unresolved_coverage: + raise ContractError(f"bundle.universe.coverage_evidence:unresolved:{unresolved_coverage}") + + if completeness == "complete": + if universe["expected_count"] is None: + raise ContractError("bundle.completeness:complete_requires_expected_count") + if universe["expected_count"] != universe["observed_count"]: + raise ContractError("bundle.completeness:count_mismatch") + if universe["terminated_by"] not in {"natural_end", "provider_pagination_end"}: + raise ContractError("bundle.completeness:invalid_termination") + if not coverage_evidence: + raise ContractError("bundle.completeness:coverage_evidence_required") + bad = [sid for sid, row in surface_by_id.items() if row["status"] != "ok" or row["truncated"]] + if bad: + raise ContractError(f"bundle.completeness:complete_with_incomplete_surfaces:{sorted(bad)}") + + claims = _list(bundle["claims"], "bundle.claims", max_items=spec["maxClaims"]) + if len(claims) > bounds["claim_cap"]: + raise ContractError("bundle.claims:declared_cap_exceeded") + claim_by_id: dict[str, dict[str, Any]] = {} + universal_claims = 0 + hypothesis_counts: dict[str, int] = {} + for index, row in enumerate(claims): + field = f"bundle.claims[{index}]" + claim = _exact_keys( + row, + {"id", "kind", "state", "method", "polarity", "scope", "statement", "evidence", "negative_control_evidence", "confidence", "falsifier", "expires_at", "supersedes"}, + field, + ) + claim_id = _string(claim["id"], f"{field}.id", max_chars=128) + if claim_id in claim_by_id: + raise ContractError(f"bundle.claims:duplicate_id:{claim_id}") + kind = _one_of(claim["kind"], set(spec["claimKinds"]), f"{field}.kind") + state = _one_of(claim["state"], set(spec["claimStates"]), f"{field}.state") + method = _one_of(claim["method"], set(spec["claimMethods"]), f"{field}.method") + polarity = _one_of(claim["polarity"], set(spec["claimPolarities"]), f"{field}.polarity") + scope = _one_of(claim["scope"], set(spec["claimScopes"]), f"{field}.scope") + _string(claim["statement"], f"{field}.statement", max_chars=spec["maxTextChars"]) + evidence = _unique_strings(claim["evidence"], f"{field}.evidence") + if not evidence: + raise ContractError(f"{field}.evidence:empty") + unresolved = sorted(set(evidence) - set(surface_by_id)) + if unresolved: + raise ContractError(f"{field}.evidence:unresolved:{unresolved}") + controls = _unique_strings(claim["negative_control_evidence"], f"{field}.negative_control_evidence") + unresolved_controls = sorted(set(controls) - set(surface_by_id)) + if unresolved_controls: + raise ContractError(f"{field}.negative_control_evidence:unresolved:{unresolved_controls}") + if polarity == "absence" and not controls: + raise ContractError(f"{field}:absence_requires_negative_control") + if kind == "fact": + if state != "observed": + raise ContractError(f"{field}.state:fact_must_be_observed") + if claim["confidence"] is not None: + raise ContractError(f"{field}.confidence:fact_must_be_null") + if method not in {"direct_observation", "derived_computation"}: + raise ContractError(f"{field}.method:fact_not_direct_or_derived") + _optional_string(claim["falsifier"], f"{field}.falsifier", max_chars=spec["maxTextChars"]) + else: + _number(claim["confidence"], f"{field}.confidence", minimum=0.0, maximum=1.0) + _string(claim["falsifier"], f"{field}.falsifier", max_chars=spec["maxTextChars"]) + hypothesis_counts[state] = hypothesis_counts.get(state, 0) + 1 + expires_at = _time(claim["expires_at"], f"{field}.expires_at") + if expires_at <= observed_at: + raise ContractError(f"{field}.expires_at:not_after_observation") + if expires_at > observed_at + max_age: + raise ContractError(f"{field}.expires_at:exceeds_subject_freshness") + if effective_now > expires_at: + raise ContractError(f"{field}:expired") + supersedes = _unique_strings(claim["supersedes"], f"{field}.supersedes") + if claim_id in supersedes: + raise ContractError(f"{field}.supersedes:self") + if scope == "universal": + universal_claims += 1 + if completeness != "complete": + raise ContractError(f"{field}:universal_requires_complete_bundle") + bad = sorted(sid for sid in evidence + controls if surface_by_id[sid]["status"] != "ok") + if bad: + raise ContractError(f"{field}:universal_uses_non_ok_surface:{bad}") + claim_by_id[claim_id] = claim + + for claim_id, claim in claim_by_id.items(): + unresolved = sorted(set(claim["supersedes"]) - set(claim_by_id)) + if unresolved: + raise ContractError(f"bundle.claims.{claim_id}.supersedes:unresolved:{unresolved}") + + unknowns = _list(bundle["unknowns"], "bundle.unknowns", max_items=spec["maxUnknowns"]) + if len(unknowns) > bounds["unknown_cap"]: + raise ContractError("bundle.unknowns:declared_cap_exceeded") + unknown_ids: set[str] = set() + open_unknowns = 0 + for index, row in enumerate(unknowns): + field = f"bundle.unknowns[{index}]" + unknown = _exact_keys( + row, + {"id", "state", "question", "blocks", "owner", "probe", "resolution_evidence"}, + field, + ) + unknown_id = _string(unknown["id"], f"{field}.id", max_chars=128) + if unknown_id in unknown_ids: + raise ContractError(f"bundle.unknowns:duplicate_id:{unknown_id}") + unknown_ids.add(unknown_id) + state = _one_of(unknown["state"], set(spec["unknownStates"]), f"{field}.state") + _string(unknown["question"], f"{field}.question", max_chars=spec["maxTextChars"]) + _unique_strings(unknown["blocks"], f"{field}.blocks") + _string(unknown["owner"], f"{field}.owner", max_chars=256) + resolution = _unique_strings(unknown["resolution_evidence"], f"{field}.resolution_evidence") + unresolved = sorted(set(resolution) - set(surface_by_id)) + if unresolved: + raise ContractError(f"{field}.resolution_evidence:unresolved:{unresolved}") + if state == "open": + open_unknowns += 1 + probe = _exact_keys( + unknown["probe"], + {"adapter", "action", "deadline_ms", "expected_discriminator", "abort_conditions"}, + f"{field}.probe", + ) + _string(probe["adapter"], f"{field}.probe.adapter", max_chars=256) + _string(probe["action"], f"{field}.probe.action", max_chars=spec["maxTextChars"]) + _int(probe["deadline_ms"], f"{field}.probe.deadline_ms", minimum=1) + _string(probe["expected_discriminator"], f"{field}.probe.expected_discriminator", max_chars=spec["maxTextChars"]) + _unique_strings(probe["abort_conditions"], f"{field}.probe.abort_conditions") + if resolution: + raise ContractError(f"{field}:open_unknown_has_resolution") + else: + if unknown["probe"] is not None: + raise ContractError(f"{field}:closed_unknown_probe_must_be_null") + if state == "answered" and not resolution: + raise ContractError(f"{field}:answered_unknown_needs_resolution") + + transition = _exact_keys( + bundle["next_transition"], + {"type", "owner", "preconditions", "success_receipt", "abort_conditions"}, + "bundle.next_transition", + ) + transition_type = _one_of(transition["type"], set(spec["transitionTypes"]), "bundle.next_transition.type") + _string(transition["owner"], "bundle.next_transition.owner", max_chars=256) + if not _unique_strings(transition["preconditions"], "bundle.next_transition.preconditions"): + raise ContractError("bundle.next_transition.preconditions:empty") + _string(transition["success_receipt"], "bundle.next_transition.success_receipt", max_chars=256) + if not _unique_strings(transition["abort_conditions"], "bundle.next_transition.abort_conditions"): + raise ContractError("bundle.next_transition.abort_conditions:empty") + + supplied_digest = bundle["bundle_digest"] + if require_digest: + _hex(supplied_digest, 64, "bundle.bundle_digest") + if supplied_digest != bundle_digest(bundle): + raise ContractError("bundle.bundle_digest:mismatch") + elif supplied_digest not in ("", None): + _hex(supplied_digest, 64, "bundle.bundle_digest") + + return { + "verdict": "PASS", + "schema": bundle["schema"], + "bundle_id": bundle["bundle_id"], + "program": bundle["program"], + "subject_kind": subject_kind, + "observed_at": clock["observed_at"], + "completeness": completeness, + "surfaces": len(surfaces), + "claims": len(claims), + "universal_claims": universal_claims, + "hypothesis_states": dict(sorted(hypothesis_counts.items())), + "open_unknowns": open_unknowns, + "next_transition": transition_type, + "bundle_digest": supplied_digest if require_digest else bundle_digest(bundle), + } + + +def seal_bundle(bundle: dict[str, Any], contract: dict[str, Any], *, now: dt.datetime | None = None) -> dict[str, Any]: + sealed = copy.deepcopy(bundle) + sealed["bundle_digest"] = "" + validate_bundle(sealed, contract, now=now, require_digest=False) + sealed["bundle_digest"] = bundle_digest(sealed) + validate_bundle(sealed, contract, now=now, require_digest=True) + return sealed + + +def _indexed(rows: list[dict[str, Any]], key: str) -> dict[str, dict[str, Any]]: + return {str(row[key]): row for row in rows} + + +def diff_bundles(before: dict[str, Any], after: dict[str, Any]) -> dict[str, Any]: + if before["subject"] != after["subject"]: + raise ContractError("diff:subject_identity_changed") + + def changes(section: str, key: str) -> dict[str, Any]: + left, right = _indexed(before[section], key), _indexed(after[section], key) + return { + "added": sorted(set(right) - set(left)), + "removed": sorted(set(left) - set(right)), + "changed": sorted( + item + for item in set(left) & set(right) + if json.dumps(left[item], sort_keys=True, separators=(",", ":")) + != json.dumps(right[item], sort_keys=True, separators=(",", ":")) + ), + } + + return { + "schema": "bbi.process-truth-diff/v1", + "subject": before["subject"], + "before": { + "bundle_id": before["bundle_id"], + "digest": before["bundle_digest"], + "observed_at": before["clock"]["observed_at"], + }, + "after": { + "bundle_id": after["bundle_id"], + "digest": after["bundle_digest"], + "observed_at": after["clock"]["observed_at"], + }, + "surfaces": changes("surfaces", "id"), + "claims": changes("claims", "id"), + "unknowns": changes("unknowns", "id"), + "universe_changed": before["universe"] != after["universe"], + "transition_changed": before["next_transition"] != after["next_transition"], + "completeness_changed": before["completeness"] != after["completeness"], + } + + +def handoff_view(bundle: dict[str, Any]) -> dict[str, Any]: + return { + "schema": "bbi.agent-handoff/v1", + "program": bundle["program"], + "subject": bundle["subject"], + "observation": { + "bundle_id": bundle["bundle_id"], + "digest": bundle["bundle_digest"], + "observed_at": bundle["clock"]["observed_at"], + "completeness": bundle["completeness"], + "universe": bundle["universe"], + }, + "facts": [ + { + "id": row["id"], + "statement": row["statement"], + "evidence": row["evidence"], + "expires_at": row["expires_at"], + } + for row in bundle["claims"] + if row["kind"] == "fact" + ], + "open_hypotheses": [ + { + "id": row["id"], + "state": row["state"], + "statement": row["statement"], + "confidence": row["confidence"], + "falsifier": row["falsifier"], + } + for row in bundle["claims"] + if row["kind"] in {"hypothesis", "inference"} and row["state"] == "open" + ], + "open_unknowns": [ + { + "id": row["id"], + "question": row["question"], + "blocks": row["blocks"], + "owner": row["owner"], + "probe": row["probe"], + } + for row in bundle["unknowns"] + if row["state"] == "open" + ], + "next_transition": bundle["next_transition"], + } + + +def validate_workgraph(graph: dict[str, Any]) -> dict[str, Any]: + _exact_keys(graph, {"schema", "programme", "authority", "wip", "status_vocabulary", "tranches"}, "workgraph") + if graph["schema"] != "bbi.pstack-program-workgraph/v1": + raise ContractError("workgraph:schema") + _string(graph["programme"], "workgraph.programme", max_chars=128) + _string(graph["authority"], "workgraph.authority", max_chars=2048) + wip = _exact_keys(graph["wip"], {"mutation_slots", "classes"}, "workgraph.wip") + _int(wip["mutation_slots"], "workgraph.wip.mutation_slots", minimum=1, maximum=8) + _unique_strings(wip["classes"], "workgraph.wip.classes") + statuses = set(_unique_strings(graph["status_vocabulary"], "workgraph.status_vocabulary")) + tranches = _list(graph["tranches"], "workgraph.tranches") + by_id: dict[str, dict[str, Any]] = {} + for index, row in enumerate(tranches): + field = f"workgraph.tranches[{index}]" + tranche = _exact_keys( + row, + {"id", "title", "priority", "owner_role", "mutation_class", "status", "dependencies", "coordination_key", "owned_paths", "first_legal_action", "success_receipt", "abort_conditions", "non_goals"}, + field, + ) + tid = _string(tranche["id"], f"{field}.id", max_chars=64) + if tid in by_id: + raise ContractError(f"workgraph:duplicate_id:{tid}") + _string(tranche["title"], f"{field}.title", max_chars=256) + _int(tranche["priority"], f"{field}.priority", minimum=0, maximum=9) + _string(tranche["owner_role"], f"{field}.owner_role", max_chars=128) + _string(tranche["mutation_class"], f"{field}.mutation_class", max_chars=128) + _one_of(tranche["status"], statuses, f"{field}.status") + _unique_strings(tranche["dependencies"], f"{field}.dependencies") + _string(tranche["coordination_key"], f"{field}.coordination_key", max_chars=128) + if not _unique_strings(tranche["owned_paths"], f"{field}.owned_paths"): + raise ContractError(f"{field}.owned_paths:empty") + _string(tranche["first_legal_action"], f"{field}.first_legal_action", max_chars=2048) + _string(tranche["success_receipt"], f"{field}.success_receipt", max_chars=256) + if not _unique_strings(tranche["abort_conditions"], f"{field}.abort_conditions"): + raise ContractError(f"{field}.abort_conditions:empty") + _unique_strings(tranche["non_goals"], f"{field}.non_goals") + by_id[tid] = tranche + + for tid, tranche in by_id.items(): + missing = sorted(set(tranche["dependencies"]) - set(by_id)) + if missing: + raise ContractError(f"workgraph.{tid}.dependencies:unknown:{missing}") + + visiting: set[str] = set() + visited: set[str] = set() + + def visit(tid: str) -> None: + if tid in visited: + return + if tid in visiting: + raise ContractError(f"workgraph:dependency_cycle:{tid}") + visiting.add(tid) + for dependency in by_id[tid]["dependencies"]: + visit(dependency) + visiting.remove(tid) + visited.add(tid) + + for tid in by_id: + visit(tid) + + terminal = {"MERGED", "REALIZED", "VERIFIED", "SOAKED", "SUPERSEDED", "CLOSED"} + for tid, tranche in by_id.items(): + if tranche["status"] in {"READY", "ACTIVE"}: + blocked_by = [dep for dep in tranche["dependencies"] if by_id[dep]["status"] not in terminal] + if blocked_by: + raise ContractError(f"workgraph.{tid}:ready_with_unsatisfied_dependencies:{blocked_by}") + + ids = list(by_id) + for left_index, left_id in enumerate(ids): + left = by_id[left_id] + for right_id in ids[left_index + 1 :]: + right = by_id[right_id] + if left["status"] not in {"ACTIVE", "READY"} or right["status"] not in {"ACTIVE", "READY"}: + continue + overlap = set(left["owned_paths"]) & set(right["owned_paths"]) + sequenced = left_id in right["dependencies"] or right_id in left["dependencies"] + coordinated = left["coordination_key"] == right["coordination_key"] + if overlap and not (sequenced or coordinated): + raise ContractError(f"workgraph:uncoordinated_active_path_overlap:{left_id}:{right_id}:{sorted(overlap)}") + + return { + "verdict": "PASS", + "programme": graph["programme"], + "tranches": len(by_id), + "frontier": sorted( + tid + for tid, tranche in by_id.items() + if tranche["status"] == "READY" + ), + } + + +def _write_json(value: dict[str, Any], path: Path | None) -> None: + text = json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False, allow_nan=False) + "\n" + if path is None: + sys.stdout.write(text) + else: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + +def _read_bundle(path: Path, contract: dict[str, Any]) -> tuple[dict[str, Any], bytes]: + return _load_json(path, max_bytes=contract["spec"]["maxBundleBytes"]) + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + sub = parser.add_subparsers(dest="command", required=True) + + def common(name: str) -> argparse.ArgumentParser: + item = sub.add_parser(name) + item.add_argument("--contract", required=True, type=Path) + item.add_argument("--input", required=True, type=Path) + item.add_argument("--now", help="deterministic UTC timestamp ending in Z") + return item + + seal = common("seal") + seal.add_argument("--output", required=True, type=Path) + verify = common("verify") + verify.add_argument("--output", type=Path) + handoff = common("handoff") + handoff.add_argument("--output", type=Path) + + diff = sub.add_parser("diff") + diff.add_argument("--contract", required=True, type=Path) + diff.add_argument("--before", required=True, type=Path) + diff.add_argument("--after", required=True, type=Path) + diff.add_argument("--now", help="deterministic UTC timestamp ending in Z") + diff.add_argument("--output", type=Path) + + graph = sub.add_parser("verify-workgraph") + graph.add_argument("--input", required=True, type=Path) + graph.add_argument("--output", type=Path) + + args = parser.parse_args(argv) + try: + if args.command == "verify-workgraph": + workgraph, _ = _load_json(args.input, max_bytes=1_048_576) + _write_json(validate_workgraph(workgraph), args.output) + return 0 + + contract = load_contract(args.contract) + now = _parse_now(getattr(args, "now", None)) + if args.command == "seal": + bundle, _ = _read_bundle(args.input, contract) + _write_json(seal_bundle(bundle, contract, now=now), args.output) + elif args.command == "verify": + bundle, raw = _read_bundle(args.input, contract) + _write_json(validate_bundle(bundle, contract, now=now, raw_size=len(raw)), args.output) + elif args.command == "handoff": + bundle, raw = _read_bundle(args.input, contract) + validate_bundle(bundle, contract, now=now, raw_size=len(raw)) + _write_json(handoff_view(bundle), args.output) + elif args.command == "diff": + before, before_raw = _read_bundle(args.before, contract) + after, after_raw = _read_bundle(args.after, contract) + validate_bundle(before, contract, now=now, raw_size=len(before_raw)) + validate_bundle(after, contract, now=now, raw_size=len(after_raw)) + _write_json(diff_bundles(before, after), args.output) + except ContractError as exc: + sys.stderr.write(json.dumps({"verdict": "SELF_ERROR", "reason": str(exc)}, sort_keys=True) + "\n") + return 2 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/qa/test-pstack-evidence-contract.py b/scripts/qa/test-pstack-evidence-contract.py new file mode 100755 index 0000000000000000000000000000000000000000..1190b3be702db94d27561264a70db24393f48d0c --- /dev/null +++ b/scripts/qa/test-pstack-evidence-contract.py @@ -0,0 +1,342 @@ +#!/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)