research/claimgraph.py

"""Claim Graph — deterministic contradiction elevation for the FSI suit.

The model does NOT resolve contradictions. The system SURFACES them:
"Doc A says X. Doc B says NOT X." → elevated CONTRADICTION card for the human.

Big-tech basis (tiny-model-suit): claim graph extraction + provenance tiering.
Deterministic, no model call needed for the elevation step.
"""
import re
import json
from collections import defaultdict
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional
from pathlib import Path

# --- Claim extraction (deterministic) ---

CLAIM_PATTERNS = [
    # "X is Y" / "X was Y" / "X will be Y"
    (re.compile(r'\b([A-Z][a-zA-Z\s]{2,50}?)\s+(is|was|will be|has been|had been)\s+([^.]{5,100})', re.IGNORECASE), 'assertion'),
    # "X said Y" / "X stated Y" / "X claimed Y"
    (re.compile(r'\b([A-Z][a-zA-Z\s]{2,50}?)\s+(said|stated|claimed|asserted|reported|wrote)\s+(?:that\s+)?([^.]{5,100})', re.IGNORECASE), 'attribution'),
    # Time expressions: "at 9am", "ended at 11am", "ran until noon", "by 5pm" - FULL sentence capture
    (re.compile(r'(?:^|[.!?]\s+)([^.]*?(?:at|by|until|ended at|ran until)\s+\d{1,2}:\d{2}|\d{1,2}(?:am|pm)|noon|midnight)[^.]*\.', re.IGNORECASE), 'time'),
    # Numbers/dates that are checkable
    (re.compile(r'\b(\d{1,3}(?:,\d{3})*(?:\.\d+)?%?)\b'), 'number'),
    (re.compile(r'\b(?:19|20)\d{2}\b'), 'year'),
    # Quoted claims
    (re.compile(r'"([^"]{10,200})"'), 'quote'),
]

# Key nouns that indicate the SUBJECT of a claim (for cross-doc matching)
SUBJECT_NOUNS = {
    'meeting', 'budget', 'file', 'incident', 'report', 'document', 'memo', 'record',
    'account', 'statement', 'testimony', 'evidence', 'claim', 'assertion', 'allegation',
    'event', 'occurrence', 'deletion', 'modification', 'change', 'update', 'revision',
    'investigation', 'audit', 'review', 'analysis', 'study', 'survey', 'assessment',
    'timeline', 'schedule', 'agenda', 'minutes', 'transcript', 'recording', 'log',
    'entry', 'transaction', 'transfer', 'payment', 'deposit', 'withdrawal', 'balance',
    'amount', 'total', 'sum', 'figure', 'number', 'value', 'cost', 'price', 'fee',
    'date', 'time', 'year', 'month', 'day', 'hour', 'minute', 'deadline', 'period',
    'person', 'individual', 'official', 'witness', 'source', 'author', 'speaker',
    'agency', 'department', 'organization', 'company', 'institution', 'government',
    'program', 'project', 'operation', 'initiative', 'policy', 'rule', 'regulation',
    'law', 'statute', 'order', 'directive', 'memo', 'memorandum', 'letter', 'email',
    'message', 'communication', 'notification', 'alert', 'warning', 'report', 'filing'
}

@dataclass
class Claim:
    text: str
    claim_type: str
    source_doc: str
    source_id: str
    entities: List[str]
    subjects: List[str]
    values: List[str]
    times: List[str]
    span_start: int
    span_end: int

@dataclass
class Contradiction:
    claim_a: Claim
    claim_b: Claim
    contradiction_type: str  # 'direct' | 'numeric' | 'temporal' | 'attribution' | 'time'
    severity: str  # 'high' | 'medium' | 'low'
    explanation: str

def extract_entities(text: str) -> List[str]:
    """Extract proper nouns and key entities from text."""
    words = re.findall(r'\b[A-Z][a-zA-Z]{2,}\b', text)
    stop = {'The', 'This', 'That', 'These', 'Those', 'It', 'He', 'She', 'We', 'They', 'You', 'I', 
            'At', 'By', 'Until', 'Ended', 'Ran', 'Official', 'Witness', 'Stated', 'Said'}
    return list(set(w for w in words if w not in stop))

def extract_subjects(text: str) -> List[str]:
    """Extract subject nouns (meeting, budget, file, etc.) for cross-doc matching."""
    text_lower = text.lower()
    subjects = []
    for noun in SUBJECT_NOUNS:
        if noun in text_lower:
            subjects.append(noun)
    return subjects

def extract_values(text: str) -> List[str]:
    """Extract checkable values: numbers, years, percentages."""
    values = []
    values.extend(re.findall(r'\b\d{1,3}(?:,\d{3})*(?:\.\d+)?%?\b', text))
    values.extend(re.findall(r'\b(?:19|20)\d{2}\b', text))
    return values

def extract_times(text: str) -> List[str]:
    """Extract time expressions."""
    times = re.findall(r'\d{1,2}:\d{2}|\d{1,2}(?:am|pm)|noon|midnight', text, re.IGNORECASE)
    return [t.lower() for t in times]

def extract_claims(text: str, source_doc: str, source_id: str) -> List[Claim]:
    """Extract atomic claims from document text."""
    claims = []
    seen_texts = set()
    for pattern, ctype in CLAIM_PATTERNS:
        for match in pattern.finditer(text):
            claim_text = match.group(0).strip()
            if len(claim_text) < 15:
                continue
            # Deduplicate by normalized text
            norm_text = re.sub(r'\s+', ' ', claim_text.lower())
            if norm_text in seen_texts:
                continue
            seen_texts.add(norm_text)
            entities = extract_entities(claim_text)
            subjects = extract_subjects(claim_text)
            values = extract_values(claim_text)
            times = extract_times(claim_text)
            claims.append(Claim(
                text=claim_text,
                claim_type=ctype,
                source_doc=source_doc,
                source_id=source_id,
                entities=entities,
                subjects=subjects,
                values=values,
                times=times,
                span_start=match.start(),
                span_end=match.end()
            ))
    return claims

def claims_overlap(claim_a: Claim, claim_b: Claim) -> bool:
    """Check if two claims are about the same subject."""
    # Check subject overlap (primary for cross-doc matching)
    shared_subjects = set(claim_a.subjects) & set(claim_b.subjects)
    if shared_subjects:
        return True
    # Also check entity overlap
    shared_entities = set(claim_a.entities) & set(claim_b.entities)
    if shared_entities:
        return True
    # Also check value overlap for numeric claims
    if claim_a.values and claim_b.values:
        shared_values = set(claim_a.values) & set(claim_b.values)
        if shared_values:
            return True
    return False

def detect_contradiction(claim_a: Claim, claim_b: Claim) -> Optional[Contradiction]:
    """Detect if two claims about the same subject contradict."""
    if not claims_overlap(claim_a, claim_b):
        return None
    
    text_a = claim_a.text.lower()
    text_b = claim_b.text.lower()
    
    # Direct negation patterns
    negations = [
        (r'\bis\b', r'\bis not\b|\bwas not\b|\bwill not be\b'),
        (r'\bwas\b', r'\bwas not\b|\bis not\b'),
        (r'\bhas\b', r'\bhas not\b|\bhave not\b'),
        (r'\bcan\b', r'\bcannot\b|\bcan\'t\b'),
        (r'\bwill\b', r'\bwill not\b|\bwon\'t\b'),
        (r'\btrue\b', r'\bfalse\b'),
        (r'\byes\b', r'\bno\b'),
    ]
    
    for pos, neg in negations:
        if re.search(pos, text_a) and re.search(neg, text_b):
            return Contradiction(
                claim_a=claim_a, claim_b=claim_b,
                contradiction_type='direct',
                severity='high',
                explanation=f"Direct negation: '{claim_a.text[:80]}...' vs '{claim_b.text[:80]}...'"
            )
        if re.search(pos, text_b) and re.search(neg, text_a):
            return Contradiction(
                claim_a=claim_a, claim_b=claim_b,
                contradiction_type='direct',
                severity='high',
                explanation=f"Direct negation: '{claim_a.text[:80]}...' vs '{claim_b.text[:80]}...'"
            )
    
    # Numeric contradiction (same subject, different numbers)
    if claim_a.values and claim_b.values:
        shared_vals = set(claim_a.values) & set(claim_b.values)
        a_only = set(claim_a.values) - set(claim_b.values)
        b_only = set(claim_b.values) - set(claim_a.values)
        if a_only and b_only and not shared_vals:
            return Contradiction(
                claim_a=claim_a, claim_b=claim_b,
                contradiction_type='numeric',
                severity='high',
                explanation=f"Conflicting values: {claim_a.values} vs {claim_b.values}"
            )
    
    # Time contradiction (same subject, different times)
    if claim_a.times and claim_b.times:
        shared_times = set(claim_a.times) & set(claim_b.times)
        a_only = set(claim_a.times) - set(claim_b.times)
        b_only = set(claim_b.times) - set(claim_a.times)
        if a_only and b_only and not shared_times:
            return Contradiction(
                claim_a=claim_a, claim_b=claim_b,
                contradiction_type='time',
                severity='high',
                explanation=f"Conflicting times: {claim_a.times} vs {claim_b.times}"
            )
    
    # Temporal contradiction (same subject, different dates)
    years_a = [v for v in claim_a.values if re.match(r'^(?:19|20)\d{2}$', v)]
    years_b = [v for v in claim_b.values if re.match(r'^(?:19|20)\d{2}$', v)]
    if years_a and years_b:
        if set(years_a) != set(years_b):
            return Contradiction(
                claim_a=claim_a, claim_b=claim_b,
                contradiction_type='temporal',
                severity='medium',
                explanation=f"Conflicting dates: {years_a} vs {years_b}"
            )
    
    # Attribution contradiction (same quote, different sources)
    if claim_a.claim_type == 'attribution' and claim_b.claim_type == 'attribution':
        if claim_a.text == claim_b.text and claim_a.source_id != claim_b.source_id:
            return Contradiction(
                claim_a=claim_a, claim_b=claim_b,
                contradiction_type='attribution',
                severity='medium',
                explanation=f"Same claim attributed to different sources: {claim_a.source_id} vs {claim_b.source_id}"
            )
    
    return None

def build_claim_graph(documents: Dict[str, str]) -> Dict:
    """Build claim graph from documents, return all claims and contradictions.
    
    Args:
        documents: {source_id: text} mapping
    Returns:
        {
            'claims': [...],
            'contradictions': [...],
            'by_subject': {subject: [claim_ids]},
            'by_entity': {entity: [claim_ids]},
            'by_source': {source_id: [claim_ids]}
        }
    """
    all_claims = []
    claim_id = 0
    by_subject = defaultdict(list)
    by_entity = defaultdict(list)
    by_source = defaultdict(list)
    
    for source_id, text in documents.items():
        claims = extract_claims(text, source_id, source_id)
        for claim in claims:
            claim_dict = asdict(claim)
            claim_dict['id'] = claim_id
            all_claims.append(claim_dict)
            for subject in claim.subjects:
                by_subject[subject].append(claim_id)
            for entity in claim.entities:
                by_entity[entity].append(claim_id)
            by_source[source_id].append(claim_id)
            claim_id += 1
    
    # Find contradictions
    contradictions = []
    for i, claim_a in enumerate(all_claims):
        for j, claim_b in enumerate(all_claims[i+1:], i+1):
            if claim_a['source_id'] == claim_b['source_id']:
                continue  # Same source, skip
            contr = detect_contradiction(
                Claim(**{k:v for k,v in claim_a.items() if k!='id'}),
                Claim(**{k:v for k,v in claim_b.items() if k!='id'})
            )
            if contr:
                c_dict = asdict(contr)
                c_dict['claim_a_id'] = i
                c_dict['claim_b_id'] = j
                contradictions.append(c_dict)
    
    return {
        'claims': all_claims,
        'contradictions': contradictions,
        'by_subject': dict(by_subject),
        'by_entity': dict(by_entity),
        'by_source': dict(by_source),
        'stats': {
            'total_claims': len(all_claims),
            'total_contradictions': len(contradictions),
            'by_type': defaultdict(int, {c['contradiction_type']: 1 for c in contradictions})
        }
    }

def format_contradiction_card(contradiction: Dict, claims: List[Dict]) -> str:
    """Format a contradiction as a human-readable card for the TUI."""
    a = claims[contradiction['claim_a_id']]
    b = claims[contradiction['claim_b_id']]
    
    severity_marker = {'high': '🔴', 'medium': '🟡', 'low': '🟢'}.get(contradiction['severity'], '⚪')
    
    card = f"""
{severity_marker} CONTRADICTION [{contradiction['contradiction_type'].upper()}] {severity_marker}
────────────────────────────────────────
Source A ({a['source_id']}):
  {a['text'][:120]}...

Source B ({b['source_id']}):
  {b['text'][:120]}...

Explanation: {contradiction['explanation']}

Shared Subjects: {', '.join(set(a['subjects']) & set(b['subjects']))}
Shared Entities: {', '.join(set(a['entities']) & set(b['entities']))}
Values A: {a['values'] or 'none'}
Values B: {b['values'] or 'none'}
Times A: {a['times'] or 'none'}
Times B: {b['times'] or 'none'}
"""
    return card

if __name__ == "__main__":
    # Demo with test documents
    test_docs = {
        "doc_a": "The meeting started at 9am and ended at 11am. Official John Smith stated the budget was $50 million.",
        "doc_b": "The meeting started at 9am and ran until noon. Witness Jane Doe said the budget was $75 million.",
        "doc_c": "The 1993 incident file was deleted. Agency reports confirm the deletion occurred in 1995.",
    }
    
    graph = build_claim_graph(test_docs)
    print(f"Claims extracted: {graph['stats']['total_claims']}")
    print(f"Contradictions found: {graph['stats']['total_contradictions']}")
    print(f"By type: {dict(graph['stats']['by_type'])}")
    print()
    for contr in graph['contradictions']:
        print(format_contradiction_card(contr, graph['claims']))