#!/usr/bin/env python3 """Build and verify Leo's pinned, reproducible identity manifest.""" from __future__ import annotations import argparse import json import re import sys from pathlib import Path from typing import Any if __package__ in {None, ""}: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts import leo_behavior_manifest as behavior SCHEMA = "livingip.leoIdentityManifest.v1" CONSTITUTION_SCHEMA = "livingip.leoConstitutionSource.v1" DATABASE_IDENTITY_SCHEMA = "livingip.leoDatabaseIdentitySource.v1" STRUCTURED_VIEW_SCHEMA = "livingip.leoCompiledIdentityView.v1" DATABASE_FINGERPRINT_SCHEMA = "livingip.teleoCanonicalDatabaseFingerprint.v1" SYNTHETIC_DATABASE_MODE = "synthetic_noncanonical_fixture" CANONICAL_DATABASE_MODE = "canonical_database_same_transaction_export" HEX_64 = re.compile(r"^[0-9a-f]{64}$") HEX_40 = re.compile(r"^[0-9a-f]{40}$") IDENTITY_TABLES = frozenset( { "public.personas", "public.strategies", "public.beliefs", "public.shared_root", "public.strategy_nodes", "public.strategy_node_anchors", "public.claims", } ) def expected_runtime_policy() -> dict[str, Any]: return { "network_send_enabled": False, "database_write_enabled": False, "allowed_tools": ["identity_readback"], } def expected_session_boundary() -> dict[str, Any]: return { "classification": "temporary_noncanonical", "included_in_identity_inputs": False, "may_be_used_as_evidence": False, "restart_policy": "fresh_process_with_session_state_outside_identity_authority", } def expected_gatewayrunner_contract() -> dict[str, Any]: return { "profile_surfaces": ["config.yaml", "generated SOUL.md", "skills", "plugins", "bin tools"], "required_absent_or_empty": { "persistent_memory": behavior.canonical_sha256([]), "pairing_store": behavior.canonical_sha256([]), "project_context": behavior.canonical_sha256([]), "generated_skill_prompt_cache": behavior.canonical_sha256([]), "prior_sessions_at_first_start": behavior.canonical_sha256([]), }, "fixed_message_context": { "auto_skill": None, "reply_context": None, "media_or_file_context": None, }, "required_t3_turn_receipt_fields": [ "runner_class", "authorization_decision", "effective_system_prompt_sha256", "compiled_skills_prompt_sha256", "tool_registry_schema_sha256", "plugin_registry_sha256", "selected_model", "selected_provider", "api_mode", "base_url_host", "database_retrieval_receipt", "session_key_sha256", "persisted_session_id_sha256", ], } class IdentityManifestError(ValueError): """An identity input is incomplete, inconsistent, or drifted.""" def _require(condition: bool, message: str) -> None: if not condition: raise IdentityManifestError(message) def _is_sha256(value: Any) -> bool: return isinstance(value, str) and bool(HEX_64.fullmatch(value)) def load_json(path: Path, label: str) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise IdentityManifestError(f"cannot read {label}: {type(exc).__name__}") from exc if not isinstance(value, dict): raise IdentityManifestError(f"{label} must be a JSON object") return value def canonical_rules(value: dict[str, Any]) -> list[dict[str, str]]: _require(value.get("schema") == CONSTITUTION_SCHEMA, "unsupported constitution schema") _require(value.get("identity_name") == "Leo", "constitution identity_name must be Leo") raw_rules = value.get("rules") _require(isinstance(raw_rules, list) and bool(raw_rules), "constitution rules must be non-empty") rules: list[dict[str, str]] = [] for raw in raw_rules: _require(isinstance(raw, dict), "each constitutional rule must be an object") rule_id = str(raw.get("id") or "").strip() text = str(raw.get("text") or "").strip() _require(bool(rule_id and text), "each constitutional rule needs id and text") rules.append({"id": rule_id, "text": text}) _require(len({item["id"] for item in rules}) == len(rules), "constitutional rule ids must be unique") return sorted(rules, key=lambda item: item["id"]) def canonical_database_records(value: dict[str, Any], *, fingerprint_sha256: str) -> list[dict[str, str | int]]: _require(value.get("schema") == DATABASE_IDENTITY_SCHEMA, "unsupported database identity schema") _require(value.get("identity_name") == "Leo", "database identity_name must be Leo") _require( value.get("database_fingerprint_sha256") == fingerprint_sha256, "database identity export is not bound to the supplied database fingerprint", ) raw_records = value.get("records") _require(isinstance(raw_records, list) and bool(raw_records), "database identity records must be non-empty") records: list[dict[str, str | int]] = [] for raw in raw_records: _require(isinstance(raw, dict), "each database identity record must be an object") rank = raw.get("rank", 0) record: dict[str, str | int] = { "table": str(raw.get("table") or "").strip(), "row_id": str(raw.get("row_id") or "").strip(), "kind": str(raw.get("kind") or "").strip(), "rank": rank if isinstance(rank, int) and not isinstance(rank, bool) else -1, "text": str(raw.get("text") or "").strip(), "status": str(raw.get("status") or "").strip(), "evidence_sha256": str(raw.get("evidence_sha256") or "").strip(), } _require(record["table"] in IDENTITY_TABLES, "database identity table is outside the allowlist") _require(bool(record["row_id"] and record["kind"] and record["text"]), "database record fields are incomplete") _require(isinstance(record["rank"], int) and record["rank"] >= 0, "database record rank is invalid") _require(record["status"] == "active", "database identity records must be active selected rows") _require(_is_sha256(record["evidence_sha256"]), "database record evidence_sha256 is invalid") records.append(record) keys = {(item["table"], item["row_id"]) for item in records} _require(len(keys) == len(records), "database identity row bindings must be unique") return sorted(records, key=lambda item: (str(item["table"]), int(item["rank"]), str(item["row_id"]))) def database_fingerprint_semantic(value: dict[str, Any]) -> dict[str, Any]: """Bind immutable capture provenance while excluding only the volatile WAL position.""" result = {key: value[key] for key in sorted(value) if key != "read_consistency"} consistency = value["read_consistency"] result["read_consistency"] = { "status": consistency["status"], "before": {key: item for key, item in consistency["before"].items() if key != "wal_lsn"}, "after": {key: item for key, item in consistency["after"].items() if key != "wal_lsn"}, } return result def database_identity_provenance( value: dict[str, Any], *, fingerprint: dict[str, Any], ) -> dict[str, Any]: raw = value.get("source_provenance") _require(isinstance(raw, dict), "database identity source provenance is missing") mode = raw.get("mode") if mode == SYNTHETIC_DATABASE_MODE: _require( raw == { "mode": SYNTHETIC_DATABASE_MODE, "canonical_authority_granted": False, "same_transaction_as_fingerprint": False, "export_receipt_sha256": None, }, "synthetic database provenance contract is invalid", ) _require( fingerprint.get("environment") == "disposable_fixture", "synthetic rows require a fixture fingerprint", ) _require("export_receipt" not in value, "synthetic database identity must not claim an export receipt") return {**raw, "authority": "synthetic_noncanonical_fixture"} if mode == CANONICAL_DATABASE_MODE: raise IdentityManifestError( "canonical authority requires independently verified output from the database-owning same-transaction " "exporter; caller-supplied T2 JSON cannot grant it" ) raise IdentityManifestError("unsupported database identity provenance mode") def validate_database_fingerprint(value: dict[str, Any]) -> None: _require(value.get("schema") == DATABASE_FINGERPRINT_SCHEMA, "unsupported database fingerprint schema") _require(value.get("status") == "ok", "database fingerprint status must be ok") for field in ("fingerprint_sha256", "table_rows_sha256", "structure_sha256"): _require(_is_sha256(value.get(field)), f"database fingerprint {field} is invalid") _require(isinstance(value.get("table_count"), int) and value["table_count"] > 0, "table_count must be positive") _require(isinstance(value.get("total_rows"), int) and value["total_rows"] >= 0, "total_rows is invalid") consistency = value.get("read_consistency") _require(isinstance(consistency, dict), "database read_consistency is missing") before = consistency.get("before") after = consistency.get("after") _require(consistency.get("status") == "stable_wal_marker", "database fingerprint was not read consistently") _require(isinstance(before, dict) and before == after and bool(before), "database read markers differ") for field in ("database", "database_user", "system_identifier", "wal_lsn"): _require(bool(before.get(field)), f"database read marker {field} is missing") def validate_content_manifest(value: Any, *, label: str) -> None: """Recompute the structural metadata of a replayable content manifest.""" _require(isinstance(value, dict), f"{label} is invalid") _require( set(value) == {"files", "missing", "symlinks", "file_count", "total_bytes", "sha256"}, f"{label} fields are invalid", ) files = value.get("files") missing = value.get("missing") symlinks = value.get("symlinks") _require(isinstance(files, list) and bool(files), f"{label} is empty") _require(missing == [], f"{label} has missing inputs") _require(symlinks == [], f"{label} contains unsupported symlinks") paths: list[str] = [] total_bytes = 0 for item in files: _require(isinstance(item, dict), f"{label} contains unreadable inputs") _require(set(item) == {"path", "bytes", "mode", "sha256"}, f"{label} contains unreadable inputs") path = item.get("path") size = item.get("bytes") mode = item.get("mode") _require(isinstance(path, str) and bool(path), f"{label} contains an invalid path") _require(isinstance(size, int) and not isinstance(size, bool) and size >= 0, f"{label} byte size is invalid") _require(isinstance(mode, str) and bool(re.fullmatch(r"0o[0-7]{3}", mode)), f"{label} mode is invalid") _require(_is_sha256(item.get("sha256")), f"{label} contains unreadable inputs") paths.append(path) total_bytes += size _require(paths == sorted(paths) and len(paths) == len(set(paths)), f"{label} paths are not unique and sorted") _require(value.get("file_count") == len(files), f"{label} file count is invalid") _require(value.get("total_bytes") == total_bytes, f"{label} total bytes is invalid") stable = {"files": files, "missing": missing, "symlinks": symlinks} _require(behavior.canonical_sha256(stable) == value.get("sha256"), f"{label} content hash is invalid") def validate_behavior_manifest(value: dict[str, Any]) -> None: _require(value.get("schema") == behavior.SCHEMA, "unsupported behavior manifest schema") for field in ("behavior_sha256", "static_runtime_sha256", "identity_runtime_sha256"): _require(_is_sha256(value.get(field)), f"behavior manifest {field} is invalid") _require( behavior.canonical_sha256(behavior.identity_runtime_payload(value)) == value["identity_runtime_sha256"], "behavior identity runtime hash drift detected", ) _require( behavior.canonical_sha256(behavior.static_runtime_payload(value)) == value["static_runtime_sha256"], "behavior static runtime hash drift detected", ) _require( behavior.canonical_sha256(behavior.stable_manifest_payload(value)) == value["behavior_sha256"], "behavior manifest stable payload hash drift detected", ) model = value.get("model_runtime") _require(isinstance(model, dict) and model.get("status") == "observed", "model runtime was not observed") _require(bool(model.get("provider") and model.get("default_model")), "provider and default model must be pinned") for field in ( "semantic_config_sha256", "identity_config_sha256", "model_section_sha256", "gateway_permissions_sha256", "tool_permissions_sha256", "memory_section_sha256", "agent_runtime_sha256", ): _require(_is_sha256(model.get(field)), f"model runtime {field} is invalid") _require( model.get("base_model_weights") == "external_provider_managed_not_locally_hashable", "hosted model weight boundary is missing", ) _require( model.get("actual_per_turn_model_must_be_recorded_by_the_call_receipt") is True, "model receipt gate missing" ) policy = model.get("identity_runtime_policy") _require(policy == expected_runtime_policy(), "identity runtime policy must be the exact local readback policy") for field in ("hermes_runtime", "teleo_infrastructure_runtime"): runtime = value.get(field) _require(isinstance(runtime, dict), f"{field} is missing") _require(bool(HEX_40.fullmatch(str(runtime.get("git_head") or ""))), f"{field} git_head is invalid") git_state = runtime.get("git_state") _require( isinstance(git_state, dict) and git_state.get("worktree_clean") is True, f"{field} is not reconstructible from clean Git", ) tree = runtime.get("identity_source_tree" if field == "teleo_infrastructure_runtime" else "source_tree") validate_content_manifest(tree, label=f"{field} source tree") components = value.get("components") _require(isinstance(components, dict), "behavior components are missing") for name in behavior.IDENTITY_RUNTIME_COMPONENTS: content = (components.get(name) or {}).get("content") validate_content_manifest(content, label=f"component {name}") python_runtime = value.get("python_runtime") _require(isinstance(python_runtime, dict), "Python runtime binding is missing") _require( bool(python_runtime.get("implementation") and python_runtime.get("python_version")), "Python runtime is incomplete", ) _require(_is_sha256(python_runtime.get("runtime_distributions_sha256")), "runtime package binding is invalid") _require( all(python_runtime.get("runtime_distributions", {}).values()), "a required runtime distribution is missing" ) def _repo_path(path: Path, source_root: Path) -> str: try: return path.resolve(strict=True).relative_to(source_root.resolve(strict=True)).as_posix() except (OSError, ValueError) as exc: raise IdentityManifestError(f"identity source is outside the pinned source root: {path}") from exc def source_binding( path: Path, *, source_root: Path, semantic_value: Any, count: int, pin_content: bool = True, ) -> dict[str, Any]: result = { "repo_path": _repo_path(path, source_root), "semantic_sha256": behavior.canonical_sha256(semantic_value), "record_count": count, } if pin_content: result["content_sha256"] = behavior.file_sha256(path) return result def render_identity_views( *, identity_inputs_sha256: str, database_fingerprint_sha256: str, database_provenance: dict[str, Any], rules: list[dict[str, str]], records: list[dict[str, str | int]], ) -> dict[str, str]: soul = [ "", "# Leo", "", f"Identity inputs: `{identity_inputs_sha256}`", "", "## Static constitutional rules", "", "These rules are static constitutional inputs, not database claims.", "", ] soul.extend(f"- **{item['id']}**: {item['text']}" for item in rules) database_description = ( "canonical database capture" if database_provenance["canonical_authority_granted"] else "synthetic noncanonical fixture" ) soul.extend( [ "", "## Database-derived identity", "", f"Generated from {database_description} fingerprint `{database_fingerprint_sha256}`.", "", ] ) soul.extend( f"- `{item['table']}/{item['row_id']}` [{item['kind']}, rank {item['rank']}]: {item['text']} " f"(evidence `{item['evidence_sha256']}`)" for item in records ) soul.extend( [ "", "## Authority and session boundary", "", "- This `SOUL.md` is a generated view and is never an authority by itself.", "- Generated database identity is authoritative only when its same-transaction export receipt grants canonical authority.", "- Runtime/session memory is temporary, noncanonical, and cannot serve as evidence.", "- Hosted model weights are provider-managed and must be attributed by each turn receipt.", "", ] ) structured = { "schema": STRUCTURED_VIEW_SCHEMA, "identity_name": "Leo", "identity_inputs_sha256": identity_inputs_sha256, "static_constitution": {"authority": "pinned_static_source", "rules": rules}, "database_identity": { "authority": database_provenance["authority"], "canonical_authority_granted": database_provenance["canonical_authority_granted"], "source_mode": database_provenance["mode"], "database_fingerprint_sha256": database_fingerprint_sha256, "records": records, }, "session_boundary": { "authority": "none", "classification": "temporary_noncanonical", "may_be_used_as_evidence": False, }, } return { "SOUL.md": "\n".join(soul), "identity.json": json.dumps(structured, indent=2, sort_keys=True) + "\n", } def build_identity_manifest( *, behavior_manifest: dict[str, Any], database_fingerprint_path: Path, constitution_path: Path, database_identity_path: Path, source_root: Path, source_commit: str | None = None, ) -> dict[str, Any]: validate_behavior_manifest(behavior_manifest) database_fingerprint = load_json(database_fingerprint_path, "database fingerprint") validate_database_fingerprint(database_fingerprint) constitution = load_json(constitution_path, "constitution source") database_identity = load_json(database_identity_path, "database identity source") rules = canonical_rules(constitution) records = canonical_database_records( database_identity, fingerprint_sha256=database_fingerprint["fingerprint_sha256"] ) database_provenance = database_identity_provenance( database_identity, fingerprint=database_fingerprint, ) runtime_commit = behavior_manifest["teleo_infrastructure_runtime"]["git_head"] source_commit = source_commit or runtime_commit _require(bool(HEX_40.fullmatch(str(source_commit))), "source commit is invalid") _require(source_commit == runtime_commit, "source commit does not match the behavior manifest") model = behavior_manifest["model_runtime"] components = behavior_manifest["components"] marker = database_fingerprint["read_consistency"]["before"] core = { "identity_name": "Leo", "source_commit": source_commit, "runtime": { "identity_runtime_sha256": behavior_manifest["identity_runtime_sha256"], "hermes_git_head": behavior_manifest["hermes_runtime"]["git_head"], "hermes_source_sha256": behavior_manifest["hermes_runtime"]["source_tree"]["sha256"], "teleo_git_head": runtime_commit, "teleo_source_sha256": behavior_manifest["teleo_infrastructure_runtime"]["identity_source_tree"]["sha256"], "python": behavior_manifest["python_runtime"], }, "model": { "provider": model["provider"], "default_model": model["default_model"], "smart_routing": model.get("smart_routing"), "smart_routing_model": model.get("smart_routing_model"), "gateway_smart_model_routing": model.get("gateway_smart_model_routing"), "model_section_sha256": model["model_section_sha256"], "hosted_weights": "external_provider_managed_not_locally_hashable", "actual_model_required_in_turn_receipt": True, }, "skills": { "content_sha256": components["procedural_skills"]["content"]["sha256"], "file_count": components["procedural_skills"]["content"]["file_count"], }, "permissions": { "identity_config_sha256": model["identity_config_sha256"], "gateway_permissions_sha256": model["gateway_permissions_sha256"], "tool_permissions_sha256": model["tool_permissions_sha256"], "runtime_policy": model["identity_runtime_policy"], "authorization_decision_required_in_runtime_receipt": True, }, "canonical_database": { "database": marker["database"], "database_user": marker["database_user"], "system_identifier": marker["system_identifier"], "authority": database_provenance["authority"], "canonical_authority_granted": database_provenance["canonical_authority_granted"], **{ key: database_fingerprint[key] for key in ("fingerprint_sha256", "table_rows_sha256", "structure_sha256", "table_count", "total_rows") }, "source": source_binding( database_fingerprint_path, source_root=source_root, semantic_value=database_fingerprint_semantic(database_fingerprint), count=database_fingerprint["table_count"], pin_content=False, ), }, "identity_sources": { "static_constitution": { "authority": "pinned_static_source", **source_binding(constitution_path, source_root=source_root, semantic_value=rules, count=len(rules)), }, "database_derived_identity": { "authority": database_provenance["authority"], "provenance": database_provenance, "database_fingerprint_sha256": database_fingerprint["fingerprint_sha256"], **source_binding( database_identity_path, source_root=source_root, semantic_value={"provenance": database_provenance, "records": records}, count=len(records), ), }, }, "session_boundary": expected_session_boundary(), "gatewayrunner_execution_contract": expected_gatewayrunner_contract(), } identity_inputs_sha256 = behavior.canonical_sha256(core) views = render_identity_views( identity_inputs_sha256=identity_inputs_sha256, database_fingerprint_sha256=database_fingerprint["fingerprint_sha256"], database_provenance=database_provenance, rules=rules, records=records, ) stable = { "schema": SCHEMA, "identity_inputs": core, "identity_inputs_sha256": identity_inputs_sha256, "compiled_views": { name: { "role": "generated_view_not_authority", "sha256": behavior.sha256_bytes(content.encode("utf-8")), "generated_from_identity_inputs_sha256": identity_inputs_sha256, } for name, content in sorted(views.items()) }, } return {**stable, "manifest_sha256": behavior.canonical_sha256(stable)} def validate_identity_manifest(value: dict[str, Any]) -> None: _require(value.get("schema") == SCHEMA, "unsupported identity manifest schema") expected = value.get("manifest_sha256") _require(_is_sha256(expected), "identity manifest sha256 is invalid") stable = {key: item for key, item in value.items() if key != "manifest_sha256"} _require(behavior.canonical_sha256(stable) == expected, "identity manifest hash drift detected") core = value.get("identity_inputs") _require(isinstance(core, dict), "identity inputs are missing") _require( set(core) == { "identity_name", "source_commit", "runtime", "model", "skills", "permissions", "canonical_database", "identity_sources", "session_boundary", "gatewayrunner_execution_contract", }, "identity input contract fields are invalid", ) identity_hash = value.get("identity_inputs_sha256") _require(_is_sha256(identity_hash), "identity input hash is invalid") _require(behavior.canonical_sha256(core) == identity_hash, "identity input hash drift detected") _require(core.get("identity_name") == "Leo", "identity name must be Leo") runtime = core.get("runtime") _require(isinstance(runtime, dict), "runtime binding is missing") _require( set(runtime) == { "identity_runtime_sha256", "hermes_git_head", "hermes_source_sha256", "teleo_git_head", "teleo_source_sha256", "python", }, "runtime binding fields are invalid", ) _require(_is_sha256(runtime.get("identity_runtime_sha256")), "identity runtime binding is invalid") for field in ("hermes_git_head", "teleo_git_head"): _require(bool(HEX_40.fullmatch(str(runtime.get(field) or ""))), f"runtime {field} is invalid") for field in ("hermes_source_sha256", "teleo_source_sha256"): _require(_is_sha256(runtime.get(field)), f"runtime {field} is invalid") _require(core.get("source_commit") == runtime["teleo_git_head"], "source commit is not bound to Teleo Git") python_runtime = runtime.get("python") _require(isinstance(python_runtime, dict), "Python runtime binding is missing") _require( bool(python_runtime.get("implementation") and python_runtime.get("python_version")), "Python runtime binding is incomplete", ) _require(_is_sha256(python_runtime.get("runtime_distributions_sha256")), "Python distribution binding is invalid") model = core.get("model") _require(isinstance(model, dict), "model binding is missing") _require( set(model) == { "provider", "default_model", "smart_routing", "smart_routing_model", "gateway_smart_model_routing", "model_section_sha256", "hosted_weights", "actual_model_required_in_turn_receipt", }, "model binding fields are invalid", ) _require(bool(model.get("provider") and model.get("default_model")), "model provider and default must be pinned") _require(_is_sha256(model.get("model_section_sha256")), "model section binding is invalid") _require( model.get("hosted_weights") == "external_provider_managed_not_locally_hashable", "hosted model boundary is invalid", ) _require(model.get("actual_model_required_in_turn_receipt") is True, "per-turn model receipt gate is invalid") skills = core.get("skills") _require(isinstance(skills, dict) and set(skills) == {"content_sha256", "file_count"}, "skills binding is invalid") _require(_is_sha256(skills.get("content_sha256")), "skills content hash is invalid") _require( isinstance(skills.get("file_count"), int) and not isinstance(skills.get("file_count"), bool) and skills["file_count"] > 0, "skills file count is invalid", ) permissions = core.get("permissions") _require(isinstance(permissions, dict), "permission binding is missing") _require( set(permissions) == { "identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256", "runtime_policy", "authorization_decision_required_in_runtime_receipt", }, "permission binding fields are invalid", ) for field in ("identity_config_sha256", "gateway_permissions_sha256", "tool_permissions_sha256"): _require(_is_sha256(permissions.get(field)), f"permission binding {field} is invalid") _require( permissions.get("runtime_policy") == expected_runtime_policy(), "identity runtime policy must be the exact local readback policy", ) _require( permissions.get("authorization_decision_required_in_runtime_receipt") is True, "runtime authorization receipt gate is invalid", ) database = core.get("canonical_database") _require(isinstance(database, dict), "canonical database binding is missing") for field in ("fingerprint_sha256", "table_rows_sha256", "structure_sha256"): _require(_is_sha256(database.get(field)), f"canonical database {field} is invalid") _require( bool(database.get("database") and database.get("database_user") and database.get("system_identifier")), "canonical database identity is incomplete", ) allowed_authorities = {"synthetic_noncanonical_fixture": False} _require(database.get("authority") in allowed_authorities, "database authority is invalid") _require( database.get("canonical_authority_granted") is allowed_authorities[database["authority"]], "database canonical authority grant is invalid", ) _require( isinstance(database.get("table_count"), int) and not isinstance(database.get("table_count"), bool) and database["table_count"] > 0, "canonical database table_count is invalid", ) _require( isinstance(database.get("total_rows"), int) and not isinstance(database.get("total_rows"), bool) and database["total_rows"] >= 0, "canonical database total_rows is invalid", ) def validate_source_binding(binding: Any, *, label: str, content_required: bool) -> None: _require(isinstance(binding, dict), f"{label} source binding is missing") repo_path = binding.get("repo_path") _require( isinstance(repo_path, str) and bool(repo_path) and not Path(repo_path).is_absolute() and ".." not in Path(repo_path).parts, f"{label} source path is invalid", ) _require(_is_sha256(binding.get("semantic_sha256")), f"{label} semantic binding is invalid") _require( isinstance(binding.get("record_count"), int) and not isinstance(binding.get("record_count"), bool) and binding["record_count"] > 0, f"{label} record count is invalid", ) if content_required: _require(_is_sha256(binding.get("content_sha256")), f"{label} content binding is invalid") else: _require("content_sha256" not in binding, f"{label} content binding must exclude volatile capture markers") validate_source_binding(database.get("source"), label="database fingerprint", content_required=False) sources = core.get("identity_sources") _require( isinstance(sources, dict) and set(sources) == {"static_constitution", "database_derived_identity"}, "identity source bindings are invalid", ) constitution = sources["static_constitution"] derived = sources["database_derived_identity"] _require(isinstance(constitution, dict), "constitution source binding is invalid") _require(isinstance(derived, dict), "database identity source binding is invalid") _require(constitution.get("authority") == "pinned_static_source", "constitution authority is invalid") _require(derived.get("authority") == database["authority"], "database identity authority is invalid") provenance = derived.get("provenance") _require(isinstance(provenance, dict), "database identity provenance is missing") _require(provenance.get("authority") == database["authority"], "database identity provenance authority is invalid") _require( provenance.get("canonical_authority_granted") is database["canonical_authority_granted"], "database identity provenance grant is invalid", ) _require(provenance.get("mode") == SYNTHETIC_DATABASE_MODE, "synthetic database provenance mode is invalid") _require( provenance.get("same_transaction_as_fingerprint") is False, "synthetic fixture transaction claim is invalid", ) _require(provenance.get("export_receipt_sha256") is None, "synthetic fixture must not claim an export receipt") _require( derived.get("database_fingerprint_sha256") == database["fingerprint_sha256"], "database identity source is misbound", ) validate_source_binding(constitution, label="constitution", content_required=True) validate_source_binding(derived, label="database identity", content_required=True) _require(core.get("session_boundary") == expected_session_boundary(), "session authority boundary is invalid") _require( core.get("gatewayrunner_execution_contract") == expected_gatewayrunner_contract(), "GatewayRunner execution contract is invalid", ) views = value.get("compiled_views") _require(isinstance(views, dict) and set(views) == {"SOUL.md", "identity.json"}, "compiled views are incomplete") for name, view in views.items(): _require(isinstance(view, dict) and _is_sha256(view.get("sha256")), f"compiled view {name} is invalid") _require( set(view) == {"role", "sha256", "generated_from_identity_inputs_sha256"}, f"compiled view {name} fields are invalid", ) _require(view.get("role") == "generated_view_not_authority", f"compiled view {name} authority is invalid") _require( view.get("generated_from_identity_inputs_sha256") == identity_hash, f"compiled view {name} is misbound" ) def verify_bound_sources(value: dict[str, Any], source_root: Path) -> dict[str, str]: validate_identity_manifest(value) core = value["identity_inputs"] bindings = { "database fingerprint": core["canonical_database"]["source"], "constitution source": core["identity_sources"]["static_constitution"], "database identity source": core["identity_sources"]["database_derived_identity"], } paths = {label: source_root / binding["repo_path"] for label, binding in bindings.items()} for label, binding in bindings.items(): path = paths[label] _require(path.is_file(), f"bound {label} is missing") if binding.get("content_sha256"): _require(behavior.file_sha256(path) == binding["content_sha256"], f"bound {label} content drift detected") fingerprint = load_json(paths["database fingerprint"], "database fingerprint") validate_database_fingerprint(fingerprint) _require( behavior.canonical_sha256(database_fingerprint_semantic(fingerprint)) == bindings["database fingerprint"]["semantic_sha256"], "database fingerprint semantic drift detected", ) marker = fingerprint["read_consistency"]["before"] for field in ("database", "database_user", "system_identifier"): _require(marker[field] == core["canonical_database"][field], f"canonical database {field} drift detected") rules = canonical_rules(load_json(paths["constitution source"], "constitution source")) database_identity = load_json(paths["database identity source"], "database identity source") records = canonical_database_records( database_identity, fingerprint_sha256=fingerprint["fingerprint_sha256"], ) database_provenance = database_identity_provenance( database_identity, fingerprint=fingerprint, ) _require( behavior.canonical_sha256(rules) == bindings["constitution source"]["semantic_sha256"], "constitution semantic drift detected", ) _require( behavior.canonical_sha256({"provenance": database_provenance, "records": records}) == bindings["database identity source"]["semantic_sha256"], "database identity semantic drift detected", ) _require( database_provenance == core["identity_sources"]["database_derived_identity"]["provenance"], "database identity provenance drift detected", ) _require( fingerprint["fingerprint_sha256"] == core["canonical_database"]["fingerprint_sha256"], "canonical database fingerprint drift detected", ) views = render_identity_views( identity_inputs_sha256=value["identity_inputs_sha256"], database_fingerprint_sha256=fingerprint["fingerprint_sha256"], database_provenance=database_provenance, rules=rules, records=records, ) for name, content in views.items(): _require( behavior.sha256_bytes(content.encode("utf-8")) == value["compiled_views"][name]["sha256"], f"compiled view contract drift detected for {name}", ) return views def write_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, indent=2, sort_keys=True) + "\n", encoding="utf-8") def main() -> int: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) generate = subparsers.add_parser("generate") generate.add_argument("--behavior-manifest", type=Path, required=True) generate.add_argument("--database-fingerprint", type=Path, required=True) generate.add_argument("--constitution", type=Path, required=True) generate.add_argument("--database-identity", type=Path, required=True) generate.add_argument("--source-root", type=Path, required=True) generate.add_argument("--source-commit") generate.add_argument("--output", type=Path, required=True) verify = subparsers.add_parser("verify") verify.add_argument("--manifest", type=Path, required=True) verify.add_argument("--source-root", type=Path) args = parser.parse_args() try: if args.command == "generate": manifest = build_identity_manifest( behavior_manifest=load_json(args.behavior_manifest, "behavior manifest"), database_fingerprint_path=args.database_fingerprint, constitution_path=args.constitution, database_identity_path=args.database_identity, source_root=args.source_root, source_commit=args.source_commit, ) write_json(args.output, manifest) else: manifest = load_json(args.manifest, "identity manifest") validate_identity_manifest(manifest) if args.source_root: verify_bound_sources(manifest, args.source_root) print(json.dumps({"status": "ok", "manifest_sha256": manifest["manifest_sha256"]})) return 0 except IdentityManifestError as exc: print(json.dumps({"status": "error", "error": "identity_manifest_invalid", "message": str(exc)})) return 2 if __name__ == "__main__": raise SystemExit(main())