tenant.openagents/omega

1#!/usr/bin/env python3
2"""Produce, or truthfully refuse, an installed Omega RC proof."""
3
4from __future__ import annotations
5
6import argparse
7import hashlib
8import ipaddress
9import json
10import os
11import plistlib
12import re
13import selectors
14import stat
15import subprocess
16import sys
17import tempfile
18import time
19from datetime import datetime
20from pathlib import Path
21from typing import Any
22from xml.parsers.expat import ExpatError
23
24from omega_identity_evidence import (
25    IdentityEvidenceError,
26    resolve_evidence_reference,
27    validate_identity_matrix,
28    validate_installed_observations,
29    validate_installed_tripwires,
30    validate_recovery_evidence,
31)
32
33
34ROOT = Path(__file__).resolve().parent.parent
35DEFAULT_APP = Path("/Applications/Omega.app")
36DEFAULT_OUTPUT = ROOT / "target/omega-rc-proof"
37EXPECTED_PRODUCT = "Omega"
38
39
40def _expected_rc_version() -> str:
41    """Read the RC version from the packaging script, which owns it.
42
43    This constant used to be a literal here as well. It went stale — the
44    bundler moved to 0.2.0-rc3 while this proof still pinned 0.2.0-rc2, so the
45    proof would have refused the only artifact the repository can actually
46    build. A pinned expectation is right; a second copy of the pin is not.
47    """
48    source = ROOT / "script/bundle-omega-rc"
49    match = re.search(
50        r'^readonly OMEGA_RC_VERSION="([^"]+)"$', source.read_text(encoding="utf-8"), re.MULTILINE
51    )
52    if match is None:
53        raise SystemExit(
54            "cannot read OMEGA_RC_VERSION from script/bundle-omega-rc; "
55            "the packaging script owns the RC version"
56        )
57    return match.group(1)
58
59
60EXPECTED_VERSION = _expected_rc_version()
61EXPECTED_BUNDLE_ID = "com.openagents.omega.rc"
62EXPECTED_TEAM_ID = "HQWSG26L43"
63EXPECTED_SIGNER = "Developer ID Application: OpenAgents, Inc. (HQWSG26L43)"
64IDENTITY_EVIDENCE_SCHEMA = (
65    ROOT / "crates/omega_identity/fixtures/identity_candidate_evidence_schema_v1.json"
66)
67FULL_AUTO_EVIDENCE_SCHEMA = (
68    ROOT
69    / "crates/omega_identity/fixtures/full_auto_candidate_evidence_schema_v1.json"
70)
71ENDPOINT_ALLOWLIST = ROOT / "crates/app_identity/fixtures/endpoint_allowlist.json"
72EXPECTED_IDENTITY_GATES = (
73    "assurance_spec_admission",
74    "custody_scenarios",
75    "recovery_scenarios",
76    "forged_request_rejection",
77    "stale_task_fencing",
78    "installed_secret_tripwires",
79    "manual_journey",
80    "owner_observation",
81    "independent_verification",
82    "accessibility",
83    "install_lifecycle",
84)
85EXPECTED_ASSURANCE_BINDING = {
86    "assurance_spec_id": "assurance.omega.identity.first.onboarding",
87    "assurance_revision": 2,
88    "admitted_document_sha256": "84dfc9daa7b3f6f92b1895b78854835ce58530c2b81f220f6ca3b62c799f72b1",
89    "admitted_proposal_sha256": "86a0b414be71df319c1fbf705eae5b7c1cc7d8405caa67478b11ab84baa0f6cd",
90    "admission_receipt_ref": "authority.decision.2d660f7602ee99ac6c4a7a2fe906ba0f",
91    "admission_receipt_sha256": "4336ef7443cba679f2aac0176761b8a8e3ed7e2b3e93cefc25c66948e80f2f96",
92    "product_spec_sha256": "fdbf66ee5c9c89a357be14d19aa21197313ab39b2c95bdc330743bda37a91c12",
93}
94EXPECTED_FULL_AUTO_ASSURANCE_BINDING = {
95    "assurance_spec_id": "assurance.omega.full.auto.host",
96    "assurance_revision": 5,
97    "admitted_document_sha256": "a612e2fe875c30b5346e81bc0b897312e5a12690fe0e2e1b56d2fa35ea10b7ee",
98    "admitted_proposal_sha256": "b5b84098e820d0dd146b368f224ef7a10b107bf5b383ea2b6740c6d64b6bfc5f",
99    "admission_receipt_ref": "authority.decision.1954518244492185756509b3cfec6e3e",
100    "admission_receipt_sha256": "9e58770c258833904396f294e0824a1235bbb6181e5843066703be69cfc387a2",
101    "product_spec_sha256": "09f8c2c2c14df6f5272737e26b85dbe3f20704ce66345a9377353710a8d6dddc",
102}
103EXPECTED_FULL_AUTO_OBLIGATIONS = tuple(
104    f"AO-OMEGA-FA-AC-{number:02d}-01" for number in range(1, 9)
105)
106EXPECTED_FULL_AUTO_GATES = (
107    "incident_replay",
108    "owner_real_multi_turn",
109    "restart_reconciliation",
110    "control_matrix",
111    "visible_cross_provider_handoff",
112    "offline_and_sync_gap",
113    "mobile_control_outcomes",
114    "ordinary_chat_separation",
115    "redaction",
116    "independent_exact_candidate_review",
117)
118FULL_AUTO_REFERENCE_KEYS = {"path", "sha256"}
119FULL_AUTO_DECISION_KEYS = {
120    "schema",
121    "decision_ref",
122    "action",
123    "actor",
124    "actor_role",
125    "decided_at",
126    "outcome",
127    "candidate_digest",
128    "artifact_sha256",
129    "release_record_sha256",
130    "omega_commit",
131    "predecessor_decision_ref",
132    "evidence_refs",
133}
134EXPECTED_LIFECYCLE_RECEIPTS = (
135    "before_snapshot",
136    "after_removal_snapshot",
137    "after_reinstall_snapshot",
138    "zed_after_removal_comparison",
139    "zed_after_reinstall_comparison",
140    "omega_after_reinstall_comparison",
141    "app_removal",
142    "app_reinstall",
143)
144EXPECTED_EFFECTD_ARCHIVE_SHA256 = "e3ab92ffe0baa0cb3082ea4bc83720d1677b633f3f0725b7c7571930ae19f7ac"
145EXPECTED_EFFECTD_MANIFEST_SHA256 = "d797f749f57d2ab88cdad74f5ff81e882bc6052951a026c028a84fb7b022b5eb"
146EXPECTED_EFFECTD_SOURCE_COMMIT = "e3280b4795efe189474e12b4c6cd06123cfda359"
147EXPECTED_EFFECTD_SOURCE_TREE = "af28ecd0a57523640a96b6f3ed7411f07a7da228"
148MANUAL_CHECKS = (
149    "clean_profile_side_by_side_zed",
150    "offline_identity_first_start",
151    "editor_open_edit_save_git_terminal",
152    "restart_project_and_layout_restoration",
153    "visible_brand_and_accessibility_labels",
154    "icon_bundle_and_file_association_branding",
155    "shell_installer_and_disk_image_branding",
156    "legal_licenses_and_notices",
157    "network_destinations_against_allowlist",
158    "data_cache_log_browser_update_and_credential_isolation",
159    "disabled_services_and_update_behavior",
160    "uninstall_preserves_zed_data",
161)
162
163
164class ProofError(RuntimeError):
165    pass
166
167
168def sha256_file(path: Path) -> str:
169    digest = hashlib.sha256()
170    with path.open("rb") as handle:
171        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
172            digest.update(chunk)
173    return digest.hexdigest()
174
175
176def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
177    value: dict[str, Any] = {}
178    for key, item in pairs:
179        if key in value:
180            raise ProofError(f"duplicate JSON key: {key}")
181        value[key] = item
182    return value
183
184
185def load_json(path: Path, description: str) -> dict[str, Any]:
186    try:
187        value = json.loads(
188            path.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicate_keys
189        )
190    except (OSError, json.JSONDecodeError, ProofError) as error:
191        raise ProofError(f"cannot read {description} {path}: {error}") from error
192    if not isinstance(value, dict):
193        raise ProofError(f"{description} must contain a JSON object")
194    return value
195
196
197def schema_type_matches(value: Any, expected: str) -> bool:
198    return {
199        "object": isinstance(value, dict),
200        "array": isinstance(value, list),
201        "string": isinstance(value, str),
202        "boolean": isinstance(value, bool),
203        "null": value is None,
204        "integer": isinstance(value, int) and not isinstance(value, bool),
205        "number": isinstance(value, (int, float)) and not isinstance(value, bool),
206    }.get(expected, False)
207
208
209def resolve_schema_ref(root_schema: dict[str, Any], reference: str) -> dict[str, Any]:
210    if not reference.startswith("#/"):
211        raise ProofError(f"identity schema uses unsupported reference: {reference}")
212    node: Any = root_schema
213    for component in reference[2:].split("/"):
214        component = component.replace("~1", "/").replace("~0", "~")
215        if not isinstance(node, dict) or component not in node:
216            raise ProofError(f"identity schema reference is unresolved: {reference}")
217        node = node[component]
218    if not isinstance(node, dict):
219        raise ProofError(f"identity schema reference is not an object: {reference}")
220    return node
221
222
223def validate_json_schema(
224    value: Any,
225    schema: dict[str, Any],
226    root_schema: dict[str, Any],
227    location: str = "$",
228) -> None:
229    if "$ref" in schema:
230        validate_json_schema(
231            value, resolve_schema_ref(root_schema, schema["$ref"]), root_schema, location
232        )
233        return
234    if "const" in schema and value != schema["const"]:
235        raise ProofError(f"{location} differs from the identity evidence schema constant")
236    if "enum" in schema and value not in schema["enum"]:
237        raise ProofError(f"{location} is not an allowed identity evidence schema value")
238    expected_type = schema.get("type")
239    if isinstance(expected_type, str) and not schema_type_matches(value, expected_type):
240        raise ProofError(f"{location} must have JSON type {expected_type}")
241    if isinstance(value, str):
242        if len(value) < schema.get("minLength", 0):
243            raise ProofError(f"{location} is shorter than the identity evidence schema permits")
244        pattern = schema.get("pattern")
245        if isinstance(pattern, str) and re.fullmatch(pattern, value) is None:
246            raise ProofError(f"{location} does not match the identity evidence schema")
247        if schema.get("format") == "date-time":
248            try:
249                parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
250            except ValueError as error:
251                raise ProofError(f"{location} is not an ISO-8601 date-time") from error
252            if parsed.tzinfo is None:
253                raise ProofError(f"{location} date-time must include a timezone")
254    if isinstance(value, list):
255        if len(value) < schema.get("minItems", 0) or len(value) > schema.get(
256            "maxItems", len(value)
257        ):
258            raise ProofError(f"{location} has an invalid item count")
259        item_schema = schema.get("items")
260        if isinstance(item_schema, dict):
261            for index, item in enumerate(value):
262                validate_json_schema(item, item_schema, root_schema, f"{location}[{index}]")
263    if isinstance(value, dict):
264        if len(value) < schema.get("minProperties", 0) or len(value) > schema.get(
265            "maxProperties", len(value)
266        ):
267            raise ProofError(f"{location} has an invalid property count")
268        required = schema.get("required", [])
269        missing = [name for name in required if name not in value]
270        if missing:
271            raise ProofError(f"{location} is missing schema properties: {', '.join(missing)}")
272        properties = schema.get("properties", {})
273        additional = schema.get("additionalProperties", True)
274        for name, item in value.items():
275            if name in properties:
276                validate_json_schema(item, properties[name], root_schema, f"{location}.{name}")
277            elif additional is False:
278                raise ProofError(f"{location} contains unknown schema property: {name}")
279            elif isinstance(additional, dict):
280                validate_json_schema(item, additional, root_schema, f"{location}.{name}")
281        property_names = schema.get("propertyNames")
282        if isinstance(property_names, dict):
283            for name in value:
284                validate_json_schema(name, property_names, root_schema, f"{location}.<name>")
285    if "oneOf" in schema:
286        matches = 0
287        for candidate_schema in schema["oneOf"]:
288            try:
289                validate_json_schema(value, candidate_schema, root_schema, location)
290                matches += 1
291            except ProofError:
292                pass
293        if matches != 1:
294            raise ProofError(f"{location} must match exactly one identity schema branch")
295    for nested_schema in schema.get("allOf", []):
296        condition = nested_schema.get("if")
297        if condition is None:
298            validate_json_schema(value, nested_schema, root_schema, location)
299            continue
300        branch = "then"
301        try:
302            validate_json_schema(value, condition, root_schema, location)
303        except ProofError:
304            branch = "else"
305        selected = nested_schema.get(branch)
306        if isinstance(selected, dict):
307            validate_json_schema(value, selected, root_schema, location)
308
309
310def validate_identity_evidence_schema(evidence: dict[str, Any]) -> None:
311    schema = load_json(IDENTITY_EVIDENCE_SCHEMA, "identity evidence schema")
312    validate_json_schema(evidence, schema, schema)
313
314
315def validate_full_auto_evidence_schema(evidence: dict[str, Any]) -> None:
316    schema = load_json(FULL_AUTO_EVIDENCE_SCHEMA, "Full Auto evidence schema")
317    validate_json_schema(evidence, schema, schema)
318
319
320def run(command: list[str], *, description: str) -> str:
321    result = subprocess.run(command, capture_output=True, text=True, check=False)
322    output = "\n".join(part for part in (result.stdout, result.stderr) if part).strip()
323    if result.returncode != 0:
324        detail = f": {output}" if output else ""
325        raise ProofError(f"{description} failed{detail}")
326    return output
327
328
329def validate_release_record(record: dict[str, Any], record_path: Path, artifact: Path) -> None:
330    required = {
331        "schema_version",
332        "product",
333        "channel",
334        "version",
335        "bundle_version",
336        "artifact_name",
337        "bundle_identifier",
338        "team_id",
339        "signing_identity",
340        "source",
341        "digests",
342        "components",
343        "notarization",
344        "legal",
345        "publication",
346    }
347    missing = sorted(required.difference(record))
348    if missing:
349        raise ProofError(f"release record missing keys: {', '.join(missing)}")
350    expected = {
351        "schema_version": 1,
352        "product": EXPECTED_PRODUCT,
353        "channel": "rc",
354        "version": EXPECTED_VERSION,
355        "artifact_name": artifact.name,
356        "bundle_identifier": EXPECTED_BUNDLE_ID,
357        "team_id": EXPECTED_TEAM_ID,
358        "signing_identity": EXPECTED_SIGNER,
359        "bundle_version": "0.2.0",
360    }
361    for key, expected_value in expected.items():
362        if record.get(key) != expected_value:
363            raise ProofError(
364                f"release record {key}={record.get(key)!r}, expected {expected_value!r}"
365            )
366    if record.get("source", {}).get("dirty") is not False:
367        raise ProofError("release record must bind a clean source commit")
368    if record.get("legal", {}).get("commercial_terms_attached") is not False:
369        raise ProofError("release record must exclude commercial terms")
370    notices = record.get("legal", {}).get("notices")
371    if not isinstance(notices, dict) or set(notices) != {
372        "LICENSE-GPL",
373        "LICENSE-APACHE",
374        "licenses.md",
375    }:
376        raise ProofError("release record legal notice digest inventory is incomplete")
377    packaged_icons = record.get("digests", {}).get("packaged_icon_outputs")
378    if not isinstance(packaged_icons, dict) or set(packaged_icons) != {
379        "Contents/Resources/openagents-icon.icns",
380        "Contents/Resources/Document.icns",
381    }:
382        raise ProofError("release record packaged icon digest inventory is incomplete")
383    for label, bindings in (("legal notice", notices), ("packaged icon", packaged_icons)):
384        if not all(
385            isinstance(value, str)
386            and re.fullmatch(r"[0-9a-f]{64}", value) is not None
387            for value in bindings.values()
388        ):
389            raise ProofError(f"release record {label} digest is invalid")
390    publication = record.get("publication", {})
391    if publication.get("repository") != "OpenAgentsInc/omega":
392        raise ProofError("release publication repository must be OpenAgentsInc/omega")
393    if publication.get("tag") != f"v{EXPECTED_VERSION}":
394        raise ProofError(f"release publication tag must be v{EXPECTED_VERSION}")
395    if publication.get("prerelease") is not True or publication.get("latest") is not False:
396        raise ProofError("release publication must be prerelease=true and latest=false")
397    notarization = record.get("notarization", {})
398    if notarization.get("attempted") is not True or notarization.get("stapled") is not True:
399        raise ProofError("release record notarization must be attempted and stapled")
400    expected_digest = record.get("digests", {}).get("package_sha256")
401    if expected_digest != sha256_file(artifact):
402        raise ProofError("artifact SHA-256 does not match release record")
403    effectd = record.get("components", {}).get("omega_effectd", {})
404    expected_effectd = {
405        "package": "@openagentsinc/omega-effectd",
406        "service_version": "0.1.0",
407        "protocol": "openagents.omega.effectd.v1",
408        "platform": "darwin-arm64",
409        "release_tag": "omega-effectd-v0.1.0-rc.10",
410        "asset_name": "omega-effectd-v0.1.0-macos-arm64.tar.gz",
411        "archive_sha256": EXPECTED_EFFECTD_ARCHIVE_SHA256,
412        "manifest_sha256": EXPECTED_EFFECTD_MANIFEST_SHA256,
413        "source_commit": EXPECTED_EFFECTD_SOURCE_COMMIT,
414        "source_tree": EXPECTED_EFFECTD_SOURCE_TREE,
415    }
416    for key, expected_value in expected_effectd.items():
417        if effectd.get(key) != expected_value:
418            raise ProofError(f"omega-effectd release record {key} differs from the pin")
419    for key in (
420        "wrapper_sha256",
421        "bundle_sha256",
422        "source_node_sha256",
423        "packaged_node_sha256",
424    ):
425        value = effectd.get(key)
426        if (
427            not isinstance(value, str)
428            or len(value) != 64
429            or any(character not in "0123456789abcdef" for character in value)
430        ):
431            raise ProofError(f"omega-effectd release record {key} is not a SHA-256 digest")
432    identity_proof = record.get("components", {}).get("identity_proof_driver", {})
433    expected_identity_proof = {
434        "path": "Contents/MacOS/omega-identity-proof",
435        "protocol": "openagents.omega.identity-proof.v1",
436        "keyring_service": "com.openagents.omega.identity-proof.v1",
437        "keyring_account": "disposable-proof-only",
438    }
439    for key, expected_value in expected_identity_proof.items():
440        if identity_proof.get(key) != expected_value:
441            raise ProofError(f"identity proof driver release record {key} differs from the pin")
442    identity_proof_digest = identity_proof.get("binary_sha256")
443    if not isinstance(identity_proof_digest, str) or re.fullmatch(
444        r"[0-9a-f]{64}", identity_proof_digest
445    ) is None:
446        raise ProofError("identity proof driver release record digest is invalid")
447    if record_path.resolve() == artifact.resolve():
448        raise ProofError("release record and artifact must be distinct files")
449
450
451def validate_identity_evidence_binding(
452    evidence: dict[str, Any], record: dict[str, Any], record_path: Path, artifact: Path
453) -> str:
454    validate_identity_evidence_schema(evidence)
455    digest = evidence.get("candidate_digest", {}).get("value")
456    if (
457        not isinstance(digest, str)
458        or len(digest) != 64
459        or any(character not in "0123456789abcdef" for character in digest)
460    ):
461        raise ProofError("identity evidence candidate digest is invalid")
462    candidate = evidence.get("candidate", {})
463    observed_digest = hashlib.sha256(
464        json.dumps(candidate, sort_keys=True, separators=(",", ":")).encode("utf-8")
465    ).hexdigest()
466    if digest != observed_digest:
467        raise ProofError("identity candidate digest does not match canonical candidate JSON")
468    candidate_artifact = candidate.get("artifact", {})
469    if candidate_artifact.get("name") != artifact.name:
470        raise ProofError("identity evidence names a different artifact")
471    if candidate_artifact.get("sha256") != sha256_file(artifact):
472        raise ProofError("identity evidence artifact digest does not match explicit artifact")
473    if candidate_artifact.get("release_record_sha256") != sha256_file(record_path):
474        raise ProofError("identity evidence release-record digest does not match")
475    if candidate.get("omega", {}).get("commit") != record.get("source", {}).get("commit"):
476        raise ProofError("identity evidence and release record bind different source commits")
477    if candidate.get("assurance_spec_admission") != EXPECTED_ASSURANCE_BINDING:
478        raise ProofError(
479            "identity evidence does not bind the admitted AssuranceSpec proposal and receipt"
480        )
481    return digest
482
483
484def validate_identity_evidence(
485    evidence: dict[str, Any], record: dict[str, Any], record_path: Path, artifact: Path
486) -> tuple[str, str, str]:
487    digest = validate_identity_evidence_binding(evidence, record, record_path, artifact)
488    if evidence.get("status") != "admitted" or evidence.get("candidate_admitted") is not True:
489        raise ProofError("identity candidate is not admitted")
490    if evidence.get("attestation_errors") not in ({}, None):
491        raise ProofError("identity candidate evidence contains attestation errors")
492    gates = evidence.get("gates")
493    if not isinstance(gates, dict) or set(gates) != set(EXPECTED_IDENTITY_GATES):
494        observed = set(gates) if isinstance(gates, dict) else set()
495        missing = sorted(set(EXPECTED_IDENTITY_GATES).difference(observed))
496        unknown = sorted(observed.difference(EXPECTED_IDENTITY_GATES))
497        detail = []
498        if missing:
499            detail.append(f"missing={','.join(missing)}")
500        if unknown:
501            detail.append(f"unknown={','.join(unknown)}")
502        raise ProofError(
503            "identity evidence gate inventory is not exact"
504            + (f": {'; '.join(detail)}" if detail else "")
505        )
506    pending = sorted(
507        name
508        for name, gate in gates.items()
509        if not isinstance(gate, dict) or gate.get("status") != "passed"
510    )
511    if pending:
512        raise ProofError(f"identity evidence has gates not passed: {', '.join(pending)}")
513    for required_gate in ("manual_journey", "owner_observation", "independent_verification"):
514        if gates.get(required_gate, {}).get("status") != "passed":
515            raise ProofError(f"identity evidence gate {required_gate} is not passed")
516    owner = gates["owner_observation"].get("attestor")
517    independent = gates["independent_verification"].get("attestor")
518    if (
519        not isinstance(owner, str)
520        or not owner.strip()
521        or not isinstance(independent, str)
522        or not independent.strip()
523        or owner == independent
524    ):
525        raise ProofError("identity owner and independent verifier must be distinct named attestors")
526    return digest, owner, independent
527
528
529def validate_identity_receipt_references(
530    evidence: dict[str, Any],
531    evidence_root: Path,
532    matrix_digest: str,
533    tripwire_digest: str,
534    observation_digest: str,
535    recovery_digest: str,
536) -> None:
537    if not evidence_root.is_dir() or evidence_root.is_symlink():
538        raise ProofError("identity evidence root is missing or unsafe")
539    expected = {
540        "custody_scenarios": matrix_digest,
541        "forged_request_rejection": matrix_digest,
542        "stale_task_fencing": matrix_digest,
543        "installed_secret_tripwires": tripwire_digest,
544        "manual_journey": observation_digest,
545        "accessibility": observation_digest,
546        "recovery_scenarios": recovery_digest,
547    }
548    for gate_name, expected_digest in expected.items():
549        gate = evidence.get("gates", {}).get(gate_name, {})
550        references = gate.get("evidence")
551        if not isinstance(references, dict) or not references:
552            raise ProofError(f"identity evidence gate {gate_name} has no receipt references")
553        try:
554            resolved = [
555                resolve_evidence_reference(reference, evidence_root)
556                for reference in references.values()
557            ]
558        except IdentityEvidenceError as error:
559            raise ProofError(str(error)) from error
560        if any(digest != expected_digest for _path, digest in resolved):
561            raise ProofError(
562                f"identity evidence gate {gate_name} substitutes another receipt"
563            )
564
565
566def validate_manual_evidence(
567    evidence: dict[str, Any],
568    candidate_digest: str,
569    identity_owner: str,
570    identity_independent: str,
571) -> None:
572    if evidence.get("schema") != "openagents.omega.installed-brand-evidence.v1":
573        raise ProofError("installed brand evidence schema is not supported")
574    if evidence.get("candidate_digest") != candidate_digest:
575        raise ProofError("installed brand evidence binds a different candidate digest")
576    checks = evidence.get("checks")
577    if not isinstance(checks, dict) or set(checks) != set(MANUAL_CHECKS):
578        raise ProofError("installed brand evidence check inventory is incomplete or unexpected")
579    for name in MANUAL_CHECKS:
580        check = checks[name]
581        if (
582            not isinstance(check, dict)
583            or check.get("status") != "passed"
584            or not isinstance(check.get("evidence"), str)
585            or not check["evidence"].strip()
586        ):
587            raise ProofError(f"installed brand evidence check {name} is not passed with evidence")
588    owner = evidence.get("owner_observation")
589    independent = evidence.get("independent_verification")
590    for name, attestation in (
591        ("owner_observation", owner),
592        ("independent_verification", independent),
593    ):
594        if (
595            not isinstance(attestation, dict)
596            or attestation.get("status") != "passed"
597            or not isinstance(attestation.get("attestor"), str)
598            or not attestation["attestor"].strip()
599            or not isinstance(attestation.get("observed_at"), str)
600            or not attestation["observed_at"].strip()
601            or not isinstance(attestation.get("evidence"), str)
602            or not attestation["evidence"].strip()
603        ):
604            raise ProofError(f"installed brand evidence {name} is incomplete")
605    if owner["attestor"] == independent["attestor"]:
606        raise ProofError("installed brand owner and independent verifier must be distinct")
607    if owner["attestor"] != identity_owner:
608        raise ProofError("installed brand owner differs from identity candidate owner")
609    if independent["attestor"] != identity_independent:
610        raise ProofError(
611            "installed brand independent verifier differs from identity candidate verifier"
612        )
613
614
615def validate_network_evidence(path: Path, candidate_digest: str) -> None:
616    evidence = load_json(path, "network capture evidence")
617    if evidence.get("schema") != "openagents.omega.network-capture.v1":
618        raise ProofError("network capture evidence schema is not supported")
619    if evidence.get("candidate_digest") != candidate_digest:
620        raise ProofError("network capture evidence binds a different candidate")
621    if evidence.get("status") != "passed" or evidence.get("blocked_reasons") != []:
622        raise ProofError("network capture evidence is blocked")
623    if evidence.get("journey") != "online_first_codex":
624        raise ProofError("network capture did not exercise online first-Codex setup")
625    allowlist = load_json(ENDPOINT_ALLOWLIST, "endpoint allowlist")
626    binding = evidence.get("allowlist", {})
627    if binding.get("sha256") != sha256_file(ENDPOINT_ALLOWLIST):
628        raise ProofError("network capture evidence binds a different endpoint allowlist")
629    if binding.get("enforcement") != allowlist.get("enforcement"):
630        raise ProofError("network capture enforcement mode differs from the allowlist")
631    processes = evidence.get("processes")
632    if not isinstance(processes, list) or not any(
633        isinstance(process, dict) and process.get("command_basename") == "omega"
634        for process in processes
635    ):
636        raise ProofError("network capture has no Omega process ancestry")
637    destinations = evidence.get("destinations")
638    if not isinstance(destinations, list) or not destinations:
639        raise ProofError("network capture has no observed destinations")
640    approved_hosts = {
641        entry.get("host")
642        for entry in allowlist.get("entries", [])
643        if entry.get("disposition") == "approved"
644    }
645    forbidden_hosts = allowlist.get("normal_start_forbidden_hosts", [])
646    for destination in destinations:
647        host = destination.get("host") if isinstance(destination, dict) else None
648        if not isinstance(host, str) or not host:
649            raise ProofError("network capture contains an invalid destination")
650        try:
651            ipaddress.ip_address(host)
652        except ValueError:
653            pass
654        else:
655            raise ProofError("network capture contains an unresolved IP destination")
656        if any(host == item or host.endswith(f".{item}") for item in forbidden_hosts):
657            raise ProofError(f"network capture reached forbidden host: {host}")
658        if not any(host == item or host.endswith(f".{item}") for item in approved_hosts):
659            raise ProofError(f"network capture reached an unreviewed host: {host}")
660    for field in ("forbidden_hosts_detected", "unreviewed_hosts", "unresolved_destinations"):
661        if evidence.get(field) != []:
662            raise ProofError(f"network capture {field} must be empty")
663
664
665def validate_full_auto_evidence_root(value: Any) -> Path:
666    if not isinstance(value, str):
667        raise ProofError("Full Auto evidence root is missing")
668    root = Path(value)
669    if not root.is_absolute() or root.is_symlink() or not root.is_dir():
670        raise ProofError("Full Auto evidence root must be an absolute non-symlink directory")
671    try:
672        return root.resolve(strict=True)
673    except OSError as error:
674        raise ProofError(f"cannot resolve Full Auto evidence root: {error}") from error
675
676
677def validate_full_auto_reference(
678    reference: Any,
679    label: str,
680    evidence_root: Path,
681    candidate_digest: str,
682) -> tuple[str, Path]:
683    if not isinstance(reference, dict) or set(reference) != FULL_AUTO_REFERENCE_KEYS:
684        raise ProofError(f"{label} is not an exact structured evidence reference")
685    relative = reference.get("path")
686    expected_digest = reference.get("sha256")
687    if (
688        not isinstance(relative, str)
689        or not relative
690        or Path(relative).is_absolute()
691        or ".." in Path(relative).parts
692        or not isinstance(expected_digest, str)
693        or re.fullmatch(r"[0-9a-f]{64}", expected_digest) is None
694    ):
695        raise ProofError(f"{label} is invalid")
696    candidate = evidence_root / relative
697    current = evidence_root
698    for component in Path(relative).parts:
699        current = current / component
700        if current.is_symlink():
701            raise ProofError(f"{label} traverses a symbolic link")
702    try:
703        metadata = candidate.stat()
704    except OSError as error:
705        raise ProofError(f"cannot inspect {label}: {error}") from error
706    if not stat.S_ISREG(metadata.st_mode):
707        raise ProofError(f"{label} must resolve to a regular file")
708    if sha256_file(candidate) != expected_digest:
709        raise ProofError(f"{label} digest differs from the referenced file")
710    if candidate.suffix.lower() == ".json":
711        document = load_json(candidate, label)
712        if document.get("candidate_digest") != candidate_digest:
713            raise ProofError(f"{label} JSON binds a different candidate")
714    return expected_digest, candidate
715
716
717def validate_full_auto_references(
718    references: Any,
719    label: str,
720    evidence_root: Path,
721    candidate_digest: str,
722) -> None:
723    if not isinstance(references, list) or not references:
724        raise ProofError(f"{label} has no evidence references")
725    identities: set[tuple[str, str]] = set()
726    for index, reference in enumerate(references):
727        validate_full_auto_reference(
728            reference, f"{label}[{index}]", evidence_root, candidate_digest
729        )
730        identity = (reference["path"], reference["sha256"])
731        if identity in identities:
732            raise ProofError(f"{label} contains a duplicate evidence reference")
733        identities.add(identity)
734
735
736def validate_full_auto_authority_decision(
737    reference: Any,
738    label: str,
739    expected_action: str,
740    expected_role: str,
741    expected_actor: str,
742    expected_predecessor: str | None,
743    bindings: dict[str, Any],
744    evidence_root: Path,
745) -> dict[str, Any]:
746    _digest, path = validate_full_auto_reference(
747        reference, label, evidence_root, bindings["candidate_digest"]
748    )
749    decision = load_json(path, label)
750    if set(decision) != FULL_AUTO_DECISION_KEYS:
751        raise ProofError(f"{label} keys are not exact")
752    decision_ref = decision.get("decision_ref")
753    if (
754        decision.get("schema") != "openagents.omega.full-auto-authority-decision.v1"
755        or not isinstance(decision_ref, str)
756        or re.fullmatch(r"authority\.decision\.[0-9a-z._-]+", decision_ref) is None
757        or decision.get("action") != expected_action
758        or decision.get("actor_role") != expected_role
759        or decision.get("actor") != expected_actor
760        or decision.get("outcome") != "succeeded"
761        or decision.get("predecessor_decision_ref") != expected_predecessor
762    ):
763        raise ProofError(f"{label} has invalid authority or disposition")
764    try:
765        decided_at = datetime.fromisoformat(decision["decided_at"].replace("Z", "+00:00"))
766    except (KeyError, AttributeError, ValueError) as error:
767        raise ProofError(f"{label} has an invalid decision timestamp") from error
768    if decided_at.tzinfo is None:
769        raise ProofError(f"{label} decision timestamp has no timezone")
770    for key, value in bindings.items():
771        if decision.get(key) != value:
772            raise ProofError(f"{label} binds a different {key}")
773    validate_full_auto_references(
774        decision.get("evidence_refs"),
775        f"{label}.evidence_refs",
776        evidence_root,
777        bindings["candidate_digest"],
778    )
779    return decision
780
781
782def validate_full_auto_evidence(
783    evidence: dict[str, Any],
784    record: dict[str, Any],
785    record_path: Path,
786    artifact: Path,
787    candidate_digest: str,
788    identity_owner: str,
789) -> tuple[str, str]:
790    validate_full_auto_evidence_schema(evidence)
791    if (
792        evidence.get("status") != "verified"
793        or evidence.get("candidate_verified") is not True
794        or evidence.get("candidate_released") is not True
795    ):
796        raise ProofError("Full Auto candidate evidence is not verified and released")
797    evidence_root = validate_full_auto_evidence_root(evidence.get("evidence_root"))
798    candidate = evidence.get("candidate", {})
799    if candidate.get("candidate_digest") != candidate_digest:
800        raise ProofError("Full Auto evidence binds a different candidate digest")
801    omega = candidate.get("omega", {})
802    if (
803        omega.get("commit") != record.get("source", {}).get("commit")
804        or omega.get("dirty") is not False
805    ):
806        raise ProofError("Full Auto evidence binds a different Omega source")
807    candidate_artifact = candidate.get("artifact", {})
808    if (
809        candidate_artifact.get("name") != artifact.name
810        or candidate_artifact.get("sha256") != sha256_file(artifact)
811        or candidate_artifact.get("release_record_sha256") != sha256_file(record_path)
812    ):
813        raise ProofError("Full Auto evidence binds a different artifact or release record")
814    if candidate.get("assurance_spec_admission") != EXPECTED_FULL_AUTO_ASSURANCE_BINDING:
815        raise ProofError("Full Auto evidence does not bind the admitted revision 5 design")
816    effectd = candidate.get("omega_effectd", {})
817    release_effectd = record.get("components", {}).get("omega_effectd", {})
818    expected_effectd_keys = {
819        "release_tag",
820        "archive_sha256",
821        "manifest_sha256",
822        "source_commit",
823        "source_tree",
824        "wrapper_sha256",
825        "bundle_sha256",
826        "source_node_sha256",
827        "packaged_node_sha256",
828    }
829    if set(effectd) != expected_effectd_keys or any(
830        effectd.get(key) != release_effectd.get(key) for key in expected_effectd_keys
831    ):
832        raise ProofError("Full Auto evidence binds a different omega-effectd component")
833    obligations = evidence.get("obligations")
834    if not isinstance(obligations, dict) or set(obligations) != set(
835        EXPECTED_FULL_AUTO_OBLIGATIONS
836    ):
837        raise ProofError("Full Auto evidence obligation inventory is not exact")
838    gates = evidence.get("issue_matrix")
839    if not isinstance(gates, dict) or set(gates) != set(EXPECTED_FULL_AUTO_GATES):
840        raise ProofError("Full Auto evidence issue-matrix inventory is not exact")
841    for name, observation in {**obligations, **gates}.items():
842        if (
843            not isinstance(observation, dict)
844            or observation.get("status") != "passed"
845            or not isinstance(observation.get("evidence_refs"), list)
846            or not observation["evidence_refs"]
847        ):
848            raise ProofError(f"Full Auto evidence {name} is not passed with evidence")
849        if observation.get("evidence_tier") != "installed_candidate":
850            raise ProofError(f"Full Auto evidence {name} is not installed-candidate evidence")
851        validate_full_auto_references(
852            observation["evidence_refs"],
853            f"Full Auto evidence {name}",
854            evidence_root,
855            candidate_digest,
856        )
857    owner = evidence.get("owner_observation", {})
858    independent = evidence.get("independent_verification", {})
859    bindings = {
860        "candidate_digest": candidate_digest,
861        "artifact_sha256": sha256_file(artifact),
862        "release_record_sha256": sha256_file(record_path),
863        "omega_commit": record.get("source", {}).get("commit"),
864    }
865    for label, attestation, role in (
866        ("owner observation", owner, "openagents.owner"),
867        ("independent verification", independent, "openagents.assurance_reviewer"),
868    ):
869        if (
870            not isinstance(attestation, dict)
871            or attestation.get("status") != "passed"
872            or attestation.get("role") != role
873            or not isinstance(attestation.get("actor"), str)
874            or not attestation["actor"].strip()
875            or any(attestation.get(key) != value for key, value in bindings.items())
876        ):
877            raise ProofError(f"Full Auto {label} is incomplete or candidate-mismatched")
878        validate_full_auto_references(
879            attestation.get("evidence_refs"),
880            f"Full Auto {label}",
881            evidence_root,
882            candidate_digest,
883        )
884    if owner["actor"] != identity_owner:
885        raise ProofError("Full Auto owner differs from the admitted identity candidate owner")
886    if owner["actor"] == independent["actor"]:
887        raise ProofError("Full Auto owner and independent verifier must be distinct")
888    decisions = evidence.get("authority_decisions")
889    if not isinstance(decisions, dict) or set(decisions) != {"verification", "release"}:
890        raise ProofError("Full Auto authority decision inventory is not exact")
891    verification = validate_full_auto_authority_decision(
892        decisions["verification"],
893        "Full Auto verification decision",
894        "verify_omega_full_auto_candidate",
895        "openagents.assurance_reviewer",
896        independent["actor"],
897        None,
898        bindings,
899        evidence_root,
900    )
901    if independent.get("authority_decision_ref") != verification["decision_ref"]:
902        raise ProofError("Full Auto reviewer does not bind the verification decision")
903    release = validate_full_auto_authority_decision(
904        decisions["release"],
905        "Full Auto release decision",
906        "release_omega_full_auto_candidate",
907        "openagents.owner",
908        owner["actor"],
909        verification["decision_ref"],
910        bindings,
911        evidence_root,
912    )
913    if owner.get("authority_decision_ref") != release["decision_ref"]:
914        raise ProofError("Full Auto owner does not bind the release decision")
915    verified_at = datetime.fromisoformat(verification["decided_at"].replace("Z", "+00:00"))
916    released_at = datetime.fromisoformat(release["decided_at"].replace("Z", "+00:00"))
917    if released_at <= verified_at:
918        raise ProofError("Full Auto release decision does not follow verification")
919    return owner["actor"], independent["actor"]
920
921
922def load_bound_lifecycle_receipt(
923    manifest_path: Path,
924    reference: Any,
925    label: str,
926    candidate_digest: str,
927) -> tuple[dict[str, Any], str]:
928    if not isinstance(reference, dict) or set(reference) != {"path", "sha256"}:
929        raise ProofError(f"lifecycle receipt reference {label} is not exact")
930    relative = reference.get("path")
931    expected_digest = reference.get("sha256")
932    if (
933        not isinstance(relative, str)
934        or not relative
935        or Path(relative).is_absolute()
936        or ".." in Path(relative).parts
937        or not isinstance(expected_digest, str)
938        or re.fullmatch(r"[0-9a-f]{64}", expected_digest) is None
939    ):
940        raise ProofError(f"lifecycle receipt reference {label} is invalid")
941    receipt_path = manifest_path.parent / relative
942    if receipt_path.is_symlink() or not receipt_path.is_file():
943        raise ProofError(f"lifecycle receipt {label} is missing or unsafe")
944    observed_digest = sha256_file(receipt_path)
945    if observed_digest != expected_digest:
946        raise ProofError(f"lifecycle receipt {label} digest differs from its manifest")
947    receipt = load_json(receipt_path, f"lifecycle receipt {label}")
948    if (
949        receipt.get("schema") != "openagents.omega.rc-lifecycle-proof.v1"
950        or receipt.get("candidate_digest") != candidate_digest
951    ):
952        raise ProofError(f"lifecycle receipt {label} binds a different candidate")
953    return receipt, observed_digest
954
955
956def validate_lifecycle_evidence(
957    path: Path,
958    candidate_digest: str,
959    record_path: Path,
960    artifact: Path,
961) -> None:
962    evidence = load_json(path, "installed lifecycle evidence")
963    if set(evidence) != {
964        "schema",
965        "status",
966        "candidate_digest",
967        "artifact_sha256",
968        "release_record_sha256",
969        "receipts",
970    }:
971        raise ProofError("installed lifecycle evidence keys are not exact")
972    if (
973        evidence.get("schema") != "openagents.omega.installed-lifecycle-evidence.v1"
974        or evidence.get("status") != "passed"
975        or evidence.get("candidate_digest") != candidate_digest
976        or evidence.get("artifact_sha256") != sha256_file(artifact)
977        or evidence.get("release_record_sha256") != sha256_file(record_path)
978    ):
979        raise ProofError("installed lifecycle evidence is incomplete or candidate-mismatched")
980    references = evidence.get("receipts")
981    if not isinstance(references, dict) or set(references) != set(
982        EXPECTED_LIFECYCLE_RECEIPTS
983    ):
984        raise ProofError("installed lifecycle receipt inventory is not exact")
985    loaded = {
986        name: load_bound_lifecycle_receipt(path, reference, name, candidate_digest)
987        for name, reference in references.items()
988    }
989    for name in ("before_snapshot", "after_removal_snapshot", "after_reinstall_snapshot"):
990        if loaded[name][0].get("kind") != "snapshot":
991            raise ProofError(f"lifecycle {name} is not a snapshot")
992    comparisons = {
993        "zed_after_removal_comparison": ("zed", "before_snapshot", "after_removal_snapshot"),
994        "zed_after_reinstall_comparison": ("zed", "before_snapshot", "after_reinstall_snapshot"),
995        "omega_after_reinstall_comparison": (
996            "omega",
997            "before_snapshot",
998            "after_reinstall_snapshot",
999        ),
1000    }
1001    for name, (scope, before_name, after_name) in comparisons.items():
1002        receipt = loaded[name][0]
1003        if (
1004            receipt.get("kind") != "comparison"
1005            or receipt.get("scope") != scope
1006            or receipt.get("status") != "passed"
1007            or receipt.get("changed_roots") != []
1008            or receipt.get("before_snapshot_sha256") != loaded[before_name][1]
1009            or receipt.get("after_snapshot_sha256") != loaded[after_name][1]
1010        ):
1011            raise ProofError(f"lifecycle comparison {name} is not a passing bound comparison")
1012    removal = loaded["app_removal"][0]
1013    if (
1014        removal.get("kind") != "app_removal"
1015        or removal.get("source_path") != str(DEFAULT_APP)
1016        or removal.get("recoverable") is not True
1017        or removal.get("zed_targets_touched") is not False
1018    ):
1019        raise ProofError("lifecycle app removal receipt is not safe and recoverable")
1020    reinstall = loaded["app_reinstall"][0]
1021    if (
1022        reinstall.get("kind") != "app_reinstall"
1023        or reinstall.get("installed_path") != str(DEFAULT_APP)
1024        or reinstall.get("artifact_sha256") != sha256_file(artifact)
1025        or reinstall.get("release_record_sha256") != sha256_file(record_path)
1026        or reinstall.get("zed_targets_touched") is not False
1027    ):
1028        raise ProofError("lifecycle app reinstall receipt binds a different candidate")
1029
1030
1031def bundle_manifest(app: Path) -> dict[str, str]:
1032    if not app.is_dir():
1033        raise ProofError(f"application bundle is missing: {app}")
1034    manifest: dict[str, str] = {}
1035    for path in sorted(app.rglob("*")):
1036        relative = path.relative_to(app).as_posix()
1037        if path.is_symlink():
1038            manifest[relative] = f"symlink:{os.readlink(path)}"
1039        elif path.is_file():
1040            manifest[relative] = f"sha256:{sha256_file(path)}"
1041    if not manifest:
1042        raise ProofError(f"application bundle is empty: {app}")
1043    return manifest
1044
1045
1046def validate_app_metadata(app: Path, record: dict[str, Any]) -> None:
1047    info_path = app / "Contents/Info.plist"
1048    try:
1049        info = plistlib.loads(info_path.read_bytes())
1050    except (OSError, plistlib.InvalidFileException) as error:
1051        raise ProofError(f"cannot read application Info.plist: {error}") from error
1052    if info.get("CFBundleName") != EXPECTED_PRODUCT:
1053        raise ProofError("application CFBundleName is not Omega")
1054    if info.get("CFBundleIdentifier") != EXPECTED_BUNDLE_ID:
1055        raise ProofError("application bundle identifier is not the Omega RC identifier")
1056    version = info.get("CFBundleShortVersionString")
1057    if version != record.get("bundle_version"):
1058        raise ProofError(f"application version {version!r} differs from release record")
1059
1060
1061def validate_binary_digests(app: Path, record: dict[str, Any]) -> None:
1062    digests = record.get("digests", {})
1063    paths = (
1064        ("omega_binary_sha256", app / "Contents/MacOS/omega", digests),
1065        ("cli_binary_sha256", app / "Contents/MacOS/cli", digests),
1066        (
1067            "binary_sha256",
1068            app / "Contents/MacOS/omega-identity-proof",
1069            record.get("components", {}).get("identity_proof_driver", {}),
1070        ),
1071    )
1072    for key, path, bindings in paths:
1073        if not path.is_file():
1074            raise ProofError(f"packaged executable is missing: {path}")
1075        if bindings.get(key) != sha256_file(path):
1076            raise ProofError(f"{path.name} digest differs from release record")
1077
1078
1079def validate_packaged_legal_and_icons(
1080    packaged_app: Path, installed_app: Path, record: dict[str, Any]
1081) -> None:
1082    for relative, expected_digest in record["legal"]["notices"].items():
1083        notice = packaged_app.parent / relative
1084        if not notice.is_file() or sha256_file(notice) != expected_digest:
1085            raise ProofError(f"packaged legal notice differs from release record: {relative}")
1086    for relative, expected_digest in record["digests"]["packaged_icon_outputs"].items():
1087        packaged_icon = packaged_app / relative
1088        installed_icon = installed_app / relative
1089        if not packaged_icon.is_file() or sha256_file(packaged_icon) != expected_digest:
1090            raise ProofError(f"packaged icon differs from release record: {relative}")
1091        if not installed_icon.is_file() or sha256_file(installed_icon) != expected_digest:
1092            raise ProofError(f"installed icon differs from release record: {relative}")
1093
1094
1095def validate_effectd_component(
1096    app: Path, record: dict[str, Any], output_dir: Path, label: str
1097) -> None:
1098    component = app / "Contents/Resources/omega-effectd"
1099    manifest_path = component / "component-manifest.json"
1100    wrapper_path = component / "bin/omega-effectd"
1101    bundle_path = component / "dist/omega-effectd.mjs"
1102    node_path = component / "runtime/bin/node"
1103    for path in (manifest_path, wrapper_path, bundle_path, node_path):
1104        if not path.is_file():
1105            raise ProofError(f"{label} omega-effectd file is missing: {path}")
1106    if not os.access(wrapper_path, os.X_OK) or not os.access(node_path, os.X_OK):
1107        raise ProofError(f"{label} omega-effectd executables are not executable")
1108
1109    manifest = load_json(manifest_path, f"{label} omega-effectd manifest")
1110    expected_manifest = {
1111        "schema": "openagents.omega.effectd.component.v1",
1112        "package": "@openagentsinc/omega-effectd",
1113        "serviceVersion": "0.1.0",
1114        "protocol": "openagents.omega.effectd.v1",
1115        "platform": "darwin-arm64",
1116    }
1117    for key, expected_value in expected_manifest.items():
1118        if manifest.get(key) != expected_value:
1119            raise ProofError(f"{label} omega-effectd manifest {key} differs from the pin")
1120    if manifest.get("source", {}).get("commit") != EXPECTED_EFFECTD_SOURCE_COMMIT:
1121        raise ProofError(f"{label} omega-effectd source commit differs from the pin")
1122    if manifest.get("source", {}).get("tree") != EXPECTED_EFFECTD_SOURCE_TREE:
1123        raise ProofError(f"{label} omega-effectd source tree differs from the pin")
1124
1125    effectd = record["components"]["omega_effectd"]
1126    observed = {
1127        "manifest_sha256": sha256_file(manifest_path),
1128        "wrapper_sha256": sha256_file(wrapper_path),
1129        "bundle_sha256": sha256_file(bundle_path),
1130        "packaged_node_sha256": sha256_file(node_path),
1131    }
1132    for key, digest in observed.items():
1133        if effectd.get(key) != digest:
1134            raise ProofError(f"{label} omega-effectd {key} differs from the release record")
1135    validate_effectd_file_digests(component, manifest, effectd, label)
1136
1137    verify_signature(node_path, output_dir, f"{label}-effectd-node")
1138    validate_effectd_node_entitlements(node_path, output_dir, label)
1139    health_receipt = probe_effectd_initialize_health(wrapper_path)
1140    (output_dir / f"omega-effectd-health-{label}.json").write_text(
1141        json.dumps(health_receipt, indent=2, sort_keys=True) + "\n", encoding="utf-8"
1142    )
1143
1144
1145def validate_effectd_file_digests(
1146    component: Path, manifest: dict[str, Any], effectd: dict[str, Any], label: str
1147) -> None:
1148    source_node_digest = manifest.get("node", {}).get("binarySha256")
1149    if source_node_digest != effectd.get("source_node_sha256"):
1150        raise ProofError(f"{label} omega-effectd source Node digest differs from the pin")
1151    for relative, expected_digest in manifest.get("files", {}).items():
1152        if relative == "runtime/bin/node":
1153            if expected_digest != source_node_digest:
1154                raise ProofError(f"{label} omega-effectd manifest Node digests disagree")
1155            continue
1156        path = component / relative
1157        if not path.is_file() or sha256_file(path) != expected_digest:
1158            raise ProofError(f"{label} omega-effectd manifest file differs: {relative}")
1159
1160
1161def validate_effectd_node_entitlements(
1162    node_path: Path, output_dir: Path, label: str
1163) -> None:
1164    details = run(
1165        ["codesign", "-d", "--entitlements", "-", "--xml", str(node_path)],
1166        description=f"{label} omega-effectd Node entitlement inspection",
1167    )
1168    entitlements = parse_codesign_entitlements(details, label)
1169    expected = {
1170        "com.apple.security.cs.allow-jit": True,
1171        "com.apple.security.cs.allow-unsigned-executable-memory": True,
1172    }
1173    if entitlements != expected:
1174        raise ProofError(f"{label} omega-effectd Node entitlements differ from the minimal set")
1175    (output_dir / f"entitlements-{label}-effectd-node.plist").write_bytes(
1176        plistlib.dumps(entitlements, sort_keys=True)
1177    )
1178
1179
1180def parse_codesign_entitlements(details: str, label: str) -> dict[str, Any]:
1181    xml_start = details.find("<?xml")
1182    xml_end_marker = "</plist>"
1183    xml_end = details.find(xml_end_marker, xml_start)
1184    if xml_start < 0 or xml_end < 0:
1185        raise ProofError(f"{label} omega-effectd Node entitlements are unreadable")
1186    xml_end += len(xml_end_marker)
1187    try:
1188        entitlements = plistlib.loads(details[xml_start:xml_end].encode("utf-8"))
1189    except (plistlib.InvalidFileException, ExpatError) as error:
1190        raise ProofError(f"{label} omega-effectd Node entitlements are invalid") from error
1191    if not isinstance(entitlements, dict):
1192        raise ProofError(f"{label} omega-effectd Node entitlements are not a dictionary")
1193    return entitlements
1194
1195
1196def probe_effectd_initialize_health(program: Path) -> dict[str, Any]:
1197    with tempfile.TemporaryDirectory(prefix="omega-effectd-proof-data-") as data_root:
1198        environment = os.environ.copy()
1199        environment["OPENAGENTS_OMEGA_EFFECTD_DATA_ROOT"] = data_root
1200        process = subprocess.Popen(
1201            [str(program)],
1202            stdin=subprocess.PIPE,
1203            stdout=subprocess.PIPE,
1204            stderr=subprocess.PIPE,
1205            text=True,
1206            env=environment,
1207        )
1208        if process.stdin is None or process.stdout is None:
1209            process.kill()
1210            raise ProofError("omega-effectd proof process did not expose stdio")
1211        selector = selectors.DefaultSelector()
1212        selector.register(process.stdout, selectors.EVENT_READ)
1213
1214        def exchange(identifier: str, method: str, params: dict[str, Any] | None) -> dict[str, Any]:
1215            request = {
1216                "schema": "openagents.omega.effectd.v1",
1217                "kind": "request",
1218                "id": identifier,
1219                "generation": 1,
1220                "method": method,
1221            }
1222            if params is not None:
1223                request["params"] = params
1224            process.stdin.write(json.dumps(request, separators=(",", ":")) + "\n")
1225            process.stdin.flush()
1226            deadline = time.monotonic() + 30
1227            while True:
1228                remaining = deadline - time.monotonic()
1229                if remaining <= 0 or not selector.select(remaining):
1230                    raise ProofError(f"omega-effectd {method} response timed out")
1231                line = process.stdout.readline()
1232                if not line:
1233                    raise ProofError(f"omega-effectd closed stdout during {method}")
1234                response = json.loads(line)
1235                if response.get("kind") == "response" and response.get("id") == identifier:
1236                    break
1237            if (
1238                response.get("schema") != "openagents.omega.effectd.v1"
1239                or response.get("generation") != 1
1240                or response.get("ok") is not True
1241                or not isinstance(response.get("result"), dict)
1242            ):
1243                raise ProofError(f"omega-effectd {method} returned an invalid response")
1244            return response["result"]
1245
1246        try:
1247            initialized = exchange("proof-initialize", "initialize", {"generation": 1})
1248            health = exchange("proof-health", "health", None)
1249        except (OSError, json.JSONDecodeError) as error:
1250            raise ProofError(f"omega-effectd initialize/health probe failed: {error}") from error
1251        finally:
1252            selector.close()
1253            process.terminate()
1254            try:
1255                process.wait(timeout=5)
1256            except subprocess.TimeoutExpired:
1257                process.kill()
1258                process.wait(timeout=5)
1259
1260        if (
1261            initialized.get("schema") != "openagents.omega.effectd.v1"
1262            or initialized.get("protocolVersion") != 1
1263            or initialized.get("serviceVersion") != "0.1.0"
1264            or initialized.get("generation") != 1
1265            or "health" not in initialized.get("capabilities", [])
1266        ):
1267            raise ProofError("omega-effectd initialize result differs from the pinned contract")
1268        if (
1269            health.get("ok") is not True
1270            or health.get("status") != "running"
1271            or health.get("generation") != 1
1272            or not isinstance(health.get("activeRunCount"), int)
1273        ):
1274            raise ProofError("omega-effectd health result is not running")
1275        return {
1276            "schema": "openagents.omega.effectd.proof.v1",
1277            "initialize": {
1278                "protocolVersion": initialized["protocolVersion"],
1279                "serviceVersion": initialized["serviceVersion"],
1280                "generation": initialized["generation"],
1281                "capabilities": initialized["capabilities"],
1282            },
1283            "health": {
1284                "ok": health["ok"],
1285                "status": health["status"],
1286                "generation": health["generation"],
1287                "activeRunCount": health["activeRunCount"],
1288            },
1289        }
1290
1291
1292def verify_signature(app: Path, output_dir: Path, label: str) -> None:
1293    run(
1294        ["codesign", "--verify", "--deep", "--strict", "--verbose=2", str(app)],
1295        description=f"{label} code-signature verification",
1296    )
1297    details = run(
1298        ["codesign", "-dv", "--verbose=4", str(app)],
1299        description=f"{label} code-signature inspection",
1300    )
1301    (output_dir / f"codesign-{label}.txt").write_text(details + "\n", encoding="utf-8")
1302    if f"TeamIdentifier={EXPECTED_TEAM_ID}" not in details:
1303        raise ProofError(f"{label} TeamIdentifier is not {EXPECTED_TEAM_ID}")
1304    if app.suffix == ".app" and f"Identifier={EXPECTED_BUNDLE_ID}" not in details:
1305        raise ProofError(f"{label} Identifier is not {EXPECTED_BUNDLE_ID}")
1306
1307
1308def mounted_application(artifact: Path):
1309    class MountedApplication:
1310        def __init__(self) -> None:
1311            self.temp = tempfile.TemporaryDirectory(prefix="omega-installed-proof-")
1312            self.mount = Path(self.temp.name) / "mount"
1313            self.mount.mkdir()
1314            self.attached = False
1315
1316        def __enter__(self) -> Path:
1317            run(
1318                [
1319                    "hdiutil",
1320                    "attach",
1321                    "-readonly",
1322                    "-nobrowse",
1323                    "-mountpoint",
1324                    str(self.mount),
1325                    str(artifact),
1326                ],
1327                description="read-only DMG mount",
1328            )
1329            self.attached = True
1330            try:
1331                applications = list(self.mount.glob("*.app"))
1332                if len(applications) != 1 or applications[0].name != "Omega.app":
1333                    raise ProofError("DMG must contain exactly one top-level Omega.app")
1334                return applications[0]
1335            except Exception:
1336                subprocess.run(
1337                    ["hdiutil", "detach", str(self.mount)],
1338                    capture_output=True,
1339                    text=True,
1340                    check=False,
1341                )
1342                self.attached = False
1343                self.temp.cleanup()
1344                raise
1345
1346        def __exit__(self, exc_type, exc_value, traceback) -> None:
1347            if self.attached:
1348                result = subprocess.run(
1349                    ["hdiutil", "detach", str(self.mount)],
1350                    capture_output=True,
1351                    text=True,
1352                    check=False,
1353                )
1354                if result.returncode != 0 and exc_type is None:
1355                    raise ProofError("failed to detach proof DMG")
1356            self.temp.cleanup()
1357
1358    return MountedApplication()
1359
1360
1361def pending_checks() -> dict[str, dict[str, str]]:
1362    checks = {
1363        "artifact_release_binding": {"status": "pending"},
1364        "notarization_and_gatekeeper": {"status": "pending"},
1365        "packaged_installed_bundle_match": {"status": "pending"},
1366        "packaged_legal_and_icon_binding": {"status": "pending"},
1367        "packaged_effectd_initialize_health": {"status": "pending"},
1368        "installed_effectd_initialize_health": {"status": "pending"},
1369        "identity_candidate_admission": {"status": "pending"},
1370        "identity_matrix_binding": {"status": "pending"},
1371        "installed_tripwire_binding": {"status": "pending"},
1372        "installed_identity_observations_binding": {"status": "pending"},
1373        "identity_recovery_evidence_binding": {"status": "pending"},
1374        "full_auto_candidate_assurance": {"status": "pending"},
1375        "manual_evidence_binding": {"status": "pending"},
1376        "network_capture_binding": {"status": "pending"},
1377        "install_lifecycle_binding": {"status": "pending"},
1378        "owner_observation": {"status": "pending"},
1379        "independent_verification": {"status": "pending"},
1380    }
1381    checks.update({name: {"status": "pending"} for name in MANUAL_CHECKS})
1382    return checks
1383
1384
1385def write_proof(
1386    output_dir: Path,
1387    status: str,
1388    checks: dict[str, dict[str, str]],
1389    blockers: list[str],
1390    paths: dict[str, Path],
1391    candidate_digest: str | None,
1392) -> None:
1393    output_dir.mkdir(parents=True, exist_ok=True)
1394    input_file_sha256 = {
1395        name: sha256_file(path)
1396        for name, path in paths.items()
1397        if not path.is_symlink()
1398        and path.is_file()
1399        and stat.S_ISREG(path.stat().st_mode)
1400    }
1401    proof = {
1402        "schema": "openagents.omega.installed-proof.v3",
1403        "schema_version": 3,
1404        "status": status,
1405        "product": EXPECTED_PRODUCT,
1406        "version": EXPECTED_VERSION,
1407        "candidate_digest": candidate_digest,
1408        "paths": {name: str(path.resolve()) for name, path in paths.items()},
1409        "input_file_sha256": input_file_sha256,
1410        "checks": checks,
1411        "blockers": blockers,
1412    }
1413    (output_dir / "installed-proof.json").write_text(
1414        json.dumps(proof, indent=2, sort_keys=True) + "\n", encoding="utf-8"
1415    )
1416    notes = [
1417        "# Omega RC installed proof",
1418        "",
1419        f"- status: `{status}`",
1420        f"- candidate digest: `{candidate_digest or 'pending'}`",
1421        "",
1422    ]
1423    if blockers:
1424        notes.extend(["## Blockers", "", *(f"- {blocker}" for blocker in blockers), ""])
1425    notes.extend(["## Gate status", ""])
1426    notes.extend(f"- {name}: `{check['status']}`" for name, check in checks.items())
1427    notes.append("")
1428    (output_dir / "NOTES.md").write_text("\n".join(notes), encoding="utf-8")
1429
1430
1431def prove(arguments: argparse.Namespace) -> int:
1432    arguments.out.mkdir(parents=True, exist_ok=True)
1433    paths = {
1434        "release_record": arguments.release_record,
1435        "artifact": arguments.artifact,
1436        "installed_app": arguments.app,
1437        "identity_evidence": arguments.identity_evidence,
1438        "identity_matrix": arguments.identity_matrix,
1439        "installed_tripwires": arguments.installed_tripwires,
1440        "installed_identity_observations": arguments.installed_identity_observations,
1441        "identity_recovery_evidence": arguments.identity_recovery_evidence,
1442        "full_auto_evidence": arguments.full_auto_evidence,
1443        "manual_evidence": arguments.manual_evidence,
1444        "network_evidence": arguments.network_evidence,
1445        "lifecycle_evidence": arguments.lifecycle_evidence,
1446    }
1447    checks = pending_checks()
1448    blockers: list[str] = []
1449    candidate_digest: str | None = None
1450    for name, path in paths.items():
1451        expected = (
1452            path.is_dir()
1453            if name == "installed_app"
1454            else path.is_file() and not path.is_symlink()
1455        )
1456        if not expected:
1457            blockers.append(f"missing {name.replace('_', ' ')}: {path}")
1458            if name in ("release_record", "artifact"):
1459                checks["artifact_release_binding"] = {"status": "blocked"}
1460            elif name == "installed_app":
1461                checks["packaged_installed_bundle_match"] = {"status": "blocked"}
1462            elif name == "identity_evidence":
1463                checks["identity_candidate_admission"] = {"status": "blocked"}
1464            elif name == "identity_matrix":
1465                checks["identity_matrix_binding"] = {"status": "blocked"}
1466            elif name == "installed_tripwires":
1467                checks["installed_tripwire_binding"] = {"status": "blocked"}
1468            elif name == "installed_identity_observations":
1469                checks["installed_identity_observations_binding"] = {"status": "blocked"}
1470            elif name == "identity_recovery_evidence":
1471                checks["identity_recovery_evidence_binding"] = {"status": "blocked"}
1472            elif name == "full_auto_evidence":
1473                checks["full_auto_candidate_assurance"] = {"status": "blocked"}
1474            elif name == "manual_evidence":
1475                checks["manual_evidence_binding"] = {"status": "blocked"}
1476            elif name == "network_evidence":
1477                checks["network_capture_binding"] = {"status": "blocked"}
1478            elif name == "lifecycle_evidence":
1479                checks["install_lifecycle_binding"] = {"status": "blocked"}
1480    core_paths = ("release_record", "artifact", "installed_app")
1481    if any(
1482        not (
1483            paths[name].is_dir()
1484            if name == "installed_app"
1485            else paths[name].is_file()
1486        )
1487        for name in core_paths
1488    ):
1489        write_proof(arguments.out, "incomplete", checks, blockers, paths, candidate_digest)
1490        return 1
1491
1492    try:
1493        record = load_json(arguments.release_record, "release record")
1494        validate_release_record(record, arguments.release_record, arguments.artifact)
1495        checks["artifact_release_binding"] = {"status": "passed"}
1496    except ProofError as error:
1497        blockers.append(str(error))
1498        checks["artifact_release_binding"] = {"status": "blocked", "reason": str(error)}
1499        write_proof(arguments.out, "incomplete", checks, blockers, paths, candidate_digest)
1500        return 1
1501
1502    try:
1503        verify_signature(arguments.artifact, arguments.out, "dmg")
1504        run(
1505            ["xcrun", "stapler", "validate", str(arguments.artifact)],
1506            description="DMG staple validation",
1507        )
1508        verify_signature(arguments.app, arguments.out, "installed-app")
1509        gatekeeper = run(
1510            ["spctl", "--assess", "--type", "execute", "--verbose=4", str(arguments.app)],
1511            description="installed app Gatekeeper assessment",
1512        )
1513        (arguments.out / "spctl-installed-app.txt").write_text(gatekeeper + "\n", encoding="utf-8")
1514        checks["notarization_and_gatekeeper"] = {"status": "passed"}
1515    except ProofError as error:
1516        blockers.append(str(error))
1517        checks["notarization_and_gatekeeper"] = {"status": "blocked", "reason": str(error)}
1518
1519    try:
1520        with mounted_application(arguments.artifact) as packaged_app:
1521            verify_signature(packaged_app, arguments.out, "packaged-app")
1522            validate_app_metadata(packaged_app, record)
1523            validate_binary_digests(packaged_app, record)
1524            validate_app_metadata(arguments.app, record)
1525            validate_binary_digests(arguments.app, record)
1526            verify_signature(
1527                packaged_app / "Contents/MacOS/omega-identity-proof",
1528                arguments.out,
1529                "packaged-identity-proof-driver",
1530            )
1531            verify_signature(
1532                arguments.app / "Contents/MacOS/omega-identity-proof",
1533                arguments.out,
1534                "installed-identity-proof-driver",
1535            )
1536            if bundle_manifest(packaged_app) != bundle_manifest(arguments.app):
1537                raise ProofError("installed Omega.app differs from the signed DMG application tree")
1538        checks["packaged_installed_bundle_match"] = {"status": "passed"}
1539    except ProofError as error:
1540        blockers.append(str(error))
1541        checks["packaged_installed_bundle_match"] = {"status": "blocked", "reason": str(error)}
1542
1543    try:
1544        with mounted_application(arguments.artifact) as packaged_app:
1545            validate_packaged_legal_and_icons(packaged_app, arguments.app, record)
1546        checks["packaged_legal_and_icon_binding"] = {"status": "passed"}
1547    except ProofError as error:
1548        blockers.append(str(error))
1549        checks["packaged_legal_and_icon_binding"] = {
1550            "status": "blocked",
1551            "reason": str(error),
1552        }
1553
1554    try:
1555        with mounted_application(arguments.artifact) as packaged_app:
1556            validate_effectd_component(packaged_app, record, arguments.out, "packaged-app")
1557        checks["packaged_effectd_initialize_health"] = {"status": "passed"}
1558    except ProofError as error:
1559        blockers.append(str(error))
1560        checks["packaged_effectd_initialize_health"] = {
1561            "status": "blocked",
1562            "reason": str(error),
1563        }
1564
1565    try:
1566        validate_effectd_component(arguments.app, record, arguments.out, "installed-app")
1567        checks["installed_effectd_initialize_health"] = {"status": "passed"}
1568    except ProofError as error:
1569        blockers.append(str(error))
1570        checks["installed_effectd_initialize_health"] = {
1571            "status": "blocked",
1572            "reason": str(error),
1573        }
1574
1575    identity_owner: str | None = None
1576    identity_independent: str | None = None
1577    identity_evidence: dict[str, Any] | None = None
1578    if arguments.identity_evidence.is_file():
1579        try:
1580            evidence = load_json(arguments.identity_evidence, "identity candidate evidence")
1581            identity_evidence = evidence
1582            candidate_digest = validate_identity_evidence_binding(
1583                evidence, record, arguments.release_record, arguments.artifact
1584            )
1585            _, identity_owner, identity_independent = validate_identity_evidence(
1586                evidence, record, arguments.release_record, arguments.artifact
1587            )
1588            checks["identity_candidate_admission"] = {"status": "passed"}
1589        except ProofError as error:
1590            blockers.append(str(error))
1591            checks["identity_candidate_admission"] = {"status": "blocked", "reason": str(error)}
1592
1593    if candidate_digest and arguments.identity_matrix.is_file():
1594        try:
1595            digest = validate_identity_matrix(arguments.identity_matrix, candidate_digest)
1596            checks["identity_matrix_binding"] = {
1597                "status": "passed",
1598                "sha256": digest,
1599            }
1600        except IdentityEvidenceError as error:
1601            blockers.append(str(error))
1602            checks["identity_matrix_binding"] = {
1603                "status": "blocked",
1604                "reason": str(error),
1605            }
1606
1607    if candidate_digest and arguments.installed_tripwires.is_file():
1608        try:
1609            digest = validate_installed_tripwires(
1610                arguments.installed_tripwires, candidate_digest
1611            )
1612            checks["installed_tripwire_binding"] = {
1613                "status": "passed",
1614                "sha256": digest,
1615            }
1616        except IdentityEvidenceError as error:
1617            blockers.append(str(error))
1618            checks["installed_tripwire_binding"] = {
1619                "status": "blocked",
1620                "reason": str(error),
1621            }
1622
1623    if candidate_digest and arguments.installed_identity_observations.is_file():
1624        try:
1625            digest = validate_installed_observations(
1626                arguments.installed_identity_observations,
1627                candidate_digest,
1628                arguments.evidence_root,
1629            )
1630            checks["installed_identity_observations_binding"] = {
1631                "status": "passed",
1632                "sha256": digest,
1633            }
1634        except IdentityEvidenceError as error:
1635            blockers.append(str(error))
1636            checks["installed_identity_observations_binding"] = {
1637                "status": "blocked",
1638                "reason": str(error),
1639            }
1640
1641    if (
1642        candidate_digest
1643        and identity_evidence is not None
1644        and arguments.identity_recovery_evidence.is_file()
1645    ):
1646        candidate = identity_evidence.get("candidate", {})
1647        artifact = candidate.get("artifact", {})
1648        packaged_app = artifact.get("packaged_app", {})
1649        try:
1650            digest = validate_recovery_evidence(
1651                arguments.identity_recovery_evidence,
1652                candidate_digest,
1653                artifact.get("sha256"),
1654                packaged_app.get("omega_binary_sha256"),
1655                arguments.evidence_root,
1656            )
1657            checks["identity_recovery_evidence_binding"] = {
1658                "status": "passed",
1659                "sha256": digest,
1660            }
1661        except IdentityEvidenceError as error:
1662            blockers.append(str(error))
1663            checks["identity_recovery_evidence_binding"] = {
1664                "status": "blocked",
1665                "reason": str(error),
1666            }
1667
1668    if (
1669        identity_evidence is not None
1670        and checks["identity_matrix_binding"].get("status") == "passed"
1671        and checks["installed_tripwire_binding"].get("status") == "passed"
1672        and checks["installed_identity_observations_binding"].get("status") == "passed"
1673        and checks["identity_recovery_evidence_binding"].get("status") == "passed"
1674    ):
1675        try:
1676            validate_identity_receipt_references(
1677                identity_evidence,
1678                arguments.evidence_root,
1679                checks["identity_matrix_binding"]["sha256"],
1680                checks["installed_tripwire_binding"]["sha256"],
1681                checks["installed_identity_observations_binding"]["sha256"],
1682                checks["identity_recovery_evidence_binding"]["sha256"],
1683            )
1684        except ProofError as error:
1685            blockers.append(str(error))
1686            checks["identity_candidate_admission"] = {
1687                "status": "blocked",
1688                "reason": str(error),
1689            }
1690
1691    if (
1692        arguments.manual_evidence.is_file()
1693        and identity_owner
1694        and identity_independent
1695        and candidate_digest
1696    ):
1697        try:
1698            manual = load_json(arguments.manual_evidence, "installed brand evidence")
1699            validate_manual_evidence(
1700                manual, candidate_digest, identity_owner, identity_independent
1701            )
1702            checks["manual_evidence_binding"] = {"status": "passed"}
1703            for name in MANUAL_CHECKS:
1704                checks[name] = {"status": "passed", "evidence": manual["checks"][name]["evidence"]}
1705            checks["owner_observation"] = {
1706                "status": "passed",
1707                "attestor": manual["owner_observation"]["attestor"],
1708                "evidence": manual["owner_observation"]["evidence"],
1709            }
1710            checks["independent_verification"] = {
1711                "status": "passed",
1712                "attestor": manual["independent_verification"]["attestor"],
1713                "evidence": manual["independent_verification"]["evidence"],
1714            }
1715        except ProofError as error:
1716            blockers.append(str(error))
1717            checks["manual_evidence_binding"] = {"status": "blocked", "reason": str(error)}
1718
1719    if arguments.network_evidence.is_file() and candidate_digest:
1720        try:
1721            validate_network_evidence(arguments.network_evidence, candidate_digest)
1722            checks["network_capture_binding"] = {"status": "passed"}
1723        except ProofError as error:
1724            blockers.append(str(error))
1725            checks["network_capture_binding"] = {
1726                "status": "blocked",
1727                "reason": str(error),
1728            }
1729
1730    if arguments.lifecycle_evidence.is_file() and candidate_digest:
1731        try:
1732            validate_lifecycle_evidence(
1733                arguments.lifecycle_evidence,
1734                candidate_digest,
1735                arguments.release_record,
1736                arguments.artifact,
1737            )
1738            checks["install_lifecycle_binding"] = {"status": "passed"}
1739        except ProofError as error:
1740            blockers.append(str(error))
1741            checks["install_lifecycle_binding"] = {
1742                "status": "blocked",
1743                "reason": str(error),
1744            }
1745
1746    if arguments.full_auto_evidence.is_file() and candidate_digest and identity_owner:
1747        try:
1748            full_auto = load_json(arguments.full_auto_evidence, "Full Auto candidate evidence")
1749            full_auto_owner, full_auto_independent = validate_full_auto_evidence(
1750                full_auto,
1751                record,
1752                arguments.release_record,
1753                arguments.artifact,
1754                candidate_digest,
1755                identity_owner,
1756            )
1757            checks["full_auto_candidate_assurance"] = {
1758                "status": "passed",
1759                "owner": full_auto_owner,
1760                "independent": full_auto_independent,
1761            }
1762        except ProofError as error:
1763            blockers.append(str(error))
1764            checks["full_auto_candidate_assurance"] = {
1765                "status": "blocked",
1766                "reason": str(error),
1767            }
1768
1769    status = "complete" if not blockers and all(
1770        check.get("status") == "passed" for check in checks.values()
1771    ) else "incomplete"
1772    if status != "complete" and not blockers:
1773        blockers.append("one or more required gates remain pending")
1774    write_proof(arguments.out, status, checks, blockers, paths, candidate_digest)
1775    return 0 if status == "complete" else 1
1776
1777
1778def self_test() -> None:
1779    with tempfile.TemporaryDirectory(prefix="omega-proof-self-test-") as directory:
1780        root = Path(directory)
1781        artifact = root / "Omega-v0.2.0-rc2-macos-arm64.dmg"
1782        artifact.write_bytes(b"signed candidate bytes")
1783        record_path = root / "release.json"
1784        record = {
1785            "schema_version": 1,
1786            "product": EXPECTED_PRODUCT,
1787            "channel": "rc",
1788            "version": EXPECTED_VERSION,
1789            "bundle_version": "0.2.0",
1790            "artifact_name": artifact.name,
1791            "bundle_identifier": EXPECTED_BUNDLE_ID,
1792            "team_id": EXPECTED_TEAM_ID,
1793            "signing_identity": EXPECTED_SIGNER,
1794            "source": {"commit": "a" * 40, "dirty": False},
1795            "digests": {
1796                "package_sha256": sha256_file(artifact),
1797                "packaged_icon_outputs": {
1798                    "Contents/Resources/openagents-icon.icns": "a" * 64,
1799                    "Contents/Resources/Document.icns": "b" * 64,
1800                },
1801            },
1802            "components": {
1803                "identity_proof_driver": {
1804                    "path": "Contents/MacOS/omega-identity-proof",
1805                    "protocol": "openagents.omega.identity-proof.v1",
1806                    "keyring_service": "com.openagents.omega.identity-proof.v1",
1807                    "keyring_account": "disposable-proof-only",
1808                    "binary_sha256": "2" * 64,
1809                },
1810                "omega_effectd": {
1811                    "package": "@openagentsinc/omega-effectd",
1812                    "service_version": "0.1.0",
1813                    "protocol": "openagents.omega.effectd.v1",
1814                    "platform": "darwin-arm64",
1815                    "release_tag": "omega-effectd-v0.1.0-rc.10",
1816                    "asset_name": "omega-effectd-v0.1.0-macos-arm64.tar.gz",
1817                    "archive_sha256": EXPECTED_EFFECTD_ARCHIVE_SHA256,
1818                    "source_commit": EXPECTED_EFFECTD_SOURCE_COMMIT,
1819                    "source_tree": EXPECTED_EFFECTD_SOURCE_TREE,
1820                    "manifest_sha256": EXPECTED_EFFECTD_MANIFEST_SHA256,
1821                    "wrapper_sha256": "d" * 64,
1822                    "bundle_sha256": "e" * 64,
1823                    "source_node_sha256": "f" * 64,
1824                    "packaged_node_sha256": "1" * 64,
1825                }
1826            },
1827            "notarization": {"attempted": True, "stapled": True},
1828            "legal": {
1829                "commercial_terms_attached": False,
1830                "notices": {
1831                    "LICENSE-GPL": "c" * 64,
1832                    "LICENSE-APACHE": "d" * 64,
1833                    "licenses.md": "e" * 64,
1834                },
1835            },
1836            "publication": {
1837                "repository": "OpenAgentsInc/omega",
1838                "tag": f"v{EXPECTED_VERSION}",
1839                "prerelease": True,
1840                "latest": False,
1841            },
1842        }
1843        record_path.write_text(json.dumps(record), encoding="utf-8")
1844        validate_release_record(record, record_path, artifact)
1845
1846        candidate = {
1847            "omega": {
1848                "repository": "https://github.com/OpenAgentsInc/omega",
1849                "commit": "a" * 40,
1850                "dirty": False,
1851            },
1852            "identity_contract": {
1853                "version": 1,
1854                "source_path": "crates/omega_identity/src/contract.rs",
1855                "source_sha256": "3" * 64,
1856            },
1857            "public_vector": {
1858                "version": 1,
1859                "path": "crates/omega_identity/fixtures/omega_nostr_identity_v1.json",
1860                "sha256": "4" * 64,
1861            },
1862            "reviewed_source": {
1863                "repository": "https://github.com/OpenAgentsInc/buzz",
1864                "commit": "b" * 40,
1865            },
1866            "assurance_spec_admission": EXPECTED_ASSURANCE_BINDING,
1867            "packages": [
1868                {"name": name, "version": "1.0.0", "cargo_checksum": "5" * 64}
1869                for name in ("atomic-write-file", "keyring", "nostr")
1870            ],
1871            "build_cargo_lock": {
1872                "name": "Cargo.lock.omega-v0.2.0-rc2",
1873                "sha256": "6" * 64,
1874            },
1875            "artifact": {
1876                "name": artifact.name,
1877                "version": EXPECTED_VERSION,
1878                "sha256": sha256_file(artifact),
1879                "release_record_sha256": sha256_file(record_path),
1880                "packaged_app": {
1881                    "application_name": "Omega",
1882                    "bundle_identifier": EXPECTED_BUNDLE_ID,
1883                    "bundle_version": "0.2.0",
1884                    "bundle_build_version": "20260724",
1885                    "omega_binary_sha256": "7" * 64,
1886                    "cli_binary_sha256": "8" * 64,
1887                    "identity_proof_binary_sha256": "2" * 64,
1888                    "source_commit_marker": "a" * 40,
1889                    "signing": {
1890                        name: {
1891                            "verified": True,
1892                            "authority": EXPECTED_SIGNER,
1893                            "team_id": EXPECTED_TEAM_ID,
1894                        }
1895                        for name in ("dmg", "application", "identity_proof_driver")
1896                    },
1897                    "fresh_boundary_scan": {
1898                        "status": "passed",
1899                        "forbidden_literals": ["BUZZ_PRIVATE_KEY", "identity.key", "get_nsec"],
1900                        "matches": {},
1901                        "scope": "self-test",
1902                    },
1903                },
1904            },
1905            "automated_tripwires": {
1906                "verifier_path": "script/verify-omega-identity",
1907                "verifier_script_sha256": "9" * 64,
1908                "source_verification": "passed",
1909                "packaged_boundary_scan": "passed",
1910                "packaged_forbidden_literals": ["BUZZ_PRIVATE_KEY", "identity.key", "get_nsec"],
1911                "fresh_mounted_app_scan": {
1912                    "status": "passed",
1913                    "forbidden_literals": ["BUZZ_PRIVATE_KEY", "identity.key", "get_nsec"],
1914                    "matches": {},
1915                    "scope": "self-test",
1916                },
1917                "scope": "self-test",
1918            },
1919        }
1920        candidate_digest = hashlib.sha256(
1921            json.dumps(candidate, sort_keys=True, separators=(",", ":")).encode("utf-8")
1922        ).hexdigest()
1923        identity = {
1924            "schema": "openagents.omega.identity-candidate-evidence.v1",
1925            "schema_version": 1,
1926            "status": "admitted",
1927            "candidate_admitted": True,
1928            "candidate": candidate,
1929            "candidate_digest": {
1930                "algorithm": "sha256",
1931                "canonicalization": "self-test canonical JSON",
1932                "value": candidate_digest,
1933            },
1934            "gates": {
1935                name: {
1936                    "status": "passed",
1937                    "attestor": "Reviewer" if name == "independent_verification" else "Owner",
1938                    "observed_at": "2026-07-24T12:00:00Z",
1939                    "evidence": {f"self-test evidence for {name}": "sha256:" + "2" * 64},
1940                    "required_evidence": [f"self-test evidence for {name}"],
1941                }
1942                for name in EXPECTED_IDENTITY_GATES
1943            },
1944            "attestation_errors": {},
1945        }
1946        observed = validate_identity_evidence(identity, record, record_path, artifact)
1947        if observed != (candidate_digest, "Owner", "Reviewer"):
1948            raise ProofError("self-test identity binding returned unexpected values")
1949
1950        full_auto_root = root / "full-auto-evidence"
1951        full_auto_root.mkdir()
1952        observation_path = full_auto_root / "observation.json"
1953        observation_path.write_text(
1954            json.dumps(
1955                {
1956                    "schema": "openagents.omega.self-test-observation.v1",
1957                    "candidate_digest": candidate_digest,
1958                }
1959            ),
1960            encoding="utf-8",
1961        )
1962        full_auto_ref = {
1963            "path": observation_path.name,
1964            "sha256": sha256_file(observation_path),
1965        }
1966        full_auto_observation = {
1967            "status": "passed",
1968            "evidence_tier": "installed_candidate",
1969            "observed_at": "2026-07-24T12:02:00Z",
1970            "evidence_refs": [full_auto_ref],
1971        }
1972        full_auto_binding = {
1973            "candidate_digest": candidate_digest,
1974            "artifact_sha256": sha256_file(artifact),
1975            "release_record_sha256": sha256_file(record_path),
1976            "omega_commit": "a" * 40,
1977        }
1978        full_auto_attestation = {
1979            "status": "passed",
1980            "actor": "Owner",
1981            "role": "openagents.owner",
1982            "observed_at": "2026-07-24T12:03:00Z",
1983            "evidence_refs": [full_auto_ref],
1984            "authority_decision_ref": "authority.decision.owner-release-self-test",
1985            **full_auto_binding,
1986        }
1987        effectd_binding_keys = (
1988            "release_tag",
1989            "archive_sha256",
1990            "manifest_sha256",
1991            "source_commit",
1992            "source_tree",
1993            "wrapper_sha256",
1994            "bundle_sha256",
1995            "source_node_sha256",
1996            "packaged_node_sha256",
1997        )
1998        def write_authority_decision(
1999            name: str,
2000            decision_ref: str,
2001            action: str,
2002            actor: str,
2003            role: str,
2004            decided_at: str,
2005            predecessor: str | None,
2006        ) -> dict[str, str]:
2007            path = full_auto_root / name
2008            path.write_text(
2009                json.dumps(
2010                    {
2011                        "schema": "openagents.omega.full-auto-authority-decision.v1",
2012                        "decision_ref": decision_ref,
2013                        "action": action,
2014                        "actor": actor,
2015                        "actor_role": role,
2016                        "decided_at": decided_at,
2017                        "outcome": "succeeded",
2018                        **full_auto_binding,
2019                        "predecessor_decision_ref": predecessor,
2020                        "evidence_refs": [full_auto_ref],
2021                    }
2022                ),
2023                encoding="utf-8",
2024            )
2025            return {"path": name, "sha256": sha256_file(path)}
2026
2027        verification_decision = write_authority_decision(
2028            "verification.json",
2029            "authority.decision.verifier-self-test",
2030            "verify_omega_full_auto_candidate",
2031            "Euler",
2032            "openagents.assurance_reviewer",
2033            "2026-07-24T12:04:00Z",
2034            None,
2035        )
2036        release_decision = write_authority_decision(
2037            "release.json",
2038            "authority.decision.owner-release-self-test",
2039            "release_omega_full_auto_candidate",
2040            "Owner",
2041            "openagents.owner",
2042            "2026-07-24T12:05:00Z",
2043            "authority.decision.verifier-self-test",
2044        )
2045        full_auto = {
2046            "schema": "openagents.omega.full-auto-candidate-evidence.v1",
2047            "schema_version": 1,
2048            "status": "verified",
2049            "candidate_verified": True,
2050            "candidate_released": True,
2051            "evidence_root": str(full_auto_root),
2052            "candidate": {
2053                "candidate_digest": candidate_digest,
2054                "omega": {
2055                    "repository": "https://github.com/OpenAgentsInc/omega",
2056                    "commit": "a" * 40,
2057                    "dirty": False,
2058                },
2059                "artifact": {
2060                    "name": artifact.name,
2061                    "sha256": sha256_file(artifact),
2062                    "release_record_sha256": sha256_file(record_path),
2063                },
2064                "omega_effectd": {
2065                    key: record["components"]["omega_effectd"][key]
2066                    for key in effectd_binding_keys
2067                },
2068                "assurance_spec_admission": EXPECTED_FULL_AUTO_ASSURANCE_BINDING,
2069            },
2070            "obligations": {
2071                key: dict(full_auto_observation)
2072                for key in EXPECTED_FULL_AUTO_OBLIGATIONS
2073            },
2074            "issue_matrix": {
2075                key: dict(full_auto_observation) for key in EXPECTED_FULL_AUTO_GATES
2076            },
2077            "owner_observation": full_auto_attestation,
2078            "independent_verification": {
2079                **full_auto_attestation,
2080                "actor": "Euler",
2081                "role": "openagents.assurance_reviewer",
2082                "authority_decision_ref": "authority.decision.verifier-self-test",
2083            },
2084            "authority_decisions": {
2085                "verification": verification_decision,
2086                "release": release_decision,
2087            },
2088        }
2089        if validate_full_auto_evidence(
2090            full_auto, record, record_path, artifact, candidate_digest, "Owner"
2091        ) != ("Owner", "Euler"):
2092            raise ProofError("self-test Full Auto binding returned unexpected actors")
2093        mismatched_full_auto = json.loads(json.dumps(full_auto))
2094        mismatched_full_auto["independent_verification"]["candidate_digest"] = "0" * 64
2095        try:
2096            validate_full_auto_evidence(
2097                mismatched_full_auto,
2098                record,
2099                record_path,
2100                artifact,
2101                candidate_digest,
2102                "Owner",
2103            )
2104        except ProofError:
2105            pass
2106        else:
2107            raise ProofError("self-test accepted mismatched Full Auto assurance")
2108        duplicate_full_auto = json.loads(json.dumps(full_auto))
2109        duplicate_full_auto["issue_matrix"]["incident_replay"]["evidence_refs"] *= 2
2110        try:
2111            validate_full_auto_evidence(
2112                duplicate_full_auto,
2113                record,
2114                record_path,
2115                artifact,
2116                candidate_digest,
2117                "Owner",
2118            )
2119        except ProofError:
2120            pass
2121        else:
2122            raise ProofError("self-test accepted duplicate Full Auto evidence references")
2123        source_incident = json.loads(json.dumps(full_auto))
2124        source_incident["issue_matrix"]["incident_replay"]["evidence_tier"] = "source"
2125        try:
2126            validate_full_auto_evidence(
2127                source_incident,
2128                record,
2129                record_path,
2130                artifact,
2131                candidate_digest,
2132                "Owner",
2133            )
2134        except ProofError:
2135            pass
2136        else:
2137            raise ProofError("self-test accepted source-only incident replay")
2138        write_proof(
2139            root / "input-digest-proof",
2140            "incomplete",
2141            pending_checks(),
2142            ["self-test"],
2143            {"full_auto_evidence": observation_path},
2144            candidate_digest,
2145        )
2146        digest_proof = load_json(
2147            root / "input-digest-proof/installed-proof.json", "input digest proof"
2148        )
2149        if digest_proof.get("input_file_sha256", {}).get(
2150            "full_auto_evidence"
2151        ) != sha256_file(observation_path):
2152            raise ProofError("self-test proof did not bind its input file digest")
2153        identity["candidate_admitted"] = False
2154        try:
2155            validate_identity_evidence(identity, record, record_path, artifact)
2156        except ProofError:
2157            pass
2158        else:
2159            raise ProofError("self-test accepted a non-admitted identity candidate")
2160        identity["candidate_admitted"] = True
2161
2162        truncated = json.loads(json.dumps(identity))
2163        truncated["gates"].pop("custody_scenarios")
2164        try:
2165            validate_identity_evidence(truncated, record, record_path, artifact)
2166        except ProofError:
2167            pass
2168        else:
2169            raise ProofError("self-test accepted truncated identity gate evidence")
2170
2171        unknown = json.loads(json.dumps(identity))
2172        unknown["gates"]["unknown_gate"] = unknown["gates"]["manual_journey"]
2173        try:
2174            validate_identity_evidence(unknown, record, record_path, artifact)
2175        except ProofError:
2176            pass
2177        else:
2178            raise ProofError("self-test accepted an unknown identity gate")
2179
2180        forged = json.loads(json.dumps(identity))
2181        forged["candidate"]["assurance_spec_admission"]["admitted_proposal_sha256"] = "0" * 64
2182        forged["candidate_digest"]["value"] = hashlib.sha256(
2183            json.dumps(
2184                forged["candidate"], sort_keys=True, separators=(",", ":")
2185            ).encode("utf-8")
2186        ).hexdigest()
2187        try:
2188            validate_identity_evidence(forged, record, record_path, artifact)
2189        except ProofError:
2190            pass
2191        else:
2192            raise ProofError("self-test accepted a forged AssuranceSpec binding")
2193
2194        duplicate_path = root / "duplicate.json"
2195        duplicate_path.write_text('{"schema":1,"schema":1}', encoding="utf-8")
2196        try:
2197            load_json(duplicate_path, "duplicate self-test")
2198        except ProofError:
2199            pass
2200        else:
2201            raise ProofError("self-test accepted a duplicate JSON object key")
2202
2203        network_path = root / "network.json"
2204        network = {
2205            "schema": "openagents.omega.network-capture.v1",
2206            "candidate_digest": candidate_digest,
2207            "allowlist": {
2208                "sha256": sha256_file(ENDPOINT_ALLOWLIST),
2209                "enforcement": "release-proof-evidence-only",
2210            },
2211            "journey": "online_first_codex",
2212            "status": "passed",
2213            "blocked_reasons": [],
2214            "processes": [{"pid": 1, "ppid": 0, "command_basename": "omega"}],
2215            "destinations": [
2216                {
2217                    "pid": 1,
2218                    "host": "registry.npmjs.org",
2219                    "port": 443,
2220                    "protocol": "tcp",
2221                }
2222            ],
2223            "forbidden_hosts_detected": [],
2224            "unreviewed_hosts": [],
2225            "unresolved_destinations": [],
2226        }
2227        network_path.write_text(json.dumps(network), encoding="utf-8")
2228        validate_network_evidence(network_path, candidate_digest)
2229        network["destinations"][0]["host"] = "api.zed.dev"
2230        network_path.write_text(json.dumps(network), encoding="utf-8")
2231        try:
2232            validate_network_evidence(network_path, candidate_digest)
2233        except ProofError:
2234            pass
2235        else:
2236            raise ProofError("self-test accepted a forbidden network destination")
2237
2238        lifecycle_root = root / "lifecycle"
2239        lifecycle_root.mkdir()
2240        snapshots: dict[str, Path] = {}
2241        for name in (
2242            "before_snapshot",
2243            "after_removal_snapshot",
2244            "after_reinstall_snapshot",
2245        ):
2246            snapshot_path = lifecycle_root / f"{name}.json"
2247            snapshot_path.write_text(
2248                json.dumps(
2249                    {
2250                        "schema": "openagents.omega.rc-lifecycle-proof.v1",
2251                        "kind": "snapshot",
2252                        "candidate_digest": candidate_digest,
2253                    }
2254                ),
2255                encoding="utf-8",
2256            )
2257            snapshots[name] = snapshot_path
2258        lifecycle_values = {
2259            "zed_after_removal_comparison": {
2260                "kind": "comparison",
2261                "scope": "zed",
2262                "status": "passed",
2263                "changed_roots": [],
2264                "before_snapshot_sha256": sha256_file(snapshots["before_snapshot"]),
2265                "after_snapshot_sha256": sha256_file(snapshots["after_removal_snapshot"]),
2266            },
2267            "zed_after_reinstall_comparison": {
2268                "kind": "comparison",
2269                "scope": "zed",
2270                "status": "passed",
2271                "changed_roots": [],
2272                "before_snapshot_sha256": sha256_file(snapshots["before_snapshot"]),
2273                "after_snapshot_sha256": sha256_file(snapshots["after_reinstall_snapshot"]),
2274            },
2275            "omega_after_reinstall_comparison": {
2276                "kind": "comparison",
2277                "scope": "omega",
2278                "status": "passed",
2279                "changed_roots": [],
2280                "before_snapshot_sha256": sha256_file(snapshots["before_snapshot"]),
2281                "after_snapshot_sha256": sha256_file(snapshots["after_reinstall_snapshot"]),
2282            },
2283            "app_removal": {
2284                "kind": "app_removal",
2285                "source_path": str(DEFAULT_APP),
2286                "recoverable": True,
2287                "zed_targets_touched": False,
2288            },
2289            "app_reinstall": {
2290                "kind": "app_reinstall",
2291                "installed_path": str(DEFAULT_APP),
2292                "artifact_sha256": sha256_file(artifact),
2293                "release_record_sha256": sha256_file(record_path),
2294                "zed_targets_touched": False,
2295            },
2296        }
2297        lifecycle_paths = dict(snapshots)
2298        for name, value in lifecycle_values.items():
2299            receipt_path = lifecycle_root / f"{name}.json"
2300            receipt_path.write_text(
2301                json.dumps(
2302                    {
2303                        "schema": "openagents.omega.rc-lifecycle-proof.v1",
2304                        "candidate_digest": candidate_digest,
2305                        **value,
2306                    }
2307                ),
2308                encoding="utf-8",
2309            )
2310            lifecycle_paths[name] = receipt_path
2311        lifecycle_manifest = lifecycle_root / "manifest.json"
2312        lifecycle = {
2313            "schema": "openagents.omega.installed-lifecycle-evidence.v1",
2314            "status": "passed",
2315            "candidate_digest": candidate_digest,
2316            "artifact_sha256": sha256_file(artifact),
2317            "release_record_sha256": sha256_file(record_path),
2318            "receipts": {
2319                name: {
2320                    "path": receipt_path.name,
2321                    "sha256": sha256_file(receipt_path),
2322                }
2323                for name, receipt_path in lifecycle_paths.items()
2324            },
2325        }
2326        lifecycle_manifest.write_text(json.dumps(lifecycle), encoding="utf-8")
2327        validate_lifecycle_evidence(
2328            lifecycle_manifest, candidate_digest, record_path, artifact
2329        )
2330        lifecycle["receipts"]["app_reinstall"]["sha256"] = "0" * 64
2331        lifecycle_manifest.write_text(json.dumps(lifecycle), encoding="utf-8")
2332        try:
2333            validate_lifecycle_evidence(
2334                lifecycle_manifest, candidate_digest, record_path, artifact
2335            )
2336        except ProofError:
2337            pass
2338        else:
2339            raise ProofError("self-test accepted a forged lifecycle receipt")
2340
2341        tampered = dict(record)
2342        tampered["digests"] = {"package_sha256": "0" * 64}
2343        try:
2344            validate_release_record(tampered, record_path, artifact)
2345        except ProofError:
2346            pass
2347        else:
2348            raise ProofError("self-test accepted a tampered artifact digest")
2349
2350        stale_candidate = json.loads(json.dumps(record))
2351        stale_candidate["version"] = "0.2.0-rc1"
2352        stale_candidate["publication"]["tag"] = "v0.2.0-rc1"
2353        try:
2354            validate_release_record(stale_candidate, record_path, artifact)
2355        except ProofError:
2356            pass
2357        else:
2358            raise ProofError("self-test accepted the withdrawn rc1 candidate")
2359
2360        packaged = root / "packaged.app"
2361        installed = root / "installed.app"
2362        for app in (packaged, installed):
2363            (app / "Contents/Resources").mkdir(parents=True)
2364            (app / "Contents/Resources/brand.txt").write_text("Omega", encoding="utf-8")
2365        if bundle_manifest(packaged) != bundle_manifest(installed):
2366            raise ProofError("self-test equal bundles did not compare equal")
2367        (installed / "Contents/Resources/brand.txt").write_text("Zed", encoding="utf-8")
2368        if bundle_manifest(packaged) == bundle_manifest(installed):
2369            raise ProofError("self-test accepted a tampered installed resource")
2370
2371        component = root / "effectd"
2372        (component / "bin").mkdir(parents=True)
2373        (component / "runtime/bin").mkdir(parents=True)
2374        wrapper = component / "bin/omega-effectd"
2375        signed_node = component / "runtime/bin/node"
2376        wrapper.write_bytes(b"verified wrapper")
2377        signed_node.write_bytes(b"source node plus a Developer ID signature")
2378        source_node_digest = hashlib.sha256(b"pristine source node").hexdigest()
2379        manifest = {
2380            "node": {"binarySha256": source_node_digest},
2381            "files": {
2382                "bin/omega-effectd": sha256_file(wrapper),
2383                "runtime/bin/node": source_node_digest,
2384            },
2385        }
2386        effectd = {
2387            "source_node_sha256": source_node_digest,
2388            "packaged_node_sha256": sha256_file(signed_node),
2389        }
2390        validate_effectd_file_digests(component, manifest, effectd, "self-test")
2391        wrapper.write_bytes(b"tampered wrapper")
2392        try:
2393            validate_effectd_file_digests(component, manifest, effectd, "self-test")
2394        except ProofError:
2395            pass
2396        else:
2397            raise ProofError("self-test accepted a tampered non-Node component file")
2398        wrapper.write_bytes(b"verified wrapper")
2399        mismatched_effectd = dict(effectd)
2400        mismatched_effectd["source_node_sha256"] = "0" * 64
2401        try:
2402            validate_effectd_file_digests(
2403                component, manifest, mismatched_effectd, "self-test"
2404            )
2405        except ProofError:
2406            pass
2407        else:
2408            raise ProofError("self-test accepted a mismatched source Node pin")
2409
2410        entitlement_fixture = plistlib.dumps(
2411            {
2412                "com.apple.security.cs.allow-jit": True,
2413                "com.apple.security.cs.allow-unsigned-executable-memory": True,
2414            }
2415        ).decode("utf-8")
2416        parsed_entitlements = parse_codesign_entitlements(
2417            f"{entitlement_fixture}\nExecutable=/fixture/node", "self-test"
2418        )
2419        if parsed_entitlements.get("com.apple.security.cs.allow-jit") is not True:
2420            raise ProofError("self-test failed to isolate codesign entitlement XML")
2421
2422        manual = {
2423            "schema": "openagents.omega.installed-brand-evidence.v1",
2424            "candidate_digest": "b" * 64,
2425            "checks": {
2426                name: {"status": "passed", "evidence": f"receipt:{name}"}
2427                for name in MANUAL_CHECKS
2428            },
2429            "owner_observation": {
2430                "status": "passed",
2431                "attestor": "Owner",
2432                "observed_at": "2026-07-24T12:00:00Z",
2433                "evidence": "receipt:owner",
2434            },
2435            "independent_verification": {
2436                "status": "passed",
2437                "attestor": "Reviewer",
2438                "observed_at": "2026-07-24T12:01:00Z",
2439                "evidence": "receipt:reviewer",
2440            },
2441        }
2442        validate_manual_evidence(manual, "b" * 64, "Owner", "Reviewer")
2443        manual["checks"][MANUAL_CHECKS[0]]["status"] = "pending"
2444        try:
2445            validate_manual_evidence(manual, "b" * 64, "Owner", "Reviewer")
2446        except ProofError:
2447            pass
2448        else:
2449            raise ProofError("self-test accepted pending manual evidence")
2450
2451        checks = pending_checks()
2452        checks["artifact_release_binding"] = {"status": "passed"}
2453        if all(check.get("status") == "passed" for check in checks.values()):
2454            raise ProofError("self-test allowed automated checks to imply complete")
2455
2456    print("Omega installed-proof harness self-test passed")
2457
2458
2459def parse_arguments() -> argparse.Namespace:
2460    parser = argparse.ArgumentParser()
2461    parser.add_argument("--harness-check", action="store_true")
2462    parser.add_argument("--self-test", action="store_true")
2463    parser.add_argument("--release-record", type=Path)
2464    parser.add_argument("--artifact", type=Path)
2465    parser.add_argument("--app", type=Path, default=DEFAULT_APP)
2466    parser.add_argument("--identity-evidence", type=Path)
2467    parser.add_argument("--identity-matrix", type=Path)
2468    parser.add_argument("--installed-tripwires", type=Path)
2469    parser.add_argument("--installed-identity-observations", type=Path)
2470    parser.add_argument("--identity-recovery-evidence", type=Path)
2471    parser.add_argument("--evidence-root", type=Path)
2472    parser.add_argument("--full-auto-evidence", type=Path)
2473    parser.add_argument("--manual-evidence", type=Path)
2474    parser.add_argument("--network-evidence", type=Path)
2475    parser.add_argument("--lifecycle-evidence", type=Path)
2476    parser.add_argument("--out", type=Path, default=DEFAULT_OUTPUT)
2477    arguments = parser.parse_args()
2478    if arguments.self_test or arguments.harness_check:
2479        return arguments
2480    missing = [
2481        option
2482        for option in (
2483            "release_record",
2484            "artifact",
2485            "identity_evidence",
2486            "identity_matrix",
2487            "installed_tripwires",
2488            "installed_identity_observations",
2489            "identity_recovery_evidence",
2490            "evidence_root",
2491            "full_auto_evidence",
2492            "manual_evidence",
2493            "network_evidence",
2494            "lifecycle_evidence",
2495        )
2496        if getattr(arguments, option) is None
2497    ]
2498    if missing:
2499        options = ", ".join(f"--{name.replace('_', '-')}" for name in missing)
2500        parser.error(f"prove mode requires {options}")
2501    return arguments
2502
2503
2504def main() -> int:
2505    arguments = parse_arguments()
2506    try:
2507        if arguments.self_test:
2508            self_test()
2509            return 0
2510        if arguments.harness_check:
2511            self_test()
2512            if not (ROOT / "docs/src/development/omega-rc-installed-proof.md").is_file():
2513                raise ProofError("installed-proof documentation is missing")
2514            if "omega-rc-installed-proof.md" not in (ROOT / "docs/src/SUMMARY.md").read_text():
2515                raise ProofError("documentation summary link is missing")
2516            checks = pending_checks()
2517            write_proof(arguments.out, "harness_ok", checks, [], {}, None)
2518            print(f"Harness OK: {arguments.out / 'installed-proof.json'}")
2519            return 0
2520        return prove(arguments)
2521    except ProofError as error:
2522        print(f"error: {error}", file=sys.stderr)
2523        return 1
2524
2525
2526if __name__ == "__main__":
2527    sys.exit(main())
2528