grain-core

Grain's storage-agnostic workflow engine: a pure advance() reducer and a RunStore port, with a conformance kit that any store can prove itself against.

Zero dependencies beyond grain-contracts (the vocabulary), so a service can drive Grain workflows without installing Grain's CLI. The reducer is pure — no clock, no filesystem, no store — and every side effect comes back as data for the driver to apply. That is how one engine serves grain-kit's filesystem runs and Diwa's Postgres missions.

from grain_contracts.workflow import Run
from grain_core import advance, RunStore
from grain_core.kernel import StepStarted

transition = advance(run, StepStarted(step_id=run.cursor), now="2026-07-14T00:00:00Z")
store.save(transition.run, expected_version=version)   # driver applies the result
for effect in transition.effects:                      # ...and any effects, as data
    ...

Implementing a store

RunStore is a Protocol with four methods — load, save(run, *, expected_version), discard_artifact, list_runs. Prove your implementation against the shared kit:

from grain_core.conformance import run_conformance

def test_my_store_conforms():
    run_conformance(lambda: MyRunStore(...))   # a fresh, empty store per call

The kit pins the version contract the port leaves to implementers: version tokens are opaque; save(expected_version=None) creates (and conflicts if the run already exists); save(expected_version=token) updates iff the token is current, else raises ConcurrentModification; list_runs is newest-first by created.

Source and issues · MIT