LlamaIndex Workflows — Deep Dive

Think of a workflow like a factory assembly line: each worker station receives a specific type of package, does a task on it, and then sends it to the next station. The purpose is to turn a messy, long AI task into a clear series of small steps that automatically trigger each other. In practice, you mark each station using a special Python label called a @step decorator. The first station receives a StartEvent package (the initial request), and the last station hands back a StopEvent package (the final answer). Each station’s input and output package types are like blueprints that define which station connects to which. Before the line ever runs, the system automatically checks that every package produced is actually received by some later station and that no station waits for a package that nobody sends—this validation catches wiring mistakes early. The trickiest part is that sometimes a station might send packages dynamically on the fly rather than only through its blueprint. For example, a station can call ctx.send_event to shoot out packages during a loop. A static check would flag those as missing because it can’t see the dynamic sends. To handle this, you can add skip_graph_checks=["reachability"] to the station’s label, telling the system to trust the dynamic behavior. Without this subsystem, a beginner trying to build loops or branches would face validation errors complaining about “unreachable” steps—a confusing failure that stops the whole design cold.

A LlamaIndex Workflow is built from steps that receive and return events, with the @step decorator.

class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4.1")

    @step
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        topic = ev.topic
        prompt = f"Write your best joke about {topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

    @step
    async def critique_joke(self, ev: JokeEvent) -> StopEvent:
        joke = ev.joke
        prompt = f"Give a thorough analysis and critique of the following joke: {joke}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))

In a LlamaIndex workflow, execution proceeds through an ordered sequence of steps connected by events. The mechanism starts when .run() is called: keyword arguments become fields of a StartEvent, which is automatically emitted to the first step marked with the @step decorator. Each step receives an event, performs logic in plain Python, and returns a new event; that returned event triggers the next step whose type annotation accepts it. This continues until a step returns a StopEvent, which immediately terminates the workflow and returns the value in its result parameter. On failure—for example, if a step raises an unhandled exception—the workflow halts and the exception propagates to the caller of .run(). The framework also allows early validation: calling workflow.validate() before the first run checks that every produced event has a consumer and every consumed event has a producer, catching design problems such as orphaned events or missing terminal steps.

The central invariant the design preserves is the consumer–producer guarantee enforced by static validation. Before execution, the @step decorator infers the input and output event types from the method’s type annotations; the workflow then verifies that for every event type emitted by a step, there exists exactly one step (or multiple, via branching) that accepts it, and that every step’s input event is produced by some other step. This guarantees that no event is lost and no step is starved, ensuring the workflow’s edges are sound without relying on runtime checks. If a step intentionally dispatches events dynamically via ctx.send_event, the static check can be skipped with skip_graph_checks=["reachability"] on that step, but the default invariant remains strict graph connectivity.

The key trade-off is between type‑first static validation and fully dynamic flexibility. The design explicitly rejects the previous DAG pattern used by other frameworks, where loops and branches had to be encoded into graph edges—making them hard to read and understand—and where passing data between nodes created complexity around optional values and parameter passing. By adopting an event‑driven, typed‑step model, the workflow system forces the developer to declare event types explicitly, which enables automatic validation and eliminates the “complexity hidden in edges” cost. The alternative rejected is an entirely untyped, dynamic graph that offers flexibility at the cost of runtime surprises and difficult debugging. This trade‑off pays off because the static checks surface errors before a single LLM call is made, and the plain Python logic inside steps remains natural to write and read.

A concrete failure mode occurs when a step produces an event that no other step consumes. For example, a step generate_joke returns JokeEvent, but no step in the workflow has a type annotation accepting JokeEvent. The signal an operator would see is a validation error when calling workflow.validate()—the framework raises an exception stating that the event has no consumer. If validation is skipped, the workflow would simply hang at runtime because the event is never handled, and the operator would observe a timeout (the timeout parameter passed to Workflow constructor, e.g., timeout=60, would eventually raise an asyncio.TimeoutError). This explicit failure signal, combined with the static check, allows operators to catch missing edges long before production deployment.

A step returns an event type that has no consumer

• Trigger: A developer writes a step whose return annotation is an event type (e.g., MyCustomEvent) that no other step in the workflow accepts as input.
• Guard: Workflow.validate() — the validation routine that “checks that every produced event has a consumer”. If validation is run before execution, the workflow aborts immediately.
• Posture: fail-hard — validation prevents the workflow from starting.
• Operator signal: A validation error message, such as “A step produces an event that nobody consumes”. The error identifies the unconnected event type.
• Recovery: The developer must either add a step that consumes that event, change the return type to a type that is consumed, or, if the event is intentionally unreachable, apply skip_graph_checks=["reachability"] to the step and re‑run validation.

The workflow has no terminal event

• Trigger: All steps in the workflow return custom event types (e.g., JokeEvent, AnotherEvent) and none returns StopEvent or a subclass of StopEvent.
• Guard: Workflow.validate() — the same validation routine that checks “the workflow has [a] terminal event”. This failure is caught statically.
• Posture: fail-hard — validation fails, and the workflow does not execute.
• Operator signal: A validation error, specifically “The workflow has no terminal event”.
• Recovery: Add a step that returns StopEvent (or a custom StopEvent subclass) to end the workflow. The developer can then re‑run validation.

A step consumes an event that is never produced (static mismatch)

• Trigger: A step’s input annotation declares an event type that is never returned by any other step. This often happens when the step is intended to receive an event sent dynamically via ctx.send_event, but static analysis does not see a producer.
• Guard: Workflow.validate() — it checks that “every consumed event has a producer”. To bypass the static check, the developer can add skip_graph_checks=["reachability"] to the step decorator.
• Posture: fail-hard if validation is run without skip_graph_checks; the workflow is aborted before start. If validation is skipped or skip_graph_checks is used, the workflow may run but the step never receives the event, leading to a deadlock.
• Operator signal: A validation error: “A step consumes an event that is never produced”. At runtime, if no guard is in place, the operator sees the workflow hang until the configurable timeout is reached.
• Recovery: Either add a step that produces the missing event statically, or add skip_graph_checks=["reachability"] and verify that the event is indeed sent via ctx.send_event at the correct time. If a timeout occurs, increase timeout or fix the missing send.

A dynamic event sent via ctx.send_event has no consumer at runtime

• Trigger: A step uses ctx.send_event(...) to emit an event type that no step in the workflow is annotated to accept. Because the event is sent dynamically, static validation does not necessarily catch it (or it may be bypassed).
• Guard: None shown in the source. The source only notes that “a step produces an event that nobody consumes” is a symptom but does not describe any runtime handler for unhandled dynamic events.
• Posture: fail-soft — the event is lost (no handler processes it), and the workflow continues with the rest of its steps. Downstream operations that depend on that event will never execute.
• Operator signal: Silent absence of expected work. If a step uses ctx.collect_events to wait for that event, the collection may never complete, resulting in a timeout. Otherwise, the workflow completes but with missing results.
• Recovery: The developer must add a step that accepts the event type, or correct the ctx.send_event call to emit a handled event. Manual inspection of the event graph is required.

A step returns a list of events but the downstream consumer expects a single event

• Trigger: A step is annotated to return list[SomeEvent] (e.g., -> list[Retrieved]), but the next step declares its input as a single event type SomeEvent instead of list[SomeEvent].
• Guard: Workflow.validate() — the validation routine checks that “produced events have consumers”. A type mismatch between list[SomeEvent] and SomeEvent will cause validation to flag that the produced event (the list type) is not consumed by any step, or that the consumer’s input type does not match.
• Posture: fail-hard — validation fails, preventing execution.
• Operator signal: A validation error similar to “A step produces an event that nobody consumes”, or an explicit type‑mismatch error (the source does not give the exact message, but the symptom matches).
• Recovery: Change the downstream step’s input to accept list[SomeEvent] (e.g., ev: list[SomeEvent]), or have the producing step emit individual events one at a time using ctx.send_event so that the consumer can accept a single event.

Imagine a relay race: each runner waits for a specific colored baton, runs their leg, then hands off a new baton to the next runner. A coach checks that every baton color matches the next runner’s hand before the race starts. That is exactly what steps and events do in a workflow—they define the sequence of tasks, each triggered by a typed message (the baton), so work flows correctly in order.

Each step is a Python method marked with a @step decorator. The decorator reads the method’s type annotations—the first parameter tells what event the step expects, and the return annotation tells what event it produces. For example, a step might take a SearchEvent and return a StopEvent. The workflow then builds a map of these connections automatically. Custom events are defined as Pydantic objects with typed fields, like JokeEvent(joke: str). The race begins when you call run(), which creates a StartEvent that holds your keyword arguments. It ends when a step returns a StopEvent, which carries the result. For branching, a step’s return annotation uses the pipe symbol, like LoopEvent | StopEvent, so the workflow knows which path to follow based on the actual event returned.

The trickiest part is that the decorator infers everything from the method signature alone—no separate wiring file. This enables static validation before the race: the workflow checks that every event type is consumed by some step and that no step returns an event nobody handles (unless you mark it with skip_graph_checks=["reachability"] for intentionally dynamic steps). Without this subsystem, you would have to manually connect every step, likely mismatching baton colors—a step would wait for an event that never arrives, or the workflow would never produce a StopEvent, leaving your program stuck forever. A beginner would feel that as silence: the workflow hangs or crashes, and they have no idea which step missed its baton.

Steps, events, and wiring are shown in this two-step workflow using custom events, StartEvent, and StopEvent.

class JokeEvent(Event):
    joke: str

class JokeFlow(Workflow):
    llm = OpenAI(model="gpt-4.1")

    @step
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        prompt = f"Write your best joke about {ev.topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

    @step
    async def critique_joke(self, ev: JokeEvent) -> StopEvent:
        prompt = f"Give a thorough analysis and critique of the following joke: {ev.joke}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))

The ordered mechanism of the Steps Events And Routing subsystem begins when the user calls workflow.run() with keyword arguments. Those arguments are automatically wrapped into a special StartEvent, which is emitted to start execution. The first step decorated with @step receives that StartEvent and processes it, returning a custom Event (a Pydantic object). The framework then forwards that event to the next step whose input type annotation matches the returned event type. Steps can branch by returning one of several event types, written with the pipe symbol (e.g., LoopEvent | StopEvent). If a step returns a StopEvent, the workflow terminates and returns the result attribute. On failure—for example, when a step produces an event that no other step consumes, or when a branch is unfinished—the static graph validation catches the issue before execution. The mechanism relies entirely on the @step decorator reading the method’s type annotations to infer input and output events, and the graph is traversed purely by event type matching.

The invariant the design preserves is that the workflow is valid before running. The source explicitly states that the inferred input and output types “are also used to verify for you that the workflow is valid before running!” This guarantee is enforced by the workflow.validate() method, which can be called directly in tests or startup code. The validation checks that every produced event has a consuming step, that no step expects an event that is never produced, and that a terminal StopEvent is reachable from all branches. If a step is intentionally unreachable from static analysis, the developer can opt out of a specific check using @step(skip_graph_checks=["reachability"]). This ensures that the event-routing graph is sound before any LLM call or I/O occurs.

The key trade-off is that the event-driven, plain-Python approach rejects the directed-acyclic-graph (DAG) pattern used by earlier frameworks and LlamaIndex itself. The source lists three pain points with DAGs: logic like loops and branches had to be encoded into edges, making them hard to read; passing data between nodes created complexity around optional and default parameters; and DAGs did not feel natural to developers building complex, looping, branching AI applications. By using ordinary if statements for branches, returning events for loops, and Python type annotations for routing, Workflows avoid those costs entirely. The cost of this rejection is the need for static graph validation, which can be overly strict for truly dynamic flows (hence the skip_graph_checks escape hatch), but the trade-off is a drastically simpler developer experience that mirrors natural control flow.

A concrete failure mode is when a step returns an event type that no other step is annotated to accept. The operator would see a validation error from workflow.validate() (or from the implicit validation on run()) indicating that the event is unconsumed. For instance, if a SearchEvent is returned but no step has @step with an input of SearchEvent, the validation fails with a message about an unreachable event. The exact signal is a raised exception from the workflow’s graph-checking logic, referencing the unmatched event type. An operator debugging the system would also notice the workflow never reaches a StopEvent and instead throws an error before any step executes, making the problem immediately visible during development or testing.

An event produced by a step that no other step consumes

• Trigger – A step decorated with @step returns a custom Event subclass, but no other step in the same Workflow accepts that event type in its input annotation.
• Guard – The pre-run validation built into the @step decorator; specifically, workflow.validate() (called automatically before run() or manually) raises an error.
• Posture – Fail‑hard – the workflow never starts; an exception is raised before any step executes.
• Operator signal – A validation error message similar to “A step produces an event that nobody consumes” (from the source’s table of common failures).
• Recovery – Add a step whose input annotation matches the produced event, or change the producer step’s output to an event that a downstream step already accepts. Retry by re‑running after the fix.

The workflow contains no reachable terminal event

• Trigger – No step in the Workflow returns StopEvent or a subclass of StopEvent under any reachable branch.
• Guard – The same validation path (workflow.validate()) checks that at least one step can emit a terminal event.
• Posture – Fail‑hard – the workflow is rejected before execution because it cannot end.
• Operator signal – A validation warning or error stating “The workflow has no terminal event” (from the source’s table).
• Recovery – Add a step that returns StopEvent (or a custom subclass) and ensure it is reachable from the start. Re‑run after editing.

Duplicate step name when using unbound functions

• Trigger – A free function decorated with @step(workflow=SomeWorkflow) uses a name that already exists as a method on the workflow class or as another registered free‑function step.
• Guard – The registration logic inside @step raises a validation error at import time.
• Posture – Fail‑hard – the Workflow class cannot be assembled; an exception is raised during module import.
• Operator signal – A ValueError or similar validation error stating that the step name is duplicate.
• Recovery – Rename the offending step or the existing method so all step names are unique. Re‑import or reload the module.

Missing required keyword argument in StartEvent

• Trigger – The caller invokes workflow.run() without supplying a keyword argument that the first step accesses directly, e.g., ev.topic.
• Guard – The step can optionally use ev.get('topic') instead of direct attribute access, as noted in the source: “You could also do ev.get('topic') to handle the case where the attribute might not be there without raising an error.”
• Posture – If the step uses direct access (ev.topic): fail‑hard (AttributeError). If the step uses ev.get('topic'): fail‑soft (returns None and continues).
• Operator signal – In the fail‑hard case, an AttributeError traceback pointing to ev.topic. In the fail‑soft case, no error – the step proceeds with None silently.
• Recovery – For fail‑hard: call run() with the missing keyword. For fail‑soft: the step should check for None and either fall back to a default or raise its own meaningful error. No automatic retry.

A step is unreachable because of a mis‑typed branch

• Trigger – A branch step returns one of several event types using the pipe annotation, but the condition logic never leads to one of those events, or the event type produced does not match the annotation (e.g., a runtime value that is not among the declared types).
• Guard – Static validation via workflow.validate() catches mismatches between annotated output and actual return types only when the mismatch is statically detectable; dynamic mismatches are not caught upfront. There is no built‑in retry or fallback for this case.
• Posture – Static mismatch: fail‑hard (validation error). Dynamic mismatch at runtime: fail‑hard – the workflow halts when the event type is unexpected and no downstream step consumes it, or an unexpected event is silently ignored (fail‑soft if the event is never consumed, but the workflow may hang because no terminal step is reached).
• Operator signal – For static mismatches: validation error referencing the branch annotation. For dynamic mismatches: either a TypeError when the event is sent to a step that doesn’t accept it, or a silent hang because no step fires.
• Recovery – Correct the branch logic or the output annotation. Re‑run after the fix. No automatic recovery.

Think of a workflow like a team of people working on a multi-step project. Each person does a small task and then passes the work to the next person. But sometimes people need to remember a running total or a shared note without passing it along in every handoff. That's what the context is for: it's a shared whiteboard where anyone can write a number, read it back, or update it during the whole project. Every time you start a project (a "run"), you get a fresh whiteboard, so different projects don't mix up their numbers.

Now let's look at how the whiteboard actually works. Someone can write a value with ctx.store.set("count", 5) and later pull it out with ctx.store.get("count"). If two people try to update the same number at the same time, the whiteboard has a special guard called edit_state() that locks it so only one person writes at a time—like putting a "do not disturb" sign on the whiteboard. But the whiteboard is only for simple stuff like numbers or short text. Heavy equipment like a database connection or an AI model doesn't go on the whiteboard; those go into special bins called resources, which are built once at the start and handed to each person (step) automatically.

The trickiest part is that the whiteboard is meant to be saved for later—like taking a photo of it so you can continue the project tomorrow. That means everything on it must be "JSON-friendly": numbers, text, lists, and plain dictionaries. If you accidentally try to store a live connection or a huge AI index there, it will break when you try to save the photo. The source even warns: "State is not a place for heavyweight clients… Put those in resources." Without the context whiteboard, each step would have to pass every shared value through event messages—cluttering the handoffs and making it impossible to resume a project where you left off. You'd either lose the running count or have to rebuild it from scratch every time.

Steps share state through ctx.store using get and set.

from workflows import Workflow, Context, step
from workflows.events import StartEvent, StopEvent

class MyWorkflow(Workflow):

    @step
    async def my_step(self, ctx: Context, ev: StartEvent) -> StopEvent:
        current_count = await ctx.store.get("count", default=0)
        current_count += 1
        await ctx.store.set("count", current_count)
        return StopEvent(result=current_count)

In this subsystem, state sharing across workflow steps is mediated by the Context object, which holds a store accessible as ctx.store. The ordered mechanism begins when a workflow run is launched; each run creates its own Context. Steps read and write values using await ctx.store.get("count", default=0) and await ctx.store.set("count", current_count + 1). For atomic, multi‑step updates, the mechanism prescribes an async with ctx.store.edit_state() as state: block. While inside that block, no other step can edit the state, enforcing a sequential write boundary. After the block exits, the updated state is visible to subsequent steps. If a failure occurs inside the edit_state block, the state remains unchanged, and the exception propagates to the caller. For cross‑run state, you explicitly create a Context(workflow) and pass it to .run(ctx=ctx). The context’s state can be serialized via ctx.to_dict() and restored via Context.from_dict(workflow, ctx_dict), enabling snapshots and durable resumes.

The central invariant this design preserves is atomicity of state updates. The edit_state() method guarantees that the read‑modify‑write cycle of the state is isolated from concurrent step execution. Without this lock, two steps running in parallel could both read the same value, increment it, and write back, losing one update. The design also enforces a serializability constraint: any value stored in ctx.store must be JSON‑serializable, because it may be serialized during durable snapshots. Heavyweight, non‑serializable objects such as LLM clients, indexes, or database handles are expressly forbidden from the store.

The key trade‑off is rejecting the obvious alternative of placing runtime dependencies—like LLM clients or database connections—directly in the shared state store. That alternative would simplify step access, because every step could just read the client from ctx.store. However, it would make durable snapshots impossible: serializing a live database handle or an LLM client would likely produce a TypeError or corrupt the snapshot. The cost avoided is the need to implement custom serializers for every non‑serializable object, which would be brittle and error‑prone. Instead, the design pushes those dependencies into resources, which are injected via factory functions (e.g., Resource(get_llm)) and are never serialized. The price of this separation is the extra wiring required: steps must declare dependencies using Annotated annotations, and resource factories must be chained explicitly.

One concrete failure mode arises when a developer mistakenly stores a non‑serializable object in the state—for instance, after a step does await ctx.store.set("db", database_connection). Later, when the workflow snapshot is taken (either automatically for durability or manually via ctx.to_dict()), the Python serialization call will throw an exception. The signal an operator would see is a TypeError or JSONEncodeError with a message similar to "Object of type DatabaseConnection is not JSON serializable". The traceback would point to the to_dict call or the durable snapshot mechanism. To recover, the operator must remove the offending value from the store and relocate it to a proper Resource pattern, such as defining a get_db factory and annotating the step with Resource(get_db).

1. Concurrent state update without locking

• Trigger — Two or more steps run simultaneously and each calls ctx.store.get("count") and ctx.store.set("count", …) without using ctx.store.edit_state(). The reads and writes interleave, causing lost updates.
• Guard — No guard in the source for this scenario; the ctx.store.edit_state() context manager is the designed guard, but it must be explicitly used by the developer. Without it, there is no protection.
• Posture — fail‑soft. The workflow continues without raising an exception, but the final value in the store is incorrect (e.g., an increment of 1 happens only once instead of twice). The overall run may produce wrong results while appearing successful.
• Operator signal — Silent absence of any error; the operator would observe an unexpected final value (e.g., a counter that should be 2 is 1). No log line or metric is automatically emitted.
• Recovery — No automatic recovery. The developer must identify the race condition and wrap concurrent read‑modify‑write operations inside async with ctx.store.edit_state() as state:.

2. Unserializable object stored in state

• Trigger — A developer puts a heavyweight, non‑JSON‑serializable object (e.g., an LLM client, a database handle, or a file descriptor) into ctx.store using ctx.store.set("client", llm_client) instead of placing it in a Resource factory.
• Guard — No runtime guard prevents the assignment. The source states: “State is not a place for heavyweight clients, indexes, file handles … put those in resources.” This is a documentation rule, not enforced by the code.
• Posture — fail‑hard if the state must be serialized, for example when calling Context.to_dict() for snapshotting or durable workflow persistence. The JSON serializer raises a TypeError and the run aborts. If no serialization is ever attempted, the failure does not manifest.
• Operator signal — A TypeError traceback: Object of type <class> is not JSON serializable. The error occurs at the point of serialization, not at the set call.
• Recovery — Manual correction. Remove the non‑serializable object from ctx.store and inject it via a Resource factory function (e.g., get_llm returning an Anthropic client). Then re‑run with a fresh context.

3. Typed state field assignment fails validation

• Trigger — A step of a workflow that uses typed state (e.g., ctx: Context[CounterState]) attempts to assign a value of the wrong type to a Pydantic field. For example, writing state.count = "abc" when CounterState.count is declared as int.
• Guard — Pydantic’s BaseModel validation. The assignment triggers a PydanticValidationError (or ValidationError) automatically before the value is stored.
• Posture — fail‑hard. The validation exception is raised inside the step, causing that step to fail and the entire workflow run to abort (assuming no outer try/except).
• Operator signal — A ValidationError traceback showing the field name, expected type, and received value. For example: validation error for CounterState\ncount\n Input should be a valid integer [type=int_type, ...].
• Recovery — No automatic recovery. The developer must fix the assignment to provide the correct type, then re‑run the workflow.

4. Context reused across runs with stale state

• Trigger — A developer completes a workflow run with a Context (e.g., ctx = Context(workflow); handler = workflow.run(ctx=ctx)), then invokes workflow.run(ctx=ctx) again expecting the state to be fresh (e.g., a counter reset to zero). The source says: “If the previous run has completed, run(ctx=ctx) starts a new run with the same stored state.”
• Guard — No guard exists. The runtime does not reset ctx.store between runs; the developer must explicitly clear state or create a new Context.
• Posture — fail‑soft. The new run proceeds without any error, but the state carries over previous values. The output may be incorrect (e.g., a counter starts from 5 instead of 0), yet no exception is raised.
• Operator signal — Silent incorrect output. The operator sees a result that is not from a clean state—for example, repeated accumulations across runs.
• Recovery — Manual: either create a new Context for each run or explicitly reset the store before the second run (e.g., using ctx.store.set("count", 0) or calling Context.from_dict(workflow, {}) to restore an empty state).

5. Long‑running work inside edit_state block causes other steps to stall

• Trigger — A step enters async with ctx.store.edit_state() as state: and inside the block performs slow I/O or an LLM call (e.g., await llm.acomplete(...)) before exiting. The source warns: “Keep the block small: do the read‑modify‑write work inside it, and keep slow LLM or network calls outside it.”
• Guard — No runtime guard enforces this rule. The lock held by edit_state() prevents all other steps from editing the state while the block is running, but there is no timeout or detection.
• Posture — fail‑soft. The workflow does not abort, but other steps that need to edit state will wait indefinitely (or until the outer workflow times out). The system degrades with increased latency or apparent hanging.
• Operator signal — No direct log. The operator may observe stalled progress in the workflow (e.g., a step waiting for a state lock never completes). If the workflow has total timeout, a TimeoutError may appear.
• Recovery — No automatic recovery. The developer must refactor the step to move slow operations out of the edit_state block, keeping only the atomic read‑modify‑write inside.

Imagine you are watching a cooking show where the chef prepares a multi‑course meal. The chef works behind the scenes, but the host announces each dish as it’s finished so you know what is happening before the meal is done. This chapter’s subsystem lets a running workflow broadcast progress updates step by step while it is still cooking, so you can see what it has accomplished without waiting for the final result.

How it works: The workflow starts a run, which returns a handler object that acts like the host. You loop over handler.stream_events() to receive each event the workflow writes (using ctx.write_event_to_stream inside a step). Every event is delivered as a separate announcement — including a final StopEvent when the meal is complete. To speed things up, the chef can split a task into pieces by returning a list from a step, and each piece becomes its own event that can be handled in parallel. You control how many cooks work at once with num_workers=5 — that limits how many copies of a step run simultaneously, like having five assistants chopping vegetables at the same time. When all pieces are done, you collect them by accepting a list parameter in the next step, which fires only after the whole batch arrives.

The trickiest part is that the stream does not just report happy progress — it also delivers failure events. If a step times out, the stream emits a WorkflowTimedOutEvent with details like how many seconds it waited and which steps were still running. A cancelled workflow sends a WorkflowCancelledEvent, and a permanently failed step sends a WorkflowFailedEvent with the exception and attempt count. You must watch for these termination events to know if the workflow finished normally or broke down. Without this streaming subsystem, you would be stuck waiting blindly for the final result, with no way to show partial progress, see which steps are stuck, or catch failures early — like sitting through an entire cooking show only to learn the kitchen caught fire at the end.

Streaming progress events from a workflow while steps run concurrently.

class FirstEvent(Event):
    first_output: str
class SecondEvent(Event):
    second_output: str
    response: str
class ProgressEvent(Event):
    msg: str

class MyWorkflow(Workflow):
    @step
    async def step_one(self, ctx: Context, ev: StartEvent) -> FirstEvent:
        ctx.write_event_to_stream(ProgressEvent(msg="Step one is happening"))
        return FirstEvent(first_output="First step complete.")

    @step
    async def step_two(self, ctx: Context, ev: FirstEvent) -> SecondEvent:
        llm = OpenAI(model="gpt-4o-mini")
        generator = await llm.astream_complete("Please give me ...")
        full_resp = ""
        async for response in generator:
            ctx.write_event_to_stream(ProgressEvent(msg=response.delta))
            full_resp += response.delta
        return SecondEvent(second_output="...", response=full_resp)

    @step
    async def step_three(self, ctx: Context, ev: SecondEvent) -> StopEvent:
        ctx.write_event_to_stream(ProgressEvent(msg="Step three is happening"))
        return StopEvent(result="Workflow complete.")

async def main():
    w = MyWorkflow(timeout=30, verbose=True)
    handler = w.run(first_input="Start the workflow.")
    async for ev in handler.stream_events():
        if isinstance(ev, ProgressEvent):
            print(ev.msg)
    final_result = await handler
    print("Final result", final_result)

A workflow begins when asyncio.run(main()) invokes run, which schedules the workflow in the background. The caller then enters a loop over handler.stream_events(), which yields every event the workflow writes to the stream as steps execute. Events are produced in the order the runtime dispatches them: first any events from initial steps, then fan-out events if a step returns a list, then events from concurrent workers (bounded by num_workers), and finally a StopEvent subclass when the workflow terminates normally or abnormally. On abnormal termination, a specific subclass — WorkflowTimedOutEvent, WorkflowCancelledEvent, or WorkflowFailedEvent — is published to the stream before the exception is raised from await handler. This ensures the stream consumer sees the failure event and can react (e.g., print details) before the error propagates.

The design preserves the invariant that "a handler stream can be consumed once." The stream is a single-use channel: after the consumer finishes iterating over stream_events(), the stream is exhausted. The guarantee is enforced by the runtime: the stream delivers exactly one sequence of events from start to terminal stop event, and no duplicate consumption is possible. Additionally, for fan-out/fan-in, the framework guarantees that a list parameter step fires exactly once after all matching workers have completed or returned None, preserving the batch semantics even when workers drop branches.

The key trade-off is automatic type-driven fan-out versus explicit event sending. The design chooses type inference: returning a list[Event] from a step automatically fans out each element as a separate event, and taking a list[Event] parameter automatically fans in the whole batch. This rejects the alternative of requiring developers to manually manage event routing via ctx.send_event calls for every concurrent split. By doing so, it avoids the cost of complex, error-prone orchestration code where developers would have to track event types, deduplicate, and manually synchronize collection. The cost of this automatic wiring is reduced flexibility — for non‑standard patterns (e.g., emitting events that do not match the return type), the developer must fall back to the dynamic ctx.send_event API, but the common case is simple and declarative.

A concrete failure mode occurs when a step permanently fails after exhausting retries. The runtime publishes a WorkflowFailedEvent containing the exact step_name, the exception object, the number of attempts, and elapsed_seconds. An operator who has written the stream‑handling code from the source — checking isinstance(ev, WorkflowFailedEvent) — would see a print like "Step 'fetch_data' failed after 3 attempts: ConnectionTimeout('...')". This signal appears in the stream_events() loop before the await handler raises, giving the operator a chance to log, alert, or clean up without catching a raw exception.

Step failure after exhausting retries

• Trigger – A step raises an exception repeatedly until the configured retry limit is exceeded.
• Guard – The runtime publishes an WorkflowFailedEvent with step_name, exception, attempts, and elapsed_seconds, then raises the exception from await handler. No additional guard is shown; the event itself is the only notification before the hard failure.
• Posture – Fail‑hard: the workflow stops and the exception propagates upward.
• Operator signal – The WorkflowFailedEvent delivers the exact step name and the exception object; the operator sees the exception raised from await handler.
• Recovery – No automatic recovery; the operator must inspect the event fields, fix the root cause, and re‑run the workflow.

Workflow timeout

• Trigger – The workflow’s wall‑clock timeout is exceeded while steps are still running.
• Guard – A WorkflowTimedOutEvent is written to the stream, containing timeout (seconds) and active_steps (list of step names that were running). After the event is consumed, an exception is raised from await handler.
• Posture – Fail‑hard: the workflow is aborted and an exception surfaces.
• Operator signal – The WorkflowTimedOutEvent includes the exact timeout value and the list of steps that were still active.
• Recovery – No automatic retry; the operator must increase the timeout or optimize slow steps, then re‑run.

Stream consumption exhaustion

• Trigger – Two or more clients attempt to iterate over the same handler.stream_events() instance.
• Guard – The source states “A handler stream can be consumed once” and advises to “consume the stream once … and fan those events out yourself.” No built‑in guard prevents double consumption.
• Posture – Fail‑soft: the first consumer sees all events; the second consumer sees none (the stream is exhausted). The workflow continues unaffected.
• Operator signal – Silent absence of events for the second consumer; no error or log is generated.
• Recovery – Manual: the application must implement a fan‑out layer (e.g., using asyncio.Queue or a pub‑sub pattern) so that multiple clients receive events from a single stream pass.

Workflow cancellation

• Trigger – An external cancellation request is issued (the source does not show the cancellation mechanism, but the event is defined).
• Guard – A WorkflowCancelledEvent is published to the stream before the workflow is terminated. No explicit guard for preventing cancellation is present.
• Posture – Fail‑hard: the workflow stops immediately.
• Operator signal – The WorkflowCancelledEvent appears in the event stream, and await handler raises an exception (implied by the pattern of other terminal events).
• Recovery – Manual: the operator must decide whether to re‑run the workflow; no automatic restart.

Race condition in ctx.send_event with downstream list join

• Trigger – A step uses ctx.send_event to emit multiple events, then fails (raises an exception) before returning. The already‑sent events remain in the stream. A downstream step that accepts list[Event] may start processing with an incomplete set of events.
• Guard – The source explicitly remarks “anything already sent stays out even if the step later fails” but provides no guard or rollback for the downstream list. No WorkflowFailedEvent is generated for the incomplete send sequence.
• Posture – Fail‑soft but dangerous: the downstream step proceeds with partial data, potentially producing incorrect results. The upstream failure is not propagated to the downstream collector.
• Operator signal – The downstream step finishes with a partial list (no error logged for the inconsistency); the workflow may complete successfully with wrong output, or the downstream step itself may raise an error later if the data shape is invalid.
• Recovery – None automatic; the developer must design steps to be idempotent or use transactional patterns (e.g., send events only after all computation succeeds), or add manual validation inside the list‑collecting step.

Imagine you're making a sandwich and realize you need a specific ingredient from the store. You write down where you left off—the bread, the knife—then go get the ingredient, return, and continue exactly from that spot. That's what this subsystem does for a computer program: it lets a workflow pause mid‑step, ask a person a question, wait for an answer, then pick up right where it stopped. Its purpose is to let human decisions fit into automated processes without losing progress.

In the workflow, two steps work together like a note‑passing game. The first step sends an "input required" event—a note saying "give me a number". The second step waits for a "human response" event—the note with the answer. The program's controller watches for these notes in a stream; when it sees the request, it asks the person (maybe by showing a prompt or a web form) and sends the reply back into the workflow. For waits that take a while—like overnight—you can snapshot the entire state of the workflow after sending the prompt. That snapshot is like a photo of the recipe step and all the ingredients so far. You store it (in a file or database) and later restore it when the response arrives. The Context object, with its state store and methods like to_dict(), carries the work forward.

The non‑obvious rule is that you don't have to keep the program running the whole time. Without snapshotting, a web app would either hang forever waiting for a person to type, or lose every bit of progress if the connection drops or the process restarts. The subsystem's Context.from_dict() rebuilds everything from the saved picture, so the workflow can resume later from the exact same point—even if the original server has been recycled. Without it, any long pause would break the whole recipe.

A two‑step workflow that pauses for human input, snapshots the context to allow stop/resume, and later restores the context to continue.

handler = workflow.run()
async for event in handler.stream_events():
    if isinstance(event, InputRequiredEvent):
        # snapshot and store context, then cancel the in‑memory run
        ctx_dict = handler.ctx.to_dict()
        # e.g. await db.save("run-123", json.dumps(ctx_dict))
        await handler.cancel_run()
        break


# ctx_dict = json.loads(await db.load("run-123"))
restored_ctx = Context.from_dict(workflow, ctx_dict)
handler = workflow.run(ctx=restored_ctx)

await handler.send_event(HumanResponseEvent(response=response))
async for event in handler.stream_events():
    continue

final_result = await handler

The subsystem operates through an ordered pair of steps: the ask step returns an InputRequiredEvent to signal that human input is needed, and the answer step consumes a HumanResponseEvent to produce the final result. The caller watches the workflow’s event stream via handler.stream_events(), then replies by calling handler.send_event(HumanResponseEvent(response=...)). For workflows that must survive long pauses—such as those that wait for a person in a separate web request—the design allows the caller to snapshot the workflow’s Context after the InputRequiredEvent is produced using ctx.to_dict(), store that dictionary, and later restore it with Context.from_dict(workflow, ctx_dict) before calling workflow.run(ctx=ctx) again. This ensures the workflow resumes from exactly where it paused, with no need to keep the original process open.

The invariant preserved is the serializable state guarantee: the workflow’s state is always serializable to JSON so it can be durably snapshotted and restored. The source explicitly warns that state should be “values a JSON serializer can encode” because it will be serialized when making a run durable. This guarantee ensures that any pause-and-resume cycle—even across different servers or requests—will carry the exact same workflow progress forward, preventing lost work or duplicated human prompts.

The key trade-off rejects the simpler alternative of a blocking, synchronous wait (e.g., a direct input() call inside the workflow). That alternative would be straightforward but would require the calling process to stay alive for the entire human delay, tying up resources like database connections, memory, and request handlers. Instead, the snapshot/restore pattern avoids the cost of holding an open process for an arbitrary duration, which is especially costly in web applications where each request must be short-lived. The price paid is the added complexity of explicit context persistence and the need to ensure that all state is serializable, but the gain is resilience to long pauses and the ability to decouple the prompt from the response.

A concrete failure mode occurs when a step stores a non‑serializable object in the context (for example, an open file handle or a database client object). When the caller attempts to snapshot via ctx.to_dict(), the serialization will fail with a TypeError such as Object of type <class '_io.TextIOWrapper'> is not JSON serializable. An operator would observe this error logged by the workflow runner, often accompanied by a stack trace pointing to the step that added the non‑serializable value. The workflow would not be able to pause, forcing the system either to keep the process alive or to crash. This highlights why the source explicitly advises to keep heavyweight runtime objects in resources rather than in state.

Missing Human Response (Indefinite Wait)

• Trigger — The second step calls wait_for_event to pause for a human response, but the person never sends the event (e.g., because they are unavailable or the notification is lost).
• Guard — No explicit guard exists in the source; the wait_for_event mechanism relies on the caller sending the event, and there is no timeout, retry, or fallback defined for a missing response.
• Posture — Fail‑hard: the workflow run hangs indefinitely, never progressing beyond the waiter step. No output is produced and no further steps execute.
• Operator signal — The workflow remains in a paused state with no new events observed in the event stream; logs would show no activity after the prompt step completed.
• Recovery — Manual intervention is required: the operator must locate the pending human‑in‑the‑loop task and either send the missing response or cancel and restart the run from the last snapshot.

Non‑Serializable State During Snapshot

• Trigger — A step stores a value in ctx.store (or a typed state field) that is not JSON‑serializable (e.g., a file handle, LLM client, or custom object without a Pydantic serializer). When the operator attempts to snapshot the context via Context.to_dict(), the serialization raises an exception.
• Guard — The source advises that state “should be data you are willing to serialize” and that you can “provide a custom serializer when calling Context.to_dict() and Context.from_dict()”, but no automatic validation or guard is enforced at write time.
• Posture — Fail‑hard: the to_dict() call raises an error, aborting the snapshot operation and therefore preventing durable storage or later restoration of the run.
• Operator signal — A serialization error is logged (e.g., TypeError: Object of type X is not JSON serializable), and the snapshot is not created.
• Recovery — The operater must fix the state by removing the non‑serializable object (or providing a custom serializer), then retake the snapshot. They may need to discard the current progress or replay from a prior valid snapshot.

Unprotected Concurrent State Modification

• Trigger — Two steps (or two runs sharing the same Context) read and write ctx.store using store.get() and store.set() without wrapping the update in edit_state(). Because the workflow may allow concurrent step execution, one step’s write overwrites another’s, leading to lost updates.
• Guard — The source provides edit_state() for atomic read‑modify‑write: “async with ctx.store.edit_state() as state: ... No other step can edit the state while the block is running.” However, its use is optional; no guard prevents steps from bypassing it.
• Posture — Fail‑soft: the workflow continues executing, but the state becomes silently corrupted (e.g., a counter value is lower than expected). Subsequent steps operate on stale or inconsistent data, producing wrong results.
• Operator signal — No immediate error is raised; the operator would observe unexpected final results or, if monitoring metrics, see that the state value diverges from the intended count.
• Recovery — The system does not auto‑recover. The operator must detect the inconsistency (e.g., by auditing the final state) and restart the workflow with a fresh context, ensuring all concurrent‑safe steps use edit_state.

Lost Human Response Event

• Trigger — The caller watches the event stream for the human response, but the event is lost due to network failure, process crash, or a bug in the event‑stream listener. The waiter step (wait_for_event) never receives the event.
• Guard — The source does not describe any delivery acknowledgment, retry, or timeout for the event‑stream watcher. The only resilience mechanism mentioned is the ability to snapshot and restore the context, but that does not re‑deliver the missed event.
• Posture — Fail‑hard: the workflow remains paused permanently because the expected event never arrives, and no fallback behavior is defined.
• Operator signal — The workflow run stays in a paused state; the event stream shows no activity for the expected response; no error is logged because the framework considers the wait normal.
• Recovery — The operator must manually inspect the pending human‑in‑the‑loop task, re‑send the response (if possible), or discard the run and restore from the last snapshot before the prompt step.

Internal Exception Handling Failure in wait_for_event

• Trigger — The step containing wait_for_event raises an internal exception to pause execution, but that exception is not properly caught by the workflow runtime (e.g., due to an unexpected state or a bug in the framework), causing the step to fail outright instead of pausing.
• Guard — The source states that “the step always runs at least once up to the waiter, which then raises an internal exception to pause execution.” However, it does not name a specific except clause or retry logic that guards against this exception escaping incorrectly. The guard is implied to be inside the framework, but no identifier is given.
• Posture — Fail‑hard: the step raises an unhandled exception, the workflow run is aborted, and no human‑in‑the‑loop pause occurs.
• Operator signal — An unhandled exception traceback is logged; the workflow run terminates with an error status.
• Recovery — No automatic recovery. The operator must diagnose the cause (e.g., code before wait_for_event that is not safe to repeat) and fix the step implementation. A new run must be initiated from scratch or from the last durable snapshot if one existed.

Imagine you are baking a multi‑layer cake. If one step flops—the sponge doesn’t rise—you can redo just that step instead of trashing the whole cake. And between every step you take a photo, so if the kitchen timer goes off and you lose power, you can pick up from the last photo, not start from scratch. This subsystem lets long, multi‑step processes survive failures without losing all their progress.

Behind the scenes, each step can be wrapped in a retry policy: you set rules like stop_after_attempt(3) to give up only after three tries, or wait_fixed(5) to wait five seconds between attempts. The workflow automatically repeats the step if it hits a glitch, using the same input. For longer runs, you take a snapshot of the entire current state—the Context—by calling to_dict(), which serializes everything into JSON. Later, from_dict() rebuilds that context, and the workflow continues exactly where it left off, re‑dispatching only the events that were still in flight.

The tricky part is what happens when you resume: a step that was mid‑execution when you snapped the photo is rewound and runs again from the top. That means any side effect—like adding eggs or calling an external service—might happen twice. So each step must be designed to be safe to repeat (idempotent). Also, heavy objects like an oven or an API client are not stored in the snapshot; they live in a Resource factory that gets re‑created on restart, keeping the snapshot small and portable. Without this subsystem, any transient error or crash would force you to throw away the entire cake—every completed step wasted—and begin again from the very first ingredient.

A retry policy attached to a step, used within a checkpoint loop for durability.

class MyWorkflow(Workflow):
    @step(
        retry_policy=retry_policy(
            wait=wait_fixed(5),
            stop=stop_after_attempt(10),
        )
    )
    async def flaky_step(self, ctx: Context, ev: StartEvent) -> StopEvent:
        result = flaky_call()  # this might raise
        return StopEvent(result=result)

The subsystem begins with a step annotated by @step(retry_policy=...) that composes a retry_policy from wait strategies like wait_fixed and stop conditions like stop_after_attempt. On failure, the workflow re-executes the entire step body using the policy’s retry logic; Context.retry_info() lets the step inspect retry_number and last_exception to adapt behavior. After all retries are exhausted, the exception would normally fail the workflow, but a @catch_error handler accepting a StepFailedEvent can recover gracefully, optionally scoped via for_steps=... or as a wildcard. For durability, after a step completes the operator calls handler.ctx.to_dict() to serialize the context—including in‑flight events and the state store—and persists it. On resume, Context.from_dict(w, json.loads(...)) rebuilds the context, and the workflow continues from the restored state by calling run(ctx=ctx). The mechanism is ordered: first attempt, then retries, then handler, then optional checkpoint; resume re‑dispatches pending events but does not replay completed steps.

The design preserves an at‑least‑once invariance: completed steps do not re‑run because their outputs are already in the restored state, but a step that was mid‑execution when the snapshot was taken is rewound and runs from the top. Therefore step side effects must be safe to repeat—the source explicitly states that resume is at‑least‑once and idempotency is required. There is no exactly‑once guarantee; the invariant accepts duplicate work for a small window of steps in flight in exchange for not losing already‑finished work.

The key trade‑off is that checkpointing is explicit and manual, not automatic. The obvious alternative would be a built‑in periodic or transactional checkpointer that snaps state after every step without user intervention. That alternative is rejected because it would force serialization of every event and state value after each step, potentially with heavy or non‑serializable objects. The source advises keeping large objects in a Resource to avoid this cost. By requiring the operator to call to_dict() themselves, the workflow avoids the overhead of snapshotting after every step and gives control over when persistence happens—at the cost of operator diligence. If the operator forgets to checkpoint, a crash loses all progress.

A concrete failure mode is a serialization error during checkpointing: if an in‑flight event or state store entry contains a value that cannot be encoded by the JSON serializer (e.g., raw bytes or an open connection), to_dict() raises an exception and the whole snapshot fails. The operator would see a traceback from to_dict() indicating which field could not be serialized, and the checkpoint would not be written. Without a saved snapshot, the next restart would start the workflow from scratch, losing all intermediate work.

Retries exhausted after maximum attempts

• Trigger: A step raises an exception repeatedly; the stop condition stop_after_attempt(3) fires after the third attempt.
• Guard: A @catch_error handler—if one is declared for that step (either scoped via for_steps or as a wildcard). If no handler exists, there is no guard.
• Posture: fail-soft if a catch_error handler is present and returns a StopEvent (the run continues with a fallback value). fail-hard if no handler is defined—the exception propagates and the workflow aborts.
• Operator signal: The log line "retry %d after %s" (written inside the step using ctx.retry_info()) appears for each attempt. On exhaustion the operator sees either a StepFailedEvent (if a handler catches it) or an unhandled exception terminating the run.
• Recovery: No more retries are attempted. If a catch_error handler exists, it returns a StopEvent with a fallback result (e.g., {"failed_step": ev.step_name, "error": str(ev.exception)}). Otherwise a manual restart is required.

Non‑retryable exception not caught by a retry condition

• Trigger: A step raises an exception that does not match any condition in the retry_policy’s retry parameter—or the step has no retry policy at all.
• Guard: A @catch_error handler (wildcard or step‑scoped) that receives the resulting StepFailedEvent. Without such a handler, no guard.
• Posture: fail-hard if no catch_error handler exists; the exception propagates and the workflow fails. fail-soft if a handler returns a StopEvent fallback.
• Operator signal: No retry log lines appear. The operator sees the workflow terminate with an exception (or a StepFailedEvent if a handler intercepts it).
• Recovery: No retries occur. A catch_error handler can produce a fallback value; otherwise the operator must manually inspect and restart the workflow.

Crash before the durable checkpoint snapshot

• Trigger: The process crashes after a step’s execution completes but before Context.to_dict() is called to persist the state.
• Guard: None. The source explicitly says “there is no built‑in checkpointer to enable” and that the user must snapshot “as the run progresses.”
• Posture: fail-hard—the run’s in‑flight state is lost entirely, and the workflow cannot resume from that point.
• Operator signal: Silent absence of a checkpoint entry in the external storage (e.g., database). On restart, the workflow does not appear in the list of recoverable runs.
• Recovery: Manual restart from the last valid checkpoint (or from scratch if no prior snapshot exists). The operator must accept that any work completed after the last snapshot is lost.

Replay‑induced duplicate side effects on resume

• Trigger: A step is interrupted before its output journal entry is committed. During resume the runtime replays the journal and re‑executes the step from the top.
• Guard: None in the runtime. The source advises “design steps to be idempotent” but provides no programmatic guard.
• Posture: fail-soft—the run continues, but external effects (API calls, file writes, state mutations) may be duplicated.
• Operator signal: No error or warning is logged by the framework. Duplication is visible only through external side effects (e.g., duplicate records, repeated API charges).
• Recovery: No automatic recovery; the operator must rely on the step’s own idempotency logic. If side effects are not idempotent, manual cleanup is required.

Recovery budget exhausted in a catch_error handler

• Trigger: A catch_error handler returns an event that re‑routes back to the same handler, causing re‑entry. The counter exceeds max_recoveries (default 1).
• Guard: The max_recoveries parameter on the @catch_error decorator stops re‑entry after the allowed count.
• Posture: fail-hard—the workflow fails instead of re‑entering the handler again. (The source: “caps how many times that can happen … before the workflow fails instead of re‑entering.”)
• Operator signal: No dedicated log line from the source, but the operator sees the workflow halt with an exception related to the exhausted recovery budget.
• Recovery: Manual intervention required—typically by adjusting max_recoveries, redesigning the handler to avoid loops, or providing an alternative fallback path.

Code changes causing non‑determinism during workflow resume

• Trigger: A workflow’s step logic is modified between the original run and a resume attempt. The resumed workflow replays the journal, then executes the new code against the restored state.
• Guard: None. The source warns: “changing a workflow's code while historical runs are still in progress can cause non‑determinism.”
• Posture: fail-hard—the resumed run may produce unexpected errors, incorrect outputs, or silently compute a different result.
• Operator signal: Errors such as Event type mismatches or unexpected AttributeErrors; or, in the worst case, no error at all but a wrong final StopEvent.
• Recovery: No automated recovery. The operator must abort the affected run, roll back the code change, or manually reconcile the state. Versioning of workflow code is left to the user.

Imagine a team of chefs in a kitchen: the head chef reads the order, then hands tasks to specialists—one chops vegetables, another grills meat. The head chef decides when to pass the plate to the next cook. In LlamaIndex, an agent is actually a built-in workflow that coordinates these hand-offs automatically. This system works as a series of steps, each triggered by an event. The FunctionAgent runs a tool-calling loop: the agent picks a tool, calls it, sees the result, then decides the next step. That decision might be to hand off to another agent, using a "handoff" mechanism configured in can_handoff_to. The root agent starts the entire process, and when it returns a StopEvent, the workflow stops with a result. If human input is needed, the workflow can pause and wait for an event. This all happens as a single AgentWorkflow object. The tricky part is that the workflow doesn't require a fixed path. Steps communicate via custom event types—like JokeEvent or ResearchEvent—rather than predefined edges. This means a writer agent can hand off back to a researcher agent for more facts, creating a loop that feels natural. Without this event-driven architecture, you would have to encode loops and branches into rigid code, making it hard to change or extend. The core classes FunctionAgent and AgentWorkflow handle this complexity, so you just define agents and their handoff permissions. What would go wrong without this subsystem? You'd end up writing your own messy state machine, manually passing data between agents, and accidentally creating deadlocks or infinite loops when agents try to hand off incorrectly.

A FunctionAgent is a built-in workflow that runs a tool-calling loop with streaming and state management.

def multiply(a: float, b: float) -> float:
    """Useful for multiplying two numbers."""
    return a * b

agent = FunctionAgent(
    tools=[multiply],
    llm=OpenAI(model="gpt-4o-mini"),
    system_prompt="You are a helpful assistant that can multiply two numbers.",
)

async def main():
    response = await agent.run("What is 1234 * 4567?")
    print(str(response))

if __name__ == "__main__":
    asyncio.run(main())

In this subsystem, the ordered mechanism begins with the FunctionAgent loop: the agent receives the latest user message along with the accumulated chat history, sends the tool schemas and that history over the API, and then the LLM responds either with a direct reply or a list of tool calls. Every tool call is executed, the results are appended to the chat history, and the agent is invoked again with the updated history—repeating until the agent returns a final answer. When multiple agents collaborate via AgentWorkflow, the root agent starts, executes tools, and can hand off control to another agent via the can_handoff_to configuration; the hand-off repeats until an agent returns a final result. If a step fails permanently after exhausting retries, the workflow publishes a WorkflowFailedEvent before raising the exception from await handler. Timing out or cancellation also publish specific StopEvent subclasses: WorkflowTimedOutEvent or WorkflowCancelledEvent. These events are consumed through the single-use handler.stream_events() generator, which yields every event until a StopEvent arrives, after which the final result is obtained by awaiting the handler.

The invariant the design preserves is that every workflow run has a dedicated Context whose ctx.store holds serializable state shared across steps. This state store is explicitly not for heavyweight clients or file handles—it is data the operator is willing to serialize for snapshotting or resuming the workflow. The guarantee is that the workflow will always produce a StopEvent (or one of its defined subclasses) and will deliver that event through the stream before termination, whether successful, cancelled, timed out, or failed. There is no exactly-once execution guarantee for individual steps (retries are possible), but the termination events provide a deterministic signal to the consumer about the outcome.

The key trade-off is the adoption of an event-driven, step-based workflow over the alternative of directed acyclic graphs (DAGs). The context explicitly states that DAGs required logic like loops and branches to be encoded into graph edges, created complexity around optional and default parameters, and felt unnatural for developing complex, looping, branching AI applications. The event-based pattern rejects that DAG approach in favor of plain Python: branches are ordinary if statements returning different event types, loops are steps that return events handled by earlier steps, and concurrent work is a step that returns list[Event]. This rejection avoids the cost of encoding control flow into graph structure, the fragility of optional parameter passing between nodes, and the general impedance mismatch between DAGs and the natural iterative behavior of agents.

A concrete failure mode is a step that exhausts its retry policy without success. The operator will observe a WorkflowFailedEvent on the stream, containing the exact step_name, the exception object, the number of attempts made, and the elapsed_seconds the step ran before failing. The operator would then inspect the exception field to understand the root cause. Similarly, if the entire workflow exceeds its configured timeout, the signal is a WorkflowTimedOutEvent which includes the timeout value in seconds and a list of active_steps—the names of the steps that were still running when the timeout fired, enabling the operator to diagnose which step(s) blocked the workflow.

Step Failure Exhausting Retries (Caught by catch_error)

• Trigger — A step decorated with @step(retry_policy=retry_policy(stop=stop_after_attempt(N), wait=wait_fixed(seconds))) raises an exception on every attempt, and after N attempts the retry policy stops.
• Guard — The @catch_error decorator on a method that accepts StepFailedEvent. The handler may return a StopEvent to gracefully end the workflow.
• Posture — fail-soft: the workflow continues by using the fallback result returned from the catch_error handler, instead of aborting.
• Operator signal — A WorkflowFailedEvent is published to the stream, containing the step_name, exception, attempts, and elapsed_seconds fields. If the handler returns a StopEvent, that event ends the stream.
• Recovery — The handler returns a StopEvent with a fallback value (e.g., {"fallback": True}). The workflow ends without additional retries.

Workflow Timeout

• Trigger — The workflow exceeds its configured timeout (mechanism not shown in source; the resulting WorkflowTimedOutEvent is described).
• Guard — No user-defined guard is provided; the framework publishes WorkflowTimedOutEvent automatically when the timeout expires.
• Posture — fail-hard: the workflow aborts and raises an exception from await handler. The WorkflowTimedOutEvent is published before the exception.
• Operator signal — WorkflowTimedOutEvent with fields timeout (seconds) and active_steps (list of running step names).
• Recovery — No automatic recovery. The operator must restart or reconfigure the workflow.

Human-in-Loop No Response

• Trigger — A step returns InputRequiredEvent (e.g., InputRequiredEvent(prefix="Enter a number: ")), and the caller never sends a HumanResponseEvent back into the handler.
• Guard — No guard is shown in the source. There is no timeout or retry on the InputRequiredEvent; the workflow pauses indefinitely.
• Posture — fail-soft: the workflow hangs (does not abort or close), waiting for input that will not arrive.
• Operator signal — The event stream stops producing new events after the InputRequiredEvent; no StopEvent or error is emitted.
• Recovery — Manual: the operator must send a HumanResponseEvent (e.g., via the handler’s stream) or cancel the workflow.

State Store Serialization Failure

• Trigger — A step writes an unserializable object (e.g., a network client, file handle) to ctx.store. The source explicitly warns: "State should be data you are willing to serialize when you snapshot or resume a workflow."
• Guard — No guard is shown. The framework assumes ctx.store content is serializable; no retry or exception handler is provided for serialization errors.
• Posture — fail-hard: if snapshot or resume is attempted, serialization fails and the workflow run aborts.
• Operator signal — A serialization-related error (likely a TypeError or AttributeError) raised during snapshot/resume. The exact log line is not specified.
• Recovery — Manual: move the unserializable object to a Resource(...) and keep only plain data in ctx.store. No automatic recovery.

Workflow Graph Validation Failure

• Trigger — At construction time (or when workflow.validate() is called), the graph validation detects that a step consumes an event that is never produced, or produces an event that nobody consumes, or the workflow has no terminal StopEvent.
• Guard — The workflow.validate() method (source: "You can also call workflow.validate() directly in tests or startup code."). The @step(skip_graph_checks=["reachability"]) decorator can selectively skip checks.
• Posture — fail-closed: the workflow refuses to run; construction raises an error and no run starts.
• Operator signal — Validation error message identifying the missing producer/consumer or dead end.
• Recovery — Manual: correct the step annotations, add missing events, or apply skip_graph_checks for intentionally dynamic steps.

Imagine a workshop where you build a complex machine—the workflow. The workflow server is like a reception desk that lets outside people operate that machine over the phone. It gives each operation a ticket number (handler ID) so you can call back to check progress or watch it run. This subsystem lets you turn a notebook workflow into a service that any HTTP client can talk to.

When you’re ready to deploy, the server uses endpoints like /run to start a workflow and wait for the final result, or /run-nowait to kick it off and immediately hand you a handler ID. That ID is your key: you can later stream events using GET /events/{handler_id} or poll for the finished result. If you need to test just one step—say, a step that pauses for human feedback—you send a POST to /events/{handler_id} with the event data and an optional step name like "process_human_feedback". This lets an external system inject input exactly where the workflow is waiting, without going through the whole chain.

The trickiest part is the after_sequence parameter when streaming events. Without it, you might miss events that already happened or re‑read old ones. By passing "now", you skip all past events and only get new ones; passing an integer resumes from that sequence number. The server tracks every event in an EventStream object with a last_sequence property, so you can save your place and reconnect after a drop. Without this subsystem, you would have no way to run your workflow outside of Python, no way to pause for human input mid‑run, and no way to stream progress to a frontend—you would be stuck debugging with print statements, unable to hand your workflow to anyone else.

The workflow server makes workflows available over HTTP; you can then test steps individually or observe internal events via the runtime instrumentation.

class MyWorkflow(Workflow):
    @step
    async def my_step(self, ev: StartEvent) -> StopEvent:
        return StopEvent(result="Done!")


async def main():
    server = WorkflowServer()
    server.add_workflow("my_workflow", MyWorkflow())
    await server.serve("0.0.0.0", 8080)

When a workflow leaves the notebook, the WorkflowServer exposes it over HTTP. A run is launched—either synchronously or asynchronously—and the server returns a handler_id. The client then uses get_workflow_events(handler_id) to stream emitted events. That call accepts an after_sequence parameter: pass "now" to skip past events, or any integer to resume from that sequence number. On connection drops, the method automatically reconnects from the last received last_sequence, up to max_reconnect_attempts (default 3). For human‑in‑the‑loop or isolated step testing, the client issues a POST /events/{handler_id} with the serialized event and an optional step name; the runtime delivers that event directly to the named step. If the server is restarted with the default MemoryWorkflowStore, all handler state and events are lost, so resumption becomes impossible.

The design preserves event‑ordering through sequence numbers. The get_workflow_events method provides a last_sequence property, and the after_sequence parameter lets clients pick up exactly where they left off. This guarantees that, as long as the client persists the last seen sequence and reconnects within the retry limit, no event is missed—a form of at‑least‑once delivery with resumability. The WorkflowServer’s instrumentation layer further emits StepStateChanged internal events that could be used for checkpointing in durable workflows, but the core streaming guarantee is driven by sequence numbers.

The key trade‑off is the choice of MemoryWorkflowStore versus a database‑backed store. The obvious alternative would be to require a durable persistence layer for every deployment. That alternative is rejected because it forces developers to set up and manage a database even for early prototyping or transient workloads. By defaulting to memory, the system avoids the operational cost of database configuration, schema migrations, and serialization concerns. The price is lossiness: any handler state and events vanish when the process restarts. This is a deliberate choice for simplicity during development, with the option to promote to a durable store when production needs demand it.

A concrete failure mode is a server crash under the default MemoryWorkflowStore. The operator sees that a previously valid handler_id no longer resolves—POST /events/{handler_id} returns a 404 or the stream from get_workflow_events immediately ends with a 204 response because the handler record is gone. Another signal appears during streaming: if a network interruption exceeds max_reconnect_attempts, the client observes an abrupt 204 and the stream stops. The operator would then need to restart the entire workflow from scratch because the in‑memory store offered no checkpoint to resume from.

MemoryWorkflowStore data loss on process restart

• Trigger – The WorkflowServer process crashes or is restarted while using the default in‑memory persistence. All handler state, events, and results that were stored in MemoryWorkflowStore are lost.
• Guard – None. The source explicitly states that state is lost on restart; no exception handler, retry, or fallback is described for this condition.
• Posture – Fail‑closed. The server refuses to accept requests for any previous handler_id because the store no longer contains them. New runs start from a clean state.
• Operator signal – A GET /results/{handler_id} or GET /events/{handler_id} with a previously valid handler_id returns HTTP 404 (resource not found). No log line is mandated by the source.
• Recovery – No automatic recovery. The operator must manually rerun the workflow and, if external input was required, re‑send the necessary events (e.g., again POST to /events/{handler_id}). To prevent, the operator must configure a workflow_store backed by a database, as mentioned in the source: “For durable persistence, pass a workflow_store backed by a database.”

Concurrent reader limit on streaming events

• Trigger – A second HTTP client attempts to GET /events/{handler_id} while another client is already streaming events for the same handler_id, exceeding the documented single‑reader restriction.
• Guard – None. The source notes “Only one reader is allowed to stream the events per workflow run” as a consideration, but does not define a concrete exception handler, lock, or retry mechanism.
• Posture – Fail‑soft. The first reader continues uninterrupted; the second reader receives an error (likely HTTP 409 Conflict), degrading the experience for the second client only.
• Operator signal – The second client receives an error response. No server‑side log is specified; the operator would observe the client‑side failure.
• Recovery – No automated retry. The operator must wait until the first reader finishes, then re‑initiate the stream. The source acknowledges they are “working to improve” this.

Invalid handler_id in event‑sending endpoint

• Trigger – An external system POSTs to /events/{handler_id} with a handler_id that does not exist (never created, already canceled with purge=true, or expired from the store).
• Guard – None explicitly named. The server likely performs a lookup in the workflow store; when not found, it returns an HTTP 404.
• Posture – Fail‑hard. The request is rejected; no event is delivered to any step.
• Operator signal – HTTP 404 in the response body. No further error field is specified.
• Recovery – The caller must verify the handler_id (e.g., from a prior POST /run-nowait response) and ensure the workflow run still exists. No automatic retry is described.

Timeout acquiring the event‑streaming lock

• Trigger – A GET /events/{handler_id} request cannot acquire the internal lock to iterate over events within the time specified by the acquire_timeout parameter (e.g., due to prolonged contention from another reader).
• Guard – The acquire_timeout parameter itself acts as a timeout guard. When exceeded, the request fails.
• Posture – Fail‑hard. The streaming request is aborted; no events are sent to the caller.
• Operator signal – The client receives an HTTP error (likely 408 Request Timeout). No specific log line is given.
• Recovery – The client can retry with a longer acquire_timeout value after a brief delay. The source does not specify any automatic backoff.

Unrecognized step name in event‑sending request

• Trigger – A client POSTs to POST /events/{handler_id} with a step field that does not match any step in the workflow’s definition (e.g., misspelled “process_human_feedback”).
• Guard – None. The source states the step is optional and documents only the successful "status": "sent" response; no validation or error handling for invalid step names is described.
• Posture – Fail‑hard. The server rejects the request (likely HTTP 400 Bad Request), and the event is not delivered.
• Operator signal – HTTP 400 with an error message (inferred, e.g., “Step ‘foo’ not found”). No exact field is mandated.
• Recovery – The operator must correct the step name by checking the workflow’s step decorators and re‑send the event. No automatic retry.