#!/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())