feat: reproduce Leo identity across disposable restarts

This commit is contained in:
twentyOne2x 2026-07-15 03:17:38 +02:00
parent 626bac493e
commit 3afd1dbb5e
20 changed files with 3259 additions and 17 deletions

View file

@ -0,0 +1,206 @@
# Reproducible Leo identity
## Definition of Working
- **Working target:** generate one pinned Leo identity manifest, compile a
disposable profile, answer the fixed identity query in a fresh process,
restart in a second process, and reject any drift before an answer.
- **Operator path:** run
`python -m scripts.run_leo_identity_reconstruction_canary` with the committed
identity fixtures from a clean checkout.
- **Done means:** the T2 receipt reports two distinct stopped processes with the
same manifest, identity inputs, query, answer hash, and output provenance;
the second process starts from a newly compiled empty profile; the drift
attempt exits before answering; every process group and temporary profile is
removed.
- **Not done:** a hand-edited `SOUL.md`, a manifest self-hash, unit tests without
process restart, or prior VPS restart evidence that did not reconstruct from
this manifest.
- **Required tier:** `T2_runtime`. T3 VPS restart/readback is a post-merge
graduation target.
## Identity input inventory
`GatewayRunner` and the surrounding Hermes profile can change an answer through
the following surfaces. The manifest either pins each surface or explicitly
keeps it outside identity authority.
| Surface | Binding | Authority |
| --- | --- | --- |
| Model provider, route, limits, and reasoning settings | secret-free semantic config hash | runtime input; the actual model remains a per-turn receipt |
| Hosted model weights | provider-managed, explicitly not locally hashable | never claimed as a local hash |
| Hermes and Teleo code | clean Git commits plus executable source-tree hashes | runtime input |
| Python ABI and imported runtime packages | exact implementation/version and curated package versions | runtime input |
| Skills, plugins, and database tools | content hashes | runtime input |
| Gateway/tool permissions | strict allowlisted config hash plus exact no-send/read-only policy | runtime input; this local compiler does not claim an authorizer readback |
| Static constitutional rules | committed source path, byte hash, and semantic hash | static authority |
| Database snapshot version | database/user/system identity plus content, row, structure, and capture-provenance hashes; only WAL position is excluded | T2 pins a noncanonical fixture; it cannot grant canonical authority |
| Database-derived identity rows | typed table/row/kind/rank/evidence bindings plus source mode | synthetic fixtures are explicitly noncanonical; T3 requires independently verified database-exporter output |
| `SOUL.md` and `identity.json` | deterministic compiled-view hashes | generated views, never authorities |
| Sessions, state, and memory | excluded from the identity input hash and labeled temporary | continuity only; never evidence |
The disposable profile is compiled from a strict non-secret configuration
allowlist; unrecognized transport/credential fields are not copied. Credential
values are omitted rather than hashed. Rotating a bot token changes the broad
operational behavior observation but does not change Leo's identity.
Changing a non-secret permission, model setting, skill, code file, database
fingerprint, constitutional rule, database identity row, or compiled view does
change a pinned input and fails closed.
### Current `GatewayRunner` consumption map
The T2 compiler contract was checked against the current live runtime source.
This inventory defines what the post-merge T3 receipt must observe rather than
silently assuming that a profile file is the whole prompt:
- Startup resolves `-p leoclean` to `HERMES_HOME`, then loads profile and project
environment files before `config.yaml`; environment expansion, legacy
`gateway.json`, and defaults can change the effective configuration. The T2
compiler therefore enforces an exact top-level profile allowlist (including
rejection of dangling symlinks) rather than relying on a filename denylist.
- Routing consumes the requested default model, the top-level
`smart_model_routing` switch, provider endpoints/defaults, auth pool choice,
reasoning settings, token/turn limits, and fallbacks. The pinned top-level
routing switch must be a boolean; arbitrary nested routing data is rejected.
Credentials are runtime capability, not identity, and their values are never
retained.
- Prompt construction consumes generated `SOUL.md`, the gateway/agent system
message, persistent `MEMORY.md`/`USER.md`, compiled skills, project-context
files, current time/timezone, and the effective model/provider. T2 requires
memory and project context absent; T3 records the effective system-prompt and
compiled-skills-prompt hashes.
- Plugin discovery can include profile plugins, optional project plugins, and
installed entry points. The live database-context hook can inject a
query-bound contract before the model and can reject or replace the reply
afterward, so T3 retains plugin registry, contract, retrieval, and delivered
response hashes.
- Effective tools come from platform toolsets and the runtime registry, not a
descriptive fixture field. T3 must read back exact tool schemas, prove the
send tool absent, and keep terminal restricted to the clone-bound read-only
wrapper.
- Authorization consumes chat/user allowlists and pairing-store state. T2 starts
with pairing absent; T3 records the fixed synthetic source and the actual
authorization decision without exposing IDs or tokens.
- Session keys consume platform/chat/user/thread/topic fields. Auto-skill,
reply/media/file context, persisted transcripts, and a reused system prompt
can all change a turn. Verification therefore happens before every child
starts, and T3 binds session key, persisted session ID, message-context
absence, and prompt hashes.
The committed T2 database/identity inputs are synthetic fixtures and the
compiled view labels them `synthetic_noncanonical_fixture`; they do not stand
in for approved production rows. This T2 schema never grants canonical
authority from caller-supplied or merely self-hashed JSON. T3 must use output
that is independently verified by the database-owning same-transaction exporter
and bound to the database fingerprint, database user, system identifier, and
row digest.
Every pinned source tree and profile component is structurally revalidated:
the verifier recomputes its file count, total bytes, sorted unique paths, and
canonical content hash, and rejects missing files, unreadable entries, or
symlinks. Rehashing forged structural metadata cannot make it authoritative.
The current live Hermes checkout is not clean, so its Git SHA alone is not a
reproducible source bundle. This T2 receipt intentionally uses the committed
clean local runtime and the committed database/identity snapshot fixture. It
does not claim to reconstruct the current dirty VPS runtime. T3 must run only
after the merged identity code is deployed and must bind the deployed clean
content (or a content-addressed dirty-source bundle if the live checkout has not
yet been repaired).
## Commands
The complete canary is the preferred operator command:
```bash
python -m scripts.run_leo_identity_reconstruction_canary \
--profile-template fixtures/working-leo/leo-identity-v1/profile-template \
--hermes-root fixtures/working-leo/leo-identity-v1/hermes-runtime \
--deployment-root . \
--source-root . \
--database-fingerprint fixtures/working-leo/leo-identity-v1/leo-database-fingerprint-v1.json \
--constitution fixtures/working-leo/leo-identity-v1/leo-constitution-v1.json \
--database-identity fixtures/working-leo/leo-identity-v1/leo-database-identity-v1.json \
--output docs/reports/leo-working-state-20260709/leo-reproducible-identity-t2-current.json
```
The negative lifecycle is separately runnable. It proves drift is caught before
the second process starts:
```bash
python -m scripts.run_leo_identity_reconstruction_canary \
--profile-template fixtures/working-leo/leo-identity-v1/profile-template \
--hermes-root fixtures/working-leo/leo-identity-v1/hermes-runtime \
--deployment-root . \
--source-root . \
--database-fingerprint fixtures/working-leo/leo-identity-v1/leo-database-fingerprint-v1.json \
--constitution fixtures/working-leo/leo-identity-v1/leo-constitution-v1.json \
--database-identity fixtures/working-leo/leo-identity-v1/leo-database-identity-v1.json \
--inject-drift-before-restart \
--output /tmp/leo-identity-drift-rejection.json
```
For inspection, the lifecycle can be decomposed into explicit manifest and
compile commands:
```bash
python -m scripts.leo_behavior_manifest \
--profile fixtures/working-leo/leo-identity-v1/profile-template \
--hermes-root fixtures/working-leo/leo-identity-v1/hermes-runtime \
--deployment-root . \
--output /tmp/leo-behavior.json
python -m scripts.leo_identity_manifest generate \
--behavior-manifest /tmp/leo-behavior.json \
--database-fingerprint fixtures/working-leo/leo-identity-v1/leo-database-fingerprint-v1.json \
--constitution fixtures/working-leo/leo-identity-v1/leo-constitution-v1.json \
--database-identity fixtures/working-leo/leo-identity-v1/leo-database-identity-v1.json \
--source-root . \
--output /tmp/leo-identity-manifest.json
python -m scripts.leo_identity_profile compile \
--manifest /tmp/leo-identity-manifest.json \
--source-root . \
--profile-template fixtures/working-leo/leo-identity-v1/profile-template \
--profile /tmp/leo-disposable-profile \
--hermes-root fixtures/working-leo/leo-identity-v1/hermes-runtime \
--deployment-root .
```
All commands require a clean Git worktree because a commit plus an unretained
dirty source tree is not reproducible.
## State inventory and transitions
| State | Meaning | Next valid transition |
| --- | --- | --- |
| `inputs_unverified` | source paths exist but are not yet bound | validate clean Git, runtime, DB, rules, rows, and permissions |
| `manifest_pinned` | every material identity input has a stable hash | compile deterministic views |
| `profile_compiled` | static profile copied and generated views match the manifest | verify immediately before start |
| `runtime_verified` | runtime/code/view hashes match and session state is non-authoritative | answer fixed query |
| `stopped_cleanly` | child and process group are absent | compile a new empty profile from the manifest |
| `restart_reproduced` | child from the newly compiled profile has the same identity/query/provenance inputs | retain receipt and clean every profile |
| `drift_detected` | any input or generated view differs | fail closed; no answer and no next start |
The irreversible boundary is not a database or production mutation: this T2
canary is a local compiler/process runtime, no-send, database-read-only, and
disposable. The successful-restart receipt reaches `T2_runtime`; the standalone
pre-restart drift receipt is a passing negative component but reports
`T1_model` and `goal_tier_satisfied=false`. Neither proves the live VPS
`GatewayRunner`, Telegram delivery, hosted-model behavior, production database
state, or T3 restart recovery.
Here `T2_runtime` uses the required Capability Tier Proof vocabulary: local
runtime behavior with restart. It does not imply Hermes/GatewayRunner or a
hosted model; those exclusions are explicit in the receipt's `runtime_variant`,
`tier_basis`, and `claim_ceiling`.
## T3 graduation
After merge and rollback proof, extend the schema with independent verification
of the database-owning same-transaction exporter, generate the manifest from
that live read-only canonical capture and deployed clean commits, compile a
disposable VPS profile, invoke the real no-send `GatewayRunner`, terminate that
child, open a new child from the same manifest, and retain model-call, DB-read,
service, cleanup, and drift-negative readbacks. Do not replace the production
profile during that graduation.

View file

@ -0,0 +1,3 @@
"""Pinned fixture prompt boundary."""
GENERATED_SOUL_IS_AUTHORITY = False

View file

@ -0,0 +1,3 @@
"""Pinned fixture process boundary; the canary uses the real profile verifier."""
SESSION_STATE_IS_IDENTITY = False

View file

@ -0,0 +1,3 @@
"""Pinned fixture tool configuration."""
ALLOWED_TOOLS = ("identity_readback",)

View file

@ -0,0 +1,3 @@
"""Pinned fixture entrypoint for the disposable identity runtime."""
RUNTIME_NAME = "leo-identity-readback-v1"

View file

@ -0,0 +1,3 @@
"""Pinned fixture registry for the no-send identity readback tool."""
REGISTERED_TOOLS = ("identity_readback",)

View file

@ -0,0 +1,18 @@
{
"identity_name": "Leo",
"rules": [
{
"id": "canonical-evidence",
"text": "Treat canonical database rows and their bound evidence as durable knowledge; never promote session memory to evidence."
},
{
"id": "review-before-apply",
"text": "Proposals may be prepared, but review and guarded apply remain separate authorized actions."
},
{
"id": "truthful-proof-tier",
"text": "State exactly what was observed, separate inference from proof, and fail closed when required provenance drifts."
}
],
"schema": "livingip.leoConstitutionSource.v1"
}

View file

@ -0,0 +1,25 @@
{
"environment": "disposable_fixture",
"fingerprint_sha256": "1111111111111111111111111111111111111111111111111111111111111111",
"read_consistency": {
"after": {
"database": "leo_identity_fixture",
"database_user": "leo_identity_reader",
"system_identifier": "7649789040005668902",
"wal_lsn": "0/16B6C50"
},
"before": {
"database": "leo_identity_fixture",
"database_user": "leo_identity_reader",
"system_identifier": "7649789040005668902",
"wal_lsn": "0/16B6C50"
},
"status": "stable_wal_marker"
},
"schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
"status": "ok",
"structure_sha256": "3333333333333333333333333333333333333333333333333333333333333333",
"table_count": 7,
"table_rows_sha256": "2222222222222222222222222222222222222222222222222222222222222222",
"total_rows": 7
}

View file

@ -0,0 +1,58 @@
{
"database_fingerprint_sha256": "1111111111111111111111111111111111111111111111111111111111111111",
"identity_name": "Leo",
"source_provenance": {
"canonical_authority_granted": false,
"export_receipt_sha256": null,
"mode": "synthetic_noncanonical_fixture",
"same_transaction_as_fingerprint": false
},
"records": [
{
"evidence_sha256": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
"kind": "persona",
"rank": 0,
"row_id": "11111111-1111-4111-8111-111111111111",
"status": "active",
"table": "public.personas",
"text": "Leo is the evidence-bound reasoning interface for the Teleo collective."
},
{
"evidence_sha256": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb",
"kind": "strategy",
"rank": 0,
"row_id": "22222222-2222-4222-8222-222222222222",
"status": "active",
"table": "public.strategies",
"text": "Prefer source-grounded synthesis that exposes uncertainty and review boundaries."
},
{
"evidence_sha256": "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc",
"kind": "belief",
"rank": 0,
"row_id": "33333333-3333-4333-8333-333333333333",
"status": "active",
"table": "public.beliefs",
"text": "A plausible answer without provenance is weaker than a bounded answer with an exact receipt."
},
{
"evidence_sha256": "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd",
"kind": "shared_root",
"rank": 0,
"row_id": "44444444-4444-4444-8444-444444444444",
"status": "active",
"table": "public.shared_root",
"text": "Collective intelligence must remain inspectable, reversible, and accountable to evidence."
},
{
"evidence_sha256": "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee",
"kind": "strategy_node",
"rank": 1,
"row_id": "55555555-5555-4555-8555-555555555555",
"status": "active",
"table": "public.strategy_nodes",
"text": "Compile identity from exact canonical inputs and reject drift before runtime startup."
}
],
"schema": "livingip.leoDatabaseIdentitySource.v1"
}

View file

@ -0,0 +1,5 @@
#!/usr/bin/env python3
"""Fixture marker for the no-send identity readback tool."""
ACCESS_MODE = "read_only"
NETWORK_SEND_ENABLED = False

View file

@ -0,0 +1,24 @@
model:
provider: fixture-local
default: leo-identity-readback-v1
smart_routing: false
max_tokens: 512
agent:
reasoning_effort: deterministic
memory:
enabled: false
search: disabled
gateway:
execution_mode: local_no_send
telegram:
posting_enabled: false
bot_token: fixture-value-is-never-emitted-or-identity-hashed
tools:
allowed:
- identity_readback
write_enabled: false
identity_runtime:
network_send_enabled: false
database_write_enabled: false
allowed_tools:
- identity_readback

View file

@ -0,0 +1,4 @@
"""Fixture marker for the pinned identity-boundary middleware."""
IDENTITY_VIEW_IS_AUTHORITY = False
SESSION_MEMORY_IS_EVIDENCE = False

View file

@ -0,0 +1,3 @@
name: identity-boundary
version: 1
description: Reject startup when a compiled identity view drifts.

View file

@ -0,0 +1,8 @@
---
name: identity-readback
description: Explain the pinned Leo identity and its noncanonical session boundary.
---
Read only the compiled identity view and its manifest receipt. Never treat a
session transcript, generated SOUL file, or unverified runtime memory as a
canonical source.

View file

@ -11,9 +11,13 @@ 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
@ -24,6 +28,54 @@ 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:
@ -50,7 +102,9 @@ def _safe_relative(path: Path, root: Path) -> str:
return str(path)
def _resolved_node_fingerprint(path: Path, seen_directories: frozenset[tuple[int, int]] = frozenset()) -> dict[str, Any]:
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:
@ -141,21 +195,139 @@ def _nested(config: dict[str, Any], *path: str) -> Any:
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",
"provider": _nested(raw, "model", "provider"),
"default_model": _nested(raw, "model", "default"),
"smart_routing": _nested(raw, "model", "smart_routing"),
"smart_routing_model": _nested(raw, "model", "smart_routing_model"),
"max_tokens": _nested(raw, "model", "max_tokens"),
"reasoning_effort": _nested(raw, "agent", "reasoning_effort"),
"memory_enabled": _nested(raw, "memory", "enabled"),
"memory_search": _nested(raw, "memory", "search"),
"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"),
}
@ -196,7 +368,13 @@ def git_state(repo: Path) -> dict[str, Any]:
}
def source_code_manifest(profile: Path, paths: tuple[Path, ...]) -> dict[str, Any]:
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 = {
@ -216,12 +394,21 @@ def source_code_manifest(profile: Path, paths: tuple[Path, ...]) -> dict[str, An
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(_safe_relative(requested, profile))
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()
@ -229,14 +416,14 @@ def source_code_manifest(profile: Path, paths: tuple[Path, ...]) -> dict[str, An
continue
files.append(
{
"path": _safe_relative(path, profile),
"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)}
stable = {"files": files, "missing": sorted(missing), "symlinks": sorted(symlinks)}
return {
**stable,
"file_count": len(files),
@ -261,6 +448,72 @@ def component(
}
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,
@ -270,6 +523,12 @@ def build_manifest(
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,
@ -294,10 +553,10 @@ def build_manifest(
),
"runtime_middleware": component(
profile,
role="Pre-LLM context injection and post-LLM validation or response replacement.",
role="Profile plugins for pre-LLM context injection and post-LLM validation or response replacement.",
mutability="deployment_static",
replayability="fully_hashable",
paths=(profile / "plugins", hermes_root / "run_agent.py"),
paths=(profile / "plugins",),
),
"database_tools": component(
profile,
@ -353,14 +612,28 @@ def build_manifest(
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_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.",
@ -372,7 +645,9 @@ def build_manifest(
"profile_root": str(profile),
"hermes_root": str(hermes_root),
"deployment_root": str(deployment_root),
"behavior_sha256": canonical_sha256(stable),
"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)),
}

View file

@ -0,0 +1,878 @@
#!/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 = [
"<!-- GENERATED FILE: edit the pinned sources, never this view. -->",
"# 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())

View file

@ -0,0 +1,495 @@
#!/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())

View file

@ -0,0 +1,439 @@
#!/usr/bin/env python3
"""Prove manifest-driven Leo identity reconstruction across a cold restart."""
from __future__ import annotations
import argparse
import json
import os
import shutil
import signal
import subprocess
import sys
import tempfile
import time
from datetime import UTC, datetime
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
from scripts import leo_identity_manifest as identity
from scripts import leo_identity_profile as profile_runtime
SCHEMA = "livingip.leoIdentityReconstructionReceipt.v1"
class CanaryError(RuntimeError):
"""The disposable identity runtime did not satisfy its T2 contract."""
def _require(condition: bool, message: str) -> None:
if not condition:
raise CanaryError(message)
def _process_group_absent(process_group: int) -> bool:
try:
os.killpg(process_group, 0)
except ProcessLookupError:
return True
except PermissionError:
return False
return False
def _wait_for_process_group_absence(process_group: int, timeout_seconds: float = 5) -> bool:
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
if _process_group_absent(process_group):
return True
time.sleep(0.02)
return _process_group_absent(process_group)
def _signal_process_group(process_group: int, signal_number: signal.Signals) -> bool:
try:
os.killpg(process_group, signal_number)
return True
except (ProcessLookupError, PermissionError):
return False
def _cleanup_remaining_process_group(process_group: int) -> str | None:
if not _signal_process_group(process_group, signal.SIGTERM):
return None
if _wait_for_process_group_absence(process_group, timeout_seconds=1):
return "SIGTERM"
if not _signal_process_group(process_group, signal.SIGKILL):
return "SIGTERM"
return "SIGKILL"
def _run_query_child(
*,
deployment_root: Path,
profile: Path,
source_root: Path,
hermes_root: Path,
timeout_seconds: float = 60,
) -> dict[str, Any]:
command = [
sys.executable,
"-m",
"scripts.leo_identity_profile",
"query",
"--profile",
str(profile),
"--source-root",
str(source_root),
"--hermes-root",
str(hermes_root),
"--deployment-root",
str(deployment_root),
"--query",
profile_runtime.FIXED_QUERY,
]
process = subprocess.Popen(
command,
cwd=deployment_root,
env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"},
text=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
start_new_session=True,
)
timed_out = False
terminated_with: str | None = None
stdout = ""
stderr = ""
process_group_absent_after_wait = False
orphan_cleanup_required = False
try:
try:
stdout, stderr = process.communicate(timeout=timeout_seconds)
except subprocess.TimeoutExpired:
timed_out = True
if _signal_process_group(process.pid, signal.SIGTERM):
terminated_with = "SIGTERM"
try:
stdout, stderr = process.communicate(timeout=5)
except subprocess.TimeoutExpired:
if _signal_process_group(process.pid, signal.SIGKILL):
terminated_with = "SIGKILL"
stdout, stderr = process.communicate(timeout=5)
finally:
if process.poll() is None:
if _signal_process_group(process.pid, signal.SIGTERM):
terminated_with = terminated_with or "SIGTERM"
try:
process.communicate(timeout=5)
except subprocess.TimeoutExpired:
if _signal_process_group(process.pid, signal.SIGKILL):
terminated_with = "SIGKILL"
process.communicate(timeout=5)
orphan_cleanup_required = not _process_group_absent(process.pid)
if orphan_cleanup_required:
terminated_with = _cleanup_remaining_process_group(process.pid) or terminated_with
process_group_absent_after_wait = _wait_for_process_group_absence(process.pid)
try:
payload = json.loads(stdout.strip().splitlines()[-1]) if stdout.strip() else {}
except json.JSONDecodeError as exc:
raise CanaryError("identity child returned invalid JSON") from exc
return {
"command": command,
"launcher_pid": process.pid,
"returncode": process.returncode,
"timed_out": timed_out,
"terminated_with": terminated_with,
"orphan_cleanup_required": orphan_cleanup_required,
"stdout": payload,
"stderr_sha256": behavior.sha256_bytes(stderr.encode("utf-8")),
"process_group_absent_after_wait": process_group_absent_after_wait,
}
def _query_passed(child: dict[str, Any]) -> bool:
payload = child.get("stdout") or {}
return bool(
child.get("returncode") == 0
and child.get("timed_out") is False
and child.get("orphan_cleanup_required") is False
and child.get("process_group_absent_after_wait") is True
and payload.get("schema") == profile_runtime.QUERY_RECEIPT_SCHEMA
and payload.get("status") == "pass"
)
def _drift_attempt(
*,
deployment_root: Path,
source_root: Path,
hermes_root: Path,
drift_profile: Path,
) -> dict[str, Any]:
with (drift_profile / "SOUL.md").open("a", encoding="utf-8") as handle:
handle.write("\nInjected drift must fail closed.\n")
child = _run_query_child(
deployment_root=deployment_root,
profile=drift_profile,
source_root=source_root,
hermes_root=hermes_root,
)
child["fail_closed"] = bool(
child["returncode"] == 3
and child["timed_out"] is False
and child["orphan_cleanup_required"] is False
and child["process_group_absent_after_wait"] is True
and (child.get("stdout") or {}).get("error") == "identity_drift"
and "compiled view drift detected" in str((child.get("stdout") or {}).get("message"))
)
child["answer_returned"] = bool((child.get("stdout") or {}).get("answer"))
return child
def run_canary(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
output = args.output.resolve()
temporary_root = Path(tempfile.mkdtemp(prefix="leo-identity-reconstruction-"))
report: dict[str, Any] = {
"schema": SCHEMA,
"generated_at_utc": datetime.now(UTC).isoformat(),
"required_tier": "T2_runtime",
"mode": "drift_before_restart"
if args.inject_drift_before_restart
else "successful_restart_plus_drift_negative",
"fixed_query": profile_runtime.FIXED_QUERY,
"fixed_query_sha256": behavior.sha256_bytes(profile_runtime.FIXED_QUERY.encode("utf-8")),
"production_profile_replaced": False,
"production_database_contacted": False,
"telegram_message_sent": False,
"actual_hosted_model_call_performed": False,
"errors": [],
}
exit_code = 1
try:
behavior_manifest = behavior.build_manifest(args.profile_template, args.hermes_root, args.deployment_root)
identity.validate_behavior_manifest(behavior_manifest)
manifest = identity.build_identity_manifest(
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,
)
report["manifest"] = manifest
report["behavior_observation_before_compile"] = {
"behavior_sha256": behavior_manifest["behavior_sha256"],
"identity_runtime_sha256": behavior_manifest["identity_runtime_sha256"],
"source_commit": behavior_manifest["teleo_infrastructure_runtime"]["git_head"],
}
compiled_profile = temporary_root / "first-profile"
report["compile"] = profile_runtime.compile_profile(
manifest,
source_root=args.source_root,
profile_template=args.profile_template,
profile=compiled_profile,
hermes_root=args.hermes_root,
deployment_root=args.deployment_root,
)
compiled_behavior_before_query = behavior.build_manifest(
compiled_profile, args.hermes_root, args.deployment_root
)
report["compiled_profile_before_query"] = {
"behavior_sha256": compiled_behavior_before_query["behavior_sha256"],
"identity_runtime_sha256": compiled_behavior_before_query["identity_runtime_sha256"],
}
first = _run_query_child(
deployment_root=args.deployment_root,
profile=compiled_profile,
source_root=args.source_root,
hermes_root=args.hermes_root,
)
report["first_start"] = first
_require(_query_passed(first), "first disposable identity process did not answer successfully")
after_first = behavior.build_manifest(compiled_profile, args.hermes_root, args.deployment_root)
report["session_boundary_readback"] = {
"full_behavior_changed_after_session": after_first["behavior_sha256"]
!= compiled_behavior_before_query["behavior_sha256"],
"identity_runtime_unchanged_after_session": after_first["identity_runtime_sha256"]
== compiled_behavior_before_query["identity_runtime_sha256"],
"session_file_count": len(list((compiled_profile / "sessions").glob("*.json"))),
"session_classification": "temporary_noncanonical",
"session_used_as_output_provenance": False,
}
if args.inject_drift_before_restart:
with (compiled_profile / "SOUL.md").open("a", encoding="utf-8") as handle:
handle.write("\nInjected pre-restart drift.\n")
rejected = _run_query_child(
deployment_root=args.deployment_root,
profile=compiled_profile,
source_root=args.source_root,
hermes_root=args.hermes_root,
)
report["pre_restart_drift"] = rejected
report["drift_target"] = "compiled SOUL.md generated view"
report["second_start_attempted"] = False
report["gates"] = {
"first_start_passed": True,
"first_process_stopped": first["process_group_absent_after_wait"],
"drift_detected_before_restart": rejected["returncode"] == 3
and (rejected.get("stdout") or {}).get("error") == "identity_drift",
"drift_returned_no_answer": not bool((rejected.get("stdout") or {}).get("answer")),
"drift_process_stopped": rejected["process_group_absent_after_wait"],
"drift_process_had_no_orphan_cleanup": rejected["orphan_cleanup_required"] is False,
"second_start_not_attempted": True,
}
report["status"] = "pass" if all(report["gates"].values()) else "fail"
else:
profile_runtime.verify_profile(
compiled_profile,
source_root=args.source_root,
hermes_root=args.hermes_root,
deployment_root=args.deployment_root,
)
report["pre_restart_verification"] = "pass"
report["second_start_attempted"] = True
restarted_profile = temporary_root / "restart-profile"
report["restart_compile"] = profile_runtime.compile_profile(
manifest,
source_root=args.source_root,
profile_template=args.profile_template,
profile=restarted_profile,
hermes_root=args.hermes_root,
deployment_root=args.deployment_root,
)
profile_runtime.verify_profile(
restarted_profile,
source_root=args.source_root,
hermes_root=args.hermes_root,
deployment_root=args.deployment_root,
)
second = _run_query_child(
deployment_root=args.deployment_root,
profile=restarted_profile,
source_root=args.source_root,
hermes_root=args.hermes_root,
)
report["restart"] = second
_require(_query_passed(second), "restarted disposable identity process did not answer successfully")
first_payload = first["stdout"]
second_payload = second["stdout"]
drift_profile = temporary_root / "drift-profile"
report["drift_compile"] = profile_runtime.compile_profile(
manifest,
source_root=args.source_root,
profile_template=args.profile_template,
profile=drift_profile,
hermes_root=args.hermes_root,
deployment_root=args.deployment_root,
)
drift = _drift_attempt(
deployment_root=args.deployment_root,
source_root=args.source_root,
hermes_root=args.hermes_root,
drift_profile=drift_profile,
)
report["drift_negative"] = drift
report["drift_target"] = "compiled SOUL.md generated view"
report["gates"] = {
"manifest_generated": identity._is_sha256(manifest["manifest_sha256"]),
"views_compiled_and_bound": report["compile"]["status"] == "pass",
"first_start_passed": True,
"first_process_stopped": first["process_group_absent_after_wait"],
"pre_restart_verification_passed": True,
"restart_profile_recompiled_from_manifest": report["restart_compile"]["status"] == "pass",
"restart_profile_started_without_sessions": report["restart_compile"][
"session_directories_started_empty"
],
"restart_passed": True,
"restart_process_is_distinct": first_payload["process_id"] != second_payload["process_id"]
and first_payload["runtime_instance_id"] != second_payload["runtime_instance_id"],
"identity_inputs_reproduced": first_payload["identity_inputs_sha256"]
== second_payload["identity_inputs_sha256"],
"manifest_reproduced": first_payload["manifest_sha256"] == second_payload["manifest_sha256"],
"query_reproduced": first_payload["query_sha256"] == second_payload["query_sha256"],
"answer_and_provenance_explainable": first_payload["answer_sha256"] == second_payload["answer_sha256"]
and first_payload["output_provenance"] == second_payload["output_provenance"],
"session_excluded_from_identity": report["session_boundary_readback"][
"identity_runtime_unchanged_after_session"
],
"drift_failed_closed": drift["fail_closed"] and not drift["answer_returned"],
"drift_profile_started_without_sessions": report["drift_compile"]["session_directories_started_empty"],
"drift_process_stopped": drift["process_group_absent_after_wait"],
"drift_process_had_no_orphan_cleanup": drift["orphan_cleanup_required"] is False,
"restart_process_stopped": second["process_group_absent_after_wait"],
}
report["status"] = "pass" if all(report["gates"].values()) else "fail"
if report["status"] == "pass":
exit_code = 0
except (CanaryError, identity.IdentityManifestError, OSError, subprocess.SubprocessError) as exc:
report["status"] = "fail"
report["errors"].append({"type": type(exc).__name__, "message": str(exc)})
finally:
shutil.rmtree(temporary_root, ignore_errors=True)
report["cleanup"] = {
"temporary_root_removed": not temporary_root.exists(),
"no_profile_retained": not temporary_root.exists(),
}
if not all(report["cleanup"].values()):
report["status"] = "fail"
exit_code = 1
positive_restart_passed = report.get("status") == "pass" and not args.inject_drift_before_restart
report["current_tier"] = "T2_runtime" if positive_restart_passed else "T1_model"
report["goal_tier_satisfied"] = positive_restart_passed
report["runtime_component_proven"] = (
"fresh_profile_recompile_plus_distinct_process_restart"
if positive_restart_passed
else "first_local_process_plus_pre_restart_drift_refusal"
)
report["runtime_variant"] = "local_deterministic_identity_compiler_readback"
report["tier_basis"] = (
"Capability Tier Proof defines T2_runtime as local API/runtime behavior with restart where relevant; "
"this is compiler-process runtime proof and explicitly excludes GatewayRunner and hosted-model proof."
if positive_restart_passed
else "This negative component does not complete the required restart path, so it remains below T2_runtime."
)
report["claim_ceiling"] = (
"Local disposable identity compilation, fresh-profile reconstruction, and cold-process restart only; "
"not a live GatewayRunner, hosted-model, Telegram, production database, VPS restart, or T3 proof."
if positive_restart_passed
else "Local first-process and pre-restart drift-refusal component only; no successful restart proof, "
"GatewayRunner, hosted model, production database, VPS, or T3 claim."
)
stable = {key: value for key, value in report.items() if key != "receipt_sha256"}
report["receipt_sha256"] = behavior.canonical_sha256(stable)
identity.write_json(output, report)
return report, exit_code
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--profile-template", type=Path, required=True)
parser.add_argument("--hermes-root", type=Path, required=True)
parser.add_argument("--deployment-root", type=Path, required=True)
parser.add_argument("--source-root", type=Path, required=True)
parser.add_argument("--database-fingerprint", type=Path, required=True)
parser.add_argument("--constitution", type=Path, required=True)
parser.add_argument("--database-identity", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("--inject-drift-before-restart", action="store_true")
args = parser.parse_args()
report, exit_code = run_canary(args)
print(
json.dumps(
{
"status": report.get("status"),
"current_tier": report.get("current_tier"),
"receipt_sha256": report.get("receipt_sha256"),
"output": str(args.output),
},
sort_keys=True,
)
)
return exit_code
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -4,6 +4,8 @@ import json
import subprocess
from pathlib import Path
import pytest
from scripts import leo_behavior_manifest as manifest
ROOT = Path(__file__).resolve().parents[1]
@ -58,6 +60,11 @@ gateway:
assert '{"private":"conversation"}' not in encoded
def test_identity_config_rejects_nonboolean_smart_model_routing() -> None:
with pytest.raises(ValueError, match="smart_model_routing must be boolean"):
manifest.identity_config_projection({"smart_model_routing": {"route": "untrusted"}})
def test_behavior_manifest_changes_when_a_behavior_layer_changes(tmp_path: Path) -> None:
profile = tmp_path / "profile"
hermes = tmp_path / "hermes"
@ -156,3 +163,64 @@ def test_git_state_marks_untracked_files_dirty(tmp_path: Path) -> None:
assert state["worktree_clean"] is False
assert len(str(state["status_sha256"])) == 64
def test_identity_runtime_omits_secret_values_but_pins_semantic_model_settings(tmp_path: Path) -> None:
profile = tmp_path / "profile"
hermes = tmp_path / "hermes"
deployment = tmp_path / "deployment"
config = """model:
provider: fixture
default: identity-v1
max_tokens: 512
gateway:
telegram:
bot_token: token-a
botToken: camel-token-a
provider:
apiKey: camel-api-key-a
transport:
endpoint: https://fixture-user:fixture-password-a@example.invalid/v1
headers: ['Authorization: Bearer fixture-header-a']
tools:
allowed: [identity_readback]
identity_runtime:
network_send_enabled: false
database_write_enabled: false
allowed_tools: [identity_readback]
"""
_write(profile / "config.yaml", config)
_write(profile / "skills" / "identity" / "SKILL.md", "identity\n")
_write(profile / "plugins" / "identity" / "__init__.py", "BOUNDARY = True\n")
_write(profile / "bin" / "identity.py", "READ_ONLY = True\n")
_write(hermes / "run_agent.py", "runtime\n")
_write(hermes / "agent" / "prompt_builder.py", "prompts\n")
for name in (
"leo_behavior_manifest.py",
"leo_identity_manifest.py",
"leo_identity_profile.py",
"run_leo_identity_reconstruction_canary.py",
):
_write(deployment / "scripts" / name, f"# {name}\n")
before = manifest.build_manifest(profile, hermes, deployment)
_write(
profile / "config.yaml",
config.replace("token-a", "token-b")
.replace("camel-api-key-a", "camel-api-key-b")
.replace("fixture-password-a", "fixture-password-b")
.replace("fixture-header-a", "fixture-header-b"),
)
secret_rotated = manifest.build_manifest(profile, hermes, deployment)
_write(profile / "config.yaml", config.replace("max_tokens: 512", "max_tokens: 1024"))
model_changed = manifest.build_manifest(profile, hermes, deployment)
assert before["behavior_sha256"] != secret_rotated["behavior_sha256"]
assert before["identity_runtime_sha256"] == secret_rotated["identity_runtime_sha256"]
assert before["identity_runtime_sha256"] != model_changed["identity_runtime_sha256"]
encoded = json.dumps(before)
assert "token-a" not in encoded
assert "camel-token-a" not in encoded
assert "camel-api-key-a" not in encoded
assert "fixture-password-a" not in encoded
assert "fixture-header-a" not in encoded

View file

@ -0,0 +1,721 @@
from __future__ import annotations
import argparse
import copy
import json
import os
import subprocess
import time
from pathlib import Path
import pytest
from scripts import leo_behavior_manifest as behavior
from scripts import leo_identity_manifest as identity
from scripts import leo_identity_profile as profile_runtime
from scripts import run_leo_identity_reconstruction_canary as canary
ROOT = Path(__file__).resolve().parents[1]
def _write(path: Path, value: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(value, encoding="utf-8")
def _write_json(path: Path, value: dict) -> None:
_write(path, json.dumps(value, indent=2, sort_keys=True) + "\n")
def _fixture(tmp_path: Path) -> dict[str, Path | dict]:
repo = tmp_path / "repo"
profile = repo / "profile-template"
hermes = repo / "hermes-runtime"
compiled = tmp_path / "compiled-profile"
_write(
profile / "config.yaml",
"""model:
provider: fixture-local
default: identity-v1
max_tokens: 512
memory:
enabled: false
gateway:
execution_mode: local_no_send
telegram:
bot_token: never-emit-this-fixture-value
transport:
endpoint: https://fixture-user:fixture-password@example.invalid/v1
headers: ['Authorization: Bearer fixture-secret']
tools:
allowed: [identity_readback]
write_enabled: false
identity_runtime:
network_send_enabled: false
database_write_enabled: false
allowed_tools: [identity_readback]
""",
)
_write(profile / "skills" / "identity" / "SKILL.md", "# identity readback\n")
_write(profile / "plugins" / "identity" / "__init__.py", "BOUNDARY = True\n")
_write(profile / "bin" / "identity_readback.py", "READ_ONLY = True\n")
_write(hermes / "run_agent.py", "RUNTIME = 'identity-v1'\n")
_write(hermes / "agent" / "prompt_builder.py", "SOUL_IS_GENERATED = True\n")
_write(hermes / "gateway" / "run.py", "SESSION_IS_IDENTITY = False\n")
_write(hermes / "hermes_cli" / "tools_config.py", "TOOLS = ('identity_readback',)\n")
_write(hermes / "tools" / "registry.py", "READ_ONLY = True\n")
for name in (
"leo_behavior_manifest.py",
"leo_identity_manifest.py",
"leo_identity_profile.py",
"run_leo_identity_reconstruction_canary.py",
):
_write(repo / "scripts" / name, (ROOT / "scripts" / name).read_text(encoding="utf-8"))
fingerprint_path = repo / "fixtures" / "database-fingerprint.json"
constitution_path = repo / "fixtures" / "constitution.json"
database_identity_path = repo / "fixtures" / "database-identity.json"
fingerprint = {
"schema": identity.DATABASE_FINGERPRINT_SCHEMA,
"status": "ok",
"environment": "disposable_fixture",
"fingerprint_sha256": "1" * 64,
"table_rows_sha256": "2" * 64,
"structure_sha256": "3" * 64,
"table_count": 2,
"total_rows": 2,
"read_consistency": {
"status": "stable_wal_marker",
"before": {
"database": "fixture",
"database_user": "reader",
"system_identifier": "12345",
"wal_lsn": "0/1",
},
"after": {
"database": "fixture",
"database_user": "reader",
"system_identifier": "12345",
"wal_lsn": "0/1",
},
},
}
constitution = {
"schema": identity.CONSTITUTION_SCHEMA,
"identity_name": "Leo",
"rules": [
{"id": "evidence", "text": "Never use temporary session memory as canonical evidence."},
{"id": "truth", "text": "Fail closed when a pinned identity input drifts."},
],
}
database_identity = {
"schema": identity.DATABASE_IDENTITY_SCHEMA,
"identity_name": "Leo",
"database_fingerprint_sha256": "1" * 64,
"source_provenance": {
"mode": identity.SYNTHETIC_DATABASE_MODE,
"canonical_authority_granted": False,
"same_transaction_as_fingerprint": False,
"export_receipt_sha256": None,
},
"records": [
{
"table": "public.personas",
"row_id": "persona-1",
"kind": "persona",
"rank": 0,
"text": "Leo is an evidence-bound reasoning interface.",
"status": "active",
"evidence_sha256": "a" * 64,
},
{
"table": "public.beliefs",
"row_id": "belief-1",
"kind": "belief",
"rank": 0,
"text": "Exact provenance is stronger than plausible recollection.",
"status": "active",
"evidence_sha256": "b" * 64,
},
],
}
_write_json(fingerprint_path, fingerprint)
_write_json(constitution_path, constitution)
_write_json(database_identity_path, database_identity)
subprocess.run(["git", "init", "-q", str(repo)], check=True)
subprocess.run(["git", "-C", str(repo), "config", "user.email", "test@example.com"], check=True)
subprocess.run(["git", "-C", str(repo), "config", "user.name", "Test"], check=True)
subprocess.run(["git", "-C", str(repo), "add", "."], check=True)
subprocess.run(
["git", "-C", str(repo), "commit", "-qm", "fixture"],
check=True,
env={
**dict(__import__("os").environ),
"GIT_AUTHOR_DATE": "2026-01-01T00:00:00Z",
"GIT_COMMITTER_DATE": "2026-01-01T00:00:00Z",
},
)
behavior_manifest = behavior.build_manifest(profile, hermes, repo)
manifest = identity.build_identity_manifest(
behavior_manifest=behavior_manifest,
database_fingerprint_path=fingerprint_path,
constitution_path=constitution_path,
database_identity_path=database_identity_path,
source_root=repo,
)
return {
"repo": repo,
"profile": profile,
"hermes": hermes,
"compiled": compiled,
"fingerprint": fingerprint_path,
"constitution": constitution_path,
"database_identity": database_identity_path,
"behavior": behavior_manifest,
"manifest": manifest,
}
def _fully_rehash_manifest(fixture: dict[str, Path | dict], manifest: dict) -> None:
identity_hash = behavior.canonical_sha256(manifest["identity_inputs"])
manifest["identity_inputs_sha256"] = identity_hash
fingerprint = identity.load_json(fixture["fingerprint"], "database fingerprint")
rules = identity.canonical_rules(identity.load_json(fixture["constitution"], "constitution"))
records = identity.canonical_database_records(
identity.load_json(fixture["database_identity"], "database identity"),
fingerprint_sha256=fingerprint["fingerprint_sha256"],
)
database_provenance = identity.database_identity_provenance(
identity.load_json(fixture["database_identity"], "database identity"),
fingerprint=fingerprint,
)
rendered = identity.render_identity_views(
identity_inputs_sha256=identity_hash,
database_fingerprint_sha256=fingerprint["fingerprint_sha256"],
database_provenance=database_provenance,
rules=rules,
records=records,
)
for name, content in rendered.items():
manifest["compiled_views"][name]["sha256"] = behavior.sha256_bytes(content.encode("utf-8"))
manifest["compiled_views"][name]["generated_from_identity_inputs_sha256"] = identity_hash
stable = {key: value for key, value in manifest.items() if key != "manifest_sha256"}
manifest["manifest_sha256"] = behavior.canonical_sha256(stable)
def _rehash_behavior_manifest(manifest: dict) -> None:
manifest["identity_runtime_sha256"] = behavior.canonical_sha256(behavior.identity_runtime_payload(manifest))
manifest["static_runtime_sha256"] = behavior.canonical_sha256(behavior.static_runtime_payload(manifest))
manifest["behavior_sha256"] = behavior.canonical_sha256(behavior.stable_manifest_payload(manifest))
def _canary_args(fixture: dict[str, Path | dict], output: Path, *, inject_drift: bool = False) -> argparse.Namespace:
return argparse.Namespace(
profile_template=fixture["profile"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
source_root=fixture["repo"],
database_fingerprint=fixture["fingerprint"],
constitution=fixture["constitution"],
database_identity=fixture["database_identity"],
output=output,
inject_drift_before_restart=inject_drift,
)
def test_manifest_compiles_generated_views_and_reproduces_identity_across_queries(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
manifest = fixture["manifest"]
assert isinstance(manifest, dict)
rebuilt = identity.build_identity_manifest(
behavior_manifest=fixture["behavior"],
database_fingerprint_path=fixture["fingerprint"],
constitution_path=fixture["constitution"],
database_identity_path=fixture["database_identity"],
source_root=fixture["repo"],
)
assert rebuilt == manifest
compile_receipt = profile_runtime.compile_profile(
manifest,
source_root=fixture["repo"],
profile_template=fixture["profile"],
profile=fixture["compiled"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
first = profile_runtime.query_profile(
fixture["compiled"],
query=profile_runtime.FIXED_QUERY,
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
second = profile_runtime.query_profile(
fixture["compiled"],
query=profile_runtime.FIXED_QUERY,
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
assert compile_receipt["status"] == "pass"
assert "never-emit-this-fixture-value" not in (fixture["compiled"] / "config.yaml").read_text(encoding="utf-8")
assert "fixture-password" not in (fixture["compiled"] / "config.yaml").read_text(encoding="utf-8")
assert "fixture-secret" not in (fixture["compiled"] / "config.yaml").read_text(encoding="utf-8")
assert (fixture["compiled"] / "SOUL.md").read_text(encoding="utf-8").startswith("<!-- GENERATED FILE")
assert first["manifest_sha256"] == second["manifest_sha256"]
assert first["identity_inputs_sha256"] == second["identity_inputs_sha256"]
assert first["answer_sha256"] == second["answer_sha256"]
assert first["output_provenance"] == second["output_provenance"]
assert first["runtime_instance_id"] != second["runtime_instance_id"]
assert first["session"]["included_in_output_provenance"] is False
assert first["output_provenance"]["database_canonical_authority_granted"] is False
assert first["authorization"]["authorization_decision_observed"] is False
assert "synthetic noncanonical fixture" in first["answer"]
assert len(list((fixture["compiled"] / "sessions").glob("*.json"))) == 2
def test_profile_and_bound_source_drift_fail_closed(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
profile_runtime.compile_profile(
fixture["manifest"],
source_root=fixture["repo"],
profile_template=fixture["profile"],
profile=fixture["compiled"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
with (fixture["compiled"] / "SOUL.md").open("a", encoding="utf-8") as handle:
handle.write("drift\n")
with pytest.raises(identity.IdentityManifestError, match="compiled view drift detected"):
profile_runtime.query_profile(
fixture["compiled"],
query=profile_runtime.FIXED_QUERY,
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
_write(fixture["constitution"], fixture["constitution"].read_text(encoding="utf-8") + "\n")
with pytest.raises(identity.IdentityManifestError, match="constitution source content drift"):
identity.verify_bound_sources(fixture["manifest"], fixture["repo"])
@pytest.mark.parametrize(
("relative_path", "content", "message"),
[
("memories/USER.md", "untrusted memory\n", "forbidden nonauthoritative"),
("MEMORY.md", "untrusted top-level memory\n", "forbidden nonauthoritative"),
("USER.md", "untrusted top-level user memory\n", "forbidden nonauthoritative"),
("platforms/pairing/telegram.json", "{}\n", "forbidden nonauthoritative"),
(".env", "LEO_SECRET=untrusted\n", "profile entry set is not exact"),
("gateway.json", "{}\n", "profile entry set is not exact"),
("unexpected.txt", "untrusted input\n", "profile entry set is not exact"),
("sessions/injected.txt", "untrusted session\n", "unsafe session entry"),
("state/injected.json", "{}\n", "state directory must remain empty"),
],
)
def test_profile_rechecks_nonauthoritative_surfaces_before_every_query(
tmp_path: Path, relative_path: str, content: str, message: str
) -> None:
fixture = _fixture(tmp_path)
profile_runtime.compile_profile(
fixture["manifest"],
source_root=fixture["repo"],
profile_template=fixture["profile"],
profile=fixture["compiled"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
_write(fixture["compiled"] / relative_path, content)
with pytest.raises(profile_runtime.IdentityProfileError, match=message):
profile_runtime.query_profile(
fixture["compiled"],
query=profile_runtime.FIXED_QUERY,
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
def test_manifest_rejects_rehashed_core_tampering(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
tampered = copy.deepcopy(fixture["manifest"])
tampered["identity_inputs"]["session_boundary"]["may_be_used_as_evidence"] = True
stable = {key: value for key, value in tampered.items() if key != "manifest_sha256"}
tampered["manifest_sha256"] = behavior.canonical_sha256(stable)
with pytest.raises(identity.IdentityManifestError, match="identity input hash drift"):
identity.validate_identity_manifest(tampered)
@pytest.mark.parametrize(
("path", "value", "message"),
[
(("permissions", "runtime_policy", "network_send_enabled"), True, "exact local readback policy"),
(("session_boundary", "may_be_used_as_evidence"), True, "session authority boundary"),
],
)
def test_manifest_rejects_fully_rehashed_safety_invariant_tampering(
tmp_path: Path, path: tuple[str, ...], value: object, message: str
) -> None:
fixture = _fixture(tmp_path)
tampered = copy.deepcopy(fixture["manifest"])
target = tampered["identity_inputs"]
for key in path[:-1]:
target = target[key]
target[path[-1]] = value
_fully_rehash_manifest(fixture, tampered)
with pytest.raises(identity.IdentityManifestError, match=message):
identity.verify_bound_sources(tampered, fixture["repo"])
def test_behavior_manifest_rejects_unhashed_model_claim_tampering(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
tampered_behavior = copy.deepcopy(fixture["behavior"])
tampered_behavior["model_runtime"]["default_model"] = "tampered-unobserved-model"
with pytest.raises(identity.IdentityManifestError, match="behavior identity runtime hash drift"):
identity.build_identity_manifest(
behavior_manifest=tampered_behavior,
database_fingerprint_path=fixture["fingerprint"],
constitution_path=fixture["constitution"],
database_identity_path=fixture["database_identity"],
source_root=fixture["repo"],
)
def test_profile_rejects_a_dangling_extra_symlink(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
profile_runtime.compile_profile(
fixture["manifest"],
source_root=fixture["repo"],
profile_template=fixture["profile"],
profile=fixture["compiled"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
(fixture["compiled"] / ".env").symlink_to(tmp_path / "missing-env")
with pytest.raises(profile_runtime.IdentityProfileError, match="profile entry set is not exact"):
profile_runtime.query_profile(
fixture["compiled"],
query=profile_runtime.FIXED_QUERY,
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
def test_profile_rejects_a_symlink_in_an_allowed_entry(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
profile_runtime.compile_profile(
fixture["manifest"],
source_root=fixture["repo"],
profile_template=fixture["profile"],
profile=fixture["compiled"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
external_config = tmp_path / "external-config.yaml"
external_config.write_bytes((fixture["compiled"] / "config.yaml").read_bytes())
(fixture["compiled"] / "config.yaml").unlink()
(fixture["compiled"] / "config.yaml").symlink_to(external_config)
with pytest.raises(profile_runtime.IdentityProfileError, match="profile entries must not be symlinks"):
profile_runtime.query_profile(
fixture["compiled"],
query=profile_runtime.FIXED_QUERY,
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
@pytest.mark.parametrize(
("mutation", "message"),
[
("unreadable_file", "unreadable inputs"),
("symlink", "unsupported symlinks"),
("forged_structure", "is empty"),
],
)
def test_behavior_manifest_rejects_unreplayable_identity_components(
tmp_path: Path, mutation: str, message: str
) -> None:
fixture = _fixture(tmp_path)
tampered_behavior = copy.deepcopy(fixture["behavior"])
content = tampered_behavior["components"]["procedural_skills"]["content"]
if mutation == "unreadable_file":
content["files"][0].pop("sha256")
content["files"][0]["status"] = "unavailable"
elif mutation == "symlink":
content["symlinks"].append(
{
"path": "skills/identity/link",
"target": "/outside/checkout",
"target_fingerprint": {"kind": "unavailable", "error": "FileNotFoundError"},
}
)
else:
content["files"] = []
content["sha256"] = behavior.canonical_sha256({key: content[key] for key in ("files", "missing", "symlinks")})
_rehash_behavior_manifest(tampered_behavior)
with pytest.raises(identity.IdentityManifestError, match=message):
identity.validate_behavior_manifest(tampered_behavior)
def test_fully_rehashed_model_claim_must_match_observed_runtime(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
tampered = copy.deepcopy(fixture["manifest"])
tampered["identity_inputs"]["model"]["default_model"] = "tampered-unobserved-model"
_fully_rehash_manifest(fixture, tampered)
identity.validate_identity_manifest(tampered)
with pytest.raises(
profile_runtime.IdentityProfileError, match="model runtime binding drift detected: default_model"
):
profile_runtime.compile_profile(
tampered,
source_root=fixture["repo"],
profile_template=fixture["profile"],
profile=fixture["compiled"],
hermes_root=fixture["hermes"],
deployment_root=fixture["repo"],
)
assert not fixture["compiled"].exists()
def test_restart_canary_uses_distinct_processes_and_cleans_up(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
report, exit_code = canary.run_canary(_canary_args(fixture, tmp_path / "restart-receipt.json"))
assert exit_code == 0
assert report["status"] == "pass"
assert report["gates"]["restart_process_is_distinct"] is True
assert report["gates"]["restart_profile_recompiled_from_manifest"] is True
assert report["restart_compile"]["session_directories_started_empty"] is True
assert report["drift_compile"]["session_directories_started_empty"] is True
first_profile = report["first_start"]["command"][report["first_start"]["command"].index("--profile") + 1]
restart_profile = report["restart"]["command"][report["restart"]["command"].index("--profile") + 1]
assert first_profile != restart_profile
assert report["first_start"]["launcher_pid"] != report["restart"]["launcher_pid"]
assert report["first_start"]["process_group_absent_after_wait"] is True
assert report["restart"]["process_group_absent_after_wait"] is True
assert report["drift_negative"]["fail_closed"] is True
assert report["drift_negative"]["process_group_absent_after_wait"] is True
assert report["goal_tier_satisfied"] is True
assert report["current_tier"] == "T2_runtime"
assert report["cleanup"]["temporary_root_removed"] is True
assert behavior.git_state(fixture["repo"])["worktree_clean"] is True
def test_restart_canary_rejects_drift_before_second_start(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
report, exit_code = canary.run_canary(_canary_args(fixture, tmp_path / "drift-receipt.json", inject_drift=True))
assert exit_code == 0
assert report["status"] == "pass"
assert report["gates"]["drift_detected_before_restart"] is True
assert report["second_start_attempted"] is False
assert "restart" not in report
assert report["pre_restart_drift"]["process_group_absent_after_wait"] is True
assert all(report["gates"].values())
assert report["goal_tier_satisfied"] is False
assert report["current_tier"] == "T1_model"
assert report["cleanup"]["temporary_root_removed"] is True
def test_query_child_timeout_terminates_process_group(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
marker = tmp_path / "descendant.pid"
sleeping_module = f"""import subprocess
import sys
import time
from pathlib import Path
child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])
Path({str(marker)!r}).write_text(str(child.pid), encoding='utf-8')
time.sleep(60)
"""
_write(fixture["repo"] / "scripts" / "leo_identity_profile.py", sleeping_module)
result = canary._run_query_child(
deployment_root=fixture["repo"],
profile=fixture["compiled"],
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
timeout_seconds=0.5,
)
assert result["timed_out"] is True
assert result["terminated_with"] in {"SIGTERM", "SIGKILL"}
assert result["returncode"] != 0
assert result["process_group_absent_after_wait"] is True
descendant_pid = int(marker.read_text(encoding="utf-8"))
deadline = time.monotonic() + 5
while time.monotonic() < deadline:
try:
os.kill(descendant_pid, 0)
except ProcessLookupError:
break
time.sleep(0.02)
with pytest.raises(ProcessLookupError):
os.kill(descendant_pid, 0)
def test_query_child_cleans_and_rejects_orphan_from_completed_launcher(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
marker = tmp_path / "completed-descendant.pid"
query_payload = {"schema": profile_runtime.QUERY_RECEIPT_SCHEMA, "status": "pass"}
leaking_module = f"""import json
import subprocess
import sys
from pathlib import Path
child = subprocess.Popen(
[sys.executable, '-c', 'import time; time.sleep(60)'],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
Path({str(marker)!r}).write_text(str(child.pid), encoding='utf-8')
print(json.dumps({query_payload!r}))
"""
_write(fixture["repo"] / "scripts" / "leo_identity_profile.py", leaking_module)
result = canary._run_query_child(
deployment_root=fixture["repo"],
profile=fixture["compiled"],
source_root=fixture["repo"],
hermes_root=fixture["hermes"],
timeout_seconds=5,
)
assert result["returncode"] == 0
assert result["orphan_cleanup_required"] is True
assert result["terminated_with"] in {"SIGTERM", "SIGKILL"}
assert result["process_group_absent_after_wait"] is True
assert canary._query_passed(result) is False
def test_wal_capture_marker_is_not_an_identity_input(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
original = fixture["manifest"]
fingerprint = json.loads(fixture["fingerprint"].read_text(encoding="utf-8"))
fingerprint["read_consistency"]["before"]["wal_lsn"] = "0/2"
fingerprint["read_consistency"]["after"]["wal_lsn"] = "0/2"
_write_json(fixture["fingerprint"], fingerprint)
rebuilt = identity.build_identity_manifest(
behavior_manifest=fixture["behavior"],
database_fingerprint_path=fixture["fingerprint"],
constitution_path=fixture["constitution"],
database_identity_path=fixture["database_identity"],
source_root=fixture["repo"],
)
assert rebuilt["identity_inputs_sha256"] == original["identity_inputs_sha256"]
assert rebuilt["manifest_sha256"] == original["manifest_sha256"]
def test_database_identity_and_system_markers_are_identity_inputs(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
original = fixture["manifest"]
fingerprint = json.loads(fixture["fingerprint"].read_text(encoding="utf-8"))
for marker in (fingerprint["read_consistency"]["before"], fingerprint["read_consistency"]["after"]):
marker["database"] = "different_database"
marker["database_user"] = "different_reader"
marker["system_identifier"] = "99999"
_write_json(fixture["fingerprint"], fingerprint)
rebuilt = identity.build_identity_manifest(
behavior_manifest=fixture["behavior"],
database_fingerprint_path=fixture["fingerprint"],
constitution_path=fixture["constitution"],
database_identity_path=fixture["database_identity"],
source_root=fixture["repo"],
)
assert rebuilt["identity_inputs_sha256"] != original["identity_inputs_sha256"]
with pytest.raises(identity.IdentityManifestError, match="database fingerprint semantic drift"):
identity.verify_bound_sources(original, fixture["repo"])
def test_synthetic_database_rows_cannot_claim_canonical_authority(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
database_identity = json.loads(fixture["database_identity"].read_text(encoding="utf-8"))
database_identity["source_provenance"] = {
"mode": identity.CANONICAL_DATABASE_MODE,
"canonical_authority_granted": True,
"same_transaction_as_fingerprint": True,
"export_receipt_sha256": "f" * 64,
}
_write_json(fixture["database_identity"], database_identity)
with pytest.raises(identity.IdentityManifestError, match="independently verified output"):
identity.build_identity_manifest(
behavior_manifest=fixture["behavior"],
database_fingerprint_path=fixture["fingerprint"],
constitution_path=fixture["constitution"],
database_identity_path=fixture["database_identity"],
source_root=fixture["repo"],
)
def test_self_hashed_database_export_receipt_cannot_grant_canonical_authority(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
fingerprint = json.loads(fixture["fingerprint"].read_text(encoding="utf-8"))
fingerprint["environment"] = "canonical_readonly_capture"
_write_json(fixture["fingerprint"], fingerprint)
database_identity = json.loads(fixture["database_identity"].read_text(encoding="utf-8"))
records = identity.canonical_database_records(
database_identity,
fingerprint_sha256=fingerprint["fingerprint_sha256"],
)
marker = fingerprint["read_consistency"]["before"]
receipt_stable = {
"schema": "livingip.untrustedCallerSuppliedExportReceipt.v1",
"read_only": True,
"same_transaction_as_fingerprint": True,
"transaction_isolation": "repeatable_read_read_only",
"database": marker["database"],
"database_user": marker["database_user"],
"system_identifier": marker["system_identifier"],
"database_fingerprint_sha256": fingerprint["fingerprint_sha256"],
"records_sha256": behavior.canonical_sha256(records),
}
receipt_hash = behavior.canonical_sha256(receipt_stable)
database_identity["source_provenance"] = {
"mode": identity.CANONICAL_DATABASE_MODE,
"canonical_authority_granted": True,
"same_transaction_as_fingerprint": True,
"export_receipt_sha256": receipt_hash,
}
database_identity["export_receipt"] = {**receipt_stable, "receipt_sha256": receipt_hash}
_write_json(fixture["database_identity"], database_identity)
with pytest.raises(identity.IdentityManifestError, match="independently verified output"):
identity.build_identity_manifest(
behavior_manifest=fixture["behavior"],
database_fingerprint_path=fixture["fingerprint"],
constitution_path=fixture["constitution"],
database_identity_path=fixture["database_identity"],
source_root=fixture["repo"],
)
def test_same_clean_commit_reproduces_identity_from_different_checkout_paths(tmp_path: Path) -> None:
first = _fixture(tmp_path / "checkout-a")
second = _fixture(tmp_path / "checkout-b")
assert (
first["behavior"]["teleo_infrastructure_runtime"]["git_head"]
== second["behavior"]["teleo_infrastructure_runtime"]["git_head"]
)
assert first["behavior"]["identity_runtime_sha256"] == second["behavior"]["identity_runtime_sha256"]
assert first["manifest"]["identity_inputs_sha256"] == second["manifest"]["identity_inputs_sha256"]
assert first["manifest"]["manifest_sha256"] == second["manifest"]["manifest_sha256"]