diff --git a/scripts/leo_behavior_manifest.py b/scripts/leo_behavior_manifest.py index 3f766dd..d8f6cad 100644 --- a/scripts/leo_behavior_manifest.py +++ b/scripts/leo_behavior_manifest.py @@ -175,6 +175,76 @@ def git_head(repo: Path) -> str | None: return value if completed.returncode == 0 and len(value) == 40 else None +def git_state(repo: Path) -> dict[str, Any]: + head = git_head(repo) + try: + completed = subprocess.run( + ["git", "-C", str(repo), "status", "--porcelain", "--untracked-files=all"], + check=False, + capture_output=True, + text=True, + timeout=10, + env={**os.environ, "GIT_OPTIONAL_LOCKS": "0"}, + ) + except (OSError, subprocess.TimeoutExpired): + return {"git_head": head, "worktree_clean": None, "status_sha256": None} + status = completed.stdout if completed.returncode == 0 else "" + return { + "git_head": head, + "worktree_clean": completed.returncode == 0 and not status.strip(), + "status_sha256": sha256_bytes(status.encode("utf-8")) if completed.returncode == 0 else None, + } + + +def source_code_manifest(profile: Path, paths: tuple[Path, ...]) -> dict[str, Any]: + """Hash executable source while excluding generated caches and environments.""" + + allowed_suffixes = { + ".cfg", + ".ini", + ".j2", + ".jinja", + ".json", + ".py", + ".sh", + ".sql", + ".tmpl", + ".toml", + ".yaml", + ".yml", + } + excluded_parts = {".git", ".pytest_cache", "__pycache__", "node_modules", "venv", ".venv"} + files: list[dict[str, Any]] = [] + missing: list[str] = [] + for requested in paths: + if not requested.exists() and not requested.is_symlink(): + missing.append(_safe_relative(requested, profile)) + continue + candidates = [requested] if requested.is_file() else sorted(requested.rglob("*")) + for path in candidates: + if not path.is_file() or excluded_parts & set(path.parts): + continue + stat = path.stat() + if path.suffix not in allowed_suffixes and not bool(stat.st_mode & 0o111): + continue + files.append( + { + "path": _safe_relative(path, profile), + "bytes": stat.st_size, + "mode": oct(stat.st_mode & 0o777), + "sha256": file_sha256(path), + } + ) + files.sort(key=lambda item: item["path"]) + stable = {"files": files, "missing": sorted(missing)} + return { + **stable, + "file_count": len(files), + "total_bytes": sum(int(item["bytes"]) for item in files), + "sha256": canonical_sha256(stable), + } + + def component( profile: Path, *, @@ -273,19 +343,22 @@ def build_manifest( }, "hermes_runtime": { "git_head": git_head(hermes_root), - "source_tree": path_manifest( - profile, (hermes_root / "run_agent.py", hermes_root / "agent" / "prompt_builder.py") + "git_state": git_state(hermes_root), + "source_tree": source_code_manifest( + profile, + ( + hermes_root / "run_agent.py", + hermes_root / "agent", + hermes_root / "gateway", + hermes_root / "hermes_cli", + hermes_root / "tools", + ), ), }, "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", - ), - ), + "git_state": git_state(deployment_root), + "source_tree": source_code_manifest(profile, (deployment_root,)), }, "components": components, "canonical_database": { diff --git a/scripts/leo_turn_execution_manifest.py b/scripts/leo_turn_execution_manifest.py index 3596f9a..964548c 100644 --- a/scripts/leo_turn_execution_manifest.py +++ b/scripts/leo_turn_execution_manifest.py @@ -12,6 +12,7 @@ from __future__ import annotations import argparse import hashlib import json +import re import subprocess from datetime import UTC, datetime from pathlib import Path @@ -19,6 +20,26 @@ 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: @@ -40,21 +61,35 @@ def git_state(repo: Path) -> dict[str, Any]: timeout=10, ) status = subprocess.run( - ["git", "-C", str(repo), "status", "--porcelain", "--untracked-files=no"], + ["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, "tracked_tree_clean": None} + 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, - "tracked_tree_clean": status.returncode == 0 and not status.stdout.strip(), + "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 {} @@ -75,11 +110,35 @@ def _safe_usage(value: Any) -> dict[str, int | float | None]: 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]] = [] - for raw in result.get("model_call_trace") or []: - if not isinstance(raw, dict) or raw.get("event") != "post_api_request": - continue + 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"), @@ -97,16 +156,37 @@ def _model_calls(result: dict[str, Any]) -> list[dict[str, Any]]: "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) -> dict[str, Any] | None: +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"), @@ -128,16 +208,25 @@ def _safe_retrieval_receipt(value: Any) -> dict[str, Any] | None: "wal_lsn_before": consistency.get("wal_lsn_before"), "wal_lsn_after": consistency.get("wal_lsn_after"), }, - "receipt_sha256": canonical_sha256(value), + "receipt_sha256": value.get("receipt_sha256"), + "trace_sha256": canonical_sha256(value), } -def _context_receipts(result: dict[str, Any]) -> list[dict[str, Any]]: +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": + 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")) + receipt = _safe_retrieval_receipt( + record.get("retrieval_receipt"), expected_query_sha256=expected_query_sha256 + ) if receipt is not None: receipts.append(receipt) return receipts @@ -150,7 +239,13 @@ def _tool_receipts(result: dict[str, Any]) -> list[dict[str, Any]]: 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"): + 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( { @@ -164,7 +259,10 @@ def _tool_receipts(result: dict[str, Any]) -> list[dict[str, Any]]: return receipts -def _database_required(result: dict[str, Any]) -> bool: +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 @@ -175,6 +273,51 @@ def _database_required(result: dict[str, Any]) -> bool: 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 = [] @@ -182,26 +325,63 @@ def _database_tool_binding(result: dict[str, Any]) -> dict[str, Any]: 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": call.get("database_invocations") or [], + "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 [])), } ) - all_results_bound = bool(calls) and all( - call.get("result_content_sha256") and call.get("result_error_detected") is False for call in calls + 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": 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"), + "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), @@ -210,20 +390,163 @@ def _database_tool_binding(result: dict[str, Any]) -> dict[str, Any]: 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 value.get("fingerprint_sha256") - and value.get("table_count") - and value.get("total_rows") + 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], - calls: list[dict[str, Any]], - receipts: list[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, @@ -232,34 +555,71 @@ def _required_missing( 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": 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") + "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")) ), - "harness_git_head": bool(harness_source.get("git_head")), "actual_model_call": bool( - calls + model_execution.get("calls") and all( - call.get("model") and call.get("provider") and call.get("response_model") - for call in calls + 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 [] ) ), - "database_retrieval_receipt": not database_required or bool(receipts), + "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_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"), + "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: @@ -280,17 +640,32 @@ def build_turn_manifest( 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, + 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 "") - calls = _model_calls(result) - context_receipts = _context_receipts(result) + 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) - receipts = context_receipts + tool_receipts - database_required = _database_required(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 {} @@ -299,10 +674,18 @@ def build_turn_manifest( 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, - calls=calls, - receipts=receipts, + model_execution=model_execution, + context_receipts=context_receipts, + database_context_binding=database_context_binding, database_required=database_required, session_key=session_key, prompt=prompt, @@ -311,6 +694,12 @@ def build_turn_manifest( 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, @@ -320,8 +709,9 @@ def build_turn_manifest( "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_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")), }, @@ -329,8 +719,9 @@ def build_turn_manifest( "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, + "fresh_temp_profile_for_suite": True, + "prior_dynamic_state_excluded_from_suite": True, + "conversation": conversation_binding, }, "runtime": { "behavior_sha256": behavior_manifest.get("behavior_sha256"), @@ -343,21 +734,17 @@ def build_turn_manifest( }, "harness_source": harness_source, }, - "model_execution": { - "call_count": len(calls), - "calls": calls, - "externally_hosted_weights_hashable": False, - "provider_response_id_required_for_attribution": False, - }, + "model_execution": model_execution, "canonical_database": { "required_for_turn": database_required, "binding_status": "retrieval_receipt_bound" - if receipts + 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, @@ -374,6 +761,8 @@ def build_turn_manifest( "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, @@ -384,8 +773,9 @@ def build_turn_manifest( "missing_required_bindings": missing, }, "replay_claim": { - "status": "bounded_recorded_inputs_pinned" if not missing else "incomplete_inputs", - "local_runtime_inputs_reconstructible": not missing, + "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") ), @@ -393,8 +783,9 @@ def build_turn_manifest( "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." + "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." ), }, } @@ -427,11 +818,21 @@ def attach_execution_manifests(report: dict[str, Any], *, repo_root: Path) -> di 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 [] + 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: - if not isinstance(result, dict): - continue manifest = build_turn_manifest( result=result, behavior_manifest=behavior, @@ -444,13 +845,19 @@ def attach_execution_manifests(report: dict[str, Any], *, repo_root: Path) -> di 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")), + 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), diff --git a/scripts/run_leo_direct_claim_handler_suite.py b/scripts/run_leo_direct_claim_handler_suite.py index 4ed679d..98909e6 100755 --- a/scripts/run_leo_direct_claim_handler_suite.py +++ b/scripts/run_leo_direct_claim_handler_suite.py @@ -18,7 +18,10 @@ import uuid from pathlib import Path from typing import Any -import working_leo_open_ended_benchmark as benchmark +try: + import working_leo_open_ended_benchmark as benchmark +except ImportError: # pragma: no cover - imported as scripts.* in tests + from scripts import working_leo_open_ended_benchmark as benchmark try: from leo_turn_execution_manifest import attach_execution_manifests @@ -39,15 +42,23 @@ CHAT_ID = "-5146042086" USER_ID = "9070919" USER_NAME = "codex handler direct claim" -SECRET_PATTERNS = [ - re.compile(r"\b\d{6,}:[A-Za-z0-9_-]{20,}\b"), - re.compile(r"(Authorization: Bearer )([A-Za-z0-9._-]+)", re.IGNORECASE), - re.compile(r"((?:api[_-]?key|token|secret|password)[=:]\s*)([A-Za-z0-9._-]+)", re.IGNORECASE), -] +JSON_SECRET_PATTERN = re.compile( + r'("(?:api[_-]?key|access[_-]?token|refresh[_-]?token|bot[_-]?token|token|secret|password)"\s*:\s*)' + r'("(?:\\.|[^"\\])*"|null|[^\s,}\]]+)', + re.IGNORECASE, +) +TELEGRAM_TOKEN_PATTERN = re.compile(r"\b\d{6,}:[A-Za-z0-9_-]{20,}\b") +AUTHORIZATION_PATTERN = re.compile(r"(Authorization:\s*Bearer\s+)([^\s\"']+)", re.IGNORECASE) +ASSIGNMENT_SECRET_PATTERN = re.compile( + r"((?:api[_-]?key|access[_-]?token|refresh[_-]?token|bot[_-]?token|token|secret|password)" + r"\s*[=:]\s*)([^\s,;\"']+)", + re.IGNORECASE, +) DB_CONTEXT_PLUGIN = ( ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py" ) +BEHAVIOR_MANIFEST = ROOT / "scripts" / "leo_behavior_manifest.py" REMOTE_SCRIPT = r''' @@ -60,6 +71,7 @@ import subprocess import sys import tempfile import traceback +import types from datetime import datetime, timezone from pathlib import Path @@ -74,6 +86,7 @@ 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__ +BEHAVIOR_MANIFEST_SOURCE = __BEHAVIOR_MANIFEST_SOURCE_JSON__ TURN_TRACE_PLUGIN_SOURCE = """\ import hashlib import json @@ -106,12 +119,31 @@ def _sha(value): return hashlib.sha256(str(value or "").encode("utf-8")).hexdigest() -def _post_api_request(**kwargs): +def _write(record): raw_path = os.getenv("LEO_TURN_API_TRACE_PATH", "").strip() if not raw_path: - return None + return path = Path(raw_path) path.parent.mkdir(parents=True, exist_ok=True) + 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 + + +def _pre_llm_call(**kwargs): + _write({ + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "event": "turn_pre_llm_call", + "session_id_sha256": _sha(kwargs.get("session_id")), + "prompt_sha256": _sha(kwargs.get("user_message")), + }) + return None + + +def _post_api_request(**kwargs): record = { "generated_at_utc": datetime.now(timezone.utc).isoformat(), "event": "post_api_request", @@ -131,24 +163,34 @@ def _post_api_request(**kwargs): "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 + _write(record) + return None + + +def _post_llm_call(**kwargs): + _write({ + "generated_at_utc": datetime.now(timezone.utc).isoformat(), + "event": "turn_post_llm_call", + "session_id_sha256": _sha(kwargs.get("session_id")), + "prompt_sha256": _sha(kwargs.get("user_message")), + "raw_assistant_response_sha256": _sha(kwargs.get("assistant_response")), + }) return None def register(ctx): + ctx.register_hook("pre_llm_call", _pre_llm_call) ctx.register_hook("post_api_request", _post_api_request) + ctx.register_hook("post_llm_call", _post_llm_call) """ 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: + - pre_llm_call - post_api_request + - post_llm_call """ @@ -420,6 +462,7 @@ def remove_tree(path): def write_report_snapshot(report): try: REPORT_PATH.write_text(json.dumps(report, sort_keys=True), encoding="utf-8") + REPORT_PATH.chmod(0o600) except Exception: pass @@ -443,41 +486,71 @@ def current_agent_messages(runner, session_key): return [dict(message) for message in messages if isinstance(message, dict)] +def conversation_trace(messages): + rows = [] + for message in messages: + content = message.get("content") + content_text = content if isinstance(content, str) else json.dumps(content, sort_keys=True, default=str) + tool_calls = message.get("tool_calls") or [] + rows.append({ + "role": message.get("role"), + "content_sha256": hashlib.sha256(content_text.encode("utf-8")).hexdigest(), + "content_bytes": len(content_text.encode("utf-8")), + "tool_calls_sha256": hashlib.sha256( + json.dumps(tool_calls, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + ).hexdigest(), + "tool_call_count": len(tool_calls), + }) + return { + "message_count": len(rows), + "messages_sha256": hashlib.sha256( + json.dumps(rows, sort_keys=True, separators=(",", ":")).encode("utf-8") + ).hexdigest(), + "last_message_role": rows[-1]["role"] if rows else None, + "last_message_content_sha256": rows[-1]["content_sha256"] if rows else None, + } + + +def model_response_trace(messages): + return [ + { + "content_sha256": row["content_sha256"], + "content_bytes": row["content_bytes"], + "tool_calls_sha256": row["tool_calls_sha256"], + "tool_call_count": row["tool_call_count"], + } + for message, row in zip(messages, conversation_trace_rows(messages)) + if message.get("role") == "assistant" + ] + + +def conversation_trace_rows(messages): + rows = [] + for message in messages: + content = message.get("content") + content_text = content if isinstance(content, str) else json.dumps(content, sort_keys=True, default=str) + tool_calls = message.get("tool_calls") or [] + rows.append({ + "content_sha256": hashlib.sha256(content_text.encode("utf-8")).hexdigest(), + "content_bytes": len(content_text.encode("utf-8")), + "tool_calls_sha256": hashlib.sha256( + json.dumps(tool_calls, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + ).hexdigest(), + "tool_call_count": len(tool_calls), + }) + return rows + + async def run_suite(): sys.path.insert(0, str(DEPLOY_ROOT / "scripts")) - import leo_behavior_manifest as behavior_manifest_module from leo_tool_trace import extract_kb_tool_trace + behavior_manifest_module = types.ModuleType("leo_behavior_manifest_harness") + behavior_manifest_module.__file__ = "" + exec(compile(BEHAVIOR_MANIFEST_SOURCE, behavior_manifest_module.__file__, "exec"), behavior_manifest_module.__dict__) + 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 + return behavior_manifest_module.build_manifest(profile, AGENT_ROOT, DEPLOY_ROOT) before_service = service_state() before_counts = db_counts() @@ -570,6 +643,7 @@ async def run_suite(): 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)) + messages_before = current_agent_messages(runner, session_key) event = MessageEvent( text=prompt["message"], message_type=MessageType.TEXT, @@ -579,7 +653,10 @@ async def run_suite(): started = datetime.now(timezone.utc) reply = await asyncio.wait_for(runner._handle_message(event), timeout=300) ended = datetime.now(timezone.utc) - database_tool_trace = extract_kb_tool_trace(current_agent_messages(runner, session_key)) + messages_after = current_agent_messages(runner, session_key) + history_prefix_preserved = messages_after[: len(messages_before)] == messages_before + new_messages = messages_after[len(messages_before) :] if history_prefix_preserved else messages_after + database_tool_trace = extract_kb_tool_trace(new_messages) results.append( { "turn": index, @@ -596,6 +673,10 @@ async def run_suite(): "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:], + "model_response_trace": model_response_trace(new_messages), + "conversation_before": conversation_trace(messages_before), + "conversation_after": conversation_trace(messages_after), + "conversation_history_prefix_preserved": history_prefix_preserved, } ) report["results"] = results @@ -633,18 +714,19 @@ async def run_suite(): } database_context_trace = read_database_context_trace(context_trace_path) model_call_trace = read_jsonl(model_trace_path) + api_model_trace = [item for item in model_call_trace if item.get("event") == "post_api_request"] 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( + report["model_call_trace_count"] = len(api_model_trace) + report["model_call_trace_all_bound"] = bool(api_model_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 + for item in api_model_trace ) context_injections = [ item for item in database_context_trace if item.get("event", "pre_llm_call") == "pre_llm_call" @@ -724,16 +806,15 @@ def build_remote_script( .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"))) + .replace("__BEHAVIOR_MANIFEST_SOURCE_JSON__", json.dumps(BEHAVIOR_MANIFEST.read_text(encoding="utf-8"))) ) def redact(text: str) -> str: - result = text - for pattern in SECRET_PATTERNS: - if pattern.groups >= 2: - result = pattern.sub(r"\1[REDACTED]", result) - else: - result = pattern.sub("[REDACTED_TOKEN]", result) + result = JSON_SECRET_PATTERN.sub(lambda match: f'{match.group(1)}"[REDACTED]"', text) + result = AUTHORIZATION_PATTERN.sub(lambda match: f"{match.group(1)}[REDACTED]", result) + result = ASSIGNMENT_SECRET_PATTERN.sub(lambda match: f"{match.group(1)}[REDACTED]", result) + result = TELEGRAM_TOKEN_PATTERN.sub("[REDACTED_TOKEN]", result) return result @@ -817,7 +898,7 @@ def run_remote( if not candidate.startswith("{"): continue try: - result["parsed"] = json.loads(candidate) + result["parsed"] = json.loads(redact(candidate)) break except json.JSONDecodeError: continue @@ -833,8 +914,83 @@ def run_remote( return result +def _tool_trace_is_safe(trace: Any) -> bool: + if not isinstance(trace, dict): + return False + count = trace.get("database_tool_call_count") + completed = trace.get("database_tool_completed_count") + calls = trace.get("calls") if isinstance(trace.get("calls"), list) else [] + if not isinstance(count, int) or isinstance(count, bool) or count < 0: + return False + if not isinstance(completed, int) or isinstance(completed, bool) or completed != count: + return False + if len(calls) != count: + return False + if count and trace.get("database_tool_calls_read_only") is not True: + return False + for call in calls: + if not isinstance(call, dict): + return False + invocations = call.get("database_invocations") + result = call.get("result") + if not isinstance(invocations, list) or not invocations: + return False + if not all(isinstance(item, dict) and item.get("access_mode") == "read_only" for item in invocations): + return False + if ( + not isinstance(result, dict) + or result.get("error_detected") is not False + or result.get("nonempty") is not True + or not re.fullmatch(r"[0-9a-f]{64}", str(result.get("content_sha256") or "")) + ): + return False + return True + + +def build_safety_gate(report: dict[str, Any]) -> dict[str, Any]: + results = [item for item in report.get("results") or [] if isinstance(item, dict)] + handler = report.get("handler") if isinstance(report.get("handler"), dict) else {} + service = report.get("service_before_after") if isinstance(report.get("service_before_after"), dict) else {} + execution_summary = ( + report.get("execution_manifest_summary") + if isinstance(report.get("execution_manifest_summary"), dict) + else {} + ) + checks = { + "remote_command_succeeded": report.get("remote_returncode") == 0, + "runtime_completed": report.get("pass_runtime") is True, + "handler_authorized": handler.get("authorized") is True, + "results_nonempty": bool(results), + "all_turns_returned_replies": bool(results) and all(item.get("ok") is True for item in results), + "all_turns_declared_no_mutation": bool(results) + and all(item.get("mutates_kb") is False for item in results), + "all_tool_traces_read_only_and_bound": bool(results) + and all(_tool_trace_is_safe(item.get("database_tool_trace")) for item in results), + "no_telegram_post": report.get("posted_to_telegram") is False, + "harness_declared_no_kb_mutation": report.get("mutates_kb_by_harness") is False, + "database_counts_unchanged": report.get("db_counts_changed") is False, + "database_fingerprint_unchanged": report.get("db_fingerprint_unchanged") is True, + "live_behavior_unchanged": report.get("live_behavior_manifest_unchanged") is True, + "service_unchanged": service.get("unchanged_from_preexisting_live_readback") is True, + "temporary_profile_removed": report.get("temp_profile_removed") is True, + "model_trace_metadata_bound": report.get("model_call_trace_all_bound") is True, + "all_turn_manifests_complete": execution_summary.get("all_turns_attribution_complete") is True, + "no_runtime_error": not report.get("error") and not report.get("failure"), + } + failed = [name for name, passed in checks.items() if not passed] + return { + "status": "pass" if not failed else "fail", + "checks": checks, + "failed_checks": failed, + } + + +def report_passes_safety(report: dict[str, Any]) -> bool: + return build_safety_gate(report)["status"] == "pass" + + def write_output(remote: dict[str, Any], *, output_json: Path = OUTPUT_JSON) -> dict[str, Any]: - REPORT_DIR.mkdir(parents=True, exist_ok=True) + output_json.parent.mkdir(parents=True, exist_ok=True) parsed = remote.get("parsed") or {} report = parsed if isinstance(parsed, dict) else {} report["source_report_path"] = str(output_json) @@ -843,6 +999,7 @@ def write_output(remote: dict[str, Any], *, output_json: Path = OUTPUT_JSON) -> if remote.get("stderr"): report["remote_stderr"] = remote["stderr"] attach_execution_manifests(report, repo_root=ROOT) + report["safety_gate"] = build_safety_gate(report) output_json.write_text(json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="utf-8") return report @@ -850,8 +1007,17 @@ def write_output(remote: dict[str, Any], *, output_json: Path = OUTPUT_JSON) -> def main() -> int: remote = run_remote() report = write_output(remote) - print(json.dumps({"json": str(OUTPUT_JSON), "pass_runtime": report.get("pass_runtime")}, indent=2)) - return 0 if remote.get("returncode") == 0 and report.get("pass_runtime") else 1 + print( + json.dumps( + { + "json": str(OUTPUT_JSON), + "pass_runtime": report.get("pass_runtime"), + "safety_gate": report.get("safety_gate"), + }, + indent=2, + ) + ) + return 0 if report_passes_safety(report) else 1 if __name__ == "__main__": diff --git a/scripts/verify_leo_db_first_oos_canary.py b/scripts/verify_leo_db_first_oos_canary.py index cb85776..9676fa2 100644 --- a/scripts/verify_leo_db_first_oos_canary.py +++ b/scripts/verify_leo_db_first_oos_canary.py @@ -133,6 +133,16 @@ def verify_report( report.get("db_counts_changed") is False and report.get("db_counts_before") == report.get("db_counts_after") ), + "canonical_content_fingerprint_unchanged": ( + report.get("db_fingerprint_unchanged") is True + and (report.get("db_fingerprint_before") or {}).get("fingerprint_sha256") + == (report.get("db_fingerprint_after") or {}).get("fingerprint_sha256") + and bool((report.get("db_fingerprint_before") or {}).get("fingerprint_sha256")) + ), + "execution_receipt_complete": ( + (report.get("execution_manifest_summary") or {}).get("all_turns_attribution_complete") is True + and (report.get("safety_gate") or {}).get("status") == "pass" + ), "live_behavior_profile_unchanged": ( report.get("changed_live_profile") is False and report.get("live_behavior_manifest_unchanged") is True @@ -165,7 +175,10 @@ def verify_report( "did_not_turn_the_test_conversation_into_hidden_training": ( checks["live_behavior_profile_unchanged"] and checks["temporary_profile_removed"] ), - "did_not_change_canonical_knowledge": checks["canonical_counts_unchanged"], + "did_not_change_canonical_knowledge": ( + checks["canonical_counts_unchanged"] + and checks["canonical_content_fingerprint_unchanged"] + ), } passed = all(checks.values()) and all(outcomes.values()) return { @@ -194,6 +207,8 @@ def verify_report( "state_readback": { "database_counts_before": report.get("db_counts_before"), "database_counts_after": report.get("db_counts_after"), + "database_fingerprint_before": report.get("db_fingerprint_before"), + "database_fingerprint_after": report.get("db_fingerprint_after"), "behavior_sha256_before": behavior_before, "behavior_sha256_after": behavior_after, "service_before": service_before, diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py index 8a41443..209eaf6 100644 --- a/tests/test_hermes_leoclean_db_context_plugin.py +++ b/tests/test_hermes_leoclean_db_context_plugin.py @@ -498,7 +498,14 @@ def test_handler_harness_retains_database_context_proof_fields() -> None: assert '"database_tool_calls_read_only"' in source assert "LEO_TURN_API_TRACE_PATH" in source assert '"model_call_trace"' in source + assert '"turn_pre_llm_call"' in source + assert '"turn_post_llm_call"' in source + assert '"model_response_trace"' in source + assert '"conversation_before"' in source + assert '"conversation_after"' in source assert "attach_execution_manifests" in source + assert "report_passes_safety" in source + assert "json.loads(redact(candidate))" 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 2948dd0..5b87992 100644 --- a/tests/test_leo_behavior_manifest.py +++ b/tests/test_leo_behavior_manifest.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import subprocess from pathlib import Path from scripts import leo_behavior_manifest as manifest @@ -114,3 +115,44 @@ def test_live_handler_harness_excludes_prior_dynamic_state_and_compares_live_fin assert "build_behavior_manifest" in source assert 'report["changed_live_profile"] =' in source assert '"changed_live_profile": False' not in source + + +def test_runtime_source_manifest_covers_nested_executable_sources(tmp_path: Path) -> None: + profile = tmp_path / "profile" + hermes = tmp_path / "hermes" + deployment = tmp_path / "deployment" + _write(profile / "config.yaml", "model:\n provider: test\n") + _write(profile / "SOUL.md", "identity\n") + _write(hermes / "run_agent.py", "runtime\n") + _write(hermes / "gateway" / "run.py", "gateway-a\n") + _write(hermes / "tools" / "terminal.py", "terminal-a\n") + _write(deployment / "scripts" / "leo_tool_trace.py", "trace-a\n") + _write(deployment / "ansible" / "roles" / "leo" / "tasks" / "main.yml", "---\n") + + before = manifest.build_manifest(profile, hermes, deployment) + _write(deployment / "ops" / "new_runtime_helper.py", "helper-b\n") + after = manifest.build_manifest(profile, hermes, deployment) + + assert before["hermes_runtime"]["source_tree"]["file_count"] == 3 + assert before["teleo_infrastructure_runtime"]["source_tree"]["file_count"] == 2 + assert after["teleo_infrastructure_runtime"]["source_tree"]["file_count"] == 3 + assert ( + before["teleo_infrastructure_runtime"]["source_tree"]["sha256"] + != after["teleo_infrastructure_runtime"]["source_tree"]["sha256"] + ) + + +def test_git_state_marks_untracked_files_dirty(tmp_path: Path) -> None: + subprocess.run(["git", "init", "-q", str(tmp_path)], check=True) + subprocess.run(["git", "-C", str(tmp_path), "config", "user.email", "test@example.com"], check=True) + subprocess.run(["git", "-C", str(tmp_path), "config", "user.name", "Test"], check=True) + _write(tmp_path / "tracked.py", "tracked\n") + subprocess.run(["git", "-C", str(tmp_path), "add", "tracked.py"], check=True) + subprocess.run(["git", "-C", str(tmp_path), "commit", "-qm", "fixture"], check=True) + + assert manifest.git_state(tmp_path)["worktree_clean"] is True + _write(tmp_path / "untracked.py", "untracked\n") + state = manifest.git_state(tmp_path) + + assert state["worktree_clean"] is False + assert len(str(state["status_sha256"])) == 64 diff --git a/tests/test_leo_turn_execution_manifest.py b/tests/test_leo_turn_execution_manifest.py index 823baf6..d4045e8 100644 --- a/tests/test_leo_turn_execution_manifest.py +++ b/tests/test_leo_turn_execution_manifest.py @@ -4,14 +4,22 @@ import copy from scripts import leo_turn_execution_manifest as manifest +PROMPT = "Challenge the weak acquisition claim." +REPLY = "The current evidence is shallow; here is a narrower candidate claim." +SESSION_SHA = "6" * 64 +EMPTY_TOOL_CALLS_SHA = manifest.canonical_sha256([]) + def behavior_manifest() -> dict: return { "behavior_sha256": "a" * 64, - "hermes_runtime": {"git_head": "b" * 40, "source_tree": {"sha256": "c" * 64}}, + "hermes_runtime": { + "git_head": "b" * 40, + "source_tree": {"sha256": "c" * 64, "file_count": 123}, + }, "teleo_infrastructure_runtime": { "git_head": "d" * 40, - "source_tree": {"sha256": "e" * 64}, + "source_tree": {"sha256": "e" * 64, "file_count": 42}, }, "components": { "runtime_identity": {"content": {"sha256": "f" * 64}}, @@ -20,12 +28,13 @@ def behavior_manifest() -> dict: } -def retrieval_receipt() -> dict: +def retrieval_receipt(prompt: str = PROMPT) -> dict: return { "schema": "livingip.teleoKbRetrievalReceipt.v1", - "query_sha256": "2" * 64, + "query_sha256": manifest.text_sha256(prompt[:16_000]), "semantic_context_sha256": "3" * 64, "artifact_state_sha256": "4" * 64, + "receipt_sha256": "5" * 64, "claim_ids": ["claim-b", "claim-a"], "source_ids": ["source-a"], "counts": {"claims": 2, "context_rows": 1}, @@ -41,23 +50,81 @@ def retrieval_receipt() -> dict: } -def turn_result() -> dict: +def conversation_after(reply: str = REPLY, *, message_count: int = 2, marker: str = "a") -> 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.", + "message_count": message_count, + "messages_sha256": marker * 64, + "last_message_role": "assistant", + "last_message_content_sha256": manifest.text_sha256(reply), + } + + +def empty_conversation() -> dict: + return { + "message_count": 0, + "messages_sha256": manifest.canonical_sha256([]), + "last_message_role": None, + "last_message_content_sha256": None, + } + + +def turn_result( + *, + prompt: str = PROMPT, + reply: str = REPLY, + before: dict | None = None, + after_marker: str = "a", + turn: int = 1, +) -> dict: + prompt_sha = manifest.text_sha256(prompt) + reply_sha = manifest.text_sha256(reply) + before_trace = copy.deepcopy(before or empty_conversation()) + after_trace = conversation_after( + reply, + message_count=int(before_trace["message_count"]) + 2, + marker=after_marker, + ) + return { + "turn": turn, + "prompt_id": f"OOS-{turn:02d}", + "prompt": prompt, + "reply": reply, "started_at_utc": "2026-07-14T12:00:00+00:00", "ended_at_utc": "2026-07-14T12:00:02+00:00", + "mutates_kb": False, "database_context_trace": [ { "event": "pre_llm_call", + "status": "ok", + "injected": True, + "query_sha256": prompt_sha, "contract_ids": ["reply_budget", "source_link_audit"], - "retrieval_receipt": retrieval_receipt(), - } + "contract_sha256": "7" * 64, + "retrieval_receipt": retrieval_receipt(prompt), + }, + { + "event": "post_llm_call", + "status": "ok", + "validated": True, + "transformed": False, + "query_sha256": prompt_sha, + "response_sha256": reply_sha, + "delivered_response_sha256": reply_sha, + }, ], - "database_tool_trace": {"database_tool_call_count": 0, "calls": []}, + "database_tool_trace": { + "schema": "livingip.leoKbToolTrace.v1", + "database_tool_call_count": 0, + "database_tool_completed_count": 0, + "database_tool_calls_read_only": True, + "calls": [], + }, "model_call_trace": [ + { + "event": "turn_pre_llm_call", + "session_id_sha256": SESSION_SHA, + "prompt_sha256": prompt_sha, + }, { "event": "post_api_request", "api_call_count": 1, @@ -67,16 +134,33 @@ def turn_result() -> dict: "response_model": "anthropic/claude-sonnet-4-6", "finish_reason": "stop", "message_count": 4, - "assistant_content_chars": 120, + "assistant_content_chars": len(reply), "assistant_tool_call_count": 0, - "task_id_sha256": "5" * 64, - "session_id_sha256": "6" * 64, - "base_url_sha256": "7" * 64, + "task_id_sha256": SESSION_SHA, + "session_id_sha256": SESSION_SHA, + "base_url_sha256": "8" * 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", + }, + { + "event": "turn_post_llm_call", + "session_id_sha256": SESSION_SHA, + "prompt_sha256": prompt_sha, + "raw_assistant_response_sha256": reply_sha, + }, + ], + "model_response_trace": [ + { + "content_sha256": reply_sha, + "content_bytes": len(reply.encode()), + "tool_calls_sha256": EMPTY_TOOL_CALLS_SHA, + "tool_call_count": 0, } ], + "conversation_before": before_trace, + "conversation_after": after_trace, + "conversation_history_prefix_preserved": True, } @@ -99,7 +183,26 @@ def database_fingerprint() -> dict: } -def build(result: dict | None = None, *, fingerprint: dict | None = None) -> dict: +def suite_safety() -> dict: + return { + "remote_returncode": 0, + "pass_runtime": True, + "live_behavior_manifest_unchanged": True, + "temp_profile_removed": True, + "service_unchanged": True, + "db_fingerprint_unchanged": True, + "model_call_trace_all_bound": True, + } + + +def build( + result: dict | None = None, + *, + fingerprint: dict | None = None, + harness_source: dict | None = None, + previous_execution_sha256: str | None = None, + expected_previous_conversation: dict | None = None, +) -> dict: selected_fingerprint = database_fingerprint() if fingerprint is None else fingerprint return manifest.build_turn_manifest( result=result or turn_result(), @@ -115,7 +218,11 @@ def build(result: dict | None = None, *, fingerprint: dict | None = None) -> dic db_fingerprint_after=selected_fingerprint, posted_to_telegram=False, mutates_kb_by_harness=False, - harness_source={"git_head": "8" * 40, "tracked_tree_clean": True}, + harness_source=harness_source + or {"git_head": "8" * 40, "worktree_clean": True, "status_sha256": manifest.text_sha256("")}, + suite_safety=suite_safety(), + previous_execution_sha256=previous_execution_sha256, + expected_previous_conversation=expected_previous_conversation, ) @@ -123,7 +230,10 @@ def test_turn_manifest_binds_model_runtime_session_and_exact_database_read() -> value = build() assert value["attribution"]["status"] == "complete" - assert value["replay_claim"]["status"] == "bounded_recorded_inputs_pinned" + assert value["replay_claim"]["status"] == "content_addressed_attribution" + assert value["model_execution"]["prompt_bound"] is True + assert value["model_execution"]["raw_response_bound"] is True + assert value["model_execution"]["delivered_response_bound"] is True 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" @@ -134,10 +244,11 @@ def test_turn_manifest_binds_model_runtime_session_and_exact_database_read() -> 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 value["replay_claim"]["runtime_source_artifacts_embedded"] is False assert manifest.validate_turn_manifest(value) == [] encoded = str(value) - assert "Challenge the weak acquisition claim" not in encoded + assert PROMPT not in encoded assert "narrower candidate claim" not in encoded assert "agent:main:telegram" not in encoded assert "secret-shaped" not in encoded @@ -151,7 +262,7 @@ def test_turn_manifest_detects_missing_actual_model_call() -> None: 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 + assert value["replay_claim"]["runtime_and_turn_inputs_content_addressed"] is False def test_turn_manifest_hash_detects_tampering() -> None: @@ -169,3 +280,134 @@ def test_database_question_requires_a_stable_full_database_fingerprint() -> None assert "database_fingerprint_before" in value["attribution"]["missing_required_bindings"] assert "database_fingerprint_after" in value["attribution"]["missing_required_bindings"] assert value["canonical_database"]["fingerprint_unchanged"] is False + + +def test_failed_database_context_lookup_cannot_be_attributed_as_complete() -> None: + result = turn_result() + result["database_context_trace"][0] = { + "event": "pre_llm_call", + "status": "error", + "injected": True, + "query_sha256": manifest.text_sha256(PROMPT), + "error": "timeout", + } + result["database_context_trace"][1]["status"] = "error" + + value = build(result) + + assert value["attribution"]["status"] == "incomplete" + assert "database_retrieval_receipt" in value["attribution"]["missing_required_bindings"] + assert "database_context_response_binding" in value["attribution"]["missing_required_bindings"] + + +def test_schema_only_or_wrong_query_receipt_is_rejected() -> None: + result = turn_result() + result["database_context_trace"][0]["retrieval_receipt"] = { + "schema": "livingip.teleoKbRetrievalReceipt.v1", + "query_sha256": "0" * 64, + } + + value = build(result) + + assert value["canonical_database"]["context_retrieval_receipts"] == [] + assert "database_retrieval_receipt" in value["attribution"]["missing_required_bindings"] + + +def test_delivered_reply_hash_mismatch_fails_closed() -> None: + result = turn_result() + result["database_context_trace"][1]["delivered_response_sha256"] = "0" * 64 + + value = build(result) + + assert "database_context_response_binding" in value["attribution"]["missing_required_bindings"] + + +def test_valid_response_transform_binds_raw_and_delivered_hashes() -> None: + result = turn_result() + raw_reply = "A much longer raw answer that the response contract replaces." + raw_sha = manifest.text_sha256(raw_reply) + result["database_context_trace"][1]["response_sha256"] = raw_sha + result["database_context_trace"][1]["transformed"] = True + result["model_call_trace"][-1]["raw_assistant_response_sha256"] = raw_sha + + value = build(result) + + assert value["attribution"]["status"] == "complete" + assert value["model_execution"]["raw_response_bound"] is True + assert value["model_execution"]["delivered_response_bound"] is True + assert value["canonical_database"]["context_binding"]["post_transformed"] is True + + +def test_dirty_or_untracked_harness_cannot_produce_complete_manifest() -> None: + value = build( + harness_source={"git_head": "8" * 40, "worktree_clean": False, "status_sha256": "1" * 64} + ) + + assert "harness_worktree_clean" in value["attribution"]["missing_required_bindings"] + + +def test_write_capable_database_tool_trace_fails_closed() -> None: + result = turn_result() + result["database_tool_trace"] = { + "schema": "livingip.leoKbToolTrace.v1", + "database_tool_call_count": 1, + "database_tool_completed_count": 1, + "database_tool_calls_read_only": False, + "calls": [ + { + "tool_call_id_sha256": "1" * 64, + "arguments_sha256": "2" * 64, + "database_invocations": [ + { + "access_mode": "write", + "command_sha256": "3" * 64, + "executable": "teleo-kb", + "subcommand": "apply", + } + ], + "result": { + "content_sha256": "4" * 64, + "content_bytes": 10, + "error_detected": False, + "nonempty": True, + "row_ids": [], + "sha256_values": [], + }, + } + ], + } + + value = build(result) + + assert "database_tools_read_only" in value["attribution"]["missing_required_bindings"] + + +def test_later_turn_binds_previous_execution_and_conversation_state() -> None: + first_result = turn_result() + first = build(first_result) + second_result = turn_result( + prompt="Now revise that candidate after my challenge.", + reply="The revision is narrower and still remains a proposal.", + before=first_result["conversation_after"], + after_marker="b", + turn=2, + ) + + second = build( + second_result, + previous_execution_sha256=first["execution_sha256"], + expected_previous_conversation=first_result["conversation_after"], + ) + + assert second["attribution"]["status"] == "complete" + conversation = second["session_boundary"]["conversation"] + assert conversation["previous_execution_sha256"] == first["execution_sha256"] + assert conversation["prior_turn_state_bound"] is True + + second_result["conversation_before"]["messages_sha256"] = "0" * 64 + broken = build( + second_result, + previous_execution_sha256=first["execution_sha256"], + expected_previous_conversation=first_result["conversation_after"], + ) + assert "conversation_prior_turn_binding" in broken["attribution"]["missing_required_bindings"] diff --git a/tests/test_run_leo_direct_claim_handler_suite.py b/tests/test_run_leo_direct_claim_handler_suite.py new file mode 100644 index 0000000..3732ac2 --- /dev/null +++ b/tests/test_run_leo_direct_claim_handler_suite.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import json +import subprocess + +from scripts import run_leo_direct_claim_handler_suite as suite + + +def safe_report() -> dict: + return { + "remote_returncode": 0, + "pass_runtime": True, + "handler": {"authorized": True}, + "results": [ + { + "ok": True, + "mutates_kb": False, + "database_tool_trace": { + "database_tool_call_count": 1, + "database_tool_completed_count": 1, + "database_tool_calls_read_only": True, + "calls": [ + { + "database_invocations": [{"access_mode": "read_only"}], + "result": { + "content_sha256": "a" * 64, + "error_detected": False, + "nonempty": True, + }, + } + ], + }, + } + ], + "posted_to_telegram": False, + "mutates_kb_by_harness": False, + "db_counts_changed": False, + "db_fingerprint_unchanged": True, + "live_behavior_manifest_unchanged": True, + "service_before_after": {"unchanged_from_preexisting_live_readback": True}, + "temp_profile_removed": True, + "model_call_trace_all_bound": True, + "execution_manifest_summary": {"all_turns_attribution_complete": True}, + } + + +def test_redaction_preserves_json_and_removes_secret_values() -> None: + raw = json.dumps( + { + "bot_token": "123456789:abcdefghijklmnopqrstuvwxyz_ABCD", + "api_key": "sk-live-secret", + "nested": {"password": "hunter2", "access_token": "bearer-secret"}, + "message": "Authorization: Bearer top-secret-value", + } + ) + + redacted = suite.redact(raw) + parsed = json.loads(redacted) + + assert parsed["bot_token"] == "[REDACTED]" + assert parsed["api_key"] == "[REDACTED]" + assert parsed["nested"]["password"] == "[REDACTED]" + assert parsed["nested"]["access_token"] == "[REDACTED]" + assert parsed["message"] == "Authorization: Bearer [REDACTED]" + assert "hunter2" not in redacted + assert "top-secret-value" not in redacted + + +def test_successful_stdout_is_redacted_before_json_parsing(monkeypatch) -> None: + secret = "123456789:abcdefghijklmnopqrstuvwxyz_ABCD" + calls = iter( + [ + subprocess.CompletedProcess([], 0, stdout=json.dumps({"bot_token": secret}) + "\n", stderr=""), + subprocess.CompletedProcess([], 0, stdout="", stderr=""), + ] + ) + monkeypatch.setattr(suite.subprocess, "run", lambda *args, **kwargs: next(calls)) + + result = suite.run_remote(prompts=[]) + + assert result["parsed"]["bot_token"] == "[REDACTED]" + assert secret not in json.dumps(result) + + +def test_safety_gate_passes_only_for_complete_no_send_no_write_run() -> None: + report = safe_report() + + assert suite.report_passes_safety(report) is True + assert suite.build_safety_gate(report)["failed_checks"] == [] + + +def test_safety_gate_rejects_write_capable_tool_and_incomplete_receipt() -> None: + report = safe_report() + report["results"][0]["database_tool_trace"]["database_tool_calls_read_only"] = False + report["results"][0]["database_tool_trace"]["calls"][0]["database_invocations"][0][ + "access_mode" + ] = "write" + report["execution_manifest_summary"]["all_turns_attribution_complete"] = False + + gate = suite.build_safety_gate(report) + + assert gate["status"] == "fail" + assert "all_tool_traces_read_only_and_bound" in gate["failed_checks"] + assert "all_turn_manifests_complete" in gate["failed_checks"] + + +def test_safety_gate_rejects_missing_no_send_declaration() -> None: + report = safe_report() + report.pop("posted_to_telegram") + + gate = suite.build_safety_gate(report) + + assert gate["status"] == "fail" + assert "no_telegram_post" in gate["failed_checks"] diff --git a/tests/test_verify_leo_db_first_oos_canary.py b/tests/test_verify_leo_db_first_oos_canary.py index 16924fc..7d89fb6 100644 --- a/tests/test_verify_leo_db_first_oos_canary.py +++ b/tests/test_verify_leo_db_first_oos_canary.py @@ -46,6 +46,7 @@ def passing_report() -> dict: "NRestarts": "0", } counts = {"public.claims": 1, "public.sources": 2} + fingerprint = {"fingerprint_sha256": "d" * 64} behavior = {"behavior_sha256": "b" * 64} return { "generated_at_utc": "2026-07-14T08:05:27+00:00", @@ -55,11 +56,16 @@ def passing_report() -> dict: "db_counts_changed": False, "db_counts_before": counts, "db_counts_after": counts, + "db_fingerprint_before": fingerprint, + "db_fingerprint_after": fingerprint, + "db_fingerprint_unchanged": True, "changed_live_profile": False, "live_behavior_manifest_unchanged": True, "live_behavior_manifest_before": behavior, "live_behavior_manifest_after": behavior, "database_response_validation_all_ok": True, + "execution_manifest_summary": {"all_turns_attribution_complete": True}, + "safety_gate": {"status": "pass"}, "temp_profile_removed": True, "service_before_after": { "before": service,