AegisSight-Monitor/tests/test_eu_factcheck.py

"""Testet die EU-Nachhak-Schleife im Faktencheck und sichert den CLI-Weg ab.

Laeuft ohne Netzzugriff und ohne Kosten, die Belegsuche und der Modellaufruf
sind durch Testdoubles ersetzt.

Aufruf aus dem Projektstamm:
    venv/bin/python tests/test_eu_factcheck.py

Seit der Belegtiefe-Pruefung (Block 2) traegt eine einzelne Quelle keine
Bestaetigung mehr. Die Faelle H und J liefern deshalb Belege aus mehreren
unabhaengigen Haeusern, sonst pruefen sie nicht mehr die Zuordnung, sondern
nur noch die Mindestzahl.
"""
import asyncio
import os
import sys

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src"))

from agents.factchecker import FactCheckerAgent  # noqa: E402
import agents.factchecker as fcmod  # noqa: E402
import agents.eu_researcher as eur  # noqa: E402
import agents.claude_client as cc  # noqa: E402
from agents.claude_client import _ai_backend_var, ClaudeUsage  # noqa: E402

ok = 0
fail = 0


def pruefe(name, bedingung, extra=""):
    global ok, fail
    if bedingung:
        ok += 1
        print("  OK    " + name)
    else:
        fail += 1
        print("  FEHL  " + name + " " + str(extra))


aufrufe = {"such": [], "modell": []}


SUCHQUELLE = {
    "headline": "Innenministerkonferenz beschliesst Fahrplan",
    "source": "example.org",
    "source_url": "https://example.org/a",
}


async def fake_plan(*, title, search_items, output_language="Deutsch",
                    max_queries=7, avoid_queries=None, label="Stuetzsuche",
                    nur_gegenproben=False):
    aufrufe["such"].append({
        "label": label,
        "items": list(search_items),
        "max": max_queries,
        "avoid": list(avoid_queries or []),
        "nur_gegenproben": nur_gegenproben,
    })
    u = ClaudeUsage(input_tokens=10, output_tokens=5, cost_usd=0.01)
    block = "\n\nBELEGE [" + label + "]\n[S1] Beleg | example.org | https://example.org/a"
    return eur.Belegsuche(block, u, [label + "-q1", label + "-q2"], [dict(SUCHQUELLE)])


antworten = []


async def fake_call_claude(prompt, tools="WebSearch,WebFetch", model=None,
                           raw_text=False, timeout=None):
    aufrufe["modell"].append({"tools": tools, "hat_belege": "BELEGE" in prompt})
    if not antworten:
        raise AssertionError("unerwarteter Modellaufruf")
    return antworten.pop(0), ClaudeUsage(input_tokens=100, output_tokens=50, cost_usd=0.05)


eur.plan_verification_searches = fake_plan
fcmod.call_claude = fake_call_claude  # CLI-Zweig nutzt das Modul-Symbol
cc.call_claude = fake_call_claude     # EU-Zweig importiert erst zur Laufzeit

fc = FactCheckerAgent()
ARTS = [
    {"headline": "Ceuta Grenze Tote gemeldet", "source": "tagesschau",
     "source_url": "https://tagesschau.de/ceuta", "content_original": "t"},
    {"headline": "Sanchez spricht ueber Angriff in Ceuta", "source": "elpais",
     "source_url": "https://elpais.com/ceuta", "content_original": "t"},
]

F_TOTE = "In Ceuta wurden 57 Tote gemeldet"
F_SANCHEZ = "Sanchez sprach in Ceuta von einem Angriff"
F_GRENZE = "Die Grenze in Ceuta wurde geschlossen"


def neu():
    aufrufe["such"].clear()
    aufrufe["modell"].clear()
    antworten.clear()


def json_fakten(eintraege):
    teile = []
    for claim, status, ev in eintraege:
        teile.append('{"claim": "%s", "status": "%s", "evidence": "%s"}' % (claim, status, ev))
    return "[" + ", ".join(teile) + "]"


print("\nA) EU-Modus, Nachhak-Runde stuft offene Behauptung hoch")
_ai_backend_var.set("bedrock")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "nur eine Quelle"),
    (F_SANCHEZ, "unconfirmed", "unklar"),
    (F_GRENZE, "confirmed", "belegt"),
]))
antworten.append(json_fakten([
    (F_TOTE, "confirmed", "laut [S1] https://example.org/a bestaetigt"),
]))
facts, usage = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
status = {f.get("claim", "")[:20]: f.get("status") for f in facts}
pruefe("A1 Erstsuche plant 7 Anfragen", aufrufe["such"][0]["max"] == 7)
pruefe("A2 Erstsuche kennt Titel und Schlagzeilen",
       len(aufrufe["such"][0]["items"]) == 3, aufrufe["such"][0]["items"])
pruefe("A3 Nachhak-Runde ausgeloest",
       len(aufrufe["such"]) == 2 and aufrufe["such"][1]["label"] == "Nachhak-Suche")
pruefe("A4 Nachhak nur fuer die 2 offenen Punkte",
       len(aufrufe["such"][1]["items"]) == 2, aufrufe["such"][1]["items"])
pruefe("A5 Nachhak meidet die alten Anfragen",
       aufrufe["such"][1]["avoid"] == ["Stuetzsuche-q1", "Stuetzsuche-q2"],
       aufrufe["such"][1]["avoid"])
pruefe("A6 belegter Fakt hochgestuft", status.get(F_TOTE[:20]) == "confirmed", status)
pruefe("A7 unbelegter Fakt bleibt offen", status.get(F_SANCHEZ[:20]) == "unconfirmed", status)
pruefe("A8 bestaetigter Fakt unangetastet", status.get(F_GRENZE[:20]) == "confirmed", status)
ev = [f["evidence"] for f in facts if f["claim"].startswith("In Ceuta")][0]
pruefe("A9 Nachrecherche mit URL in der Evidenz",
       "Nachrecherche" in ev and "https://example.org/a" in ev, ev[:160])
pruefe("A10 kein Faktenverlust", len(facts) == 3, len(facts))
pruefe("A11 beide Modellaufrufe ohne Werkzeuge",
       [m["tools"] for m in aufrufe["modell"]] == [None, None], aufrufe["modell"])
pruefe("A12 beide Modellaufrufe mit Belegblock",
       all(m["hat_belege"] for m in aufrufe["modell"]))
pruefe("A13 Kosten aufsummiert", round(usage.cost_usd, 3) == 0.12, usage.cost_usd)

print("\nB) Ist alles belegt, folgt keine Nachbelegung sondern eine Belastungsprobe")
neu()
antworten.append(json_fakten([
    (F_TOTE, "confirmed", "belegt"),
    (F_GRENZE, "confirmed", "belegt"),
]))
antworten.append(json_fakten([(F_TOTE, "confirmed", "haelt der Gegenprobe stand")]))
asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
pruefe("B1 keine Nachbelegung, sondern Belastungsprobe",
       [a["label"] for a in aufrufe["such"]] == ["Stuetzsuche", "Belastungsprobe"],
       [a["label"] for a in aufrufe["such"]])
pruefe("B2 Belastungsprobe sucht nur Gegenproben",
       aufrufe["such"][1]["nur_gegenproben"] is True)
pruefe("B3 zwei Modellaufrufe", len(aufrufe["modell"]) == 2, len(aufrufe["modell"]))

print("\nC) Ein einzelner offener Punkt loest keine Nachbelegung aus")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_GRENZE, "confirmed", "belegt"),
]))
antworten.append(json_fakten([(F_GRENZE, "confirmed", "haelt")]))
asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
pruefe("C1 Schwelle von zwei offenen Punkten greift weiterhin",
       aufrufe["such"][1]["label"] == "Belastungsprobe",
       [a["label"] for a in aufrufe["such"]])
pruefe("C2 nur der belegte Punkt wird angegriffen",
       aufrufe["such"][1]["items"] == [F_GRENZE], aufrufe["such"][1]["items"])

print("\nD) Regression, CLI-Weg unveraendert")
_ai_backend_var.set("cli")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_SANCHEZ, "unconfirmed", "unklar"),
]))
facts_d, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
pruefe("D1 CLI macht keine Belegsuche", len(aufrufe["such"]) == 0, aufrufe["such"])
pruefe("D2 CLI genau ein Modellaufruf", len(aufrufe["modell"]) == 1, len(aufrufe["modell"]))
pruefe("D3 CLI behaelt die WebSearch-Werkzeuge",
       aufrufe["modell"][0]["tools"] == "WebSearch,WebFetch", aufrufe["modell"])
pruefe("D4 CLI bekommt keinen Belegblock", aufrufe["modell"][0]["hat_belege"] is False)
pruefe("D5 CLI liefert Fakten", len(facts_d) == 2, len(facts_d))

print("\nE) EU-Modus, inkrementelle Pruefung")
_ai_backend_var.set("bedrock")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_SANCHEZ, "unconfirmed", "unklar"),
]))
antworten.append(json_fakten([
    (F_SANCHEZ, "confirmed", "[S1] https://example.org/a belegt die Aussage"),
]))
alt = [{"claim": "Alter offener Punkt aus Ceuta", "status": "unconfirmed", "evidence": "e"}]
facts_e, _ = asyncio.run(fc.check_incremental("Ceuta", ARTS, alt, "adhoc", "Deutsch"))
pruefe("E1 inkrementell nutzt die Belegsuche", len(aufrufe["such"]) >= 1, aufrufe["such"])
pruefe("E2 alte offene Punkte fliessen in die Suche",
       any("Alter offener Punkt" in i for i in aufrufe["such"][0]["items"]),
       aufrufe["such"][0]["items"])
pruefe("E3 inkrementell hakt nach", len(aufrufe["such"]) == 2, len(aufrufe["such"]))
pruefe("E4 inkrementelle Hochstufung greift",
       any(f["claim"].startswith("Sanchez") and f["status"] == "confirmed" for f in facts_e),
       [(f["claim"][:25], f["status"]) for f in facts_e])

print("\nF) Regression, CLI-Weg inkrementell unveraendert")
_ai_backend_var.set("cli")
neu()
antworten.append(json_fakten([(F_TOTE, "unconfirmed", "unklar")]))
asyncio.run(fc.check_incremental("Ceuta", ARTS, alt, "adhoc", "Deutsch"))
pruefe("F1 CLI inkrementell ohne Belegsuche", len(aufrufe["such"]) == 0, aufrufe["such"])
pruefe("F2 CLI inkrementell mit Werkzeugen",
       aufrufe["modell"][0]["tools"] == "WebSearch,WebFetch", aufrufe["modell"])

print("\nG) EU-Modus, Nachhak-Runde faellt aus, Fakten bleiben erhalten")
_ai_backend_var.set("bedrock")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_SANCHEZ, "unconfirmed", "unklar"),
]))
antworten.append("Das ist kein JSON, sondern Fliesstext.")
facts_g, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
pruefe("G1 Fakten ueberleben eine unbrauchbare Nachhak-Antwort", len(facts_g) == 2, len(facts_g))
pruefe("G2 Status bleibt offen",
       all(f["status"] in ("unconfirmed", "unverified") for f in facts_g),
       [(f["claim"][:25], f["status"]) for f in facts_g])

print("\nH) EU-Modus, Gruppenpruefung nutzt die kleinere Suchmenge")
from config import EU_VERIFY_GROUP_QUERIES, EU_VERIFY_GROUP_FOLLOWUP_QUERIES  # noqa: E402

_ai_backend_var.set("bedrock")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_SANCHEZ, "unconfirmed", "unklar"),
]))
antworten.append(json_fakten([(F_TOTE, "confirmed",
    "[S1] https://example.org/a und [S2] https://beispiel.de/b belegen")]))
facts_h, _ = asyncio.run(fc._eu_pruefen(
    "Auftrag", title="Ceuta (Opfer)", articles=None, output_language="Deutsch",
    search_items=[F_TOTE, F_SANCHEZ], incident_type="adhoc",
    max_queries=EU_VERIFY_GROUP_QUERIES,
    followup_queries=EU_VERIFY_GROUP_FOLLOWUP_QUERIES,
))
pruefe("H1 Gruppe plant die kleinere Suchmenge",
       aufrufe["such"][0]["max"] == EU_VERIFY_GROUP_QUERIES, aufrufe["such"][0]["max"])
pruefe("H2 Gruppe hakt mit kleinerer Menge nach",
       len(aufrufe["such"]) == 2 and aufrufe["such"][1]["max"] == EU_VERIFY_GROUP_FOLLOWUP_QUERIES,
       [a["max"] for a in aufrufe["such"]])
pruefe("H3 Gruppe stuft belegten Fakt hoch",
       any(f["claim"].startswith("In Ceuta") and f["status"] == "confirmed" for f in facts_h),
       [(f["claim"][:25], f["status"]) for f in facts_h])

print("\nI) Nachhaken abschaltbar")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_SANCHEZ, "unconfirmed", "unklar"),
]))
asyncio.run(fc._eu_pruefen(
    "Auftrag", title="Ceuta", articles=None, output_language="Deutsch",
    search_items=[F_TOTE], incident_type="adhoc", followup_queries=0,
))
pruefe("I1 followup_queries=0 unterbindet die Nachrunde", len(aufrufe["such"]) == 1, len(aufrufe["such"]))

print("\nJ) Suchbelege zaehlen bei der Quellenzuordnung")
_ai_backend_var.set("bedrock")
neu()
# Behauptung, die zu KEINEM gespeicherten Artikel passt, wohl aber zur
# Suchquelle. Vor der Korrektur wurde so ein Fakt herabgestuft.
NUR_SUCHE = "Die Innenministerkonferenz beschliesst einen Fahrplan"
antworten.append(json_fakten([
    (NUR_SUCHE, "established",
     "belegt durch die Suche: https://example.org/a, https://beispiel.de/b, https://muster.at/c"),
    (F_GRENZE, "confirmed", "belegt"),
]))
facts_j, _ = asyncio.run(fc.check("Ceuta", ARTS, "research", "Deutsch"))
treffer_j = [f for f in facts_j if f["claim"].startswith("Die Innenministerkonferenz")]
pruefe("J1 Fakt allein aus der Suche bleibt belegt",
       treffer_j and treffer_j[0]["status"] == "established",
       [(f["claim"][:35], f["status"]) for f in facts_j])
pruefe("J2 Adresse der Suchquelle steht in der Evidenz",
       treffer_j and "https://example.org/a" in treffer_j[0]["evidence"],
       treffer_j[0]["evidence"][:160] if treffer_j else "")

print("\nK) Ohne Suchbelege bleibt die Zuordnung wie bisher")
_ai_backend_var.set("cli")
neu()
antworten.append(json_fakten([(NUR_SUCHE, "established", "belegt")]))
facts_k, _ = asyncio.run(fc.check("Ceuta", ARTS, "research", "Deutsch"))
pruefe("K1 CLI stuft ohne zuordenbare Quelle weiterhin herab",
       facts_k and facts_k[0]["status"] != "established",
       [(f["claim"][:35], f["status"]) for f in facts_k])

print("\nL) Belastungsprobe greift, wenn alles bestaetigt aussieht")
_ai_backend_var.set("bedrock")
neu()
antworten.append(json_fakten([
    (F_TOTE, "confirmed", "belegt"),
    (F_GRENZE, "confirmed", "belegt"),
    (F_SANCHEZ, "confirmed", "belegt"),
]))
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "[S1] https://example.org/a nennt nur eine Schaetzung"),
]))
facts_l, usage_l = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
status_l = {f["claim"][:20]: f["status"] for f in facts_l}
pruefe("L1 zweite Runde ist eine Belastungsprobe",
       len(aufrufe["such"]) == 2 and aufrufe["such"][1]["label"] == "Belastungsprobe",
       [a["label"] for a in aufrufe["such"]])
pruefe("L2 Belastungsprobe sucht ausschliesslich Gegenproben",
       aufrufe["such"][1]["nur_gegenproben"] is True)
pruefe("L3 Belastungsprobe meidet die schon gestellten Anfragen",
       aufrufe["such"][1]["avoid"] == ["Stuetzsuche-q1", "Stuetzsuche-q2"],
       aufrufe["such"][1]["avoid"])
pruefe("L4 nicht haltbare Bestaetigung wird zurueckgenommen",
       status_l.get(F_TOTE[:20]) == "unconfirmed", status_l)
pruefe("L5 haltbare Bestaetigungen bleiben",
       status_l.get(F_GRENZE[:20]) == "confirmed" and status_l.get(F_SANCHEZ[:20]) == "confirmed",
       status_l)
pruefe("L6 kein Faktenverlust", len(facts_l) == 3, len(facts_l))
pruefe("L7 Begruendung vermerkt die Belastungsprobe",
       any("Belastungsprobe" in (f.get("evidence") or "") for f in facts_l),
       [f.get("evidence", "")[:70] for f in facts_l])

print("\nM) Die Belastungsprobe kann nur zurueckstufen")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_GRENZE, "confirmed", "belegt"),
    (F_SANCHEZ, "confirmed", "belegt"),
]))
antworten.append(json_fakten([
    (F_GRENZE, "confirmed", "haelt"),
    (F_SANCHEZ, "established", "sogar noch besser belegt"),
]))
facts_m, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
status_m = {f["claim"][:20]: f["status"] for f in facts_m}
pruefe("M1 Belastungsprobe laeuft auch bei einem offenen Punkt",
       len(aufrufe["such"]) == 2 and aufrufe["such"][1]["label"] == "Belastungsprobe",
       [a["label"] for a in aufrufe["such"]])
pruefe("M2 keine Hochstufung durch die Belastungsprobe",
       status_m.get(F_SANCHEZ[:20]) == "confirmed", status_m)
pruefe("M3 offener Punkt bleibt offen", status_m.get(F_TOTE[:20]) == "unconfirmed", status_m)

print("\nN) Widerspruch in der Nachhak-Runde stuft zurueck")
neu()
antworten.append(json_fakten([
    (F_TOTE, "unconfirmed", "unklar"),
    (F_SANCHEZ, "unconfirmed", "unklar"),
    (F_GRENZE, "confirmed", "belegt"),
]))
antworten.append(json_fakten([
    (F_TOTE, "contradicted", "[S1] https://example.org/a nennt eine andere Zahl"),
]))
facts_n, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
status_n = {f["claim"][:20]: f["status"] for f in facts_n}
pruefe("N1 Nachhak-Runde nimmt den Widerspruch auf",
       status_n.get(F_TOTE[:20]) == "contradicted", status_n)

print("\nO) Faellt die zweite Runde aus, bleibt der erste Durchgang erhalten")
neu()
antworten.append(json_fakten([
    (F_TOTE, "confirmed", "belegt"),
    (F_GRENZE, "confirmed", "belegt"),
]))


async def plan_kaputt(**kw):
    raise RuntimeError("Suche nicht erreichbar")


eur.plan_verification_searches = fake_plan
_alt_plan = eur.plan_verification_searches


async def plan_erst_gut_dann_kaputt(**kw):
    if kw.get("label") == "Belastungsprobe":
        raise RuntimeError("Suche nicht erreichbar")
    return await fake_plan(**kw)


eur.plan_verification_searches = plan_erst_gut_dann_kaputt
facts_o, _ = asyncio.run(fc.check("Ceuta", ARTS, "adhoc", "Deutsch"))
eur.plan_verification_searches = fake_plan
pruefe("O1 Fakten ueberleben den Ausfall", len(facts_o) == 2, len(facts_o))
pruefe("O2 Bewertungen bleiben unveraendert",
       all(f["status"] == "confirmed" for f in facts_o),
       [(f["claim"][:25], f["status"]) for f in facts_o])

print("\nErgebnis: " + str(ok) + " bestanden, " + str(fail) + " fehlgeschlagen")
sys.exit(1 if fail else 0)