#!/usr/bin/env python3 """Contract regression tests for trusted artifact/review/workflow mutators.""" import json import hashlib import os import shutil import sys import yaml ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) HOOKS = os.path.join(ROOT, ".claude", "hooks") WS = os.path.join(ROOT, ".claude", "tests", "fixtures", "workflow-contract-ws") os.environ["CLAUDE_PROJECT_DIR"] = ROOT os.environ["ORGOS_WORKSPACE"] = WS sys.path.insert(0, HOOKS) import acceptance_log as AL # noqa: E402 import context_package as CP # noqa: E402 import state_engine as SE # noqa: E402 shutil.rmtree(WS, ignore_errors=True) os.makedirs(WS, exist_ok=True) passed = failed = 0 def check(name, condition): global passed, failed if condition: passed += 1 print(f" PASS {name}") else: failed += 1 print(f" FAIL {name}") def init(wf, plan="cascade", tier="standard"): SE.init_ledger(wf, plan=plan, tier=tier) def write_report(wf, artifact_id, kind, producer, payload, stage="intake"): directory = os.path.join(WS, "completion-records", wf) os.makedirs(directory, exist_ok=True) path = os.path.join(directory, artifact_id + ".report.yaml") report = { "report-type": "workflow-artifact", "artifact-kind": kind, "artifact-version": 1, "tier": (SE.read_ledger(wf) or {}).get("tier", "standard"), "identity": { "artifact-id": artifact_id, "workflow-id": wf, "stage": stage, "producer-role-id": producer, }, "payload": payload, "report-header": { "bottom-line": f"{kind} fixture", "decision-needed": {"needed": False}, "confidence": {"value": "Med", "derived-from": "evidence"}, "risks": [], "evidence": [{"source-uri": "README.md", "grade": "E3"}], }, } with open(path, "w", encoding="utf-8") as fh: yaml.safe_dump(report, fh, allow_unicode=True, sort_keys=False) return path def submit(wf, path, actor="OPS-ORCH"): return SE.submit_report(wf, path, actor) def open_stage(wf, stage): """Test fixture helper: open a stage without re-testing unrelated upstream gates.""" current = SE.read_ledger(wf).get("stage") event = { "state-event-id": f"fixture-{wf}-{stage}", "event-type": "state-transition", "workflow-id": wf, "from": current, "to": stage, "actor": "OPS-ORCH", "effective-at": SE._now(), } return SE._atomic_event_transaction(wf, workflow_event=event)[0] def add_receipt(wf, receipt_id, command="fixture-check", subject=None, assertion_status="passed", category="test"): directory = os.path.join(WS, "evidence") os.makedirs(directory, exist_ok=True) with open(os.path.join(directory, "ledger.jsonl"), "a", encoding="utf-8") as fh: fh.write(json.dumps({ "tool_use_id": receipt_id, "tool_name": "Bash", "command": command, "exit_code": 0 if assertion_status == "passed" else 1, "workflow_id": wf, "session_id": "fixture-session", "agent_id": "fixture-agent", "ts": SE._now(), "receipt_type": "verification-run", "verification_category": category, "verification_subject": subject or receipt_id, "assertion_status": assertion_status, }) + "\n") def file_sha(path): return hashlib.sha256(open(path, "rb").read()).hexdigest() print("== trusted submit derives gate facts ==") init("wf-contract", tier="light") brief = write_report("wf-contract", "brief-1", "decision-brief", "EXEC-CEO", {"mode": "converge", "tier": "light", "candidate-families": ["FAM-CPO", "FAM-CTO", "FAM-CFO"]}) profile = write_report("wf-contract", "profile-1", "workload-profile", "EXEC-CEO", { "surfaces": {"ui": False, "public-api": False, "persistence": False, "infrastructure": False}, "risk": {"security-bearing": False, "data-migration": False, "external-side-effect": False, "risk-level": "Low", "reversibility": "two-way-door", "blast-radius": "single-role", "privacy": False, "regulatory": False, "slo-impact": False}, "required-capabilities": ["product"], "product-feature": False, }) check("submit decision brief", submit("wf-contract", brief)[0]) check("submit workload profile", submit("wf-contract", profile)[0]) facts = SE._facts("wf-contract", SE.read_ledger("wf-contract")) check("intake facts come from artifacts", facts["decision_brief_present"] and facts["workload_profile_present"]) check("fixture opens discovery stage", open_stage("wf-contract", "discovery")) ground = write_report("wf-contract", "ground-1", "grounding-package", "STR-ANALYST", {"problem-structure": {"question": "target scope"}, "analysis-synthesis": {"insight": "scope trades off against cost"}, "options": [ {"id": "a", "problem": "narrow", "tradeoffs": ["scope"], "evidence-refs": ["README.md"]}, {"id": "b", "problem": "broad", "tradeoffs": ["cost"], "evidence-refs": ["README.md"]}], "evidence": ["README.md"], "source-contributions": [ {"report-id": f"fixture-source-{index}", "report-ref": f"fixture/source-{index}.report.yaml", "report-sha256": str(index) * 64, "producer-role-id": "EXEC-CEO", "context-package-ref": f"fixture/context-{index}.yaml", "context-package-sha256": str(index + 3) * 64, "assigned-lens": lens, "producer-run-id": f"fixture-run-{index}"} for index, lens in enumerate( ["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH"], 1)], "lens-coverage": {"required-min": 3, "covered": ["LENS-VALUE", "LENS-PRODUCT", "LENS-TECH"], "contrarian-report-id": None}}, stage="discovery") check("grounding-package submit", submit("wf-contract", ground)[0]) check("option count derived from report", SE._facts("wf-contract", SE.read_ledger("wf-contract"))["option_set_count"] == 2) ok, _ = SE.record_artifact("wf-contract", design_type="completion-record", option_count=99, evidence_grade="E5") check("unsafe record API removed", not ok) print("== exact effective decision + authorization ==") check("fixture opens decide stage", open_stage("wf-contract", "decide")) packet = write_report("wf-contract", "packet-1", "executive-decision-packet", "EXEC-CEO", {"recommendation": "A"}, stage="decide") check("decision packet submit", submit("wf-contract", packet)[0]) bad_stage_packet = write_report("wf-contract", "packet-bad-stage", "executive-decision-packet", "EXEC-CEO", {"recommendation": "A"}, stage="discovery") check("artifact kind cannot claim an undeclared output stage", not submit("wf-contract", bad_stage_packet)[0]) check("unrelated accepted artifact cannot approve packet", not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"]) check("self-review rejected", not SE.review_artifact("wf-contract", packet, "accepted", "EXEC-CEO")[0]) check("unauthorized reviewer rejected", not SE.review_artifact("wf-contract", packet, "accepted", "ENG-BE")[0]) ok, accepted = SE.review_artifact("wf-contract", packet, "accepted", "HUMAN-001") check("authorized exact revision accepted", ok and accepted.get("artifact-sha256")) check("packet gate accepted", SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"]) check("later changes-requested invalidates acceptance", SE.review_artifact("wf-contract", packet, "changes-requested", "HUMAN-001")[0] and not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"]) check("re-accept latest decision", SE.review_artifact("wf-contract", packet, "accepted", "HUMAN-001")[0]) packet2 = write_report("wf-contract", "packet-2", "executive-decision-packet", "EXEC-CEO", {"recommendation": "B"}, stage="decide") check("new packet revision submitted", submit("wf-contract", packet2)[0]) check("newer unreviewed packet makes old accepted revision stale", not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["decision_packet_accepted"]) check("new packet supersedes old on acceptance", SE.review_artifact("wf-contract", packet2, "accepted", "HUMAN-001", supersedes="packet-1")[0]) old_sha = next(a["artifact-sha256"] for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "packet-1") check("old revision effective decision is Superseded", AL.effective_decision("wf-contract", "packet-1", old_sha) == "Superseded") print("== current-stage lifecycle + event projection ==") direction_brief = os.path.join(WS, "direction-input-brief.yaml") with open(direction_brief, "w", encoding="utf-8") as fh: yaml.safe_dump({ "product-goal": "ship a coherent interface", "core-users": ["operator"], "core-tasks": ["review"], "information-density": "high", "required-accessibility": "WCAG AA", "brand-constraints": ["legible"], "avoid-cliches": ["generic dashboard"], "representative-screen-requirement": "core-task", "tech-platform-constraints": ["web"], }, fh, allow_unicode=True, sort_keys=False) SE.init_ledger("dd-contract", plan="design-direction", parent_workflow="wf-contract", product_decision="packet-2", direction_input_brief=direction_brief) check("new workflow starts with current stage running", SE.read_ledger("dd-contract").get("stage") == "design-direction-intake" and SE.read_ledger("dd-contract").get("stage-status") == "running") check("cannot enter next stage before current is completed", not SE.enter_stage("dd-contract", "design-direction-discovery", actor="OPS-ORCH")[0]) check("complete-stage validates selected exit gate", SE.complete_stage("dd-contract", "OPS-ORCH", to="design-direction-discovery")[0] and SE.read_ledger("dd-contract").get("stage-status") == "completed") check("enter-stage opens next stage as running", SE.enter_stage("dd-contract", "design-direction-discovery", actor="OPS-ORCH")[0] and SE.read_ledger("dd-contract").get("stage") == "design-direction-discovery" and SE.read_ledger("dd-contract").get("stage-status") == "running") ledger_path = SE._ledger_path("dd-contract", create=False) tampered = yaml.safe_load(open(ledger_path, encoding="utf-8")) tampered["stage"] = "design-direction-approved" with open(ledger_path, "w", encoding="utf-8") as fh: yaml.safe_dump(tampered, fh, allow_unicode=True, sort_keys=False) check("workflow.yaml stage edit cannot override canonical events", SE.read_ledger("dd-contract").get("stage") == "design-direction-discovery") print("== design/spec bundles are all-of ==") check("fixture opens design stage", open_stage("wf-contract", "design")) packet_basis = next(a for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "packet-2") design = write_report("wf-contract", "design-1", "overall-design", "ARCH-SOLUTION", {"summary": "overall design", "basis-artifact-id": "packet-2", "basis-artifact-sha256": packet_basis["artifact-sha256"]}, stage="design") check("overall design submitted", submit("wf-contract", design)[0]) check("design bundle not accepted on presence", not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["design_accepted"]) check("design bundle accepted after exact architecture review", SE.review_artifact("wf-contract", design, "accepted", "ARCH-EA")[0] and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["design_accepted"]) design_basis = next(a for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "design-1") check("fixture opens spec stage", open_stage("wf-contract", "spec")) api = write_report("wf-contract", "api-1", "api-contract", "ARCH-TECH", {"summary": "api contract", "basis-artifact-id": "design-1", "basis-artifact-sha256": design_basis["artifact-sha256"]}, stage="spec") check("api contract alone accepted", submit("wf-contract", api)[0] and SE.review_artifact("wf-contract", api, "accepted", "ARCH-APP")[0] and not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["spec_accepted"]) criteria = write_report("wf-contract", "criteria-1", "acceptance-criteria", "PROD-PO", {"basis-artifact-id": "design-1", "basis-artifact-sha256": design_basis["artifact-sha256"], "criteria": [{"criterion-id": "AC-1", "preconditions": [], "input": {}, "expected-result": "works", "risk-level": "Low", "verification-method": "automated-test"}]}, stage="spec") check("required spec bundle accepted", submit("wf-contract", criteria)[0] and SE.review_artifact("wf-contract", criteria, "accepted", "EXEC-CPO")[0] and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["spec_accepted"]) print("== quality/release event mutators ==") check("fixture opens build stage", open_stage("wf-contract", "build")) implementation = os.path.join(WS, "implemented.txt") with open(implementation, "w", encoding="utf-8") as fh: fh.write("implemented\n") add_receipt("wf-contract", "build-test-receipt") completion = write_report("wf-contract", "build-1", "completion-record", "ENG-BE", {"summary": "implementation complete", "source-revision": {"kind": "workspace-tree", "sha256": "a" * 64}, "primary-artifacts": [{"path": implementation, "kind": "code", "sha256": file_sha(implementation)}], "acceptance-criteria-coverage": [{"criterion-id": "AC-1", "status": "Passed", "evidence-receipt-ids": ["build-test-receipt"]}], "verification-receipt-ids": ["build-test-receipt"], "remaining-risks": []}, stage="build") check("completion submitted", submit("wf-contract", completion)[0]) target = next(a for a in SE._trusted_artifacts("wf-contract") if a["artifact-id"] == "build-1") check("fixture opens verification stage", open_stage("wf-contract", "verification")) add_receipt("wf-contract", "quality-test-receipt", subject="AC-1") quality = write_report("wf-contract", "quality-1", "quality-gate-review", "QA", { "quality-gate": {"status": "Passed"}, "blocker-open": False, "reviewed-artifact-id": "build-1", "reviewed-artifact-sha256": target["artifact-sha256"], "checks": [{"check-id": "AC-1", "category": "test", "status": "Passed", "evidence-receipt-ids": ["quality-test-receipt"]}], "findings": [], }, stage="verification") check("quality gate event records protected facts", SE.record_quality_gate("wf-contract", quality, "QA")[0] and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["quality_gate_status"] == "Passed") check("Passed quality gate cannot trigger verification rework", not SE.can_transition("wf-contract", "build", actor="OPS-ORCH")[0]) check("caller fact injection cannot trigger verification rework", not SE.can_transition("wf-contract", "build", actor="OPS-ORCH", ctx={"facts": {"quality-gate-failed": True}})[0]) add_receipt("wf-contract", "quality-failure-receipt", subject="AC-1", assertion_status="failed") quality_failure = write_report("wf-contract", "quality-2", "quality-gate-review", "QA", { "quality-gate": {"status": "Failed"}, "blocker-open": True, "reviewed-artifact-id": "build-1", "reviewed-artifact-sha256": target["artifact-sha256"], "checks": [{"check-id": "AC-1", "category": "test", "status": "Failed", "evidence-receipt-ids": ["quality-failure-receipt"]}], "findings": [{"finding-id": "QF-REWORK", "severity": "High", "blocking": True, "resolved": False, "summary": "verification regression requires build rework"}], }, stage="verification") check("trusted current Failed quality gate enables rework predicate", SE.record_quality_gate("wf-contract", quality_failure, "QA")[0] and SE.can_transition("wf-contract", "build", actor="OPS-ORCH")[0]) check("non-executor cannot enter rework transition", not SE.can_transition("wf-contract", "build", actor="QA")[0]) check("verification rework transition returns workflow to running build", SE.transition("wf-contract", "build", actor="OPS-ORCH")[0] and SE.read_ledger("wf-contract").get("stage") == "build" and SE.read_ledger("wf-contract").get("stage-status") == "running") check("fixture reopens verification after corrected build", open_stage("wf-contract", "verification")) add_receipt("wf-contract", "quality-recovery-receipt", subject="AC-1") quality_recovery = write_report("wf-contract", "quality-3", "quality-gate-review", "QA", { "quality-gate": {"status": "Passed"}, "blocker-open": False, "reviewed-artifact-id": "build-1", "reviewed-artifact-sha256": target["artifact-sha256"], "checks": [{"check-id": "AC-1", "category": "test", "status": "Passed", "evidence-receipt-ids": ["quality-recovery-receipt"]}], "findings": [{"finding-id": "QF-REWORK", "severity": "High", "blocking": True, "resolved": True, "summary": "verification regression corrected and rerun"}], }, stage="verification") check("fresh Passed quality gate supersedes the failed rework review", SE.record_quality_gate("wf-contract", quality_recovery, "QA")[0] and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["quality_gate_status"] == "Passed" and not SE._facts("wf-contract", SE.read_ledger("wf-contract"))["blocker_open"]) check("fixture opens acceptance stage", open_stage("wf-contract", "acceptance")) quality_state = SE.read_ledger("wf-contract") release = write_report("wf-contract", "release-1", "release-decision", "HUMAN-001", { "release-decision": {"status": "Approved"}, "unresolved-critical-risks": False, "reviewed-completion-artifact-id": quality_state["current-completion-artifact-id"], "reviewed-completion-artifact-sha256": quality_state["current-completion-artifact-sha256"], "reviewed-quality-event-id": quality_state["current-quality-event-id"], "reviewed-quality-artifact-id": quality_state["current-quality-artifact-id"], "reviewed-quality-artifact-sha256": quality_state["current-quality-artifact-sha256"], }, stage="acceptance") check("release decision event records protected facts", SE.record_release_decision("wf-contract", release, "HUMAN-001")[0] and SE._facts("wf-contract", SE.read_ledger("wf-contract"))["release_acceptance_status"] == "Approved") print("== blocked/resume and actors ==") blocked = write_report("wf-contract", "blocked-1", "blocked-report", "OPS-ORCH", {"blocker": "dependency", "resume-condition": "receipt exists"}, stage="blocked") check("block command requires valid report and enters blocked", SE.block_workflow("wf-contract", blocked)[0] and SE.current_stage("wf-contract") == "blocked") resume = write_report("wf-contract", "resume-1", "resume-evidence", "OPS-ORCH", {"resume-condition-satisfied": True}, stage="blocked") check("resume command restores blocked-from", SE.resume_workflow("wf-contract", resume)[0] and SE.current_stage("wf-contract") != "blocked") check("invented actor rejected", not SE._actor_allowed({"allowed-by": ["OPS-ORCH"]}, "fake-role-agent")[0]) check("placeholder-only authorization rejected", not SE._actor_allowed({"allowed-by": ["design-lead-role-agent"]}, "OPS-ORCH")[0]) print("== family candidate metadata only ==") resolved = SE.resolve_family("FAM-DESIGN") check("family resolves to concrete workers", resolved and "DES-DIRECTOR" in resolved["resolved-workers"]) pkg = {"target-role-agent": "fam-design"} check("family target rejected by context-package", any("candidate metadata" in error for error in CP._semantic_errors(pkg))) print("== contract-derived invariants ==") contract = SE.load_contracts() kinds = contract.get("artifact-kinds", {}) bundled = set() for bundle in (contract.get("artifact-bundles", {}) or {}).values(): bundled.update(bundle.get("always") or []) for conditional in bundle.get("conditional") or []: bundled.update(conditional.get("require") or []) check("every bundle artifact-kind is declared", bundled.issubset(set(kinds))) mapped_must_reads = { kind for mapping in (SE.load_cmap().get("design-to-build-contract", {}).get("mappings", []) or []) for kind in (mapping.get("must-read-designs") or []) } check("compatibility must-read vocabulary contains only canonical artifact-kinds", mapped_must_reads.issubset(set(kinds))) default_schema = contract.get("default-payload-schema-ref") schema_dir = os.path.join(ROOT, ".claude", "schemas") check("every artifact-kind resolves to an existing payload schema", all( os.path.isfile(os.path.join(schema_dir, definition.get("payload-schema-ref") or default_schema)) for definition in kinds.values())) cascade_pairs = {("intake", "discovery"), ("discovery", "decide"), ("decide", "design"), ("design", "spec"), ("spec", "build"), ("build", "verification"), ("verification", "acceptance"), ("acceptance", "released")} runtime = {(t.get("from"), t.get("to")): t for t in SE._ws_transitions()} check("every cascade transition has concrete OPS-ORCH executor", all( pair in runtime and "OPS-ORCH" in ((runtime[pair].get("allowed-by") or {}).get("executor") or []) for pair in cascade_pairs)) check("every protected mutable gate has an official mutator", all(callable(getattr(SE, name, None)) for name in ( "init_workflow", "submit_artifact", "review_artifact", "record_human_signoff", "record_quality_gate", "record_release_decision", "block_workflow", "resume_workflow", "complete_stage", "enter_stage"))) workflow_outputs = set() workflow_conditions = set() for workflow in (contract.get("workflows", {}) or {}).values(): for stage in (workflow.get("stages", {}) or {}).values(): outputs = stage.get("outputs") or {} workflow_outputs.update(outputs.get("bundle") or []) if outputs.get("dynamic-bundle"): bundle = (contract.get("artifact-bundles", {}) or {}).get(outputs["dynamic-bundle"], {}) workflow_outputs.update(bundle.get("always") or []) for condition in bundle.get("conditional") or []: workflow_outputs.update(condition.get("require") or []) workflow_conditions.update(stage.get("exit-gate") or []) for transition in workflow.get("additional-transitions") or []: workflow_conditions.update(transition.get("required-conditions") or []) for transition in contract.get("side-transitions") or []: workflow_conditions.update(transition.get("required-conditions") or []) check("every declared stage output has an artifact contract", workflow_outputs.issubset(set(kinds))) check("every declared exit gate has a runtime predicate", workflow_conditions.issubset(set(SE._PREDICATES))) print(f"\n{passed} passed, {failed} failed") shutil.rmtree(WS, ignore_errors=True) sys.exit(0 if failed == 0 else 1)