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

from scripts.working_leo_open_ended_benchmark import (
    M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS,
    score_reply,
)

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()
    vps_text = (BRIDGE_DIR / "kb_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 "status" in vps_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",
                "proposal_type": "attach_evidence",
                "payload": {},
            }
        ]

    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 result["proposals"][0]["readiness"] == {
        "review_state": "approved_needs_apply_payload",
        "has_apply_payload": False,
        "worker_supported_type": True,
        "worker_contract_applyable": False,
        "production_worker_enabled": None,
        "guidance": (
            "This is proposal-contract readiness only. It does not prove the production apply worker is enabled, "
            "the payload passes strict validation, or apply is authorized."
        ),
    }
    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
    assert "approved_needs_apply_payload" in output
    assert "worker_contract_applyable: `False`" in output


def test_vps_bridge_proposal_readiness_separates_contract_from_live_worker_state() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")

    legacy_approved = module.classify_proposal_readiness(
        {"status": "approved", "proposal_type": "approve_claim", "payload": {}}
    )
    strict_approved = module.classify_proposal_readiness(
        {
            "status": "approved",
            "proposal_type": "approve_claim",
            "payload": {"apply_payload": {"contract_version": 2}},
        }
    )
    pending_strict = module.classify_proposal_readiness(
        {
            "status": "pending_review",
            "proposal_type": "approve_claim",
            "payload": {"apply_payload": {"contract_version": 2}},
        }
    )
    unsupported_approved = module.classify_proposal_readiness(
        {"status": "approved", "proposal_type": "change_policy", "payload": {}}
    )

    assert legacy_approved["review_state"] == "approved_needs_apply_payload"
    assert legacy_approved["worker_contract_applyable"] is False
    assert strict_approved["review_state"] == "approved_contract_present"
    assert strict_approved["worker_contract_applyable"] is True
    assert pending_strict["review_state"] == "needs_human_review"
    assert pending_strict["worker_contract_applyable"] is False
    assert unsupported_approved["review_state"] == "unsupported_by_apply_worker_contract"
    assert unsupported_approved["worker_contract_applyable"] is False
    assert strict_approved["production_worker_enabled"] is None
    assert "does not prove the production apply worker is enabled" in strict_approved["guidance"]


def test_vps_bridge_operational_contracts_are_query_specific_and_schema_exact() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")

    reply_budget = module.operational_contracts("How should Leo answer?")[0]
    assert reply_budget["target_words"] == 150
    assert reply_budget["hard_max_words"] == 200
    assert "omission is better" in reply_budget["shape"]

    source_contracts = {
        item["id"]: item
        for item in module.operational_contracts("I hand Leo a document. What can it absorb and stage before approval?")
    }
    source = source_contracts["source_intake"]
    assert source["required_lead"] == "Local compiler proven; live VPS/chat intake not shipped."
    assert source["capability_tier"] == "build-local compiler only"
    assert source["current_public_sources_columns"] == list(module.CURRENT_PUBLIC_SCHEMA["sources"])
    assert "author/title/publisher/publication date are current public.sources columns" in source["must_not_claim"]

    mixed_contracts = {
        item["id"]: item
        for item in module.operational_contracts(
            "Compose a packet with a factual observation, strategic framework, governance rule, and old belief."
        )
    }
    mixed = mixed_contracts["mixed_packet_composition"]
    assert mixed["map"]["reusable_framework"] == "reasoning_tools"
    assert mixed["approve_claim_supported"] == ["claims", "sources", "evidence", "edges", "reasoning_tools"]
    assert "belief updates" in mixed["keep_staged_until_separate_capability"]
    assert "reasoning_tools has no scope field" in mixed["schema_guards"]
    assert mixed["current_columns"]["reasoning_tools"] == list(module.CURRENT_PUBLIC_SCHEMA["reasoning_tools"])
    assert mixed["current_columns"]["behavioral_rules"] == list(module.CURRENT_PUBLIC_SCHEMA["behavioral_rules"])

    runtime_ids = {
        item["id"]
        for item in module.operational_contracts(
            "Do unchanged database totals after restart prove answer behavior and erase the prior session?"
        )
    }
    assert {"reply_budget", "runtime_persistence"} <= runtime_ids


def test_vps_bridge_broad_db_questions_select_query_specific_live_contracts() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    cases = {
        "What has actually changed in the KB rather than sitting in proposed state?": "proposal_state_readback",
        "What changed in the KB?": "proposal_state_readback",
        "What has changed in the knowledge base?": "proposal_state_readback",
        "Has Helmer's Seven Powers framework landed as live knowledge?": "named_packet_readback",
        "Was that accepted through the decision-matrix?": "decision_matrix_readback",
        "Is the proposal backlog stuck on missing document-to-source links?": "source_link_audit",
        "Can we demo a real database write from Leo?": "demo_capability_readback",
        "Does editing the SOUL file alter the source of truth for identity?": "identity_canonicality_readback",
    }

    for query, expected_id in cases.items():
        contract_ids = {item["id"] for item in module.operational_contracts(query)}
        assert expected_id in contract_ids, query


def test_vps_bridge_direct_intents_reject_unrelated_and_resolve_overlaps() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    direct_ids = set(module.DIRECT_READBACK_CONTRACTS)

    for query in (
        "What changed in the KB scorer?",
        "Who is Helmer?",
        "Can I demo the database dashboard?",
        "Is the KB live search broken?",
    ):
        contract_ids = {item["id"] for item in module.operational_contracts(query)}
        assert not contract_ids & direct_ids, query

    overlap_ids = {
        item["id"]
        for item in module.operational_contracts(
            "Did the decision matrix reject the pending proposal because its document pointer is missing?"
        )
    }
    assert overlap_ids & direct_ids == {"decision_matrix_readback"}
    assert "source_intake" not in overlap_ids

    lifecycle_ids = {
        item["id"]
        for item in module.operational_contracts("What changed in the KB after the benchmark proposal was approved?")
    }
    assert lifecycle_ids & direct_ids == {"proposal_state_readback"}

    substring_ids = {
        item["id"]
        for item in module.operational_contracts("What changed in the KB after the capital allocation update?")
    }
    assert substring_ids & direct_ids == {"proposal_state_readback"}


def _direct_contract_fixtures() -> tuple[dict, dict, dict, dict]:
    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},
    }
    approved = {
        "id": "a64df080-8502-42e2-98f4-9bbdecb8da73",
        "proposal_type": "approve_claim",
        "status": "approved",
        "source_ref": "telegram:helmer-7-powers",
        "created_at": "2026-06-29T00:00:00+00:00",
        "reviewed_at": "2026-06-29T00:10:00+00:00",
        "applied_at": None,
        "readiness": {"review_state": "approved_needs_apply_payload"},
    }
    matrix = {
        "decision_matrix_tables": {
            f"{schema}.{table}": False
            for schema in ("public", "kb_stage")
            for table in ("matrix_voters", "proposal_votes", "proposal_decisions")
        }
    }
    source_audit = {
        "pending_or_approved": 17,
        "with_source_ref": 16,
        "without_source_ref": 1,
        "exact_public_source_matches": 2,
        "samples": [],
    }
    return database_status, approved, matrix, source_audit


def test_vps_bridge_enriches_and_compiles_all_direct_questions_from_live_rows(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    database_status, approved, matrix, source_audit = _direct_contract_fixtures()
    applied = {
        **approved,
        "id": "11111111-1111-4111-8111-111111111111",
        "status": "applied",
        "applied_at": "2026-07-01",
    }

    monkeypatch.setattr(
        module,
        "load_direct_readback_snapshot",
        lambda _args: {
            "snapshot_id": "100:100:",
            "statement_timestamp": "2026-07-13T12:00:00+00:00",
            "database_status": database_status,
            "proposals": [approved, applied],
            "named_packet_proposals": [approved],
            "named_packet_canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []},
            "decision_matrix_status": matrix,
            "source_link_audit": source_audit,
        },
    )

    for prompt in M3TAVERSAL_DIRECT_CLAIM_FOLLOWUP_SCENARIOS:
        contracts = module.context_operational_contracts(SimpleNamespace(), prompt["message"])
        response = module.compile_operational_response(contracts)

        assert response is not None, prompt["id"]
        assert len(re.findall(r"\b\w+(?:[-']\w+)*\b", response)) <= 200
        score = score_reply(prompt, response)
        assert score["pass"] is True, score


def test_vps_bridge_direct_readback_uses_one_statement_snapshot(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    database_status, approved, matrix, source_audit = _direct_contract_fixtures()
    captured_sql: list[str] = []
    expected = {
        "snapshot_id": "200:200:",
        "statement_timestamp": "2026-07-13T12:00:00+00:00",
        "database_status": database_status,
        "proposals": [approved],
        "named_packet_proposals": [approved],
        "named_packet_canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []},
        "decision_matrix_status": matrix,
        "source_link_audit": source_audit,
    }

    def fake_psql_json(_args, sql):
        captured_sql.append(sql)
        return [expected]

    monkeypatch.setattr(module, "psql_json", fake_psql_json)
    assert module.load_direct_readback_snapshot(SimpleNamespace()) == expected
    assert len(captured_sql) == 1
    sql = captured_sql[0]
    assert "txid_current_snapshot()" in sql
    assert "named_packet_proposals" in sql
    assert "search_text like '%helmer%' and search_text like '%power%'" in sql
    assert "decision_matrix_status" in sql
    assert "source_link_audit" in sql


def test_vps_bridge_named_packet_compiler_uses_observed_applied_state() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    database_status, approved, _matrix, _source_audit = _direct_contract_fixtures()
    applied = {
        **approved,
        "status": "applied",
        "applied_at": "2026-07-01T00:00:00+00:00",
        "readiness": {"review_state": "applied"},
    }
    contracts = module.operational_contracts("Is Helmer's 7 Powers in Leo now?")
    named = next(item for item in contracts if item["id"] == "named_packet_readback")
    named.update(
        {
            "database_status": database_status,
            "matching_proposals": [applied],
            "canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []},
        }
    )

    response = module.compile_operational_response(contracts)
    assert response is not None
    assert "status: applied" in response
    assert "applied_at: 2026-07-01" in response
    assert "applied_at is empty" not in response
    assert "does not prove which canonical rows were written" in response


def test_vps_bridge_matrix_compiler_handles_all_tables_present_without_overclaim() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    database_status, _approved, matrix, _source_audit = _direct_contract_fixtures()
    matrix["decision_matrix_tables"] = {name: True for name in matrix["decision_matrix_tables"]}
    contracts = module.operational_contracts("Did the decision matrix approve this already?")
    contract = next(item for item in contracts if item["id"] == "decision_matrix_readback")
    contract.update({"database_status": database_status, "live_matrix_status": matrix})

    response = module.compile_operational_response(contracts)

    assert response is not None
    assert "all required matrix tables are present" in response
    assert "proposal-specific vote or decision rows" in response
    assert "matrix approval is not proven" in response
    assert "0 matrix tables are absent ()" not in response


def test_vps_bridge_matrix_compiler_treats_empty_or_partial_table_state_as_unobserved() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    database_status, _approved, _matrix, _source_audit = _direct_contract_fixtures()

    for table_states, unobserved_count in (({}, 6), ({"public.matrix_voters": True}, 5)):
        contracts = module.operational_contracts("Did the decision matrix approve this already?")
        contract = next(item for item in contracts if item["id"] == "decision_matrix_readback")
        contract.update(
            {
                "database_status": database_status,
                "live_matrix_status": {"decision_matrix_tables": table_states},
            }
        )

        response = module.compile_operational_response(contracts)

        assert response is not None
        assert f"{unobserved_count} required tables were not observed" in response
        assert "presence or absence cannot be inferred" in response
        assert "matrix tables are absent" not in response
        assert "all required matrix tables are present" not in response


def test_vps_bridge_source_link_audit_checks_exact_staging_to_canonical_links(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    _database_status, _approved, _matrix, source_audit = _direct_contract_fixtures()
    captured_sql: list[str] = []

    def fake_psql_json(_args, sql):
        captured_sql.append(sql)
        return [source_audit]

    monkeypatch.setattr(module, "psql_json", fake_psql_json)
    assert module.load_source_link_audit(SimpleNamespace()) == source_audit
    sql = captured_sql[0]
    assert "p.source_ref" in sql
    assert "s.id::text = p.source_ref" in sql
    assert "s.url = p.source_ref" in sql
    assert "s.storage_path = p.source_ref" in sql
    assert "p.status in ('pending_review', 'approved')" in sql


def test_vps_bridge_compiles_safe_mixed_packet_response_from_contracts() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    contracts = module.operational_contracts(
        "Compose a packet with a factual observation, strategic framework, disputed interpretation, governance "
        "rule, and old belief."
    )

    response = module.compile_operational_response(contracts)

    assert response is not None
    assert len(re.findall(r"\b\w+(?:[-']\w+)*\b", response)) <= 200
    assert "approve_claim applies only claims, sources, evidence, edges, and reasoning_tools" in response
    assert "approve_claim supports neither behavioral_rules nor governance_gates" in response
    assert "belief updates, and existing-row updates remain staged" in response
    assert "proposal-level applied_at" in response


def test_vps_bridge_context_markdown_surfaces_runtime_contracts(capsys) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    data = {
        "query": "absorb this document",
        "operational_contracts": module.operational_contracts("absorb this document"),
        "context_rows": [],
        "claims": [],
    }

    module.print_claim_bundle(data)
    output = capsys.readouterr().out

    assert "## Current Runtime Contracts" in output
    assert '"id": "source_intake"' in output
    assert "live VPS/chat intake not shipped" in output


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

    def fake_psql_json(_args, sql):
        captured_sql.append(sql)
        if "information_schema.columns" in sql:
            return [{"table_name": "sources", "columns": ["id", "source_type", "url", "hash", "new_live_column"]}]
        return [{"table_name": "sources", "constraints": {"sources_pkey": "PRIMARY KEY (id)"}}]

    monkeypatch.setattr(module, "psql_json", fake_psql_json)
    contracts = {
        item["id"]: item
        for item in module.context_operational_contracts(
            SimpleNamespace(), "Absorb this document into a staged packet."
        )
    }

    assert contracts["source_intake"]["current_public_sources_columns"] == [
        "id",
        "source_type",
        "url",
        "hash",
        "new_live_column",
    ]
    assert len(captured_sql) == 2
    assert "information_schema.columns" in captured_sql[0]
    assert "table_name = any" in captured_sql[0]
    assert "pg_constraint" in captured_sql[1]


def test_vps_bridge_mixed_contract_reads_live_postgres_constraints(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")

    def fake_psql_json(_args, sql):
        if "information_schema.columns" in sql:
            return [{"table_name": "claims", "columns": ["id", "type", "text", "status"]}]
        return [
            {
                "table_name": "claims",
                "constraints": {
                    "claims_type_check": "CHECK ((type = ANY (ARRAY['empirical', 'structural'])))",
                    "claims_status_check": "CHECK ((status = ANY (ARRAY['open', 'retired'])))",
                },
            }
        ]

    monkeypatch.setattr(module, "psql_json", fake_psql_json)

    contracts = {
        item["id"]: item
        for item in module.context_operational_contracts(
            SimpleNamespace(),
            "Compose a packet with factual observations, a strategic framework, and a governance rule.",
        )
    }
    mixed = contracts["mixed_packet_composition"]
    assert "'empirical'" in mixed["current_constraints"]["claims"]["claims_type_check"]
    assert "'open'" in mixed["current_constraints"]["claims"]["claims_status_check"]
    assert "binding operating/governance rules" in mixed["table_semantics"]["behavioral_rules"]
    assert "evaluative pass/fail gates" in mixed["table_semantics"]["governance_gates"]
    assert "supports neither behavioral_rules nor governance_gates" in mixed["unsupported_apply_boundary"]
    assert "proposal-level applied_at" in mixed["review_apply_sequence"][-1]


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_vps_bridge_status_reads_all_direct_claim_counts(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    captured_sql: list[str] = []
    expected = {
        "artifact": "teleo_kb_status",
        "backend": "teleo|postgres",
        "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},
    }

    def fake_psql_json(_args, sql):
        captured_sql.append(sql)
        return [expected]

    monkeypatch.setattr(module, "psql_json", fake_psql_json)

    assert module.status(SimpleNamespace()) == expected
    sql = captured_sql[0]
    for table in (
        "public.claims",
        "public.sources",
        "public.claim_edges",
        "public.claim_evidence",
        "kb_stage.kb_proposals",
    ):
        assert table in sql
    assert not re.search(r"\b(?:insert|update|delete|alter|drop|create)\b", sql, re.I)


def test_vps_bridge_status_prints_copyable_db_readback(capsys) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    module.print_status(
        {
            "artifact": "teleo_kb_status",
            "backend": "teleo|postgres",
            "high_signal_rows": {
                "claims": 1837,
                "sources": 4145,
                "claim_edges": 4916,
                "claim_evidence": 4670,
                "kb_proposals": 26,
            },
            "proposal_status_counts": {"applied": 2},
        }
    )

    assert (
        "DB readback: claims: `1837`; sources: `4145`; claim_edges: `4916`; "
        "claim_evidence: `4670`; kb_proposals: `26`." in capsys.readouterr().out
    )


def test_cloudsql_status_prints_copyable_canonical_db_readback(capsys) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    module.emit_markdown(
        {
            "artifact": "teleo_cloudsql_memory_status",
            "backend": "cloudsql:test",
            "db_identity": "audit|reader",
            "canonical": {
                "db_identity": "teleo|reader",
                "schema_tables": {"public": 33, "kb_stage": 6},
                "high_signal_rows": {
                    "claims": 1837,
                    "sources": 4145,
                    "claim_edges": 4916,
                    "claim_evidence": 4670,
                    "kb_proposals": 26,
                },
            },
            "extensions": [],
            "teleo_restore_tables": 8,
            "total_rows": 52164,
            "high_signal_rows": {},
        }
    )

    assert (
        "DB readback: claims: `1837`; sources: `4145`; claim_edges: `4916`; "
        "claim_evidence: `4670`; kb_proposals: `26`." in capsys.readouterr().out
    )


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
    )