tenant.openagents/omega

1#!/usr/bin/env python3
2"""Collect public-safe Full Auto receipts from one exact installed Omega candidate."""
3
4from __future__ import annotations
5
6import argparse
7import hashlib
8import json
9import os
10import plistlib
11import re
12import stat
13import subprocess
14import sys
15import tempfile
16from datetime import datetime
17from pathlib import Path
18from typing import Any
19
20
21SCHEMA = "openagents.omega.full-auto-installed-run-evidence.v1"
22RUNS_SCHEMA = "openagents.desktop.full_auto_run_registry.v1"
23REPORTS_SCHEMA = "openagents.desktop.full_auto_run_report.v1"
24HANDOFFS_SCHEMA = "openagents.desktop.provider_handoff_transition.v1"
25NATIVE_BINDINGS_SCHEMA = "openagents.omega.full_auto_native_binding.v1"
26REGISTRY_FILES = {
27    "runs": ("runs.json", RUNS_SCHEMA, "runs"),
28    "reports": ("run-reports.json", REPORTS_SCHEMA, "reports"),
29    "handoffs": ("provider-handoffs.json", HANDOFFS_SCHEMA, "transitions"),
30    "native_bindings": ("native-bindings.json", NATIVE_BINDINGS_SCHEMA, "bindings"),
31}
32RUN_REF_RE = re.compile(r"run\.full-auto\.[a-z0-9.-]+")
33REF_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{0,255}")
34SHA256_RE = re.compile(r"[0-9a-f]{64}")
35COMMIT_RE = re.compile(r"[0-9a-f]{40}")
36ALLOWED_STATES = {
37    "draft", "running", "pausing", "paused", "stalled", "stopping",
38    "stopped", "completed", "failed",
39}
40ALLOWED_ACTORS = {"control_api", "effectd", "host", "owner", "system"}
41ALLOWED_LANES = {"codex-local", "claude-local"}
42ALLOWED_DISPOSITIONS = {
43    "complete_within_bounds", "completed", "failed", "interrupted",
44    "stopped", "cancelled", "timed_out", "unknown",
45}
46SENSITIVE_KEY_PARTS = {
47    "objective", "prompt", "message", "reason", "title", "path", "content",
48    "credential", "token", "secret", "password", "transcript", "summary",
49}
50SECRET_PATTERNS = (
51    re.compile(r"sk-[A-Za-z0-9_-]{12,}"),
52    re.compile(r"(?i)bearer\s+[A-Za-z0-9._~+/-]{12,}"),
53    re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----"),
54)
55
56
57class EvidenceError(RuntimeError):
58    pass
59
60
61def duplicate_key_guard(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
62    result: dict[str, Any] = {}
63    for key, value in pairs:
64        if key in result:
65            raise EvidenceError(f"duplicate JSON key: {key}")
66        result[key] = value
67    return result
68
69
70def canonical_bytes(value: Any) -> bytes:
71    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
72
73
74def digest_bytes(value: bytes) -> str:
75    return hashlib.sha256(value).hexdigest()
76
77
78def digest_file(path: Path) -> str:
79    digest = hashlib.sha256()
80    with path.open("rb") as handle:
81        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
82            digest.update(chunk)
83    return digest.hexdigest()
84
85
86def digest_value(value: Any) -> str:
87    return digest_bytes(canonical_bytes(value))
88
89
90def digest_ref(value: str, label: str) -> str:
91    require_ref(value, label)
92    return digest_bytes(value.encode())
93
94
95def require_ref(value: Any, label: str) -> str:
96    if not isinstance(value, str) or REF_RE.fullmatch(value) is None:
97        raise EvidenceError(f"{label} is not a safe public reference")
98    return value
99
100
101def require_sha256(value: Any, label: str) -> str:
102    if not isinstance(value, str) or SHA256_RE.fullmatch(value) is None:
103        raise EvidenceError(f"{label} must be a lowercase SHA-256 digest")
104    return value
105
106
107def require_time(value: Any, label: str) -> str:
108    if not isinstance(value, str):
109        raise EvidenceError(f"{label} must be an ISO-8601 timestamp")
110    try:
111        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
112    except ValueError as error:
113        raise EvidenceError(f"{label} must be an ISO-8601 timestamp") from error
114    if parsed.tzinfo is None:
115        raise EvidenceError(f"{label} must include a timezone")
116    return value
117
118
119def require_integer(value: Any, label: str) -> int:
120    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
121        raise EvidenceError(f"{label} must be a nonnegative integer")
122    return value
123
124
125def require_no_symlink_components(path: Path, label: str) -> None:
126    absolute = path.absolute()
127    current = Path(absolute.anchor)
128    for component in absolute.parts[1:]:
129        current /= component
130        if current.is_symlink():
131            raise EvidenceError(f"{label} traverses a symbolic link")
132
133
134def require_regular(path: Path, label: str, *, private: bool = False) -> os.stat_result:
135    require_no_symlink_components(path, label)
136    try:
137        metadata = path.stat()
138    except OSError as error:
139        raise EvidenceError(f"cannot inspect {label}: {error}") from error
140    if not stat.S_ISREG(metadata.st_mode):
141        raise EvidenceError(f"{label} must be a regular non-symlink file")
142    mode = stat.S_IMODE(metadata.st_mode)
143    if private and mode & 0o077:
144        raise EvidenceError(f"{label} must not grant group or other permissions")
145    if mode & 0o022:
146        raise EvidenceError(f"{label} must not be group- or world-writable")
147    return metadata
148
149
150def load_json(path: Path, label: str, *, private: bool = False) -> dict[str, Any]:
151    metadata = require_regular(path, label, private=private)
152    if metadata.st_size > 16 * 1024 * 1024:
153        raise EvidenceError(f"{label} is unexpectedly large")
154    try:
155        value = json.loads(path.read_text(encoding="utf-8"), object_pairs_hook=duplicate_key_guard)
156    except (OSError, UnicodeError, json.JSONDecodeError, EvidenceError) as error:
157        raise EvidenceError(f"cannot read {label}: {error}") from error
158    if not isinstance(value, dict):
159        raise EvidenceError(f"{label} must be a JSON object")
160    return value
161
162
163def exact_object(value: Any, keys: set[str], label: str) -> dict[str, Any]:
164    if not isinstance(value, dict):
165        raise EvidenceError(f"{label} must be an object")
166    if set(value) != keys:
167        raise EvidenceError(f"{label} keys are not exact")
168    return value
169
170
171def validate_registry_root(root: Path, release: dict[str, Any]) -> Path:
172    if not root.is_absolute():
173        raise EvidenceError("registry root must be absolute")
174    require_no_symlink_components(root, "registry root")
175    try:
176        metadata = root.stat()
177    except OSError as error:
178        raise EvidenceError(f"cannot inspect registry root: {error}") from error
179    if not stat.S_ISDIR(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) & 0o077:
180        raise EvidenceError("registry root must be a private non-symlink directory")
181    if sys.platform == "darwin":
182        volume_name = release.get("volume_name")
183        if not isinstance(volume_name, str) or not volume_name.startswith("Omega"):
184            raise EvidenceError("release record has no valid Omega release namespace")
185        expected = Path.home() / "Library/Application Support" / volume_name / "openagents/full-auto"
186        if root != expected:
187            raise EvidenceError("registry root is outside the release-record namespace")
188    return root
189
190
191def candidate_bindings(release_path: Path, identity_path: Path, app: Path) -> dict[str, Any]:
192    release = load_json(release_path, "release record")
193    identity = load_json(identity_path, "identity candidate evidence")
194    if app != Path("/Applications/Omega.app"):
195        raise EvidenceError("installed application must be exactly /Applications/Omega.app")
196    require_no_symlink_components(app, "installed application")
197    if not app.is_dir():
198        raise EvidenceError("installed application is missing")
199    if stat.S_IMODE(app.stat().st_mode) & 0o022:
200        raise EvidenceError("installed application must not be group- or world-writable")
201    info_path = app / "Contents/Info.plist"
202    binary_path = app / "Contents/MacOS/omega"
203    require_regular(info_path, "installed Info.plist")
204    require_regular(binary_path, "installed Omega executable")
205    try:
206        with info_path.open("rb") as handle:
207            info = plistlib.load(handle)
208    except (OSError, plistlib.InvalidFileException) as error:
209        raise EvidenceError(f"cannot read installed Info.plist: {error}") from error
210    if info.get("CFBundleIdentifier") != release.get("bundle_identifier"):
211        raise EvidenceError("installed bundle identifier differs from release record")
212    bundle_version = release.get("bundle_version", release.get("version"))
213    if info.get("CFBundleShortVersionString") != bundle_version:
214        raise EvidenceError("installed version differs from release record")
215    binary_sha256 = digest_file(binary_path)
216    expected_binary = release.get("digests", {}).get("omega_binary_sha256")
217    if binary_sha256 != expected_binary:
218        raise EvidenceError("installed executable digest differs from release record")
219    artifact_sha256 = require_sha256(
220        release.get("digests", {}).get("package_sha256"), "release artifact digest"
221    )
222    omega_commit = release.get("source", {}).get("commit")
223    if not isinstance(omega_commit, str) or COMMIT_RE.fullmatch(omega_commit) is None:
224        raise EvidenceError("release record source commit is invalid")
225    candidate = identity.get("candidate")
226    candidate_digest = identity.get("candidate_digest", {}).get("value")
227    if not isinstance(candidate, dict):
228        raise EvidenceError("identity evidence has no candidate object")
229    require_sha256(candidate_digest, "identity candidate digest")
230    if digest_value(candidate) != candidate_digest:
231        raise EvidenceError("identity candidate digest is not canonical")
232    artifact = candidate.get("artifact", {})
233    if (
234        candidate.get("omega", {}).get("commit") != omega_commit
235        or artifact.get("sha256") != artifact_sha256
236        or artifact.get("release_record_sha256") != digest_file(release_path)
237    ):
238        raise EvidenceError("identity evidence binds a different candidate")
239    if sys.platform == "darwin":
240        result = subprocess.run(
241            ["codesign", "--verify", "--deep", "--strict", str(app)],
242            stdout=subprocess.PIPE, stderr=subprocess.PIPE, check=False,
243        )
244        if result.returncode != 0:
245            raise EvidenceError("installed application fails strict code-signature verification")
246    return {
247        "candidate_digest": candidate_digest,
248        "artifact_sha256": artifact_sha256,
249        "release_record_sha256": digest_file(release_path),
250        "omega_commit": omega_commit,
251        "installed_app_name": "Omega.app",
252        "installed_binary_sha256": binary_sha256,
253        "bundle_identifier": release["bundle_identifier"],
254        "release_version": release["version"],
255        "bundle_version": bundle_version,
256    }
257
258
259def transition_projection(value: Any, label: str) -> dict[str, Any]:
260    transition = exact_object(value, {"from", "to", "actor", "at", "reason"}, label)
261    source = transition["from"]
262    target = transition["to"]
263    actor = transition["actor"]
264    if source not in ALLOWED_STATES or target not in ALLOWED_STATES:
265        raise EvidenceError(f"{label} has an unknown state")
266    if actor not in ALLOWED_ACTORS:
267        raise EvidenceError(f"{label} has an unknown actor")
268    if not isinstance(transition["reason"], str):
269        raise EvidenceError(f"{label}.reason must be a string")
270    return {
271        "from": source, "to": target, "actor": actor,
272        "at": require_time(transition["at"], f"{label}.at"),
273        "record_sha256": digest_value(transition),
274    }
275
276
277def handoff_projection(value: Any, label: str, run_ref: str) -> dict[str, Any]:
278    keys = {
279        "handoffRef", "runRef", "threadRef", "sourceThreadRef", "targetThreadRef",
280        "from", "to", "actor", "at", "reason", "disposition", "truncated",
281        "envelopeSchema",
282    }
283    handoff = exact_object(value, keys, label)
284    if handoff["runRef"] != run_ref:
285        raise EvidenceError(f"{label} binds a different run")
286    if handoff["from"] not in ALLOWED_LANES or handoff["to"] not in ALLOWED_LANES:
287        raise EvidenceError(f"{label} has an unknown provider lane")
288    if handoff["from"] == handoff["to"]:
289        raise EvidenceError(f"{label} does not change provider lane")
290    if handoff["actor"] not in ALLOWED_ACTORS:
291        raise EvidenceError(f"{label} has an unknown actor")
292    if handoff["disposition"] not in ALLOWED_DISPOSITIONS:
293        raise EvidenceError(f"{label} has an unknown disposition")
294    if not isinstance(handoff["truncated"], bool) or not isinstance(handoff["reason"], str):
295        raise EvidenceError(f"{label} has invalid typed fields")
296    if handoff["envelopeSchema"] != "openagents.desktop.provider_handoff_envelope.v1":
297        raise EvidenceError(f"{label} has an unknown envelope schema")
298    return {
299        "handoff_ref": require_ref(handoff["handoffRef"], f"{label}.handoffRef"),
300        "from": handoff["from"], "to": handoff["to"], "actor": handoff["actor"],
301        "at": require_time(handoff["at"], f"{label}.at"),
302        "disposition": handoff["disposition"], "truncated": handoff["truncated"],
303        "envelope_schema": handoff["envelopeSchema"],
304        "thread_ref_sha256": digest_ref(handoff["threadRef"], f"{label}.threadRef"),
305        "source_thread_ref_sha256": digest_ref(handoff["sourceThreadRef"], f"{label}.sourceThreadRef"),
306        "target_thread_ref_sha256": digest_ref(handoff["targetThreadRef"], f"{label}.targetThreadRef"),
307        "record_sha256": digest_value(handoff),
308    }
309
310
311def one_by_run(records: list[Any], run_ref: str, label: str) -> dict[str, Any]:
312    matches = [record for record in records if isinstance(record, dict) and record.get("runRef") == run_ref]
313    if len(matches) != 1:
314        raise EvidenceError(f"{label} must contain exactly one record for {run_ref}")
315    return matches[0]
316
317
318def require_unique_registry_refs(records: list[Any], key: str, label: str) -> None:
319    observed: set[str] = set()
320    for index, record in enumerate(records):
321        if not isinstance(record, dict):
322            raise EvidenceError(f"{label}[{index}] must be an object")
323        value = require_ref(record.get(key), f"{label}[{index}].{key}")
324        if value in observed:
325            raise EvidenceError(f"{label} contains duplicate {key}: {value}")
326        observed.add(value)
327
328
329def collect(release_path: Path, identity_path: Path, app: Path, root: Path, run_refs: list[str]) -> dict[str, Any]:
330    if not run_refs or len(run_refs) != len(set(run_refs)):
331        raise EvidenceError("run refs must be a nonempty unique list")
332    for run_ref in run_refs:
333        if RUN_REF_RE.fullmatch(run_ref) is None:
334            raise EvidenceError(f"invalid run ref: {run_ref}")
335    bindings = candidate_bindings(release_path, identity_path, app)
336    release = load_json(release_path, "release record")
337    root = validate_registry_root(root, release)
338    registries: dict[str, dict[str, Any]] = {}
339    registry_evidence: dict[str, Any] = {}
340    for key, (filename, schema_name, array_key) in REGISTRY_FILES.items():
341        path = root / filename
342        document = load_json(path, f"{key} registry", private=True)
343        exact_object(document, {"schema", array_key}, f"{key} registry")
344        if document["schema"] != schema_name or not isinstance(document[array_key], list):
345            raise EvidenceError(f"{key} registry schema is invalid")
346        registries[key] = document
347        registry_evidence[key] = {
348            "schema": schema_name,
349            "sha256": digest_file(path),
350            "mode": "0600",
351        }
352    require_unique_registry_refs(registries["runs"]["runs"], "runRef", "runs registry")
353    require_unique_registry_refs(registries["reports"]["reports"], "runRef", "reports registry")
354    require_unique_registry_refs(
355        registries["native_bindings"]["bindings"], "runRef", "native bindings registry"
356    )
357    require_unique_registry_refs(
358        registries["handoffs"]["transitions"], "handoffRef", "handoffs registry"
359    )
360    runs = []
361    for run_ref in run_refs:
362        run = one_by_run(registries["runs"]["runs"], run_ref, "runs registry")
363        report = one_by_run(registries["reports"]["reports"], run_ref, "reports registry")
364        native = one_by_run(registries["native_bindings"]["bindings"], run_ref, "native bindings registry")
365        state = run.get("state")
366        if state not in ALLOWED_STATES or report.get("state") != state:
367            raise EvidenceError(f"run/report state mismatch for {run_ref}")
368        thread_ref = require_ref(run.get("threadRef"), f"{run_ref}.threadRef")
369        if report.get("threadRef") != thread_ref:
370            raise EvidenceError(f"run/report thread mismatch for {run_ref}")
371        transitions = run.get("transitions")
372        if not isinstance(transitions, list) or not transitions:
373            raise EvidenceError(f"{run_ref} has no lifecycle transitions")
374        projected_transitions = [
375            transition_projection(value, f"{run_ref}.transitions[{index}]")
376            for index, value in enumerate(transitions)
377        ]
378        handoffs = [
379            value for value in registries["handoffs"]["transitions"]
380            if isinstance(value, dict) and value.get("runRef") == run_ref
381        ]
382        projected_handoffs = [
383            handoff_projection(value, f"{run_ref}.handoffs[{index}]", run_ref)
384            for index, value in enumerate(handoffs)
385        ]
386        report_handoffs = report.get("providerTransitions")
387        if not isinstance(report_handoffs, list):
388            raise EvidenceError(f"{run_ref} report has invalid provider transitions")
389        if [digest_value(value) for value in report_handoffs] != [digest_value(value) for value in handoffs]:
390            raise EvidenceError(f"{run_ref} report and handoff registry disagree")
391        workspace_ref = require_ref(run.get("workspaceRef"), f"{run_ref}.workspaceRef")
392        if report.get("workspaceRef") != workspace_ref or native.get("workspaceRef") != workspace_ref:
393            raise EvidenceError(f"{run_ref} workspace binding mismatch")
394        if native.get("runRef") != run_ref or not isinstance(native.get("rebaseUnsafe"), bool):
395            raise EvidenceError(f"{run_ref} native binding is invalid")
396        worktree_path_digest = native.get("worktreePathDigest")
397        if worktree_path_digest is not None:
398            require_sha256(worktree_path_digest, f"{run_ref}.worktreePathDigest")
399        git_head = native.get("gitHead")
400        if git_head is not None and (not isinstance(git_head, str) or COMMIT_RE.fullmatch(git_head) is None):
401            raise EvidenceError(f"{run_ref}.gitHead is invalid")
402        objective_digest = require_sha256(report.get("objectiveDigest"), f"{run_ref}.objectiveDigest")
403        done_digest = require_sha256(report.get("doneConditionDigest"), f"{run_ref}.doneConditionDigest")
404        run_projection = {
405            "run_ref": run_ref,
406            "thread_ref_sha256": digest_ref(thread_ref, f"{run_ref}.threadRef"),
407            "workspace_ref_sha256": digest_ref(workspace_ref, f"{run_ref}.workspaceRef"),
408            "objective_sha256": objective_digest,
409            "done_condition_sha256": done_digest,
410            "state": state,
411            "state_revision": require_integer(run.get("stateRevision"), f"{run_ref}.stateRevision"),
412            "turn_cap": require_integer(run.get("turnCap"), f"{run_ref}.turnCap"),
413            "successful_attempts": require_integer(run.get("successfulAttempts"), f"{run_ref}.successfulAttempts"),
414            "failed_attempts": require_integer(run.get("failedAttempts"), f"{run_ref}.failedAttempts"),
415            "created_at": require_time(run.get("createdAt"), f"{run_ref}.createdAt"),
416            "started_at": require_time(run.get("startedAt"), f"{run_ref}.startedAt"),
417            "updated_at": require_time(report.get("updatedAt"), f"{run_ref}.updatedAt"),
418            "lifecycle_transitions": projected_transitions,
419            "provider_handoffs": projected_handoffs,
420            "native_binding": {
421                "project_ref_sha256": digest_ref(native.get("projectRef"), f"{run_ref}.projectRef"),
422                "worktree_ref_sha256": digest_ref(native.get("worktreeRef"), f"{run_ref}.worktreeRef"),
423                "worktree_path_sha256": worktree_path_digest,
424                "git_head": git_head,
425                "rebase_unsafe": native["rebaseUnsafe"],
426                "bound_at": require_time(native.get("boundAt"), f"{run_ref}.boundAt"),
427                "record_sha256": digest_value(native),
428            },
429            "run_record_sha256": digest_value(run),
430            "report_record_sha256": digest_value(report),
431        }
432        runs.append(run_projection)
433    evidence = {
434        "schema": SCHEMA,
435        **bindings,
436        "collector": {
437            "name": "collect-omega-full-auto-installed-evidence",
438            "version": 1,
439            "sha256": digest_file(Path(__file__)),
440        },
441        "registry_namespace": release["volume_name"],
442        "registries": registry_evidence,
443        "selected_run_count": len(runs),
444        "runs": runs,
445    }
446    validate_public_output(evidence)
447    evidence["evidence_sha256"] = digest_value(evidence)
448    return evidence
449
450
451def validate_public_output(value: Any, path: str = "$") -> None:
452    if isinstance(value, dict):
453        for key, child in value.items():
454            lowered = key.lower()
455            if any(part in lowered for part in SENSITIVE_KEY_PARTS):
456                allowed = key in {"objective_sha256", "done_condition_sha256", "worktree_path_sha256"}
457                if not allowed:
458                    raise EvidenceError(f"public output contains forbidden field at {path}.{key}")
459            validate_public_output(child, f"{path}.{key}")
460    elif isinstance(value, list):
461        for index, child in enumerate(value):
462            validate_public_output(child, f"{path}[{index}]")
463    elif isinstance(value, str):
464        for pattern in SECRET_PATTERNS:
465            if pattern.search(value):
466                raise EvidenceError(f"public output contains secret-shaped text at {path}")
467        if value.startswith("/") or re.search(r"(?:^|/)Users/", value):
468            raise EvidenceError(f"public output contains a raw path at {path}")
469
470
471def write_output(path: Path, evidence: dict[str, Any]) -> None:
472    if not path.is_absolute():
473        raise EvidenceError("output path must be absolute")
474    require_no_symlink_components(path.parent, "output parent")
475    try:
476        path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
477    except OSError as error:
478        raise EvidenceError(f"cannot create output directory: {error}") from error
479    if path.is_symlink():
480        raise EvidenceError("output must not be a symbolic link")
481    temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp")
482    try:
483        try:
484            descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
485            with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
486                json.dump(evidence, handle, indent=2, sort_keys=True)
487                handle.write("\n")
488                handle.flush()
489                os.fsync(handle.fileno())
490            os.replace(temporary, path)
491            os.chmod(path, 0o600)
492        except OSError as error:
493            raise EvidenceError(f"cannot write output: {error}") from error
494    finally:
495        if temporary.exists():
496            temporary.unlink()
497
498
499def self_test() -> None:
500    base_transition = {
501        "from": "running", "to": "paused", "actor": "control_api",
502        "at": "2026-07-25T00:00:00Z", "reason": "owner request",
503    }
504    projected = transition_projection(base_transition, "test transition")
505    if "reason" in projected or projected["to"] != "paused":
506        raise EvidenceError("transition projection leaked a reason")
507    if projected != transition_projection(dict(base_transition), "test transition"):
508        raise EvidenceError("transition projection is not deterministic")
509    handoff = {
510        "handoffRef": "handoff.provider.test", "runRef": "run.full-auto.test",
511        "threadRef": "thread.test", "sourceThreadRef": "thread.source",
512        "targetThreadRef": "thread.target", "from": "codex-local", "to": "claude-local",
513        "actor": "control_api", "at": "2026-07-25T00:00:01Z", "reason": "switch",
514        "disposition": "complete_within_bounds", "truncated": False,
515        "envelopeSchema": "openagents.desktop.provider_handoff_envelope.v1",
516    }
517    projected_handoff = handoff_projection(handoff, "test handoff", "run.full-auto.test")
518    if "reason" in projected_handoff or "thread.test" in canonical_bytes(projected_handoff).decode():
519        raise EvidenceError("handoff projection leaked private text or opaque refs")
520    for forged in (
521        '{"schema":"x","schema":"y"}',
522        '{"token":"sk-testSecretValue123456789"}',
523    ):
524        try:
525            value = json.loads(forged, object_pairs_hook=duplicate_key_guard)
526            validate_public_output(value)
527        except EvidenceError:
528            pass
529        else:
530            raise EvidenceError("duplicate-key or redaction forgery was accepted")
531    with tempfile.TemporaryDirectory() as temporary_directory:
532        root = Path(temporary_directory)
533        private = root / "private.json"
534        private.write_text('{"schema":"x"}\n', encoding="utf-8")
535        private.chmod(0o644)
536        try:
537            load_json(private, "private fixture", private=True)
538        except EvidenceError:
539            pass
540        else:
541            raise EvidenceError("weak registry mode was accepted")
542        target = root / "target.json"
543        target.write_text('{}\n', encoding="utf-8")
544        link = root / "link.json"
545        link.symlink_to(target)
546        try:
547            load_json(link, "symlink fixture")
548        except EvidenceError:
549            pass
550        else:
551            raise EvidenceError("symlink registry was accepted")
552        tampered = dict(handoff)
553        tampered["disposition"] = "secret: sk-testSecretValue123456789"
554        try:
555            handoff_projection(tampered, "tampered handoff", "run.full-auto.test")
556        except EvidenceError:
557            pass
558        else:
559            raise EvidenceError("tampered handoff was accepted")
560    print("Full Auto installed evidence collector self-test passed")
561
562
563def parse_args() -> argparse.Namespace:
564    parser = argparse.ArgumentParser(description=__doc__)
565    parser.add_argument("--release-record", type=Path)
566    parser.add_argument("--identity-evidence", type=Path)
567    parser.add_argument("--app", type=Path, default=Path("/Applications/Omega.app"))
568    parser.add_argument("--registry-root", type=Path)
569    parser.add_argument("--run-ref", action="append", dest="run_refs")
570    parser.add_argument("--output", type=Path)
571    parser.add_argument("--self-test", action="store_true")
572    return parser.parse_args()
573
574
575def main() -> int:
576    args = parse_args()
577    try:
578        if args.self_test:
579            self_test()
580            return 0
581        required = {
582            "--release-record": args.release_record,
583            "--identity-evidence": args.identity_evidence,
584            "--registry-root": args.registry_root,
585            "--run-ref": args.run_refs,
586            "--output": args.output,
587        }
588        missing = [name for name, value in required.items() if not value]
589        if missing:
590            raise EvidenceError(f"missing required arguments: {', '.join(missing)}")
591        evidence = collect(
592            args.release_record.absolute(), args.identity_evidence.absolute(),
593            args.app, args.registry_root, args.run_refs,
594        )
595        write_output(args.output, evidence)
596        print(f"wrote public-safe Full Auto installed evidence: {args.output}")
597        return 0
598    except EvidenceError as error:
599        print(f"Full Auto installed evidence refused: {error}", file=sys.stderr)
600        return 2
601
602
603if __name__ == "__main__":
604    raise SystemExit(main())
605