tenant.openagents/omega

1#!/usr/bin/env python3
2"""Collect Omega/Zed lifecycle evidence and perform narrowly confirmed RC moves."""
3
4from __future__ import annotations
5
6import argparse
7import hashlib
8import json
9import os
10import plistlib
11import re
12import shutil
13import stat
14import subprocess
15import tempfile
16from datetime import datetime, timezone
17from pathlib import Path
18from typing import Any
19
20
21EXPECTED_APP = Path("/Applications/Omega.app")
22EXPECTED_BUNDLE_ID = "com.openagents.omega.rc"
23EXPECTED_KEYCHAIN = {
24    "service": "com.openagents.omega.credentials.rc",
25    "account": "omega-sovereign-identity-v1",
26    "secret_read": False,
27}
28SCHEMA = "openagents.omega.rc-lifecycle-proof.v1"
29COLLECTOR_VERSION = 2
30
31
32class LifecycleError(RuntimeError):
33    pass
34
35
36def sha256_file(path: Path) -> str:
37    digest = hashlib.sha256()
38    with path.open("rb") as handle:
39        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
40            digest.update(chunk)
41    return digest.hexdigest()
42
43
44def sha256_json(value: Any) -> str:
45    encoded = json.dumps(
46        value, sort_keys=True, separators=(",", ":"), ensure_ascii=False
47    ).encode("utf-8")
48    return hashlib.sha256(encoded).hexdigest()
49
50
51def collector_binding() -> dict[str, Any]:
52    collector = Path(__file__).resolve()
53    return {
54        "path": "script/omega-rc-lifecycle-proof",
55        "version": COLLECTOR_VERSION,
56        "sha256": sha256_file(collector),
57    }
58
59
60def validate_candidate_digest(value: str) -> str:
61    if re.fullmatch(r"[0-9a-f]{64}", value) is None:
62        raise LifecycleError("candidate digest must be 64 lowercase hexadecimal characters")
63    return value
64
65
66def roots() -> dict[str, dict[str, Path]]:
67    user_root = Path.home()
68    library = user_root / "Library"
69    omega = {
70        "application": EXPECTED_APP,
71        "data": library / "Application Support/Omega RC",
72        "config": user_root / ".config/omega-rc",
73        "state": user_root / ".local/state/omega-rc",
74        "cache": library / "Caches/omega-rc",
75        "logs": library / "Logs/omega-rc",
76        "http_storage": library / "HTTPStorages/com.openagents.omega.rc",
77        "preferences": library / "Preferences/com.openagents.omega.rc.plist",
78        "saved_state": library / "Saved Application State/com.openagents.omega.rc.savedState",
79    }
80    zed = {
81        "application_stable": Path("/Applications/Zed.app"),
82        "application_preview": Path("/Applications/Zed Preview.app"),
83        "application_nightly": Path("/Applications/Zed Nightly.app"),
84        "application_dev": Path("/Applications/Zed Dev.app"),
85        "data": library / "Application Support/Zed",
86        "config": user_root / ".config/zed",
87        "logs": library / "Logs/Zed",
88        "cache_stable": library / "Caches/dev.zed.Zed",
89        "cache_preview": library / "Caches/dev.zed.Zed-Preview",
90        "cache_nightly": library / "Caches/dev.zed.Zed-Nightly",
91        "cache_dev": library / "Caches/dev.zed.Zed-Dev",
92        "http_storage_stable": library / "HTTPStorages/dev.zed.Zed",
93        "http_storage_preview": library / "HTTPStorages/dev.zed.Zed-Preview",
94        "preferences_stable": library / "Preferences/dev.zed.Zed.plist",
95        "preferences_preview": library / "Preferences/dev.zed.Zed-Preview.plist",
96        "saved_state_stable": library / "Saved Application State/dev.zed.Zed.savedState",
97        "saved_state_preview": library / "Saved Application State/dev.zed.Zed-Preview.savedState",
98        "remote_server": user_root / ".zed_server",
99    }
100    return {"omega": omega, "zed": zed}
101
102
103def path_digest(path: Path) -> str:
104    return hashlib.sha256(os.fsencode(path.absolute())).hexdigest()
105
106
107def manifest_root(path: Path) -> dict[str, Any]:
108    result: dict[str, Any] = {
109        "path_digest": path_digest(path),
110        "status": "absent",
111        "files": 0,
112        "directories": 0,
113        "symlinks": 0,
114        "bytes": 0,
115        "manifest_sha256": hashlib.sha256(b"absent").hexdigest(),
116    }
117    if not path.exists() and not path.is_symlink():
118        return result
119
120    digest = hashlib.sha256()
121    errors: list[str] = []
122
123    def record(candidate: Path, relative: str) -> None:
124        try:
125            metadata = candidate.lstat()
126            mode = stat.S_IMODE(metadata.st_mode)
127            digest.update(relative.encode("utf-8", errors="surrogateescape"))
128            digest.update(str(mode).encode())
129            if stat.S_ISLNK(metadata.st_mode):
130                result["symlinks"] += 1
131                digest.update(b"symlink:")
132                digest.update(os.fsencode(os.readlink(candidate)))
133            elif stat.S_ISREG(metadata.st_mode):
134                result["files"] += 1
135                result["bytes"] += metadata.st_size
136                digest.update(b"file:")
137                digest.update(sha256_file(candidate).encode())
138            elif stat.S_ISDIR(metadata.st_mode):
139                result["directories"] += 1
140                digest.update(b"directory")
141            else:
142                digest.update(b"other")
143        except OSError as error:
144            errors.append(type(error).__name__)
145
146    if path.is_symlink() or path.is_file():
147        record(path, ".")
148    else:
149        record(path, ".")
150        for directory, directory_names, file_names in os.walk(path, followlinks=False):
151            directory_path = Path(directory)
152            for name in sorted(directory_names + file_names, key=os.fsencode):
153                candidate = directory_path / name
154                record(candidate, candidate.relative_to(path).as_posix())
155    result["manifest_sha256"] = digest.hexdigest()
156    result["status"] = "blocked" if errors else "captured"
157    if errors:
158        result["error_count"] = len(errors)
159        result["error_types"] = sorted(set(errors))
160    return result
161
162
163def inventory_summary(manifests: dict[str, dict[str, Any]]) -> dict[str, Any]:
164    root_names = sorted(manifests)
165    return {
166        "root_names": root_names,
167        "root_count": len(root_names),
168        "manifests_sha256": sha256_json(
169            {name: manifests[name] for name in root_names}
170        ),
171    }
172
173
174def build_snapshot(
175    candidate_digest: str,
176    inventory: dict[str, dict[str, Path]] | None = None,
177) -> dict[str, Any]:
178    inventory = inventory or roots()
179    manifests = {
180        scope: {name: manifest_root(path) for name, path in paths.items()}
181        for scope, paths in inventory.items()
182    }
183    return {
184        "schema": SCHEMA,
185        "kind": "snapshot",
186        "candidate_digest": validate_candidate_digest(candidate_digest),
187        "captured_at": datetime.now(timezone.utc).isoformat(),
188        "collector": collector_binding(),
189        "user": {"uid": os.getuid()},
190        "keychain_locator": EXPECTED_KEYCHAIN,
191        "root_inventory": {
192            scope: inventory_summary(scope_manifests)
193            for scope, scope_manifests in manifests.items()
194        },
195        "roots": manifests,
196    }
197
198
199def write_new_json(path: Path, value: dict[str, Any]) -> None:
200    path.parent.mkdir(parents=True, exist_ok=True)
201    with path.open("x", encoding="utf-8") as handle:
202        json.dump(value, handle, indent=2, sort_keys=True)
203        handle.write("\n")
204
205
206def read_json(path: Path) -> dict[str, Any]:
207    try:
208        value = json.loads(path.read_text(encoding="utf-8"))
209    except (OSError, json.JSONDecodeError) as error:
210        raise LifecycleError(f"cannot read proof JSON {path}: {error}") from error
211    if not isinstance(value, dict):
212        raise LifecycleError(f"proof JSON is not an object: {path}")
213    return value
214
215
216def validate_collector(value: dict[str, Any]) -> None:
217    if value.get("collector") != collector_binding():
218        raise LifecycleError("lifecycle receipt collector binding differs")
219
220
221def validate_manifest_summary(
222    name: str, path: Path, summary: Any
223) -> dict[str, Any]:
224    if not isinstance(summary, dict):
225        raise LifecycleError(f"lifecycle root {name} has no manifest summary")
226    status = summary.get("status")
227    base_fields = {
228        "path_digest",
229        "status",
230        "files",
231        "directories",
232        "symlinks",
233        "bytes",
234        "manifest_sha256",
235    }
236    expected_fields = (
237        base_fields | {"error_count", "error_types"}
238        if status == "blocked"
239        else base_fields
240    )
241    if set(summary) != expected_fields:
242        raise LifecycleError(f"lifecycle root {name} manifest fields differ")
243    if summary.get("path_digest") != path_digest(path):
244        raise LifecycleError(f"lifecycle root {name} path binding differs")
245    for field in ("files", "directories", "symlinks", "bytes"):
246        if not isinstance(summary.get(field), int) or summary[field] < 0:
247            raise LifecycleError(f"lifecycle root {name} {field} is invalid")
248    if re.fullmatch(r"[0-9a-f]{64}", str(summary.get("manifest_sha256", ""))) is None:
249        raise LifecycleError(f"lifecycle root {name} manifest digest is invalid")
250    if status == "absent":
251        if any(summary[field] != 0 for field in ("files", "directories", "symlinks", "bytes")):
252            raise LifecycleError(f"absent lifecycle root {name} has content")
253        if summary["manifest_sha256"] != hashlib.sha256(b"absent").hexdigest():
254            raise LifecycleError(f"absent lifecycle root {name} digest differs")
255    elif status == "captured":
256        pass
257    elif status == "blocked":
258        if (
259            not isinstance(summary.get("error_count"), int)
260            or summary["error_count"] <= 0
261            or not isinstance(summary.get("error_types"), list)
262            or not summary["error_types"]
263            or not all(isinstance(error, str) and error for error in summary["error_types"])
264        ):
265            raise LifecycleError(f"blocked lifecycle root {name} errors are invalid")
266    else:
267        raise LifecycleError(f"lifecycle root {name} status is invalid")
268    return summary
269
270
271def validate_snapshot(
272    value: dict[str, Any],
273    candidate_digest: str,
274    inventory: dict[str, dict[str, Path]] | None = None,
275) -> None:
276    inventory = inventory or roots()
277    digest = validate_candidate_digest(candidate_digest)
278    if value.get("schema") != SCHEMA or value.get("kind") != "snapshot":
279        raise LifecycleError("input is not a lifecycle snapshot")
280    if value.get("candidate_digest") != digest:
281        raise LifecycleError("lifecycle snapshot binds a different candidate")
282    validate_collector(value)
283    if value.get("keychain_locator") != EXPECTED_KEYCHAIN:
284        raise LifecycleError("lifecycle snapshot keychain locator differs")
285    captured_roots = value.get("roots")
286    captured_inventory = value.get("root_inventory")
287    if not isinstance(captured_roots, dict) or set(captured_roots) != set(inventory):
288        raise LifecycleError("lifecycle snapshot scope inventory differs")
289    if not isinstance(captured_inventory, dict) or set(captured_inventory) != set(
290        inventory
291    ):
292        raise LifecycleError("lifecycle snapshot root inventory summary differs")
293    for scope, expected_paths in inventory.items():
294        scope_roots = captured_roots.get(scope)
295        if not isinstance(scope_roots, dict) or set(scope_roots) != set(expected_paths):
296            raise LifecycleError(f"lifecycle snapshot {scope} root inventory differs")
297        validated = {
298            name: validate_manifest_summary(name, expected_paths[name], scope_roots[name])
299            for name in sorted(expected_paths)
300        }
301        if captured_inventory.get(scope) != inventory_summary(validated):
302            raise LifecycleError(
303                f"lifecycle snapshot {scope} manifest summary digest differs"
304            )
305
306
307def validate_exact_app(path: Path, *, must_exist: bool) -> None:
308    if path != EXPECTED_APP:
309        raise LifecycleError(f"app path must be exactly {EXPECTED_APP}")
310    if not path.exists():
311        if must_exist:
312            raise LifecycleError(f"Omega RC app is missing: {path}")
313        return
314    if path.is_symlink() or not path.is_dir():
315        raise LifecycleError("Omega RC app must be a real application directory")
316    try:
317        info = plistlib.loads((path / "Contents/Info.plist").read_bytes())
318    except (OSError, plistlib.InvalidFileException) as error:
319        raise LifecycleError(f"cannot inspect Omega RC application: {error}") from error
320    if info.get("CFBundleIdentifier") != EXPECTED_BUNDLE_ID:
321        raise LifecycleError("refusing application with a non-Omega RC bundle identifier")
322
323
324def validate_holding_directory(path: Path) -> None:
325    if not path.is_absolute() or path in (Path("/"), Path.home(), Path("/Applications")):
326        raise LifecycleError("holding directory must be an explicit narrow absolute path")
327    for owned in roots()["omega"].values():
328        if path == owned or owned in path.parents or path in owned.parents:
329            raise LifecycleError("holding directory must be outside every Omega-owned root")
330    if path.is_symlink():
331        raise LifecycleError("holding directory must not be a symlink")
332
333
334def snapshot_command(arguments: argparse.Namespace) -> int:
335    write_new_json(arguments.output, build_snapshot(arguments.candidate_digest))
336    print(f"wrote lifecycle snapshot: {arguments.output}")
337    return 0
338
339
340def compare_command(arguments: argparse.Namespace) -> int:
341    before = read_json(arguments.before)
342    after = read_json(arguments.after)
343    digest = validate_candidate_digest(arguments.candidate_digest)
344    validate_snapshot(before, digest)
345    validate_snapshot(after, digest)
346    before_scope = before.get("roots", {}).get(arguments.scope)
347    after_scope = after.get("roots", {}).get(arguments.scope)
348    if not isinstance(before_scope, dict) or not isinstance(after_scope, dict):
349        raise LifecycleError(f"snapshot has no {arguments.scope} scope")
350    changed = sorted(
351        name
352        for name in set(before_scope) | set(after_scope)
353        if before_scope.get(name) != after_scope.get(name)
354    )
355    blocked_roots = sorted(
356        name
357        for name in set(before_scope) | set(after_scope)
358        if before_scope.get(name, {}).get("status") == "blocked"
359        or after_scope.get(name, {}).get("status") == "blocked"
360    )
361    status = "blocked" if blocked_roots else ("passed" if not changed else "changed")
362    report = {
363        "schema": SCHEMA,
364        "kind": "comparison",
365        "candidate_digest": digest,
366        "collector": collector_binding(),
367        "scope": arguments.scope,
368        "expected_root_names": sorted(roots()[arguments.scope]),
369        "before_snapshot_sha256": sha256_file(arguments.before),
370        "after_snapshot_sha256": sha256_file(arguments.after),
371        "before_manifests_sha256": before["root_inventory"][arguments.scope][
372            "manifests_sha256"
373        ],
374        "after_manifests_sha256": after["root_inventory"][arguments.scope][
375            "manifests_sha256"
376        ],
377        "status": status,
378        "blocked_roots": blocked_roots,
379        "changed_roots": changed,
380    }
381    validate_comparison(
382        report,
383        digest,
384        arguments.before,
385        arguments.after,
386    )
387    write_new_json(arguments.output, report)
388    print(f"wrote lifecycle comparison: {arguments.output}")
389    if blocked_roots:
390        return 2
391    return 0 if not changed else 1
392
393
394def validate_comparison(
395    value: dict[str, Any],
396    candidate_digest: str,
397    before_path: Path,
398    after_path: Path,
399    inventory: dict[str, dict[str, Path]] | None = None,
400) -> None:
401    inventory = inventory or roots()
402    digest = validate_candidate_digest(candidate_digest)
403    if value.get("schema") != SCHEMA or value.get("kind") != "comparison":
404        raise LifecycleError("input is not a lifecycle comparison")
405    if value.get("candidate_digest") != digest:
406        raise LifecycleError("lifecycle comparison binds a different candidate")
407    validate_collector(value)
408    scope = value.get("scope")
409    if scope not in inventory:
410        raise LifecycleError("lifecycle comparison scope is invalid")
411    before = read_json(before_path)
412    after = read_json(after_path)
413    validate_snapshot(before, digest, inventory)
414    validate_snapshot(after, digest, inventory)
415    before_scope = before["roots"][scope]
416    after_scope = after["roots"][scope]
417    changed = sorted(
418        name
419        for name in inventory[scope]
420        if before_scope.get(name) != after_scope.get(name)
421    )
422    blocked_roots = sorted(
423        name
424        for name in inventory[scope]
425        if before_scope.get(name, {}).get("status") == "blocked"
426        or after_scope.get(name, {}).get("status") == "blocked"
427    )
428    expected = {
429        "expected_root_names": sorted(inventory[scope]),
430        "before_snapshot_sha256": sha256_file(before_path),
431        "after_snapshot_sha256": sha256_file(after_path),
432        "before_manifests_sha256": before["root_inventory"][scope][
433            "manifests_sha256"
434        ],
435        "after_manifests_sha256": after["root_inventory"][scope][
436            "manifests_sha256"
437        ],
438        "status": (
439            "blocked" if blocked_roots else ("passed" if not changed else "changed")
440        ),
441        "blocked_roots": blocked_roots,
442        "changed_roots": changed,
443    }
444    for field, expected_value in expected.items():
445        if value.get(field) != expected_value:
446            raise LifecycleError(f"lifecycle comparison {field} differs")
447
448
449def remove_app_command(arguments: argparse.Namespace) -> int:
450    if arguments.confirm != "REMOVE-OMEGA-RC-APP":
451        raise LifecycleError("refusing app removal without --confirm REMOVE-OMEGA-RC-APP")
452    validate_exact_app(arguments.app, must_exist=True)
453    validate_holding_directory(arguments.holding_dir)
454    arguments.holding_dir.mkdir(parents=True, mode=0o700, exist_ok=False)
455    held_app = arguments.holding_dir / "Omega.app"
456    if os.stat(arguments.app.parent).st_dev != os.stat(arguments.holding_dir).st_dev:
457        raise LifecycleError("holding directory must be on the same filesystem as /Applications")
458    os.replace(arguments.app, held_app)
459    receipt = {
460        "schema": SCHEMA,
461        "kind": "app_removal",
462        "candidate_digest": validate_candidate_digest(arguments.candidate_digest),
463        "collector": collector_binding(),
464        "performed_at": datetime.now(timezone.utc).isoformat(),
465        "source_path": str(EXPECTED_APP),
466        "holding_path": str(held_app),
467        "recoverable": True,
468        "zed_targets_touched": False,
469    }
470    write_new_json(arguments.output, receipt)
471    print(f"moved Omega RC app to recoverable holding path: {held_app}")
472    return 0
473
474
475def run(command: list[str], description: str) -> str:
476    result = subprocess.run(command, capture_output=True, text=True, check=False)
477    if result.returncode != 0:
478        detail = "\n".join(part for part in (result.stdout, result.stderr) if part).strip()
479        raise LifecycleError(f"{description} failed" + (f": {detail}" if detail else ""))
480    return result.stdout
481
482
483def reinstall_command(arguments: argparse.Namespace) -> int:
484    if arguments.confirm != "REINSTALL-OMEGA-RC-APP":
485        raise LifecycleError("refusing reinstall without --confirm REINSTALL-OMEGA-RC-APP")
486    validate_exact_app(arguments.app, must_exist=False)
487    if arguments.app.exists():
488        raise LifecycleError("refusing to overwrite an existing Omega.app")
489    record = read_json(arguments.release_record)
490    if record.get("bundle_identifier") != EXPECTED_BUNDLE_ID:
491        raise LifecycleError("release record is not for Omega RC")
492    expected_digest = record.get("digests", {}).get("package_sha256")
493    if expected_digest != sha256_file(arguments.artifact):
494        raise LifecycleError("artifact digest differs from the release record")
495    mount_root = Path(tempfile.mkdtemp(prefix="omega-rc-lifecycle-mount-"))
496    mounted = False
497    temporary_app = arguments.app.parent / f".Omega.app.install-{os.getpid()}"
498    try:
499        run(
500            [
501                "hdiutil",
502                "attach",
503                "-readonly",
504                "-nobrowse",
505                "-mountpoint",
506                str(mount_root),
507                str(arguments.artifact),
508            ],
509            "read-only DMG mount",
510        )
511        mounted = True
512        packaged_app = mount_root / "Omega.app"
513        validate_exact_bundle_identifier(packaged_app)
514        run(["ditto", str(packaged_app), str(temporary_app)], "copy exact packaged app")
515        validate_exact_bundle_identifier(temporary_app)
516        os.replace(temporary_app, arguments.app)
517    finally:
518        if temporary_app.exists():
519            shutil.rmtree(temporary_app)
520        if mounted:
521            run(["hdiutil", "detach", str(mount_root)], "DMG detach")
522        mount_root.rmdir()
523    receipt = {
524        "schema": SCHEMA,
525        "kind": "app_reinstall",
526        "candidate_digest": validate_candidate_digest(arguments.candidate_digest),
527        "collector": collector_binding(),
528        "performed_at": datetime.now(timezone.utc).isoformat(),
529        "artifact_sha256": expected_digest,
530        "release_record_sha256": sha256_file(arguments.release_record),
531        "installed_path": str(arguments.app),
532        "zed_targets_touched": False,
533    }
534    write_new_json(arguments.output, receipt)
535    print(f"reinstalled exact Omega RC app: {arguments.app}")
536    return 0
537
538
539def validate_exact_bundle_identifier(app: Path) -> None:
540    if app.is_symlink() or not app.is_dir():
541        raise LifecycleError(f"application bundle is missing or unsafe: {app}")
542    try:
543        info = plistlib.loads((app / "Contents/Info.plist").read_bytes())
544    except (OSError, plistlib.InvalidFileException) as error:
545        raise LifecycleError(f"cannot inspect application bundle: {error}") from error
546    if info.get("CFBundleIdentifier") != EXPECTED_BUNDLE_ID:
547        raise LifecycleError("application bundle identifier is not Omega RC")
548
549
550def cleanup_data_command(arguments: argparse.Namespace) -> int:
551    if arguments.confirm != "MOVE-OMEGA-RC-DATA-TO-HOLDING":
552        raise LifecycleError(
553            "refusing cleanup without --confirm MOVE-OMEGA-RC-DATA-TO-HOLDING"
554        )
555    validate_holding_directory(arguments.holding_dir)
556    arguments.holding_dir.mkdir(parents=True, mode=0o700, exist_ok=False)
557    moved: list[str] = []
558    for name, source in roots()["omega"].items():
559        if name == "application" or (not source.exists() and not source.is_symlink()):
560            continue
561        if source.is_symlink():
562            raise LifecycleError(f"refusing symlinked Omega data root: {name}")
563        target = arguments.holding_dir / name
564        target.parent.mkdir(parents=True, exist_ok=True)
565        if os.stat(source.parent).st_dev != os.stat(arguments.holding_dir).st_dev:
566            raise LifecycleError(f"holding directory is not on the same filesystem as {name}")
567        os.replace(source, target)
568        moved.append(name)
569    receipt = {
570        "schema": SCHEMA,
571        "kind": "data_cleanup",
572        "candidate_digest": validate_candidate_digest(arguments.candidate_digest),
573        "collector": collector_binding(),
574        "performed_at": datetime.now(timezone.utc).isoformat(),
575        "moved_omega_roots": moved,
576        "holding_directory": str(arguments.holding_dir),
577        "recoverable": True,
578        "keychain_touched": False,
579        "zed_targets_touched": False,
580    }
581    write_new_json(arguments.output, receipt)
582    print(f"moved {len(moved)} Omega data roots to recoverable holding")
583    return 0
584
585
586def validate_command(arguments: argparse.Namespace) -> int:
587    value = read_json(arguments.input)
588    digest = validate_candidate_digest(arguments.candidate_digest)
589    kind = value.get("kind")
590    if kind == "snapshot":
591        validate_snapshot(value, digest)
592    elif kind == "comparison":
593        if arguments.before is None or arguments.after is None:
594            raise LifecycleError(
595                "comparison validation requires --before and --after snapshots"
596            )
597        validate_comparison(value, digest, arguments.before, arguments.after)
598    elif kind in {"app_removal", "app_reinstall", "data_cleanup"}:
599        if value.get("schema") != SCHEMA or value.get("candidate_digest") != digest:
600            raise LifecycleError("lifecycle operation receipt binding differs")
601        validate_collector(value)
602    else:
603        raise LifecycleError("lifecycle receipt kind is unsupported")
604    print(f"lifecycle receipt valid: {arguments.input}")
605    return 0
606
607
608def self_test_command(_arguments: argparse.Namespace) -> int:
609    digest = "a" * 64
610    with tempfile.TemporaryDirectory(prefix="omega-lifecycle-self-test-") as directory:
611        test_root = Path(directory)
612        sample = test_root / "sample"
613        sample.mkdir()
614        (sample / "public.txt").write_text("public evidence", encoding="utf-8")
615        os.symlink("public.txt", sample / "link")
616        first = manifest_root(sample)
617        if first["status"] != "captured" or first["files"] != 1 or first["symlinks"] != 1:
618            raise LifecycleError("self-test manifest did not capture files and symlinks")
619        (sample / "public.txt").write_text("changed evidence", encoding="utf-8")
620        second = manifest_root(sample)
621        if first["manifest_sha256"] == second["manifest_sha256"]:
622            raise LifecycleError("self-test manifest did not detect changed content")
623        control = test_root / "control"
624        control.mkdir()
625        test_inventory = {
626            "omega": {"sample": sample},
627            "zed": {"control": control},
628        }
629        snapshot = build_snapshot(digest, test_inventory)
630        validate_snapshot(snapshot, digest, test_inventory)
631
632        def reject_snapshot(tampered: dict[str, Any], description: str) -> None:
633            try:
634                validate_snapshot(tampered, digest, test_inventory)
635            except LifecycleError:
636                return
637            raise LifecycleError(f"self-test accepted {description}")
638
639        tampered = json.loads(json.dumps(snapshot))
640        del tampered["roots"]["omega"]["sample"]
641        reject_snapshot(tampered, "missing expected root")
642        tampered = json.loads(json.dumps(snapshot))
643        tampered["roots"]["omega"]["sample"]["manifest_sha256"] = "b" * 64
644        reject_snapshot(tampered, "forged manifest summary")
645        tampered = json.loads(json.dumps(snapshot))
646        tampered["collector"]["sha256"] = "b" * 64
647        reject_snapshot(tampered, "forged collector digest")
648
649        before_path = test_root / "before.json"
650        after_path = test_root / "after.json"
651        before_path.write_text(json.dumps(snapshot), encoding="utf-8")
652        after_path.write_text(json.dumps(snapshot), encoding="utf-8")
653        comparison = {
654            "schema": SCHEMA,
655            "kind": "comparison",
656            "candidate_digest": digest,
657            "collector": collector_binding(),
658            "scope": "zed",
659            "expected_root_names": ["control"],
660            "before_snapshot_sha256": sha256_file(before_path),
661            "after_snapshot_sha256": sha256_file(after_path),
662            "before_manifests_sha256": snapshot["root_inventory"]["zed"][
663                "manifests_sha256"
664            ],
665            "after_manifests_sha256": snapshot["root_inventory"]["zed"][
666                "manifests_sha256"
667            ],
668            "status": "passed",
669            "blocked_roots": [],
670            "changed_roots": [],
671        }
672        validate_comparison(
673            comparison, digest, before_path, after_path, test_inventory
674        )
675        blocked_snapshot = json.loads(json.dumps(snapshot))
676        blocked_manifest = blocked_snapshot["roots"]["zed"]["control"]
677        blocked_manifest["status"] = "blocked"
678        blocked_manifest["error_count"] = 1
679        blocked_manifest["error_types"] = ["PermissionError"]
680        blocked_snapshot["root_inventory"]["zed"] = inventory_summary(
681            blocked_snapshot["roots"]["zed"]
682        )
683        before_path.write_text(json.dumps(blocked_snapshot), encoding="utf-8")
684        after_path.write_text(json.dumps(blocked_snapshot), encoding="utf-8")
685        forged_pass = {
686            **comparison,
687            "before_snapshot_sha256": sha256_file(before_path),
688            "after_snapshot_sha256": sha256_file(after_path),
689            "before_manifests_sha256": blocked_snapshot["root_inventory"]["zed"][
690                "manifests_sha256"
691            ],
692            "after_manifests_sha256": blocked_snapshot["root_inventory"]["zed"][
693                "manifests_sha256"
694            ],
695        }
696        try:
697            validate_comparison(
698                forged_pass, digest, before_path, after_path, test_inventory
699            )
700        except LifecycleError:
701            pass
702        else:
703            raise LifecycleError("self-test accepted blocked manifests as passed")
704        before_path.write_text(json.dumps(snapshot), encoding="utf-8")
705        after_path.write_text(json.dumps(snapshot), encoding="utf-8")
706        comparison["expected_root_names"] = []
707        try:
708            validate_comparison(
709                comparison, digest, before_path, after_path, test_inventory
710            )
711        except LifecycleError:
712            pass
713        else:
714            raise LifecycleError("self-test accepted incomplete comparison inventory")
715        if validate_candidate_digest(digest) != digest:
716            raise LifecycleError("self-test candidate digest validation failed")
717        try:
718            validate_holding_directory(Path.home())
719        except LifecycleError:
720            pass
721        else:
722            raise LifecycleError("self-test accepted the home directory as a holding path")
723    print("Omega RC lifecycle proof helper self-test passed")
724    return 0
725
726
727def parse_arguments() -> argparse.Namespace:
728    parser = argparse.ArgumentParser(description=__doc__)
729    subparsers = parser.add_subparsers(dest="command", required=True)
730
731    self_test = subparsers.add_parser("self-test")
732    self_test.set_defaults(handler=self_test_command)
733
734    snapshot = subparsers.add_parser("snapshot")
735    snapshot.add_argument("--candidate-digest", required=True)
736    snapshot.add_argument("--output", type=Path, required=True)
737    snapshot.set_defaults(handler=snapshot_command)
738
739    validate = subparsers.add_parser("validate")
740    validate.add_argument("--candidate-digest", required=True)
741    validate.add_argument("--input", type=Path, required=True)
742    validate.add_argument("--before", type=Path)
743    validate.add_argument("--after", type=Path)
744    validate.set_defaults(handler=validate_command)
745
746    compare = subparsers.add_parser("compare")
747    compare.add_argument("--candidate-digest", required=True)
748    compare.add_argument("--before", type=Path, required=True)
749    compare.add_argument("--after", type=Path, required=True)
750    compare.add_argument("--scope", choices=("omega", "zed"), required=True)
751    compare.add_argument("--output", type=Path, required=True)
752    compare.set_defaults(handler=compare_command)
753
754    remove_app = subparsers.add_parser("remove-app")
755    remove_app.add_argument("--candidate-digest", required=True)
756    remove_app.add_argument("--app", type=Path, default=EXPECTED_APP)
757    remove_app.add_argument("--holding-dir", type=Path, required=True)
758    remove_app.add_argument("--confirm", required=True)
759    remove_app.add_argument("--output", type=Path, required=True)
760    remove_app.set_defaults(handler=remove_app_command)
761
762    reinstall = subparsers.add_parser("reinstall")
763    reinstall.add_argument("--candidate-digest", required=True)
764    reinstall.add_argument("--artifact", type=Path, required=True)
765    reinstall.add_argument("--release-record", type=Path, required=True)
766    reinstall.add_argument("--app", type=Path, default=EXPECTED_APP)
767    reinstall.add_argument("--confirm", required=True)
768    reinstall.add_argument("--output", type=Path, required=True)
769    reinstall.set_defaults(handler=reinstall_command)
770
771    cleanup = subparsers.add_parser("cleanup-data")
772    cleanup.add_argument("--candidate-digest", required=True)
773    cleanup.add_argument("--holding-dir", type=Path, required=True)
774    cleanup.add_argument("--confirm", required=True)
775    cleanup.add_argument("--output", type=Path, required=True)
776    cleanup.set_defaults(handler=cleanup_data_command)
777    return parser.parse_args()
778
779
780def main() -> int:
781    arguments = parse_arguments()
782    try:
783        return arguments.handler(arguments)
784    except LifecycleError as error:
785        print(f"error: {error}", file=os.sys.stderr)
786        return 2
787
788
789if __name__ == "__main__":
790    raise SystemExit(main())
791