流式（SSE）

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"}

event: message
data: { …每 frame JSON 信封… }

event: message
data: { …另一个 frame… }

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

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

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

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