# Model response contracts


class ProductResponse(StrictModel):
    """The semantic output accepted from initial manufacture."""

    product: NonEmptyText = Field(description="Complete replacement content for the product revision.")


class RepairChange(StrictModel):
    """A repair operation's account of the change made for one assigned defect."""

    defect_id: DefectId = Field(description="Assigned defect this change was intended to correct.")
    summary: NonEmptyText = Field(description="Concise description of the material change made to the product.")


class RepairResponse(StrictModel):
    """Replacement product and structured account returned by repair."""

    product: NonEmptyText = Field(description="Complete replacement content for the product revision.")
    changes: tuple[RepairChange, ...] = Field(
        min_length=1,
        description="One account of the material change made for each assigned defect.",
    )


class InspectionObservation(StrictModel):
    """One semantic judgment against one declared acceptance criterion."""

    criterion_id: CriterionId = Field(description="Acceptance criterion being assessed.")
    outcome: ObservationOutcome = Field(description="Semantic assessment of the criterion for this product.")
    explanation: NonEmptyText = Field(description="Reason the inspecting model recorded this outcome.")
    evidence: tuple[EvidenceReference, ...] = Field(
        min_length=1,
        description="Recorded product, material or evidence identities supporting the observation.",
    )


class InspectionResponse(StrictModel):
    """Inspection must return complete observations, never a pass/fail decision."""

    observations: tuple[InspectionObservation, ...] = Field(
        min_length=1,
        description="One observation for every criterion in the assigned acceptance basis.",
    )


# Each semantic stage receives one explicit input contract. This avoids
# constructing ad-hoc prompt dictionaries and makes the handoffs inspectable.


class ManufactureInputs(StrictModel):
    """Recorded inputs supplied to the manufacturing operation."""

    source_material: SourceMaterial
    product_assignment: ProductAssignment
    acceptance_basis: AcceptanceBasis


# Retained production records


class MachineryRecord(StrictModel):
    """The machinery identity retained for one semantic operation."""

    provider: ProviderName = Field(description="Adapter used to invoke the model.")
    model: NonEmptyText = Field(description="Requested or provider-selected model identity.")
    client: NonEmptyText = Field(description="Client and version which performed the invocation.")
    authentication: NonEmptyText = Field(description="Authentication mechanism, never the credential itself.")
    response_id: NonEmptyText | None = Field(default=None, description="Provider response identity when available.")
    usage: JsonObject | None = Field(
        default=None,
        description="Provider-reported usage metadata when it is available.",
    )


class ProductRevision(StrictModel):
    """One immutable manufactured or repaired version of the product."""

    id: ProductRevisionId = Field(description="Identity of this exact product revision.")
    revision: PositiveInt = Field(description="Monotonically increasing revision number within the run.")
    created_by: OperationId = Field(description="Semantic operation which produced this revision.")
    source_material: MaterialSetId = Field(description="Source-material set from which production began.")
    acceptance_basis: AcceptanceBasisId = Field(description="Acceptance standard assigned to this product.")
    content_sha256: Sha256Digest = Field(description="Digest binding the record to its exact content.")
    content: NonEmptyText = Field(description="Complete product content for this revision.")
    replaces: ProductRevisionId | None = Field(default=None, description="Rejected revision replaced by this one.")
    supporting_information: SupportingInformationId | None = Field(
        default=None,
        description="Supporting information used during repair, when applicable.",
    )

    @model_validator(mode="after")
    def validate_revision_lineage(self) -> Self:
        """Keep initial manufacture distinct from later repaired revisions."""

        if self.revision == 1:
            if self.replaces is not None or self.supporting_information is not None:
                raise ValueError("The first manufactured revision cannot replace earlier work or use repair evidence.")
            return self
        if self.replaces is None or self.supporting_information is None:
            raise ValueError(
                "A repaired revision must identify the revision it replaces and the supporting information used."
            )
        return self


class SupportingInformation(StrictModel):
    """Deterministically gathered evidence supplied to inspection and repair."""

    id: SupportingInformationId = Field(description="Identity of this gathered information set.")
    source: EvidenceStoreId = Field(description="Evidence store from which the records were gathered.")
    ticket_evidence: dict[TicketId, tuple[EvidenceRecord, ...]] = Field(
        description="Exact supporting records gathered for each source ticket."
    )


class DefectRecord(StrictModel):
    """An unmet requirement derived deterministically from an inspection observation."""

    id: DefectId = Field(description="Stable identity used to assign this defect to repair.")
    product_id: ProductRevisionId = Field(description="Rejected product revision containing the defect.")
    inspection_id: InspectionId = Field(description="Inspection which observed the unmet requirement.")
    criterion_id: CriterionId = Field(description="Acceptance criterion which was not satisfied.")
    outcome: ObservationOutcome = Field(description="Original inspection outcome; defects are always unsatisfied.")
    tolerance: Tolerance = Field(description="Declared release effect copied from the acceptance basis.")
    repair_group: PositiveInt = Field(description="Declared policy group used to schedule repair.")
    description: NonEmptyText = Field(description="Inspection explanation passed to repair.")
    evidence: tuple[EvidenceReference, ...] = Field(
        min_length=1, description="Evidence cited by the failed observation."
    )

    @model_validator(mode="after")
    def require_unsatisfied_outcome(self) -> Self:
        """Prevent unresolved observations being silently converted into repair defects."""

        if self.outcome is not ObservationOutcome.UNSATISFIED:
            raise ValueError("A defect record must originate from an unsatisfied observation.")
        return self


class InspectionAssessment(StrictModel):
    """Complete semantic observations awaiting a deterministic disposition."""

    id: InspectionId = Field(description="Identity of this complete inspection attempt.")
    product_id: ProductRevisionId = Field(description="Exact product revision inspected.")
    acceptance_basis: AcceptanceBasisId = Field(description="Standard applied during this inspection.")
    created_by: OperationId = Field(description="Semantic operation which produced the observations.")
    observations: tuple[InspectionObservation, ...] = Field(min_length=1, description="Recorded semantic judgments.")


class InspectionRecord(InspectionAssessment):
    """Retained inspection after the decision table has assigned its disposition."""

    disposition: Disposition = Field(description="Deterministic result applied to the semantic observations.")
    decision_reason: NonEmptyText = Field(description="Why the decision table chose this disposition.")
    assigned_repair_group: PositiveInt | None = Field(
        default=None,
        description="Earliest blocking repair group selected for the next attempt.",
    )
    assigned_defects: tuple[DefectId, ...] = Field(
        default=(),
        description="Exact defects assigned to the next repair operation.",
    )

    @model_validator(mode="after")
    def validate_repair_assignment(self) -> Self:
        """Require repair routing and its bounded defect assignment to agree."""

        if len(self.assigned_defects) != len(set(self.assigned_defects)):
            raise ValueError("An inspection cannot assign the same defect twice.")
        has_assignment = bool(self.assigned_defects)
        if self.disposition is Disposition.REPAIR:
            if self.assigned_repair_group is None or not has_assignment:
                raise ValueError("A repair disposition requires a repair group and at least one assigned defect.")
        elif self.assigned_repair_group is not None or has_assignment:
            raise ValueError("Only a repair disposition may carry an assigned repair group or defects.")
        return self


class DefectSet(StrictModel):
    """All defects recorded for a revision and the bounded subset sent to repair."""

    product_id: ProductRevisionId = Field(description="Product revision in which these defects were observed.")
    defects: tuple[DefectRecord, ...] = Field(
        min_length=1,
        description="All unmet requirements recorded by this inspection.",
    )
    assigned_to_repair: tuple[DefectId, ...] = Field(
        description="Blocking defects in the earliest repair group, if any."
    )

    @model_validator(mode="after")
    def validate_assignment(self) -> Self:
        """Bind the repair assignment to defects actually present in this set."""

        defect_ids = tuple(defect.id for defect in self.defects)
        if len(defect_ids) != len(set(defect_ids)):
            raise ValueError("Defect IDs must be unique within a defect set.")
        if len(self.assigned_to_repair) != len(set(self.assigned_to_repair)):
            raise ValueError("A defect cannot be assigned to repair more than once.")
        unknown = set(self.assigned_to_repair) - set(defect_ids)
        if unknown:
            raise ValueError(f"Repair assignment references unknown defects: {sorted(unknown)}")
        return self


class RepairRecord(StrictModel):
    """Handoff connecting rejected work and assigned defects to its replacement."""

    id: RepairId = Field(description="Identity of this repair handoff.")
    input_product: ProductRevisionId = Field(description="Rejected revision supplied to repair.")
    defects: tuple[DefectId, ...] = Field(min_length=1, description="Exact defects assigned to this repair attempt.")
    changes: tuple[RepairChange, ...] = Field(
        min_length=1,
        description="Repair operation's account of the material change made for each assigned defect.",
    )
    repair_group: PositiveInt = Field(description="Policy group shared by every assigned defect.")
    output_product: ProductRevisionId = Field(description="Replacement revision produced by repair.")
    created_by: OperationId = Field(description="Semantic operation which produced the replacement.")

    @model_validator(mode="after")
    def require_replacement(self) -> Self:
        if len(self.defects) != len(set(self.defects)):
            raise ValueError("Repair cannot receive the same defect more than once.")
        changed_defects = tuple(change.defect_id for change in self.changes)
        if changed_defects != self.defects:
            raise ValueError("Repair changes must describe every assigned defect exactly once and in order.")
        if self.input_product == self.output_product:
            raise ValueError("Repair must create a new product revision.")
        return self


class ReleaseRecord(StrictModel):
    """Terminal record naming the accepted product and the evidence behind it."""

    id: ReleaseId = Field(description="Identity of this release decision.")
    released_at: datetime = Field(description="UTC time at which release completed.")
    product_id: ProductRevisionId = Field(description="Exact product revision made available for use.")
    product_sha256: Sha256Digest = Field(description="Digest binding release to the product content.")
    passing_inspection: InspectionId = Field(description="Passing inspection of the released revision.")
    source_material: MaterialSetId = Field(description="Source-material set from which production began.")
    supporting_information: SupportingInformationId = Field(
        description="Supporting information used by inspection and repair."
    )
    acceptance_basis: AcceptanceBasisId = Field(description="Acceptance standard satisfied by the product.")
    evidence: tuple[RecordPath, ...] = Field(
        min_length=1,
        description="Retained files supporting the release decision.",
    )


class OperationRecord(StrictModel):
    """Prompt, response, machinery and outcome retained for one semantic call."""

    id: OperationId = Field(description="Identity of this semantic operation.")
    operation: OperationName = Field(description="Bounded semantic job performed.")
    status: OperationStatus = Field(description="Whether the operation completed.")
    started_at: datetime = Field(description="UTC time at which invocation began.")
    duration_ms: int = Field(ge=0, description="Elapsed invocation and validation time in milliseconds.")
    prompt: RecordPath = Field(description="Path to the exact retained prompt.")
    prompt_sha256: Sha256Digest = Field(description="Digest of the retained prompt file.")
    response: RecordPath | None = Field(default=None, description="Path to the raw response, when one was received.")
    response_sha256: Sha256Digest | None = Field(default=None, description="Digest of the retained raw response file.")
    machinery: MachineryRecord | None = Field(
        default=None, description="Provider and model identity, when invocation began."
    )
    failure_type: NonEmptyText | None = Field(
        default=None, description="Stable exception class for a failed operation."
    )
    failure: NonEmptyText | None = Field(default=None, description="Actionable failure detail for diagnosis.")

    @model_validator(mode="after")
    def validate_terminal_state(self) -> Self:
        """Prevent completed and failed operation fields from being mixed."""

        if (self.response is None) != (self.response_sha256 is None):
            raise ValueError("A retained response path and its digest must be recorded together.")
        response_complete = self.response is not None and self.machinery is not None
        failure_complete = self.failure_type is not None and self.failure is not None
        if self.status is OperationStatus.COMPLETED:
            if not response_complete or failure_complete:
                raise ValueError("A completed operation requires response evidence and cannot contain a failure.")
        elif not failure_complete:
            raise ValueError("A failed operation requires failure type and detail.")
        return self


class ReleasedRunRecord(StrictModel):
    """Top-level record written when a product revision is released."""

    status: RunStatus
    finished_at: datetime
    provider: ProviderName
    requested_model: str | None = None
    input_file: NonEmptyText
    input_sha256: Sha256Digest
    release: ReleaseRecord

    @model_validator(mode="after")
    def require_released_status(self) -> Self:
        if self.status is not RunStatus.RELEASED:
            raise ValueError("A released run record must have released status.")
        return self


class StoppedRunRecord(StrictModel):
    """Top-level record written when the process releases nothing."""

    status: RunStatus
    finished_at: datetime
    provider: ProviderName
    requested_model: str | None = None
    input_file: NonEmptyText
    input_sha256: Sha256Digest
    reason: NonEmptyText

    @model_validator(mode="after")
    def require_stopped_status(self) -> Self:
        if self.status is not RunStatus.STOPPED:
            raise ValueError("A stopped run record must have stopped status.")
        return self


# Stage input contracts depend on retained-record types, so they are declared
# after those records rather than relying on implicit forward-model rebuilding.


class InspectionInputs(StrictModel):
    """Recorded inputs supplied to the inspecting operation."""

    product: ProductRevision
    source_material: SourceMaterial
    supporting_information: SupportingInformation
    acceptance_basis: AcceptanceBasis


class RepairInputs(StrictModel):
    """Recorded inputs supplied to the repair operation."""

    failed_product: ProductRevision
    assigned_defects: tuple[DefectRecord, ...] = Field(min_length=1)
    source_material: SourceMaterial
    supporting_information: SupportingInformation
    acceptance_basis: AcceptanceBasis

def manufacture(
    definition: ExampleDefinition,
    run_directory: Path,
    provider: ProviderName,
    model: str | None,
    operation_id: OperationId,
    product_id: ProductRevisionId,
) -> tuple[ProductRevision, OperationArtifacts]:
    """Manufacture revision 1 without allowing the model to accept its own work."""

    prompt = operation_prompt(
        "Manufacture the security queue described by the product assignment from the incoming tickets. "
        "This is the initial product revision, so make the best report the supplied tickets support without "
        "inventing independent evidence. Return the complete report as Markdown in the `product` field.",
        ManufactureInputs(
            source_material=definition.source_material,
            product_assignment=definition.product_assignment,
            acceptance_basis=definition.acceptance_basis,
        ),
    )

    operation = run_model_operation(
        run_directory=run_directory,
        operation_id=operation_id,
        name=OperationName.MANUFACTURE,
        provider=provider,
        model=model,
        prompt=prompt,
        response_type=ProductResponse,
    )
    content = operation.payload.product
    return (
        ProductRevision(
            id=product_id,
            revision=1,
            created_by=operation.artifacts.operation_id,
            source_material=definition.source_material.id,
            acceptance_basis=definition.acceptance_basis.id,
            content_sha256=digest(content),
            content=content,
        ),
        operation.artifacts,
    )

def validate_inspection_response(
    response: InspectionResponse,
    definition: ExampleDefinition,
    product: ProductRevision,
    supporting_information: SupportingInformation,
) -> None:
    """Validate dynamic inspection references which a JSON schema cannot express.

    The response schema establishes the shape of each observation. This check
    binds those observations to the particular product, acceptance basis and
    evidence set assigned to the stage, preventing invented references or an
    incomplete inspection from reaching the decision table.
    """

    expected_criteria = tuple(criterion.id for criterion in definition.acceptance_basis.criteria)
    observed_criteria = tuple(observation.criterion_id for observation in response.observations)
    if observed_criteria != expected_criteria:
        raise ResponseContractError(
            "Inspection must assess every acceptance criterion exactly once and in declared order."
        )

    allowed_references = {
        str(product.id),
        str(definition.source_material.id),
        str(definition.evidence_store.id),
        str(definition.acceptance_basis.id),
        str(supporting_information.id),
        *(str(ticket.id) for ticket in definition.source_material.incoming_tickets),
        *(str(record.id) for record in definition.evidence_store.records),
    }
    for observation in response.observations:
        unknown_references = {
            str(reference) for reference in observation.evidence if str(reference) not in allowed_references
        }
        if unknown_references:
            raise ResponseContractError(f"Inspection cited unknown evidence: {sorted(unknown_references)}")

    outcomes = {observation.criterion_id: observation.outcome for observation in response.observations}
    for criterion, observation in zip(
        definition.acceptance_basis.criteria,
        response.observations,
        strict=True,
    ):
        unresolved_dependency = any(
            outcomes[dependency] is not ObservationOutcome.SATISFIED for dependency in criterion.depends_on
        )
        if unresolved_dependency and observation.outcome is not ObservationOutcome.UNRESOLVED:
            raise ResponseContractError(f"{criterion.id} must remain unresolved until its dependencies are satisfied.")


def inspect(
    definition: ExampleDefinition,
    run_directory: Path,
    product: ProductRevision,
    supporting_information: SupportingInformation,
    provider: ProviderName,
    model: str | None,
    operation_id: OperationId,
    inspection_id: InspectionId,
) -> tuple[InspectionAssessment, OperationArtifacts]:
    """Record semantic observations for every criterion without asking the model for a disposition."""

    prompt = operation_prompt(
        "Inspect the supplied product against every acceptance criterion. Record one semantic observation "
        "for each criterion. When a criterion depends on another criterion which is unsatisfied or unresolved, "
        "record the dependent criterion as unresolved rather than guessing. Cite supporting record IDs in the "
        "evidence list. Do not decide whether the product passes, should be repaired, or may be released.",
        InspectionInputs(
            product=product,
            source_material=definition.source_material,
            supporting_information=supporting_information,
            acceptance_basis=definition.acceptance_basis,
        ),
    )

    def validate_payload(response: InspectionResponse) -> None:
        validate_inspection_response(
            response,
            definition,
            product,
            supporting_information,
        )

    operation = run_model_operation(
        run_directory=run_directory,
        operation_id=operation_id,
        name=OperationName.INSPECT,
        provider=provider,
        model=model,
        prompt=prompt,
        response_type=InspectionResponse,
        payload_validator=validate_payload,
    )
    return (
        InspectionAssessment(
            id=inspection_id,
            product_id=product.id,
            acceptance_basis=definition.acceptance_basis.id,
            created_by=operation.artifacts.operation_id,
            observations=operation.payload.observations,
        ),
        operation.artifacts,
    )


def apply_decision_table(
    acceptance_basis: AcceptanceBasis,
    inspection: InspectionAssessment,
    next_defect_id: Callable[[], DefectId],
) -> InspectionDecision:
    """Turn typed observations and declared tolerance into one reproducible route."""

    criteria = {criterion.id: criterion for criterion in acceptance_basis.criteria}
    observed_ids = [observation.criterion_id for observation in inspection.observations]
    expected_ids = set(criteria)
    if len(observed_ids) != len(set(observed_ids)) or set(observed_ids) != expected_ids:
        return InspectionDecision(
            route=Route.STOP,
            disposition=Disposition.INCOMPLETE,
            reason="Inspection did not record exactly one observation for every criterion.",
        )

    defects: list[DefectRecord] = []
    unresolved_blocking = False
    for observation in inspection.observations:
        criterion = criteria[observation.criterion_id]
        if observation.outcome is ObservationOutcome.SATISFIED:
            continue
        if observation.outcome is ObservationOutcome.UNRESOLVED:
            unresolved_blocking = unresolved_blocking or criterion.tolerance is Tolerance.BLOCKING
            continue
        defects.append(
            DefectRecord(
                id=next_defect_id(),
                product_id=inspection.product_id,
                inspection_id=inspection.id,
                criterion_id=observation.criterion_id,
                outcome=ObservationOutcome.UNSATISFIED,
                tolerance=criterion.tolerance,
                repair_group=criterion.repair_group,
                description=observation.explanation,
                evidence=observation.evidence,
            )
        )

    blocking_defects = tuple(defect for defect in defects if defect.tolerance is Tolerance.BLOCKING)
    if blocking_defects:
        # Repair groups encode a declared dependency order. We retain every
        # observed defect, but only assign the earliest blocking group so repair
        # cannot improvise its own scope or work around a prerequisite failure.
        earliest_group = min(defect.repair_group for defect in blocking_defects)
        repair_batch = tuple(defect for defect in blocking_defects if defect.repair_group == earliest_group)
        return InspectionDecision(
            route=Route.REPAIR,
            disposition=Disposition.REPAIR,
            reason="At least one blocking criterion is unsatisfied.",
            defects=tuple(defects),
            repair_batch=repair_batch,
        )
    if unresolved_blocking:
        return InspectionDecision(
            route=Route.STOP,
            disposition=Disposition.INCONCLUSIVE,
            reason="At least one blocking criterion remains unresolved.",
            defects=tuple(defects),
        )
    return InspectionDecision(
        route=Route.RELEASE,
        disposition=Disposition.PASS,
        reason="No blocking criterion is unsatisfied or unresolved.",
        defects=tuple(defects),
    )


def finalize_inspection(inspection: InspectionAssessment, decision: InspectionDecision) -> InspectionRecord:
    """Bind the deterministic decision to the inspection record used by pipeline routing."""

    return InspectionRecord(
        id=inspection.id,
        product_id=inspection.product_id,
        acceptance_basis=inspection.acceptance_basis,
        created_by=inspection.created_by,
        observations=inspection.observations,
        disposition=decision.disposition,
        decision_reason=decision.reason,
        assigned_repair_group=(decision.repair_batch[0].repair_group if decision.repair_batch else None),
        assigned_defects=tuple(defect.id for defect in decision.repair_batch),
    )

def validate_repair_response(response: RepairResponse, defects: tuple[DefectRecord, ...]) -> None:
    """Bind every reported change to the exact defects assigned to this repair."""

    expected_ids = tuple(defect.id for defect in defects)
    reported_ids = tuple(change.defect_id for change in response.changes)
    if reported_ids != expected_ids:
        raise ResponseContractError("Repair must describe every assigned defect exactly once and in order.")


def repair(
    definition: ExampleDefinition,
    run_directory: Path,
    product: ProductRevision,
    defects: tuple[DefectRecord, ...],
    supporting_information: SupportingInformation,
    provider: ProviderName,
    model: str | None,
    operation_id: OperationId,
    repair_id: RepairId,
    output_product_id: ProductRevisionId,
) -> tuple[ProductRevision, RepairRecord, OperationArtifacts]:
    """Repair one declared defect group and return a new, still-unaccepted product revision."""

    if not defects:
        raise ProcessStopped("Repair was selected without an assigned defect batch.")
    prompt = operation_prompt(
        "Repair the supplied product using the assigned defects. Make the smallest complete changes needed to "
        "correct this repair group and keep the report internally consistent. Preserve supported work and do not "
        "independently rewrite findings outside the assigned repair group. Return the complete replacement report "
        "as Markdown in the `product` field, and record one concise summary of the material change made for each "
        "assigned defect in the `changes` field.",
        RepairInputs(
            failed_product=product,
            assigned_defects=defects,
            source_material=definition.source_material,
            supporting_information=supporting_information,
            acceptance_basis=definition.acceptance_basis,
        ),
    )

    def validate_payload(response: RepairResponse) -> None:
        validate_repair_response(response, defects)

    operation = run_model_operation(
        run_directory=run_directory,
        operation_id=operation_id,
        name=OperationName.REPAIR,
        provider=provider,
        model=model,
        prompt=prompt,
        response_type=RepairResponse,
        payload_validator=validate_payload,
    )
    revision = product.revision + 1
    content = operation.payload.product
    repaired_product = ProductRevision(
        id=output_product_id,
        revision=revision,
        created_by=operation.artifacts.operation_id,
        replaces=product.id,
        source_material=definition.source_material.id,
        supporting_information=supporting_information.id,
        acceptance_basis=definition.acceptance_basis.id,
        content_sha256=digest(content),
        content=content,
    )
    repair_record = RepairRecord(
        id=repair_id,
        input_product=product.id,
        defects=tuple(defect.id for defect in defects),
        changes=operation.payload.changes,
        repair_group=defects[0].repair_group,
        output_product=repaired_product.id,
        created_by=operation.artifacts.operation_id,
    )
    return repaired_product, repair_record, operation.artifacts

# Provider adapters own authentication and transport, but never pipeline routing.


def invoke_codex(prompt: str, schema: JsonObject, model: str | None) -> ProviderResponse:
    """Invoke Codex with stored ChatGPT authentication and a strict output schema."""

    command = shutil.which("codex")
    if not command:
        raise ConfigurationError("Codex CLI was not found. Install it and run `codex login` first.")

    with tempfile.TemporaryDirectory(prefix="industrial-ai-codex-") as temporary_directory:
        workdir = Path(temporary_directory)
        schema_path = workdir / "response.schema.json"
        output_path = workdir / "response.json"
        write_json_object(schema_path, schema)
        arguments = [
            command,
            "exec",
            "--ephemeral",
            "--ignore-rules",
            "--ignore-user-config",
            "--sandbox",
            "read-only",
            "--skip-git-repo-check",
            "--color",
            "never",
            "--output-schema",
            str(schema_path),
            "--output-last-message",
            str(output_path),
        ]
        if model:
            arguments.extend(["--model", model])
        arguments.append("-")
        try:
            completed = subprocess.run(
                arguments,
                input=prompt,
                cwd=workdir,
                check=False,
                capture_output=True,
                encoding="utf-8",
                errors="replace",
                timeout=600,
            )
        except OSError as error:
            raise ProviderInvocationError(f"Codex could not be started: {error}") from error
        if completed.returncode != 0:
            raise ProviderInvocationError(f"Codex failed:\n{command_failure_detail(completed)}")
        try:
            raw = output_path.read_text(encoding="utf-8")
        except OSError as error:
            raise ProviderInvocationError("Codex completed without a readable response file.") from error

    return ProviderResponse(
        raw=raw,
        machinery=MachineryRecord(
            provider=ProviderName.CODEX,
            model=model or "CLI default",
            client=command_version(command),
            authentication="stored Codex CLI credentials",
        ),
    )


def invoke_copilot(prompt: str, schema: JsonObject, model: str | None) -> ProviderResponse:
    """Invoke GitHub Copilot with stored OAuth credentials and local validation."""

    command = shutil.which("copilot")
    if not command:
        raise ConfigurationError("GitHub Copilot CLI was not found. Install it and run `copilot login` first.")

    structured_prompt = (
        f"{prompt}\n\nReturn only one JSON object matching this schema. Do not use tools.\n"
        f"{json.dumps(schema, indent=2)}"
    )
    arguments = [
        command,
        "-p",
        structured_prompt,
        "--silent",
        "--no-ask-user",
        "--output-format",
        "text",
    ]
    if model:
        arguments.extend(["--model", model])

    environment = os.environ.copy()
    environment.pop("COPILOT_ALLOW_ALL", None)
    environment["COPILOT_AUTO_UPDATE"] = "false"
    environment["GITHUB_COPILOT_PROMPT_MODE_EXTENSIONS"] = "false"
    environment["GITHUB_COPILOT_PROMPT_MODE_REPO_HOOKS"] = "false"
    environment["GITHUB_COPILOT_PROMPT_MODE_WORKSPACE_MCP"] = "false"
    with tempfile.TemporaryDirectory(prefix="industrial-ai-copilot-") as temporary_directory:
        try:
            completed = subprocess.run(
                arguments,
                cwd=temporary_directory,
                check=False,
                capture_output=True,
                encoding="utf-8",
                errors="replace",
                timeout=600,
                env=environment,
            )
        except OSError as error:
            raise ProviderInvocationError(f"GitHub Copilot could not be started: {error}") from error
    if completed.returncode != 0:
        raise ProviderInvocationError(f"GitHub Copilot failed:\n{command_failure_detail(completed)}")

    return ProviderResponse(
        raw=completed.stdout,
        machinery=MachineryRecord(
            provider=ProviderName.COPILOT,
            model=model or "CLI default",
            client=command_version(command),
            authentication="stored GitHub Copilot CLI credentials",
        ),
    )


def invoke_openai(prompt: str, schema: JsonObject, model: str | None) -> ProviderResponse:
    """Invoke the OpenAI Responses API for automated environments."""

    requested_model = model or os.environ.get("OPENAI_MODEL")
    if not requested_model:
        raise ConfigurationError("The OpenAI API provider requires `--model` or OPENAI_MODEL.")
    if not os.environ.get("OPENAI_API_KEY"):
        raise ConfigurationError("The OpenAI API provider requires OPENAI_API_KEY.")

    from openai import OpenAI, OpenAIError

    text_config = cast(
        "ResponseTextConfigParam",
        {
            "format": {
                "type": "json_schema",
                "name": "industrial_ai_operation",
                "strict": True,
                "schema": schema,
            }
        },
    )
    try:
        response = OpenAI().responses.create(
            model=requested_model,
            input=prompt,
            text=text_config,
            store=False,
        )
    except OpenAIError as error:
        raise ProviderInvocationError(f"OpenAI API request failed: {error}") from error
    usage = JSON_OBJECT_ADAPTER.validate_python(response.usage.model_dump(mode="json")) if response.usage else None
    return ProviderResponse(
        raw=response.output_text,
        machinery=MachineryRecord(
            provider=ProviderName.OPENAI,
            model=response.model,
            client="OpenAI Python SDK",
            authentication="OPENAI_API_KEY",
            response_id=response.id,
            usage=usage,
        ),
    )


@cache
def replay_responses() -> dict[str, ProviderResponse]:
    """Load retained responses indexed by the prompt which originally produced them."""

    responses: dict[str, ProviderResponse] = {}
    for filename in ("replay_responses.json", "method_check_responses.json"):
        replay_path = Path(__file__).parent / "data" / filename
        try:
            replay_data = JSON_OBJECT_ADAPTER.validate_json(replay_path.read_text(encoding="utf-8"))
        except (OSError, ValidationError) as error:
            raise ConfigurationError(f"Unable to load retained replay responses: {error}") from error

        raw_responses = replay_data.get("responses_by_prompt_sha256")
        if not isinstance(raw_responses, dict):
            raise ConfigurationError("Retained replay responses are missing their prompt index.")
        source_model = replay_data.get("model", "retained Codex response")
        if not isinstance(source_model, str) or not source_model.strip():
            raise ConfigurationError("Retained replay responses have an invalid model identity.")

        for prompt_sha256, response in raw_responses.items():
            if not isinstance(response, dict):
                raise ConfigurationError("Retained replay response index is invalid.")
            retained = ProviderResponse(
                raw=json.dumps(response, indent=2, sort_keys=True),
                machinery=MachineryRecord(
                    provider=ProviderName.REPLAY,
                    model=source_model,
                    client="industrialized-ai-example replay",
                    authentication="not required",
                ),
            )
            if prompt_sha256 in responses and responses[prompt_sha256] != retained:
                raise ConfigurationError("Retained replay responses conflict for the same prompt.")
            responses[prompt_sha256] = retained
    return responses


def invoke_replay(prompt: str, schema: JsonObject, model: str | None) -> ProviderResponse:
    """Return the retained response for exactly the bounded prompt being replayed."""

    del schema, model
    prompt_sha256 = str(digest(prompt))
    response = replay_responses().get(prompt_sha256)
    if response is None:
        raise ProviderInvocationError(
            "The retained run has no response for this operation input. "
            "Replay only supports the packaged example without modification."
        )
    return response


PROVIDER_ADAPTERS: Mapping[ProviderName, ProviderAdapter] = MappingProxyType(
    {
        ProviderName.REPLAY: invoke_replay,
        ProviderName.CODEX: invoke_codex,
        ProviderName.COPILOT: invoke_copilot,
        ProviderName.OPENAI: invoke_openai,
    }
)


def invoke_provider(
    provider: ProviderName,
    prompt: str,
    schema: JsonObject,
    model: str | None,
) -> ProviderResponse:
    """Invoke a provider adapter without exposing provider choices to the pipeline."""

    return PROVIDER_ADAPTERS[provider](prompt, schema, model)