#!/usr/bin/env python3 """Candidate-bound, event-driven Task26 lifecycle evidence observers. This tool is read-only with respect to the profile. It uses Linux inotify and stable O_NOFOLLOW descriptors; only private redacted evidence receipts are created. """ from __future__ import annotations import argparse import ctypes import hashlib import json import os import selectors import stat import struct import sys import time from collections.abc import Mapping from dataclasses import dataclass from pathlib import Path from typing import Final SCHEMA: Final = "task26-lifecycle-observer-manifest-v1" RECEIPT_SCHEMA: Final = "task26-lifecycle-observer-receipt-v1" CANDIDATE: Final = "2e0894eac92bc396cc4723bf1f18ebc653b95018dd41574df435941c235da925" WHEEL: Final = "af4a9d0a1ffffb6eb7551c1d6dc2b32853ca6d024332a4f8f5702bbf992f141b" PLAN: Final = "7ace03c6dad33d2fc3ef223621cbca68a150fde8429932138e252fb8498ac582" GOLDEN_V4: Final = "e0979931fb863d957403f71305d33470697813c5e740622b6b05a23a65c76359" RECOVERY_V4: Final = "f539460f91096cfe32140dd6d5b5a940ff38a13ad67bf6d83302f6732aa719a1" DIGEST_LENGTH: Final = 64 PRIVATE_FILE: Final = 0o600 PRIVATE_DIR: Final = 0o700 O_NOFOLLOW: Final = getattr(os, "O_NOFOLLOW", 0) O_DIRECTORY: Final = getattr(os, "O_DIRECTORY", 0) IN_MODIFY = 0x00000002 IN_CLOSE_WRITE = 0x00000008 IN_MOVED_TO = 0x00000080 IN_CREATE = 0x00000100 IN_Q_OVERFLOW = 0x00004000 IN_IGNORED = 0x00008000 WATCH_MASK = IN_MODIFY | IN_CLOSE_WRITE | IN_MOVED_TO | IN_CREATE _EVENT = struct.Struct("iIII") MODE_PATHS: Final = { "subscribe-outbox": "data/onboarding/telegram-publication-outbox-v1/ledger.json", "audit-tail": "data/customer-activation-audit.jsonl", "watch-deliveries": "data/owner-actions/draft-deliveries.json", } OUTBOX_SCHEMA: Final = "telegram-nutrition-onboarding-publication-outbox-v2" OUTBOX_FIELDS: Final = frozenset({ "session_id", "generation", "payload", "route", "role", "render_identity", "payload_digest", "dispatch_identity", "state", "message_id", "receipt_integrity", }) AUDIT_REQUIRED: Final = frozenset({ "event", "customer_id", "enabled", "registry_path", "data_root", "checklist_evidence_path", "recorded_at", }) AUDIT_OPTIONAL: Final = frozenset({ "transaction_id", "registry_sha256", "nutrition_activation_receipt", }) DELIVERY_ALLOWED: Final = frozenset({ "draft_id", "customer_key", "session_id", "text", "revision", "approved_revision", "approved_event_id", "checkin_revision", "idempotency_key", "canonical_payload_digest", "provider_chat_id", "provider_topic_id", "status", "message_id", "provider_receipt", }) class ObserverError(RuntimeError): """Fail-closed observer contract violation.""" class ObserverTimeout(ObserverError): """The bounded monotonic deadline expired.""" @dataclass(frozen=True) class Snapshot: present: bool inode: int | None digest: str revision: int value: object size: int @dataclass(frozen=True) class ObserverSpec: name: str mode: str after_seq: int expected: str output: str def _digest_bytes(raw: bytes) -> str: return hashlib.sha256(raw).hexdigest() def _id_hash(value: str) -> str: return _digest_bytes(value.encode("utf-8")) def _is_digest(value: object) -> bool: return isinstance(value, str) and len(value) == DIGEST_LENGTH and all(c in "0123456789abcdef" for c in value) def _private_regular(fd: int, label: str) -> os.stat_result: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode): raise ObserverError(f"{label} is not a regular file") if info.st_uid != os.getuid(): raise ObserverError(f"{label} has the wrong owner") if info.st_nlink != 1: raise ObserverError(f"{label} has a hard link") if stat.S_IMODE(info.st_mode) != PRIVATE_FILE: raise ObserverError(f"{label} is not private 0600") return info def _private_directory(fd: int, label: str) -> os.stat_result: info = os.fstat(fd) if not stat.S_ISDIR(info.st_mode): raise ObserverError(f"{label} is not a directory") if info.st_uid != os.getuid(): raise ObserverError(f"{label} has the wrong owner") if stat.S_IMODE(info.st_mode) != PRIVATE_DIR: raise ObserverError(f"{label} is public or not private 0700") return info def _read_fd_once(fd: int, size: int) -> bytes: if size > 16 * 1024 * 1024: raise ObserverError("authority file exceeds the evidence bound") os.lseek(fd, 0, os.SEEK_SET) raw = os.read(fd, size + 1) if len(raw) != size: raise ObserverError("authority changed during its descriptor read") return raw def _load_private_json(path: Path, label: str) -> dict[str, object]: try: fd = os.open(path, os.O_RDONLY | os.O_CLOEXEC | O_NOFOLLOW) except OSError as exc: raise ObserverError(f"{label} cannot be stable-opened") from exc try: info = _private_regular(fd, label) raw = _read_fd_once(fd, info.st_size) finally: os.close(fd) try: value = json.loads(raw) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ObserverError(f"{label} is invalid JSON") from exc if not isinstance(value, dict): raise ObserverError(f"{label} must be a JSON object") return value def load_manifest(path: Path) -> dict[str, object]: value = _load_private_json(path, "observer manifest") if value.get("schema") != SCHEMA or set(value) != { "schema", "bindings", "receipts", "expected_ids", "observers" }: raise ObserverError("observer manifest schema drift") bindings = value.get("bindings") if not isinstance(bindings, dict) or set(bindings) != { "candidate", "wheel_sha256", "plan_sha256", "runbook_sha256" }: raise ObserverError("observer manifest binding schema drift") runbooks = bindings.get("runbook_sha256") if ( bindings.get("candidate") != CANDIDATE or bindings.get("wheel_sha256") != WHEEL or bindings.get("plan_sha256") != PLAN or runbooks != {"golden_v4": GOLDEN_V4, "recovery_v4": RECOVERY_V4} ): raise ObserverError("observer manifest is not bound to current candidate 2e/wheel af4/plan 7ace/runbook v4") receipts = value.get("receipts") if not isinstance(receipts, dict) or set(receipts) != {"invite", "reset", "cleanup", "provider"}: raise ObserverError("observer receipt pin schema drift") if any(not _is_digest(item) for item in receipts.values()): raise ObserverError("observer receipt pin is invalid") ids = value.get("expected_ids") expected_id_fields = {"session_id", "generation", "customer_key", "transaction_id", "draft_id", "delivery_key"} if not isinstance(ids, dict) or set(ids) != expected_id_fields: raise ObserverError("expected lifecycle ID schema drift") if any(not isinstance(ids[key], str) or not ids[key] or len(ids[key]) > 256 for key in expected_id_fields - {"generation"}): raise ObserverError("expected lifecycle ID is invalid") if type(ids["generation"]) is not int or int(ids["generation"]) < 0: raise ObserverError("expected lifecycle generation is invalid") observer_specs(value) return value def observer_specs(manifest: Mapping[str, object]) -> list[ObserverSpec]: raw_specs = manifest.get("observers") if not isinstance(raw_specs, list) or not raw_specs: raise ObserverError("observer manifest has no observers") result: list[ObserverSpec] = [] for raw in raw_specs: if not isinstance(raw, dict) or set(raw) != {"name", "mode", "after_seq", "expected", "output"}: raise ObserverError("observer entry schema drift") name, mode, expected, output = raw["name"], raw["mode"], raw["expected"], raw["output"] after = raw["after_seq"] if not all(isinstance(item, str) and item and len(item) <= 256 for item in (name, mode, expected, output)): raise ObserverError("observer entry value is invalid") if mode not in MODE_PATHS or type(after) is not int or after < 0: raise ObserverError("observer mode or sequence is invalid") output_path = Path(output) if output_path.is_absolute() or len(output_path.parts) != 1 or output_path.name in {".", ".."}: raise ObserverError("observer output must be one relative private filename") result.append(ObserverSpec(name, mode, after, expected, output)) if len({item.name for item in result}) != len(result): raise ObserverError("duplicate observer name") if len({item.mode for item in result}) != len(result): raise ObserverError("duplicate observer mode") return result def _parse_json(raw: bytes, label: str) -> object: try: return json.loads(raw) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ObserverError(f"{label} schema drift: invalid JSON") from exc def _outbox(raw: bytes) -> tuple[object, int]: value = _parse_json(raw, "outbox") if not isinstance(value, dict) or set(value) != {"schema", "records"} or value.get("schema") != OUTBOX_SCHEMA: raise ObserverError("outbox schema drift") rows = value.get("records") if not isinstance(rows, list): raise ObserverError("outbox schema drift") keys: set[tuple[str, int]] = set() revision = 0 ranks = {"DISPATCHING": 1, "RECEIPTED": 2, "COMMITTED": 3} for row in rows: if not isinstance(row, dict) or set(row) != OUTBOX_FIELDS: raise ObserverError("outbox record schema drift") session, generation, state = row.get("session_id"), row.get("generation"), row.get("state") if not isinstance(session, str) or not session or type(generation) is not int: raise ObserverError("outbox record identity schema drift") key = (session, generation) if key in keys: raise ObserverError("duplicate outbox lifecycle record") keys.add(key) if state not in ranks: raise ObserverError("outbox has unknown outcome") revision += ranks[str(state)] return value, revision def _jsonl(raw: bytes, label: str) -> list[dict[str, object]]: if raw and not raw.endswith(b"\n"): raise ObserverError(f"{label} has a partial append") rows: list[dict[str, object]] = [] for line in raw.splitlines(): try: row = json.loads(line) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ObserverError(f"{label} schema drift") from exc if not isinstance(row, dict): raise ObserverError(f"{label} schema drift") rows.append(row) return rows def _audit(raw: bytes) -> tuple[object, int]: rows = _jsonl(raw, "activation audit") seen: set[str] = set() for row in rows: if not AUDIT_REQUIRED <= set(row) or not set(row) <= AUDIT_REQUIRED | AUDIT_OPTIONAL: raise ObserverError("activation audit schema drift") transaction = row.get("transaction_id") if not isinstance(transaction, str) or not transaction: raise ObserverError("activation audit transaction schema drift") if transaction in seen: raise ObserverError("duplicate activation audit transaction") seen.add(transaction) if row.get("event") != "customer_activation" or row.get("enabled") is not True: raise ObserverError("activation audit has unknown outcome") return rows, len(rows) def _deliveries(raw: bytes) -> tuple[object, int]: value = _parse_json(raw, "delivery ledger") if not isinstance(value, dict): raise ObserverError("delivery ledger schema drift") revision = 0 ranks = {"pending": 1, "sent": 2, "delivered": 2, "sent_audited": 3} for key, row in value.items(): if not isinstance(key, str) or not key or not isinstance(row, dict): raise ObserverError("delivery ledger schema drift") if not {"draft_id", "customer_key", "session_id", "status", "idempotency_key"} <= set(row): raise ObserverError("delivery record schema drift") if not set(row) <= DELIVERY_ALLOWED: raise ObserverError("delivery record schema drift") state = row.get("status") if state not in ranks: raise ObserverError("delivery ledger has unknown outcome") revision += ranks[str(state)] return value, revision def _decode(mode: str, raw: bytes) -> tuple[object, int]: if mode == "subscribe-outbox": return _outbox(raw) if mode == "audit-tail": return _audit(raw) return _deliveries(raw) def _transition(mode: str, before: Snapshot, after: Snapshot, manifest: Mapping[str, object], spec: ObserverSpec) -> dict[str, object]: ids = manifest["expected_ids"] assert isinstance(ids, dict) if after.revision <= before.revision or after.revision <= spec.after_seq: raise ObserverError("stale authority event has no new transition") if after.revision != before.revision + 1 and mode != "watch-deliveries": raise ObserverError("duplicate or skipped authority transition") if mode == "subscribe-outbox": old = before.value if isinstance(before.value, dict) else {"records": []} new = after.value assert isinstance(old, dict) and isinstance(new, dict) old_map = {(r["session_id"], r["generation"]): r for r in old["records"]} new_map = {(r["session_id"], r["generation"]): r for r in new["records"]} changed = [key for key in set(old_map) | set(new_map) if old_map.get(key) != new_map.get(key)] if len(changed) != 1: raise ObserverError("duplicate or stale outbox transition") expected_key = (ids["session_id"], ids["generation"]) if changed[0] != expected_key: raise ObserverError("wrong lifecycle ID in outbox transition") row = new_map.get(expected_key) if row is None or row.get("state") != spec.expected: raise ObserverError("outbox transition has unknown outcome") state = str(row["state"]) elif mode == "audit-tail": old_rows = before.value if isinstance(before.value, list) else [] new_rows = after.value assert isinstance(new_rows, list) if new_rows[: len(old_rows)] != old_rows or len(new_rows) != len(old_rows) + 1: raise ObserverError("activation audit is not one exact append transition") row = new_rows[-1] if row.get("customer_id") != ids["customer_key"] or row.get("transaction_id") != ids["transaction_id"]: raise ObserverError("wrong lifecycle ID in activation audit transition") if row.get("event") != spec.expected: raise ObserverError("activation audit transition has unknown outcome") state = str(row["event"]) else: old = before.value if isinstance(before.value, dict) else {} new = after.value assert isinstance(new, dict) changed = [key for key in set(old) | set(new) if old.get(key) != new.get(key)] if len(changed) != 1 or changed[0] != ids["delivery_key"]: raise ObserverError("duplicate, stale, or wrong lifecycle ID delivery transition") row = new.get(str(ids["delivery_key"])) if not isinstance(row, dict) or ( row.get("draft_id") != ids["draft_id"] or row.get("customer_key") != ids["customer_key"] or row.get("session_id") != ids["session_id"] ): raise ObserverError("wrong lifecycle ID in delivery transition") terminal = [item for item in new.values() if isinstance(item, dict) and item.get("status") == "sent_audited"] if row.get("status") != spec.expected or len(terminal) != 1 or not row.get("message_id"): raise ObserverError("delivery transition is not exactly one sent_audited record") state = str(row["status"]) return { "state": state, "revision_before": before.revision, "revision_after": after.revision, "authority_sha256_before": before.digest, "authority_sha256_after": after.digest, "candidate_id_hashes": { key: _id_hash(str(ids[key])) for key in ("session_id", "customer_key", "transaction_id", "draft_id", "delivery_key") }, } class Observer: def __init__(self, profile: Path, manifest: Mapping[str, object], spec: ObserverSpec) -> None: self.profile = profile self.manifest = manifest self.spec = spec self.relative = Path(MODE_PATHS[spec.mode]) self.root_fd = -1 self.parent_fd = -1 self.file_fd = -1 self.inotify_fd = -1 self.watch = -1 self.baseline: Snapshot | None = None def arm(self) -> Snapshot: if not self.profile.is_absolute(): raise ObserverError("profile path must be absolute") try: self.root_fd = os.open(self.profile, os.O_RDONLY | os.O_CLOEXEC | O_DIRECTORY | O_NOFOLLOW) except OSError as exc: raise ObserverError("profile root cannot be stable-opened") from exc _private_directory(self.root_fd, "profile root") current = os.dup(self.root_fd) try: for component in self.relative.parts[:-1]: next_fd = os.open(component, os.O_RDONLY | os.O_CLOEXEC | O_DIRECTORY | O_NOFOLLOW, dir_fd=current) _private_directory(next_fd, f"authority directory {component}") os.close(current) current = next_fd self.parent_fd = current current = -1 finally: if current >= 0: os.close(current) self.inotify_fd = _inotify_init() self.watch = _inotify_add(self.inotify_fd, f"/proc/self/fd/{self.parent_fd}", WATCH_MASK) self.baseline = self._snapshot(allow_absent=True) if self.baseline.revision != self.spec.after_seq: raise ObserverError( f"stale --after-seq {self.spec.after_seq}; baseline revision is {self.baseline.revision}" ) return self.baseline def _snapshot(self, *, allow_absent: bool) -> Snapshot: name = self.relative.name try: fd = os.open(name, os.O_RDONLY | os.O_CLOEXEC | O_NOFOLLOW, dir_fd=self.parent_fd) except FileNotFoundError: if not allow_absent: raise ObserverError("authority event did not publish its target file") from None value, revision = _decode(self.spec.mode, b"" if self.spec.mode == "audit-tail" else b"{}") if self.spec.mode != "subscribe-outbox" else ({"schema": OUTBOX_SCHEMA, "records": []}, 0) return Snapshot(False, None, _digest_bytes(b""), revision, value, 0) except OSError as exc: raise ObserverError("authority file cannot be stable-opened") from exc try: info = _private_regular(fd, "authority file") raw = _read_fd_once(fd, info.st_size) value, revision = _decode(self.spec.mode, raw) if self.file_fd >= 0: os.close(self.file_fd) self.file_fd = os.dup(fd) return Snapshot(True, info.st_ino, _digest_bytes(raw), revision, value, info.st_size) finally: os.close(fd) def wait(self, deadline: float) -> dict[str, object]: if self.baseline is None: raise ObserverError("observer is not armed") selector = selectors.DefaultSelector() selector.register(self.inotify_fd, selectors.EVENT_READ) try: while True: remaining = deadline - time.monotonic() if remaining <= 0: raise ObserverTimeout("bounded monotonic deadline expired") ready = selector.select(remaining) if not ready: raise ObserverTimeout("bounded monotonic deadline expired") raw = os.read(self.inotify_fd, 65536) offset = 0 matched = False while offset + _EVENT.size <= len(raw): watch, mask, _cookie, length = _EVENT.unpack_from(raw, offset) offset += _EVENT.size name_raw = raw[offset : offset + length] offset += length name = name_raw.split(b"\0", 1)[0].decode("utf-8", "strict") if mask & IN_Q_OVERFLOW: raise ObserverError("inotify queue overflow") if mask & IN_IGNORED or watch != self.watch: raise ObserverError("inotify authority watch was invalidated") if name == self.relative.name: if mask & IN_MOVED_TO: matched = True elif mask & (IN_CREATE | IN_MODIFY | IN_CLOSE_WRITE): raise ObserverError("authority changed without exact atomic replace event") if not matched: continue after = self._snapshot(allow_absent=False) if self.baseline.present and after.inode == self.baseline.inode: raise ObserverError("atomic replacement reused the baseline inode") return _transition(self.spec.mode, self.baseline, after, self.manifest, self.spec) finally: selector.close() def close(self) -> None: for fd_name in ("file_fd", "inotify_fd", "parent_fd", "root_fd"): fd = getattr(self, fd_name) if fd >= 0: os.close(fd) setattr(self, fd_name, -1) def _inotify_init() -> int: libc = ctypes.CDLL(None, use_errno=True) fd = libc.inotify_init1(os.O_CLOEXEC | os.O_NONBLOCK) if fd < 0: error = ctypes.get_errno() raise ObserverError(f"inotify_init1 failed: errno {error}") return int(fd) def _inotify_add(fd: int, path: str, mask: int) -> int: libc = ctypes.CDLL(None, use_errno=True) watch = libc.inotify_add_watch(fd, os.fsencode(path), mask) if watch < 0: error = ctypes.get_errno() raise ObserverError(f"inotify_add_watch failed: errno {error}") return int(watch) def _secure_output(path: Path, value: Mapping[str, object]) -> None: if not path.is_absolute(): path = Path.cwd() / path try: parent_fd = os.open(path.parent, os.O_RDONLY | os.O_CLOEXEC | O_DIRECTORY | O_NOFOLLOW) except OSError as exc: raise ObserverError("receipt parent cannot be stable-opened") from exc try: _private_directory(parent_fd, "receipt parent") flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | O_NOFOLLOW fd = os.open(path.name, flags, PRIVATE_FILE, dir_fd=parent_fd) try: raw = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() + b"\n" os.write(fd, raw) os.fsync(fd) _private_regular(fd, "receipt") finally: os.close(fd) os.fsync(parent_fd) finally: os.close(parent_fd) def _binding_receipt(manifest: Mapping[str, object]) -> dict[str, object]: return { "candidate": CANDIDATE, "wheel_sha256": WHEEL, "plan_sha256": PLAN, "runbook_sha256": {"golden_v4": GOLDEN_V4, "recovery_v4": RECOVERY_V4}, "source_receipt_sha256": manifest["receipts"], } def _ready(fd: int | None, names: list[str]) -> None: if fd is None: return raw = json.dumps({"status": "READY", "observers": names}, sort_keys=True).encode() os.write(fd, raw) os.close(fd) def _choose_spec(manifest: Mapping[str, object], mode: str, after: int, observer_name: str | None) -> ObserverSpec: matches = [item for item in observer_specs(manifest) if item.mode == mode and (observer_name is None or item.name == observer_name)] if len(matches) != 1: raise ObserverError("manifest does not select exactly one observer") spec = matches[0] if spec.after_seq != after: raise ObserverError("CLI --after-seq differs from the manifest") return spec def run_single(args: argparse.Namespace) -> dict[str, object]: manifest = load_manifest(args.manifest) spec = _choose_spec(manifest, args.mode, args.after_seq, args.observer) watcher = Observer(args.profile, manifest, spec) try: baseline = watcher.arm() _ready(args.ready_fd, [spec.name]) transition = watcher.wait(time.monotonic() + args.timeout) receipt: dict[str, object] = { "schema": RECEIPT_SCHEMA, "status": "PASS", "mode": spec.mode, "observer": spec.name, "event_subscription": "inotify-before-trigger", "baseline": {"present": baseline.present, "inode": baseline.inode, "sha256": baseline.digest, "revision": baseline.revision}, "transition": transition, "bindings": _binding_receipt(manifest), "privacy": "redacted-no-raw-route-content-token-or-provider-material", } _secure_output(args.output, receipt) return receipt finally: watcher.close() def run_arm_only(args: argparse.Namespace) -> dict[str, object]: manifest = load_manifest(args.manifest) watchers = [Observer(args.profile, manifest, spec) for spec in observer_specs(manifest)] try: baselines = [(watcher, watcher.arm()) for watcher in watchers] _ready(args.ready_fd, [watcher.spec.name for watcher in watchers]) receipt: dict[str, object] = { "schema": RECEIPT_SCHEMA, "status": "ARMED_READ_ONLY", "event_subscription": "inotify-established-no-wait-no-profile-mutation", "observers": [{"name": watcher.spec.name, "mode": watcher.spec.mode, "present": baseline.present, "inode": baseline.inode, "sha256": baseline.digest, "revision": baseline.revision} for watcher, baseline in baselines], "bindings": _binding_receipt(manifest), "privacy": "redacted-no-raw-route-content-token-or-provider-material", } _secure_output(args.receipt, receipt) return receipt finally: for watcher in watchers: watcher.close() def run_multi(args: argparse.Namespace) -> dict[str, object]: manifest = load_manifest(args.manifest) manifest_path = args.manifest.resolve() watchers = [Observer(args.profile, manifest, spec) for spec in observer_specs(manifest)] try: for watcher in watchers: watcher.arm() _ready(args.ready_fd, [watcher.spec.name for watcher in watchers]) deadline = time.monotonic() + args.timeout results: list[dict[str, object]] = [] for watcher in watchers: transition = watcher.wait(deadline) event_receipt = { "schema": RECEIPT_SCHEMA, "status": "PASS", "mode": watcher.spec.mode, "observer": watcher.spec.name, "event_subscription": "multi-observer-ready-before-trigger", "transition": transition, "bindings": _binding_receipt(manifest), "privacy": "redacted-no-raw-route-content-token-or-provider-material", } _secure_output(manifest_path.parent / watcher.spec.output, event_receipt) results.append({"observer": watcher.spec.name, "status": "PASS", "state": transition["state"]}) receipt: dict[str, object] = {"schema": RECEIPT_SCHEMA, "status": "PASS", "mode": "multi-observer", "ready_barrier": "all-observers-armed-before-actions", "results": results, "bindings": _binding_receipt(manifest)} _secure_output(args.receipt, receipt) return receipt finally: for watcher in watchers: watcher.close() def parser() -> argparse.ArgumentParser: result = argparse.ArgumentParser(description=__doc__) sub = result.add_subparsers(dest="mode", required=True) for mode in MODE_PATHS: command = sub.add_parser(mode) command.add_argument("--profile", type=Path, required=True) command.add_argument("--manifest", type=Path, required=True) command.add_argument("--after-seq", type=int, required=True) command.add_argument("--timeout", type=float, required=True) command.add_argument("--output", type=Path, required=True) command.add_argument("--observer") command.add_argument("--ready-fd", type=int) arm = sub.add_parser("arm-only") arm.add_argument("--profile", type=Path, required=True) arm.add_argument("--manifest", type=Path, required=True) arm.add_argument("--receipt", type=Path, required=True) arm.add_argument("--ready-fd", type=int) multi = sub.add_parser("multi-observer") multi.add_argument("--profile", type=Path, required=True) multi.add_argument("--manifest", type=Path, required=True) multi.add_argument("--timeout", type=float, required=True) multi.add_argument("--ready-fd", type=int, required=True) multi.add_argument("--receipt", type=Path, required=True) return result def main(argv: list[str] | None = None) -> int: args = parser().parse_args(argv) try: if args.mode in MODE_PATHS: if args.timeout <= 0 or args.timeout > 3600: raise ObserverError("timeout must be in (0, 3600]") receipt = run_single(args) elif args.mode == "arm-only": receipt = run_arm_only(args) else: if args.timeout <= 0 or args.timeout > 3600: raise ObserverError("timeout must be in (0, 3600]") receipt = run_multi(args) except ObserverTimeout as exc: print(f"FAIL: {exc}", file=sys.stderr) return 3 except (ObserverError, OSError, ValueError) as exc: print(f"FAIL: {exc}", file=sys.stderr) return 2 print(json.dumps(receipt, sort_keys=True, separators=(",", ":"))) return 0 if __name__ == "__main__": raise SystemExit(main())