Prove Leo database tool attribution

.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 \

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<subcommand>[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<subcommand>[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,
+    }
+

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)
 
 

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

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