From 837d57dd5a2bcbbcb94c6452db91ff6b2f02830d Mon Sep 17 00:00:00 2001 From: lyon Date: Wed, 17 Jun 2026 13:30:40 +0800 Subject: [PATCH] feat(workbench): add realtime event stream --- internal/cloud/server-workbench-http.test.ts | 98 ++++++++ internal/cloud/server-workbench-http.ts | 207 +++++++++++++++++ internal/cloud/server.ts | 7 +- web/hwlab-cloud-web/scripts/check.ts | 10 +- web/hwlab-cloud-web/src/api/index.ts | 1 + .../src/api/workbench-events.ts | 101 +++++++++ web/hwlab-cloud-web/src/stores/workbench.ts | 210 +++++++++++++++++- 7 files changed, 619 insertions(+), 15 deletions(-) create mode 100644 web/hwlab-cloud-web/src/api/workbench-events.ts diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index b0c1c63a..dc8417ed 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/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) }; +} diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 619bde01..62b4c7f3 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/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"); } diff --git a/internal/cloud/server.ts b/internal/cloud/server.ts index d162e4c3..31a44b35 100644 --- a/internal/cloud/server.ts +++ b/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; diff --git a/web/hwlab-cloud-web/scripts/check.ts b/web/hwlab-cloud-web/scripts/check.ts index e2bfdcc6..67eeaeb1 100644 --- a/web/hwlab-cloud-web/scripts/check.ts +++ b/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; devDependencies?: Record }; 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(appSource, "getAgentTurn(traceId, projectId, inactivityTimeoutMs, activityRef)", "Turn status polling must use activityRef-backed inactivity timeout"); -assertIncludes(appSource, "for (;;) ", "trace subscription must keep unbounded polling without total timeout"); +assertIncludes(workbenchStoreSource, "connectWorkbenchEvents", "Workbench must use the unified SSE realtime entry"); +assertIncludes(workbenchStoreSource, "scheduleRealtimeGapHydration", "Workbench realtime disconnect/reconnect must trigger REST gap fill"); +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"); diff --git a/web/hwlab-cloud-web/src/api/index.ts b/web/hwlab-cloud-web/src/api/index.ts index d0e09349..8237c0ae 100644 --- a/web/hwlab-cloud-web/src/api/index.ts +++ b/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"; diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts new file mode 100644 index 00000000..90451888 --- /dev/null +++ b/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 | 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; + } +} diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 7c677dcf..b4b254e5 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/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(null); const activityRef = ref({ lastActivityAt: Date.now(), lastActivityIso: new Date().toISOString(), waitingFor: "idle", lastEventLabel: null as string | null }); - const activeAbort = ref(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>({}); const turnStatusAuthority = ref>({}); const traceAuthorityById = ref>>({}); const traceHydrationInFlight = new Set(); + const terminalRealtimeRefreshInFlight = new Set(); + 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 { @@ -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 { @@ -559,19 +570,15 @@ export const useWorkbenchStore = defineStore("workbench", () => { currentRequest.value = null; const activeTraceId = activeTraceIdFromWorkspace(workspace.value); if (activeTraceId) void clearActiveTrace(activeTraceId, "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) }); + restartRealtime("clear-conversation"); } 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 { @@ -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): 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 { + 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 { + 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).conversationId, activeConversationId.value)); + restartRealtime("trace-terminal"); } async function hydrateTerminalMessageDiagnostics(): Promise { @@ -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 { @@ -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;