Stream each task's transcript over its own SSE; keep session note order

web/src/api.gen.ts

@@ -481,7 +481,7 @@ export interface paths {
             path?: never;
             cookie?: never;
         };
-        /** @description Server-sent events. Each event's id is its seq; data is the Event as JSON. Events after Last-Event-ID (or the after query parameter) are replayed before live events. */
+        /** @description Server-sent events. Each event's id is its seq; data is the Event as JSON. Events after Last-Event-ID (or the after query parameter) are replayed before live events. With task_id, only that task's events are sent. */
         get: operations["streamEvents"];
         put?: never;
         post?: never;
@@ -1629,6 +1629,7 @@ export interface operations {
         parameters: {
             query?: {
                 after?: number;
+                task_id?: string;
             };
             header?: {
                 "Last-Event-ID"?: number;
@@ -1647,6 +1648,7 @@ export interface operations {
                     "text/event-stream": string;
                 };
             };
+            404: components["responses"]["Error"];
         };
     };
     getStatus: {

web/src/api.ts

@@ -103,7 +103,12 @@ export const del = <P extends PathsWith<"delete">>(path: P, opts?: Options<Op<P,
 const staleAfterMs = 40_000;
 const retryAfterMs = 2_000;
 
-export function subscribe(after: number, onEvent: (ev: ByreEvent) => void, onState: (connected: boolean) => void): void {
+export function subscribe(
+  after: number,
+  taskId: string | undefined,
+  onEvent: (ev: ByreEvent) => void,
+  onState: (connected: boolean) => void,
+): () => void {
   let last = after;
   let heard = Date.now();
   let source: EventSource | undefined;
@@ -113,7 +118,11 @@ export function subscribe(after: number, onEvent: (ev: ByreEvent) => void, onSta
     clearTimeout(retry);
     source?.close();
     heard = Date.now();
-    const s = new EventSource(`/api/events?after=${last}`);
+    const query = new URLSearchParams({ after: String(last) });
+    if (taskId) {
+      query.set("task_id", taskId);
+    }
+    const s = new EventSource(`/api/events?${query}`);
     source = s;
     s.onopen = () => {
       heard = Date.now();
@@ -146,12 +155,21 @@ export function subscribe(after: number, onEvent: (ev: ByreEvent) => void, onSta
     connect();
   };
 
-  setInterval(check, 5_000);
-  addEventListener("online", connect);
-  document.addEventListener("visibilitychange", () => {
+  const onVisible = () => {
     if (document.visibilityState === "visible") {
       check();
     }
-  });
+  };
+
+  const interval = setInterval(check, 5_000);
+  addEventListener("online", connect);
+  document.addEventListener("visibilitychange", onVisible);
   connect();
+  return () => {
+    clearInterval(interval);
+    clearTimeout(retry);
+    removeEventListener("online", connect);
+    document.removeEventListener("visibilitychange", onVisible);
+    source?.close();
+  };
 }

web/src/app.ts

@@ -35,7 +35,7 @@ function route(): void {
   const match = /^#\/task\/(\w+)$/.exec(location.hash);
   const taskId = match?.[1];
   if (taskId !== state.selectedTaskId) {
-    void selectTask(taskId);
+    selectTask(taskId);
   }
 }
 
@@ -64,7 +64,7 @@ async function main(): Promise<void> {
   const status = await get("/api/admin/status");
   await Promise.all([loadMe(), loadProjects(), loadInbox()]);
 
-  subscribe(status.last_event_seq, handleEvent, (connected) => {
+  subscribe(status.last_event_seq, undefined, handleEvent, (connected) => {
     state.connected = connected;
     notify("connection");
   });

web/src/state.ts

@@ -1,4 +1,4 @@
-import { get, type AgentSession, type ByreEvent, type InboxItem, type Project, type Task, type User } from "./api.js";
+import { get, subscribe, type AgentSession, type ByreEvent, type InboxItem, type Project, type Task, type User } from "./api.js";
 
 export type Topic = "me" | "projects" | "tasks" | "sessions" | "inbox" | "selection" | "transcript" | "stream" | "connection" | "event";
 
@@ -87,7 +87,9 @@ export async function loadInbox(): Promise<void> {
   notify("inbox");
 }
 
-let liveDuringLoad: ByreEvent[] | undefined;
+let closeTaskStream: (() => void) | undefined;
+let taskBuffer: ByreEvent[] = [];
+let taskFlush = 0;
 
 const visitsKey = "byre.visits";
 
@@ -102,33 +104,48 @@ function markVisited(taskId: string | undefined): void {
   localStorage.setItem(visitsKey, JSON.stringify({ ...lastVisits(), [taskId]: Date.now() }));
 }
 
-export async function selectTask(taskId: string | undefined): Promise<void> {
+export function selectTask(taskId: string | undefined): void {
   markVisited(state.selectedTaskId);
   markVisited(taskId);
+  closeTaskStream?.();
+  closeTaskStream = undefined;
+  clearTimeout(taskFlush);
+  taskFlush = 0;
+  taskBuffer = [];
   state.selectedTaskId = taskId;
   state.transcript = [];
   state.lastTranscriptSeq = 0;
-  liveDuringLoad = undefined;
   notify("selection");
 
   if (!taskId) {
     return;
   }
 
-  const held: ByreEvent[] = [];
-  liveDuringLoad = held;
-  for (;;) {
-    const page = await get("/api/tasks/{taskId}/events", { path: { taskId }, query: { after: state.lastTranscriptSeq, limit: 1000 } });
-    if (state.selectedTaskId !== taskId || liveDuringLoad !== held) {
-      return;
-    }
-    appendTranscript(page);
-    if (page.length < 1000) {
-      break;
-    }
+  closeTaskStream = subscribe(0, taskId, queueTaskEvent, () => undefined);
+}
+
+function queueTaskEvent(ev: ByreEvent): void {
+  taskBuffer.push(ev);
+  if (!taskFlush) {
+    taskFlush = setTimeout(flushTaskEvents, 0);
   }
-  liveDuringLoad = undefined;
-  appendTranscript(held);
+}
+
+function flushTaskEvents(): void {
+  const events = taskBuffer;
+  taskBuffer = [];
+  taskFlush = 0;
+  let stored: ByreEvent[] = [];
+  for (const ev of events) {
+    if (ev.seq !== 0) {
+      stored.push(ev);
+      continue;
+    }
+    appendTranscript(stored);
+    stored = [];
+    notify("stream", ev);
+  }
+  appendTranscript(stored);
 }
 
 function appendTranscript(events: ByreEvent[]): void {
@@ -143,21 +160,10 @@ function appendTranscript(events: ByreEvent[]): void {
 
 export function handleEvent(ev: ByreEvent): void {
   if (ev.seq === 0) {
-    if (ev.task_id && ev.task_id === state.selectedTaskId && !liveDuringLoad) {
-      notify("stream", ev);
-    }
     notify("event", ev);
     return;
   }
 
-  if (ev.task_id && ev.task_id === state.selectedTaskId) {
-    if (liveDuringLoad) {
-      liveDuringLoad.push(ev);
-    } else {
-      appendTranscript([ev]);
-    }
-  }
-
   switch (ev.kind) {
     case "user.updated":
       if (ev.payload["login"] === state.me?.login) {

web/src/task.ts

@@ -116,13 +116,16 @@ export function mountTranscript(root: HTMLElement): () => void {
       transcript.scrollTop = transcript.scrollHeight;
     }
   };
-  const put = (el: HTMLElement) => {
-    if (spinner.parentNode === transcript) {
+  const putBefore = (el: HTMLElement, anchor: HTMLElement | undefined) => {
+    if (anchor?.parentNode === transcript) {
+      transcript.insertBefore(el, anchor);
+    } else if (spinner.parentNode === transcript) {
       transcript.insertBefore(el, spinner);
     } else {
       transcript.append(el);
     }
   };
+  const put = (el: HTMLElement) => putBefore(el, pending[0]?.el);
 
   const renderHead = () => {
     const task = state.selectedTaskId ? findTask(state.selectedTaskId) : undefined;
@@ -207,7 +210,7 @@ export function mountTranscript(root: HTMLElement): () => void {
     const entry = { el, text, imgs };
     pending.push(entry);
     ctx.run = undefined;
-    put(el);
+    putBefore(el, spinner);
     transcript.scrollTop = transcript.scrollHeight;
     return () => {
       el.remove();