253 lines
8.4 KiB
Python
253 lines
8.4 KiB
Python
"""Tests for secret-safe Hermes knowledge-base tool attribution."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
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_structured_nonzero_exit_does_not_prove_database_call() -> None:
|
|
messages = [
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [
|
|
{
|
|
"id": "call-nonzero",
|
|
"function": {
|
|
"name": "terminal",
|
|
"arguments": '{"command":"teleo-kb context broad-question"}',
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "tool",
|
|
"tool_call_id": "call-nonzero",
|
|
"content": '{"output":"usage error","exit_code":2,"error":null}',
|
|
},
|
|
]
|
|
|
|
trace = extract_kb_tool_trace(messages)
|
|
|
|
assert trace["database_tool_call_proven"] is False
|
|
assert trace["database_tool_completed_count"] == 0
|
|
assert trace["calls"][0]["result"]["error_detected"] is True
|
|
|
|
|
|
def test_actual_staging_subcommands_and_unknown_subcommands_fail_read_only_classification() -> None:
|
|
messages = []
|
|
commands = (
|
|
"teleo-kb propose-attachment-evaluation packet.json",
|
|
"teleo-kb record-document-evaluation packet.json",
|
|
"teleo-kb future-command --flag",
|
|
)
|
|
for index, command in enumerate(commands):
|
|
call_id = f"call-{index}"
|
|
messages.extend(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [
|
|
{
|
|
"id": call_id,
|
|
"function": {"name": "terminal", "arguments": json.dumps({"command": command})},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": call_id, "content": "completed"},
|
|
]
|
|
)
|
|
|
|
trace = extract_kb_tool_trace(messages)
|
|
|
|
assert trace["database_tool_call_count"] == 3
|
|
assert trace["database_tool_calls_read_only"] is False
|
|
assert trace["access_modes"] == ["staging_write", "unknown_write_risk"]
|
|
|
|
|
|
def test_direct_kb_tool_global_flags_do_not_hide_mutating_or_unknown_subcommands() -> None:
|
|
messages = []
|
|
commands = (
|
|
"python3 /profile/bin/kb_tool.py --format json --local propose-edge from supports to --rationale test",
|
|
"python3 /profile/bin/kb_tool.py --local --format=json --db teleo --container teleo-pg future-command",
|
|
"python3 /profile/bin/kb_tool.py --ssh=teleo@example --local --format markdown context topic",
|
|
)
|
|
for index, command in enumerate(commands):
|
|
call_id = f"direct-{index}"
|
|
messages.extend(
|
|
[
|
|
{
|
|
"role": "assistant",
|
|
"tool_calls": [
|
|
{
|
|
"id": call_id,
|
|
"function": {"name": "terminal", "arguments": json.dumps({"command": command})},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": call_id, "content": "completed"},
|
|
]
|
|
)
|
|
|
|
trace = extract_kb_tool_trace(messages)
|
|
|
|
assert trace["database_tool_call_count"] == 3
|
|
assert trace["database_tool_calls_read_only"] is False
|
|
assert trace["access_modes"] == ["read_only", "staging_write", "unknown_write_risk"]
|
|
assert [call["database_invocations"][0]["subcommand"] for call in trace["calls"]] == [
|
|
"propose-edge",
|
|
"future-command",
|
|
"context",
|
|
]
|
|
|
|
|
|
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
|