Build a validated structured extraction pipeline with bounded repair

• Use a strict schema with closed enums, bounded strings, and explicit nullable fields.
• Validate evidence offsets and quoted text against the original input after parsing.
• Keep normalization deterministic and separate from model generation.
• Allow no more than one repair call, and include only validation errors needed for repair.
• Return a typed terminal outcome for success, refusal, review required, or dependency failure.
• Record model and schema versions plus safe validation metrics without logging the email body.

1. Define the external schema and a separate domain result so provider response details do not leak through the application.
2. Call the model with strict structured output and classify transport, refusal, and parse outcomes.
3. Run deterministic semantic checks for evidence offsets, supported urgency, and summary grounding.
4. If only repairable validation errors remain, issue one constrained repair request against the original text.
5. Map a valid result into the domain type; otherwise return review required with safe reason codes.
6. Emit latency, token, refusal, repair, and validation-failure signals keyed by version rather than raw content.

from dataclasses import dataclass
from enum import StrEnum

class Outcome(StrEnum):
    OK = "ok"
    REFUSED = "refused"
    REVIEW = "review_required"
    DEPENDENCY_FAILED = "dependency_failed"

@dataclass(frozen=True)
class ExtractionResult:
    outcome: Outcome
    ticket: Ticket | None = None
    reason_codes: tuple[str, ...] = ()

async def extract_ticket(text: str) -> ExtractionResult:
    request = build_request(text, schema=TICKET_SCHEMA, strict=True)
    for attempt in range(2):
        try:
            response = await gateway.generate(request, timeout_seconds=8)
        except TemporaryModelFailure:
            metrics.increment("extract.dependency_failed")
            return ExtractionResult(Outcome.DEPENDENCY_FAILED)

        if response.refusal:
            return ExtractionResult(Outcome.REFUSED, reason_codes=("model_refusal",))

        parsed = parse_ticket(response.output)
        errors = validate_ticket(parsed, source=text)
        if not errors:
            metrics.increment("extract.ok", tags=version_tags(response))
            return ExtractionResult(Outcome.OK, ticket=normalize(parsed))

        if attempt == 0 and all(error.repairable for error in errors):
            request = build_repair_request(text, response.output, safe_codes(errors))
            continue
        return ExtractionResult(Outcome.REVIEW, reason_codes=tuple(safe_codes(errors)))

    raise AssertionError("bounded loop exhausted")

The deterministic work is O(n) in the input and cited evidence. Model latency dominates, while the hard two-call limit bounds cost and tail latency.

• The provider returns schema-valid JSON whose evidence quote does not match the source offsets.
• The source lacks enough information to assign urgency without guessing.
• A model refusal is syntactically valid but is not a successful extraction.
• The repair response introduces a new validation error or changes already valid evidence.
• The request times out after the provider processed it, so metrics must not imply a clean model failure.

• Return a valid ticket on the first response and assert no repair call occurs.
• Return invalid evidence offsets, then a corrected response, and assert exactly two total calls.
• Return semantic invalidity twice and assert a review outcome rather than a third call.
• Return a refusal and verify it is not parsed as an empty ticket.
• Inspect telemetry and assert neither the source email nor generated summary is recorded.

• Assuming schema adherence proves the extracted facts are supported by the source.
• Retrying until valid output creates an unbounded cost and latency loop.
• Using one exception for refusals, timeouts, parse errors, and semantic failures hides operational causes.
• Sending the whole conversation plus internal validation details into every repair call expands exposure.
• Logging prompts and outputs by default leaks customer data into the observability system.