teleo/teleo-infrastructure
8
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions 1

Harden Leo turn attribution and no-write gate

Browse source
This commit is contained in:
twentyOne2x 2026-07-15 00:33:16 +02:00
parent c2eebb9654
commit 4088c9c8f5
9 changed files with 1228 additions and 156 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

91
scripts/leo_behavior_manifest.py
View file

@ -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": {

541
scripts/leo_turn_execution_manifest.py
View file

@ -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),

280
scripts/run_leo_direct_claim_handler_suite.py
View file

@ -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__ = "<embedded leo_behavior_manifest.py>"
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
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__":

17
scripts/verify_leo_db_first_oos_canary.py
View file

@ -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,

7
tests/test_hermes_leoclean_db_context_plugin.py
View file

@ -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

42
tests/test_leo_behavior_manifest.py
View file

@ -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

284
tests/test_leo_turn_execution_manifest.py
View file

@ -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"]

114
tests/test_run_leo_direct_claim_handler_suite.py Normal file
View file

@ -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"]

6
tests/test_verify_leo_db_first_oos_canary.py
View file

@ -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,