arborist/tests/test_run_query_byte_identity.py

"""Byte-identity safety gate for the legacy-query() → run_query()
collapse (Phase 1 step 5 of #53).

The 4280-line legacy ``arborist.qa.query.query()`` computes a 9-dim
cache_key over inputs that include ``conversation_hash(messages)``
— so any drift in HOW the prompt is built (message order, whitespace,
optional gates) rotates every cache_key in the world. Phase 2 of the
collapse rewrites that prompt-build path inside ``run_query``; this
gate FREEZES the canonical bytes today so the rewrite can be byte-
checked against the captured reference.

What's pinned:
  1. Hash functions are deterministic on stable inputs (regression-
     locks the hash algos themselves).
  2. The claim-lattice prompt strings (CLAIM_LATTICE_SYSTEM_PROMPT
     and CLAIM_LATTICE_GROUNDING_REMINDER) hash to a fixed value.
     If a future edit drifts those prompt strings, the conversation_hash
     drifts and EVERY existing providence_cache record orphans on
     re-lookup. The pin is the early-warning signal.
  3. A synthetic prompt-build matching the shape ``run_query`` emits
     (sys + user with EVIDENCE + QUESTION + grounding_reminder) hashes
     to a fixed value — that's the gate Phase 2's rewrite must
     reproduce verbatim.

Fixture location: ``tests/fixtures/byte_identity/cache_key_*.json``.
Captured once via ``CAPTURE=1 pytest -v
tests/test_run_query_byte_identity.py`` — the test rewrites the
fixture file from the live values. Re-capture only when a deliberate
prompt-shape change is being made AND every prior cache record is
being treated as cold (cache invalidation event).
"""
from __future__ import annotations

import json
import os
from pathlib import Path

import pytest

from arborist.qa.keys import (
    conversation_hash,
    governance_policy_hash,
    model_profile_hash,
    question_hash,
)
from arborist.qa.prompts import (
    CLAIM_LATTICE_GROUNDING_REMINDER,
    CLAIM_LATTICE_SYSTEM_PROMPT,
)

FIXTURE_DIR = Path(__file__).parent / "fixtures" / "byte_identity"
SMOKE_QUESTIONS = [
    "when did the soviet union dissolve?",
    "where is mount kilimanjaro located?",
    "who painted the mona lisa?",
    "who were the original seven mercury astronauts?",
    "why did the dinosaurs go extinct?",
]
CAPTURE = os.environ.get("CAPTURE") == "1"


def _prompt_user_payload(question: str, evidence_text: str) -> str:
    """Mirror the user-payload shape ``run_query`` builds.

    See arborist/qa/corpus_query.py:173-177 — this MUST stay in sync
    with that template. If run_query's template changes, this string
    must change too, and the pinned hash gets re-captured.
    """
    return (
        f"EVIDENCE:\n\n{evidence_text}\n\n"
        f"QUESTION: {question}\n\n"
        f"{CLAIM_LATTICE_GROUNDING_REMINDER}"
    )


def _canonical_messages(question: str, evidence_text: str) -> list[dict]:
    """The 2-message array run_query passes to chat_completion."""
    return [
        {"role": "system", "content": CLAIM_LATTICE_SYSTEM_PROMPT},
        {"role": "user", "content": _prompt_user_payload(question, evidence_text)},
    ]


_SYNTHETIC_EVIDENCE = (
    "[E1 | Stub Title | 0000000000000000: "
    "this is the literal evidence span the test pins.]"
)


def _all_pins() -> dict:
    """Compute every pinned value from current code state."""
    out = {
        "prompt_strings": {
            "system_prompt_sha256": conversation_hash(
                [{"role": "system", "content": CLAIM_LATTICE_SYSTEM_PROMPT}]
            ),
            "grounding_reminder_sha256": conversation_hash(
                [{"role": "user", "content": CLAIM_LATTICE_GROUNDING_REMINDER}]
            ),
        },
        "per_question": [],
    }
    for q in SMOKE_QUESTIONS:
        msgs = _canonical_messages(q, _SYNTHETIC_EVIDENCE)
        out["per_question"].append({
            "question": q,
            "question_hash_strict": question_hash(q, mode="strict"),
            "question_hash_eq_class": question_hash(q, mode="equivalence_class"),
            "conversation_hash": conversation_hash(msgs),
        })
    out["governance_policy_hash_examples"] = {
        # A few well-known policy shapes. If anyone adds a field to one
        # of these dicts, the cache rotates.
        "empty_policy": governance_policy_hash({}),
        "temperature_0.1_max512": governance_policy_hash(
            {"temperature": 0.1, "max_tokens": 512}
        ),
        "answer_mode_claim_lattice": governance_policy_hash(
            {"answer_mode": "claim_lattice"}
        ),
    }
    out["model_profile_hash_examples"] = {
        "hermes": model_profile_hash(
            "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic"
        ),
        "qwen": model_profile_hash("Qwen3.6-27B-UD-Q4_K_XL.gguf"),
        "stub": model_profile_hash("stub"),
    }
    return out


def _fixture_path() -> Path:
    return FIXTURE_DIR / "claim_lattice.json"


def test_capture_or_compare():
    """Either capture the live hashes to the fixture file (CAPTURE=1)
    or assert the live values match the pinned fixture."""
    pins_now = _all_pins()
    fixture = _fixture_path()
    if CAPTURE:
        fixture.parent.mkdir(parents=True, exist_ok=True)
        fixture.write_text(json.dumps(pins_now, indent=2, sort_keys=True) + "\n")
        return
    if not fixture.exists():
        pytest.fail(
            f"fixture missing: {fixture}\n"
            "first-time setup: CAPTURE=1 pytest "
            f"{Path(__file__).name}"
        )
    pins_pinned = json.loads(fixture.read_text())
    if pins_now != pins_pinned:
        # Render a diff-style failure so the operator sees exactly
        # which hash drifted.
        diffs = []

        def walk(now, pinned, path):
            if isinstance(now, dict) and isinstance(pinned, dict):
                for k in sorted(set(now) | set(pinned)):
                    walk(now.get(k), pinned.get(k), f"{path}.{k}")
            elif isinstance(now, list) and isinstance(pinned, list):
                for i, (a, b) in enumerate(zip(now, pinned)):
                    walk(a, b, f"{path}[{i}]")
                if len(now) != len(pinned):
                    diffs.append(f"{path}: list length {len(now)} != {len(pinned)}")
            elif now != pinned:
                diffs.append(f"{path}: now={now!r}  pinned={pinned!r}")

        walk(pins_now, pins_pinned, "fixture")
        pytest.fail(
            "byte-identity drift — pinned cache-identity inputs no "
            "longer hash to the same values. If this is intentional "
            "(deliberate prompt or policy shape change with cache "
            "invalidation), re-capture: CAPTURE=1 pytest "
            f"{Path(__file__).name}\n\n"
            + "\n".join(diffs[:20])
        )


def test_hash_determinism_on_stable_inputs():
    """Hash functions stay deterministic across calls. Sanity check
    that pins the hash algorithm itself, not the input bytes."""
    assert conversation_hash([{"role": "system", "content": "x"}]) == conversation_hash(
        [{"role": "system", "content": "x"}]
    )
    assert governance_policy_hash({"a": 1, "b": 2}) == governance_policy_hash(
        {"b": 2, "a": 1}
    )  # canonical_json sorts keys; dict order must not matter
    # "what is the test?" has both a trailing '?' and an article 'the'
    # — equivalence_class strips both, strict preserves them, so the
    # canonical forms (and hashes) MUST differ.
    assert question_hash("what is the test?", mode="strict") != question_hash(
        "what is the test?", mode="equivalence_class"
    )