tests/test_escalate_live.py

"""Isolated test — just escalate(). Feeds it a constructed disagreement case
(real claim/evidence, but with deberta_verdict/llm_verdict manually forced to
disagree) so we can exercise the retry + adjudicator logic directly, rather
than waiting to stumble on a real disagreement by chance.

Requires GROQ_API_KEY and S2_API_KEY set in your environment.

Run from the veriscite/ root:
    python -m tests.test_escalate_live
"""

import asyncio
from app.graph.nodes import escalate

SAMPLE_CLAIM_CITATION = {
    "claim": "Attention mechanisms allow transformer models to capture "
             "long-range dependencies in sequential data without relying on recurrence",
    "citation_marker": "[1]",
    "reference_string": "Vaswani, A., Shazeer, N., Parmar, N., Uszkoreit, J., "
                         "Jones, L., Gomez, A. N., Kaiser, L., & Polosukhin, I. (2017). "
                         "Attention is all you need. In NeurIPS 2017.",
    "resolved_paper_id": "204e3073870fae3d05bcbc2f6a8e263d9b72e776",
    "evidence_text": (
        "The dominant sequence transduction models are based on complex "
        "recurrent or convolutional neural networks in an encoder-decoder "
        "configuration. We propose a new simple network architecture, the "
        "Transformer, based solely on attention mechanisms, dispensing with "
        "recurrence and convolutions entirely."
    ),
}

# Manually forced disagreement — real verdicts agreed (both SUPPORT) in the
# earlier chain test, so we force a mismatch here to actually exercise escalate.
FORCED_AUDIT = {
    "claim_citation": SAMPLE_CLAIM_CITATION,
    "winner_sentence": "We propose a new simple network architecture, the "
                        "Transformer, based solely on attention mechanisms, "
                        "dispensing with recurrence and convolutions entirely.",
    "attribution_available": True,
    "deberta_verdict": {"label": "SUPPORT", "confidence": 0.97, "source": "deberta"},
    "llm_verdict": {"label": "NOT_ENOUGH_INFO", "confidence": 0.6,
                     "source": "llm"},  # forced mismatch
    "agreement": False,
    "escalated": False,
    "escalation_retry_succeeded": None,
    "adjudicator_reasoning": None,
    "final_verdict": None,
    "attribution": None,
}


async def main():
    state = {
        "source_text": "",
        "claims": [SAMPLE_CLAIM_CITATION],
        "audits": [FORCED_AUDIT],
        "current_index": 0,
        "report": None,
    }
    result = await escalate(state)
    audit = result["audits"][0]

    print("AUDIT AFTER ESCALATE:")
    for k, v in audit.items():
        if k != "claim_citation":
            print(f"  {k}: {v}")


if __name__ == "__main__":
    asyncio.run(main())