diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ed708df..2ab589e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -58,6 +58,7 @@ jobs: scripts/check_llm_refinement_contract.py \ scripts/build_working_leo_m3taversal_outcome_sandbox.py \ scripts/leo_behavior_manifest.py \ + scripts/leo_tool_trace.py \ scripts/replay_decision_engine_eval.py \ scripts/prove_phase1b_local.py \ scripts/run_gcp_generated_db_direct_claim_suite.py \ @@ -86,6 +87,7 @@ jobs: tests/test_hermes_leoclean_kb_bridge_source.py \ tests/test_hermes_leoclean_skill_surfaces.py \ tests/test_leo_behavior_manifest.py \ + tests/test_leo_tool_trace.py \ tests/test_compile_kb_source_packet.py \ tests/test_verify_postgres_parity_manifest.py \ tests/test_working_leo_m3taversal_oos_benchmark.py \ diff --git a/scripts/leo_tool_trace.py b/scripts/leo_tool_trace.py new file mode 100644 index 0000000..c446247 --- /dev/null +++ b/scripts/leo_tool_trace.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +"""Extract a secret-safe proof summary from Hermes tool-call messages.""" + +from __future__ import annotations + +import hashlib +import json +import re +from typing import Any + +READ_ONLY_SUBCOMMANDS = frozenset( + { + "context", + "decision-matrix-status", + "edges", + "evidence", + "list-proposals", + "search", + "search-proposals", + "show", + "show-proposal", + "status", + } +) +MUTATING_SUBCOMMANDS = frozenset( + { + "prepare-source", + "propose-attachment-eval", + "propose-edge", + "propose-source", + "record-document-eval", + } +) +KNOWN_SUBCOMMANDS = READ_ONLY_SUBCOMMANDS | MUTATING_SUBCOMMANDS + +UUID_RE = re.compile( + r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}\b" +) +SHA256_RE = re.compile(r"\b[0-9a-fA-F]{64}\b") +TELEO_KB_COMMAND_RE = re.compile( + r"(?:^|[\s;&|()])(?:[^\s;&|()]+/)?teleo-kb\s+(?P[a-z][a-z0-9-]*)\b", + re.IGNORECASE, +) +KB_TOOL_PY_COMMAND_RE = re.compile( + r"(?:^|[\s;&|()])(?:python(?:3(?:\.\d+)?)?\s+)?(?:[^\s;&|()]+/)?kb_tool\.py\s+" + r"(?P[a-z][a-z0-9-]*)\b", + re.IGNORECASE, +) +RECEIPT_SCHEMA_RE = re.compile(r"livingip\.teleoKbRetrievalReceipt\.v\d+") +SEMANTIC_SHA_RE = re.compile(r"semantic context SHA-256:\s*`?([0-9a-fA-F]{64})", re.IGNORECASE) +ARTIFACT_SHA_RE = re.compile(r"artifact state SHA-256:\s*`?([0-9a-fA-F]{64})", re.IGNORECASE) +CONSISTENCY_RE = re.compile(r"DB read consistency:\s*`?([a-z0-9_-]+)", re.IGNORECASE) +ERROR_RE = re.compile( + r"(?:traceback \(most recent call last\)|command (?:failed|timed out)|permission denied|" + r"no such file or directory|connection (?:refused|failed))", + re.IGNORECASE, +) + + +def _sha256_bytes(value: bytes) -> str: + return hashlib.sha256(value).hexdigest() + + +def _canonical_bytes(value: Any) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + + +def _normalize_tool_call(tool_call: Any) -> tuple[str | None, str, Any]: + if not isinstance(tool_call, dict): + return None, "", None + function = tool_call.get("function") + if isinstance(function, dict): + return tool_call.get("id"), str(function.get("name") or ""), function.get("arguments") + return tool_call.get("id"), str(tool_call.get("name") or ""), tool_call.get("arguments") + + +def _parse_arguments(arguments: Any) -> Any: + if not isinstance(arguments, str): + return arguments + try: + return json.loads(arguments) + except json.JSONDecodeError: + return arguments + + +def _command_candidates(value: Any) -> list[str]: + if isinstance(value, str): + return [value] + if not isinstance(value, dict): + return [] + candidates = [] + for key in ("command", "cmd", "script", "shell_command"): + candidate = value.get(key) + if isinstance(candidate, str): + candidates.append(candidate) + return candidates + + +def _database_invocations(tool_name: str, arguments: Any) -> list[dict[str, Any]]: + parsed_arguments = _parse_arguments(arguments) + invocations: list[dict[str, Any]] = [] + for command in _command_candidates(parsed_arguments): + for executable, pattern in (("teleo-kb", TELEO_KB_COMMAND_RE), ("kb_tool.py", KB_TOOL_PY_COMMAND_RE)): + for match in pattern.finditer(command): + subcommand = match.group("subcommand").lower() + if subcommand not in KNOWN_SUBCOMMANDS: + continue + invocations.append( + { + "executable": executable, + "subcommand": subcommand, + "access_mode": "read_only" if subcommand in READ_ONLY_SUBCOMMANDS else "staging_write", + "command_sha256": _sha256_bytes(command.encode("utf-8")), + } + ) + normalized_tool_name = tool_name.lower().replace("_", "-") + if normalized_tool_name in {"teleo-kb", "kb-tool"} and isinstance(parsed_arguments, dict): + subcommand = str(parsed_arguments.get("subcommand") or parsed_arguments.get("action") or "").lower() + if subcommand in KNOWN_SUBCOMMANDS: + invocations.append( + { + "executable": normalized_tool_name, + "subcommand": subcommand, + "access_mode": "read_only" if subcommand in READ_ONLY_SUBCOMMANDS else "staging_write", + "command_sha256": _sha256_bytes(_canonical_bytes(parsed_arguments)), + } + ) + return invocations + + +def _sanitize_result(content: Any) -> dict[str, Any]: + text = content if isinstance(content, str) else json.dumps(content, sort_keys=True, default=str) + encoded = text.encode("utf-8") + uuids = sorted(set(UUID_RE.findall(text))) + hashes = sorted({value.lower() for value in SHA256_RE.findall(text)}) + schema_match = RECEIPT_SCHEMA_RE.search(text) + semantic_match = SEMANTIC_SHA_RE.search(text) + artifact_match = ARTIFACT_SHA_RE.search(text) + consistency_match = CONSISTENCY_RE.search(text) + retrieval_receipt = None + if schema_match or semantic_match or artifact_match or consistency_match: + retrieval_receipt = { + "schema": schema_match.group(0) if schema_match else None, + "semantic_context_sha256": semantic_match.group(1).lower() if semantic_match else None, + "artifact_state_sha256": artifact_match.group(1).lower() if artifact_match else None, + "read_consistency_status": consistency_match.group(1).lower() if consistency_match else None, + } + return { + "content_sha256": _sha256_bytes(encoded), + "content_bytes": len(encoded), + "nonempty": bool(text.strip()), + "error_detected": bool(ERROR_RE.search(text)), + "row_ids": uuids[:64], + "row_id_count": len(uuids), + "sha256_values": hashes[:64], + "sha256_value_count": len(hashes), + "retrieval_receipt": retrieval_receipt, + } + + +def extract_kb_tool_trace(messages: list[dict[str, Any]] | None) -> dict[str, Any]: + """Return proof metadata without retaining raw commands, arguments, or results.""" + + calls_by_raw_id: dict[str, dict[str, Any]] = {} + calls: list[dict[str, Any]] = [] + for message_index, message in enumerate(messages or []): + if not isinstance(message, dict) or message.get("role") != "assistant": + continue + for call_index, tool_call in enumerate(message.get("tool_calls") or []): + raw_id, tool_name, arguments = _normalize_tool_call(tool_call) + invocations = _database_invocations(tool_name, arguments) + if not invocations: + continue + raw_id_text = str(raw_id or f"message-{message_index}-call-{call_index}") + record = { + "message_index": message_index, + "call_index": call_index, + "tool_call_id_sha256": _sha256_bytes(raw_id_text.encode("utf-8")), + "tool_name": tool_name, + "arguments_sha256": _sha256_bytes(_canonical_bytes(_parse_arguments(arguments))), + "database_invocations": invocations, + "result": None, + } + calls.append(record) + calls_by_raw_id[raw_id_text] = record + + for message in messages or []: + if not isinstance(message, dict) or message.get("role") not in {"tool", "function"}: + continue + raw_id = message.get("tool_call_id") + record = calls_by_raw_id.get(str(raw_id)) if raw_id is not None else None + if record is not None: + record["result"] = _sanitize_result(message.get("content")) + + completed = [ + call + for call in calls + if call["result"] + and call["result"]["nonempty"] + and not call["result"]["error_detected"] + ] + receipt_calls = [ + call + for call in completed + if call["result"].get("retrieval_receipt") + and call["result"]["retrieval_receipt"].get("semantic_context_sha256") + and call["result"]["retrieval_receipt"].get("artifact_state_sha256") + ] + access_modes = { + invocation["access_mode"] + for call in calls + for invocation in call["database_invocations"] + } + return { + "schema": "livingip.leoKbToolTrace.v1", + "database_tool_call_count": len(calls), + "database_tool_completed_count": len(completed), + "database_tool_call_proven": bool(completed), + "database_retrieval_receipt_proven": bool(receipt_calls), + "database_tool_calls_read_only": bool(calls) and access_modes == {"read_only"}, + "access_modes": sorted(access_modes), + "calls": calls, + "raw_commands_retained": False, + "raw_arguments_retained": False, + "raw_results_retained": False, + } + diff --git a/scripts/run_leo_direct_claim_handler_suite.py b/scripts/run_leo_direct_claim_handler_suite.py index 1185882..44f0943 100755 --- a/scripts/run_leo_direct_claim_handler_suite.py +++ b/scripts/run_leo_direct_claim_handler_suite.py @@ -205,9 +205,25 @@ def read_database_context_trace(path): return records +def current_agent_messages(runner, session_key): + cache = getattr(runner, "_agent_cache", None) or {} + cache_lock = getattr(runner, "_agent_cache_lock", None) + if cache_lock is not None: + with cache_lock: + cached = cache.get(session_key) + else: + cached = cache.get(session_key) + if not cached: + return [] + agent = cached[0] + messages = getattr(agent, "_session_messages", None) or [] + return [dict(message) for message in messages if isinstance(message, dict)] + + async def run_suite(): sys.path.insert(0, str(DEPLOY_ROOT / "scripts")) from leo_behavior_manifest import build_manifest as build_behavior_manifest + from leo_tool_trace import extract_kb_tool_trace before_service = service_state() before_counts = db_counts() @@ -301,6 +317,7 @@ 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)) results.append( { "turn": index, @@ -315,6 +332,7 @@ async def run_suite(): "evidence_tier": "live_vps_gatewayrunner_temp_profile", "claim_ceiling": "Live VPS GatewayRunner reply from temp leoclean profile; no Telegram post; no production DB apply authorized.", "database_context_trace": read_database_context_trace(context_trace_path)[trace_start:], + "database_tool_trace": database_tool_trace, } ) report["results"] = results @@ -367,6 +385,26 @@ async def run_suite(): report["database_response_transform_count"] = sum( item.get("transformed") is True for item in response_validations ) + tool_traces = [ + result.get("database_tool_trace") or {} + for result in report.get("results") or [] + ] + report["database_tool_call_proven_count"] = sum( + trace.get("database_tool_call_proven") is True for trace in tool_traces + ) + report["database_tool_call_proven"] = any( + trace.get("database_tool_call_proven") is True for trace in tool_traces + ) + report["database_retrieval_receipt_proven"] = any( + trace.get("database_retrieval_receipt_proven") is True for trace in tool_traces + ) + tool_traces_with_calls = [ + trace for trace in tool_traces if trace.get("database_tool_call_count") + ] + report["database_tool_calls_read_only"] = bool(tool_traces_with_calls) and all( + trace.get("database_tool_calls_read_only") is True + for trace in tool_traces_with_calls + ) write_report_snapshot(report) diff --git a/tests/test_hermes_leoclean_db_context_plugin.py b/tests/test_hermes_leoclean_db_context_plugin.py index cdbf4bd..66a12ce 100644 --- a/tests/test_hermes_leoclean_db_context_plugin.py +++ b/tests/test_hermes_leoclean_db_context_plugin.py @@ -471,3 +471,7 @@ def test_handler_harness_retains_database_context_proof_fields() -> None: assert '"database_response_validation_count"' in source assert '"database_response_validation_all_ok"' in source assert '"database_response_transform_count"' in source + assert '"database_tool_trace"' in source + assert '"database_tool_call_proven"' in source + assert '"database_retrieval_receipt_proven"' in source + assert '"database_tool_calls_read_only"' in source diff --git a/tests/test_leo_tool_trace.py b/tests/test_leo_tool_trace.py new file mode 100644 index 0000000..13aee54 --- /dev/null +++ b/tests/test_leo_tool_trace.py @@ -0,0 +1,157 @@ +"""Tests for secret-safe Hermes knowledge-base tool attribution.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) + +from leo_tool_trace import extract_kb_tool_trace # noqa: E402 + + +def test_extracts_completed_read_only_context_call_and_receipt() -> None: + claim_id = "2a7ae257-d01d-46f4-b813-63f81bb9c7c7" + source_id = "15740795-1234-4abc-8def-1234567890ab" + semantic_sha = "8" * 64 + artifact_sha = "4" * 64 + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "call-secret-looking-id", + "function": { + "name": "terminal", + "arguments": '{"command":"teleo-kb context \\\"AI sandbagging\\\""}', + }, + } + ], + }, + { + "role": "tool", + "tool_call_id": "call-secret-looking-id", + "content": ( + "# Teleo KB Context\n" + "- schema: `livingip.teleoKbRetrievalReceipt.v1`\n" + f"- semantic context SHA-256: `{semantic_sha}`\n" + f"- artifact state SHA-256: `{artifact_sha}`\n" + "- DB read consistency: `stable_wal_marker`; attempts: `1`\n" + f"claim_id: {claim_id}; source_id: {source_id}\n" + ), + }, + ] + + trace = extract_kb_tool_trace(messages) + + assert trace["database_tool_call_proven"] is True + assert trace["database_retrieval_receipt_proven"] is True + assert trace["database_tool_calls_read_only"] is True + assert trace["access_modes"] == ["read_only"] + assert trace["calls"][0]["database_invocations"][0]["subcommand"] == "context" + assert trace["calls"][0]["result"]["row_ids"] == sorted([claim_id, source_id]) + assert trace["calls"][0]["result"]["retrieval_receipt"] == { + "schema": "livingip.teleoKbRetrievalReceipt.v1", + "semantic_context_sha256": semantic_sha, + "artifact_state_sha256": artifact_sha, + "read_consistency_status": "stable_wal_marker", + } + + +def test_never_retains_raw_command_arguments_or_result() -> None: + secret = "password=do-not-retain" + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "call-1", + "function": { + "name": "terminal", + "arguments": {"command": f"teleo-kb search demo; echo {secret}"}, + }, + } + ], + }, + {"role": "tool", "tool_call_id": "call-1", "content": f"result {secret}"}, + ] + + trace = extract_kb_tool_trace(messages) + rendered = str(trace) + + assert trace["database_tool_call_proven"] is True + assert secret not in rendered + assert "teleo-kb search demo" not in rendered + assert trace["raw_commands_retained"] is False + assert trace["raw_arguments_retained"] is False + assert trace["raw_results_retained"] is False + + +def test_staging_write_is_classified_and_not_called_read_only() -> None: + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "call-write", + "function": { + "name": "terminal", + "arguments": '{"command":"teleo-kb propose-edge --from a --to b"}', + }, + } + ], + }, + {"role": "tool", "tool_call_id": "call-write", "content": "pending_review"}, + ] + + trace = extract_kb_tool_trace(messages) + + assert trace["database_tool_call_proven"] is True + assert trace["database_tool_calls_read_only"] is False + assert trace["access_modes"] == ["staging_write"] + + +def test_unmatched_or_failed_result_does_not_prove_database_call() -> None: + unmatched = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "call-missing", + "function": {"name": "terminal", "arguments": '{"command":"teleo-kb show abc"}'}, + } + ], + } + ] + failed = [ + *unmatched, + { + "role": "tool", + "tool_call_id": "call-missing", + "content": "Traceback (most recent call last): connection failed", + } + ] + + assert extract_kb_tool_trace(unmatched)["database_tool_call_proven"] is False + assert extract_kb_tool_trace(failed)["database_tool_call_proven"] is False + + +def test_regular_terminal_call_is_ignored() -> None: + messages = [ + { + "role": "assistant", + "tool_calls": [ + { + "id": "call-ls", + "function": {"name": "terminal", "arguments": '{"command":"ls -la"}'}, + } + ], + }, + {"role": "tool", "tool_call_id": "call-ls", "content": "total 8"}, + ] + + trace = extract_kb_tool_trace(messages) + + assert trace["database_tool_call_count"] == 0 + assert trace["database_tool_call_proven"] is False