root
1376 lines
56 KiB
TypeScript
1376 lines
56 KiB
TypeScript
/*
|
|
* SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-20-p2-terminal-outbox-recovery; draft-2026-06-27-p0-read-model-timeline-contract; draft-2026-06-28-p0-d518-session-timeline-consistency; PJ2026-010403 API契约 draft-2026-06-20-p2-terminal-outbox-recovery; draft-2026-06-27-p0-workbench-read-api-contract; PJ2026-010401 Web工作台 draft-2026-06-20-p2-terminal-outbox-recovery; draft-2026-06-27-p0-workbench-read-model-contract; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
|
|
* 职责: Workbench SSE and Kafka realtime transport over durable read-model facts.
|
|
*/
|
|
import { createHash } from "node:crypto";
|
|
|
|
import {
|
|
parsePositiveInteger,
|
|
safeConversationId,
|
|
safeOpaqueId,
|
|
safeSessionId,
|
|
safeTraceId,
|
|
sendJson
|
|
} from "./server-http-utils.ts";
|
|
import { createWorkbenchReadModel } from "./workbench-read-model.ts";
|
|
import { createWorkbenchRuntimeClient } from "./workbench-runtime-client.ts";
|
|
import { emitLiveKafkaOtelSpan } from "./kafka-event-bridge.ts";
|
|
import { createWorkbenchKafkaRefreshHandoff, workbenchKafkaRefreshErrorPayload } from "./workbench-kafka-refresh-handoff.ts";
|
|
import { buildWorkbenchSessionDetail, compactLaunchContext, includeMessagesForSessionDetail } from "./workbench-session-detail-response.ts";
|
|
import { projectionOutboxRealtimeEvents } from "./workbench-projection-outbox-events.ts";
|
|
import { durableTraceStatus, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
|
import { emitCodeAgentOtelSpan, emitHttpServerRequestSpan } from "./otel-trace.ts";
|
|
import {
|
|
workbenchRealtimeCapabilities
|
|
} from "./workbench-realtime-capabilities.ts";
|
|
import * as workbenchFacts from "./server-workbench-facts.ts";
|
|
|
|
const DEFAULT_PAGE_LIMIT = 50;
|
|
const DEFAULT_SESSION_LIST_LIMIT = 20;
|
|
const MAX_PAGE_LIMIT = 100;
|
|
const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000;
|
|
const WORKBENCH_REALTIME_DRAIN_TIMEOUT_MS = 2500;
|
|
const WORKBENCH_SESSION_LIST_PAGE_FAMILIES = Object.freeze(["sessions"]);
|
|
const WORKBENCH_SESSION_DETAIL_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
|
|
const WORKBENCH_SESSION_MESSAGE_PAGE_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
|
|
const WORKBENCH_TRACE_EVENT_METADATA_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]);
|
|
const WORKBENCH_TRACE_EVENT_PAGE_FAMILIES = Object.freeze(["traceEvents"]);
|
|
const activeWorkbenchRealtimeConnections = new Set();
|
|
|
|
const {
|
|
factSessionSummary,
|
|
factSessionDetail,
|
|
factMessagesForSession,
|
|
canonicalFactMessageGroupsForSession,
|
|
factMessageCanonicalGroupKey,
|
|
selectCanonicalFactMessage,
|
|
factMessageCanonicalScore,
|
|
factMessageHasFinalResponsePart,
|
|
workbenchLifecycleMessageId,
|
|
factMessageDto,
|
|
factMessageAuthorityText,
|
|
factPartsHaveFinalResponse,
|
|
factSyntheticTerminalFinalResponse,
|
|
isTerminalProjectionStatus,
|
|
firstFactPartText,
|
|
factMessageFinalResponseText,
|
|
factTerminalMessageForTrace,
|
|
factTerminalStatusFromMessageProjection,
|
|
factPartDto,
|
|
factTurnForTrace,
|
|
factTurnSnapshot,
|
|
factTraceSnapshot,
|
|
factTraceTimingProjection,
|
|
factTraceEventIsTerminalAuthority,
|
|
factTraceEventTerminalTimes,
|
|
factTraceEventDto,
|
|
factProjectionForTrace,
|
|
factTimingProjection,
|
|
factCombinedTimingProjection,
|
|
factTimingSource,
|
|
firstTimestampIso,
|
|
latestTimestampIso,
|
|
elapsedFactMs,
|
|
timestampIso,
|
|
factCheckpointForTrace,
|
|
workbenchProjectionStoreError,
|
|
nonNegativeIntegerOrNull,
|
|
factSessionId,
|
|
factLastTraceId,
|
|
factLatestTraceIdForSession,
|
|
factCurrentTraceForSession,
|
|
factMessageTraceCandidate,
|
|
factTraceCandidate,
|
|
compareTraceCandidatesDesc,
|
|
nonUnknownStatus,
|
|
factTurnId,
|
|
factUpdatedAt,
|
|
factSeq,
|
|
factProjectedSeq,
|
|
normalizeProjectionStatus,
|
|
normalizeProjectionHealth,
|
|
compareFactSessionsDesc,
|
|
compareFactMessagesAsc,
|
|
compareFactMessagesForSessionAsc,
|
|
factMessageTimelineKey,
|
|
factTimelineAnchorMessageForTrace,
|
|
factMessageTimelineRoleRank,
|
|
firstFiniteNumber,
|
|
compareOptionalNumberAsc,
|
|
compareFactPartsAsc,
|
|
compareFactTraceEventsAsc,
|
|
compareFactRecordsDesc,
|
|
compareTimestampAsc,
|
|
compareTimestampDesc,
|
|
compareOptionalTimestampAsc,
|
|
compareNumberAsc,
|
|
compareNumberDesc,
|
|
compareText,
|
|
timestampMs,
|
|
optionalTimestampMs,
|
|
numericValue,
|
|
mergeFactSets,
|
|
emptyFactSet,
|
|
factArray,
|
|
uniqueText,
|
|
eventSeq,
|
|
normalizeStatus,
|
|
normalizeTerminalStatus,
|
|
firstUserPreview,
|
|
latestUserPreview,
|
|
latestValidMessagePreview,
|
|
latestAssistantPreview,
|
|
sessionTitleFromMessages,
|
|
sessionPreviewFromMessages,
|
|
boundedPreviewText,
|
|
isAssistantLikeRole,
|
|
objectValue,
|
|
textValue,
|
|
messageAuthorityTextValue,
|
|
safeTurnId,
|
|
safeMessageId,
|
|
safePartId,
|
|
hash
|
|
} = workbenchFacts;
|
|
|
|
import {
|
|
authenticateWorkbenchRead,
|
|
blockedTraceEventProjection,
|
|
classifyWorkbenchReadModelFailure,
|
|
isAgentRunAliasSessionId,
|
|
logWorkbenchReadModelFailure,
|
|
methodNotAllowed,
|
|
nonNegativeInteger,
|
|
queryFactsByRouteId,
|
|
traceEventsReadModelBlocker,
|
|
visibleFactSessionForTrace,
|
|
visibleFactSessions,
|
|
workbenchError
|
|
} from "./server-workbench-read-http.ts";
|
|
|
|
export async function drainWorkbenchRealtimeConnections(options = {}) {
|
|
const reason = textValue(options.reason) || "server_shutdown";
|
|
const signal = textValue(options.signal) || null;
|
|
const timeoutMs = parsePositiveInteger(options.timeoutMs, WORKBENCH_REALTIME_DRAIN_TIMEOUT_MS);
|
|
const connections = [...activeWorkbenchRealtimeConnections].filter((connection) => connection?.isActive?.());
|
|
for (const connection of connections) {
|
|
try {
|
|
connection.close({ reason, signal });
|
|
} catch {
|
|
// Best effort: one broken client must not block shutdown drain for the rest.
|
|
}
|
|
}
|
|
const deadline = Date.now() + timeoutMs;
|
|
while (Date.now() < deadline) {
|
|
const activeAfterClose = connections.filter((connection) => activeWorkbenchRealtimeConnections.has(connection) && connection?.isActive?.()).length;
|
|
if (activeAfterClose === 0) break;
|
|
await sleepMs(Math.min(50, Math.max(1, deadline - Date.now())));
|
|
}
|
|
const activeAfter = connections.filter((connection) => activeWorkbenchRealtimeConnections.has(connection) && connection?.isActive?.()).length;
|
|
return {
|
|
ok: activeAfter === 0,
|
|
activeBefore: connections.length,
|
|
closed: Math.max(0, connections.length - activeAfter),
|
|
activeAfter,
|
|
timeout: activeAfter > 0,
|
|
reason,
|
|
signal
|
|
};
|
|
}
|
|
|
|
|
|
export async function handleWorkbenchRealtimeHttp(request, response, url, options = {}) {
|
|
try {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
const realtimeCapabilities = workbenchRealtimeCapabilities(options.env ?? process.env);
|
|
const projectionOnly = url.pathname === "/v1/workbench/projection-events";
|
|
if (!projectionOnly && realtimeCapabilities.liveKafkaSse) {
|
|
await handleLiveKafkaWorkbenchRealtimeHttp(request, response, url, options);
|
|
return;
|
|
}
|
|
if (!realtimeCapabilities.projectionRealtime) {
|
|
sendJson(response, 503, workbenchError("workbench_realtime_disabled", "No Workbench realtime SSE capability is enabled."));
|
|
return;
|
|
}
|
|
const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId"));
|
|
const requestedTraceId = safeTraceId(url.searchParams.get("traceId"));
|
|
const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS);
|
|
attachWorkbenchRealtimeOtelContext(request, {
|
|
route: projectionOnly ? "/v1/workbench/projection-events" : "/v1/workbench/events",
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId,
|
|
heartbeatMs,
|
|
realtimeSource: "projection-outbox"
|
|
});
|
|
const perf = options.backendPerformance;
|
|
const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options);
|
|
if (!auth) return;
|
|
|
|
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) {
|
|
sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench read model is keyed by sessionId/threadId/traceId only."));
|
|
return;
|
|
}
|
|
|
|
if (!requestedSessionId && !requestedTraceId) {
|
|
sendJson(response, 400, workbenchError("workbench_realtime_scope_required", "Workbench realtime requires sessionId or traceId."));
|
|
return;
|
|
}
|
|
const runtime = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
|
if (typeof runtime?.readAtomicWorkbenchProjectionSync !== "function") {
|
|
sendJson(response, 503, workbenchError("workbench_realtime_runtime_unconfigured", "Workbench realtime requires an atomic projection snapshot reader."));
|
|
return;
|
|
}
|
|
const requestedAfterSeq = workbenchRealtimeAfterSeq(request, url);
|
|
const outboxTailBatchSize = requiredPositiveRuntimeSetting(options.env ?? process.env, "HWLAB_WORKBENCH_OUTBOX_TAIL_BATCH_SIZE");
|
|
let closed = false;
|
|
let realtimeCloseReason = "client_close";
|
|
let realtimeCloseSignal = null;
|
|
let realtimeSessionId = requestedSessionId ?? null;
|
|
let realtimeTraceId = requestedTraceId ?? null;
|
|
let realtimeThreadId = null;
|
|
const realtimeStartedAtMs = Date.now();
|
|
const cleanup = [];
|
|
|
|
let streamSessionId = requestedSessionId ?? null;
|
|
let streamThreadId = null;
|
|
let activeTraceId = requestedTraceId ?? null;
|
|
let outboxCursor = requestedAfterSeq;
|
|
let scanRunning = false;
|
|
let scanRequested = false;
|
|
let initialScanState = "pending";
|
|
let initialScanWakeRequested = false;
|
|
let realtimeConnection = null;
|
|
|
|
const isActive = () => !closed && !response.destroyed && !response.writableEnded;
|
|
const closeConnection = () => {
|
|
if (closed) return;
|
|
closed = true;
|
|
emitWorkbenchRealtimeClosedOtelSpan(request, options.env, {
|
|
reason: realtimeCloseReason,
|
|
signal: realtimeCloseSignal,
|
|
startedAtMs: realtimeStartedAtMs,
|
|
sessionId: realtimeSessionId,
|
|
threadId: realtimeThreadId,
|
|
traceId: realtimeTraceId,
|
|
activeConnectionCount: activeWorkbenchRealtimeConnections.size
|
|
});
|
|
for (const item of cleanup.splice(0)) item();
|
|
};
|
|
|
|
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"
|
|
});
|
|
response.once("close", closeConnection);
|
|
request.once?.("aborted", closeConnection);
|
|
request.socket?.once?.("close", closeConnection);
|
|
cleanup.push(() => request.off?.("aborted", closeConnection));
|
|
cleanup.push(() => request.socket?.off?.("close", closeConnection));
|
|
if (typeof response.flushHeaders === "function") response.flushHeaders();
|
|
|
|
const writeEvent = async (name, payload = {}) => {
|
|
if (!isActive()) return false;
|
|
const startedAt = nowMs();
|
|
const eventCreatedAt = realtimeEventCreatedAt(payload);
|
|
const traceSeq = realtimeTraceSeq(payload);
|
|
const eventId = realtimeEventId(payload);
|
|
const block = `event: ${name}\n${eventId ? `id: ${eventId}\n` : ""}data: ${JSON.stringify({ contractVersion: "workbench-events-v1", serverSentAt: new Date().toISOString(), eventCreatedAt, traceSeq, ...payload })}\n\n`;
|
|
const writable = response.write(block);
|
|
if (!writable) await waitForResponseDrain(response, isActive);
|
|
perf?.recordPhase({ phase: "sse_write", durationMs: nowMs() - startedAt, outcome: isActive() ? "ok" : "closed" });
|
|
return isActive();
|
|
};
|
|
|
|
realtimeConnection = {
|
|
close(fields = {}) {
|
|
if (!isActive()) return false;
|
|
const reason = textValue(fields.reason) || "server_shutdown";
|
|
realtimeCloseReason = reason;
|
|
realtimeCloseSignal = textValue(fields.signal) || null;
|
|
void writeEvent("workbench.server_draining", {
|
|
type: "server.draining",
|
|
status: "closing",
|
|
reason,
|
|
signal: realtimeCloseSignal,
|
|
sessionId: realtimeSessionId,
|
|
threadId: realtimeThreadId,
|
|
traceId: realtimeTraceId,
|
|
observedAt: new Date().toISOString()
|
|
}).finally(() => {
|
|
if (!response.writableEnded) response.end();
|
|
});
|
|
return true;
|
|
},
|
|
isActive
|
|
};
|
|
activeWorkbenchRealtimeConnections.add(realtimeConnection);
|
|
cleanup.push(() => activeWorkbenchRealtimeConnections.delete(realtimeConnection));
|
|
|
|
await writeEvent("workbench.connected", {
|
|
type: "connected",
|
|
status: "connected",
|
|
realtimeSource: "projection-outbox",
|
|
snapshotSource: "atomic-projection-sync",
|
|
heartbeatMs,
|
|
cursor: { outboxSeq: requestedAfterSeq },
|
|
filters: { sessionId: requestedSessionId, traceId: requestedTraceId }
|
|
});
|
|
|
|
realtimeSessionId = streamSessionId ?? requestedSessionId ?? null;
|
|
realtimeTraceId = activeTraceId ?? requestedTraceId ?? null;
|
|
realtimeThreadId = streamThreadId ?? null;
|
|
attachWorkbenchRealtimeOtelContext(request, {
|
|
sessionId: streamSessionId ?? requestedSessionId,
|
|
traceId: activeTraceId,
|
|
threadId: streamThreadId,
|
|
heartbeatMs,
|
|
realtimeSource: "projection-outbox"
|
|
});
|
|
emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env);
|
|
|
|
const syncParams = (extra = {}) => ({
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId,
|
|
afterOutboxSeq: outboxCursor,
|
|
afterSeq: outboxCursor,
|
|
limit: outboxTailBatchSize,
|
|
actor: { id: auth.actor?.id, role: auth.actor?.role ?? "user" },
|
|
...extra
|
|
});
|
|
const updateStreamScope = (snapshot) => {
|
|
const session = factArray(snapshot?.facts?.sessions)[0] ?? null;
|
|
streamSessionId = textValue(session?.sessionId) || requestedSessionId || streamSessionId;
|
|
streamThreadId = safeOpaqueId(session?.threadId) || textValue(session?.threadId) || streamThreadId;
|
|
activeTraceId = requestedTraceId || textValue(session?.lastTraceId) || activeTraceId;
|
|
realtimeSessionId = streamSessionId;
|
|
realtimeThreadId = streamThreadId;
|
|
realtimeTraceId = activeTraceId;
|
|
};
|
|
const emitProjectionSnapshot = async (snapshot) => {
|
|
updateStreamScope(snapshot);
|
|
for (const item of projectionOutboxRealtimeEvents(snapshot, { includeSnapshot: true, includeEvents: false })) {
|
|
if (!await writeEvent(item.name, item.payload)) break;
|
|
}
|
|
outboxCursor = nonNegativeInteger(snapshot?.cutoffOutboxSeq ?? snapshot?.cursorOutboxSeq);
|
|
};
|
|
const recoverCursor = async () => {
|
|
const snapshot = await runtime.readAtomicWorkbenchProjectionSync(syncParams({ afterOutboxSeq: 0, afterSeq: 0, snapshotOnly: true, deltaOnly: false }));
|
|
await emitProjectionSnapshot(snapshot);
|
|
};
|
|
const runProjectionScan = async ({ initial = false } = {}) => {
|
|
const snapshotOnly = initial && requestedAfterSeq <= 0;
|
|
if (snapshotOnly) {
|
|
await recoverCursor();
|
|
return;
|
|
}
|
|
let hasMore = true;
|
|
while (hasMore && isActive()) {
|
|
const snapshot = await runtime.readAtomicWorkbenchProjectionSync(syncParams({ snapshotOnly: false, deltaOnly: true }));
|
|
const nextCursor = nonNegativeInteger(snapshot?.cursorOutboxSeq);
|
|
if (nextCursor < outboxCursor) {
|
|
await recoverCursor();
|
|
return;
|
|
}
|
|
for (const item of projectionOutboxRealtimeEvents(snapshot)) {
|
|
if (!await writeEvent(item.name, item.payload)) return;
|
|
}
|
|
outboxCursor = nextCursor;
|
|
hasMore = snapshot?.hasMore === true;
|
|
}
|
|
};
|
|
const emitProjectionScanError = async (error) => {
|
|
await writeEvent("workbench.error", { type: "error", realtimeSource: "projection-outbox", sessionId: streamSessionId, traceId: activeTraceId, error: { code: error?.code ?? "workbench_outbox_scan_failed", message: error?.message ?? "Workbench projection outbox scan failed.", valuesRedacted: true } });
|
|
};
|
|
const scanProjectionOutbox = async () => {
|
|
if (!isActive()) return;
|
|
if (initialScanState !== "complete") {
|
|
if (initialScanState === "pending") initialScanWakeRequested = true;
|
|
return;
|
|
}
|
|
if (scanRunning) {
|
|
scanRequested = true;
|
|
return;
|
|
}
|
|
scanRunning = true;
|
|
try {
|
|
do {
|
|
scanRequested = false;
|
|
await runProjectionScan({ initial: false });
|
|
} while (scanRequested && isActive());
|
|
} catch (error) {
|
|
await emitProjectionScanError(error);
|
|
if (!response.writableEnded) response.end();
|
|
} finally {
|
|
scanRunning = false;
|
|
}
|
|
};
|
|
const requestProjectionScan = () => {
|
|
if (!isActive()) return;
|
|
if (initialScanState !== "complete") {
|
|
if (initialScanState === "pending") initialScanWakeRequested = true;
|
|
return;
|
|
}
|
|
scanRequested = true;
|
|
void scanProjectionOutbox();
|
|
};
|
|
const unsubscribeProjectionCommits = options.kafkaEventBridge?.subscribeProjectionCommits?.((signal = {}) => {
|
|
if (workbenchProjectionSignalMatches(signal, requestedSessionId, requestedTraceId)) requestProjectionScan();
|
|
});
|
|
if (typeof unsubscribeProjectionCommits === "function") cleanup.push(unsubscribeProjectionCommits);
|
|
|
|
try {
|
|
await runProjectionScan({ initial: true });
|
|
initialScanState = "complete";
|
|
} catch (error) {
|
|
initialScanState = "failed";
|
|
await emitProjectionScanError(error);
|
|
if (!response.writableEnded) response.end();
|
|
return;
|
|
}
|
|
if (initialScanState === "complete" && initialScanWakeRequested && isActive()) {
|
|
initialScanWakeRequested = false;
|
|
scanRequested = true;
|
|
await scanProjectionOutbox();
|
|
}
|
|
if (!isActive()) return;
|
|
|
|
const heartbeat = setInterval(() => {
|
|
void writeEvent("workbench.heartbeat", {
|
|
type: "heartbeat",
|
|
sessionId: requestedSessionId ?? null,
|
|
traceId: activeTraceId ?? null,
|
|
observedAt: new Date().toISOString()
|
|
});
|
|
}, Math.max(1000, heartbeatMs));
|
|
heartbeat.unref?.();
|
|
cleanup.push(() => clearInterval(heartbeat));
|
|
} catch (error) {
|
|
handleWorkbenchRealtimeFailure(request, response, url, options, error);
|
|
}
|
|
}
|
|
|
|
async function handleLiveKafkaWorkbenchRealtimeHttp(request, response, url, options = {}) {
|
|
const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId"));
|
|
const requestedTraceId = safeTraceId(url.searchParams.get("traceId"));
|
|
const realtimeCapabilities = workbenchRealtimeCapabilities(options.env ?? process.env);
|
|
const heartbeatMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_HEARTBEAT_MS, DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS);
|
|
attachWorkbenchRealtimeOtelContext(request, {
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId,
|
|
heartbeatMs,
|
|
realtimeSource: "live-kafka-sse",
|
|
realtimeCapabilities
|
|
});
|
|
const perf = options.backendPerformance;
|
|
const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options);
|
|
if (!auth) return;
|
|
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) {
|
|
sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench realtime is keyed by sessionId/traceId only."));
|
|
return;
|
|
}
|
|
if (!requestedSessionId && !requestedTraceId) {
|
|
sendJson(response, 400, workbenchError("workbench_realtime_scope_required", "Workbench realtime requires sessionId or traceId."));
|
|
return;
|
|
}
|
|
const authorization = await authorizeLiveKafkaWorkbenchRealtimeScope(options, auth.actor, {
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId
|
|
});
|
|
if (!authorization.ok) {
|
|
sendJson(response, authorization.status, authorization.body);
|
|
return;
|
|
}
|
|
const authorizedSessionId = authorization.sessionId;
|
|
const bridge = options.kafkaEventBridge;
|
|
if (bridge?.capabilities?.liveKafkaSse !== true || typeof bridge?.subscribeLiveHwlabEvents !== "function") {
|
|
sendJson(response, 503, workbenchError("workbench_live_kafka_unconfigured", "Workbench live Kafka fanout is not configured."));
|
|
return;
|
|
}
|
|
if (realtimeCapabilities.kafkaRefreshReplay) {
|
|
await handleKafkaRefreshReplayWorkbenchRealtimeHttp(request, response, options, {
|
|
bridge,
|
|
heartbeatMs,
|
|
realtimeCapabilities,
|
|
requestedSessionId: authorizedSessionId,
|
|
requestedTraceId
|
|
});
|
|
return;
|
|
}
|
|
|
|
let closed = false;
|
|
let realtimeCloseReason = "client_close";
|
|
let realtimeCloseSignal = null;
|
|
const realtimeStartedAtMs = Date.now();
|
|
const cleanup = [];
|
|
const isActive = () => !closed && !response.destroyed && !response.writableEnded;
|
|
const closeConnection = () => {
|
|
if (closed) return;
|
|
closed = true;
|
|
emitWorkbenchRealtimeClosedOtelSpan(request, options.env, {
|
|
reason: realtimeCloseReason,
|
|
signal: realtimeCloseSignal,
|
|
startedAtMs: realtimeStartedAtMs,
|
|
sessionId: authorizedSessionId,
|
|
threadId: null,
|
|
traceId: requestedTraceId,
|
|
activeConnectionCount: activeWorkbenchRealtimeConnections.size
|
|
});
|
|
for (const item of cleanup.splice(0)) item();
|
|
};
|
|
|
|
response.once("close", closeConnection);
|
|
request.once?.("aborted", closeConnection);
|
|
request.socket?.once?.("close", closeConnection);
|
|
cleanup.push(() => request.off?.("aborted", closeConnection));
|
|
cleanup.push(() => request.socket?.off?.("close", closeConnection));
|
|
|
|
const bufferedEnvelopes = [];
|
|
let liveDeliveryStarted = false;
|
|
let enqueueEvent = null;
|
|
const unsubscribe = bridge.subscribeLiveHwlabEvents((envelope, transport) => {
|
|
if (!liveKafkaEnvelopeMatches(envelope, authorizedSessionId, requestedTraceId)) return;
|
|
if (!liveDeliveryStarted || typeof enqueueEvent !== "function") {
|
|
bufferedEnvelopes.push({ envelope, transport });
|
|
return;
|
|
}
|
|
void enqueueEvent("hwlab.event.v1", envelope, transport);
|
|
});
|
|
if (typeof unsubscribe === "function") cleanup.push(unsubscribe);
|
|
|
|
try {
|
|
await (bridge.liveReady ?? bridge.ready);
|
|
} catch (error) {
|
|
closeConnection();
|
|
throw error;
|
|
}
|
|
if (!isActive()) {
|
|
closeConnection();
|
|
return;
|
|
}
|
|
|
|
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"
|
|
});
|
|
if (typeof response.flushHeaders === "function") response.flushHeaders();
|
|
|
|
const writeEvent = async (name, payload, transport = null) => {
|
|
if (!isActive()) return false;
|
|
const block = `event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`;
|
|
const writable = response.write(block);
|
|
if (!writable) await waitForResponseDrain(response, isActive);
|
|
const active = isActive();
|
|
if (active && name === "hwlab.event.v1") {
|
|
emitLiveKafkaOtelSpan("hwlab.workbench.live_sse.business_event_write", payload, transport, {
|
|
env: options.env,
|
|
otelSpanEmitter: options.otelSpanEmitter ?? emitCodeAgentOtelSpan
|
|
});
|
|
}
|
|
return active;
|
|
};
|
|
let writeChain = Promise.resolve(true);
|
|
enqueueEvent = (name, payload, transport = null) => {
|
|
writeChain = writeChain.then(() => writeEvent(name, payload, transport));
|
|
return writeChain;
|
|
};
|
|
|
|
const realtimeConnection = {
|
|
close(fields = {}) {
|
|
if (!isActive()) return false;
|
|
realtimeCloseReason = textValue(fields.reason) || "server_shutdown";
|
|
realtimeCloseSignal = textValue(fields.signal) || null;
|
|
void enqueueEvent("workbench.server_draining", {
|
|
type: "server.draining",
|
|
status: "closing",
|
|
reason: realtimeCloseReason,
|
|
signal: realtimeCloseSignal,
|
|
capabilities: realtimeCapabilities,
|
|
lossPossible: true
|
|
}).finally(() => {
|
|
if (!response.writableEnded) response.end();
|
|
});
|
|
return true;
|
|
},
|
|
isActive
|
|
};
|
|
activeWorkbenchRealtimeConnections.add(realtimeConnection);
|
|
cleanup.push(() => activeWorkbenchRealtimeConnections.delete(realtimeConnection));
|
|
|
|
await enqueueEvent("workbench.connected", {
|
|
type: "connected",
|
|
status: "connected",
|
|
capabilities: realtimeCapabilities,
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "live-only",
|
|
liveOnly: true,
|
|
replay: false,
|
|
replaySupported: false,
|
|
lossPossible: true,
|
|
filters: { sessionId: authorizedSessionId, traceId: requestedTraceId }
|
|
});
|
|
liveDeliveryStarted = true;
|
|
const readyBufferedEnvelopes = bufferedEnvelopes.splice(0);
|
|
for (const { envelope, transport } of readyBufferedEnvelopes) void enqueueEvent("hwlab.event.v1", envelope, transport);
|
|
await writeChain;
|
|
if (!isActive()) {
|
|
closeConnection();
|
|
return;
|
|
}
|
|
const heartbeatTimer = setInterval(() => {
|
|
void enqueueEvent("workbench.heartbeat", {
|
|
type: "heartbeat",
|
|
status: "connected",
|
|
realtimeSource: "hwlab.event.v1",
|
|
liveOnly: true,
|
|
replay: false,
|
|
lossPossible: true,
|
|
serverSentAt: new Date().toISOString(),
|
|
valuesPrinted: false
|
|
});
|
|
}, heartbeatMs);
|
|
heartbeatTimer.unref?.();
|
|
cleanup.push(() => clearInterval(heartbeatTimer));
|
|
emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env);
|
|
}
|
|
|
|
async function handleKafkaRefreshReplayWorkbenchRealtimeHttp(request, response, options, input) {
|
|
const {
|
|
bridge,
|
|
heartbeatMs,
|
|
realtimeCapabilities,
|
|
requestedSessionId,
|
|
requestedTraceId
|
|
} = input;
|
|
const refreshReplay = bridge?.refreshReplay;
|
|
if (!refreshReplay || typeof bridge?.queryHwlabEventRetention !== "function") {
|
|
sendJson(response, 503, workbenchError("workbench_kafka_refresh_unconfigured", "Workbench Kafka refresh replay is enabled without its retention query runtime."));
|
|
return;
|
|
}
|
|
let closed = false;
|
|
let realtimeCloseReason = "client_close";
|
|
let realtimeCloseSignal = null;
|
|
let failureClosing = false;
|
|
const realtimeStartedAtMs = Date.now();
|
|
const cleanup = [];
|
|
const isActive = () => !closed && !response.destroyed && !response.writableEnded;
|
|
const closeConnection = () => {
|
|
if (closed) return;
|
|
closed = true;
|
|
emitWorkbenchRealtimeClosedOtelSpan(request, options.env, {
|
|
reason: realtimeCloseReason,
|
|
signal: realtimeCloseSignal,
|
|
startedAtMs: realtimeStartedAtMs,
|
|
sessionId: requestedSessionId,
|
|
threadId: null,
|
|
traceId: requestedTraceId,
|
|
activeConnectionCount: activeWorkbenchRealtimeConnections.size
|
|
});
|
|
for (const item of cleanup.splice(0)) item();
|
|
};
|
|
|
|
response.once("close", closeConnection);
|
|
request.once?.("aborted", closeConnection);
|
|
request.socket?.once?.("close", closeConnection);
|
|
cleanup.push(() => request.off?.("aborted", closeConnection));
|
|
cleanup.push(() => request.socket?.off?.("close", closeConnection));
|
|
|
|
try {
|
|
await (bridge.liveReady ?? bridge.ready);
|
|
} catch (error) {
|
|
for (const item of cleanup.splice(0)) item();
|
|
throw error;
|
|
}
|
|
if (!isActive()) {
|
|
closeConnection();
|
|
return;
|
|
}
|
|
|
|
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"
|
|
});
|
|
if (typeof response.flushHeaders === "function") response.flushHeaders();
|
|
|
|
const writeEvent = async (name, payload, transport = null) => {
|
|
if (!isActive()) return false;
|
|
const block = `event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`;
|
|
const writable = response.write(block);
|
|
if (!writable) await waitForResponseDrain(response, isActive);
|
|
const active = isActive();
|
|
if (active && name === "hwlab.event.v1") {
|
|
emitLiveKafkaOtelSpan("hwlab.workbench.live_sse.business_event_write", payload, transport, {
|
|
env: options.env,
|
|
otelSpanEmitter: options.otelSpanEmitter ?? emitCodeAgentOtelSpan
|
|
});
|
|
}
|
|
return active;
|
|
};
|
|
let writeChain = Promise.resolve(true);
|
|
const enqueueEvent = (name, payload, transport = null) => {
|
|
writeChain = writeChain.then(() => writeEvent(name, payload, transport));
|
|
return writeChain;
|
|
};
|
|
const closeWithRefreshFailure = async (error) => {
|
|
if (failureClosing || !isActive()) return;
|
|
failureClosing = true;
|
|
await enqueueEvent("workbench.error", workbenchKafkaRefreshErrorPayload(error, {
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId
|
|
}));
|
|
await writeChain;
|
|
if (!response.writableEnded) response.end();
|
|
};
|
|
|
|
const realtimeConnection = {
|
|
close(fields = {}) {
|
|
if (!isActive()) return false;
|
|
realtimeCloseReason = textValue(fields.reason) || "server_shutdown";
|
|
realtimeCloseSignal = textValue(fields.signal) || null;
|
|
void enqueueEvent("workbench.server_draining", {
|
|
type: "server.draining",
|
|
status: "closing",
|
|
reason: realtimeCloseReason,
|
|
signal: realtimeCloseSignal,
|
|
capabilities: realtimeCapabilities,
|
|
lossPossible: false
|
|
}).finally(() => {
|
|
if (!response.writableEnded) response.end();
|
|
});
|
|
return true;
|
|
},
|
|
isActive
|
|
};
|
|
activeWorkbenchRealtimeConnections.add(realtimeConnection);
|
|
cleanup.push(() => activeWorkbenchRealtimeConnections.delete(realtimeConnection));
|
|
|
|
const handoff = createWorkbenchKafkaRefreshHandoff({
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId,
|
|
liveBufferLimit: refreshReplay.liveBufferLimit,
|
|
identityWindowLimit: refreshReplay.matchedEventLimit + refreshReplay.liveBufferLimit,
|
|
subscribeLive: (listener) => bridge.subscribeLiveHwlabEvents(listener),
|
|
queryRetention: async ({ signal }) => {
|
|
const result = await bridge.queryHwlabEventRetention({
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId,
|
|
limit: refreshReplay.matchedEventLimit,
|
|
scanLimit: refreshReplay.scanLimit,
|
|
timeoutMs: refreshReplay.timeoutMs,
|
|
groupIdPrefix: refreshReplay.groupIdPrefix,
|
|
partitionKey: requestedSessionId,
|
|
fromBeginning: true,
|
|
signal
|
|
});
|
|
(options.logger ?? console).log(JSON.stringify({
|
|
code: "workbench-kafka-refresh-query",
|
|
component: "hwlab-workbench-realtime",
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId,
|
|
completionReason: result?.completionReason ?? null,
|
|
complete: result?.completion?.complete === true,
|
|
scannedCount: result?.scannedCount ?? null,
|
|
parsedCount: result?.parsedCount ?? null,
|
|
matchedCount: result?.matchedCount ?? null,
|
|
filterRejectedCount: result?.filterRejectedCount ?? null,
|
|
keyRejectedCount: result?.keyRejectedCount ?? null,
|
|
partitionKeyScoped: result?.partitionKeyScoped === true,
|
|
targetPartition: result?.targetPartition ?? null,
|
|
timing: result?.timing ?? null,
|
|
valuesPrinted: false
|
|
}));
|
|
return result;
|
|
},
|
|
deliverEvent: (envelope, transport) => enqueueEvent("hwlab.event.v1", envelope, transport),
|
|
deliverConnected: (summary) => enqueueEvent("workbench.connected", {
|
|
type: "connected",
|
|
status: "connected",
|
|
capabilities: realtimeCapabilities,
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
replaySupported: true,
|
|
lossPossible: false,
|
|
filters: { sessionId: requestedSessionId, traceId: requestedTraceId },
|
|
refreshReplay: summary
|
|
}),
|
|
onFailure: closeWithRefreshFailure
|
|
});
|
|
cleanup.push(() => handoff.stop("connection-closed"));
|
|
|
|
try {
|
|
await handoff.start();
|
|
} catch (error) {
|
|
if (isActive()) await closeWithRefreshFailure(error);
|
|
return;
|
|
}
|
|
if (!isActive()) return;
|
|
|
|
const heartbeatTimer = setInterval(() => {
|
|
void enqueueEvent("workbench.heartbeat", {
|
|
type: "heartbeat",
|
|
status: "connected",
|
|
realtimeSource: "hwlab.event.v1",
|
|
deliverySemantics: "kafka-retention-then-live",
|
|
liveOnly: false,
|
|
replay: true,
|
|
lossPossible: false,
|
|
serverSentAt: new Date().toISOString(),
|
|
valuesPrinted: false
|
|
});
|
|
}, heartbeatMs);
|
|
heartbeatTimer.unref?.();
|
|
cleanup.push(() => clearInterval(heartbeatTimer));
|
|
emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env);
|
|
}
|
|
|
|
async function authorizeLiveKafkaWorkbenchRealtimeScope(options, actor, { sessionId, traceId }) {
|
|
const access = options.accessController;
|
|
if ((sessionId && typeof access?.getAgentSession !== "function") || (traceId && typeof access?.getAgentSessionByTraceId !== "function")) {
|
|
return {
|
|
ok: false,
|
|
status: 503,
|
|
body: workbenchError("workbench_realtime_authorization_unconfigured", "Workbench live realtime ownership authorization is not configured.")
|
|
};
|
|
}
|
|
|
|
const [sessionById, sessionByTrace] = await Promise.all([
|
|
sessionId ? access.getAgentSession(sessionId) : null,
|
|
traceId ? access.getAgentSessionByTraceId(traceId) : null
|
|
]);
|
|
const sessionByIdId = sessionById ? safeSessionId(sessionById.id ?? sessionById.sessionId) : null;
|
|
const sessionByTraceId = sessionByTrace ? safeSessionId(sessionByTrace.id ?? sessionByTrace.sessionId) : null;
|
|
const scopeMissing = (sessionId && sessionByIdId !== sessionId)
|
|
|| (traceId && !sessionByTraceId)
|
|
|| (sessionId && traceId && sessionByIdId !== sessionByTraceId);
|
|
const resolvedSessions = [sessionById, sessionByTrace].filter(Boolean);
|
|
const ownedByActor = actor?.role === "admin" || (Boolean(actor?.id)
|
|
&& resolvedSessions.length > 0
|
|
&& resolvedSessions.every((session) => textValue(session.ownerUserId) === actor.id));
|
|
if (scopeMissing || !ownedByActor) {
|
|
return {
|
|
ok: false,
|
|
status: 404,
|
|
body: workbenchError("workbench_realtime_scope_not_found", "Workbench realtime scope is not visible to the current actor.")
|
|
};
|
|
}
|
|
return { ok: true, sessionId: sessionByIdId ?? sessionByTraceId };
|
|
}
|
|
|
|
function liveKafkaEnvelopeMatches(envelope, sessionId, traceId) {
|
|
if (!envelope || envelope.schema !== "hwlab.event.v1") return false;
|
|
if (sessionId && safeSessionId(envelope.hwlabSessionId ?? envelope.sessionId) !== sessionId) return false;
|
|
if (traceId && safeTraceId(envelope.traceId) !== traceId) return false;
|
|
return true;
|
|
}
|
|
|
|
export function attachWorkbenchRealtimeOtelContext(request, fields = {}) {
|
|
const context = request?.hwlabHttpRequestContext;
|
|
if (!context) return;
|
|
context.route = textValue(fields.route) || "/v1/workbench/events";
|
|
context.otelAttributes = {
|
|
...(context.otelAttributes ?? {}),
|
|
"workbench.route": "events",
|
|
"workbench.session_id": fields.sessionId ?? null,
|
|
traceId: fields.traceId ?? null,
|
|
"workbench.trace_id": fields.traceId ?? null,
|
|
"workbench.thread_id": fields.threadId ?? null,
|
|
"workbench.sse.heartbeat_ms": fields.heartbeatMs ?? null,
|
|
"workbench.sse.realtime_source": fields.realtimeSource ?? null,
|
|
"workbench.sse.live_kafka_enabled": fields.realtimeCapabilities?.liveKafkaSse ?? null,
|
|
"workbench.sse.projection_realtime_enabled": fields.realtimeCapabilities?.projectionRealtime ?? null,
|
|
"workbench.sse.outbox_mode": fields.realtimeCapabilities?.liveKafkaSse === true ? false : true
|
|
};
|
|
}
|
|
|
|
export { projectionOutboxRealtimeEvents } from "./workbench-projection-outbox-events.ts";
|
|
|
|
export function compactObject(value) {
|
|
return Object.fromEntries(Object.entries(value || {}).filter(([, entry]) => textValue(entry)));
|
|
}
|
|
|
|
export function integerValue(value) {
|
|
const number = Number(value);
|
|
return Number.isFinite(number) ? Math.trunc(number) : null;
|
|
}
|
|
|
|
export function attachWorkbenchReadModelDiagnosticOtel(request, fields = {}) {
|
|
const context = request?.hwlabHttpRequestContext;
|
|
if (!context) return;
|
|
const code = textValue(fields.code) || "workbench_read_model_not_found";
|
|
const route = textValue(fields.route) || context.route || "/v1/workbench";
|
|
context.route = route;
|
|
context.lastHttpError = {
|
|
code,
|
|
category: "workbench_read_model",
|
|
layer: "workbench.read_model"
|
|
};
|
|
context.otelAttributes = {
|
|
...(context.otelAttributes ?? {}),
|
|
"workbench.route": route,
|
|
"workbench.read_model.route": route,
|
|
"workbench.read_model.result": code,
|
|
"workbench.read_model.count": Number.isFinite(Number(fields.count)) ? Math.trunc(Number(fields.count)) : null,
|
|
"workbench.session_id": fields.sessionId ?? null,
|
|
"workbench.trace_id": fields.traceId ?? null,
|
|
"workbench.turn_id": fields.turnId ?? null,
|
|
traceId: fields.traceId ?? null,
|
|
"error.code": code,
|
|
"error.category": "workbench_read_model",
|
|
"error.layer": "workbench.read_model"
|
|
};
|
|
}
|
|
|
|
export function emitWorkbenchRealtimeAcceptedOtelSpan(request, env = process.env) {
|
|
const context = request?.hwlabHttpRequestContext;
|
|
if (!context?.traceId) return;
|
|
const endedAtMs = Date.now();
|
|
void emitHttpServerRequestSpan(context, env, {
|
|
startTimeMs: context.startedAtMs,
|
|
endTimeMs: endedAtMs,
|
|
statusCode: 200,
|
|
method: "GET",
|
|
route: context.route ?? "/v1/workbench/events",
|
|
attributes: {
|
|
...(context.otelAttributes ?? {}),
|
|
"http.closed_by": "sse-accepted",
|
|
"workbench.sse.lifecycle": "accepted",
|
|
"workbench.sse.accept_latency_ms": Math.max(0, endedAtMs - Number(context.startedAtMs ?? endedAtMs))
|
|
}
|
|
}).catch(() => undefined);
|
|
}
|
|
|
|
export function emitWorkbenchRealtimeClosedOtelSpan(request, env = process.env, fields = {}) {
|
|
const context = request?.hwlabHttpRequestContext;
|
|
if (!context?.traceId) return;
|
|
const endedAtMs = Date.now();
|
|
const startedAtMs = Number(fields.startedAtMs ?? context.startedAtMs ?? endedAtMs);
|
|
const reason = textValue(fields.reason) || "client_close";
|
|
void emitHttpServerRequestSpan(context, env, {
|
|
startTimeMs: Number.isFinite(startedAtMs) ? startedAtMs : endedAtMs,
|
|
endTimeMs: endedAtMs,
|
|
statusCode: 200,
|
|
method: "GET",
|
|
route: context.route ?? "/v1/workbench/events",
|
|
attributes: {
|
|
...(context.otelAttributes ?? {}),
|
|
"http.closed_by": reason,
|
|
"workbench.sse.lifecycle": "closed",
|
|
"workbench.sse.close_reason": reason,
|
|
"workbench.sse.close_signal": fields.signal ?? null,
|
|
"workbench.sse.duration_ms": Math.max(0, endedAtMs - (Number.isFinite(startedAtMs) ? startedAtMs : endedAtMs)),
|
|
"workbench.sse.session_id": fields.sessionId ?? null,
|
|
"workbench.sse.thread_id": fields.threadId ?? null,
|
|
"workbench.sse.trace_id": fields.traceId ?? null,
|
|
"workbench.sse.active_connection_count": Number.isFinite(Number(fields.activeConnectionCount)) ? Number(fields.activeConnectionCount) : null
|
|
}
|
|
}).catch(() => undefined);
|
|
}
|
|
|
|
export function handleWorkbenchRealtimeFailure(request, response, url, options = {}, error) {
|
|
const classified = classifyWorkbenchReadModelFailure(error);
|
|
const context = request?.hwlabHttpRequestContext;
|
|
if (context) {
|
|
context.route = "/v1/workbench/events";
|
|
context.otelAttributes = {
|
|
...(context.otelAttributes ?? {}),
|
|
"workbench.route": "events",
|
|
"workbench.sse.lifecycle": "failed",
|
|
"error.code": classified.code,
|
|
"error.layer": "workbench.realtime"
|
|
};
|
|
}
|
|
logWorkbenchReadModelFailure(options.logger ?? console, {
|
|
event: "workbench_realtime_route_failed",
|
|
ok: false,
|
|
route: "/v1/workbench/events",
|
|
statusCode: classified.statusCode,
|
|
errorCode: classified.code,
|
|
errorName: error?.name ?? "Error",
|
|
message: error instanceof Error ? error.message : String(error ?? "unknown"),
|
|
valuesRedacted: true
|
|
});
|
|
if (response.headersSent && !response.destroyed) {
|
|
try {
|
|
response.write(`event: workbench.error\n`);
|
|
response.write(`data: ${JSON.stringify({ contractVersion: "workbench-events-v1", serverSentAt: new Date().toISOString(), type: "error", error: { code: classified.code, message: classified.message, valuesRedacted: true } })}\n\n`);
|
|
response.end();
|
|
} catch {
|
|
try { response.destroy(); } catch {}
|
|
}
|
|
return;
|
|
}
|
|
if (!response.destroyed) sendJson(response, classified.statusCode, workbenchError(classified.code, classified.message, { route: "/v1/workbench/events", errorName: error?.name ?? "Error" }));
|
|
}
|
|
|
|
export 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." }
|
|
});
|
|
}
|
|
}
|
|
|
|
export function realtimeEventCreatedAt(payload) {
|
|
const candidates = [payload?.event?.createdAt, payload?.snapshot?.lastEvent?.createdAt, payload?.turn?.updatedAt];
|
|
for (const value of candidates) {
|
|
const text = textValue(value);
|
|
if (text) return text;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
export function realtimeTraceSeq(payload) {
|
|
const seq = Number(payload?.cursor?.traceSeq ?? payload?.event?.seq ?? payload?.snapshot?.lastEvent?.seq);
|
|
return Number.isFinite(seq) && seq >= 0 ? seq : null;
|
|
}
|
|
|
|
export function realtimeEventId(payload) {
|
|
const raw = payload?.cursor?.outboxSeq ?? payload?.outboxSeq ?? payload?.id;
|
|
const text = textValue(raw);
|
|
if (!text || /[\r\n]/u.test(text)) return null;
|
|
return text.slice(0, 128);
|
|
}
|
|
|
|
export function workbenchRealtimeAfterSeq(request, url) {
|
|
for (const value of [request?.headers?.["last-event-id"], url.searchParams.get("afterSeq"), url.searchParams.get("afterOutboxSeq")]) {
|
|
const parsed = nonNegativeInteger(value, 0);
|
|
if (parsed > 0) return parsed;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
function requiredPositiveRuntimeSetting(env, name) {
|
|
const value = Number(env?.[name]);
|
|
if (!Number.isInteger(value) || value <= 0) {
|
|
const error = new Error(`${name} must be explicitly configured as a positive integer.`);
|
|
error.code = "workbench_outbox_tail_config_invalid";
|
|
throw error;
|
|
}
|
|
return value;
|
|
}
|
|
|
|
export function workbenchProjectionSignalMatches(signal = {}, sessionId = null, traceId = null) {
|
|
if (signal.recovery === true) return true;
|
|
const signalSessionId = textValue(signal.sessionId);
|
|
const signalTraceId = textValue(signal.traceId);
|
|
if (traceId && signalTraceId !== traceId) return false;
|
|
if (sessionId && signalSessionId !== sessionId) return false;
|
|
return Boolean(signalSessionId || signalTraceId);
|
|
}
|
|
|
|
export function waitForResponseDrain(response, isActive = () => true) {
|
|
if (!isActive()) return Promise.resolve(false);
|
|
return new Promise((resolve) => {
|
|
const settle = (result) => {
|
|
response.off?.("drain", onDrain);
|
|
response.off?.("close", onClose);
|
|
response.off?.("error", onClose);
|
|
resolve(result);
|
|
};
|
|
const onDrain = () => settle(isActive());
|
|
const onClose = () => settle(false);
|
|
response.once("drain", onDrain);
|
|
response.once("close", onClose);
|
|
response.once("error", onClose);
|
|
});
|
|
}
|
|
|
|
export function nowMs() {
|
|
if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now();
|
|
return Date.now();
|
|
}
|
|
|
|
export function sleepMs(ms) {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
|
|
export async function realtimeInitialSessionContext(readModel, actor, sessionId) {
|
|
const safeId = safeSessionId(sessionId);
|
|
if (!safeId || typeof readModel?.queryFacts !== "function") {
|
|
const blocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench realtime initial session must be read from durable facts; the facts query interface is unavailable.", { session: { sessionId: safeId }, route: "/v1/workbench/events" });
|
|
return { facts: emptyFactSet(), session: null, blocker };
|
|
}
|
|
try {
|
|
const result = await readModel.queryFacts({ sessionId: safeId, families: WORKBENCH_SESSION_DETAIL_FAMILIES, limit: MAX_PAGE_LIMIT });
|
|
if (result.error || result.durable === false) {
|
|
const blocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench realtime initial session facts query failed; realtime must surface the read-model error instead of falling back to legacy session snapshots.", { session: { sessionId: safeId }, route: "/v1/workbench/events" });
|
|
blocker.causeCode = result.error?.code ?? null;
|
|
blocker.causeName = result.error?.name ?? null;
|
|
blocker.message = result.error?.message ? `${blocker.message} Cause: ${result.error.message}` : result.durable === false ? `${blocker.message} Cause: durable facts reader unavailable.` : blocker.message;
|
|
return { facts: emptyFactSet(), session: null, blocker };
|
|
}
|
|
const facts = result.facts ?? emptyFactSet();
|
|
const session = visibleFactSessions(facts, actor).find((item) => factSessionId(item) === safeId) ?? null;
|
|
return { facts, session, blocker: null };
|
|
} catch (error) {
|
|
const blocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench realtime initial session facts query failed; realtime must surface the read-model error instead of falling back to legacy session snapshots.", { session: { sessionId: safeId }, route: "/v1/workbench/events" });
|
|
blocker.causeCode = error?.code ?? null;
|
|
blocker.causeName = error?.name ?? null;
|
|
blocker.message = error?.message ? `${blocker.message} Cause: ${error.message}` : blocker.message;
|
|
return { facts: emptyFactSet(), session: null, blocker };
|
|
}
|
|
}
|
|
|
|
export 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;
|
|
}
|
|
if (context.factsReadBlocker) {
|
|
writeEvent("workbench.error", {
|
|
type: "error",
|
|
traceId,
|
|
reason,
|
|
error: context.factsReadBlocker
|
|
});
|
|
}
|
|
const sessionId = factSessionId(context.factSession) ?? safeSessionId(context.session?.id) ?? null;
|
|
const threadId = safeOpaqueId(context.factSession?.threadId ?? context.session?.threadId) ?? (textValue(context.factSession?.threadId ?? context.session?.threadId) || null);
|
|
const realtimeTraceSnapshot = context.factsReadBlocker ? blockedRealtimeTraceSnapshot(context) : traceSnapshotForRealtime(context.trace);
|
|
const traceSeq = context.factsReadBlocker ? 0 : traceSnapshotLastSeq(context.trace);
|
|
writeEvent("workbench.trace.snapshot", {
|
|
type: "trace.snapshot",
|
|
reason,
|
|
sessionId,
|
|
threadId,
|
|
traceId,
|
|
snapshot: { ...realtimeTraceSnapshot, sessionId, threadId },
|
|
cursor: { traceSeq }
|
|
});
|
|
writeEvent("workbench.turn.snapshot", {
|
|
type: "turn.snapshot",
|
|
reason,
|
|
sessionId,
|
|
threadId,
|
|
traceId,
|
|
turn: realtimeTurnSnapshot(context),
|
|
cursor: { traceSeq }
|
|
});
|
|
}
|
|
|
|
export function realtimeTurnSnapshot(context) {
|
|
if (context?.factsReadBlocker) return blockedRealtimeTurnSnapshot(context, context.factsReadBlocker);
|
|
const facts = context?.facts;
|
|
const traceId = safeTraceId(context?.traceId);
|
|
if (traceId && facts && context?.factSession) {
|
|
const factTurn = factTurnForTrace(facts, traceId);
|
|
const projected = factTurnSnapshot({ turn: factTurn, session: context.factSession, facts, traceId, turnId: context.turnId });
|
|
if (projected) return projected;
|
|
}
|
|
const blocker = traceEventsReadModelBlocker("workbench_facts_turn_missing", "Workbench durable facts did not contain a turn snapshot; realtime must surface the read-model gap instead of falling back to legacy trace projection.", { traceId, session: context?.factSession, route: "/v1/workbench/events" });
|
|
return blockedRealtimeTurnSnapshot(context, blocker);
|
|
}
|
|
|
|
export function blockedRealtimeTraceSnapshot(context) {
|
|
const traceId = safeTraceId(context?.traceId ?? context?.factsReadBlocker?.traceId) ?? null;
|
|
const blocker = context?.factsReadBlocker ?? traceEventsReadModelBlocker("workbench_facts_unavailable", "Workbench durable facts are unavailable for realtime trace snapshot.", { traceId, route: "/v1/workbench/events" });
|
|
const projection = blockedTraceEventProjection({}, blocker);
|
|
return {
|
|
traceId,
|
|
status: "unknown",
|
|
traceStatus: "unknown",
|
|
events: [],
|
|
eventCount: 0,
|
|
fullTraceLoaded: true,
|
|
hasMore: false,
|
|
nextProjectedSeq: 0,
|
|
updatedAt: null,
|
|
projection,
|
|
projectionStatus: projection.projectionStatus,
|
|
projectionHealth: projection.projectionHealth,
|
|
staleMs: projection.staleMs ?? null,
|
|
blocker,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
}
|
|
|
|
export function blockedRealtimeTurnSnapshot(context, blocker) {
|
|
const traceId = safeTraceId(context?.traceId ?? blocker?.traceId) ?? null;
|
|
const turnId = safeTurnId(context?.turnId ?? blocker?.turnId) || traceId;
|
|
const sessionId = safeSessionId(blocker?.sessionId ?? factSessionId(context?.factSession) ?? context?.session?.id) ?? null;
|
|
const threadId = safeOpaqueId(context?.factSession?.threadId ?? context?.session?.threadId) ?? (textValue(context?.session?.threadId) || null);
|
|
const projection = blockedTraceEventProjection({}, blocker);
|
|
return {
|
|
turnId,
|
|
traceId,
|
|
status: "unknown",
|
|
running: false,
|
|
terminal: false,
|
|
sessionId,
|
|
threadId,
|
|
userMessageId: null,
|
|
assistantMessageId: null,
|
|
assistantText: null,
|
|
finalResponse: null,
|
|
timing: null,
|
|
startedAt: null,
|
|
lastEventAt: null,
|
|
finishedAt: null,
|
|
durationMs: null,
|
|
agentRun: null,
|
|
trace: {
|
|
traceId,
|
|
status: "unknown",
|
|
eventCount: 0,
|
|
updatedAt: null
|
|
},
|
|
urls: {
|
|
self: turnId ? `/v1/workbench/turns/${encodeURIComponent(turnId)}` : null,
|
|
traceEvents: traceId ? `/v1/workbench/traces/${encodeURIComponent(traceId)}/events` : null
|
|
},
|
|
projectionStatus: projection.projectionStatus,
|
|
projectionHealth: projection.projectionHealth,
|
|
staleMs: projection.staleMs ?? null,
|
|
blocker,
|
|
projection,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
}
|
|
|
|
export async function writeTraceRealtimeSnapshotSafe(input) {
|
|
try {
|
|
await writeTraceRealtimeSnapshot(input);
|
|
} catch (error) {
|
|
input.writeEvent("workbench.error", {
|
|
type: "error",
|
|
traceId: input.traceId,
|
|
reason: input.reason,
|
|
error: {
|
|
code: "workbench_trace_snapshot_failed",
|
|
causeCode: error?.code ?? null,
|
|
retryable: error?.data?.retryable === true,
|
|
message: error?.message ?? "Workbench trace snapshot failed.",
|
|
valuesRedacted: true
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
export async function visibleTraceContext(options, actor, traceId) {
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
if (typeof readModel.queryFacts !== "function") {
|
|
const projection = factProjectionForTrace(emptyFactSet(), traceId);
|
|
const factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench durable facts query is unavailable; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
|
return { visible: true, turnId: traceId, traceId, status: "unknown", projection, result: null, session: null, trace: factTraceSnapshot(emptyFactSet(), traceId), facts: emptyFactSet(), factSession: null, factsReadBlocker };
|
|
}
|
|
|
|
let facts = emptyFactSet();
|
|
let factsReadBlocker = null;
|
|
try {
|
|
const factResult = await readModel.queryFacts({ traceId, families: [...WORKBENCH_SESSION_DETAIL_FAMILIES, ...WORKBENCH_TRACE_EVENT_PAGE_FAMILIES], limit: MAX_PAGE_LIMIT });
|
|
if (factResult.error || factResult.durable === false) {
|
|
const projection = factProjectionForTrace(facts, traceId);
|
|
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts query failed; realtime must surface the read-model error instead of silently falling back to legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
|
factsReadBlocker.causeCode = factResult.error?.code ?? null;
|
|
factsReadBlocker.causeName = factResult.error?.name ?? null;
|
|
factsReadBlocker.message = factResult.error?.message ? `${factsReadBlocker.message} Cause: ${factResult.error.message}` : factResult.durable === false ? `${factsReadBlocker.message} Cause: durable facts reader unavailable.` : factsReadBlocker.message;
|
|
} else {
|
|
facts = factResult?.facts ?? facts;
|
|
}
|
|
} catch (error) {
|
|
const projection = factProjectionForTrace(facts, traceId);
|
|
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts query failed; realtime must surface the read-model error instead of silently falling back to legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
|
factsReadBlocker.causeCode = error?.code ?? null;
|
|
factsReadBlocker.causeName = error?.name ?? null;
|
|
factsReadBlocker.message = error?.message ? `${factsReadBlocker.message} Cause: ${error.message}` : factsReadBlocker.message;
|
|
}
|
|
|
|
const resolved = factsReadBlocker ? { session: null, facts } : await visibleFactSessionForTrace(readModel, facts, actor, traceId);
|
|
if (resolved.error) {
|
|
const projection = factProjectionForTrace(facts, traceId);
|
|
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts session lookup failed; realtime must surface the read-model error instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
|
factsReadBlocker.causeCode = resolved.error?.code ?? null;
|
|
factsReadBlocker.causeName = resolved.error?.name ?? null;
|
|
factsReadBlocker.message = resolved.error?.message ? `${factsReadBlocker.message} Cause: ${resolved.error.message}` : factsReadBlocker.message;
|
|
} else if (resolved.facts) {
|
|
facts = resolved.facts;
|
|
}
|
|
|
|
const factSession = resolved.session ?? null;
|
|
const trace = factTraceSnapshot(facts, traceId);
|
|
const turn = factTurnForTrace(facts, traceId, traceId);
|
|
let projection = factsReadBlocker ? blockedTraceEventProjection(factProjectionForTrace(facts, traceId), factsReadBlocker) : factProjectionForTrace(facts, traceId);
|
|
const visible = Boolean(factsReadBlocker || factSession || turn || factCheckpointForTrace(facts, traceId) || trace.eventCount > 0);
|
|
if (!visible) {
|
|
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_trace_missing", "Workbench realtime could not find durable facts for the requested trace; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" });
|
|
projection = blockedTraceEventProjection(projection, factsReadBlocker);
|
|
}
|
|
if (!factsReadBlocker && !factSession) {
|
|
factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_missing", "Workbench durable facts did not contain the trace session; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, turn, route: "/v1/workbench/events" });
|
|
projection = blockedTraceEventProjection(projection, factsReadBlocker);
|
|
}
|
|
const snapshot = factTurnSnapshot({ turn, session: factSession, facts, traceId, turnId: traceId });
|
|
return { visible: true, turnId: snapshot.turnId ?? traceId, traceId, status: snapshot.status, projection, result: null, session: null, trace, facts, factSession, factsReadBlocker };
|
|
}
|
|
|
|
export 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
|
|
};
|
|
}
|
|
|
|
export 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
|
|
};
|
|
}
|
|
|
|
export function traceSnapshotLastSeq(snapshot) {
|
|
const events = Array.isArray(snapshot?.events) ? snapshot.events : [];
|
|
return events.reduce((max, event, index) => Math.max(max, eventSeq(event, index)), 0);
|
|
}
|