State store#

The material lifecycle and the SQLite store that makes a run resumable.

The coordination plane: the funnel state machine + the SQLite StateStore (2.1).

Everything the orchestrator knows about “where each material is” lives here, in two parts:

  1. The state machine – the fixed lifecycle a candidate material moves through, encoded exactly once (MaterialState, next_state, is_terminal, can_transition) so there is a single definition of “what comes next” and “what is finished”. The full funnel:

    NEW -> COMPOSED -> STRUCTURED -> SCREENED -> CALCULATED -> EXTRACTED -> DONE

    plus two terminal states outside the linear chain, each reachable from any live state: FAILED (a hard failure) and EXPANDED (a seed or parent retired after fanning out into children – terminal but not a failure, so FAILED keeps meaning genuine failure in every downstream query). CALCULATED (Phase 6) is the output of the optional calculate stage: a run that skips it walks SCREENED straight to EXTRACTED via complete’s explicit target_state – which stages a run executes is run configuration, not lifecycle physics (the finalize doctrine).

  2. The store – one concrete SQLite-backed StateStore (no ABC: this is a single-user tool; the contract lives in docstrings and tests, and an interface gets extracted only if a second database ever actually arrives). SQLite runs in WAL mode with a busy_timeout, and every read-then-write sequence executes inside a BEGIN IMMEDIATE transaction, which is what makes claim an atomic lease even across separate store instances on the same file.

Three durability properties the store guarantees (the restart-safety thesis):

  • Atomic leasing. claim selects unclaimed rows and stamps them inside one immediate transaction – two claimers can never be handed the same item.

  • Idempotent progress. complete is keyed on a (material_id, stage) completions ledger: replaying a completion after a crash advances nothing twice. enqueue, expand, clear_job and terminal fail are likewise safe to replay.

  • The in-flight job ledger. record_job persists every submitted backend handle before any polling, so a restarted run re-attaches to its live jobs instead of resubmitting them. A material referenced by a live record is never swept by release_stale_leases – the recorded handle owns it until clear_job. One live job per material is enforced by the schema itself (the link table’s primary key).

This module never imports the config layer, the CLI layer, or a backend: the handle it stores is an opaque JSON-safe dict, converted from/to the compute plane’s Handle at the orchestrator boundary (2.2).

On the ``# nosec B608`` markers. Several statements below build their SQL with an f-string. In every one of them the interpolated part is only a run of ? placeholders (or a fixed WHERE fragment assembled here) – never a value, never a caller-supplied identifier; the values themselves are always passed to execute as bound parameters. That is the standard way to express a variable-length IN (...) in sqlite3, and it is why bandit’s SQL-injection heuristic is suppressed at those exact lines and nowhere else.

class matmelt.core.state.MaterialState(value)[source]#

Bases: str, Enum

The lifecycle states of one candidate material.

A str enum: members compare equal to their plain-string values and persist as text (state.value into a column, MaterialState(text) back out) with no custom conversion.

The seven forward states are listed in funnel order; FAILED and EXPANDED sit outside the chain, reachable from any live state. CALCULATED belongs to the optional calculate stage – runs that skip it never park a material there.

matmelt.core.state.LIFECYCLE: tuple[MaterialState, ...] = (MaterialState.NEW, MaterialState.COMPOSED, MaterialState.STRUCTURED, MaterialState.SCREENED, MaterialState.CALCULATED, MaterialState.EXTRACTED, MaterialState.DONE)#

The forward funnel, in order. The successor of LIFECYCLE[i] is LIFECYCLE[i + 1]; DONE (last) has none. FAILED / EXPANDED are intentionally absent – they are sinks, not chain members.

matmelt.core.state.is_terminal(state)[source]#

Say whether a state is an end state.

Parameters:

state (MaterialState) – The state to classify.

Returns:

True for DONE, FAILED and EXPANDED. A terminal material has left the funnel: it has no successor and must never be claimed or advanced again.

Return type:

bool

Examples

>>> is_terminal(MaterialState.SCREENED), is_terminal(MaterialState.EXPANDED)
(False, True)
matmelt.core.state.next_state(state)[source]#

Give the state that follows when a stage completes a material.

Parameters:

state (MaterialState) – The material’s current state.

Returns:

Its successor in LIFECYCLE.

Return type:

MaterialState

Raises:

ValueError – For a terminal state – asking for the successor of DONE, FAILED or EXPANDED is a programming error, not a recoverable condition.

Examples

>>> next_state(MaterialState.SCREENED)
<MaterialState.CALCULATED: 'CALCULATED'>
matmelt.core.state.can_transition(current, target)[source]#

Judge whether a stage-driven move is legal.

The store consults this before every write, so an out-of-order replay can never corrupt state.

Parameters:
Returns:

True for the three legal moves: the single forward step to current’s successor (success); any live state to FAILED (hard failure); and any live state to EXPANDED (retired after fan-out). Everything else – skipping a stage, going backwards, leaving a terminal state – is False.

Return type:

bool

Notes

Two moves deliberately sit outside this function: finalize and complete’s explicit target_state. Which state ends a run, and which stages a run executes, are run configuration rather than lifecycle physics.

exception matmelt.core.state.StateError[source]#

Bases: Exception

A coordination-plane contract violation.

Raised for an unknown material_id, an illegal transition (rejected by can_transition), a bad query filter, or a ledger-invariant violation (empty material set, second live job for a material, unserialisable handle). Never used for “no work available” – that is an empty list.

class matmelt.core.state.MaterialRecord(material_id, state, claimed_at=None, parent_id=None, attempts=0, error=None, fail_stage=None, fail_reason=None, completions=<factory>, created_at=0.0, updated_at=0.0)[source]#

Bases: object

One material’s snapshot as handed back by claim / query / get.

Frozen: advancing a material happens only through complete / fail / expand, never by mutating what a reader was handed.

material_id#

The material’s stable id.

Type:

str

state#

Where it is in the funnel.

Type:

MaterialState

claimed_at#

Epoch seconds when the current lease was taken, or None when unleased. The basis for release_stale_leases.

Type:

float or None

parent_id#

Lineage: the material this one was fanned out from, None for a root seed. A plain back-pointer, recorded at enqueue.

Type:

str or None

attempts#

Consecutive failed attempts at the current stage. Reset to 0 by complete.

Type:

int

error#

The latest attempt’s failure reason, set on every fail including requeues, so a requeued material still shows why its last try died.

Type:

str or None

fail_stage#

Which stage the material died at, set when it went terminal via fail.

Type:

str or None

fail_reason#

Why it died. Also reused by expand to carry its note, since a second note kind did not warrant another column.

Type:

str or None

completions#

{stage: completed_at} from the ledger – the material’s forward timeline.

Type:

dict

created_at#

Epoch seconds the row was inserted.

Type:

float

updated_at#

Epoch seconds it was last written. Terminal rows are never written again, so a FAILED material’s updated_at is its failure time; and for any material, now - updated_at is its time in the current state. That is the monitor’s clock, with no extra columns needed.

Type:

float

class matmelt.core.state.InFlightJob(job_key, stage, backend_name, handle, material_ids, submitted_at)[source]#

Bases: object

One submitted-but-unfinished backend job, as persisted by record_job.

job_key#

Caller-chosen unique identity, by convention "<backend_name>:<job_id>". Re-recording a key replaces the record.

Type:

str

stage#

The stage whose work this job carries.

Type:

str

backend_name#

Which configured backend the handle must be resolved against.

Type:

str

handle#

The backend Handle as an opaque JSON-safe dict. The orchestrator converts at the boundary, so this module never imports the compute plane.

Type:

dict

material_ids#

Every material whose result this job will produce.

Type:

tuple of str

submitted_at#

Epoch seconds when the record was written.

Type:

float

matmelt.core.state.FINALIZE_STAGE = 'finalize'#

Reserved completions-ledger stage name for the orchestrator’s stop-stage bookkeeping transition (finalize). Not a real stage; a stage must never use it as its own name.

class matmelt.core.state.StateStore(db_path, *, busy_timeout_ms=5000, read_only=False)[source]#

Bases: object

The durable coordination plane, backed by one SQLite database file.

One connection per instance, shared across threads behind an internal lock (check_same_thread=False); isolation_level=None gives explicit transaction control so read-then-write sequences run under BEGIN IMMEDIATE. Cross-instance safety (two stores on one file) comes from SQLite itself: the immediate transaction takes the write lock up front, and busy_timeout absorbs brief contention instead of raising.

Parameters:

  • db_path (str | Path) – Path to the database file, created if absent. Opening an existing file is a safe no-op re-apply of the schema, so a run resumes against the same file it was interrupted on.

  • busy_timeout_ms (int, default: 5000) – How long SQLite waits for a contended write lock before raising.

  • read_only (bool, default: False) – Open an observer store – the monitor seam. The schema script is skipped (it is a write) and PRAGMA query_only=ON makes SQLite itself reject every write on the connection, so the no-corruption guarantee is enforced by the engine rather than by caller discipline. Mutating methods additionally raise up front, for a friendlier failure than SQLITE_READONLY.

Raises:

StateError – When read_only is set and no database exists at db_path. An observer must never silently create an empty database.

Notes

query_only is chosen over the URI ?mode=ro deliberately: a WAL reader may need to create the -shm file – after the last writer died uncleanly, for instance – which mode=ro forbids and query_only allows.

close()[source]#

Close the underlying connection.

Safe to call more than once. A closed store is not reusable; open a new one against the same file, which is exactly what a resumed run does.

Return type:

None

enqueue(material_id, parent_id=None)[source]#

Add material_id to the funnel in state NEW. Idempotent.

Parameters:

  • material_id (str) – The id to enqueue.

  • parent_id (str | None, default: None) – The material this one fanned out from, recorded once as lineage. None for a root seed.

Raises:

StateError – When the store is read-only.

Return type:

None

Notes

Re-enqueueing an existing id touches nothing: its current state, attempts, and recorded parent_id all stand, first write wins. That is what lets a replayed seed or a re-run feeder never knock a material backwards or rewrite its lineage.

get(material_id)[source]#

Fetch one material’s full snapshot. Read-only.

Parameters:

material_id (str) – The id to fetch.

Returns:

The snapshot, or None when the store does not know the id.

Return type:

MaterialRecord or None

query(filters=None)[source]#

List materials matching filters, oldest first. Read-only.

No leasing, no writes.

Parameters:

filters (dict[str, Any] | None, default: None) – Field/value pairs, AND-combined. Keys must be one of material_id, state or parent_id – whitelisted, so a caller-supplied field name is never interpolated into SQL unchecked. A MaterialState value is coerced to its stored text. None or empty returns every material.

Returns:

Matching snapshots, ordered by creation time then id.

Return type:

list of MaterialRecord

Raises:

StateError – For a filter key outside the whitelist.

count_by_state()[source]#

Tally the materials per state. Read-only.

Returns:

{state: count} from one GROUP BY. States holding no materials are simply absent; callers zero-fill if they need the full set.

Return type:

dict

completion_counts_since(cutoff)[source]#

Count completions ledgered at or after a cutoff. Read-only.

The rolling-window primitive behind the throughput and speed displays: the caller picks the window and divides.

Parameters:

cutoff (float) – Epoch seconds. Pass 0.0 for all-time counts, or time.time() - window_s for a rolling window.

Returns:

{stage: count}. Reserved FINALIZE_STAGE rows appear under their own name, leaving the caller to decide whether finalizations count as stage work. Stages with no completions in the window are simply absent.

Return type:

dict

recent_failures(limit=10)[source]#

List the most recently failed materials, newest first. Read-only.

Parameters:

limit (int, default: 10) – How many to return. Zero or less returns nothing.

Returns:

Ordered by updated_at descending. A terminal row is never written again, so its updated_at is exactly when it failed, and the record carries it so callers can render “failed at”.

Return type:

list of MaterialRecord

recent_successes(limit=10)[source]#

List the most recently finished materials, newest first. Read-only.

The success mirror of recent_failures.

Parameters:

limit (int, default: 10) – How many to return. Zero or less returns nothing.

Returns:

DONE materials ordered by updated_at descending. A terminal row is never written again, so updated_at is exactly when the material finished and updated_at - created_at is how long its whole ride took; the monitor renders both.

Return type:

list of MaterialRecord

earliest_created_at()[source]#

Find when the first material entered the store. Read-only.

Returns:

Epoch seconds of the earliest created_at, or None on an empty store. This is the run’s wall-clock anchor: elapsed time is now - earliest.

Return type:

float or None

claim(state, limit, *, among=None)[source]#

Atomically lease up to limit unclaimed materials in state.

Selects the oldest unleased rows and stamps claimed_at inside one immediate transaction, so two concurrent claimers can never be handed the same material.

Parameters:

  • state (MaterialState) – Which input state to claim from.

  • limit (int) – The most materials to lease. Zero or less claims nothing.

  • among (Collection[str] | None, default: None) – Restrict the claim to these ids – the selective-run seam, where the orchestrator precomputes the eligible lineage and leases nothing outside it. None is unrestricted; an empty collection claims nothing.

Returns:

The leased snapshots, oldest first. Empty when nothing is available.

Return type:

list of MaterialRecord

Raises:

StateError – When state is terminal – claiming one is a programming error – or when the store is read-only.

Notes

The among ids ride an SQL IN clause, which very old SQLite builds cap at 999 entries. That is fine at v1 scale; chunking is a later concern.

complete(material_id, stage, *, target_state=None)[source]#

Advance material_id forward in the funnel and release its lease.

Idempotent on the (material_id, stage) completions-ledger key: if this stage already completed this material the call is a no-op, so a crash-replay never advances twice. Otherwise the material moves to the target, the lease clears, the retry budget resets (attempts = 0, error = NULL) and the ledger row is written – all in one transaction.

Parameters:

  • material_id (str) – The material to advance.

  • stage (str) – The stage that completed it, which is also the ledger’s idempotency key.

  • target_state (MaterialState | None, default: None) – Where to move it. None takes the single step to its lifecycle successor. An explicit value moves it to any strictly forward chain state, which is how a run that skips an optional stage (say calculate) walks its consumer’s completion over the skipped state – SCREENED straight to EXTRACTED. Like finalize, the jump sits deliberately outside can_transition: which stages a run executes is run configuration, not lifecycle physics.

Raises:

StateError – For an unknown material; for one already terminal; for a backward, same-state, or off-chain (FAILED / EXPANDED) target; or when the store is read-only.

Return type:

None

fail(material_id, stage, reason, *, max_attempts=1)[source]#

Spend one retry attempt; requeue while budget remains, else fail terminally.

Increments attempts and records reason verbatim in error.

Parameters:

  • material_id (str) – The material that failed.

  • stage (str) – Where it failed, recorded in fail_stage when the failure turns terminal.

  • reason (str) – Why, stored verbatim. Shown by status and monitor.

  • max_attempts (int, default: 1) – The stage’s retry budget. The default of 1 makes the first failure terminal; the orchestrator passes the configured per-stage value.

Returns:

The material’s state after the call: unchanged when it was requeued (lease cleared, so the next claim re-runs the same stage), or FAILED when the budget ran out. A failed row stays in the database – failed candidates are still data.

Return type:

MaterialState

Raises:

StateError – For an unknown material, one terminal as DONE or EXPANDED, or a read-only store. Calling on an already-FAILED material is idempotent: a no-op that spends no attempt.

reset(material_id, target_state)[source]#

Un-fail a terminal FAILED material back to a live, claimable state.

The recovery primitive behind matmelt retry. The lease clears and all failure bookkeeping is wiped (attempts = 0, and error, fail_stage, fail_reason all set to NULL), so the next run re-claims the material and re-runs that stage.

The completions ledger is left untouched: those earlier stages really did complete. Preserving the timeline is what lets the caller compute the correct target_state in the first place, and it keeps lineage and provenance intact.

Parameters:

  • material_id (str) – The failed material to revive.

  • target_state (MaterialState) – Where to put it back – the state it was claimable from before it failed, i.e. its fail_stage’s input state. The caller derives this from the material’s completions timeline, because the state layer holds no stage-to-state map.

Raises:

StateError – For an unknown material; for one that is not FAILED, since this is a recovery primitive and not a general “set state”; for a terminal target_state; or for a read-only store.

Return type:

None

expand(material_id, reason)[source]#

Retire material_id as EXPANDED – it fanned out into children.

Terminal like FAILED but not a failure. Keeping them distinct is what lets FAILED go on meaning genuine failure in every downstream query. The lease clears.

Parameters:

  • material_id (str) – The parent to retire.

  • reason (str) – A diagnostic note, stored in fail_reason – a second note kind did not warrant another column.

Raises:

StateError – For an unknown material, one terminal as DONE or FAILED, or a read-only store. Calling on an already-EXPANDED material is idempotent.

Return type:

None

finalize(material_id)[source]#

Retire a live material as first-class terminal DONE.

The orchestrator’s bookkeeping move, not a stage transition: a material that completes the run’s last stage has no consuming stage left, so this retires it as “went as far as this run was asked to go” and records a completions-ledger row under the reserved stage name FINALIZE_STAGE. The lease clears and the retry budget resets.

It sits deliberately outside can_transition, which describes stage-driven moves only: finalize is legal from any live state, because which state is “the end” is run configuration rather than lifecycle physics.

Parameters:

material_id (str) – The material to retire.

Raises:

StateError – For an unknown material, one terminal as FAILED or EXPANDED, or a read-only store. Calling on an already-DONE material is idempotent.

Return type:

None

release_stale_leases(timeout_s)[source]#

Clear leases older than timeout_s seconds.

Run at startup and between passes: a lease older than the timeout belonged to a run that died mid-stage, and clearing it returns the material to its queue with its state unchanged.

Parameters:

timeout_s (float) – Lease age, in seconds, past which a lease is considered abandoned.

Returns:

How many leases were freed.

Return type:

int

Raises:

StateError – When the store is read-only.

Notes

Only live, currently leased rows are touched, and a material referenced by a live in-flight job is never swept whatever its lease age: the recorded handle owns it until clear_job. Releasing underneath a running job is exactly the double-submission hole the ledger closes.

record_job(job_key, stage, backend_name, handle, material_ids)[source]#

Persist one submitted backend job: job row plus material links, atomically.

Called immediately after backend.submit returns and before any polling – this record is what survives a crash. Re-recording an existing job_key replaces the record and its links, so a replayed submit leaves exactly one record.

Parameters:

  • job_key (str) – The job’s unique identity, by convention "<backend_name>:<job_id>".

  • stage (str) – Whose work the job carries.

  • backend_name (str) – Which configured backend the handle resolves against.

  • handle (dict[str, Any]) – The backend Handle as a JSON-safe dict.

  • material_ids (Sequence[str]) – Every material the job will produce a result for. Duplicates are dropped, caller order preserved.

Raises:

StateError – When material_ids is empty; when the handle is not JSON-serialisable; when a material is already linked to a different live job, since the schema allows one live job per material; when a material is unknown to the store; or when the store is read-only.

Return type:

None

list_jobs(*, stage=None, backend_name=None)[source]#

List the live in-flight jobs, oldest first. Read-only.

Parameters:

  • stage (str | None, default: None) – Restrict to one stage’s jobs. None lists every stage.

  • backend_name (str | None, default: None) – Restrict to one backend’s jobs. None lists every backend.

Returns:

Ordered by submission time then key, each carrying its material links.

Return type:

list of InFlightJob

Notes

The check half of the pass loop lists everything; the submit half filters by backend_name to count live jobs against that backend’s max_parallel.

clear_job(job_key)[source]#

Remove one in-flight record, making its materials sweepable again.

Called after the job’s results were attributed – complete or fail per material – and the backend handle released.

Parameters:

job_key (str) – The record to remove. Clearing an unknown or already-cleared key is a no-op, so restart-time cleanup is always safe.

Raises:

StateError – When the store is read-only.

Return type:

None