I read this as a stop-work handoff - below is a decision-complete patch for official lead application; I performed no further tool calls or mutations. The prior RED attempt created the focused test file before this stop instruction. The patch below replaces that file and adds the controller. ```diff *** Begin Patch *** Add File: .omo/evidence/task27/task27_supplemental_cleanup_controller.py +#!/usr/bin/env python3 +"""Disposable-only Task27 cleanup for session-bound outbox/membership residue.""" +from __future__ import annotations + +import argparse +import contextlib +import hashlib +import json +import os +import shutil +import stat +import subprocess +import sys +import tempfile +import traceback +import uuid +from pathlib import Path +from typing import Any + +SCHEMA = "task27-supplemental-cleanup-controller-v1" +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" +) +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 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", + }, + ] + + +def cleanup_receipt_binding() -> dict[str, object]: + _require_private(CLEANUP_RECEIPT, directory=False) + 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 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 + ) + matches = 0 + for entry in Path("/proc").iterdir(): + if not entry.name.isdigit(): + 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) + from gateway.platforms.telegram_nutrition_onboarding_publication_outbox import ( + GatewayOnboardingPublicationOutbox, + ) + from gateway.platforms.telegram_staff_membership_gate import MembershipJournal + + return GatewayOnboardingPublicationOutbox, 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) + _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), + "publication_records": len(primary_rows), + "membership_rows": len(rows), + "membership_epoch": rows[0]["subscription_epoch_id"], + } + + +def dry_run( + profile: Path, + output: Path, + *, + service_override: Path | None = None, +) -> dict[str, object]: + profile = profile.absolute() + _require_private(profile, directory=True) + 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(), + "service": state, + "exact_state": { + "outbox": target["outbox"], + "membership": target["membership"], + }, + "observed": { + "publication_records": target["publication_records"], + "membership_rows": target["membership_rows"], + "membership_epoch": target["membership_epoch"], + }, + "proposed_mutations": proposed_mutations(), + } + seal = sha_bytes(canonical(payload)) + receipt = { + "schema": SCHEMA, + "mode": "dry-run", + "status": "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) + + +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, + service_override: Path | None = None, +) -> dict[str, object]: + if output.exists() or output.is_symlink(): + raise Refusal(f"output already exists; receipt reuse refused: {output}") + profile = profile.absolute() + if profile.resolve() == LIVE.resolve(): + raise Refusal("live execution is not implemented") + 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 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") + require_inactive(service_state(service_override)) + current = validate_target_state(profile) + for name in ("outbox", "membership"): + if current[name] != payload["exact_state"][name]: + raise Refusal(f"source drift: {name}") + + paths = mutation_paths(profile) + rollback = Path(tempfile.mkdtemp(prefix="task27-supplemental-rollback-")) + operation = uuid.uuid4().hex + archive = paths["supplemental_archive"] / "archives" / operation + backup = rollback / "state" + backup.mkdir(mode=0o700) + _copy(paths["outbox"], backup / "outbox") + _copy(paths["membership"], backup / "membership") + try: + files = archive / "files/data/onboarding" + files.mkdir(parents=True, mode=0o700) + _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)), + } + if archived != { + "outbox": { + **payload["exact_state"]["outbox"], + "root": str(files / paths["outbox"].name), + }, + "membership": { + **payload["exact_state"]["membership"], + "root": str(files / paths["membership"].name), + }, + }: + raise Refusal("supplemental archive copy verification failed") + manifest = { + "schema": "task27-supplemental-archive-v1", + "operation_id": operation, + "identity": payload["identity"], + "cleanup_receipt": payload["cleanup_receipt"], + "source": payload["exact_state"], + "archive": archived, + } + manifest_path = archive / "manifest.json" + atomic_json(manifest_path, manifest) + 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": "disposable", + "profile": str(profile), + "permission_seal": seal, + "operation_id": operation, + "cleanup_receipt": payload["cleanup_receipt"], + "observed": payload["observed"], + "archive": { + "root": str(archive), + "manifest_sha256": sha_file(manifest_path), + "source": payload["exact_state"], + "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: + _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("--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( + "--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, + service_override=args.test_service_state, + ) + else: + result = execute( + args.profile, + args.permission_file, + args.permission_seal, + args.output, + fault=args.fault, + 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()) *** Delete File: .omo/evidence/task27/test_task27_supplemental_cleanup_controller.py *** Add File: .omo/evidence/task27/test_task27_supplemental_cleanup_controller.py +from __future__ import annotations + +import hashlib +import importlib.util +import json +import os +import stat +import subprocess +import sys +from pathlib import Path + +import pytest + +HERE = Path(__file__).resolve().parent +CONTROLLER = HERE / "task27_supplemental_cleanup_controller.py" +SESSION = "cb_mYUoMIsk_CRzSDKYpPE9dg" +USER = "8527916639" +ROUTE = (USER, "0") +WHEEL = ( + HERE.parent + / "task26/task26-combined-v38-delivered-st_01a019d7/artifacts" + / "hermes_agent-0.17.0-py3-none-any.whl" +) + + +def load(): + spec = importlib.util.spec_from_file_location("task27_supplement", CONTROLLER) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def canonical(value): + return json.dumps( + value, ensure_ascii=True, sort_keys=True, separators=(",", ":") + ).encode() + + +def private_json(path: Path, value) -> None: + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + path.write_bytes(canonical(value) + b"\n") + path.chmod(0o600) + + +def profile(tmp_path: Path) -> Path: + root = tmp_path / "profile" + root.mkdir(parents=True, mode=0o700) + marker = root / ".task27-supplemental-disposable-copy" + marker.write_text("TASK27_SUPPLEMENTAL_DISPOSABLE_COPY\n") + marker.chmod(0o600) + sys.path.insert(0, str(WHEEL)) + from gateway.platforms.telegram_nutrition_onboarding_publication_outbox import ( + GatewayOnboardingPublicationOutbox, + ) + from gateway.platforms.telegram_staff_membership_gate import MembershipJournal + + payload = {"body_digest": "a" * 64, "state": "collecting"} + render = "b" * 64 + outbox = GatewayOnboardingPublicationOutbox(root) + outbox.claim( + session_id=SESSION, + generation=0, + payload=payload, + route=ROUTE, + role="customer", + render_identity=render, + ) + outbox.record_receipt( + session_id=SESSION, + generation=0, + chat_id=USER, + topic_id="0", + message_id=304, + ) + outbox.mark_committed( + session_id=SESSION, + generation=0, + payload=payload, + route=ROUTE, + role="customer", + render_identity=render, + message_id=304, + ) + MembershipJournal( + root / "data/onboarding/telegram-staff-membership-v1/events.jsonl" + ).append( + { + "event": "subscription_armed", + "subscription_epoch_id": "epoch-1", + "observed_at_utc": "2026-08-18T14:55:11+00:00", + "staff_chat_inventory_sha256": "c" * 64, + "customer_user_ids": [USER], + } + ) + return root + + +def service_state(tmp_path: Path, active: bool = False) -> Path: + path = tmp_path / ("active.json" if active else "inactive.json") + private_json( + path, + { + "active_state": "active" if active else "inactive", + "sub_state": "running" if active else "dead", + "main_pid": 42 if active else 0, + "matching_processes": 1 if active else 0, + }, + ) + return path + + +def permission(module, root: Path, tmp_path: Path): + path = tmp_path / "permission.json" + result = module.dry_run( + root, path, service_override=service_state(tmp_path) + ) + return path, result["permission_seal"], result + + +def mutation_digest(module, root: Path) -> str: + return module.sha_bytes( + canonical( + { + key: module.compact(module.tree_inventory(value)) + for key, value in module.mutation_paths(root).items() + } + ) + ) + + +def test_dry_run_binds_contract(tmp_path: Path) -> None: + module = load() + root = profile(tmp_path) + _, _, result = permission(module, root, tmp_path) + payload = result["permission_payload"] + assert set(payload["exact_state"]) == {"outbox", "membership"} + assert payload["cleanup_receipt"]["sha256"] == module.CLEANUP_RECEIPT_SHA256 + assert payload["proposed_mutations"] == module.proposed_mutations() + + +@pytest.mark.parametrize( + "kind", ["foreign_session", "foreign_route", "mixed_membership"] +) +def test_rejects_foreign_or_mixed_state(tmp_path: Path, kind: str) -> None: + module = load() + root = profile(tmp_path) + if kind.startswith("foreign"): + path = root / "data/onboarding/telegram-publication-outbox-v1/ledger.json" + document = json.loads(path.read_text()) + if kind == "foreign_session": + document["records"][0]["session_id"] = "foreign" + else: + document["records"][0]["route"] = ["999", "0"] + private_json(path, document) + else: + path = root / "data/onboarding/telegram-staff-membership-v1/events.jsonl" + row = json.loads(path.read_text()) + row["customer_user_ids"] = [USER, "999"] + row.pop("row_sha256") + row["row_sha256"] = hashlib.sha256(canonical(row)).hexdigest() + path.write_bytes(canonical(row) + b"\n") + with pytest.raises(module.Refusal): + module.dry_run( + root, + tmp_path / "bad-permission.json", + service_override=service_state(tmp_path), + ) + + +@pytest.mark.parametrize( + "attack", ["hmac", "unknown", "symlink", "hardlink", "mode"] +) +def test_rejects_unsafe_or_unknown_outbox(tmp_path: Path, attack: str) -> None: + module = load() + root = profile(tmp_path) + outbox = root / "data/onboarding/telegram-publication-outbox-v1" + ledger = outbox / "ledger.json" + if attack == "hmac": + document = json.loads(ledger.read_text()) + document["records"][0]["receipt_integrity"] = "0" * 64 + private_json(ledger, document) + elif attack == "unknown": + path = outbox / "unexpected.json" + path.write_text("{}\n") + path.chmod(0o600) + elif attack == "symlink": + ledger.unlink() + ledger.symlink_to("emergency.json") + elif attack == "hardlink": + os.link(ledger, outbox / "ledger-copy.json") + else: + ledger.chmod(0o644) + with pytest.raises(module.Refusal): + module.dry_run( + root, + tmp_path / "bad-permission.json", + service_override=service_state(tmp_path), + ) + + +@pytest.mark.parametrize( + "attack", ["unknown", "symlink", "hardlink", "mode", "malformed"] +) +def test_rejects_unsafe_membership(tmp_path: Path, attack: str) -> None: + module = load() + root = profile(tmp_path) + membership = root / "data/onboarding/telegram-staff-membership-v1" + events = membership / "events.jsonl" + if attack == "unknown": + path = membership / "unexpected" + path.write_text("x") + path.chmod(0o600) + elif attack == "symlink": + events.unlink() + events.symlink_to("events.jsonl.lock") + elif attack == "hardlink": + os.link(events, membership / "copy") + elif attack == "mode": + events.chmod(0o644) + else: + events.write_text("{") + with pytest.raises(module.Refusal): + module.dry_run( + root, + tmp_path / "bad-permission.json", + service_override=service_state(tmp_path), + ) + + +def test_rejects_active_service_and_state_drift(tmp_path: Path) -> None: + module = load() + root = profile(tmp_path) + with pytest.raises(module.Refusal): + module.dry_run( + root, + tmp_path / "active-permission.json", + service_override=service_state(tmp_path, True), + ) + permission_path, seal, _ = permission(module, root, tmp_path) + (root / "data/onboarding/telegram-publication-outbox-v1/.lock").write_text( + "drift" + ) + with pytest.raises(module.Refusal, match="drift"): + module.execute( + root, + permission_path, + seal, + tmp_path / "receipt.json", + service_override=service_state(tmp_path), + ) + + +@pytest.mark.parametrize("fault", ["after_archive_copy", "after_prune"]) +def test_fault_rollback_is_exact(tmp_path: Path, fault: str) -> None: + module = load() + root = profile(tmp_path) + permission_path, seal, _ = permission(module, root, tmp_path) + before = mutation_digest(module, root) + with pytest.raises(RuntimeError, match=fault): + module.execute( + root, + permission_path, + seal, + tmp_path / "receipt.json", + fault=fault, + service_override=service_state(tmp_path), + ) + assert mutation_digest(module, root) == before + assert not (root / "data/task27-supplemental-cleanup").exists() + + +def test_success_is_frozen_terminal_and_one_use(tmp_path: Path) -> None: + module = load() + root = profile(tmp_path) + permission_path, seal, _ = permission(module, root, tmp_path) + output = tmp_path / "receipt.json" + result = module.execute( + root, + permission_path, + seal, + output, + service_override=service_state(tmp_path), + ) + assert result["status"] == "COMMITTED" + assert result["terminal"]["active_target_matches"] == 0 + assert not ( + root / "data/onboarding/telegram-publication-outbox-v1" + ).exists() + assert not ( + root / "data/onboarding/telegram-staff-membership-v1" + ).exists() + archive = Path(result["archive"]["root"]) + assert stat.S_IMODE(archive.stat().st_mode) == 0o500 + assert all( + not (path.stat().st_mode & 0o222) + for path in [archive, *archive.rglob("*")] + ) + with pytest.raises(module.Refusal, match="output already exists|reuse"): + module.execute( + root, + permission_path, + seal, + output, + service_override=service_state(tmp_path), + ) + + +def test_execute_is_disposable_only(tmp_path: Path) -> None: + module = load() + root = profile(tmp_path) + permission_path, seal, _ = permission(module, root, tmp_path) + (root / ".task27-supplemental-disposable-copy").unlink() + with pytest.raises(module.Refusal, match="disposable"): + module.execute( + root, + permission_path, + seal, + tmp_path / "receipt.json", + service_override=service_state(tmp_path), + ) + + +def test_cli_bad_seal_and_success(tmp_path: Path) -> None: + root = profile(tmp_path) + state = service_state(tmp_path) + permission_path = tmp_path / "permission.json" + dry = subprocess.run( + [ + sys.executable, + str(CONTROLLER), + "dry-run", + "--profile", + str(root), + "--output", + str(permission_path), + "--test-service-state", + str(state), + ], + text=True, + capture_output=True, + ) + assert dry.returncode == 0, dry.stderr + seal = json.loads(permission_path.read_text())["permission_seal"] + bad = subprocess.run( + [ + sys.executable, + str(CONTROLLER), + "execute", + "--profile", + str(root), + "--permission-file", + str(permission_path), + "--permission-seal", + "0" * 64, + "--output", + str(tmp_path / "bad.json"), + "--test-service-state", + str(state), + ], + text=True, + capture_output=True, + ) + assert bad.returncode == 2 + assert "permission seal mismatch" in bad.stderr + good = subprocess.run( + [ + sys.executable, + str(CONTROLLER), + "execute", + "--profile", + str(root), + "--permission-file", + str(permission_path), + "--permission-seal", + seal, + "--output", + str(tmp_path / "good.json"), + "--test-service-state", + str(state), + ], + text=True, + capture_output=True, + ) + assert good.returncode == 0, good.stderr *** End Patch ``` ## Observed RED Command: ```bash python -m pytest -q \ .omo/evidence/task27/test_task27_supplemental_cleanup_controller.py ``` Observed summary: ```text FFFFFFFFFFFFFF [100%] ... FileNotFoundError: [Errno 2] No such file or directory: '/home/cube/projects/richard/traning coach/.omo/evidence/task27/task27_supplemental_cleanup_controller.py' ... 14 failed in 0.18s ``` This was a genuine missing-controller RED. No GREEN run occurred. ## Intended GREEN and QA ```bash python -m py_compile \ .omo/evidence/task27/task27_supplemental_cleanup_controller.py \ .omo/evidence/task27/test_task27_supplemental_cleanup_controller.py python -m pytest -q \ .omo/evidence/task27/test_task27_supplemental_cleanup_controller.py ``` Then run LSP diagnostics on both files. Live read-only dry-run: ```bash python .omo/evidence/task27/task27_supplemental_cleanup_controller.py dry-run \ --profile /home/cube/.hermes/profiles/dualcoachtest \ --output .omo/evidence/task27/task27-supplemental-live-dry-run.json chmod 0400 .omo/evidence/task27/task27-supplemental-live-dry-run.json ``` Disposable rehearsal: ```bash tmp="$(mktemp -d /tmp/task27-supplemental-rehearsal-XXXXXX)" cp -a /home/cube/.hermes/profiles/dualcoachtest "$tmp/profile" chmod 0700 "$tmp/profile" printf '%s\n' TASK27_SUPPLEMENTAL_DISPOSABLE_COPY \ > "$tmp/profile/.task27-supplemental-disposable-copy" chmod 0600 "$tmp/profile/.task27-supplemental-disposable-copy" python .omo/evidence/task27/task27_supplemental_cleanup_controller.py dry-run \ --profile "$tmp/profile" \ --output "$tmp/permission.json" seal="$( python -c 'import json,sys; print(json.load(open(sys.argv[1]))["permission_seal"])' \ "$tmp/permission.json" )" python .omo/evidence/task27/task27_supplemental_cleanup_controller.py execute \ --profile "$tmp/profile" \ --permission-file "$tmp/permission.json" \ --permission-seal "$seal" \ --output .omo/evidence/task27/task27-supplemental-rehearsal-receipt.json chmod 0400 .omo/evidence/task27/task27-supplemental-rehearsal-receipt.json ``` Manual bad-input and one-use QA: ```bash python .omo/evidence/task27/task27_supplemental_cleanup_controller.py execute \ --profile "$tmp/profile" \ --permission-file "$tmp/permission.json" \ --permission-seal "$(printf '0%.0s' {1..64})" \ --output "$tmp/bad-seal.json" # Expected: exit 2, "permission seal mismatch" python .omo/evidence/task27/task27_supplemental_cleanup_controller.py execute \ --profile "$tmp/profile" \ --permission-file "$tmp/permission.json" \ --permission-seal "$seal" \ --output .omo/evidence/task27/task27-supplemental-rehearsal-receipt.json # Expected: exit 2, output already exists / receipt reuse refused ``` Before the live dry-run and after all disposable QA, independently hash the live profile with the existing Task27 `tree_inventory()` implementation. Emit: - `task27-supplemental-verification-receipt.json` - `task27-supplemental-live-nonmutation-receipt.json` The nonmutation receipt should bind equal before/after live-profile digests, controller/test hashes, cleanup receipt hash, dry-run hash, rehearsal hash, test result, compile/LSP results, and the exact `proposed_mutations()` value. No live execution or authorization receipt should be produced.