Orchestrator

The single-threaded pass loop where stages, backends and durable state meet.

The orchestrator: a single-threaded pass loop over durable state (2.2).

This is the seam where the three planes meet – StateStore (coordination), Stage (what), ComputeBackend (where) – and it is deliberately the only place they meet. It is constructed from plain objects (an ordered list of stages, a {stage_name: backend} mapping): it never reads config (that wiring arrives at 2.3 via build_runtime), never imports the CLI layer, and never reaches inside a stage or a backend beyond their contracts.

The concurrency model (locked 2026-07-08): parallelism is jobs in flight, never threads. Jobs run as OS processes wherever the backend puts them; the orchestrator itself is one thread running short passes over the durable state in SQLite. Every fact needed to resume – who is where, which handles are live – is in the store, so the process can be killed at any point and a new one continues the same run.

Each run_pass() is one non-blocking sweep, four steps:

1. Stale-lease release – leases older than lease_timeout_s are returned to their queue (materials owned by a recorded in-flight job are never touched; the store guarantees it).

2. Check phase – every job in the in-flight ledger is polled through its recorded backend. Still-running jobs are left recorded (the pass never waits). A terminal job is fetched, judged by its stage (interpret_result then validate_outputs), attributed to the store (complete / fail – or, for a fan_out stage, children adopted from the manifest and the parent retired as EXPANDED), its handle released (best-effort), and its ledger record cleared.

3. Finalize – materials that completed the configured funnel’s last stage are retired as first-class DONE via StateStore.finalize (idempotent, so a crash between check and finalize just retries next pass). A run stopped mid-funnel (stop_after or --only naming an earlier stage) never finalizes: its materials rest at that stage’s output state, live and claimable, so a later run continues them.

4. Submit phase – per stage in funnel order up to stop_after: backlog gate (a stage whose input state is empty is never considered) -> health gate (an unhealthy backend pauses the stage; it resumes by itself on a later pass) -> claim up to the backend’s free slots (max_parallel minus its recorded in-flight jobs) -> per item: validate_inputs -> build_jobspec -> submit -> record_job. The handle is recorded before the pass ends, so a crash after submission re-polls the job instead of resubmitting it; a job whose record cannot be written is released (abandoned) so nothing runs untracked.

Failure attribution happens at exactly two helpers (the single attribution point of the design):

• _complete – a store fault while completing leaves the item leased; the lease timeout recovers it. Never raises.

• _fail – the verdict-vs-fault doctrine is applied here and only here: a StageVerdict with terminal=True is a deterministic science verdict a re-run cannot change, so it bypasses the retry budget (failed with budget 1); everything else is a fault that spends one unit of retry_budget and requeues while budget remains. StageInputError is budgeted the same way (a missing input can be a not-yet-mounted disk). A raised StageError (the stage as a whole cannot run) fails nothing: the stage is skipped this pass and its claimed items recover by lease timeout.

A pass never raises. Every phase guards its store/backend/stage calls; an unexpected error is logged (stdlib logging, logger matmelt.core.orchestrator) and the pass moves on. run() loops passes with a sleep between them and returns cleanly on KeyboardInterrupt – in-flight jobs stay recorded and the next run re-attaches to them.

class matmelt.core.orchestrator.Orchestrator(store, stages, backends, work_dir, *, stop_after=None, only=None, materials=None, retry_budget=3, lease_timeout_s=900.0)

Bases: object

Drives materials through the funnel: check in-flight work, submit new work.

Parameters:

• store (StateStore) – The durable coordination plane. The orchestrator is stateless beyond it: two consecutive processes given the same store continue one run.

• stages (Sequence[Stage]) – The pipeline stages in funnel order. Each stage consumes the previous stage’s output state (the first consumes NEW), and its completion lands the material in its own output state: a name in STAGE_OUTPUT_STATE owns that lifecycle state by name – so omitting an optional stage never shifts its neighbours’ states – while any other name takes the next chain state positionally. Output states must advance strictly (which bounds unmapped stages at len(LIFECYCLE) - 1). Names must be unique; the reserved name "finalize" is not a legal stage name.

• backends (Mapping[str, ComputeBackend]) – {stage_name: backend} – which backend runs each stage’s jobs. Every stage must be wired (missing wiring is a construction error, not a runtime surprise). Two stages may share one backend instance; its free slots are then shared too.

• work_dir (str | Path) – Root of the per-item working directories: item m works in work_dir/m (created at first submission). This is the only place the id -> directory convention lives.

• stop_after (str | None, default: None) – Name of the last stage this run executes (a first-class knob, not a debug hack): stages after it are never considered. Materials that complete it rest at its output state – live, claimable by the next stage in a later run; they are finalized to terminal DONE only when the stop stage is the configured last stage (where resting and finishing coincide). None means the full configured funnel.

• only (str | None, default: None) – Restrict the submit phase to exactly this stage (single-stage runs on demand). The check phase is untouched – any recorded in-flight job still settles – and finalize runs only when the only stage is the configured last stage: a mid-funnel only-run must never retire its output as DONE (a later run’s stages still consume it). Mutually exclusive with stop_after.

• materials (frozenset[str] | None, default: None) – Restrict the run to this selected lineage: only materials whose id – or any ancestor’s id, walking parent_id – is in the set are claimed by the submit phase or retired by finalize. Fan-out children of a selected parent are therefore eligible automatically, including on a rerun (the ancestry walk needs no memory of the adoption). The check phase is untouched: any recorded in-flight job still settles, whoever it belongs to. None = unrestricted; an empty set is a construction error (it would silently do nothing).

• retry_budget (int, default: 3) – Per-stage fault budget handed to StateStore.fail – a faulted item is requeued until the budget is exhausted, then goes terminal FAILED. Terminal verdicts bypass it entirely.

• lease_timeout_s (float, default: 900.0) – Age beyond which a lease is considered orphaned and released at the start of every pass (in-flight-ledger materials excepted).

seed(material_ids)

Enqueue material_ids as NEW roots.

Parameters:

material_ids (Iterable[str]) – The seed ids to offer, typically from core.seeds.element_combinations.

Returns:

How many ids were offered – not how many were new. The call is idempotent end to end, since enqueue never regresses an existing material, so replaying a seed list, or seeding while a run is in progress, is always safe.

Return type:

int

run(*, max_passes=None, pass_interval_s=1.0)

Run passes until a limit is reached or the user interrupts.

Parameters:

• max_passes (int | None, default: None) – How many passes to run. None runs until interrupted.

• pass_interval_s (float, default: 1.0) – How long to sleep between passes.

Return type:

None

Notes

KeyboardInterrupt is the supported way to stop an open-ended run: it is caught, logged, and the method returns cleanly. In-flight jobs stay recorded in the ledger, and the next run() re-attaches to them instead of resubmitting.

run_pass()

Sweep once, without blocking: release stale leases, check, finalize, submit.

Never raises.

Returns:

What the pass did, with every key always present and zero when nothing of that kind happened – callers may index the summary without checking:

released

Stale leases freed.

completed, failed, requeued

Materials attributed by the check phase or by submit-time input validation.

expanded, children

Fan-out parents retired as EXPANDED, and the materials those parents adopted.

finalized

Materials retired as DONE.

submitted

New jobs recorded in the ledger.

Return type:

dict