An agent stream is the same tool loop as `fx.run`, delivered while the model is still working. A support chat shows tokens, tool calls, and a finish reason as server-sent events. Return `fx.json.stream` from the HTTP Flow. Parse the frames with `okengine/client/agent`. `return fx.json.stream(fx.run(agent, input, { stream: true }))`. `stream` is the third argument. `fx.stream` yields plain text from a model, not these events. ## Quick start ### Declare the agent ```typescript title="src/core/ai.ts" import { ai } from "okengine"; export const smart = ai.model("smart", { provider: "openrouter", model: "openrouter/free", }); export const support = ai.agent("support", { model: smart, tools: ["orders.get"], maxSteps: 4, }); ``` ### Stream it from HTTP ```typescript title="src/flows/support/assist.ts" import { on, flow, http } from "okengine"; import { z } from "zod"; import { support } from "@/core/ai"; export const assist = on( http.post({ in: z.object({ message: z.string().min(1) }) }).public(), flow({ do: ({ message }, fx) => fx.json.stream(fx.run(support, { message }, { stream: true })), }), ); ``` The response is `text/event-stream`. Pass your own `threadId` when the chat already has one. The default is a unique id. It is not a per-process counter. ### Read the frames ```bash curl -N -X POST http://localhost:6530/support/assist \ -H "content-type: application/json" \ -d '{"message":"Where is order 14?"}' ``` Each `data:` line is one JSON event. A text reply with no tool call looks like this (ids vary): ```text data: {"type":"RUN_STARTED","threadId":"agent-run-1","runId":"agent-run-1"} data: {"type":"STEP_STARTED","stepName":"step-1"} data: {"type":"TEXT_MESSAGE_START","messageId":"m-1","role":"assistant"} data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"m-1","delta":"Order 14 is open."} data: {"type":"TEXT_MESSAGE_END","messageId":"m-1"} data: {"type":"STEP_FINISHED","stepName":"step-1"} data: {"type":"RUN_FINISHED","threadId":"agent-run-1","runId":"agent-run-1","result":{"cost":0.01,"stopReason":"completed","output":"Order 14 is open."},"usage":[{"inputTokens":3,"outputTokens":2}]} data: [DONE] ``` `usage` is present only when the model reported token counts. `cost` and `stopReason` are on `result`, not on `usage`. ### Parse with the client ```typescript import { readAgentEvents } from "okengine/client/agent"; const response = await fetch("http://localhost:6530/support/assist", { method: "POST", headers: { "content-type": "application/json" }, body: JSON.stringify({ message: "Where is order 14?" }), }); for await (const event of readAgentEvents(response)) { if (event.type === "TEXT_MESSAGE_CONTENT") process.stdout.write(event.delta); } ``` Import `okengine/client/agent`. That module stays off `okengine/client`. ## Event reference Core names match AG-UI. Subagent notices are `CUSTOM`, not extra types. | Type | Fields | When it fires | | ---------------------- | ---------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------- | | `RUN_STARTED` | `threadId`, `runId` | The run begins. `threadId` is the option you passed, or the run id. | | `STEP_STARTED` | `stepName` | A step starts. `maxSteps` counts tool calls, not model calls. Names are `step-1`, `step-2`, … | | `TEXT_MESSAGE_START` | `messageId`, `role` | Assistant text is about to arrive. `role` is `assistant`. Empty text is skipped. | | `TEXT_MESSAGE_CONTENT` | `messageId`, `delta` | One piece of assistant text. A streaming driver emits one frame per delta. A driver without `stream` emits the whole turn in one frame. | | `TEXT_MESSAGE_END` | `messageId` | That text message is complete. | | `TOOL_CALL_START` | `toolCallId`, `toolCallName`, `parentMessageId?` | The model named a tool. `parentMessageId` is set when that step emitted assistant text. | | `TOOL_CALL_ARGS` | `toolCallId`, `delta` | One piece of the tool arguments. Streaming drivers emit each args delta. A driver without `stream` emits one JSON string. | | `TOOL_CALL_END` | `toolCallId` | Arguments are complete. The tool has not run yet. | | `TOOL_CALL_RESULT` | `messageId`, `toolCallId`, `content`, `role` | The tool returned. `role` is `tool`. `content` is a string. | | `STEP_FINISHED` | `stepName` | That model step is done, including its tool calls. | | `RUN_FINISHED` | `threadId`, `runId`, `result?`, `usage?`, `outcome?` | The loop stopped, or a tool is waiting for approval. | | `RUN_ERROR` | `message`, `code?` | The run failed and did not finish. `code` is the error name when it is not `Error`. | | `CUSTOM` | `name`, `value` | A nested agent started, finished, or failed. | `RUN_FINISHED.result` is `{ cost, stopReason, output }`. `cost` is the sum of model-reported spend for this run. `output` is the last tool result. A text reply with no tool call may be the raw provider payload, not the assistant string. `RUN_FINISHED.usage` is a one-element array, `{ inputTokens?, outputTokens? }`, and only when those numbers came from the model. Counts are never invented. `messageId` values are `m-1`, `m-2`, … inside the run. A missing provider tool-call id becomes `agent:step:index`. ## Tool approval A tool with `approval` and `gate` parks a durable Flow. The stream emits `RUN_FINISHED` with `outcome.type` `"interrupt"`, then `data: [DONE]`, then the response ends. The browser does not stay open. The approval id is base64url of `runId.toolCallId`, so it carries the durable Flow run id. ```json { "type": "RUN_FINISHED", "threadId": "agent-run-1", "runId": "agent-run-1", "outcome": { "type": "interrupt", "interrupts": [ { "id": "VjFTdEdYUjhfWjVqZEhpNkItbXlULmNhbGxfMQ", "reason": "approval", "payload": { "tool": "refund", "args": { "amount": 10 } } } ] } } ``` `id` is base64url of `.`. The example decodes to `V1StGXR8_Z5jdHi6B-myT.call_1`. `V1StGXR8_Z5jdHi6B-myT` is the journal run id, not `agent-run-1`. `runId` on the frame is the agent stream id. Show `payload.tool` and `payload.args`. Keep `id`. That id is the only handle for the decision. Resolve it from a Flow, or from the built-in routes. The routes are public so the request can arrive. The tool's gate is checked on the call. A denying gate is 403. A missing id, or a tenant that does not match the park, is 404. ```typescript await fx.agent.approve(id, { args: { amount: 4 } }); await fx.agent.deny(id, { reason: "over the limit" }); ``` ```bash curl -X POST http://localhost:6530/agent/approvals/approve \ -H "content-type: application/json" \ -d '{"id":"VjFTdEdYUjhfWjVqZEhpNkItbXlULmNhbGxfMQ","args":{"amount":4}}' ``` ```bash curl -X POST http://localhost:6530/agent/approvals/deny \ -H "content-type: application/json" \ -d '{"id":"VjFTdEdYUjhfWjVqZEhpNkItbXlULmNhbGxfMQ","reason":"over the limit"}' ``` `args` replaces the tool input. `reason` is what the model sees after a deny. The first decision wins. The default wait is `24h`, then the tool is denied with reason `timeout`. | Outcome | HTTP | What you do | | ---------------- | ------------------------------------------------------------------------------------------------------------------------ | ---------------------------- | | `{ ok: true }` | 200 `{ "data": { "ok": true }, "error": null }` | The journal sleep wakes now. | | Already resolved | 409 `Conflict`. No `Retry-After`. Message: `That value is already in use.` | Stop. Do not retry. | | Lease held | 409 `JournalLeaseBusy` with `Retry-After`. Message: `This run is locked by another worker. Retry after the given delay.` | Retry after the delay. | Approving or denying does not continue this SSE response. Follow the run to see the rest. ## Follow a run `GET /agent/runs/:runId/events` is `text/event-stream`. Each frame has an `id`. Send that id as `Last-Event-ID` to resume without gaps or duplicates. The route checks the same gate and tenant as the Flow that started the run. Another tenant, or a gate that denies the caller, is rejected. An approval interrupt does not end this follow. The stream stays open through the interrupt and continues to the terminal `RUN_FINISHED`. The first run stream still closes after the interrupt. Disconnect, then resume with the last id you received. The run stream yields every token. The follow log coalesces text and argument deltas (about every 100 ms or 1 KB, including a timer) and stores structural events one by one. The instance that holds the journal lease is the only writer. It reads the highest seq before it appends, and a repeated seq is an error. Followers read `seq` greater than the last id, not the whole run. The log keeps 5,000 rows per run. Past that, deltas stop and one `CUSTOM` event named `oke.events.truncated` is stored. Structural events, approval interrupts, `RUN_FINISHED`, and `RUN_ERROR` are still stored. Earlier rows stay, so a follower can resume and seq numbers do not skip. A finished run's events are deleted after 24 hours. An unfinished run older than 7 days is closed with `RUN_ERROR` and deleted. The scheduler reads the store, not the process that opened the run. One instance claims the journal lease before it closes that run. The log is stored on the journal driver and keyed by the agent run id. `flow.retry` keeps that id. A follower must be the starting principal or an operator, and must pass every gate on the Flow. The file journal is single-process. It is not for large logs. Use the Postgres journal driver in production. A stored row on the live run stream carries `id:` set to that row's seq. Resume with `Last-Event-ID` so text you already showed is not sent again. ```typescript import { approve, deny, readAgentEvents } from "okengine/client/agent"; for await (const event of readAgentEvents(`http://localhost:6530/agent/runs/${runId}/events`)) { if (event.type === "RUN_FINISHED" && event.outcome?.type === "interrupt") { await approve("http://localhost:6530/agent/approvals/approve", event.outcome.interrupts[0]!.id); } } ``` `approve` and `deny` retry `JournalLeaseBusy` and throw `Conflict` when the decision is already stored. `useAgentRun` from `okengine/client-react` sends the message, keeps the text so far, and follows once after an interrupt. Approve and deny do not open a second follow. ## Patterns A string and `{ message }` are the same turn. Passing both throws `fx.run: pass message or messages, not both`. ```typescript return fx.json.stream(fx.run(support, "Where is order 14?", { stream: true })); ``` `messages` is the prior thread. Roles are `system`, `user`, `assistant`, and `tool`. ```typescript return fx.json.stream( fx.run( support, { messages: [ { role: "user", content: "Where is order 14?" }, { role: "assistant", content: "Looking it up." }, { role: "user", content: "Refund it." }, ], }, { stream: true }, ), ); ``` Pass `threadId` when the chat already has one. The default is a unique id. ```typescript return fx.json.stream(fx.run(support, { message }, { stream: true, threadId: chatId })); ``` `RUN_STARTED` and `RUN_FINISHED` both carry that `threadId`. `mock`, `anthropic`, and `openai-compatible` stream tokens and tool-call arguments. `bedrock` and `vertex` are reserved driver ids: boot throws before a call. `fx.stream` uses the same `stream` method, so it works on Anthropic too. A tool that names another agent emits `CUSTOM` on the parent stream. `okengine/client/agent` also returns typed names. | `CUSTOM.name` | Parsed `type` | `value` | | ----------------------- | ------------------- | ---------------------------- | | `oke.subagent.started` | `subagent.started` | `runId`, `parentToolCallId?` | | `oke.subagent.finished` | `subagent.finished` | `runId`, `parentToolCallId?` | | `oke.subagent.error` | `subagent.error` | `runId`, `parentToolCallId?` | Any other `CUSTOM` name stays `CUSTOM`. Nesting stops at `maxDepth` (default 3). ## Text stream or event stream | Call | What each SSE `data:` frame is | | -------------------------------------------------------- | ---------------------------------------------- | | `fx.json.stream(fx.stream(model, { prompt }))` | A JSON string of plain text (`"Hel"`, `"lo"`). | | `fx.json.stream(fx.run(agent, input, { stream: true }))` | One AG-UI event object. | `fx.stream` is the model text iterator. `fx.json.stream` is the SSE carrier. Agent UIs use the second row. ## Abort and stop reasons Disconnecting the HTTP client aborts the in-flight model call. The run records `stopReason` `aborted`. If the frame is still written, it is `RUN_ERROR` with `code` `AbortError`. `message` may be the provider's abort text, not only `aborted`. `result.stopReason` on a normal `RUN_FINISHED`: | `stopReason` | Meaning | | ------------ | ------------------------------------------------------------------------------------------ | | `completed` | The model stopped calling tools. A gate denial fed back to the model stays `completed`. | | `max_steps` | The loop hit `maxSteps` (default 6). | | `budget` | Spend reached `budget.maxCostPerRun`. The trail so far is the output. This does not throw. | `denied` (unknown tool) and `aborted` arrive as `RUN_ERROR`, not as `result.stopReason`. A thrown tool ends with `RUN_FINISHED` and `result.stopReason` `"error"`, then `data: [DONE]`. The same close follows an approval interrupt: the final frame, then `[DONE]`, then the body ends. ## Troubleshooting The tool has `approval`, and the Flow is not durable. The message names that Flow: `ai: flow "support.assist" must set durable: true to run agent "support"`. Set `durable: true`. A false `approval` predicate does not park. The model called a tool that is not on `ai.agent({tools})`. `code` is `AgentLoopHalt`. Add the Flow name to `tools`, or stop offering it in the prompt. `result.stopReason` is `budget` and `result.cost` is the spend so far. Raise `budget.maxCostPerRun`, or treat the partial `output` as the reply. A prompt's `maxCostPerCall` is different: `fx.ask` throws `ai: prompt "…" exceeded maxCostPerCall …` (`AiBudgetExceededError`). The model call aborts. You may see `RUN_ERROR` with message `aborted` and code `AbortError`, or only a closed socket. The recorded `stopReason` is `aborted`. `readAgentEvents` yields `CUSTOM` unchanged when `name` is not `oke.subagent.started`, `oke.subagent.finished`, or `oke.subagent.error`. Those three become `subagent.started`, `subagent.finished`, and `subagent.error`. An object with no `type` is skipped. `data: [DONE]` is skipped. The iterator ends when the body ends, not because of that line. A second approve or deny after the decision is stored is `Conflict`. Do not retry it. A lost race while another worker holds the run is `JournalLeaseBusy` with `Retry-After`. Retry only that code. ## Learn more - [Agents](/docs/elements/ai/agents) — `fx.run`, tools, `maxSteps`, and approval - [Prompts](/docs/elements/ai/prompts) — tool-less `fx.ask` still throws on `maxCostPerCall` - [MCP](/docs/elements/ai/mcp) — external tools in the same loop ## Next