mirrors/pikasTech-HWLAB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(workbench): stream terminal message snapshots (#1996)

Browse Source
This commit is contained in:
Lyon
2026-06-23 23:56:08 +08:00
committed by GitHub
parent aed4e9bc3e
commit 21f349f3b1
4 changed files with 88 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/cloud/server-workbench-http.ts
+41
View File
@@ -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) {
web/hwlab-cloud-web/src/api/workbench-events.ts
+3 -1
View File
@@ -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<string, unknown> | 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"
web/hwlab-cloud-web/src/stores/workbench-server-state.ts
+31
View File
@@ -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;
web/hwlab-cloud-web/src/stores/workbench.ts
+13
View File
@@ -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;