scripts/claim_audit_pipeline.py

• _stable_json · function · L58-L60 — def _stable_json(value: Any) -> str
• _hash_text · function · L63-L64 — def _hash_text(text: str | None) -> str
• _clamp_to_rationale_budget · function · L79-L99 — def _clamp_to_rationale_budget(text: str, *, reserved: int) -> str
• _bounded_failure_detail · function · L102-L111 — def _bounded_failure_detail(message: str) -> str
• _bounded_judge_rationale · function · L114-L130 — def _bounded_judge_rationale(rationale: Any) -> str
• _active_constraints_for_claim · function · L133-L157 — 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]]
• _cache_key · function · L160-L187 — 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
• _AuditInvocationError · class · L212-L225 — class _AuditInvocationError(Exception)
• init · method · L221-L225 — def init(self, fault_class: str, detail: str) -> None
• JudgeInvocationError · class · L228-L235 — class JudgeInvocationError(_AuditInvocationError)
• RetrievalInvocationError · class · L238-L245 — class RetrievalInvocationError(_AuditInvocationError)
• _validate_judge_dict · function · L248-L328 — def _validate_judge_dict( result: Any, *, allowed_judgments: frozenset[str], active_constraint_ids: frozenset[str], source: str = "judge_fn", ) -> dict[str, Any]
• _invoke_judge · function · L331-L374 — 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_retrieve · function · L377-L452 — def _invoke_retrieve( retrieve_fn: Callable[[dict[str, Any]], dict[str, Any]], citation: dict[str, Any], ) -> dict[str, Any]
• _anchorless_entry · function · L460-L487 — def _anchorless_entry(citation: dict[str, Any], *, audit_run_id: str, now_iso: str, judge_model: str) -> dict[str, Any]
• _retrieval_failure_entry · function · L490-L544 — 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]
• _judge_result_entry · function · L547-L631 — 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]
• _uncited_audit_failure_entry · function · L634-L662 — 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]
• _constraint_violation_entry · function · L665-L697 — 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]
• _uncited_assertion_entry · function · L700-L745 — 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).
• _claim_drift_entry · function · L748-L768 — 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]
• _stratified_bucket_indices · function · L776-L808 — def _stratified_bucket_indices(total: int, cap: int) -> list[int]
• _detect_drifts · function · L816-L956 — 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]]
• run_audit_pipeline · function · L964-L1525 — 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, ) -> dict[str, list[dict[str, Any]]]
• _written_scope_for · function · L1114-L1133 — def _written_scope_for(citation: dict[str, Any]) -> str
• _next_cd · function · L1495-L1499 — def _next_cd() -> str