430 lines
19 KiB
Python
430 lines
19 KiB
Python
"""Tests for automatic live database-contract injection into Leo turns."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PLUGIN = ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
|
|
|
|
|
|
def load_plugin():
|
|
spec = importlib.util.spec_from_file_location("leo_db_context_plugin", PLUGIN)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_plugin_registers_generation_and_delivery_hooks() -> None:
|
|
module = load_plugin()
|
|
registered = []
|
|
module.register(SimpleNamespace(register_hook=lambda name, callback: registered.append((name, callback))))
|
|
assert [name for name, _callback in registered] == ["pre_llm_call", "post_llm_call"]
|
|
|
|
|
|
def test_plugin_injects_only_operational_contracts_and_writes_redacted_trace(tmp_path: Path, monkeypatch) -> None:
|
|
module = load_plugin()
|
|
home = tmp_path / "profile"
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("# fixture\n", encoding="utf-8")
|
|
trace = tmp_path / "trace.jsonl"
|
|
monkeypatch.setenv("LEO_DB_CONTEXT_TRACE_PATH", str(trace))
|
|
payload = {
|
|
"query": "must not be injected",
|
|
"claims": [{"text": "must not be injected"}],
|
|
"operational_contracts": [
|
|
{"id": "reply_budget", "hard_max_words": 220},
|
|
{"id": "source_intake", "capability_tier": "build-local compiler only"},
|
|
],
|
|
"compiled_response": "must not be injected either",
|
|
}
|
|
calls = []
|
|
|
|
def fake_runner(command, **kwargs):
|
|
calls.append((command, kwargs))
|
|
return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")
|
|
|
|
context = module.build_database_context(
|
|
"operator secret-shaped but nonsecret message", hermes_home=home, runner=fake_runner
|
|
)
|
|
|
|
assert "reply_budget" in context
|
|
assert "source_intake" in context
|
|
assert "build-local compiler only" in context
|
|
assert "must not be injected" not in context
|
|
assert "operator secret-shaped" not in trace.read_text(encoding="utf-8")
|
|
record = json.loads(trace.read_text(encoding="utf-8"))
|
|
assert record["status"] == "ok"
|
|
assert record["injected"] is True
|
|
assert record["contract_ids"] == ["reply_budget", "source_intake"]
|
|
command, kwargs = calls[0]
|
|
assert command[2:5] == ["--local", "context", "operator secret-shaped but nonsecret message"]
|
|
assert command[-6:] == ["--limit", "0", "--context-limit", "0", "--format", "json"]
|
|
assert kwargs["timeout"] == module.DEFAULT_TIMEOUT_SECONDS
|
|
|
|
|
|
def test_plugin_replaces_contract_violating_draft_with_compiled_db_response(tmp_path: Path, monkeypatch) -> None:
|
|
module = load_plugin()
|
|
home = tmp_path / "profile"
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("# fixture\n", encoding="utf-8")
|
|
trace = tmp_path / "trace.jsonl"
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
monkeypatch.setenv("LEO_DB_CONTEXT_TRACE_PATH", str(trace))
|
|
compiled = (
|
|
"Map claims with sources and evidence; put the framework in reasoning_tools, the rule in "
|
|
"behavioral_rules, an evaluative gate in governance_gates, and stage the belief correction. "
|
|
"Stage one pending_review proposal. approve_claim applies only claims, sources, evidence, edges, and "
|
|
"reasoning_tools; approve_claim supports neither behavioral_rules nor governance_gates. Belief updates "
|
|
"remain staged until a separate reviewed apply capability exists. Receipt: proposal applied_at and "
|
|
"postflight row readback."
|
|
)
|
|
payload = {
|
|
"operational_contracts": [
|
|
{"id": "reply_budget", "hard_max_words": 200},
|
|
{"id": "mixed_packet_composition"},
|
|
],
|
|
"compiled_response": compiled,
|
|
}
|
|
|
|
def fake_runner(command, **_kwargs):
|
|
return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")
|
|
|
|
snapshot = module._load_database_snapshot("compose this packet", hermes_home=home, runner=fake_runner)
|
|
module._store_snapshot("session-1", snapshot)
|
|
result = module._post_llm_call(
|
|
session_id="session-1",
|
|
user_message="compose this packet",
|
|
assistant_response=(
|
|
"Claims, sources, evidence, reasoning_tools, behavioral_rules, governance_gates, and beliefs. "
|
|
"Stage and review a proposal, then apply only the supported collections (claims, sources, evidence, "
|
|
"edges, reasoning_tools, belief updates). approve_claim does not cover behavioral_rules or "
|
|
"governance_gates; use a separate reviewed apply path. Receipt: applied_at and postflight readback."
|
|
),
|
|
)
|
|
|
|
assert result == {"assistant_response": compiled}
|
|
records = [json.loads(line) for line in trace.read_text(encoding="utf-8").splitlines()]
|
|
assert records[-1]["event"] == "post_llm_call"
|
|
assert records[-1]["transformed"] is True
|
|
assert records[-1]["issues"] == [
|
|
"unsupported_collection_listed_as_supported",
|
|
"unsupported_collection_listed_in_apply",
|
|
]
|
|
assert "compose this packet" not in trace.read_text(encoding="utf-8")
|
|
|
|
|
|
def test_plugin_keeps_contract_valid_draft(tmp_path: Path) -> None:
|
|
module = load_plugin()
|
|
home = tmp_path / "profile"
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("# fixture\n", encoding="utf-8")
|
|
valid = (
|
|
"Map claims with sources and evidence; put the framework in reasoning_tools, the rule in "
|
|
"behavioral_rules, an evaluative gate in governance_gates, and stage the belief correction. "
|
|
"Stage one pending_review proposal. approve_claim applies only claims, sources, evidence, edges, and "
|
|
"reasoning_tools; approve_claim supports neither behavioral_rules nor governance_gates. Belief updates "
|
|
"remain staged until a separate reviewed apply capability exists. Receipt: proposal applied_at and "
|
|
"postflight row readback."
|
|
)
|
|
payload = {
|
|
"operational_contracts": [
|
|
{"id": "reply_budget", "hard_max_words": 200},
|
|
{"id": "mixed_packet_composition"},
|
|
],
|
|
"compiled_response": valid,
|
|
}
|
|
|
|
def fake_runner(command, **_kwargs):
|
|
return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")
|
|
|
|
snapshot = module._load_database_snapshot("compose this packet", hermes_home=home, runner=fake_runner)
|
|
module._store_snapshot("session-2", snapshot)
|
|
assert (
|
|
module._post_llm_call(session_id="session-2", user_message="compose this packet", assistant_response=valid)
|
|
is None
|
|
)
|
|
|
|
|
|
def test_source_validator_accepts_explicit_schema_and_canonical_write_prohibitions() -> None:
|
|
module = load_plugin()
|
|
response = (
|
|
"Capture and hash the artifact without authorization, retain its URL or storage_path, and stage candidates "
|
|
"in a pending_review proposal. Author, title, and publisher are not current public.sources columns. Do not "
|
|
"create public.sources or other canonical rows during staging. VPS staging is shipped for prepared inputs; "
|
|
"autonomous chat attachment extraction is not shipped. Approval begins before canonical apply."
|
|
)
|
|
contracts = [
|
|
{"id": "reply_budget", "hard_max_words": 200},
|
|
{"id": "source_intake"},
|
|
]
|
|
|
|
assert module.response_contract_issues(response, contracts) == []
|
|
|
|
|
|
def test_broad_report_learning_question_requires_database_truth() -> None:
|
|
module = load_plugin()
|
|
query = (
|
|
"If I send you a report here, can you learn it into your knowledge base and make it part of what you know? "
|
|
"What actually happens right now?"
|
|
)
|
|
|
|
assert module._requires_database_truth(query) is True
|
|
|
|
|
|
def test_plugin_replaces_memory_only_status_answer_with_exact_live_db_receipt(tmp_path: Path, monkeypatch) -> None:
|
|
module = load_plugin()
|
|
home = tmp_path / "profile"
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("# fixture\n", encoding="utf-8")
|
|
monkeypatch.setenv("HERMES_HOME", str(home))
|
|
database_status = {
|
|
"high_signal_rows": {
|
|
"claims": 1837,
|
|
"sources": 4145,
|
|
"claim_edges": 4916,
|
|
"claim_evidence": 4670,
|
|
"kb_proposals": 26,
|
|
},
|
|
"proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 14, "canceled": 7},
|
|
}
|
|
contracts = [
|
|
{"id": "reply_budget", "hard_max_words": 200},
|
|
{"id": "proposal_state_readback", "database_status": database_status},
|
|
]
|
|
compiled = (
|
|
"Partly.\n"
|
|
"DB readback: claims: 1837; sources: 4145; claim_edges: 4916; claim_evidence: 4670; kb_proposals: 26.\n"
|
|
"Proposal states are applied: 2, approved: 3, pending_review: 14, canceled: 7. Approved is not applied, "
|
|
"and applied_at plus row-level postflight is required.\n\n"
|
|
"Next proof-changing follow-up: choose the proposal UUID, compare public.* rows, and request explicit "
|
|
"guarded-apply authorization only after strict payload review."
|
|
)
|
|
payload = {"operational_contracts": contracts, "compiled_response": compiled}
|
|
|
|
assert module.response_contract_issues(compiled, contracts) == []
|
|
invalid = "Two are applied and three are approved. Want me to list them?"
|
|
invalid_issues = module.response_contract_issues(invalid, contracts)
|
|
assert "missing_live_count_readback:proposal_state_readback" in invalid_issues
|
|
assert "missing_proof_changing_followup" in invalid_issues
|
|
|
|
def fake_runner(command, **_kwargs):
|
|
return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")
|
|
|
|
query = "What changed in the KB rather than staying proposed?"
|
|
snapshot = module._load_database_snapshot(query, hermes_home=home, runner=fake_runner)
|
|
module._store_snapshot("session-direct", snapshot)
|
|
assert module._post_llm_call(session_id="session-direct", user_message=query, assistant_response=invalid) == {
|
|
"assistant_response": compiled
|
|
}
|
|
|
|
alternative_valid = compiled.replace("Partly.", "Current state.")
|
|
assert module.response_contract_issues(alternative_valid, contracts) == []
|
|
module._store_snapshot("session-alternative", snapshot)
|
|
assert module._post_llm_call(
|
|
session_id="session-alternative", user_message=query, assistant_response=alternative_valid
|
|
) == {"assistant_response": compiled}
|
|
|
|
contradictory = compiled + " All approved rows are already canonical."
|
|
assert "contradictory_approved_state_claim" in module.response_contract_issues(contradictory, contracts)
|
|
module._store_snapshot("session-contradictory", snapshot)
|
|
assert module._post_llm_call(
|
|
session_id="session-contradictory", user_message=query, assistant_response=contradictory
|
|
) == {"assistant_response": compiled}
|
|
|
|
|
|
def test_plugin_requires_named_proposal_receipt_when_live_match_exists() -> None:
|
|
module = load_plugin()
|
|
database_status = {
|
|
"high_signal_rows": {
|
|
"claims": 1837,
|
|
"sources": 4145,
|
|
"claim_edges": 4916,
|
|
"claim_evidence": 4670,
|
|
"kb_proposals": 29,
|
|
},
|
|
"proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 16, "canceled": 8},
|
|
}
|
|
contracts = [
|
|
{"id": "reply_budget", "hard_max_words": 200},
|
|
{
|
|
"id": "proposal_state_readback",
|
|
"database_status": database_status,
|
|
"matching_proposals": [
|
|
{
|
|
"id": "10bc0719-1c2b-5f42-a41e-86e6478692cb",
|
|
"status": "pending_review",
|
|
"applied_at": None,
|
|
}
|
|
],
|
|
},
|
|
]
|
|
common = (
|
|
"DB readback: claims: 1837; sources: 4145; claim_edges: 4916; claim_evidence: 4670; kb_proposals: 29.\n"
|
|
"Proposal states are applied: 2, approved: 3, pending_review: 16, canceled: 8. Approved is not applied; "
|
|
"applied_at and row-level postflight are required.\n\n"
|
|
"Next proof-changing follow-up: inspect the named proposal, then return its apply receipt and public rows."
|
|
)
|
|
valid = (
|
|
"No.\n"
|
|
"DB readback: proposal: 10bc0719-1c2b-5f42-a41e-86e6478692cb; status: pending_review; "
|
|
"applied_at: none; readiness: needs_human_review.\n"
|
|
+ common
|
|
)
|
|
|
|
missing_receipt_issues = module.response_contract_issues("No.\n" + common, contracts)
|
|
|
|
assert "missing_matching_proposal_readback" in missing_receipt_issues
|
|
assert module.response_contract_issues(valid, contracts) == []
|
|
|
|
|
|
def test_plugin_fails_closed_when_database_contract_lookup_fails(tmp_path: Path) -> None:
|
|
module = load_plugin()
|
|
home = tmp_path / "profile"
|
|
kb_tool = home / "bin" / "kb_tool.py"
|
|
kb_tool.parent.mkdir(parents=True)
|
|
kb_tool.write_text("# fixture\n", encoding="utf-8")
|
|
|
|
def fake_runner(command, **_kwargs):
|
|
return subprocess.CompletedProcess(command, 9, stdout="", stderr="database unavailable")
|
|
|
|
query = "What is live in the knowledge base?"
|
|
snapshot = module._load_database_snapshot(query, hermes_home=home, runner=fake_runner)
|
|
assert 'status="unavailable"' in snapshot["context"]
|
|
assert "Do not infer current schema" in snapshot["context"]
|
|
assert "database unavailable" not in snapshot["context"]
|
|
assert snapshot["requires_database_truth"] is True
|
|
assert snapshot["compiled_response"] == module.DATABASE_UNAVAILABLE_RESPONSE
|
|
|
|
module._store_snapshot("session-failed", snapshot)
|
|
assert module._post_llm_call(
|
|
session_id="session-failed", user_message=query, assistant_response="The KB is fully current."
|
|
) == {"assistant_response": module.DATABASE_UNAVAILABLE_RESPONSE}
|
|
|
|
for database_query in (
|
|
"Has Helmer's Seven Powers framework landed as live knowledge?",
|
|
"Does editing the SOUL file alter the source of truth for identity?",
|
|
"I hand Leo a document. What can it absorb and stage before approval?",
|
|
):
|
|
failed = module._load_database_snapshot(database_query, hermes_home=home, runner=fake_runner)
|
|
assert failed["requires_database_truth"] is True
|
|
module._store_snapshot("session-failed-expanded", failed)
|
|
assert module._post_llm_call(
|
|
session_id="session-failed-expanded",
|
|
user_message=database_query,
|
|
assistant_response="It is already live and canonical.",
|
|
) == {"assistant_response": module.DATABASE_UNAVAILABLE_RESPONSE}
|
|
|
|
|
|
def test_database_truth_fallback_covers_every_direct_readback_question() -> None:
|
|
module = load_plugin()
|
|
direct_questions = (
|
|
"Did we actually update the knowledge base, or is it still just proposals?",
|
|
"Is Helmer's 7 Powers in Leo now?",
|
|
"Did the decision matrix approve this already?",
|
|
"Are the pending proposals stuck because the documents are not pointed at the right source rows?",
|
|
"Can I demo that Leo changes the KB?",
|
|
"If we changed SOUL.md, did we change Leo's canonical identity?",
|
|
"In the current-schema, can approve_claim set superseded_by?",
|
|
"A temporary-profile GatewayRunner posted nothing to Telegram. Is delivery proven?",
|
|
)
|
|
|
|
assert all(module._requires_database_truth(question) for question in direct_questions)
|
|
assert module._requires_database_truth("Who is Helmer?") is False
|
|
assert module._requires_database_truth("How are you?") is False
|
|
|
|
|
|
def test_plugin_enforces_every_database_backed_oos_compiler() -> None:
|
|
module = load_plugin()
|
|
expected_ids = {
|
|
"runtime_persistence",
|
|
"shared_claims_agent_positions",
|
|
"forecast_resolution_schema",
|
|
"telegram_delivery_proof",
|
|
"claim_supersession_schema",
|
|
}
|
|
assert expected_ids <= module.COMPILED_CONTRACT_IDS
|
|
|
|
for index, contract_id in enumerate(sorted(expected_ids)):
|
|
query = f"database-backed query {index}"
|
|
query_sha256 = module.hashlib.sha256(query.encode("utf-8")).hexdigest()
|
|
compiled = f"Database-composed response for {contract_id}."
|
|
contracts = [{"id": "reply_budget", "hard_max_words": 200}, {"id": contract_id}]
|
|
module._store_snapshot(
|
|
f"compiled-{index}",
|
|
{
|
|
"status": "ok",
|
|
"query_sha256": query_sha256,
|
|
"contracts": contracts,
|
|
"compiled_response": compiled,
|
|
"requires_database_truth": True,
|
|
"context": "fixture",
|
|
},
|
|
)
|
|
|
|
assert module._post_llm_call(
|
|
session_id=f"compiled-{index}", user_message=query, assistant_response="Unconstrained draft."
|
|
) == {"assistant_response": compiled}
|
|
|
|
|
|
def test_plugin_fails_closed_on_missing_snapshot_or_invalid_compiler() -> None:
|
|
module = load_plugin()
|
|
query = "Did an approved proposal change the knowledge base?"
|
|
|
|
assert module._post_llm_call(session_id="missing", user_message=query, assistant_response="Yes, it changed.") == {
|
|
"assistant_response": module.DATABASE_UNAVAILABLE_RESPONSE
|
|
}
|
|
assert (
|
|
module._post_llm_call(session_id="missing-general", user_message="How are you?", assistant_response="Fine.")
|
|
is None
|
|
)
|
|
|
|
database_status = {
|
|
"high_signal_rows": {
|
|
"claims": 1,
|
|
"sources": 2,
|
|
"claim_edges": 3,
|
|
"claim_evidence": 4,
|
|
"kb_proposals": 5,
|
|
},
|
|
"proposal_status_counts": {"applied": 1, "approved": 1, "pending_review": 2, "canceled": 1},
|
|
}
|
|
contracts = [
|
|
{"id": "reply_budget", "hard_max_words": 200},
|
|
{"id": "proposal_state_readback", "database_status": database_status},
|
|
]
|
|
query_sha256 = module.hashlib.sha256(query.encode("utf-8")).hexdigest()
|
|
module._store_snapshot(
|
|
"invalid-compiler",
|
|
{
|
|
"status": "ok",
|
|
"query_sha256": query_sha256,
|
|
"contracts": contracts,
|
|
"compiled_response": "stale",
|
|
"requires_database_truth": True,
|
|
"context": "fixture",
|
|
},
|
|
)
|
|
assert module._post_llm_call(
|
|
session_id="invalid-compiler", user_message=query, assistant_response="Yes, it changed."
|
|
) == {"assistant_response": module.DATABASE_UNAVAILABLE_RESPONSE}
|
|
|
|
|
|
def test_handler_harness_retains_database_context_proof_fields() -> None:
|
|
source = (ROOT / "scripts" / "run_leo_direct_claim_handler_suite.py").read_text(encoding="utf-8")
|
|
assert "LEO_DB_CONTEXT_TRACE_PATH" in source
|
|
assert '"database_context_trace"' in source
|
|
assert '"database_context_injection_count"' in source
|
|
assert '"database_context_all_ok"' in source
|
|
assert '"database_response_validation_count"' in source
|
|
assert '"database_response_validation_all_ok"' in source
|
|
assert '"database_response_transform_count"' in source
|