arborist/tests/test_operation_sandwich.py

"""Ticket #000056 — Operation Sandwich.

The bright line as executable invariants:

- the grounded core is English-only: `answer_text` and `audit_mode`
  are produced exactly as a same-sources English run; the Spanish
  text lives in additive `display_*` keys and is NEVER the verifier's
  input;
- query MT is a retrieval transform: it binds into
  `RetrievalPlan` (→ `retrieval_plan_hash` → run-DAG), NOT
  `question_hash` / `governance_policy_hash` — with zero hash churn
  on every non-MT run;
- default OFF, gated under the Phase-0 flag, graceful-degrades when
  the `[mt]` extra is absent.

Deterministic: a `StubTranslator` is injected (`query(translator=)`),
so no network and no model weights.
"""

from __future__ import annotations

from collections.abc import Iterator

from arborist.qa.mt import StubTranslator
from arborist.qa.retrieval_plan import RetrievalPlan, retrieval_plan_hash

ES_Q = "¿Qué es el anarcocapitalismo?"
EN_Q = "What is anarcho-capitalism?"
EN_A = "Anarcho-capitalism is a political philosophy. [E1]\n"
ES_A = "El anarcocapitalismo es una filosofía política."


def _es_marker(text: str, src: str, tgt: str) -> str:
    """Deterministic, render-string-agnostic: es→en pins the question,
    en→es prefixes a marker so the test asserts *which side* got
    translated without coupling to exact rendered output."""
    if (src, tgt) == ("es", "en"):
        return EN_Q
    if (src, tgt) == ("en", "es"):
        return "[ES] " + text
    return text


def _stub():
    return StubTranslator(fn=_es_marker)


# --- Unit: RetrievalPlan MT binding + zero churn (criterion 3) -------------

def test_retrieval_plan_mt_fields_bind_and_zero_churn():
    base = RetrievalPlan(retrieval_keywords="", top_k=8, over_fetch=32,
                         max_context_chars=60000)
    # No MT → canonical() must be byte-identical to pre-#000056 (no "mt"
    # key) so every existing retrieval_plan_hash is unchanged.
    assert "mt" not in base.canonical()
    same = RetrievalPlan(retrieval_keywords="", top_k=8, over_fetch=32,
                         max_context_chars=60000)
    assert retrieval_plan_hash(base) == retrieval_plan_hash(same)

    mt = RetrievalPlan(retrieval_keywords="", top_k=8, over_fetch=32,
                       max_context_chars=60000, mt_engine="opus-mt-v1",
                       mt_manifest_hash="abc", source_lang="es")
    assert mt.canonical()["mt"] == {
        "engine": "opus-mt-v1", "manifest_hash": "abc", "source_lang": "es",
    }
    assert retrieval_plan_hash(mt) != retrieval_plan_hash(base)
    # Different engine identity → different plan hash (audit can tell
    # which MT engine pulled the English sources in).
    mt2 = RetrievalPlan(retrieval_keywords="", top_k=8, over_fetch=32,
                        max_context_chars=60000, mt_engine="other",
                        mt_manifest_hash="abc", source_lang="es")
    assert retrieval_plan_hash(mt2) != retrieval_plan_hash(mt)


# --- Unit: enabling MT cannot move the proof hashes (criterion 4 / §2#6) ---

def test_sandwich_hash_invariants():
    """The honest invariants (artifact over instruction —
    `governance_policy_hash` hashes the *whole* policy, keys.py:182):

    - `question_hash` UNCHANGED — the user's Spanish question is
      preserved as cache/question identity (the load-bearing bright
      line: MT never rewrites what the user asked);
    - `verifier_policy_hash` UNCHANGED — the verifier is byte-identical
      (crosslang flags are not in `_VERIFIER_POLICY_FIELDS`);
    - `governance_policy_hash` CHANGES — like every policy flag
      (quantifier, metacognition, soft-preflight), because it covers
      the whole policy dict. This is correct: a sandwich-on answer
      must NOT be served to a sandwich-off lookup. Cache partitions
      by config; it does not leak.
    """
    from arborist.qa.keys import (
        governance_policy_hash,
        question_hash,
        verifier_policy_hash,
    )
    from arborist.qa.query import DEFAULT_QUERY_POLICY

    off = dict(DEFAULT_QUERY_POLICY)
    on = dict(DEFAULT_QUERY_POLICY, crosslang_guard_enabled=True,
              crosslang_translate_enabled=True)
    assert verifier_policy_hash(off) == verifier_policy_hash(on)
    qh = lambda p: question_hash(  # noqa: E731
        ES_Q, mode=p.get("question_dedup", "equivalence_class"))
    assert qh(off) == qh(on)
    assert governance_policy_hash(off) != governance_policy_hash(on)


# --- Integration harness ---------------------------------------------------

def _ingest_one(tmp_path, text, title):
    from arborist.document import Document
    from arborist.ingest import ingest_source
    from arborist.source import Source
    from arborist.store import connect

    class FakeSource(Source):
        source_type = "test"

        def __init__(self, docs):
            self.docs = docs

        def iter_documents(self) -> Iterator[Document]:
            yield from self.docs

    shard = tmp_path / "shard.db"
    conn = connect(shard)
    try:
        ingest_source(
            conn,
            FakeSource([Document(uri="t://d", content=text,
                                 source_type="test", title=title)]),
        )
    finally:
        conn.close()
    return shard


def _run(tmp_path, *, translator, policy_extra):
    from arborist.qa.client import StubClient
    from arborist.qa.query import DEFAULT_QUERY_POLICY, query

    shard = _ingest_one(
        tmp_path,
        "Anarcho-capitalism is a political philosophy. " * 40,
        "Anarcho-capitalism",
    )
    return query(
        question=ES_Q,
        qa_db=tmp_path / "qa.db",
        chat_client=StubClient(answer=EN_A),
        model_id="stub",
        single_db=shard,
        translator=translator,
        policy=dict(DEFAULT_QUERY_POLICY, answer_mode="claim_lattice_pointer",
                    **policy_extra),
    )


def test_sandwich_core_is_english_display_is_spanish(tmp_path):
    r = _run(tmp_path, translator=_stub(),
             policy_extra=dict(crosslang_guard_enabled=True,
                               crosslang_translate_enabled=True))
    # Grounded core: the verified answer is ENGLISH and untranslated.
    assert r["status"] != "cross_language_unsupported"
    assert "Anarcho-capitalism" in r["answer_text"]
    assert not r["answer_text"].startswith("[ES] ")
    assert "audit_mode" in r
    # Display edge: additive Spanish rendering, banner-labelled.
    assert r["display_translated"] is True
    assert r["display_lang"] == "es" and r["display_source_lang"] == "en"
    assert r["display_answer"].startswith("[ES] ")
    assert "verific" in r["display_unverified_banner"].lower()
    # Bright line: the verifier's subject is the English core, never
    # the Spanish display string (no es-marker anywhere proof-side).
    assert r["answer_text"] != r["display_answer"]
    assert "[ES] " not in r["answer_text"]
    assert "[ES] " not in (r.get("verifier_input_text") or "")


def test_default_off_no_sandwich(tmp_path):
    # Guard on, translate OFF (default) → never engages; no display_*.
    r = _run(tmp_path, translator=_stub(),
             policy_extra=dict(crosslang_guard_enabled=True))
    assert "display_answer" not in r
    assert r.get("display_translated") is None


def test_graceful_degrade_when_mt_unavailable(tmp_path):
    # [mt] absent is modelled by an unavailable translator. The
    # sandwich must no-op to Phase-0 (no display_*), never raise.
    dead = StubTranslator({}, available=False)
    r = _run(tmp_path, translator=dead,
             policy_extra=dict(crosslang_guard_enabled=True,
                               crosslang_translate_enabled=True))
    assert "display_answer" not in r
    assert r["status"] != "cross_language_unsupported"  # had a content token


def test_sandwich_requires_a_non_english_signal(tmp_path):
    # An English question with the sandwich enabled never engages
    # (Phase-0 guard returns None → translator never consulted →
    # byte-identical English path).
    from arborist.qa.client import StubClient
    from arborist.qa.query import DEFAULT_QUERY_POLICY, query

    shard = _ingest_one(tmp_path,
                        "Anarcho-capitalism is a political philosophy. " * 40,
                        "Anarcho-capitalism")

    class _Boom:
        available = True
        engine_id = "boom"
        manifest_hash = "boom"

        def translate(self, *a, **k):
            raise AssertionError("translator consulted on an English query")

    r = query(
        question=EN_Q,
        qa_db=tmp_path / "qa.db",
        chat_client=StubClient(answer=EN_A),
        model_id="stub",
        single_db=shard,
        translator=_Boom(),
        policy=dict(DEFAULT_QUERY_POLICY, answer_mode="claim_lattice_pointer",
                    crosslang_guard_enabled=True,
                    crosslang_translate_enabled=True),
    )
    assert "display_answer" not in r