#!/usr/bin/env python3
"""Reproducible automated screening analysis for the authorized University News census.

This script implements the frozen 2026-08-10 preregistration addendum. It is a
transparent rule-based measurement pipeline, not a classifier of reporter intent.
All semantic population estimates remain screening results pending manual
validation. Copyrighted bodies are read from the private JSONL but never written.
Output evidence excerpts are capped at 25 words.
"""

from __future__ import annotations

import csv
import hashlib
import json
import math
import re
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable

import numpy as np
import pandas as pd
from scipy import stats


PACKAGE_ROOT = Path(__file__).resolve().parents[1]
WORK_ROOT = PACKAGE_ROOT.parent / "work" if (PACKAGE_ROOT.parent / "work").exists() else PACKAGE_ROOT / "private-input"
CORPUS_PATH = WORK_ROOT / "private-university-corpus-final.jsonl"
INDEX_PATH = WORK_ROOT / "authorized-university-index-dedup.json"
ADDENDUM_PATH = PACKAGE_ROOT / "preregistration-university-news-addendum.md"
ARTICLE_OUT = PACKAGE_ROOT / "article-coding.csv"
ACTOR_OUT = PACKAGE_ROOT / "actor-level-analysis.csv"
RESULTS_OUT = PACKAGE_ROOT / "automated-screening-results.json"
VALIDATION_OUT = PACKAGE_ROOT / "automated-validation-sample-selection.csv"

EXPECTED_ADDENDUM_SHA256 = "26bd2ae9dce549a6d488dd3946051841cba8ed3381b374e92f6da7ae03e72e9b"
STUDY_START = "2023-10-01"
STUDY_END = "2026-08-10"
BOOTSTRAP_SEED = 20260810
BOOTSTRAP_REPS = 5000
VALIDATION_SEED = "university-news-validation-2026-08-10-v1"
VALIDATION_N = 230


ACTOR_ORDER = [
    "administration_trustees",
    "students_orgs",
    "protesters",
    "workers_unions",
    "faculty_critics",
    "government_officials",
    "other",
]
CHALLENGERS = ["students_orgs", "protesters", "workers_unions", "faculty_critics"]

# Strict dictionaries prioritize precision. Broad dictionaries are used only in
# the preregistered sensitivity analysis. Patterns are case-insensitive.
ACTOR_PATTERNS_STRICT = {
    "administration_trustees": [
        r"\b(?:university|columbia|barnard)\s+(?:administration|administrator|officials?|spokespersons?|leadership)\b",
        r"\b(?:interim\s+|acting\s+)?(?:president|provost|dean|vice president|trustee|board of trustees|general counsel)\b",
        r"\b(?:office of the president|office of the provost|office of institutional equity|office of university life)\b",
        r"\b(?:columbia university|barnard college)\s+(?:announced|said|wrote|stated|maintained|confirmed|declined|did not)\b",
        r"\b(?:a|the)\s+(?:university|columbia|barnard)\s+spokesperson\b",
    ],
    "students_orgs": [
        r"\b(?:student|students|undergraduate|undergraduates|graduate students?|doctoral students?|student body)\b",
        r"\b(?:ccsc|esc|gssc|sga|student council|student organization|student group|student club)\b",
        r"\b(?:CC|BC|GS|SEAS|SPS|GSAS|JRN)\s*[’']\d{2}\b",
    ],
    "protesters": [
        r"\b(?:protester|protesters|demonstrator|demonstrators|activist|activists|encampment participants?|occupiers?)\b",
        r"\b(?:protest organizers?|demonstration organizers?|encampment organizers?)\b",
        r"\b(?:CUAD|Columbia University Apartheid Divest|Students for Justice in Palestine|Jewish Voice for Peace|SJP|JVP)\b",
    ],
    "workers_unions": [
        r"\b(?:union|unions|unionized|labor organizers?|bargaining committee|picketing workers?|striking workers?)\b",
        r"\b(?:workers?|employees?)\s+(?:said|wrote|told|argued|demanded|voted|organized|struck|filed|alleged)\b",
        r"\b(?:SWC[-–]UAW|UAW\s*(?:Local)?\s*2710|Local\s*2110|Student Workers of Columbia|Resident Adviser Union)\b",
    ],
    # Faculty count as critics only when a critic/advocacy signal is in the same paragraph.
    "faculty_critics": [r"__ROLE_SENSITIVE_FACULTY_CRITIC__"],
    "government_officials": [
        r"\b(?:federal|state|city)\s+(?:officials?|agency|agencies|government|administration)\b",
        r"\b(?:NYPD|Department of Education|Department of Justice|OCR|Congress|White House|governor|mayor|senator|representative|judge|court)\b",
    ],
    "other": [
        r"\b(?:alumni|alumnus|alumnae|donors?|neighbors?|residents?|community members?|outside experts?|advocacy groups?)\b"
    ],
}

ACTOR_PATTERNS_BROAD = {
    **ACTOR_PATTERNS_STRICT,
    "administration_trustees": ACTOR_PATTERNS_STRICT["administration_trustees"]
    + [r"\b(?:the university|the college|columbia|barnard|administrators?|officials?)\b"],
    "students_orgs": ACTOR_PATTERNS_STRICT["students_orgs"]
    + [r"\b(?:class of \d{4}|campus groups?|student leaders?)\b"],
    "protesters": ACTOR_PATTERNS_STRICT["protesters"]
    + [r"\b(?:protest|demonstration|rally|walkout|occupation|encampment)\b"],
    "workers_unions": ACTOR_PATTERNS_STRICT["workers_unions"]
    + [r"\b(?:staff|faculty union|labor|bargaining|picket|strike)\b"],
    "faculty_critics": [r"\b(?:faculty|professor|professors|instructors?|lecturers?|AAUP)\b"],
}

FACULTY_TERMS = re.compile(r"\b(?:faculty|professor|professors|instructors?|lecturers?|AAUP)\b", re.I)
FACULTY_CRITIC_SIGNALS = re.compile(
    r"\b(?:criticized|opposed|objected|challenged|questioned|condemned|denounced|petitioned|"
    r"signed (?:a|the) (?:letter|petition)|called (?:on|for)|demanded|urged|warned|raised concerns?|"
    r"expressed concerns?|voted against|filed (?:a|the)|protested|advocated)\b",
    re.I,
)

EVAL_TERMS = {
    "favorable": [
        "effective", "successful", "transparent", "responsive", "responsible", "constructive",
        "historic", "groundbreaking", "welcome", "commendable", "impressive", "rigorous",
    ],
    "adverse": [
        "failed", "failure", "misleading", "secretive", "chaotic", "controversial", "embattled",
        "unprecedented", "sweeping", "stunning", "crackdown", "reversal", "under fire",
        "backtracked", "violated", "ignored", "refused", "threatened", "punitive", "abrupt",
    ],
    "delegitimizing": [
        "purported", "so-called", "dubious", "baseless", "unfounded", "extreme", "radical",
        "mob", "caves", "caved", "capitulated", "unruly",
    ],
    "legitimating": [
        "credible", "authoritative", "legitimate", "necessary", "appropriate", "justified",
        "independent", "expert-led", "evidence-based",
    ],
    "conflict": [
        "clash", "clashed", "dispute", "disputed", "opposed", "criticized", "condemned",
        "protested", "demanded", "challenged", "accused", "alleged", "complaint", "lawsuit",
    ],
}
NEGATIONS = {"no", "not", "never", "neither", "nor", "without", "hardly", "denied"}

ATTRIBUTION_TERMS = {
    "neutral": ["said", "says", "wrote", "told", "announced", "stated", "according to", "reported", "explained", "added"],
    "epistemically_marked": ["claimed", "alleged", "asserted", "insisted", "maintained", "purported"],
    "concession_or_fault": ["admitted", "conceded", "acknowledged"],
    "emotive_or_adversarial": ["blasted", "attacked", "complained", "boasted", "threatened"],
}

OFFICIAL_ORIGIN_PATTERNS = [
    r"\b(?:the university|columbia|barnard|the college) (?:announced|released|said|wrote|confirmed|unveiled)\b",
    r"\b(?:email|statement|announcement|press release|message) (?:from|by) (?:the university|columbia|barnard|president|provost|dean)\b",
    r"\b(?:university|columbia|barnard) spokesperson\b",
    r"\baccording to (?:a|the) (?:university|columbia|barnard) (?:webpage|website|statement|email|announcement|report)\b",
]

DOCUMENT_PATTERNS = [
    r"\b(?:documents?|records?|filings?|court filings?|lawsuit|complaint|petition|contract|budget|minutes|policy text|archived (?:page|website)|data|dataset|survey|report)\b",
    r"\b(?:emails?|messages?) (?:obtained|reviewed|provided) (?:by|to) Spectator\b",
    r"\bSpectator (?:obtained|reviewed|analyzed|examined|found)\b",
    r"\baccording to (?:data|records|documents|a filing|the filing|the report|the policy)\b",
]

TESTING_PATTERNS = [
    r"\b(?:however|but|despite|although|contrary to|in contrast|did not|does not|had not|has not)\b",
    r"\b(?:records?|documents?|data|the filing|the policy|archived pages?) (?:show|shows|showed|indicate|indicates|contradict|contradicts|do not|did not)\b",
    r"\bSpectator (?:found|could not verify|confirmed|reviewed|obtained|analyzed|examined)\b",
    r"\b(?:independent|outside) (?:expert|experts|analysis|review)\b",
]

CONTEXT_PATTERNS = [
    r"\b(?:previously|historically|for the first time|since \d{4}|in \d{4}|last year|years? earlier)\b",
    r"\b(?:authority|power|jurisdiction|precedent|governance|resources?|funding|consequences?|impact|affect|risk|access)\b",
    r"\b(?:Spectator previously reported|according to prior reporting|in a previous article)\b",
]

FOLLOWUP_PATTERNS = [
    r"\b(?:Spectator previously reported|following up|months? after|years? after|since then|earlier reporting)\b",
    r"\b(?:deadline|will be reviewed|plans to revisit|expected to|scheduled to|set to)\b",
]

TOPIC_PATTERNS = {
    "protest_palestine": [r"\b(?:Palestin|Gaza|Israel|encampment|protest|demonstration|CUAD|SJP|JVP|divest)"],
    "discipline": [r"\b(?:disciplin|suspend|expel|sanction|probation|Rules of University Conduct|OIE)"],
    "policing_public_safety": [r"\b(?:Public Safety|NYPD|police|security|arrest|crime|surveillance)"],
    "labor": [r"\b(?:union|UAW|worker|labor|bargaining|strike|picket|contract negotiations?)"],
    "governance": [r"\b(?:University Senate|faculty vote|shared governance|senate|committee|governance)"],
    "president_trustees": [r"\b(?:president|Shafik|Armstrong|Shipman|trustee|board of trustees)"],
    "federal_pressure": [r"\b(?:federal|Trump|Congress|Department of Education|Title VI|OCR|funding freeze|settlement)"],
    "campus_access": [r"\b(?:campus access|campus gate|locked out|ID access|access restriction|campus closure)"],
    "speech": [r"\b(?:free speech|academic freedom|expression|speech|censor|event policy|demonstration policy)"],
    "institutional_policy": [r"\b(?:policy|rules?|guidelines?|procedure|regulation|code of conduct)"],
    "major_announcement": [r"\b(?:announced|launch|opened|appointed|selected|named|unveiled|new program|initiative)"],
}

SOURCE_TYPE_PATTERNS = {
    "official": [r"\b(?:spokesperson|president|provost|dean|trustee|University statement|Barnard statement)\b"],
    "student": [r"\b(?:student|undergraduate|graduate student|CC|BC|GS|SEAS)\b"],
    "protester": [r"\b(?:protester|activist|organizer|CUAD|SJP|JVP|encampment)\b"],
    "worker_union": [r"\b(?:worker|union|UAW|bargaining committee|labor organizer)\b"],
    "faculty": [r"\b(?:professor|faculty member|lecturer|instructor)\b"],
    "government": [r"\b(?:NYPD|federal official|city official|state official|judge|court|agency spokesperson)\b"],
    "outside_expert": [r"\b(?:expert|researcher at|professor at|analyst at|attorney at)\b"],
    "affected_person": [r"\b(?:affected|displaced|injured|suspended|terminated|lost access|recipient)\b"],
    "document_data_prior": DOCUMENT_PATTERNS,
}

QUOTE_RE = re.compile(r"[“\"](.{2,2000}?)[”\"]", re.S)
WORD_RE = re.compile(r"\b[\w’'-]+\b", re.UNICODE)


def sha256_file(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""):
            h.update(chunk)
    return h.hexdigest()


def article_id(url: str) -> str:
    return "spec-" + hashlib.sha256(url.encode("utf-8")).hexdigest()[:16]


def words(text: str) -> list[str]:
    return WORD_RE.findall(text or "")


def word_count(text: str) -> int:
    return len(words(text))


def excerpt(text: str, limit: int = 25) -> str:
    clean = re.sub(r"\s+", " ", text or "").strip()
    toks = clean.split()
    return " ".join(toks[:limit])


def mask_quotes(text: str) -> tuple[str, int]:
    count = 0

    def repl(m: re.Match[str]) -> str:
        nonlocal count
        count += 1
        return " " * len(m.group(0))

    return QUOTE_RE.sub(repl, text or ""), count


def compile_patterns(patterns: Iterable[str]) -> list[re.Pattern[str]]:
    return [re.compile(p, re.I) for p in patterns if not p.startswith("__")]


STRICT_RX = {k: compile_patterns(v) for k, v in ACTOR_PATTERNS_STRICT.items()}
BROAD_RX = {k: compile_patterns(v) for k, v in ACTOR_PATTERNS_BROAD.items()}


def faculty_critic_spans(text: str) -> list[tuple[int, int]]:
    spans: list[tuple[int, int]] = []
    pos = 0
    for para in re.split(r"\n\s*\n", text or ""):
        if FACULTY_TERMS.search(para) and FACULTY_CRITIC_SIGNALS.search(para):
            spans.extend((pos + m.start(), pos + m.end()) for m in FACULTY_TERMS.finditer(para))
        pos += len(para) + 2
    return spans


def actor_spans(text: str, actor: str, broad: bool = False) -> list[tuple[int, int]]:
    spans = [(m.start(), m.end()) for rx in (BROAD_RX if broad else STRICT_RX)[actor] for m in rx.finditer(text or "")]
    if actor == "faculty_critics" and not broad:
        spans.extend(faculty_critic_spans(text))
    return sorted(set(spans))


def merge_windows(text: str, spans: list[tuple[int, int]], radius: int = 180) -> str:
    if not spans:
        return ""
    ranges = sorted((max(0, s - radius), min(len(text), e + radius)) for s, e in spans)
    merged: list[list[int]] = []
    for s, e in ranges:
        if merged and s <= merged[-1][1]:
            merged[-1][1] = max(merged[-1][1], e)
        else:
            merged.append([s, e])
    return " … ".join(text[s:e] for s, e in merged)


def term_occurrences(text: str, terms: list[str]) -> tuple[int, list[str]]:
    count = 0
    hits: list[str] = []
    for term in terms:
        rx = re.compile(r"\b" + re.escape(term) + r"\b", re.I)
        for m in rx.finditer(text or ""):
            prior = words(text[max(0, m.start() - 45):m.start()].lower())[-4:]
            if any(tok in NEGATIONS for tok in prior):
                continue
            count += 1
            hits.append(m.group(0))
    return count, hits


def eval_counts(window: str) -> dict[str, Any]:
    out: dict[str, Any] = {}
    for cls, terms in EVAL_TERMS.items():
        count, hits = term_occurrences(window, terms)
        out[cls] = count
        out[f"{cls}_hits"] = hits
    return out


def actor_presence_confidence(strict_n: int, broad_n: int, actor: str) -> tuple[str, float]:
    if strict_n >= 2:
        return "high", 0.90
    if strict_n == 1:
        return ("medium", 0.72) if actor != "faculty_critics" else ("medium", 0.68)
    if broad_n:
        return "low", 0.45
    return "not_present", 1.0


def classify_topics(text: str) -> list[str]:
    labels = [name for name, pats in TOPIC_PATTERNS.items() if any(re.search(p, text, re.I) for p in pats)]
    return labels or ["other"]


TOPIC_PRIORITY = [
    "protest_palestine", "discipline", "policing_public_safety", "labor", "federal_pressure",
    "president_trustees", "governance", "campus_access", "speech", "institutional_policy",
    "major_announcement", "other",
]


def primary_topic(labels: list[str]) -> str:
    return next((x for x in TOPIC_PRIORITY if x in labels), "other")


def article_type_proxy(title: str, body: str, wc: int, documents: int) -> str:
    lower = title.lower()
    if "timeline" in lower or "live updates" in lower:
        return "live_timeline"
    if any(x in lower for x in ["what to know", "explained", "how ", "why "]) and wc >= 650:
        return "analysis_explainer"
    enterprise = len(re.findall(r"\b(?:obtained|reviewed|analyzed|examined|found)\b", body, re.I))
    if documents and wc >= 1000 and enterprise >= 2:
        return "investigative_enterprise"
    if wc < 450 and re.search(r"\b(?:breaking|announced|resigns?|appointed|selected|canceled|closed)\b", title, re.I):
        return "breaking_brief"
    return "straight_news"


def leadership_period(date: str) -> str:
    if date <= "2023-12-12":
        return "147th_board"
    if date <= "2024-12-10":
        return "148th_board"
    if date <= "2025-12-09":
        return "149th_board"
    return "150th_board"


def quarter(date: str) -> str:
    y, m, _ = map(int, date.split("-"))
    return f"{y}-Q{(m - 1) // 3 + 1}"


def author_parts(author: str) -> list[str]:
    vals = re.split(r"\s+and\s+|\s*,\s*|\s*;\s*", author or "")
    return [v.strip() for v in vals if v.strip()]


def build_person_actor_map(body: str) -> dict[str, str]:
    mapping: dict[str, str] = {}
    name = r"([A-Z][A-Za-zÀ-ÖØ-öø-ÿ.'’-]+(?:\s+[A-Z][A-Za-zÀ-ÖØ-öø-ÿ.'’-]+){1,3})"
    patterns = [
        (rf"{name},?\s+(?:CC|BC|GS|SEAS|SPS|GSAS|JRN)\s*[’']\d{{2}}", "students_orgs"),
        (rf"(?:President|Provost|Dean|Trustee|Vice President)\s+{name}", "administration_trustees"),
        (rf"{name},?\s+(?:the\s+)?(?:president|provost|dean|trustee|vice president|administrator|spokesperson)", "administration_trustees"),
        (rf"(?:Professor|professor)\s+{name}", "faculty_critics"),
        (rf"{name},?\s+(?:a\s+|an\s+)?(?:professor|faculty member|lecturer|instructor)", "faculty_critics"),
        (rf"{name},?\s+(?:a\s+|an\s+)?(?:union member|union organizer|worker|employee|bargaining committee member)", "workers_unions"),
        (rf"{name},?\s+(?:a\s+|an\s+)?(?:protester|activist|organizer|demonstrator)", "protesters"),
    ]
    for pat, actor in patterns:
        for m in re.finditer(pat, body or "", re.I if actor != "students_orgs" else 0):
            candidate = m.group(1).strip()
            if len(candidate.split()) <= 4:
                mapping[candidate.lower()] = actor
    return mapping


def classify_quote_actor(context: str, person_map: dict[str, str]) -> tuple[str, float]:
    lower = context.lower()
    mapped = {actor for name, actor in person_map.items() if name in lower}
    if len(mapped) == 1:
        return next(iter(mapped)), 0.90
    found = []
    for actor in ACTOR_ORDER:
        if actor_spans(context, actor, broad=False):
            found.append(actor)
    if len(found) == 1:
        return found[0], 0.78
    return "unattributed_or_ambiguous", 0.20


def quotes_by_actor(body: str) -> tuple[list[dict[str, Any]], float]:
    person_map = build_person_actor_map(body)
    rows: list[dict[str, Any]] = []
    for m in QUOTE_RE.finditer(body or ""):
        before = body[max(0, m.start() - 220):m.start()]
        after = body[m.end():min(len(body), m.end() + 220)]
        context = after + " " + before
        verbs: list[tuple[str, str]] = []
        for cls, terms in ATTRIBUTION_TERMS.items():
            for term in terms:
                hit = re.search(r"\b" + re.escape(term) + r"\b", context, re.I)
                if hit:
                    verbs.append((cls, term))
        actor, conf = classify_quote_actor(context, person_map)
        rows.append({
            "actor": actor,
            "confidence": conf,
            "words": word_count(m.group(1)),
            "verb_class": verbs[0][0] if verbs else "other_or_unclear",
        })
    attributed = sum(1 for r in rows if r["actor"] in ACTOR_ORDER)
    confidence = attributed / len(rows) if rows else 1.0
    return rows, confidence


def binary_patterns(text: str, patterns: list[str]) -> int:
    return int(any(re.search(p, text or "", re.I) for p in patterns))


def source_types(text: str) -> list[str]:
    return [k for k, pats in SOURCE_TYPE_PATTERNS.items() if any(re.search(p, text or "", re.I) for p in pats)]


def unique_source_proxy(body: str, quote_rows: list[dict[str, Any]]) -> int:
    # Conservative proxy: distinct mapped quoted actors plus explicit institutional/document sources.
    quoted = Counter(r["actor"] for r in quote_rows if r["actor"] in ACTOR_ORDER)
    institutional = len(set(re.findall(
        r"\b(?:University spokesperson|Barnard spokesperson|Columbia spokesperson|University statement|Barnard statement|"
        r"court filing|lawsuit|report|survey|dataset|records?)\b", body or "", re.I
    )))
    return len(quoted) + institutional


def source_in_first_half(body: str, actors: list[str]) -> bool:
    half = body[: max(1, len(body) // 2)]
    qrows, _ = quotes_by_actor(half)
    return any(q["actor"] in actors for q in qrows)


def actor_window_metrics(
    actor: str,
    full_reporter: str,
    full_unmasked: str,
    headline_reporter: str,
    lede_reporter: str,
    quote_rows: list[dict[str, Any]],
    broad: bool = False,
) -> dict[str, Any]:
    spans = actor_spans(full_reporter, actor, broad=broad)
    unmasked_spans = actor_spans(full_unmasked, actor, broad=broad)
    win = merge_windows(full_reporter, spans)
    unmasked_win = merge_windows(full_unmasked, unmasked_spans)
    e = eval_counts(win)
    eu = eval_counts(unmasked_win)
    hw = merge_windows(headline_reporter, actor_spans(headline_reporter, actor, broad=broad), radius=120)
    lw = merge_windows(lede_reporter, actor_spans(lede_reporter, actor, broad=broad), radius=150)
    he = eval_counts(hw)
    le = eval_counts(lw)
    attr = {}
    for cls, terms in ATTRIBUTION_TERMS.items():
        attr[cls] = sum(term_occurrences(win, [term])[0] for term in terms)
    actor_quotes = [q for q in quote_rows if q["actor"] == actor]
    marked = int((e["adverse"] + e["delegitimizing"]) > 0) if spans else ""
    marked_conflict = int((e["adverse"] + e["delegitimizing"] + e["conflict"]) > 0) if spans else ""
    marked_include_quotes = int((eu["adverse"] + eu["delegitimizing"]) > 0) if unmasked_spans else ""

    def frame(ec: dict[str, Any], present: bool) -> str:
        if not present:
            return "not_present"
        if ec["adverse"] + ec["delegitimizing"] > ec["favorable"] + ec["legitimating"]:
            return "adverse"
        if ec["favorable"] + ec["legitimating"] > ec["adverse"] + ec["delegitimizing"]:
            return "favorable"
        if ec["conflict"]:
            return "conflict"
        return "neutral_descriptive"

    hit_text = ""
    all_hits = e["adverse_hits"] + e["delegitimizing_hits"] + e["favorable_hits"] + e["legitimating_hits"]
    if all_hits:
        term = all_hits[0]
        pos = win.lower().find(term.lower())
        hit_text = excerpt(win[max(0, pos - 90):pos + len(term) + 90])
    wc = word_count(win)
    return {
        "mention_count": len(spans),
        "window_words": wc,
        "favorable_count": e["favorable"],
        "adverse_count": e["adverse"],
        "delegitimizing_count": e["delegitimizing"],
        "legitimating_count": e["legitimating"],
        "conflict_count": e["conflict"],
        "favorable_rate_per_1000": 1000 * e["favorable"] / wc if wc else "",
        "adverse_rate_per_1000": 1000 * (e["adverse"] + e["delegitimizing"]) / wc if wc else "",
        "marked_strict": marked,
        "marked_adverse_plus_conflict": marked_conflict,
        "marked_include_quotes": marked_include_quotes,
        "headline_frame": frame(he, bool(hw)),
        "lede_frame": frame(le, bool(lw)),
        "quote_count": len(actor_quotes),
        "direct_quote_words": sum(q["words"] for q in actor_quotes),
        "actor_quoted": int(bool(actor_quotes)) if spans else "",
        "attr_neutral": attr["neutral"],
        "attr_epistemically_marked": attr["epistemically_marked"],
        "attr_concession_or_fault": attr["concession_or_fault"],
        "attr_emotive_or_adversarial": attr["emotive_or_adversarial"],
        "marked_evidence_excerpt": hit_text,
    }


def first_actor_in_text(text: str, actors: list[str], broad: bool = False) -> str:
    candidates: list[tuple[int, str]] = []
    for actor in actors:
        spans = actor_spans(text, actor, broad=broad)
        if spans:
            candidates.append((spans[0][0], actor))
    return min(candidates)[1] if candidates else "none_observed"


def scrutiny_score(actor: str, claim: bool, actor_window: str, docs: int, opposing_claim: bool) -> str | int:
    if not claim:
        return "NO_SUBSTANTIVE_CLAIM"
    tests = sum(len(re.findall(p, actor_window, re.I)) for p in TESTING_PATTERNS)
    if docs and tests >= 2:
        return 3
    if docs and tests >= 1:
        return 2
    if opposing_claim:
        return 1
    return 0


def confidence_label(score: float) -> str:
    if score >= 0.80:
        return "high"
    if score >= 0.60:
        return "medium"
    return "low"


def build_rows(record: dict[str, Any]) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    url = record["canonical_url"]
    aid = article_id(url)
    title = record.get("title") or record.get("archive_title") or ""
    description = re.sub(r"<[^>]+>", " ", record.get("description") or "")
    body = record.get("body") or ""
    paragraphs = [p.strip() for p in re.split(r"\n\s*\n", body) if p.strip()]
    lede = "\n\n".join(paragraphs[:2])
    headline_text = f"{title}. {description}".strip()
    full_unmasked = f"{headline_text}\n\n{body}"
    full_reporter, masked_quote_n = mask_quotes(full_unmasked)
    headline_reporter, _ = mask_quotes(headline_text)
    lede_reporter, _ = mask_quotes(lede)
    quote_rows, quote_conf = quotes_by_actor(body)
    docs = binary_patterns(body, DOCUMENT_PATTERNS)
    context = int(sum(binary_patterns(body, [p]) for p in CONTEXT_PATTERNS) >= 2)
    followup = binary_patterns(body, FOLLOWUP_PATTERNS)
    official_origin = binary_patterns(lede, OFFICIAL_ORIGIN_PATTERNS)
    types = source_types(body)
    # Topic controls require substantial/early prominence. Full-body searching caused
    # incidental historical references to dominate the control; see the timestamped
    # parser-defect change log written before this recalculation.
    labels = classify_topics(f"{headline_text}\n{lede}")
    ptopic = primary_topic(labels)
    date = record.get("archive_date") or ""
    subdesk = (record.get("archive_categories") or [record.get("section_label", "unknown")]) [0]
    wc = record.get("word_count") or word_count(body)
    actor_metrics: dict[str, dict[str, Any]] = {}
    actor_metrics_broad: dict[str, dict[str, Any]] = {}
    strict_present: dict[str, bool] = {}
    broad_present: dict[str, bool] = {}
    pres_conf: dict[str, tuple[str, float]] = {}
    for actor in ACTOR_ORDER:
        actor_metrics[actor] = actor_window_metrics(
            actor, full_reporter, full_unmasked, headline_reporter, lede_reporter, quote_rows, broad=False
        )
        actor_metrics_broad[actor] = actor_window_metrics(
            actor, full_reporter, full_unmasked, headline_reporter, lede_reporter, quote_rows, broad=True
        )
        strict_n = actor_metrics[actor]["mention_count"]
        broad_n = actor_metrics_broad[actor]["mention_count"]
        strict_present[actor] = strict_n > 0
        broad_present[actor] = broad_n > 0
        pres_conf[actor] = actor_presence_confidence(strict_n, broad_n, actor)

    admin_present = strict_present["administration_trustees"]
    challenger_present = any(strict_present[a] for a in CHALLENGERS)
    admin_claim = bool(admin_present and (
        actor_metrics["administration_trustees"]["quote_count"]
        or sum(actor_metrics["administration_trustees"][f"attr_{c}"] for c in [
            "neutral", "epistemically_marked", "concession_or_fault", "emotive_or_adversarial"
        ])
    ))
    challenger_claims: dict[str, bool] = {}
    for actor in CHALLENGERS:
        m = actor_metrics[actor]
        challenger_claims[actor] = bool(strict_present[actor] and (
            m["quote_count"] or sum(m[f"attr_{c}"] for c in [
                "neutral", "epistemically_marked", "concession_or_fault", "emotive_or_adversarial"
            ])
        ))
    any_challenger_claim = any(challenger_claims.values())

    first_actor = first_actor_in_text(full_reporter, ACTOR_ORDER)
    first_quote = next((q["actor"] for q in quote_rows if q["actor"] in ACTOR_ORDER), "none_observed")
    actor_out: list[dict[str, Any]] = []
    for actor in ACTOR_ORDER:
        if not broad_present[actor]:
            continue
        m = actor_metrics[actor]
        mb = actor_metrics_broad[actor]
        awin = merge_windows(full_reporter, actor_spans(full_reporter, actor, broad=False), radius=300)
        claim = admin_claim if actor == "administration_trustees" else challenger_claims.get(actor, bool(m["quote_count"]))
        opposing = any_challenger_claim if actor == "administration_trustees" else admin_claim
        scrutiny = scrutiny_score(actor, claim, awin, docs, opposing)
        conf_num = min(pres_conf[actor][1], quote_conf if m["quote_count"] else 0.75, 0.62 if claim else 0.75)
        actor_out.append({
            "article_id": aid,
            "url": url,
            "date": date,
            "subdesk": subdesk,
            "actor_category": actor,
            "actor_group": "administration" if actor == "administration_trustees" else ("challenger" if actor in CHALLENGERS else "contextual"),
            "strict_present": int(strict_present[actor]),
            "broad_present": int(broad_present[actor]),
            "presence_confidence": pres_conf[actor][0],
            "presence_confidence_numeric": pres_conf[actor][1],
            **m,
            "broad_mention_count": mb["mention_count"],
            "broad_window_words": mb["window_words"],
            "broad_marked_strict": mb["marked_strict"],
            "broad_marked_adverse_plus_conflict": mb["marked_adverse_plus_conflict"],
            "first_actor": int(first_actor == actor) if strict_present[actor] else "",
            "first_quoted_actor": int(first_quote == actor) if strict_present[actor] else "",
            "substantive_claim_proxy": int(claim) if strict_present[actor] else "",
            "scrutiny_score_auto": scrutiny,
            "quote_attribution_confidence": quote_conf,
            "reporter_window_confidence": 0.92 if masked_quote_n == len(quote_rows) else 0.70,
            "semantic_confidence": confidence_label(conf_num),
            "semantic_confidence_numeric": conf_num,
            "validation_status": "AUTOMATED_SCREENING_PENDING_MANUAL_VALIDATION",
            "word_count": wc,
            "primary_topic": ptopic,
            "article_type_proxy": article_type_proxy(title, body, wc, docs),
            "leadership_period": leadership_period(date),
            "first_author": (author_parts(record.get("author") or "") or [""])[0],
            "author_count": len(author_parts(record.get("author") or "")),
        })

    # Article-level testing/reliance proxies use actor results already computed.
    scrutiny_by_actor = {r["actor_category"]: r["scrutiny_score_auto"] for r in actor_out}
    admin_tested = int(isinstance(scrutiny_by_actor.get("administration_trustees"), int) and scrutiny_by_actor["administration_trustees"] >= 2) if admin_claim else "NOT_APPLICABLE"
    chall_scores = [scrutiny_by_actor.get(a) for a in CHALLENGERS if challenger_claims.get(a)]
    chall_tested = int(any(isinstance(v, int) and v >= 2 for v in chall_scores)) if chall_scores else "NOT_APPLICABLE"
    challenger_first_half = source_in_first_half(body, CHALLENGERS)
    independent_test = bool(admin_tested == 1 or chall_tested == 1)
    if not official_origin:
        reliance = 0
    elif independent_test:
        reliance = 1
    elif challenger_first_half:
        reliance = 2
    else:
        reliance = 3
    high_reliance = int(official_origin and not independent_test and not challenger_first_half)
    update_notes = record.get("update_correction_notes") or []
    update_text = " ".join(map(str, update_notes))
    if re.search(r"editor.?s note", update_text, re.I):
        update_status = "editors_note"
    elif re.search(r"correct", update_text, re.I):
        update_status = "corrected"
    elif update_notes:
        update_status = "updated"
    else:
        update_status = "none_observed"
    first_authors = author_parts(record.get("author") or "")
    article_conf = np.mean([p[1] for a, p in pres_conf.items() if broad_present[a]]) if any(broad_present.values()) else 1.0
    marked_snip = next((r["marked_evidence_excerpt"] for r in actor_out if r["marked_evidence_excerpt"]), "")
    strict_admin_rows = [r for r in actor_out if r["actor_category"] == "administration_trustees" and r["strict_present"] == 1]
    strict_challenger_rows = [r for r in actor_out if r["actor_category"] in CHALLENGERS and r["strict_present"] == 1]
    if strict_admin_rows and strict_challenger_rows:
        ar0 = strict_admin_rows[0]
        admin_marked = float(ar0["marked_strict"])
        challenger_marked = float(np.mean([float(r["marked_strict"]) for r in strict_challenger_rows]))
        claim_challenger_rows = [r for r in strict_challenger_rows if r["substantive_claim_proxy"] == 1 and isinstance(r["scrutiny_score_auto"], int)]
        admin_scrutiny_value = ar0["scrutiny_score_auto"] if isinstance(ar0["scrutiny_score_auto"], int) else np.nan
        challenger_scrutiny_value = float(np.mean([r["scrutiny_score_auto"] for r in claim_challenger_rows])) if claim_challenger_rows else np.nan
        admin_qw = int(ar0["direct_quote_words"])
        challenger_qw = int(sum(r["direct_quote_words"] for r in strict_challenger_rows))
        relevant_qw = admin_qw + challenger_qw
        article_pair_values = {
            "admin_marked_auto": admin_marked,
            "challenger_marked_mean_auto": challenger_marked,
            "paired_framing_gap_auto": admin_marked - challenger_marked,
            "admin_scrutiny_score_auto": admin_scrutiny_value if pd.notna(admin_scrutiny_value) else "NO_SUBSTANTIVE_CLAIM",
            "challenger_scrutiny_mean_auto": challenger_scrutiny_value if pd.notna(challenger_scrutiny_value) else "NO_SUBSTANTIVE_CLAIM",
            "paired_scrutiny_gap_auto": admin_scrutiny_value - challenger_scrutiny_value if pd.notna(admin_scrutiny_value) and pd.notna(challenger_scrutiny_value) else "NO_SUBSTANTIVE_CLAIM",
            "admin_direct_quote_words_auto": admin_qw,
            "challenger_direct_quote_words_auto": challenger_qw,
            "direct_quote_word_share_gap_auto": (admin_qw - challenger_qw) / relevant_qw if relevant_qw else "NO_ATTRIBUTED_RELEVANT_QUOTE",
            "first_quote_gap_auto": int(first_quote == "administration_trustees") - int(first_quote in CHALLENGERS),
        }
    else:
        article_pair_values = {
            "admin_marked_auto": "NOT_APPLICABLE",
            "challenger_marked_mean_auto": "NOT_APPLICABLE",
            "paired_framing_gap_auto": "NOT_APPLICABLE",
            "admin_scrutiny_score_auto": "NOT_APPLICABLE",
            "challenger_scrutiny_mean_auto": "NOT_APPLICABLE",
            "paired_scrutiny_gap_auto": "NOT_APPLICABLE",
            "admin_direct_quote_words_auto": "NOT_APPLICABLE",
            "challenger_direct_quote_words_auto": "NOT_APPLICABLE",
            "direct_quote_word_share_gap_auto": "NOT_APPLICABLE",
            "first_quote_gap_auto": "NOT_APPLICABLE",
        }
    article = {
        "article_id": aid,
        "title": title,
        "subtitle": record.get("subtitle") or "",
        "author": record.get("author") or "",
        "first_author": first_authors[0] if first_authors else "",
        "author_count": len(first_authors),
        "date": date,
        "publication_month": date[:7],
        "quarter": quarter(date),
        "subdesk": subdesk,
        "section_label": record.get("section_label") or "",
        "url": url,
        "retrieved_at": record.get("retrieved_at") or "",
        "body_sha256": record.get("body_sha256") or "",
        "extraction_status": record.get("extraction_status") or "",
        "word_count": wc,
        "computed_word_count": word_count(body),
        "topic_labels": "|".join(labels),
        "primary_topic": ptopic,
        "article_type_proxy": article_type_proxy(title, body, wc, docs),
        "leadership_period": leadership_period(date),
        "admin_present_strict": int(admin_present),
        "challenger_present_strict": int(challenger_present),
        "paired_actor_article_strict": int(admin_present and challenger_present),
        "admin_present_broad": int(broad_present["administration_trustees"]),
        "challenger_present_broad": int(any(broad_present[a] for a in CHALLENGERS)),
        "paired_actor_article_broad": int(broad_present["administration_trustees"] and any(broad_present[a] for a in CHALLENGERS)),
        **article_pair_values,
        "first_actor_auto": first_actor,
        "first_quoted_actor_auto": first_quote,
        "quote_count_total": len(quote_rows),
        "quote_word_count_total": sum(q["words"] for q in quote_rows),
        "quote_attributed_count": sum(q["actor"] in ACTOR_ORDER for q in quote_rows),
        "quote_attribution_confidence": quote_conf,
        "source_count_proxy": unique_source_proxy(body, quote_rows),
        "source_type_count_proxy": len(types),
        "source_types_proxy": "|".join(types),
        "university_official_origin_auto": official_origin,
        "university_release_reliance_auto": reliance,
        "high_release_reliance_auto": high_reliance,
        "official_assertion_tested_auto": admin_tested,
        "challenger_assertion_tested_auto": chall_tested,
        "documents_records_data_auto": docs,
        "affected_people_consulted_auto": int(any(r["quote_count"] for r in actor_out if r["actor_category"] in CHALLENGERS)),
        "power_history_consequences_context_auto": context,
        "follow_up_orientation_auto": followup,
        "visible_update_correction_status": update_status,
        "vulnerable_identification_auto": "NOT_CODED_IMAGE_CONSENT_REVIEW_REQUIRED",
        "semantic_confidence": confidence_label(float(article_conf)),
        "semantic_confidence_numeric": float(article_conf),
        "marked_evidence_excerpt_max_25_words": marked_snip,
        "coding_status": "AUTOMATED_SCREENING_PENDING_MANUAL_VALIDATION",
        "parser_failure_reason": "",
    }
    return article, actor_out


def mean_ci_boot(values: Iterable[float], seed: int, reps: int = BOOTSTRAP_REPS) -> dict[str, Any]:
    arr = np.asarray([float(v) for v in values if pd.notna(v)], dtype=float)
    if not len(arr):
        return {"n": 0, "estimate": None, "ci95": [None, None], "method": "not_estimable"}
    rng = np.random.default_rng(seed)
    means = np.empty(reps)
    for i in range(reps):
        means[i] = rng.choice(arr, size=len(arr), replace=True).mean()
    return {
        "n": int(len(arr)),
        "estimate": float(arr.mean()),
        "ci95": [float(np.quantile(means, 0.025)), float(np.quantile(means, 0.975))],
        "method": f"paired_article_bootstrap_{reps}_reps",
    }


def wilson_ci(x: int, n: int, z: float = 1.959963984540054) -> list[float | None]:
    if n == 0:
        return [None, None]
    p = x / n
    den = 1 + z * z / n
    center = (p + z * z / (2 * n)) / den
    half = z * math.sqrt(p * (1 - p) / n + z * z / (4 * n * n)) / den
    return [center - half, center + half]


def article_pair_table(article_df: pd.DataFrame, actor_df: pd.DataFrame, broad: bool = False) -> pd.DataFrame:
    pcol = "broad_present" if broad else "strict_present"
    mcol = "broad_marked_strict" if broad else "marked_strict"
    rows = []
    for aid, grp in actor_df.groupby("article_id"):
        admin = grp[(grp.actor_category == "administration_trustees") & (grp[pcol] == 1)]
        chall = grp[(grp.actor_category.isin(CHALLENGERS)) & (grp[pcol] == 1)]
        if admin.empty or chall.empty:
            continue
        ar = article_df.loc[article_df.article_id == aid].iloc[0]
        a = admin.iloc[0]
        ach = chall.copy()
        claim_ch = ach[pd.to_numeric(ach.substantive_claim_proxy, errors="coerce") == 1]
        admin_scrutiny = pd.to_numeric(pd.Series([a.scrutiny_score_auto]), errors="coerce").iloc[0]
        challenger_scrutiny = pd.to_numeric(claim_ch.scrutiny_score_auto, errors="coerce").mean() if not claim_ch.empty else np.nan
        admin_claim = int(pd.to_numeric(pd.Series([a.substantive_claim_proxy]), errors="coerce").fillna(0).iloc[0])
        admin_qw = float(a.direct_quote_words or 0)
        chall_qw = float(pd.to_numeric(ach.direct_quote_words, errors="coerce").fillna(0).sum())
        quote_den = admin_qw + chall_qw
        admin_marked = float(pd.to_numeric(pd.Series([a[mcol]]), errors="coerce").iloc[0])
        chall_marked = float(pd.to_numeric(ach[mcol], errors="coerce").mean())
        rows.append({
            "article_id": aid,
            "admin_marked": admin_marked,
            "challenger_marked_mean": chall_marked,
            "framing_gap": admin_marked - chall_marked,
            "admin_scrutiny": admin_scrutiny,
            "challenger_scrutiny_mean": challenger_scrutiny,
            "scrutiny_gap": admin_scrutiny - challenger_scrutiny if admin_claim and not np.isnan(challenger_scrutiny) else np.nan,
            "admin_first_quote": int(ar.first_quoted_actor_auto == "administration_trustees"),
            "challenger_first_quote": int(ar.first_quoted_actor_auto in CHALLENGERS),
            "first_quote_gap": int(ar.first_quoted_actor_auto == "administration_trustees") - int(ar.first_quoted_actor_auto in CHALLENGERS),
            "admin_quote_share": admin_qw / quote_den if quote_den else np.nan,
            "challenger_quote_share": chall_qw / quote_den if quote_den else np.nan,
            "quote_share_gap": (admin_qw - chall_qw) / quote_den if quote_den else np.nan,
            "admin_actor_quoted": float(a.actor_quoted) if str(a.actor_quoted) != "" else np.nan,
            "challenger_actor_quoted_mean": float(pd.to_numeric(ach.actor_quoted, errors="coerce").mean()),
            "actor_quoted_gap": float(a.actor_quoted) - float(pd.to_numeric(ach.actor_quoted, errors="coerce").mean()) if str(a.actor_quoted) != "" else np.nan,
            "admin_first_actor": int(ar.first_actor_auto == "administration_trustees"),
            "challenger_first_actor": int(ar.first_actor_auto in CHALLENGERS),
            "first_actor_gap": int(ar.first_actor_auto == "administration_trustees") - int(ar.first_actor_auto in CHALLENGERS),
            "admin_marked_attribution": int(float(a.attr_epistemically_marked) + float(a.attr_concession_or_fault) + float(a.attr_emotive_or_adversarial) > 0),
            "challenger_marked_attribution_mean": float(((pd.to_numeric(ach.attr_epistemically_marked, errors="coerce") + pd.to_numeric(ach.attr_concession_or_fault, errors="coerce") + pd.to_numeric(ach.attr_emotive_or_adversarial, errors="coerce")) > 0).mean()),
            "marked_attribution_gap": int(float(a.attr_epistemically_marked) + float(a.attr_concession_or_fault) + float(a.attr_emotive_or_adversarial) > 0) - float(((pd.to_numeric(ach.attr_epistemically_marked, errors="coerce") + pd.to_numeric(ach.attr_concession_or_fault, errors="coerce") + pd.to_numeric(ach.attr_emotive_or_adversarial, errors="coerce")) > 0).mean()),
            "date": ar.date,
            "subdesk": ar.subdesk,
            "primary_topic": ar.primary_topic,
            "publication_month": ar.publication_month,
            "leadership_period": ar.leadership_period,
            "word_count": ar.word_count,
            "article_type_proxy": ar.article_type_proxy,
            "author_count": ar.author_count,
            "first_author": ar.first_author,
            "visible_update_correction_status": ar.visible_update_correction_status,
            "semantic_confidence_numeric": ar.semantic_confidence_numeric,
        })
    return pd.DataFrame(rows)


def design_matrix(df: pd.DataFrame, include_admin: bool = True) -> tuple[np.ndarray, list[str]]:
    parts = [pd.DataFrame({"intercept": np.ones(len(df))}, index=df.index)]
    if include_admin:
        parts.append(pd.DataFrame({"is_admin": df["is_admin"].astype(float)}, index=df.index))
    log_wc = np.log1p(pd.to_numeric(df.word_count, errors="coerce").fillna(0))
    parts.append(pd.DataFrame({
        "log_word_count_centered": log_wc - log_wc.mean(),
        "author_count_centered": pd.to_numeric(df.author_count, errors="coerce").fillna(1) - pd.to_numeric(df.author_count, errors="coerce").fillna(1).mean(),
    }, index=df.index))
    for col in ["subdesk", "primary_topic", "publication_month", "article_type_proxy"]:
        dummies = pd.get_dummies(df[col].astype(str), prefix=col, drop_first=True, dtype=float)
        parts.append(dummies)
    raw = pd.concat(parts, axis=1).astype(float)
    # Deterministically drop columns that do not increase rank.
    kept: list[str] = []
    current = np.empty((len(raw), 0))
    rank = 0
    for col in raw.columns:
        candidate = np.column_stack([current, raw[col].to_numpy()])
        new_rank = np.linalg.matrix_rank(candidate)
        if new_rank > rank:
            kept.append(col)
            current = candidate
            rank = new_rank
    return current, kept


def ols_cluster_model(long_df: pd.DataFrame, outcome: str) -> dict[str, Any]:
    d = long_df.dropna(subset=[outcome, "first_author"]).copy()
    if len(d) < 100 or d.first_author.nunique() < 20 or d[outcome].nunique() < 2:
        return {"status": "not_estimable", "reason": "insufficient observations, author clusters, or outcome variation", "n": int(len(d))}
    X, names = design_matrix(d)
    y = d[outcome].astype(float).to_numpy()
    beta = np.linalg.pinv(X.T @ X) @ X.T @ y
    resid = y - X @ beta
    bread = np.linalg.pinv(X.T @ X)
    meat = np.zeros((X.shape[1], X.shape[1]))
    groups = d.first_author.fillna("UNKNOWN").astype(str).to_numpy()
    unique_groups = np.unique(groups)
    for g in unique_groups:
        idx = groups == g
        score = X[idx].T @ resid[idx]
        meat += np.outer(score, score)
    n, k, G = len(y), X.shape[1], len(unique_groups)
    correction = (G / (G - 1)) * ((n - 1) / max(1, n - k)) if G > 1 else 1.0
    vcov = correction * bread @ meat @ bread
    se = np.sqrt(np.maximum(0, np.diag(vcov)))
    if "is_admin" not in names:
        return {"status": "not_estimable", "reason": "is_admin was collinear", "n": n}
    j = names.index("is_admin")
    z = beta[j] / se[j] if se[j] else np.nan
    return {
        "status": "estimated",
        "model": "OLS/linear-probability with first-author clustered sandwich SE",
        "outcome": outcome,
        "n": int(n),
        "author_clusters": int(G),
        "coefficient_admin_minus_pooled_challenger": float(beta[j]),
        "standard_error": float(se[j]),
        "ci95": [float(beta[j] - 1.959963984540054 * se[j]), float(beta[j] + 1.959963984540054 * se[j])],
        "p_value_two_sided": float(2 * stats.norm.sf(abs(z))) if np.isfinite(z) else None,
        "controls": ["subdesk", "primary_topic", "log word count", "publication month", "article type proxy", "number of authors"],
        "columns_retained": names,
        "note": "Publication month absorbs leadership-period variation; both were not entered together because leadership periods are deterministic functions of month.",
    }


def paired_to_long(pair_df: pd.DataFrame, admin_col: str, chall_col: str, outcome: str) -> pd.DataFrame:
    keep = [
        "article_id", "date", "subdesk", "primary_topic", "publication_month", "leadership_period",
        "word_count", "article_type_proxy", "author_count", "first_author",
    ]
    a = pair_df[keep + [admin_col]].rename(columns={admin_col: outcome}).copy()
    a["is_admin"] = 1
    c = pair_df[keep + [chall_col]].rename(columns={chall_col: outcome}).copy()
    c["is_admin"] = 0
    return pd.concat([a, c], ignore_index=True)


def bh_adjust(pvals: list[float]) -> list[float]:
    n = len(pvals)
    order = np.argsort(pvals)
    out = np.empty(n)
    running = 1.0
    for rank_from_end, idx in enumerate(order[::-1], start=1):
        rank = n - rank_from_end + 1
        running = min(running, pvals[idx] * n / rank)
        out[idx] = running
    return out.tolist()


def secondary_paired(pair_df: pd.DataFrame) -> list[dict[str, Any]]:
    specs = [
        ("actor_quoted_gap", "administration-minus-challenger direct-quote presence"),
        ("first_actor_gap", "administration-minus-challenger first substantive actor"),
        ("marked_attribution_gap", "administration-minus-challenger marked attribution"),
    ]
    rows = []
    pvals = []
    for i, (col, label) in enumerate(specs):
        vals = pd.to_numeric(pair_df[col], errors="coerce").dropna().to_numpy()
        stat = mean_ci_boot(vals, BOOTSTRAP_SEED + 400 + i)
        p = float(stats.ttest_1samp(vals, popmean=0).pvalue) if len(vals) > 1 and np.std(vals) > 0 else 1.0
        stat.update({"outcome": label, "raw_p_value": p})
        rows.append(stat)
        pvals.append(p)
    qs = bh_adjust(pvals)
    for row, q in zip(rows, qs):
        row["bh_q_value"] = q
    return rows


def robustness_checks(article_df: pd.DataFrame, actor_df: pd.DataFrame, strict_pair: pd.DataFrame) -> list[dict[str, Any]]:
    checks: list[dict[str, Any]] = []

    def add(name: str, df: pd.DataFrame, col: str, seed_offset: int, note: str = "") -> None:
        s = mean_ci_boot(pd.to_numeric(df[col], errors="coerce").dropna(), BOOTSTRAP_SEED + seed_offset)
        s.update({"check": name, "estimand": col, "note": note})
        checks.append(s)

    broad_pair = article_pair_table(article_df, actor_df, broad=True)
    add("broad_actor_dictionary", broad_pair, "framing_gap", 10)

    # Recompute paired framing under adverse-plus-conflict and quote-including thresholds.
    for altcol, label, off in [
        ("marked_adverse_plus_conflict", "adverse_plus_conflict_threshold", 20),
        ("marked_include_quotes", "include_quotation_language", 30),
    ]:
        rows = []
        for aid, grp in actor_df.groupby("article_id"):
            a = grp[(grp.actor_category == "administration_trustees") & (grp.strict_present == 1)]
            c = grp[(grp.actor_category.isin(CHALLENGERS)) & (grp.strict_present == 1)]
            if a.empty or c.empty:
                continue
            av = pd.to_numeric(a[altcol], errors="coerce").iloc[0]
            cv = pd.to_numeric(c[altcol], errors="coerce").mean()
            if pd.notna(av) and pd.notna(cv):
                rows.append(av - cv)
        add(label, pd.DataFrame({"gap": rows}), "gap", off)

    add("primary_within_article_paired_subset", strict_pair, "framing_gap", 40)
    filters = [
        ("exclude_breaking_obituary_event_recap_under_300", ~strict_pair.article_type_proxy.eq("breaking_brief") & (strict_pair.word_count >= 300)),
        ("exclude_april_2024_encampment_peak", ~strict_pair.date.between("2024-04-17", "2024-05-31")),
        ("exclude_2025_federal_funding_peak", ~strict_pair.date.between("2025-03-01", "2025-07-31")),
        ("exclude_visible_updates_corrections", strict_pair.visible_update_correction_status.eq("none_observed")),
        ("high_confidence_only", strict_pair.semantic_confidence_numeric >= 0.80),
    ]
    for j, (name, mask) in enumerate(filters, start=5):
        add(name, strict_pair[mask], "framing_gap", 10 * j)
    for j, subdesk in enumerate(sorted(strict_pair.subdesk.unique()), start=20):
        add(f"subdesk_{subdesk}", strict_pair[strict_pair.subdesk == subdesk], "framing_gap", 10 * j)
    for j, actor in enumerate(CHALLENGERS, start=30):
        vals = []
        for aid, grp in actor_df.groupby("article_id"):
            a = grp[(grp.actor_category == "administration_trustees") & (grp.strict_present == 1)]
            c = grp[(grp.actor_category == actor) & (grp.strict_present == 1)]
            if a.empty or c.empty:
                continue
            av = pd.to_numeric(a.marked_strict, errors="coerce").iloc[0]
            cv = pd.to_numeric(c.marked_strict, errors="coerce").iloc[0]
            if pd.notna(av) and pd.notna(cv):
                vals.append(av - cv)
        add(f"challenger_category_{actor}", pd.DataFrame({"gap": vals}), "gap", 10 * j)
    return checks


def validation_sample(article_df: pd.DataFrame) -> pd.DataFrame:
    d = article_df.copy()
    d["length_band"] = pd.cut(d.word_count, bins=[-1, 499, 999, np.inf], labels=["short", "medium", "long"]).astype(str)
    d["topic_screen"] = (d.primary_topic != "other").astype(int)
    d["actor_screen"] = d.paired_actor_article_strict.astype(int)
    d["release_screen"] = d.university_official_origin_auto.astype(int)
    d["marked_screen"] = d.article_id.map(
        actor_df_global.groupby("article_id").marked_strict.apply(lambda x: int(pd.to_numeric(x, errors="coerce").fillna(0).max()))
    ).fillna(0).astype(int)
    d["validation_stratum"] = d[[
        "subdesk", "quarter", "topic_screen", "actor_screen", "release_screen", "length_band", "marked_screen"
    ]].astype(str).agg("|".join, axis=1)
    counts = d.validation_stratum.value_counts().sort_index()
    expected = counts * VALIDATION_N / len(d)
    allocation = np.floor(expected).astype(int)
    remaining = VALIDATION_N - int(allocation.sum())
    remainders = (expected - allocation).sort_values(ascending=False, kind="mergesort")
    for stratum in remainders.index[:remaining]:
        allocation.loc[stratum] += 1
    d["hash_rank"] = d.apply(
        lambda r: hashlib.sha256(f"{VALIDATION_SEED}|{r.validation_stratum}|{r.url}".encode()).hexdigest(), axis=1
    )
    selected = []
    for stratum, n in allocation.items():
        if n:
            selected.append(d[d.validation_stratum == stratum].sort_values("hash_rank").head(int(n)))
    out = pd.concat(selected, ignore_index=True).sort_values(["subdesk", "quarter", "validation_stratum", "hash_rank"])
    out["manual_coding_status"] = "NOT_CODED"
    keep = [
        "article_id", "title", "author", "date", "subdesk", "url", "word_count", "quarter",
        "primary_topic", "article_type_proxy", "leadership_period", "validation_stratum", "hash_rank",
        "manual_coding_status",
    ]
    return out[keep]


def safe_records_for_csv(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    for row in rows:
        for key, value in row.items():
            if "excerpt" in key and value:
                row[key] = excerpt(str(value), 25)
    return rows


def write_csv(path: Path, rows: list[dict[str, Any]]) -> None:
    if not rows:
        raise RuntimeError(f"No rows for {path}")
    fieldnames = list(rows[0].keys())
    with path.open("w", encoding="utf-8", newline="") as f:
        writer = csv.DictWriter(f, fieldnames=fieldnames, extrasaction="raise")
        writer.writeheader()
        writer.writerows(rows)


def main() -> None:
    if sha256_file(ADDENDUM_PATH) != EXPECTED_ADDENDUM_SHA256:
        raise RuntimeError("Frozen preregistration addendum hash mismatch; refusing to calculate outcomes")
    with INDEX_PATH.open(encoding="utf-8") as f:
        index = json.load(f)
    records = [json.loads(line) for line in CORPUS_PATH.read_text(encoding="utf-8").splitlines() if line.strip()]
    index_urls = {a["url"] for a in index["articles"]}
    corpus_urls = {r["canonical_url"] for r in records}
    duplicate_records = len(records) - len(corpus_urls)
    missing_urls = sorted(index_urls - corpus_urls)
    extra_urls = sorted(corpus_urls - index_urls)
    extraction_failures = [r["canonical_url"] for r in records if r.get("extraction_status") != "success"]
    empty_bodies = [r["canonical_url"] for r in records if not (r.get("body") or "").strip()]
    if missing_urls or extra_urls or duplicate_records or extraction_failures or empty_bodies:
        raise RuntimeError(
            f"Census reconciliation failed: missing={len(missing_urls)}, extra={len(extra_urls)}, "
            f"duplicates={duplicate_records}, extraction_failures={len(extraction_failures)}, empty={len(empty_bodies)}"
        )

    article_rows: list[dict[str, Any]] = []
    actor_rows: list[dict[str, Any]] = []
    for record in records:
        article, actors = build_rows(record)
        article_rows.append(article)
        actor_rows.extend(actors)
    safe_records_for_csv(article_rows)
    safe_records_for_csv(actor_rows)
    write_csv(ARTICLE_OUT, article_rows)
    write_csv(ACTOR_OUT, actor_rows)

    article_df = pd.DataFrame(article_rows)
    actor_df = pd.DataFrame(actor_rows)
    global actor_df_global
    actor_df_global = actor_df
    pair_df = article_pair_table(article_df, actor_df, broad=False)

    primary = {
        "paired_scrutiny_gap": {
            **mean_ci_boot(pair_df.scrutiny_gap.dropna(), BOOTSTRAP_SEED + 1),
            "component_means": {
                "administration": float(pair_df.dropna(subset=["scrutiny_gap"]).admin_scrutiny.mean()),
                "pooled_challenger": float(pair_df.dropna(subset=["scrutiny_gap"]).challenger_scrutiny_mean.mean()),
            },
            "definition": "administration scrutiny score minus mean pooled-challenger scrutiny score in articles with claim proxies for both sides; positive means more scrutiny of administration",
        },
        "paired_reporter_framing_gap": {
            **mean_ci_boot(pair_df.framing_gap.dropna(), BOOTSTRAP_SEED + 2),
            "component_means": {
                "administration_marked_rate": float(pair_df.admin_marked.mean()),
                "pooled_challenger_mean_marked_rate": float(pair_df.challenger_marked_mean.mean()),
            },
            "definition": "administration marked reporter-window indicator minus mean pooled-challenger indicator; negative is consistent with hypothesized administrative deference",
        },
        "first_quote_probability_gap": {
            **mean_ci_boot(pair_df.first_quote_gap.dropna(), BOOTSTRAP_SEED + 3),
            "component_means": {
                "administration_first_probability": float(pair_df.admin_first_quote.mean()),
                "pooled_challenger_first_probability": float(pair_df.challenger_first_quote.mean()),
            },
            "definition": "administration-first minus challenger-first quote indicator among paired actor-present articles; no relevant attributed quote contributes zero to both",
        },
        "direct_quote_word_share_gap": {
            **mean_ci_boot(pair_df.quote_share_gap.dropna(), BOOTSTRAP_SEED + 4),
            "component_means": {
                "administration_share": float(pair_df.dropna(subset=["quote_share_gap"]).admin_quote_share.mean()),
                "pooled_challenger_share": float(pair_df.dropna(subset=["quote_share_gap"]).challenger_quote_share.mean()),
            },
            "definition": "administration minus pooled-challenger share of attributed direct-quote words among articles with at least one attributed quote for either group",
        },
    }

    model_specs = [
        ("paired_framing", paired_to_long(pair_df, "admin_marked", "challenger_marked_mean", "outcome")),
        ("paired_scrutiny", paired_to_long(pair_df.dropna(subset=["admin_scrutiny", "challenger_scrutiny_mean"]), "admin_scrutiny", "challenger_scrutiny_mean", "outcome")),
        ("first_quote_probability", paired_to_long(pair_df, "admin_first_quote", "challenger_first_quote", "outcome")),
        ("quote_word_share", paired_to_long(pair_df.dropna(subset=["admin_quote_share", "challenger_quote_share"]), "admin_quote_share", "challenger_quote_share", "outcome")),
    ]
    adjusted = {name: ols_cluster_model(d, "outcome") for name, d in model_specs}
    for model in adjusted.values():
        model["specification_status"] = "POST_OUTCOME_EXPLORATORY_TOPIC_PROXY_PARSER_CORRECTION"

    secondary_rates = {}
    for col in [
        "university_official_origin_auto", "high_release_reliance_auto", "documents_records_data_auto",
        "affected_people_consulted_auto", "power_history_consequences_context_auto", "follow_up_orientation_auto",
    ]:
        vals = pd.to_numeric(article_df[col], errors="coerce").dropna().astype(int)
        x, n = int(vals.sum()), int(len(vals))
        secondary_rates[col] = {"n": n, "count": x, "proportion": x / n if n else None, "wilson_ci95": wilson_ci(x, n)}

    validation = validation_sample(article_df)
    validation.to_csv(VALIDATION_OUT, index=False, encoding="utf-8")
    if len(validation) != VALIDATION_N:
        raise RuntimeError(f"Validation sample has {len(validation)} rows, expected {VALIDATION_N}")

    date_counts = article_df.groupby(["subdesk", "publication_month"]).size().reset_index(name="n").to_dict("records")
    topic_counts = article_df.groupby(["subdesk", "primary_topic"]).size().reset_index(name="n").to_dict("records")
    actor_counts = actor_df.groupby(["actor_category", "strict_present", "broad_present"]).size().reset_index(name="n").to_dict("records")
    word_diff = pd.to_numeric(article_df.word_count) - pd.to_numeric(article_df.computed_word_count)
    word_count_mismatch = int((word_diff != 0).sum())
    max_excerpt_words = 0
    for row in article_rows + actor_rows:
        for k, v in row.items():
            if "excerpt" in k and v:
                max_excerpt_words = max(max_excerpt_words, len(str(v).split()))

    results = {
        "study": {
            "name": "Authorized University News census automated screening analysis",
            "study_start": STUDY_START,
            "study_end": STUDY_END,
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "preregistration_addendum_sha256": sha256_file(ADDENDUM_PATH),
            "private_corpus_sha256": sha256_file(CORPUS_PATH),
            "index_sha256": sha256_file(INDEX_PATH),
            "analysis_script_sha256": None,
            "interpretation_status": "AUTOMATED_SCREENING_PENDING_MANUAL_VALIDATION",
            "warning": "Semantic estimates must be suppressed as population findings if relevant automated-versus-manual reliability is below kappa 0.60.",
            "method_change_log": "work/university-automated-method-change-log.md",
        },
        "corpus_reconciliation": {
            "official_archive_urls": len(index_urls),
            "private_records": len(records),
            "unique_private_urls": len(corpus_urls),
            "successful_extractions": len(records) - len(extraction_failures),
            "missing_urls": len(missing_urls),
            "extra_urls": len(extra_urls),
            "duplicate_records": duplicate_records,
            "empty_bodies": len(empty_bodies),
            "census_status": "complete_reconciled_census",
            "subdesk_counts": article_df.subdesk.value_counts().sort_index().to_dict(),
            "date_range": [article_df.date.min(), article_df.date.max()],
        },
        "parser_quality": {
            "fatal_parser_failures": 0,
            "stored_word_count_vs_computed_mismatch_articles": word_count_mismatch,
            "word_count_difference_stored_minus_regex_token_count": {
                "median": float(word_diff.median()),
                "mean": float(word_diff.mean()),
                "minimum": int(word_diff.min()),
                "maximum": int(word_diff.max()),
                "median_absolute_difference": float(word_diff.abs().median()),
                "explanation": "The crawler's stored count and the analysis regex tokenize punctuation, typographic apostrophes, and markup differently; this is a count-definition discrepancy, not missing full text.",
            },
            "quote_attribution_mean_confidence": float(article_df.quote_attribution_confidence.mean()),
            "quote_attribution_low_confidence_articles": int((article_df.quote_attribution_confidence < 0.60).sum()),
            "semantic_low_confidence_articles": int((article_df.semantic_confidence == "low").sum()),
            "maximum_output_evidence_excerpt_words": max_excerpt_words,
            "body_text_written_to_outputs": False,
            "known_limitations": [
                "Quote attribution is a rule-based proxy and can miss delayed, multi-paragraph, collective, or name-only attribution.",
                "Faculty critics require a faculty role and critic/advocacy signal in the same paragraph under the strict dictionary.",
                "Official-origin detection identifies textual provenance cues; it cannot prove that a press release supplied the wording.",
                "Assertion-testing and context variables require semantic judgment and remain screening variables pending manual validation.",
                "Reporter-language masking excludes text in typographic quotation marks but may miss malformed or unmatched quotation marks.",
                "Source counts are conservative actor/type proxies, not verified unique-human-source counts.",
                "Image-based identification, consent, foreseeable harm, and silent edits cannot be evaluated from article body text alone.",
            ],
        },
        "automated_actor_presence": {
            "paired_articles_strict": int(article_df.paired_actor_article_strict.sum()),
            "paired_articles_broad": int(article_df.paired_actor_article_broad.sum()),
            "actor_rows": int(len(actor_df)),
            "counts": actor_counts,
        },
        "primary_estimands": primary,
        "adjusted_models": adjusted,
        "secondary_article_rates": secondary_rates,
        "secondary_paired_actor_outcomes_bh_adjusted": secondary_paired(pair_df),
        "robustness_checks": robustness_checks(article_df, actor_df, pair_df),
        "manual_validation_sample": {
            "n": len(validation),
            "share_of_census": len(validation) / len(article_df),
            "seed": VALIDATION_SEED,
            "method": "proportional largest-remainder allocation across the preregistered combined strata, then SHA-256 rank within stratum",
            "output": str(VALIDATION_OUT.relative_to(PACKAGE_ROOT)),
        },
        "missingness_breakdowns": {
            "by_subdesk_month": date_counts,
            "by_subdesk_primary_topic": topic_counts,
            "note": "All 1,527 full-text records were accessible; these tables describe corpus composition, not missing full text.",
        },
        "fixed_rule_inventory": {
            "actor_strict": ACTOR_PATTERNS_STRICT,
            "actor_broad": ACTOR_PATTERNS_BROAD,
            "evaluative_lexicon": EVAL_TERMS,
            "attribution_verbs": ATTRIBUTION_TERMS,
            "official_origin_patterns": OFFICIAL_ORIGIN_PATTERNS,
            "document_patterns": DOCUMENT_PATTERNS,
            "testing_patterns": TESTING_PATTERNS,
            "context_patterns": CONTEXT_PATTERNS,
            "followup_patterns": FOLLOWUP_PATTERNS,
            "topic_patterns": TOPIC_PATTERNS,
            "peak_exclusion_windows": {
                "april_2024_encampment_nypd": ["2024-04-17", "2024-05-31"],
                "2025_federal_funding": ["2025-03-01", "2025-07-31"],
            },
        },
    }
    # Record the final script hash after calculation rules are defined. This changes only
    # if the file itself changes, not with outcome values.
    results["study"]["analysis_script_sha256"] = sha256_file(Path(__file__))
    RESULTS_OUT.write_text(json.dumps(results, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")

    # Final no-leak and referential-integrity checks.
    if max_excerpt_words > 25:
        raise RuntimeError(f"Evidence excerpt exceeded 25 words: {max_excerpt_words}")
    if set(actor_df.article_id) - set(article_df.article_id):
        raise RuntimeError("Actor output contains unknown article IDs")
    if article_df.article_id.duplicated().any():
        raise RuntimeError("Duplicate article IDs")
    if len(article_df) != 1527:
        raise RuntimeError(f"Expected 1,527 article rows, got {len(article_df)}")


if __name__ == "__main__":
    main()
