A runnable example.

Industrialized AI puts AI inside a deterministic production process: manufacture a product, inspect it against an acceptance basis, repair any defects, and inspect the replacement. This example turns that process into a small Python program you can run and take apart.

The program demonstrates the core manufacture, inspection, repair, and release loop rather than every capability of Industrialized AI. Composition, factories, and production-order planning build on the same process, but are outside this example.

The input is a deliberately unreliable set of security tickets. The first reviewed queue gets several important things wrong, and the pipeline needs four repairs before it can release revision 5.

Run it and see where the first report fails.

The replay uses responses retained from a Codex run, then puts them through the complete Python pipeline again. Inspection coverage, defect creation, repair routing, and release are all performed during the replay, and a fresh set of records is written to disk. Download the source archive, inspect it, then run it locally; you don't need an account or API key.

Download the source and tests

tar -xzf industrialized_ai_example-1.1.0.tar.gz
cd industrialized_ai_example-1.1.0
uv run --locked industrialized-ai --replay

What happened during the retained Codex run

reviewed-queue-r001 inspection-001 Defect: Attribution
The report confused staging with production and treated an internal scan as evidence of production exposure.
reviewed-queue-r002 inspection-002 Defect: Applicability
The report left component and feature applicability unresolved even though the supplied records established both.
reviewed-queue-r003 inspection-003 Defect: Severity
The report left local severity unresolved after applicability and exposure had been established.
reviewed-queue-r004 inspection-004 Defect: Priority
The report placed a medium staging finding ahead of a high production finding with demonstrated public exposure.
reviewed-queue-r005 inspection-005 Pass and release
No blocking criterion remained unsatisfied or unresolved.

Each failed revision records the defect inspection found, repair records what it changed, and the following inspection checks the complete replacement. The terminal summary shows that history before naming the report which was released.

Read the control loop

The pipeline decides what happens next.

AI handles the work which depends on meaning inside manufacture, inspection, and repair. Python supplies the recorded inputs, validates the responses, gives every revision and operation an identity, and applies the decision table to the observations returned by inspection. The model never chooses which station runs next.

A repair therefore leads back to inspection because the pipeline says it does, while a release can only name the current product revision and the inspection which passed it. That control loop is the center of the example.

The control flow implemented by the Python file

Inputs Recorded material and acceptance basis
Semantic station Manufacture a product revision
Deterministic station Gather supporting records
Semantic station Inspect every criterion
Python Apply the decision table

Pass Release this product revision with its evidence

Repair required Assign the next repair group, create a replacement, then return to inspection

def run_process(
    definition: ExampleDefinition,
    run_directory: Path,
    provider: ProviderName,
    model: str | None,
    max_revisions: int,
) -> ReleaseRecord:
    """Run manufacture, inspection and bounded repair until release or a controlled stop."""

    write_record(run_directory / "source-material.json", definition.source_material)
    write_record(run_directory / "evidence-store.json", definition.evidence_store)
    write_record(run_directory / "product-assignment.json", definition.product_assignment)
    write_record(run_directory / "acceptance-basis.json", definition.acceptance_basis)

    evidence: list[RecordPath] = [
        RecordPath("source-material.json"),
        RecordPath("evidence-store.json"),
        RecordPath("product-assignment.json"),
        RecordPath("acceptance-basis.json"),
    ]
    identities = RunIdentities()
    initial_product_id = identities.product_revision(1)

    report_manufacture_started(initial_product_id)
    product, artifacts = manufacture(
        definition=definition,
        run_directory=run_directory,
        provider=provider,
        model=model,
        operation_id=identities.next_operation(OperationName.MANUFACTURE),
        product_id=initial_product_id,
    )
    evidence.extend(artifacts.evidence_files)
    evidence.append(record_product(run_directory, product))

    supporting_information = gather_supporting_information(
        definition,
        identities.next_supporting_information(),
    )
    supporting_path = run_directory / "supporting-information.json"
    write_record(supporting_path, supporting_information)
    evidence.append(relative(supporting_path, run_directory))
    report_manufacture_completed(product, len(definition.evidence_store.records))

    while True:
        report_inspection_started(product, len(definition.acceptance_basis.criteria))
        inspection, artifacts = inspect(
            definition=definition,
            run_directory=run_directory,
            product=product,
            supporting_information=supporting_information,
            provider=provider,
            model=model,
            operation_id=identities.next_operation(OperationName.INSPECT),
            inspection_id=identities.next_inspection(),
        )
        decision = apply_decision_table(
            definition.acceptance_basis,
            inspection,
            identities.next_defect,
        )
        inspection = finalize_inspection(inspection, decision)
        report_inspection_completed(decision)
        evidence.extend(artifacts.evidence_files)
        evidence.append(record_inspection(run_directory, inspection))

        if decision.defects:
            evidence.append(record_defects(run_directory, product, decision))

        # The route is derived solely from the current inspection. Models never
        # select the next station and an earlier inspection can never be reused.
        match decision.route:
            case Route.RELEASE:
                release = release_product(
                    run_directory=run_directory,
                    definition=definition,
                    supporting_information=supporting_information,
                    product=product,
                    inspection=inspection,
                    release_id=identities.next_release(),
                    evidence=evidence,
                )
                write_record(run_directory / "release.json", release)
                return release

            case Route.STOP:
                raise ProcessStopped(decision.reason)

            case Route.REPAIR:
                if product.revision >= max_revisions:
                    raise ProcessStopped(
                        f"The product still requires repair after {max_revisions} revisions; nothing was released."
                    )
                if not decision.repair_batch:
                    raise ProcessStopped("The decision table selected repair without assigning a defect.")

                report_repair_started(product, decision.repair_batch)
                product, repair_record, artifacts = repair(
                    definition=definition,
                    run_directory=run_directory,
                    product=product,
                    defects=decision.repair_batch,
                    supporting_information=supporting_information,
                    provider=provider,
                    model=model,
                    operation_id=identities.next_operation(OperationName.REPAIR),
                    repair_id=identities.next_repair(),
                    output_product_id=identities.product_revision(product.revision + 1),
                )
                repair_path = run_directory / "repairs" / f"{repair_record.id}.json"
                write_record(repair_path, repair_record)
                evidence.extend(artifacts.evidence_files)
                evidence.append(relative(repair_path, run_directory))
                evidence.append(record_product(run_directory, product))
                report_repair_completed(product, len(definition.acceptance_basis.criteria))

            case _ as unreachable:
                assert_never(unreachable)

The input file defines the job before a model is called.

The example starts with three tickets describing two security findings, together with the records needed to check them. Those records establish the actual services and environments, installed versions, enabled features, scanner positions, and network routes. The same file also defines the reviewed queue we want back and the acceptance criteria it must satisfy.

Keeping that information outside the program makes the production order visible and replaceable. The program validates the complete definition before manufacture begins, while the typed contracts below also govern model responses and the records retained during the run.

{
  "source_material": {
    "id": "ticket-set-001",
    "incoming_tickets": [
      {
        "id": "SEC-1841",
        "title": "Critical internet-facing RCE in payments gateway",
        "claimed_severity": "critical",
        "claims": [
          "Gateway Runtime 4.2 is installed in production.",
          "The legacy request parser is enabled on port 8443.",
          "An external scanner reached the production service.",
          "Planned work should stop until the gateway is patched."
        ],
        "reported_asset": "payments-gateway.example.com",
        "scanner_event": "scan-771"
      },
      {
        "id": "SEC-1842",
        "title": "Possible information disclosure in customer export service",
        "claimed_severity": "medium",
        "claims": [
          "Customer Export 2.8 is installed in production.",
          "A debug export route may be enabled.",
          "The route is believed to be internal and requires authentication.",
          "The finding can wait for the normal maintenance window."
        ],
        "reported_asset": "customer-export.example.com",
        "scanner_event": "scan-804"
      },
      {
        "id": "SEC-1843",
        "title": "High-severity legacy parser exposure on gateway-stage-03",
        "claimed_severity": "high",
        "claims": [
          "Gateway Runtime 4.2 is installed on gateway-stage-03.",
          "The legacy parser is enabled and reachable from the corporate network.",
          "The asset may be another name for the production payments gateway."
        ],
        "reported_asset": "gateway-stage-03",
        "scanner_event": "scan-771"
      }
    ]
  },
  "evidence_store": {
    "id": "evidence-store-001",
    "records": [
      {
        "id": "service-catalog",
        "text": "payments-gateway.example.com identifies the production payments gateway. gateway-stage-03 is a separate staging service."
      },
      {
        "id": "gateway-inventory",
        "text": "Gateway Runtime 4.2 is installed in both production and staging."
      },
      {
        "id": "gateway-configuration",
        "text": "The legacy parser is disabled in production and enabled on port 8443 in staging."
      },
      {
        "id": "scan-771-record",
        "text": "scan-771 originated on the corporate network and reached gateway-stage-03. It did not target the production payments gateway."
      },
      {
        "id": "gateway-firewall",
        "text": "The recorded firewall policy denies internet traffic to port 8443 for both gateway services."
      },
      {
        "id": "gateway-advisory",
        "text": "Gateway Runtime 4.2 is affected only when the legacy parser is enabled and reachable."
      },
      {
        "id": "export-inventory",
        "text": "Customer Export 2.8 is installed in production and the debug export route is enabled."
      },
      {
        "id": "export-edge-route",
        "text": "The production edge routes public requests for /exports/debug to Customer Export."
      },
      {
        "id": "scan-804-record",
        "text": "scan-804 reached /exports/debug from outside the corporate network and downloaded a test export without authentication."
      },
      {
        "id": "export-advisory",
        "text": "Customer Export 2.8 permits unauthenticated export download when the debug route is enabled."
      }
    ],
    "ticket_index": {
      "SEC-1841": [
        "service-catalog",
        "gateway-inventory",
        "gateway-configuration",
        "scan-771-record",
        "gateway-firewall",
        "gateway-advisory"
      ],
      "SEC-1842": [
        "export-inventory",
        "export-edge-route",
        "scan-804-record",
        "export-advisory"
      ],
      "SEC-1843": [
        "service-catalog",
        "gateway-inventory",
        "gateway-configuration",
        "scan-771-record",
        "gateway-firewall",
        "gateway-advisory"
      ]
    }
  },
  "product_assignment": {
    "product": "Reviewed and prioritized security queue",
    "intended_use": "Help the team decide which reported vulnerabilities require attention first.",
    "instructions": "Consolidate duplicates, distinguish each service and environment, assess the reported findings, recommend proportionate action, and put the resulting work in priority order. Retain the source ticket IDs and state anything the available information cannot establish."
  },
  "acceptance_basis": {
    "id": "acceptance-basis-001",
    "criteria": [
      {
        "id": "attribution",
        "requirement": "Every conclusion is attributed to the correct service, environment and source ticket.",
        "tolerance": "blocking",
        "repair_group": 1
      },
      {
        "id": "duplicates",
        "requirement": "Tickets describing the same event are consolidated without losing their source identities.",
        "tolerance": "blocking",
        "depends_on": ["attribution"],
        "repair_group": 1
      },
      {
        "id": "applicability",
        "requirement": "Each finding establishes whether the affected component and vulnerable feature are present.",
        "tolerance": "blocking",
        "depends_on": ["attribution"],
        "repair_group": 2
      },
      {
        "id": "exposure",
        "requirement": "Internal and external reachability claims follow from evidence for the correct environment.",
        "tolerance": "blocking",
        "depends_on": ["attribution", "applicability"],
        "repair_group": 2
      },
      {
        "id": "severity",
        "requirement": "Local severity follows from established applicability and exposure rather than source labels.",
        "tolerance": "blocking",
        "depends_on": ["applicability", "exposure"],
        "repair_group": 3
      },
      {
        "id": "action",
        "requirement": "Recommended action is proportionate to the established local severity.",
        "tolerance": "blocking",
        "depends_on": ["severity"],
        "repair_group": 3
      },
      {
        "id": "priority",
        "requirement": "The final order puts the most important supported work first and does not count duplicates twice.",
        "tolerance": "blocking",
        "depends_on": ["duplicates", "severity"],
        "repair_group": 4
      },
      {
        "id": "limits",
        "requirement": "Missing evidence which could materially change a conclusion remains visible.",
        "tolerance": "blocking",
        "repair_group": 4
      }
    ]
  }
}

Input, response, and retained-record contracts

records.py · lines 45 to 391

# 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 station 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


# Station 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

What manufacture receives and records.

Manufacture receives the tickets, the product assignment, and the acceptance basis, then asks the selected model for the first reviewed queue. A valid response becomes reviewed-queue-r001, with its exact content, digest, source material, and producing operation recorded together. Manufacture has produced a product; it hasn't established that the product is acceptable.

Before inspection, a deterministic station follows the ticket index and gathers the supporting records declared for each ticket. The model isn't allowed to choose a convenient evidence set or search outside the run.

Manufacture

stations.py · lines 72 to 116

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 gather_supporting_information(
    definition: ExampleDefinition,
    supporting_information_id: SupportingInformationId,
) -> SupportingInformation:
    """Resolve the declared ticket-to-evidence index without asking a model to choose evidence."""

    records = {record.id: record for record in definition.evidence_store.records}
    ticket_evidence = {
        ticket_id: tuple(records[record_id] for record_id in record_ids)
        for ticket_id, record_ids in definition.evidence_store.ticket_index.items()
    }
    return SupportingInformation(
        id=supporting_information_id,
        source=definition.evidence_store.id,
        ticket_evidence=ticket_evidence,
    )

How inspection becomes a deterministic decision.

Inspection receives a specific product revision, the information used to judge it, and the acceptance basis. The model records one observation for every criterion, with an outcome, explanation, and supporting evidence; it isn't asked whether the product should pass.

Python checks that every criterion was assessed once, that the evidence references exist, and that dependent criteria remain unresolved when their foundations haven't been established. The decision table then applies the declared tolerances. A blocking unsatisfied requirement becomes a defect and sends the revision to repair; a blocking unresolved requirement stops the run if there is no blocking defect to repair; complete satisfaction permits release.

These checks ensure that every criterion has a recorded assessment and that the acceptance rules are followed. The reference cases later on this page test the accuracy of those assessments by comparing them with expected results established from the source evidence.

Inspection and the decision table

stations.py · lines 121 to 310

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 station, 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),
    )

Repair has to explain what it changed.

Repair receives the failed product revision, the exact defects assigned to this attempt, the original source material, and the supporting records. It returns a complete replacement and a concise account of the change made for each assigned defect. The response contract rejects a repair which omits a defect or claims to have repaired something it wasn't given.

That account tells us what repair attempted; the next inspection determines whether it worked. The replacement gets a new product revision and goes through the complete acceptance basis, which is why later defects in this example become visible instead of being hidden behind an earlier correction.

Repair

stations.py · lines 315 to 392

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

What the run leaves behind.

The program writes each product revision, inspection, defect, and repair while the run is happening. When an inspection passes, the release record names the exact product made available for use and links it to the work which led there. If no revision passes, there is no release.

Records written by the example
Record	What it contains
`source-material.json`	The incoming tickets used to manufacture the queue.
`supporting-information.json`	The records supplied to inspection and repair.
`product-assignment.json`	The product to make and its intended use.
`acceptance-basis.json`	The criteria, dependencies, and tolerances used by inspection.
`operations/`	Every bounded prompt, raw response, and machinery record.
`products/`	The manufactured product and every repaired revision.
`inspections/`	The observations and deterministic disposition for each revision.
`defects/` and `repairs/`	What failed, which defects were assigned, and what repair changed.
`release.json`	The released revision, passing inspection, and supporting evidence.

The example writes everything to the local filesystem so the complete run is easy to follow. A larger implementation can replace the storage and execution machinery without changing these handoffs.

Download the retained run View the run summary

Check whether a change improves the process.

Reference cases let us compare a method’s output with results established from the source evidence. The example includes correct reports, versions with known mistakes, and reports which correctly state that evidence is missing. The expected answers are kept separate from the inputs given to the model.

What the reference cases check
Stage	What the reference cases check
Manufacture	Whether the report accurately represents the information supplied to manufacture, retains the required facts, and acknowledges what it cannot establish.
Inspection	Whether each assessment matches the expected outcome, including detecting known defects and accepting correct conclusions.
Repair	Whether the replacement corrects the assigned defect while preserving conclusions that were already sound.

Inspection returns structured assessments, so the program can compare them with the expected answers and report missed defects, incorrect rejections, and unexpected unresolved assessments. Manufacture and repair produce prose, so their results appear alongside a short answer sheet for a person to review against the expected facts.

Replay the saved responses to see how the checks work:

uv run industrialized-ai-check --replay

To test a change, run the cases with a live model, save the results, then change the instructions or model used by a stage and run the same cases again. Comparing the results shows whether the change corrected the problem and whether it affected work that was previously right.¹

uv run industrialized-ai-check

The saved records identify the inputs, instructions, and model used for each attempt, so the comparison can be examined later. These few cases demonstrate how to assess and improve the methods; establishing reliability for production would require broader testing and continued monitoring.

Run the same pipeline with a live model.

Codex is the default live provider, while GitHub Copilot and the OpenAI API use the same operation contracts through separate adapters. Provider authentication and transport end at that boundary; manufacture, inspection, repair, and routing remain unchanged.

Live model options
Provider	Preparation	Command
Codex	`codex login`	`uv run --locked industrialized-ai`
GitHub Copilot	`copilot login`	`uv run --locked industrialized-ai --provider copilot`
OpenAI API	Set `OPENAI_API_KEY` and `OPENAI_MODEL`	`uv run --locked industrialized-ai --provider openai`

Provider adapters

providers.py · lines 99 to 337

# 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)

The source archive contains the package, example data, dependency lock, and tests. Start with replay, inspect the records it creates, then change the input or provider once the control loop makes sense.