tenant.openagents/omega

1#!/usr/bin/env python3
2"""Generate candidate-bound Omega Full Auto assurance evidence without inventing observations."""
3
4from __future__ import annotations
5
6import argparse
7import hashlib
8import json
9import re
10import stat
11import sys
12import tempfile
13from datetime import datetime
14from pathlib import Path
15from typing import Any
16
17
18ROOT = Path(__file__).resolve().parent.parent
19SCHEMA_PATH = (
20    ROOT
21    / "crates/omega_identity/fixtures/full_auto_candidate_evidence_schema_v1.json"
22)
23SCHEMA = "openagents.omega.full-auto-candidate-evidence.v1"
24OBSERVATION_SCHEMA = "openagents.omega.full-auto-observations.v1"
25ASSURANCE_BINDING = {
26    "assurance_spec_id": "assurance.omega.full.auto.host",
27    "assurance_revision": 5,
28    "admitted_document_sha256": "a612e2fe875c30b5346e81bc0b897312e5a12690fe0e2e1b56d2fa35ea10b7ee",
29    "admitted_proposal_sha256": "b5b84098e820d0dd146b368f224ef7a10b107bf5b383ea2b6740c6d64b6bfc5f",
30    "admission_receipt_ref": "authority.decision.1954518244492185756509b3cfec6e3e",
31    "admission_receipt_sha256": "9e58770c258833904396f294e0824a1235bbb6181e5843066703be69cfc387a2",
32    "product_spec_sha256": "09f8c2c2c14df6f5272737e26b85dbe3f20704ce66345a9377353710a8d6dddc",
33}
34OBLIGATIONS = tuple(f"AO-OMEGA-FA-AC-{number:02d}-01" for number in range(1, 9))
35ISSUE_GATES = (
36    "incident_replay",
37    "owner_real_multi_turn",
38    "restart_reconciliation",
39    "control_matrix",
40    "visible_cross_provider_handoff",
41    "offline_and_sync_gap",
42    "mobile_control_outcomes",
43    "ordinary_chat_separation",
44    "redaction",
45    "independent_exact_candidate_review",
46)
47INSTALLED_ONLY_GATES = {
48    "incident_replay",
49    "owner_real_multi_turn",
50    "restart_reconciliation",
51    "control_matrix",
52    "visible_cross_provider_handoff",
53    "offline_and_sync_gap",
54    "mobile_control_outcomes",
55    "ordinary_chat_separation",
56    "redaction",
57    "independent_exact_candidate_review",
58}
59ATTESTATION_KEYS = {
60    "status",
61    "actor",
62    "role",
63    "observed_at",
64    "evidence_refs",
65    "candidate_digest",
66    "artifact_sha256",
67    "release_record_sha256",
68    "omega_commit",
69    "authority_decision_ref",
70}
71OBSERVATION_KEYS = {
72    "status",
73    "evidence_tier",
74    "observed_at",
75    "evidence_refs",
76}
77REFERENCE_KEYS = {"path", "sha256"}
78DECISION_KEYS = {
79    "schema",
80    "decision_ref",
81    "action",
82    "actor",
83    "actor_role",
84    "decided_at",
85    "outcome",
86    "candidate_digest",
87    "artifact_sha256",
88    "release_record_sha256",
89    "omega_commit",
90    "predecessor_decision_ref",
91    "evidence_refs",
92}
93
94
95class EvidenceError(RuntimeError):
96    pass
97
98
99def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
100    result: dict[str, Any] = {}
101    for key, value in pairs:
102        if key in result:
103            raise EvidenceError(f"duplicate JSON key: {key}")
104        result[key] = value
105    return result
106
107
108def load_json(path: Path, description: str) -> dict[str, Any]:
109    try:
110        value = json.loads(
111            path.read_text(encoding="utf-8"), object_pairs_hook=reject_duplicate_keys
112        )
113    except (OSError, json.JSONDecodeError, EvidenceError) as error:
114        raise EvidenceError(f"cannot read {description} {path}: {error}") from error
115    if not isinstance(value, dict):
116        raise EvidenceError(f"{description} must be a JSON object")
117    return value
118
119
120def sha256_file(path: Path) -> str:
121    digest = hashlib.sha256()
122    with path.open("rb") as handle:
123        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
124            digest.update(chunk)
125    return digest.hexdigest()
126
127
128def require_exact_keys(value: dict[str, Any], expected: set[str], label: str) -> None:
129    observed = set(value)
130    if observed != expected:
131        missing = sorted(expected - observed)
132        unknown = sorted(observed - expected)
133        details = []
134        if missing:
135            details.append(f"missing={','.join(missing)}")
136        if unknown:
137            details.append(f"unknown={','.join(unknown)}")
138        raise EvidenceError(f"{label} keys are not exact: {'; '.join(details)}")
139
140
141def require_digest(value: Any, label: str) -> str:
142    if not isinstance(value, str) or re.fullmatch(r"[0-9a-f]{64}", value) is None:
143        raise EvidenceError(f"{label} must be a lowercase SHA-256 digest")
144    return value
145
146
147def require_time(value: Any, label: str) -> str:
148    if not isinstance(value, str):
149        raise EvidenceError(f"{label} must be an ISO-8601 timestamp")
150    try:
151        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
152    except ValueError as error:
153        raise EvidenceError(f"{label} must be an ISO-8601 timestamp") from error
154    if parsed.tzinfo is None:
155        raise EvidenceError(f"{label} must include a timezone")
156    return value
157
158
159def validate_evidence_root(path: Path) -> Path:
160    if not path.is_absolute() or path.is_symlink() or not path.is_dir():
161        raise EvidenceError("evidence root must be an absolute, non-symlink directory")
162    try:
163        return path.resolve(strict=True)
164    except OSError as error:
165        raise EvidenceError(f"cannot resolve evidence root {path}: {error}") from error
166
167
168def resolve_evidence_ref(
169    reference: Any,
170    label: str,
171    evidence_root: Path,
172    candidate_digest: str,
173) -> tuple[dict[str, str], Path]:
174    if not isinstance(reference, dict):
175        raise EvidenceError(f"{label} must be a structured evidence reference")
176    require_exact_keys(reference, REFERENCE_KEYS, label)
177    relative = reference.get("path")
178    expected_digest = reference.get("sha256")
179    if (
180        not isinstance(relative, str)
181        or not relative
182        or Path(relative).is_absolute()
183        or ".." in Path(relative).parts
184    ):
185        raise EvidenceError(f"{label}.path must be a safe relative path")
186    require_digest(expected_digest, f"{label}.sha256")
187    candidate = evidence_root / relative
188    current = evidence_root
189    for component in Path(relative).parts:
190        current = current / component
191        if current.is_symlink():
192            raise EvidenceError(f"{label} traverses a symbolic link")
193    try:
194        metadata = candidate.stat()
195    except OSError as error:
196        raise EvidenceError(f"cannot inspect {label}: {error}") from error
197    if not stat.S_ISREG(metadata.st_mode):
198        raise EvidenceError(f"{label} must resolve to a regular file")
199    if sha256_file(candidate) != expected_digest:
200        raise EvidenceError(f"{label} digest differs from the referenced file")
201    if candidate.suffix.lower() == ".json":
202        document = load_json(candidate, label)
203        if document.get("candidate_digest") != candidate_digest:
204            raise EvidenceError(f"{label} JSON binds a different candidate")
205    return {"path": relative, "sha256": expected_digest}, candidate
206
207
208def require_refs(
209    value: Any,
210    label: str,
211    evidence_root: Path,
212    candidate_digest: str,
213) -> list[dict[str, str]]:
214    if not isinstance(value, list) or not value:
215        raise EvidenceError(f"{label} must contain evidence references")
216    validated: list[dict[str, str]] = []
217    identities: set[tuple[str, str]] = set()
218    for index, reference in enumerate(value):
219        normalized, _path = resolve_evidence_ref(
220            reference, f"{label}[{index}]", evidence_root, candidate_digest
221        )
222        identity = (normalized["path"], normalized["sha256"])
223        if identity in identities:
224            raise EvidenceError(f"{label} contains a duplicate evidence reference")
225        identities.add(identity)
226        validated.append(normalized)
227    return validated
228
229
230def candidate_bindings(
231    release_record: dict[str, Any], release_path: Path, artifact: Path, identity: dict[str, Any]
232) -> dict[str, Any]:
233    if not artifact.is_file():
234        raise EvidenceError(f"candidate artifact is missing: {artifact}")
235    artifact_digest = sha256_file(artifact)
236    if release_record.get("digests", {}).get("package_sha256") != artifact_digest:
237        raise EvidenceError("artifact digest differs from the release record")
238    source = release_record.get("source")
239    if not isinstance(source, dict) or source.get("dirty") is not False:
240        raise EvidenceError("release record does not bind a clean Omega source")
241    omega_commit = source.get("commit")
242    if not isinstance(omega_commit, str) or re.fullmatch(r"[0-9a-f]{40}", omega_commit) is None:
243        raise EvidenceError("release record Omega commit is invalid")
244    identity_candidate = identity.get("candidate")
245    candidate_digest = identity.get("candidate_digest", {}).get("value")
246    if not isinstance(identity_candidate, dict):
247        raise EvidenceError("identity evidence has no candidate object")
248    require_digest(candidate_digest, "identity candidate digest")
249    observed_candidate_digest = hashlib.sha256(
250        json.dumps(identity_candidate, sort_keys=True, separators=(",", ":")).encode()
251    ).hexdigest()
252    if candidate_digest != observed_candidate_digest:
253        raise EvidenceError("identity candidate digest does not match canonical candidate JSON")
254    identity_artifact = identity_candidate.get("artifact", {})
255    if (
256        identity_candidate.get("omega", {}).get("commit") != omega_commit
257        or identity_artifact.get("name") != artifact.name
258        or identity_artifact.get("sha256") != artifact_digest
259        or identity_artifact.get("release_record_sha256") != sha256_file(release_path)
260    ):
261        raise EvidenceError("identity evidence binds a different Omega candidate")
262    effectd = release_record.get("components", {}).get("omega_effectd")
263    if not isinstance(effectd, dict):
264        raise EvidenceError("release record has no omega-effectd component binding")
265    effectd_keys = (
266        "release_tag",
267        "archive_sha256",
268        "manifest_sha256",
269        "source_commit",
270        "source_tree",
271        "wrapper_sha256",
272        "bundle_sha256",
273        "source_node_sha256",
274        "packaged_node_sha256",
275    )
276    for key in effectd_keys[1:]:
277        length = 40 if key in ("source_commit", "source_tree") else 64
278        value = effectd.get(key)
279        if not isinstance(value, str) or re.fullmatch(rf"[0-9a-f]{{{length}}}", value) is None:
280            raise EvidenceError(f"omega-effectd {key} binding is invalid")
281    if not isinstance(effectd.get("release_tag"), str) or not effectd["release_tag"]:
282        raise EvidenceError("omega-effectd release tag binding is invalid")
283    return {
284        "candidate_digest": candidate_digest,
285        "artifact_name": artifact.name,
286        "artifact_sha256": artifact_digest,
287        "release_record_sha256": sha256_file(release_path),
288        "omega_commit": omega_commit,
289        "omega_effectd": {key: effectd[key] for key in effectd_keys},
290    }
291
292
293def validate_observation(
294    observation: Any,
295    label: str,
296    *,
297    installed_required: bool,
298    evidence_root: Path,
299    candidate_digest: str,
300) -> dict[str, Any]:
301    if not isinstance(observation, dict):
302        raise EvidenceError(f"{label} must be an object")
303    require_exact_keys(observation, OBSERVATION_KEYS, label)
304    if observation["status"] != "passed":
305        raise EvidenceError(f"{label} is not passed")
306    allowed_tiers = {"source", "installed_candidate"}
307    if observation["evidence_tier"] not in allowed_tiers:
308        raise EvidenceError(f"{label} has an unsupported evidence tier")
309    if installed_required and observation["evidence_tier"] != "installed_candidate":
310        raise EvidenceError(f"{label} requires installed-candidate evidence")
311    require_time(observation["observed_at"], f"{label}.observed_at")
312    return {
313        **observation,
314        "evidence_refs": require_refs(
315            observation["evidence_refs"],
316            f"{label}.evidence_refs",
317            evidence_root,
318            candidate_digest,
319        ),
320    }
321
322
323def validate_attestation(
324    value: Any,
325    label: str,
326    expected_role: str,
327    bindings: dict[str, Any],
328    evidence_root: Path,
329) -> dict[str, Any]:
330    if not isinstance(value, dict):
331        raise EvidenceError(f"{label} must be an object")
332    require_exact_keys(value, ATTESTATION_KEYS, label)
333    if value["status"] != "passed" or value["role"] != expected_role:
334        raise EvidenceError(f"{label} has the wrong status or authority role")
335    if not isinstance(value["actor"], str) or not value["actor"].strip():
336        raise EvidenceError(f"{label} requires a named actor")
337    require_time(value["observed_at"], f"{label}.observed_at")
338    evidence_refs = require_refs(
339        value["evidence_refs"],
340        f"{label}.evidence_refs",
341        evidence_root,
342        bindings["candidate_digest"],
343    )
344    for key in (
345        "candidate_digest",
346        "artifact_sha256",
347        "release_record_sha256",
348        "omega_commit",
349    ):
350        if value[key] != bindings[key]:
351            raise EvidenceError(f"{label} binds a different {key}")
352    return {**value, "evidence_refs": evidence_refs}
353
354
355def validate_authority_decision(
356    reference: Any,
357    label: str,
358    expected_action: str,
359    expected_role: str,
360    expected_actor: str,
361    expected_predecessor: str | None,
362    bindings: dict[str, Any],
363    evidence_root: Path,
364) -> tuple[dict[str, str], dict[str, Any]]:
365    normalized, decision_path = resolve_evidence_ref(
366        reference, label, evidence_root, bindings["candidate_digest"]
367    )
368    decision = load_json(decision_path, label)
369    require_exact_keys(decision, DECISION_KEYS, label)
370    if decision.get("schema") != "openagents.omega.full-auto-authority-decision.v1":
371        raise EvidenceError(f"{label} schema is not supported")
372    decision_ref = decision.get("decision_ref")
373    if (
374        not isinstance(decision_ref, str)
375        or re.fullmatch(r"authority\.decision\.[0-9a-z._-]+", decision_ref) is None
376    ):
377        raise EvidenceError(f"{label} decision_ref is invalid")
378    if (
379        decision.get("action") != expected_action
380        or decision.get("actor_role") != expected_role
381        or decision.get("actor") != expected_actor
382        or decision.get("outcome") != "succeeded"
383        or decision.get("predecessor_decision_ref") != expected_predecessor
384    ):
385        raise EvidenceError(f"{label} has the wrong action, actor, role, outcome, or predecessor")
386    require_time(decision.get("decided_at"), f"{label}.decided_at")
387    for key in (
388        "candidate_digest",
389        "artifact_sha256",
390        "release_record_sha256",
391        "omega_commit",
392    ):
393        if decision.get(key) != bindings[key]:
394            raise EvidenceError(f"{label} binds a different {key}")
395    decision = {
396        **decision,
397        "evidence_refs": require_refs(
398            decision.get("evidence_refs"),
399            f"{label}.evidence_refs",
400            evidence_root,
401            bindings["candidate_digest"],
402        ),
403    }
404    return normalized, decision
405
406
407def build_evidence(
408    observations: dict[str, Any], bindings: dict[str, Any], evidence_root: Path
409) -> dict[str, Any]:
410    evidence_root = validate_evidence_root(evidence_root)
411    require_exact_keys(
412        observations,
413        {
414            "schema",
415            "candidate_digest",
416            "artifact_sha256",
417            "release_record_sha256",
418            "omega_commit",
419            "obligations",
420            "issue_matrix",
421            "owner_observation",
422            "independent_verification",
423            "verification_decision",
424            "release_decision",
425        },
426        "observation packet",
427    )
428    if observations["schema"] != OBSERVATION_SCHEMA:
429        raise EvidenceError("observation packet schema is not supported")
430    for key in (
431        "candidate_digest",
432        "artifact_sha256",
433        "release_record_sha256",
434        "omega_commit",
435    ):
436        if observations[key] != bindings[key]:
437            raise EvidenceError(f"observation packet binds a different {key}")
438
439    obligations = observations["obligations"]
440    issue_matrix = observations["issue_matrix"]
441    if not isinstance(obligations, dict) or set(obligations) != set(OBLIGATIONS):
442        raise EvidenceError("observation obligation inventory is not exact")
443    if not isinstance(issue_matrix, dict) or set(issue_matrix) != set(ISSUE_GATES):
444        raise EvidenceError("observation issue-matrix inventory is not exact")
445    validated_obligations = {
446        key: validate_observation(
447            value,
448            f"obligations.{key}",
449            installed_required=True,
450            evidence_root=evidence_root,
451            candidate_digest=bindings["candidate_digest"],
452        )
453        for key, value in obligations.items()
454    }
455    validated_matrix = {
456        key: validate_observation(
457            value,
458            f"issue_matrix.{key}",
459            installed_required=key in INSTALLED_ONLY_GATES,
460            evidence_root=evidence_root,
461            candidate_digest=bindings["candidate_digest"],
462        )
463        for key, value in issue_matrix.items()
464    }
465    owner = validate_attestation(
466        observations["owner_observation"],
467        "owner_observation",
468        "openagents.owner",
469        bindings,
470        evidence_root,
471    )
472    independent = validate_attestation(
473        observations["independent_verification"],
474        "independent_verification",
475        "openagents.assurance_reviewer",
476        bindings,
477        evidence_root,
478    )
479    if owner["actor"] == independent["actor"]:
480        raise EvidenceError("owner and independent verifier must be distinct actors")
481    verification_ref, verification = validate_authority_decision(
482        observations["verification_decision"],
483        "verification_decision",
484        "verify_omega_full_auto_candidate",
485        "openagents.assurance_reviewer",
486        independent["actor"],
487        None,
488        bindings,
489        evidence_root,
490    )
491    if independent["authority_decision_ref"] != verification["decision_ref"]:
492        raise EvidenceError("independent attestation does not bind the verification decision")
493    release_ref, release = validate_authority_decision(
494        observations["release_decision"],
495        "release_decision",
496        "release_omega_full_auto_candidate",
497        "openagents.owner",
498        owner["actor"],
499        verification["decision_ref"],
500        bindings,
501        evidence_root,
502    )
503    if owner["authority_decision_ref"] != release["decision_ref"]:
504        raise EvidenceError("owner attestation does not bind the release decision")
505    verified_at = datetime.fromisoformat(verification["decided_at"].replace("Z", "+00:00"))
506    released_at = datetime.fromisoformat(release["decided_at"].replace("Z", "+00:00"))
507    if released_at <= verified_at:
508        raise EvidenceError("release decision must occur after independent verification")
509
510    return {
511        "schema": SCHEMA,
512        "schema_version": 1,
513        "status": "verified",
514        "candidate_verified": True,
515        "candidate_released": True,
516        "evidence_root": str(evidence_root),
517        "candidate": {
518            "candidate_digest": bindings["candidate_digest"],
519            "omega": {
520                "repository": "https://github.com/OpenAgentsInc/omega",
521                "commit": bindings["omega_commit"],
522                "dirty": False,
523            },
524            "artifact": {
525                "name": bindings["artifact_name"],
526                "sha256": bindings["artifact_sha256"],
527                "release_record_sha256": bindings["release_record_sha256"],
528            },
529            "omega_effectd": bindings["omega_effectd"],
530            "assurance_spec_admission": ASSURANCE_BINDING,
531        },
532        "obligations": validated_obligations,
533        "issue_matrix": validated_matrix,
534        "owner_observation": owner,
535        "independent_verification": independent,
536        "authority_decisions": {
537            "verification": verification_ref,
538            "release": release_ref,
539        },
540    }
541
542
543def write_evidence(path: Path, evidence: dict[str, Any]) -> None:
544    path.parent.mkdir(parents=True, exist_ok=True)
545    temporary = path.with_name(f".{path.name}.tmp")
546    try:
547        with temporary.open("x", encoding="utf-8") as handle:
548            json.dump(evidence, handle, indent=2, sort_keys=True)
549            handle.write("\n")
550        temporary.replace(path)
551    except OSError as error:
552        temporary.unlink(missing_ok=True)
553        raise EvidenceError(f"cannot write evidence {path}: {error}") from error
554
555
556def self_test() -> None:
557    with tempfile.TemporaryDirectory(prefix="omega-fa-evidence-self-test-") as directory:
558        test_root = Path(directory)
559        evidence_root = test_root / "evidence"
560        evidence_root.mkdir()
561        bindings = {
562            "candidate_digest": "a" * 64,
563            "artifact_name": "Omega.dmg",
564            "artifact_sha256": "b" * 64,
565            "release_record_sha256": "c" * 64,
566            "omega_commit": "d" * 40,
567            "omega_effectd": {
568                "release_tag": "omega-effectd-v0.1.0-rc.test",
569                "archive_sha256": "1" * 64,
570                "manifest_sha256": "2" * 64,
571                "source_commit": "3" * 40,
572                "source_tree": "4" * 40,
573                "wrapper_sha256": "5" * 64,
574                "bundle_sha256": "6" * 64,
575                "source_node_sha256": "7" * 64,
576                "packaged_node_sha256": "8" * 64,
577            },
578        }
579        artifact_receipt = evidence_root / "installed-observation.json"
580        artifact_receipt.write_text(
581            json.dumps(
582                {
583                    "schema": "openagents.omega.self-test-observation.v1",
584                    "candidate_digest": bindings["candidate_digest"],
585                }
586            ),
587            encoding="utf-8",
588        )
589        evidence_ref = {
590            "path": artifact_receipt.name,
591            "sha256": sha256_file(artifact_receipt),
592        }
593        observation = {
594            "status": "passed",
595            "evidence_tier": "installed_candidate",
596            "observed_at": "2026-07-24T12:00:00Z",
597            "evidence_refs": [evidence_ref],
598        }
599        attestation_bindings = {
600            key: bindings[key]
601            for key in (
602                "candidate_digest",
603                "artifact_sha256",
604                "release_record_sha256",
605                "omega_commit",
606            )
607        }
608        owner_decision_ref = "authority.decision.owner-release-self-test"
609        verifier_decision_ref = "authority.decision.verifier-self-test"
610        owner = {
611            "status": "passed",
612            "actor": "Owner",
613            "role": "openagents.owner",
614            "observed_at": "2026-07-24T12:01:00Z",
615            "evidence_refs": [evidence_ref],
616            "authority_decision_ref": owner_decision_ref,
617            **attestation_bindings,
618        }
619        reviewer = {
620            **owner,
621            "actor": "Euler",
622            "role": "openagents.assurance_reviewer",
623            "authority_decision_ref": verifier_decision_ref,
624        }
625
626        def write_decision(
627            name: str,
628            decision_ref: str,
629            action: str,
630            actor: str,
631            role: str,
632            decided_at: str,
633            predecessor: str | None,
634        ) -> dict[str, str]:
635            path = evidence_root / name
636            path.write_text(
637                json.dumps(
638                    {
639                        "schema": "openagents.omega.full-auto-authority-decision.v1",
640                        "decision_ref": decision_ref,
641                        "action": action,
642                        "actor": actor,
643                        "actor_role": role,
644                        "decided_at": decided_at,
645                        "outcome": "succeeded",
646                        **attestation_bindings,
647                        "predecessor_decision_ref": predecessor,
648                        "evidence_refs": [evidence_ref],
649                    }
650                ),
651                encoding="utf-8",
652            )
653            return {"path": name, "sha256": sha256_file(path)}
654
655        verifier_decision = write_decision(
656            "verification.json",
657            verifier_decision_ref,
658            "verify_omega_full_auto_candidate",
659            "Euler",
660            "openagents.assurance_reviewer",
661            "2026-07-24T12:02:00Z",
662            None,
663        )
664        owner_decision = write_decision(
665            "release.json",
666            owner_decision_ref,
667            "release_omega_full_auto_candidate",
668            "Owner",
669            "openagents.owner",
670            "2026-07-24T12:03:00Z",
671            verifier_decision_ref,
672        )
673        packet = {
674            "schema": OBSERVATION_SCHEMA,
675            **attestation_bindings,
676            "obligations": {key: dict(observation) for key in OBLIGATIONS},
677            "issue_matrix": {key: dict(observation) for key in ISSUE_GATES},
678            "owner_observation": owner,
679            "independent_verification": reviewer,
680            "verification_decision": verifier_decision,
681            "release_decision": owner_decision,
682        }
683        evidence = build_evidence(packet, bindings, evidence_root)
684        if evidence["status"] != "verified" or evidence["candidate_released"] is not True:
685            raise EvidenceError("self-test did not build released evidence")
686        packet["issue_matrix"]["incident_replay"]["evidence_tier"] = "source"
687        try:
688            build_evidence(packet, bindings, evidence_root)
689        except EvidenceError:
690            pass
691        else:
692            raise EvidenceError("self-test accepted source-only incident evidence")
693        packet["issue_matrix"]["incident_replay"]["evidence_tier"] = "installed_candidate"
694        packet["obligations"][OBLIGATIONS[0]]["evidence_refs"] = [evidence_ref, evidence_ref]
695        try:
696            build_evidence(packet, bindings, evidence_root)
697        except EvidenceError:
698            pass
699        else:
700            raise EvidenceError("self-test accepted duplicate evidence references")
701        packet["obligations"][OBLIGATIONS[0]]["evidence_refs"] = [evidence_ref]
702        original = artifact_receipt.read_text(encoding="utf-8")
703        artifact_receipt.write_text(original + "\n", encoding="utf-8")
704        try:
705            build_evidence(packet, bindings, evidence_root)
706        except EvidenceError:
707            pass
708        else:
709            raise EvidenceError("self-test accepted a changed evidence file")
710        artifact_receipt.write_text(original, encoding="utf-8")
711        release_document = load_json(evidence_root / "release.json", "release fixture")
712        release_document["predecessor_decision_ref"] = None
713        (evidence_root / "release.json").write_text(
714            json.dumps(release_document), encoding="utf-8"
715        )
716        packet["release_decision"]["sha256"] = sha256_file(evidence_root / "release.json")
717        try:
718            build_evidence(packet, bindings, evidence_root)
719        except EvidenceError:
720            pass
721        else:
722            raise EvidenceError("self-test accepted release without verifier predecessor")
723
724        artifact = test_root / "Omega.dmg"
725        artifact.write_bytes(b"candidate")
726        release_path = test_root / "release.json"
727        release = {
728            "source": {"commit": "d" * 40, "dirty": False},
729            "digests": {"package_sha256": sha256_file(artifact)},
730            "components": {"omega_effectd": bindings["omega_effectd"]},
731        }
732        release_path.write_text(json.dumps(release), encoding="utf-8")
733        identity_candidate = {
734            "omega": {"commit": "d" * 40},
735            "artifact": {
736                "name": artifact.name,
737                "sha256": sha256_file(artifact),
738                "release_record_sha256": sha256_file(release_path),
739            },
740        }
741        identity = {
742            "candidate": identity_candidate,
743            "candidate_digest": {
744                "value": hashlib.sha256(
745                    json.dumps(
746                        identity_candidate, sort_keys=True, separators=(",", ":")
747                    ).encode()
748                ).hexdigest()
749            },
750        }
751        observed_bindings = candidate_bindings(
752            release, release_path, artifact, identity
753        )
754        if observed_bindings["omega_effectd"]["source_tree"] != "4" * 40:
755            raise EvidenceError("self-test did not preserve the source-tree binding")
756
757        duplicate = test_root / "duplicate.json"
758        duplicate.write_text('{"schema":1,"schema":1}', encoding="utf-8")
759        try:
760            load_json(duplicate, "duplicate fixture")
761        except EvidenceError:
762            pass
763        else:
764            raise EvidenceError("self-test accepted a duplicate JSON key")
765    if not SCHEMA_PATH.is_file():
766        raise EvidenceError("Full Auto candidate evidence schema is missing")
767    print("Omega Full Auto candidate evidence self-test passed")
768
769
770def parse_arguments() -> argparse.Namespace:
771    parser = argparse.ArgumentParser(description=__doc__)
772    parser.add_argument("--self-test", action="store_true")
773    parser.add_argument("--release-record", type=Path)
774    parser.add_argument("--artifact", type=Path)
775    parser.add_argument("--identity-evidence", type=Path)
776    parser.add_argument("--observations", type=Path)
777    parser.add_argument("--evidence-root", type=Path)
778    parser.add_argument(
779        "--output",
780        type=Path,
781        default=ROOT / "target/omega-full-auto-evidence/candidate-evidence.json",
782    )
783    arguments = parser.parse_args()
784    if arguments.self_test:
785        return arguments
786    missing = [
787        name
788        for name in (
789            "release_record",
790            "artifact",
791            "identity_evidence",
792            "observations",
793            "evidence_root",
794        )
795        if getattr(arguments, name) is None
796    ]
797    if missing:
798        parser.error(
799            "generate mode requires "
800            + ", ".join(f"--{name.replace('_', '-')}" for name in missing)
801        )
802    return arguments
803
804
805def main() -> int:
806    arguments = parse_arguments()
807    try:
808        if arguments.self_test:
809            self_test()
810            return 0
811        release_record = load_json(arguments.release_record, "release record")
812        identity = load_json(arguments.identity_evidence, "identity candidate evidence")
813        observations = load_json(arguments.observations, "Full Auto observations")
814        bindings = candidate_bindings(
815            release_record, arguments.release_record, arguments.artifact, identity
816        )
817        evidence = build_evidence(observations, bindings, arguments.evidence_root)
818        write_evidence(arguments.output, evidence)
819        print(f"wrote verified Full Auto candidate evidence: {arguments.output}")
820        return 0
821    except EvidenceError as error:
822        print(f"error: {error}", file=sys.stderr)
823        return 1
824
825
826if __name__ == "__main__":
827    raise SystemExit(main())
828