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

from __future__ import annotations

import ast
import base64
import contextlib
import importlib.util
import io
import json
import os
import re
import shutil
import subprocess
from pathlib import Path
from types import SimpleNamespace

import pytest

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"
DB_CONTEXT_PLUGIN = ROOT / "hermes-agent" / "leoclean-plugins" / "vps" / "leo-db-context" / "__init__.py"
GCP_RUNTIME_ROLE_SQL = ROOT / "ops" / "gcp_leoclean_runtime_role.sql"
GCP_RUNTIME_WRAPPER = BRIDGE_DIR / "teleo-kb-gcp"
GCP_RUNTIME_PROVISIONER = ROOT / "ops" / "provision_gcp_leoclean_runtime_role.sh"


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)


@pytest.mark.parametrize(
    "argument",
    [
        "--host",
        "--host=127.0.0.1",
        "--port",
        "--port=6543",
        "--db",
        "--db=postgres",
        "--canonical-db",
        "--canonical-db=postgres",
        "--user",
        "--user=postgres",
        "--password-secret",
        "--password-secret=administrator-secret",
        "--project",
        "--project=other-project",
        "--credential-mode",
        "--credential-mode=clone-readonly",
    ],
)
def test_deployed_gcp_wrapper_executable_rejects_every_identity_override(argument: str) -> None:
    injected = "sensitive-override-value"
    completed = subprocess.run(
        [str(GCP_RUNTIME_WRAPPER), argument, injected],
        capture_output=True,
        check=False,
        text=True,
    )

    assert completed.returncode == 64
    assert "identity and endpoint overrides are forbidden" in completed.stderr
    assert injected not in completed.stderr
    assert completed.stdout == ""


def test_gcp_wrapper_and_password_provisioner_are_executable_and_syntax_valid() -> None:
    assert os.access(GCP_RUNTIME_WRAPPER, os.X_OK)
    assert os.access(GCP_RUNTIME_PROVISIONER, os.X_OK)
    subprocess.run(["bash", "-n", str(GCP_RUNTIME_WRAPPER)], check=True)
    subprocess.run(["bash", "-n", str(GCP_RUNTIME_PROVISIONER)], check=True)

    wrapper = GCP_RUNTIME_WRAPPER.read_text(encoding="utf-8")
    assert "exec /usr/bin/env -i" in wrapper
    assert '/usr/bin/python3 -I "$CLOUDSQL_TOOL" "$@"' in wrapper
    assert "TELEO_CLOUDSQL_USER=leoclean_kb_runtime" in wrapper
    assert "PGSSLMODE=verify-ca" in wrapper
    assert "TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK=0" in wrapper

    provisioner = GCP_RUNTIME_PROVISIONER.read_text(encoding="utf-8")
    capture_runtime = "runtime_password=${TELEO_LEOCLEAN_DB_PASSWORD-}"
    capture_admin = "administrator_password=${PGPASSWORD-}"
    clear_ambient = "unset TELEO_LEOCLEAN_DB_PASSWORD PGPASSWORD"
    first_external_child = 'SCRIPT_DIR="$(cd -- "$script_directory" && pwd -P)"'
    assert provisioner.index(capture_runtime) < provisioner.index(first_external_child)
    assert provisioner.index(capture_admin) < provisioner.index(first_external_child)
    assert provisioner.index(clear_ambient) < provisioner.index(first_external_child)
    assert '"$SETSID_BIN" -- "$PSQL_BIN"' in provisioner
    assert 'kill -TERM -- "-$provision_pid"' in provisioner
    assert 'wait "$provision_pid"' in provisioner
    assert "SERVER_CA_SHA256=80701e" in provisioner
    assert "--no-password" in provisioner
    assert "PGSSLMODE=verify-ca" in provisioner


def test_vps_bridge_recognizes_natural_mixed_briefing_contract() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    contracts = module.operational_contracts(
        "Compose this briefing from evidence-backed facts, a reasoning framework, a contested position, "
        "an operating rule, and a correction."
    )

    assert "mixed_packet_composition" in {item["id"] for item in contracts}


def test_vps_bridge_global_options_do_not_accept_untraced_abbreviations(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    monkeypatch.setattr(
        module.sys,
        "argv",
        ["kb_tool.py", "--loc", "context", "topic"],
    )

    with pytest.raises(SystemExit):
        module.parse_args()


@pytest.mark.parametrize("command", ("search", "context", "search-proposals"))
def test_vps_bridge_joins_unquoted_multiword_queries(monkeypatch, command: str) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    monkeypatch.setattr(
        module.sys,
        "argv",
        ["kb_tool.py", command, "Atlas", "status", "snapshot", "--limit", "1"],
    )

    args = module.parse_args()

    assert args.query == "Atlas status snapshot"
    assert args.limit == 1


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 "SCRIPT_DIR=${SCRIPT_PATH%/*}" in wrapper_text
    assert 'SCRIPT_DIR="$(cd "$SCRIPT_DIR" && pwd)"' in wrapper_text
    assert 'CLOUDSQL_TOOL="$SCRIPT_DIR/cloudsql_memory_tool.py"' in wrapper_text
    assert wrapper_text.startswith("#!/bin/bash\n")
    assert "exec /usr/bin/python3" in wrapper_text
    assert "command -v psql" in wrapper_text
    assert "gcloud" not in wrapper_text
    assert '"gcloud"' not in cloudsql_text

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


def _install_wrapper_fixture(tmp_path: Path, *, missing: frozenset[str] = frozenset()) -> tuple[Path, dict[str, str]]:
    profile = tmp_path / "profile"
    bin_dir = profile / "bin"
    bin_dir.mkdir(parents=True)
    wrapper = bin_dir / "teleo-kb"
    wrapper.write_bytes((BRIDGE_DIR / "teleo-kb").read_bytes())
    wrapper.chmod(0o700)
    cloudsql = bin_dir / "cloudsql_memory_tool.py"
    if "cloudsql-tool" not in missing:
        cloudsql.write_text(
            "import json, sys\nprint(json.dumps({'backend': 'cloudsql', 'argv': sys.argv[1:]}))\n",
            encoding="utf-8",
        )
        cloudsql.chmod(0o700)
    local = bin_dir / "kb_tool.py"
    local.write_text(
        "import json, sys\nprint(json.dumps({'backend': 'local', 'argv': sys.argv[1:]}))\n",
        encoding="utf-8",
    )
    local.chmod(0o700)
    fake_bin = tmp_path / "fake-bin"
    fake_bin.mkdir()
    for name in ("bash", "python3"):
        target = shutil.which(name)
        assert target
        (fake_bin / name).symlink_to(target)
    for name in ("psql",):
        if name in missing:
            continue
        executable = fake_bin / name
        executable.write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8")
        executable.chmod(0o700)
    env = {
        **os.environ,
        "HERMES_HOME": str(profile),
        "TELEO_KB_MODE": "auto",
        "PATH": str(fake_bin),
    }
    return wrapper, env


def test_cloudsql_wrapper_auto_mode_routes_supported_reads_without_slow_probe(tmp_path: Path) -> None:
    wrapper, env = _install_wrapper_fixture(tmp_path)

    completed = subprocess.run(
        [str(wrapper), "search", "AI sandbagging M&A liability", "--format", "json"],
        text=True,
        capture_output=True,
        check=False,
        env=env,
    )

    assert completed.returncode == 0, completed.stderr
    result = json.loads(completed.stdout)
    assert result == {
        "backend": "cloudsql",
        "argv": ["search", "AI sandbagging M&A liability", "--format", "json"],
    }


def test_cloudsql_wrapper_auto_mode_never_falls_back_after_supported_command_failure(tmp_path: Path) -> None:
    wrapper, env = _install_wrapper_fixture(tmp_path)
    cloudsql = Path(env["HERMES_HOME"]) / "bin" / "cloudsql_memory_tool.py"
    cloudsql.write_text("raise SystemExit(23)\n", encoding="utf-8")

    completed = subprocess.run(
        [str(wrapper), "status", "--format", "json"],
        text=True,
        capture_output=True,
        check=False,
        env=env,
    )

    assert completed.returncode == 23
    assert "local" not in completed.stdout


@pytest.mark.parametrize("missing", ["cloudsql-tool", "psql"])
def test_cloudsql_wrapper_auto_mode_fails_closed_when_prerequisite_is_missing(tmp_path: Path, missing: str) -> None:
    wrapper, env = _install_wrapper_fixture(tmp_path, missing=frozenset({missing}))

    completed = subprocess.run(
        [str(wrapper), "search", "canonical only", "--format", "json"],
        text=True,
        capture_output=True,
        check=False,
        env=env,
    )

    assert completed.returncode == 69
    assert "local" not in completed.stdout
    assert "Cloud SQL" in completed.stderr


def test_cloudsql_wrapper_explicit_mode_rejects_unsupported_commands_without_fallback(tmp_path: Path) -> None:
    wrapper, env = _install_wrapper_fixture(tmp_path)
    env["TELEO_KB_MODE"] = "cloudsql"

    completed = subprocess.run(
        [str(wrapper), "prepare-source", "example.md"],
        text=True,
        capture_output=True,
        check=False,
        env=env,
    )

    assert completed.returncode == 64
    assert "local" not in completed.stdout
    assert "not supported by the Cloud SQL backend" in completed.stderr


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

    assert "kb_stage.stage_leoclean_proposal(" in cloudsql_text
    assert "insert into kb_stage.kb_proposals" not in cloudsql_text.lower()


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 _runtime_connection_args(module, **overrides):
    values = {
        "credential_mode": "runtime",
        "user": module.RUNTIME_DB_USER,
        "password_secret": module.RUNTIME_PASSWORD_SECRET,
        "project": module.RUNTIME_GCP_PROJECT,
        "host": module.RUNTIME_CLOUDSQL_HOST,
        "port": module.RUNTIME_CLOUDSQL_PORT,
        "db": module.RUNTIME_CLOUDSQL_DB,
        "canonical_db": module.RUNTIME_CLOUDSQL_DB,
        "command": "status",
    }
    values.update(overrides)
    return SimpleNamespace(**values)


def _configure_runtime_environment(monkeypatch: pytest.MonkeyPatch) -> None:
    for key in tuple(os.environ):
        normalized_key = key.upper()
        if (
            normalized_key.startswith("PG")
            or normalized_key.startswith("CLOUDSDK_")
            or normalized_key == "GOOGLE_APPLICATION_CREDENTIALS"
            or normalized_key.startswith("LD_")
            or normalized_key.startswith("BASH_FUNC_")
            or normalized_key.startswith("PYTHON")
            or normalized_key
            in {
                "BASH_ENV",
                "BASHOPTS",
                "CDPATH",
                "ENV",
                "GLOBIGNORE",
                "IFS",
                "PS4",
                "SHELLOPTS",
            }
            or normalized_key == "PROXY"
            or normalized_key.endswith("_PROXY")
            or normalized_key
            in {
                "CURL_CA_BUNDLE",
                "PYTHONHTTPSVERIFY",
                "REQUESTS_CA_BUNDLE",
                "SSL_CERT_DIR",
                "SSL_CERT_FILE",
                "SSLKEYLOGFILE",
                "OPENSSL_CONF",
                "OPENSSL_ENGINES",
                "OPENSSL_MODULES",
                "GCONV_PATH",
            }
        ):
            monkeypatch.delenv(key, raising=False)
    monkeypatch.setenv("TELEO_GCP_METADATA_ONLY", "1")
    monkeypatch.setenv("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", "0")


def test_runtime_environment_fixture_scrubs_github_runner_inheritance(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    monkeypatch.setenv("CI", "true")
    monkeypatch.setenv("LD_LIBRARY_PATH", "/opt/hostedtoolcache/Python/3.11.15/x64/lib")
    monkeypatch.setenv("pythonLocation", "/opt/hostedtoolcache/Python/3.11.15/x64")
    monkeypatch.setenv("HTTP_PROXY", "http://runner-proxy.invalid:8080")
    monkeypatch.setenv("BASH_FUNC_runner%%", "() { :; }")
    monkeypatch.setenv("SSL_CERT_FILE", "/tmp/runner-ca.pem")
    monkeypatch.setenv("OPENSSL_CONF", "/tmp/runner-openssl.cnf")
    monkeypatch.setenv("GCONV_PATH", "/tmp/runner-gconv")

    _configure_runtime_environment(monkeypatch)

    assert os.environ["CI"] == "true"
    module.validate_runtime_connection(args)


class _FakeHttpResponse:
    def __init__(self, body: bytes, status: int = 200) -> None:
        self.body = body
        self.status = status

    def __enter__(self):
        return self

    def __exit__(self, *_args) -> None:
        return None

    def getcode(self) -> int:
        return self.status

    def read(self, _limit: int) -> bytes:
        return self.body


def _runtime_http_bodies(module, password: bytes = b"scoped-runtime-password") -> list[bytes]:
    return [
        module.RUNTIME_SERVICE_ACCOUNT.encode("ascii"),
        json.dumps({"access_token": "metadata-access-token", "token_type": "Bearer", "expires_in": 300}).encode(),
        json.dumps({"payload": {"data": base64.b64encode(password).decode("ascii")}}).encode(),
    ]


def test_cloudsql_runtime_requires_the_exact_scoped_identity(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    monkeypatch.delenv("PGPASSWORD", raising=False)

    with pytest.raises(SystemExit, match="requires database user leoclean_kb_runtime"):
        module.validate_runtime_connection(SimpleNamespace(credential_mode="runtime", user=None, password_secret=None))
    with pytest.raises(SystemExit, match="requires database user leoclean_kb_runtime"):
        module.validate_runtime_connection(
            SimpleNamespace(
                credential_mode="runtime",
                user="postgres",
                password_secret=module.RUNTIME_PASSWORD_SECRET,
            )
        )
    with pytest.raises(SystemExit, match="requires database user leoclean_kb_runtime"):
        module.validate_runtime_connection(
            SimpleNamespace(
                credential_mode="runtime",
                user=f" {module.RUNTIME_DB_USER}",
                password_secret=module.RUNTIME_PASSWORD_SECRET,
            )
        )
    with pytest.raises(SystemExit, match="requires scoped password secret"):
        module.validate_runtime_connection(
            SimpleNamespace(
                credential_mode="runtime",
                user=module.RUNTIME_DB_USER,
                password_secret="gcp-teleo-pgvector-standby-postgres-password",
            )
        )
    with pytest.raises(SystemExit, match="requires scoped password secret"):
        module.validate_runtime_connection(
            SimpleNamespace(credential_mode="runtime", user=module.RUNTIME_DB_USER, password_secret=None)
        )


def test_cloudsql_runtime_accepts_only_scoped_credentials_and_rejects_inherited_password(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)

    module.validate_runtime_connection(args)

    monkeypatch.setenv("PGPASSWORD", "")
    with pytest.raises(SystemExit, match="Refusing inherited credential, proxy, TLS, or PostgreSQL environment"):
        module.validate_runtime_connection(args)


@pytest.mark.parametrize(
    ("name", "value"),
    [
        ("GOOGLE_APPLICATION_CREDENTIALS", "/tmp/ambient.json"),
        ("google_application_credentials", "/tmp/ambient-lowercase.json"),
        ("CLOUDSDK_CONFIG", "/tmp/gcloud"),
        ("CLOUDSDK_AUTH_ACCESS_TOKEN", "ambient-token"),
        ("cloudsdk_core_project", "attacker-project"),
        ("cloudsdk_auth_access_token", "ambient-lowercase-token"),
        ("PGHOST", "attacker.invalid"),
        ("PGOPTIONS", "-c search_path=attacker"),
        ("pgservice", "attacker-service"),
        ("ld_preload", "/tmp/attacker.so"),
        ("LD_LIBRARY_PATH", "/tmp/attacker-library"),
        ("Ld_AuDiT", "/tmp/attacker-audit.so"),
        ("BASH_ENV", "/tmp/attacker-bash-env"),
        ("ENV", "/tmp/attacker-sh-env"),
        ("BASH_FUNC_psql%%", "() { :; }"),
        ("PythonHome", "/tmp/attacker-python"),
        ("pythonpath", "/tmp/attacker-modules"),
        ("PYTHONHTTPSVERIFY", "0"),
        ("pythonstartup", "/tmp/attacker-startup.py"),
        ("https_proxy", "http://attacker.invalid:8080"),
        ("HtTp_PrOxY", "http://attacker.invalid:8080"),
        ("ssl_cert_file", "/tmp/attacker.pem"),
        ("SSL_CERT_DIR", "/tmp/attacker-certs"),
        ("REQUESTS_CA_BUNDLE", "/tmp/attacker.pem"),
        ("SSLKEYLOGFILE", "/tmp/attacker-tls-keys"),
    ],
)
def test_cloudsql_runtime_rejects_ambient_auth_and_pg_environment(
    monkeypatch: pytest.MonkeyPatch,
    name: str,
    value: str,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    monkeypatch.setenv(name, value)

    with pytest.raises(SystemExit, match="Refusing inherited credential, proxy, TLS, or PostgreSQL environment"):
        module.validate_runtime_connection(args)


def test_cloudsql_runtime_allows_only_pythonunbuffered_runtime_tuning(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    monkeypatch.setenv("PYTHONUNBUFFERED", "1")

    module.validate_runtime_connection(args)


def test_cloudsql_runtime_requires_explicit_metadata_only_mode(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    monkeypatch.delenv("TELEO_GCP_METADATA_ONLY")

    with pytest.raises(SystemExit, match="TELEO_GCP_METADATA_ONLY=1"):
        module.validate_runtime_connection(args)


@pytest.mark.parametrize(
    ("field", "value", "message"),
    [
        ("project", "other-project", "project=teleo-501523"),
        ("host", "127.0.0.1", "host=10.61.0.3"),
        ("port", "6543", "port=5432"),
        ("db", "teleo_kb", "db=teleo_canonical"),
        ("canonical_db", "teleo_kb", "canonical_db=teleo_canonical"),
    ],
)
def test_cloudsql_runtime_rejects_any_target_drift_before_credentials(
    monkeypatch: pytest.MonkeyPatch,
    field: str,
    value: str,
    message: str,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module, **{field: value})
    monkeypatch.delenv("PGPASSWORD", raising=False)
    monkeypatch.setenv("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", "0")

    with pytest.raises(SystemExit, match=message):
        module.validate_runtime_connection(args)


def test_cloudsql_runtime_rejects_audit_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    monkeypatch.delenv("PGPASSWORD", raising=False)
    monkeypatch.setenv("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", "1")

    with pytest.raises(SystemExit, match="AUDIT_FALLBACK=0"):
        module.validate_runtime_connection(args)


def test_cloudsql_clone_mode_is_read_only_and_bound_to_one_disposable_database(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = SimpleNamespace(
        credential_mode="clone-readonly",
        user="postgres",
        password_secret=None,
        canonical_db="teleo_clone_canary_123",
        db="teleo_clone_canary_123",
        command="status",
    )
    monkeypatch.setenv("PGPASSWORD", "clone-only-password")
    monkeypatch.setenv("PGOPTIONS", module.CLONE_READ_ONLY_PGOPTIONS)

    module.validate_runtime_connection(args)
    assert module.password(args) == "clone-only-password"

    args.command = "propose-core-change"
    with pytest.raises(SystemExit, match="read commands only"):
        module.validate_runtime_connection(args)
    args.command = "status"
    args.canonical_db = "teleo_canonical"
    with pytest.raises(SystemExit, match=r"teleo_clone_\*"):
        module.validate_runtime_connection(args)


def test_cloudsql_clone_mode_never_resolves_a_password_secret_argument(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = SimpleNamespace(
        credential_mode="clone-readonly",
        user="postgres",
        password_secret="gcp-teleo-pgvector-standby-postgres-password",
        canonical_db="teleo_clone_canary_123",
        db="teleo_clone_canary_123",
        command="status",
    )
    monkeypatch.delenv("PGPASSWORD", raising=False)
    monkeypatch.setenv("PGOPTIONS", module.CLONE_READ_ONLY_PGOPTIONS)
    monkeypatch.setattr(
        module.subprocess,
        "run",
        lambda *_args, **_kwargs: pytest.fail("a password-secret argument must never launch a credential helper"),
    )

    with pytest.raises(SystemExit, match="requires an explicit user and inherited test credential"):
        module.password(args)


def test_cloudsql_psql_uses_only_explicit_pg_environment_fields(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    monkeypatch.setenv("HOME", "/tmp/untrusted-home")
    monkeypatch.setenv("LANG", "attacker-locale")
    monkeypatch.setenv("SHELL", "/tmp/untrusted-shell")
    monkeypatch.setenv("UNRELATED_AMBIENT_VALUE", "must-not-reach-psql")
    captured: dict[str, object] = {}

    def fake_run(command, **kwargs):
        captured["command"] = command
        captured["kwargs"] = kwargs
        return SimpleNamespace(returncode=0, stdout="ok\n", stderr="")

    monkeypatch.setattr(module, "runtime_password_from_metadata", lambda: "scoped-runtime-password")
    monkeypatch.setattr(module.subprocess, "run", fake_run)

    assert module.run_psql(args, "select 1;") == "ok\n"
    assert captured["command"] == ["/usr/bin/psql", "-X", "-At", "-q", "-v", "ON_ERROR_STOP=1"]
    kwargs = captured["kwargs"]
    assert isinstance(kwargs, dict)
    env = kwargs["env"]
    assert env == {
        "PATH": "/usr/bin:/bin",
        "PGHOST": "10.61.0.3",
        "PGPORT": "5432",
        "PGDATABASE": "teleo_canonical",
        "PGUSER": module.RUNTIME_DB_USER,
        "PGSSLMODE": "verify-ca",
        "PGSSLROOTCERT": module.RUNTIME_SSL_ROOT_CERT,
        "PGCONNECT_TIMEOUT": "8",
        "PGPASSWORD": "scoped-runtime-password",
    }


def test_cloudsql_runtime_metadata_and_secret_requests_are_exact_and_silent(
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    bodies = _runtime_http_bodies(module)
    requests: list[tuple[str, str, dict[str, str], float]] = []

    def fake_urlopen(request, *, timeout):
        requests.append(
            (
                request.full_url,
                request.get_method(),
                {key.lower(): value for key, value in request.header_items()},
                timeout,
            )
        )
        return _FakeHttpResponse(bodies.pop(0))

    monkeypatch.setattr(module.RUNTIME_HTTP_OPENER, "open", fake_urlopen)

    assert module.password(args) == "scoped-runtime-password"
    assert module.GCE_METADATA_EMAIL_URL == (
        "http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/email"
    )
    assert module.GCE_METADATA_TOKEN_URL == (
        "http://169.254.169.254/computeMetadata/v1/instance/service-accounts/default/token"
    )
    assert requests == [
        (
            module.GCE_METADATA_EMAIL_URL,
            "GET",
            {"metadata-flavor": "Google"},
            module.HTTP_TIMEOUT_SECONDS,
        ),
        (
            module.GCE_METADATA_TOKEN_URL,
            "GET",
            {"metadata-flavor": "Google"},
            module.HTTP_TIMEOUT_SECONDS,
        ),
        (
            module.RUNTIME_SECRET_ACCESS_URL,
            "GET",
            {"accept": "application/json", "authorization": "Bearer metadata-access-token"},
            module.HTTP_TIMEOUT_SECONDS,
        ),
    ]
    assert capsys.readouterr() == ("", "")


def test_cloudsql_runtime_http_opener_disables_proxies_and_redirects() -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    proxy_handlers = [
        handler
        for handler in module.RUNTIME_HTTP_OPENER.handlers
        if isinstance(handler, module.urllib.request.ProxyHandler)
    ]
    redirect_handlers = [
        handler
        for handler in module.RUNTIME_HTTP_OPENER.handlers
        if isinstance(handler, module.urllib.request.HTTPRedirectHandler)
    ]

    # An empty ProxyHandler suppresses build_opener's environment-backed default;
    # urllib omits the empty handler itself from the final chain.
    assert proxy_handlers == []
    assert module.RUNTIME_PROXY_HANDLER.proxies == {}
    assert len(redirect_handlers) == 1
    assert type(redirect_handlers[0]) is module._RejectRedirectHandler


def test_cloudsql_runtime_rejects_wrong_metadata_service_account_before_token(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    calls: list[str] = []

    def fake_urlopen(request, *, timeout):
        assert timeout == module.HTTP_TIMEOUT_SECONDS
        calls.append(request.full_url)
        return _FakeHttpResponse(b"unexpected-sa@teleo-501523.iam.gserviceaccount.com")

    monkeypatch.setattr(module.RUNTIME_HTTP_OPENER, "open", fake_urlopen)

    with pytest.raises(SystemExit, match="did not match the required service account"):
        module.password(args)
    assert calls == [module.GCE_METADATA_EMAIL_URL]


@pytest.mark.parametrize(
    "token_body",
    [
        b"not-json",
        b"[]",
        b'{"access_token":"token","token_type":"Basic","expires_in":300}',
        b'{"access_token":"token with spaces","token_type":"Bearer","expires_in":300}',
        b'{"access_token":"token","token_type":"Bearer","expires_in":0}',
        b'{"access_token":"one","access_token":"two","token_type":"Bearer","expires_in":300}',
    ],
)
def test_cloudsql_runtime_rejects_malformed_metadata_tokens_without_leaking_them(
    monkeypatch: pytest.MonkeyPatch,
    token_body: bytes,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    bodies = [module.RUNTIME_SERVICE_ACCOUNT.encode("ascii"), token_body]
    monkeypatch.setattr(
        module.RUNTIME_HTTP_OPENER,
        "open",
        lambda _request, *, timeout: _FakeHttpResponse(bodies.pop(0)),
    )

    with pytest.raises(SystemExit, match="Runtime metadata token response was invalid") as captured:
        module.password(args)
    assert "token with spaces" not in str(captured.value)


@pytest.mark.parametrize(
    "secret_body",
    [
        b"not-json",
        b"[]",
        b"{}",
        b'{"payload":{"data":"%%%"}}',
        json.dumps({"payload": {"data": base64.b64encode(b"x" * 4097).decode("ascii")}}).encode(),
        json.dumps({"payload": {"data": base64.b64encode(b"line1\nline2").decode("ascii")}}).encode(),
        json.dumps({"payload": {"data": base64.b64encode(b"line1\rline2").decode("ascii")}}).encode(),
        json.dumps({"payload": {"data": base64.b64encode(b"nul\x00byte").decode("ascii")}}).encode(),
    ],
)
def test_cloudsql_runtime_rejects_malformed_secret_payloads_without_leaking_them(
    monkeypatch: pytest.MonkeyPatch,
    secret_body: bytes,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    bodies = _runtime_http_bodies(module)
    bodies[-1] = secret_body
    monkeypatch.setattr(
        module.RUNTIME_HTTP_OPENER,
        "open",
        lambda _request, *, timeout: _FakeHttpResponse(bodies.pop(0)),
    )

    with pytest.raises(SystemExit, match=r"Scoped runtime secret .* invalid") as captured:
        module.password(args)
    assert secret_body.decode("utf-8", errors="ignore") not in str(captured.value)


def test_cloudsql_runtime_admin_secret_is_impossible_through_arguments(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module, password_secret="gcp-teleo-pgvector-standby-postgres-password")
    _configure_runtime_environment(monkeypatch)
    monkeypatch.setattr(
        module.RUNTIME_HTTP_OPENER,
        "open",
        lambda *_args, **_kwargs: pytest.fail("metadata must not be queried for an admin secret argument"),
    )

    with pytest.raises(SystemExit, match="requires scoped password secret"):
        module.password(args)


def test_cloudsql_runtime_http_and_psql_failures_never_echo_credentials(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    _configure_runtime_environment(monkeypatch)
    dummy = "DUMMY-RUNTIME-SECRET-DO-NOT-LEAK"

    def failed_urlopen(_request, *, timeout):
        raise RuntimeError(f"provider echoed {dummy}")

    monkeypatch.setattr(module.RUNTIME_HTTP_OPENER, "open", failed_urlopen)
    with pytest.raises(SystemExit) as http_error:
        module.password(args)
    assert dummy not in str(http_error.value)

    monkeypatch.setattr(module, "runtime_password_from_metadata", lambda: dummy)
    monkeypatch.setattr(
        module.subprocess,
        "run",
        lambda *_args, **_kwargs: SimpleNamespace(returncode=2, stdout=dummy, stderr=dummy),
    )
    with pytest.raises(SystemExit, match="Cloud SQL query failed with exit status 2") as psql_error:
        module.run_psql(args, "select 1;")
    assert dummy not in str(psql_error.value)


def test_cloudsql_runtime_read_marker_binds_database_user_private_server_and_ssl(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    marker = {
        "database": module.RUNTIME_CLOUDSQL_DB,
        "database_user": module.RUNTIME_DB_USER,
        "server_addr": module.RUNTIME_CLOUDSQL_HOST,
        "server_port": int(module.RUNTIME_CLOUDSQL_PORT),
        "ssl": True,
        "ssl_version": "TLSv1.3",
        "system_identifier": module.RUNTIME_SYSTEM_IDENTIFIER,
        "wal_lsn": "0/1",
    }
    captured: dict[str, object] = {}

    def fake_psql_json_lines(_args, sql, db=None):
        captured["sql"] = sql
        captured["db"] = db
        return [marker]

    monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines)

    assert module.database_read_marker(args) == marker
    assert captured["db"] == module.RUNTIME_CLOUDSQL_DB
    assert "inet_server_addr()" in str(captured["sql"])
    assert "pg_stat_ssl" in str(captured["sql"])


@pytest.mark.parametrize(
    ("field", "value", "message"),
    [
        ("database", "teleo_kb", "unexpected database identity"),
        ("database_user", "postgres", "unexpected database identity"),
        ("server_addr", "127.0.0.1", "unexpected server address"),
        ("server_port", 6543, "unexpected server port"),
        ("ssl", False, "encrypted connection"),
    ],
)
def test_cloudsql_runtime_read_marker_rejects_wrong_live_identity(
    monkeypatch: pytest.MonkeyPatch,
    field: str,
    value: object,
    message: str,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = _runtime_connection_args(module)
    marker = {
        "database": module.RUNTIME_CLOUDSQL_DB,
        "database_user": module.RUNTIME_DB_USER,
        "server_addr": module.RUNTIME_CLOUDSQL_HOST,
        "server_port": int(module.RUNTIME_CLOUDSQL_PORT),
        "ssl": True,
        "ssl_version": "TLSv1.3",
        "system_identifier": module.RUNTIME_SYSTEM_IDENTIFIER,
        "wal_lsn": "0/1",
    }
    marker[field] = value
    monkeypatch.setattr(module, "psql_json_lines", lambda *_args, **_kwargs: [marker])

    with pytest.raises(SystemExit, match=message):
        module.database_read_marker(args)


def test_cloudsql_proposal_calls_leave_leo_identity_to_the_database(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    captured: list[str] = []

    def fake_psql_json_lines(_args, sql, db=None):
        captured.append(sql)
        return [{"id": "proposal-id", "database": db}]

    monkeypatch.setattr(module, "psql_json_lines", fake_psql_json_lines)
    core_args = SimpleNamespace(
        proposal_type="revise_claim",
        target_kind="claim",
        target_ref="claim-id",
        current="old",
        proposed="new",
        evidence=[],
        implication=[],
        originator="",
        channel="telegram",
        source_ref="telegram:1",
        rationale="Grounded correction",
        canonical_db="teleo_canonical",
    )
    edge_args = SimpleNamespace(
        from_claim="11111111-1111-4111-8111-111111111111",
        to_claim="22222222-2222-4222-8222-222222222222",
        edge_type="supports",
        channel="telegram",
        source_ref="telegram:2",
        rationale="Grounded relation",
        canonical_db="teleo_canonical",
    )

    module.propose_core_change(core_args)
    module.propose_edge(edge_args)

    assert "p.proposed_by_handle" not in captured[0]
    assert "as proposed_by_handle" not in captured[0]
    assert "p.proposed_by_handle" not in captured[1]
    assert "as proposed_by_handle" not in captured[1]
    assert "stage_leoclean_proposal(\n    p.proposal_type,\n    p.channel," in captured[0]
    assert "stage_leoclean_proposal(\n    'add_edge',\n    p.channel," in captured[1]


def test_gcp_runtime_role_uses_one_narrow_staging_function() -> None:
    sql = GCP_RUNTIME_ROLE_SQL.read_text()

    assert "security definer" in sql.lower()
    assert "stage_leoclean_proposal" in sql
    assert "'pending_review'" in sql
    assert "revoke all privileges on all tables in schema public, kb_stage" in sql
    assert "grant insert" in sql.lower()
    assert "unexpectedly has database CREATE" in sql
    assert "unexpectedly has schema CREATE" in sql
    assert "runtime CONNECT reaches a noncanonical database" in sql
    assert "PUBLIC retains database CONNECT" in sql
    assert "PUBLIC retains large-object mutator EXECUTE" in sql
    assert "REVOKE TEMP FROM PUBLIC would be a" in sql
    assert "can execute unexpected SECURITY DEFINER functions" in sql
    assert "revoke connect on database %I from public" in sql
    assert "revoke execute on function pg_catalog.%I(%s) from public" in sql
    assert re.search(
        r"grant\s+insert\s*\([^)]*proposal_type[^)]*payload[^)]*\)\s+"
        r"on\s+kb_stage\.kb_proposals\s+to\s+leoclean_kb_stage_owner",
        sql,
        re.IGNORECASE | re.DOTALL,
    )
    assert not re.search(
        r"grant\s+insert(?:\s*\([^)]*\))?\s+on\s+[^;]+\s+to\s+leoclean_kb_runtime",
        sql,
        re.IGNORECASE | re.DOTALL,
    )
    assert "grant update" not in sql.lower()
    assert "grant delete" not in sql.lower()
    assert "create role leoclean_kb_runtime nologin" in sql.lower()
    disable_statement = "alter role leoclean_kb_runtime\n  with nologin nosuperuser"
    password_statement = "\\password leoclean_kb_runtime"
    assert sql.lower().index(disable_statement) < sql.index(password_statement)
    assert "\\getenv runtime_password" not in sql
    assert "password :'runtime_password'" not in sql.lower()
    assert sql.lower().index("with nologin nosuperuser") < sql.index("\\connect teleo_canonical")
    assert sql.index("\\connect teleo_canonical") < sql.lower().rindex("with login nosuperuser")
    assert sql.lower().rindex("with login nosuperuser") < sql.rindex("commit;")
    assert "leoclean_kb_runtime final login attributes are unsafe" in sql


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)
    if path.name == "cloudsql_memory_tool.py":
        module.runtime_ssl_root_cert = lambda: module.RUNTIME_SSL_ROOT_CERT
        module.runtime_cloudsdk_config = lambda: module.RUNTIME_CLOUDSDK_CONFIG
    return module


def test_cloudsql_reads_emit_stable_source_bound_retrieval_receipts(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    args = SimpleNamespace(
        command="search",
        query="AI sandbagging M&A liability",
        claim_id=None,
        limit=5,
        context_limit=5,
    )
    marker = {
        "database": "teleo_clone_test",
        "database_user": "postgres",
        "system_identifier": "system-1",
        "wal_lsn": "0/123",
    }
    monkeypatch.setattr(module, "database_read_marker", lambda _args: dict(marker))
    payload = {
        "artifact": "teleo_cloudsql_canonical_kb_context",
        "query": args.query,
        "claims": [
            {
                "id": "2a7ae257-d01d-46f4-b813-63f81bb9c7c7",
                "evidence": [
                    {
                        "source_id": "15740795-ecc6-40fa-9a01-3d6bc7c54f79",
                        "source_hash": "abc123",
                        "storage_path": "inbox/archive/source.md",
                    }
                ],
                "edges": [],
            }
        ],
    }

    result = module.read_with_retrieval_receipt(args, lambda: json.loads(json.dumps(payload)))
    receipt = result["retrieval_receipt"]

    assert receipt["schema"] == "livingip.teleoKbRetrievalReceipt.v1"
    assert receipt["claim_ids"] == ["2a7ae257-d01d-46f4-b813-63f81bb9c7c7"]
    assert receipt["source_ids"] == ["15740795-ecc6-40fa-9a01-3d6bc7c54f79"]
    assert receipt["read_consistency"]["status"] == "stable_wal_marker"
    rebuilt = module.build_retrieval_receipt(
        payload,
        query_descriptor=module.receipt_query_descriptor(args),
        before=marker,
        after=marker,
        attempts=1,
        consistency_status="stable_wal_marker",
    )
    assert rebuilt["semantic_context_sha256"] == receipt["semantic_context_sha256"]
    assert rebuilt["artifact_state_sha256"] == receipt["artifact_state_sha256"]


def test_cloudsql_evidence_rows_include_source_identity_and_hash(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    captured = {}

    def fake_query(_args, sql, db=None):
        captured.update(sql=sql, db=db)
        return []

    monkeypatch.setattr(module, "psql_json_lines", fake_query)
    args = SimpleNamespace(canonical_db="teleo_clone_test", include_excerpts=False, redacted=False)

    assert module.canonical_evidence(args, "2a7ae257-d01d-46f4-b813-63f81bb9c7c7", 8) == []
    assert "s.id as source_id" in captured["sql"]
    assert "s.hash as source_hash" in captured["sql"]
    assert "'source_id', source_id::text" in captured["sql"]
    assert "'source_hash', source_hash" in captured["sql"]
    assert captured["db"] == "teleo_clone_test"


def test_cloudsql_markdown_prints_retrieval_receipt_before_claim_context() -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    value = {
        "artifact": "teleo_cloudsql_canonical_kb_context",
        "backend": "cloudsql:test",
        "query": "test",
        "terms": ["test"],
        "privacy": "redacted",
        "claims": [],
        "context_rows": [],
        "retrieval_receipt": {
            "schema": "livingip.teleoKbRetrievalReceipt.v1",
            "semantic_context_sha256": "a" * 64,
            "artifact_state_sha256": "b" * 64,
            "read_consistency": {"status": "stable_wal_marker", "attempts": 1},
        },
    }
    output = io.StringIO()

    with contextlib.redirect_stdout(output):
        module.emit_markdown(value)

    text = output.getvalue()
    assert text.index("# Teleo KB Retrieval Receipt") < text.index("# Teleo KB Context")
    assert "semantic context SHA-256" in text
    assert "artifact state SHA-256" in text
    assert "DB read consistency: `stable_wal_marker`" in text


def test_cloudsql_all_read_commands_are_receipted() -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")

    expected = {
        "search",
        "context",
        "show",
        "evidence",
        "edges",
        "status",
        "list-proposals",
        "search-proposals",
        "show-proposal",
        "decision-matrix-status",
    }
    assert expected == module.RECEIPTED_READ_COMMANDS


def test_cloudsql_status_works_without_optional_audit_fallback(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    canonical = {
        "db_identity": "teleo_clone_test|postgres",
        "extensions": ["plpgsql"],
        "schema_tables": {"kb_stage": 15, "public": 32},
        "high_signal_rows": {"claims": 1837, "sources": 4145, "kb_proposals": 29},
    }
    calls = []

    def fake_run_psql(_args, sql, db=None):
        calls.append((sql, db))
        if "json_build_object" in sql:
            return json.dumps(canonical)
        if "to_regclass" in sql:
            return "f\n"
        raise AssertionError(sql)

    monkeypatch.setattr(module, "run_psql", fake_run_psql)
    args = SimpleNamespace(db="teleo_clone_test", canonical_db="teleo_clone_test")

    result = module.status(args)

    assert result["canonical"] == canonical
    assert result["audit_fallback_available"] is False
    assert result["teleo_restore_tables"] == 0
    assert all("teleo_restore.response_audit" not in sql or "to_regclass" in sql for sql, _db in calls)


def test_cloudsql_zero_hit_search_stays_canonical_when_audit_is_disabled(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    canonical = {"artifact": "teleo_cloudsql_canonical_kb_context", "hit_count_total": 0, "hits": []}
    monkeypatch.setattr(module, "query_canonical_rows", lambda *_args: canonical.copy())
    monkeypatch.delenv("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", raising=False)
    monkeypatch.setattr(
        module,
        "audit_restore_available",
        lambda _args: (_ for _ in ()).throw(AssertionError("audit availability must not be probed")),
    )
    monkeypatch.setattr(
        module,
        "query_audit_rows",
        lambda *_args: (_ for _ in ()).throw(AssertionError("audit fallback must not run")),
    )

    result = module.query_rows(SimpleNamespace(), "unseen query", 8)

    assert result["hits"] == []
    assert "audit fallback disabled" in result["canonical_fallback_reason"]


def test_cloudsql_zero_hit_search_uses_audit_only_when_explicitly_enabled(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "cloudsql_memory_tool.py")
    canonical = {"artifact": "teleo_cloudsql_canonical_kb_context", "hit_count_total": 0, "hits": []}
    audit = {"artifact": "teleo_cloudsql_memory_context", "hits": [{"row_id": "audit-1"}]}
    monkeypatch.setenv("TELEO_CLOUDSQL_ENABLE_AUDIT_FALLBACK", "1")
    monkeypatch.setattr(module, "query_canonical_rows", lambda *_args: canonical.copy())
    monkeypatch.setattr(module, "audit_restore_available", lambda _args: True)
    monkeypatch.setattr(module, "query_audit_rows", lambda *_args: audit.copy())

    result = module.query_rows(SimpleNamespace(), "unseen query", 8)

    assert result["hits"] == [{"row_id": "audit-1"}]
    assert result["canonical_fallback_reason"] == "no matching canonical public/persona/strategy/belief rows"


def test_vps_bridge_verifies_relative_source_artifact_against_canonical_hash(tmp_path: Path) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    root = tmp_path / "workspace"
    artifact = root / "domains" / "research.md"
    artifact.parent.mkdir(parents=True)
    artifact.write_text("hash-bound evidence\n", encoding="utf-8")
    digest = module.hashlib.sha256(artifact.read_bytes()).hexdigest()

    verification = module.verify_source_artifact(
        {
            "storage_path": "domains/research.md",
            "url": None,
            "source_hash": digest,
        },
        roots=(root,),
    )

    assert verification["status"] == "verified"
    assert verification["artifact_exists"] is True
    assert verification["artifact_sha256"] == digest
    assert verification["hash_matches_db"] is True
    assert verification["resolved_storage_path"] == str(artifact)


def test_vps_bridge_does_not_invent_local_artifact_state(tmp_path: Path) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    root = tmp_path / "workspace"
    root.mkdir()
    outside = tmp_path / "outside.md"
    outside.write_text("outside", encoding="utf-8")

    rejected = module.verify_source_artifact(
        {"storage_path": str(outside), "source_hash": module.hashlib.sha256(outside.read_bytes()).hexdigest()},
        roots=(root,),
    )
    missing = module.verify_source_artifact(
        {"storage_path": "domains/missing.md", "source_hash": "0" * 64},
        roots=(root,),
    )
    locator = module.verify_source_artifact(
        {"storage_path": "telegram://chat/-5146042086/message/12345", "source_hash": "0" * 64},
        roots=(root,),
    )

    assert rejected["status"] == "rejected_outside_allowlisted_roots"
    assert rejected["artifact_exists"] is False
    assert missing["status"] == "missing"
    assert missing["artifact_exists"] is False
    assert locator["status"] == "locator_only"
    assert locator["artifact_exists"] is None


def test_vps_bridge_reports_source_hash_mismatch(tmp_path: Path) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    root = tmp_path / "workspace"
    artifact = root / "evidence.txt"
    artifact.parent.mkdir()
    artifact.write_text("actual bytes", encoding="utf-8")

    verification = module.verify_source_artifact(
        {"storage_path": "evidence.txt", "source_hash": "0" * 64},
        roots=(root,),
    )

    assert verification["status"] == "hash_mismatch"
    assert verification["hash_matches_db"] is False
    assert verification["artifact_sha256"] != verification["expected_sha256"]


def test_vps_bridge_source_artifact_verification_is_size_bounded(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    root = tmp_path / "sources"
    artifact = root / "large.md"
    artifact.parent.mkdir(parents=True)
    artifact.write_bytes(b"0123456789")
    monkeypatch.setenv("TELEO_KB_MAX_ARTIFACT_BYTES", "4")

    result = module.verify_source_artifact(
        {"storage_path": "large.md", "source_hash": module.hashlib.sha256(artifact.read_bytes()).hexdigest()},
        roots=(root,),
    )

    assert result["status"] == "too_large_to_verify"
    assert result["artifact_exists"] is True
    assert result["artifact_bytes"] == 10
    assert result["max_artifact_bytes"] == 4
    assert result["artifact_sha256"] is None


def test_vps_bridge_evidence_fetch_includes_source_identity_and_verification(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    root = tmp_path / "workspace"
    artifact = root / "source.md"
    artifact.parent.mkdir()
    artifact.write_text("source bytes", encoding="utf-8")
    digest = module.hashlib.sha256(artifact.read_bytes()).hexdigest()
    claim_id = "11111111-1111-4111-8111-111111111111"
    source_id = "22222222-2222-4222-8222-222222222222"
    captured_sql: list[str] = []

    def fake_psql_json(_args, sql):
        captured_sql.append(sql)
        return [
            {
                "claim_id": claim_id,
                "role": "grounds",
                "weight": 1,
                "source_id": source_id,
                "source_type": "article",
                "url": None,
                "storage_path": "source.md",
                "source_hash": digest,
                "captured_at": "2026-07-14T00:00:00+00:00",
                "excerpt": "source bytes",
            }
        ]

    monkeypatch.setenv("TELEO_KB_SOURCE_ROOTS", str(root))
    monkeypatch.setattr(module, "psql_json", fake_psql_json)

    rows = module.load_evidence(SimpleNamespace(), [claim_id], 4)[claim_id]

    assert rows[0]["source_id"] == source_id
    assert rows[0]["source_hash"] == digest
    assert rows[0]["artifact_verification"]["status"] == "verified"
    assert "s.id as source_id" in captured_sql[0]
    assert "s.hash as source_hash" in captured_sql[0]
    assert "s.captured_at" in captured_sql[0]


def test_vps_bridge_context_receipt_is_deterministic_and_retries_moving_wal(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    markers = iter(
        [
            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/1"},
            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/2"},
            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/2"},
            {"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/3"},
        ]
    )
    contract_calls = 0

    def contracts(_args, _query):
        nonlocal contract_calls
        contract_calls += 1
        return [{"id": "reply_budget", "hard_max_words": 200, "statement_timestamp": f"volatile-{contract_calls}"}]

    monkeypatch.setattr(module, "database_read_marker", lambda _args: next(markers))
    monkeypatch.setattr(module, "context_operational_contracts", contracts)
    monkeypatch.setattr(module, "compile_operational_response", lambda _contracts: None)
    monkeypatch.setattr(module, "find_claims", lambda _args, _query, _limit: [])
    monkeypatch.setattr(module, "load_evidence", lambda _args, _ids, _limit: {})
    monkeypatch.setattr(module, "load_edges", lambda _args, _ids, _limit: {})
    monkeypatch.setattr(module, "find_context_rows", lambda _args, _query, _limit: [])

    result = module.bundle(SimpleNamespace(), "same database question", 4, 6)
    receipt = result["retrieval_receipt"]

    assert contract_calls == 2
    assert receipt["read_consistency"]["status"] == "stable_content_across_wal_change_retry"
    assert receipt["read_consistency"]["attempts"] == 2
    stable_copy = {**result, "operational_contracts": [{"id": "reply_budget", "hard_max_words": 200}]}
    rebuilt = module.build_retrieval_receipt(
        {key: value for key, value in stable_copy.items() if key != "retrieval_receipt"},
        before={"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/9"},
        after={"database": "teleo", "database_user": "postgres", "system_identifier": "sys", "wal_lsn": "0/9"},
        attempts=1,
        consistency_status="stable_wal_marker",
    )
    assert rebuilt["semantic_context_sha256"] == receipt["semantic_context_sha256"]


def test_vps_bridge_retrieval_queries_have_deterministic_final_tie_breakers() -> None:
    source = (BRIDGE_DIR / "kb_tool.py").read_text(encoding="utf-8")

    assert "length(text), text, id" in source
    assert "order by score desc, source, owner, title, body" in source
    assert "s.url nulls last,\n                        s.id" in source
    assert "other.text,\n                        other.id" in source


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"] == 170
    assert reply_budget["hard_max_words"] == 220
    assert "omission is better" in reply_budget["shape"]
    constrained_budget = module.operational_contracts("Audit the restart claim under 180 words.")[0]
    assert constrained_budget["target_words"] == 170
    assert constrained_budget["hard_max_words"] == 180

    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"] == (
        "VPS filesystem document preparation and source-to-proposal staging are shipped; automatic Telegram "
        "attachment capture is not shipped."
    )
    assert source["capability_tier"] == (
        "VPS-local document preparation, canonical dedupe retrieval, and pending-review staging"
    )
    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"]

    broad_source_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?"
    )
    broad_source_contracts = {item["id"]: item for item in module.operational_contracts(broad_source_query)}
    assert "source_intake" in broad_source_contracts
    broad_source_response = module.compile_operational_response(list(broad_source_contracts.values()))
    assert broad_source_response is not None
    assert (
        "Sending a report in chat does not by itself create a proposal or persistent knowledge" in broad_source_response
    )
    assert "Telegram attachment downloader is not connected" in broad_source_response
    assert "teleo-kb prepare-source" in broad_source_response
    assert "teleo-kb propose-source" in broad_source_response
    assert "writes only kb_stage.kb_proposals" in broad_source_response

    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

    receipt_queries = [
        "Recall the same blocker and give the canonical receipt plus applied_at postflight readback.",
        "Return Closure proof for the same approved-versus-applied canonical gap.",
        "Say which before/after canonical receipt and applied_at readback would resolve it.",
    ]
    for query in receipt_queries:
        receipt_ids = {item["id"] for item in module.operational_contracts(query)}
        assert "apply_receipt_invariants" in receipt_ids
    receipt_contract = next(
        item
        for item in module.operational_contracts("Give the applied_at postflight proof.")
        if item["id"] == "apply_receipt_invariants"
    )
    assert receipt_contract["approve_claim_supported"] == [
        "claims",
        "sources",
        "evidence",
        "edges",
        "reasoning_tools",
    ]
    assert "aggregate table counts need not change" in receipt_contract["count_boundary"]


def test_vps_bridge_binds_claim_challenge_to_exact_retrieved_rows() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    contracts = module.operational_contracts(
        "Inspect one exact claim body and its evidence, challenge an unsupported assumption, and give a narrower "
        "replacement."
    )
    claim_id = "11111111-1111-4111-8111-111111111111"
    source_id = "22222222-2222-4222-8222-222222222222"
    claim_text = "The observed association proves a universal causal mechanism."
    excerpt = "The retained study observed an association in one bounded cohort."

    module.bind_claim_evidence_challenge(
        contracts,
        [{"id": claim_id, "text": claim_text}],
        {claim_id: [{"source_id": source_id, "excerpt": excerpt, "role": "grounds"}]},
    )
    response = module.compile_operational_response(contracts)

    assert response is not None
    assert claim_id in response
    assert source_id in response
    assert f'Exact claim body: "{claim_text}"' in response
    assert f'Exact evidence excerpt: "{excerpt}"' in response
    assert "Unsupported leap:" in response
    assert "Narrower revision:" in response

    unbound_contracts = module.operational_contracts(
        "Inspect one exact claim body and its evidence, challenge an unsupported assumption, and give a narrower "
        "replacement."
    )
    module.bind_claim_evidence_challenge(unbound_contracts, [], {})
    unbound_response = module.compile_operational_response(unbound_contracts)
    assert unbound_response is not None
    assert "no claim/source pair" in unbound_response
    assert "no database write was made" in unbound_response


def test_vps_prepare_source_retrieves_canonical_candidates_before_model_extraction(tmp_path: Path, monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    preparer = tmp_path / "prepare_kb_source_manifest.py"
    preparer.write_text("# fixture\n", encoding="utf-8")
    inbox = tmp_path / "source-inbox"
    preparation_root = tmp_path / "source-preparation"
    inbox.mkdir()
    preparation_root.mkdir()
    artifact = inbox / "source.md"
    artifact.write_text("Canonical review precedes guarded apply.", encoding="utf-8")
    output_dir = preparation_root / "private-output"
    candidate_id = "11111111-1111-4111-8111-111111111111"
    captured: dict = {}

    monkeypatch.setattr(module, "SOURCE_PREPARER_PATH", preparer)
    monkeypatch.setattr(module, "SOURCE_INBOX_ROOT", inbox)
    monkeypatch.setattr(module, "SOURCE_PREPARATION_ROOT", preparation_root)
    monkeypatch.setattr(
        module,
        "find_claims",
        lambda _args, query, limit: [
            {
                "id": candidate_id,
                "text": "Canonical writes require guarded apply.",
                "score": 2,
                "evidence_count": 3,
                "edge_count": 1,
            }
        ],
    )

    def fake_run(command, **_kwargs):
        context_path = Path(command[command.index("--kb-context") + 1])
        captured["context"] = json.loads(context_path.read_text(encoding="utf-8"))
        captured["command"] = command
        receipt = {
            "schema": module.SOURCE_PREPARATION_RECEIPT_SCHEMA,
            "status": "prepared",
            "database_write_performed": False,
            "canonical_apply_performed": False,
            "retrieval": {"canonical_candidate_ids": [candidate_id]},
            "extraction": {"model": "fixture/model", "claim_count": 1},
            "outputs": {},
        }
        return subprocess.CompletedProcess(command, 0, stdout=json.dumps(receipt), stderr="")

    monkeypatch.setattr(module.subprocess, "run", fake_run)
    args = SimpleNamespace(
        local=True,
        query="canonical review guarded apply",
        artifact=artifact,
        text=None,
        identity="document:test",
        source_key="test_source",
        source_type="article",
        title="Test source",
        locator="artifact://test/source",
        output_dir=output_dir,
        model="fixture/model",
    )

    receipt = module.prepare_source(args)

    assert receipt["status"] == "prepared"
    assert captured["context"]["database_search_query"] == args.query
    assert captured["context"]["candidate_claims"][0]["id"] == candidate_id
    assert "--text" not in captured["command"]


def test_vps_prepare_source_rejects_files_and_outputs_outside_private_roots(tmp_path: Path) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    inbox = tmp_path / "source-inbox"
    preparation = tmp_path / "source-preparation"
    outside = tmp_path / "outside.md"
    inbox.mkdir()
    preparation.mkdir()
    outside.write_text("private", encoding="utf-8")

    with pytest.raises(SystemExit, match="must stay under"):
        module._scoped_source_path(outside, inbox, "--artifact", must_exist=True)
    with pytest.raises(SystemExit, match="must stay under"):
        module._scoped_source_path(tmp_path / "outside-output", preparation, "--output-dir", must_exist=False)


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",
        (
            "Reviewers signed off on three database proposals. Challenge that conclusion from current state "
            "semantics and name the closing receipt."
        ): "proposal_state_readback",
        (
            "A source packet is attached to a pending proposal, so provenance is supposedly finished. Audit the "
            "canonical evidence link and distinguish it from a weak locator."
        ): "source_link_audit",
        (
            "Someone says the V3 architecture document is already part of Leo's live knowledge because a proposal "
            "exists. What is actually true right now, what would you inspect, and what must happen before those "
            "claims are canonical? Do not mutate anything."
        ): "proposal_state_readback",
        "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",
        (
            "The current visible Telegram sender is @m3taversal. What should Leo call this participant, which "
            "identity sources are allowed, and how should Leo avoid mixing identities?"
        ): "telegram_participant_identity",
    }

    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

    runtime_ids = {
        item["id"]
        for item in module.operational_contracts(
            "After a fresh process launch, do unchanged totals prove behavioral parity and a blank session?"
        )
    }
    assert "runtime_persistence" in runtime_ids


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

    v3_ids = {
        item["id"]
        for item in module.operational_contracts(
            "Someone says the V3 architecture document is already part of Leo's live knowledge because a proposal "
            "exists. What is actually true right now, what would you inspect, and what must happen before those "
            "claims are canonical? Do not mutate anything."
        )
    }
    assert v3_ids & direct_ids == {"proposal_state_readback"}
    assert "source_link_audit" not in v3_ids

    reasoning_followup_ids = {
        item["id"]
        for item in module.operational_contracts(
            "Inspect the exact claim body and evidence: what is assumption versus observed support, what would "
            "falsify it, and which replacement claims would you put into review? Keep them as proposals, not live "
            "knowledge."
        )
    }
    assert reasoning_followup_ids & direct_ids == set()

    chat_memory_ids = {
        item["id"]
        for item in module.operational_contracts(
            "Use chat memory only: bind the approved-versus-applied canonical blocker to TEMP until my next "
            "question. Do not stage or apply anything."
        )
    }
    assert chat_memory_ids == {"reply_budget"}
    memory_set_queries = [
        "Remember the approved-versus-applied blocker as TEMP for the next turn only. Chat memory only.",
        "Bind the approved-versus-applied canonical gap to TEMP until my next question; no staging or apply.",
        "Name the blocker precisely enough to distinguish approval, applied_at, and canonical readback. It must not "
        "become a source, memory record, or database write.",
    ]
    assert all(
        "apply_receipt_invariants" not in {item["id"] for item in module.operational_contracts(query)}
        for query in memory_set_queries
    )


def test_vps_bridge_named_document_question_matches_exact_proposal_and_compiles_receipt(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    database_status, approved, matrix, source_audit = _direct_contract_fixtures()
    database_status = {
        **database_status,
        "high_signal_rows": {**database_status["high_signal_rows"], "kb_proposals": 29},
        "proposal_status_counts": {"applied": 2, "approved": 3, "pending_review": 16, "canceled": 8},
    }
    source_metadata = {
        "apply_payload": {
            "sources": [
                {
                    "excerpt": json.dumps(
                        {
                            "metadata": {
                                "title": "LivingIP KB Agent Identity Architecture v3",
                                "source_key": "livingip_kb_agent_identity_architecture_v3",
                            }
                        }
                    )
                }
            ]
        }
    }
    corrected = {
        **approved,
        "id": "10bc0719-1c2b-5f42-a41e-86e6478692cb",
        "status": "pending_review",
        "source_ref": "normalized:12e38cb0-b72b-557c-95e5-ffc789a29f62:72895c9b58b79d41",
        "payload": source_metadata,
        "created_at": "2026-07-13T04:00:00+00:00",
        "reviewed_at": None,
        "applied_at": None,
    }
    canceled = {
        **corrected,
        "id": "a9eb4158-5aa2-5484-b5a7-3d249588b3ec",
        "status": "canceled",
        "created_at": "2026-07-13T03:00:00+00:00",
    }
    monkeypatch.setattr(
        module,
        "load_direct_readback_snapshot",
        lambda _args: {
            "snapshot_id": "300:300:",
            "statement_timestamp": "2026-07-13T05:00:00+00:00",
            "database_status": database_status,
            "proposals": [approved, canceled, corrected],
            "named_packet_proposals": [approved],
            "named_packet_canonical_matches": {"reasoning_tools": [], "claims": [], "sources": []},
            "decision_matrix_status": matrix,
            "source_link_audit": source_audit,
        },
    )
    prompt = (
        "Someone says the V3 architecture document is already part of Leo's live knowledge because a proposal "
        "exists. What is actually true right now, what would you inspect, and what must happen before those claims "
        "are canonical? Do not mutate anything."
    )

    contracts = module.context_operational_contracts(SimpleNamespace(), prompt)
    proposal_state = next(item for item in contracts if item["id"] == "proposal_state_readback")
    response = module.compile_operational_response(contracts)

    assert proposal_state["proposal_query_terms"] == ["v3", "architecture"]
    assert [item["id"] for item in proposal_state["matching_proposals"]] == [corrected["id"], canceled["id"]]
    assert response is not None
    assert response.startswith("No.\nFresh live readback.")
    assert f"proposal: {corrected['id']}; status: pending_review; applied_at: none" in response
    assert corrected["source_ref"] in response
    assert "proposal states are applied: 2, approved: 3, pending_review: 16, canceled: 8" in response.lower()
    assert "Approved is not applied." in response
    assert "has not reached canonical apply" in response
    assert "telegram_file_refs" not in response
    assert len(re.findall(r"\b\w+(?:[-']\w+)*\b", response)) <= 200
    assert _load_module(DB_CONTEXT_PLUGIN).response_contract_issues(response, contracts) == []


def test_vps_bridge_oos_intents_preserve_specific_contracts_and_negated_actions() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    cases = {
        (
            "I corrected Leo's worldview in chat and someone also edited SOUL.md. Will that correction be part of "
            "Leo's canonical identity tomorrow after a restart? Give me the exact truth test from database row to "
            "runtime artifact. Do not change either surface."
        ): {"identity_canonicality_readback", "runtime_persistence"},
        (
            "The five database totals are unchanged after a restart. Does that prove Leo's answer behavior is "
            "unchanged and that every fact from the prior session was erased?"
        ): {"runtime_persistence"},
        (
            "An old claim recorded a 60% forecast but never defined resolution criteria. The event is now over. "
            "What needs a schema proposal?"
        ): {"forecast_resolution_schema"},
        (
            "Should the factual claim have one copy per agent, or should each agent's conclusion live in beliefs? "
            "Can we add edges from beliefs to the shared claim in the current schema?"
        ): {"shared_claims_agent_positions"},
        (
            "A canonical claim is wrong. I want the replacement and the old claim visibly retired. In current v1, "
            "which writes fit approve_claim and which require a separate reviewed apply capability?"
        ): {"claim_supersession_schema"},
        ("A temporary-profile GatewayRunner answered, but posted nothing to Telegram. Is delivery proven live?"): {
            "telegram_delivery_proof"
        },
        (
            "The current visible Telegram sender is @m3taversal. An earlier answer shortened that handle. What "
            "should Leo call this participant, which identity sources are allowed, and how should Leo avoid mixing "
            "identities when another user replies? Do not write to memory or the KB."
        ): {"telegram_participant_identity"},
    }

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

    memory_query = "Remember this blocker under __MEMORY_TOKEN__. This is chat memory only; do not write it to the KB."
    memory_ids = {item["id"] for item in module.operational_contracts(memory_query)}
    assert not memory_ids & module.DIRECT_READBACK_CONTRACTS
    assert memory_ids == {"reply_budget"}


def test_vps_bridge_restart_with_soul_reference_prefers_runtime_causality_and_session_proof() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    prompt = (
        "After a restart, does the fact that SOUL.md exists prove what this same session will remember and what the "
        "handler will answer? Separate canonical rows, runtime/profile artifacts, session continuity, and visible "
        "delivery."
    )

    contracts = module.operational_contracts(prompt)
    contract_ids = {item["id"] for item in contracts}
    response = module.compile_operational_response(contracts)

    assert "runtime_persistence" in contract_ids
    assert "identity_canonicality_readback" not in contract_ids
    assert response is not None
    assert "Proof tiers:" in response
    assert "state.db/session JSONL" in response
    assert "handler" in response
    assert "Telegram" in response
    assert "runtime_persistence" not in {
        item["id"] for item in module.operational_contracts("Where is SOUL.md located on the filesystem?")
    }


def test_vps_bridge_chat_only_memory_with_approved_applied_vocabulary_avoids_db_lifecycle_contracts() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    prompts = (
        (
            "For this chat only, remember the temporary label OXBOW for the phrase 'approved is not applied'. "
            "Do not query or write the database."
        ),
        (
            "Memory check for the preceding turn only: return the temporary label attached to 'approved is not "
            "applied'. This is not a proposal-state or readiness question."
        ),
    )

    for prompt in prompts:
        contract_ids = {item["id"] for item in module.operational_contracts(prompt)}
        assert contract_ids == {"reply_budget"}, prompt
        assert not contract_ids & {"proposal_state_readback", "proposal_apply_readiness"}


def test_vps_bridge_negated_database_scope_does_not_activate_lifecycle_readback() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    prompt = (
        "The database is not relevant; remember that approval and apply are distinct lifecycle words for this "
        "conversation only."
    )

    contract_ids = {item["id"] for item in module.operational_contracts(prompt)}

    assert contract_ids == {"reply_budget"}


def test_vps_bridge_query_terms_extend_beyond_the_old_ten_term_cutoff() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    prompt = (
        "Review alpha beta gamma delta epsilon zeta eta theta iota kappa lambda before examining ceramic turbine "
        "housings for hydraulic resonance anomalies."
    )

    terms = module.query_terms(prompt)

    assert {"ceramic", "turbine", "housings", "hydraulic", "resonance", "anomalies"} <= set(terms)
    assert terms.index("ceramic") > 9
    assert len(terms) <= 24


def test_vps_bridge_contract_receipt_collects_only_typed_row_ids() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    proposal_id = "11111111-1111-4111-8111-111111111111"
    source_ref = "22222222-2222-4222-8222-222222222222"

    row_ids = module._contract_row_ids(
        [
            {
                "id": "proposal_state_readback",
                "matching_proposals": [{"id": proposal_id, "source_ref": source_ref}],
                "untyped_uuid": "33333333-3333-4333-8333-333333333333",
            }
        ]
    )

    assert row_ids == [proposal_id]


def test_vps_bridge_context_corpus_does_not_expose_persona_source_ref(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    captured = {}

    def capture_sql(_args, sql):
        captured["sql"] = sql
        return []

    monkeypatch.setattr(module, "psql_json", capture_sql)
    assert module.find_context_rows(SimpleNamespace(), "ceramic turbine", 4) == []
    assert "p.source_ref" not in captured["sql"]
    assert "p.voice, p.role" in captured["sql"]


def test_vps_bridge_ranked_retrieval_keeps_one_match_recall_floor(monkeypatch) -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    sql_statements = []

    def capture_sql(_args, sql):
        sql_statements.append(sql)
        return []

    monkeypatch.setattr(module, "psql_json", capture_sql)
    module.find_claims(SimpleNamespace(), "singular ceramic topic with several wrapper terms", 4)
    module.find_context_rows(SimpleNamespace(), "singular ceramic topic with several wrapper terms", 4)

    assert len(sql_statements) == 2
    for sql in sql_statements:
        assert "max(score) over () as max_score" in sql
        assert "score >= 1" in sql
        assert "case when max_score >= 2 then 2 else 1 end" in sql


def test_vps_bridge_combines_identity_and_restart_contracts_for_independent_paraphrase() -> None:
    module = _load_module(BRIDGE_DIR / "kb_tool.py")
    prompt = (
        "A maintainer changed SOUL.md and plans a service restart. Contrast durable profile and session inputs with "
        "canonical identity rows, then give the row, render, hash, and handler receipt needed to verify the result."
    )

    contracts = module.operational_contracts(prompt)
    contract_ids = {item["id"] for item in contracts}
    response = module.compile_operational_response(contracts)

    assert {"identity_canonicality_readback", "runtime_persistence"} <= contract_ids
    assert response is not None
    assert "personas, strategies, beliefs" in response
    assert "state.db and session JSONL" in response
    assert "render/sync" in response
    assert "restart" in response


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)

        demo_contract = next(
            (contract for contract in contracts if contract["id"] == "demo_capability_readback"),
            None,
        )
        if demo_contract:
            assert [row["status"] for row in demo_contract["applied_and_approved_proposals"]] == [
                "approved",
                "applied",
            ]
            assert "1/1 approved rows are approved_needs_apply_payload" in response

        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 "automatic Telegram attachment capture is 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_schema_contract_reads_live_columns_and_edge_enum(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 "pg_enum" in sql:
            return [{"type_name": "edge_type", "values": ["supports", "supersedes"]}]
        if "information_schema.columns" in sql:
            return [
                {"table_name": "claims", "columns": ["id", "text", "superseded_by"]},
                {"table_name": "claim_edges", "columns": ["id", "from_claim", "to_claim", "edge_type"]},
            ]
        if "pg_constraint" in sql:
            return []
        raise AssertionError(sql)

    monkeypatch.setattr(module, "psql_json", fake_psql_json)
    contracts = {
        item["id"]: item
        for item in module.context_operational_contracts(
            SimpleNamespace(),
            "A canonical claim is wrong. In current v1, can approve_claim insert its replacement and supersedes edge?",
        )
    }

    supersession = contracts["claim_supersession_schema"]
    assert supersession["current_columns"]["claims"] == ["id", "text", "superseded_by"]
    assert supersession["current_columns"]["claim_edges"] == ["id", "from_claim", "to_claim", "edge_type"]
    assert supersession["current_edge_types"] == ["supports", "supersedes"]
    assert any("pg_enum" in sql and "t.typname" in sql for sql in captured_sql)
    assert all(not re.search(r"\b(?:insert|update|delete|alter|drop|create)\b", sql, re.I) for sql in captured_sql)


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
    )