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"]