Streaming

​

1. The three SSE surfaces

​

2. The invoke SSE flow (one-shot)

POST /api/v1/agents/agent_abc/invoke
Authorization: Bearer oag_...
Accept: text/event-stream
Content-Type: application/json

{ "message": "Tell me a haiku" }

data: {"type":"delta","text":"Quiet "}

data: {"type":"delta","text":"morning "}

data: {"type":"delta","text":"breeze..."}

data: {"type":"done","text":"Quiet morning breeze...","context_id":"ch-..."}

data: {"type":"done","is_error":true,
       "code":"agent_reply_error","error":"index out of range",
       "text":"index out of range","context_id":"ch-..."}

data: {"type":"error","code":"agent_offline","status_code":503,
       "message":"agent is offline"}

data: {"type":"done","is_error":true,"code":"agent_offline",
       "error":"agent is offline"}

​

Required client logic (invoke)

1. Append delta.text to a running buffer as deltas arrive.
2. Treat the first done frame as terminal — close the connection.
3. If done.is_error === true, surface done.code + done.error to the caller. Don’t retry blindly on agent_reply_error (in-band) but DO retry on service_timeout / agent_offline (transport).

​

3. The task / conversation SSE flow (named events)

event: message
data: { …per-frame JSON envelope… }

event: message
data: { …another frame… }

event: end
data: {"reason":"task_terminal"}

​

Terminal event: end

​

Required client logic (task / conversation)

const es = new EventSource(url);
let lastOffset = startingSince ?? 0;

es.addEventListener("message", (evt) => {
  const frame = JSON.parse(evt.data);
  lastOffset = frame.offset ?? lastOffset;
  handle(frame);  // UI render, etc.
});

es.addEventListener("end", (evt) => {
  const { reason } = JSON.parse(evt.data);
  console.log("stream ended:", reason);
  es.close();
  // Do NOT reconnect — terminal.
});

es.onerror = () => {
  // Connection dropped before `end`. Reconnect with since=lastOffset.
  setTimeout(() => connect(`${url}?since=${lastOffset}`), backoffMs);
};

​

4. The since cursor (drop / reconnect compensation)

• since=0 (or omitted) — replay the full history of the channel, then keep streaming new events. Useful for late attachers who want the entire transcript.
• since=N (N > 0) — replay every event at offset > N, then keep streaming. Use this on reconnect: pass the last observed frame.offset so you don’t see duplicates or miss any frames.

​

Edge cases

1. Connection dropped before any frame. lastOffset is still 0 — reconnect with since=0, which replays from the beginning. On invoke SSE (no offsets) you have to redrive the invoke altogether (the underlying chat_message wasn’t durable because you never got a message_id to retry idempotently against).
2. Last frame was agent_message_chunk then connection dropped. Reconnect with since=<that chunk's offset>. You will get every remaining chunk PLUS the final agent_reply. There is no risk of duplicating already-rendered text — each chunk has a distinct message_id.
3. Same client opens two SSE connections to the same task. Both get the full live stream. Message Service is fan-out — there’s no “you already have a subscription” semantic on the gateway.
4. backfill_truncated frame on reconnect (since OpenAPI v1.1 / ADR-0022). When the channel has been idle long enough that the ephemeral stream has aged out before your Last-Event-ID, the server emits a single backfill_truncated event before the normal replay_complete. Shape:
   event: backfill_truncated
   data: { "oldest_redis_offset": 4711, "since": 3120,
           "hint": "ephemeral chunks before oldest_redis_offset have aged out of the Redis stream" }
   
   Recovery options, in order of decreasing fidelity:
   • Replay surviving durable rows — fetch GET /messages?since=<since> (durable rows only); you’ll miss the per-token chunks but recover the final reply / non-chunk state. The server only emits this frame for ephemeral types, so chat_message / agent_reply / agent.input_required etc. are still in the durable log.
   • Fast-forward — resume with since=replay_complete.latest_offset and accept that the intermediate token chunks are gone. This is what the official SDK does by default since the chunks are rendering UX, not data of record.
   SDKs that don’t recognise backfill_truncated SHOULD treat the frame as a no-op message — it’s purely informational and doesn’t change the rest of the protocol.

​

5. Keepalive

​

Browser / EventSource

• Hand-roll the request via fetch + manual SSE parsing (recommended for production — gives you precise control over headers, retries, and reconnect timing); or
• Embed the token in the URL (?access_token=...) — avoid this: the URL ends up in CDN logs / proxy access logs / browser history.

​

Node.js (server-to-server)

let attempt = 0;
let lastOffset = 0;
let stop = false;

while (!stop) {
  try {
    const url = `${BASE}/api/v1/agents/${agentId}/tasks/${taskId}/events?since=${lastOffset}`;
    const res = await fetch(url, { headers: { Authorization: `Bearer ${TOKEN}` } });
    if (!res.body) throw new Error("no body");
    const reader = res.body.getReader();
    // ...parse SSE frames, update lastOffset on each `message`...
    // ...break out on `event: end`...
    attempt = 0;  // reset on a clean run
  } catch (e) {
    attempt++;
    await new Promise((r) => setTimeout(r, Math.min(2 ** attempt * 250, 30_000)));
  }
}

​

Go

​

Server-side considerations

​

6. Error frames vs done (invoke) / end (task & conversation)

• invoke: branch on data.type === "done" to decide “stop”.
• task: branch on event === "end" (the named SSE event) to decide “stop”; everything before is a message.
• conversation: same as task, but end fires only on explicit delete / upstream close, not on any per-turn terminal.

​

7. Worked example — robust task watcher (TypeScript)

import { Buffer } from "node:buffer";

interface Frame {
  type: string;
  message_id?: string;
  offset?: number;
  in_reply_to?: string;
  payload?: any;
  created_at?: string;
}

async function watchTask(taskId: string, agentId: string, token: string) {
  let lastOffset = 0;
  let attempt = 0;

  loop: while (true) {
    const url =
      `https://openapi.beeos.ai/api/v1/agents/${agentId}/tasks/${taskId}/events` +
      `?since=${lastOffset}`;
    try {
      const res = await fetch(url, {
        headers: { Authorization: `Bearer ${token}`, Accept: "text/event-stream" },
      });
      if (!res.ok || !res.body) throw new Error(`http ${res.status}`);
      const reader = res.body.getReader();
      const dec = new TextDecoder();
      let buf = "";

      while (true) {
        const { value, done } = await reader.read();
        if (done) break;
        buf += dec.decode(value, { stream: true });
        // SSE: frames separated by blank lines
        let idx: number;
        while ((idx = buf.indexOf("\n\n")) >= 0) {
          const block = buf.slice(0, idx);
          buf = buf.slice(idx + 2);
          let evName = "message";
          let data = "";
          for (const line of block.split("\n")) {
            if (line.startsWith("event:")) evName = line.slice(6).trim();
            else if (line.startsWith("data:")) data += line.slice(5).trim();
          }
          if (evName === "end") {
            console.log("task ended:", JSON.parse(data).reason);
            break loop;          // <- terminal; do NOT retry
          }
          const f: Frame = JSON.parse(data);
          if (f.offset != null) lastOffset = f.offset;
          render(f);
        }
      }
      // EOF without `end` — reconnect (atypical; treat as transient)
    } catch (e) {
      attempt++;
      const wait = Math.min(2 ** attempt * 250, 30_000);
      await new Promise((r) => setTimeout(r, wait));
      continue;
    }
    attempt = 0;
  }
}

​

8. Frame replay & idempotency

​

9. Common mistakes

• Polling lastOffset = 0 on reconnect. Re-replays the entire channel history, costs you bandwidth, and floods your UI with duplicate frames you’ve already rendered. Always track the latest observed frame.offset and pass it as since=.
• Treating event: end as a recoverable error. It’s not — it means there will be NO more frames. Reconnecting just yields another end (or 404 if the channel was evicted). Close the connection and move on.
• Branching on event name for invoke SSE. Invoke SSE emits unnamed frames; clients keyed on event === "message" will treat every frame as ignorable. Use JSON.parse(data).type === "done" for invoke; use event === "end" for task / conversation.
• Mixing Accept: text/event-stream with POST /tasks/.... The task SSE surface is the separate GET /events endpoint, not the create call. POST /tasks always returns a JSON 200 with the task ID; you then open SSE on GET /tasks/{id}/events.
• Embedding the token in the URL. Set Authorization: via a custom-header SSE client (or fetch + manual parsing). URLs end up in proxy / CDN access logs.

​

10. See also

• Calling Agents — non-streaming variants
• Conversations vs Tasks — when to pick which SSE surface
• Error Reference — every code that can appear in error / done.code frames
• OpenAPI contract: backend/openapi/beeos-platform-v1.yaml (search text/event-stream for the schemas)
• SSE DTO definitions: backend/services/openapi-gateway/internal/dto/sse.go