teleo-infrastructure/tests/test_gcp_generated_db_blind_claim_canary.py

from __future__ import annotations

from scripts import leo_tool_trace
from scripts.run_gcp_generated_db_blind_claim_canary import (
    DISCOVERY_SUBCOMMANDS,
    EXPECTED_CLAIM_ID,
    EXPECTED_SOURCE_IDS,
    EXPECTED_SUBCOMMANDS,
    PROMPT,
    reply_outcomes,
    sanitize_gateway,
    should_reexec_in_hermes,
    summarize_wrapper_tool_proof,
    trace_facts,
    transcript_events_to_messages,
)


def retrieval_content(*row_ids: str) -> str:
    return "\n".join(
        [
            "schema: livingip.teleoKbRetrievalReceipt.v1",
            f"semantic context SHA-256: `{'1' * 64}`",
            f"artifact state SHA-256: `{'2' * 64}`",
            "DB read consistency: `stable_wal_marker`",
            *row_ids,
        ]
    )


def test_blind_prompt_contains_no_row_identifier() -> None:
    assert EXPECTED_CLAIM_ID not in PROMPT
    assert not EXPECTED_SOURCE_IDS.intersection(PROMPT.split())


def test_blind_runner_reexecs_when_venv_python_symlinks_to_system_python(tmp_path) -> None:
    system_python = tmp_path / "system-python"
    system_python.touch()
    hermes_python = tmp_path / "hermes-python"
    hermes_python.symlink_to(system_python)

    assert should_reexec_in_hermes(system_python, hermes_python) is True
    assert should_reexec_in_hermes(hermes_python, hermes_python) is False


def test_transcript_proves_three_receipted_read_calls_and_expected_rows() -> None:
    source_ids = sorted(EXPECTED_SOURCE_IDS)
    events = []
    for index, (subcommand, content) in enumerate(
        (
            ("search", retrieval_content(EXPECTED_CLAIM_ID)),
            ("show", retrieval_content(EXPECTED_CLAIM_ID)),
            ("evidence", retrieval_content(EXPECTED_CLAIM_ID, *source_ids)),
        )
    ):
        call_id = f"call-{index}"
        events.extend(
            [
                {
                    "phase": "call",
                    "tool_call_id": call_id,
                    "tool_name": "terminal",
                    "arguments": {"command": f"teleo-kb {subcommand} query"},
                },
                {
                    "phase": "result",
                    "tool_call_id": call_id,
                    "tool_name": "terminal",
                    "content": content,
                },
            ]
        )

    trace = leo_tool_trace.extract_kb_tool_trace(transcript_events_to_messages(events))
    subcommands, row_ids, receipt_count = trace_facts(trace)

    assert subcommands >= EXPECTED_SUBCOMMANDS
    assert subcommands & DISCOVERY_SUBCOMMANDS
    assert EXPECTED_CLAIM_ID in row_ids
    assert row_ids >= EXPECTED_SOURCE_IDS
    assert receipt_count == 3
    assert trace["database_tool_calls_read_only"] is True


def test_gateway_sanitization_hashes_events_without_retaining_them() -> None:
    secret = "raw-secret-that-must-not-survive"
    sanitized = sanitize_gateway(
        {
            "reply": "safe reply",
            "transcript_tool_trace": {
                "event_count": 1,
                "events": [{"content": secret}],
            },
        }
    )

    assert sanitized["reply"] == "safe reply"
    assert sanitized["transcript_tool_trace"]["event_count"] == 1
    assert sanitized["transcript_tool_trace"]["raw_events_retained"] is False
    assert len(sanitized["transcript_tool_trace"]["events_sha256"]) == 64
    assert secret not in str(sanitized)


def test_wrapper_summary_drops_raw_arguments_but_keeps_database_binding() -> None:
    secret_argument = "do-not-retain-this-query"
    summarized = summarize_wrapper_tool_proof(
        {
            "event_count": 2,
            "invocation_count": 1,
            "complete_start_end_pairing": True,
            "all_bound_to_supplied_target": True,
            "database_read_only_required": True,
            "default_read_only_required": True,
            "all_completed_successfully": True,
            "invocations": [
                {
                    "argv": ["search", secret_argument],
                    "container": "teleo-prod-1",
                    "database": "teleo_clone_test",
                    "database_identity": {"ssl": True},
                    "returncode": 0,
                }
            ],
        }
    )

    assert summarized["invocations"][0]["subcommand"] == "search"
    assert summarized["invocations"][0]["argument_count"] == 1
    assert summarized["invocations"][0]["database"] == "teleo_clone_test"
    assert summarized["raw_argv_retained"] is False
    assert secret_argument not in str(summarized)


def test_reply_outcomes_accepts_reasoned_alternative_decomposition() -> None:
    reply = """
    The support is thin: the two sources do not establish the full legal conclusion.
    I would split it into candidate claims about acquisition due diligence and valuation risk,
    plus a separate representations and warranties claim. Please provide the filing or URL,
    then review and confirm the candidates before any apply step or database write goes live.
    """

    assert all(reply_outcomes(reply).values())


def test_reply_outcomes_rejects_assertion_without_decomposition_or_iteration() -> None:
    reply = "The claim is weak. I propose a better claim."

    outcomes = reply_outcomes(reply)

    assert outcomes["challenges_weak_support"] is False
    assert outcomes["decomposes_the_bundled_legal_claim"] is False
    assert outcomes["asks_for_user_iteration_before_live"] is False


def test_reply_outcomes_rejects_forbidden_stakeholder_alias() -> None:
    otherwise_valid = """
    The evidence is thin. I propose candidate claims about acquisition due diligence and
    representations and warranties. Please review the source URL before any apply step goes live.
    Next Cory-style follow-up: choose the claim framing.
    """

    assert reply_outcomes(otherwise_valid)["uses_only_m3taversal_handle"] is False