#!/usr/bin/env python3 """Verification-only v5 successor for sealed Task26 lifecycle cleanup.""" from __future__ import annotations import argparse import hashlib import json import os import shutil import stat import subprocess import sys import tempfile from pathlib import Path from typing import Any FLAGS = os.O_CLOEXEC | getattr(os, "O_NOFOLLOW", 0) DIR_FLAGS = os.O_RDONLY | os.O_DIRECTORY | FLAGS PROC_READ_LIMIT = 1024 * 1024 SERVICE_NAME = "hermes-gateway-dualcoachtest.service" V4_ROOT = Path(__file__).parent.parent / "task26-post-lifecycle-cleanup-v4-4a6c7ee5-st_01a00e68" V4_SEAL_SHA256 = "188e9e630b53e9329ae185fc8c007aecb7988c16d174d325d68e8e004edb38e9" OBSERVER_ARTIFACTS = ( "gateway_start_observer.py", "lifecycle_observer.py", "observer_v2.py", "observer_v3.py", "observer_v4.py", "observer_v5.py", "observer_v6.py", "observer_v61.py", "observer_v62.py", "observer_v63.py", "observer_v64.py", ) CUSTOMER = "task26_live_2e_r2_20260815_8527916639" ACTOR = "8527916639" OWNER = "8693203710" SESSION = "005959f8d90c5a57f664d8022a9d5049" DRAFT = "e7d63548ceecf2fd" REVISION = "58f2cda7abd532c0d6e145bbb3f3216609b004374adab8bfae2ee28dc5d7d660" DELIVERY_KEY = f"{DRAFT}:{REVISION}" class CleanupError(RuntimeError): pass def canon(value: Any) -> bytes: return (json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n").encode() def digest(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def file_digest(path: Path) -> str: return digest(path.read_bytes()) def write_all(fd: int, raw: bytes) -> None: """Write every byte or fail before the caller can publish the temp file.""" remaining = memoryview(raw) while remaining: try: written = os.write(fd, remaining) except InterruptedError: continue if isinstance(written, bool) or not isinstance(written, int) or written <= 0 or written > len(remaining): raise CleanupError(f"invalid write count: {written!r}") remaining = remaining[written:] def atomic(path: Path, raw: bytes, mode: int = 0o600) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) fd, tmp = tempfile.mkstemp(prefix=".task26-cleanup-", dir=path.parent) try: try: os.fchmod(fd, mode) write_all(fd, raw) os.fsync(fd) finally: try: if fd >= 0: os.close(fd) finally: fd = -1 os.replace(tmp, path) dfd = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY | FLAGS) try: os.fsync(dfd) finally: os.close(dfd) finally: try: os.unlink(tmp) except FileNotFoundError: pass def safe_read(path: Path, modes: set[int] | None = None) -> bytes: fd = os.open(path, os.O_RDONLY | FLAGS) try: before = os.fstat(fd) allowed = modes or {0o400, 0o500, 0o600, 0o644, 0o664, 0o700, 0o755, 0o775} if ( not stat.S_ISREG(before.st_mode) or before.st_uid != os.getuid() or before.st_nlink != 1 or stat.S_IMODE(before.st_mode) not in allowed ): raise CleanupError(f"unsafe file: {path}") chunks: list[bytes] = [] while chunk := os.read(fd, 1024 * 1024): chunks.append(chunk) after = os.fstat(fd) if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != ( after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns ): raise CleanupError(f"concurrent file mutation: {path}") return b"".join(chunks) finally: os.close(fd) def load_json(path: Path) -> Any: try: return json.loads(safe_read(path)) except json.JSONDecodeError as exc: raise CleanupError(f"invalid JSON: {path}") from exc def stable_tree_read(path: Path) -> bytes: """Read unrelated-profile content stably; Git object hardlinks are valid.""" before = path.stat(follow_symlinks=False) if not stat.S_ISREG(before.st_mode) or before.st_uid != os.getuid(): raise CleanupError(f"unsafe unrelated-profile file: {path}") raw = path.read_bytes() after = path.stat(follow_symlinks=False) if (before.st_dev, before.st_ino, before.st_size, before.st_mtime_ns, before.st_ctime_ns) != (after.st_dev, after.st_ino, after.st_size, after.st_mtime_ns, after.st_ctime_ns): raise CleanupError(f"concurrent unrelated-profile mutation: {path}") return raw def tree_inventory(root: Path, *, exclude: tuple[str, ...] = ()) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for base, dirs, files in os.walk(root, followlinks=False): dirs.sort() files.sort() relbase = Path(base).relative_to(root) dirs[:] = [d for d in dirs if not any((relbase / d).as_posix().startswith(x) for x in exclude)] for name in list(dirs): path = Path(base) / name rel = path.relative_to(root).as_posix() st = path.lstat() if stat.S_ISLNK(st.st_mode): rows.append({"path": rel, "type": "L", "target": os.readlink(path), "mode": f"{stat.S_IMODE(st.st_mode):04o}"}) dirs.remove(name) elif stat.S_ISDIR(st.st_mode): rows.append({"path": rel, "type": "D", "mode": f"{stat.S_IMODE(st.st_mode):04o}"}) else: raise CleanupError(f"unsupported unrelated-profile entry: {path}") for name in files: path = Path(base) / name rel = path.relative_to(root).as_posix() if any(rel.startswith(x) for x in exclude): continue st = path.lstat() if stat.S_ISLNK(st.st_mode): rows.append({"path": rel, "type": "L", "target": os.readlink(path), "mode": f"{stat.S_IMODE(st.st_mode):04o}"}) continue if not stat.S_ISREG(st.st_mode): raise CleanupError(f"unsupported tree entry: {path}") raw = stable_tree_read(path) rows.append({"path": rel, "type": "F", "size": len(raw), "sha256": digest(raw), "mode": f"{stat.S_IMODE(st.st_mode):04o}"}) return rows def inventory_digest(rows: list[dict[str, Any]]) -> str: return digest(canon(rows)) def sealed_tree_digest(root: Path) -> str: rows: list[dict[str, str]] = [] for base, dirs, files in os.walk(root, followlinks=False): dirs.sort() files.sort() for name in dirs: path = Path(base) / name st = path.lstat() if not stat.S_ISDIR(st.st_mode): raise CleanupError(f"unsafe sealed archive directory: {path}") rows.append({"path": path.relative_to(root).as_posix(), "type": "D", "mode": f"{stat.S_IMODE(st.st_mode):04o}"}) for name in files: path = Path(base) / name st = path.lstat() if not stat.S_ISREG(st.st_mode): raise CleanupError(f"unsafe sealed archive entry: {path}") rows.append({"path": path.relative_to(root).as_posix(), "type": "F", "mode": f"{stat.S_IMODE(st.st_mode):04o}", "sha256": file_digest(path)}) return digest(canon(rows)) def authority_paths(profile: Path, contract: dict[str, Any]) -> list[str]: paths: list[str] = [] for rel in contract["clear_scopes"]: if (profile / rel).exists(): paths.append(rel) data = profile / "data" protected = set(contract["protected_data_roots"]) if data.exists(): for child in sorted(data.iterdir()): rel = f"data/{child.name}" if child.name in protected or rel in paths: continue if child.name in contract["allowed_dynamic_data_roots"] or any( child.name.startswith(prefix) for prefix in contract["allowed_dynamic_data_prefixes"] ): paths.append(rel) else: raise CleanupError(f"unknown data authority root: {rel}") outer: list[str] = [] for rel in sorted(set(paths), key=lambda value: (value.count("/"), value)): if not any(rel == parent or rel.startswith(parent + "/") for parent in outer): outer.append(rel) return sorted(outer) def scoped_rows(profile: Path, scopes: list[str]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for rel in scopes: path = profile / rel if path.is_file(): raw = safe_read(path) rows.append({"path": rel, "size": len(raw), "sha256": digest(raw), "mode": f"{stat.S_IMODE(path.stat().st_mode):04o}"}) elif path.is_dir(): for row in tree_inventory(path): if row.get("type") == "D": continue if row.get("type") != "F": raise CleanupError(f"symlink in active authority: {rel}/{row['path']}") row["path"] = f"{rel}/{row['path']}" rows.append(row) else: raise CleanupError(f"unsupported authority scope: {rel}") return sorted(rows, key=lambda row: row["path"]) def service_show(service: str) -> dict[str, str]: result = subprocess.run( ["/usr/bin/systemctl", "--user", "show", service, "--property=ActiveState,SubState,MainPID,ControlPID,ExecMainPID,FragmentPath,UnitFileState"], text=True, capture_output=True, check=False, ) if result.returncode != 0: raise CleanupError(f"systemctl show failed: {result.stderr.strip()}") values = dict(line.split("=", 1) for line in result.stdout.splitlines() if "=" in line) if set(values) != {"ActiveState", "SubState", "MainPID", "ControlPID", "ExecMainPID", "FragmentPath", "UnitFileState"}: raise CleanupError("unexpected systemctl authority output") return { "active": values["ActiveState"], "sub": values["SubState"], "pid": values["MainPID"], "control_pid": values["ControlPID"], "exec_main_pid": values["ExecMainPID"], "fragment": values["FragmentPath"], "unit_file_state": values["UnitFileState"], } def _read_proc_file(proc_fd: int, name: str) -> bytes: """Read one proc pseudo-file through a pinned process directory.""" fd = os.open(name, os.O_RDONLY | FLAGS, dir_fd=proc_fd) try: chunks: list[bytes] = [] total = 0 while True: chunk = os.read(fd, min(65536, PROC_READ_LIMIT + 1 - total)) if not chunk: break chunks.append(chunk) total += len(chunk) if total > PROC_READ_LIMIT: raise CleanupError(f"oversized /proc entry: {name}") return b"".join(chunks) finally: os.close(fd) def _parse_proc_stat(raw: bytes, expected_pid: int) -> dict[str, int]: try: text = raw.decode("ascii").strip() prefix, fields_text = text.rsplit(") ", 1) pid_text, _comm = prefix.split(" (", 1) fields = fields_text.split() if int(pid_text) != expected_pid or len(fields) < 20: raise ValueError # fields starts at kernel stat field 3 (state). return { "ppid": int(fields[1]), "session_id": int(fields[3]), "start_time_ticks": int(fields[19]), } except (UnicodeDecodeError, ValueError) as exc: raise CleanupError(f"malformed /proc/{expected_pid}/stat") from exc def _decode_cmdline(raw: bytes, pid: int) -> list[str]: if not raw: return [] if not raw.endswith(b"\0"): raise CleanupError(f"malformed /proc/{pid}/cmdline") return [part.decode("utf-8", "surrogateescape") for part in raw[:-1].split(b"\0")] def _follow_option(argv: list[str]) -> bool: for arg in argv[1:]: if arg == "--follow" or (arg.startswith("-") and not arg.startswith("--") and "f" in arg[1:]): return True return False def _process_reasons(argv: list[str], profile: Path, service: str) -> list[str]: if not argv: return [] joined = "\0".join(argv) lower = joined.lower() executable = Path(argv[0]).name.lower() profile_text = str(profile.resolve()) target_context = service in joined or profile_text in joined or "dualcoachtest" in lower or "task26" in lower reasons: list[str] = [] if service in joined: reasons.append("target_service_reference") if profile_text in joined: reasons.append("exact_profile_path_reference") if any(name in joined for name in OBSERVER_ARTIFACTS) or ( "live-v" in lower and "-events.jsonl" in lower ): reasons.append("observer_script_or_event") if executable == "inotifywait" and target_context: reasons.append("target_inotify_watcher") if executable == "journalctl" and _follow_option(argv) and target_context: reasons.append("target_journal_follower") if "task26" in lower and any(token in lower for token in ("monitor", "watch", "follow", "observer")): reasons.append("task_owned_monitor") return sorted(set(reasons)) def enumerate_processes( profile: Path, service: str, *, proc_root: Path = Path("/proc"), self_pid: int | None = None, ) -> dict[str, Any]: """Capture a stable same-UID process snapshot and target classifications.""" excluded_pid = os.getpid() if self_pid is None else self_pid evidence: list[dict[str, Any]] = [] races: list[dict[str, Any]] = [] violations: list[dict[str, Any]] = [] try: names = sorted((name for name in os.listdir(proc_root) if name.isdecimal()), key=int) except OSError as exc: raise CleanupError(f"cannot enumerate {proc_root}") from exc for name in names: pid = int(name) try: proc_fd = os.open(proc_root / name, DIR_FLAGS) except FileNotFoundError: races.append({"pid": pid, "race": "disappeared_before_open"}) continue except PermissionError: continue except OSError as exc: if exc.errno in (2, 3): races.append({"pid": pid, "race": "disappeared_before_open"}) continue raise CleanupError(f"cannot open /proc/{pid}") from exc try: owner = os.fstat(proc_fd).st_uid if owner != os.getuid(): continue try: before = _parse_proc_stat(_read_proc_file(proc_fd, "stat"), pid) cmdline = _decode_cmdline(_read_proc_file(proc_fd, "cmdline"), pid) after = _parse_proc_stat(_read_proc_file(proc_fd, "stat"), pid) except (FileNotFoundError, ProcessLookupError): races.append({"pid": pid, "race": "disappeared_during_read"}) continue except OSError as exc: if exc.errno in (2, 3): races.append({"pid": pid, "race": "disappeared_during_read"}) continue raise CleanupError(f"cannot inspect same-uid /proc/{pid}") from exc if before["start_time_ticks"] != after["start_time_ticks"]: races.append({"pid": pid, "race": "pid_reused_during_read", "start_time_ticks_before": before["start_time_ticks"], "start_time_ticks_after": after["start_time_ticks"]}) continue evidence.append({"pid": pid, **after, "cmdline": cmdline}) finally: os.close(proc_fd) by_pid = {row["pid"]: row for row in evidence} ancestors: set[int] = set() cursor = excluded_pid while cursor in by_pid and cursor not in ancestors: ancestors.add(cursor) cursor = by_pid[cursor]["ppid"] excluded_identities: list[dict[str, int]] = [] controller_name = Path(__file__).name for row in evidence: pid = row["pid"] verifier_ancestor = pid in ancestors and any(controller_name in arg for arg in row["cmdline"]) if pid == excluded_pid or verifier_ancestor: row["classification"] = "self_excluded" if pid == excluded_pid else "verifier_command_excluded" excluded_identities.append({"pid": pid, "start_time_ticks": row["start_time_ticks"]}) continue reasons = _process_reasons(row["cmdline"], profile, service) row["classification"] = "target_owned" if reasons else "unrelated" if reasons: row["reasons"] = reasons violations.append(row) return { "same_uid_processes": evidence, "process_races": races, "violations": violations, "self_pid": excluded_pid, "excluded_verifier_identities": excluded_identities, } def process_poststate(profile: Path, service: str, *, proc_root: Path = Path("/proc")) -> dict[str, Any]: """Use two stable snapshots so births, exits, and PID reuse are observable.""" first = enumerate_processes(profile, service, proc_root=proc_root) second = enumerate_processes(profile, service, proc_root=proc_root) violations = first["violations"] + second["violations"] if violations: details = [(row["pid"], row["start_time_ticks"], row["cmdline"], row["reasons"]) for row in violations] raise CleanupError(f"session-owned background processes remain: {details}") stable_first = {(row["pid"], row["start_time_ticks"]) for row in first["same_uid_processes"]} stable_second = {(row["pid"], row["start_time_ticks"]) for row in second["same_uid_processes"]} return { "schema": "task26-session-process-poststate-v1", "status": "PASS", "passes": 2, "target_processes": [], "first_snapshot": first, "second_snapshot": second, "identity_births": sorted([{"pid": pid, "start_time_ticks": start} for pid, start in stable_second - stable_first], key=lambda row: row["pid"]), "identity_exits": sorted([{"pid": pid, "start_time_ticks": start} for pid, start in stable_first - stable_second], key=lambda row: row["pid"]), } def validate_bindings(args: argparse.Namespace) -> dict[str, Any]: contract_raw = safe_read(args.contract, {0o600}) permission = load_json(args.permission) contract = json.loads(contract_raw) expected_permission = { "schema": "task26-post-lifecycle-cleanup-permission-v5", "approval": "TASK26_COMPLETED_CLEANUP_VERIFICATION_ONLY", "candidate_digest": args.candidate, "hermes_wheel_sha256": args.hermes_wheel, "profile_wheel_sha256": args.profile_wheel, "plan_sha256": args.plan_sha256, "lifecycle_root_sha256": args.lifecycle_root_sha256, "lifecycle_seal_sha256": args.lifecycle_seal_sha256, "actor_id": ACTOR, "customer_key": CUSTOMER, "controller_sha256": file_digest(Path(__file__)), "contract_sha256": digest(contract_raw), "execute_allowed_once": False, "run_id": args.run_id, "v3_seal_sha256": contract["completed_cleanup"]["v3_seal_sha256"], "archive_root_sha256": contract["completed_cleanup"]["archive_root_sha256"], "v4_seal_sha256": V4_SEAL_SHA256, } if permission != expected_permission: raise CleanupError("verification-only permission pins mismatch") if ( contract.get("predecessor", {}).get("v4_seal_sha256") != V4_SEAL_SHA256 or file_digest(V4_ROOT / "SEAL.json") != V4_SEAL_SHA256 ): raise CleanupError("v4 predecessor seal mismatch") pins = contract.get("bindings", {}) required = { "candidate_digest": args.candidate, "hermes_wheel_sha256": args.hermes_wheel, "profile_wheel_sha256": args.profile_wheel, "plan_sha256": args.plan_sha256, "lifecycle_root_sha256": args.lifecycle_root_sha256, "lifecycle_seal_sha256": args.lifecycle_seal_sha256, } if contract.get("schema") != "task26-post-lifecycle-cleanup-contract-v5" or any(pins.get(k) != v for k, v in required.items()): raise CleanupError("contract binding mismatch") completed = contract.get("completed_cleanup", {}) if ( completed.get("status") != "PASS_VERIFIED_DISABLED_ARCHIVED_CLEAN" or completed.get("v3_seal_sha256") != "84d937ef67c1517bfdbca4cabce9fb4bfbe39aa1165458bb3aedd14decc4bff0" or completed.get("archive_root_sha256") != "9f5d72840b4cc1fcb57720a61ebc53529fa00202d9201615e4ebdbb3e047a635" or file_digest(args.v3_seal) != completed["v3_seal_sha256"] ): raise CleanupError("completed cleanup binding mismatch") for path, expected in contract["source_authorities"].items(): if file_digest(Path(path)) != expected: raise CleanupError(f"source authority drift: {path}") service = contract["service"] if file_digest(Path("/usr/bin/systemctl")) != service["systemctl_sha256"]: raise CleanupError("systemctl authority drift") if file_digest(Path(service["fragment_path"])) != service["fragment_sha256"]: raise CleanupError("service unit authority drift") if file_digest(args.apply_patch) != contract["apply_patch_sha256"]: raise CleanupError("apply_patch authority drift") return contract def validate_lifecycle(profile: Path, contract: dict[str, Any], *, require_running: bool) -> dict[str, Any]: if file_digest(profile / "config.yaml") != contract["bindings"]["config_sha256"]: raise CleanupError("current config hash mismatch") if stat.S_IMODE((profile / "config.yaml").stat().st_mode) != 0o664: raise CleanupError("current config mode mismatch") for rel, expected in contract["sealed_live_file_sha256"].items(): if file_digest(profile / rel) != expected: raise CleanupError(f"sealed lifecycle drift: {rel}") registry = load_json(profile / "customers/registry.json") customers = registry.get("customers") if not isinstance(customers, list) or len(customers) != 1: raise CleanupError("requires exactly one customer") customer = customers[0] if customer.get("customer_key") != CUSTOMER or customer.get("enabled") is not True: raise CleanupError("target customer is not uniquely enabled") if str(customer.get("telegram", {}).get("user_id")) != ACTOR or str(registry.get("owner", {}).get("user_id")) != OWNER: raise CleanupError("customer/actor/owner mismatch") bootstrap = load_json(profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json") active = [row for row in bootstrap.get("sessions", []) if isinstance(row, dict) and row.get("state") == "ACTIVE"] if len(active) != 1 or active[0].get("customer_draft", {}).get("customer_key") != CUSTOMER: raise CleanupError("requires exactly one bound ACTIVE lifecycle") delivery_doc = load_json(profile / "data/owner-actions/draft-deliveries.json") if list(delivery_doc) != [DELIVERY_KEY]: raise CleanupError("delivery authority is not exactly one bound row") delivery = delivery_doc[DELIVERY_KEY] if delivery.get("status") != "sent_audited" or str(delivery.get("message_id")) != "232" or delivery.get("customer_key") != CUSTOMER: raise CleanupError("audited delivery mismatch") jobs = profile / "cron/jobs.json" if jobs.exists() and load_json(jobs).get("jobs") != []: raise CleanupError("future jobs remain") emergency = load_json(profile / "data/onboarding/telegram-publication-outbox-v1/emergency.json") if emergency.get("records") != []: raise CleanupError("emergency publications remain") state = load_json(profile / "gateway_state.json") boot_id = Path("/proc/sys/kernel/random/boot_id").read_text().strip() if state.get("boot_id") != boot_id or state.get("gateway_state") != "running": raise CleanupError("gateway runtime authority mismatch") service = service_show(contract["service"]["name"]) if require_running and (service["active"], service["sub"], service["pid"]) != ("active", "running", str(state.get("pid"))): raise CleanupError("service is not the bound active gateway") if require_running and ( service["pid"] != contract["service"]["initial_main_pid"] or state.get("boot_id") != contract["service"]["initial_boot_id"] or service["fragment"] != contract["service"]["fragment_path"] ): raise CleanupError("service/runtime pin mismatch") return {"customer_key": CUSTOMER, "actor_id": ACTOR, "owner_id": OWNER, "session_id": SESSION, "draft_id": DRAFT, "revision": REVISION, "message_id": "232", "boot_id": boot_id, "main_pid": str(state["pid"])} def baseline_other_profiles(contract: dict[str, Any]) -> dict[str, str]: result: dict[str, str] = {} for root in contract["unrelated_profiles"]: path = Path(root) result[root] = inventory_digest(tree_inventory(path)) return result def archive_prestate(args: argparse.Namespace, contract: dict[str, Any], scopes: list[str], rows: list[dict[str, Any]], ids: dict[str, Any], unrelated: dict[str, str]) -> Path: final = args.archive_root / args.run_id pending = args.archive_root / f".pending-{args.run_id}" if final.exists() or pending.exists(): raise CleanupError("one-use archive already exists") pending.mkdir(parents=True, mode=0o700) payload = pending / "payload" payload.mkdir(mode=0o700) try: for row in rows: raw = safe_read(args.profile / row["path"]) if digest(raw) != row["sha256"]: raise CleanupError(f"archive source drift: {row['path']}") atomic(payload / row["path"], raw) config_raw = safe_read(args.profile / "config.yaml", {0o664}) atomic(payload / "config.yaml", config_raw) service = service_show(contract["service"]["name"]) atomic(payload / "service-prestate.json", canon(service)) manifest = { "schema": "task26-post-lifecycle-cleanup-archive-v3", "run_id": args.run_id, "bindings": contract["bindings"], "active_lifecycle": ids, "scopes": scopes, "entries": rows, "config_entry": {"path": "config.yaml", "size": len(config_raw), "sha256": digest(config_raw), "mode": "0664"}, "service_prestate_sha256": file_digest(payload / "service-prestate.json"), "unrelated_profiles_pre_sha256": unrelated, "archive_committed_before_first_mutation": True, "active_and_audited_history_byte_exact": True, } manifest["evidence_digest"] = digest(canon(manifest)) atomic(pending / "manifest.json", canon(manifest)) os.rename(pending, final) return final except Exception: if pending.exists(): shutil.rmtree(pending) raise def run_checked(command: list[str], *, cwd: Path | None = None, env: dict[str, str] | None = None) -> dict[str, Any]: result = subprocess.run(command, cwd=cwd, env=env, text=True, capture_output=True, check=False) receipt = {"command": command, "cwd": str(cwd) if cwd else None, "exit_code": result.returncode, "stdout": result.stdout, "stderr": result.stderr} if result.returncode != 0: raise CleanupError(f"command failed: {command!r}: {result.stderr.strip()}") return receipt def disable_customer(args: argparse.Namespace, archive: Path) -> dict[str, Any]: command = [str(args.python), "-m", "checkin_cli.customer_admin", "--registry", str(args.profile / "customers/registry.json"), "disable", CUSTOMER] env = dict(os.environ) env["PYTHONPATH"] = str(args.customer_admin_root) receipt = run_checked(command, cwd=args.customer_admin_root, env=env) registry = load_json(args.profile / "customers/registry.json") if registry["customers"][0].get("enabled") is not False: raise CleanupError("canonical disable did not persist") receipt.update({"schema": "task26-canonical-disable-receipt-v1", "status": "PASS", "customer_key": CUSTOMER, "registry_sha256": file_digest(args.profile / "customers/registry.json")}) atomic(archive / "disable-receipt.json", canon(receipt)) return receipt def patch_delivery_gate(args: argparse.Namespace, archive: Path) -> dict[str, Any]: patch = "*** Begin Patch\n*** Update File: config.yaml\n@@\n- delivery_enabled: true\n+ delivery_enabled: false\n*** End Patch\n" result = subprocess.run([str(args.apply_patch)], cwd=args.profile, input=patch, text=True, capture_output=True, check=False) if result.returncode != 0: raise CleanupError(f"apply_patch config edit failed: {result.stderr.strip()}") raw = safe_read(args.profile / "config.yaml", {0o664}) text = raw.decode() required = (" enabled: true", " delivery_enabled: false", " activation: false", " delivery: false") if any(token not in text for token in required) or stat.S_IMODE((args.profile / "config.yaml").stat().st_mode) != 0o664: raise CleanupError("delivery config postcondition mismatch") receipt = {"schema": "task26-delivery-gate-disable-receipt-v1", "status": "PASS", "tool": str(args.apply_patch), "patch_sha256": digest(patch.encode()), "config_sha256": digest(raw), "mode": "0664", "stdout": result.stdout, "stderr": result.stderr} atomic(archive / "config-disable-receipt.json", canon(receipt)) return receipt def qa_and_stop(args: argparse.Namespace, contract: dict[str, Any], archive: Path, delivery_before: str, unrelated: dict[str, str]) -> dict[str, Any]: service = contract["service"]["name"] cursor_result = run_checked(["/usr/bin/journalctl", "--user", "-u", service, "-n", "0", "--show-cursor", "--no-pager", "-o", "cat"]) cursor_line = next((line for line in cursor_result["stdout"].splitlines() if line.startswith("-- cursor:")), None) if not cursor_line: raise CleanupError("could not arm journal cursor") cursor = cursor_line.split(":", 1)[1].strip() active_after_disable = service_show(service) if (active_after_disable["active"], active_after_disable["sub"]) != ("active", "running"): raise CleanupError("gateway did not remain active after disable") if file_digest(args.profile / "data/owner-actions/draft-deliveries.json") != delivery_before: raise CleanupError("delivery authority changed after disable") if baseline_other_profiles(contract) != unrelated: raise CleanupError("unrelated profile changed during disable QA") stop = run_checked([str(args.hermes_python), "-m", "hermes_cli.main", "--profile", "dualcoachtest", "gateway", "stop"], cwd=args.hermes_source) stopped = service_show(service) if (stopped["active"], stopped["sub"], stopped["pid"]) != ("inactive", "dead", "0"): raise CleanupError("gateway did not stop inactive/dead MainPID0") journal = run_checked(["/usr/bin/journalctl", "--user", "-u", service, "--after-cursor", cursor, "--no-pager", "-o", "cat"]) suspicious = [line for line in journal["stdout"].splitlines() if CUSTOMER in line and any(token in line.lower() for token in ("send", "deliver", "message_id"))] if suspicious: raise CleanupError("customer delivery observed after disable") receipt = {"schema": "task26-restart-after-disable-qa-v1", "status": "PASS", "restart_required": False, "gateway_remained_running_after_disable": True, "journal_cursor": cursor, "customer_delivery_events_after_disable": 0, "delivery_authority_unchanged": True, "unrelated_profiles_unchanged": True, "stop_command": stop["command"], "service_post_stop": stopped, "journal_sha256": digest(journal["stdout"].encode())} atomic(archive / "restart-after-disable-qa.json", canon(receipt)) atomic(archive / "journal-after-disable.txt", journal["stdout"].encode()) return receipt def move_scopes(profile: Path, archive: Path, scopes: list[str]) -> list[str]: dest = archive / "cleared-originals" dest.mkdir(mode=0o700) moved: list[str] = [] try: for rel in scopes: source = profile / rel if not source.exists(): continue target = dest / rel target.parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.rename(source, target) moved.append(rel) if any((profile / rel).exists() for rel in scopes): raise CleanupError("reset scope remains live") return moved except Exception: for rel in reversed(moved): source = dest / rel target = profile / rel target.parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.rename(source, target) raise def restore_prestate(args: argparse.Namespace, archive: Path, moved: list[str], was_running: bool) -> None: cleared = archive / "cleared-originals" for rel in reversed(moved): source = cleared / rel if source.exists() and not (args.profile / rel).exists(): (args.profile / rel).parent.mkdir(parents=True, exist_ok=True, mode=0o700) os.rename(source, args.profile / rel) for rel in ("customers/registry.json", "config.yaml"): source = archive / "payload" / rel if source.exists(): mode = 0o664 if rel == "config.yaml" else 0o600 atomic(args.profile / rel, safe_read(source, {0o600}), mode) if was_running and service_show("hermes-gateway-dualcoachtest.service")["active"] != "active": run_checked([str(args.hermes_python), "-m", "hermes_cli.main", "--profile", "dualcoachtest", "gateway", "start"], cwd=args.hermes_source) def postconditions(args: argparse.Namespace, contract: dict[str, Any], archive: Path, unrelated: dict[str, str]) -> dict[str, Any]: service = service_show(contract["service"]["name"]) forbidden = ["customers", "data/customers", "data/onboarding", "data/owner-actions", "gateway.lock", "gateway.pid", "gateway_state.json", "state.db", "state.db-shm", "state.db-wal"] present = [rel for rel in forbidden if (args.profile / rel).exists()] if present: raise CleanupError(f"live cleanup artifacts remain: {present}") jobs = args.profile / "cron/jobs.json" if jobs.exists() and load_json(jobs).get("jobs") != []: raise CleanupError("future jobs remain post-cleanup") if (service["active"], service["sub"], service["pid"], service["control_pid"]) != ("inactive", "dead", "0", "0"): raise CleanupError("service postcondition mismatch") processes = process_poststate(args.profile, contract["service"]["name"]) config = safe_read(args.profile / "config.yaml", {0o664}) if b"delivery_enabled: false" not in config or b" enabled: true" not in config or b" activation: false" not in config or b" delivery: false" not in config: raise CleanupError("config postcondition mismatch") unrelated_after = baseline_other_profiles(contract) if unrelated_after != unrelated: raise CleanupError("unrelated profile changed") protected = {path: file_digest(Path(path)) for path in contract["protected_files"]} if protected != contract["protected_files"]: raise CleanupError("token/provider authority changed") trainer_matches: list[str] = [] for root in (args.profile / "customers", args.profile / "data/customers", args.profile / "data/onboarding", args.profile / "data/owner-actions"): if root.exists(): trainer_matches.extend(str(path) for path in root.rglob("*trainer*")) if trainer_matches: raise CleanupError("trainer authority remains") return {"schema": "task26-clean-baseline-poststate-v1", "status": "PASS", "registry": "absent", "future_jobs": 0, "future_publications": 0, "gateway_state": "absent", "service": service, "profile_processes": [], "session_process_proof": processes, "customer_live_artifacts": "absent", "trainer_artifacts": [], "config_sha256": digest(config), "config_mode": "0664", "unrelated_profiles_sha256": unrelated_after, "protected_files_sha256": protected, "archive": str(archive)} def execute(args: argparse.Namespace, contract: dict[str, Any]) -> dict[str, Any]: if contract.get("completed_cleanup", {}).get("status") == "PASS_VERIFIED_DISABLED_ARCHIVED_CLEAN": raise CleanupError("cleanup already completed; v4 forbids mutation replay") ids = validate_lifecycle(args.profile, contract, require_running=True) unrelated = baseline_other_profiles(contract) scopes = authority_paths(args.profile, contract) rows = scoped_rows(args.profile, scopes) archive = archive_prestate(args, contract, scopes, rows, ids, unrelated) moved: list[str] = [] try: delivery_before = file_digest(args.profile / "data/owner-actions/draft-deliveries.json") disable = disable_customer(args, archive) config = patch_delivery_gate(args, archive) qa = qa_and_stop(args, contract, archive, delivery_before, unrelated) moved = move_scopes(args.profile, archive, scopes) post = postconditions(args, contract, archive, unrelated) atomic(archive / "poststate.json", canon(post)) receipt = {"schema": "task26-post-lifecycle-cleanup-receipt-v3", "status": "PASS", "mode": "execute", "run_id": args.run_id, "archive_committed_before_first_mutation": True, "canonical_disable_receipt_sha256": file_digest(archive / "disable-receipt.json"), "config_receipt_sha256": file_digest(archive / "config-disable-receipt.json"), "qa_receipt_sha256": file_digest(archive / "restart-after-disable-qa.json"), "poststate_sha256": file_digest(archive / "poststate.json"), "active_lifecycle": ids, "moved_scopes": moved, "disable_registry_sha256": disable["registry_sha256"], "post_config_sha256": config["config_sha256"], "customer_deliveries_after_disable": qa["customer_delivery_events_after_disable"]} atomic(archive / "receipt.json", canon(receipt)) return receipt except Exception as exc: try: restore_prestate(args, archive, moved, True) failure = {"schema": "task26-post-lifecycle-cleanup-failure-v3", "status": "FAIL_RESTORED_PRESTATE", "error": str(exc), "archive_retained": True} except Exception as restore_exc: failure = {"schema": "task26-post-lifecycle-cleanup-failure-v3", "status": "FAIL_RESTORE_ERROR", "error": str(exc), "restore_error": str(restore_exc), "archive_retained": True} atomic(archive / "failure-receipt.json", canon(failure)) raise def verify_archive(args: argparse.Namespace, contract: dict[str, Any]) -> dict[str, Any]: archive = args.archive_root / args.run_id manifest_raw = safe_read(archive / "manifest.json", {0o600}) manifest = json.loads(manifest_raw) receipt = load_json(archive / "receipt.json") if manifest.get("schema") != "task26-post-lifecycle-cleanup-archive-v3" or receipt.get("status") != "PASS": raise CleanupError("archive schema/status mismatch") claimed = manifest["evidence_digest"] check = dict(manifest) del check["evidence_digest"] if claimed != digest(canon(check)): raise CleanupError("archive manifest seal mismatch") for row in manifest["entries"]: raw = safe_read(archive / "payload" / row["path"], {0o600}) if len(raw) != row["size"] or digest(raw) != row["sha256"]: raise CleanupError(f"archive payload mismatch: {row['path']}") config = manifest["config_entry"] raw = safe_read(archive / "payload/config.yaml", {0o600}) if len(raw) != config["size"] or digest(raw) != config["sha256"]: raise CleanupError("archived config mismatch") archive_root_sha256 = sealed_tree_digest(archive) if archive_root_sha256 != contract["completed_cleanup"]["archive_root_sha256"]: raise CleanupError("sealed archive root mismatch") post = postconditions(args, contract, archive, manifest["unrelated_profiles_pre_sha256"]) return {"schema": "task26-post-lifecycle-cleanup-verification-v5", "status": "PASS", "archive": str(archive), "entry_count": len(manifest["entries"]), "archive_manifest_sha256": digest(manifest_raw), "archive_root_sha256": contract["completed_cleanup"]["archive_root_sha256"], "v3_seal_sha256": contract["completed_cleanup"]["v3_seal_sha256"], "delivery_history_preserved": True, "active_lifecycle_preserved": True, "poststate": post} def parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser() p.add_argument("mode", choices=("dry-run", "verify")) p.add_argument("--profile", type=Path, required=True) p.add_argument("--archive-root", type=Path, required=True) p.add_argument("--contract", type=Path, required=True) p.add_argument("--permission", type=Path, required=True) p.add_argument("--run-id", required=True) p.add_argument("--candidate", required=True) p.add_argument("--hermes-wheel", required=True) p.add_argument("--profile-wheel", required=True) p.add_argument("--plan-sha256", required=True) p.add_argument("--lifecycle-root-sha256", required=True) p.add_argument("--lifecycle-seal-sha256", required=True) p.add_argument("--v3-seal", type=Path, required=True) p.add_argument("--python", type=Path, default=Path(sys.executable)) p.add_argument("--customer-admin-root", type=Path, required=True) p.add_argument("--hermes-python", type=Path, required=True) p.add_argument("--hermes-source", type=Path, required=True) p.add_argument("--apply-patch", type=Path, required=True) return p def main() -> int: args = parser().parse_args() try: contract = validate_bindings(args) if args.mode == "verify": result = verify_archive(args, contract) else: verification = verify_archive(args, contract) result = {"schema": "task26-post-lifecycle-cleanup-dry-run-v5", "status": "PASS", "mode": "dry-run", "mutations": 0, "archive_created": False, "cleanup_replayed": False, "completed_archive_verified": True, "archive_root_sha256": verification["archive_root_sha256"], "v3_seal_sha256": verification["v3_seal_sha256"], "poststate": verification["poststate"]} sys.stdout.buffer.write(canon(result)) return 0 except (CleanupError, OSError, ValueError, KeyError, TypeError) as exc: sys.stderr.write(f"FAIL: {exc}\n") return 2 if __name__ == "__main__": raise SystemExit(main())