teleo-infrastructure/tests/test_hermes_leoclean_db_context_plugin.py

"""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. The local compiler is proven; live VPS/chat "
        "intake 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_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
    }


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")

    context = module.build_database_context("What is live?", hermes_home=home, runner=fake_runner)
    assert 'status="unavailable"' in context
    assert "Do not infer current schema" in context
    assert "database unavailable" not in context


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