305 lines
11 KiB
Python
305 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Fingerprint every local runtime layer that can make Leo answer differently.
|
|
|
|
The manifest deliberately records hashes and a small allowlist of non-secret
|
|
configuration fields. It never emits profile file contents or credentials.
|
|
Canonical Postgres state is outside this manifest and must be fingerprinted by
|
|
the database harness that owns the connection.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
SCHEMA = "livingip.leoBehaviorManifest.v1"
|
|
DEFAULT_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean")
|
|
DEFAULT_HERMES_ROOT = Path("/home/teleo/.hermes/hermes-agent")
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def canonical_sha256(value: Any) -> str:
|
|
encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
|
|
return sha256_bytes(encoded)
|
|
|
|
|
|
def file_sha256(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _safe_relative(path: Path, root: Path) -> str:
|
|
try:
|
|
return path.relative_to(root).as_posix()
|
|
except ValueError:
|
|
return str(path)
|
|
|
|
|
|
def _resolved_node_fingerprint(path: Path, seen_directories: frozenset[tuple[int, int]] = frozenset()) -> dict[str, Any]:
|
|
"""Hash a symlink target, following nested links while stopping directory cycles."""
|
|
|
|
try:
|
|
if path.is_symlink():
|
|
target = os.readlink(path)
|
|
resolved = path.resolve(strict=True)
|
|
return {
|
|
"kind": "symlink",
|
|
"target": target,
|
|
"resolved": _resolved_node_fingerprint(resolved, seen_directories),
|
|
}
|
|
stat = path.stat()
|
|
if path.is_file():
|
|
return {
|
|
"kind": "file",
|
|
"bytes": stat.st_size,
|
|
"mode": oct(stat.st_mode & 0o777),
|
|
"sha256": file_sha256(path),
|
|
}
|
|
if path.is_dir():
|
|
identity = (stat.st_dev, stat.st_ino)
|
|
if identity in seen_directories:
|
|
return {"kind": "directory_cycle"}
|
|
child_seen = seen_directories | {identity}
|
|
children = [
|
|
{"name": child.name, "fingerprint": _resolved_node_fingerprint(child, child_seen)}
|
|
for child in sorted(path.iterdir(), key=lambda item: item.name)
|
|
]
|
|
return {
|
|
"kind": "directory",
|
|
"entry_count": len(children),
|
|
"sha256": canonical_sha256(children),
|
|
}
|
|
return {"kind": "other", "mode": oct(stat.st_mode & 0o777)}
|
|
except (OSError, RuntimeError) as exc:
|
|
return {"kind": "unavailable", "error": type(exc).__name__}
|
|
|
|
|
|
def path_manifest(profile: Path, paths: tuple[Path, ...]) -> dict[str, Any]:
|
|
files: list[dict[str, Any]] = []
|
|
missing: list[str] = []
|
|
symlinks: list[dict[str, Any]] = []
|
|
for requested in paths:
|
|
if not requested.exists() and not requested.is_symlink():
|
|
missing.append(_safe_relative(requested, profile))
|
|
continue
|
|
candidates = [requested] if requested.is_file() or requested.is_symlink() else sorted(requested.rglob("*"))
|
|
for path in candidates:
|
|
relative = _safe_relative(path, profile)
|
|
if path.is_symlink():
|
|
symlinks.append(
|
|
{
|
|
"path": relative,
|
|
"target": os.readlink(path),
|
|
"target_fingerprint": _resolved_node_fingerprint(path),
|
|
}
|
|
)
|
|
continue
|
|
if not path.is_file():
|
|
continue
|
|
try:
|
|
stat = path.stat()
|
|
files.append(
|
|
{
|
|
"path": relative,
|
|
"bytes": stat.st_size,
|
|
"mode": oct(stat.st_mode & 0o777),
|
|
"sha256": file_sha256(path),
|
|
}
|
|
)
|
|
except OSError as exc:
|
|
files.append({"path": relative, "status": "unavailable", "error": type(exc).__name__})
|
|
files.sort(key=lambda item: item["path"])
|
|
symlinks.sort(key=lambda item: item["path"])
|
|
stable = {"files": files, "missing": sorted(missing), "symlinks": symlinks}
|
|
return {
|
|
**stable,
|
|
"file_count": len(files),
|
|
"total_bytes": sum(int(item.get("bytes", 0)) for item in files),
|
|
"sha256": canonical_sha256(stable),
|
|
}
|
|
|
|
|
|
def _nested(config: dict[str, Any], *path: str) -> Any:
|
|
value: Any = config
|
|
for key in path:
|
|
value = value.get(key) if isinstance(value, dict) else None
|
|
return value
|
|
|
|
|
|
def 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__}
|
|
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"),
|
|
}
|
|
|
|
|
|
def git_head(repo: Path) -> str | None:
|
|
try:
|
|
completed = subprocess.run(
|
|
["git", "-C", str(repo), "rev-parse", "HEAD"],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=10,
|
|
env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"},
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return None
|
|
value = completed.stdout.strip()
|
|
return value if completed.returncode == 0 and len(value) == 40 else None
|
|
|
|
|
|
def component(
|
|
profile: Path,
|
|
*,
|
|
role: str,
|
|
mutability: str,
|
|
replayability: str,
|
|
paths: tuple[Path, ...],
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"role": role,
|
|
"mutability": mutability,
|
|
"replayability": replayability,
|
|
"content": path_manifest(profile, paths),
|
|
}
|
|
|
|
|
|
def build_manifest(profile: Path = DEFAULT_PROFILE, hermes_root: Path = DEFAULT_HERMES_ROOT) -> dict[str, Any]:
|
|
profile = profile.resolve()
|
|
hermes_root = hermes_root.resolve()
|
|
config = safe_model_config(profile / "config.yaml")
|
|
components = {
|
|
"profile_config": component(
|
|
profile,
|
|
role="Selects model routing, tools, memory, and gateway behavior.",
|
|
mutability="deployment_static",
|
|
replayability="hashable_but_may_contain_unpinned_provider_defaults",
|
|
paths=(profile / "config.yaml",),
|
|
),
|
|
"runtime_identity": component(
|
|
profile,
|
|
role="SOUL.md identity text injected into the Hermes system prompt.",
|
|
mutability="file_static_until_edit_or_render",
|
|
replayability="fully_hashable",
|
|
paths=(profile / "SOUL.md",),
|
|
),
|
|
"procedural_skills": component(
|
|
profile,
|
|
role="On-demand operating instructions available to the model.",
|
|
mutability="deployment_static",
|
|
replayability="fully_hashable",
|
|
paths=(profile / "skills",),
|
|
),
|
|
"runtime_middleware": component(
|
|
profile,
|
|
role="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"),
|
|
),
|
|
"database_tools": component(
|
|
profile,
|
|
role="Commands that query Postgres, stage proposals, and return receipts.",
|
|
mutability="deployment_static",
|
|
replayability="fully_hashable",
|
|
paths=(profile / "bin",),
|
|
),
|
|
"persistent_memory": component(
|
|
profile,
|
|
role="Hermes MEMORY.md and USER.md content injected independently of canonical Postgres.",
|
|
mutability="conversation_dynamic_noncanonical",
|
|
replayability="hashable_but_should_not_define_durable_knowledge",
|
|
paths=(profile / "memories",),
|
|
),
|
|
"conversation_sessions": component(
|
|
profile,
|
|
role="Session transcripts and local state used for conversation continuity.",
|
|
mutability="request_dynamic_noncanonical",
|
|
replayability="hashable_but_not_a_knowledge_source",
|
|
paths=(
|
|
profile / "sessions",
|
|
profile / "state",
|
|
profile / "state.db",
|
|
profile / "state.db-wal",
|
|
profile / "state.db-shm",
|
|
),
|
|
),
|
|
"generated_prompt_cache": component(
|
|
profile,
|
|
role="Hermes-generated cache of the skills prompt.",
|
|
mutability="generated_cache",
|
|
replayability="discard_and_rebuild",
|
|
paths=(profile / ".skills_prompt_snapshot.json",),
|
|
),
|
|
}
|
|
stable = {
|
|
"schema": SCHEMA,
|
|
"model_runtime": {
|
|
**config,
|
|
"base_model_weights": "external_provider_managed_not_locally_hashable",
|
|
"actual_per_turn_model_must_be_recorded_by_the_call_receipt": True,
|
|
},
|
|
"hermes_runtime": {
|
|
"git_head": git_head(hermes_root),
|
|
"source_tree": path_manifest(
|
|
profile, (hermes_root / "run_agent.py", hermes_root / "agent" / "prompt_builder.py")
|
|
),
|
|
},
|
|
"components": components,
|
|
"canonical_database": {
|
|
"included": False,
|
|
"reason": "Fingerprint through a read-only Postgres manifest in the database-owning harness.",
|
|
},
|
|
}
|
|
return {
|
|
**stable,
|
|
"generated_at_utc": datetime.now(timezone.utc).isoformat(),
|
|
"profile_root": str(profile),
|
|
"hermes_root": str(hermes_root),
|
|
"behavior_sha256": canonical_sha256(stable),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--profile", type=Path, default=DEFAULT_PROFILE)
|
|
parser.add_argument("--hermes-root", type=Path, default=DEFAULT_HERMES_ROOT)
|
|
parser.add_argument("--output", type=Path)
|
|
args = parser.parse_args()
|
|
manifest = build_manifest(args.profile, args.hermes_root)
|
|
encoded = json.dumps(manifest, indent=2, sort_keys=True) + "\n"
|
|
if args.output:
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(encoded, encoding="utf-8")
|
|
print(encoded, end="")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|