Skip to content

Quality gates

The gate contract and helper. See the Quality gates guide, and the idempotency recipe for artifact_key / skip_if_done.

agentflow.prebuilt.gate

Quality gates: a checked "good enough?" decision on a step of work.

A gate is not a new kind of node in the graph API and not a new backend. It is a convention plus a thin helper over the primitives you already have: conditional edges, a per-gate attempt counter, and ctx.emit for telemetry. The rule it enforces is the one that keeps long autonomous runs debuggable: a work node writes an artifact; a gate decides what happens next — accept it, send it back for revision, or escalate to a human.

The contract is one dataclass and one helper. Mix :class:GateState into your state schema so the gate has channels for its bookkeeping::

class ReviewState(GateState):
    artifact: Annotated[str, last]
    feedback: Annotated[str, last]

async def evaluate(state, ctx) -> GateResult:
    score = await grade(state["artifact"])
    return GateResult(passed=score >= 0.85, score=score,
                      feedback=None if score >= 0.85 else "add tests")

g = Graph(ReviewState)
g.add_node("revise", revise)
add_quality_gate(
    g, "review", evaluate,
    on_pass=END, on_fail="revise", on_escalate="ask_human",
    max_attempts=3,
)
g.add_edge("revise", "review")

GateResult vs Candidate

:class:GateResult and :class:~agentflow.prebuilt.Candidate look similar but answer different questions; pick by what you are building.

  • :class:~agentflow.prebuilt.Candidate is the output of the worker inside :func:~agentflow.prebuilt.iterate_until_converged. Candidate.done means "the loop may stop now"; the loop owns the stop policy (best-so-far, patience, max iterations).
  • :class:GateResult is the output of an evaluator consulted by :func:add_quality_gate. GateResult.passed means "this artifact meets the bar"; the gate owns the routing (pass / fail-and-revise / escalate).

Use iterate_until_converged for a self-contained optimize loop with a score to climb. Use add_quality_gate for an explicit checkpoint in a larger graph that routes three ways and can hand off to a human. They compose: a gate's on_fail can point back into a revise node, which is its own small loop.

GateResult dataclass

An evaluator's verdict on an artifact.

passed is the only field the router requires. score and reasons are for observability and human review; feedback is the actionable text a revise node can consume to improve the artifact on the next attempt.

The gate stores this verdict in state as a plain dict (via :meth:to_dict), not as this object, so it is JSON-serializable and survives a durable checkpointer. Read it from state["gate_results"][name] as a mapping: result["passed"], result["feedback"], and so on.

Source code in agentflow/prebuilt/gate.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
@dataclass(frozen=True)
class GateResult:
    """An evaluator's verdict on an artifact.

    ``passed`` is the only field the router requires. ``score`` and ``reasons``
    are for observability and human review; ``feedback`` is the actionable text
    a revise node can consume to improve the artifact on the next attempt.

    The gate stores this verdict in state as a plain dict (via :meth:`to_dict`),
    not as this object, so it is JSON-serializable and survives a durable
    checkpointer. Read it from ``state["gate_results"][name]`` as a mapping:
    ``result["passed"]``, ``result["feedback"]``, and so on.
    """

    passed: bool
    score: float | None = None
    reasons: list[str] = field(default_factory=list)
    feedback: str | None = None

    def to_dict(self) -> dict[str, Any]:
        """A JSON-serializable view, as stored in the ``gate_results`` channel."""
        return {
            "passed": self.passed,
            "score": self.score,
            "reasons": list(self.reasons),
            "feedback": self.feedback,
        }

to_dict 

to_dict() -> dict[str, Any]

A JSON-serializable view, as stored in the gate_results channel.

Source code in agentflow/prebuilt/gate.py 
 96
 97
 98
 99
100
101
102
103
def to_dict(self) -> dict[str, Any]:
    """A JSON-serializable view, as stored in the ``gate_results`` channel."""
    return {
        "passed": self.passed,
        "score": self.score,
        "reasons": list(self.reasons),
        "feedback": self.feedback,
    }

GateState

Bases: State

Mix into a state schema to add the channels quality gates need.

One pair of channels serves any number of gates in the graph — the gate name is the dict key, so you never declare per-gate fields. Both use last-wins/merge semantics and hold only gate bookkeeping.

Source code in agentflow/prebuilt/gate.py 
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class GateState(State):
    """Mix into a state schema to add the channels quality gates need.

    One pair of channels serves any number of gates in the graph — the gate
    name is the dict key, so you never declare per-gate fields. Both use
    last-wins/merge semantics and hold only gate bookkeeping.
    """

    #: Attempt count per gate name.
    gate_attempts: Annotated[dict, merge]
    #: The most recent verdict per gate name, stored as a plain dict
    #: (``{"passed", "score", "reasons", "feedback"}``) so it is
    #: JSON-serializable and survives a durable checkpointer. Read it as a
    #: mapping — ``state["gate_results"]["name"]["passed"]``.
    gate_results: Annotated[dict, last]

Evaluator

Bases: Protocol

A gate's decision function: async (state, ctx) -> GateResult.

It may be pure code or call a backend (an LLMBackend scoring the artifact, an AgentBackend reviewing a diff). It is an ordinary node body, not a backend — do not confuse it with a token source. Keep it read-only with respect to the artifact: it judges, it does not fix.

Source code in agentflow/prebuilt/gate.py 
123
124
125
126
127
128
129
130
131
132
133
@runtime_checkable
class Evaluator(Protocol):
    """A gate's decision function: ``async (state, ctx) -> GateResult``.

    It may be pure code or call a backend (an ``LLMBackend`` scoring the
    artifact, an ``AgentBackend`` reviewing a diff). It is an ordinary node
    body, not a backend — do not confuse it with a token source. Keep it
    read-only with respect to the artifact: it judges, it does not fix.
    """

    async def __call__(self, state: Mapping[str, Any], ctx: Any) -> GateResult: ...

add_quality_gate 

add_quality_gate(g: Graph, name: str, evaluator: Evaluator | Callable[[Mapping[str, Any], Any], Awaitable[GateResult]], *, on_pass: str, on_fail: str, on_escalate: str, max_attempts: int = 3) -> Graph

Add a quality-gate node name and its three-way routing to g.

The gate runs evaluator once per visit and routes on the verdict:

  • passed -> on_pass
  • failed with attempts remaining -> on_fail (typically a revise node that loops back into this gate)
  • failed at max_attempts -> on_escalate (often a human-in-the-loop node, or :data:~agentflow.graph.END)

on_pass / on_fail / on_escalate are node names already in the graph, or :data:~agentflow.graph.END. All three are required so a gate never silently accepts a failing artifact once it runs out of attempts — for an autonomous run point on_escalate at a human gate or a park-and-report node.

The gate emits :class:~agentflow.events.GateEvents (started then one of passed / failed / escalated) via ctx.emit, so telemetry and the run stream show each decision and how many times a process looped on quality. It keeps its attempt count in the gate_attempts channel and its last verdict in gate_results — mix :class:GateState into your schema to declare both. Returns g for chaining.

Source code in agentflow/prebuilt/gate.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
def add_quality_gate(
    g: Graph,
    name: str,
    evaluator: Evaluator | Callable[[Mapping[str, Any], Any], Awaitable[GateResult]],
    *,
    on_pass: str,
    on_fail: str,
    on_escalate: str,
    max_attempts: int = 3,
) -> Graph:
    """Add a quality-gate node ``name`` and its three-way routing to ``g``.

    The gate runs ``evaluator`` once per visit and routes on the verdict:

    - passed                              -> ``on_pass``
    - failed with attempts remaining      -> ``on_fail`` (typically a revise
      node that loops back into this gate)
    - failed at ``max_attempts``          -> ``on_escalate`` (often a
      human-in-the-loop node, or :data:`~agentflow.graph.END`)

    ``on_pass`` / ``on_fail`` / ``on_escalate`` are node names already in the
    graph, or :data:`~agentflow.graph.END`. All three are required so a gate
    never silently accepts a failing artifact once it runs out of attempts —
    for an autonomous run point ``on_escalate`` at a human gate or a
    park-and-report node.

    The gate emits :class:`~agentflow.events.GateEvent`s (``started`` then one
    of ``passed`` / ``failed`` / ``escalated``) via ``ctx.emit``, so telemetry
    and the run stream show each decision and how many times a process looped on
    quality. It keeps its attempt count in the ``gate_attempts`` channel and its
    last verdict in ``gate_results`` — mix :class:`GateState` into your schema
    to declare both. Returns ``g`` for chaining.
    """
    if name in g.nodes:
        raise ValueError(f"node {name!r} already exists")
    if max_attempts < 1:
        raise ValueError("max_attempts must be >= 1")

    async def gate(state: Mapping[str, Any], ctx: Any) -> dict[str, Any]:
        attempts = dict(state.get("gate_attempts") or {})
        attempt = int(attempts.get(name, 0)) + 1
        ctx.emit(GateEvent(gate=name, phase="started", attempt=attempt))

        result = await evaluator(state, ctx)
        reasons = tuple(result.reasons)

        if result.passed:
            phase = "passed"
        elif attempt >= max_attempts:
            phase = "escalated"
        else:
            phase = "failed"
        ctx.emit(
            GateEvent(gate=name, phase=phase, attempt=attempt, score=result.score, reasons=reasons)
        )

        results = dict(state.get("gate_results") or {})
        results[name] = result.to_dict()  # store a plain dict, not the dataclass
        return {"gate_attempts": {name: attempt}, "gate_results": results}

    def route(state: Mapping[str, Any]) -> str:
        results = state.get("gate_results") or {}
        result: dict[str, Any] | None = results.get(name)
        attempts = state.get("gate_attempts") or {}
        attempt = int(attempts.get(name, 0))
        if result is not None and result["passed"]:
            return "pass"
        if attempt >= max_attempts:
            return "escalate"
        return "fail"

    g.add_node(name, gate)
    g.add_conditional_edges(
        name,
        route,
        {"pass": on_pass, "fail": on_fail, "escalate": on_escalate},
    )
    return g

Idempotency helpers

skip_if_done avoids recomputing deterministic work; IdempotentOp protects a side effect. See long-running loops for when to use which.

agentflow.prebuilt.idempotent

Idempotency for long loops: skip a node when its work is already done.

Long autonomous runs re-run the same steps after a crash, a resume, or another loop iteration. Expensive, deterministic work (a fetch, a build, an LLM call you do not want to pay for twice) should run once per distinct input. This is a recipe over the :class:~agentflow.store.Store you already have — not a new runtime feature: hash the node's input, look for a cached result under that hash, and store the fresh result keyed by it.

store = MemoryStore()

async def expensive(state, ctx):
    return {"result": await do_costly_thing(state["input"])}

node = skip_if_done(
    expensive, store,
    namespace=("cache", "expensive"),
    key_from=lambda state: artifact_key(state["input"]),
)
g.add_node("expensive", node)

On the first visit the node runs and its update is cached; on any later visit with the same key the cached update is returned without running the node. The cache is a Store, so it can be in-memory for a single process or Postgres for a shared, durable one, and it can expire with a ttl.

Only wrap deterministic work whose result depends solely on the hashed input. A node that must run every time (it has side effects you always want, or reads live state the key does not capture) is a poor fit.

artifact_key 

artifact_key(value: Any) -> str

Return a stable content hash for a JSON-serializable value.

The value is serialized with sorted keys so equal content always hashes the same, then hashed with SHA-256. Use it to key a cache by what the input is rather than when it was produced. Raises TypeError if value is not JSON-serializable — hash something you can round-trip, not a live object.

Source code in agentflow/prebuilt/idempotent.py 
60
61
62
63
64
65
66
67
68
69
70
def artifact_key(value: Any) -> str:
    """Return a stable content hash for a JSON-serializable ``value``.

    The value is serialized with sorted keys so equal content always hashes the
    same, then hashed with SHA-256. Use it to key a cache by *what* the input
    is rather than when it was produced. Raises ``TypeError`` if ``value`` is
    not JSON-serializable — hash something you can round-trip, not a live
    object.
    """
    encoded = json.dumps(value, sort_keys=True, ensure_ascii=False, default=str)
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()

skip_if_done 

skip_if_done(node: Node, store: Store, *, namespace: Namespace, key_from: Callable[[Mapping[str, Any]], str], ttl: float | None = None) -> Node

Wrap node to run once per distinct key, caching its update in store.

key_from(state) computes the cache key from the node's input — pair it with :func:artifact_key to key by content hash. namespace scopes the cache in the store. ttl optionally expires a cached entry after that many seconds.

The cached value is the node's update dict. A node returning None is cached as an empty update, so a "did nothing" result still short-circuits. The wrapper is a plain (state, ctx) -> update node, transparent to the engine.

Source code in agentflow/prebuilt/idempotent.py 
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def skip_if_done(
    node: Node,
    store: Store,
    *,
    namespace: Namespace,
    key_from: Callable[[Mapping[str, Any]], str],
    ttl: float | None = None,
) -> Node:
    """Wrap ``node`` to run once per distinct key, caching its update in ``store``.

    ``key_from(state)`` computes the cache key from the node's input — pair it
    with :func:`artifact_key` to key by content hash. ``namespace`` scopes the
    cache in the store. ``ttl`` optionally expires a cached entry after that
    many seconds.

    The cached value is the node's update dict. A node returning ``None`` is
    cached as an empty update, so a "did nothing" result still short-circuits.
    The wrapper is a plain ``(state, ctx) -> update`` node, transparent to the
    engine.
    """

    async def wrapped(state: Mapping[str, Any], ctx: Any) -> Mapping[str, Any] | None:
        key = key_from(state)
        cached = await store.get(namespace, key)
        if cached is not None:
            return cached.value

        update = await node(state, ctx)
        await store.put(namespace, key, dict(update) if update else {}, ttl=ttl)
        return update

    wrapped.__name__ = f"skip_if_done({getattr(node, '__name__', 'node')})"
    return wrapped

agentflow.prebuilt.effects

Exactly-once effects in an at-least-once world.

A super-step is atomic over state: if any node in the step raises, the step's state updates are discarded and no checkpoint is written. That transaction covers the checkpointer and nothing else. A side effect a node already performed — a file written, an HTTP POST sent, a row inserted in someone else's database — is not rolled back, because the engine has no transaction over it. So on a resume the node runs again and the effect can happen twice. This is not a bug to fix; it is the delivery contract of every durable executor without distributed transactions: at-least-once execution, and exactly-once only if the effect itself is idempotent.

IdempotentOp is the primitive that makes that idempotency explicit. It wraps one specific effect inside a node (not the whole node — a node usually does an effect plus other work) with a claim -> run -> commit marker in a :class:~agentflow.store.Store:

op = IdempotentOp(store, ("effects", "send_email"))

async def notify(state, ctx):
    async def send():
        return await email_api.send(state["to"], state["body"],
                                    idempotency_key=key)  # <- pass it downstream!
    key = state["message_id"]
    receipt = await op.run(key, send, on_incomplete="error")
    return {"receipt": receipt}

State machine per key, all in a single store entry so a plain get reads it in one shot:

  • absent -> write in_flight, run fn, write done(result), return it.
  • done -> return the stored result without running fn again.
  • in_flight -> a previous attempt did not reach done. What to do is a policy you choose per effect (on_incomplete), because the SDK cannot know whether the effect actually completed.

Two windows this does NOT close

First, the crash gap. There is an irreducible gap between "the effect happened" and "the marker says done". If the process dies inside fn — the POST left, the file is half-written — no done marker exists and the next run sees in_flight; the SDK cannot know if the effect landed.

Second, the lost-response case, which happens even without a crash: fn performs the effect and then raises (a TimeoutError reading the response of a POST that already succeeded). An exception does not prove the effect did not happen. on_error controls the marker here — "keep" (default) preserves in_flight so the next attempt decides via on_incomplete; "release" drops it and the retry re-runs the effect, which duplicates it unless the effect is idempotent. So on_incomplete="error" protects the crash gap, but with on_error="release" it does not protect the lost-response case — the marker is already gone.

The real fix for both lives in the effect itself, keyed idempotency downstream: pass the same key as an idempotency token to the API, write to a temp file at a stable destination path with deterministic content and atomically rename, use INSERT ... ON CONFLICT. Use the key this primitive gives you as that token — that is what actually collapses duplicates.

Concurrency: an atomic claim

The claim is an atomic conditional write — :meth:Store.put(..., if_absent=True) <agentflow.store.Store.put> — not a plain get-then-put. Two workers racing the same key give exactly one winner (who runs the effect) and one :class:~agentflow.errors.StoreConflict (who re-reads and defers to the winner's marker). So the "both saw absent and both ran" race is closed, as far as the store can see it: MemoryStore is atomic within a process, and PostgresStore is atomic across processes via a single conditional INSERT.

That still does not make the effect exactly-once — the winner can crash or lose its response inside fn (the windows above). The atomic claim removes the duplicate-starter race; keyed idempotency downstream removes the rest.

The claim's single-runner guarantee also applies only to that first attempt. on_incomplete="rerun" (see :meth:IdempotentOp.run) overwrites the marker unconditionally, so a rerun can execute fn concurrently with a winner that is still running. Reserve "rerun" for idempotent effects that deduplicate downstream by the key.

IdempotentOp vs skip_if_done

:func:~agentflow.prebuilt.skip_if_done wraps a whole node and caches its update to avoid recomputing deterministic work; its contract is "recomputing would be wasteful", and re-running on a lost cache is harmless. IdempotentOp guards a side effect, where re-running is not harmless, and forces you to handle the interrupted-in-flight case. Reach for skip_if_done to save work; reach for IdempotentOp to protect an effect.

IdempotentOp

Run a side effect at most once per key, with an explicit interrupted case.

store holds the claim/commit markers; namespace scopes them (one namespace per logical effect reads well, e.g. ("effects", "send_email")). Markers can expire with default_ttl — set it when a stale in_flight from a long-dead process should eventually be reclaimable.

Source code in agentflow/prebuilt/effects.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
class IdempotentOp:
    """Run a side effect at most once per key, with an explicit interrupted case.

    ``store`` holds the claim/commit markers; ``namespace`` scopes them (one
    namespace per logical effect reads well, e.g. ``("effects", "send_email")``).
    Markers can expire with ``default_ttl`` — set it when a stale ``in_flight``
    from a long-dead process should eventually be reclaimable.
    """

    def __init__(
        self,
        store: Store,
        namespace: Namespace,
        *,
        default_ttl: float | None = None,
    ) -> None:
        self._store = store
        self._namespace = namespace
        self._default_ttl = default_ttl

    async def run(
        self,
        key: str,
        fn: Callable[[], Awaitable[T]],
        *,
        on_incomplete: OnIncomplete = "error",
        on_error: OnError = "keep",
        ttl: float | None = None,
    ) -> T:
        """Run ``fn`` unless this ``key`` already completed; return its result.

        - Completed before: returns the stored result, ``fn`` is not called.
        - Never started: writes an ``in_flight`` marker, awaits ``fn``, stores
          the result as ``done``, returns it.
        - Left ``in_flight`` by a prior attempt: dispatch on ``on_incomplete`` —
          ``"error"`` raises :class:`IncompleteEffectError` (safe default);
          ``"rerun"`` runs ``fn`` again (only when the effect is idempotent);
          ``"skip"`` assumes it completed and returns ``None``.

        Note on ``"rerun"`` and concurrency: the atomic claim (see the class
        docstring) only guarantees a single runner for the *first* attempt.
        ``"rerun"`` overwrites the existing ``in_flight`` marker
        unconditionally, so if the marker belongs to a winner that is *still
        running* (not a dead attempt), the rerun runs ``fn`` concurrently with
        it. Two simultaneous executions are the result. That is why ``"rerun"``
        is only for a genuinely idempotent effect whose safety rests on
        downstream deduplication by the key, not on this primitive's claim.

        If ``fn`` raises, ``on_error`` decides what the marker is left as, and
        this is a genuine choice you must make — an exception does **not** prove
        the effect did not happen. A `TimeoutError` reading the response of a
        `POST` that already succeeded is an exception *after* the effect landed.

        - ``"keep"`` (default): leave the ``in_flight`` marker, so the next
          attempt hits ``on_incomplete`` and decides deliberately. This is the
          safe default for a non-idempotent effect, because it does not assume
          the effect was skipped.
        - ``"release"``: remove the marker so the next attempt starts from
          ``absent`` and re-runs ``fn`` on the clean path. Correct **only** when
          the exception reliably means the effect did not happen (e.g. a
          connect error before any request was sent). With ``"release"``,
          ``on_incomplete="error"`` cannot protect you from an
          effect-with-a-lost-response — the marker is gone, so the retry just
          runs the effect again. Use ``"release"`` only with downstream
          idempotency (see the class docstring).

        In both cases the exception propagates after the marker is handled.

        The stored result must be JSON-serializable (it round-trips through the
        store). Return an id or receipt, not a live object.
        """
        effective_ttl = ttl if ttl is not None else self._default_ttl
        existing = await self._store.get(self._namespace, key)

        if existing is not None:
            resolved = self._resolve(existing, key, on_incomplete)
            if resolved is not _RUN:
                return resolved  # type: ignore[return-value]
            return await self._handle_incomplete(key, fn, on_incomplete, on_error, effective_ttl)

        # Absent: attempt an atomic claim. If a racing writer claimed first, the
        # conditional write raises StoreConflict; re-read and dispatch on what
        # they left (in_flight or done) instead of running the effect too.
        try:
            return await self._claim_run_commit(key, fn, on_error, effective_ttl, conditional=True)
        except StoreConflict:
            current = await self._store.get(self._namespace, key)
            if current is None:
                # The winner released/expired between our conflict and re-read;
                # treat as a fresh interrupted claim for the policy to decide.
                raise IncompleteEffectError(self._namespace, key) from None
            resolved = self._resolve(current, key, on_incomplete)
            if resolved is not _RUN:
                return resolved  # type: ignore[return-value]
            return await self._handle_incomplete(key, fn, on_incomplete, on_error, effective_ttl)

    def _resolve(self, item: Any, key: str, on_incomplete: OnIncomplete) -> Any:
        """Map a stored marker to a return value, or _RUN to signal 'act'.

        ``done`` -> the stored result; ``in_flight`` -> _RUN so the caller runs
        the ``on_incomplete`` path; anything else -> _RUN (treat as absent).
        """
        record = item.value
        status = record.get("status") if isinstance(record, dict) else None
        if status == _DONE:
            return record.get("result")
        return _RUN

    async def _handle_incomplete(
        self,
        key: str,
        fn: Callable[[], Awaitable[T]],
        on_incomplete: OnIncomplete,
        on_error: OnError,
        ttl: float | None,
    ) -> T:
        if on_incomplete == "error":
            raise IncompleteEffectError(self._namespace, key)
        if on_incomplete == "skip":
            return None  # type: ignore[return-value]
        # "rerun": the effect is declared idempotent. We already hold an
        # in_flight marker for this key, so overwrite it unconditionally rather
        # than trying to claim again (which would conflict with ourselves).
        return await self._claim_run_commit(key, fn, on_error, ttl, conditional=False)

    async def _claim_run_commit(
        self,
        key: str,
        fn: Callable[[], Awaitable[T]],
        on_error: OnError,
        ttl: float | None,
        *,
        conditional: bool,
    ) -> T:
        # Claim: mark in-flight before touching the outside world. When
        # conditional, this is an ATOMIC create (put if_absent) — two writers
        # racing the same key give one winner and one StoreConflict, so only one
        # runs the effect. The marker also makes an interrupted attempt visible
        # as in_flight on the next try.
        await self._store.put(
            self._namespace, key, {"status": _IN_FLIGHT}, ttl=ttl, if_absent=conditional
        )
        try:
            result = await fn()
        except BaseException:
            # fn raised, but that does NOT prove the effect did not happen (the
            # response may have been lost after the effect landed). on_error is
            # the deliberate choice: "release" drops the marker (retry re-runs
            # on the clean path — only safe if the exception means "not done");
            # "keep" leaves in_flight so the next attempt's on_incomplete
            # decides. Either way the exception propagates.
            if on_error == "release":
                await self._store.delete(self._namespace, key)
            raise
        # Commit: record the result so any later attempt short-circuits.
        await self._store.put(self._namespace, key, {"status": _DONE, "result": result}, ttl=ttl)
        return result

run async 

run(key: str, fn: Callable[[], Awaitable[T]], *, on_incomplete: OnIncomplete = 'error', on_error: OnError = 'keep', ttl: float | None = None) -> T

Run fn unless this key already completed; return its result.

  • Completed before: returns the stored result, fn is not called.
  • Never started: writes an in_flight marker, awaits fn, stores the result as done, returns it.
  • Left in_flight by a prior attempt: dispatch on on_incomplete — "error" raises :class:IncompleteEffectError (safe default); "rerun" runs fn again (only when the effect is idempotent); "skip" assumes it completed and returns None.

Note on "rerun" and concurrency: the atomic claim (see the class docstring) only guarantees a single runner for the first attempt. "rerun" overwrites the existing in_flight marker unconditionally, so if the marker belongs to a winner that is still running (not a dead attempt), the rerun runs fn concurrently with it. Two simultaneous executions are the result. That is why "rerun" is only for a genuinely idempotent effect whose safety rests on downstream deduplication by the key, not on this primitive's claim.

If fn raises, on_error decides what the marker is left as, and this is a genuine choice you must make — an exception does not prove the effect did not happen. A TimeoutError reading the response of a POST that already succeeded is an exception after the effect landed.

  • "keep" (default): leave the in_flight marker, so the next attempt hits on_incomplete and decides deliberately. This is the safe default for a non-idempotent effect, because it does not assume the effect was skipped.
  • "release": remove the marker so the next attempt starts from absent and re-runs fn on the clean path. Correct only when the exception reliably means the effect did not happen (e.g. a connect error before any request was sent). With "release", on_incomplete="error" cannot protect you from an effect-with-a-lost-response — the marker is gone, so the retry just runs the effect again. Use "release" only with downstream idempotency (see the class docstring).

In both cases the exception propagates after the marker is handled.

The stored result must be JSON-serializable (it round-trips through the store). Return an id or receipt, not a live object.

Source code in agentflow/prebuilt/effects.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
async def run(
    self,
    key: str,
    fn: Callable[[], Awaitable[T]],
    *,
    on_incomplete: OnIncomplete = "error",
    on_error: OnError = "keep",
    ttl: float | None = None,
) -> T:
    """Run ``fn`` unless this ``key`` already completed; return its result.

    - Completed before: returns the stored result, ``fn`` is not called.
    - Never started: writes an ``in_flight`` marker, awaits ``fn``, stores
      the result as ``done``, returns it.
    - Left ``in_flight`` by a prior attempt: dispatch on ``on_incomplete`` —
      ``"error"`` raises :class:`IncompleteEffectError` (safe default);
      ``"rerun"`` runs ``fn`` again (only when the effect is idempotent);
      ``"skip"`` assumes it completed and returns ``None``.

    Note on ``"rerun"`` and concurrency: the atomic claim (see the class
    docstring) only guarantees a single runner for the *first* attempt.
    ``"rerun"`` overwrites the existing ``in_flight`` marker
    unconditionally, so if the marker belongs to a winner that is *still
    running* (not a dead attempt), the rerun runs ``fn`` concurrently with
    it. Two simultaneous executions are the result. That is why ``"rerun"``
    is only for a genuinely idempotent effect whose safety rests on
    downstream deduplication by the key, not on this primitive's claim.

    If ``fn`` raises, ``on_error`` decides what the marker is left as, and
    this is a genuine choice you must make — an exception does **not** prove
    the effect did not happen. A `TimeoutError` reading the response of a
    `POST` that already succeeded is an exception *after* the effect landed.

    - ``"keep"`` (default): leave the ``in_flight`` marker, so the next
      attempt hits ``on_incomplete`` and decides deliberately. This is the
      safe default for a non-idempotent effect, because it does not assume
      the effect was skipped.
    - ``"release"``: remove the marker so the next attempt starts from
      ``absent`` and re-runs ``fn`` on the clean path. Correct **only** when
      the exception reliably means the effect did not happen (e.g. a
      connect error before any request was sent). With ``"release"``,
      ``on_incomplete="error"`` cannot protect you from an
      effect-with-a-lost-response — the marker is gone, so the retry just
      runs the effect again. Use ``"release"`` only with downstream
      idempotency (see the class docstring).

    In both cases the exception propagates after the marker is handled.

    The stored result must be JSON-serializable (it round-trips through the
    store). Return an id or receipt, not a live object.
    """
    effective_ttl = ttl if ttl is not None else self._default_ttl
    existing = await self._store.get(self._namespace, key)

    if existing is not None:
        resolved = self._resolve(existing, key, on_incomplete)
        if resolved is not _RUN:
            return resolved  # type: ignore[return-value]
        return await self._handle_incomplete(key, fn, on_incomplete, on_error, effective_ttl)

    # Absent: attempt an atomic claim. If a racing writer claimed first, the
    # conditional write raises StoreConflict; re-read and dispatch on what
    # they left (in_flight or done) instead of running the effect too.
    try:
        return await self._claim_run_commit(key, fn, on_error, effective_ttl, conditional=True)
    except StoreConflict:
        current = await self._store.get(self._namespace, key)
        if current is None:
            # The winner released/expired between our conflict and re-read;
            # treat as a fresh interrupted claim for the policy to decide.
            raise IncompleteEffectError(self._namespace, key) from None
        resolved = self._resolve(current, key, on_incomplete)
        if resolved is not _RUN:
            return resolved  # type: ignore[return-value]
        return await self._handle_incomplete(key, fn, on_incomplete, on_error, effective_ttl)

IncompleteEffectError

Bases: AgentFlowError

A prior attempt at an effect died between claim and commit.

Raised by :meth:IdempotentOp.run when it finds an in_flight marker and on_incomplete="error" — the safe default for a non-idempotent effect, where blindly re-running could duplicate it. The run suspends the decision to a human or a higher-level policy rather than guessing. key and namespace identify the stuck effect.

Source code in agentflow/prebuilt/effects.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
class IncompleteEffectError(AgentFlowError):
    """A prior attempt at an effect died between claim and commit.

    Raised by :meth:`IdempotentOp.run` when it finds an ``in_flight`` marker and
    ``on_incomplete="error"`` — the safe default for a non-idempotent effect,
    where blindly re-running could duplicate it. The run suspends the decision
    to a human or a higher-level policy rather than guessing. ``key`` and
    ``namespace`` identify the stuck effect.
    """

    def __init__(self, namespace: Namespace, key: str):
        self.namespace = namespace
        self.key = key
        super().__init__(
            f"effect {key!r} under {namespace} was left in-flight by a prior attempt; "
            f"cannot tell if it completed. Choose on_incomplete='rerun' if the effect "
            f"is idempotent, or resolve it by hand."
        )

effect_key 

effect_key(*parts: Any) -> str

Build a stable key for an effect from any JSON-serializable parts.

Serializes with sorted keys and hashes with SHA-256, so the same logical effect always maps to the same key regardless of dict ordering. Feed it the things that identify the effect (a message id, a target path, the payload) — and pass the returned key to the downstream system as its idempotency token so duplicates collapse there too.

Source code in agentflow/prebuilt/effects.py 
148
149
150
151
152
153
154
155
156
157
158
def effect_key(*parts: Any) -> str:
    """Build a stable key for an effect from any JSON-serializable parts.

    Serializes with sorted keys and hashes with SHA-256, so the same logical
    effect always maps to the same key regardless of dict ordering. Feed it the
    things that identify the effect (a message id, a target path, the payload) —
    and pass the returned key to the downstream system as its idempotency token
    so duplicates collapse there too.
    """
    encoded = json.dumps(parts, sort_keys=True, ensure_ascii=False, default=str)
    return hashlib.sha256(encoded.encode("utf-8")).hexdigest()