Firm Supply Orders — Developer Reference¶

This document covers the internals of the firm supply orders feature: code paths, data structures, contracts between layers, and gotchas. For the planner-facing documentation see User Guide → Firm Supply Orders.

1. Why this exists¶

Before this feature, the only way to mark a supply order as "already committed by the ERP" was to write status='firm' (or 'released' / 'in_transit') into pipe_supply_orders. That table is owned by the engine: every supply run drops and rewrites its partition. There was no separate place to durably store ERP-originated orders, no way to distinguish them in the UI, no way to attach release_dt/arrival_dt (the table only has weekly indices), and no engine flag to constrain recommendations around them.

master_supply_orders_firm solves all of this:

a per-tenant master table that lives outside the engine partition;
datetime-anchored (release_dt, arrival_dt) so the data model is granularity-ready;
full ERP lifecycle (firm → released → in_transit → received);
two new scenario parameters that change how the engine interacts with the firm rows.

2. Code paths¶

2.1 Layer overview¶

Layer	File	Responsibility
PG DDL	`files/DDL/supply_schema.sql`	Table + indexes + one-time legacy migration
CH DDL	`files/DDL/ch_schema.sql`	`MASTER_supply_orders_firm` PostgreSQL-engine mirror
CH DDL	`files/DDL/allocation_ch_schema.sql`	`is_firm` column on `PIPE_allocation_results`
Python loader	`files/supply_runner.py`	`_load_firm_supply_orders`, `_load_min_lt_by_type`
Python config	`files/supply_runner.py:_load_supply_config`	Defaults + bool coercion for the two new flags
Python → Rust	`files/supply_runner.py` engine_cfg dict	Forwards flags to Rust msgpack payload
Rust types	`files/supply/src/types.rs`	`OrderType::idx()`, `SkuPlan.latest_firm_by_type[7]`, `SkuPlan.min_lt_by_type[7]`
Rust config	`files/supply/src/engine.rs:EngineConfig`	Two new fields + Default impl
Rust sourcing	`files/supply/src/sourcing.rs:sourcing_pass`	Per-source-type skip check in Pass-2 main loop
Allocation	`files/allocation_runner.py:load_supplies`	UNION firm rows from CH with `is_firm=1`, `reliability=1.0`
API list	`files/api/supply_router.py:list_supply_orders`	`include_firm` / `firm_only` query params
API CRUD	`files/api/supply_router.py`	`GET/POST/PUT/DELETE /supply/firm-orders[/{id}]`, `POST /supply/firm-orders/import`
Frontend	`files/frontend/src/components/SupplyPlan.jsx`	`OrdersTab` toggle + firm row styling
Frontend	`files/frontend/src/components/DetailSupplyTab.jsx`	SKU drill-down firm row styling
Frontend	`files/frontend/src/components/SupplyScenarios.jsx`	Two new fields in Planning Horizon group

2.2 OrderType ↔ index mapping¶

The per-SKU arrays latest_firm_by_type[7] and min_lt_by_type[7] are indexed by OrderType::idx(). The mapping is duplicated across the boundary:

Index	Rust `OrderType`	Python `_ORDER_TYPE_IDX` key
0	`Buy`	`"BUY"`
1	`Build`	`"BUILD"`
2	`Repair`	`"REPAIR"`
3	`Transfer`	`"TRANSFER"`
4	`Balancing`	`"BALANCING"`
5	`Replenish`	`"REPLENISH"`
6	`Return`	`"RETURN"`

Drift hazard

The mapping is hand-coded in three files: files/supply/src/types.rs (OrderType::idx), files/supply/src/sourcing.rs (the match source_type { ... } block in sourcing_pass), and files/supply_runner.py (_ORDER_TYPE_IDX dict). If you renumber OrderType, all three must change in lock-step. The sourcing.rs match also intentionally omits Replenish (slot 5) because no SourceType::Replenish exists in routes — Replenish orders are produced post-sourcing.

3. Engine flags — what happens at sourcing time¶

3.1 Per-type bound construction (Python side)¶

# files/supply_runner.py — _load_firm_supply_orders
for (item_id, site_id, order_type, release_dt, arrival_dt, ...) in rows:
    week_idx = floor((release_dt.date - today).days / 7)  # clamped to [0, 51]
    ot_idx   = _ORDER_TYPE_IDX[order_type.upper()]
    out[(item_id, site_id)]["latest_firm_by_type"][ot_idx] = \
        max(out[(item_id, site_id)]["latest_firm_by_type"][ot_idx], week_idx)

min_lt_by_type is built similarly from master.route joined to master.route_type:

# files/supply_runner.py — _load_min_lt_by_type
SELECT r.item_id, r.site_id,
       UPPER(COALESCE(rt.planning_type, 'BUY')) AS order_type,
       MIN(COALESCE(r.lead_time, 0))            AS min_lt
FROM master.route r
LEFT JOIN master.route_type rt ON rt.id = r.type_id
GROUP BY r.item_id, r.site_id, rt.planning_type

The MIN across routes is intentionally permissive — the engine assumes the planner has access to the fastest route in that order_type.

3.2 Sentinels¶

Array	Sentinel value	Meaning
`latest_firm_by_type[ot_idx]`	`255`	No firm order of that type exists for that SKU. The check is skipped.
`min_lt_by_type[ot_idx]`	`0`	No route of that type exists for that SKU. The check is skipped.

3.3 Sourcing skip check (Rust side)¶

// files/supply/src/sourcing.rs — sourcing_pass Pass-2 main loop
'source_loop: for source_type in priority_order {
    let ot_idx: usize = match source_type {
        SourceType::Buy       => 0,
        SourceType::Build     => 1,
        SourceType::Repair    => 2,
        SourceType::Transfer  => 3,
        SourceType::Balancing => 4,
        SourceType::Return    => 6,
    };

    if ctx.firm_orders_block_horizon {
        let latest = ctx.plan.latest_firm_by_type[ot_idx];
        if latest != 255 && (t as u8) < latest {
            continue 'source_loop;  // skip this source_type at week t
        }
    }
    if ctx.respect_lead_time {
        let min_lt = ctx.plan.min_lt_by_type[ot_idx];
        if min_lt > 0 && (t as u8) < min_lt {
            continue 'source_loop;
        }
    }
    // ... existing route iteration ...
}

The check is per-source-type (not per-route) and happens before the route iteration, so it's O(1) per (week × source_type).

4. Replacing the legacy `scheduled_receipts` source¶

Before this change, supply_runner read pre-existing firm receipts from CH PIPE_supply_orders WHERE status IN ('firm','released','in_transit'). That worked because the engine itself wrote status='firm' for confirmed orders, but it created a circular dependency: orders were both an input and an output of the engine.

After this change:

PIPE_supply_orders is engine output only. The engine writes planned, constrained, delayed, expedited.
master_supply_orders_firm is the single source of truth for ERP-committed orders.
supply_runner._load_firm_supply_orders reads from PG (via the CH PG-engine mirror for the allocation runner, or directly from PG for the supply runner).
The legacy CH-based loader in supply_runner.py is removed.

The migration block in supply_schema.sql does a one-time lift of any historical firm/released/in_transit rows in pipe_supply_orders:

DO $$
BEGIN
    -- Skipped if master_supply_orders_firm already has rows.
    IF EXISTS (SELECT 1 FROM master_supply_orders_firm LIMIT 1) THEN
        RETURN;
    END IF;
    INSERT INTO master_supply_orders_firm (
        external_id, item_id, site_id, order_type, qty,
        release_dt, arrival_dt, erp_status, source_system, ...
    )
    SELECT
        'legacy:' || pso.id, pso.item_id, pso.site_id, pso.order_type, pso.qty,
        v_anchor + (pso.release_week * INTERVAL '7 days'),
        v_anchor + (pso.arrival_week * INTERVAL '7 days'),
        pso.status, 'legacy_pipe_supply_orders', ...
    FROM pipe_supply_orders pso
    WHERE pso.status IN ('firm','released','in_transit');
END $$;

v_anchor = MIN(started_at) of completed supply runs, falling back to CURRENT_DATE.

5. Allocation engine integration¶

5.1 Supply pool UNION¶

# files/allocation_runner.py — load_supplies
# 1) engine recommendations (no firm/released/in_transit anymore)
SELECT 'SO_' || toString(id), item_id, site_id, qty, arrival_week, order_type, source_site_id
FROM PIPE_supply_orders
WHERE pipeline_id = {pid} AND scenario_id = {sid}
  AND status IN ('planned','constrained','delayed','expedited')

# 2) firm rows from CH mirror — UNION into the supply pool
SELECT 'FIRM_' || toString(id), item_id, site_id,
       max(qty - if(isNull(received_qty),0,received_qty), 0) AS qty_remaining,
       toUInt8(greatest(0, least(51,
           intDiv(dateDiff('day', toDateTime(now()), toDateTime(arrival_dt)), 7)
       ))) AS arrival_week,
       order_type, source_site_id
FROM MASTER_supply_orders_firm
WHERE erp_status IN ('firm','released','in_transit')
  AND qty - if(isNull(received_qty),0,received_qty) > 0

5.2 `supply_id` prefix convention¶

Prefix	Meaning	Reliability
`SO_<id>`	Engine recommendation (id from `PIPE_supply_orders`)	`0.85`
`FIRM_<id>`	ERP-committed (id from `master_supply_orders_firm`)	`1.0`

The allocation runner derives is_firm from the prefix when writing PIPE_allocation_results:

"is_firm": 1 if str(a.get("supply_id", "")).startswith("FIRM_") else 0

5.3 Reliability score in the scoring engine¶

The Rust allocation engine receives each supply with a reliability_score and uses it in the criticality component. Higher reliability = preferred when both supplies can satisfy the same demand. Firm orders therefore tend to be picked first when demand priorities tie.

6. Process log integration¶

The auto-triggered allocation step (after supply) is logged as a child of the supply step:

# files/run_pipeline.py — supply step
log_id = _supply_pl.start_step('supply')
# ... run supply engine ...
# Then auto-trigger allocation as a CHILD of the supply step:
_alloc_result = run_allocation_pipeline(
    pipeline_id=pipeline_id, scenario_id=supply_sid,
    parent_log_id=log_id,
)

# files/allocation_runner.py — run_allocation_pipeline
plog_step_id = plog.start_step("allocation", parent_id=parent_log_id)

In the UI this renders as a nested log entry under the supply step.

Manual standalone allocation runs (triggered from POST /api/pipeline/run/allocation) do NOT pass parent_log_id, so they remain top-level entries — which is correct, because they have no parent step.

7. API contract reference¶

7.1 Pydantic models¶

class FirmSupplyOrderIn(BaseModel):
    external_id:    Optional[str]   = None
    item_id:        int
    site_id:        int
    source_site_id: Optional[int]   = None
    order_type:     str          # validated against allow-list
    qty:            float        # must be > 0
    release_dt:     str          # ISO 8601 — PG will coerce to TIMESTAMPTZ
    arrival_dt:     str          # ISO 8601
    route_id:       Optional[int]   = None
    unit_cost:      Optional[float] = None
    order_cost:     Optional[float] = None
    total_cost:     Optional[float] = None
    erp_status:     Optional[str]   = "firm"
    received_qty:   Optional[float] = 0
    expiry_date:    Optional[str]   = None
    attributes:     Optional[dict]  = None
    source_system:  Optional[str]   = "manual_upload"


class FirmSupplyOrderUpdate(BaseModel):
    # explicit allow-list of mutable fields
    qty:            Optional[float] = None
    arrival_dt:     Optional[str]   = None
    release_dt:     Optional[str]   = None
    erp_status:     Optional[str]   = None
    received_qty:   Optional[float] = None
    expiry_date:    Optional[str]   = None
    attributes:     Optional[dict]  = None

7.2 `GET /supply/orders` extension¶

Query param	Default	Effect
`include_firm`	`true`	UNION firm rows into the response
`firm_only`	`false`	Skip `PIPE_supply_orders` entirely; only return firm rows

Both can be combined with the existing filters (pipeline_id, scenario_id, item_id, site_id, source_site_id, order_type, week_from, week_to). Pagination is applied to the merged result.

8. Testing — manual smoke test¶

# 1. Insert a firm BUY for item 2 site 100 at release_dt = 2026-09-01
curl -X POST http://localhost:8002/api/supply/firm-orders \
     -H "Authorization: Bearer $TOKEN" \
     -H "Content-Type: application/json" \
     -d '{"external_id":"PO-TEST-1","item_id":2,"site_id":100,
          "order_type":"BUY","qty":500,
          "release_dt":"2026-09-01T00:00:00Z",
          "arrival_dt":"2026-10-15T00:00:00Z",
          "erp_status":"firm","source_system":"manual_test"}'

# 2. Enable both flags on the supply scenario
psql -c "UPDATE scenario.config_supply_scenario
         SET param_overrides = param_overrides ||
             '{\"firm_orders_block_horizon\":true,\"respect_lead_time\":true}'::jsonb
         WHERE id = 1"

# 3. Run supply for the pipeline
curl -X POST http://localhost:8002/api/pipeline/run/supply \
     -H "Authorization: Bearer $TOKEN" \
     -H "Content-Type: application/json" \
     -d '{"pipeline_id":4}'

# 4. Verify the engine produced no BUY recommendations before week 12
clickhouse-client -q "
  SELECT release_week, count()
  FROM PIPE_supply_orders
  WHERE pipeline_id = 4 AND item_id = 2 AND site_id = 100
    AND order_type = 'BUY'
  GROUP BY release_week ORDER BY release_week"

Expected: no rows with release_week < 12 (or sparse rows only — depending on whether BUY_EXPEDITE post-processor fires).

9. Known limitations / follow-up work¶

Topic	Status	Notes
Multi-granularity engine (hourly/daily)	Future	Rust arrays are hard-coded `[f32; 52]`; the firm table stores full datetimes so the data model is ready.
Approval → firm promotion	Future	`interact_order_approvals` records approvals but does not auto-promote to `master_supply_orders_firm`.
`BUY_EXPEDITE` / `TRANSFER_PREPOS` blocking	Future	These post-processor orders currently ignore `firm_orders_block_horizon`.
Scheduled ERP sync ETL	Future	Bulk import endpoint exists; a scheduled sync job is not in scope.
Partial receipt inventory consumption	Partial	`received_qty` exists, but engine logic for partially-received firm orders is simplified (subtraction only).