Merge pull request #1386 from pikasTech/codex/1379-workbench-events

feat(workbench): add realtime event stream

internal/cloud/server-workbench-http.test.ts

@@ -237,6 +237,69 @@ test("workbench read model treats active session state as running", async () =>
   }
 });
 
+test("workbench realtime stream emits workspace, turn, trace, and live trace events", async () => {
+  const traceStore = createCodeAgentTraceStore();
+  const results = createCodeAgentChatResultStore();
+  const traceId = "trc_workbench_realtime";
+  const session = {
+    id: "ses_workbench_realtime",
+    projectId: "prj_hwpod_workbench",
+    agentId: "hwlab-code-agent",
+    status: "running",
+    ownerUserId: ACTOR.id,
+    conversationId: "cnv_workbench_realtime",
+    threadId: "thread-workbench-realtime",
+    lastTraceId: traceId,
+    updatedAt: "2026-06-17T02:00:00.000Z",
+    session: { sessionStatus: "running", lastTraceId: traceId, messages: [{ role: "user", text: "stream", traceId }] }
+  };
+  const workspace = {
+    workspaceId: "wsp_workbench_realtime",
+    ownerUserId: ACTOR.id,
+    projectId: "prj_hwpod_workbench",
+    revision: 7,
+    selectedConversationId: session.conversationId,
+    selectedAgentSessionId: session.id,
+    activeTraceId: traceId,
+    updatedAt: "2026-06-17T02:00:00.000Z",
+    workspace: { selectedConversationId: session.conversationId, selectedAgentSessionId: session.id, lastTraceId: traceId }
+  };
+  traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" });
+  results.set(traceId, { status: "running", traceId, ownerUserId: ACTOR.id, conversationId: session.conversationId, sessionId: session.id, threadId: session.threadId });
+  const accessController = {
+    store: {
+      async getOrCreateDefaultWorkspace() { return workspace; },
+      async getWorkspaceForUser() { return workspace; },
+      async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; }
+    },
+    async ensureBootstrap() {},
+    async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; },
+    async publicWorkbenchWorkspace(value) { return value; }
+  };
+  const server = createCloudApiServer({ accessController, traceStore, codeAgentChatResults: results, env: { HWLAB_WORKBENCH_SSE_HEARTBEAT_MS: "60000" } });
+  await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
+
+  try {
+    const { port } = server.address();
+    const eventsPromise = getSseEvents(port, `/v1/workbench/events?projectId=prj_hwpod_workbench&workspaceId=${encodeURIComponent(workspace.workspaceId)}&traceId=${encodeURIComponent(traceId)}`, 5);
+    setTimeout(() => traceStore.append(traceId, { type: "backend", status: "running", label: "backend:live" }), 50);
+    const events = await eventsPromise;
+    assert.deepEqual(events.slice(0, 4).map((event) => event.event), [
+      "workbench.connected",
+      "workbench.workspace.snapshot",
+      "workbench.trace.snapshot",
+      "workbench.turn.snapshot"
+    ]);
+    assert.equal(events[1].data.workspace.workspaceId, workspace.workspaceId);
+    assert.equal(events[2].data.traceId, traceId);
+    assert.equal(events[3].data.turn.traceId, traceId);
+    assert.equal(events[4].event, "workbench.trace.event");
+    assert.equal(events[4].data.event.label, "backend:live");
+  } finally {
+    await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve()));
+  }
+});
+
 async function getJson(port, path) {
   const response = await fetch(`http://127.0.0.1:${port}${path}`);
   return {
@@ -244,3 +307,38 @@ async function getJson(port, path) {
     body: await response.json()
   };
 }
+
+async function getSseEvents(port, path, count) {
+  const controller = new AbortController();
+  const response = await fetch(`http://127.0.0.1:${port}${path}`, { signal: controller.signal });
+  assert.equal(response.status, 200);
+  const reader = response.body.getReader();
+  const decoder = new TextDecoder();
+  let buffer = "";
+  const events = [];
+  try {
+    while (events.length < count) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      buffer += decoder.decode(value, { stream: true });
+      for (;;) {
+        const index = buffer.indexOf("\n\n");
+        if (index < 0) break;
+        const block = buffer.slice(0, index);
+        buffer = buffer.slice(index + 2);
+        events.push(parseSseBlock(block));
+        if (events.length >= count) break;
+      }
+    }
+  } finally {
+    controller.abort();
+  }
+  return events;
+}
+
+function parseSseBlock(block) {
+  const lines = block.split("\n");
+  const event = lines.find((line) => line.startsWith("event:"))?.slice(6).trim() ?? "message";
+  const data = lines.filter((line) => line.startsWith("data:")).map((line) => line.slice(5).trim()).join("\n");
+  return { event, data: JSON.parse(data) };
+}

internal/cloud/server-workbench-http.ts

@@ -17,6 +17,7 @@ import {
 const DEFAULT_PROJECT_ID = "prj_hwpod_workbench";
 const DEFAULT_PAGE_LIMIT = 50;
 const MAX_PAGE_LIMIT = 100;
+const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000;
 const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]);
 const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active"]);
 
@@ -61,6 +62,118 @@ export async function handleWorkbenchReadModelHttp(request, response, url, optio
   sendJson(response, 404, workbenchError("workbench_route_not_found", "Workbench read model route is not implemented.", { route: url.pathname }));
 }
 
+export async function handleWorkbenchRealtimeHttp(request, response, url, options = {}) {
+  if (request.method !== "GET") return methodNotAllowed(response, "GET");
+  const auth = await authenticateWorkbenchRead(request, response, options);
+  if (!auth) return;
+
+  const projectId = textValue(url.searchParams.get("projectId")) || DEFAULT_PROJECT_ID;
+  const requestedWorkspaceId = safeWorkspaceId(url.searchParams.get("workspaceId"));
+  const requestedTraceId = safeTraceId(url.searchParams.get("traceId"));
+  const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS);
+  const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
+  let closed = false;
+  let latestWorkspaceRevision = 0;
+  const cleanup = [];
+
+  response.writeHead(200, {
+    "content-type": "text/event-stream; charset=utf-8",
+    "cache-control": "no-store, no-transform",
+    connection: "keep-alive",
+    "x-accel-buffering": "no",
+    "x-content-type-options": "nosniff"
+  });
+
+  const writeEvent = (name, payload = {}) => {
+    if (closed || response.destroyed) return;
+    response.write(`event: ${name}\n`);
+    response.write(`data: ${JSON.stringify({ contractVersion: "workbench-events-v1", projectId, ...payload })}\n\n`);
+  };
+
+  writeEvent("workbench.connected", {
+    type: "connected",
+    status: "connected",
+    heartbeatMs,
+    filters: {
+      workspaceId: requestedWorkspaceId,
+      traceId: requestedTraceId
+    }
+  });
+
+  const initialWorkspace = await visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: requestedWorkspaceId, projectId });
+  if (initialWorkspace) {
+    latestWorkspaceRevision = Number(initialWorkspace.revision ?? 0) || 0;
+    writeEvent("workbench.workspace.snapshot", {
+      type: "workspace.snapshot",
+      reason: "initial",
+      workspace: await publicWorkbenchWorkspace(options, initialWorkspace, auth.actor),
+      cursor: { workspaceRevision: latestWorkspaceRevision }
+    });
+  }
+
+  const activeTraceId = requestedTraceId ?? safeTraceId(initialWorkspace?.activeTraceId ?? initialWorkspace?.workspace?.activeTraceId ?? initialWorkspace?.workspace?.lastTraceId);
+  if (activeTraceId) {
+    await writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" });
+    const unsubscribe = traceStore.subscribe(activeTraceId, (event, snapshot) => {
+      if (event) {
+        writeEvent("workbench.trace.event", {
+          type: "trace.event",
+          traceId: activeTraceId,
+          event,
+          snapshot: traceSnapshotSummary(snapshot),
+          cursor: { traceSeq: eventSeq(event, Number(snapshot?.eventCount ?? 1) - 1) }
+        });
+      } else {
+        writeEvent("workbench.trace.snapshot", {
+          type: "trace.snapshot",
+          traceId: activeTraceId,
+          reason: "trace-store-update",
+          snapshot: traceSnapshotForRealtime(snapshot),
+          cursor: { traceSeq: traceSnapshotLastSeq(snapshot) }
+        });
+      }
+      if (event?.terminal === true || TERMINAL_STATUSES.has(normalizeStatus(event?.status))) {
+        void writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "terminal" });
+      }
+    });
+    cleanup.push(unsubscribe);
+  }
+
+  const heartbeat = setInterval(async () => {
+    try {
+      const workspace = await visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: requestedWorkspaceId ?? initialWorkspace?.workspaceId, projectId });
+      const revision = Number(workspace?.revision ?? 0) || 0;
+      if (workspace && revision > latestWorkspaceRevision) {
+        latestWorkspaceRevision = revision;
+        writeEvent("workbench.workspace.snapshot", {
+          type: "workspace.snapshot",
+          reason: "revision-change",
+          workspace: await publicWorkbenchWorkspace(options, workspace, auth.actor),
+          cursor: { workspaceRevision: revision }
+        });
+      }
+      writeEvent("workbench.heartbeat", {
+        type: "heartbeat",
+        workspaceId: workspace?.workspaceId ?? requestedWorkspaceId ?? null,
+        workspaceRevision: revision || latestWorkspaceRevision,
+        traceId: activeTraceId ?? null,
+        observedAt: new Date().toISOString()
+      });
+    } catch (error) {
+      writeEvent("workbench.error", {
+        type: "error",
+        error: { code: "workbench_realtime_heartbeat_failed", message: error?.message ?? "Workbench realtime heartbeat failed." }
+      });
+    }
+  }, Math.max(1000, heartbeatMs));
+  cleanup.push(() => clearInterval(heartbeat));
+
+  response.on("close", () => {
+    closed = true;
+    for (const item of cleanup.splice(0)) item();
+  });
+}
+
 async function authenticateWorkbenchRead(request, response, options) {
   const access = options.accessController;
   if (!access?.authenticate) {
@@ -449,6 +562,95 @@ function durableTraceStatus(events) {
   return "running";
 }
 
+async function visibleWorkbenchWorkspace(options, actor, input = {}) {
+  const store = options.accessController?.store;
+  if (!store) return null;
+  const workspaceId = safeWorkspaceId(input.workspaceId);
+  if (workspaceId) {
+    return await store.getWorkspaceForUser?.({ workspaceId, ownerUserId: actor.id, actorRole: actor.role }) ?? null;
+  }
+  return await store.getOrCreateDefaultWorkspace?.({
+    ownerUserId: actor.id,
+    projectId: textValue(input.projectId) || DEFAULT_PROJECT_ID,
+    now: new Date().toISOString()
+  }) ?? null;
+}
+
+async function publicWorkbenchWorkspace(options, workspace, actor) {
+  if (!workspace) return null;
+  if (typeof options.accessController?.publicWorkbenchWorkspace === "function") {
+    return await options.accessController.publicWorkbenchWorkspace(workspace, actor);
+  }
+  return workspace;
+}
+
+async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, reason }) {
+  const context = await visibleTraceContext(options, actor, traceId);
+  if (!context.visible) {
+    writeEvent("workbench.trace.unavailable", {
+      type: "trace.unavailable",
+      reason,
+      traceId,
+      error: { code: "workbench_trace_not_found", message: "Workbench trace is not visible to the current actor." }
+    });
+    return;
+  }
+  writeEvent("workbench.trace.snapshot", {
+    type: "trace.snapshot",
+    reason,
+    traceId,
+    snapshot: traceSnapshotForRealtime(context.trace),
+    cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
+  });
+  writeEvent("workbench.turn.snapshot", {
+    type: "turn.snapshot",
+    reason,
+    traceId,
+    turn: turnSnapshot(context),
+    cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
+  });
+}
+
+async function visibleTraceContext(options, actor, traceId) {
+  const result = options.codeAgentChatResults?.get?.(traceId) ?? null;
+  const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor);
+  if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) return { visible: false };
+  const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
+  if (!resultVisible && !session) return { visible: false };
+  const trace = await traceSnapshot(options, traceId);
+  const status = turnStatus(result, session, trace);
+  return { visible: true, turnId: traceId, traceId, status, result, session, trace };
+}
+
+function traceSnapshotForRealtime(snapshot) {
+  const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
+  return {
+    ...snapshot,
+    events,
+    eventCount: Number.isFinite(Number(snapshot?.eventCount)) ? Number(snapshot.eventCount) : events.length,
+    lastEvent: snapshot?.lastEvent ?? events.at(-1) ?? null,
+    valuesRedacted: true,
+    secretMaterialStored: false
+  };
+}
+
+function traceSnapshotSummary(snapshot) {
+  const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
+  return {
+    traceId: snapshot?.traceId ?? null,
+    status: snapshot?.status ?? "missing",
+    eventCount: Number.isFinite(Number(snapshot?.eventCount)) ? Number(snapshot.eventCount) : events.length,
+    lastEvent: snapshot?.lastEvent ?? events.at(-1) ?? null,
+    updatedAt: snapshot?.updatedAt ?? null,
+    valuesRedacted: true
+  };
+}
+
+function traceSnapshotLastSeq(snapshot) {
+  const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
+  return events.reduce((max, event, index) => Math.max(max, eventSeq(event, index)), 0);
+}
+
 function turnStatus(result, session, trace) {
   return normalizeStatus(
     result?.status ??
@@ -550,6 +752,11 @@ function safePartId(value) {
   return /^prt_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
 }
 
+function safeWorkspaceId(value) {
+  const text = textValue(value);
+  return /^wsp_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
+}
+
 function hash(value) {
   return createHash("sha256").update(String(value)).digest("hex");
 }

internal/cloud/server.ts

@@ -66,7 +66,7 @@ import {
   handleCodeAgentTurnHttp,
   handleCodeAgentTraceHttp
 } from "./server-code-agent-http.ts";
-import { handleWorkbenchReadModelHttp } from "./server-workbench-http.ts";
+import { handleWorkbenchReadModelHttp, handleWorkbenchRealtimeHttp } from "./server-workbench-http.ts";
 import { handleM3IoControlHttp } from "./server-m3-http.ts";
 import { handleSkillsHttp } from "./server-skills-http.ts";
 import { handleProviderProfileCatalogHttp, handleProviderProfilesHttp } from "./provider-profile-management.ts";
@@ -486,6 +486,11 @@ async function handleRestAdapter(request, response, url, options) {
     return;
   }
 
+  if (url.pathname === "/v1/workbench/events") {
+    await handleWorkbenchRealtimeHttp(request, response, url, options);
+    return;
+  }
+
   if (url.pathname === "/v1/workbench/sessions" || url.pathname.startsWith("/v1/workbench/sessions/") || url.pathname.startsWith("/v1/workbench/turns/") || url.pathname.startsWith("/v1/workbench/traces/")) {
     await handleWorkbenchReadModelHttp(request, response, url, options);
     return;

web/hwlab-cloud-web/scripts/check.ts

@@ -17,6 +17,7 @@ const requiredFiles = Object.freeze([
   "src/api/client.ts",
   "src/api/auth.ts",
   "src/api/workbench.ts",
+  "src/api/workbench-events.ts",
   "src/api/agent.ts",
   "src/stores/index.ts",
   "src/stores/app.ts",
@@ -58,6 +59,7 @@ for (const file of requiredFiles) {
 const html = readWeb("index.html");
 const pkg = JSON.parse(readWeb("package.json")) as { dependencies?: Record<string, string>; devDependencies?: Record<string, string> };
 const appSource = readCloudWebAppSource(rootDir);
+const workbenchStoreSource = readWeb("src/stores/workbench.ts");
 const traceTimelineSource = readWeb("src/components/agent/TraceTimeline.vue");
 const messageTraceDebugSource = readWeb("src/components/workbench/MessageTraceDebugPanel.vue");
 
@@ -94,9 +96,11 @@ assertIncludes(appSource, "useTableLoader", "table loader composable must remain
 assertIncludes(appSource, "useForm", "form composable must remain available for CRUD dialogs");
 assertIncludes(appSource, "hwlab_session", "auth comments/code must preserve Web session cookie boundary");
 assertIncludes(appSource, "activityRef", "Code Agent inactivity-timeout activityRef must be preserved");
-assertIncludes(appSource, "activityRef: () => activityRef.value", "Workbench must pass activityRef into trace subscription");
+assertIncludes(workbenchStoreSource, "connectWorkbenchEvents", "Workbench must use the unified SSE realtime entry");
-assertIncludes(appSource, "getAgentTurn(traceId, projectId, inactivityTimeoutMs, activityRef)", "Turn status polling must use activityRef-backed inactivity timeout");
+assertIncludes(workbenchStoreSource, "scheduleRealtimeGapHydration", "Workbench realtime disconnect/reconnect must trigger REST gap fill");
-assertIncludes(appSource, "for (;;) ", "trace subscription must keep unbounded polling without total timeout");
+assertIncludes(workbenchStoreSource, "hydrateRealtimeGap", "Workbench realtime recovery must use REST snapshot/page gap fill");
+assert.doesNotMatch(workbenchStoreSource, /subscribeToTrace|TRACE_POLL_INTERVAL_MS/u, "Workbench store must not reintroduce active trace polling");
+assertIncludes(appSource, "/v1/workbench/events", "Workbench realtime client must use the RESTful same-origin events endpoint");
 assertIncludes(appSource, "/v1/agent/traces/", "trace hydration must use RESTful Cloud Web same-origin trace API");
 assert.doesNotMatch(appSource, /\/v1\/agent\/chat\/trace\//u, "Cloud Web must not use the legacy action-style chat trace API");
 assertIncludes(appSource, "/v1/workbench/workspace", "workspace bootstrap must use Cloud Web same-origin workbench API");

web/hwlab-cloud-web/src/api/index.ts

@@ -1,6 +1,7 @@
 export { fetchJson, fetchText, type ActivityRef, type ActivityRefSource, type ApiRequestOptions } from "./client";
 export { authAPI, HWLAB_WEB_SESSION_COOKIE, type WorkspaceBootstrapResponse } from "./auth";
 export { workbenchAPI } from "./workbench";
+export { connectWorkbenchEvents, type WorkbenchEventStream, type WorkbenchRealtimeEvent } from "./workbench-events";
 export { agentAPI } from "./agent";
 export { hwpodAPI } from "./hwpod";
 export { accessAPI } from "./access";

web/hwlab-cloud-web/src/api/workbench-events.ts

@@ -0,0 +1,101 @@
+// SPEC: PJ2026-010403 API契约 draft-2026-06-17-r0; PJ2026-010401 Web工作台 draft-2026-06-17-r0.
+// Responsibility: Workbench SSE client. Realtime events accelerate UI projection; REST snapshots remain gap-fill authority.
+
+import type { TraceEvent, WorkspaceRecord } from "@/types";
+
+export interface WorkbenchRealtimeTraceSnapshot {
+  traceId?: string | null;
+  status?: string | null;
+  events?: TraceEvent[];
+  eventCount?: number | null;
+  lastEvent?: TraceEvent | null;
+  updatedAt?: string | null;
+  [key: string]: unknown;
+}
+
+export interface WorkbenchRealtimeEvent {
+  type?: string;
+  status?: string;
+  projectId?: string;
+  workspace?: WorkspaceRecord | null;
+  workspaceId?: string | null;
+  workspaceRevision?: number | null;
+  traceId?: string | null;
+  event?: TraceEvent | null;
+  snapshot?: WorkbenchRealtimeTraceSnapshot | null;
+  turn?: Record<string, unknown> | null;
+  reason?: string | null;
+  error?: { code?: string; message?: string } | null;
+  cursor?: { workspaceRevision?: number | null; traceSeq?: number | null; [key: string]: unknown };
+  [key: string]: unknown;
+}
+
+export interface WorkbenchEventStreamOptions {
+  projectId: string;
+  workspaceId?: string | null;
+  conversationId?: string | null;
+  sessionId?: string | null;
+  traceId?: string | null;
+  onEvent: (event: WorkbenchRealtimeEvent, eventName: string) => void;
+  onOpen?: () => void;
+  onError?: (event: Event) => void;
+}
+
+export interface WorkbenchEventStream {
+  close: () => void;
+}
+
+const WORKBENCH_EVENT_NAMES = [
+  "workbench.connected",
+  "workbench.workspace.snapshot",
+  "workbench.trace.snapshot",
+  "workbench.trace.event",
+  "workbench.trace.unavailable",
+  "workbench.turn.snapshot",
+  "workbench.heartbeat",
+  "workbench.error"
+];
+
+export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): WorkbenchEventStream | null {
+  if (typeof EventSource === "undefined") return null;
+  const params = new URLSearchParams({ projectId: options.projectId });
+  appendParam(params, "workspaceId", options.workspaceId);
+  appendParam(params, "conversationId", options.conversationId);
+  appendParam(params, "sessionId", options.sessionId);
+  appendParam(params, "traceId", options.traceId);
+  const source = new EventSource(`/v1/workbench/events?${params.toString()}`, { withCredentials: true });
+  source.onopen = () => options.onOpen?.();
+  source.onerror = (event) => options.onError?.(event);
+  const listeners = WORKBENCH_EVENT_NAMES.map((name) => {
+    const listener = (event: MessageEvent) => {
+      const payload = parseRealtimeEvent(event.data);
+      if (payload) options.onEvent(payload, name);
+    };
+    source.addEventListener(name, listener);
+    return { name, listener };
+  });
+  source.onmessage = (event) => {
+    const payload = parseRealtimeEvent(event.data);
+    if (payload) options.onEvent(payload, "message");
+  };
+  return {
+    close() {
+      for (const { name, listener } of listeners) source.removeEventListener(name, listener);
+      source.close();
+    }
+  };
+}
+
+function appendParam(params: URLSearchParams, key: string, value: string | null | undefined): void {
+  const text = typeof value === "string" ? value.trim() : "";
+  if (text) params.set(key, text);
+}
+
+function parseRealtimeEvent(raw: string): WorkbenchRealtimeEvent | null {
+  try {
+    const value = JSON.parse(raw);
+    return value && typeof value === "object" ? value as WorkbenchRealtimeEvent : null;
+  } catch {
+    return null;
+  }
+}

web/hwlab-cloud-web/src/stores/workbench.ts

@@ -4,7 +4,8 @@
 import { computed, ref } from "vue";
 import { defineStore } from "pinia";
 import { api } from "@/api";
-import { mergeRunnerTrace, snapshotToRunnerTrace, subscribeToTrace, type TraceSnapshot } from "@/composables/useTraceSubscription";
+import { connectWorkbenchEvents, type WorkbenchEventStream, type WorkbenchRealtimeEvent } from "@/api/workbench-events";
+import { mergeRunnerTrace, snapshotToRunnerTrace, type TraceSnapshot } from "@/composables/useTraceSubscription";
 import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, ApiResult, ChatMessage, ConversationRecord, LiveSurface, ProviderProfile, TraceEvent, WorkspaceRecord } from "@/types";
 import { DEFAULT_WORKBENCH_PROJECT_ID, firstNonEmptyString, nextProtocolId, normalizeWorkbenchConversationId, rememberWorkbenchProjectId, resolveInitialWorkbenchProjectId, workspaceProjectId } from "@/utils";
 import { RECENT_DRAFTS_STORAGE_KEY, activeTraceIdFromWorkspace, defaultProviderProfileOptions, mergeConversationIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, selectedConversationIdFromWorkspace, shouldApplyWorkspaceSnapshot, shouldShowSessionListLoading, sortSessionTabs, stableConversationList, workspaceWithClearedActiveTrace, type DraftEntry, type ProviderProfileOption, type SessionStatusAuthority, type TurnStatusAuthority } from "./workbench-session";
@@ -38,13 +39,16 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   const chatPending = ref(false);
   const error = ref<string | null>(null);
   const activityRef = ref({ lastActivityAt: Date.now(), lastActivityIso: new Date().toISOString(), waitingFor: "idle", lastEventLabel: null as string | null });
-  const activeAbort = ref<AbortController | null>(null);
   const currentRequest = ref<{ traceId: string; conversationId: string | null; sessionId: string | null; threadId: string | null; status?: string | null } | null>(null);
   const workspaceSelectionEpoch = ref(0);
   const sessionStatusAuthority = ref<Record<string, SessionStatusAuthority>>({});
   const turnStatusAuthority = ref<Record<string, TurnStatusAuthority>>({});
   const traceAuthorityById = ref<Record<string, NonNullable<ChatMessage["runnerTrace"]>>>({});
   const traceHydrationInFlight = new Set<string>();
+  const terminalRealtimeRefreshInFlight = new Set<string>();
+  let realtimeStream: WorkbenchEventStream | null = null;
+  let realtimeKey = "";
+  let realtimeGapTimer: number | null = null;
 
   const visibleConversations = computed(() => conversations.value);
   const activeConversationId = computed(() => firstNonEmptyString(switchingConversationId.value, workspace.value?.selectedConversationId, workspace.value?.workspace?.selectedConversationId, messages.value.find((message) => message.conversationId)?.conversationId));
@@ -119,6 +123,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
       error.value = conversations.value.length > 0 ? null : conversationsResult.error ?? "session list unavailable";
     }
     await refreshProviderOptions();
+    restartRealtime("hydrate");
   }
 
   async function refreshLive(): Promise<void> {
@@ -148,6 +153,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
       void hydrateTraceEvents(messages.value);
       await refreshSelectedSessionStatus();
       await refreshConversations(selectedConversationIdFromWorkspace(workspace.value));
+      restartRealtime("create-session");
     }
   }
 
@@ -383,7 +389,12 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     alignOptimisticTurnMessages(traceId, response.data);
     applyTurnStatusSnapshot(traceId, response.data);
     void refreshConversations(conversationId);
-    subscribe(traceId, response.data);
+    if ((response.data as AgentChatResultResponse).terminal === true || isTerminalMessageStatus(response.data.status)) {
+      completeTrace(traceId, response.data as AgentChatResultResponse);
+      return;
+    }
+    restartRealtime("submit");
+    scheduleRealtimeGapHydration("submit");
   }
 
   async function cancelAgentMessage(message: ChatMessage): Promise<void> {
@@ -559,19 +570,15 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     currentRequest.value = null;
     const activeTraceId = activeTraceIdFromWorkspace(workspace.value);
     if (activeTraceId) void clearActiveTrace(activeTraceId, "clear-conversation");
-  }
+    restartRealtime("clear-conversation");
-
-  function subscribe(traceId: string, initial: AgentChatResponse): void {
-    activeAbort.value?.abort();
-    const controller = new AbortController();
-    activeAbort.value = controller;
-    void subscribeToTrace({ traceId, projectId: activeProjectId.value, initial, inactivityTimeoutMs: codeAgentTimeoutMs.value, activityRef: () => activityRef.value, signal: controller.signal, onActivity: () => recordActivity("trace-poll"), onSnapshot: (snapshot) => applyTraceSnapshot(traceId, snapshot), onComplete: (result) => completeTrace(traceId, result), onInfrastructureError: (message) => failTrace(traceId, message) });
   }
 
   function reattachTrace(traceId: string): void {
     const initial: AgentChatResponse = { status: "running", traceId, turnUrl: `/v1/agent/turns/${encodeURIComponent(traceId)}?projectId=${encodeURIComponent(activeProjectId.value)}` };
     if (!messages.value.some((message) => message.traceId === traceId)) messages.value.push(makeMessage("agent", "", "running", { traceId, title: "Code Agent", traceAutoLifecycle: "running" }));
-    subscribe(traceId, initial);
+    currentRequest.value = { traceId, conversationId: activeConversationId.value ?? null, sessionId: selectedSessionId.value ?? null, threadId: selectedThreadId.value ?? null, status: initial.status };
+    restartRealtime("reattach");
+    scheduleRealtimeGapHydration("reattach");
   }
 
   async function validateAndReattachTrace(traceId: string): Promise<void> {
@@ -588,6 +595,156 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     await clearActiveTrace(traceId, "reattach-terminal-result", workspaceSessionStatusFromChatStatus(statusFromResult(result.data.status)));
   }
 
+  function restartRealtime(reason: string): void {
+    const traceId = realtimeTraceId();
+    const workspaceId = workspace.value?.workspaceId ?? null;
+    const conversationId = activeConversationId.value ?? null;
+    const sessionId = selectedSessionId.value ?? null;
+    const key = [activeProjectId.value, workspaceId ?? "", conversationId ?? "", sessionId ?? "", traceId ?? ""].join("|");
+    if (key === realtimeKey && realtimeStream) return;
+    stopRealtime();
+    realtimeKey = key;
+    if (!workspaceId && !traceId) return;
+    realtimeStream = connectWorkbenchEvents({
+      projectId: activeProjectId.value,
+      workspaceId,
+      conversationId,
+      sessionId,
+      traceId,
+      onOpen: () => scheduleRealtimeGapHydration(`${reason}:open`),
+      onError: () => scheduleRealtimeGapHydration(`${reason}:error`),
+      onEvent: (event, eventName) => applyRealtimeEvent(event, eventName)
+    });
+    if (!realtimeStream) scheduleRealtimeGapHydration(`${reason}:eventsource-unavailable`);
+  }
+
+  function stopRealtime(): void {
+    realtimeStream?.close();
+    realtimeStream = null;
+    realtimeKey = "";
+  }
+
+  function realtimeTraceId(): string | null {
+    return firstNonEmptyString(currentRequest.value?.traceId, activeTraceIdFromMessages(messages.value, turnStatusAuthority.value), activeTraceIdFromWorkspace(workspace.value));
+  }
+
+  function applyRealtimeEvent(event: WorkbenchRealtimeEvent, eventName: string): void {
+    recordActivity(event.type ? `realtime:${event.type}` : `realtime:${eventName}`);
+    if (event.type === "workspace.snapshot" && event.workspace) {
+      applyRealtimeWorkspaceSnapshot(event.workspace);
+      return;
+    }
+    if (event.type === "trace.snapshot") {
+      applyRealtimeTraceSnapshot(event.traceId, event.snapshot);
+      return;
+    }
+    if (event.type === "trace.event") {
+      applyRealtimeTraceEvent(event.traceId, event.event, event.snapshot);
+      return;
+    }
+    if (event.type === "turn.snapshot" && event.turn) {
+      applyRealtimeTurnSnapshot(event.turn);
+      return;
+    }
+    if (event.type === "trace.unavailable") {
+      const traceId = firstNonEmptyString(event.traceId);
+      if (traceId) void clearActiveTrace(traceId, "realtime-trace-unavailable");
+    }
+  }
+
+  function applyRealtimeWorkspaceSnapshot(nextWorkspace: WorkspaceRecord): void {
+    if (!shouldApplyWorkspaceSnapshot({ requestEpoch: workspaceSelectionEpoch.value, currentEpoch: workspaceSelectionEpoch.value, currentConversationId: activeConversationId.value, workspace: nextWorkspace })) return;
+    const previousTraceId = realtimeTraceId();
+    workspace.value = nextWorkspace;
+    providerProfile.value = firstNonEmptyString(workspace.value?.providerProfile, workspace.value?.workspace?.providerProfile, providerProfile.value) ?? providerProfile.value;
+    rememberWorkbenchProjectId(workspaceProjectId(workspace.value, activeProjectId.value));
+    const nextTraceId = realtimeTraceId();
+    if (firstNonEmptyString(previousTraceId) !== firstNonEmptyString(nextTraceId)) {
+      restartRealtime("workspace-snapshot");
+      scheduleRealtimeGapHydration("workspace-snapshot");
+    }
+  }
+
+  function applyRealtimeTraceSnapshot(traceId: string | null | undefined, snapshot: WorkbenchRealtimeEvent["snapshot"]): void {
+    const id = firstNonEmptyString(traceId, snapshot?.traceId);
+    if (!id || !snapshot) return;
+    applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, snapshot));
+    if (isTerminalMessageStatus(snapshot.status)) void refreshTerminalTraceFromRest(id, "realtime-trace-snapshot");
+  }
+
+  function applyRealtimeTraceEvent(traceId: string | null | undefined, event: WorkbenchRealtimeEvent["event"], snapshot: WorkbenchRealtimeEvent["snapshot"]): void {
+    const id = firstNonEmptyString(traceId, event?.traceId, snapshot?.traceId);
+    if (!id) return;
+    const events = event ? [event] : Array.isArray(snapshot?.events) ? snapshot.events : [];
+    applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, snapshot ?? { traceId: id, status: event?.status, events }, events));
+    if (event?.terminal === true || isTerminalMessageStatus(event?.status) || isTerminalMessageStatus(snapshot?.status)) void refreshTerminalTraceFromRest(id, "realtime-trace-event");
+  }
+
+  function applyRealtimeTurnSnapshot(turn: Record<string, unknown>): void {
+    const traceId = firstNonEmptyString(turn.traceId);
+    if (!traceId) return;
+    const status = firstNonEmptyString(turn.status) ?? undefined;
+    applyTurnStatusSnapshot(traceId, {
+      traceId,
+      status,
+      running: turn.running === true,
+      terminal: turn.terminal === true,
+      conversationId: firstNonEmptyString(turn.conversationId) ?? undefined,
+      sessionId: firstNonEmptyString(turn.sessionId) ?? undefined,
+      threadId: firstNonEmptyString(turn.threadId) ?? undefined,
+      agentRun: turn.agentRun as AgentRunProvenance | undefined
+    } as AgentChatResultResponse);
+    if (turn.terminal === true || isTerminalMessageStatus(status)) void refreshTerminalTraceFromRest(traceId, "realtime-turn-snapshot");
+  }
+
+  async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise<void> {
+    if (terminalRealtimeRefreshInFlight.has(traceId)) return;
+    terminalRealtimeRefreshInFlight.add(traceId);
+    try {
+      const result = await api.agent.getAgentTurn(traceId, activeProjectId.value, 8000, () => activityRef.value);
+      if (!result.ok || !result.data) return;
+      applyTurnStatusSnapshot(traceId, result.data);
+      if (result.data.terminal === true || isTerminalMessageStatus(result.data.status)) {
+        completeTrace(traceId, result.data);
+      }
+    } finally {
+      terminalRealtimeRefreshInFlight.delete(traceId);
+      restartRealtime(reason);
+    }
+  }
+
+  function scheduleRealtimeGapHydration(reason: string): void {
+    if (typeof window === "undefined") {
+      void hydrateRealtimeGap(reason);
+      return;
+    }
+    if (realtimeGapTimer) window.clearTimeout(realtimeGapTimer);
+    realtimeGapTimer = window.setTimeout(() => {
+      realtimeGapTimer = null;
+      void hydrateRealtimeGap(reason);
+    }, 200);
+  }
+
+  async function hydrateRealtimeGap(reason: string): Promise<void> {
+    recordActivity(`realtime-gap:${reason}`);
+    const [workspaceResult] = await Promise.all([
+      api.workbench.workspace(activeProjectId.value),
+      hydrateTurnStatusAuthority(messages.value),
+      hydrateTraceEvents(messages.value)
+    ]);
+    if (workspaceResult.ok && workspaceResult.data?.workspace) applyRealtimeWorkspaceSnapshot(workspaceResult.data.workspace);
+    restartRealtime(`gap:${reason}`);
+  }
+
+  function installRealtimeVisibilityHandler(): void {
+    if (typeof document === "undefined") return;
+    document.addEventListener("visibilitychange", () => {
+      if (document.visibilityState !== "visible") return;
+      restartRealtime("visibility");
+      scheduleRealtimeGapHydration("visibility");
+    });
+  }
+
   function applyTraceSnapshot(traceId: string, snapshot: TraceSnapshot): void {
     const trace = snapshotToRunnerTrace(snapshot);
     rememberTurnStatus(traceId, snapshot);
@@ -619,6 +776,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     void clearActiveTrace(traceId, "trace-terminal", workspaceSessionStatusFromChatStatus(terminalStatus));
     void refreshSessionStatusById(result.sessionId ?? selectedSessionId.value);
     void refreshConversations(firstNonEmptyString((result as Record<string, unknown>).conversationId, activeConversationId.value));
+    restartRealtime("trace-terminal");
   }
 
   async function hydrateTerminalMessageDiagnostics(): Promise<void> {
@@ -726,6 +884,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     void hydrateTerminalMessageDiagnostics();
     void hydrateTraceEvents(messages.value);
     reattachRestoredActiveTrace();
+    restartRealtime("apply-selected-conversation");
   }
 
   async function persistSelectedConversation(conversation: ConversationRecord, selectedProjectId: string, existingResponse?: ApiResult<{ workspace?: WorkspaceRecord }>): Promise<void> {
@@ -795,6 +954,8 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     if (traceId) void validateAndReattachTrace(traceId);
   }
 
+  installRealtimeVisibilityHandler();
+
   return { projectId, workspace, conversations, messages, providerProfile, providerOptions, recentDrafts, codeAgentTimeoutMs, gatewayShellTimeoutMs, live, loading, sessionListLoading, conversationDetailLoading, chatPending, error, activeConversationId, selectedSessionId, selectedThreadId, sessionTabs, composer, hydrate, refreshLive, createSession, selectConversation, selectConversationById, deleteCurrentSession, refreshConversations, submitMessage, cancelAgentMessage, cancelRunningTrace, retryAgentMessage, setProviderProfile, refreshProviderOptions, pickDraft, clearRecentDrafts, clearConversation, recordActivity };
 });
 
@@ -912,6 +1073,33 @@ function activeTraceIdFromMessages(messages: ChatMessage[], turnStatusAuthority:
   return null;
 }
 
+function realtimeSnapshotToTraceSnapshot(traceId: string, snapshot: WorkbenchRealtimeEvent["snapshot"], overrideEvents?: TraceEvent[]): TraceSnapshot {
+  const source = snapshot ?? { traceId };
+  const events = Array.isArray(overrideEvents) ? overrideEvents : Array.isArray(source.events) ? source.events : [];
+  return {
+    ...source,
+    traceId: firstNonEmptyString(source.traceId, traceId) ?? traceId,
+    status: firstNonEmptyString(source.status) ?? undefined,
+    events,
+    eventCount: firstFiniteNumber(source.eventCount, events.length),
+    fullTraceLoaded: source.fullTraceLoaded === true,
+    hasMore: source.hasMore === true,
+    truncated: source.truncated === true,
+    nextSinceSeq: firstFiniteNumber(source.nextSinceSeq) ?? null,
+    range: recordValue(source.range) as TraceSnapshot["range"],
+    agentRun: asAgentRun(source.agentRun) ?? undefined,
+    traceStatus: firstNonEmptyString(source.traceStatus) ?? undefined,
+    retention: source.retention,
+    terminalEvidence: source.terminalEvidence,
+    finalResponse: source.finalResponse,
+    traceSummary: source.traceSummary,
+    error: normalizeAgentError(source.error) ?? undefined,
+    lastEventLabel: firstNonEmptyString(source.lastEventLabel, source.lastEvent?.label, source.lastEvent?.type, events.at(-1)?.label, events.at(-1)?.type) ?? undefined,
+    eventSource: "trace-api",
+    updatedAt: firstNonEmptyString(source.updatedAt) ?? new Date().toISOString()
+  };
+}
+
 function normalizedStatusText(value: unknown): string | null {
   const text = firstNonEmptyString(value);
   return text ? text.trim().toLowerCase().replace(/_/gu, "-") : null;