#!/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 re 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" DATABASE_PROMPT_TERMS = ( "knowledge base", "database", "postgres", "claim", "belief", "evidence", "source", "proposal", "canonical", "approve", "apply", "document", "identity", "soul", "strategy", "row", ) HEX_64 = re.compile(r"^[0-9a-f]{64}$") HEX_40 = re.compile(r"^[0-9a-f]{40}$") 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=all"], check=False, capture_output=True, text=True, timeout=10, ) except (OSError, subprocess.TimeoutExpired): return {"git_head": None, "worktree_clean": None, "status_sha256": None} git_head = head.stdout.strip() if head.returncode == 0 and len(head.stdout.strip()) == 40 else None status_text = status.stdout if status.returncode == 0 else "" return { "git_head": git_head, "worktree_clean": status.returncode == 0 and not status_text.strip(), "status_sha256": text_sha256(status_text) if status.returncode == 0 else None, } def _is_sha256(value: Any) -> bool: return isinstance(value, str) and bool(HEX_64.fullmatch(value)) def _is_git_sha(value: Any) -> bool: return isinstance(value, str) and bool(HEX_40.fullmatch(value)) def _non_negative_int(value: Any) -> bool: return isinstance(value, int) and not isinstance(value, bool) and value >= 0 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 _trace_events(result: dict[str, Any], event: str) -> list[dict[str, Any]]: return [ item for item in result.get("model_call_trace") or [] if isinstance(item, dict) and item.get("event") == event ] def _response_trace(result: dict[str, Any]) -> list[dict[str, Any]]: responses = [] for raw in result.get("model_response_trace") or []: if not isinstance(raw, dict): continue responses.append( { "content_sha256": raw.get("content_sha256"), "content_bytes": raw.get("content_bytes"), "tool_calls_sha256": raw.get("tool_calls_sha256"), "tool_call_count": raw.get("tool_call_count"), } ) return responses def _model_calls(result: dict[str, Any]) -> list[dict[str, Any]]: calls: list[dict[str, Any]] = [] responses = _response_trace(result) for index, raw in enumerate(_trace_events(result, "post_api_request")): response = responses[index] if index < len(responses) else {} 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"), "assistant_message": response, } ) return calls def _safe_retrieval_receipt(value: Any, *, expected_query_sha256: str) -> 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 {} required_hashes = ( value.get("query_sha256"), value.get("semantic_context_sha256"), value.get("artifact_state_sha256"), value.get("receipt_sha256"), ) stable_consistency = bool( consistency.get("status") == "stable_wal_marker" and consistency.get("database") and consistency.get("database_user") and consistency.get("system_identifier") and consistency.get("wal_lsn_before") and consistency.get("wal_lsn_before") == consistency.get("wal_lsn_after") ) if ( not all(_is_sha256(item) for item in required_hashes) or value.get("query_sha256") != expected_query_sha256 or not stable_consistency ): return None 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": value.get("receipt_sha256"), "trace_sha256": canonical_sha256(value), } def _context_receipts(result: dict[str, Any], *, expected_query_sha256: str) -> 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" or record.get("status") != "ok" or record.get("injected") is not True or record.get("query_sha256") != expected_query_sha256 ): continue receipt = _safe_retrieval_receipt( record.get("retrieval_receipt"), expected_query_sha256=expected_query_sha256 ) 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 raw.get("schema") != RETRIEVAL_RECEIPT_SCHEMA or not _is_sha256(raw.get("semantic_context_sha256")) or not _is_sha256(raw.get("artifact_state_sha256")) or raw.get("read_consistency_status") != "stable_wal_marker" ): 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], *, prompt: str) -> bool: prompt_casefold = prompt.casefold() if any(term in prompt_casefold for term in DATABASE_PROMPT_TERMS): return True 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_context_binding( result: dict[str, Any], *, prompt_sha256: str, reply_sha256: str ) -> dict[str, Any]: records = [item for item in result.get("database_context_trace") or [] if isinstance(item, dict)] pre_records = [item for item in records if item.get("event") == "pre_llm_call"] post_records = [item for item in records if item.get("event") == "post_llm_call"] pre = pre_records[0] if len(pre_records) == 1 else {} post = post_records[0] if len(post_records) == 1 else {} raw_response_sha256 = post.get("response_sha256") delivered_response_sha256 = post.get("delivered_response_sha256") return { "pre_record_count": len(pre_records), "post_record_count": len(post_records), "query_sha256": pre.get("query_sha256"), "contract_ids": sorted(str(item) for item in pre.get("contract_ids") or []), "contract_sha256": pre.get("contract_sha256"), "pre_status": pre.get("status"), "pre_injected": pre.get("injected"), "post_status": post.get("status"), "post_validated": post.get("validated"), "post_transformed": post.get("transformed"), "raw_response_sha256": raw_response_sha256, "delivered_response_sha256": delivered_response_sha256, "query_bound": bool( len(pre_records) == 1 and len(post_records) == 1 and pre.get("query_sha256") == prompt_sha256 and post.get("query_sha256") == prompt_sha256 ), "response_bound": bool( _is_sha256(raw_response_sha256) and delivered_response_sha256 == reply_sha256 and post.get("validated") is True and post.get("status") == "ok" ), "context_available": bool( len(pre_records) == 1 and pre.get("status") == "ok" and pre.get("injected") is True and _is_sha256(pre.get("contract_sha256")) ), "trace_sha256": canonical_sha256(records), } 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 {} invocations = [ { "access_mode": invocation.get("access_mode"), "command_sha256": invocation.get("command_sha256"), "executable": invocation.get("executable"), "subcommand": invocation.get("subcommand"), } for invocation in call.get("database_invocations") or [] if isinstance(invocation, dict) ] calls.append( { "tool_call_id_sha256": call.get("tool_call_id_sha256"), "arguments_sha256": call.get("arguments_sha256"), "database_invocations": invocations, "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_nonempty": call_result.get("nonempty"), "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 [])), } ) tool_call_count = trace.get("database_tool_call_count", 0) completed_count = trace.get("database_tool_completed_count", 0) all_results_bound = bool( _non_negative_int(tool_call_count) and _non_negative_int(completed_count) and completed_count == tool_call_count and len(calls) == tool_call_count and all( _is_sha256(call.get("tool_call_id_sha256")) and _is_sha256(call.get("arguments_sha256")) and _is_sha256(call.get("result_content_sha256")) and _non_negative_int(call.get("result_content_bytes")) and call.get("result_error_detected") is False and call.get("result_nonempty") is True for call in calls ) ) all_calls_read_only = bool( trace.get("database_tool_calls_read_only") is True and all( call.get("database_invocations") and all( invocation.get("access_mode") == "read_only" and _is_sha256(invocation.get("command_sha256")) for invocation in call.get("database_invocations") or [] ) for call in calls ) ) return { "schema": trace.get("schema"), "tool_call_count": tool_call_count, "completed_count": completed_count, "all_calls_read_only": all_calls_read_only if tool_call_count else True, "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 {} before = consistency.get("before") if isinstance(consistency.get("before"), dict) else {} after = consistency.get("after") if isinstance(consistency.get("after"), dict) else {} return bool( value.get("status") == "ok" and _is_sha256(value.get("fingerprint_sha256")) and _is_sha256(value.get("table_rows_sha256")) and _is_sha256(value.get("structure_sha256")) and isinstance(value.get("table_count"), int) and value.get("table_count") > 0 and _non_negative_int(value.get("total_rows")) and consistency.get("status") == "stable_wal_marker" and before and before == after and before.get("database") and before.get("database_user") and before.get("system_identifier") and before.get("wal_lsn") ) def _source_tree_complete(runtime: Any) -> bool: if not isinstance(runtime, dict): return False tree = runtime.get("source_tree") if isinstance(runtime.get("source_tree"), dict) else {} return bool( _is_git_sha(runtime.get("git_head")) and _is_sha256(tree.get("sha256")) and isinstance(tree.get("file_count"), int) and tree.get("file_count") > 0 ) def _conversation_binding( result: dict[str, Any], *, reply_sha256: str, previous_execution_sha256: str | None, expected_previous_conversation: dict[str, Any] | None, ) -> dict[str, Any]: before = result.get("conversation_before") if isinstance(result.get("conversation_before"), dict) else {} after = result.get("conversation_after") if isinstance(result.get("conversation_after"), dict) else {} expected = expected_previous_conversation or {} before_valid = bool( _non_negative_int(before.get("message_count")) and _is_sha256(before.get("messages_sha256")) ) after_valid = bool( _non_negative_int(after.get("message_count")) and after.get("message_count", 0) > before.get("message_count", -1) and _is_sha256(after.get("messages_sha256")) and after.get("last_message_role") == "assistant" and after.get("last_message_content_sha256") == reply_sha256 ) if previous_execution_sha256 is None: prior_state_bound = before.get("message_count") == 0 else: prior_state_bound = bool( _is_sha256(previous_execution_sha256) and before.get("message_count") == expected.get("message_count") and before.get("messages_sha256") == expected.get("messages_sha256") and before.get("last_message_role") == expected.get("last_message_role") and before.get("last_message_content_sha256") == expected.get("last_message_content_sha256") ) return { "before": before, "after": after, "history_prefix_preserved": result.get("conversation_history_prefix_preserved"), "previous_execution_sha256": previous_execution_sha256, "starts_from_empty_conversation": before.get("message_count") == 0, "prior_turn_state_bound": prior_state_bound, "conversation_hashes_valid": before_valid and after_valid, } def _model_execution_binding( result: dict[str, Any], *, prompt_sha256: str, reply_sha256: str, raw_response_sha256: str | None, ) -> dict[str, Any]: calls = _model_calls(result) responses = _response_trace(result) pre_events = _trace_events(result, "turn_pre_llm_call") post_events = _trace_events(result, "turn_post_llm_call") pre = pre_events[0] if len(pre_events) == 1 else {} post = post_events[0] if len(post_events) == 1 else {} session_hashes = { str(value) for value in [ pre.get("session_id_sha256"), post.get("session_id_sha256"), *(call.get("session_id_sha256") for call in calls), ] if value } call_sequence = [call.get("api_call_count") for call in calls] response_hashes_valid = bool( responses and len(responses) == len(calls) and all( _is_sha256(response.get("content_sha256")) and _is_sha256(response.get("tool_calls_sha256")) and _non_negative_int(response.get("content_bytes")) and _non_negative_int(response.get("tool_call_count")) for response in responses ) ) return { "call_count": len(calls), "calls": calls, "pre_llm_trace_count": len(pre_events), "post_llm_trace_count": len(post_events), "prompt_sha256": pre.get("prompt_sha256"), "raw_assistant_response_sha256": post.get("raw_assistant_response_sha256"), "session_id_sha256": next(iter(session_hashes)) if len(session_hashes) == 1 else None, "prompt_bound": bool( len(pre_events) == 1 and len(post_events) == 1 and pre.get("prompt_sha256") == prompt_sha256 and post.get("prompt_sha256") == prompt_sha256 ), "raw_response_bound": bool( _is_sha256(raw_response_sha256) and post.get("raw_assistant_response_sha256") == raw_response_sha256 ), "delivered_response_bound": bool( response_hashes_valid and responses[-1].get("content_sha256") == reply_sha256 ), "response_trace_count_matches_api_calls": len(responses) == len(calls), "api_call_sequence_valid": call_sequence == list(range(1, len(calls) + 1)), "session_binding_valid": len(session_hashes) == 1 and all(_is_sha256(item) for item in session_hashes), "response_hashes_valid": response_hashes_valid, "externally_hosted_weights_hashable": False, "provider_response_id_required_for_attribution": False, } def _suite_safety_binding(value: dict[str, Any] | None) -> dict[str, Any]: source = value or {} return { "remote_returncode": source.get("remote_returncode"), "pass_runtime": source.get("pass_runtime"), "live_behavior_manifest_unchanged": source.get("live_behavior_manifest_unchanged"), "temp_profile_removed": source.get("temp_profile_removed"), "service_unchanged": source.get("service_unchanged"), "db_fingerprint_unchanged": source.get("db_fingerprint_unchanged"), "model_call_trace_all_bound": source.get("model_call_trace_all_bound"), } def _required_missing( *, behavior: dict[str, Any], model_execution: dict[str, Any], context_receipts: list[dict[str, Any]], database_context_binding: 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], db_counts_changed: bool | None, posted_to_telegram: bool | None, mutates_kb_by_harness: bool | None, result_mutates_kb: bool | None, conversation_binding: dict[str, Any], suite_safety: dict[str, Any], ) -> list[str]: missing = [] checks = { "prompt": bool(prompt), "reply": bool(reply), "session_boundary": bool(session_key), "behavior_sha256": _is_sha256(behavior.get("behavior_sha256")), "hermes_runtime_content_addressed": _source_tree_complete(behavior.get("hermes_runtime")), "teleo_infrastructure_runtime_content_addressed": _source_tree_complete( behavior.get("teleo_infrastructure_runtime") ), "harness_git_head": _is_git_sha(harness_source.get("git_head")), "harness_worktree_clean": bool( harness_source.get("worktree_clean") is True and _is_sha256(harness_source.get("status_sha256")) ), "actual_model_call": bool( model_execution.get("calls") and all( call.get("model") and call.get("provider") and call.get("response_model") and _is_sha256(call.get("task_id_sha256")) and _is_sha256(call.get("session_id_sha256")) and _is_sha256(call.get("base_url_sha256")) for call in model_execution.get("calls") or [] ) ), "model_prompt_binding": model_execution.get("prompt_bound") is True, "model_raw_response_binding": model_execution.get("raw_response_bound") is True, "model_delivered_response_binding": model_execution.get("delivered_response_bound") is True, "model_response_trace_count": model_execution.get("response_trace_count_matches_api_calls") is True, "model_api_call_sequence": model_execution.get("api_call_sequence_valid") is True, "model_session_binding": model_execution.get("session_binding_valid") is True, "database_context_query_binding": database_context_binding.get("query_bound") is True, "database_context_available": database_context_binding.get("context_available") is True, "database_context_response_binding": database_context_binding.get("response_bound") is True, "database_retrieval_receipt": not database_required or len(context_receipts) == 1, "database_tool_results": not database_tool_binding.get("tool_call_count") or bool(database_tool_binding.get("all_results_content_bound")), "database_tools_read_only": database_tool_binding.get("all_calls_read_only") is True, "database_fingerprint_before": _database_fingerprint_complete(db_fingerprint_before), "database_fingerprint_after": _database_fingerprint_complete(db_fingerprint_after), "database_fingerprint_unchanged": db_fingerprint_before.get("fingerprint_sha256") == db_fingerprint_after.get("fingerprint_sha256"), "database_counts_unchanged": db_counts_changed is False, "telegram_not_posted": posted_to_telegram is False, "harness_did_not_mutate_kb": mutates_kb_by_harness is False, "turn_did_not_mutate_kb": result_mutates_kb is False, "conversation_history_prefix": conversation_binding.get("history_prefix_preserved") is True, "conversation_hashes": conversation_binding.get("conversation_hashes_valid") is True, "conversation_prior_turn_binding": conversation_binding.get("prior_turn_state_bound") is True, "suite_remote_success": suite_safety.get("remote_returncode") == 0, "suite_runtime_success": suite_safety.get("pass_runtime") is True, "suite_live_behavior_unchanged": suite_safety.get("live_behavior_manifest_unchanged") is True, "suite_temp_profile_removed": suite_safety.get("temp_profile_removed") is True, "suite_service_unchanged": suite_safety.get("service_unchanged") is True, "suite_db_fingerprint_unchanged": suite_safety.get("db_fingerprint_unchanged") is True, "suite_model_trace_bound": suite_safety.get("model_call_trace_all_bound") is True, } 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 | None, mutates_kb_by_harness: bool | None, harness_source: dict[str, Any], suite_safety: dict[str, Any] | None = None, previous_execution_sha256: str | None = None, expected_previous_conversation: dict[str, Any] | None = None, ) -> dict[str, Any]: prompt = str(result.get("prompt") or "") reply = str(result.get("reply") or "") prompt_sha256 = text_sha256(prompt) database_query_sha256 = text_sha256(prompt[:16_000]) reply_sha256 = text_sha256(reply) database_context_binding = _database_context_binding( result, prompt_sha256=database_query_sha256, reply_sha256=reply_sha256, ) model_execution = _model_execution_binding( result, prompt_sha256=prompt_sha256, reply_sha256=reply_sha256, raw_response_sha256=database_context_binding.get("raw_response_sha256"), ) context_receipts = _context_receipts(result, expected_query_sha256=database_query_sha256) tool_receipts = _tool_receipts(result) database_required = _database_required(result, prompt=prompt) database_tool_binding = _database_tool_binding(result) fingerprint_before = db_fingerprint_before or {} fingerprint_after = db_fingerprint_after or {} fingerprint_unchanged = bool( fingerprint_before.get("fingerprint_sha256") and fingerprint_after.get("fingerprint_sha256") and fingerprint_before.get("fingerprint_sha256") == fingerprint_after.get("fingerprint_sha256") ) conversation_binding = _conversation_binding( result, reply_sha256=reply_sha256, previous_execution_sha256=previous_execution_sha256, expected_previous_conversation=expected_previous_conversation, ) safety_binding = _suite_safety_binding(suite_safety) missing = _required_missing( behavior=behavior_manifest, model_execution=model_execution, context_receipts=context_receipts, database_context_binding=database_context_binding, 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, db_counts_changed=db_counts_changed, posted_to_telegram=posted_to_telegram, mutates_kb_by_harness=mutates_kb_by_harness, result_mutates_kb=result.get("mutates_kb"), conversation_binding=conversation_binding, suite_safety=safety_binding, ) 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": prompt_sha256, "database_query_sha256": database_query_sha256, "reply_sha256": reply_sha256, "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_for_suite": True, "prior_dynamic_state_excluded_from_suite": True, "conversation": conversation_binding, }, "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": model_execution, "canonical_database": { "required_for_turn": database_required, "binding_status": "retrieval_receipt_bound" if context_receipts else "not_required" if not database_required else "missing", "context_retrieval_receipts": context_receipts, "explicit_tool_retrieval_receipts": tool_receipts, "context_binding": database_context_binding, "database_tool_binding": database_tool_binding, "fingerprint_before": fingerprint_before, "fingerprint_after": fingerprint_after, "fingerprint_unchanged": fingerprint_unchanged, "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, "turn_mutates_kb": result.get("mutates_kb"), "suite_safety": safety_binding, "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": "content_addressed_attribution" if not missing else "incomplete_attribution", "runtime_and_turn_inputs_content_addressed": not missing, "runtime_source_artifacts_embedded": False, "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 receipt binds the tested prompt, conversation chain, response hashes, source hashes, " "model-call metadata, and database reads. It identifies those inputs; it does not embed the " "runtime source, database snapshot, or hosted model weights needed to reconstruct them." ), }, } 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 = [item for item in report.get("results") or [] if isinstance(item, dict)] service = report.get("service_before_after") if isinstance(report.get("service_before_after"), dict) else {} suite_safety = { "remote_returncode": report.get("remote_returncode"), "pass_runtime": report.get("pass_runtime"), "live_behavior_manifest_unchanged": report.get("live_behavior_manifest_unchanged"), "temp_profile_removed": report.get("temp_profile_removed"), "service_unchanged": service.get("unchanged_from_preexisting_live_readback"), "db_fingerprint_unchanged": report.get("db_fingerprint_unchanged"), "model_call_trace_all_bound": report.get("model_call_trace_all_bound"), } complete = 0 previous_execution_sha256: str | None = None expected_previous_conversation: dict[str, Any] | None = None for result in results: 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=report.get("posted_to_telegram"), mutates_kb_by_harness=report.get("mutates_kb_by_harness"), harness_source=harness_source, suite_safety=suite_safety, previous_execution_sha256=previous_execution_sha256, expected_previous_conversation=expected_previous_conversation, ) result["execution_manifest"] = manifest if manifest["attribution"]["status"] == "complete" and not validate_turn_manifest(manifest): complete += 1 previous_execution_sha256 = manifest.get("execution_sha256") conversation = result.get("conversation_after") expected_previous_conversation = conversation if isinstance(conversation, dict) else None 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())