diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 883f9a2e..e8986c98 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -177,6 +177,16 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq }, projectionStatus: { terminal: row.terminal, sealed: row.sealed } }); + await writeMessageRealtimeSnapshotSafe({ + writeEvent, + readModel, + actor: auth.actor, + sessionId: rowSessionId, + threadId: streamThreadId, + traceId: rowTraceId, + reason: "outbox-terminal", + cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq } + }); if (rowTraceId) void writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "terminal" }); } else { writeEvent("workbench.trace.event", { @@ -352,6 +362,37 @@ function handleWorkbenchRealtimeFailure(request, response, url, options = {}, er if (!response.destroyed) sendJson(response, classified.statusCode, workbenchError(classified.code, classified.message, { route: "/v1/workbench/events", errorName: error?.name ?? "Error" })); } +async function writeMessageRealtimeSnapshotSafe({ writeEvent, readModel, actor, sessionId, threadId = null, traceId = null, reason = "message", cursor = null } = {}) { + const safeSession = safeSessionId(sessionId); + if (!safeSession || typeof readModel?.queryFacts !== "function") return; + try { + const result = await queryFactsByRouteId(readModel, safeSession, { families: WORKBENCH_SESSION_MESSAGE_PAGE_FAMILIES }); + if (result.error) return; + const session = visibleFactSessions(result.facts, actor)[0] ?? null; + if (!session) return; + const messages = factMessagesForSession(session, result.facts); + const message = [...messages].reverse().find((item) => { + if (!isAssistantLikeRole(item?.role)) return false; + return !traceId || item.traceId === traceId; + }) ?? null; + if (!message) return; + writeEvent("workbench.message.snapshot", { + type: "message.snapshot", + sessionId: factSessionId(session), + threadId: threadId ?? session.threadId ?? null, + traceId: message.traceId ?? traceId ?? null, + reason, + cursor, + message + }); + } catch (error) { + writeEvent("workbench.error", { + type: "error", + error: { code: "workbench_message_snapshot_failed", message: error?.message ?? "Workbench realtime message snapshot failed." } + }); + } +} + function realtimeEventCreatedAt(payload) { const candidates = [payload?.event?.createdAt, payload?.snapshot?.lastEvent?.createdAt, payload?.turn?.updatedAt]; for (const value of candidates) { diff --git a/web/hwlab-cloud-web/src/api/workbench-events.ts b/web/hwlab-cloud-web/src/api/workbench-events.ts index 712aee32..00f47a36 100644 --- a/web/hwlab-cloud-web/src/api/workbench-events.ts +++ b/web/hwlab-cloud-web/src/api/workbench-events.ts @@ -1,7 +1,7 @@ // SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0. // Responsibility: Workbench SSE client. Realtime events accelerate UI projection; REST snapshots remain gap-fill authority. -import type { ProjectionDiagnostic, TraceEvent } from "@/types"; +import type { ChatMessage, ProjectionDiagnostic, TraceEvent } from "@/types"; import { recordWorkbenchSseLifecycle } from "@/utils/workbench-performance"; export interface WorkbenchRealtimeTraceSnapshot { @@ -29,6 +29,7 @@ export interface WorkbenchRealtimeEvent { traceId?: string | null; event?: TraceEvent | null; snapshot?: WorkbenchRealtimeTraceSnapshot | null; + message?: ChatMessage | null; turn?: Record | null; reason?: string | null; error?: { code?: string; message?: string; [key: string]: unknown } | null; @@ -54,6 +55,7 @@ const WORKBENCH_EVENT_NAMES = [ "workbench.trace.snapshot", "workbench.trace.event", "workbench.trace.unavailable", + "workbench.message.snapshot", "workbench.turn.snapshot", "workbench.heartbeat", "workbench.error" diff --git a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts index c024b117..a87e65e0 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-server-state.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-server-state.ts @@ -17,6 +17,7 @@ export type WorkbenchServerAction = | { type: "session.list"; sessions: WorkbenchSessionRecord[] } | { type: "session.detail"; session: WorkbenchSessionRecord | null | undefined } | { type: "session.messages"; sessionId: string | null | undefined; messages: ChatMessage[] } + | { type: "message.snapshot"; sessionId: string | null | undefined; message: ChatMessage | null | undefined } | { type: "session.status"; session: SessionStatusAuthority } | { type: "session.forget"; sessionId: string } | { type: "turn.status"; turn: TurnStatusAuthority } @@ -42,6 +43,8 @@ export function reduceWorkbenchServerState(state: WorkbenchServerState, action: return reduceSessionDetail(state, action.session); case "session.messages": return reduceSessionMessages(state, action.sessionId, action.messages); + case "message.snapshot": + return reduceMessageSnapshot(state, action.sessionId, action.message); case "session.status": return { ...state, sessionStatusById: { ...state.sessionStatusById, [action.session.sessionId]: action.session } }; case "session.forget": @@ -140,6 +143,34 @@ function reduceSessionMessages(state: WorkbenchServerState, sessionId: string | }; } +function reduceMessageSnapshot(state: WorkbenchServerState, sessionId: string | null | undefined, message: ChatMessage | null | undefined): WorkbenchServerState { + const id = sessionId || message?.sessionId; + if (!id || !message) return state; + const existing = state.messagesBySessionId[id] ?? state.sessionsById[id]?.messages ?? []; + const index = existing.findIndex((item) => messageMatchesSnapshot(item, message)); + const merged = index >= 0 + ? existing.map((item, itemIndex) => itemIndex === index ? mergeMessageSnapshot(item, message) : item) + : [...existing, message]; + return reduceSessionMessages(state, id, merged); +} + +function messageMatchesSnapshot(existing: ChatMessage, incoming: ChatMessage): boolean { + const existingId = existing.messageId ?? existing.id; + const incomingId = incoming.messageId ?? incoming.id; + if (existingId && incomingId && existingId === incomingId) return true; + return Boolean(existing.traceId && incoming.traceId && existing.traceId === incoming.traceId && existing.role === incoming.role); +} + +function mergeMessageSnapshot(existing: ChatMessage, incoming: ChatMessage): ChatMessage { + return { + ...existing, + ...incoming, + runnerTrace: existing.runnerTrace ?? incoming.runnerTrace ?? null, + traceAutoLifecycle: existing.traceAutoLifecycle, + updatedAt: incoming.updatedAt ?? existing.updatedAt + }; +} + function mergeSessionRecord(existing: WorkbenchSessionRecord | undefined, incoming: WorkbenchSessionRecord): WorkbenchSessionRecord { if (!existing) return incoming; const incomingMessages = Array.isArray(incoming.messages) ? incoming.messages : undefined; diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 63323bae..46b6ac71 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -1178,6 +1178,10 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project applyRealtimeTraceEvent(event.traceId, event.event, event.snapshot, event); return; } + if (event.type === "message.snapshot" && event.message) { + applyRealtimeMessageSnapshot(event); + return; + } if (event.type === "turn.snapshot" && event.turn) { applyRealtimeTurnSnapshot(event.turn); return; @@ -1192,6 +1196,15 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project } } + function applyRealtimeMessageSnapshot(event: WorkbenchRealtimeEvent): void { + const message = event.message ? normalizeChatMessage(event.message) : null; + const sessionId = normalizeWorkbenchSessionId(firstNonEmptyString(event.sessionId, message?.sessionId)); + if (!message || !sessionId) return; + const activeId = activeSessionId.value; + if (activeId && sessionId !== activeId) return; + reduceServerState({ type: "message.snapshot", sessionId, message }); + } + function applyRealtimeTraceSnapshot(traceId: string | null | undefined, snapshot: WorkbenchRealtimeEvent["snapshot"]): void { const id = firstNonEmptyString(traceId, snapshot?.traceId); if (!id || !snapshot) return;