Custom Substrate — Plugin Example¶

Bring your own storage. Implement StoreSubstrate — the one seam that decides where functualize keeps its own bookkeeping — and install it at boot. Every store follows, because there is one place that decides and one object handed to all of them.

This example's implementation is a dict. The shape is what transfers: it is the same shape functualize-substrate-sqlite fills with a database.

Source¶

examples/plugins/custom_state_backend/

The Protocol¶

Six members. A substrate is asked for a document by key; it is never asked what a scope is or which document refuses an unreadable read — those are decisions about meaning and they stay on the stores.

from functualize._types.protocols import Stored, StoreSubstrate

class MySubstrate:
    def read(self, key: str) -> Stored | None: ...
    def write(self, key: str, payload: dict, *, expect: int | None = None) -> bool: ...
    def lock(self, *keys: str): ...                 # a context manager
    def clear(self, key: str) -> str | None: ...    # `func builtin data clear`
    def delete(self, key: str) -> bool: ...         # the scope purge
    def describe(self, key: str) -> str: ...        # `func builtin data show`

assert isinstance(MySubstrate(), StoreSubstrate)

Two distinctions the stores depend on:

Nothing stored is not an empty document. A missing scopes reads as "no scopes"; an empty one reads as "a scope file that happens to be empty". Return None for the first and a Stored for the second.
clear is not delete. clear is the documented way out of a document that cannot be read and may keep a copy aside; delete is the scope purge and must not.

Two ways to be safe, because backends differ. A filesystem gets mutual exclusion from flock; a remote store often cannot offer it and needs compare-and-swap instead, so write(..., expect=) is in the port from the start and reports refusal rather than assuming a lock was held.

Plugin Boot Class¶

from functualize.plugin import PluginHost

class MyPlugin:
    name = "state-my-substrate"

    def __call__(self, app: PluginHost) -> None:
        app.offer_substrate(self._choose)

    def _choose(self, app: PluginHost) -> MySubstrate:
        return MySubstrate()

Offer at registration; boot asks. Boot calls the offer once, while it selects the store — after configuration has resolved, so _choose can read the plugin's own settings through app.configuration, and before the engine is built, so what it returns is the storage every store uses. A plugin whose choice needs no configuration may call app.install_substrate(MySubstrate()) from __call__ instead.

APP_READY is too late: the engine already holds its storage by then, and an install from there is refused — boot stops with SubstrateInstallError rather than half-applying it. Some of a run's documents in one backend and some in another is the state this seam exists to make unreachable. So is choosing between two storage plugins: if two claim storage, boot refuses and names both.

Why this is not a `StateBackend`¶

It used to be. ADR-022 records why the key-value domain was retired: a backend-agnostic key-value protocol can only offer the intersection of every backend, which is worth least exactly where having a real database is worth most. The MCP history tools were the proof — get_job_history could not ask for "recent executions", so it probed four method names and took whichever the installed backend had.

Durable state for a job is a different question, answered by rc.state and by the workflow scope — not by a storage plugin.

Key Concepts¶

Protocol compliance — isinstance(substrate, StoreSubstrate) validates the shape
One decision — the substrate is chosen once; fingerprints, scopes and the run log all follow it
Entry points — auto-discovery without manual configuration