Ordering Platform Design

services/ai_ordering is a Python Temporal worker that drives candidate selection through a recommendation → submit → confirm loop, one candidate per iteration, per OEM. This page describes the target architecture agreed in the design review. The code converges on it phase by phase — see Decisions for what is settled. Until a phase ships, the current code may differ; services/ai_ordering/CLAUDE.md marks which rules are enforced today versus decided-but-not-shipped.

Terminology

• OEM — the manufacturer we order from (Stellantis, Mercedes); the unit of ordering: one profile, one service class, one private API login per (oem, market).
• Brand (make) — a consumer marque an OEM owns (Stellantis: Jeep, RAM, …; BMW: BMW, MINI). Many brands per OEM; brands appear inside run scoping and product data, never as a dispatch key.
• Private dealer API — the authenticated, undocumented HTTP API of a dealer application we automate (DealerConnect, NetSTAR); the transport the injected client (packages/ai_dealer_connect, packages/ai_netstar) speaks. Sessions, logins, and auth failures are private API concepts.
• OMS — the order management system behind the private API: the logic that validates orders, adds defaults, refuses what it will not build, and accepts the rest. Submission outcomes are OMS concepts. Cure is our name for handling that response; the OMS does not use the word, so CureResult is an ai_ordering concept.

Architecture

One OEM-agnostic workflow drives per-OEM services (Strategy pattern) through a single generic activity layer. An OEM is fully defined by an OrderingOEMProfile (job config) plus its OrderingService implementation, both bound at the worker's composition root.

The ordering service interface and the factories that resolve an OEM's service live in services/ordering.py; types/ holds everything else that crosses the boundary:

Module	Contents
services/ordering.py	The OrderingService interface; OrderingServiceFactory with live and simulated implementations
types/collection.py	CollectionJob — the OEM-declared collection jobs an ordering run triggers
types/snapshots.py	SnapshotRef — key of one stored snapshot table; ResolvedSnapshot pairs one with its session timestamp
types/curing.py	CureResult — validation/curing outcome: the cured build normalized, whether it is submittable, and the target-scoped signals the pipeline records
types/target_scopes.py	Mode contracts: volume fill, order modification
types/run_datasets.py	RunDatasets — the bucket and keys holding the warehouse tables a run reads once for its dealer. Names and keys only, so no row structure crosses the boundary
types/errors.py	Canonical taxonomy: SessionConflict, AuthExpired, OrderingNotPermitted, ConfigurationError, SnapshotUnavailable

Activity names are fixed; the task queue selects the run kind — one live queue and one simulated queue, each serving every OEM, so the number of workers stays constant as OEMs are added. A run names its OEM in its input and the activity layer dispatches on (oem, market).

OEM is a runtime lookup; the simulated boundary is not. The simulated queue registers no activity that can reach a private API, and the services composed onto it hold a mock private API client. So a simulated run cannot touch one, whatever its input says.

Operating modes

Two modes cover every current OEM; a new OEM declares which it is. "Volume" and "modification" for short:

Mode	Pattern	OEMs
Order modification	Pick modifiable orders from inventory → one target per slot → modify	Mercedes
Volume fill	Volume optimization over projected supply → one scope per product carrying a volume count → place	Stellantis; BMW (dry-run only)

Run kinds

Three kinds run the same logic almost end to end. Two workflows implement them, one per queue: OrderingRunWorkflow covers both live kinds, SimulatedOrderingRunWorkflow the simulated one. They diverge in exactly two places, each converging immediately after:

Step	Simulated	Live dry run	Live submission
Collection jobs	Resolved as of the run's as_of_date; never triggered	Triggered when past refresh_after	Triggered when past refresh_after
Validate and cure against the private API	Skipped — the recommendation stands as selected	Executed	Executed
Submit	Skipped	Skipped	Executed

A simulated run reads whatever snapshot was current at its as_of_date, which makes a backtest a date-parameterized replay rather than a second code path.

Scopes and targets

A scope is the product × dealer space one selection pipeline initializes over. A target is one thing ordered: an existing order to modify, or one from-scratch order to place. construct_targets emits one mode-tagged OrderingScope per pipeline space, carrying its production windows as inclusive date intervals — types/target_scopes.py is the contract.

initialize_scope_pipeline runs once per scope and the workflow drops the pipeline at scope exit, so a run's in-memory DataFrames belong to one scope however many products it orders (ADR-004). Targets within a scope share that pipeline.

The cost of that pattern is per-scope reads. A volume-fill run with a cap of one unit gives every model its own scope, so a brand's worth of models is that many initializations. A table partitioned only by date costs the same on each one, because a scope's predicate narrows the rows returned rather than the bytes scanned. candidate_selection names the tables where that is worth paying once per run in RUN_LEVEL_TABLES, and the ordering service supplies them to each scope as run_datasets rather than letting the assembler read them again. See the context layer.

Run flow

Every box in the phases below is one or more activities (or child workflows) invoked by OrderingWorkflow. The workflow owns the sequence, the per-target loop, signals, and continue-as-new, and holds run state only as opaque activity output:

Each step's logic has one code home. Where it executes is the worker whose queue the activity arrives on, so selection steps run on the ordering worker as generic activities delegating into the library:

Step	Logic lives in
Run sequence, per-target loop, signals, continue-as-new	ai_ordering/workflows (Temporal sandbox)
Collect OEM data	OEM-declared collection jobs (profile config), executed by collection workers
Freshness gate, pull inventory candidates, emit/persist	Generic activities: ai_ordering/activities, services/, persistence/
Construct targets, simulate/validate/order, report data	OrderingService impls in ai_ordering/oem/
Volume optimization (inside construct targets, volume mode)	packages/ai_ppm
Pipeline init, recommendations, supply-state confirm	packages/candidate_selection — ordering only triggers
Which tables a run reads once vs per scope	RUN_LEVEL_TABLES in packages/candidate_selection; ordering decides when (once per run) because it owns the scope loop
Private API transport + parsers (collection and submit)	packages/ai_dealer_connect (DealerConnect), packages/ai_netstar (NetSTAR)
Report build & delivery	ai_ordering/reports, services/ (SendGrid, Slack)

The full logic those phases run, mode branching included:

Loop mechanics:

• Pipelines are bounded at one scope. A scope's SelectionPipeline exists only while its scope loop runs — memory stays at ~one scope regardless of how many models a dealer orders. The pipelines ride in activity output through external storage. Initialization happens at scope entry — the workflow knows its next scope before any recommendation is asked for.
• A rejection extends the current target's restrictions, and the loop retries with the next candidate. A placement updates the pipeline's supply state, and supply take rates are derived from that state at the next scoring pass.
• A scope can end before its targets do. A submit that finds the scope's supply gone fails with SupplyExhausted; the workflow records that attempt, makes no attempt on the scope's remaining targets, and continues with the next scope. Those targets count as skipped, and the exit reason is unchanged (ADR-015).
• The loop is legible from the event history alone. Every activity in one try carries the activity id t<target_index>-a<attempt>-<step>, so which slot a row belongs to and whether it is a retry read off the timeline without decoding a payload. A refused build fails validate_and_cure — non-retryable, with the CureResult in the failure's details, which the workflow reads back. So the candidates the OMS turned down are the failing rows, and a target that needed four tries looks like one.
• Cluster-shared supply state is deferred. Current supply formulas give nearby-dealer supply negligible weight, so a run's loop reads only its own confirms. The per-iteration shared-state refresh and (dealer, candidate) write-back activate when that weighting requires realtime granularity (see ADR-012).

How a run ends

A run that reaches the end of its targets returns an OrderingResult, and its exit_reason says which completed outcome it reached: QUOTA_EXHAUSTED, NO_CANDIDATES, DRY_RUN_COMPLETE or CANCELLED.

Every other ending fails the workflow, so a stopped run is red in Temporal rather than green with the failure recorded only in Mongo. Before failing, the run persists its state, notifies Slack, and closes its event stream, so no ending is silent. It ships no report: a run that stopped early has no complete allocation to report on, and the attempts it did place are already persisted. Two shapes:

Ending	Failure the workflow raises
A reason the run itself decided — expired session, stale collection session, no landed partition, a pause nobody lifted	Non-retryable ApplicationError typed by the exit_reason, carrying the OrderingResult in its details
Any other activity failure	The activity's own error, unchanged

A caller therefore reads a stopped run's outcome from the failure rather than from a return value.

Data flow

An ordering run never pulls a private API for data. It resolves one collection_id per freshness-checked table of each declared collection job (CollectionJob.tables). Sessions are identified per table, so each freshness-checked table resolves independently; a job may write more tables than it freshness-checks. It then reads exactly that snapshot, freshness-checked against the job's two staleness quantities. Data younger than refresh_after is reused as-is; past it, a live run triggers the job as a child workflow. Data older than max_staleness is meaningless to order against and blocks the run — the only check a simulated run applies, since it never triggers and always resolves to the most recent completed session. Collection workers execute the jobs; snapshots are written to BigQuery keyed (dealer, table, date, collection_id). The private API is touched again only at validate/submit. The run's warehouse reference data is freshness-checked based on presence rather than age: the probe resolves the newest date, at or before the run's as_of_date and within partition_lookback_days, where every table the run reads has completed. A brand's ingestion writes its partition hours into the UTC day, so a run starting before that reads the day before rather than terminating. Nothing in the window terminates the run — live and simulated alike. A date is taken whole or not at all: reading one table on today and another on yesterday would join rows the warehouse never held together. The resolved date replaces the run's as_of_date before any activity that reads data is dispatched, so one date covers the partition, the production window and the collection reference time of a simulated run. The dependency list is derived, not restated, from the IcebergTableMap the selection config declares plus the service's own volume_optimization_namespaces, so a table added to a read is covered without a second list to keep in step.

Those warehouse reads happen at two scopes. Most tables are read per scope, where the partition is worth scanning for one model's rows. The tables in candidate_selection.RUN_LEVEL_TABLES are not: their partitions carry every dealer and model of the market and prune no further than the day. So a scope's predicate narrows the rows returned and not the bytes scanned — the read costs the same whether it asks for one model or all of them. load_run_dealer_datasets reads those once for the dealer at the start of the ordering loop and the workflow hands the result to every scope, which narrows them in memory. A run therefore pays that scan once rather than once per model it orders.

RunDatasets crosses as bucket and keys. The tables themselves stay in the cache bucket, and a scope fetches the whole-table payloads plus its own model's slices. It carries no row model, so SnapshotRef remains the only row reference between activities.

The comparison itself is CollectionJob.assess_freshness — a pure, sandbox-safe verdict the workflow computes inline against its reference time (its own clock for a live run, the input simulation moment for a backtest). So resolution stays one activity returning data, and the trigger/terminate branching is workflow control flow:

Verdict	Age	Live run	Simulated run
fresh	≤ refresh_after	reuse	reuse
stale	≤ max_staleness	trigger the job	reuse
expired	> max_staleness	trigger the job	terminate
missing	no completed session	trigger the job	terminate

A dealer with no completed session resolves to None, not an error — absence is a freshness input. The store's errors are reserved for genuine faults: ConfigurationError for a logical table the source does not configure, SnapshotUnavailable for a ref naming a session its table never recorded.

While the TypeScript collectors write the snapshots, a session's identity is its fetch_date — one timestamp stamped on every row a session writes to a freshness-checked table. SnapshotStore.resolve mints collection_id from that timestamp, and reads translate it back into a fetch_date filter. Once collectors are migrated to Python, the triggering workflow mints the child's workflow id as the collection_id and stamps it at write instead. A table without a per-session stamp cannot be freshness-checked — feature_catalog is stamped per combo, sales_* per row. Stellantis freshness-checks only inventory: allocations are collected for analysis but are not an ordering upstream — the volume optimizer runs without an allocation ceiling. SnapshotRefs are the only snapshot reference in activity inputs and outputs — rows are read inside activities and never cross a boundary. The Temporal converter deserializes payload fields to their annotated base type, so subclass-only fields would silently drop.

Reads go through services/snapshots.py — a single generic SnapshotStore (resolve plus a typed read_rows(ref, row_model)). Row structures are OEM-internal: every row consumer lives inside an OEM service (orders ride the inventory pull, and target construction assesses which are modifiable). So each OEM defines its own row models beside its service and binds them at the callsite. Nothing standardizes row vocabulary, because rows never cross the interface. The ref is the only cross-OEM snapshot contract.

Run state

State	Where it lives
Pulled snapshots (inventory — orders included)	BigQuery (dealer, table, date, collection_id); read via SnapshotStore
Warehouse reference data (models, dealers, stats)	Iceberg/Nessie via services/warehouse.py, read at the run's partition
OEM payload between a service's own activities	Temporal External Storage — typed activity args/returns; over-threshold payloads offload to S3
Live SelectionPipeline	Passed between the selection activities, offloaded to Temporal External Storage like the OEM payload. The workflow shuttles it as an opaque RawValue and never decodes one — unpickling imports polars, which the workflow sandbox forbids
Cluster-shared supply state (future)	candidate_selection library — the durable extension of the supply state confirm_order already mutates
Runs, attempts, events, restriction evidence, volume decisions, report builds	MongoDB; events sync to Iceberg

There is no general-purpose "stash anything by run_id" store: each kind of state has a home in the table above.

Target layout

services/ai_ordering/src/ai_ordering/
├── worker.py                  # composition root: profile → services → activities → queue
├── profiles.py                # OrderingOEMProfile per OEM (job config only)
├── queues.py                  # the live and simulated ordering queues
├── types/                     # the entire boundary: everything that crosses it
├── activities/                # generic Temporal layer — only temporalio importer
├── workflows/                 # sandbox loop internals
├── oem/                       # per-OEM OrderingService implementations + their private API clients
├── services/                  # storage + integrations: warehouse, snapshots, sendgrid, …
├── persistence/               # Mongo: runs, attempts, events, oem_restrictions, volume, reports
└── state/                     # what a run must hold in memory, and nothing else

Restrictions

Two kinds, one principle — a live restriction is evidence, not fact:

• Static — known ahead of the run (oem_restrictions, operator-transcribed today; target: fed by ingestion).
• Discovered — found when a submit is rejected; slot-scoped today. Mercedes restrictions are date- and threshold-dependent, so moving into shared scope needs a per-OEM policy (scope / TTL / confidence) — moving them too aggressively suppresses the attempts that would prove or disprove the restriction.

Deployment and scaling

One codebase, one image, one ECS service definition — instantiated per (OEM, run kind) with config selecting queue and OEM. Ordering services scale 0↔1 with the run trigger (the trigger already knows work is coming). Collection workers scale on task-queue backlog. ResourceBasedSlotSupplier caps per-worker concurrency by memory/CPU. desired_count > 1 is conditioned on durable run state. Scaling can never be finer than routing: per-OEM scaling requires per-OEM queues, which is why they stay.