M27B: Cert Domain 5.5 + 5.6 Deep Dive — Human Review, Confidence, Provenance, Temporal Data, Synthesis

import anthropic
from dataclasses import dataclass, field

@dataclass
class Claim:
    text: str
    source_id: str
    confidence: float  # 0.0 - 1.0
    valid_from: str = ""  # ISO date
    valid_to: str | None = None

EXTRACT_TOOL = {
    "name": "extract_claims",
    "description": "Extract every distinct factual claim from a source chunk.",
    "input_schema": {
        "type": "object",
        "properties": {
            "claims": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "text": {"type": "string"},
                        "confidence": {"type": "number"},
                        "valid_from": {"type": "string"},
                        "valid_to": {"type": "string"},
                    },
                    "required": ["text", "confidence"],
                },
            },
        },
        "required": ["claims"],
    },
}

def extract_claims(chunk_id: str, chunk_text: str) -> list[Claim]:
    """Send one chunk to Claude; return a list of Claim objects."""
    client = anthropic.Anthropic()
    try:
        result = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=800,
            tools=[EXTRACT_TOOL],
            tool_choice={"type": "tool", "name": "extract_claims"},
            messages=[{"role": "user",
                       "content": f"Extract every factual claim from this source.\n\n{chunk_text}"}],
        )
    except anthropic.APIError as e:
        print(f"Extraction failed for {chunk_id}: {e.message}")
        return []
    # tool_use block contains the structured output
    raw = next(b.input for b in result.content if b.type == "tool_use")
    return [Claim(text=c["text"], source_id=chunk_id,
                  confidence=c["confidence"],
                  valid_from=c.get("valid_from", ""),
                  valid_to=c.get("valid_to"))
            for c in raw["claims"]]

CLASSIFY_TOOL = {
    "name": "classify_pair",
    "description": "Classify whether two claims agree, contradict, or are unrelated.",
    "input_schema": {
        "type": "object",
        "properties": {
            "relation": {"type": "string", "enum": ["agree", "contradict", "unrelated"]},
            "topic": {"type": "string", "description": "If agree or contradict, the shared topic."},
        },
        "required": ["relation"],
    },
}

def classify_pair(client, c1: Claim, c2: Claim) -> tuple[str, str]:
    """Return (relation, topic). relation in {agree, contradict, unrelated}."""
    try:
        result = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=200,
            tools=[CLASSIFY_TOOL],
            tool_choice={"type": "tool", "name": "classify_pair"},
            messages=[{"role": "user", "content":
                f"Claim A (source {c1.source_id}): {c1.text}\n"
                f"Claim B (source {c2.source_id}): {c2.text}\n\n"
                "Classify the relation."}],
        )
        raw = next(b.input for b in result.content if b.type == "tool_use")
        return raw["relation"], raw.get("topic", "")
    except anthropic.APIError as e:
        return "unrelated", ""  # fail safe: don't cluster on errors

def cluster_claims(client, claims: list[Claim]) -> tuple[list, list, list]:
    """Return (established, contested, single_source) lists.
    Pairwise classification is O(n^2); fine for tens of claims, switch to
    embedding-based clustering for hundreds.
    """
    n = len(claims)
    visited = [False] * n
    established, contested, single = [], [], []

    for i in range(n):
        if visited[i]:
            continue
        agree_with = [i]
        contradicts = []
        for j in range(i + 1, n):
            if visited[j]:
                continue
            relation, topic = classify_pair(client, claims[i], claims[j])
            if relation == "agree":
                agree_with.append(j); visited[j] = True
            elif relation == "contradict":
                contradicts.append(j)

        visited[i] = True
        if contradicts:
            contested.append({
                "claim_a": claims[i].text,
                "sources_a": [claims[k].source_id for k in agree_with],
                "claim_b": claims[contradicts[0]].text,
                "sources_b": [claims[contradicts[0]].source_id],
            })
            for k in contradicts: visited[k] = True
        elif len(agree_with) >= 2:
            established.append({
                "claim": claims[i].text,
                "sources": [claims[k].source_id for k in agree_with],
                "confidence": sum(claims[k].confidence for k in agree_with) / len(agree_with),
            })
        else:
            single.append({
                "claim": claims[i].text,
                "source": claims[i].source_id,
                "confidence": claims[i].confidence,
                "needs_review": claims[i].confidence < 0.85,
            })
    return established, contested, single

from datetime import date

def flag_temporal(claims: list[Claim], as_of: date) -> list[dict]:
    """Return a list of temporal warnings for stale claims."""
    warnings = []
    for c in claims:
        # valid_to is set AND in the past relative to query time -> stale
        if c.valid_to:
            try:
                vt = date.fromisoformat(c.valid_to)
                if vt < as_of:
                    warnings.append({
                        "claim": c.text,
                        "as_of": c.valid_to,
                        "query_as_of": as_of.isoformat(),
                        "freshness": "stale",
                        "source": c.source_id,
                    })
            except ValueError:
                pass  # malformed date, skip
        # valid_to is None means "current" - don't flag
        # No valid_from at all means "unknown temporal" - that's a single-source review trigger
    return warnings

import random

def stratified_sample(synthesis: dict, per_bucket: int = 1) -> list[dict]:
    """Pick `per_bucket` items from each output category for human review."""
    review_queue = []
    for bucket_name in ("established", "contested", "single", "temporal_warnings"):
        bucket = synthesis.get(bucket_name, [])
        n = min(len(bucket), per_bucket)
        if n == 0:
            continue
        chosen = random.sample(bucket, n)
        for item in chosen:
            review_queue.append({"bucket": bucket_name, **item})
    return review_queue

"""run_lab.py - Process the 8-chunk fixture, emit synthesis, verify shape."""
import json
import anthropic
from datetime import date
from synthesizer import extract_claims, cluster_claims, flag_temporal, stratified_sample

def main():
    with open("preauth_chunks.json") as f:
        fixture = json.load(f)

    print(f"Query: {fixture['query']}")
    print(f"As-of: {fixture['as_of']}\n")

    # Step A: extract claims from each chunk
    all_claims = []
    for chunk in fixture["chunks"]:
        claims = extract_claims(chunk["chunk_id"], chunk["text"])
        for c in claims:
            c.valid_to = chunk.get("valid_to")
        all_claims.extend(claims)

    print(f"Extracted {len(all_claims)} claims across {len(fixture['chunks'])} chunks\n")

    # Step B: cluster into established / contested / single
    client = anthropic.Anthropic()
    established, contested, single = cluster_claims(client, all_claims)

    # Step C: temporal flagging
    as_of = date.fromisoformat(fixture["as_of"])
    temporal_warnings = flag_temporal(all_claims, as_of)

    synthesis = {
        "established": established,
        "contested": contested,
        "single": single,
        "temporal_warnings": temporal_warnings,
    }

    # Step D: print synthesis
    print("=" * 60)
    print("SYNTHESIS OUTPUT")
    print("=" * 60)
    for bucket in ("established", "contested", "single", "temporal_warnings"):
        print(f"\n[{bucket}] {len(synthesis[bucket])} item(s):")
        for item in synthesis[bucket]:
            print(f"  - {json.dumps(item, indent=2, default=str)[:200]}")

    return synthesis

if __name__ == "__main__":
    main()

# Append to run_lab.py or run interactively after main()
def review_step(synthesis: dict):
    queue = stratified_sample(synthesis, per_bucket=1)
    print("\n" + "=" * 60)
    print("HUMAN REVIEW QUEUE (stratified, 1 per bucket)")
    print("=" * 60)
    for item in queue:
        print(f"\n  [bucket: {item['bucket']}]")
        print(f"  {json.dumps({k: v for k, v in item.items() if k != 'bucket'}, default=str)[:180]}")
    print(f"\nTotal queue size: {len(queue)} items")

if __name__ == "__main__":
    synthesis = main()
    review_step(synthesis)

def assert_contested_have_both_sources(synthesis: dict):
    """Cert-correct contested claims must include BOTH source pointers."""
    failures = []
    for c in synthesis.get("contested", []):
        if not c.get("sources_a") or not c.get("sources_b"):
            failures.append(c)
        if not c.get("claim_a") or not c.get("claim_b"):
            failures.append(c)
    if failures:
        raise AssertionError(
            f"REGRESSION: {len(failures)} contested item(s) missing both sources/claims. "
            f"This is the canonical Domain 5.6 trap. Fix the synthesizer.")
    print(f"\n✅ All {len(synthesis.get('contested', []))} contested claim(s) have both source pointers.")

if __name__ == "__main__":
    synthesis = main()
    review_step(synthesis)
    assert_contested_have_both_sources(synthesis)