teleo-infrastructure/tests/test_hermes_leoclean_kb_bridge_source.py

"""Source-control checks for the Hermes leoclean KB bridge files."""

from __future__ import annotations

import ast
import contextlib
import importlib.util
import io
import re
import subprocess
from pathlib import Path
from types import SimpleNamespace

ROOT = Path(__file__).resolve().parents[1]
BRIDGE_DIR = ROOT / "hermes-agent" / "leoclean-bin"


def test_leoclean_bridge_files_are_present_and_parseable() -> None:
    cloudsql_tool = BRIDGE_DIR / "cloudsql_memory_tool.py"
    vps_tool = BRIDGE_DIR / "kb_tool.py"
    wrapper = BRIDGE_DIR / "teleo-kb"

    assert cloudsql_tool.exists()
    assert vps_tool.exists()
    assert wrapper.exists()

    ast.parse(cloudsql_tool.read_text())
    ast.parse(vps_tool.read_text())
    subprocess.run(["bash", "-n", str(wrapper)], check=True)


def test_cloudsql_wrapper_supports_expected_review_gated_commands() -> None:
    wrapper_text = (BRIDGE_DIR / "teleo-kb").read_text()
    cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text()

    for command in [
        "status",
        "search",
        "context",
        "show",
        "evidence",
        "edges",
        "propose-core-change",
        "propose-edge",
        "list-proposals",
        "search-proposals",
        "show-proposal",
        "decision-matrix-status",
    ]:
        assert command in wrapper_text
        assert command in cloudsql_text

    assert "kb_stage.kb_proposals" in cloudsql_text
    assert "canonical apply" in cloudsql_text.lower()


def test_bridge_edge_proposals_stage_strict_apply_payloads() -> None:
    cloudsql_text = (BRIDGE_DIR / "cloudsql_memory_tool.py").read_text()
    vps_text = (BRIDGE_DIR / "kb_tool.py").read_text()

    for text in (cloudsql_text, vps_text):
        assert "'proposal_type', proposal_type" in text or "select 'add_edge'" in text
        assert "'apply_payload', jsonb_build_object(" in text
        assert "'from_claim', from_row.id::text" in text
        assert "'to_claim', to_row.id::text" in text
        assert "'edge_type', p.edge_type::text" in text


def test_bridge_source_does_not_commit_raw_secret_values() -> None:
    combined = "\n".join(path.read_text(errors="replace") for path in BRIDGE_DIR.iterdir() if path.is_file())
    forbidden_patterns = [
        r"bot\d+:[A-Za-z0-9_-]{20,}",
        r"postgres://[^\\s]+:[^\\s]+@",
        r"-----BEGIN [A-Z ]*PRIVATE KEY-----",
        r"sk-[A-Za-z0-9]{20,}",
        r"xox[baprs]-[A-Za-z0-9-]{20,}",
        r"AIza[0-9A-Za-z_-]{20,}",
    ]
    for pattern in forbidden_patterns:
        assert not re.search(pattern, combined), pattern


def _load_module(path: Path):
    spec = importlib.util.spec_from_file_location(path.stem, path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def test_vps_bridge_search_proposals_finds_approved_rows_by_payload(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    captured_sql: list[str] = []

    def fake_psql_json(_args, sql):
        captured_sql.append(sql)
        return [{"id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "status": "approved"}]

    monkeypatch.setattr(module, "psql_json", fake_psql_json)
    args = SimpleNamespace(query="Helmer 7 Powers", status="all", limit=20)
    result = module.search_proposals(args)

    assert result["proposals"][0]["status"] == "approved"
    assert {"helmer", "powers"} <= set(result["terms"])
    sql = captured_sql[0]
    assert "payload::text ilike any" in sql
    assert "coalesce(rationale, '') ilike any" in sql
    assert "status =" not in sql


def test_vps_bridge_proposal_list_prints_rationale_for_non_edge_rows(capsys) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    module.print_proposal_list(
        [
            {
                "id": "a64df080-8502-42e2-98f4-9bbdecb8da73",
                "proposal_type": "attach_evidence",
                "status": "approved",
                "proposed_by_handle": "leo",
                "channel": "telegram",
                "created_at": "2026-07-09 00:00:00+00",
                "reviewed_at": "2026-07-09 00:10:00+00",
                "applied_at": None,
                "payload": {"title": "Helmer 7 Powers"},
                "rationale": "Revised Helmer packet should remain visible as approved but unapplied.",
            }
        ]
    )

    output = capsys.readouterr().out

    assert "attach_evidence" in output
    assert "approved" in output
    assert "applied_at: `-`" in output
    assert "Revised Helmer packet" in output


def test_vps_bridge_decision_matrix_status_checks_schema_tables(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    captured_sql: list[str] = []

    def fake_psql_json(_args, sql):
        captured_sql.append(sql)
        return [
            {
                "artifact": "teleo_kb_decision_matrix_status",
                "all_required_tables_present": False,
                "any_decision_matrix_table_present": False,
                "decision_matrix_tables": {},
                "proposal_status_counts": {},
                "guidance": "Decision-matrix approval schema is absent or incomplete.",
            }
        ]

    monkeypatch.setattr(module, "psql_json", fake_psql_json)
    result = module.decision_matrix_status(SimpleNamespace())

    assert result["all_required_tables_present"] is False
    sql = captured_sql[0]
    for table in (
        "public.matrix_voters",
        "public.proposal_votes",
        "public.proposal_decisions",
        "kb_stage.matrix_voters",
        "kb_stage.proposal_votes",
        "kb_stage.proposal_decisions",
    ):
        assert table in sql


def test_cloudsql_bridge_matches_direct_claim_readback_commands(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    captured_sql: list[str] = []

    def fake_psql_json_lines(_args, sql, db=None):
        captured_sql.append(sql)
        if "matrix_voters" in sql:
            return [
                {
                    "artifact": "teleo_cloudsql_kb_decision_matrix_status",
                    "all_required_tables_present": False,
                    "any_decision_matrix_table_present": False,
                    "decision_matrix_tables": {},
                    "proposal_status_counts": {},
                    "guidance": "Decision-matrix approval schema is absent or incomplete.",
                }
            ]
        return [{"id": "a64df080-8502-42e2-98f4-9bbdecb8da73", "status": "approved"}]

    monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines)
    args = SimpleNamespace(query="Helmer 7 Powers", status="all", limit=20, canonical_db="teleo")
    proposal_result = module.search_proposals(args)
    matrix_result = module.decision_matrix_status(SimpleNamespace(canonical_db="teleo"))

    assert proposal_result["artifact"] == "teleo_cloudsql_kb_proposal_search"
    assert proposal_result["proposals"][0]["status"] == "approved"
    assert matrix_result["artifact"] == "teleo_cloudsql_kb_decision_matrix_status"
    assert any("payload::text ilike any" in sql for sql in captured_sql)
    assert any("public.proposal_decisions" in sql for sql in captured_sql)


def test_kb_bridges_emit_public_claim_links_for_telegram_rendering() -> None:
    claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1"
    expected = f"https://leo.livingip.xyz/kb/claims/{claim_id}"

    for filename in ("kb_tool.py", "cloudsql_memory_tool.py"):
        module = _load_module(BRIDGE_DIR / filename)

        assert module.claim_url(claim_id) == expected
        assert module.markdown_claim_link(claim_id, "d3fb892b") == (
            f"[`d3fb892b`]({expected})"
        )
        assert module.markdown_claim_link(claim_id, "claim `with ticks`") == (
            f"[`claim 'with ticks'`]({expected})"
        )


def test_vps_bridge_markdown_links_claim_text_and_edges() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    claim_id = "d3fb892b-3c5a-4700-9512-55e5c680eec1"
    connected_id = "9d5281fe-7ee4-4fe1-b1bf-c54c4d7fb6a5"
    data = {
        "query": "strategy kernel",
        "context_rows": [],
        "claims": [
            {
                "id": claim_id,
                "text": "Claims should be easy to scan in Telegram.",
                "type": "belief",
                "confidence": 0.8,
                "tags": ["telegram"],
                "score": 4,
                "evidence_count": 2,
                "edge_count": 1,
                "evidence": [],
                "edges": [
                    {
                        "direction": "outgoing",
                        "edge_type": "supports",
                        "connected_id": connected_id,
                        "connected_text": "Full claim pages expose body, evidence, and graph edges.",
                    }
                ],
            }
        ],
    }

    buffer = io.StringIO()
    with contextlib.redirect_stdout(buffer):
        module.print_claim_bundle(data)
    markdown = buffer.getvalue()

    assert (
        f"[`Claims should be easy to scan in Telegram.`](https://leo.livingip.xyz/kb/claims/{claim_id})"
        in markdown
    )
    assert (
        f"- open full claim/body/edges: [`claim page`](https://leo.livingip.xyz/kb/claims/{claim_id})"
        in markdown
    )
    assert (
        f"[`Full claim pages expose body, evidence, and graph edges.`](https://leo.livingip.xyz/kb/claims/{connected_id})"
        in markdown
    )