Streaming Task events

export TASK_ID='YOUR_TASK_UUID'
curl --fail-with-body -sSN "$MELSO_URL/api/tasks/$TASK_ID/stream" \
  -H "Authorization: Bearer $MELSO_TOKEN" -H "X-Workspace-ID: $WORKSPACE_ID" -H "Accept: text/event-stream"

event: run
id: eyJ2IjoxLCJrIjoi…
data: {"id":"4f5c…","task_id":"9b1e…","status":"completed","attempt":1,"failure_reason":"",…}

event: stream.end
data: {"reason":"max_duration"}

const base = process.env.MELSO_URL ?? "https://melso.ai";
const key = process.env.MELSO_API_KEY!; // workspace API key (mwk_…)
const taskId = process.env.MELSO_TASK_ID!;
let cursor = process.env.MELSO_STREAM_CURSOR ?? ""; // persist it with the Task ID

function handle(event: string, data: any) {
  if (event === "run") console.log("Run", data.id, data.status, data.failure_reason ?? "");
  if (event === "task.message" && data.role === "assistant") console.log("Reply:", data.content);
  if (event === "interaction" && data.status === "pending") console.log("Needs input:", data.message);
  if (event === "task" && (data.completed_at || data.cancelled_at)) return "closed";
}

for (let failures = 0; ; ) {
  const abort = new AbortController();
  let silence = setTimeout(() => abort.abort(), 45_000);
  let stop = false;
  try {
    const headers: Record<string, string> = { Authorization: `Bearer ${key}`, Accept: "text/event-stream" };
    if (cursor) headers["Last-Event-ID"] = cursor;
    const res = await fetch(`${base}/api/tasks/${taskId}/stream`, { headers, signal: abort.signal });
    if ([400, 401, 403, 404].includes(res.status)) throw Object.assign(new Error(await res.text()), { fatal: true });
    if (!res.ok || !res.body) throw new Error(`HTTP ${res.status}`);
    let buffer = "";
    const decoder = new TextDecoder();
    for await (const chunk of res.body) {
      clearTimeout(silence);
      silence = setTimeout(() => abort.abort(), 45_000);
      buffer += decoder.decode(chunk, { stream: true });
      for (let end = buffer.indexOf("\n\n"); end >= 0; end = buffer.indexOf("\n\n")) {
        const block = buffer.slice(0, end);
        buffer = buffer.slice(end + 2);
        let event = "", id = "", data = "";
        for (const line of block.split("\n")) {
          if (line.startsWith("event: ")) event = line.slice(7);
          else if (line.startsWith("id: ")) id = line.slice(4);
          else if (line.startsWith("data: ")) data += line.slice(6);
        }
        if (!event) continue; // a ping or the retry hint
        const payload = JSON.parse(data);
        if (event === "stream.end") {
          stop = payload.reason === "task_unavailable" || payload.reason === "credential_invalid";
          continue;
        }
        if (handle(event, payload) === "closed") stop = true;
        if (id) cursor = id;
      }
      if (stop) break;
    }
    failures = 0;
  } catch (error: any) {
    if (error?.fatal) throw error;
    failures++;
  } finally {
    clearTimeout(silence);
    abort.abort();
  }
  if (stop) break;
  await new Promise((resolve) => setTimeout(resolve, Math.min(30_000, 1_000 * 2 ** failures)));
}

{"type":"auth","payload":{"token":"YOUR_MELSO_API_KEY"}}