teleo-infrastructure/tests/test_leoclean_nosend_runtime.py

"""Fail-closed contracts for the source-pinned leoclean no-send runtime."""

from __future__ import annotations

import ast
import copy
import importlib.util
import io
import json
import re
import shutil
import subprocess
import sys
import tarfile
from pathlib import Path
from types import SimpleNamespace

import pytest
import yaml

from scripts import compile_leoclean_nosend_runtime as compiler
from scripts import leo_behavior_manifest
from scripts import verify_leoclean_nosend_runtime as verifier

ROOT = Path(__file__).resolve().parents[1]
RUNTIME = ROOT / "hermes-agent" / "runtime" / "leoclean-nosend"
LOCK = json.loads((RUNTIME / "runtime-lock.json").read_text(encoding="utf-8"))
CONTRACT = json.loads((RUNTIME / "runtime-contract.json").read_text(encoding="utf-8"))


def load_bootstrap():
    spec = importlib.util.spec_from_file_location("leoclean_nosend_bootstrap_test", RUNTIME / "bootstrap.py")
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def load_module(path: Path, name: str):
    spec = importlib.util.spec_from_file_location(name, path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    sys.modules[name] = module
    spec.loader.exec_module(module)
    return module


def copy_profile(tmp_path: Path, bootstrap) -> Path:
    runtime_root = tmp_path / "runtime"
    runtime_root.mkdir(parents=True)
    compiler._copy_profile(ROOT, runtime_root, LOCK)
    bootstrap._runtime_root = lambda: runtime_root
    profile = tmp_path / "profile"
    shutil.copytree(runtime_root / "profile-template", profile)
    return profile


def test_committed_lock_and_contract_are_exactly_pinned() -> None:
    compiler._validate_lock(LOCK)
    load_bootstrap()._validate_contract(CONTRACT)
    assert LOCK["hermes"]["commit"] == "b2f477a30b3c05d0f383c543af98496ae8a96070"
    assert LOCK["hermes"]["archive_sha256"] == "a1c3455e65d7948746046314c69bf39833d915ca6da75d74e47a89289a36c742"
    assert LOCK["base_images"]["python"].startswith("python:3.11.9-slim-bookworm@sha256:")
    assert CONTRACT["runtime_mode"] == "gcp_staging_no_send"
    assert CONTRACT["transport_authority"] == "absent"
    assert CONTRACT["toolset"]["allowed_tools"] == ["skills_list", "skill_view", "terminal"]
    assert "send_message" in CONTRACT["toolset"]["forbidden_tools"]
    assert CONTRACT["database"]["runtime_role"] == "leoclean_kb_runtime"
    assert CONTRACT["database"]["automatic_context_modes"] == [
        "claim_evidence_challenge",
        "review_only_candidate",
    ]
    assert CONTRACT["database"]["unsupported_context_policy"] == "fail_closed"
    assert CONTRACT["profile"]["learning_policy"] == {
        "automatic_memory_review_interval": 0,
        "automatic_skill_review_interval": 0,
        "memory_enabled": False,
        "user_profile_enabled": False,
    }
    assert CONTRACT["profile"]["routing_policy"] == {
        "cheap_model": "absent",
        "smart_model_routing": False,
    }
    config = yaml.safe_load((RUNTIME / "profile-template" / "config.yaml").read_text(encoding="utf-8"))
    assert config["memory"] == {
        "enabled": False,
        "memory_enabled": False,
        "nudge_interval": 0,
        "user_profile_enabled": False,
    }
    assert config["skills"] == {"creation_nudge_interval": 0}
    assert config["smart_model_routing"] is False
    assert leo_behavior_manifest.safe_model_config(RUNTIME / "profile-template" / "config.yaml")["status"] == (
        "observed"
    )
    identity = json.loads((RUNTIME / "profile-template" / "gcp-runtime-identity.json").read_text(encoding="utf-8"))
    assert identity == {
        "expected_service_account": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",
        "schema": "livingip.leocleanGcpRuntimeIdentity.v1",
    }


def test_profile_sources_keep_gcp_adapters_out_of_vps_deploy_paths() -> None:
    by_target = {entry["target"]: entry for entry in LOCK["profile_sources"]}
    assert by_target["bin/cloudsql_memory_tool.py"]["source"] == (
        "hermes-agent/runtime/leoclean-nosend/profile-sources/cloudsql_memory_runtime.py"
    )
    assert by_target["plugins/leo-db-context/__init__.py"]["source"] == (
        "hermes-agent/runtime/leoclean-nosend/profile-sources/leo_db_context.py"
    )
    assert by_target["bin/cloudsql_memory_base.py"]["source"] == ("hermes-agent/leoclean-bin/cloudsql_memory_tool.py")
    assert by_target["bin/operational_contracts.py"]["source"] == (
        "hermes-agent/runtime/leoclean-nosend/profile-sources/operational_contracts.py"
    )
    assert by_target["plugins/leo-db-context/base.py"]["source"] == (
        "hermes-agent/leoclean-plugins/vps/leo-db-context/__init__.py"
    )


@pytest.mark.parametrize(
    ("section", "field", "unsafe"),
    [
        ("toolset", "allowed_tools", ["skills_list", "skill_view", "terminal", "send_message"]),
        ("toolset", "forbidden_tools", ["process"]),
        ("profile", "forbidden_entries", [".env"]),
        ("environment", "forbidden_exact", ["TELEGRAM_BOT_TOKEN"]),
        ("database", "direct_canonical_writes", "allowed"),
    ],
)
def test_runtime_rejects_a_weakened_contract(section: str, field: str, unsafe: object) -> None:
    bootstrap = load_bootstrap()
    mutated = copy.deepcopy(CONTRACT)
    mutated[section][field] = unsafe

    with pytest.raises(bootstrap.NoSendContractError):
        bootstrap._validate_contract(mutated)


@pytest.mark.parametrize(
    ("section", "index", "field", "unsafe"),
    [
        ("patches", 0, "source", "/tmp/patch.py"),
        ("patches", 0, "source", "../../patch.py"),
        ("patches", 0, "target", "/tmp/run_agent.py"),
        ("patches", 0, "target", "../../run_agent.py"),
        ("profile_sources", 0, "source", "/etc/passwd"),
        ("profile_sources", 0, "source", "../../outside"),
        ("profile_sources", 0, "target", "/tmp/config.yaml"),
        ("profile_sources", 0, "target", "../config.yaml"),
        ("profile_sources", 0, "mode", "0777"),
    ],
)
def test_lock_rejects_escaping_paths_and_unsafe_modes(
    section: str,
    index: int,
    field: str,
    unsafe: str,
) -> None:
    mutated = copy.deepcopy(LOCK)
    mutated[section][index][field] = unsafe

    with pytest.raises(compiler.CompileError):
        compiler._validate_lock(mutated)


def test_lock_rejects_non_hex_hash_and_unbound_archive_url() -> None:
    bad_hash = copy.deepcopy(LOCK)
    bad_hash["hermes"]["archive_sha256"] = "z" * 64
    with pytest.raises(compiler.CompileError, match="archive_sha256"):
        compiler._validate_lock(bad_hash)

    bad_url = copy.deepcopy(LOCK)
    bad_url["hermes"]["archive_url"] = "https://example.invalid/archive.tar.gz"
    with pytest.raises(compiler.CompileError, match="archive URL"):
        compiler._validate_lock(bad_url)


@pytest.mark.parametrize("kind", ["traversal", "absolute", "symlink", "hardlink", "device"])
def test_archive_extraction_rejects_unsafe_members(tmp_path: Path, kind: str) -> None:
    archive = tmp_path / "unsafe.tar.gz"
    root = LOCK["hermes"]["archive_root"]
    with tarfile.open(archive, "w:gz") as bundle:
        if kind == "traversal":
            member = tarfile.TarInfo(f"{root}/../../escape")
        elif kind == "absolute":
            member = tarfile.TarInfo("/tmp/escape")
        elif kind == "symlink":
            member = tarfile.TarInfo(f"{root}/link")
            member.type = tarfile.SYMTYPE
            member.linkname = "/etc/passwd"
        elif kind == "hardlink":
            member = tarfile.TarInfo(f"{root}/hardlink")
            member.type = tarfile.LNKTYPE
            member.linkname = f"{root}/target"
        else:
            member = tarfile.TarInfo(f"{root}/device")
            member.type = tarfile.CHRTYPE
        if member.isreg():
            payload = b"unsafe"
            member.size = len(payload)
            bundle.addfile(member, io.BytesIO(payload))
        else:
            bundle.addfile(member)

    with pytest.raises(compiler.CompileError):
        compiler._safe_extract(archive, tmp_path / "output", root)


def test_profile_template_and_runtime_state_are_exact(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    bootstrap._validate_profile(profile, CONTRACT, template=True)

    for name in CONTRACT["profile"]["required_runtime_entries"]:
        (profile / name).write_text("{}\n", encoding="utf-8")
    for name in ("logs", "sessions", "platforms"):
        (profile / name).mkdir()
    (profile / "state.db").write_bytes(b"sqlite fixture")
    bootstrap._validate_profile(profile, CONTRACT, template=False)


@pytest.mark.parametrize(
    ("section", "field", "unsafe"),
    [
        ("memory", "memory_enabled", True),
        ("memory", "memory_enabled", 0),
        ("memory", "user_profile_enabled", True),
        ("memory", "nudge_interval", 10),
        ("memory", "nudge_interval", False),
        ("skills", "creation_nudge_interval", 10),
        ("skills", "creation_nudge_interval", False),
    ],
)
def test_profile_rejects_learning_policy_drift(
    tmp_path: Path,
    section: str,
    field: str,
    unsafe: object,
) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    config_path = profile / "config.yaml"
    config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    config[section][field] = unsafe
    config_path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8")

    with pytest.raises(bootstrap.NoSendContractError):
        bootstrap._load_profile_config(profile, CONTRACT)
    with pytest.raises(verifier.VerificationError):
        verifier._validate_profile_config(profile, CONTRACT)


def test_profile_rejects_smart_model_routing(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    config_path = profile / "config.yaml"
    config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    config["smart_model_routing"] = {
        "enabled": True,
        "cheap_model": {"provider": "openrouter", "model": "unreviewed/model"},
    }
    config_path.write_text(yaml.safe_dump(config, sort_keys=False), encoding="utf-8")

    with pytest.raises(bootstrap.NoSendContractError, match="smart-model routing"):
        bootstrap._load_profile_config(profile, CONTRACT)
    with pytest.raises(verifier.VerificationError, match="smart-model routing"):
        verifier._validate_profile_config(profile, CONTRACT)


def test_real_cloudsql_producer_feeds_gcp_plugin_with_private_bounded_excerpts(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_cloudsql_runtime_test")
    plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_db_plugin_test")
    query = "Challenge one exact claim body with its evidence source and offer a narrower revision."
    result = {
        "artifact": "teleo_cloudsql_canonical_kb_context",
        "backend": "cloudsql:teleo-pgvector-standby/teleo_canonical/public+kb_stage",
        "query": query,
        "terms": ["challenge", "claim", "evidence"],
        "claims": [
            {
                "id": "11111111-1111-1111-1111-111111111111",
                "type": "fact",
                "text": "Exact canonical claim body.",
                "status": "open",
                "confidence": 0.6,
                "tags": ["test"],
                "score": 3,
                "evidence": [
                    {
                        "source_id": "22222222-2222-2222-2222-222222222222",
                        "role": "supports",
                        "source_type": "document",
                        "excerpt": "Exact linked evidence excerpt.",
                    }
                ],
                "edges": [],
            }
        ],
        "context_rows": [],
        "hit_count_total": 1,
        "hits": [],
        "privacy": "private agent context mode",
    }
    args = SimpleNamespace(command="context", canonical_db="teleo_canonical")
    payload = cloudsql._enrich_context(args, query, result)
    contract_ids = {item["id"] for item in payload["operational_contracts"]}
    assert "claim_evidence_challenge" in contract_ids
    assert 'Exact claim body: "Exact canonical claim body."' in payload["compiled_response"]
    assert 'Exact evidence excerpt: "Exact linked evidence excerpt."' in payload["compiled_response"]

    observed = {}

    def fake_runner(command, **kwargs):
        observed.update({"command": command, **kwargs})
        return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")

    snapshot = plugin._load_database_snapshot(query, hermes_home=profile, runner=fake_runner)
    assert snapshot["status"] == "ok"
    assert snapshot["requires_database_truth"] is True
    assert "Exact canonical claim body." in snapshot["context"]
    assert "Exact linked evidence excerpt." in snapshot["context"]
    assert observed["command"][:3] == [str(profile / "bin" / "teleo-kb"), "context", query]
    assert "--include-excerpts" in observed["command"]
    assert "--local" not in observed["command"]
    assert observed["check"] is False


def test_gcp_runtime_identity_parser_accepts_only_the_exact_staging_account(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_identity_parser_test")
    valid = (
        b'{"expected_service_account":"sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",'
        b'"schema":"livingip.leocleanGcpRuntimeIdentity.v1"}'
    )
    assert cloudsql._parse_runtime_identity(valid) == cloudsql.STAGING_SERVICE_ACCOUNT
    for invalid in (
        b'{"expected_service_account":"unknown@teleo-501523.iam.gserviceaccount.com","schema":"livingip.leocleanGcpRuntimeIdentity.v1"}',
        b'{"expected_service_account":"sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com","schema":"wrong"}',
        b'{"expected_service_account":"sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com","schema":"livingip.leocleanGcpRuntimeIdentity.v1","extra":true}',
        b'{"expected_service_account":"one","expected_service_account":"two","schema":"livingip.leocleanGcpRuntimeIdentity.v1"}',
    ):
        with pytest.raises(SystemExit, match=r"Runtime identity manifest .*invalid"):
            cloudsql._parse_runtime_identity(invalid)


def test_packaged_current_main_contract_keeps_review_only_candidate_out_of_database_intents(
    tmp_path: Path,
) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_candidate_contract_test")
    ids = {
        item["id"]
        for item in cloudsql.contracts.operational_contracts(
            "Turn the surviving claim into a review-only proposal: exact body, confidence, falsifier, required "
            "sources, and links to claims that support or contradict it. Do not stage or write anything."
        )
    }
    assert ids == {"reply_budget"}


def test_sql_free_contract_compiler_matches_current_main_for_supported_modes(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_contract_parity_test")
    current_main = load_module(ROOT / "hermes-agent" / "leoclean-bin" / "kb_tool.py", "kb_tool_contract_parity_test")
    prompts = [
        "Summarize this in under 80 words.",
        (
            "Turn the surviving claim into a review-only proposal: exact body, confidence, falsifier, required "
            "sources, and links to claims that support or contradict it. Do not stage or write anything."
        ),
        "Challenge one exact claim body with its evidence source and offer a narrower revision.",
    ]
    for prompt in prompts:
        assert cloudsql.contracts.operational_contracts(prompt) == current_main.operational_contracts(prompt)

    source_path = profile / "bin" / "operational_contracts.py"
    source = source_path.read_text(encoding="utf-8")
    ast.parse(source)
    forbidden = re.compile(r"\b(?:delete|insert|psql|select|socket|subprocess|truncate|update)\b", re.I)
    assert not forbidden.search(source)


def test_gcp_plugin_preserves_current_main_subject_anchors(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_subject_plugin_test")
    query = "How should we revise that conclusion?"
    history = [
        {"role": "user", "content": "Compare LivingIP ownership models."},
        {"role": "assistant", "content": "I found a claim to inspect."},
        {"role": "user", "content": query},
    ]
    observed = {}

    def fake_runner(command, **kwargs):
        observed.update({"command": command, **kwargs})
        payload = {
            "claims": [],
            "compiled_response": None,
            "context_rows": [],
            "operational_contracts": [{"id": "reply_budget", "hard_max_words": 220}],
            "runtime_contract_mode": "general",
        }
        return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")

    snapshot = plugin._load_database_snapshot(
        query,
        conversation_history=history,
        hermes_home=profile,
        runner=fake_runner,
    )
    command = observed["command"]
    assert snapshot["status"] == "ok"
    assert command[command.index("--retrieval-query") + 1] == query
    anchors = [command[index + 1] for index, item in enumerate(command) if item == "--retrieval-anchor"]
    assert "LivingIP" in anchors
    assert "--include-excerpts" in command


def test_gcp_plugin_fails_closed_for_unimplemented_database_lifecycle_contract(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_gap_plugin_test")
    query = "Which database proposal is currently approved or applied?"

    def fake_runner(command, **_kwargs):
        payload = {
            "claims": [],
            "compiled_response": None,
            "context_rows": [],
            "operational_contracts": [{"id": "reply_budget", "hard_max_words": 220}],
            "runtime_contract_mode": "general",
        }
        return subprocess.CompletedProcess(command, 0, stdout=json.dumps(payload), stderr="")

    snapshot = plugin._load_database_snapshot(query, hermes_home=profile, runner=fake_runner)
    assert snapshot["status"] == "error"
    assert snapshot["requires_database_truth"] is True
    assert "read-only current database contract lookup failed" in snapshot["context"]


def test_gcp_plugin_registers_callbacks_with_the_gcp_loader_installed(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    plugin = load_module(profile / "plugins" / "leo-db-context" / "__init__.py", "leoclean_registered_plugin_test")
    registered = {}

    class Context:
        def register_hook(self, name, callback):
            registered[name] = callback

    plugin.register(Context())
    assert plugin.base._load_database_snapshot is plugin._load_database_snapshot
    assert set(registered) == {"post_llm_call", "pre_llm_call"}


def test_gcp_cloudsql_runtime_backfills_subject_claim_without_losing_raw_results(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_anchor_backfill_test")
    query = "How should we revise that conclusion?"
    raw_claim = {"id": "raw", "text": "A cross-domain claim", "tags": [], "evidence": [], "edges": []}
    subject_claim = {
        "id": "subject",
        "text": "LivingIP ownership claim",
        "tags": ["LivingIP"],
        "evidence": [],
        "edges": [],
    }
    calls = []

    def fake_query(_args, semantic_query, _limit):
        calls.append(semantic_query)
        claims = [subject_claim] if semantic_query == "LivingIP" else [raw_claim]
        return {
            "artifact": "teleo_cloudsql_canonical_kb_context",
            "claims": claims,
            "context_rows": [],
            "hit_count_total": len(claims),
            "hits": [
                {"source_table": "public.claims", "row_id": claim["id"], "excerpt": claim["text"]} for claim in claims
            ],
            "privacy": "private agent context mode: short excerpts included",
            "query": semantic_query,
        }

    cloudsql._BASE_QUERY_ROWS = fake_query
    args = SimpleNamespace(
        command="context",
        canonical_db="teleo_canonical",
        retrieval_anchor=["LivingIP"],
        retrieval_query=query,
    )
    payload = cloudsql._query_rows(args, query, 4)
    assert calls == [query, "LivingIP"]
    assert [claim["id"] for claim in payload["claims"]] == ["subject", "raw"]
    assert payload["query"] == query
    assert payload["retrieval_query_sha256"]
    assert payload["retrieval_anchors_sha256"]


def test_gcp_cloudsql_parser_accepts_only_bounded_contextual_retrieval_flags(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    cloudsql = load_module(profile / "bin" / "cloudsql_memory_tool.py", "leoclean_context_parser_test")
    monkeypatch.setattr(
        cloudsql.sys,
        "argv",
        [
            "cloudsql_memory_tool.py",
            "context",
            "revise that",
            "--retrieval-query",
            "revise that",
            "--retrieval-anchor",
            "LivingIP",
            "--include-excerpts",
        ],
    )
    parsed = cloudsql._parse_args()
    assert parsed.query == "revise that"
    assert parsed.retrieval_query == "revise that"
    assert parsed.retrieval_anchor == ["LivingIP"]
    assert parsed.include_excerpts is True

    monkeypatch.setattr(
        cloudsql.sys,
        "argv",
        [
            "cloudsql_memory_tool.py",
            "context",
            "revise that",
            "--retrieval-query",
            "first",
            "--retrieval-query",
            "second",
        ],
    )
    with pytest.raises(SystemExit, match="Only one retrieval query"):
        cloudsql._parse_args()


@pytest.mark.parametrize("entry", [".env", "BOOT.md", "gateway.json", "hooks", "processes.json", "secrets"])
def test_runtime_profile_rejects_startup_control_entries(tmp_path: Path, entry: str) -> None:
    bootstrap = load_bootstrap()
    profile = copy_profile(tmp_path, bootstrap)
    for name in CONTRACT["profile"]["required_runtime_entries"]:
        (profile / name).write_text("{}\n", encoding="utf-8")
    target = profile / entry
    if "." in entry and entry != "hooks":
        target.write_text("unsafe\n", encoding="utf-8")
    else:
        target.mkdir()

    with pytest.raises(bootstrap.NoSendContractError):
        bootstrap._validate_profile(profile, CONTRACT, template=False)


def test_profile_rejects_nested_mutation_and_symlink(tmp_path: Path) -> None:
    bootstrap = load_bootstrap()
    mutated = copy_profile(tmp_path / "mutated", bootstrap)
    skill = mutated / "skills" / "teleo-kb-bridge" / "SKILL.md"
    skill.write_text(skill.read_text(encoding="utf-8") + "\nunsafe\n", encoding="utf-8")
    with pytest.raises(bootstrap.NoSendContractError, match="immutable profile tree drifted"):
        bootstrap._validate_profile(mutated, CONTRACT, template=True)

    linked = copy_profile(tmp_path / "linked", bootstrap)
    (linked / "skills" / "escape").symlink_to("/tmp")
    with pytest.raises(bootstrap.NoSendContractError, match="symlink"):
        bootstrap._validate_profile(linked, CONTRACT, template=True)


def forbidden_environment_cases() -> list[str]:
    policy = CONTRACT["environment"]
    cases = list(policy["forbidden_exact"])
    cases.extend(f"{prefix}INJECTED" for prefix in policy["forbidden_prefixes"])
    cases.extend(f"INJECTED{suffix}" for suffix in policy["forbidden_suffixes"])
    return sorted(set(cases))


@pytest.mark.parametrize("name", forbidden_environment_cases())
def test_each_forbidden_environment_shape_fails_closed(monkeypatch: pytest.MonkeyPatch, name: str) -> None:
    bootstrap = load_bootstrap()
    monkeypatch.setattr(bootstrap.os, "environ", {name: "secret-shaped-sentinel"})

    with pytest.raises(bootstrap.NoSendContractError) as caught:
        bootstrap._validate_environment(CONTRACT)

    assert name in str(caught.value)
    assert "secret-shaped-sentinel" not in str(caught.value)


def test_only_reviewed_provider_and_runtime_environment_are_accepted(monkeypatch: pytest.MonkeyPatch) -> None:
    bootstrap = load_bootstrap()
    monkeypatch.setattr(
        bootstrap.os,
        "environ",
        {
            "OPENROUTER_API_KEY": "secret-shaped-sentinel",
            "HERMES_HOME": "/profile",
            "PYTHONNOUSERSITE": "1",
        },
    )
    bootstrap._validate_environment(CONTRACT)


def test_sealed_tool_map_rejects_registration_and_overwrite() -> None:
    bootstrap = load_bootstrap()
    tools = bootstrap._SealedToolMap({"terminal": object()})

    with pytest.raises(bootstrap.NoSendContractError):
        tools["send_message"] = object()
    with pytest.raises(bootstrap.NoSendContractError):
        tools["terminal"] = object()
    with pytest.raises(bootstrap.NoSendContractError):
        tools.update({"process": object()})
    assert list(tools) == ["terminal"]


def test_restricted_terminal_uses_exact_argv_and_sanitized_environment(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    bootstrap = load_bootstrap()
    profile = tmp_path / "profile"
    wrapper = profile / "bin" / "teleo-kb"
    wrapper.parent.mkdir(parents=True)
    wrapper.write_text("#!/bin/sh\n", encoding="utf-8")
    wrapper.chmod(0o755)
    observed = {}

    def fake_run(argv, **kwargs):
        observed.update({"argv": argv, **kwargs})
        return subprocess.CompletedProcess(argv, 0, stdout="ok", stderr="")

    monkeypatch.setattr(bootstrap.subprocess, "run", fake_run)
    monkeypatch.setenv("TELEGRAM_BOT_TOKEN", "must-not-propagate")
    monkeypatch.setenv("PGPASSWORD", "must-not-propagate")
    result = json.loads(
        bootstrap._restricted_terminal_handler(
            profile,
            set(CONTRACT["toolset"]["terminal_commands"]),
            {"command": "teleo-kb search 'claim; literal' --limit 2", "timeout": 500},
        )
    )

    assert observed["argv"] == [str(wrapper.resolve()), "search", "claim; literal", "--limit", "2"]
    assert observed["cwd"] == profile
    assert observed["timeout"] == 120
    assert observed["env"] == {
        "HOME": str(profile / "state"),
        "HERMES_HOME": str(profile),
        "LANG": "C",
        "LC_ALL": "C",
        "PATH": "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin",
    }
    assert result == {"error": None, "exit_code": 0, "output": "ok", "truncated": False}


@pytest.mark.parametrize(
    "tool_args",
    [
        {"command": "sh -c whoami"},
        {"command": "teleo-kb apply-proposal 123"},
        {"command": "teleo-kb status ; whoami"},
        {"command": "teleo-kb status --user=postgres"},
        {"command": "teleo-kb status\nwhoami"},
        {"command": "teleo-kb status", "workdir": "/tmp"},
        {"command": "teleo-kb status", "timeout": "not-an-integer"},
    ],
)
def test_restricted_terminal_rejects_bypasses_without_subprocess(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    tool_args: dict[str, object],
) -> None:
    bootstrap = load_bootstrap()
    profile = tmp_path / "profile"
    wrapper = profile / "bin" / "teleo-kb"
    wrapper.parent.mkdir(parents=True)
    wrapper.write_text("#!/bin/sh\n", encoding="utf-8")
    wrapper.chmod(0o755)

    def forbidden_call(*_args, **_kwargs):
        pytest.fail("subprocess must not be called for a rejected terminal request")

    monkeypatch.setattr(bootstrap.subprocess, "run", forbidden_call)
    result = json.loads(
        bootstrap._restricted_terminal_handler(
            profile,
            set(CONTRACT["toolset"]["terminal_commands"]),
            tool_args,
        )
    )
    assert result["exit_code"] == 126


def test_verifier_preserves_virtual_environment_python_symlink(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    capsys: pytest.CaptureFixture[str],
) -> None:
    interpreter = tmp_path / "venv" / "bin" / "python"
    interpreter.parent.mkdir(parents=True)
    interpreter.symlink_to(sys.executable)
    observed = {}

    def fake_verify_artifact(**kwargs):
        observed.update(kwargs)
        return {"schema": verifier.RECEIPT_SCHEMA, "status": "pass"}

    monkeypatch.setattr(verifier, "verify_artifact", fake_verify_artifact)
    monkeypatch.setattr(
        sys,
        "argv",
        [
            "verify",
            "--repo-root",
            str(ROOT),
            "--artifact",
            str(tmp_path / "artifact"),
            "--runtime-python",
            str(interpreter),
        ],
    )

    assert verifier.main() == 0
    assert observed["runtime_python"] == interpreter.absolute()
    assert observed["runtime_python"] != interpreter.resolve()
    assert json.loads(capsys.readouterr().out)["status"] == "pass"


def test_ci_runs_the_release_canary_and_retains_its_receipt() -> None:
    workflow = (ROOT / ".github" / "workflows" / "ci.yml").read_text(encoding="utf-8")
    wrapper = (ROOT / "scripts" / "run_leoclean_nosend_runtime_canary.sh").read_text(encoding="utf-8")

    assert "leoclean-nosend-runtime:" in workflow
    assert "scripts/run_leoclean_nosend_runtime_canary.sh" in workflow
    assert 'python-version: "3.11.9"' in workflow
    assert "uv==0.9.30" in workflow
    assert ".crabbox-results/leoclean-nosend-runtime.json" in workflow
    assert "--development" not in wrapper
    assert "--allow-dirty" not in wrapper
    assert 'rm -f "$receipt"' in wrapper
    assert 'repeat_artifact="$temporary/artifact-repeat"' in wrapper
    assert "first != second" in wrapper
    assert "compile-primary.json" in wrapper
    assert '--runtime-python "$runtime_venv/bin/python"' in wrapper
    assert 'receipt.get("verification_mode") != "release"' in wrapper