# Streaming

Render typed events and handle provisional output.

## How it works

Two views exist over one run. `Agent.stream_text()` yields the answer's text as
plain strings; `Agent.run_stream()` yields typed `RunEvent`s and exposes the
final `RunResult`. Both use the same engine, tools, and permission checks as
`agent.run()`.

Text deltas are provisional. A retried model call restarts the answer and emits
`ATTEMPT_RESET`: discard the text already shown, and treat the final
`RunResult` as authoritative.

### Stalled streams and continued replies

A stream that never starts (`RetryPolicy.stream_first_event_timeout_seconds`, on
Anthropic), stalls for `RetryPolicy.stream_idle_timeout_seconds`, or ends
without the provider saying it is complete, is retried the same way, with an
`ATTEMPT_RESET`. See [Streams that stall or end
early](../retries-and-fallbacks/#streams-that-stall-or-end-early).

A reply cut off at its token budget and then continued is not a reset. The
continuation streams after the text already shown, and the final output joins
the pieces.

## Stream text

```python
import asyncio
from harnessx import Agent, AgentConfig, RunFailed

agent = Agent(config=AgentConfig(system_prompt="You are helpful."))

async def main():
    async with agent:
        try:
            async for chunk in agent.stream_text(
                "Tell me a story",
                on_reset=lambda: print("\n[answer restarted]"),
            ):
                print(chunk, end="", flush=True)
        except RunFailed as exc:
            print("run failed:", exc.error["message"])

asyncio.run(main())
```

A run that does not end with a whole answer raises a `RunError` out of
`stream_text()`: `RunFailed`, `RunTruncated` when the reply hit the token
budget, `RunAwaitingInput` when a tool needs approval, or `RunCancelled`. The
`on_reset` callback fires on `ATTEMPT_RESET`; without it the answer simply
starts over in the stream.

With fallbacks configured in `AgentConfig.fallbacks`, failover inside the chain
happens only before the first chunk, and a provider switch is not itself a
reset event. After a chunk arrives, a transient failure goes back to the
engine, and any further attempt resets as above. See [Retries & fallback
models](../retries-and-fallbacks/#streaming-behavior).

### Refusals and the step limit

`stream_text()` also raises `RunRefused` when the model declined or a filter
stopped it, and `RunLimitReached` when the run reached its step limit.
`RunTruncated` is raised only after truncation recovery has run out.

### Stop a stream early

Breaking out of the loop early leaves the run to be cancelled when the generator
closes. Wrap the generator in `contextlib.aclosing()` to make that immediate:

```python
import contextlib

async def first_line(agent, prompt):
    text = ""
    async with contextlib.aclosing(agent.stream_text(prompt)) as chunks:
        async for chunk in chunks:
            text += chunk
            if "\n" in text:
                break
    return text.partition("\n")[0]
```

## Stream events

```python
import asyncio
from harnessx import Agent, AgentConfig, RunEventType

agent = Agent(config=AgentConfig(system_prompt="You are helpful."))

async def main():
    async with agent, agent.run_stream("Tell me a story") as stream:
        async for event in stream:
            if event.type == RunEventType.TEXT_DELTA:
                print(event.data, end="", flush=True)
            elif event.type == RunEventType.TOOL_CALL_START:
                print(f"\n> calling {event.data.name}({event.data.input})")
            elif event.type == RunEventType.TOOL_RESULT:
                if event.data.is_error:
                    print(f"  tool error: {event.data.content}")
            elif event.type == RunEventType.TURN_COMPLETE:
                print()
        result = await stream.result()
        result.raise_for_status()

asyncio.run(main())
```

Use the stream's async context manager to close or cancel an abandoned stream.
Durable runs also emit approval and recovery events. `RunEventType` members:

| Event | `event.data` |
| --- | --- |
| `TEXT_DELTA` | Answer text fragment |
| `TEXT_COMPLETE` | Finished text of a turn |
| `THINKING_DELTA` | Reasoning text fragment |
| `TOOL_CALL_DELTA` | `ToolCallDelta(id, name, partial_json)` while the model writes a tool call; see below |
| `TOOL_CALL_START` | Tool name and input, when the call is complete and about to run |
| `TOOL_CALL_COMPLETE` | Finished call |
| `TOOL_RESULT` | Content and `is_error` |
| `TURN_COMPLETE` | End of a model turn |
| `ATTEMPT_RESET` | Restarted answer |
| `WAITING` | `{"on": "model", "seconds": ...}` while a model call is slow, `{"on": "retry", ...}` during a long wait before a retry; see below |
| `TODOS_UPDATED` | The whole task list; see below |
| `APPROVAL_REQUIRED` | A tool call that needs approval (durable runs) |
| `RECOVERY_REQUIRED` | A tool call whose outcome is unknown (durable runs) |
| `GAP` | `{"reason": "event_history_unavailable"}` when a Temporal run's event history cannot be read |
| `ERROR` | Run error |
| `RUN_RESULT` | Final `RunResult` |

## Show a tool call while it is being written

A model can spend a while writing a single tool call, say a whole report passed
to `write_file`. `TOOL_CALL_DELTA` lets you show something in the meantime. The
first delta for a call carries the tool's name and an empty `partial_json`,
which is a good moment to show "calling write_file…". Each delta after that
carries the next piece of the arguments, and joined together they make the
call's JSON. `TOOL_CALL_START` still comes once the call is complete and about
to run.

```python
from harnessx import RunEventType

async def show_tool_calls(agent, prompt):
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.TOOL_CALL_DELTA and not event.data.partial_json:
                print(f"\n[writing a call to {event.data.name}...]")
            elif event.type == RunEventType.TOOL_CALL_START:
                print(f"[running {event.data.name}]")
        return await stream.result()
```

Anthropic, OpenAI, Azure and OpenRouter send the arguments in pieces. Gemini
sends each call in one go, so you get two deltas: one with the name and one
with all the arguments. Handle them like text deltas: when `ATTEMPT_RESET`
arrives, drop anything you showed from that attempt. `stream_text()` leaves them
out, and so does a stream whose reply is buffered by response middleware.

`RunStream.text(on_reset=...)` is the same text view over any event stream,
including a durable `runtime.run_stream()`:

```python
async def text_only(stream):
    async with stream:
        async for chunk in stream.text():
            print(chunk, end="", flush=True)
```

`AgentRuntime` offers the same three verbs: `runtime.run()`,
`runtime.run_stream()`, and `runtime.stream_text()`.

## Show that a slow call is still working

A provider that is slow rather than broken sends nothing: no text, no error, no
retry. To the stream it looks like a hung agent. `WAITING` events fill that
silence. The first arrives after `first_after_seconds` without a chunk, then
one every `repeat_every_seconds`, each carrying the seconds waited so far.

```python
from harnessx import Agent, AgentConfig, ProgressPolicy, RunEventType

agent = Agent(config=AgentConfig(
    progress=ProgressPolicy(first_after_seconds=10.0, repeat_every_seconds=15.0),
))

async def show_progress(agent, prompt):
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.WAITING:
                print(f"\n[still waiting on the {event.data['on']}: {event.data['seconds']}s]")
            elif event.type == RunEventType.TEXT_DELTA:
                print(event.data, end="", flush=True)
        return await stream.result()
```

Those values are the defaults. `ProgressPolicy(first_after_seconds=None)` turns
the events off. You get the same events, with `"on": "retry"`, while the agent
waits before trying again, for example after a server's `Retry-After`. They are notices, not deadlines: nothing is cancelled, and the
call is still bounded by `RetryPolicy.call_timeout_seconds`. `stream_text()`
skips them.

Try it with no API key: `uv run python examples/01-basics/progress_and_waiting.py`.

## Task list updates

`RunEventType.TODOS_UPDATED` fires when the agent changes its [task
list](../loop-reliability/#a-plan-the-model-keeps-seeing) with `write_todos`.
`event.data` is a dict with the whole list every time: `todos` (each
`{"content", "status"}`), `completed`, `total`, and `in_progress`, the item being
worked on or `None`.

```python
from harnessx import RunEventType

async def show_plan(agent, prompt):
    async with agent.run_stream(prompt) as stream:
        async for event in stream:
            if event.type == RunEventType.TODOS_UPDATED:
                plan = event.data
                print(f"[{plan['completed']}/{plan['total']}] {plan['in_progress'] or ''}")
        return await stream.result()
```

The event is checked once per batch of tool calls, so a durable or Temporal run
reports the plan exactly as a direct one does.