confident_wrong.py

# =============================================================================
# DEPRECATED — JEPA-era research artifact. Kept for provenance, not for use.
#
# This file is part of the original experiments (2026-06-30 to 2026-07-01) that
# the Aporia Engine grew out of: bridging a frozen LLM with a JEPA energy model
# to detect confident-wrong claims. The JEPA fact-grammar approach was SHELVED
# on 2026-07-01 (per-fact density beat it under a pre-registered bar); its
# surviving legacy is the typicality signal in the lacunae ledger's triage.
# Nothing here is wired into the running system. Where it started: this file.
# Where it went: docs/design/ and the verification/ladder/orchestrator code.
# =============================================================================
"""The confident-wrong experiment: can any signal rank plausible-but-FALSE RiftSuite claims
above true ones? Three detectors on claims.py:

  1. DENSITY   - kNN energy vs the RiftSuite corpus (the cartridge signal). Baseline; predicted
                 to FAIL because fakes are RiftSuite-styled (on-manifold).
  2. SYMBOLIC  - retrieve nothing; just check whether the claim's identifiers actually EXIST in
                 the corpus. Should catch nonexistent-identifier fakes cheaply.
  3. JUDGE     - retrieval + local llama3.2 judging the claim against the real retrieved code
                 (RAG-as-calibration). Should catch the SEMANTIC fakes symbolic can't.

Metric: real should pass (not flagged), fake should be flagged. Report catch/pass rates.
"""

from __future__ import annotations

import json
import re
import sys
import urllib.request

import numpy as np

from claims import CLAIMS
from corpus import CPP_EXTS, RIFTSUITE_ROOTS, file_chunk_lists
from embed import Embedder

GEN_URL = "http://localhost:11434/api/generate"
JUDGE_MODEL = "qwen2.5:7b"
USE_GENERATED = "gen" in sys.argv


def unit(x):
    return x / (np.linalg.norm(x, axis=-1, keepdims=True) + 1e-8)


def auroc(higher, lower):
    allv = np.concatenate([higher, lower])
    ranks = np.empty(len(allv))
    ranks[allv.argsort()] = np.arange(1, len(allv) + 1)
    return (ranks[: len(higher)].sum() - len(higher) * (len(higher) + 1) / 2) / (len(higher) * len(lower))


def ollama_generate(prompt: str) -> str:
    payload = json.dumps({
        "model": JUDGE_MODEL, "prompt": prompt, "stream": False,
        "options": {"temperature": 0, "num_predict": 12},
    }).encode()
    req = urllib.request.Request(GEN_URL, data=payload, headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.loads(r.read())["response"]


def main() -> int:
    from generate_claims import generate
    claim_set = generate() if USE_GENERATED else CLAIMS
    print(f"claims: {'generated' if USE_GENERATED else 'hand-written'}  n={len(claim_set)}", flush=True)

    emb = Embedder()
    corpus_texts = [c for f in file_chunk_lists(RIFTSUITE_ROOTS, CPP_EXTS) for c in f]
    corpus_vecs = unit(emb.embed(corpus_texts))
    corpus_blob = "\n".join(corpus_texts)

    claim_texts = [c["text"] for c in claim_set]
    claim_vecs = unit(emb.embed(claim_texts))
    labels = np.array([c["label"] == "fake" for c in claim_set])  # True = should be flagged

    # ---- 1. DENSITY ----
    sims = claim_vecs @ corpus_vecs.T
    energy = 1.0 - np.partition(sims, -10, axis=1)[:, -10:].mean(axis=1)

    # ---- 2. SYMBOLIC (identifier existence) ----
    def identifiers(text):
        ids = set(re.findall(r"[UAFE]Rift[A-Za-z0-9]+", text))       # Rift type names
        ids |= set(re.findall(r"::\s*([A-Za-z_][A-Za-z0-9_]*)", text))  # ::Method calls
        ids |= set(re.findall(r"\b([A-Z][A-Za-z0-9]{2,})\(", text))     # bare CamelCase calls: Foo(
        return ids

    symbolic_flag = np.array([
        any(i not in corpus_blob for i in identifiers(t)) for t in claim_texts
    ])

    # ---- 3. RETRIEVAL + JUDGE (symbolic-guided retrieval + contradiction-only prompt) ----
    def retrieve(text, vec, k=5):
        scored = vec @ corpus_vecs.T
        want = {i for i in identifiers(text) if i in corpus_blob}     # real identifiers the claim names
        id_hits = sorted((i for i in range(len(corpus_texts)) if any(w in corpus_texts[i] for w in want)),
                         key=lambda i: -scored[i])
        chosen = id_hits[:3]                                          # the actual named-method code
        for i in np.argsort(scored)[::-1]:                           # top up with embedding-nearest
            if len(chosen) >= k:
                break
            if int(i) not in chosen:
                chosen.append(int(i))
        return "\n---\n".join(corpus_texts[i] for i in chosen)

    judge_flag = []
    for t, v in zip(claim_texts, claim_vecs):
        ctx = retrieve(t, v)
        prompt = (
            "You are checking a claim against real C++ source from the RiftSuite codebase.\n"
            f"REAL CODE:\n{ctx}\n\n"
            f'CLAIM: "{t}"\n\n'
            "Answer CONTRADICTED only if the code DIRECTLY contradicts the claim "
            "(different return type, signature, owner class, or behavior). "
            "If the code confirms the claim, or does not clearly address it, answer CONSISTENT. "
            "Reply with exactly one word: CONSISTENT or CONTRADICTED."
        )
        judge_flag.append("CONTRADICT" in ollama_generate(prompt).upper())
    judge_flag = np.array(judge_flag)

    combined = symbolic_flag | judge_flag  # full retrieval-grounded detector

    def rates(flag):
        real, fake = ~labels, labels
        return (~flag[real]).mean(), flag[fake].mean(), ((~flag[real]).mean() + flag[fake].mean()) / 2

    print("=== DENSITY (baseline) ===")
    print(f"  AUROC fake-vs-real energy = {auroc(energy[labels], energy[~labels]):.3f}  (0.5 = no signal)")
    print(f"  real mean={energy[~labels].mean():.4f}  fake mean={energy[labels].mean():.4f}")

    print("\n=== detector accuracy (real should PASS, fake should be FLAGGED) ===")
    print(f"  {'detector':<20}{'real-pass':>10}{'fake-catch':>12}{'balanced':>10}")
    for name, flag in [("symbolic", symbolic_flag), ("judge(RAG)", judge_flag), ("symbolic+judge", combined)]:
        rp, fc, ba = rates(flag)
        print(f"  {name:<20}{rp:>10.0%}{fc:>12.0%}{ba:>10.0%}")

    print("\n=== fake-catch by type (combined) ===")
    for ty in sorted({c["type"] for c in claim_set if c["label"] == "fake"}):
        idx = [i for i, c in enumerate(claim_set) if c["label"] == "fake" and c["type"] == ty]
        print(f"  {ty:12} {combined[idx].mean():.0%}  (n={len(idx)})")

    print("\n=== FALSE POSITIVES (real flagged) — instrument these ===")
    for i, c in enumerate(claim_set):
        if not labels[i] and combined[i]:
            print(f"  sym={'F' if symbolic_flag[i] else '.'} judge={'F' if judge_flag[i] else '.'}  {c['text'][:78]}")
    print("=== FALSE NEGATIVES (fake missed) ===")
    for i, c in enumerate(claim_set):
        if labels[i] and not combined[i]:
            print(f"  [{c['type']}]  {c['text'][:78]}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())