teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Merge remote-tracking branch 'origin/main' into codex/leo-observatory-private-staging-followup-w2-20260715

Browse source
This commit is contained in:
twentyOne2x 2026-07-15 22:58:40 +02:00
commit 2631dcf4a5
2 changed files with 653 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

422
scripts/leo_identity_version_manifest.py Normal file
View file

@ -0,0 +1,422 @@
#!/usr/bin/env python3
"""Generate and independently rebuild Leo's deterministic identity version manifest.
The version manifest is an operator-facing projection over the stricter identity
compiler. It names the runtime, Git commit, individual skills, database
version, compiled identity, and session boundary without turning generated
views or session history into provenance.
"""
from __future__ import annotations
import argparse
import json
import re
import sys
from pathlib import Path, PurePosixPath
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
SCHEMA = "livingip.leoIdentityVersionManifest.v1"
DEFAULT_PROFILE = Path("fixtures/working-leo/leo-identity-v1/profile-template")
DEFAULT_HERMES_ROOT = Path("fixtures/working-leo/leo-identity-v1/hermes-runtime")
DEFAULT_DEPLOYMENT_ROOT = Path(".")
DEFAULT_SOURCE_ROOT = Path(".")
DEFAULT_DATABASE_FINGERPRINT = Path("fixtures/working-leo/leo-identity-v1/leo-database-fingerprint-v1.json")
DEFAULT_CONSTITUTION = Path("fixtures/working-leo/leo-identity-v1/leo-constitution-v1.json")
DEFAULT_DATABASE_IDENTITY = Path("fixtures/working-leo/leo-identity-v1/leo-database-identity-v1.json")
HEX_64 = re.compile(r"^[0-9a-f]{64}$")
class IdentityVersionManifestError(identity.IdentityManifestError):
"""The identity version receipt is incomplete, inconsistent, or drifted."""
def _require(condition: bool, message: str) -> None:
if not condition:
raise IdentityVersionManifestError(message)
def expected_authority_contract() -> dict[str, str]:
return {
"durable_knowledge_authority": "approved_postgres_rows_with_bound_evidence",
"compiled_identity_role": "derived_view_from_versioned_inputs_not_authority",
"session_history_role": "temporary_or_persisted_context_not_provenance",
}
def _runtime_version(identity_manifest: dict[str, Any]) -> dict[str, Any]:
core = identity_manifest["identity_inputs"]
runtime = core["runtime"]
model = core["model"]
return {
"source_commit": core["source_commit"],
"identity_runtime_sha256": runtime["identity_runtime_sha256"],
"teleo_infrastructure": {
"git_head": runtime["teleo_git_head"],
"identity_source_sha256": runtime["teleo_source_sha256"],
},
"hermes": {
"git_head": runtime["hermes_git_head"],
"source_sha256": runtime["hermes_source_sha256"],
},
"python": runtime["python"],
"model": {
"provider": model["provider"],
"default_model": model["default_model"],
"model_section_sha256": model["model_section_sha256"],
"hosted_weights": model["hosted_weights"],
"actual_model_required_in_turn_receipt": model["actual_model_required_in_turn_receipt"],
},
}
def _canonical_database_version(identity_manifest: dict[str, Any]) -> dict[str, Any]:
core = identity_manifest["identity_inputs"]
database = core["canonical_database"]
source = core["identity_sources"]["database_derived_identity"]
provenance = source["provenance"]
return {
"authority": database["authority"],
"canonical_authority_granted": database["canonical_authority_granted"],
"database": database["database"],
"database_user": database["database_user"],
"system_identifier": database["system_identifier"],
"fingerprint_sha256": database["fingerprint_sha256"],
"table_rows_sha256": database["table_rows_sha256"],
"structure_sha256": database["structure_sha256"],
"table_count": database["table_count"],
"total_rows": database["total_rows"],
"identity_records_sha256": source["semantic_sha256"],
"identity_record_count": source["record_count"],
"source_mode": provenance["mode"],
"same_transaction_as_fingerprint": provenance["same_transaction_as_fingerprint"],
"export_receipt_sha256": provenance["export_receipt_sha256"],
}
def _compiled_identity_version(identity_manifest: dict[str, Any]) -> dict[str, Any]:
return {
"role": "derived_view_from_versioned_inputs_not_authority",
"identity_inputs_sha256": identity_manifest["identity_inputs_sha256"],
"identity_manifest_sha256": identity_manifest["manifest_sha256"],
"views": identity_manifest["compiled_views"],
}
def _skill_versions(behavior_manifest: dict[str, Any], identity_manifest: dict[str, Any]) -> dict[str, Any]:
content = behavior_manifest["components"]["procedural_skills"]["content"]
identity.validate_content_manifest(content, label="procedural skills")
groups: dict[str, list[dict[str, Any]]] = {}
for item in content["files"]:
parts = PurePosixPath(item["path"]).parts
_require(
len(parts) >= 3 and parts[0] == "skills" and bool(parts[1]),
"procedural skill file is not inside a named skill directory",
)
relative_path = PurePosixPath(*parts[2:]).as_posix()
groups.setdefault(parts[1], []).append(
{
"path": relative_path,
"bytes": item["bytes"],
"mode": item["mode"],
"sha256": item["sha256"],
}
)
_require(bool(groups), "at least one named skill is required")
skills = []
for name in sorted(groups):
files = sorted(groups[name], key=lambda item: item["path"])
skills.append(
{
"name": name,
"files": files,
"file_count": len(files),
"total_bytes": sum(item["bytes"] for item in files),
"content_sha256": behavior.canonical_sha256(files),
}
)
result = {
"content_sha256": content["sha256"],
"file_count": content["file_count"],
"total_bytes": content["total_bytes"],
"skills": skills,
}
expected = identity_manifest["identity_inputs"]["skills"]
_require(result["content_sha256"] == expected["content_sha256"], "skill aggregate hash drift detected")
_require(result["file_count"] == expected["file_count"], "skill aggregate file count drift detected")
return result
def _validate_skill_versions(value: Any, identity_manifest: dict[str, Any]) -> None:
_require(isinstance(value, dict), "skill versions are missing")
_require(
set(value) == {"content_sha256", "file_count", "total_bytes", "skills"},
"skill version fields are invalid",
)
_require(isinstance(value.get("skills"), list) and bool(value["skills"]), "named skills are missing")
names: list[str] = []
reconstructed_files: list[dict[str, Any]] = []
total_bytes = 0
for skill in value["skills"]:
_require(
isinstance(skill, dict) and set(skill) == {"name", "files", "file_count", "total_bytes", "content_sha256"},
"named skill fields are invalid",
)
name = skill.get("name")
files = skill.get("files")
_require(
isinstance(name, str) and bool(name) and PurePosixPath(name).parts == (name,) and name not in {".", ".."},
"skill name is invalid",
)
_require(isinstance(files, list) and bool(files), f"skill {name} has no files")
names.append(name)
paths: list[str] = []
skill_bytes = 0
for item in files:
_require(
isinstance(item, dict) and set(item) == {"path", "bytes", "mode", "sha256"},
f"skill {name} file fields are invalid",
)
path = item.get("path")
size = item.get("bytes")
mode = item.get("mode")
_require(
isinstance(path, str)
and bool(path)
and not path.startswith("/")
and ".." not in PurePosixPath(path).parts,
f"skill {name} path is invalid",
)
_require(isinstance(size, int) and not isinstance(size, bool) and size >= 0, "skill byte size is invalid")
_require(isinstance(mode, str) and bool(re.fullmatch(r"0o[0-7]{3}", mode)), "skill mode is invalid")
_require(
isinstance(item.get("sha256"), str) and bool(HEX_64.fullmatch(item["sha256"])), "skill hash is invalid"
)
paths.append(path)
skill_bytes += size
reconstructed_files.append({**item, "path": f"skills/{name}/{path}"})
_require(paths == sorted(paths) and len(paths) == len(set(paths)), f"skill {name} paths are not sorted")
_require(skill.get("file_count") == len(files), f"skill {name} file count is invalid")
_require(skill.get("total_bytes") == skill_bytes, f"skill {name} byte count is invalid")
_require(skill.get("content_sha256") == behavior.canonical_sha256(files), f"skill {name} hash drift detected")
total_bytes += skill_bytes
_require(names == sorted(names) and len(names) == len(set(names)), "skill names are not unique and sorted")
reconstructed_files.sort(key=lambda item: item["path"])
aggregate = behavior.canonical_sha256({"files": reconstructed_files, "missing": [], "symlinks": []})
_require(value.get("content_sha256") == aggregate, "skill aggregate content drift detected")
_require(value.get("file_count") == len(reconstructed_files), "skill aggregate file count is invalid")
_require(value.get("total_bytes") == total_bytes, "skill aggregate byte count is invalid")
expected = identity_manifest["identity_inputs"]["skills"]
_require(value["content_sha256"] == expected["content_sha256"], "skills are not bound to the identity manifest")
_require(value["file_count"] == expected["file_count"], "skill count is not bound to the identity manifest")
def assemble_version_manifest(
*, behavior_manifest: dict[str, Any], identity_manifest: dict[str, Any]
) -> dict[str, Any]:
identity.validate_behavior_manifest(behavior_manifest)
identity.validate_identity_manifest(identity_manifest)
core = identity_manifest["identity_inputs"]
_require(
behavior_manifest["identity_runtime_sha256"] == core["runtime"]["identity_runtime_sha256"],
"behavior runtime is not bound to the identity manifest",
)
_require(
behavior_manifest["teleo_infrastructure_runtime"]["git_head"] == core["source_commit"],
"behavior source commit is not bound to the identity manifest",
)
stable = {
"schema": SCHEMA,
"identity_name": "Leo",
"runtime_version": _runtime_version(identity_manifest),
"skill_versions": _skill_versions(behavior_manifest, identity_manifest),
"canonical_database_version": _canonical_database_version(identity_manifest),
"compiled_identity_version": _compiled_identity_version(identity_manifest),
"session_boundary": core["session_boundary"],
"authority_contract": expected_authority_contract(),
"identity_manifest": identity_manifest,
}
return {**stable, "manifest_sha256": behavior.canonical_sha256(stable)}
def build_version_manifest(
*,
profile: Path,
hermes_root: Path,
deployment_root: Path,
source_root: Path,
database_fingerprint: Path,
constitution: Path,
database_identity: Path,
) -> dict[str, Any]:
behavior_manifest = behavior.build_manifest(profile, hermes_root, deployment_root)
identity_manifest = identity.build_identity_manifest(
behavior_manifest=behavior_manifest,
database_fingerprint_path=database_fingerprint,
constitution_path=constitution,
database_identity_path=database_identity,
source_root=source_root,
)
identity.verify_bound_sources(identity_manifest, source_root)
return assemble_version_manifest(
behavior_manifest=behavior_manifest,
identity_manifest=identity_manifest,
)
def validate_version_manifest(value: dict[str, Any]) -> None:
_require(
set(value)
== {
"schema",
"identity_name",
"runtime_version",
"skill_versions",
"canonical_database_version",
"compiled_identity_version",
"session_boundary",
"authority_contract",
"identity_manifest",
"manifest_sha256",
},
"identity version manifest fields are invalid",
)
_require(value.get("schema") == SCHEMA, "unsupported identity version manifest schema")
_require(value.get("identity_name") == "Leo", "identity version must describe Leo")
expected_hash = value.get("manifest_sha256")
_require(isinstance(expected_hash, str) and bool(HEX_64.fullmatch(expected_hash)), "manifest hash is invalid")
stable = {key: item for key, item in value.items() if key != "manifest_sha256"}
_require(behavior.canonical_sha256(stable) == expected_hash, "identity version manifest hash drift detected")
embedded = value.get("identity_manifest")
_require(isinstance(embedded, dict), "embedded identity manifest is missing")
identity.validate_identity_manifest(embedded)
core = embedded["identity_inputs"]
_require(value.get("runtime_version") == _runtime_version(embedded), "runtime version projection drift detected")
_validate_skill_versions(value.get("skill_versions"), embedded)
_require(
value.get("canonical_database_version") == _canonical_database_version(embedded),
"canonical database version projection drift detected",
)
_require(
value.get("compiled_identity_version") == _compiled_identity_version(embedded),
"compiled identity version projection drift detected",
)
_require(value.get("session_boundary") == core["session_boundary"], "session boundary projection drift detected")
_require(
value.get("authority_contract") == expected_authority_contract(), "identity authority contract drift detected"
)
def verify_against_inputs(
value: dict[str, Any],
*,
profile: Path,
hermes_root: Path,
deployment_root: Path,
source_root: Path,
database_fingerprint: Path,
constitution: Path,
database_identity: Path,
) -> None:
validate_version_manifest(value)
rebuilt = build_version_manifest(
profile=profile,
hermes_root=hermes_root,
deployment_root=deployment_root,
source_root=source_root,
database_fingerprint=database_fingerprint,
constitution=constitution,
database_identity=database_identity,
)
_require(rebuilt == value, "identity version manifest does not match the supplied versioned inputs")
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 summary(value: dict[str, Any]) -> dict[str, Any]:
return {
"status": "ok",
"manifest_sha256": value["manifest_sha256"],
"source_commit": value["runtime_version"]["source_commit"],
"skills": [item["name"] for item in value["skill_versions"]["skills"]],
"canonical_database": {
"authority": value["canonical_database_version"]["authority"],
"canonical_authority_granted": value["canonical_database_version"]["canonical_authority_granted"],
"fingerprint_sha256": value["canonical_database_version"]["fingerprint_sha256"],
},
"compiled_identity_sha256": value["compiled_identity_version"]["identity_inputs_sha256"],
"session_boundary": value["session_boundary"]["classification"],
}
def _add_input_arguments(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--profile", type=Path, default=DEFAULT_PROFILE)
parser.add_argument("--hermes-root", type=Path, default=DEFAULT_HERMES_ROOT)
parser.add_argument("--deployment-root", type=Path, default=DEFAULT_DEPLOYMENT_ROOT)
parser.add_argument("--source-root", type=Path, default=DEFAULT_SOURCE_ROOT)
parser.add_argument("--database-fingerprint", type=Path, default=DEFAULT_DATABASE_FINGERPRINT)
parser.add_argument("--constitution", type=Path, default=DEFAULT_CONSTITUTION)
parser.add_argument("--database-identity", type=Path, default=DEFAULT_DATABASE_IDENTITY)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
subparsers = parser.add_subparsers(dest="command", required=True)
generate = subparsers.add_parser("generate", help="generate a deterministic identity version manifest")
_add_input_arguments(generate)
generate.add_argument("--output", type=Path, required=True)
verify = subparsers.add_parser("verify", help="rebuild and compare a manifest from its versioned inputs")
verify.add_argument("--manifest", type=Path, required=True)
_add_input_arguments(verify)
args = parser.parse_args()
try:
if args.command == "generate":
manifest = build_version_manifest(
profile=args.profile,
hermes_root=args.hermes_root,
deployment_root=args.deployment_root,
source_root=args.source_root,
database_fingerprint=args.database_fingerprint,
constitution=args.constitution,
database_identity=args.database_identity,
)
write_json(args.output, manifest)
else:
manifest = identity.load_json(args.manifest, "identity version manifest")
verify_against_inputs(
manifest,
profile=args.profile,
hermes_root=args.hermes_root,
deployment_root=args.deployment_root,
source_root=args.source_root,
database_fingerprint=args.database_fingerprint,
constitution=args.constitution,
database_identity=args.database_identity,
)
print(json.dumps(summary(manifest), sort_keys=True))
return 0
except (identity.IdentityManifestError, OSError, ValueError) as exc:
print(
json.dumps(
{
"status": "error",
"error": "identity_version_manifest_invalid",
"message": str(exc),
},
sort_keys=True,
)
)
return 2
if __name__ == "__main__":
raise SystemExit(main())

231
tests/test_leo_identity_version_manifest.py Normal file
View file

@ -0,0 +1,231 @@
from __future__ import annotations
import json
import os
import subprocess
import sys
from pathlib import Path
import pytest
from scripts import leo_behavior_manifest as behavior
from scripts import leo_identity_version_manifest as version
from tests.test_leo_identity_manifest import _fixture, _write
ROOT = Path(__file__).resolve().parents[1]
def _build(fixture: dict[str, Path | dict]) -> dict:
return version.build_version_manifest(
profile=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"],
)
def _rehash(value: dict) -> None:
stable = {key: item for key, item in value.items() if key != "manifest_sha256"}
value["manifest_sha256"] = behavior.canonical_sha256(stable)
def _input_args(fixture: dict[str, Path | dict]) -> list[str]:
return [
"--profile",
str(fixture["profile"]),
"--hermes-root",
str(fixture["hermes"]),
"--deployment-root",
str(fixture["repo"]),
"--source-root",
str(fixture["repo"]),
"--database-fingerprint",
str(fixture["fingerprint"]),
"--constitution",
str(fixture["constitution"]),
"--database-identity",
str(fixture["database_identity"]),
]
def test_version_manifest_names_every_required_identity_boundary(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
assert manifest["schema"] == version.SCHEMA
assert (
manifest["runtime_version"]["source_commit"] == manifest["runtime_version"]["teleo_infrastructure"]["git_head"]
)
assert [item["name"] for item in manifest["skill_versions"]["skills"]] == ["identity"]
assert manifest["canonical_database_version"] == {
"authority": "synthetic_noncanonical_fixture",
"canonical_authority_granted": False,
"database": "fixture",
"database_user": "reader",
"system_identifier": "12345",
"fingerprint_sha256": "1" * 64,
"table_rows_sha256": "2" * 64,
"structure_sha256": "3" * 64,
"table_count": 2,
"total_rows": 2,
"identity_records_sha256": manifest["identity_manifest"]["identity_inputs"]["identity_sources"][
"database_derived_identity"
]["semantic_sha256"],
"identity_record_count": 2,
"source_mode": "synthetic_noncanonical_fixture",
"same_transaction_as_fingerprint": False,
"export_receipt_sha256": None,
}
assert manifest["compiled_identity_version"]["role"] == "derived_view_from_versioned_inputs_not_authority"
assert manifest["session_boundary"]["included_in_identity_inputs"] is False
assert manifest["session_boundary"]["may_be_used_as_evidence"] is False
assert manifest["authority_contract"] == version.expected_authority_contract()
version.validate_version_manifest(manifest)
def test_version_manifest_is_identical_across_checkout_paths(tmp_path: Path) -> None:
first = _build(_fixture(tmp_path / "checkout-a"))
second = _build(_fixture(tmp_path / "checkout-b"))
assert first == second
assert str(tmp_path) not in json.dumps(first, sort_keys=True)
@pytest.mark.parametrize(
("field", "replacement", "message"),
[
(
"authority_contract",
{
"durable_knowledge_authority": "session_history",
"compiled_identity_role": "authority",
"session_history_role": "provenance",
},
"authority contract drift",
),
(
"session_boundary",
{
"classification": "canonical",
"included_in_identity_inputs": True,
"may_be_used_as_evidence": True,
"restart_policy": "reuse_history_as_identity",
},
"session boundary projection drift",
),
],
)
def test_rehashed_authority_or_session_tampering_fails_closed(
tmp_path: Path, field: str, replacement: dict, message: str
) -> None:
manifest = _build(_fixture(tmp_path))
manifest[field] = replacement
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match=message):
version.validate_version_manifest(manifest)
def test_rehashed_database_authority_tampering_fails_closed(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
manifest["canonical_database_version"]["canonical_authority_granted"] = True
manifest["canonical_database_version"]["authority"] = "canonical_database"
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match="canonical database version projection drift"):
version.validate_version_manifest(manifest)
def test_named_skill_manifest_rejects_fully_rehashed_path_or_hash_tampering(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
skill = manifest["skill_versions"]["skills"][0]
skill["files"][0]["path"] = "../MEMORY.md"
skill["content_sha256"] = behavior.canonical_sha256(skill["files"])
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match="skill identity path is invalid"):
version.validate_version_manifest(manifest)
def test_named_skill_manifest_rejects_a_rehashed_parent_name(tmp_path: Path) -> None:
manifest = _build(_fixture(tmp_path))
manifest["skill_versions"]["skills"][0]["name"] = ".."
_rehash(manifest)
with pytest.raises(version.IdentityVersionManifestError, match="skill name is invalid"):
version.validate_version_manifest(manifest)
def test_rebuilder_rejects_a_different_clean_input_commit(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
manifest = _build(fixture)
_write(fixture["profile"] / "skills" / "identity" / "SKILL.md", "# changed identity readback\n")
subprocess.run(["git", "-C", str(fixture["repo"]), "add", "."], check=True)
subprocess.run(
["git", "-C", str(fixture["repo"]), "commit", "-qm", "change identity input"],
check=True,
env={
**dict(os.environ),
"GIT_AUTHOR_DATE": "2026-01-02T00:00:00Z",
"GIT_COMMITTER_DATE": "2026-01-02T00:00:00Z",
},
)
with pytest.raises(version.IdentityVersionManifestError, match="does not match the supplied versioned inputs"):
version.verify_against_inputs(
manifest,
profile=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"],
)
def test_cli_generate_is_deterministic_and_verify_rebuilds_inputs(tmp_path: Path) -> None:
fixture = _fixture(tmp_path)
first = tmp_path / "identity-version-first.json"
second = tmp_path / "identity-version-second.json"
common = _input_args(fixture)
for output in (first, second):
result = subprocess.run(
[
sys.executable,
"-m",
"scripts.leo_identity_version_manifest",
"generate",
*common,
"--output",
str(output),
],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
summary = json.loads(result.stdout)
assert summary["status"] == "ok"
assert summary["skills"] == ["identity"]
assert first.read_bytes() == second.read_bytes()
verify = subprocess.run(
[
sys.executable,
"-m",
"scripts.leo_identity_version_manifest",
"verify",
"--manifest",
str(first),
*common,
],
cwd=ROOT,
check=True,
capture_output=True,
text=True,
)
assert json.loads(verify.stdout)["manifest_sha256"] == json.loads(first.read_text())["manifest_sha256"]