#!/usr/bin/env python3 """R03-A3 isolated Frozen ChangePlan and typed synthetic staging loop. Only opaque CouncilCore workspace object IDs are accepted. There is no URL, OS-path, shell, argv, package, service, or production-host operation surface. """ from __future__ import annotations import argparse import ast import base64 import contextlib from copy import deepcopy from dataclasses import dataclass import hashlib import hmac import io import json import os from pathlib import Path import re import signal import socket import sqlite3 import stat import tempfile import threading from typing import Any, Callable from council_core import ( CouncilCore, CouncilError, DeterministicTokens, MutableClock, SessionContext, SessionCredential, canonical_json, expected_answers, read_json, sha256, ) DATA_CLASS = "SYNTHETIC_PUBLIC_ONLY" HEX32 = re.compile(r"[0-9a-f]{32}\Z") HEX64 = re.compile(r"[0-9a-f]{64}\Z") ID = re.compile(r"[A-Za-z0-9_.:-]{1,64}\Z") CALLBACK = re.compile(r"ARC1:([AR]):([A-Za-z0-9_-]{16}):([A-Za-z0-9_-]{8}):([A-Za-z0-9_-]{22})\Z") REVIEW_ROWS = frozenset({"correctness", "scope", "security", "tests", "rollback", "non_target_preservation"}) A3_ROUTES = ( "POST /v1/workspaces", "POST /v1/workspaces/{run}/objects", "PUT /v1/workspaces/{run}/objects/{object}", "GET /v1/workspaces/{run}/objects/{object}", "POST /v1/tests", "POST /v1/artifacts", "GET /v1/artifacts/{artifact}", "POST /v1/reviews", "GET /v1/changeplans/{plan}", ) class StagingError(CouncilError): pass class StagingHTTPError(StagingError): def __init__(self, status: int, code: str) -> None: super().__init__(code) self.status = status self.code = code class InjectedFailure(StagingError): pass def require(value: bool, message: str) -> None: if not value: raise StagingError(message) def b64u(data: bytes) -> str: return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") def strict_object(raw: bytes) -> dict[str, Any]: def pairs(items: list[tuple[str, Any]]) -> dict[str, Any]: result: dict[str, Any] = {} for key, value in items: if key in result: raise ValueError("duplicate key") result[key] = value return result try: value = json.loads(raw.decode(), object_pairs_hook=pairs, parse_constant=lambda _x: (_ for _ in ()).throw(ValueError("constant"))) except (UnicodeError, json.JSONDecodeError, ValueError) as error: raise StagingHTTPError(400, "INVALID_JSON") from error if not isinstance(value, dict): raise StagingHTTPError(400, "INVALID_SCHEMA") return value def exact(value: dict[str, Any], fields: set[str]) -> None: if set(value) != fields: raise StagingHTTPError(400, "INVALID_SCHEMA") def apply_single_file_unified_diff(before: bytes, patch: bytes) -> bytes: """Apply one strict UTF-8 unified diff in memory.""" try: old = before.decode().splitlines(keepends=True) lines = patch.decode().splitlines(keepends=True) except UnicodeError as error: raise StagingError("diff UTF-8") from error require(len(lines) >= 3 and lines[0].startswith("--- a/") and lines[1].startswith("+++ b/") and "\x00" not in "".join(lines), "diff headers") header = re.compile(r"@@ -(\d+)(?:,(\d+))? \+(\d+)(?:,(\d+))? @@(?: .*)?\n?\Z") output: list[str] = [] old_cursor = 0 index = 2 hunks = 0 while index < len(lines): match = header.fullmatch(lines[index]) require(match is not None, "diff hunk") hunks += 1 old_start = int(match.group(1)); old_count = int(match.group(2) or "1") new_count = int(match.group(4) or "1") require(old_start >= 1 and old_start - 1 >= old_cursor, "diff order") output.extend(old[old_cursor:old_start - 1]); old_cursor = old_start - 1 index += 1; used_old = 0; used_new = 0 while index < len(lines) and not lines[index].startswith("@@ "): line = lines[index] require(bool(line) and line[0] in " +-", "diff marker") marker, content = line[0], line[1:] if marker in " -": require(old_cursor < len(old) and old[old_cursor] == content, "diff old content") old_cursor += 1; used_old += 1 if marker in " +": output.append(content); used_new += 1 index += 1 require(used_old == old_count and used_new == new_count, "diff counts") require(hunks > 0, "diff hunk required") output.extend(old[old_cursor:]) return "".join(output).encode() def validate_policy(policy: dict[str, Any]) -> None: expected = { "schema": "AI_COUNCIL_R03_A3_POLICY_V1", "version": 1, "mode": "ISOLATED_SYNTHETIC_STAGING_ONLY_NOT_DEPLOYED", "pilot_result": "PROVISIONAL_PILOT", "allowed_data_class": DATA_CLASS, "rejected_data_classes": ["PRIVATE", "PRODUCTION", "SECRET", "CREDENTIAL", "MARKET_LIVE", "VOICE_RAW", "PERSONAL_DATA", "UNKNOWN"], "routes": list(A3_ROUTES), "changeplan_ttl_seconds": 900, "callback_nonce_bytes": 6, "callback_mac_bytes": 16, "callback_data_max_bytes": 64, "artifact_review_max_bytes": 1048576, "typed_operations": ["ATOMIC_REPLACE_ONE_SYNTHETIC_OBJECT"], "apply_commit_points": 5, "rollback_commit_points": 5, "independent_review_required": True, "all_review_rows_must_pass": True, "test_receipt_required": True, "owner_decision_once": True, "approval_callback_operations": ["send_approval_card", "answer_callback"], "message_body_persisted": False, "arbitrary_url_api": False, "arbitrary_path_api": False, "arbitrary_command_api": False, "shell_or_argv_execution": False, "private_payload_encryption_implemented": False, "target_install_authorized": False, "target_ubuntu_changes": 0, } pins = { "capability_gate_url": "https://paste.rs/efmTH", "capability_gate_sha256": "596b60b7c4efb28c30f095978c21b73ce91038d223cf6b4cd07388b053e7ac0d", "a2_policy_sha256": "e12566998b930b0ca2c94de9d773c868149372236c711b2a70845a7b18fa8590", "core_policy_sha256": "ce793b1a1054eb1c57ab86a6bb0f676866e5f2f33d9c3f2c89c13a3c368f4341", "gateway_a2_before_sha256": "6851db5e584b527fff6c7399bf572b5cf47bed3ff3f477ca166878ca82caf568", "relay_a2_before_sha256": "cb8735bc53e863c50943ab77a3b36ccce131eb4ef2f2ea93038caac6bc2d40e3", } require(set(policy) == set(expected) | set(pins), "policy fields") for key, value in {**expected, **pins}.items(): require(policy[key] == value, f"policy value {key}") @dataclass(frozen=True) class ArtifactRecord: artifact_id: str author: SessionContext canonical: bytes @dataclass(frozen=True) class ReviewRecord: review_id: str reviewer: SessionContext canonical: bytes @dataclass(frozen=True) class PlanRecord: plan_id: str canonical: bytes class StagingController: def __init__(self, core: CouncilCore, policy: dict[str, Any], clock: Callable[[], int] | None = None, token_bytes: Callable[[int], bytes] | None = None) -> None: validate_policy(policy) self.core = core; self.policy = policy self.clock = clock or core.clock; self.token_bytes = token_bytes or core.token_bytes self.artifacts: dict[str, ArtifactRecord] = {} self.reviews: dict[str, ReviewRecord] = {} self.plans: dict[str, PlanRecord] = {} self.callback_nonces: dict[str, str] = {} def now(self) -> int: return int(self.clock()) def _active(self, context: SessionContext, room: str, tool: str) -> None: self.core._require_admission(context, room, tool) def _row(self, object_id: str, owner: str | None = None) -> sqlite3.Row: require(isinstance(object_id, str) and HEX32.fullmatch(object_id) is not None, "object id") row = self.core.db.execute("SELECT * FROM workspace_objects WHERE object_id=? AND kind='OBJECT'", (object_id,)).fetchone() require(row is not None and row["state"] == "READY" and row["deleted_at"] is None and row["expires_at"] >= self.now(), "object unavailable") if owner is not None: require(row["owner_agent_digest"] == owner, "object owner") run = self.core.db.execute("SELECT * FROM workspace_objects WHERE object_id=? AND kind='RUN' AND run_id=?", (row["run_id"], row["run_id"])).fetchone() require(run is not None and run["owner_agent_digest"] == row["owner_agent_digest"] and run["state"] == "ACTIVE" and run["deleted_at"] is None and run["expires_at"] >= self.now(), "run unavailable") return row def _path(self, row: sqlite3.Row) -> Path: run = self.core.db.execute("SELECT * FROM workspace_objects WHERE object_id=? AND kind='RUN'", (row["run_id"],)).fetchone() require(run is not None and HEX32.fullmatch(run["relative_storage_name"]) is not None and HEX32.fullmatch(row["relative_storage_name"]) is not None, "storage metadata") directory = self.core.workspace_root / run["relative_storage_name"] require(directory.is_dir() and not directory.is_symlink() and directory.resolve(strict=True).parent == self.core.workspace_root, "run path") info = directory.lstat() require(stat.S_ISDIR(info.st_mode) and info.st_uid == os.geteuid() and stat.S_IMODE(info.st_mode) == 0o700, "run mode") path = directory / row["relative_storage_name"] require(path.parent.resolve(strict=True) == directory.resolve(strict=True), "object containment") return path def _read(self, object_id: str, owner: str | None = None) -> bytes: row = self._row(object_id, owner); path = self._path(row) flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0) try: descriptor = os.open(path, flags) except OSError as error: raise StagingError("object open") from error try: before = os.fstat(descriptor) require(stat.S_ISREG(before.st_mode) and before.st_nlink == 1 and before.st_uid == os.geteuid() and stat.S_IMODE(before.st_mode) == 0o600, "object metadata") with os.fdopen(descriptor, "rb", closefd=False) as handle: data = handle.read(self.core.policy["workspace_single_object_bytes"] + 1) after = os.fstat(descriptor) require((before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns) == (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns), "object changed") finally: os.close(descriptor) require(len(data) == row["size_bytes"] and sha256(data) == row["actual_sha256"], "object integrity") return data def register_artifact_bundle(self, context: SessionContext, job_id: str, target_object_id: str, before_object_id: str, after_object_id: str, full_file_object_id: str, diff_object_id: str, rollback_object_id: str, test_receipt_ids: list[str]) -> dict[str, Any]: self._active(context, "ENGINEERING", "artifact.export") job = self.core.db.execute("SELECT * FROM jobs WHERE job_id=?", (job_id,)).fetchone() require(job is not None and job["lease_agent_digest"] == context.agent_digest and job["status"] == "LEASED" and job["expires_at"] >= self.now(), "artifact job") names = {"target_object_id": target_object_id, "before_object_id": before_object_id, "after_object_id": after_object_id, "full_file_object_id": full_file_object_id, "diff_object_id": diff_object_id, "rollback_object_id": rollback_object_id} rows = {name: self._row(value, context.agent_digest) for name, value in names.items()} run_ids = {row["run_id"] for row in rows.values()}; require(len(run_ids) == 1, "one run") values = {name: self._read(value, context.agent_digest) for name, value in names.items()} require(values["target_object_id"] == values["before_object_id"], "target before") require(values["rollback_object_id"] == values["before_object_id"], "rollback before") require(values["full_file_object_id"] == values["after_object_id"], "full after") require(values["before_object_id"] != values["after_object_id"], "no-op") require(sum(map(len, values.values())) <= self.policy["artifact_review_max_bytes"], "review size") require(apply_single_file_unified_diff(values["before_object_id"], values["diff_object_id"]) == values["after_object_id"], "diff mismatch") require(isinstance(test_receipt_ids, list) and 1 <= len(test_receipt_ids) <= 16 and len(set(test_receipt_ids)) == len(test_receipt_ids), "receipt ids") receipts: list[dict[str, Any]] = [] for receipt_id in test_receipt_ids: require(isinstance(receipt_id, str) and HEX32.fullmatch(receipt_id) is not None, "receipt id") receipt = self.core.db.execute("SELECT * FROM test_receipts WHERE receipt_id=?", (receipt_id,)).fetchone() require(receipt is not None and receipt["job_id"] == job_id and receipt["passed"] == 1, "passing receipt") receipts.append({key: receipt[key] for key in ("receipt_id", "validator_id", "input_set_sha256", "result_sha256")}) required = { ("FROZEN_ENGINEERING_FIXTURE_V1", sha256(canonical_json([sha256(values["before_object_id"]), sha256(values["after_object_id"])]))), ("UNIFIED_DIFF_STRUCTURE_V1", sha256(canonical_json([sha256(values["diff_object_id"])]))), ("PYTHON_AST_PARSE_V1", sha256(canonical_json([sha256(values["after_object_id"])]))), } require(required <= {(item["validator_id"], item["input_set_sha256"]) for item in receipts}, "receipt binding") artifact_id = self.token_bytes(16).hex(); require(artifact_id not in self.artifacts, "artifact collision") descriptor = { "schema": "ARC_FROZEN_ARTIFACT_SET_V1", "artifact_id": artifact_id, "job_id": job_id, "run_id": next(iter(run_ids)), "author_agent_digest": context.agent_digest, "data_class": DATA_CLASS, "operation": "ATOMIC_REPLACE_ONE_SYNTHETIC_OBJECT", "objects": {name: {"object_id": names[name], "sha256": sha256(values[name]), "size_bytes": len(values[name])} for name in sorted(names)}, "test_receipts": sorted(receipts, key=lambda item: item["receipt_id"]), "created_at": self.now(), } frozen = canonical_json(descriptor); artifact_hash = sha256(frozen) self.artifacts[artifact_id] = ArtifactRecord(artifact_id, context, frozen) self.core._audit("ARTIFACT_FROZEN", context.agent_digest, artifact_hash, "PASS"); self.core.db.commit() return {"artifact_id": artifact_id, "artifact_set_sha256": artifact_hash, "descriptor": deepcopy(descriptor)} def read_artifact(self, context: SessionContext, artifact_id: str) -> dict[str, Any]: self._active(context, "REVIEW", "artifact.read") record = self.artifacts.get(artifact_id); require(record is not None, "artifact unavailable") descriptor = json.loads(record.canonical) require(context.agent_digest != descriptor["author_agent_digest"], "self review") bodies: dict[str, str] = {}; total = 0 for name, metadata in descriptor["objects"].items(): data = self._read(metadata["object_id"], descriptor["author_agent_digest"]) require(sha256(data) == metadata["sha256"] and len(data) == metadata["size_bytes"], "artifact changed") total += len(data); require(total <= self.policy["artifact_review_max_bytes"], "review size") bodies[name] = base64.b64encode(data).decode() return {"schema": "ARC_REVIEW_ARTIFACT_V1", "artifact_set_sha256": sha256(record.canonical), "descriptor": descriptor, "object_bodies_base64": bodies} def submit_review(self, context: SessionContext, artifact_id: str, decision: str, rows: dict[str, dict[str, str]]) -> dict[str, Any]: self._active(context, "REVIEW", "review.submit") record = self.artifacts.get(artifact_id); require(record is not None, "artifact unavailable") artifact = json.loads(record.canonical) require(context.agent_digest != artifact["author_agent_digest"], "self review") require(decision in {"APPROVE_STAGING", "REJECT"} and isinstance(rows, dict) and set(rows) == REVIEW_ROWS, "review shape") all_pass = True for value in rows.values(): require(isinstance(value, dict) and set(value) == {"result", "evidence_sha256"} and value["result"] in {"PASS", "FAIL"} and isinstance(value["evidence_sha256"], str) and HEX64.fullmatch(value["evidence_sha256"]) is not None, "review row") all_pass &= value["result"] == "PASS" require((decision == "APPROVE_STAGING") == all_pass, "review decision") artifact_hash = sha256(record.canonical) require(self.core.db.execute("SELECT 1 FROM reviews WHERE artifact_set_sha256=? AND reviewer_agent_digest=?", (artifact_hash, context.agent_digest)).fetchone() is None, "duplicate review") review_hash = self.core.submit_review(context, artifact["job_id"], artifact_hash, decision, rows) row = self.core.db.execute("SELECT * FROM reviews WHERE review_sha256=?", (review_hash,)).fetchone(); require(row is not None, "review persistence") review = {"schema": "ARC_FROZEN_REVIEW_V1", "review_id": row["review_id"], "artifact_id": artifact_id, "artifact_set_sha256": artifact_hash, "job_id": artifact["job_id"], "author_agent_digest": artifact["author_agent_digest"], "reviewer_agent_digest": context.agent_digest, "decision": decision, "rows": rows, "review_sha256": review_hash, "created_at": row["created_at"]} self.reviews[review_hash] = ReviewRecord(row["review_id"], context, canonical_json(review)) result: dict[str, Any] = {"review_id": row["review_id"], "review_sha256": review_hash, "decision": decision} if decision == "APPROVE_STAGING": result["changeplan"] = self.freeze_changeplan(artifact_id, review_hash) return result def freeze_changeplan(self, artifact_id: str, review_hash: str) -> dict[str, Any]: artifact_record = self.artifacts.get(artifact_id); review_record = self.reviews.get(review_hash) require(artifact_record is not None and review_record is not None, "freeze inputs") artifact = json.loads(artifact_record.canonical); review = json.loads(review_record.canonical) job = self.core.db.execute("SELECT * FROM jobs WHERE job_id=?", (artifact["job_id"],)).fetchone() require(job is not None and job["status"] == "LEASED" and job["lease_agent_digest"] == artifact["author_agent_digest"] and job["expires_at"] >= self.now(), "plan job") require(review["artifact_id"] == artifact_id and review["artifact_set_sha256"] == sha256(artifact_record.canonical) and review["decision"] == "APPROVE_STAGING" and review["reviewer_agent_digest"] != artifact["author_agent_digest"] and all(value["result"] == "PASS" for value in review["rows"].values()), "passing independent review") plan_id = b64u(self.token_bytes(12)); require(len(plan_id) == 16 and ID.fullmatch(plan_id) is not None and plan_id not in self.plans, "plan id") nonce = b64u(self.token_bytes(self.policy["callback_nonce_bytes"])); require(len(nonce) == 8, "nonce") objects = artifact["objects"] target = {"target_object_id": objects["target_object_id"]["object_id"], "before_sha256": objects["before_object_id"]["sha256"], "after_object_id": objects["after_object_id"]["object_id"], "after_sha256": objects["after_object_id"]["sha256"], "full_file_object_id": objects["full_file_object_id"]["object_id"], "full_file_sha256": objects["full_file_object_id"]["sha256"], "diff_object_id": objects["diff_object_id"]["object_id"], "diff_sha256": objects["diff_object_id"]["sha256"], "rollback_object_id": objects["rollback_object_id"]["object_id"], "rollback_sha256": objects["rollback_object_id"]["sha256"]} scope_hash = sha256(canonical_json({"operation": "ATOMIC_REPLACE_ONE_SYNTHETIC_OBJECT", "run_id": artifact["run_id"], "target_object_id": target["target_object_id"], "before_sha256": target["before_sha256"], "after_sha256": target["after_sha256"]})) created = self.now(); expires = created + self.policy["changeplan_ttl_seconds"] plan = {"schema": "ARC_FROZEN_CHANGEPLAN_V1", "plan_id": plan_id, "job_id": artifact["job_id"], "run_id": artifact["run_id"], "artifact_id": artifact_id, "artifact_set_sha256": sha256(artifact_record.canonical), "review_id": review["review_id"], "review_sha256": review_hash, "author_agent_digest": artifact["author_agent_digest"], "reviewer_agent_digest": review["reviewer_agent_digest"], "data_class": DATA_CLASS, "operation": "ATOMIC_REPLACE_ONE_SYNTHETIC_OBJECT", "target": target, "test_receipts": artifact["test_receipts"], "target_scope_sha256": scope_hash, "created_at": created, "expires_at": expires, "constraints": {"owner_decision_once": True, "verify_after_apply": True, "automatic_rollback": True, "target_host": False, "shell_or_argv": False}} frozen = canonical_json(plan); plan_hash = sha256(frozen) self.core.db.execute("INSERT INTO changeplans VALUES(?,?,?,?,?,?,NULL,NULL,NULL)", (plan_id, artifact["job_id"], plan_hash, scope_hash, sha256(nonce.encode()), expires)) self.plans[plan_id] = PlanRecord(plan_id, frozen); self.callback_nonces[plan_id] = nonce self.core._audit("CHANGEPLAN_FROZEN", artifact["author_agent_digest"], plan_hash, "PASS"); self.core.db.commit() return {"plan_id": plan_id, "plan_sha256": plan_hash, "target_scope_sha256": scope_hash, "expires_at": expires} def read_changeplan(self, context: SessionContext, plan_id: str) -> dict[str, Any]: self._active(context, "APPROVAL", "changeplan.read") record = self.plans.get(plan_id); require(record is not None, "plan unavailable") row = self.core.db.execute("SELECT * FROM changeplans WHERE plan_id=?", (plan_id,)).fetchone() require(row is not None and row["plan_sha256"] == sha256(record.canonical), "plan integrity") return {"schema": "ARC_CHANGEPLAN_RESPONSE_V1", "plan": json.loads(record.canonical), "plan_sha256": row["plan_sha256"], "owner_decision": row["owner_decision"], "decided_at": row["decided_at"], "consumed_at": row["consumed_at"]} def _callback_mac(self, action: str, plan_id: str, nonce: str) -> str: payload = f"ARC1\0{action}\0{plan_id}\0{nonce}".encode() digest = hmac.new(self.core.master_key, b"R03-A3-OWNER-CALLBACK\0" + payload, hashlib.sha256).digest() return b64u(digest[:self.policy["callback_mac_bytes"]]) def callback_data(self, plan_id: str, action: str) -> str: require(action in {"A", "R"} and plan_id in self.callback_nonces, "callback") nonce = self.callback_nonces[plan_id] value = f"ARC1:{action}:{plan_id}:{nonce}:{self._callback_mac(action, plan_id, nonce)}" require(len(value.encode()) <= 64 and CALLBACK.fullmatch(value) is not None, "callback encoding") return value def build_approval_card(self, plan_id: str) -> dict[str, Any]: record = self.plans.get(plan_id); require(record is not None, "plan unavailable") row = self.core.db.execute("SELECT * FROM changeplans WHERE plan_id=?", (plan_id,)).fetchone() require(row is not None and row["owner_decision"] is None and row["consumed_at"] is None and row["expires_at"] >= self.now(), "not approvable") return {"schema": "ARC_APPROVAL_CARD_V1", "plan_id": plan_id, "plan_sha256": row["plan_sha256"], "target_scope_sha256": row["target_scope_sha256"], "expires_at": row["expires_at"], "approve_callback_data": self.callback_data(plan_id, "A"), "reject_callback_data": self.callback_data(plan_id, "R")} def approval_gateway_frame(self, plan_id: str) -> dict[str, Any]: card = self.build_approval_card(plan_id) return {"schema": "ARC_CONTROLLER_TO_GATEWAY_V1", "op": "send_approval_card", "request_id": card["plan_sha256"][:32], "room_id": "APPROVAL", "data_class": DATA_CLASS, "plan_id": card["plan_id"], "plan_sha256": card["plan_sha256"], "target_scope_sha256": card["target_scope_sha256"], "expires_at": card["expires_at"], "approve_callback_data": card["approve_callback_data"], "reject_callback_data": card["reject_callback_data"]} def consume_owner_callback(self, frame: dict[str, Any]) -> dict[str, Any]: fields = {"schema", "op", "request_id", "gateway_event_sha256", "callback_query_id_sha256", "room_id", "data_class", "callback_data", "callback_data_sha256"} require(isinstance(frame, dict) and set(frame) == fields, "callback fields") data = frame["callback_data"] require(frame["schema"] == "ARC_GATEWAY_TO_CONTROLLER_V1" and frame["op"] == "owner_changeplan_callback" and isinstance(frame["request_id"], str) and HEX32.fullmatch(frame["request_id"]) is not None and isinstance(frame["gateway_event_sha256"], str) and HEX64.fullmatch(frame["gateway_event_sha256"]) is not None and frame["request_id"] == frame["gateway_event_sha256"][:32] and isinstance(frame["callback_query_id_sha256"], str) and HEX64.fullmatch(frame["callback_query_id_sha256"]) is not None and frame["room_id"] == "APPROVAL" and frame["data_class"] == DATA_CLASS and isinstance(data, str) and frame["callback_data_sha256"] == sha256(data.encode()), "callback values") match = CALLBACK.fullmatch(data); require(match is not None, "callback encoding") action, plan_id, nonce, supplied = match.groups() require(hmac.compare_digest(supplied, self._callback_mac(action, plan_id, nonce)), "callback MAC") record = self.plans.get(plan_id); require(record is not None, "plan unavailable") row = self.core.db.execute("SELECT * FROM changeplans WHERE plan_id=?", (plan_id,)).fetchone() require(row is not None and row["plan_sha256"] == sha256(record.canonical) and row["nonce_digest"] == sha256(nonce.encode()) and row["expires_at"] >= self.now() and row["owner_decision"] is None and row["decided_at"] is None and row["consumed_at"] is None, "callback unavailable") decision = "APPROVE" if action == "A" else "REJECT" self.core.db.execute("BEGIN IMMEDIATE") try: cursor = self.core.db.execute("UPDATE changeplans SET owner_decision=?,decided_at=? WHERE plan_id=? AND owner_decision IS NULL AND decided_at IS NULL AND consumed_at IS NULL", (decision, self.now(), plan_id)) require(cursor.rowcount == 1, "callback race") self.core._audit("OWNER_DECISION", None, row["plan_sha256"], decision); self.core.db.commit() except BaseException: self.core.db.rollback(); raise return {"schema": "ARC_OWNER_DECISION_ACK_V1", "request_id": frame["request_id"], "status": "DECIDED", "decision": decision, "plan_id": plan_id, "plan_sha256": row["plan_sha256"]} def _atomic_write(self, path: Path, data: bytes, label: str) -> None: temporary = path.with_name("." + path.name + "." + label + "-" + self.token_bytes(8).hex()) descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600) try: with os.fdopen(descriptor, "wb") as handle: handle.write(data); handle.flush(); os.fsync(handle.fileno()) os.chmod(temporary, 0o600); os.replace(temporary, path) parent_fd = os.open(path.parent, os.O_RDONLY) try: os.fsync(parent_fd) finally: os.close(parent_fd) finally: try: temporary.unlink() except FileNotFoundError: pass info = path.lstat(); require(stat.S_ISREG(info.st_mode) and info.st_nlink == 1 and info.st_uid == os.geteuid() and stat.S_IMODE(info.st_mode) == 0o600, "atomic metadata") def _restore(self, row: sqlite3.Row, path: Path, before: bytes, fail_after: int | None) -> None: if fail_after == 1: raise InjectedFailure("rollback 1") self._atomic_write(path, before, "rollback") if fail_after == 2: raise InjectedFailure("rollback 2") digest = sha256(before) self.core.db.execute("UPDATE workspace_objects SET expected_sha256=?,actual_sha256=?,size_bytes=?,state='READY' WHERE object_id=?", (digest, digest, len(before), row["object_id"])); self.core.db.commit() if fail_after == 3: raise InjectedFailure("rollback 3") require(self._read(row["object_id"], row["owner_agent_digest"]) == before, "rollback verify") if fail_after == 4: raise InjectedFailure("rollback 4") self.core._audit("STAGING_ROLLBACK", row["owner_agent_digest"], digest, "PASS"); self.core.db.commit() if fail_after == 5: raise InjectedFailure("rollback 5") def apply_approved_plan(self, plan_id: str, fail_after: int | None = None, fail_after_rollback: int | None = None) -> dict[str, Any]: require(fail_after is None or fail_after in range(1, 6), "apply point") require(fail_after_rollback is None or fail_after_rollback in range(1, 6), "rollback point") record = self.plans.get(plan_id); require(record is not None, "plan unavailable") plan = json.loads(record.canonical) plan_row = self.core.db.execute("SELECT * FROM changeplans WHERE plan_id=?", (plan_id,)).fetchone() require(plan_row is not None and plan_row["plan_sha256"] == sha256(record.canonical) and plan_row["owner_decision"] == "APPROVE" and plan_row["decided_at"] is not None and plan_row["consumed_at"] is None and plan_row["expires_at"] >= self.now(), "plan not executable") require(plan["operation"] == "ATOMIC_REPLACE_ONE_SYNTHETIC_OBJECT" and plan["data_class"] == DATA_CLASS, "operation") target = plan["target"]; row = self._row(target["target_object_id"], plan["author_agent_digest"]); path = self._path(row) before = self._read(target["target_object_id"], plan["author_agent_digest"]) after = self._read(target["after_object_id"], plan["author_agent_digest"]) full = self._read(target["full_file_object_id"], plan["author_agent_digest"]) rollback = self._read(target["rollback_object_id"], plan["author_agent_digest"]) require(sha256(before) == target["before_sha256"] and sha256(after) == target["after_sha256"] and after == full and sha256(full) == target["full_file_sha256"] and rollback == before and sha256(rollback) == target["rollback_sha256"], "plan artifacts") if fail_after == 1: raise InjectedFailure("apply 1") temporary = path.with_name("." + path.name + ".stage-" + self.token_bytes(8).hex()) descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o600) replaced = False try: with os.fdopen(descriptor, "wb") as handle: handle.write(after); handle.flush(); os.fsync(handle.fileno()) os.chmod(temporary, 0o600) if fail_after == 2: raise InjectedFailure("apply 2") os.replace(temporary, path); replaced = True parent_fd = os.open(path.parent, os.O_RDONLY) try: os.fsync(parent_fd) finally: os.close(parent_fd) if fail_after == 3: raise InjectedFailure("apply 3") digest = sha256(after) self.core.db.execute("UPDATE workspace_objects SET expected_sha256=?,actual_sha256=?,size_bytes=?,state='READY' WHERE object_id=?", (digest, digest, len(after), target["target_object_id"])); self.core.db.commit() if fail_after == 4: raise InjectedFailure("apply 4") require(self._read(target["target_object_id"], plan["author_agent_digest"]) == after, "apply verify") if fail_after == 5: raise InjectedFailure("apply 5") self.core.db.execute("BEGIN IMMEDIATE") try: current = self.core.db.execute("SELECT * FROM changeplans WHERE plan_id=?", (plan_id,)).fetchone() require(current is not None and current["owner_decision"] == "APPROVE" and current["consumed_at"] is None, "consume race") self.core.db.execute("UPDATE changeplans SET consumed_at=? WHERE plan_id=?", (self.now(), plan_id)) self.core.db.execute("UPDATE jobs SET status='STAGED_VERIFIED' WHERE job_id=?", (plan["job_id"],)) self.core._audit("STAGING_APPLY", plan["author_agent_digest"], plan_row["plan_sha256"], "PASS"); self.core.db.commit() except BaseException: self.core.db.rollback(); raise except BaseException as original: try: temporary.unlink() except FileNotFoundError: pass if replaced: interrupted = False try: self._restore(row, path, before, fail_after_rollback) except BaseException: interrupted = True; self._restore(row, path, before, None) if interrupted: raise StagingError("apply failed; rollback recovered") from original raise finally: try: temporary.unlink() except FileNotFoundError: pass return {"schema": "ARC_STAGING_APPLY_RECEIPT_V1", "status": "STAGED_VERIFIED", "plan_id": plan_id, "plan_sha256": plan_row["plan_sha256"], "target_object_id": target["target_object_id"], "before_sha256": target["before_sha256"], "after_sha256": target["after_sha256"], "verified": True, "rollback_used": False, "target_host_changes": 0} def dispatch_http(self, context: SessionContext, method: str, path: str, body: bytes) -> tuple[int, dict[str, Any]]: try: if method == "POST" and path == "/v1/workspaces": value = strict_object(body); exact(value, {"job_id", "fencing_token"}) run_id = self.core.create_workspace_run(context, value["job_id"], value["fencing_token"]) return 201, {"schema": "ARC_WORKSPACE_RUN_V1", "run_id": run_id} match = re.fullmatch(r"/v1/workspaces/([0-9a-f]{32})/objects", path) if method == "POST" and match: value = strict_object(body); exact(value, {"expected_sha256"}) object_id = self.core.allocate_object(context, match.group(1), value["expected_sha256"]) return 201, {"schema": "ARC_WORKSPACE_OBJECT_V1", "object_id": object_id} match = re.fullmatch(r"/v1/workspaces/([0-9a-f]{32})/objects/([0-9a-f]{32})", path) if match and method in {"PUT", "GET"}: row = self.core._object_row(context, match.group(2)); require(row["run_id"] == match.group(1), "run object") if method == "PUT": digest = self.core.write_object(context, match.group(2), body) return 200, {"schema": "ARC_OBJECT_WRITE_RECEIPT_V1", "object_id": match.group(2), "sha256": digest} data = self.core.read_object(context, match.group(2)) return 200, {"schema": "ARC_OBJECT_READ_V1", "object_id": match.group(2), "sha256": sha256(data), "content_base64": base64.b64encode(data).decode()} if method == "POST" and path == "/v1/tests": value = strict_object(body); exact(value, {"job_id", "validator_id", "object_ids", "expected_sha256"}) return 201, {"schema": "ARC_TEST_RECEIPT_V1", **self.core.run_validator(context, value["job_id"], value["validator_id"], value["object_ids"], value["expected_sha256"])} if method == "POST" and path == "/v1/artifacts": value = strict_object(body); exact(value, {"job_id", "target_object_id", "before_object_id", "after_object_id", "full_file_object_id", "diff_object_id", "rollback_object_id", "test_receipt_ids"}) result = self.register_artifact_bundle(context, **value) return 201, {"schema": "ARC_ARTIFACT_RECEIPT_V1", "artifact_id": result["artifact_id"], "artifact_set_sha256": result["artifact_set_sha256"]} match = re.fullmatch(r"/v1/artifacts/([0-9a-f]{32})", path) if method == "GET" and match: return 200, self.read_artifact(context, match.group(1)) if method == "POST" and path == "/v1/reviews": value = strict_object(body); exact(value, {"artifact_id", "decision", "rows"}) return 201, {"schema": "ARC_REVIEW_RECEIPT_V1", **self.submit_review(context, value["artifact_id"], value["decision"], value["rows"])} match = re.fullmatch(r"/v1/changeplans/([A-Za-z0-9_-]{16})", path) if method == "GET" and match: return 200, self.read_changeplan(context, match.group(1)) except StagingHTTPError: raise except (CouncilError, KeyError, TypeError, ValueError) as error: raise StagingHTTPError(403, "REQUEST_REJECTED") from error raise StagingHTTPError(404, "NOT_FOUND") def build_callback_frame(query_id: str, data: str) -> dict[str, Any]: require(isinstance(query_id, str) and 1 <= len(query_id) <= 128 and isinstance(data, str) and CALLBACK.fullmatch(data) is not None, "callback frame") routing = {"callback_query_id_sha256": sha256(query_id.encode()), "callback_data_sha256": sha256(data.encode())} event_hash = sha256(canonical_json(routing)) return {"schema": "ARC_GATEWAY_TO_CONTROLLER_V1", "op": "owner_changeplan_callback", "request_id": event_hash[:32], "gateway_event_sha256": event_hash, "callback_query_id_sha256": routing["callback_query_id_sha256"], "room_id": "APPROVAL", "data_class": DATA_CLASS, "callback_data": data, "callback_data_sha256": routing["callback_data_sha256"]} def context_for(core: CouncilCore, credential: SessionCredential, clock: MutableClock, nonce: str) -> SessionContext: mac = core.sign_request(credential, "GET", "/v1/health", b"", clock(), nonce) return core.authenticate_request(credential.session_id, credential.agent_id, "GET", "/v1/health", b"", clock(), nonce, mac) def run_self_test(policy_path: Path, core_policy_path: Path, expected_path: Path) -> dict[str, Any]: import council_relay as relay import telegram_gateway as gateway policy = read_json(policy_path); core_policy = read_json(core_policy_path) relay_policy = read_json(policy_path.parent / "r03_a2_policy_v1.json") validate_policy(policy); expected = read_json(expected_path) tests: list[str] = [] def passed(name: str, value: bool = True) -> None: require(value, name); tests.append(name) def rejected(name: str, call: Callable[[], Any]) -> None: try: call() except (CouncilError, sqlite3.Error, gateway.GatewayError): tests.append(name) else: raise StagingError("expected rejection " + name) clock = MutableClock(1_788_667_200); tokens = DeterministicTokens() master = hashlib.sha256(b"R03-A3-ISOLATED-MASTER").digest() manifests = core_policy_path.parent / core_policy["manifest_directory"] temporary_path = "" with tempfile.TemporaryDirectory(prefix="ai-council-r03-a3-", dir="/tmp") as temporary: temporary_path = temporary; root = Path(temporary) / "state" core = CouncilCore(root, manifests, core_policy, master, clock, tokens) staging = StagingController(core, policy, clock, tokens) try: passed("policy_schema"); passed("nine_routes", len(A3_ROUTES) == 9) source_tree = ast.parse(Path(__file__).read_text()) roots = {(n.module or "").split(".")[0] for n in ast.walk(source_tree) if isinstance(n, ast.ImportFrom)} | {a.name.split(".")[0] for n in ast.walk(source_tree) if isinstance(n, ast.Import) for a in n.names} passed("no_subprocess", "subprocess" not in roots); passed("no_http_client", not {"http", "urllib", "requests"} & roots) passed("no_private_payload", policy["private_payload_encryption_implemented"] is False) credentials: list[SessionCredential] = []; contexts: list[SessionContext] = [] for slot in (1, 2, 3): code = core.create_enrollment(slot, f"A3-PROMPT-{slot}") credential = core.consume_enrollment(code, f"A3-AGENT-{slot}", f"A3-PROMPT-{slot}") credentials.append(credential); contexts.append(context_for(core, credential, clock, f"{slot:032x}")) author, reviewer, observer = contexts; passed("three_fake_agents", len({x.session_id for x in contexts}) == 3) commons = core.get_manifest("COMMONS") for participant in contexts: core.ack_room(participant, "COMMONS", commons["content_sha256"], expected_answers(commons)) discussions = ["A3 author proposes synthetic change", "A3 reviewer requires tests and rollback", "A3 chair fixes one-object scope"] for participant, text in zip(contexts, discussions): core.publish_message(participant, "COMMONS", text, DATA_CLASS) passed("three_agent_commons", len(core.poll_events(observer, 0)) == 3) passed("discussion_not_in_db", all(text.encode() not in core.database_path.read_bytes() for text in discussions)) for room in ("JOB", "ENGINEERING", "REVIEW"): manifest = core.get_manifest(room); core.ack_room(author, room, manifest["content_sha256"], expected_answers(manifest)) manifest = core.get_manifest("REVIEW"); core.ack_room(reviewer, "REVIEW", manifest["content_sha256"], expected_answers(manifest)) manifest = core.get_manifest("APPROVAL"); core.ack_room(observer, "APPROVAL", manifest["content_sha256"], expected_answers(manifest)) rejected("private_rejected", lambda: core.create_job(b"x", "PRIVATE")) job = core.create_job(b"change one to two", DATA_CLASS); fence = core.lease_job(author, job) rejected("stale_fence", lambda: core.create_workspace_run(author, job, 0)) run = core.create_workspace_run(author, job, fence) before = b"def answer():\n return 1\n"; after = b"def answer():\n return 2\n" diff = b"--- a/answer.py\n+++ b/answer.py\n@@ -1,2 +1,2 @@\n def answer():\n- return 1\n+ return 2\n" object_ids = [] for data in (before, before, after, after, diff, before): oid = core.allocate_object(author, run, sha256(data)); core.write_object(author, oid, data); object_ids.append(oid) target, before_id, after_id, full_id, diff_id, rollback_id = object_ids; passed("six_objects") fixture = core.run_validator(author, job, "FROZEN_ENGINEERING_FIXTURE_V1", [before_id, after_id]) diff_test = core.run_validator(author, job, "UNIFIED_DIFF_STRUCTURE_V1", [diff_id]) ast_test = core.run_validator(author, job, "PYTHON_AST_PARSE_V1", [after_id]); passed("three_tests") wrong = core.run_validator(author, job, "PYTHON_AST_PARSE_V1", [before_id]) rejected("unbound_receipt", lambda: staging.register_artifact_bundle(author, job, target, before_id, after_id, full_id, diff_id, rollback_id, [fixture["receipt_id"], diff_test["receipt_id"], wrong["receipt_id"]])) wrong_diff = b"--- a/answer.py\n+++ b/answer.py\n@@ -1,2 +1,2 @@\n def answer():\n- return 1\n+ return 3\n" wrong_diff_id = core.allocate_object(author, run, sha256(wrong_diff)); core.write_object(author, wrong_diff_id, wrong_diff) wrong_diff_test = core.run_validator(author, job, "UNIFIED_DIFF_STRUCTURE_V1", [wrong_diff_id]) rejected("incorrect_diff", lambda: staging.register_artifact_bundle(author, job, target, before_id, after_id, full_id, wrong_diff_id, rollback_id, [fixture["receipt_id"], wrong_diff_test["receipt_id"], ast_test["receipt_id"]])) artifact = staging.register_artifact_bundle(author, job, target, before_id, after_id, full_id, diff_id, rollback_id, [fixture["receipt_id"], diff_test["receipt_id"], ast_test["receipt_id"]]); passed("artifact_frozen") rejected("self_artifact_read", lambda: staging.read_artifact(author, artifact["artifact_id"])) view = staging.read_artifact(reviewer, artifact["artifact_id"]); passed("review_read", base64.b64decode(view["object_bodies_base64"]["after_object_id"]) == after) evidence = sha256(b"evidence"); rows = {name: {"result": "PASS", "evidence_sha256": evidence} for name in REVIEW_ROWS} rejected("self_review", lambda: staging.submit_review(author, artifact["artifact_id"], "APPROVE_STAGING", rows)) failed = deepcopy(rows); failed["security"]["result"] = "FAIL" rejected("failed_row_approve", lambda: staging.submit_review(reviewer, artifact["artifact_id"], "APPROVE_STAGING", failed)) review = staging.submit_review(reviewer, artifact["artifact_id"], "APPROVE_STAGING", rows); plan = review["changeplan"]; plan_id = plan["plan_id"] passed("independent_review"); rejected("duplicate_review", lambda: staging.submit_review(reviewer, artifact["artifact_id"], "APPROVE_STAGING", rows)) plan_view = staging.read_changeplan(observer, plan_id); passed("plan_read", plan_view["plan_sha256"] == plan["plan_sha256"]) mutated = plan_view; mutated["plan"]["operation"] = "SHELL"; passed("plan_immutable", staging.read_changeplan(observer, plan_id)["plan"]["operation"].startswith("ATOMIC")) card = staging.build_approval_card(plan_id); passed("callback_size", len(card["approve_callback_data"]) <= 64 and len(card["reject_callback_data"]) <= 64) config = {"allowed_chat_id": -100123, "owner_user_id": 4242, "allowed_thread_id": None} update = {"update_id": 700, "callback_query": {"id": "CB-OWNER", "from": {"id": 4242, "is_bot": False}, "message": {"message_id": 88, "chat": {"id": -100123, "type": "supergroup"}}, "data": card["approve_callback_data"]}} extracted = gateway.extract_owner_changeplan_callback(update, config, True); passed("gateway_extract", extracted is not None) passed("default_callback_off", gateway.extract_owner_changeplan_callback(update, config, False) is None) wrong_owner = deepcopy(update); wrong_owner["callback_query"]["from"]["id"] = 9; passed("wrong_owner", gateway.extract_owner_changeplan_callback(wrong_owner, config, True) is None) wrong_chat = deepcopy(update); wrong_chat["callback_query"]["message"]["chat"]["id"] = -9; passed("wrong_chat", gateway.extract_owner_changeplan_callback(wrong_chat, config, True) is None) passed("wrong_thread", gateway.extract_owner_changeplan_callback(update, {**config, "allowed_thread_id": 7}, True) is None) frame = gateway.build_gateway_callback_frame(extracted); passed("frame_match", frame == build_callback_frame("CB-OWNER", card["approve_callback_data"])) tampered = card["approve_callback_data"][:-1] + ("A" if card["approve_callback_data"][-1] != "A" else "B") rejected("tampered_mac", lambda: staging.consume_owner_callback(build_callback_frame("T", tampered))) relay_app = relay.RelayApplication(core, relay_policy, clock=clock, monotonic=lambda: float(clock()), staging_controller=staging) runtime = Path(temporary) / "controller-runtime"; runtime.mkdir(mode=0o700); controller_socket = runtime / "in.sock" ingress = relay.GatewayIngressServer(relay_app, controller_socket, os.geteuid()); ingress.start() try: decision = gateway.forward_council_callback(frame, controller_socket, os.geteuid(), 2) finally: ingress.close() passed("callback_unix_e2e", decision["decision"] == "APPROVE") rejected("callback_replay", lambda: staging.consume_owner_callback(frame)) rejected("opposite_action", lambda: staging.consume_owner_callback(build_callback_frame("R", card["reject_callback_data"]))) # Freeze the reject/expiry branch plans while the job lease is still valid. rejected_plan = staging.freeze_changeplan(artifact["artifact_id"], review["review_sha256"]) reject_card = staging.build_approval_card(rejected_plan["plan_id"]) reject_decision = staging.consume_owner_callback(build_callback_frame("REJECT", reject_card["reject_callback_data"])) passed("owner_reject", reject_decision["decision"] == "REJECT") rejected("rejected_not_apply", lambda: staging.apply_approved_plan(rejected_plan["plan_id"])) expiring = staging.freeze_changeplan(artifact["artifact_id"], review["review_sha256"]) expiring_card = staging.build_approval_card(expiring["plan_id"]) protected = {oid: sha256(staging._read(oid, author.agent_digest)) for oid in (before_id, after_id, full_id, diff_id, rollback_id, wrong_diff_id)} for point in range(1, 6): rejected(f"apply_fail_{point}", lambda point=point: staging.apply_approved_plan(plan_id, fail_after=point)) passed(f"apply_restore_{point}", staging._read(target, author.agent_digest) == before) for point in range(1, 6): rejected(f"rollback_interrupt_{point}", lambda point=point: staging.apply_approved_plan(plan_id, fail_after=4, fail_after_rollback=point)) passed(f"rollback_restore_{point}", staging._read(target, author.agent_digest) == before) passed("non_target_preserved", protected == {oid: sha256(staging._read(oid, author.agent_digest)) for oid in protected}) receipt = staging.apply_approved_plan(plan_id); passed("staged_verified", receipt["status"] == "STAGED_VERIFIED") passed("target_after", staging._read(target, author.agent_digest) == after) metadata = core.db.execute("SELECT expected_sha256,actual_sha256 FROM workspace_objects WHERE object_id=?", (target,)).fetchone(); passed("metadata_after", tuple(metadata) == (sha256(after), sha256(after))) rejected("apply_twice", lambda: staging.apply_approved_plan(plan_id)) core.publish_message(observer, "COMMONS", "verified " + sha256(canonical_json(receipt)), DATA_CLASS); passed("chair_closes_loop") clock.advance(901) rejected("expired_callback", lambda: staging.consume_owner_callback(build_callback_frame("EXPIRE", expiring_card["approve_callback_data"]))) passed("no_source_in_db", before not in core.database_path.read_bytes() and after not in core.database_path.read_bytes()) passed("three_plans", core.db.execute("SELECT COUNT(*) FROM changeplans").fetchone()[0] == 3) passed("zero_target_changes", policy["target_ubuntu_changes"] == 0) finally: core.close() reopened = CouncilCore(root, manifests, core_policy, master, clock, tokens) try: restarted = StagingController(reopened, policy, clock, tokens) passed("restart_metadata_only", reopened.db.execute("SELECT COUNT(*) FROM changeplans").fetchone()[0] == 3 and not restarted.plans) rejected("restart_plan_fail_closed", lambda: restarted.build_approval_card(plan_id)) finally: reopened.close() passed("temporary_removed", not Path(temporary_path).exists()) receipt = {"schema": "AI_COUNCIL_R03_A3_SELF_TEST_V1", "status": "PASS", "test_count": len(tests), "tests": tests, "fake_agent_sessions": 3, "independent_review": True, "frozen_changeplan": True, "owner_callback_decisions_tested": ["APPROVE", "REJECT"], "owner_callback_once": True, "apply_failure_points": 5, "rollback_failure_points": 5, "typed_staging_operation": "ATOMIC_REPLACE_ONE_SYNTHETIC_OBJECT", "final_apply_status": "STAGED_VERIFIED", "allowed_data_class": DATA_CLASS, "private_or_production_payload": False, "external_network_calls": 0, "subprocess_calls": 0, "production_listeners_started": 0, "production_writes": 0, "target_ubuntu_changes": 0, "services_changed": 0, "ports_opened": 0, "pilot_result": "PROVISIONAL_PILOT", "temporary_artifacts_after_cleanup": 0} require(receipt == expected, "self-test fixture mismatch") return receipt def main() -> int: parser = argparse.ArgumentParser(); parser.add_argument("--self-test", action="store_true") parser.add_argument("--policy", type=Path, required=True); parser.add_argument("--core-policy", type=Path, required=True); parser.add_argument("--expected", type=Path, required=True) args = parser.parse_args(); require(args.self_test, "only self-test enabled") print(json.dumps(run_self_test(args.policy, args.core_policy, args.expected), ensure_ascii=False, indent=2)); return 0 if __name__ == "__main__": raise SystemExit(main())