"""Advisory unit validation and retry receipts for cite-check orchestration. This module is the small seam between a host runtime and the provider-neutral contracts. It deliberately does not perform citation matching, compare worker rows with a parent inventory, or reinterpret parent-owned assignment metadata. Those are outside the citation-level contract. """ from __future__ import annotations import re from copy import deepcopy from typing import Any from common.contracts import ( HARD_ERROR_CODES, PACKAGE_DIR, _load_json, validate_schema, validate_unit_result, ) _SOURCE_RESOLUTION_VALUES = frozenset( { "matched_supplied_source", "not_supplied_confirmed_exists_elsewhere", "not_supplied_not_found_potential_hallucination", "not_supplied_search_unavailable", } ) _FABRICATION_INDICATOR_VALUES = frozenset( { "reporter_coordinates_belong_to_different_supplied_case", "not_found_in_external_search", "citation_malformed_or_impossible", } ) def _add_validation_flag(citation: dict[str, Any], flag: str) -> None: flags = citation.get("validation_flags") if not isinstance(flags, list): flags = [] citation["validation_flags"] = flags if flag not in flags: flags.append(flag) def _annotate_allowed_evidence_defects(validation: dict[str, Any]) -> None: """Mark only allowed evidence-link defects for the one targeted retry.""" normalized = validation.get("normalized") if not isinstance(normalized, dict): return citations = normalized.get("citations") if not isinstance(citations, list): return for citation in citations: if not isinstance(citation, dict): continue source_id = citation.get("matched_source_id") if "matched_source_id" not in citation or ( source_id is not None and (not isinstance(source_id, str) or not source_id) ): _add_validation_flag(citation, "malformed_source_id") resolution = citation.get("source_resolution") if ( "source_resolution" not in citation or not isinstance(resolution, str) or resolution not in _SOURCE_RESOLUTION_VALUES ): _add_validation_flag(citation, "malformed_citation") indicators = citation.get("fabrication_indicators") invalid_indicators = "fabrication_indicators" not in citation or not isinstance( indicators, list ) if not invalid_indicators: invalid_indicators = any( not isinstance(item, str) or item not in _FABRICATION_INDICATOR_VALUES for item in indicators ) if not invalid_indicators: invalid_indicators = len(indicators) != len(set(indicators)) if invalid_indicators: _add_validation_flag(citation, "malformed_citation") for field in ("colliding_source_id", "existence_check_notes"): if ( field in citation and citation[field] is not None and not isinstance(citation[field], str) ): _add_validation_flag(citation, "malformed_citation") def _worker_schema_view(value: dict[str, Any]) -> dict[str, Any]: """Remove validator-owned fields before checking the worker contract.""" candidate = deepcopy(value) candidate.pop("validation", None) citations = candidate.get("citations") if isinstance(citations, list): for citation in citations: if isinstance(citation, dict): citation.pop("validation_flags", None) return candidate def validate_attempt( value: Any, expected_unit_id: str, *, authority_ids: set[str] | None = None, ) -> dict[str, Any]: """Validate one worker attempt as advisory diagnostics plus normalization.""" validation = validate_unit_result( value, expected_unit_id, authority_ids=authority_ids, ) _annotate_allowed_evidence_defects(validation) return validation def should_retry(validation: dict[str, Any], attempts_remaining: int) -> bool: """Retry only report-blocking defects, and only within the caller's budget.""" hard_errors = validation.get("hard_errors", []) return ( attempts_remaining > 0 and isinstance(hard_errors, list) and any( isinstance(item, dict) and item.get("code") in HARD_ERROR_CODES for item in hard_errors ) ) def attempt_receipt( attempt_number: int, raw_value: Any, validation: dict[str, Any], *, elapsed_ms: int | None = None, runtime_failure: dict[str, Any] | None = None, requested_model: str = "unknown", requested_effort: str = "unknown", observed_model: str = "unknown", observed_effort: str = "unknown", ) -> dict[str, Any]: """Build an audit receipt retaining the raw attempt and normalized outcome.""" hard_errors = validation.get("hard_errors", []) receipt: dict[str, Any] = { "attempt": attempt_number, "status": ( "failed" if runtime_failure is not None else "accepted" if not hard_errors else "rejected" ), "rawResult": deepcopy(raw_value), "warnings": deepcopy(validation.get("warnings", [])), "potentialIssues": deepcopy(validation.get("potential_issues", [])), "hardErrors": deepcopy(hard_errors), "requestedModel": requested_model, "requestedReasoningEffort": requested_effort, "observedModel": observed_model, "observedReasoningEffort": observed_effort, } normalized = validation.get("normalized") if normalized is not None: receipt["normalizedResult"] = deepcopy(normalized) if runtime_failure is not None: receipt["runtimeFailure"] = deepcopy(runtime_failure) if isinstance(elapsed_ms, int) and not isinstance(elapsed_ms, bool): receipt["elapsedMs"] = max(0, elapsed_ms) return receipt def normalized_report_result(validation: dict[str, Any]) -> dict[str, Any] | None: """Return the normalized result, retaining the advisory layer separately.""" value = validation.get("normalized") return value if isinstance(value, dict) else None def strict_normalized_report_result( validation: dict[str, Any], ) -> dict[str, Any] | None: """Return a normalized result only when it passes the worker schema. Advisory validation deliberately preserves citation-level warnings, but a resume receipt is terminal only when its underlying worker result is also structurally valid. Keeping that check here makes fresh results and same-run resume use the same acceptance boundary. """ normalized = normalized_report_result(validation) if normalized is None or validation.get("hard_errors"): return None if terminal_schema_errors(validation): return None return normalized def strict_schema_errors(validation: dict[str, Any]) -> list[str]: """Return structural worker-schema errors for an advisory validation.""" normalized = normalized_report_result(validation) if normalized is None or validation.get("hard_errors"): return [] schema = _load_json( PACKAGE_DIR / "schemas" / "cite-check-unit-result.schema.json", "unit result schema", ) if not isinstance(schema, dict): return ["unit result schema is not an object"] return validate_schema(_worker_schema_view(normalized), schema) _ROW_EVIDENCE_ERROR = re.compile( r"^\$?\.citations\[\d+\]\.(?:matched_source_id|source_resolution|" r"fabrication_indicators(?:\[\d+\])?|colliding_source_id|" r"existence_check_notes)$" ) _MISSING_ROW_EVIDENCE_ERROR = re.compile( r"^\$?\.citations\[\d+\]: missing required property " r"'(?:matched_source_id|source_resolution|fabrication_indicators)'$" ) def _is_row_evidence_error(error: str) -> bool: path = error.split(":", 1)[0] return bool( _ROW_EVIDENCE_ERROR.match(path) or _MISSING_ROW_EVIDENCE_ERROR.match(error) ) def terminal_schema_errors(validation: dict[str, Any]) -> list[str]: """Return schema errors that make a whole unit unreportable. The worker schema is intentionally strict, but malformed evidence values belong to a citation row. The validator preserves those raw values, marks the row amber, and may request one mechanical repair. They must not make the entire prepared unit disappear from the report. Wrapper errors, unknown properties, and malformed legal-judgment fields remain terminal. """ return [ error for error in strict_schema_errors(validation) if not _is_row_evidence_error(error) ]