# 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()`.

## 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())
```

`stream_text()` raises `RunFailed` (or `RunTruncated` when the reply hit the token budget) (or another `RunError`) if the run does not
complete, so a consumer that only wants text never has to inspect the result.
The `on_reset` callback fires on `ATTEMPT_RESET`, when a retried model call
restarts the answer; without it the answer simply starts over in the stream.
Clear any provisional text you have shown when it fires.

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.
Event types include `TEXT_DELTA`, `TEXT_COMPLETE`, `TOOL_CALL_START`,
`TOOL_CALL_COMPLETE`, `TOOL_RESULT`, `THINKING_DELTA`, `TURN_COMPLETE`,
`ERROR`, `ATTEMPT_RESET`, and `RUN_RESULT`. Durable runs also emit approval and
recovery events. Text deltas are provisional; discard an interrupted attempt's
text when `ATTEMPT_RESET` arrives. `RunEvent(data=...)` is required when you
construct events yourself.

`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()`. The 0.3 `harnessx.streaming`
module and `StreamingAgent` were removed in 0.4; they had been a shim over
`run_stream()`.

Usage stats accumulate on the guardrails engine:

```python
agent.guardrails.usage_summary
# {'iterations': 3, 'input_tokens': 5120, 'output_tokens': 640, 'estimated_cost': '$0.0250'}
```