#!/usr/bin/env python3 """Fingerprint every local runtime layer that can make Leo answer differently. The manifest deliberately records hashes and a small allowlist of non-secret configuration fields. It never emits profile file contents or credentials. Canonical Postgres state is outside this manifest and must be fingerprinted by the database harness that owns the connection. """ from __future__ import annotations import argparse import hashlib import importlib.metadata import json import os import platform import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Any import yaml SCHEMA = "livingip.leoBehaviorManifest.v1" DEFAULT_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean") DEFAULT_HERMES_ROOT = Path("/home/teleo/.hermes/hermes-agent") DEFAULT_DEPLOYMENT_ROOT = Path("/opt/teleo-eval/workspaces/deploy-infra") STATIC_RUNTIME_COMPONENTS = ( "profile_config", "procedural_skills", "runtime_middleware", "database_tools", ) IDENTITY_RUNTIME_COMPONENTS = ( "procedural_skills", "runtime_middleware", "database_tools", ) STABLE_MANIFEST_FIELDS = ( "schema", "model_runtime", "hermes_runtime", "teleo_infrastructure_runtime", "components", "python_runtime", "canonical_database", ) SECRET_KEYS = frozenset( { "access_key", "access_token", "api_key", "authorization", "bot_token", "client_secret", "credential", "credentials", "password", "private_key", "refresh_token", "secret", "token", } ) SECRET_KEY_SUFFIXES = ( "api_key", "access_key", "private_key", "password", "secret", "token", "credential", ) RUNTIME_DISTRIBUTIONS = ("PyYAML",) SAFE_CONFIG_SCALAR = re.compile(r"^[A-Za-z0-9._:/-]+$") def sha256_bytes(value: bytes) -> str: return hashlib.sha256(value).hexdigest() def canonical_sha256(value: Any) -> str: encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") return sha256_bytes(encoded) def file_sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def _safe_relative(path: Path, root: Path) -> str: try: return path.relative_to(root).as_posix() except ValueError: return str(path) def _resolved_node_fingerprint( path: Path, seen_directories: frozenset[tuple[int, int]] = frozenset() ) -> dict[str, Any]: """Hash a symlink target, following nested links while stopping directory cycles.""" try: if path.is_symlink(): target = os.readlink(path) resolved = path.resolve(strict=True) return { "kind": "symlink", "target": target, "resolved": _resolved_node_fingerprint(resolved, seen_directories), } stat = path.stat() if path.is_file(): return { "kind": "file", "bytes": stat.st_size, "mode": oct(stat.st_mode & 0o777), "sha256": file_sha256(path), } if path.is_dir(): identity = (stat.st_dev, stat.st_ino) if identity in seen_directories: return {"kind": "directory_cycle"} child_seen = seen_directories | {identity} children = [ {"name": child.name, "fingerprint": _resolved_node_fingerprint(child, child_seen)} for child in sorted(path.iterdir(), key=lambda item: item.name) ] return { "kind": "directory", "entry_count": len(children), "sha256": canonical_sha256(children), } return {"kind": "other", "mode": oct(stat.st_mode & 0o777)} except (OSError, RuntimeError) as exc: return {"kind": "unavailable", "error": type(exc).__name__} def path_manifest(profile: Path, paths: tuple[Path, ...]) -> dict[str, Any]: files: list[dict[str, Any]] = [] missing: list[str] = [] symlinks: list[dict[str, Any]] = [] for requested in paths: if not requested.exists() and not requested.is_symlink(): missing.append(_safe_relative(requested, profile)) continue candidates = [requested] if requested.is_file() or requested.is_symlink() else sorted(requested.rglob("*")) for path in candidates: relative = _safe_relative(path, profile) if path.is_symlink(): symlinks.append( { "path": relative, "target": os.readlink(path), "target_fingerprint": _resolved_node_fingerprint(path), } ) continue if not path.is_file(): continue try: stat = path.stat() files.append( { "path": relative, "bytes": stat.st_size, "mode": oct(stat.st_mode & 0o777), "sha256": file_sha256(path), } ) except OSError as exc: files.append({"path": relative, "status": "unavailable", "error": type(exc).__name__}) files.sort(key=lambda item: item["path"]) symlinks.sort(key=lambda item: item["path"]) stable = {"files": files, "missing": sorted(missing), "symlinks": symlinks} return { **stable, "file_count": len(files), "total_bytes": sum(int(item.get("bytes", 0)) for item in files), "sha256": canonical_sha256(stable), } def _nested(config: dict[str, Any], *path: str) -> Any: value: Any = config for key in path: value = value.get(key) if isinstance(value, dict) else None return value def secret_free_config(value: Any) -> Any: """Return a semantic config projection with secret-bearing fields omitted.""" if isinstance(value, dict): return { str(key): secret_free_config(item) for key, item in sorted(value.items(), key=lambda pair: str(pair[0])) if not _is_secret_key(str(key)) } if isinstance(value, list): return [secret_free_config(item) for item in value] return value def _validate_identity_config_value(value: Any, *, path: str) -> None: if value is None or isinstance(value, (bool, int)): return if isinstance(value, str): if not SAFE_CONFIG_SCALAR.fullmatch(value) or "://" in value or "@" in value: raise ValueError(f"unsafe identity config scalar: {path}") return if isinstance(value, list): for index, item in enumerate(value): _validate_identity_config_value(item, path=f"{path}[{index}]") return if isinstance(value, dict): for key, item in value.items(): if _is_secret_key(str(key)): raise ValueError(f"secret-bearing identity config key: {path}.{key}") _validate_identity_config_value(item, path=f"{path}.{key}") return raise ValueError(f"unsupported identity config value: {path}") def identity_config_projection(raw: dict[str, Any]) -> dict[str, Any]: """Build the only configuration surface copied into a disposable identity profile.""" section_fields = { "model": ("provider", "default", "smart_routing", "smart_routing_model", "max_tokens"), "agent": ("reasoning_effort",), "memory": ("enabled", "search"), "tools": ("allowed", "write_enabled"), "identity_runtime": ("network_send_enabled", "database_write_enabled", "allowed_tools"), } projection: dict[str, Any] = {} for section, fields in section_fields.items(): source = raw.get(section) if source is None: continue if not isinstance(source, dict): raise ValueError(f"identity config section must be a mapping: {section}") selected = {field: source[field] for field in fields if field in source} _validate_identity_config_value(selected, path=section) projection[section] = selected gateway = raw.get("gateway") if gateway is not None: if not isinstance(gateway, dict): raise ValueError("identity config section must be a mapping: gateway") selected_gateway = {key: gateway[key] for key in ("execution_mode",) if key in gateway} telegram = gateway.get("telegram") if telegram is not None: if not isinstance(telegram, dict): raise ValueError("identity config section must be a mapping: gateway.telegram") selected_telegram = {key: telegram[key] for key in ("posting_enabled",) if key in telegram} if selected_telegram: selected_gateway["telegram"] = selected_telegram _validate_identity_config_value(selected_gateway, path="gateway") projection["gateway"] = selected_gateway if "smart_model_routing" in raw: routing = raw["smart_model_routing"] if not isinstance(routing, bool): raise ValueError("smart_model_routing must be boolean") projection["smart_model_routing"] = routing return projection def _is_secret_key(key: str) -> bool: snake_case = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", "_", key) normalized = re.sub(r"[^a-z0-9]+", "_", snake_case.casefold()).strip("_") return normalized in SECRET_KEYS or any(normalized.endswith(f"_{suffix}") for suffix in SECRET_KEY_SUFFIXES) def python_runtime_manifest() -> dict[str, Any]: distributions: dict[str, str | None] = {} for name in RUNTIME_DISTRIBUTIONS: try: distributions[name] = importlib.metadata.version(name) except importlib.metadata.PackageNotFoundError: distributions[name] = None return { "implementation": platform.python_implementation(), "python_version": platform.python_version(), "abi_tag": sys.implementation.cache_tag, "runtime_distributions": distributions, "runtime_distributions_sha256": canonical_sha256(distributions), } def safe_model_config(config_path: Path) -> dict[str, Any]: try: raw = yaml.safe_load(config_path.read_text(encoding="utf-8")) or {} except (OSError, TypeError, yaml.YAMLError) as exc: return {"status": "unavailable", "error": type(exc).__name__} if not isinstance(raw, dict): return {"status": "unavailable", "error": "ConfigNotMapping"} semantic = secret_free_config(raw) try: identity_config = identity_config_projection(raw) except ValueError: return {"status": "unavailable", "error": "UnsafeIdentityConfig"} model_semantic = identity_config.get("model") or {} return { "status": "observed", "config_sha256": file_sha256(config_path), "semantic_config_sha256": canonical_sha256(semantic), "identity_config_sha256": canonical_sha256(identity_config), "model_section_sha256": canonical_sha256(model_semantic), "gateway_permissions_sha256": canonical_sha256(identity_config.get("gateway")), "tool_permissions_sha256": canonical_sha256(identity_config.get("tools")), "memory_section_sha256": canonical_sha256(identity_config.get("memory")), "agent_runtime_sha256": canonical_sha256(identity_config.get("agent")), "identity_runtime_policy": identity_config.get("identity_runtime"), "provider": _nested(identity_config, "model", "provider"), "default_model": _nested(identity_config, "model", "default"), "smart_routing": _nested(identity_config, "model", "smart_routing"), "smart_routing_model": _nested(identity_config, "model", "smart_routing_model"), "gateway_smart_model_routing": identity_config.get("smart_model_routing"), "max_tokens": _nested(identity_config, "model", "max_tokens"), "reasoning_effort": _nested(identity_config, "agent", "reasoning_effort"), "memory_enabled": _nested(identity_config, "memory", "enabled"), "memory_search": _nested(identity_config, "memory", "search"), } def git_head(repo: Path) -> str | None: try: completed = subprocess.run( ["git", "-C", str(repo), "rev-parse", "HEAD"], check=False, capture_output=True, text=True, timeout=10, env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, ) except (OSError, subprocess.TimeoutExpired): return None value = completed.stdout.strip() return value if completed.returncode == 0 and len(value) == 40 else None def git_state(repo: Path) -> dict[str, Any]: head = git_head(repo) try: completed = subprocess.run( ["git", "-C", str(repo), "status", "--porcelain", "--untracked-files=all"], check=False, capture_output=True, text=True, timeout=10, env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, ) except (OSError, subprocess.TimeoutExpired): return {"git_head": head, "worktree_clean": None, "status_sha256": None} status = completed.stdout if completed.returncode == 0 else "" return { "git_head": head, "worktree_clean": completed.returncode == 0 and not status.strip(), "status_sha256": sha256_bytes(status.encode("utf-8")) if completed.returncode == 0 else None, } def source_code_manifest( profile: Path, paths: tuple[Path, ...], *, source_root: Path | None = None, namespace: str = "source", ) -> dict[str, Any]: """Hash executable source while excluding generated caches and environments.""" allowed_suffixes = { ".cfg", ".ini", ".j2", ".jinja", ".json", ".py", ".sh", ".sql", ".tmpl", ".toml", ".yaml", ".yml", } excluded_parts = {".git", ".pytest_cache", "__pycache__", "node_modules", "venv", ".venv"} files: list[dict[str, Any]] = [] missing: list[str] = [] symlinks: list[str] = [] source_root = (source_root or profile).resolve() def source_label(path: Path) -> str: return f"{namespace}:{_safe_relative(path, source_root)}" for requested in paths: if not requested.exists() and not requested.is_symlink(): missing.append(source_label(requested)) continue candidates = [requested] if requested.is_file() else sorted(requested.rglob("*")) for path in candidates: if path.is_symlink(): symlinks.append(source_label(path)) continue if not path.is_file() or excluded_parts & set(path.parts): continue stat = path.stat() if path.suffix not in allowed_suffixes and not bool(stat.st_mode & 0o111): continue files.append( { "path": source_label(path), "bytes": stat.st_size, "mode": oct(stat.st_mode & 0o777), "sha256": file_sha256(path), } ) files.sort(key=lambda item: item["path"]) stable = {"files": files, "missing": sorted(missing), "symlinks": sorted(symlinks)} return { **stable, "file_count": len(files), "total_bytes": sum(int(item["bytes"]) for item in files), "sha256": canonical_sha256(stable), } def component( profile: Path, *, role: str, mutability: str, replayability: str, paths: tuple[Path, ...], ) -> dict[str, Any]: return { "role": role, "mutability": mutability, "replayability": replayability, "content": path_manifest(profile, paths), } def stable_manifest_payload(manifest: dict[str, Any]) -> dict[str, Any]: """Return the canonical, path-independent behavior payload.""" return {field: manifest.get(field) for field in STABLE_MANIFEST_FIELDS} def static_runtime_payload(manifest: dict[str, Any]) -> dict[str, Any]: """Return the exact payload committed by ``static_runtime_sha256``.""" components = manifest.get("components") or {} return { "schema": manifest.get("schema"), "model_runtime": manifest.get("model_runtime"), "hermes_runtime": manifest.get("hermes_runtime"), "teleo_infrastructure_runtime": manifest.get("teleo_infrastructure_runtime"), "components": {name: components.get(name) for name in STATIC_RUNTIME_COMPONENTS}, "python_runtime": manifest.get("python_runtime"), "canonical_database": manifest.get("canonical_database"), } def identity_runtime_payload(manifest: dict[str, Any]) -> dict[str, Any]: """Return the exact payload committed by ``identity_runtime_sha256``.""" model = manifest.get("model_runtime") or {} components = manifest.get("components") or {} teleo = manifest.get("teleo_infrastructure_runtime") or {} identity_model_runtime = { key: model.get(key) for key in ( "status", "identity_config_sha256", "model_section_sha256", "gateway_permissions_sha256", "tool_permissions_sha256", "memory_section_sha256", "agent_runtime_sha256", "identity_runtime_policy", "provider", "default_model", "smart_routing", "smart_routing_model", "gateway_smart_model_routing", "max_tokens", "reasoning_effort", "memory_enabled", "memory_search", "base_model_weights", "actual_per_turn_model_must_be_recorded_by_the_call_receipt", ) } return { "schema": manifest.get("schema"), "model_runtime": identity_model_runtime, "hermes_runtime": manifest.get("hermes_runtime"), "teleo_infrastructure_runtime": { "git_head": teleo.get("git_head"), "git_state": teleo.get("git_state"), "source_tree": teleo.get("identity_source_tree"), }, "components": {name: components.get(name) for name in IDENTITY_RUNTIME_COMPONENTS}, "python_runtime": manifest.get("python_runtime"), "canonical_database": manifest.get("canonical_database"), } def build_manifest( profile: Path = DEFAULT_PROFILE, hermes_root: Path = DEFAULT_HERMES_ROOT, deployment_root: Path = DEFAULT_DEPLOYMENT_ROOT, ) -> dict[str, Any]: profile = profile.resolve() hermes_root = hermes_root.resolve() deployment_root = deployment_root.resolve() config = safe_model_config(profile / "config.yaml") identity_runtime_sources = ( deployment_root / "scripts" / "leo_behavior_manifest.py", deployment_root / "scripts" / "leo_identity_manifest.py", deployment_root / "scripts" / "leo_identity_profile.py", deployment_root / "scripts" / "run_leo_identity_reconstruction_canary.py", ) components = { "profile_config": component( profile, role="Selects model routing, tools, memory, and gateway behavior.", mutability="deployment_static", replayability="hashable_but_may_contain_unpinned_provider_defaults", paths=(profile / "config.yaml",), ), "runtime_identity": component( profile, role="SOUL.md identity text injected into the Hermes system prompt.", mutability="file_static_until_edit_or_render", replayability="fully_hashable", paths=(profile / "SOUL.md",), ), "procedural_skills": component( profile, role="On-demand operating instructions available to the model.", mutability="deployment_static", replayability="fully_hashable", paths=(profile / "skills",), ), "runtime_middleware": component( profile, role="Profile plugins for pre-LLM context injection and post-LLM validation or response replacement.", mutability="deployment_static", replayability="fully_hashable", paths=(profile / "plugins",), ), "database_tools": component( profile, role="Commands that query Postgres, stage proposals, and return receipts.", mutability="deployment_static", replayability="fully_hashable", paths=(profile / "bin",), ), "persistent_memory": component( profile, role="Hermes MEMORY.md and USER.md content injected independently of canonical Postgres.", mutability="conversation_dynamic_noncanonical", replayability="hashable_but_should_not_define_durable_knowledge", paths=(profile / "memories",), ), "conversation_sessions": component( profile, role="Session transcripts and local state used for conversation continuity.", mutability="request_dynamic_noncanonical", replayability="hashable_but_not_a_knowledge_source", paths=( profile / "sessions", profile / "state", profile / "state.db", profile / "state.db-wal", profile / "state.db-shm", ), ), "generated_prompt_cache": component( profile, role="Hermes-generated cache of the skills prompt.", mutability="generated_cache", replayability="discard_and_rebuild", paths=(profile / ".skills_prompt_snapshot.json",), ), } stable = { "schema": SCHEMA, "model_runtime": { **config, "base_model_weights": "external_provider_managed_not_locally_hashable", "actual_per_turn_model_must_be_recorded_by_the_call_receipt": True, }, "hermes_runtime": { "git_head": git_head(hermes_root), "git_state": git_state(hermes_root), "source_tree": source_code_manifest( profile, ( hermes_root / "run_agent.py", hermes_root / "agent", hermes_root / "gateway", hermes_root / "hermes_cli", hermes_root / "tools", ), source_root=hermes_root, namespace="hermes", ), }, "teleo_infrastructure_runtime": { "git_head": git_head(deployment_root), "git_state": git_state(deployment_root), "source_tree": source_code_manifest( profile, (deployment_root,), source_root=deployment_root, namespace="teleo", ), "identity_source_tree": source_code_manifest( profile, identity_runtime_sources, source_root=deployment_root, namespace="teleo-identity", ), }, "components": components, "python_runtime": python_runtime_manifest(), "canonical_database": { "included": False, "reason": "Fingerprint through a read-only Postgres manifest in the database-owning harness.", }, } return { **stable, "generated_at_utc": datetime.now(timezone.utc).isoformat(), "profile_root": str(profile), "hermes_root": str(hermes_root), "deployment_root": str(deployment_root), "static_runtime_sha256": canonical_sha256(static_runtime_payload(stable)), "identity_runtime_sha256": canonical_sha256(identity_runtime_payload(stable)), "behavior_sha256": canonical_sha256(stable_manifest_payload(stable)), } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--profile", type=Path, default=DEFAULT_PROFILE) parser.add_argument("--hermes-root", type=Path, default=DEFAULT_HERMES_ROOT) parser.add_argument("--deployment-root", type=Path, default=DEFAULT_DEPLOYMENT_ROOT) parser.add_argument("--output", type=Path) args = parser.parse_args() manifest = build_manifest(args.profile, args.hermes_root, args.deployment_root) encoded = json.dumps(manifest, indent=2, sort_keys=True) + "\n" if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(encoded, encoding="utf-8") print(encoded, end="") return 0 if __name__ == "__main__": raise SystemExit(main())