Events

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.

Quick start

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,
});

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 })),
  }),
);

curl -N -X POST http://localhost:6530/support/assist \
  -H "content-type: application/json" \
  -d '{"message":"Where is order 14?"}'

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]

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);
}

Event reference

Core names match AG-UI. Subagent notices are CUSTOM, not extra types.

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.

{
  "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 <durable Flow run id>.<tool call id>. 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.

await fx.agent.approve(id, { args: { amount: 4 } });
await fx.agent.deny(id, { reason: "over the limit" });

curl -X POST http://localhost:6530/agent/approvals/approve \
  -H "content-type: application/json" \
  -d '{"id":"VjFTdEdYUjhfWjVqZEhpNkItbXlULmNhbGxfMQ","args":{"amount":4}}'

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.

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.

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

return fx.json.stream(fx.run(support, "Where is order 14?", { stream: true }));

Text stream or event stream

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:

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

Learn more

• Agents — fx.run, tools, maxSteps, and approval
• Prompts — tool-less fx.ask still throws on maxCostPerCall
• MCP — external tools in the same loop

Next