#!/usr/bin/env python3 """Compile, verify, and query a disposable Leo identity profile.""" from __future__ import annotations import argparse import json import os import shutil import sys import uuid from datetime import UTC, datetime from pathlib import Path from typing import Any import yaml if __package__ in {None, ""}: sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts import leo_behavior_manifest as behavior from scripts import leo_identity_manifest as identity LOCK_SCHEMA = "livingip.leoIdentityProfileLock.v1" COMPILE_RECEIPT_SCHEMA = "livingip.leoIdentityCompileReceipt.v1" QUERY_RECEIPT_SCHEMA = "livingip.leoIdentityQueryReceipt.v1" SESSION_RECORD_SCHEMA = "livingip.leoTemporarySessionRecord.v1" FIXED_QUERY = "What defines Leo's identity, and what is temporary?" STATIC_PROFILE_ENTRIES = ("config.yaml", "skills", "plugins", "bin") COMPILED_PROFILE_ENTRIES = frozenset( { "config.yaml", "skills", "plugins", "bin", "SOUL.md", "identity.json", "identity-manifest.json", "identity-lock.json", "compile-receipt.json", "sessions", "state", } ) OMITTED_NONAUTHORITATIVE_ENTRIES = ( "memories", "MEMORY.md", "USER.md", "platforms/pairing", ".skills_prompt_snapshot.json", ".hermes.md", "HERMES.md", "AGENTS.md", "CLAUDE.md", ".cursorrules", ) class IdentityProfileError(identity.IdentityManifestError): """A compiled profile cannot be trusted or started.""" def _require(condition: bool, message: str) -> None: if not condition: raise IdentityProfileError(message) def _copy_static_profile(template: Path, target: Path) -> list[str]: _require(template.is_dir(), "profile template is missing") _require(not target.exists(), "compiled profile target already exists") target.mkdir(parents=True, mode=0o700) copied: list[str] = [] for name in STATIC_PROFILE_ENTRIES: source = template / name if not source.exists(): continue destination = target / name _require(not source.is_symlink(), f"unsafe symlinked static profile entry: {name}") if source.is_dir(): _require( not any(path.is_symlink() for path in source.rglob("*")), f"unsafe symlink inside static profile entry: {name}", ) shutil.copytree(source, destination, symlinks=False) elif source.is_file() and not source.is_symlink(): if name == "config.yaml": try: raw_config = yaml.safe_load(source.read_text(encoding="utf-8")) or {} except (OSError, TypeError, yaml.YAMLError) as exc: raise IdentityProfileError(f"cannot sanitize profile config: {type(exc).__name__}") from exc _require(isinstance(raw_config, dict), "profile config must be a mapping") destination.write_text( yaml.safe_dump(behavior.identity_config_projection(raw_config), sort_keys=False), encoding="utf-8", ) destination.chmod(0o600) else: shutil.copy2(source, destination) else: raise IdentityProfileError(f"unsafe static profile entry: {name}") copied.append(name) _require("config.yaml" in copied, "profile template config.yaml is missing") return copied def _behavior(profile: Path, hermes_root: Path, deployment_root: Path) -> dict[str, Any]: return behavior.build_manifest(profile, hermes_root, deployment_root) def _verify_runtime_binding( manifest: dict[str, Any], *, profile: Path, hermes_root: Path, deployment_root: Path, ) -> dict[str, Any]: observed = _behavior(profile, hermes_root, deployment_root) identity.validate_behavior_manifest(observed) expected = manifest["identity_inputs"]["runtime"] _require( observed["identity_runtime_sha256"] == expected["identity_runtime_sha256"], "identity runtime drift detected" ) _require(observed["hermes_runtime"]["git_head"] == expected["hermes_git_head"], "Hermes Git drift detected") _require( observed["hermes_runtime"]["source_tree"]["sha256"] == expected["hermes_source_sha256"], "Hermes source drift detected", ) _require( observed["teleo_infrastructure_runtime"]["git_head"] == expected["teleo_git_head"], "Teleo Git drift detected", ) _require( observed["teleo_infrastructure_runtime"]["identity_source_tree"]["sha256"] == expected["teleo_source_sha256"], "Teleo identity source drift detected", ) _require(observed["python_runtime"] == expected["python"], "Python runtime drift detected") _require( manifest["identity_inputs"]["source_commit"] == observed["teleo_infrastructure_runtime"]["git_head"], "source commit drift detected", ) expected_model = manifest["identity_inputs"]["model"] observed_model = observed["model_runtime"] model_bindings = { "provider": observed_model.get("provider"), "default_model": observed_model.get("default_model"), "smart_routing": observed_model.get("smart_routing"), "smart_routing_model": observed_model.get("smart_routing_model"), "gateway_smart_model_routing": observed_model.get("gateway_smart_model_routing"), "model_section_sha256": observed_model.get("model_section_sha256"), "hosted_weights": observed_model.get("base_model_weights"), "actual_model_required_in_turn_receipt": observed_model.get( "actual_per_turn_model_must_be_recorded_by_the_call_receipt" ), } for field, observed_value in model_bindings.items(): _require(expected_model.get(field) == observed_value, f"model runtime binding drift detected: {field}") expected_permissions = manifest["identity_inputs"]["permissions"] permission_bindings = { "identity_config_sha256": observed_model.get("identity_config_sha256"), "gateway_permissions_sha256": observed_model.get("gateway_permissions_sha256"), "tool_permissions_sha256": observed_model.get("tool_permissions_sha256"), "runtime_policy": observed_model.get("identity_runtime_policy"), "authorization_decision_required_in_runtime_receipt": True, } for field, observed_value in permission_bindings.items(): _require( expected_permissions.get(field) == observed_value, f"permission runtime binding drift detected: {field}" ) observed_skills = observed["components"]["procedural_skills"]["content"] expected_skills = manifest["identity_inputs"]["skills"] _require(expected_skills["content_sha256"] == observed_skills["sha256"], "skills runtime binding drift detected") _require(expected_skills["file_count"] == observed_skills["file_count"], "skills file count drift detected") return observed def _view_file_sha256(profile: Path, name: str) -> str: path = profile / name _require(path.is_file() and not path.is_symlink(), f"compiled view is missing or unsafe: {name}") return behavior.file_sha256(path) def _verify_nonauthoritative_surfaces( profile: Path, *, require_empty_sessions: bool, require_compile_receipt: bool, ) -> dict[str, Any]: omitted = { name: not (profile / name).exists() and not (profile / name).is_symlink() for name in OMITTED_NONAUTHORITATIVE_ENTRIES } _require(all(omitted.values()), "profile contains a forbidden nonauthoritative input surface") expected_entries = set(COMPILED_PROFILE_ENTRIES) if not require_compile_receipt: expected_entries.remove("compile-receipt.json") profile_entries = list(profile.iterdir()) actual_entries = {path.name for path in profile_entries} _require(actual_entries == expected_entries, "compiled profile entry set is not exact") _require(not any(path.is_symlink() for path in profile_entries), "compiled profile entries must not be symlinks") if require_compile_receipt: compile_receipt = identity.load_json(profile / "compile-receipt.json", "identity compile receipt") _require(compile_receipt.get("schema") == COMPILE_RECEIPT_SCHEMA, "identity compile receipt schema is invalid") _require(compile_receipt.get("status") == "pass", "identity compile receipt status is invalid") state = profile / "state" sessions = profile / "sessions" for path, label in ((state, "state"), (sessions, "sessions")): _require(path.is_dir() and not path.is_symlink(), f"profile {label} directory is missing or unsafe") _require(not any(state.iterdir()), "profile state directory must remain empty") session_files = list(sessions.iterdir()) if require_empty_sessions: _require(not session_files, "profile sessions must start empty") for path in session_files: _require(path.is_file() and not path.is_symlink() and path.suffix == ".json", "unsafe session entry detected") record = identity.load_json(path, "temporary session record") _require( set(record) == { "schema", "authority", "classification", "query_sha256", "answer_sha256", "may_be_used_as_evidence", }, "temporary session record fields are invalid", ) _require(record.get("schema") == SESSION_RECORD_SCHEMA, "temporary session record schema is invalid") _require(record.get("authority") == "none", "temporary session authority is invalid") _require(record.get("classification") == "temporary_noncanonical", "temporary session class is invalid") _require(record.get("may_be_used_as_evidence") is False, "temporary session cannot be evidence") _require(identity._is_sha256(record.get("query_sha256")), "temporary session query hash is invalid") _require(identity._is_sha256(record.get("answer_sha256")), "temporary session answer hash is invalid") return { "nonauthoritative_inputs_omitted": omitted, "session_file_count": len(session_files), "state_empty": True, } def compile_profile( manifest: dict[str, Any], *, source_root: Path, profile_template: Path, profile: Path, hermes_root: Path, deployment_root: Path, ) -> dict[str, Any]: views = identity.verify_bound_sources(manifest, source_root) template_behavior = _behavior(profile_template, hermes_root, deployment_root) identity.validate_behavior_manifest(template_behavior) _require( template_behavior["identity_runtime_sha256"] == manifest["identity_inputs"]["runtime"]["identity_runtime_sha256"], "profile template does not match the pinned identity runtime", ) copied = _copy_static_profile(profile_template, profile) try: for name, content in views.items(): path = profile / name path.write_text(content, encoding="utf-8") path.chmod(0o600) identity.write_json(profile / "identity-manifest.json", manifest) (profile / "identity-manifest.json").chmod(0o600) for directory in (profile / "sessions", profile / "state"): directory.mkdir(mode=0o700) lock = { "schema": LOCK_SCHEMA, "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], "compiled_views": {name: {"sha256": _view_file_sha256(profile, name)} for name in sorted(views)}, "session_boundary": manifest["identity_inputs"]["session_boundary"], } identity.write_json(profile / "identity-lock.json", lock) (profile / "identity-lock.json").chmod(0o600) observed = _verify_runtime_binding( manifest, profile=profile, hermes_root=hermes_root, deployment_root=deployment_root, ) for name, expected in manifest["compiled_views"].items(): _require(_view_file_sha256(profile, name) == expected["sha256"], f"compiled {name} hash mismatch") soul_files = observed["components"]["runtime_identity"]["content"]["files"] soul = next((item for item in soul_files if item.get("path") == "SOUL.md"), None) _require( isinstance(soul, dict) and soul.get("sha256") == manifest["compiled_views"]["SOUL.md"]["sha256"], "SOUL runtime binding failed", ) surface_readback = _verify_nonauthoritative_surfaces( profile, require_empty_sessions=True, require_compile_receipt=False, ) receipt = { "schema": COMPILE_RECEIPT_SCHEMA, "status": "pass", "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], "profile_static_entries": copied, "compiled_view_sha256": { name: manifest["compiled_views"][name]["sha256"] for name in sorted(manifest["compiled_views"]) }, "runtime_identity_sha256": observed["identity_runtime_sha256"], "session_directories_started_empty": all( not any((profile / name).iterdir()) for name in ("sessions", "state") ), "nonauthoritative_inputs_omitted": surface_readback["nonauthoritative_inputs_omitted"], "generated_views_are_authority": False, } identity.write_json(profile / "compile-receipt.json", receipt) return receipt except BaseException: shutil.rmtree(profile, ignore_errors=True) raise def verify_profile( profile: Path, *, source_root: Path, hermes_root: Path, deployment_root: Path, ) -> tuple[dict[str, Any], dict[str, str], dict[str, Any]]: manifest = identity.load_json(profile / "identity-manifest.json", "compiled identity manifest") views = identity.verify_bound_sources(manifest, source_root) lock = identity.load_json(profile / "identity-lock.json", "identity profile lock") _verify_nonauthoritative_surfaces( profile, require_empty_sessions=False, require_compile_receipt=True, ) _require(lock.get("schema") == LOCK_SCHEMA, "identity profile lock schema is invalid") _require( lock.get("manifest_sha256") == manifest["manifest_sha256"], "identity profile lock manifest drift detected" ) _require( lock.get("identity_inputs_sha256") == manifest["identity_inputs_sha256"], "identity profile lock input drift detected", ) compile_receipt = identity.load_json(profile / "compile-receipt.json", "identity compile receipt") _require( compile_receipt.get("manifest_sha256") == manifest["manifest_sha256"], "identity compile receipt manifest drift detected", ) _require( compile_receipt.get("identity_inputs_sha256") == manifest["identity_inputs_sha256"], "identity compile receipt input drift detected", ) for name, expected_content in views.items(): expected_hash = behavior.sha256_bytes(expected_content.encode("utf-8")) _require(expected_hash == manifest["compiled_views"][name]["sha256"], f"manifest view drift detected: {name}") _require(_view_file_sha256(profile, name) == expected_hash, f"compiled view drift detected: {name}") _require( (lock.get("compiled_views") or {}).get(name, {}).get("sha256") == expected_hash, f"profile lock view drift detected: {name}", ) observed = _verify_runtime_binding( manifest, profile=profile, hermes_root=hermes_root, deployment_root=deployment_root, ) return manifest, views, observed def query_profile( profile: Path, *, query: str, source_root: Path, hermes_root: Path, deployment_root: Path, ) -> dict[str, Any]: _require(query == FIXED_QUERY, "only the pinned identity readback query is allowed") manifest, _views, observed = verify_profile( profile, source_root=source_root, hermes_root=hermes_root, deployment_root=deployment_root, ) structured = identity.load_json(profile / "identity.json", "compiled structured identity view") rule_count = len(structured["static_constitution"]["rules"]) record_count = len(structured["database_identity"]["records"]) fingerprint = structured["database_identity"]["database_fingerprint_sha256"] canonical_authority = structured["database_identity"]["canonical_authority_granted"] database_description = "canonical database" if canonical_authority else "synthetic noncanonical fixture" answer = ( f"Leo is defined by {rule_count} pinned static constitutional rules and {record_count} active " f"database-derived identity rows from a {database_description} at fingerprint {fingerprint}. " "SOUL.md is a generated view; " "runtime/session memory is temporary, noncanonical, and cannot be evidence." ) query_sha256 = behavior.sha256_bytes(query.encode("utf-8")) answer_sha256 = behavior.sha256_bytes(answer.encode("utf-8")) instance_id = uuid.uuid4().hex session_record = { "schema": SESSION_RECORD_SCHEMA, "authority": "none", "classification": "temporary_noncanonical", "query_sha256": query_sha256, "answer_sha256": answer_sha256, "may_be_used_as_evidence": False, } session_path = profile / "sessions" / f"{instance_id}.json" identity.write_json(session_path, session_record) receipt = { "schema": QUERY_RECEIPT_SCHEMA, "status": "pass", "started_at_utc": datetime.now(UTC).isoformat(), "process_id": os.getpid(), "runtime_instance_id": instance_id, "query": query, "query_sha256": query_sha256, "answer": answer, "answer_sha256": answer_sha256, "manifest_sha256": manifest["manifest_sha256"], "identity_inputs_sha256": manifest["identity_inputs_sha256"], "output_provenance": { "static_constitution_sha256": manifest["identity_inputs"]["identity_sources"]["static_constitution"][ "semantic_sha256" ], "database_identity_sha256": manifest["identity_inputs"]["identity_sources"]["database_derived_identity"][ "semantic_sha256" ], "database_fingerprint_sha256": fingerprint, "database_authority": structured["database_identity"]["authority"], "database_canonical_authority_granted": canonical_authority, "compiled_identity_sha256": manifest["compiled_views"]["identity.json"]["sha256"], "compiled_soul_sha256": manifest["compiled_views"]["SOUL.md"]["sha256"], "runtime_identity_sha256": observed["identity_runtime_sha256"], "actual_model_call_performed": False, }, "authorization": { "declared_runtime_policy": manifest["identity_inputs"]["permissions"]["runtime_policy"], "authorization_decision_observed": False, "claim": "local_policy_binding_not_an_authorizer_readback", }, "session": { "record_sha256": behavior.file_sha256(session_path), "classification": "temporary_noncanonical", "included_in_output_provenance": False, "may_be_used_as_evidence": False, }, } return receipt def main() -> int: parser = argparse.ArgumentParser(description=__doc__) subparsers = parser.add_subparsers(dest="command", required=True) compile_command = subparsers.add_parser("compile") compile_command.add_argument("--manifest", type=Path, required=True) compile_command.add_argument("--source-root", type=Path, required=True) compile_command.add_argument("--profile-template", type=Path, required=True) compile_command.add_argument("--profile", type=Path, required=True) compile_command.add_argument("--hermes-root", type=Path, required=True) compile_command.add_argument("--deployment-root", type=Path, required=True) query_command = subparsers.add_parser("query") query_command.add_argument("--profile", type=Path, required=True) query_command.add_argument("--query", default=FIXED_QUERY) query_command.add_argument("--source-root", type=Path, required=True) query_command.add_argument("--hermes-root", type=Path, required=True) query_command.add_argument("--deployment-root", type=Path, required=True) args = parser.parse_args() try: if args.command == "compile": result = compile_profile( identity.load_json(args.manifest, "identity manifest"), source_root=args.source_root, profile_template=args.profile_template, profile=args.profile, hermes_root=args.hermes_root, deployment_root=args.deployment_root, ) else: result = query_profile( args.profile, query=args.query, source_root=args.source_root, hermes_root=args.hermes_root, deployment_root=args.deployment_root, ) print(json.dumps(result, sort_keys=True)) return 0 except identity.IdentityManifestError as exc: print(json.dumps({"status": "error", "error": "identity_drift", "message": str(exc)})) return 3 if __name__ == "__main__": raise SystemExit(main())