Handle partial results, completion and cancellation in one action

Run the local action to receive output before completion and a separate final status. The caller can render partial work, cancel the action and decide what to retain.

A stream has more than values#

A useful stream has an order, a completion signal and a failure signal. It also needs an answer to “what happens when the consumer no longer wants this?” Without those, streaming is a sequence of callbacks whose lifecycle the application must reconstruct.

Imagine a transcription service that emits words, timestamps and a final confidence report. Concatenating every event into one text channel makes each consumer decode the service's private event convention. Named outputs let a caption view read text while a timeline reads timing information. The two consumers do not need to agree on the same rendering.

Here is the smaller version of that idea: a named string stream, using only local computation.

import asyncio
from typing import Annotated

import a11

REGISTRY = a11.ActionRegistry()


@REGISTRY.action
async def read(
    text: str,
    words: Annotated[a11.AsyncNode, a11.OutputPort(mimetype="text/plain")],
):
    for word in text.split():
        await words.put(f"{word} ")


async def main():
    action = REGISTRY.make_action("read").run()
    await action["text"].finalize(
        "Your functions. Your architecture. AI-native."
    )
    async for word in action["words"]:
        print(word, flush=True)
    await action.wait()


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

Install with pip install a11-kit, download the file, and run python first_stream.py. Local text processing keeps the example focused on the stream lifecycle.

Check completion separately#

A quiet stream may mean the producer is computing, a network is stalled, or the producer has finished. A11's action lifecycle gives the caller an explicit terminal status. Read the stream, then await the action's completion.

This separation makes partial results useful without misrepresenting them. An indexing task might emit 900 records before record 901 fails validation. Keeping the 900 records may be correct, but calling the entire indexing job successful is not. Your application decides whether to retain, retry or roll back the partial work.

Set a retention policy#

Incremental production reduces latency. It does not by itself guarantee bounded memory or backpressure all the way to a provider. A consumer that appends every fragment to an array still grows with the response. A producer with an unbounded queue can still overwhelm a slow consumer.

Decide what you retain. A caption view may replace old lines; a transcript editor may persist every word; a debug panel may keep a bounded preview and export the full record elsewhere. A11's ports carry the values; retention remains a product and runtime configuration decision.

Propagate cancellation to the producer#

Hiding a spinner is not cancellation. The producing operation needs a cancellation signal, and a provider request or local resource needs to stop when that signal arrives. A11 can cancel nested actions; handlers must also cooperate with external operations they own.

The browser example on the staged landing runs an actual TypeScript action. Its Stop button cancels that action, and its handler checks the action signal before each write. You can edit the input and run it again. That is a small but complete lifecycle: input, incremental output, stop or completion, and a new invocation.

Keep outputs stable across processes#

When the same producer moves to another process, its named outputs can stay stable. The remote caller changes how it connects and dispatches, while the rendering logic continues reading the same stream.

Streaming through the action interface carries early results through the application. A stream confined to the final web response remains unavailable while upstream operations run.

Try the remote version, or start with the complete local walkthrough.