#!/usr/bin/env python3 """Disposable-only Task27 cleanup for session-bound onboarding residue.""" from __future__ import annotations import argparse import contextlib import fcntl import hashlib import importlib import json import os import shutil import stat import subprocess import sys import tempfile import traceback import uuid from collections.abc import Iterator from pathlib import Path from typing import Any SCHEMA = "task27-supplemental-cleanup-controller-v1" CANDIDATE = "d1109d8f78aaccf949ec4f664d9e62584c3cca33032030518bc3e2d712239112" CUSTOMER = "task26_claim_20260818145508_1b96b23d" SESSION = "cb_mYUoMIsk_CRzSDKYpPE9dg" USER_ID = "8527916639" ROUTE = (USER_ID, "0") LIVE = Path("/home/cube/.hermes/profiles/dualcoachtest") EVIDENCE = Path(__file__).resolve().parent CLEANUP_RECEIPT = EVIDENCE / "task27-live-cleanup-receipt.json" CLEANUP_RECEIPT_SHA256 = ( "7848def41100aa4f4ef27239dfaae3cc634d5ad2b24df7966d4d80838f692a3a" ) HERMES_WHEEL = ( EVIDENCE.parent / "task26/task26-combined-v38-delivered-st_01a019d7/artifacts" / "hermes_agent-0.17.0-py3-none-any.whl" ) HERMES_WHEEL_SHA256 = ( "f8b3c779c58435dd33f8f9bbc89e9823bd09f81367a27366c6d678270c1860b1" ) APPROVAL_MESSAGE = "ㅇㅇ" LIVE_PREVIEW = EVIDENCE / "task27-supplemental-live-final-preview-v2.json" LIVE_AUTHORIZATION = ( EVIDENCE / "task27-supplemental-live-cleanup-authorization-final.json" ) LIVE_OUTPUT = EVIDENCE / "task27-supplemental-live-cleanup-receipt.json" MARKER = ".task27-supplemental-disposable-copy" MARKER_VALUE = "TASK27_SUPPLEMENTAL_DISPOSABLE_COPY" OUTBOX_NAMES = { ".lock", ".emergency.lock", ".owner-callbacks.lock", ".receipt-key", "ledger.json", "emergency.json", "owner-callbacks.json", } MEMBERSHIP_NAMES = {"events.jsonl", "events.jsonl.lock"} ARCHIVE_ROOT_NAMES = { "customer-cleanup", "task27-final-cleanup", "task27-supplemental-cleanup", "post-lifecycle-cleanup-archives", "profile-reset-archives", "rehearsal-reset-archives", "recovery-audits", } class Refusal(RuntimeError): pass def canonical(value: object) -> bytes: return json.dumps( value, ensure_ascii=True, sort_keys=True, separators=(",", ":") ).encode() def sha_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def sha_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: while block := handle.read(1024 * 1024): digest.update(block) return digest.hexdigest() def atomic_json(path: Path, value: object, mode: int = 0o600) -> None: if path.exists() or path.is_symlink(): raise Refusal(f"output already exists: {path}") path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) descriptor, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: os.fchmod(descriptor, mode) with os.fdopen(descriptor, "wb", closefd=True) as handle: descriptor = -1 handle.write(canonical(value) + b"\n") handle.flush() os.fsync(handle.fileno()) os.replace(temporary, path) path.chmod(mode) finally: if descriptor >= 0: os.close(descriptor) with contextlib.suppress(FileNotFoundError): os.unlink(temporary) def _require_private(path: Path, *, directory: bool) -> os.stat_result: try: info = path.lstat() except OSError as exc: raise Refusal(f"required cleanup state is unavailable: {path}") from exc expected_kind = stat.S_ISDIR if directory else stat.S_ISREG expected_mode = 0o700 if directory else 0o600 if ( stat.S_ISLNK(info.st_mode) or not expected_kind(info.st_mode) or info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != expected_mode or (not directory and info.st_nlink != 1) ): raise Refusal(f"cleanup state is unsafe: {path}") return info def _require_owner_receipt(path: Path) -> None: try: info = path.lstat() except OSError as exc: raise Refusal(f"cleanup receipt is unavailable: {path}") from exc if ( stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) not in {0o400, 0o600} or info.st_nlink != 1 ): raise Refusal(f"cleanup receipt is unsafe: {path}") def tree_inventory(root: Path) -> dict[str, Any]: if not root.exists() and not root.is_symlink(): return { "root": str(root), "exists": False, "entry_count": 0, "digest": sha_bytes(b"[]"), } _require_private(root, directory=True) rows: list[dict[str, Any]] = [] for path in [root, *sorted(root.rglob("*"))]: relative = "." if path == root else path.relative_to(root).as_posix() info = path.lstat() if stat.S_ISLNK(info.st_mode): raise Refusal(f"cleanup state contains a symlink: {path}") if stat.S_ISDIR(info.st_mode): if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700: raise Refusal(f"cleanup directory is not owner-only: {path}") rows.append({"path": relative, "type": "dir", "mode": 0o700}) elif stat.S_ISREG(info.st_mode): if ( info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o600 or info.st_nlink != 1 ): raise Refusal(f"cleanup file is unsafe: {path}") rows.append( { "path": relative, "type": "file", "mode": 0o600, "size": info.st_size, "sha256": sha_file(path), } ) else: raise Refusal(f"cleanup state has an unsupported entry: {path}") return { "root": str(root), "exists": True, "entry_count": len(rows), "digest": sha_bytes(canonical(rows)), "entries": rows, } def compact(snapshot: dict[str, Any]) -> dict[str, Any]: return { key: snapshot[key] for key in ("root", "exists", "entry_count", "digest") } def mutation_paths(root: Path) -> dict[str, Path]: return { "outbox": root / "data/onboarding/telegram-publication-outbox-v1", "membership": root / "data/onboarding/telegram-staff-membership-v1", "supplemental_archive": root / "data/task27-supplemental-cleanup", } def proposed_mutations() -> list[dict[str, str]]: return [ { "path": "data/onboarding/telegram-publication-outbox-v1/**", "action": ( "archive the exact target-only operational directory, verify it, " "then prune it" ), }, { "path": "data/onboarding/telegram-staff-membership-v1/**", "action": ( "archive the exact target-only subscription directory, verify it, " "then prune it" ), }, { "path": "data/task27-supplemental-cleanup/archives//**", "action": "create and freeze a byte-recoverable supplemental archive", }, { "path": "/tmp/task27-supplemental-rollback-*/**", "action": "create and remove a private rollback buffer during execution", }, { "path": ".omo/evidence/task27/task27-supplemental-live-cleanup-receipt.json", "action": "create and freeze the one-use terminal live receipt", }, { "path": ( ".omo/evidence/task27/" ".task27-supplemental-live-cleanup-receipt.json.*" ), "action": "create and remove the private atomic receipt temporary", }, ] def tool_binding() -> dict[str, str]: controller = Path(__file__).resolve() wheel_hash = sha_file(HERMES_WHEEL) if wheel_hash != HERMES_WHEEL_SHA256: raise Refusal("delivered Hermes wheel hash mismatch") return { "candidate_digest": CANDIDATE, "controller_sha256": sha_file(controller), "hermes_wheel_sha256": wheel_hash, } def cleanup_receipt_binding() -> dict[str, object]: _require_owner_receipt(CLEANUP_RECEIPT) actual_hash = sha_file(CLEANUP_RECEIPT) if actual_hash != CLEANUP_RECEIPT_SHA256: raise Refusal("current Task27 cleanup receipt hash mismatch") try: receipt = json.loads(CLEANUP_RECEIPT.read_bytes()) except (OSError, json.JSONDecodeError) as exc: raise Refusal("current Task27 cleanup receipt is invalid") from exc if ( receipt.get("status") != "COMMITTED" or receipt.get("operation_id") != "e90fcc341e9f49c2a2c57ce46dd50c12" or receipt.get("before", {}).get("expected", {}).get("customer_key") != CUSTOMER or receipt.get("before", {}).get("expected", {}).get("session_id") != SESSION or receipt.get("before", {}).get("expected", {}).get("user_id") != USER_ID ): raise Refusal("current Task27 cleanup receipt binding mismatch") return { "path": str(CLEANUP_RECEIPT), "sha256": actual_hash, "operation_id": receipt["operation_id"], } def approved_preview() -> tuple[dict[str, Any], str, str]: _require_owner_receipt(LIVE_PREVIEW) preview_sha = sha_file(LIVE_PREVIEW) try: preview = json.loads(LIVE_PREVIEW.read_bytes()) except (OSError, json.JSONDecodeError) as exc: raise Refusal("approved live preview is invalid") from exc payload = preview.get("permission_payload") seal = preview.get("permission_seal") if ( preview.get("schema") != SCHEMA or preview.get("mode") != "dry-run" or preview.get("status") != "AWAITING_AUTHORIZATION" or not isinstance(payload, dict) or payload.get("target") != str(LIVE) or payload.get("live_authorization") is not None or payload.get("tools") != tool_binding() or payload.get("proposed_mutations") != proposed_mutations() or not isinstance(seal, str) or seal != sha_bytes(canonical(payload)) ): raise Refusal("approved live preview binding mismatch") return payload, preview_sha, seal def live_authorization_binding(path: Path) -> dict[str, object]: if path.resolve() != LIVE_AUTHORIZATION.resolve(): raise Refusal("live authorization path mismatch") _require_owner_receipt(path) authorization_hash = sha_file(path) try: document = json.loads(path.read_bytes()) except (OSError, json.JSONDecodeError) as exc: raise Refusal("live authorization is invalid") from exc approval = document.get("approval") expected_identity = { "customer_key": CUSTOMER, "session_id": SESSION, "user_id": USER_ID, "route": list(ROUTE), } expected_constraints = { "activation": False, "commit": "none", "customer_delivery": False, "external_network_actions": 0, "gateway_must_remain_stopped": True, "provider_actions": 0, "push": "none", "release": False, "telegram_actions": 0, } if not isinstance(approval, dict): raise Refusal("live authorization binding mismatch") approved_preview_path = approval.get("approved_dry_run_path") if ( not isinstance(approved_preview_path, str) or Path(approved_preview_path).resolve() != LIVE_PREVIEW.resolve() ): raise Refusal("approved live preview path mismatch") _, approved_preview_sha, approved_preview_seal = approved_preview() current_tools = tool_binding() if ( document.get("schema") != "task27-supplemental-live-cleanup-authorization-v1" or document.get("status") != "APPROVED_ONCE" or document.get("target") != str(LIVE) or document.get("candidate_digest") != CANDIDATE or document.get("cleanup_receipt_sha256") != CLEANUP_RECEIPT_SHA256 or document.get("identity") != expected_identity or document.get("proposed_mutations") != proposed_mutations() or document.get("constraints") != expected_constraints or approval.get("approved_controller_sha256") != current_tools["controller_sha256"] or approval.get("approved_dry_run_sha256") != approved_preview_sha or approval.get("approved_permission_seal") != approved_preview_seal or approval.get("user_message") != APPROVAL_MESSAGE or not isinstance(approval.get("approved_at"), str) or not approval["approved_at"] or not isinstance(approval.get("thread_session_id"), str) or not approval["thread_session_id"] ): raise Refusal("live authorization binding mismatch") return { "path": str(path), "sha256": authorization_hash, "approved_controller_sha256": current_tools["controller_sha256"], "approved_dry_run_sha256": approved_preview_sha, "approved_permission_seal": approved_preview_seal, "approved_at": approval["approved_at"], "thread_session_id": approval["thread_session_id"], "user_message": APPROVAL_MESSAGE, } def service_state(override: Path | None = None) -> dict[str, object]: if override is not None: value = json.loads(override.read_bytes()) return { "active_state": value["active_state"], "sub_state": value["sub_state"], "main_pid": int(value["main_pid"]), "matching_processes": int(value["matching_processes"]), "source": "test_override", } result = subprocess.run( [ "systemctl", "--user", "show", "hermes-agent@dualcoachtest.service", "-p", "ActiveState", "-p", "SubState", "-p", "MainPID", ], check=False, capture_output=True, text=True, ) if result.returncode: raise Refusal(f"service state unavailable: {result.stderr.strip()}") fields = dict( line.split("=", 1) for line in result.stdout.splitlines() if "=" in line ) ancestors = {os.getpid()} parent = os.getppid() while parent > 1 and parent not in ancestors: ancestors.add(parent) try: parent = int((Path("/proc") / str(parent) / "stat").read_text().split()[3]) except (OSError, ValueError, IndexError): break matches = 0 for entry in Path("/proc").iterdir(): if not entry.name.isdigit() or int(entry.name) in ancestors: continue try: command = (entry / "cmdline").read_bytes().replace(b"\0", b" ") except OSError: continue if b"dualcoachtest" in command or b"hermes-agent" in command: matches += 1 return { "active_state": fields.get("ActiveState"), "sub_state": fields.get("SubState"), "main_pid": int(fields.get("MainPID", "-1")), "matching_processes": matches, "source": "systemd_and_proc", } def require_inactive(value: dict[str, object]) -> None: if ( value["active_state"], value["sub_state"], value["main_pid"], value["matching_processes"], ) != ("inactive", "dead", 0, 0): raise Refusal("service must be inactive/dead with no matching process") def _strict_names(root: Path, expected: set[str]) -> None: actual = {path.name for path in root.iterdir()} if actual != expected: raise Refusal( f"cleanup directory has unknown or missing entries: {root}: " f"{sorted(actual ^ expected)}" ) def _import_runtime_types() -> tuple[type[Any], type[Any]]: wheel = str(HERMES_WHEEL) if wheel not in sys.path: sys.path.insert(0, wheel) outbox_module = importlib.import_module( "gateway.platforms.telegram_nutrition_onboarding_publication_outbox" ) membership_module = importlib.import_module( "gateway.platforms.telegram_staff_membership_gate" ) return ( getattr(outbox_module, "GatewayOnboardingPublicationOutbox"), getattr(membership_module, "MembershipJournal"), ) def validate_target_state(root: Path) -> dict[str, object]: paths = mutation_paths(root) outbox = paths["outbox"] membership = paths["membership"] outbox_tree = tree_inventory(outbox) membership_tree = tree_inventory(membership) supplemental_tree = compact(tree_inventory(paths["supplemental_archive"])) if not outbox_tree["exists"] or not membership_tree["exists"]: raise Refusal("target operational directories are absent") if supplemental_tree["exists"]: raise Refusal("supplemental cleanup root must be absent before execution") _strict_names(outbox, OUTBOX_NAMES) _strict_names(membership, MEMBERSHIP_NAMES) try: primary = json.loads((outbox / "ledger.json").read_bytes()) emergency = json.loads((outbox / "emergency.json").read_bytes()) callbacks = json.loads((outbox / "owner-callbacks.json").read_bytes()) except (OSError, json.JSONDecodeError) as exc: raise Refusal("outbox JSON is malformed") from exc primary_rows = primary.get("records") if ( primary.get("schema") != "telegram-nutrition-onboarding-publication-outbox-v2" or not isinstance(primary_rows, list) or not primary_rows or emergency != { "schema": "telegram-nutrition-onboarding-publication-outbox-v2", "records": [], } or callbacks != { "schema": "telegram-nutrition-onboarding-owner-callback-v1", "records": [], } ): raise Refusal("outbox does not match the exact supplemental state") for row in primary_rows: if ( not isinstance(row, dict) or row.get("session_id") != SESSION or row.get("route") != list(ROUTE) or row.get("role") != "customer" or row.get("state") != "COMMITTED" ): raise Refusal("foreign, mixed, or nonterminal outbox row") outbox_type, journal_type = _import_runtime_types() try: authenticated = outbox_type(root, initialize=False) parsed = authenticated.records() if authenticated.emergency_records(): raise Refusal("emergency outbox is not empty") rows = journal_type(membership / "events.jsonl").verify() except Refusal: raise except Exception as exc: raise Refusal("outbox HMAC or membership hash-chain validation failed") from exc if len(parsed) != len(primary_rows): raise Refusal("outbox parser projection mismatch") if ( len(rows) != 1 or rows[0].get("event") != "subscription_armed" or rows[0].get("customer_user_ids") != [USER_ID] or not isinstance(rows[0].get("subscription_epoch_id"), str) ): raise Refusal("foreign or mixed membership state") return { "outbox": compact(outbox_tree), "membership": compact(membership_tree), "supplemental_archive": supplemental_tree, "publication_records": len(primary_rows), "membership_rows": len(rows), "membership_epoch": rows[0]["subscription_epoch_id"], } def dry_run( profile: Path, output: Path, *, live_authorization_file: Path | None = None, service_override: Path | None = None, ) -> dict[str, Any]: profile = profile.absolute() _require_private(profile, directory=True) is_live = profile.resolve() == LIVE.resolve() if is_live: authorization = ( None if live_authorization_file is None else live_authorization_binding(live_authorization_file) ) else: if live_authorization_file is not None: raise Refusal("live authorization cannot target a disposable profile") authorization = None state = service_state(service_override) require_inactive(state) target = validate_target_state(profile) payload = { "schema": "task27-supplemental-cleanup-permission-v1", "target": str(profile), "identity": { "customer_key": CUSTOMER, "session_id": SESSION, "user_id": USER_ID, "route": list(ROUTE), }, "cleanup_receipt": cleanup_receipt_binding(), "live_authorization": authorization, "tools": tool_binding(), "service": state, "exact_state": { name: target[name] for name in ("outbox", "membership", "supplemental_archive") }, "observed": { "publication_records": target["publication_records"], "membership_rows": target["membership_rows"], "membership_epoch": target["membership_epoch"], }, "proposed_mutations": proposed_mutations(), } if is_live and authorization is not None: preview_payload, _, _ = approved_preview() preview_projection = dict(payload) preview_projection["live_authorization"] = None if preview_projection != preview_payload: raise Refusal("live state drifted from the approved preview") seal = sha_bytes(canonical(payload)) receipt = { "schema": SCHEMA, "mode": "dry-run", "status": ( "AWAITING_AUTHORIZATION" if is_live and authorization is None else "READY" ), "permission_payload": payload, "permission_seal": seal, } atomic_json(output, receipt) return receipt def _remove(path: Path) -> None: if path.is_symlink(): raise Refusal(f"refusing to remove symlink: {path}") if path.exists(): for item in sorted(path.rglob("*"), reverse=True): if not item.is_symlink(): item.chmod(0o700 if item.is_dir() else 0o600) path.chmod(0o700) shutil.rmtree(path) def _copy(source: Path, destination: Path) -> None: shutil.copytree(source, destination, symlinks=False) def _freeze(root: Path) -> None: for path in sorted(root.rglob("*"), key=lambda item: len(item.parts), reverse=True): if path.is_symlink(): raise Refusal(f"archive contains a symlink: {path}") path.chmod(0o500 if path.is_dir() else 0o400) root.chmod(0o500) @contextlib.contextmanager def authority_lock(root: Path) -> Iterator[None]: data = root / "data" _require_private(data, directory=True) path = data / ".profile-authority.lock" _require_private(path, directory=False) descriptor = os.open( path, os.O_RDWR | os.O_NOFOLLOW | os.O_CLOEXEC, ) try: info = os.fstat(descriptor) named = path.lstat() if ( not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_nlink != 1 or (info.st_dev, info.st_ino) != (named.st_dev, named.st_ino) ): raise Refusal("profile authority lock is unsafe") fcntl.flock(descriptor, fcntl.LOCK_EX) yield finally: fcntl.flock(descriptor, fcntl.LOCK_UN) os.close(descriptor) def _active_target_matches(root: Path) -> list[str]: needles = (CUSTOMER.encode(), SESSION.encode(), USER_ID.encode()) matches: list[str] = [] data = root / "data" if not data.exists(): return matches for path in data.rglob("*"): if not path.is_file(): continue relative = path.relative_to(data) if any(part in ARCHIVE_ROOT_NAMES for part in relative.parts): continue try: payload = path.read_bytes() except OSError as exc: raise Refusal(f"terminal scan cannot read {path}") from exc if any(needle in payload for needle in needles): matches.append(relative.as_posix()) return matches def execute( profile: Path, permission_file: Path, seal: str, output: Path, *, fault: str | None = None, live_authorization_file: Path | None = None, service_override: Path | None = None, ) -> dict[str, Any]: if output.exists() or output.is_symlink(): raise Refusal(f"output already exists; receipt reuse refused: {output}") profile = profile.absolute() is_live = profile.resolve() == LIVE.resolve() if is_live: if output.resolve() != LIVE_OUTPUT.resolve(): raise Refusal("live receipt output path mismatch") if live_authorization_file is None: raise Refusal("live execution requires exact authorization") authorization = live_authorization_binding(live_authorization_file) else: if live_authorization_file is not None: raise Refusal("live authorization cannot target a disposable profile") authorization = None marker = profile / MARKER if ( not marker.is_file() or marker.is_symlink() or marker.read_text().strip() != MARKER_VALUE ): raise Refusal("execution requires an explicit disposable-copy marker") _require_private(profile, directory=True) try: permission = json.loads(permission_file.read_bytes()) except (OSError, json.JSONDecodeError) as exc: raise Refusal("permission receipt is invalid") from exc payload = permission.get("permission_payload") expected_seal = permission.get("permission_seal") if ( not isinstance(payload, dict) or permission.get("schema") != SCHEMA or permission.get("mode") != "dry-run" or permission.get("status") != "READY" or expected_seal != sha_bytes(canonical(payload)) or seal != expected_seal ): raise Refusal("permission seal mismatch") if payload.get("target") != str(profile): raise Refusal("permission target mismatch") if payload.get("cleanup_receipt") != cleanup_receipt_binding(): raise Refusal("cleanup receipt drift") if payload.get("live_authorization") != authorization: raise Refusal("live authorization drift") if payload.get("tools") != tool_binding(): raise Refusal("controller or delivered tool drift") if is_live: preview_payload, _, _ = approved_preview() preview_projection = dict(payload) preview_projection["live_authorization"] = None if preview_projection != preview_payload: raise Refusal("permission is not the approved preview transformation") require_inactive(service_state(service_override)) current = validate_target_state(profile) for name in ("outbox", "membership", "supplemental_archive"): if current[name] != payload["exact_state"][name]: raise Refusal(f"source drift: {name}") paths = mutation_paths(profile) operation = ( sha_bytes(f"{SCHEMA}:{seal}".encode())[:32] if is_live else uuid.uuid4().hex ) archive = paths["supplemental_archive"] / "archives" / operation with authority_lock(profile): require_inactive(service_state(service_override)) locked = validate_target_state(profile) for name in ("outbox", "membership", "supplemental_archive"): if locked[name] != payload["exact_state"][name]: raise Refusal(f"source drift under authority lock: {name}") rollback = Path(tempfile.mkdtemp(prefix="task27-supplemental-rollback-")) backup = rollback / "state" mutation_started = False try: backup.mkdir(mode=0o700) _copy(paths["outbox"], backup / "outbox") _copy(paths["membership"], backup / "membership") files = archive / "files/data/onboarding" files.mkdir(parents=True, mode=0o700) mutation_started = True _copy(paths["outbox"], files / paths["outbox"].name) _copy(paths["membership"], files / paths["membership"].name) archived = { "outbox": compact(tree_inventory(files / paths["outbox"].name)), "membership": compact(tree_inventory(files / paths["membership"].name)), } expected_archived = { name: { **payload["exact_state"][name], "root": str(files / paths[name].name), } for name in ("outbox", "membership") } if archived != expected_archived: raise Refusal("supplemental archive copy verification failed") manifest = { "schema": "task27-supplemental-archive-v1", "operation_id": operation, "identity": payload["identity"], "cleanup_receipt": payload["cleanup_receipt"], "live_authorization": authorization, "receipt_output": str(output), "tools": payload["tools"], "source": { name: payload["exact_state"][name] for name in ("outbox", "membership") }, "archive": archived, } manifest_path = archive / "manifest.json" atomic_json(manifest_path, manifest) consumption = None if is_live: consumption = { "schema": "task27-supplemental-authorization-consumption-v1", "status": "CONSUMED_ON_COMMIT", "authorization": authorization, "permission_seal": seal, "operation_id": operation, "receipt_output": str(output), } atomic_json( archive / "authorization-consumed.json", consumption, ) if fault == "after_archive_copy": raise RuntimeError("injected fault after_archive_copy") _remove(paths["outbox"]) _remove(paths["membership"]) if fault == "after_prune": raise RuntimeError("injected fault after_prune") matches = _active_target_matches(profile) if matches: raise Refusal(f"active target residue remains: {matches}") require_inactive(service_state(service_override)) terminal = { "schema": SCHEMA, "mode": "execute", "status": "COMMITTED", "execution_target": "live" if is_live else "disposable", "profile": str(profile), "permission_seal": seal, "operation_id": operation, "cleanup_receipt": payload["cleanup_receipt"], "live_authorization": authorization, "authorization_consumption": consumption, "tools": payload["tools"], "observed": payload["observed"], "archive": { "root": str(archive), "manifest_sha256": sha_file(manifest_path), "source": { name: payload["exact_state"][name] for name in ("outbox", "membership") }, "copied": archived, }, "terminal": { "outbox_absent": not paths["outbox"].exists(), "membership_absent": not paths["membership"].exists(), "active_target_matches": 0, "service_inactive": True, }, "proposed_live_mutations": proposed_mutations(), } atomic_json(archive / "receipt.json", terminal) _freeze(archive) atomic_json(output, terminal, 0o400) return terminal except BaseException: if mutation_started: _remove(paths["outbox"]) _remove(paths["membership"]) _remove(paths["supplemental_archive"]) _copy(backup / "outbox", paths["outbox"]) _copy(backup / "membership", paths["membership"]) raise finally: shutil.rmtree(rollback) def parser() -> argparse.ArgumentParser: value = argparse.ArgumentParser() commands = value.add_subparsers(dest="command", required=True) dry = commands.add_parser("dry-run") dry.add_argument("--profile", type=Path, required=True) dry.add_argument("--output", type=Path, required=True) dry.add_argument("--live-authorization-file", type=Path) dry.add_argument("--test-service-state", type=Path, help=argparse.SUPPRESS) run = commands.add_parser("execute") run.add_argument("--profile", type=Path, required=True) run.add_argument("--permission-file", type=Path, required=True) run.add_argument("--permission-seal", required=True) run.add_argument("--output", type=Path, required=True) run.add_argument("--live-authorization-file", type=Path) run.add_argument("--fault", choices=("after_archive_copy", "after_prune")) run.add_argument("--test-service-state", type=Path, help=argparse.SUPPRESS) return value def main(argv: list[str] | None = None) -> int: args = parser().parse_args(argv) try: if args.command == "dry-run": result = dry_run( args.profile, args.output, live_authorization_file=args.live_authorization_file, service_override=args.test_service_state, ) else: result = execute( args.profile, args.permission_file, args.permission_seal, args.output, fault=args.fault, live_authorization_file=args.live_authorization_file, service_override=args.test_service_state, ) print( json.dumps( { "status": result["status"], "permission_seal": result["permission_seal"], "output": str(args.output), }, sort_keys=True, ) ) return 0 except BaseException as exc: print( json.dumps( { "status": "REFUSED", "error": str(exc), "type": type(exc).__name__, }, sort_keys=True, ), file=sys.stderr, ) if os.environ.get("TASK27_TRACEBACK") == "1": traceback.print_exc() return 2 if __name__ == "__main__": raise SystemExit(main())