root
307 lines
13 KiB
TypeScript
307 lines
13 KiB
TypeScript
// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; 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; PJ2026-010401080313 Workbench实时权威 draft-2026-07-08-p0-workbench-realtime-authority-v2.
|
|
// Responsibility: Workbench SSE client. Realtime events and sync replay are projection authority; REST routes are detail/history reads.
|
|
|
|
import { fetchJson, type ApiRequestOptions } from "@/api/client";
|
|
import type { ApiResult, ChatMessage, ProjectionDiagnostic, TraceEvent } from "@/types";
|
|
import { createCoalescedEventQueue } from "@/utils/scheduler/coalesced-event-queue";
|
|
import { composeWorkbenchScopedKey, firstScopePart } from "@/utils/workbench-key";
|
|
import { recordWorkbenchRuntimeDiagnostic, recordWorkbenchSseLifecycle } from "@/utils/workbench-performance";
|
|
import type { WorkbenchRealtimeCapabilities } from "@/config/runtime";
|
|
|
|
export interface WorkbenchRealtimeTraceSnapshot {
|
|
traceId?: string | null;
|
|
status?: string | null;
|
|
events?: TraceEvent[];
|
|
eventCount?: number | null;
|
|
lastEvent?: TraceEvent | null;
|
|
updatedAt?: string | null;
|
|
projection?: ProjectionDiagnostic | null;
|
|
projectionStatus?: string | null;
|
|
projectionHealth?: string | null;
|
|
staleMs?: number | null;
|
|
blocker?: ProjectionDiagnostic["blocker"] | null;
|
|
[key: string]: unknown;
|
|
}
|
|
|
|
export interface WorkbenchRealtimeEvent {
|
|
schema?: string | null;
|
|
eventType?: string | null;
|
|
eventId?: string | null;
|
|
sourceEventId?: string | null;
|
|
capabilities?: WorkbenchRealtimeCapabilities | null;
|
|
liveOnly?: boolean | null;
|
|
replay?: boolean | null;
|
|
lossPossible?: boolean | null;
|
|
type?: string;
|
|
contractVersion?: string | null;
|
|
realtimeAuthority?: string | null;
|
|
status?: string;
|
|
serverSentAt?: string | null;
|
|
eventCreatedAt?: string | null;
|
|
hwlabSessionId?: string | null;
|
|
sessionId?: string | null;
|
|
threadId?: string | null;
|
|
traceId?: string | null;
|
|
runId?: string | null;
|
|
commandId?: 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;
|
|
traceSeq?: number | null;
|
|
outboxSeq?: number | null;
|
|
cursor?: { traceSeq?: number | null; outboxSeq?: number | null; [key: string]: unknown };
|
|
entity?: {
|
|
family?: string | null;
|
|
id?: string | null;
|
|
version?: number | string | null;
|
|
entityVersion?: number | string | null;
|
|
outboxSeq?: number | string | null;
|
|
traceSeq?: number | string | null;
|
|
projectionRevision?: string | null;
|
|
committedAt?: string | null;
|
|
serverCommittedAt?: string | null;
|
|
authority?: string | null;
|
|
detailProjection?: boolean | null;
|
|
[key: string]: unknown;
|
|
} | null;
|
|
authority?: string | null;
|
|
detailProjection?: boolean | null;
|
|
projectionRevision?: string | null;
|
|
[key: string]: unknown;
|
|
}
|
|
|
|
export interface WorkbenchSyncReplayRequest {
|
|
sessionId?: string | null;
|
|
traceId?: string | null;
|
|
since?: number | null;
|
|
}
|
|
|
|
export interface WorkbenchSyncReplayResponse {
|
|
contractVersion?: string | null;
|
|
realtimeAuthority?: string | null;
|
|
scope?: Record<string, unknown> | null;
|
|
cursor?: Record<string, unknown> | null;
|
|
events?: WorkbenchRealtimeEvent[];
|
|
delta?: WorkbenchRealtimeEvent[] | {
|
|
sessions?: Record<string, unknown>[];
|
|
messages?: Record<string, unknown>[];
|
|
parts?: Record<string, unknown>[];
|
|
turns?: Record<string, unknown>[];
|
|
traceEvents?: Record<string, unknown>[];
|
|
checkpoints?: Record<string, unknown>[];
|
|
[key: string]: unknown;
|
|
};
|
|
families?: Record<string, unknown> | null;
|
|
authority?: Record<string, unknown> | null;
|
|
[key: string]: unknown;
|
|
}
|
|
|
|
export interface WorkbenchEventStreamOptions {
|
|
realtimeCapabilities: WorkbenchRealtimeCapabilities;
|
|
sessionId?: string | null;
|
|
traceId?: string | null;
|
|
afterSeq?: number | null;
|
|
flushMaxItemsPerChunk?: number | null;
|
|
flushMaxChunkMs?: number | null;
|
|
flushYieldMs?: number | null;
|
|
onIngress?: (frame: WorkbenchSseIngressFrame) => void;
|
|
onEvent: (event: WorkbenchRealtimeEvent, eventName: string) => void;
|
|
onOpen?: () => void;
|
|
onError?: (event: Event) => void;
|
|
}
|
|
|
|
export interface WorkbenchEventStream {
|
|
close: () => void;
|
|
}
|
|
|
|
export interface WorkbenchSseIngressFrame {
|
|
eventName: string;
|
|
rawText: string;
|
|
decodeStatus: "accepted" | "invalid-json" | "not-object";
|
|
}
|
|
|
|
interface QueuedRealtimeEvent {
|
|
payload: WorkbenchRealtimeEvent;
|
|
eventName: string;
|
|
}
|
|
|
|
const WORKBENCH_EVENT_NAMES = [
|
|
"hwlab.event.v1",
|
|
"workbench.connected",
|
|
"workbench.trace.snapshot",
|
|
"workbench.trace.event",
|
|
"workbench.trace.unavailable",
|
|
"workbench.message.snapshot",
|
|
"workbench.turn.snapshot",
|
|
"workbench.heartbeat",
|
|
"workbench.error"
|
|
];
|
|
|
|
export function connectWorkbenchEvents(options: WorkbenchEventStreamOptions): WorkbenchEventStream | null {
|
|
if (typeof EventSource === "undefined") return null;
|
|
const eventRoute = workbenchEventStreamPath(options);
|
|
const connectStartedAt = typeof performance === "undefined" ? Date.now() : performance.now();
|
|
recordWorkbenchSseLifecycle({ state: "connect", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId });
|
|
const source = new EventSource(eventRoute, { withCredentials: true });
|
|
const queue = createCoalescedEventQueue<QueuedRealtimeEvent>({
|
|
keyOf: (item) => realtimeCoalesceKey(item.payload, item.eventName),
|
|
maxItemsPerChunk: options.flushMaxItemsPerChunk,
|
|
maxChunkMs: options.flushMaxChunkMs,
|
|
yieldSchedule: (flush) => scheduleRealtimeFlushYield(flush, options.flushYieldMs),
|
|
onFlush: (items) => {
|
|
for (const item of items) options.onEvent(item.payload, item.eventName);
|
|
},
|
|
onDrain: (info) => {
|
|
if (info.eventCount === 0 && info.droppedCount === 0) return;
|
|
recordWorkbenchRuntimeDiagnostic({
|
|
module: "workbench-events",
|
|
sessionId: options.sessionId,
|
|
traceId: options.traceId,
|
|
outcome: "ok",
|
|
diagnostic: {
|
|
code: "workbench_sse_flush",
|
|
reason: info.reason,
|
|
source: "eventsource-coalesced-queue",
|
|
eventCount: info.eventCount,
|
|
deliveredCount: info.deliveredCount,
|
|
dropped: info.droppedCount,
|
|
chunkCount: info.chunkCount,
|
|
replacedByKey: info.replacedByKey,
|
|
flushDurationMs: Math.round(info.flushDurationMs),
|
|
maxItemsPerChunk: info.maxItemsPerChunk,
|
|
maxChunkMs: Math.round(info.maxChunkMs),
|
|
valuesRedacted: true
|
|
}
|
|
});
|
|
}
|
|
});
|
|
source.onopen = () => {
|
|
const openedAt = typeof performance === "undefined" ? Date.now() : performance.now();
|
|
recordWorkbenchSseLifecycle({ state: "open", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId, valueMs: Math.max(0, Math.round(openedAt - connectStartedAt)) });
|
|
options.onOpen?.();
|
|
};
|
|
source.onerror = (event) => {
|
|
recordWorkbenchSseLifecycle({ state: "error", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId, errorName: event.type });
|
|
options.onError?.(event);
|
|
};
|
|
const listeners = WORKBENCH_EVENT_NAMES.map((name) => {
|
|
const listener = (event: MessageEvent) => {
|
|
enqueueRealtimeFrame(event.data, name, options, queue);
|
|
};
|
|
source.addEventListener(name, listener);
|
|
return { name, listener };
|
|
});
|
|
source.onmessage = (event) => {
|
|
enqueueRealtimeFrame(event.data, "message", options, queue);
|
|
};
|
|
return {
|
|
close() {
|
|
queue.clear("stream-close");
|
|
for (const { name, listener } of listeners) source.removeEventListener(name, listener);
|
|
recordWorkbenchSseLifecycle({ state: "close", route: eventRoute, sessionId: options.sessionId, traceId: options.traceId });
|
|
source.close();
|
|
}
|
|
};
|
|
}
|
|
|
|
export function workbenchEventStreamPath(options: Pick<WorkbenchEventStreamOptions, "realtimeCapabilities" | "sessionId" | "traceId" | "afterSeq">): string {
|
|
const params = new URLSearchParams();
|
|
appendParam(params, "sessionId", options.sessionId);
|
|
appendParam(params, "traceId", options.traceId);
|
|
if (!options.realtimeCapabilities.liveKafkaSse && options.realtimeCapabilities.projectionRealtime) appendNumberParam(params, "afterSeq", options.afterSeq);
|
|
return `/v1/workbench/events?${params.toString()}`;
|
|
}
|
|
|
|
export function workbenchProjectionEventStreamPath(options: Pick<WorkbenchEventStreamOptions, "sessionId" | "traceId" | "afterSeq">): string {
|
|
const params = new URLSearchParams();
|
|
appendParam(params, "sessionId", options.sessionId);
|
|
appendParam(params, "traceId", options.traceId);
|
|
appendNumberParam(params, "afterSeq", options.afterSeq);
|
|
return `/v1/workbench/projection-events?${params.toString()}`;
|
|
}
|
|
|
|
export async function fetchWorkbenchSyncReplay(input: WorkbenchSyncReplayRequest, options: ApiRequestOptions = {}): Promise<ApiResult<WorkbenchSyncReplayResponse>> {
|
|
return fetchJson<WorkbenchSyncReplayResponse>(workbenchSyncReplayPath(input), {
|
|
...options,
|
|
timeoutName: options.timeoutName ?? "workbench sync replay"
|
|
});
|
|
}
|
|
|
|
export function workbenchSyncReplayPath(input: WorkbenchSyncReplayRequest): string {
|
|
const params = new URLSearchParams();
|
|
appendParam(params, "sessionId", input.sessionId);
|
|
appendParam(params, "traceId", input.traceId);
|
|
appendNumberParam(params, "since", input.since);
|
|
return `/v1/workbench/sync?${params.toString()}`;
|
|
}
|
|
|
|
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 appendNumberParam(params: URLSearchParams, key: string, value: number | null | undefined): void {
|
|
const parsed = Number(value);
|
|
if (!Number.isFinite(parsed) || parsed <= 0) return;
|
|
params.set(key, String(Math.trunc(parsed)));
|
|
}
|
|
|
|
function enqueueRealtimeFrame(raw: unknown, eventName: string, options: WorkbenchEventStreamOptions, queue: ReturnType<typeof createCoalescedEventQueue<QueuedRealtimeEvent>>): void {
|
|
const rawText = typeof raw === "string" ? raw : String(raw ?? "");
|
|
const decoded = decodeRealtimeEvent(rawText);
|
|
options.onIngress?.({ eventName, rawText, decodeStatus: decoded.status });
|
|
if (decoded.payload) queue.push({ payload: decoded.payload, eventName });
|
|
}
|
|
|
|
function decodeRealtimeEvent(raw: string): { payload: WorkbenchRealtimeEvent | null; status: WorkbenchSseIngressFrame["decodeStatus"] } {
|
|
try {
|
|
const value = JSON.parse(raw);
|
|
return value && typeof value === "object" && !Array.isArray(value)
|
|
? { payload: value as WorkbenchRealtimeEvent, status: "accepted" }
|
|
: { payload: null, status: "not-object" };
|
|
} catch {
|
|
return { payload: null, status: "invalid-json" };
|
|
}
|
|
}
|
|
|
|
function scheduleRealtimeFlushYield(flush: () => void, yieldMs: number | null | undefined): () => void {
|
|
if (typeof window !== "undefined" && typeof window.requestAnimationFrame === "function" && (!Number.isFinite(Number(yieldMs)) || Number(yieldMs) <= 0)) {
|
|
const id = window.requestAnimationFrame(() => flush());
|
|
return () => window.cancelAnimationFrame(id);
|
|
}
|
|
const delay = Math.max(0, Math.trunc(Number(yieldMs ?? 0)));
|
|
const id = setTimeout(() => flush(), delay);
|
|
return () => clearTimeout(id);
|
|
}
|
|
|
|
export function realtimeCoalesceKey(event: WorkbenchRealtimeEvent, eventName: string): string | null {
|
|
if (eventName === "hwlab.event.v1" || event.schema === "hwlab.event.v1") {
|
|
const eventId = firstScopePart(event.eventId, event.sourceEventId, event.event?.sourceEventId);
|
|
return eventId ? composeWorkbenchScopedKey("workbench.sse.live-kafka-event", eventId) : null;
|
|
}
|
|
const entity = event.entity;
|
|
const entityFamily = firstScopePart(entity?.family);
|
|
const entityId = firstScopePart(entity?.id);
|
|
const entityVersion = numericCursor(entity?.version ?? entity?.entityVersion);
|
|
if (entityFamily && entityId && entityVersion !== null) return composeWorkbenchScopedKey("workbench.realtime.entity", entityFamily, entityId, entityVersion);
|
|
const sessionId = firstScopePart(event.sessionId, event.message?.sessionId, event.snapshot?.sessionId, event.event?.sessionId, event.turn?.sessionId);
|
|
const traceId = firstScopePart(event.traceId, event.message?.traceId, event.snapshot?.traceId, event.event?.traceId, event.turn?.traceId);
|
|
const outboxSeq = numericCursor(event.cursor?.outboxSeq ?? event.outboxSeq);
|
|
const traceSeq = numericCursor(event.cursor?.traceSeq ?? event.traceSeq ?? event.event?.seq ?? event.event?.projectedSeq);
|
|
if (eventName === "workbench.turn.snapshot" && traceId) return composeWorkbenchScopedKey("workbench.sse.turn-snapshot", sessionId, traceId);
|
|
if (eventName === "workbench.trace.event" || eventName === "message") {
|
|
const eventSeq = outboxSeq ?? traceSeq;
|
|
return eventSeq === null ? null : composeWorkbenchScopedKey("workbench.sse.event", sessionId, traceId, eventName, eventSeq);
|
|
}
|
|
const messageId = firstScopePart(event.message?.messageId, event.message?.id);
|
|
return composeWorkbenchScopedKey("workbench.sse.snapshot", sessionId, traceId, eventName, messageId, outboxSeq ?? traceSeq);
|
|
}
|
|
|
|
function numericCursor(value: unknown): number | null {
|
|
const parsed = Number(value);
|
|
return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : null;
|
|
}
|