diff --git a/docs/deployment.md b/docs/deployment.md index 4ac4671..a588b2d 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -354,9 +354,10 @@ to lean on: content_authenticity for high-volume bulk runs; reserve them for high-risk paths. 2. **Model choice.** `FLYDOCS_MODEL` is the default; override per - request via `options.model`. Lighter models (haiku, gpt-4o-mini) - are fine for high-volume extraction; reserve the heavier ones for - adversarial or low-confidence cases. + request via `options.model` (stages pinned via a + `FLYDOCS__MODEL` keep their model regardless). Lighter + models (haiku, gpt-4o-mini) are fine for high-volume extraction; + reserve the heavier ones for adversarial or low-confidence cases. 3. **Fallback.** `FLYDOCS_FALLBACK_MODEL` is used when the primary errors out. Setting it to a cheaper model avoids double-paying for a single failed call — and gives you a graceful degradation path. diff --git a/docs/payload-reference.md b/docs/payload-reference.md index 9c1585c..807d079 100644 --- a/docs/payload-reference.md +++ b/docs/payload-reference.md @@ -437,7 +437,7 @@ for validators to fire. |------------------------|----------|-----------------------|--------------------------------------------------------------------| | `return_bboxes` | boolean | `true` | `false` strips bboxes from the response (cheaper to transfer). | | `language_hint` | string? | `null` | ISO 639-1, ≤ 16 chars. Guides multilingual OCR / extraction. | -| `model` | string? | `null` (env default) | Per-request primary model id (`anthropic:claude-sonnet-4-6`, …). | +| `model` | string? | `null` (env default) | Per-request primary model id (`anthropic:claude-sonnet-4-6`, …). Applies only to stages without a pinned `FLYDOCS__MODEL`; a pinned stage keeps its model. | | `declared_media_type` | string? | `null` | Override sniffing; rare. | | `stages` | object | server defaults | See `StageToggles` below. | | `escalation` | object? | `null` | `{ threshold, model }`. Required when `stages.judge_escalation=true`. | diff --git a/docs/pipeline.md b/docs/pipeline.md index b8ad801..9f65d0e 100644 --- a/docs/pipeline.md +++ b/docs/pipeline.md @@ -55,7 +55,7 @@ reflect exactly what executed. | 9 | `visual_authenticity` | no | LLM evaluates caller-defined visual validators (signature present, stamp present, …). | 180 s | | 10 | `content_authenticity` | no | LLM audit: dates consistent, totals add up, expected boilerplate, tampering signals. | 180 s | | 11 | `judge` | no | Second LLM pass re-grades every extracted value against the source. | 300 s | -| 12 | `repair` | no | Targeted repair: re-extract ONLY the fields that failed the judge or a validator, quoting the failure evidence; accept per field only when the re-check passes. | 300 s | +| 12 | `repair` | no | Targeted repair: re-extract ONLY the fields the judge marked FAIL or flag_for_review, or that failed a validator, quoting the failure evidence; accept per field only when the re-check passes. For PDFs with known failing pages the pass runs on a sub-slice of those pages (plus one page of margin). | 300 s | | 13 | `judge_escalation` | no | When the judge's failure rate exceeds `escalation_threshold`, re-run extract + judge with `escalation_model` and keep the better result. | 600 s | | 14 | `transform` | no | Caller-declared post-extraction transformations: declarative entity resolution + free-form LLM transformations. See [transformations.md](transformations.md). | 600 s | | 15 | `rules` | no | LLM evaluates the business-rule DAG, level by level. | 180 s | @@ -79,6 +79,16 @@ Other short-circuits: - `repair` is silently skipped when both `judge` and `field_validation` are off (no failure signals to act on); the request validator emits a `repair_no_verification_stage` warning. + Per task, repair also steps aside when more than + `FLYDOCS_REPAIR_MAX_FAILING_FRACTION` of the fields are failing -- + the extraction is globally untrustworthy and `judge_escalation`'s + full re-run is the right tool (`RepairInfo.tasks_skipped` records + it). Concurrent task repairs are bounded by + `FLYDOCS_REPAIR_TASK_CONCURRENCY`, and + `FLYDOCS_REPAIR_INCLUDE_FLAGGED=false` restricts repair to hard + failures. On the sync channel the validator emits a + `repair_sync_latency` warning -- the pass eats into the + `FLYDOCS_SYNC_TIMEOUT_S` wall. - `judge_escalation` is silently skipped when `judge` is off. - `transform` is silently a no-op when `options.transformations` is empty even with the toggle on. diff --git a/env_template b/env_template index 1b8e74f..bf59c1a 100644 --- a/env_template +++ b/env_template @@ -55,16 +55,19 @@ FLYDOCS_MODEL=anthropic:claude-sonnet-4-6 FLYDOCS_FALLBACK_MODEL=openai:gpt-4o # Default models for each pipeline stage. Any stage left unset falls back -# to the global FLYDOCS_MODEL (or the request's ``options.model``). -FLYDOCS_SPLITTER_MODEL=anthropic:claude-sonnet-4-6 -FLYDOCS_CLASSIFIER_MODEL=anthropic:claude-sonnet-4-6 +# to the global FLYDOCS_MODEL (or the request's ``options.model``); a set +# stage wins over the request's ``options.model``. Tiering: mechanical +# stages (splitting, classification, bbox value matching) run on haiku, +# extraction and reasoning on sonnet, the judge on opus. +FLYDOCS_SPLITTER_MODEL=anthropic:claude-haiku-4-5 +FLYDOCS_CLASSIFIER_MODEL=anthropic:claude-haiku-4-5 FLYDOCS_EXTRACT_MODEL=anthropic:claude-sonnet-4-6 FLYDOCS_VISUAL_AUTHENTICITY_MODEL=anthropic:claude-sonnet-4-6 FLYDOCS_CONTENT_AUTHENTICITY_MODEL=anthropic:claude-sonnet-4-6 FLYDOCS_JUDGE_MODEL=anthropic:claude-opus-4-8 FLYDOCS_RULE_ENGINE_MODEL=anthropic:claude-sonnet-4-6 FLYDOCS_TRANSFORM_MODEL=anthropic:claude-sonnet-4-6 -FLYDOCS_BBOX_MATCHER_MODEL=anthropic:claude-sonnet-4-6 +FLYDOCS_BBOX_MATCHER_MODEL=anthropic:claude-haiku-4-5 # Maximum pages per single extraction request. Documents above this go through # the async API only. The document bytes are sent directly to the multimodal @@ -75,15 +78,25 @@ FLYDOCS_MAX_SYNC_PAGES=10 FLYDOCS_MAX_BYTES=33554432 # Closed-loop targeted repair (stage is opt-in per request via -# ``options.stages.repair``). Fields that failed the judge or a validator -# are re-extracted once with the failure evidence quoted back to the -# model. Default model for the repair pass; unset falls back to the -# global FLYDOCS_MODEL. A stronger model than the extractor is -# recommended -- it is fixing what the extractor got wrong. Note the -# repair pass consumes part of the FLYDOCS_ASYNC_TIMEOUT_S budget on -# async jobs -- raise that budget if you also raise this timeout. +# ``options.stages.repair``). Fields the judge marked FAIL or +# flag_for_review, or that failed a validator, are re-extracted once +# with the failure evidence quoted back to the model. Default model for +# the repair pass; unset falls back to the global FLYDOCS_MODEL. A +# stronger model than the extractor is recommended -- it is fixing what +# the extractor got wrong. Note the repair pass consumes part of the +# FLYDOCS_ASYNC_TIMEOUT_S budget on async jobs -- raise that budget if +# you also raise this timeout. Sync requests stay capped by +# FLYDOCS_SYNC_TIMEOUT_S, so repair usually warrants the async API. FLYDOCS_REPAIR_MODEL=anthropic:claude-opus-4-8 FLYDOCS_REPAIR_TIMEOUT_S=300 +# Set to false to repair only hard failures (judge FAIL or a validator +# error), skipping flagged-but-PASS fields that are usually correct. +FLYDOCS_REPAIR_INCLUDE_FLAGGED=true +# Above this failing-field fraction per task, targeted repair steps +# aside and leaves the full re-run to judge_escalation (1.0 = no cap). +FLYDOCS_REPAIR_MAX_FAILING_FRACTION=0.5 +# Max tasks repaired concurrently; lower it under provider rate limits. +FLYDOCS_REPAIR_TASK_CONCURRENCY=4 # Per-call timeouts (seconds). Sync requests get the shorter one; async jobs # retry up to ``JOB_MAX_ATTEMPTS`` times on timeout. diff --git a/src/flydocs/config.py b/src/flydocs/config.py index 56db855..5a56878 100644 --- a/src/flydocs/config.py +++ b/src/flydocs/config.py @@ -234,8 +234,17 @@ class IDPSettings(BaseSettings): # ``repair_model`` pins the pass to its own model (``None`` = the # request model). Runs BEFORE judge_escalation, so the full re-run # only fires when targeted repair was not enough. + # ``repair_include_flagged=false`` restricts repair to hard failures + # (judge FAIL or a validator error), skipping ambiguous flagged-PASS + # fields. ``repair_max_failing_fraction`` caps scope: above it the doc + # is globally untrustworthy, so repair steps aside for judge_escalation + # (1.0 disables). ``repair_task_concurrency`` bounds parallel repairs + # under provider rate limits, mirroring ``bbox_refine_doc_concurrency``. repair_model: str | None = None repair_timeout_s: int = 300 + repair_include_flagged: bool = True + repair_max_failing_fraction: float = 0.5 + repair_task_concurrency: int = 4 # -- Webhook -------------------------------------------------------- # The result webhook delivers the full extraction (split docs + fields + diff --git a/src/flydocs/core/configuration.py b/src/flydocs/core/configuration.py index ed2b9a5..2c4cc14 100644 --- a/src/flydocs/core/configuration.py +++ b/src/flydocs/core/configuration.py @@ -372,6 +372,9 @@ def field_repairer( judge=judge, field_validator=field_validator, default_model=settings.repair_model, + include_flagged=settings.repair_include_flagged, + max_failing_fraction=settings.repair_max_failing_fraction, + task_concurrency=settings.repair_task_concurrency, ) # ------------------------------------------------------------------ diff --git a/src/flydocs/core/services/repair/field_repairer.py b/src/flydocs/core/services/repair/field_repairer.py index 68d1eba..a4cd980 100644 --- a/src/flydocs/core/services/repair/field_repairer.py +++ b/src/flydocs/core/services/repair/field_repairer.py @@ -25,7 +25,11 @@ re-check when the judge stage is enabled) and replace the originals only when they come back clean -- with the judge on, that means an explicit PASS, and a null candidate never replaces a value. The repair -is monotonic: it can fix fields, never degrade them. +is monotonic at the field level: a field is only replaced by a verified +candidate. Array fields are the caveat: they are re-extracted and +accepted whole, so a previously-correct row may be replaced by a new +verifier-passing value -- ``RepairInfo.rows_changed`` records how many +rows differ per repaired array so table repairs can be diffed. Opt-in via ``options.stages.repair``. Runs after judge and BEFORE judge_escalation, so the expensive full re-run only fires when targeted @@ -41,6 +45,7 @@ from typing import Any from flydocs.core.services.extraction.extractor import MultimodalExtractor +from flydocs.core.services.extraction.pdf_slicer import PageRange, slice_pdf from flydocs.core.services.judge import Judge from flydocs.core.services.validation.field_validator import FieldValidator from flydocs.interfaces.dtos.document_type import DocumentTypeSpec @@ -59,23 +64,30 @@ class FailingField: field: str value: Any evidence: str + # 1-indexed slice-relative pages the value was reported on; empty when + # unknown (typical for null-value failures, whose page is discarded). + pages: tuple[int, ...] = () -def collect_failing_fields(groups: list[ExtractedFieldGroup]) -> list[FailingField]: +def collect_failing_fields( + groups: list[ExtractedFieldGroup], *, include_flagged: bool = True +) -> list[FailingField]: """Top-level fields whose judge verdict or validation failed. A field fails when ``judge.status == FAIL``, ``judge.flag_for_review`` - is set, or ``validation.valid`` is false. The evidence string joins - the judge's reasoning with every validator error message. Array - fields carry ``valid=False`` with an empty parent error list (the - messages live on the row sub-fields), so the evidence for arrays is - harvested from the rows. + is set (unless ``include_flagged`` is false -- flagged-but-PASS fields + are usually ambiguous-but-correct, so cost-sensitive deployments can + restrict repair to hard failures), or ``validation.valid`` is false. + The evidence string joins the judge's reasoning with every validator + error message. Array fields carry ``valid=False`` with an empty parent + error list (the messages live on the row sub-fields), so the evidence + for arrays is harvested from the rows. """ failures: list[FailingField] = [] for group in groups: for field in group.fields: reasons: list[str] = [] - if field.judge.status == JudgeStatus.FAIL or field.judge.flag_for_review: + if field.judge.status == JudgeStatus.FAIL or (include_flagged and field.judge.flag_for_review): reasons.append(field.judge.evidence or field.judge.notes or "judge rejected the value") if not field.validation.valid: messages = [e.message for e in field.validation.errors] @@ -89,11 +101,22 @@ def collect_failing_fields(groups: list[ExtractedFieldGroup]) -> list[FailingFie field=field.name, value=field.value, evidence="; ".join(reasons), + pages=_field_pages(field), ) ) return failures +def _field_pages(field: ExtractedField) -> tuple[int, ...]: + """Every page the field (or, for arrays, its rows) was reported on.""" + pages = set(field.pages) + if isinstance(field.value, list): + for sub in field.value: + if isinstance(sub, ExtractedField): + pages.update(_field_pages(sub)) + return tuple(sorted(pages)) + + def _row_error_messages(rows: list[Any]) -> list[str]: """Validator messages stamped on the row sub-fields of an array field.""" messages: list[str] = [] @@ -117,11 +140,17 @@ def __init__( judge: Judge, field_validator: FieldValidator, default_model: str | None, + include_flagged: bool = True, + max_failing_fraction: float = 0.5, + task_concurrency: int = 4, ) -> None: self._extractor = extractor self._judge = judge self._field_validator = field_validator self._default_model = default_model + self._include_flagged = include_flagged + self._max_failing_fraction = max_failing_fraction + self._task_concurrency = task_concurrency async def maybe_repair(self, ctx: Any, request: ExtractionRequest) -> RepairInfo | None: """Return a :class:`RepairInfo` when at least one field was flagged, else ``None``. @@ -132,20 +161,42 @@ async def maybe_repair(self, ctx: Any, request: ExtractionRequest) -> RepairInfo tasks: list[Any] = ctx.metadata.get("tasks", []) model = self._default_model or ctx.metadata.get("model_id") flagged = 0 + skipped_tasks = 0 repaired_paths: list[str] = [] + rows_changed: dict[str, int] = {} + # Same rationale as bbox_refine_doc_concurrency: each repaired task + # multiplies in-flight LLM calls; lower it under provider rate limits. + semaphore = asyncio.Semaphore(max(1, self._task_concurrency)) async def _repair_task(task: Any) -> None: - nonlocal flagged - failures = collect_failing_fields(task.extracted_groups) + nonlocal flagged, skipped_tasks + failures = collect_failing_fields(task.extracted_groups, include_flagged=self._include_flagged) if not failures: return flagged += len(failures) + total = sum(len(group.fields) for group in task.extracted_groups) + if total and len(failures) / total > self._max_failing_fraction: + # The extraction is globally untrustworthy: a focused pass + # would re-extract nearly everything on top of the eventual + # escalation re-run. Leave the FAIL verdicts in place so + # judge_escalation triggers as before. + skipped_tasks += 1 + logger.info( + "repair skipped for %s: %d/%d fields failing exceeds max fraction %.2f", + task.task_id, + len(failures), + total, + self._max_failing_fraction, + ) + return try: - accepted = await self._repair_one(task, failures, request, model) + async with semaphore: + accepted, task_rows_changed = await self._repair_one(task, failures, request, model) except Exception as exc: # noqa: BLE001 -- repair must never break the pipeline logger.warning("repair failed for %s: %s; keeping original fields", task.task_id, exc) return repaired_paths.extend(accepted) + rows_changed.update(task_rows_changed) await asyncio.gather(*(_repair_task(t) for t in tasks)) @@ -157,11 +208,14 @@ async def _repair_task(task: Any) -> None: fields_flagged=flagged, fields_repaired=len(repaired_paths), repaired_fields=sorted(repaired_paths), + tasks_skipped=skipped_tasks, + rows_changed=rows_changed, ) logger.info( - "repair flagged=%d repaired=%d model=%s", + "repair flagged=%d repaired=%d skipped_tasks=%d model=%s", info.fields_flagged, info.fields_repaired, + info.tasks_skipped, model, ) return info @@ -172,28 +226,30 @@ async def _repair_one( failures: list[FailingField], request: ExtractionRequest, model: str | None, - ) -> list[str]: - """Repair one task; return the ``group.field`` paths that were accepted.""" + ) -> tuple[list[str], dict[str, int]]: + """Repair one task; return the accepted ``group.field`` paths and, + for accepted array fields, how many rows differ from the original.""" failing_keys = {(f.group, f.field) for f in failures} subset = _subset_spec(task.doc_spec, failing_keys) + doc_bytes, page_count, page_offset = _repair_slice(task, failures) repaired_groups = await self._extractor.extract_repair( - document_bytes=task.slice_bytes, + document_bytes=doc_bytes, media_type=task.segment.media_type, - page_count=task.slice_pages, + page_count=page_count, doc=subset, failing_fields_text=_failures_text(failures), language_hint=request.options.language_hint, model=model, ) if not repaired_groups: - return [] + return [], {} # Re-verify: validators always (deterministic, free); judge subset # re-check only when the caller enabled the judge stage. self._field_validator.validate(subset.field_groups, repaired_groups) if request.options.stages.judge: await self._judge.judge( - document_bytes=task.slice_bytes, + document_bytes=doc_bytes, media_type=task.segment.media_type, doc=subset, extracted_groups=repaired_groups, @@ -206,6 +262,7 @@ async def _repair_one( } judged = request.options.stages.judge accepted: list[str] = [] + rows_changed: dict[str, int] = {} for group in task.extracted_groups: for index, field in enumerate(group.fields): key = (group.name, field.name) @@ -214,9 +271,16 @@ async def _repair_one( candidate = repaired_by_key.get(key) if candidate is None or not _is_clean(candidate, require_judge_pass=judged): continue + if page_offset: + _shift_pages(candidate, page_offset) + path = f"{group.name}.{field.name}" + if isinstance(field.value, list) and isinstance(candidate.value, list): + changed = _changed_row_count(field.value, candidate.value) + if changed: + rows_changed[path] = changed group.fields[index] = candidate - accepted.append(f"{group.name}.{field.name}") - return accepted + accepted.append(path) + return accepted, rows_changed def _is_clean(field: ExtractedField, *, require_judge_pass: bool) -> bool: @@ -237,6 +301,66 @@ def _is_clean(field: ExtractedField, *, require_judge_pass: bool) -> bool: return field.judge.status != JudgeStatus.FAIL and not field.judge.flag_for_review +def _repair_slice(task: Any, failures: list[FailingField]) -> tuple[bytes, int, int]: + """Bytes, page count and page offset for the focused repair pass. + + The dominant repair cost is the input pages, not the schema subset, + so when every failing field carries page provenance the segment PDF + is sliced to the failing span plus one page of margin. Falls back to + the full segment slice when the media is not PDF, any failing field + has no known pages (nulled values lose theirs), the span would not + actually shrink the document, or slicing fails. + """ + full = (task.slice_bytes, task.slice_pages, 0) + if task.segment.media_type != "application/pdf": + return full + if not failures or any(not f.pages for f in failures): + return full + pages = sorted({page for f in failures for page in f.pages}) + start = max(1, pages[0] - 1) + end = min(task.slice_pages, pages[-1] + 1) + if end - start + 1 >= task.slice_pages: + return full + try: + sliced = slice_pdf(task.slice_bytes, PageRange(start=start, end=end)) + except Exception as exc: # noqa: BLE001 -- slicing is an optimization, never fatal + logger.warning("repair page slicing failed for %s: %s; using full slice", task.task_id, exc) + return full + return sliced, end - start + 1, start - 1 + + +def _shift_pages(field: ExtractedField, offset: int) -> None: + """Map slice-relative page numbers of an accepted candidate back to + segment coordinates (recursing into array rows and sub-fields).""" + field.pages = [page + offset for page in field.pages] + if isinstance(field.value, list): + for sub in field.value: + if isinstance(sub, ExtractedField): + _shift_pages(sub, offset) + + +def _changed_row_count(old_rows: list[Any], new_rows: list[Any]) -> int: + """Rows of a repaired array whose content differs from the original. + + Compared positionally by (name, value) signature -- rows have no + stable keys, so an inserted or dropped row also counts as a change. + """ + changed = sum( + 1 + for old, new in zip(old_rows, new_rows, strict=False) + if _row_signature(old) != _row_signature(new) + ) + return changed + abs(len(old_rows) - len(new_rows)) + + +def _row_signature(row: Any) -> Any: + if isinstance(row, ExtractedField): + if isinstance(row.value, list): + return (row.name, tuple(_row_signature(sub) for sub in row.value)) + return (row.name, row.value) + return row + + def _subset_spec(doc: DocumentTypeSpec, failing_keys: set[tuple[str, str]]) -> DocumentTypeSpec: """A copy of ``doc`` whose field_groups contain only the failing fields.""" groups = [] diff --git a/src/flydocs/core/services/validation/request_validator.py b/src/flydocs/core/services/validation/request_validator.py index 96fc60c..cccfc33 100644 --- a/src/flydocs/core/services/validation/request_validator.py +++ b/src/flydocs/core/services/validation/request_validator.py @@ -100,13 +100,13 @@ class RequestValidator: share one instance. """ - def validate(self, request: ExtractionRequest) -> ValidationReport: + def validate(self, request: ExtractionRequest, *, sync: bool = False) -> ValidationReport: report = ValidationReport() self._check_files(request, report) self._check_document_types(request, report) self._check_rule_references(request, report) self._check_rule_dag(request, report) - self._check_stage_consistency(request, report) + self._check_stage_consistency(request, report, sync=sync) return report # -- file-level checks (multi-file shape) ---------------------------- @@ -338,7 +338,9 @@ def _check_rule_dag(self, request: ExtractionRequest, report: ValidationReport) # -- stage / toggle consistency -------------------------------------- - def _check_stage_consistency(self, request: ExtractionRequest, report: ValidationReport) -> None: + def _check_stage_consistency( + self, request: ExtractionRequest, report: ValidationReport, *, sync: bool = False + ) -> None: stages = request.options.stages # rule_engine on but no rules => the stage is a no-op. Warn. @@ -385,6 +387,24 @@ def _check_stage_consistency(self, request: ExtractionRequest, report: Validatio ) ) + # repair on a synchronous request => the extra LLM pass eats into + # the hard sync wall (FLYDOCS_SYNC_TIMEOUT_S) and raises the odds + # of a 408. Warn; the async API has the budget for it. + if sync and stages.repair: + report.issues.append( + ValidationIssue( + severity="warning", + code="repair_sync_latency", + message=( + "stages.repair adds an extra LLM pass per failing " + "field; synchronous requests are hard-capped at " + "FLYDOCS_SYNC_TIMEOUT_S and are likely to return " + "408 -- prefer the async POST /api/v1/extractions." + ), + path="options.stages.repair", + ) + ) + # repair on but no stage produces failure signals => the node is # never added to the DAG. Warn. if stages.repair and not (stages.judge or stages.field_validation): diff --git a/src/flydocs/interfaces/dtos/extract.py b/src/flydocs/interfaces/dtos/extract.py index 22ae2a3..340e512 100644 --- a/src/flydocs/interfaces/dtos/extract.py +++ b/src/flydocs/interfaces/dtos/extract.py @@ -260,6 +260,13 @@ class RepairInfo(BaseModel): fields_flagged: int = Field(default=0, ge=0) fields_repaired: int = Field(default=0, ge=0) repaired_fields: list[str] = Field(default_factory=list) + # Tasks whose repair was skipped because the failing-field fraction + # exceeded FLYDOCS_REPAIR_MAX_FAILING_FRACTION (escalation handles them). + tasks_skipped: int = Field(default=0, ge=0) + # Array fields are accepted whole, so previously-correct rows may be + # replaced; per repaired ``group.field`` path, how many rows differ + # from the original (row count changes included). + rows_changed: dict[str, int] = Field(default_factory=dict) class UsageBreakdown(BaseModel): diff --git a/src/flydocs/web/controllers/extract_controller.py b/src/flydocs/web/controllers/extract_controller.py index 4e4a56b..32600de 100644 --- a/src/flydocs/web/controllers/extract_controller.py +++ b/src/flydocs/web/controllers/extract_controller.py @@ -99,7 +99,7 @@ async def validate(self, request: Valid[Body[ExtractionRequest]]) -> ValidationR whether to proceed. Identical schema for both errors and warnings: ``[{severity, code, message, path}]``. """ - report = self._validator.validate(request) + report = self._validator.validate(request, sync=True) return ValidationResponse( ok=not report.has_errors, error_count=len(report.errors), @@ -163,7 +163,7 @@ def _enforce_size_limits(request: ExtractionRequest, *, max_bytes: int) -> None: def _enforce_semantic_validation(request: ExtractionRequest, validator: RequestValidator) -> None: """Reject the request with a 422 when the semantic validator finds errors.""" - report: ValidationReport = validator.validate(request) + report: ValidationReport = validator.validate(request, sync=True) if report.has_errors: raise _http_problem_with_payload( status_code=422, diff --git a/tests/unit/test_field_repairer.py b/tests/unit/test_field_repairer.py index e43d887..84b90e7 100644 --- a/tests/unit/test_field_repairer.py +++ b/tests/unit/test_field_repairer.py @@ -22,11 +22,14 @@ from __future__ import annotations +import asyncio import base64 +import io from types import SimpleNamespace from typing import Any from unittest.mock import AsyncMock, MagicMock +import pypdf import pytest from flydocs.config import IDPSettings @@ -169,6 +172,38 @@ def test_collect_failing_fields_includes_flag_for_review() -> None: assert [(f.group, f.field) for f in failures] == [("identity", "number")] +def test_collect_failing_fields_can_exclude_flagged_pass_fields() -> None: + """include_flagged=False restricts collection to hard failures: a PASS + field that is merely flag_for_review (ambiguous-but-correct) is skipped.""" + flagged = ExtractedField( + name="number", + value="X123", + judge=JudgeOutcome(status=JudgeStatus.PASS, flag_for_review=True), + ) + hard_fail = _judge_failed_field("iban", "ES00", "checksum") + groups = [ExtractedFieldGroup(name="identity", fields=[flagged, hard_fail])] + failures = collect_failing_fields(groups, include_flagged=False) + assert [(f.group, f.field) for f in failures] == [("identity", "iban")] + + +@pytest.mark.asyncio +async def test_maybe_repair_skips_flagged_pass_fields_when_configured() -> None: + flagged = ExtractedField( + name="number", + value="X123", + judge=JudgeOutcome(status=JudgeStatus.PASS, flag_for_review=True), + ) + task = _task([ExtractedFieldGroup(name="identity", fields=[flagged])]) + extractor = MagicMock() + extractor.extract_repair = AsyncMock(return_value=[]) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock()), include_flagged=False) + info = await repairer.maybe_repair(_ctx([task]), _request()) + + assert info is None + extractor.extract_repair.assert_not_awaited() + + def test_collect_failing_fields_empty_when_all_clean() -> None: groups = [ExtractedFieldGroup(name="identity", fields=[_clean_field("name", "JOHN")])] assert collect_failing_fields(groups) == [] @@ -227,12 +262,18 @@ def _repairer( judge: Any, *, default_model: str | None = None, + include_flagged: bool = True, + max_failing_fraction: float = 1.0, + task_concurrency: int = 4, ) -> FieldRepairer: return FieldRepairer( extractor=extractor, judge=judge, field_validator=FieldValidator(), default_model=default_model, + include_flagged=include_flagged, + max_failing_fraction=max_failing_fraction, + task_concurrency=task_concurrency, ) @@ -403,6 +444,214 @@ async def test_maybe_repair_falls_back_to_request_model() -> None: assert extractor.extract_repair.await_args.kwargs["model"] == "request-model" +# --------------------------------------------------------------------------- +# Page-sliced repair +# --------------------------------------------------------------------------- + + +def _pdf(pages: int) -> bytes: + writer = pypdf.PdfWriter() + for _ in range(pages): + writer.add_blank_page(width=72, height=72) + buf = io.BytesIO() + writer.write(buf) + return buf.getvalue() + + +def _paged_task(groups: list[ExtractedFieldGroup], *, pages: int) -> Any: + task = _task(groups) + task.slice_bytes = _pdf(pages) + task.slice_pages = pages + return task + + +@pytest.mark.asyncio +async def test_repair_slices_pdf_to_failing_pages_and_remaps_candidate() -> None: + """A failure localized on page 3 of 5 re-sends only pages 2-4 (one page + of margin); the accepted candidate's slice-relative pages are shifted + back to document coordinates.""" + failing = _judge_failed_field("number", "X123", "misread") + failing.pages = [3] + task = _paged_task([ExtractedFieldGroup(name="identity", fields=[failing])], pages=5) + extractor = MagicMock() + extractor.extract_repair = AsyncMock( + return_value=[ + ExtractedFieldGroup( + name="identity", fields=[ExtractedField(name="number", value="Y456", pages=[2])] + ) + ] + ) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock())) + info = await repairer.maybe_repair(_ctx([task]), _request(judge=False)) + + kwargs = extractor.extract_repair.await_args.kwargs + assert kwargs["page_count"] == 3 + assert len(pypdf.PdfReader(io.BytesIO(kwargs["document_bytes"])).pages) == 3 + assert info is not None and info.fields_repaired == 1 + assert task.extracted_groups[0].fields[0].pages == [3] + + +@pytest.mark.asyncio +async def test_repair_sends_full_document_when_failing_pages_unknown() -> None: + """Null-value failures carry pages=[]; without provenance the repair + falls back to the full segment slice.""" + failing = _judge_failed_field("number", "X123", "misread") # pages=[] by default + task = _paged_task([ExtractedFieldGroup(name="identity", fields=[failing])], pages=5) + extractor = MagicMock() + extractor.extract_repair = AsyncMock(return_value=[]) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock())) + await repairer.maybe_repair(_ctx([task]), _request(judge=False)) + + kwargs = extractor.extract_repair.await_args.kwargs + assert kwargs["document_bytes"] == task.slice_bytes + assert kwargs["page_count"] == 5 + + +@pytest.mark.asyncio +async def test_repair_sends_full_document_for_non_pdf_media() -> None: + failing = _judge_failed_field("number", "X123", "misread") + failing.pages = [3] + task = _paged_task([ExtractedFieldGroup(name="identity", fields=[failing])], pages=5) + task.segment = SimpleNamespace(media_type="image/png") + extractor = MagicMock() + extractor.extract_repair = AsyncMock(return_value=[]) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock())) + await repairer.maybe_repair(_ctx([task]), _request(judge=False)) + + kwargs = extractor.extract_repair.await_args.kwargs + assert kwargs["document_bytes"] == task.slice_bytes + assert kwargs["page_count"] == 5 + + +# --------------------------------------------------------------------------- +# Array audit +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_maybe_repair_records_changed_rows_for_arrays() -> None: + """Arrays are accepted whole, so previously-correct rows may be replaced; + rows_changed records how many rows differ so QA can diff table repairs.""" + old_rows = [ + ExtractedField(name="row", value=[ExtractedField(name="a", value="1")]), + ExtractedField(name="row", value=[ExtractedField(name="b", value="-3")]), + ] + array_field = ExtractedField( + name="line_items", + value=old_rows, + validation=FieldValidation(valid=False, errors=[]), + ) + task = _task([ExtractedFieldGroup(name="items", fields=[array_field])]) + new_rows = [ + ExtractedField(name="row", value=[ExtractedField(name="a", value="1")]), # untouched + ExtractedField(name="row", value=[ExtractedField(name="b", value="3")]), # fixed + ] + extractor = MagicMock() + extractor.extract_repair = AsyncMock( + return_value=[ + ExtractedFieldGroup(name="items", fields=[ExtractedField(name="line_items", value=new_rows)]) + ] + ) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock())) + info = await repairer.maybe_repair(_ctx([task]), _request(judge=False)) + + assert info is not None and info.fields_repaired == 1 + assert info.rows_changed == {"items.line_items": 1} + + +# --------------------------------------------------------------------------- +# Scope cap +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_maybe_repair_skips_task_above_max_failing_fraction() -> None: + """When most fields failed the extraction is globally untrustworthy: + targeted repair is skipped and judge_escalation's full re-run (whose + trigger rate is unchanged) is the right tool.""" + task = _task( + [ + ExtractedFieldGroup( + name="identity", + fields=[ + _judge_failed_field("number", "X123", "misread"), + _judge_failed_field("name", "J0HN", "misread"), + ], + ) + ] + ) + extractor = MagicMock() + extractor.extract_repair = AsyncMock(return_value=[]) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock()), max_failing_fraction=0.5) + info = await repairer.maybe_repair(_ctx([task]), _request()) + + extractor.extract_repair.assert_not_awaited() + assert info is not None and info.triggered + assert info.fields_flagged == 2 + assert info.fields_repaired == 0 + assert info.tasks_skipped == 1 + + +@pytest.mark.asyncio +async def test_maybe_repair_runs_when_failing_fraction_at_or_below_cap() -> None: + task = _task( + [ + ExtractedFieldGroup( + name="identity", + fields=[ + _clean_field("name", "JOHN"), + _judge_failed_field("number", "X123", "misread"), + ], + ) + ] + ) + extractor = MagicMock() + extractor.extract_repair = AsyncMock(return_value=[]) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock()), max_failing_fraction=0.5) + info = await repairer.maybe_repair(_ctx([task]), _request()) + + extractor.extract_repair.assert_awaited_once() + assert info is not None and info.tasks_skipped == 0 + + +# --------------------------------------------------------------------------- +# Concurrency bound +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_maybe_repair_bounds_concurrent_task_repairs() -> None: + tasks = [ + _task([ExtractedFieldGroup(name="identity", fields=[_judge_failed_field("number", "X", "bad")])]) + for _ in range(3) + ] + in_flight = 0 + peak = 0 + + async def _tracked_repair(**_: Any) -> list[ExtractedFieldGroup]: + nonlocal in_flight, peak + in_flight += 1 + peak = max(peak, in_flight) + await asyncio.sleep(0.01) + in_flight -= 1 + return [] + + extractor = MagicMock() + extractor.extract_repair = AsyncMock(side_effect=_tracked_repair) + + repairer = _repairer(extractor, MagicMock(judge=AsyncMock()), task_concurrency=1) + await repairer.maybe_repair(_ctx(tasks), _request()) + + assert extractor.extract_repair.await_count == 3 + assert peak == 1 + + # --------------------------------------------------------------------------- # Prompt catalog # --------------------------------------------------------------------------- diff --git a/tests/unit/test_request_validator.py b/tests/unit/test_request_validator.py index 50846e3..5e6774c 100644 --- a/tests/unit/test_request_validator.py +++ b/tests/unit/test_request_validator.py @@ -267,6 +267,24 @@ def test_repair_with_judge_produces_no_warning(validator: RequestValidator) -> N assert "repair_no_verification_stage" not in codes +# -- warning: repair on a synchronous request --------------------------------- + + +def test_repair_on_sync_request_warns_about_latency(validator: RequestValidator) -> None: + options = ExtractionOptions(stages=StageToggles(repair=True, judge=True)) + report = validator.validate(_request(options=options), sync=True) + assert not report.has_errors + codes = [i.code for i in report.warnings] + assert "repair_sync_latency" in codes + + +def test_repair_on_async_request_has_no_sync_warning(validator: RequestValidator) -> None: + options = ExtractionOptions(stages=StageToggles(repair=True, judge=True)) + report = validator.validate(_request(options=options)) + codes = [i.code for i in report.warnings] + assert "repair_sync_latency" not in codes + + # -- warning: visual_authenticity on but no visual checks --------------------