feat: bind tested Leo turns to reproducible inputs

deploy/auto-deploy.sh

@@ -103,7 +103,7 @@ fi
 
 # Syntax check all Python files before copying
 ERRORS=0
-for f in lib/*.py *.py diagnostics/*.py telegram/*.py tests/*.py hermes-agent/leoclean-bin/*.py hermes-agent/leoclean-plugins/vps/*/*.py hermes-agent/patches/*.py scripts/compile_kb_source_packet.py scripts/kb_proposal_normalize.py scripts/prepare_kb_source_manifest.py scripts/leo_behavior_manifest.py; do
+for f in lib/*.py *.py diagnostics/*.py telegram/*.py tests/*.py hermes-agent/leoclean-bin/*.py hermes-agent/leoclean-plugins/vps/*/*.py hermes-agent/patches/*.py scripts/compile_kb_source_packet.py scripts/kb_proposal_normalize.py scripts/prepare_kb_source_manifest.py scripts/leo_behavior_manifest.py scripts/leo_turn_execution_manifest.py; do
   [ -f "$f" ] || continue
   if ! python3 -c "import ast, sys; ast.parse(open(sys.argv[1]).read())" "$f" 2>&1; then
     log "SYNTAX ERROR: $f"

deploy/deploy.sh

@@ -48,7 +48,7 @@ echo ""
 # Syntax check all Python files before deploying
 echo "=== Pre-deploy syntax check ==="
 ERRORS=0
-for f in "$REPO_ROOT/lib/"*.py "$REPO_ROOT/"*.py "$REPO_ROOT/diagnostics/"*.py "$REPO_ROOT/telegram/"*.py "$REPO_ROOT/hermes-agent/leoclean-bin/"*.py "$REPO_ROOT/hermes-agent/leoclean-plugins/vps/"*/*.py "$REPO_ROOT/hermes-agent/patches/"*.py "$REPO_ROOT/scripts/compile_kb_source_packet.py" "$REPO_ROOT/scripts/kb_proposal_normalize.py" "$REPO_ROOT/scripts/prepare_kb_source_manifest.py" "$REPO_ROOT/scripts/leo_behavior_manifest.py"; do
+for f in "$REPO_ROOT/lib/"*.py "$REPO_ROOT/"*.py "$REPO_ROOT/diagnostics/"*.py "$REPO_ROOT/telegram/"*.py "$REPO_ROOT/hermes-agent/leoclean-bin/"*.py "$REPO_ROOT/hermes-agent/leoclean-plugins/vps/"*/*.py "$REPO_ROOT/hermes-agent/patches/"*.py "$REPO_ROOT/scripts/compile_kb_source_packet.py" "$REPO_ROOT/scripts/kb_proposal_normalize.py" "$REPO_ROOT/scripts/prepare_kb_source_manifest.py" "$REPO_ROOT/scripts/leo_behavior_manifest.py" "$REPO_ROOT/scripts/leo_turn_execution_manifest.py"; do
   [ -f "$f" ] || continue
   if ! python3 -c "import ast, sys; ast.parse(open(sys.argv[1]).read())" "$f" 2>/dev/null; then
     echo "SYNTAX ERROR: $f"

hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py

@@ -71,6 +71,41 @@ def _trace(record: dict[str, Any]) -> None:
         pass
 
 
+def _retrieval_receipt_trace(value: Any) -> dict[str, Any] | None:
+    """Retain the exact read receipt without retaining claim or source bodies."""
+
+    if not isinstance(value, dict) or value.get("schema") != "livingip.teleoKbRetrievalReceipt.v1":
+        return None
+    consistency = value.get("read_consistency") if isinstance(value.get("read_consistency"), dict) else {}
+    counts = value.get("counts") if isinstance(value.get("counts"), dict) else {}
+    safe = {
+        "schema": value.get("schema"),
+        "query_sha256": value.get("query_sha256"),
+        "semantic_context_sha256": value.get("semantic_context_sha256"),
+        "artifact_state_sha256": value.get("artifact_state_sha256"),
+        "claim_ids": sorted(str(item) for item in value.get("claim_ids") or []),
+        "source_ids": sorted(str(item) for item in value.get("source_ids") or []),
+        "counts": {
+            key: item
+            for key, item in sorted(counts.items())
+            if isinstance(key, str) and isinstance(item, int) and not isinstance(item, bool)
+        },
+        "read_consistency": {
+            "status": consistency.get("status"),
+            "attempts": consistency.get("attempts"),
+            "database": consistency.get("database"),
+            "database_user": consistency.get("database_user"),
+            "system_identifier": consistency.get("system_identifier"),
+            "wal_lsn_before": consistency.get("wal_lsn_before"),
+            "wal_lsn_after": consistency.get("wal_lsn_after"),
+        },
+    }
+    safe["receipt_sha256"] = hashlib.sha256(
+        json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    ).hexdigest()
+    return safe
+
+
 def _failure_context(reason: str) -> str:
     return (
         '<leo_current_runtime_contracts source="teleo-kb" status="unavailable">\n'
@@ -246,6 +281,7 @@ def _load_database_snapshot(
         compiled_response = payload.get("compiled_response")
         if compiled_response is not None and not isinstance(compiled_response, str):
             raise ValueError("compiled_response_invalid")
+        retrieval_receipt = _retrieval_receipt_trace(payload.get("retrieval_receipt"))
     except (json.JSONDecodeError, TypeError, ValueError) as exc:
         record = base_record | {"status": "error", "error": str(exc), "injected": True}
         _trace(record)
@@ -259,6 +295,8 @@ def _load_database_snapshot(
         "contract_sha256": hashlib.sha256(contract_json.encode("utf-8")).hexdigest(),
         "compiled_response_available": bool(compiled_response),
     }
+    if retrieval_receipt is not None:
+        record["retrieval_receipt"] = retrieval_receipt
     _trace(record)
     return {
         "status": "ok",

scripts/leo_behavior_manifest.py

@@ -23,6 +23,7 @@ import yaml
 SCHEMA = "livingip.leoBehaviorManifest.v1"
 DEFAULT_PROFILE = Path("/home/teleo/.hermes/profiles/leoclean")
 DEFAULT_HERMES_ROOT = Path("/home/teleo/.hermes/hermes-agent")
+DEFAULT_DEPLOYMENT_ROOT = Path("/opt/teleo-eval/workspaces/deploy-infra")
 
 
 def sha256_bytes(value: bytes) -> str:
@@ -190,9 +191,14 @@ def component(
     }
 
 
-def build_manifest(profile: Path = DEFAULT_PROFILE, hermes_root: Path = DEFAULT_HERMES_ROOT) -> dict[str, Any]:
+def build_manifest(
+    profile: Path = DEFAULT_PROFILE,
+    hermes_root: Path = DEFAULT_HERMES_ROOT,
+    deployment_root: Path = DEFAULT_DEPLOYMENT_ROOT,
+) -> dict[str, Any]:
     profile = profile.resolve()
     hermes_root = hermes_root.resolve()
+    deployment_root = deployment_root.resolve()
     config = safe_model_config(profile / "config.yaml")
     components = {
         "profile_config": component(
@@ -271,6 +277,16 @@ def build_manifest(profile: Path = DEFAULT_PROFILE, hermes_root: Path = DEFAULT_
                 profile, (hermes_root / "run_agent.py", hermes_root / "agent" / "prompt_builder.py")
             ),
         },
+        "teleo_infrastructure_runtime": {
+            "git_head": git_head(deployment_root),
+            "source_tree": path_manifest(
+                profile,
+                (
+                    deployment_root / "scripts" / "leo_behavior_manifest.py",
+                    deployment_root / "scripts" / "leo_tool_trace.py",
+                ),
+            ),
+        },
         "components": components,
         "canonical_database": {
             "included": False,
@@ -282,6 +298,7 @@ def build_manifest(profile: Path = DEFAULT_PROFILE, hermes_root: Path = DEFAULT_
         "generated_at_utc": datetime.now(timezone.utc).isoformat(),
         "profile_root": str(profile),
         "hermes_root": str(hermes_root),
+        "deployment_root": str(deployment_root),
         "behavior_sha256": canonical_sha256(stable),
     }
 
@@ -290,9 +307,10 @@ def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--profile", type=Path, default=DEFAULT_PROFILE)
     parser.add_argument("--hermes-root", type=Path, default=DEFAULT_HERMES_ROOT)
+    parser.add_argument("--deployment-root", type=Path, default=DEFAULT_DEPLOYMENT_ROOT)
     parser.add_argument("--output", type=Path)
     args = parser.parse_args()
-    manifest = build_manifest(args.profile, args.hermes_root)
+    manifest = build_manifest(args.profile, args.hermes_root, args.deployment_root)
     encoded = json.dumps(manifest, indent=2, sort_keys=True) + "\n"
     if args.output:
         args.output.parent.mkdir(parents=True, exist_ok=True)

scripts/leo_turn_execution_manifest.py

@@ -0,0 +1,476 @@
+#!/usr/bin/env python3
+"""Bind one tested Leo turn to its runtime, model call, session, and KB read.
+
+The manifest is deliberately content-minimizing: prompts, replies, raw database
+rows, tool arguments, provider URLs, and session identifiers are represented by
+hashes. It proves attribution of a tested turn, not deterministic regeneration
+of externally hosted model weights.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import subprocess
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+SCHEMA = "livingip.leoTurnExecutionManifest.v1"
+RETRIEVAL_RECEIPT_SCHEMA = "livingip.teleoKbRetrievalReceipt.v1"
+
+
+def canonical_sha256(value: Any) -> str:
+    encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def text_sha256(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8")).hexdigest()
+
+
+def git_state(repo: Path) -> dict[str, Any]:
+    try:
+        head = subprocess.run(
+            ["git", "-C", str(repo), "rev-parse", "HEAD"],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+        status = subprocess.run(
+            ["git", "-C", str(repo), "status", "--porcelain", "--untracked-files=no"],
+            check=False,
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        return {"git_head": None, "tracked_tree_clean": None}
+    git_head = head.stdout.strip() if head.returncode == 0 and len(head.stdout.strip()) == 40 else None
+    return {
+        "git_head": git_head,
+        "tracked_tree_clean": status.returncode == 0 and not status.stdout.strip(),
+    }
+
+
+def _safe_usage(value: Any) -> dict[str, int | float | None]:
+    if not isinstance(value, dict):
+        return {}
+    safe: dict[str, int | float | None] = {}
+    for key in (
+        "input_tokens",
+        "output_tokens",
+        "prompt_tokens",
+        "completion_tokens",
+        "total_tokens",
+        "cache_read_tokens",
+        "cache_write_tokens",
+        "reasoning_tokens",
+    ):
+        item = value.get(key)
+        if isinstance(item, (int, float)) and not isinstance(item, bool):
+            safe[key] = item
+    return safe
+
+
+def _model_calls(result: dict[str, Any]) -> list[dict[str, Any]]:
+    calls: list[dict[str, Any]] = []
+    for raw in result.get("model_call_trace") or []:
+        if not isinstance(raw, dict) or raw.get("event") != "post_api_request":
+            continue
+        calls.append(
+            {
+                "api_call_count": raw.get("api_call_count"),
+                "api_mode": raw.get("api_mode"),
+                "model": raw.get("model"),
+                "provider": raw.get("provider"),
+                "response_model": raw.get("response_model"),
+                "finish_reason": raw.get("finish_reason"),
+                "message_count": raw.get("message_count"),
+                "assistant_content_chars": raw.get("assistant_content_chars"),
+                "assistant_tool_call_count": raw.get("assistant_tool_call_count"),
+                "task_id_sha256": raw.get("task_id_sha256"),
+                "session_id_sha256": raw.get("session_id_sha256"),
+                "base_url_sha256": raw.get("base_url_sha256"),
+                "usage": _safe_usage(raw.get("usage")),
+                "provider_response_id": raw.get("provider_response_id"),
+                "provider_response_id_status": raw.get("provider_response_id_status"),
+            }
+        )
+    return calls
+
+
+def _safe_retrieval_receipt(value: Any) -> dict[str, Any] | None:
+    if not isinstance(value, dict) or value.get("schema") != RETRIEVAL_RECEIPT_SCHEMA:
+        return None
+    consistency = value.get("read_consistency") if isinstance(value.get("read_consistency"), dict) else {}
+    counts = value.get("counts") if isinstance(value.get("counts"), dict) else {}
+    return {
+        "schema": value.get("schema"),
+        "query_sha256": value.get("query_sha256"),
+        "semantic_context_sha256": value.get("semantic_context_sha256"),
+        "artifact_state_sha256": value.get("artifact_state_sha256"),
+        "claim_ids_sha256": canonical_sha256(sorted(str(item) for item in value.get("claim_ids") or [])),
+        "source_ids_sha256": canonical_sha256(sorted(str(item) for item in value.get("source_ids") or [])),
+        "counts": {
+            key: item
+            for key, item in sorted(counts.items())
+            if isinstance(key, str) and isinstance(item, int) and not isinstance(item, bool)
+        },
+        "read_consistency": {
+            "status": consistency.get("status"),
+            "attempts": consistency.get("attempts"),
+            "database": consistency.get("database"),
+            "database_user": consistency.get("database_user"),
+            "system_identifier": consistency.get("system_identifier"),
+            "wal_lsn_before": consistency.get("wal_lsn_before"),
+            "wal_lsn_after": consistency.get("wal_lsn_after"),
+        },
+        "receipt_sha256": canonical_sha256(value),
+    }
+
+
+def _context_receipts(result: dict[str, Any]) -> list[dict[str, Any]]:
+    receipts = []
+    for record in result.get("database_context_trace") or []:
+        if not isinstance(record, dict) or record.get("event") != "pre_llm_call":
+            continue
+        receipt = _safe_retrieval_receipt(record.get("retrieval_receipt"))
+        if receipt is not None:
+            receipts.append(receipt)
+    return receipts
+
+
+def _tool_receipts(result: dict[str, Any]) -> list[dict[str, Any]]:
+    receipts = []
+    trace = result.get("database_tool_trace") or {}
+    for call in trace.get("calls") or []:
+        if not isinstance(call, dict):
+            continue
+        raw = ((call.get("result") or {}).get("retrieval_receipt") or {})
+        if not isinstance(raw, dict) or not raw.get("semantic_context_sha256"):
+            continue
+        receipts.append(
+            {
+                "schema": raw.get("schema"),
+                "semantic_context_sha256": raw.get("semantic_context_sha256"),
+                "artifact_state_sha256": raw.get("artifact_state_sha256"),
+                "read_consistency": {"status": raw.get("read_consistency_status")},
+                "receipt_sha256": canonical_sha256(raw),
+            }
+        )
+    return receipts
+
+
+def _database_required(result: dict[str, Any]) -> bool:
+    for record in result.get("database_context_trace") or []:
+        if not isinstance(record, dict) or record.get("event") != "pre_llm_call":
+            continue
+        contract_ids = {str(item) for item in record.get("contract_ids") or []}
+        if contract_ids - {"reply_budget"}:
+            return True
+    trace = result.get("database_tool_trace") or {}
+    return bool(trace.get("database_tool_call_count"))
+
+
+def _database_tool_binding(result: dict[str, Any]) -> dict[str, Any]:
+    trace = result.get("database_tool_trace") or {}
+    calls = []
+    for call in trace.get("calls") or []:
+        if not isinstance(call, dict):
+            continue
+        call_result = call.get("result") if isinstance(call.get("result"), dict) else {}
+        calls.append(
+            {
+                "tool_call_id_sha256": call.get("tool_call_id_sha256"),
+                "arguments_sha256": call.get("arguments_sha256"),
+                "database_invocations": call.get("database_invocations") or [],
+                "result_content_sha256": call_result.get("content_sha256"),
+                "result_content_bytes": call_result.get("content_bytes"),
+                "result_error_detected": call_result.get("error_detected"),
+                "result_row_ids_sha256": canonical_sha256(sorted(call_result.get("row_ids") or [])),
+                "result_sha256_values_sha256": canonical_sha256(sorted(call_result.get("sha256_values") or [])),
+            }
+        )
+    all_results_bound = bool(calls) and all(
+        call.get("result_content_sha256") and call.get("result_error_detected") is False for call in calls
+    )
+    return {
+        "schema": trace.get("schema"),
+        "tool_call_count": trace.get("database_tool_call_count", 0),
+        "completed_count": trace.get("database_tool_completed_count", 0),
+        "all_calls_read_only": trace.get("database_tool_calls_read_only"),
+        "all_results_content_bound": all_results_bound,
+        "calls": calls,
+        "trace_sha256": canonical_sha256(trace),
+    }
+
+
+def _database_fingerprint_complete(value: dict[str, Any]) -> bool:
+    consistency = value.get("read_consistency") if isinstance(value.get("read_consistency"), dict) else {}
+    return bool(
+        value.get("status") == "ok"
+        and value.get("fingerprint_sha256")
+        and value.get("table_count")
+        and value.get("total_rows")
+        and consistency.get("status") == "stable_wal_marker"
+    )
+
+
+def _required_missing(
+    *,
+    behavior: dict[str, Any],
+    calls: list[dict[str, Any]],
+    receipts: list[dict[str, Any]],
+    database_required: bool,
+    session_key: str,
+    prompt: str,
+    reply: str,
+    harness_source: dict[str, Any],
+    database_tool_binding: dict[str, Any],
+    db_fingerprint_before: dict[str, Any],
+    db_fingerprint_after: dict[str, Any],
+) -> list[str]:
+    missing = []
+    checks = {
+        "prompt": bool(prompt),
+        "reply": bool(reply),
+        "session_boundary": bool(session_key),
+        "behavior_sha256": bool(behavior.get("behavior_sha256")),
+        "hermes_git_head": bool((behavior.get("hermes_runtime") or {}).get("git_head")),
+        "teleo_infrastructure_git_head": bool(
+            (behavior.get("teleo_infrastructure_runtime") or {}).get("git_head")
+        ),
+        "harness_git_head": bool(harness_source.get("git_head")),
+        "actual_model_call": bool(
+            calls
+            and all(
+                call.get("model") and call.get("provider") and call.get("response_model")
+                for call in calls
+            )
+        ),
+        "database_retrieval_receipt": not database_required or bool(receipts),
+        "database_tool_results": not database_tool_binding.get("tool_call_count")
+        or bool(database_tool_binding.get("all_results_content_bound")),
+        "database_fingerprint_before": not database_required
+        or _database_fingerprint_complete(db_fingerprint_before),
+        "database_fingerprint_after": not database_required
+        or _database_fingerprint_complete(db_fingerprint_after),
+        "database_fingerprint_unchanged": not database_required
+        or db_fingerprint_before.get("fingerprint_sha256") == db_fingerprint_after.get("fingerprint_sha256"),
+    }
+    for label, present in checks.items():
+        if not present:
+            missing.append(label)
+    return missing
+
+
+def build_turn_manifest(
+    *,
+    result: dict[str, Any],
+    behavior_manifest: dict[str, Any],
+    session_key: str,
+    source: dict[str, Any],
+    suite_mode: str,
+    remote_run_id: str | None,
+    db_counts_before: dict[str, Any] | None,
+    db_counts_after: dict[str, Any] | None,
+    db_counts_changed: bool | None,
+    db_fingerprint_before: dict[str, Any] | None,
+    db_fingerprint_after: dict[str, Any] | None,
+    posted_to_telegram: bool,
+    mutates_kb_by_harness: bool,
+    harness_source: dict[str, Any],
+) -> dict[str, Any]:
+    prompt = str(result.get("prompt") or "")
+    reply = str(result.get("reply") or "")
+    calls = _model_calls(result)
+    context_receipts = _context_receipts(result)
+    tool_receipts = _tool_receipts(result)
+    receipts = context_receipts + tool_receipts
+    database_required = _database_required(result)
+    database_tool_binding = _database_tool_binding(result)
+    fingerprint_before = db_fingerprint_before or {}
+    fingerprint_after = db_fingerprint_after or {}
+    missing = _required_missing(
+        behavior=behavior_manifest,
+        calls=calls,
+        receipts=receipts,
+        database_required=database_required,
+        session_key=session_key,
+        prompt=prompt,
+        reply=reply,
+        harness_source=harness_source,
+        database_tool_binding=database_tool_binding,
+        db_fingerprint_before=fingerprint_before,
+        db_fingerprint_after=fingerprint_after,
+    )
+    stable = {
+        "schema": SCHEMA,
+        "turn": {
+            "prompt_id": result.get("prompt_id"),
+            "turn_index": result.get("turn"),
+            "remote_run_id": remote_run_id,
+            "started_at_utc": result.get("started_at_utc"),
+            "ended_at_utc": result.get("ended_at_utc"),
+            "prompt_sha256": text_sha256(prompt),
+            "reply_sha256": text_sha256(reply),
+            "prompt_bytes": len(prompt.encode("utf-8")),
+            "reply_bytes": len(reply.encode("utf-8")),
+        },
+        "session_boundary": {
+            "session_key_sha256": text_sha256(session_key) if session_key else None,
+            "source_platform": source.get("platform"),
+            "chat_type": source.get("chat_type"),
+            "fresh_temp_profile": True,
+            "prior_dynamic_state_excluded": True,
+        },
+        "runtime": {
+            "behavior_sha256": behavior_manifest.get("behavior_sha256"),
+            "hermes_runtime": behavior_manifest.get("hermes_runtime"),
+            "teleo_infrastructure_runtime": behavior_manifest.get("teleo_infrastructure_runtime"),
+            "profile_component_hashes": {
+                name: (component.get("content") or {}).get("sha256")
+                for name, component in sorted((behavior_manifest.get("components") or {}).items())
+                if isinstance(component, dict)
+            },
+            "harness_source": harness_source,
+        },
+        "model_execution": {
+            "call_count": len(calls),
+            "calls": calls,
+            "externally_hosted_weights_hashable": False,
+            "provider_response_id_required_for_attribution": False,
+        },
+        "canonical_database": {
+            "required_for_turn": database_required,
+            "binding_status": "retrieval_receipt_bound"
+            if receipts
+            else "not_required"
+            if not database_required
+            else "missing",
+            "context_retrieval_receipts": context_receipts,
+            "explicit_tool_retrieval_receipts": tool_receipts,
+            "database_tool_binding": database_tool_binding,
+            "fingerprint_before": fingerprint_before,
+            "fingerprint_after": fingerprint_after,
+            "fingerprint_unchanged": fingerprint_before.get("fingerprint_sha256")
+            == fingerprint_after.get("fingerprint_sha256"),
+            "suite_counts_before_sha256": canonical_sha256(db_counts_before or {}),
+            "suite_counts_after_sha256": canonical_sha256(db_counts_after or {}),
+            "suite_counts_changed": db_counts_changed,
+            "full_database_content_fingerprint_included": bool(
+                fingerprint_before.get("fingerprint_sha256") and fingerprint_after.get("fingerprint_sha256")
+            ),
+            "full_database_snapshot_artifact_included": False,
+        },
+        "delivery_and_safety": {
+            "suite_mode": suite_mode,
+            "posted_to_telegram": posted_to_telegram,
+            "kb_mutation_by_harness": mutates_kb_by_harness,
+            "raw_prompt_retained_in_manifest": False,
+            "raw_reply_retained_in_manifest": False,
+            "raw_database_rows_retained_in_manifest": False,
+            "raw_session_identifier_retained_in_manifest": False,
+        },
+        "attribution": {
+            "status": "complete" if not missing else "incomplete",
+            "missing_required_bindings": missing,
+        },
+        "replay_claim": {
+            "status": "bounded_recorded_inputs_pinned" if not missing else "incomplete_inputs",
+            "local_runtime_inputs_reconstructible": not missing,
+            "database_state_identified_by_content_fingerprint": bool(
+                not missing and fingerprint_before.get("fingerprint_sha256")
+            ),
+            "database_snapshot_reconstructible_from_this_manifest_alone": False,
+            "external_model_state_reconstructible": False,
+            "bitwise_identical_model_output_guaranteed": False,
+            "reason": (
+                "The tested turn is attributable to exact local/runtime inputs and DB reads, but hosted model "
+                "weights and sampling are not locally hashable or deterministic."
+            ),
+        },
+    }
+    return {
+        **stable,
+        "generated_at_utc": datetime.now(UTC).isoformat(),
+        "execution_sha256": canonical_sha256(stable),
+    }
+
+
+def validate_turn_manifest(manifest: dict[str, Any]) -> list[str]:
+    problems = []
+    if manifest.get("schema") != SCHEMA:
+        problems.append("schema_mismatch")
+    stable = {
+        key: value
+        for key, value in manifest.items()
+        if key not in {"generated_at_utc", "execution_sha256"}
+    }
+    if manifest.get("execution_sha256") != canonical_sha256(stable):
+        problems.append("execution_sha256_mismatch")
+    missing = (manifest.get("attribution") or {}).get("missing_required_bindings")
+    expected_status = "complete" if not missing else "incomplete"
+    if (manifest.get("attribution") or {}).get("status") != expected_status:
+        problems.append("attribution_status_mismatch")
+    return problems
+
+
+def attach_execution_manifests(report: dict[str, Any], *, repo_root: Path) -> dict[str, Any]:
+    behavior = report.get("executed_behavior_manifest") or report.get("live_behavior_manifest_before") or {}
+    handler = report.get("handler") or {}
+    harness_source = git_state(repo_root)
+    results = report.get("results") or []
+    complete = 0
+    for result in results:
+        if not isinstance(result, dict):
+            continue
+        manifest = build_turn_manifest(
+            result=result,
+            behavior_manifest=behavior,
+            session_key=str(handler.get("session_key") or ""),
+            source=report.get("source") or {},
+            suite_mode=str(report.get("mode") or ""),
+            remote_run_id=report.get("remote_run_id"),
+            db_counts_before=report.get("db_counts_before"),
+            db_counts_after=report.get("db_counts_after"),
+            db_counts_changed=report.get("db_counts_changed"),
+            db_fingerprint_before=report.get("db_fingerprint_before"),
+            db_fingerprint_after=report.get("db_fingerprint_after"),
+            posted_to_telegram=bool(report.get("posted_to_telegram")),
+            mutates_kb_by_harness=bool(report.get("mutates_kb_by_harness")),
+            harness_source=harness_source,
+        )
+        result["execution_manifest"] = manifest
+        if manifest["attribution"]["status"] == "complete" and not validate_turn_manifest(manifest):
+            complete += 1
+    report["execution_manifest_summary"] = {
+        "schema": SCHEMA,
+        "turn_count": len(results),
+        "attribution_complete_count": complete,
+        "all_turns_attribution_complete": bool(results) and complete == len(results),
+        "harness_source": harness_source,
+    }
+    return report
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--suite-report", required=True, type=Path)
+    parser.add_argument("--repo-root", default=Path(__file__).resolve().parents[1], type=Path)
+    parser.add_argument("--output", required=True, type=Path)
+    args = parser.parse_args()
+    report = json.loads(args.suite_report.read_text(encoding="utf-8"))
+    attach_execution_manifests(report, repo_root=args.repo_root)
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
+    summary = report["execution_manifest_summary"]
+    print(json.dumps(summary, indent=2, sort_keys=True))
+    return 0 if summary["all_turns_attribution_complete"] else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_leo_direct_claim_handler_suite.py

@@ -20,6 +20,11 @@ from typing import Any
 
 import working_leo_open_ended_benchmark as benchmark
 
+try:
+    from leo_turn_execution_manifest import attach_execution_manifests
+except ImportError:  # pragma: no cover - imported as scripts.* in tests
+    from scripts.leo_turn_execution_manifest import attach_execution_manifests
+
 ROOT = Path(__file__).resolve().parents[1]
 REPORT_DIR = ROOT / "docs" / "reports" / "leo-working-state-20260709"
 OUTPUT_JSON = REPORT_DIR / "telegram-handler-direct-claim-suite-current.json"
@@ -40,9 +45,14 @@ SECRET_PATTERNS = [
     re.compile(r"((?:api[_-]?key|token|secret|password)[=:]\s*)([A-Za-z0-9._-]+)", re.IGNORECASE),
 ]
 
+DB_CONTEXT_PLUGIN = (
+    ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
+)
+
 
 REMOTE_SCRIPT = r'''
 import asyncio
+import hashlib
 import json
 import os
 import shutil
@@ -63,6 +73,83 @@ RUN_ID = "__RUN_ID__"
 PROMPTS = __PROMPTS_JSON__
 REPORT_PREFIX = __REPORT_PREFIX_JSON__
 REPORT_PATH = Path(f"/tmp/{REPORT_PREFIX}-{RUN_ID}.json")
+DB_CONTEXT_PLUGIN_SOURCE = __DB_CONTEXT_PLUGIN_SOURCE_JSON__
+TURN_TRACE_PLUGIN_SOURCE = """\
+import hashlib
+import json
+import os
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+def _safe_usage(value):
+    if not isinstance(value, dict):
+        return {}
+    safe = {}
+    for key in (
+        "input_tokens",
+        "output_tokens",
+        "prompt_tokens",
+        "completion_tokens",
+        "total_tokens",
+        "cache_read_tokens",
+        "cache_write_tokens",
+        "reasoning_tokens",
+    ):
+        item = value.get(key)
+        if isinstance(item, (int, float)) and not isinstance(item, bool):
+            safe[key] = item
+    return safe
+
+
+def _sha(value):
+    return hashlib.sha256(str(value or "").encode("utf-8")).hexdigest()
+
+
+def _post_api_request(**kwargs):
+    raw_path = os.getenv("LEO_TURN_API_TRACE_PATH", "").strip()
+    if not raw_path:
+        return None
+    path = Path(raw_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    record = {
+        "generated_at_utc": datetime.now(timezone.utc).isoformat(),
+        "event": "post_api_request",
+        "task_id_sha256": _sha(kwargs.get("task_id")),
+        "session_id_sha256": _sha(kwargs.get("session_id")),
+        "model": kwargs.get("model"),
+        "provider": kwargs.get("provider"),
+        "base_url_sha256": _sha(kwargs.get("base_url")),
+        "api_mode": kwargs.get("api_mode"),
+        "api_call_count": kwargs.get("api_call_count"),
+        "finish_reason": kwargs.get("finish_reason"),
+        "message_count": kwargs.get("message_count"),
+        "response_model": kwargs.get("response_model"),
+        "usage": _safe_usage(kwargs.get("usage")),
+        "assistant_content_chars": kwargs.get("assistant_content_chars"),
+        "assistant_tool_call_count": kwargs.get("assistant_tool_call_count"),
+        "provider_response_id": None,
+        "provider_response_id_status": "not_exposed_by_hermes_post_api_request_hook",
+    }
+    with path.open("a", encoding="utf-8") as handle:
+        handle.write(json.dumps(record, sort_keys=True) + "\\n")
+    try:
+        path.chmod(0o600)
+    except OSError:
+        pass
+    return None
+
+
+def register(ctx):
+    ctx.register_hook("post_api_request", _post_api_request)
+"""
+TURN_TRACE_PLUGIN_MANIFEST = """\
+name: leo-turn-execution-trace
+version: "1.0.0"
+description: Secret-safe model-call trace for no-send Leo harnesses.
+provides_hooks:
+  - post_api_request
+"""
 
 
 def service_state():
@@ -114,6 +201,127 @@ select jsonb_build_object(
         return {"error": str(exc)}
 
 
+def db_read_marker():
+    sql = """
+select jsonb_build_object(
+  'database', current_database(),
+  'database_user', current_user,
+  'system_identifier', (select system_identifier::text from pg_control_system()),
+  'wal_lsn', pg_current_wal_lsn()::text
+)::text;
+"""
+    raw = subprocess.check_output(
+        ["docker", "exec", "-i", "teleo-pg", "psql", "-U", "postgres", "-d", "teleo", "-At", "-q"],
+        input=sql,
+        text=True,
+        stderr=subprocess.STDOUT,
+    )
+    return json.loads(raw.strip())
+
+
+def db_fingerprint():
+    sql_path = DEPLOY_ROOT / "ops" / "postgres_parity_manifest.sql"
+    before = db_read_marker()
+    raw = subprocess.check_output(
+        ["docker", "exec", "-i", "teleo-pg", "psql", "-U", "postgres", "-d", "teleo", "-At", "-q"],
+        input=sql_path.read_text(encoding="utf-8"),
+        text=True,
+        stderr=subprocess.STDOUT,
+        timeout=180,
+    )
+    after = db_read_marker()
+    rows = []
+    for line in raw.splitlines():
+        candidate = line.strip()
+        if not candidate or candidate in {"BEGIN", "SET", "ROLLBACK"}:
+            continue
+        value = json.loads(candidate)
+        if isinstance(value, dict):
+            rows.append(value)
+    tables = {
+        f"{row['schema']}.{row['table']}": {
+            "row_count": int(row["row_count"]),
+            "rowset_md5": row["rowset_md5"],
+        }
+        for row in rows
+        if row.get("kind") == "table"
+    }
+    singleton = {
+        str(row["kind"]): row.get("items", [])
+        for row in rows
+        if row.get("kind")
+        in {
+            "schemas",
+            "extensions",
+            "application_roles",
+            "columns",
+            "constraints",
+            "indexes",
+            "sequences",
+            "views",
+            "functions",
+            "triggers",
+            "types",
+            "policies",
+        }
+    }
+    identity_rows = [row for row in rows if row.get("kind") == "identity"]
+    if len(identity_rows) != 1 or not tables:
+        raise RuntimeError("canonical database fingerprint manifest is incomplete")
+    identity = identity_rows[0]
+    stable = {
+        "identity": {
+            key: identity.get(key)
+            for key in (
+                "database",
+                "server_version_num",
+                "server_encoding",
+                "database_collation",
+                "database_ctype",
+                "transaction_read_only",
+            )
+        },
+        "singleton_sha256": {
+            key: hashlib.sha256(
+                json.dumps(value, sort_keys=True, separators=(",", ":")).encode("utf-8")
+            ).hexdigest()
+            for key, value in sorted(singleton.items())
+        },
+        "tables": tables,
+    }
+    return {
+        "schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
+        "status": "ok",
+        "fingerprint_sha256": hashlib.sha256(
+            json.dumps(stable, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        ).hexdigest(),
+        "table_count": len(tables),
+        "total_rows": sum(item["row_count"] for item in tables.values()),
+        "table_rows_sha256": hashlib.sha256(
+            json.dumps(tables, sort_keys=True, separators=(",", ":")).encode("utf-8")
+        ).hexdigest(),
+        "structure_sha256": hashlib.sha256(
+            json.dumps(stable["singleton_sha256"], sort_keys=True, separators=(",", ":")).encode("utf-8")
+        ).hexdigest(),
+        "read_consistency": {
+            "status": "stable_wal_marker" if before == after else "wal_changed_during_fingerprint",
+            "before": before,
+            "after": after,
+        },
+    }
+
+
+def safe_db_fingerprint():
+    try:
+        return db_fingerprint()
+    except Exception as exc:
+        return {
+            "schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
+            "status": "error",
+            "error": type(exc).__name__,
+        }
+
+
 def copy_profile() -> Path:
     tmp_root = Path(tempfile.mkdtemp(prefix="leo-direct-claim-handler-"))
     target = tmp_root / "profile"
@@ -155,6 +363,31 @@ def copy_profile() -> Path:
     return target
 
 
+def install_harness_plugins(profile):
+    db_context = profile / "plugins" / "leo-db-context" / "__init__.py"
+    db_context.parent.mkdir(parents=True, exist_ok=True)
+    db_context.write_text(DB_CONTEXT_PLUGIN_SOURCE, encoding="utf-8")
+
+    turn_trace = profile / "plugins" / "leo-turn-execution-trace"
+    turn_trace.mkdir(parents=True, exist_ok=True)
+    (turn_trace / "__init__.py").write_text(TURN_TRACE_PLUGIN_SOURCE, encoding="utf-8")
+    (turn_trace / "plugin.yaml").write_text(TURN_TRACE_PLUGIN_MANIFEST, encoding="utf-8")
+
+
+def read_jsonl(path):
+    if not path or not path.exists():
+        return []
+    records = []
+    for line in path.read_text(encoding="utf-8").splitlines():
+        try:
+            value = json.loads(line)
+        except json.JSONDecodeError:
+            continue
+        if isinstance(value, dict):
+            records.append(value)
+    return records
+
+
 def remove_tree(path):
     if not path or not path.exists():
         return True
@@ -192,17 +425,7 @@ def write_report_snapshot(report):
 
 
 def read_database_context_trace(path):
-    if not path or not path.exists():
+    return read_jsonl(path)
-        return []
-    records = []
-    for line in path.read_text(encoding="utf-8").splitlines():
-        try:
-            value = json.loads(line)
-        except json.JSONDecodeError:
-            continue
-        if isinstance(value, dict):
-            records.append(value)
-    return records
 
 
 def current_agent_messages(runner, session_key):
@@ -222,14 +445,47 @@ def current_agent_messages(runner, session_key):
 
 async def run_suite():
     sys.path.insert(0, str(DEPLOY_ROOT / "scripts"))
-    from leo_behavior_manifest import build_manifest as build_behavior_manifest
+    import leo_behavior_manifest as behavior_manifest_module
     from leo_tool_trace import extract_kb_tool_trace
 
+    def build_behavior_manifest(profile):
+        try:
+            return behavior_manifest_module.build_manifest(profile, AGENT_ROOT, DEPLOY_ROOT)
+        except TypeError:
+            manifest = behavior_manifest_module.build_manifest(profile, AGENT_ROOT)
+            manifest["teleo_infrastructure_runtime"] = {
+                "git_head": behavior_manifest_module.git_head(DEPLOY_ROOT),
+                "source_tree": behavior_manifest_module.path_manifest(
+                    profile,
+                    (
+                        DEPLOY_ROOT / "scripts" / "leo_behavior_manifest.py",
+                        DEPLOY_ROOT / "scripts" / "leo_tool_trace.py",
+                    ),
+                ),
+            }
+            stable = {
+                key: value
+                for key, value in manifest.items()
+                if key
+                not in {
+                    "generated_at_utc",
+                    "profile_root",
+                    "hermes_root",
+                    "deployment_root",
+                    "behavior_sha256",
+                }
+            }
+            manifest["deployment_root"] = str(DEPLOY_ROOT)
+            manifest["behavior_sha256"] = behavior_manifest_module.canonical_sha256(stable)
+            return manifest
+
     before_service = service_state()
     before_counts = db_counts()
-    behavior_before = build_behavior_manifest(LIVE_PROFILE, AGENT_ROOT)
+    before_fingerprint = safe_db_fingerprint()
+    behavior_before = build_behavior_manifest(LIVE_PROFILE)
     temp_profile = None
     context_trace_path = None
+    model_trace_path = None
     temp_removed = False
     report = {
         "generated_at_utc": datetime.now(timezone.utc).isoformat(),
@@ -264,16 +520,21 @@ async def run_suite():
             "This proves handler-level live VPS GatewayRunner behavior, not human-visible Telegram delivery.",
         ],
         "db_counts_before": before_counts,
+        "db_fingerprint_before": before_fingerprint,
         "before_service": before_service,
         "remote_report_path": str(REPORT_PATH),
     }
     write_report_snapshot(report)
     try:
         temp_profile = copy_profile()
+        install_harness_plugins(temp_profile)
         os.environ["HERMES_HOME"] = str(temp_profile)
         os.environ["HOME"] = "/home/teleo"
         context_trace_path = temp_profile / "state" / "leo-db-context-trace.jsonl"
+        model_trace_path = temp_profile / "state" / "leo-turn-api-trace.jsonl"
         os.environ["LEO_DB_CONTEXT_TRACE_PATH"] = str(context_trace_path)
+        os.environ["LEO_TURN_API_TRACE_PATH"] = str(model_trace_path)
+        report["executed_behavior_manifest"] = build_behavior_manifest(temp_profile)
         sys.path.insert(0, str(AGENT_ROOT))
 
         from gateway.config import Platform
@@ -308,6 +569,7 @@ async def run_suite():
         results = []
         for index, prompt in enumerate(PROMPTS, start=1):
             trace_start = len(read_database_context_trace(context_trace_path))
+            model_trace_start = len(read_jsonl(model_trace_path))
             event = MessageEvent(
                 text=prompt["message"],
                 message_type=MessageType.TEXT,
@@ -333,6 +595,7 @@ async def run_suite():
                     "claim_ceiling": "Live VPS GatewayRunner reply from temp leoclean profile; no Telegram post; no production DB apply authorized.",
                     "database_context_trace": read_database_context_trace(context_trace_path)[trace_start:],
                     "database_tool_trace": database_tool_trace,
+                    "model_call_trace": read_jsonl(model_trace_path)[model_trace_start:],
                 }
             )
             report["results"] = results
@@ -348,10 +611,15 @@ async def run_suite():
         return report
     finally:
         after_counts = db_counts()
+        after_fingerprint = safe_db_fingerprint()
         after_service = service_state()
-        behavior_after = build_behavior_manifest(LIVE_PROFILE, AGENT_ROOT)
+        behavior_after = build_behavior_manifest(LIVE_PROFILE)
         report["db_counts_after"] = after_counts
         report["db_counts_changed"] = after_counts != before_counts
+        report["db_fingerprint_after"] = after_fingerprint
+        report["db_fingerprint_unchanged"] = (
+            before_fingerprint.get("fingerprint_sha256") == after_fingerprint.get("fingerprint_sha256")
+        )
         report["live_behavior_manifest_after"] = behavior_after
         report["changed_live_profile"] = behavior_before["behavior_sha256"] != behavior_after["behavior_sha256"]
         report["live_behavior_manifest_unchanged"] = not report["changed_live_profile"]
@@ -361,11 +629,20 @@ async def run_suite():
             "unchanged_from_preexisting_live_readback": before_service == after_service,
         }
         database_context_trace = read_database_context_trace(context_trace_path)
+        model_call_trace = read_jsonl(model_trace_path)
         if temp_profile is not None:
             tmp_root = temp_profile.parent
             temp_removed = remove_tree(tmp_root)
         report["temp_profile_removed"] = temp_removed
         report["database_context_trace"] = database_context_trace
+        report["model_call_trace_count"] = len(model_call_trace)
+        report["model_call_trace_all_bound"] = bool(model_call_trace) and all(
+            item.get("event") == "post_api_request"
+            and item.get("model")
+            and item.get("provider")
+            and item.get("response_model")
+            for item in model_call_trace
+        )
         context_injections = [
             item for item in database_context_trace if item.get("event", "pre_llm_call") == "pre_llm_call"
         ]
@@ -443,6 +720,7 @@ def build_remote_script(
         .replace("__SUITE_MODE_JSON__", json.dumps(suite_mode))
         .replace("__REPORT_PREFIX_JSON__", json.dumps(report_prefix))
         .replace("__PROMPT_NOTE_JSON__", json.dumps(prompt_note))
+        .replace("__DB_CONTEXT_PLUGIN_SOURCE_JSON__", json.dumps(DB_CONTEXT_PLUGIN.read_text(encoding="utf-8")))
     )
 
 
@@ -561,6 +839,7 @@ def write_output(remote: dict[str, Any], *, output_json: Path = OUTPUT_JSON) ->
     report["remote_run_id"] = remote.get("run_id")
     if remote.get("stderr"):
         report["remote_stderr"] = remote["stderr"]
+    attach_execution_manifests(report, repo_root=ROOT)
     output_json.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8")
     return report
 

tests/test_hermes_leoclean_db_context_plugin.py

@@ -43,6 +43,24 @@ def test_plugin_injects_only_operational_contracts_and_writes_redacted_trace(tmp
             {"id": "source_intake", "capability_tier": "build-local compiler only"},
         ],
         "compiled_response": "must not be injected either",
+        "retrieval_receipt": {
+            "schema": "livingip.teleoKbRetrievalReceipt.v1",
+            "query_sha256": "1" * 64,
+            "semantic_context_sha256": "2" * 64,
+            "artifact_state_sha256": "3" * 64,
+            "claim_ids": ["claim-1"],
+            "source_ids": ["source-1"],
+            "counts": {"claims": 1, "sources": 1},
+            "read_consistency": {
+                "status": "stable_wal_marker",
+                "attempts": 1,
+                "database": "teleo",
+                "database_user": "runtime",
+                "system_identifier": "system-1",
+                "wal_lsn_before": "0/123",
+                "wal_lsn_after": "0/123",
+            },
+        },
     }
     calls = []
 
@@ -63,6 +81,9 @@ def test_plugin_injects_only_operational_contracts_and_writes_redacted_trace(tmp
     assert record["status"] == "ok"
     assert record["injected"] is True
     assert record["contract_ids"] == ["reply_budget", "source_intake"]
+    assert record["retrieval_receipt"]["semantic_context_sha256"] == "2" * 64
+    assert record["retrieval_receipt"]["read_consistency"]["system_identifier"] == "system-1"
+    assert len(record["retrieval_receipt"]["receipt_sha256"]) == 64
     command, kwargs = calls[0]
     assert command[2:5] == ["--local", "context", "operator secret-shaped but nonsecret message"]
     assert command[-6:] == ["--limit", "0", "--context-limit", "0", "--format", "json"]
@@ -475,3 +496,9 @@ def test_handler_harness_retains_database_context_proof_fields() -> None:
     assert '"database_tool_call_proven"' in source
     assert '"database_retrieval_receipt_proven"' in source
     assert '"database_tool_calls_read_only"' in source
+    assert "LEO_TURN_API_TRACE_PATH" in source
+    assert '"model_call_trace"' in source
+    assert "attach_execution_manifests" in source
+    assert '"db_fingerprint_before"' in source
+    assert '"db_fingerprint_after"' in source
+    assert '"db_fingerprint_unchanged"' in source

tests/test_leo_behavior_manifest.py

@@ -16,6 +16,7 @@ def _write(path: Path, value: str) -> None:
 def test_behavior_manifest_is_deterministic_and_never_emits_config_secrets(tmp_path: Path) -> None:
     profile = tmp_path / "profile"
     hermes = tmp_path / "hermes"
+    deployment = tmp_path / "deployment"
     _write(
         profile / "config.yaml",
         """model:
@@ -40,12 +41,15 @@ gateway:
     _write(profile / "state.db", "runtime-state")
     _write(hermes / "run_agent.py", "# runtime\n")
     _write(hermes / "agent" / "prompt_builder.py", "# prompts\n")
+    _write(deployment / "scripts" / "leo_behavior_manifest.py", "# manifest\n")
+    _write(deployment / "scripts" / "leo_tool_trace.py", "# tool trace\n")
 
-    first = manifest.build_manifest(profile, hermes)
+    first = manifest.build_manifest(profile, hermes, deployment)
-    second = manifest.build_manifest(profile, hermes)
+    second = manifest.build_manifest(profile, hermes, deployment)
 
     assert first["behavior_sha256"] == second["behavior_sha256"]
     assert first["model_runtime"]["default_model"] == "anthropic/claude-sonnet-4-6"
+    assert first["teleo_infrastructure_runtime"]["source_tree"]["file_count"] == 2
     assert first["components"]["persistent_memory"]["content"]["file_count"] == 1
     encoded = json.dumps(first)
     assert "super-secret-token" not in encoded
@@ -56,14 +60,17 @@ gateway:
 def test_behavior_manifest_changes_when_a_behavior_layer_changes(tmp_path: Path) -> None:
     profile = tmp_path / "profile"
     hermes = tmp_path / "hermes"
+    deployment = tmp_path / "deployment"
     _write(profile / "config.yaml", "model:\n  provider: openrouter\n  default: model-a\n")
     _write(profile / "SOUL.md", "identity-a\n")
     _write(hermes / "run_agent.py", "runtime\n")
     _write(hermes / "agent" / "prompt_builder.py", "prompts\n")
+    _write(deployment / "scripts" / "leo_behavior_manifest.py", "manifest\n")
+    _write(deployment / "scripts" / "leo_tool_trace.py", "tool trace\n")
 
-    before = manifest.build_manifest(profile, hermes)
+    before = manifest.build_manifest(profile, hermes, deployment)
     _write(profile / "SOUL.md", "identity-b\n")
-    after = manifest.build_manifest(profile, hermes)
+    after = manifest.build_manifest(profile, hermes, deployment)
 
     assert before["behavior_sha256"] != after["behavior_sha256"]
     assert (
@@ -75,6 +82,7 @@ def test_behavior_manifest_changes_when_a_behavior_layer_changes(tmp_path: Path)
 def test_behavior_manifest_hashes_content_behind_symlinks(tmp_path: Path) -> None:
     profile = tmp_path / "profile"
     hermes = tmp_path / "hermes"
+    deployment = tmp_path / "deployment"
     target = tmp_path / "deployed-skill.md"
     _write(profile / "config.yaml", "model:\n  provider: test\n")
     _write(profile / "SOUL.md", "identity\n")
@@ -83,10 +91,12 @@ def test_behavior_manifest_hashes_content_behind_symlinks(tmp_path: Path) -> Non
     (profile / "skills" / "SKILL.md").symlink_to(target)
     _write(hermes / "run_agent.py", "runtime\n")
     _write(hermes / "agent" / "prompt_builder.py", "prompts\n")
+    _write(deployment / "scripts" / "leo_behavior_manifest.py", "manifest\n")
+    _write(deployment / "scripts" / "leo_tool_trace.py", "tool trace\n")
 
-    before = manifest.build_manifest(profile, hermes)
+    before = manifest.build_manifest(profile, hermes, deployment)
     _write(target, "version-b\n")
-    after = manifest.build_manifest(profile, hermes)
+    after = manifest.build_manifest(profile, hermes, deployment)
 
     assert before["behavior_sha256"] != after["behavior_sha256"]
     symlinks = after["components"]["procedural_skills"]["content"]["symlinks"]

tests/test_leo_turn_execution_manifest.py

@@ -0,0 +1,170 @@
+from __future__ import annotations
+
+import copy
+
+from scripts import leo_turn_execution_manifest as manifest
+
+
+def behavior_manifest() -> dict:
+    return {
+        "behavior_sha256": "a" * 64,
+        "hermes_runtime": {"git_head": "b" * 40, "source_tree": {"sha256": "c" * 64}},
+        "teleo_infrastructure_runtime": {
+            "git_head": "d" * 40,
+            "source_tree": {"sha256": "e" * 64},
+        },
+        "components": {
+            "runtime_identity": {"content": {"sha256": "f" * 64}},
+            "procedural_skills": {"content": {"sha256": "1" * 64}},
+        },
+    }
+
+
+def retrieval_receipt() -> dict:
+    return {
+        "schema": "livingip.teleoKbRetrievalReceipt.v1",
+        "query_sha256": "2" * 64,
+        "semantic_context_sha256": "3" * 64,
+        "artifact_state_sha256": "4" * 64,
+        "claim_ids": ["claim-b", "claim-a"],
+        "source_ids": ["source-a"],
+        "counts": {"claims": 2, "context_rows": 1},
+        "read_consistency": {
+            "status": "stable_wal_marker",
+            "attempts": 1,
+            "database": "teleo",
+            "database_user": "runtime",
+            "system_identifier": "system-1",
+            "wal_lsn_before": "0/123",
+            "wal_lsn_after": "0/123",
+        },
+    }
+
+
+def turn_result() -> dict:
+    return {
+        "turn": 1,
+        "prompt_id": "OOS-01",
+        "prompt": "Challenge the weak acquisition claim.",
+        "reply": "The current evidence is shallow; here is a narrower candidate claim.",
+        "started_at_utc": "2026-07-14T12:00:00+00:00",
+        "ended_at_utc": "2026-07-14T12:00:02+00:00",
+        "database_context_trace": [
+            {
+                "event": "pre_llm_call",
+                "contract_ids": ["reply_budget", "source_link_audit"],
+                "retrieval_receipt": retrieval_receipt(),
+            }
+        ],
+        "database_tool_trace": {"database_tool_call_count": 0, "calls": []},
+        "model_call_trace": [
+            {
+                "event": "post_api_request",
+                "api_call_count": 1,
+                "api_mode": "openai_chat_completions",
+                "model": "anthropic/claude-sonnet-4-6",
+                "provider": "openrouter",
+                "response_model": "anthropic/claude-sonnet-4-6",
+                "finish_reason": "stop",
+                "message_count": 4,
+                "assistant_content_chars": 120,
+                "assistant_tool_call_count": 0,
+                "task_id_sha256": "5" * 64,
+                "session_id_sha256": "6" * 64,
+                "base_url_sha256": "7" * 64,
+                "usage": {"input_tokens": 100, "output_tokens": 20, "ignored": "secret-shaped"},
+                "provider_response_id": None,
+                "provider_response_id_status": "not_exposed_by_hermes_post_api_request_hook",
+            }
+        ],
+    }
+
+
+def database_fingerprint() -> dict:
+    marker = {
+        "database": "teleo",
+        "database_user": "runtime",
+        "system_identifier": "system-1",
+        "wal_lsn": "0/123",
+    }
+    return {
+        "schema": "livingip.teleoCanonicalDatabaseFingerprint.v1",
+        "status": "ok",
+        "fingerprint_sha256": "9" * 64,
+        "table_count": 39,
+        "total_rows": 52167,
+        "table_rows_sha256": "a" * 64,
+        "structure_sha256": "b" * 64,
+        "read_consistency": {"status": "stable_wal_marker", "before": marker, "after": marker},
+    }
+
+
+def build(result: dict | None = None, *, fingerprint: dict | None = None) -> dict:
+    selected_fingerprint = database_fingerprint() if fingerprint is None else fingerprint
+    return manifest.build_turn_manifest(
+        result=result or turn_result(),
+        behavior_manifest=behavior_manifest(),
+        session_key="agent:main:telegram:group:private",
+        source={"platform": "telegram", "chat_type": "group"},
+        suite_mode="live_vps_gatewayrunner_temp_profile",
+        remote_run_id="run-1",
+        db_counts_before={"public.claims": 1837},
+        db_counts_after={"public.claims": 1837},
+        db_counts_changed=False,
+        db_fingerprint_before=selected_fingerprint,
+        db_fingerprint_after=selected_fingerprint,
+        posted_to_telegram=False,
+        mutates_kb_by_harness=False,
+        harness_source={"git_head": "8" * 40, "tracked_tree_clean": True},
+    )
+
+
+def test_turn_manifest_binds_model_runtime_session_and_exact_database_read() -> None:
+    value = build()
+
+    assert value["attribution"]["status"] == "complete"
+    assert value["replay_claim"]["status"] == "bounded_recorded_inputs_pinned"
+    assert value["model_execution"]["calls"][0]["response_model"] == "anthropic/claude-sonnet-4-6"
+    assert value["model_execution"]["calls"][0]["usage"] == {"input_tokens": 100, "output_tokens": 20}
+    assert value["canonical_database"]["binding_status"] == "retrieval_receipt_bound"
+    assert value["canonical_database"]["fingerprint_before"]["table_count"] == 39
+    assert value["canonical_database"]["fingerprint_unchanged"] is True
+    receipt = value["canonical_database"]["context_retrieval_receipts"][0]
+    assert receipt["read_consistency"]["system_identifier"] == "system-1"
+    assert receipt["read_consistency"]["wal_lsn_before"] == receipt["read_consistency"]["wal_lsn_after"]
+    assert value["delivery_and_safety"]["posted_to_telegram"] is False
+    assert value["delivery_and_safety"]["kb_mutation_by_harness"] is False
+    assert manifest.validate_turn_manifest(value) == []
+
+    encoded = str(value)
+    assert "Challenge the weak acquisition claim" not in encoded
+    assert "narrower candidate claim" not in encoded
+    assert "agent:main:telegram" not in encoded
+    assert "secret-shaped" not in encoded
+
+
+def test_turn_manifest_detects_missing_actual_model_call() -> None:
+    result = turn_result()
+    result["model_call_trace"] = []
+
+    value = build(result)
+
+    assert value["attribution"]["status"] == "incomplete"
+    assert "actual_model_call" in value["attribution"]["missing_required_bindings"]
+    assert value["replay_claim"]["local_runtime_inputs_reconstructible"] is False
+
+
+def test_turn_manifest_hash_detects_tampering() -> None:
+    value = build()
+    tampered = copy.deepcopy(value)
+    tampered["turn"]["reply_sha256"] = "0" * 64
+
+    assert manifest.validate_turn_manifest(tampered) == ["execution_sha256_mismatch"]
+
+
+def test_database_question_requires_a_stable_full_database_fingerprint() -> None:
+    value = build(fingerprint={"status": "error"})
+
+    assert value["attribution"]["status"] == "incomplete"
+    assert "database_fingerprint_before" in value["attribution"]["missing_required_bindings"]
+    assert "database_fingerprint_after" in value["attribution"]["missing_required_bindings"]