imbad0202__academic-research-skills/scripts/claim_audit_pipeline.py

"""claim_audit_pipeline — Python implementation of the §4 Step 1-6 pipeline.

This module is the executable face of `claim_ref_alignment_audit_agent.md`.
The agent prompt narrates the pipeline contract; this module runs it
under test so cross-field invariants and emission routing can be pinned
without dispatching the agent to a live model.

Retrieval and judge invocation are dependency-injected (`retrieve_fn` /
`judge_fn`) so tests can drive every error and decision path — paywall,
audit_tool_failure, not_found, SUPPORTED, UNSUPPORTED with each
defect_stage hint, VIOLATED. Production callers wire these to real
retrieval/judge clients in their own dispatch layer.

The full spec is in
docs/design/2026-05-15-issue-103-claim-alignment-audit-spec.md §4-§5.
"""
from __future__ import annotations

import hashlib
import json
import sys
from pathlib import Path
from typing import Any, Callable

# Allow both CLI invocations (`python3 scripts/claim_audit_pipeline.py`) AND
# package-style invocations (`python -m unittest scripts.test_*`) to resolve
# the shared constants module via the same import.
sys.path.insert(0, str(Path(__file__).resolve().parent))
from _claim_audit_constants import (  # noqa: E402
    DRIFT_RULE_VERSION,
    INV6_RATIONALE_PREFIX,
    JUDGE_PROMPT_SHA256,
    RE_NC_CONSTRAINT,
    SAMPLING_STRATEGY,
    SENTINEL_MANIFEST_ID,
    UAF_RULE_VERSION,
    UNCITED_RULE_VERSION,
    is_emittable_partial_breakdown,
)

# Permitted UNSUPPORTED defect_stages for non-constraint paths (§3.1 matrix).
_UNSUPPORTED_NON_CONSTRAINT_DEFECTS = {
    "source_description",
    "metadata",
    "citation_anchor",
    "synthesis_overclaim",
}

# Permitted AMBIGUOUS defect_stages (§3.1 matrix).
_AMBIGUOUS_DEFECTS = {"source_description", "citation_anchor", "synthesis_overclaim", None}


# ---------------------------------------------------------------------------
# Cache helpers.
# ---------------------------------------------------------------------------


def _stable_json(value: Any) -> str:
    """JCS-style canonicalization sufficient for cache-key hashing."""
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False)


def _hash_text(text: str | None) -> str:
    return hashlib.sha256((text or "").encode("utf-8")).hexdigest()


# claim_audit_result.rationale schema maxLength (#355 P2#3). A judge/retrieval
# failure detail that embeds the offending payload's repr must fit here, or the
# "clean inconclusive" fallback row it produces is itself schema-invalid. Every
# JudgeInvocationError / RetrievalInvocationError detail ends up in a row's
# rationale via `f"{fault_class}: {detail}"`, so the bound lives in those
# exceptions' constructors (single choke point) rather than at each raise site.
_RATIONALE_MAX_LEN = 2000
_RATIONALE_TRUNC_MARK = "…[truncated]"
# Widest fault-class prefix across both exception families ("retrieval_network_error: ").
_WIDEST_FAULT_PREFIX = "retrieval_network_error: "

# #512 PDF read-integrity rationale tag. Appended (never prepended — INV-6/INV-14
# startswith contracts stay intact) to a completed manual_pdf page-anchor row whose
# preflight sidecar is missing or non-PASS. The finalizer keys its advisory
# annotation on this exact substring; keep in lockstep with
# claim_audit_finalizer.PDF_READ_INTEGRITY_TAG.
PDF_READ_INTEGRITY_TAG = "[pdf_read_integrity_unverified]"


def _tag_pdf_read_integrity(entry: dict[str, Any]) -> None:
    """#512: mark a manual_pdf page-anchor row whose preflight did not vouch.

    Applied at the single Step-6 emission point AFTER cache resolution, so a cache
    hit cannot bypass it (the tag is run-context, never written into the judge
    cache body). Idempotent; respects the rationale maxLength budget."""
    rationale = entry.get("rationale") or ""
    if PDF_READ_INTEGRITY_TAG in rationale:
        return
    if rationale:
        clamped = _clamp_to_rationale_budget(
            rationale, reserved=len(PDF_READ_INTEGRITY_TAG) + 1
        )
        entry["rationale"] = f"{clamped} {PDF_READ_INTEGRITY_TAG}"
    else:
        entry["rationale"] = PDF_READ_INTEGRITY_TAG


def _clamp_to_rationale_budget(text: str, *, reserved: int) -> str:
    """Clamp `text` so that `reserved + len(result)` fits the rationale maxLength.

    Single length-budgeting choke point for every untrusted string that lands in
    a row's `rationale` (#355 P2#3 / #360). `reserved` is the number of chars the
    caller will prepend before this text reaches the rationale field:

    - Failure paths emit ``f"{fault_class}: {detail}"`` → reserved = widest
      fault-class prefix width, so the worst-case composed rationale still fits.
    - Success paths copy a judge's own `rationale` verbatim onto the row → no
      prefix → reserved = 0.

    Truncation preserves the diagnostic head (which says WHAT the string is) and
    marks the dropped tail, so a short string that already fits passes through
    byte-for-byte.
    """
    budget = _RATIONALE_MAX_LEN - reserved
    if len(text) <= budget:
        return text
    keep = budget - len(_RATIONALE_TRUNC_MARK)
    return text[:keep] + _RATIONALE_TRUNC_MARK


def _bounded_failure_detail(message: str) -> str:
    """Clamp a failure detail so ``f"{fault_class}: {detail}"`` fits the maxLength.

    A malformed payload's repr embedded in `message` (a >1000-char
    sub_claim_text, an over-decomposed breakdown, a giant non-string
    judgment/method) can push the detail past the rationale maxLength, making the
    fallback row schema-invalid (#355 P2#3). Budgets against the widest
    fault-class prefix so the composed rationale fits for every fault class.
    """
    return _clamp_to_rationale_budget(message, reserved=len(_WIDEST_FAULT_PREFIX))


def _bounded_judge_rationale(rationale: Any) -> str:
    """Clamp a judge-supplied `rationale` copied verbatim onto a SUCCESS-path row.

    The judge is an LLM with no pre-emission length guarantee; an over-long
    `rationale` makes a *clean* completed / constraint_violation row
    schema-invalid (#360 — the success-path parallel to the #359 fallback fix).
    No fault-class prefix is prepended on the success path, so reserved = 0.

    `_validate_judge_dict` only checks that the `rationale` key is present, not
    that its value is a string — a JSON-null (or otherwise non-string) rationale
    passes that gate. Return "" for a non-string value so each caller's existing
    fallback (`... or "(no rationale provided)"` / the constraint default) takes
    over, rather than calling len() on it and aborting the audit run.
    """
    if not isinstance(rationale, str):
        return ""
    return _clamp_to_rationale_budget(rationale, reserved=0)


def _active_constraints_for_claim(
    *,
    scoped_manifest_id: str,
    claim_id: str,
    claim_by_mc_id: dict[tuple[str, str], dict[str, Any]],
    mncs_by_manifest_id: dict[str, list[dict[str, Any]]],
) -> list[dict[str, Any]]:
    """Return the manifest-scoped + claim-scoped negative-constraint set for a citation.

    Reads from pre-built indexes (built once per audit run in
    `run_audit_pipeline`) instead of rescanning the manifest tree per
    citation — at realistic workloads (~150 citations × ~100 manifest
    claims) that saves ~30k Python ops per run with no behavioral delta.
    """
    constraints: list[dict[str, Any]] = []
    for mnc in mncs_by_manifest_id.get(scoped_manifest_id, []):
        constraints.append({"constraint_id": mnc["constraint_id"], "rule": mnc["rule"], "scope": "MNC"})
    claim = claim_by_mc_id.get((scoped_manifest_id, claim_id))
    if claim is not None:
        for nc in claim.get("negative_constraints", []) or []:
            constraints.append(
                {"constraint_id": nc["constraint_id"], "rule": nc["rule"], "scope": "NC"}
            )
    constraints.sort(key=lambda c: c["constraint_id"])
    return constraints


def _cache_key(
    *,
    claim_text: str,
    ref_slug: str,
    anchor_kind: str,
    anchor_value: str,
    retrieved_excerpt: str | None,
    active_constraints: list[dict[str, Any]],
    judge_model: str,
    prompt_version: str,
) -> str:
    payload = {
        "claim_text_hash": _hash_text(claim_text),
        "ref_slug": ref_slug,
        "anchor_kind": anchor_kind,
        "anchor_value_hash": _hash_text(anchor_value),
        "retrieved_excerpt_hash": _hash_text(retrieved_excerpt),
        "active_constraints_hash": _hash_text(
            _stable_json([{"constraint_id": c["constraint_id"], "rule": c["rule"]} for c in active_constraints])
        ),
        "judge_model": judge_model,
        # #361: a judge-prompt revision partitions the keyspace — a verdict
        # cached under one prompt is never served against new prompt logic.
        # judge_model and prompt_version stay separate components (independent
        # axes of judge behavior).
        "prompt_version": prompt_version,
    }
    return hashlib.sha256(_stable_json(payload).encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# Judge + retrieval invocation — wraps callables so transient failures become
# INV-14 audit_tool_failure rows instead of aborting the audit. Spec §4 step 2
# + INV-14 + Step 13 R1+R2 codex findings (transient errors on either external
# call must surface as MED-WARN advisory rows).
# ---------------------------------------------------------------------------

# Legal judge verdicts per path. claim_audit_result schema enum + constraint-side
# verdicts (VIOLATED / NOT_VIOLATED). Cited and uncited paths route different
# subsets — passing path-specific allow-lists into `_invoke_judge` rejects an
# off-path verdict at the invocation boundary instead of letting it propagate
# into _judge_result_entry where the ValueError would abort the audit
# (Step 13 R3 codex P2 #2).
_CITED_PATH_JUDGMENTS: frozenset[str] = frozenset(
    {"SUPPORTED", "UNSUPPORTED", "AMBIGUOUS", "PARTIAL", "VIOLATED"}
)
# PARTIAL is a cited-path-only verdict (a reference can support some sub-claims
# but not others). The uncited path has no reference to be partial against, so
# it stays the constraint VIOLATED/NOT_VIOLATED binary (#213).
_UNCITED_PATH_JUDGMENTS: frozenset[str] = frozenset({"VIOLATED", "NOT_VIOLATED"})


class _AuditInvocationError(Exception):
    """Base for audit-tool invocation failures (judge / retrieval).

    Carries an INV-14 fault-class tag + detail. The detail is bounded so the
    `f"{fault_class}: {detail}"` rationale it becomes fits the claim_audit_result
    schema maxLength (#355 P2#3) — every subclass's detail flows to a row
    rationale, so the bound lives here (single choke point).
    """

    def __init__(self, fault_class: str, detail: str) -> None:
        detail = _bounded_failure_detail(detail)
        super().__init__(f"{fault_class}: {detail}")
        self.fault_class = fault_class
        self.detail = detail


class JudgeInvocationError(_AuditInvocationError):
    """Raised by `_invoke_judge` when judge_fn fails or returns malformed output.

    Carries the INV-14 fault-class tag + detail so the caller can emit a
    `RETRIEVAL_FAILED + inconclusive + not_applicable + audit_tool_failure`
    row per spec §4 step 2 + INV-14 instead of letting the exception abort the
    audit pass.
    """


class RetrievalInvocationError(_AuditInvocationError):
    """Raised by `_invoke_retrieve` when retrieve_fn fails or returns malformed output.

    Mirrors JudgeInvocationError but tags faults with the INV-14 retrieval_*
    family (retrieval_api_error / retrieval_timeout / retrieval_network_error)
    so a transient retrieval outage surfaces as audit_tool_failure rather than
    aborting the audit pass (Step 13 R2 codex P2 finding).
    """


def _validate_judge_dict(
    result: Any,
    *,
    allowed_judgments: frozenset[str],
    active_constraint_ids: frozenset[str],
    source: str = "judge_fn",
) -> dict[str, Any]:
    """Validate a judge-output dict (fresh or cache-hit).

    Raises `JudgeInvocationError` with the appropriate fault-class tag for any
    shape violation. Pulled out of `_invoke_judge` so cache hits can reuse the
    same validation surface — without it a malformed cache entry would crash
    `_judge_result_entry` and abort the audit (Step 13 R3 codex P2 #4).

    Validation surface:
      - non-dict                                  → judge_parse_error
      - missing `judgment` or `rationale`         → judge_parse_error
      - judgment not in `allowed_judgments`       → judge_parse_error
      - VIOLATED without non-blank string id      → judge_parse_error
      - VIOLATED id not in `active_constraint_ids`→ judge_parse_error
        (Step 13 R3 codex P2 #1 — prevents formatter gate-refuse on a
        hallucinated constraint the author never declared)
    """
    if not isinstance(result, dict):
        raise JudgeInvocationError(
            "judge_parse_error",
            f"{source} returned {type(result).__name__}, expected dict",
        )
    if "judgment" not in result or "rationale" not in result:
        raise JudgeInvocationError(
            "judge_parse_error",
            f"{source} returned dict missing required key(s); got keys={sorted(result)}",
        )
    judgment = result.get("judgment")
    # Step 13 R8 codex P2-3: guard isinstance(str) before set membership so a
    # malformed return like {"judgment": [1, 2], ...} surfaces as a clean
    # judge_parse_error instead of bubbling TypeError("unhashable type") out
    # past the translation boundary and aborting the audit.
    if not isinstance(judgment, str):
        raise JudgeInvocationError(
            "judge_parse_error",
            f"{source} returned non-string judgment={judgment!r} (type={type(judgment).__name__}); expected one of {sorted(allowed_judgments)}",
        )
    if judgment not in allowed_judgments:
        raise JudgeInvocationError(
            "judge_parse_error",
            f"{source} returned judgment={judgment!r}; expected one of {sorted(allowed_judgments)} on this path",
        )
    if judgment == "VIOLATED":
        vcid = result.get("violated_constraint_id")
        if not isinstance(vcid, str) or not vcid.strip():
            raise JudgeInvocationError(
                "judge_parse_error",
                f"{source} returned VIOLATED without a valid violated_constraint_id (got {vcid!r}); INV-7 requires non-null id",
            )
        if vcid not in active_constraint_ids:
            raise JudgeInvocationError(
                "judge_parse_error",
                f"{source} returned VIOLATED with violated_constraint_id={vcid!r} outside the active constraint set {sorted(active_constraint_ids)}; rejecting hallucinated id (Step 13 R3 codex P2 #1)",
            )
    if judgment == "PARTIAL":
        # #213: a PARTIAL MUST carry a well-formed true-partial sub_claim_breakdown.
        # A malformed PARTIAL is a judge-output parse failure — routing it through
        # judge_parse_error yields the existing (RETRIEVAL_FAILED, inconclusive,
        # not_applicable, audit_tool_failure) row, never a silent bare UNSUPPORTED
        # (which would recreate the invisible-trap failure). The malformed-PARTIAL
        # path has no new matrix triple; it reuses the judge_parse_error contract.
        # is_emittable_*: true-partial mix AND every item schema-shaped (non-empty
        # sub_claim_text + valid sub_verdict). The item-shape half is required
        # because _judge_result_entry copies items onto a *completed* row; a
        # mix-valid-but-malformed item (e.g. missing sub_claim_text) would emit a
        # schema-invalid row instead of taking the judge_parse_error path
        # (ship-gate round-2 finding).
        if not is_emittable_partial_breakdown(result.get("sub_claim_breakdown")):
            raise JudgeInvocationError(
                "judge_parse_error",
                f"{source} returned PARTIAL without an emittable true-partial sub_claim_breakdown "
                f"(>=2 schema-shaped items, >=1 SUPPORTED AND >=1 non-SUPPORTED, each with a "
                f"non-empty sub_claim_text); got {result.get('sub_claim_breakdown')!r}",
            )
    return result


def _invoke_judge(
    judge_fn: Callable[..., dict[str, Any]],
    *,
    allowed_judgments: frozenset[str],
    active_constraint_ids: frozenset[str],
    **call_kwargs: Any,
) -> dict[str, Any]:
    """Invoke `judge_fn` and translate transient failures + malformed output into INV-14 tags.

    Exception → fault class mapping:
      - TimeoutError                          → judge_timeout
      - json.JSONDecodeError / ValueError     → judge_parse_error
      - any other Exception                   → judge_api_error

    Return-value validation delegates to `_validate_judge_dict` so cache hits
    reuse the same surface (Step 13 R3 codex P2 #4).

    `allowed_judgments` is path-specific: cited path passes
    `_CITED_PATH_JUDGMENTS`, uncited (constraint) path passes
    `_UNCITED_PATH_JUDGMENTS`. RETRIEVAL_FAILED / NOT_VIOLATED on the cited
    path is rejected here instead of crashing in `_judge_result_entry`
    (Step 13 R3 codex P2 #2).

    `active_constraint_ids` carries the in-scope MNC/NC ids for this call
    so a VIOLATED with a hallucinated id is rejected at the boundary
    (Step 13 R3 codex P2 #1).

    Does NOT swallow `SystemExit` / `KeyboardInterrupt`.
    """
    try:
        result = judge_fn(**call_kwargs)
    except TimeoutError as exc:
        raise JudgeInvocationError("judge_timeout", str(exc) or "judge timed out") from exc
    except (json.JSONDecodeError, ValueError) as exc:
        raise JudgeInvocationError("judge_parse_error", str(exc) or "judge returned malformed payload") from exc
    except Exception as exc:  # noqa: BLE001 — translation boundary; the source class is captured
        raise JudgeInvocationError("judge_api_error", f"{type(exc).__name__}: {exc}") from exc

    return _validate_judge_dict(
        result,
        allowed_judgments=allowed_judgments,
        active_constraint_ids=active_constraint_ids,
        source="judge_fn",
    )


def _invoke_retrieve(
    retrieve_fn: Callable[[dict[str, Any]], dict[str, Any]],
    citation: dict[str, Any],
) -> dict[str, Any]:
    """Invoke `retrieve_fn` and translate transient failures + malformed output into INV-14 retrieval_* tags.

    Exception → fault class mapping (mirrors `_invoke_judge`):
      - TimeoutError                          → retrieval_timeout
      - OSError / ConnectionError             → retrieval_network_error
      - json.JSONDecodeError / ValueError     → retrieval_api_error
      - any other Exception                   → retrieval_api_error

    Return-value validation:
      - non-dict                              → retrieval_api_error
      - missing `ref_retrieval_method` key    → retrieval_api_error
      - unknown `ref_retrieval_method` value  → retrieval_api_error

    Returns the retrieval dict on success; raises RetrievalInvocationError
    otherwise so the caller can map it to an audit_tool_failure row.
    """
    try:
        result = retrieve_fn(citation)
    except TimeoutError as exc:
        raise RetrievalInvocationError("retrieval_timeout", str(exc) or "retrieve_fn timed out") from exc
    except (ConnectionError, OSError) as exc:
        raise RetrievalInvocationError(
            "retrieval_network_error",
            f"{type(exc).__name__}: {exc}" if str(exc) else type(exc).__name__,
        ) from exc
    except (json.JSONDecodeError, ValueError) as exc:
        raise RetrievalInvocationError(
            "retrieval_api_error",
            str(exc) or "retrieve_fn returned malformed payload",
        ) from exc
    except Exception as exc:  # noqa: BLE001 — translation boundary
        raise RetrievalInvocationError("retrieval_api_error", f"{type(exc).__name__}: {exc}") from exc

    if not isinstance(result, dict):
        raise RetrievalInvocationError(
            "retrieval_api_error",
            f"retrieve_fn returned {type(result).__name__}, expected dict",
        )
    method = result.get("ref_retrieval_method")
    if method is None:
        raise RetrievalInvocationError(
            "retrieval_api_error",
            f"retrieve_fn return missing ref_retrieval_method; got keys={sorted(result)}",
        )
    # Step 13 R8 codex P2-4: guard isinstance(str) before set membership so a
    # malformed return like {"ref_retrieval_method": [...], ...} surfaces as a
    # clean retrieval_api_error instead of bubbling TypeError("unhashable type")
    # out past the translation boundary and aborting the audit (symmetric to
    # P2-3 on the judge side).
    if not isinstance(method, str):
        raise RetrievalInvocationError(
            "retrieval_api_error",
            f"retrieve_fn returned non-string ref_retrieval_method={method!r} (type={type(method).__name__})",
        )
    if method not in {"api", "manual_pdf", "failed", "not_found", "audit_tool_failure"}:
        raise RetrievalInvocationError(
            "retrieval_api_error",
            f"retrieve_fn returned unknown ref_retrieval_method={method!r}",
        )
    # Step 13 R3 codex P2 #3: a successful retrieval pathway MUST carry a
    # non-empty excerpt — otherwise the judge would be invoked with
    # `retrieved_excerpt=None`/empty and could mark a claim SUPPORTED with no
    # source text. Map this shape violation to retrieval_api_error so it
    # surfaces as audit_tool_failure instead of silently degrading the audit.
    if method in {"api", "manual_pdf"}:
        excerpt = result.get("retrieved_excerpt")
        if not isinstance(excerpt, str) or not excerpt.strip():
            raise RetrievalInvocationError(
                "retrieval_api_error",
                f"retrieve_fn returned ref_retrieval_method={method!r} with empty/missing retrieved_excerpt; successful retrievals must carry source text",
            )
    return result


# ---------------------------------------------------------------------------
# Emission helpers — each builds one entry dict.
# ---------------------------------------------------------------------------


def _anchorless_entry(citation: dict[str, Any], *, audit_run_id: str, now_iso: str, judge_model: str) -> dict[str, Any]:
    """§4 Step 1: anchor=none short-circuits to RETRIEVAL_FAILED+inconclusive+not_applicable+not_attempted.

    INV-6 sentinel: anchor_kind=none MUST carry anchor_value="" (empty sentinel
    per claim_audit_result.schema.json). We pin the empty string here rather
    than passing through citation.anchor_value — a stale residual anchor like
    "123" on an anchor_kind=none row violates the schema contract.
    """
    return {
        "claim_id": citation["claim_id"],
        "scoped_manifest_id": citation.get("scoped_manifest_id", SENTINEL_MANIFEST_ID),
        "claim_text": citation["claim_text"],
        "ref_slug": citation["ref_slug"],
        "anchor_kind": "none",
        "anchor_value": "",
        "judgment": "RETRIEVAL_FAILED",
        "audit_status": "inconclusive",
        "defect_stage": "not_applicable",
        "rationale": (
            f"{INV6_RATIONALE_PREFIX}: cited claim {citation['claim_id']} carries anchor=none; "
            "v3.7.3 finalizer should have gate-refused upstream — defense-in-depth row."
        ),
        "judge_model": judge_model,
        "judge_run_at": now_iso,
        "ref_retrieval_method": "not_attempted",
        "upstream_owner_agent": citation.get("upstream_owner_agent"),
        "audit_run_id": audit_run_id,
    }


def _retrieval_failure_entry(
    citation: dict[str, Any],
    *,
    method: str,
    audit_run_id: str,
    now_iso: str,
    judge_model: str,
    fault_class: str | None = None,
) -> dict[str, Any]:
    """§4 Step 2: retrieval-side failure routes that skip the judge."""
    if method == "failed":
        # D2 paywall — LOW-WARN advisory; INV-10.
        entry = {
            "judgment": "RETRIEVAL_FAILED",
            "audit_status": "inconclusive",
            "defect_stage": "not_applicable",
            "rationale": "Reference full text not retrievable (paywall / license-restricted access).",
        }
    elif method == "not_found":
        # Fabricated reference — HIGH-WARN; INV-12.
        entry = {
            "judgment": "RETRIEVAL_FAILED",
            "audit_status": "completed",
            "defect_stage": "retrieval_existence",
            "rationale": "Retrieval API reports the cited reference does not exist (suspected fabrication).",
        }
    elif method == "audit_tool_failure":
        # Transient infrastructure outage — MED-WARN; INV-14.
        tag = fault_class or "retrieval_api_error"
        entry = {
            "judgment": "RETRIEVAL_FAILED",
            "audit_status": "inconclusive",
            "defect_stage": "not_applicable",
            "rationale": f"{tag}: transient audit-infrastructure failure during retrieval; retry on next pipeline pass.",
        }
    else:  # pragma: no cover — should be unreachable given Step 2 caller dispatch
        raise ValueError(f"_retrieval_failure_entry called with non-failure method={method!r}")

    return {
        "claim_id": citation["claim_id"],
        "scoped_manifest_id": citation.get("scoped_manifest_id", SENTINEL_MANIFEST_ID),
        "claim_text": citation["claim_text"],
        "ref_slug": citation["ref_slug"],
        "anchor_kind": citation["anchor_kind"],
        "anchor_value": citation.get("anchor_value", ""),
        "judgment": entry["judgment"],
        "audit_status": entry["audit_status"],
        "defect_stage": entry["defect_stage"],
        "rationale": entry["rationale"],
        "judge_model": judge_model,
        "judge_run_at": now_iso,
        "ref_retrieval_method": method,
        "upstream_owner_agent": citation.get("upstream_owner_agent"),
        "audit_run_id": audit_run_id,
    }


def _judge_result_entry(
    citation: dict[str, Any],
    *,
    judge_result: dict[str, Any],
    ref_retrieval_method: str,
    audit_run_id: str,
    now_iso: str,
    judge_model: str,
) -> dict[str, Any]:
    """§4 Steps 5-6: route judge verdict to the right (judgment, defect_stage) row."""
    verdict = judge_result["judgment"]
    # #360: a judge is an LLM with no length guarantee; clamp its rationale to
    # the schema maxLength before it lands on the completed row (success path).
    rationale = _bounded_judge_rationale(judge_result.get("rationale", ""))

    if verdict == "SUPPORTED":
        judgment, defect_stage, violated_id = "SUPPORTED", None, None
    elif verdict == "AMBIGUOUS":
        hint = judge_result.get("defect_stage_hint")
        if hint not in _AMBIGUOUS_DEFECTS:
            hint = None  # AMBIGUOUS+disallowed defect → coerce to null (INV-3 protection)
        judgment, defect_stage, violated_id = "AMBIGUOUS", hint, None
    elif verdict == "UNSUPPORTED":
        hint = judge_result.get("defect_stage_hint") or "source_description"
        if hint not in _UNSUPPORTED_NON_CONSTRAINT_DEFECTS:
            hint = "source_description"
        judgment, defect_stage, violated_id = "UNSUPPORTED", hint, None
    elif verdict == "VIOLATED":
        # Cited constraint violation — INV-7/INV-8 path.
        judgment = "UNSUPPORTED"
        defect_stage = "negative_constraint_violation"
        violated_id = judge_result.get("violated_constraint_id")
    elif verdict == "PARTIAL":
        # #213 B1 normalization: a prompt-verdict PARTIAL becomes
        # judgment=UNSUPPORTED, defect_stage=source_description, carrying the
        # sub_claim_breakdown[] (the machine-readable partial signal). Routing
        # to UNSUPPORTED puts the unsupported sub-claim through the same
        # gate-refuse path a fully-unsupported claim takes. The breakdown shape
        # was validated true-partial in _validate_judge_dict, so INV-19 holds on
        # the emitted completed row.
        judgment = "UNSUPPORTED"
        defect_stage = "source_description"
        violated_id = None
    else:
        raise ValueError(f"unknown judge verdict: {verdict!r}")

    entry: dict[str, Any] = {
        "claim_id": citation["claim_id"],
        "scoped_manifest_id": citation.get("scoped_manifest_id", SENTINEL_MANIFEST_ID),
        "claim_text": citation["claim_text"],
        "ref_slug": citation["ref_slug"],
        "anchor_kind": citation["anchor_kind"],
        "anchor_value": citation.get("anchor_value", ""),
        "judgment": judgment,
        "audit_status": "completed",
        "defect_stage": defect_stage,
        "rationale": rationale or "(no rationale provided)",
        "judge_model": judge_model,
        "judge_run_at": now_iso,
        "ref_retrieval_method": ref_retrieval_method,
        "upstream_owner_agent": citation.get("upstream_owner_agent"),
        "audit_run_id": audit_run_id,
    }
    if violated_id is not None:
        entry["violated_constraint_id"] = violated_id
    if verdict == "PARTIAL":
        # Carry the decomposition onto the emitted row. Normalize each item to
        # the schema item shape (sub_claim_text, sub_verdict, optional
        # evidence_pointer), dropping any extra keys the judge added so the
        # additionalProperties:false item schema holds. Presence of this field
        # is the machine-readable partial-support signal (#213).
        entry["sub_claim_breakdown"] = [
            {
                "sub_claim_text": item.get("sub_claim_text"),
                "sub_verdict": item.get("sub_verdict"),
                **(
                    {"evidence_pointer": item["evidence_pointer"]}
                    if "evidence_pointer" in item
                    else {}
                ),
            }
            for item in judge_result["sub_claim_breakdown"]
            if isinstance(item, dict)
        ]
    return entry


def _uncited_audit_failure_entry(
    *,
    sentence: dict[str, Any],
    scoped_manifest_id: str,
    manifest_claim_id: str | None,
    fault_class: str,
    detail: str,
    finding_id: str,
    judge_model: str,
    now_iso: str,
) -> dict[str, Any]:
    """§3.6 (v3.8.2 / #118): uncited sentence × manifest pair where the
    constraint judge raised JudgeInvocationError. Mirrors INV-14 row on the
    cited path but rides in the uncited_audit_failures[] aggregate because
    claim_audit_result.ref_slug is required."""
    rationale = f"{fault_class}: {detail}" if detail else f"{fault_class}:"
    return {
        "finding_id": finding_id,
        "claim_text": sentence["sentence_text"],
        "section_path": sentence.get("section_path", ""),
        "scoped_manifest_id": scoped_manifest_id,
        "manifest_claim_id": manifest_claim_id,
        "fault_class": fault_class,
        "rationale": rationale,
        "judge_model": judge_model,
        "judge_run_at": now_iso,
        "rule_version": UAF_RULE_VERSION,
        "upstream_owner_agent": sentence.get("upstream_owner_agent"),
    }


def _constraint_violation_entry(
    *,
    sentence: dict[str, Any],
    judge_result: dict[str, Any],
    scoped_manifest_id: str,
    finding_id: str,
    judge_model: str,
    now_iso: str,
) -> dict[str, Any]:
    """§3.5 / §5 stream (d): uncited claim with VIOLATED judge verdict."""
    violated_id = judge_result.get("violated_constraint_id")
    manifest_claim_id = None
    if violated_id:
        nc_match = RE_NC_CONSTRAINT.match(violated_id)
        if nc_match:
            manifest_claim_id = f"C-{nc_match.group(1)}"
    return {
        "finding_id": finding_id,
        "claim_text": sentence["sentence_text"],
        "section_path": sentence.get("section_path", ""),
        "violated_constraint_id": violated_id,
        "scoped_manifest_id": scoped_manifest_id,
        "manifest_claim_id": manifest_claim_id,
        "judge_verdict": "VIOLATED",
        # #360: clamp the judge-supplied rationale (success path) to maxLength;
        # `or` default catches a missing/null/empty rationale (schema minLength=1).
        "rationale": _bounded_judge_rationale(judge_result.get("rationale"))
        or "Constraint violated by uncited claim.",
        "judge_model": judge_model,
        "judge_run_at": now_iso,
        "rule_version": DRIFT_RULE_VERSION,
        "upstream_owner_agent": sentence.get("upstream_owner_agent"),
    }


def _uncited_assertion_entry(
    *,
    sentence: dict[str, Any],
    finding_id: str,
    now_iso: str,
    trigger_tokens: list[str] | None = None,
) -> dict[str, Any]:
    # Resolve trigger_tokens with strict semantics: prefer the explicit
    # keyword arg, fall back to the sentence dict, and raise if both are
    # absent. The prior `["uncited"]` sentinel passed U-INV-2 minItems=1
    # but carried no semantic content — callers who skipped the detector
    # silently emitted meaningless tokens into the passport. Raise instead
    # so the contract is enforced at write-time, not discovered at audit-
    # read-time (per codex R1 P1-4).
    tokens = trigger_tokens or sentence.get("trigger_tokens")
    if not tokens:
        raise ValueError(
            f"_uncited_assertion_entry: finding_id={finding_id!r} has no "
            "trigger_tokens. Caller must pre-process draft sentences "
            "through detect_uncited_assertions (or supply trigger_tokens "
            "explicitly); the schema's U-INV-2 minItems=1 invariant is "
            "an audit-quality contract, not a placeholder slot."
        )
    manifest_claim_id = sentence.get("manifest_claim_id")
    # Step 7 codex R1 CO-3 / U-INV-4 pair rule: scoped_manifest_id is the
    # disambiguator for a specific manifest claim. When no claim_id is bound
    # (the uncited sentence is in scope for a manifest-level MNC but is NOT
    # itself a manifest claim — runtime contract for stream-d uncited
    # constraint-violation routing), the uncited_assertion row MUST drop the
    # manifest scope. The companion constraint_violations[] row owns the
    # manifest pointer in that case; carrying scope on both rows would fail
    # U-INV-4 (manifest_claim_id null ↔ scoped_manifest_id null).
    scoped_manifest_id = (
        sentence.get("scoped_manifest_id") if manifest_claim_id is not None else None
    )
    return {
        "finding_id": finding_id,
        "sentence_text": sentence["sentence_text"],
        "section_path": sentence.get("section_path", ""),
        "trigger_tokens": tokens,
        "detected_at": now_iso,
        "rule_version": UNCITED_RULE_VERSION,
        "upstream_owner_agent": sentence.get("upstream_owner_agent"),
        "manifest_claim_id": manifest_claim_id,
        "scoped_manifest_id": scoped_manifest_id,
    }


def _claim_drift_entry(
    *,
    drift_kind: str,
    claim_text: str,
    finding_id: str,
    now_iso: str,
    manifest_claim_id: str | None = None,
    scoped_manifest_id: str | None = None,
    section_path: str | None = None,
) -> dict[str, Any]:
    entry = {
        "finding_id": finding_id,
        "drift_kind": drift_kind,
        "claim_text": claim_text,
        "detected_at": now_iso,
        "rule_version": DRIFT_RULE_VERSION,
        "manifest_claim_id": manifest_claim_id,
        "scoped_manifest_id": scoped_manifest_id,
        "section_path": section_path,
    }
    return entry


# ---------------------------------------------------------------------------
# Sampling helper.
# ---------------------------------------------------------------------------


def _stratified_bucket_indices(total: int, cap: int) -> list[int]:
    """Pick `cap` indices in [0, total) via stratified buckets in document order.

    Divides [0, total) into `cap` equal-ish buckets and picks the first index of
    each bucket. The result is strictly ascending and has length min(cap, total).

    Why two-stage fill: a naive `int(i * width)` for `width = total / cap`
    silently collapses adjacent picks when `total/cap < 2` (e.g. N=101,
    cap=100 → `int(99 * 1.01) == int(100 * 1.0)`-class duplicates after
    dedup). The S-INV-1 invariant ties `audited_count` to `len(audited_indices)`
    so a silent dedup would shrink audited_count below cap with no surface.
    We dedup first, then fill the remaining slots from un-picked indices in
    ascending document order — keeping the bucket-first-pick bias for spread
    while honoring the contract that `audited_count == min(cap, N)` whenever
    that is achievable.
    """
    if total <= 0 or cap <= 0:
        return []
    k = min(cap, total)
    width = total / k
    picks: set[int] = set()
    for i in range(k):
        picks.add(int(i * width))
    # Fill missing slots from un-picked indices in ascending order so the
    # final result is exactly k strictly-ascending unique picks whenever
    # k ≤ total (which is guaranteed by `k = min(cap, total)`).
    if len(picks) < k:
        for j in range(total):
            if j not in picks:
                picks.add(j)
                if len(picks) == k:
                    break
    return sorted(picks)


# ---------------------------------------------------------------------------
# Drift detection (manifest set-diff).
# ---------------------------------------------------------------------------


def _detect_drifts(
    *,
    manifests: list[dict[str, Any]],
    emitted_citations: list[dict[str, Any]],
    uncited_sentence_texts: set[str],
    constraint_absorbed_claim_ids: set[tuple[str, str]],
    constraint_absorbed_manifest_scopes: set[str],
    now_iso: str,
    next_finding_id: Callable[[], str],
) -> list[dict[str, Any]]:
    """§4 step 5 manifest set-diff producing claim_drift entries.

    Precedence:
      - T-P8 + Step 13 R5 codex P3: constraint-violation absorbs drift in
        **full** for the violating manifest. `constraint_absorbed_claim_ids`
        captures (manifest, declared-claim-id) pairs; `constraint_absorbed_manifest_scopes`
        captures the manifest_id itself so a same-manifest citation with a
        drifted (non-manifest) claim_id is also suppressed — preserves the
        "absorbed in full" spec promise. A violation in manifest A does NOT
        silence drift in manifest B.
      - T-P10 / D-INV-4: uncited sentence takes precedence over drift — no
        drift entry whose claim_text matches an uncited sentence_text.

    MANIFEST-MISSING fallback (Step 13 R5 codex P2 #2): when no manifest
    carries any claims, there's no pre-commitment baseline to drift FROM —
    every emitted citation would be classified EMITTED_NOT_INTENDED, layering
    spurious LOW-WARN noise on top of the MANIFEST-MISSING advisory the
    formatter already surfaces. Short-circuit and return no drifts so the
    fallback run remains audit-only.
    """
    has_baseline = any((m.get("claims") or []) for m in manifests)
    if not has_baseline:
        return []
    drifts: list[dict[str, Any]] = []

    # Index emitted citations by (scoped_manifest_id, claim_id) — these are
    # the "supported" set candidates the prose actually produced. Use
    # .get(SENTINEL_MANIFEST_ID) so MANIFEST-MISSING callers that omit
    # scoped_manifest_id still build a coherent emitted_pairs set per the
    # sentinel fallback contract (Step 13 R4 codex P2 #3).
    emitted_pairs: set[tuple[str, str]] = {
        (c.get("scoped_manifest_id", SENTINEL_MANIFEST_ID), c.get("claim_id", ""))
        for c in emitted_citations
    }
    emitted_texts = {c.get("claim_text", "") for c in emitted_citations}

    # INTENDED_NOT_EMITTED — manifest claims missing from emitted set.
    # D6 defines Emitted as a set of claim_text values; the dropped-claim side
    # MUST mirror that. A stale or re-numbered claim_id where the claim_text
    # still appears in the draft would otherwise show up as INTENDED_NOT_EMITTED
    # even though the prose carries the claim — a false drift signal
    # (Step 13 R2 codex P2 finding).
    for m in manifests:
        mid = m.get("manifest_id")
        # If this manifest had any constraint violation, absorb ALL of its
        # drift — Step 13 R5 codex P3. The (mid, cid) pair-level absorption
        # below is preserved for backwards compat, but the scope-level skip
        # is the load-bearing rule per "absorbed in full".
        if mid in constraint_absorbed_manifest_scopes:
            continue
        for claim in m.get("claims", []) or []:
            cid = claim.get("claim_id")
            claim_text = claim.get("claim_text", "")
            if (mid, cid) in emitted_pairs:
                continue
            if claim_text and claim_text in emitted_texts:
                # The draft carries the claim under a different claim_id (e.g.
                # claim_id was re-numbered between manifest emission and prose).
                # D6 set-of-text semantics — not a drop.
                continue
            if (mid, cid) in constraint_absorbed_claim_ids:
                continue
            if claim_text in uncited_sentence_texts:
                # Step 7 codex R1 CO-1 / D-INV-4 cross-aggregate exclusivity:
                # when a manifest claim appears as an uncited sentence in the
                # draft, the uncited_assertion row takes priority. Emitting an
                # INTENDED_NOT_EMITTED drift alongside would fail the D-INV-4
                # consistency lint (one finding per sentence across both
                # aggregates). Mirrors the EMITTED_NOT_INTENDED skip in the
                # loop below.
                continue
            drifts.append(
                _claim_drift_entry(
                    drift_kind="INTENDED_NOT_EMITTED",
                    claim_text=claim_text,
                    finding_id=next_finding_id(),
                    now_iso=now_iso,
                    manifest_claim_id=cid,
                    scoped_manifest_id=mid,
                    section_path=None,
                )
            )

    # EMITTED_NOT_INTENDED — emitted citations whose claim_text is not in any
    # manifest's claim_text set. D6 defines Emitted as a SET of claim_text
    # values, so a drifted claim that carries multiple citations (e.g. one
    # sentence with two ref markers) MUST emit ONE drift row, not one per
    # citation. We dedupe by claim_text here while keeping the first
    # encountered section_path as the representative (Step 13 R1 codex P2).
    all_manifest_texts: set[str] = set()
    for m in manifests:
        for claim in m.get("claims", []) or []:
            text = claim.get("claim_text")
            if text:
                all_manifest_texts.add(text)

    seen_drift_texts: set[str] = set()
    for c in emitted_citations:
        text = c.get("claim_text", "")
        if text in all_manifest_texts:
            continue
        c_scope = c.get("scoped_manifest_id", SENTINEL_MANIFEST_ID)
        c_claim_id = c.get("claim_id", "")
        if c_scope in constraint_absorbed_manifest_scopes:
            # Step 13 R5 codex P3 — manifest scope absorbed in full when any
            # citation in it violated a negative constraint. Suppresses
            # same-manifest drift for emitted citations whose claim_id was
            # NOT in the manifest's declared claim set.
            continue
        if (c_scope, c_claim_id) in constraint_absorbed_claim_ids:
            continue
        if text in uncited_sentence_texts:
            # Precedence rule 3 / D-INV-4 — uncited takes priority.
            continue
        if text in seen_drift_texts:
            # D6 set semantics — one drift row per drifted claim_text.
            continue
        seen_drift_texts.add(text)
        drifts.append(
            _claim_drift_entry(
                drift_kind="EMITTED_NOT_INTENDED",
                claim_text=text,
                finding_id=next_finding_id(),
                now_iso=now_iso,
                manifest_claim_id=None,
                scoped_manifest_id=None,
                section_path=c.get("section_path", "unknown"),
            )
        )

    return drifts


# ---------------------------------------------------------------------------
# Public entry point.
# ---------------------------------------------------------------------------


def run_audit_pipeline(
    *,
    citations: list[dict[str, Any]],
    manifests: list[dict[str, Any]],
    # Reserved for the production retrieval-driver wiring per spec §4 step 2.
    # The current implementation injects retrieval via `retrieve_fn` so the
    # corpus is read inside that callback; keeping the parameter on the
    # signature lets the orchestrator pass corpus through without changing
    # the public API when the production retrieve_fn lands.
    corpus: list[dict[str, Any]] | None = None,
    config: dict[str, Any],
    retrieve_fn: Callable[[dict[str, Any]], dict[str, Any]],
    judge_fn: Callable[..., dict[str, Any]],
    audit_run_id: str,
    now_iso: str,
    cache: dict[str, Any] | None = None,
    uncited_sentences: list[dict[str, Any]] | None = None,
    all_uncited_sentences: list[dict[str, Any]] | None = None,
    pdf_preflight_sidecars: dict[str, dict[str, Any]] | None = None,
) -> dict[str, list[dict[str, Any]]]:
    """Run §4 Step 1-6 + manifest set-diff over caller-supplied inputs.

    ``config.judge_model`` is the caller-supplied judge identity (effort
    included where it changes judgment). Omitted / null / blank / ``unknown``
    is recorded as ``unknown`` and binds the cache key to this run (#826), so a
    verdict can never be reused across runs under an unattributed identity.

    Two uncited streams (Step 13 R4 codex P1 #2):

    - `uncited_sentences`: D4-c detector positives — output of
      `detect_uncited_assertions` (sentences matching the quantifier /
      empirical-trigger filter). Drives `uncited_assertions[]` LOW-WARN
      advisory emission only.
    - `all_uncited_sentences`: the full uncited sentence set (every draft
      sentence with no in-text citation marker). Drives stream (d) —
      constraint judging for `constraint_violations[]` HIGH-WARN. The full
      set is needed because a manifest negative constraint like "No causal
      language" can be violated by a sentence ("The program caused
      improvement") that the D4-c detector filters OUT (no quantifier, no
      empirical trigger token). Routing constraint judging through the
      D4-c-filtered subset silently drops those HIGH-WARN cases.

    When `all_uncited_sentences` is omitted, it defaults to
    `uncited_sentences` (legacy callers preserved — but a warning band:
    those callers miss the R4 P1 expansion and should pass both).

    The uncited token-rule detector
    (`scripts/uncited_assertion_detector.py`, §"Uncited-assertion detector
    (D4-c)" in claim_ref_alignment_audit_agent.md) is NOT invoked here.
    Callers pre-process the full uncited set through
    `detect_uncited_assertions` to get `uncited_sentences`; pass the raw
    full set as `all_uncited_sentences`. `scripts/test_e2e_claim_audit.py`
    exercises the full detector → pipeline → finalizer chain end-to-end.

    Sentence-dict shape:
      - `uncited_sentences[]`: `sentence_text` + `section_path` +
        `trigger_tokens` (non-empty per U-INV-2). D4-c output guarantees
        these fields.
      - `all_uncited_sentences[]`: `sentence_text` + `section_path` only.
        Constraint judging does not consult `trigger_tokens`.

    `pdf_preflight_sidecars` (#512): optional map of `ref_slug` →
    `pdf_read_preflight/1` sidecar dict (see `scripts/pdf_read_preflight.py`).
    `None` (default) = the caller has not wired the preflight layer — legacy
    byte-equivalent behavior, no tagging. When provided (even empty), every
    completed `manual_pdf` row with `anchor_kind == "page"` whose sidecar is
    missing or non-`PASS` gets `PDF_READ_INTEGRITY_TAG` appended to its
    rationale at the Step-6 emission point, cache hit or not. Additionally,
    page-anchor citations handed to `retrieve_fn` carry a
    `pdf_preflight_verdict` key (`PASS`/`FAIL`/`UNAVAILABLE`/`MISSING`):
    retrieve_fn implementations SHOULD locate the passage by content rather
    than by the page scope when the verdict is not `PASS` — the judge must
    never be fed a passage selected by an untrusted page number. Sidecar
    freshness is the CALLER's contract: this function does no file I/O, so
    the orchestrator re-hashes each PDF against its sidecar `sha256` before
    dispatch and re-runs the preflight on mismatch (a stale `PASS` from a
    replaced file must not license the new bytes).

    Returns:
        dict with six aggregate arrays keyed by passport-aggregate name:
        claim_audit_results, uncited_assertions, claim_drifts,
        constraint_violations, audit_sampling_summaries, plus
        claim_intent_manifests (echoed for downstream consumption).

    Raises:
        ValueError: when config validation fails (e.g. max_claims_per_paper <= 0).
    """
    # Intentional no-op: `corpus` is reserved for the production retrieval-driver
    # wiring (spec §4 step 2). Mark as read so static analysers (ruff ARG002,
    # mypy strict unused-arg) do not flag the forward-compat parameter.
    _ = corpus

    # ---- Config sanity ----
    cap = config.get("max_claims_per_paper", 100)
    if not isinstance(cap, int) or cap <= 0:
        raise ValueError(
            f"max_claims_per_paper must be positive integer; got {cap!r} "
            "(spec §4 step 3 + S-INV-2 / T-P11 cap=0 rejected)"
        )
    supplied_judge_model = config.get("judge_model")
    if supplied_judge_model is not None and not isinstance(supplied_judge_model, str):
        raise ValueError("judge_model must be a string identity or null")
    raw_judge_model = (supplied_judge_model or "").strip()
    judge_identity_known = bool(raw_judge_model) and raw_judge_model.lower() != "unknown"
    judge_model = raw_judge_model if judge_identity_known else "unknown"
    # #826: an unattributed judge is never a cache-key component on its own —
    # same fail-closed shape as the unknown prompt_version below: bind a
    # run-local component so cross-run hits are impossible while within-run
    # dedup of repeated citations still holds.
    cache_judge_model = judge_model if judge_identity_known else f"__unknown__:{audit_run_id}"
    # #361: prompt_version is a judge-cache-key component. Absent key → default
    # to JUDGE_PROMPT_SHA256, the prompt's own fingerprint and the SINGLE SOURCE
    # OF TRUTH for cache invalidation: check_judge_prompt_version.py keeps this
    # hash in lockstep with the judge-prompt text, so any prompt edit changes the
    # hash and AUTOMATICALLY invalidates stale entries (the human-readable
    # JUDGE_PROMPT_VERSION label is decoupled and must NOT gate the cache).
    # Present-but-None → the caller declares the prompt version UNKNOWN; fail
    # CLOSED by binding a run-local component (audit_run_id is per-run unique) so
    # a stale entry is never served across an unknown-version boundary — cross-run
    # hits are disabled, but within-run dedup for repeated citations still holds.
    prompt_version = config.get("judge_prompt_version", JUDGE_PROMPT_SHA256)
    if prompt_version is None:
        prompt_version = f"__unknown__:{audit_run_id}"

    # Build the three lookup indexes once per run. Used by per-citation
    # constraint resolution + manifest-level absorption + drift detection.
    manifests_by_id: dict[str, dict[str, Any]] = {
        m["manifest_id"]: m for m in manifests if m.get("manifest_id")
    }
    claim_by_mc_id: dict[tuple[str, str], dict[str, Any]] = {
        (m["manifest_id"], claim["claim_id"]): claim
        for m in manifests
        if m.get("manifest_id")
        for claim in (m.get("claims") or [])
        if claim.get("claim_id")
    }
    mncs_by_manifest_id: dict[str, list[dict[str, Any]]] = {
        m["manifest_id"]: list(m.get("manifest_negative_constraints") or [])
        for m in manifests
        if m.get("manifest_id")
    }
    cache = cache if cache is not None else {}
    uncited_sentences = uncited_sentences or []
    # Step 13 R4 codex P1 #2: constraint judging needs the FULL uncited set,
    # not the D4-c-filtered subset. When the caller omits the full set we
    # fall back to the D4-c subset for backwards compatibility — but that
    # path silently drops constraint violations on sentences outside D4-c
    # trigger tokens. New callers should pass both.
    all_uncited_sentences = (
        all_uncited_sentences if all_uncited_sentences is not None else list(uncited_sentences)
    )

    # ---- Sampling decision ----
    total = len(citations)
    if total > cap:
        sampled_indices = _stratified_bucket_indices(total, cap)
        audited_citations = [citations[i] for i in sampled_indices]
        sampling_summaries = [
            {
                "audit_run_id": audit_run_id,
                "max_claims_per_paper": cap,
                "total_citation_count": total,
                "audited_count": len(sampled_indices),
                "audited_indices": sampled_indices,
                "sampling_strategy": SAMPLING_STRATEGY,
                "emitted_at": now_iso,
            }
        ]
    else:
        audited_citations = list(citations)
        sampling_summaries = []

    # ---- Per-citation §4 Step 1-6 ----
    claim_audit_results: list[dict[str, Any]] = []
    constraint_violations: list[dict[str, Any]] = []
    constraint_absorbed_claim_ids: set[tuple[str, str]] = set()
    # Step 13 R5 codex P3: track manifest_id scopes whose drift is absorbed
    # "in full". A constraint violation in manifest M suppresses ALL of M's
    # drift — declared claims (by id), the violating citation's pair (which
    # may itself be drifted), AND any other emitted citation in M with a
    # non-manifest claim_id.
    constraint_absorbed_manifest_scopes: set[str] = set()

    def _written_scope_for(citation: dict[str, Any]) -> str:
        """Return the scoped_manifest_id that goes onto the claim_audit_result row.

        Step 7 codex R1 CO-2: a drifted-cited citation's `claim_id` is not in
        any manifest, but the citation still arrives with the active
        `scoped_manifest_id` for runtime constraint resolution (so global MNCs
        still apply per M-INV-3). The row written to the passport, however,
        MUST carry the sentinel manifest id whenever the (scope, claim_id)
        pair is not present in the manifest index — otherwise INV-15 dangling
        check rejects the passport. Runtime constraint lookup stays untouched
        (it reads citation.scoped_manifest_id directly); only the persisted
        row is normalized.
        """
        runtime_scope = citation.get("scoped_manifest_id", SENTINEL_MANIFEST_ID)
        cid = citation.get("claim_id")
        if runtime_scope == SENTINEL_MANIFEST_ID:
            return SENTINEL_MANIFEST_ID
        if (runtime_scope, cid) in claim_by_mc_id:
            return runtime_scope
        return SENTINEL_MANIFEST_ID

    for citation in audited_citations:
        anchor_kind = citation.get("anchor_kind")
        scoped_manifest_id = citation.get("scoped_manifest_id", SENTINEL_MANIFEST_ID)
        claim_id = citation.get("claim_id")
        written_scope = _written_scope_for(citation)

        # Step 1 — anchor=none firm-rule short-circuit.
        if anchor_kind == "none":
            entry = _anchorless_entry(
                citation,
                audit_run_id=audit_run_id,
                now_iso=now_iso,
                judge_model=judge_model,
            )
            entry["scoped_manifest_id"] = written_scope
            claim_audit_results.append(entry)
            continue

        # Step 2 — retrieval. Wrap in `_invoke_retrieve` so transient failures
        # surface as INV-14 retrieval_* audit_tool_failure rows instead of
        # aborting the pass (Step 13 R2 codex P2 finding, symmetric to the
        # R1 _invoke_judge wrapper).
        # #512 r2: when the preflight layer is wired, page-anchor citations carry
        # the sidecar verdict INTO retrieval so implementations can locate the
        # passage by content instead of the untrusted page scope BEFORE the judge
        # sees it — tagging after the fact cannot fix a judge that already read
        # the wrong passage. Copy-on-write; API-path retrieve_fns ignore the key.
        retrieval_citation = citation
        if pdf_preflight_sidecars is not None and anchor_kind == "page":
            _sc = pdf_preflight_sidecars.get(citation["ref_slug"])
            retrieval_citation = dict(citation)
            retrieval_citation["pdf_preflight_verdict"] = (
                _sc.get("verdict") if _sc else "MISSING"
            )
        try:
            retrieval = _invoke_retrieve(retrieve_fn, retrieval_citation)
        except RetrievalInvocationError as ret_err:
            entry = _retrieval_failure_entry(
                citation,
                method="audit_tool_failure",
                audit_run_id=audit_run_id,
                now_iso=now_iso,
                judge_model=judge_model,
                fault_class=ret_err.fault_class,
            )
            entry["rationale"] = f"{ret_err.fault_class}: {ret_err.detail}"
            entry["scoped_manifest_id"] = written_scope
            claim_audit_results.append(entry)
            continue
        method = retrieval["ref_retrieval_method"]
        excerpt = retrieval.get("retrieved_excerpt")

        if method in {"failed", "not_found", "audit_tool_failure"}:
            entry = _retrieval_failure_entry(
                citation,
                method=method,
                audit_run_id=audit_run_id,
                now_iso=now_iso,
                judge_model=judge_model,
                fault_class=retrieval.get("fault_class"),
            )
            entry["scoped_manifest_id"] = written_scope
            claim_audit_results.append(entry)
            continue

        if method not in {"api", "manual_pdf"}:
            raise ValueError(f"unexpected ref_retrieval_method: {method!r}")

        # Step 3 — cache lookup. Active constraints scoped by (manifest, claim).
        active_constraints = _active_constraints_for_claim(
            scoped_manifest_id=scoped_manifest_id,
            claim_id=claim_id,
            claim_by_mc_id=claim_by_mc_id,
            mncs_by_manifest_id=mncs_by_manifest_id,
        )
        key = _cache_key(
            claim_text=citation["claim_text"],
            ref_slug=citation["ref_slug"],
            anchor_kind=anchor_kind,
            anchor_value=citation.get("anchor_value", ""),
            retrieved_excerpt=excerpt,
            active_constraints=active_constraints,
            judge_model=cache_judge_model,
            prompt_version=prompt_version,
        )
        # In-scope constraint ids for this call. Both fresh judge invocations
        # AND cache hits validate VIOLATED ids against this set so a
        # hallucinated id never reaches `_judge_result_entry` (Step 13 R3
        # codex P2 #1).
        active_ids: frozenset[str] = frozenset(
            c["constraint_id"] for c in active_constraints if c.get("constraint_id")
        )

        cached = cache.get(key)
        try:
            if cached is not None:
                # Step 13 R3 codex P2 #4: cache may carry a corrupted/partial
                # dict from a prior session. Re-validate every hit through the
                # same surface as fresh invocations so a stale entry surfaces
                # as cache_corruption instead of crashing in
                # `_judge_result_entry`.
                judge_result = _validate_judge_dict(
                    cached,
                    allowed_judgments=_CITED_PATH_JUDGMENTS,
                    active_constraint_ids=active_ids,
                    source="cache",
                )
            else:
                # Step 4-5 — passage location is implicit (excerpt is the
                # located passage); invoke judge. Wrap in `_invoke_judge` so
                # transient failures surface as INV-14 `audit_tool_failure`
                # rows instead of aborting the audit (Step 13 R1 codex).
                judge_result = _invoke_judge(
                    judge_fn,
                    allowed_judgments=_CITED_PATH_JUDGMENTS,
                    active_constraint_ids=active_ids,
                    claim_text=citation["claim_text"],
                    retrieved_excerpt=excerpt,
                    anchor_kind=anchor_kind,
                    anchor_value=citation.get("anchor_value", ""),
                    active_constraints=active_constraints,
                    judge_model=judge_model,
                )
                cache[key] = judge_result
        except JudgeInvocationError as judge_err:
            # Cache-hit validation failures map to cache_corruption per INV-14.
            fault_class = "cache_corruption" if cached is not None else judge_err.fault_class
            entry = _retrieval_failure_entry(
                citation,
                method="audit_tool_failure",
                audit_run_id=audit_run_id,
                now_iso=now_iso,
                judge_model=judge_model,
                fault_class=fault_class,
            )
            entry["rationale"] = f"{fault_class}: {judge_err.detail}"
            entry["scoped_manifest_id"] = written_scope
            claim_audit_results.append(entry)
            continue

        # Step 6 — defect_stage routing + emission.
        entry = _judge_result_entry(
            citation,
            judge_result=judge_result,
            ref_retrieval_method=method,
            audit_run_id=audit_run_id,
            now_iso=now_iso,
            judge_model=judge_model,
        )
        entry["scoped_manifest_id"] = written_scope
        # #512 PDF read-integrity precondition, executable path. Sits AFTER cache
        # resolution so hits and fresh invocations are tagged identically (codex
        # #512 P1: a cache hit must not bypass the check). `None` = caller has not
        # wired the preflight layer → byte-equivalent legacy behavior; a provided
        # dict (even empty) means the orchestrator ran the layer, so a missing or
        # non-PASS sidecar for a manual_pdf page-anchor row is tagged.
        if (
            pdf_preflight_sidecars is not None
            and method == "manual_pdf"
            and anchor_kind == "page"
        ):
            sidecar = pdf_preflight_sidecars.get(citation["ref_slug"])
            if not (sidecar and sidecar.get("verdict") == "PASS"):
                _tag_pdf_read_integrity(entry)
        claim_audit_results.append(entry)

        # Precedence rule 1: cited constraint violation absorbs the drift signal.
        # Spec §6 lint rule 6 + §7.2 T-P8: when a citation in this manifest
        # judges VIOLATED, that manifest's drift findings are absorbed in
        # full — the constraint violation has already surfaced the L3
        # faithfulness failure at HIGH-WARN, so layering LOW-WARN drift
        # noise on top of it just reports the same paper-level problem twice.
        # Absorption is manifest-scoped (not global) so VIOLATED in
        # manifest A does NOT silence legitimate drift signal in manifest B.
        if entry["defect_stage"] == "negative_constraint_violation":
            manifest = manifests_by_id.get(scoped_manifest_id)
            if manifest is not None:
                for claim in manifest.get("claims", []) or []:
                    cid_in_manifest = claim.get("claim_id")
                    if cid_in_manifest:
                        constraint_absorbed_claim_ids.add(
                            (scoped_manifest_id, cid_in_manifest)
                        )
            # Also absorb the emitted citation's own (manifest, claim) pair
            # so a drifted-yet-violated citation does not produce a
            # companion EMITTED_NOT_INTENDED row.
            constraint_absorbed_claim_ids.add((scoped_manifest_id, claim_id))
            # Step 13 R5 codex P3 — record the full manifest scope so any
            # other emitted citation in M with a drifted claim_id is also
            # absorbed by the drift detector.
            constraint_absorbed_manifest_scopes.add(scoped_manifest_id)

    # ---- Stream (d): uncited constraint judging over FULL uncited set ----
    # Step 13 R4 codex P1 #2: constraint judging MUST see every uncited
    # sentence (not just D4-c detector positives). An MNC like "No causal
    # language" can be violated by an uncited sentence that the D4-c filter
    # drops (no quantifier, no empirical trigger token), so routing
    # constraint judging through the LOW-WARN advisory loop below would
    # silently miss HIGH-WARN cases. We run constraint judging here on
    # `all_uncited_sentences` and emit LOW-WARN uncited_assertion rows
    # separately on `uncited_sentences` (the D4-c subset).
    #
    # Step 13 R6 codex P1: the documented Stage 4 draft sentence shape carries
    # only `sentence_text` / `section_path` / optional `adjacent_text` — NOT
    # a sentence-level `scoped_manifest_id`. Pre-fix this loop required the
    # caller to populate scope on every sentence; absent that, the
    # constraint judge was skipped and the HIGH-WARN-CONSTRAINT-VIOLATION-UNCITED
    # gate was a no-op for orchestrator callers following the contract.
    # The runtime now derives a default scope set per sentence: if the caller
    # provides `scoped_manifest_id` on the sentence dict, only that manifest's
    # MNCs apply; otherwise the pipeline applies EVERY manifest's MNCs
    # (uncited sentences have no claim-level binding, so manifest-scoped
    # MNCs reach them universally per spec §3.5 D4-c stream (d) semantics).
    # Per-manifest constraint sets — MNCs (manifest-wide) PLUS, when the
    # sentence carries a `manifest_claim_id`, that claim's NC-C entries.
    # NC-C (claim-level) is the R7 codex P1 gap: spec §3.5 D4-c stream (d)
    # covers BOTH MNC and NC-C for uncited violations, but the R6 closure
    # only wired MNCs. We resolve NC-C from claim_by_mc_id at call time
    # when the sentence binds a claim_id.
    #
    # Step 13 R7 codex P2 also applies here: when MNC ids collide across
    # manifests (two manifests both have "MNC-1"), passing a flat list to
    # one judge call makes the returned `violated_constraint_id` ambiguous
    # — we'd have to first-match-wins which mis-attributes the row to the
    # wrong manifest. Solution: run the judge ONCE PER MANIFEST, with that
    # manifest's MNC + NC-C set. Each return is unambiguous by construction;
    # cost is N judge calls per sentence (cf. cited path where each
    # citation already takes one judge call).
    manifest_mncs_by_id: dict[str, list[dict[str, Any]]] = {}
    for m in manifests:
        mid = m.get("manifest_id")
        if not mid:
            continue
        mncs_for_mid: list[dict[str, Any]] = []
        for mnc in m.get("manifest_negative_constraints") or []:
            if mnc.get("constraint_id"):
                mncs_for_mid.append(
                    {"constraint_id": mnc["constraint_id"], "rule": mnc["rule"], "scope": "MNC"}
                )
        manifest_mncs_by_id[mid] = mncs_for_mid

    constraint_violation_texts: set[str] = set()
    cv_counter = 1
    # v3.8.2 / #118 — uncited_audit_failures[] aggregate for JudgeInvocationError
    # on the uncited constraint-judging path. Mirrors INV-14 audit_tool_failure
    # on the cited path. Pre-v3.8.2 the failure was silently substituted as
    # NOT_VIOLATED, suppressing HIGH-WARN constraint checks; the UAF aggregate
    # surfaces the operational signal at MED-WARN advisory tier without
    # dropping audit coverage (option 4 — re-raise and abort — was rejected
    # for that exact coverage reason). See spec §3.6.
    uncited_audit_failures: list[dict[str, Any]] = []
    uaf_counter = 1
    for sentence in all_uncited_sentences:
        scoped_manifest_id_for_sentence = sentence.get("scoped_manifest_id")
        sentence_claim_id = sentence.get("manifest_claim_id")

        # Decide which manifests this sentence is constraint-judged against.
        # Caller-pinned scope → that one manifest. Otherwise every manifest.
        if scoped_manifest_id_for_sentence:
            target_manifest_ids = [scoped_manifest_id_for_sentence] if scoped_manifest_id_for_sentence in manifest_mncs_by_id else []
        else:
            target_manifest_ids = list(manifest_mncs_by_id.keys())

        if not target_manifest_ids:
            continue

        # One judge call per (sentence, manifest) pair so MNC id collisions
        # across manifests cannot misattribute the violation (R7 codex P2).
        sentence_violation_recorded = False
        for mid in target_manifest_ids:
            per_manifest_constraints: list[dict[str, Any]] = list(manifest_mncs_by_id[mid])
            # R7 codex P1: also include NC-C for this manifest's bound claim.
            # v3.8.2 / #118 codex P2-2 + R2 P2-1: only set the UAF row's
            # manifest_claim_id when THIS manifest actually owns the claim
            # binding AND at least one NC constraint was added. When sentence
            # carries a sentence_claim_id but the current manifest doesn't
            # have that claim_id, OR the claim exists but has no NC entries
            # (so the failed judge call was MNC-only), the UAF row's
            # manifest_claim_id MUST stay null — the failure is MNC-only at
            # that point and a non-null binding would mislabel downstream
            # consumers about which constraint set the outage hit.
            uaf_manifest_claim_id: str | None = None
            if sentence_claim_id:
                claim = claim_by_mc_id.get((mid, sentence_claim_id))
                if claim is not None:
                    for nc in claim.get("negative_constraints") or []:
                        cid = nc.get("constraint_id")
                        if cid:
                            per_manifest_constraints.append(
                                {"constraint_id": cid, "rule": nc["rule"], "scope": "NC"}
                            )
                            # Only mark NC-C binding when at least one NC
                            # constraint actually entered the judge call.
                            uaf_manifest_claim_id = sentence_claim_id

            if not per_manifest_constraints:
                continue

            applicable_ids: frozenset[str] = frozenset(
                c["constraint_id"] for c in per_manifest_constraints if c.get("constraint_id")
            )
            # Wrap in `_invoke_judge` so transient failures don't abort the
            # uncited stream. v3.8.2 / #118 — JudgeInvocationError now routes
            # to an uncited_audit_failures[] row (MED-WARN advisory, gate
            # passes) instead of synthesising a NOT_VIOLATED verdict. The
            # synthesis substitution shipped pre-v3.8.2 was silently
            # suppressing HIGH-WARN constraint checks on transient judge
            # outage — see spec §3.6 + §4 step 9 fourth bullet for routing,
            # and the design memo at docs/superpowers/plans/2026-05-17-issue-118-*
            # for the option 1-4 trade-off analysis.
            try:
                judge_result = _invoke_judge(
                    judge_fn,
                    allowed_judgments=_UNCITED_PATH_JUDGMENTS,
                    active_constraint_ids=applicable_ids,
                    claim_text=sentence["sentence_text"],
                    retrieved_excerpt=None,
                    anchor_kind=None,
                    anchor_value=None,
                    active_constraints=per_manifest_constraints,
                    judge_model=judge_model,
                )
            except JudgeInvocationError as judge_err:
                uncited_audit_failures.append(
                    _uncited_audit_failure_entry(
                        sentence=sentence,
                        scoped_manifest_id=mid,
                        manifest_claim_id=uaf_manifest_claim_id,
                        fault_class=judge_err.fault_class,
                        detail=judge_err.detail,
                        finding_id=f"UAF-{uaf_counter:03d}",
                        judge_model=judge_model,
                        now_iso=now_iso,
                    )
                )
                uaf_counter += 1
                continue  # no fake NOT_VIOLATED, no CV row, skip to next manifest
            if judge_result.get("judgment") == "VIOLATED":
                constraint_violations.append(
                    _constraint_violation_entry(
                        sentence=sentence,
                        judge_result=judge_result,
                        scoped_manifest_id=mid,
                        finding_id=f"CV-{cv_counter:03d}",
                        judge_model=judge_model,
                        now_iso=now_iso,
                    )
                )
                cv_counter += 1
                sentence_violation_recorded = True

        if sentence_violation_recorded:
            constraint_violation_texts.add(sentence["sentence_text"])

    # ---- Step 6 stream (uncited_assertion LOW-WARN advisory) ----
    # D4-c detector positives only — uncited_sentences is the filtered set.
    uncited_assertions: list[dict[str, Any]] = []
    uncited_sentence_texts: set[str] = set()
    ua_counter = 1

    for sentence in uncited_sentences:
        uncited_sentence_texts.add(sentence["sentence_text"])
        # Always emit uncited_assertion (LOW-WARN advisory). CV-INV-4
        # explicitly permits a sentence to appear in both uncited_assertions[]
        # and constraint_violations[] simultaneously.
        uncited_assertions.append(
            _uncited_assertion_entry(
                sentence=sentence,
                finding_id=f"UA-{ua_counter:03d}",
                now_iso=now_iso,
            )
        )
        ua_counter += 1

    # Also surface uncited_sentence_texts for any constraint-violation
    # sentence text that wasn't a D4-c positive but did violate an MNC.
    # The drift detector reads `uncited_sentence_texts` to apply the
    # D-INV-4 uncited-takes-precedence rule; without including CV-text
    # entries here, a sentence outside D4-c but matching a manifest claim
    # would produce both a constraint_violation row AND a drift row,
    # violating D-INV-4.
    uncited_sentence_texts.update(constraint_violation_texts)

    # ---- Manifest set-diff drift detection ----
    cd_counter = 1

    def _next_cd() -> str:
        nonlocal cd_counter
        out = f"CD-{cd_counter:03d}"
        cd_counter += 1
        return out

    # Step 7 codex R1 CO-4: drift detection's emitted-side index MUST use
    # the FULL citation list, not the sampled subset. Sampling caps judge
    # invocations (spec §4 step 3) — it does NOT shrink the prose visible to
    # the manifest set-diff. Passing audited_citations here made every
    # unsampled-but-present citation look dropped from manifest, producing
    # false INTENDED_NOT_EMITTED rows in proportion to (total - cap).
    claim_drifts = _detect_drifts(
        manifests=manifests,
        emitted_citations=citations,
        uncited_sentence_texts=uncited_sentence_texts,
        constraint_absorbed_claim_ids=constraint_absorbed_claim_ids,
        constraint_absorbed_manifest_scopes=constraint_absorbed_manifest_scopes,
        now_iso=now_iso,
        next_finding_id=_next_cd,
    )

    return {
        "claim_intent_manifests": manifests,
        "claim_audit_results": claim_audit_results,
        "uncited_assertions": uncited_assertions,
        "claim_drifts": claim_drifts,
        "constraint_violations": constraint_violations,
        "audit_sampling_summaries": sampling_summaries,
        "uncited_audit_failures": uncited_audit_failures,
    }