From cc46d66713f18ce880a921c7bc46546e5f7e7bf2 Mon Sep 17 00:00:00 2001 From: twentyOne2x Date: Tue, 14 Jul 2026 23:45:22 +0200 Subject: [PATCH] feat: bind tested Leo turns to reproducible inputs --- deploy/auto-deploy.sh | 2 +- deploy/deploy.sh | 2 +- .../vps/leo-db-context/__init__.py | 38 ++ scripts/leo_behavior_manifest.py | 22 +- scripts/leo_turn_execution_manifest.py | 476 ++++++++++++++++++ scripts/run_leo_direct_claim_handler_suite.py | 307 ++++++++++- .../test_hermes_leoclean_db_context_plugin.py | 27 + tests/test_leo_behavior_manifest.py | 22 +- tests/test_leo_turn_execution_manifest.py | 170 +++++++ 9 files changed, 1042 insertions(+), 24 deletions(-) create mode 100644 scripts/leo_turn_execution_manifest.py create mode 100644 tests/test_leo_turn_execution_manifest.py diff --git a/deploy/auto-deploy.sh b/deploy/auto-deploy.sh index 6468513..1b7d517 100755 --- a/deploy/auto-deploy.sh +++ b/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" diff --git a/deploy/deploy.sh b/deploy/deploy.sh index ccde590..5d0950a 100755 --- a/deploy/deploy.sh +++ b/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" diff --git a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py b/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py index d886545..728c8d6 100644 --- a/hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py +++ b/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 ( '\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", diff --git a/scripts/leo_behavior_manifest.py b/scripts/leo_behavior_manifest.py index 39bae4e..3f766dd 100644 --- a/scripts/leo_behavior_manifest.py +++ b/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) diff --git a/scripts/leo_turn_execution_manifest.py b/scripts/leo_turn_execution_manifest.py new file mode 100644 index 0000000..b497721 --- /dev/null +++ b/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()) diff --git a/scripts/run_leo_direct_claim_handler_suite.py b/scripts/run_leo_direct_claim_handler_suite.py index 44f0943..cd8053f 100755 --- a/scripts/run_leo_direct_claim_handler_suite.py +++ b/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 [] - 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 + return read_jsonl(path) 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 diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py index 66a12ce..8a41443 100644 --- a/tests/test_hermes_leoclean_db_context_plugin.py +++ b/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 diff --git a/tests/test_leo_behavior_manifest.py b/tests/test_leo_behavior_manifest.py index 08287ad..2948dd0 100644 --- a/tests/test_leo_behavior_manifest.py +++ b/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) - second = manifest.build_manifest(profile, hermes) + first = manifest.build_manifest(profile, hermes, deployment) + 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"] diff --git a/tests/test_leo_turn_execution_manifest.py b/tests/test_leo_turn_execution_manifest.py new file mode 100644 index 0000000..c74a169 --- /dev/null +++ b/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"]