lyon
2561 lines
114 KiB
TypeScript
2561 lines
114 KiB
TypeScript
/*
|
|
* SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010403 API契约 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-010401 Web工作台 draft-2026-06-20-p2-terminal-outbox-recovery; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
|
|
* 职责: Workbench REST read model。只读投影 session/message/turn/trace facts，不执行 AgentRun sync、billing finalize 或 workspace repair。
|
|
*/
|
|
import { createHash } from "node:crypto";
|
|
|
|
import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts";
|
|
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 { createWorkbenchTurnProjection, createWorkbenchTurnTimingProjection, durableTraceStatus, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts";
|
|
import { emitHttpServerRequestSpan } from "./otel-trace.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", "turns", "checkpoints"]);
|
|
const WORKBENCH_TRACE_EVENT_PAGE_FAMILIES = Object.freeze(["traceEvents"]);
|
|
const activeWorkbenchRealtimeConnections = new Set();
|
|
|
|
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 handleWorkbenchReadModelHttp(request, response, url, options = {}) {
|
|
try {
|
|
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.pathname === "/v1/workbench/sessions") {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
await (perf ? perf.measure("workbench_session_list", () => handleWorkbenchSessionList(request, response, url, options, auth.actor)) : handleWorkbenchSessionList(request, response, url, options, auth.actor));
|
|
return;
|
|
}
|
|
|
|
const sessionMessagesMatch = url.pathname.match(/^\/v1\/workbench\/sessions\/([^/]+)\/messages$/u);
|
|
if (sessionMessagesMatch) {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
await (perf ? perf.measure("workbench_message_page", () => handleWorkbenchMessagePage(request, response, url, options, auth.actor, decodeURIComponent(sessionMessagesMatch[1]))) : handleWorkbenchMessagePage(request, response, url, options, auth.actor, decodeURIComponent(sessionMessagesMatch[1])));
|
|
return;
|
|
}
|
|
|
|
const sessionMatch = url.pathname.match(/^\/v1\/workbench\/sessions\/([^/]+)$/u);
|
|
if (sessionMatch) {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
await (perf ? perf.measure("workbench_session_detail", () => handleWorkbenchSessionDetail(request, response, options, auth.actor, decodeURIComponent(sessionMatch[1]))) : handleWorkbenchSessionDetail(request, response, options, auth.actor, decodeURIComponent(sessionMatch[1])));
|
|
return;
|
|
}
|
|
|
|
const turnMatch = url.pathname.match(/^\/v1\/workbench\/turns\/([^/]+)$/u);
|
|
if (turnMatch) {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
await (perf ? perf.measure("workbench_turn_snapshot", () => handleWorkbenchTurnSnapshot(request, response, url, options, auth.actor, decodeURIComponent(turnMatch[1]))) : handleWorkbenchTurnSnapshot(request, response, url, options, auth.actor, decodeURIComponent(turnMatch[1])));
|
|
return;
|
|
}
|
|
|
|
const traceEventsMatch = url.pathname.match(/^\/v1\/workbench\/traces\/([^/]+)\/events$/u);
|
|
if (traceEventsMatch) {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
await (perf ? perf.measure("workbench_trace_events", () => handleWorkbenchTraceEventPage(request, response, url, options, auth.actor, decodeURIComponent(traceEventsMatch[1]))) : handleWorkbenchTraceEventPage(request, response, url, options, auth.actor, decodeURIComponent(traceEventsMatch[1])));
|
|
return;
|
|
}
|
|
|
|
sendJson(response, 404, workbenchError("workbench_route_not_found", "Workbench read model route is not implemented.", { route: url.pathname }));
|
|
} catch (error) {
|
|
handleWorkbenchReadModelFailure(response, url, options, error);
|
|
}
|
|
}
|
|
|
|
export async function handleWorkbenchRealtimeHttp(request, response, url, options = {}) {
|
|
try {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
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);
|
|
const realtimeOutboxReader = workbenchRuntimeOutboxReader(request, options);
|
|
attachWorkbenchRealtimeOtelContext(request, {
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId,
|
|
heartbeatMs,
|
|
outboxMode: typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
|
|
});
|
|
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;
|
|
}
|
|
|
|
const traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
|
|
const readModel = createWorkbenchReadModel(options, auth.actor);
|
|
const requestedAfterSeq = workbenchRealtimeAfterSeq(request, url);
|
|
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 = [];
|
|
|
|
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 = (name, payload = {}) => {
|
|
if (closed || response.destroyed) return;
|
|
const startedAt = nowMs();
|
|
const eventCreatedAt = realtimeEventCreatedAt(payload);
|
|
const traceSeq = realtimeTraceSeq(payload);
|
|
const eventId = realtimeEventId(payload);
|
|
response.write(`event: ${name}\n`);
|
|
if (eventId) response.write(`id: ${eventId}\n`);
|
|
response.write(`data: ${JSON.stringify({ contractVersion: "workbench-events-v1", serverSentAt: new Date().toISOString(), eventCreatedAt, traceSeq, ...payload })}\n\n`);
|
|
perf?.recordPhase({ phase: "sse_write", durationMs: nowMs() - startedAt, outcome: "ok" });
|
|
};
|
|
|
|
const realtimeConnection = {
|
|
close(fields = {}) {
|
|
if (closed || response.destroyed || response.writableEnded) return false;
|
|
const reason = textValue(fields.reason) || "server_shutdown";
|
|
realtimeCloseReason = reason;
|
|
realtimeCloseSignal = textValue(fields.signal) || null;
|
|
try {
|
|
writeEvent("workbench.server_draining", {
|
|
type: "server.draining",
|
|
status: "closing",
|
|
reason,
|
|
signal: realtimeCloseSignal,
|
|
sessionId: realtimeSessionId,
|
|
threadId: realtimeThreadId,
|
|
traceId: realtimeTraceId,
|
|
observedAt: new Date().toISOString()
|
|
});
|
|
response.end();
|
|
return true;
|
|
} catch {
|
|
try { response.destroy(); } catch {}
|
|
return false;
|
|
}
|
|
},
|
|
isActive() {
|
|
return !closed && !response.destroyed && !response.writableEnded;
|
|
}
|
|
};
|
|
activeWorkbenchRealtimeConnections.add(realtimeConnection);
|
|
cleanup.push(() => activeWorkbenchRealtimeConnections.delete(realtimeConnection));
|
|
|
|
writeEvent("workbench.connected", {
|
|
type: "connected",
|
|
status: "connected",
|
|
heartbeatMs,
|
|
cursor: { outboxSeq: requestedAfterSeq },
|
|
filters: {
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId
|
|
}
|
|
});
|
|
|
|
const initialSession = requestedSessionId ? await readModel.getSessionById(requestedSessionId) : null;
|
|
const initialSessionSnapshot = objectValue(initialSession?.session);
|
|
const streamSessionId = safeSessionId(initialSession?.id ?? requestedSessionId) ?? null;
|
|
const streamThreadId = safeOpaqueId(initialSession?.threadId ?? initialSessionSnapshot.threadId) ?? (textValue(initialSession?.threadId ?? initialSessionSnapshot.threadId) || null);
|
|
const activeTraceId = requestedTraceId
|
|
?? safeTraceId(initialSession?.lastTraceId ?? initialSessionSnapshot.lastTraceId ?? initialSessionSnapshot.currentTraceId)
|
|
?? null;
|
|
realtimeSessionId = streamSessionId ?? requestedSessionId ?? null;
|
|
realtimeTraceId = activeTraceId ?? requestedTraceId ?? null;
|
|
realtimeThreadId = streamThreadId ?? null;
|
|
attachWorkbenchRealtimeOtelContext(request, {
|
|
sessionId: streamSessionId ?? requestedSessionId,
|
|
traceId: activeTraceId,
|
|
threadId: streamThreadId,
|
|
heartbeatMs,
|
|
outboxMode: typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function"
|
|
});
|
|
emitWorkbenchRealtimeAcceptedOtelSpan(request, options.env);
|
|
if (activeTraceId && requestedAfterSeq <= 0) {
|
|
await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" }));
|
|
}
|
|
const outboxPollMs = parsePositiveInteger(options.env?.HWLAB_WORKBENCH_SSE_OUTBOX_POLL_MS, 500);
|
|
let outboxCursor = requestedAfterSeq;
|
|
const outboxTraceSnapshots = new Set(activeTraceId ? [activeTraceId] : []);
|
|
if (typeof realtimeOutboxReader?.readWorkbenchProjectionOutbox === "function" && (streamSessionId || activeTraceId)) {
|
|
const pollOutbox = async () => {
|
|
if (closed || response.destroyed) return;
|
|
try {
|
|
const query = { afterSeq: outboxCursor, limit: 50 };
|
|
if (streamSessionId) query.sessionId = streamSessionId;
|
|
else query.traceId = activeTraceId;
|
|
const rows = await realtimeOutboxReader.readWorkbenchProjectionOutbox(query);
|
|
for (const row of rows) {
|
|
if (closed || response.destroyed) break;
|
|
outboxCursor = row.outboxSeq;
|
|
const rowTraceId = safeTraceId(row.traceId) ?? activeTraceId;
|
|
const rowSessionId = safeSessionId(row.sessionId ?? streamSessionId) ?? streamSessionId;
|
|
if (rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) {
|
|
outboxTraceSnapshots.add(rowTraceId);
|
|
await writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "outbox-session" });
|
|
}
|
|
if (row.commitType === "message") {
|
|
await writeMessageRealtimeSnapshotSafe({
|
|
writeEvent,
|
|
readModel,
|
|
actor: auth.actor,
|
|
sessionId: rowSessionId,
|
|
threadId: streamThreadId,
|
|
traceId: rowTraceId,
|
|
reason: "outbox-message",
|
|
cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq }
|
|
});
|
|
continue;
|
|
}
|
|
if (row.commitType === "terminal") {
|
|
writeEvent("workbench.trace.snapshot", {
|
|
type: "trace.snapshot",
|
|
sessionId: rowSessionId,
|
|
threadId: streamThreadId,
|
|
traceId: rowTraceId,
|
|
reason: "outbox-terminal",
|
|
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", {
|
|
type: "trace.event",
|
|
sessionId: rowSessionId,
|
|
threadId: streamThreadId,
|
|
traceId: rowTraceId,
|
|
cursor: { traceSeq: row.projectedSeq, outboxSeq: row.outboxSeq }
|
|
});
|
|
}
|
|
}
|
|
} catch (outboxError) {
|
|
writeEvent("workbench.error", {
|
|
type: "error",
|
|
error: { code: "workbench_outbox_poll_failed", message: outboxError?.message ?? "Workbench outbox poll failed." }
|
|
});
|
|
}
|
|
};
|
|
await pollOutbox();
|
|
const outboxTimer = setInterval(() => { void pollOutbox(); }, Math.max(500, outboxPollMs));
|
|
cleanup.push(() => clearInterval(outboxTimer));
|
|
} else if (activeTraceId) {
|
|
const unsubscribe = traceStore.subscribe(activeTraceId, (event, snapshot) => {
|
|
if (event) {
|
|
writeEvent("workbench.trace.event", {
|
|
type: "trace.event",
|
|
sessionId: streamSessionId,
|
|
threadId: streamThreadId,
|
|
traceId: activeTraceId,
|
|
event,
|
|
snapshot: { ...traceSnapshotSummary(snapshot), sessionId: streamSessionId, threadId: streamThreadId },
|
|
cursor: { traceSeq: eventSeq(event, Number(snapshot?.eventCount ?? 1) - 1) }
|
|
});
|
|
} else {
|
|
writeEvent("workbench.trace.snapshot", {
|
|
type: "trace.snapshot",
|
|
sessionId: streamSessionId,
|
|
threadId: streamThreadId,
|
|
traceId: activeTraceId,
|
|
reason: "trace-store-update",
|
|
snapshot: { ...traceSnapshotForRealtime(snapshot), sessionId: streamSessionId, threadId: streamThreadId },
|
|
cursor: { traceSeq: traceSnapshotLastSeq(snapshot) }
|
|
});
|
|
}
|
|
if (event?.terminal === true || TERMINAL_STATUSES.has(normalizeStatus(event?.status))) {
|
|
void writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "terminal" });
|
|
}
|
|
});
|
|
cleanup.push(unsubscribe);
|
|
}
|
|
|
|
const heartbeat = setInterval(async () => {
|
|
try {
|
|
writeEvent("workbench.heartbeat", {
|
|
type: "heartbeat",
|
|
sessionId: requestedSessionId ?? null,
|
|
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", () => {
|
|
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();
|
|
});
|
|
} catch (error) {
|
|
handleWorkbenchRealtimeFailure(request, response, url, options, error);
|
|
}
|
|
}
|
|
|
|
function attachWorkbenchRealtimeOtelContext(request, fields = {}) {
|
|
const context = request?.hwlabHttpRequestContext;
|
|
if (!context) return;
|
|
context.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.outbox_mode": fields.outboxMode === true
|
|
};
|
|
}
|
|
|
|
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"
|
|
};
|
|
}
|
|
|
|
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: "/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);
|
|
}
|
|
|
|
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: "/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);
|
|
}
|
|
|
|
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" }));
|
|
}
|
|
|
|
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) {
|
|
const text = textValue(value);
|
|
if (text) return text;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function realtimeTraceSeq(payload) {
|
|
const seq = Number(payload?.cursor?.traceSeq ?? payload?.event?.seq ?? payload?.snapshot?.lastEvent?.seq);
|
|
return Number.isFinite(seq) && seq >= 0 ? seq : null;
|
|
}
|
|
|
|
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);
|
|
}
|
|
|
|
function workbenchRealtimeAfterSeq(request, url) {
|
|
for (const value of [url.searchParams.get("afterSeq"), url.searchParams.get("afterOutboxSeq"), request?.headers?.["last-event-id"]]) {
|
|
const parsed = nonNegativeInteger(value, 0);
|
|
if (parsed > 0) return parsed;
|
|
}
|
|
return 0;
|
|
}
|
|
|
|
function nowMs() {
|
|
if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now();
|
|
return Date.now();
|
|
}
|
|
|
|
function sleepMs(ms) {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
function recordWorkbenchTurnReadMetric(options = {}, url, statusCode, body = {}, startedAt = nowMs()) {
|
|
const performanceStore = options.backendPerformanceStore ?? options.backendPerformance;
|
|
if (!performanceStore?.recordWorkbenchTurnRead) return;
|
|
performanceStore.recordWorkbenchTurnRead({
|
|
route: url?.pathname ?? "/v1/workbench/turns/:id",
|
|
status: body?.status ?? statusClassLabel(statusCode),
|
|
degradedReason: workbenchTurnDegradedReason(body),
|
|
durationMs: nowMs() - startedAt,
|
|
responseBytes: responseBodyBytes(body),
|
|
timedOut: statusCode === 504 || body?.error?.code === "timeout"
|
|
});
|
|
}
|
|
|
|
function recordWorkbenchCacheMetric(options = {}, url, body = {}, startedAt = nowMs()) {
|
|
const performanceStore = options.backendPerformanceStore ?? options.backendPerformance;
|
|
if (!performanceStore?.recordWorkbenchCache) return;
|
|
const cache = body?.cache && typeof body.cache === "object" ? body.cache : null;
|
|
if (!cache) return;
|
|
performanceStore.recordWorkbenchCache({
|
|
route: url?.pathname ?? "/v1/workbench",
|
|
cacheKeyClass: cache.class,
|
|
cacheStatus: cache.status,
|
|
operation: "read",
|
|
durationMs: nowMs() - startedAt,
|
|
payloadBytes: cache.payloadBytes,
|
|
cacheAgeMs: cache.cacheAgeMs,
|
|
dbQueryAvoided: cache.dbQueryAvoided,
|
|
freshnessSloMs: cache.freshnessSloMs,
|
|
projectionSeq: cache.projectionSeq
|
|
});
|
|
}
|
|
|
|
function recordWorkbenchProjectionMetric(options = {}, { result = null, trace = null, projection = null, diagnostic = null } = {}) {
|
|
const performanceStore = options.backendPerformanceStore ?? options.backendPerformance;
|
|
if (!performanceStore?.recordWorkbenchProjection || !projection) return;
|
|
const sourceLatestSeq = workbenchProjectionSourceLatestSeq(result, trace, projection);
|
|
const lastProjectedSeq = Number(diagnostic?.lastProjectedSeq ?? projection?.lastProjectedSeq ?? 0);
|
|
if (!Number.isFinite(sourceLatestSeq) && !Number.isFinite(lastProjectedSeq)) return;
|
|
performanceStore.recordWorkbenchProjection({
|
|
sourceLatestSeq: Number.isFinite(sourceLatestSeq) ? sourceLatestSeq : lastProjectedSeq,
|
|
lastProjectedSeq: Number.isFinite(lastProjectedSeq) ? lastProjectedSeq : 0,
|
|
sourceLatestEventAt: workbenchProjectionSourceLatestAt(result, trace, projection),
|
|
projectedLatestEventAt: diagnostic?.updatedAt ?? projection?.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null,
|
|
terminalSourceAt: workbenchTerminalSourceAt(result, projection),
|
|
terminalProjectedAt: diagnostic?.projectionStatus === "caught-up" ? diagnostic?.updatedAt ?? projection?.updatedAt ?? trace?.updatedAt ?? result?.updatedAt ?? null : null,
|
|
status: projection?.status ?? "unknown",
|
|
reason: workbenchProjectionReason(diagnostic),
|
|
projectionStatus: diagnostic?.projectionStatus ?? "unknown",
|
|
source: "workbench_read_model"
|
|
});
|
|
}
|
|
|
|
function workbenchProjectionSourceLatestSeq(result = null, trace = null, projection = null) {
|
|
const candidates = [
|
|
result?.agentRun?.lastSeq,
|
|
result?.traceSummary?.agentRun?.lastSeq,
|
|
result?.traceSummary?.lastSeq,
|
|
result?.providerTrace?.lastSeq,
|
|
result?.runnerTrace?.lastEvent?.sourceSeq,
|
|
result?.runnerTrace?.lastEvent?.seq,
|
|
result?.runnerTrace?.eventCount,
|
|
trace?.lastSeq,
|
|
trace?.lastEvent?.sourceSeq,
|
|
trace?.lastEvent?.seq,
|
|
trace?.eventCount,
|
|
projection?.lastProjectedSeq,
|
|
projection?.eventCount
|
|
];
|
|
for (const value of candidates) {
|
|
const seq = Number(value);
|
|
if (Number.isFinite(seq) && seq >= 0) return Math.floor(seq);
|
|
}
|
|
return NaN;
|
|
}
|
|
|
|
function workbenchProjectionSourceLatestAt(result = null, trace = null, projection = null) {
|
|
return result?.traceSummary?.updatedAt
|
|
?? result?.agentRun?.updatedAt
|
|
?? result?.providerTrace?.updatedAt
|
|
?? result?.runnerTrace?.updatedAt
|
|
?? trace?.updatedAt
|
|
?? result?.updatedAt
|
|
?? projection?.updatedAt
|
|
?? null;
|
|
}
|
|
|
|
function workbenchTerminalSourceAt(result = null, projection = null) {
|
|
if (projection?.terminal !== true) return null;
|
|
return result?.traceSummary?.updatedAt
|
|
?? result?.agentRun?.updatedAt
|
|
?? result?.updatedAt
|
|
?? projection?.updatedAt
|
|
?? null;
|
|
}
|
|
|
|
function workbenchProjectionReason(diagnostic = null) {
|
|
if (diagnostic?.blocker) return diagnostic.blocker.code ?? "projection_blocked";
|
|
if (diagnostic?.projectionStatus === "projecting") return "projecting";
|
|
if (diagnostic?.projectionStatus === "caught-up") return "none";
|
|
return diagnostic?.projectionStatus ?? "unknown";
|
|
}
|
|
|
|
function workbenchTurnDegradedReason(body = {}) {
|
|
if (body?.projection?.blocker?.code) return body.projection.blocker.code;
|
|
if (body?.blocker?.code) return body.blocker.code;
|
|
if (body?.error?.code) return body.error.code;
|
|
if (body?.projectionStatus === "projecting") return "projecting";
|
|
if (body?.projectionStatus === "blocked") return "projection_blocked";
|
|
return "none";
|
|
}
|
|
|
|
function responseBodyBytes(body = {}) {
|
|
try { return Buffer.byteLength(JSON.stringify(body)); } catch { return 0; }
|
|
}
|
|
|
|
function statusClassLabel(value) {
|
|
const status = Number(value);
|
|
if (!Number.isFinite(status) || status <= 0) return "unknown";
|
|
return `${Math.floor(status / 100)}xx`;
|
|
}
|
|
|
|
async function authenticateWorkbenchRead(request, response, options) {
|
|
const access = options.accessController;
|
|
if (!access?.authenticate) {
|
|
sendJson(response, 503, workbenchError("workbench_access_controller_missing", "Workbench read model requires access controller."));
|
|
return null;
|
|
}
|
|
await access.ensureBootstrap?.();
|
|
const auth = await access.authenticate(request, { required: true });
|
|
if (!auth?.ok) {
|
|
sendJson(response, auth?.status ?? 401, auth ?? workbenchError("auth_required", "Authentication is required."));
|
|
return null;
|
|
}
|
|
return auth;
|
|
}
|
|
|
|
async function handleWorkbenchSessionList(request, response, url, options, actor) {
|
|
const startedAt = nowMs();
|
|
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) return sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench session list is keyed by sessionId only."));
|
|
const limit = boundedSessionListLimit(url.searchParams.get("limit"));
|
|
const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after"));
|
|
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
|
|
const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId"));
|
|
const runtime = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
|
if (typeof runtime?.listWorkbenchSessions !== "function") {
|
|
throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for session list.", false);
|
|
}
|
|
const payload = await runtime.listWorkbenchSessions({
|
|
limit,
|
|
offset,
|
|
cursor: url.searchParams.get("cursor") ?? url.searchParams.get("after") ?? null,
|
|
includeSessionId: includeRouteId ?? null,
|
|
actor: { id: actor.id, role: actor.role ?? "user", valuesRedacted: true }
|
|
});
|
|
recordWorkbenchCacheMetric(options, url, payload, startedAt);
|
|
return sendJson(response, 200, payload);
|
|
}
|
|
|
|
function runtimeDependencyError(code, message, retryable = true) {
|
|
const error = new Error(message || code);
|
|
error.name = "WorkbenchRuntimeDependencyError";
|
|
error.code = code;
|
|
error.data = { serviceId: "hwlab-workbench-runtime", retryable, transient: retryable, valuesRedacted: true };
|
|
return error;
|
|
}
|
|
|
|
function workbenchRuntimeFactsReadModel(request, options, actor) {
|
|
const runtime = options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
|
if (typeof runtime?.queryWorkbenchFacts !== "function") {
|
|
throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for Workbench facts reads.", false);
|
|
}
|
|
return {
|
|
queryFacts: (query = {}) => runtime.queryWorkbenchFacts({
|
|
...query,
|
|
actor: { id: actor.id, role: actor.role ?? "user", valuesRedacted: true }
|
|
})
|
|
};
|
|
}
|
|
|
|
function workbenchRuntimeOutboxReader(request, options) {
|
|
return options.workbenchRuntime ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent });
|
|
}
|
|
|
|
async function handleWorkbenchSessionListBunLegacy(response, url, options, actor) {
|
|
if (url.searchParams.has("projectId") || url.searchParams.has("workspaceId")) return sendJson(response, 400, workbenchError("workbench_authority_removed", "Workbench session list is keyed by sessionId only."));
|
|
const perf = options.backendPerformance;
|
|
const limit = boundedSessionListLimit(url.searchParams.get("limit"));
|
|
const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after"));
|
|
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
|
|
const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId"));
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
const pageQuery = { ownerUserId: actor.role === "admin" ? undefined : actor.id, limit: limit + offset + 1, families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES, sessionsOrder: "updated_desc", sessionProjection: "summary" };
|
|
const includeQueryPromise = includeRouteId ? queryWorkbenchSessionInclude(readModel, includeRouteId, perf) : null;
|
|
const pageResult = await (perf ? perf.measure("workbench_session_page_query", () => readModel.queryFacts(pageQuery)) : readModel.queryFacts(pageQuery));
|
|
if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error));
|
|
let pageFacts = pageResult.facts;
|
|
let naturalPage = visibleFactSessions(pageFacts, actor).slice(offset, offset + limit + 1);
|
|
if (includeRouteId && !naturalPage.some((session) => factSessionMatchesRouteId(session, includeRouteId))) {
|
|
const includeResult = await includeQueryPromise;
|
|
if (includeResult.error) return sendJson(response, 503, workbenchProjectionStoreError(includeResult.error));
|
|
const included = visibleFactSessions(includeResult.facts, actor)[0] ?? null;
|
|
if (included) {
|
|
pageFacts = mergeFactSets(pageFacts, includeResult.facts);
|
|
naturalPage = [included, ...naturalPage];
|
|
}
|
|
}
|
|
const pageSessions = naturalPage.slice(0, limit);
|
|
const summaryResult = await (perf ? perf.measure("workbench_session_summary_query", () => queryFactsForSessionSummaries(readModel, pageSessions)) : queryFactsForSessionSummaries(readModel, pageSessions));
|
|
if (summaryResult.error) return sendJson(response, 503, workbenchProjectionStoreError(summaryResult.error));
|
|
pageFacts = mergeFactSets(pageFacts, summaryResult.facts);
|
|
const summaries = pageSessions.map((session) => factSessionSummary(session, pageFacts)).filter(Boolean);
|
|
const hasMore = naturalPage.length > limit;
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "succeeded",
|
|
contractVersion: "workbench-sessions-v1",
|
|
sessions: summaries,
|
|
count: summaries.length,
|
|
cursor: offset > 0 ? cursorFromOffset(offset) : null,
|
|
hasMore,
|
|
nextCursor: hasMore ? cursorFromOffset(offset + limit) : null,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
});
|
|
}
|
|
|
|
function queryWorkbenchSessionInclude(readModel, includeRouteId, perf = null) {
|
|
const run = () => queryFactsByRouteId(readModel, includeRouteId, { families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES });
|
|
const promise = perf ? perf.measure("workbench_session_include_query", run) : run();
|
|
return promise.catch((error) => ({ error }));
|
|
}
|
|
|
|
function sessionMatchesRouteId(session, routeId) {
|
|
const sessionId = safeSessionId(routeId);
|
|
if (sessionId && session?.id === sessionId) return true;
|
|
const conversationId = safeConversationId(routeId);
|
|
return Boolean(conversationId && session?.conversationId === conversationId);
|
|
}
|
|
|
|
async function queryFactsByRouteId(readModel, routeId, query = {}) {
|
|
const baseQuery = { ...query, limit: query.limit ?? MAX_PAGE_LIMIT };
|
|
const sessionId = safeSessionId(routeId);
|
|
if (sessionId) return readModel.queryFacts({ ...baseQuery, sessionId });
|
|
const conversationId = safeConversationId(routeId);
|
|
if (conversationId) {
|
|
const sessionResult = await readModel.queryFacts({ ...baseQuery, families: WORKBENCH_SESSION_LIST_PAGE_FAMILIES, conversationId });
|
|
if (sessionResult.error || queryRequestsOnlySessionFacts(baseQuery)) return sessionResult;
|
|
const resolvedSessionId = factSessionId(factArray(sessionResult.facts?.sessions)[0] ?? null);
|
|
if (!resolvedSessionId) return sessionResult;
|
|
const relatedResult = await readModel.queryFacts({ ...baseQuery, sessionId: resolvedSessionId });
|
|
if (relatedResult.error) return relatedResult;
|
|
const facts = mergeFactSets(sessionResult.facts, relatedResult.facts);
|
|
return { facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0), durable: relatedResult.durable ?? sessionResult.durable, persistence: relatedResult.persistence ?? sessionResult.persistence ?? null };
|
|
}
|
|
return { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
|
}
|
|
|
|
function queryRequestsOnlySessionFacts(query = {}) {
|
|
const families = Array.isArray(query.families) ? query.families : [];
|
|
return families.length === 1 && families[0] === "sessions";
|
|
}
|
|
|
|
async function queryFactsForSessionSummaries(readModel, sessions = []) {
|
|
const sessionIds = uniqueText(sessions.map(factSessionId));
|
|
const traceIds = uniqueText(sessions.map(factLastTraceId));
|
|
if (sessionIds.length === 0 && traceIds.length === 0) return { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
|
// Session list summary is hot and is also sampled by the two-page web-probe
|
|
// observer. Avoid parallel fan-out here so one list request cannot consume
|
|
// multiple durable-read connections while another page is refreshing.
|
|
const bySession = sessionIds.length > 0
|
|
? await readModel.queryFacts({ sessionIds, families: ["messages", "turns"] })
|
|
: { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
|
if (bySession.error) return bySession;
|
|
const byTrace = traceIds.length > 0
|
|
? await readModel.queryFacts({ traceIds, families: ["checkpoints"] })
|
|
: { facts: emptyFactSet(), count: 0, durable: false, persistence: null };
|
|
if (byTrace.error) return byTrace;
|
|
const facts = mergeFactSets(bySession.facts, byTrace.facts);
|
|
return { facts, count: Object.values(facts).reduce((sum, rows) => sum + rows.length, 0), durable: bySession.durable ?? byTrace.durable, persistence: bySession.persistence ?? byTrace.persistence ?? null };
|
|
}
|
|
|
|
function visibleFactSessions(facts, actor) {
|
|
return factArray(facts?.sessions)
|
|
.filter((session) => canActorReadFactSession(session, actor))
|
|
.sort(compareFactSessionsDesc);
|
|
}
|
|
|
|
function canActorReadFactSession(session, actor) {
|
|
if (!session || !actor) return false;
|
|
if (normalizeStatus(session.status) === "archived") return false;
|
|
if (actor.role === "admin") return true;
|
|
return session.ownerUserId === actor.id;
|
|
}
|
|
|
|
async function visibleFactSessionForTrace(readModel, facts, actor, traceId) {
|
|
const visible = visibleFactSessions(facts, actor);
|
|
const byLastTrace = visible.find((item) => item.lastTraceId === traceId) ?? null;
|
|
if (byLastTrace) return { session: byLastTrace, facts };
|
|
const relatedSessionIds = uniqueText([
|
|
factTurnForTrace(facts, traceId)?.sessionId,
|
|
factArray(facts.messages).find((message) => message.traceId === traceId)?.sessionId,
|
|
factCheckpointForTrace(facts, traceId)?.sessionId
|
|
]);
|
|
const alreadyLoaded = visible.find((item) => relatedSessionIds.includes(factSessionId(item))) ?? null;
|
|
if (alreadyLoaded) return { session: alreadyLoaded, facts };
|
|
for (const sessionId of relatedSessionIds) {
|
|
const result = await queryFactsByRouteId(readModel, sessionId, { families: ["sessions"], limit: 1 });
|
|
if (result.error) return { error: result.error };
|
|
const session = visibleFactSessions(result.facts, actor)[0] ?? null;
|
|
if (session) return { session, facts: mergeFactSets(facts, result.facts) };
|
|
}
|
|
return { session: null, facts };
|
|
}
|
|
|
|
function factSessionMatchesRouteId(session, routeId) {
|
|
const sessionId = safeSessionId(routeId);
|
|
if (sessionId && factSessionId(session) === sessionId) return true;
|
|
const conversationId = safeConversationId(routeId);
|
|
return Boolean(conversationId && session?.conversationId === conversationId);
|
|
}
|
|
|
|
function factSessionSummary(session, facts = {}) {
|
|
const sessionId = factSessionId(session);
|
|
if (!sessionId) return null;
|
|
const traceId = factLastTraceId(session) ?? factLatestTraceIdForSession(facts, sessionId);
|
|
const projection = traceId ? factProjectionForTrace(facts, traceId) : null;
|
|
const turn = traceId ? factTurnForTrace(facts, traceId) : null;
|
|
const checkpoint = traceId ? factCheckpointForTrace(facts, traceId) : null;
|
|
const trace = traceId ? factTraceSnapshot(facts, traceId) : null;
|
|
const messages = factMessagesForSession(session, facts);
|
|
const traceStatus = normalizeTerminalStatus(trace?.status);
|
|
const checkpointStatus = normalizeTerminalStatus(checkpoint?.status);
|
|
const turnStatus = normalizeStatus(turn?.status);
|
|
const launchContext = compactLaunchContext(session?.sessionJson?.launchContext);
|
|
// Terminal checkpoint/trace evidence is the turn lifecycle authority. A
|
|
// stale running turn row must not keep the visible timer running after the
|
|
// terminal event has already been projected.
|
|
const status = normalizeStatus(checkpointStatus ?? traceStatus ?? turnStatus ?? session?.status);
|
|
const timingSource = checkpointStatus ? checkpoint : turn ?? checkpoint ?? (traceId ? null : session);
|
|
const timing = factTimingProjection(timingSource, status);
|
|
return {
|
|
sessionId,
|
|
threadId: safeOpaqueId(session?.threadId) ?? (textValue(session?.threadId) || null),
|
|
agentId: session?.agentId ?? "hwlab-code-agent",
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
|
lastTraceId: traceId,
|
|
projection,
|
|
projectionStatus: projection?.projectionStatus ?? null,
|
|
projectionHealth: projection?.projectionHealth ?? null,
|
|
staleMs: projection?.staleMs ?? null,
|
|
blocker: projection?.blocker ?? null,
|
|
providerProfile: textValue(session?.providerProfile ?? session?.sessionJson?.providerProfile) || null,
|
|
launchContext,
|
|
messageCount: messages.length,
|
|
firstUserMessagePreview: firstUserPreview(messages),
|
|
updatedAt: factUpdatedAt(session),
|
|
turnSummary: turn ? {
|
|
turnId: factTurnId(turn, traceId),
|
|
traceId,
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
|
eventCount: projection?.lastProjectedSeq ?? trace?.eventCount,
|
|
projection,
|
|
projectionStatus: projection?.projectionStatus ?? null,
|
|
projectionHealth: projection?.projectionHealth ?? null,
|
|
staleMs: projection?.staleMs ?? null,
|
|
blocker: projection?.blocker ?? null,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
updatedAt: factUpdatedAt(turn)
|
|
} : null,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function factSessionDetail(session, facts = {}) {
|
|
const sessionId = factSessionId(session);
|
|
return {
|
|
...factSessionSummary(session, facts),
|
|
metadata: {
|
|
startedAt: session?.startedAt ?? session?.createdAt ?? null,
|
|
endedAt: session?.endedAt ?? null,
|
|
ownerUserId: session?.ownerUserId ?? null,
|
|
launchContext: compactLaunchContext(session?.sessionJson?.launchContext)
|
|
},
|
|
messagePageUrl: sessionId ? `/v1/workbench/sessions/${encodeURIComponent(sessionId)}/messages` : null,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
}
|
|
|
|
function compactLaunchContext(value) {
|
|
if (!value || typeof value !== "object" || Array.isArray(value)) return null;
|
|
const projectId = textValue(value.projectId) || null;
|
|
const taskRef = textValue(value.taskRef) || null;
|
|
if (!projectId && !taskRef) return null;
|
|
return {
|
|
source: textValue(value.source) || "project-management",
|
|
projectId,
|
|
taskRef,
|
|
sourceId: textValue(value.sourceId) || null,
|
|
fileRef: textValue(value.fileRef) || null,
|
|
taskId: textValue(value.taskId) || null,
|
|
title: textValue(value.title) || null,
|
|
status: textValue(value.status) || null,
|
|
route: textValue(value.route) || "/projects/mdtodo",
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function factMessagesForSession(session, facts = {}) {
|
|
const sessionId = factSessionId(session);
|
|
if (!sessionId) return [];
|
|
const partsByMessageId = new Map();
|
|
for (const part of factArray(facts.parts)) {
|
|
if (part.sessionId !== sessionId) continue;
|
|
const messageId = textValue(part.messageId);
|
|
if (!messageId) continue;
|
|
const existing = partsByMessageId.get(messageId) ?? [];
|
|
existing.push(part);
|
|
partsByMessageId.set(messageId, existing);
|
|
}
|
|
return factArray(facts.messages)
|
|
.filter((message) => message.sessionId === sessionId)
|
|
.sort(compareFactMessagesAsc)
|
|
.map((message) => factMessageDto(message, partsByMessageId.get(message.messageId) ?? [], facts))
|
|
.filter(Boolean);
|
|
}
|
|
|
|
function factMessageDto(message, parts = [], facts = {}) {
|
|
const messageId = safeMessageId(message?.messageId) || textValue(message?.messageId);
|
|
if (!messageId) return null;
|
|
const traceId = safeTraceId(message?.traceId) ?? null;
|
|
const role = textValue(message?.role) || "agent";
|
|
const turn = traceId ? factTurnForTrace(facts, traceId) : null;
|
|
const checkpoint = traceId ? factCheckpointForTrace(facts, traceId) : null;
|
|
const messageStatus = normalizeStatus(message?.status);
|
|
const assistantLike = isAssistantLikeRole(role);
|
|
const checkpointStatus = assistantLike ? normalizeTerminalStatus(checkpoint?.status) : null;
|
|
const status = normalizeStatus(checkpointStatus ?? (assistantLike ? turn?.status : null) ?? messageStatus);
|
|
const timing = assistantLike
|
|
? factCombinedTimingProjection(status, checkpoint, turn, message)
|
|
: factTimingProjection(message, status);
|
|
const legacyText = projectionText(message?.text, message?.content, message?.message, message?.finalResponse);
|
|
const normalizedParts = parts.length > 0
|
|
? [...parts].sort(compareFactPartsAsc).map((part) => factPartDto(part, messageId, traceId)).filter(Boolean)
|
|
: !assistantLike && legacyText ? [partFact({ type: "text", text: legacyText, status: message?.status }, 0, messageId, traceId)] : [];
|
|
const text = factMessageAuthorityText({ role, status, parts: normalizedParts, legacyText });
|
|
return {
|
|
messageId,
|
|
role,
|
|
sessionId: (safeSessionId(message?.sessionId) ?? textValue(message?.sessionId)) || null,
|
|
traceId,
|
|
turnId: safeTurnId(message?.turnId) || traceId,
|
|
status,
|
|
parts: normalizedParts,
|
|
text: text || "",
|
|
textPreview: text ? text.slice(0, 240) : null,
|
|
createdAt: textValue(message?.createdAt) || null,
|
|
updatedAt: factUpdatedAt(message),
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
projectionStatus: null,
|
|
projectionHealth: null,
|
|
valuesRedacted: message?.valuesRedacted !== false
|
|
};
|
|
}
|
|
|
|
function factMessageAuthorityText({ role, status, parts = [], legacyText = null } = {}) {
|
|
if (!isAssistantLikeRole(role)) return firstFactPartText(parts) ?? legacyText ?? "";
|
|
if (!isTerminalProjectionStatus(status)) return "";
|
|
return firstFactPartText(parts, "final_response") ?? "";
|
|
}
|
|
|
|
function isTerminalProjectionStatus(status) {
|
|
return TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status);
|
|
}
|
|
|
|
function firstFactPartText(parts = [], type = null) {
|
|
for (const part of parts) {
|
|
if (type && part?.type !== type) continue;
|
|
const text = projectionText(part?.text);
|
|
if (text) return text;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function factMessageFinalResponseText(message) {
|
|
if (!message || !isAssistantLikeRole(message.role) || !isTerminalProjectionStatus(message.status)) return null;
|
|
return firstFactPartText(message.parts, "final_response");
|
|
}
|
|
|
|
function factPartDto(part, messageId, traceId) {
|
|
const partId = safePartId(part?.partId) || textValue(part?.partId) || `prt_${hash(`${messageId}:${part?.partIndex ?? 0}:${part?.partType ?? part?.type ?? "text"}`).slice(0, 24)}`;
|
|
const text = projectionText(part?.text, part?.content, part?.message);
|
|
return {
|
|
partId,
|
|
messageId,
|
|
traceId: safeTraceId(part?.traceId) ?? traceId,
|
|
type: textValue(part?.partType ?? part?.type) || "text",
|
|
text: text || null,
|
|
status: normalizeStatus(part?.status),
|
|
toolName: textValue(part?.toolName ?? part?.name) || null,
|
|
createdAt: textValue(part?.createdAt ?? part?.occurredAt) || null,
|
|
valuesRedacted: part?.valuesRedacted !== false
|
|
};
|
|
}
|
|
|
|
function factTurnForTrace(facts = {}, traceId, turnId = null) {
|
|
const safeTrace = safeTraceId(traceId);
|
|
if (!safeTrace) return null;
|
|
const requestedTurn = safeTurnId(turnId);
|
|
const matches = factArray(facts.turns)
|
|
.filter((turn) => turn.traceId === safeTrace && (!requestedTurn || turn.turnId === requestedTurn || turn.turnId === safeTrace))
|
|
.sort(compareFactRecordsDesc);
|
|
return matches[0] ?? null;
|
|
}
|
|
|
|
function factTurnSnapshot({ turn = null, session = null, facts = {}, traceId, turnId = null } = {}) {
|
|
const safeTrace = safeTraceId(traceId);
|
|
const resolvedTurnId = factTurnId(turn, safeTrace) ?? safeTurnId(turnId) ?? safeTrace;
|
|
const messages = session ? factMessagesForSession(session, facts) : [];
|
|
const userMessage = messages.find((message) => message.role === "user") ?? null;
|
|
const assistantMessage = [...messages].reverse().find((message) => isAssistantLikeRole(message.role)) ?? null;
|
|
const checkpoint = safeTrace ? factCheckpointForTrace(facts, safeTrace) : null;
|
|
const trace = factTraceSnapshot(facts, safeTrace);
|
|
const checkpointStatus = normalizeTerminalStatus(checkpoint?.status);
|
|
const traceStatus = normalizeTerminalStatus(trace.status);
|
|
const status = normalizeStatus(checkpointStatus ?? traceStatus ?? turn?.status ?? session?.status);
|
|
const terminal = isTerminalProjectionStatus(status);
|
|
const assistantText = terminal ? factMessageFinalResponseText(assistantMessage) : null;
|
|
const timing = factCombinedTimingProjection(status, checkpoint, turn, trace);
|
|
return {
|
|
turnId: resolvedTurnId,
|
|
traceId: safeTrace,
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal,
|
|
sessionId: factSessionId(session) ?? turn?.sessionId ?? checkpoint?.sessionId ?? null,
|
|
threadId: safeOpaqueId(session?.threadId) ?? (textValue(session?.threadId) || null),
|
|
userMessageId: userMessage?.messageId ?? null,
|
|
assistantMessageId: assistantMessage?.messageId ?? turn?.messageId ?? null,
|
|
assistantText: assistantText ?? null,
|
|
finalResponse: assistantText ? { text: assistantText, sealed: true, source: "message-part" } : null,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
agentRun: checkpoint ? {
|
|
runId: textValue(checkpoint.runId) || null,
|
|
commandId: textValue(checkpoint.commandId) || null,
|
|
status,
|
|
lastSeq: factSeq(checkpoint),
|
|
valuesRedacted: true
|
|
} : null,
|
|
trace: {
|
|
traceId: safeTrace,
|
|
status: trace.status,
|
|
eventCount: trace.eventCount,
|
|
timing: trace.timing,
|
|
startedAt: trace.startedAt,
|
|
lastEventAt: trace.lastEventAt,
|
|
finishedAt: trace.finishedAt,
|
|
durationMs: trace.durationMs,
|
|
updatedAt: trace.updatedAt
|
|
},
|
|
urls: {
|
|
self: resolvedTurnId ? `/v1/workbench/turns/${encodeURIComponent(resolvedTurnId)}` : null,
|
|
traceEvents: safeTrace ? `/v1/workbench/traces/${encodeURIComponent(safeTrace)}/events` : null
|
|
}
|
|
};
|
|
}
|
|
|
|
function factTraceSnapshot(facts = {}, traceId) {
|
|
const safeTrace = safeTraceId(traceId);
|
|
const events = factArray(facts.traceEvents)
|
|
.filter((event) => !safeTrace || event.traceId === safeTrace)
|
|
.sort(compareFactTraceEventsAsc)
|
|
.map(factTraceEventDto)
|
|
.filter(Boolean);
|
|
const status = durableTraceStatus(events);
|
|
const lastEvent = events.at(-1) ?? null;
|
|
const checkpoint = factCheckpointForTrace(facts, safeTrace);
|
|
const timing = factTraceTimingProjection(events, checkpoint, status);
|
|
return {
|
|
traceId: safeTrace,
|
|
status,
|
|
eventCount: events.length,
|
|
events,
|
|
lastEvent,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
updatedAt: lastEvent?.updatedAt ?? lastEvent?.createdAt ?? checkpoint?.updatedAt ?? null,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
}
|
|
|
|
function factTraceTimingProjection(events = [], checkpoint = null, status = null) {
|
|
const checkpointTiming = factTimingProjection(checkpoint, status);
|
|
const normalizedStatus = normalizeStatus(status ?? checkpoint?.status);
|
|
const terminal = TERMINAL_STATUSES.has(normalizedStatus) && !RUNNING_STATUSES.has(normalizedStatus);
|
|
const observedAt = new Date().toISOString();
|
|
const eventStartTimes = factArray(events).flatMap((event) => [event?.createdAt, event?.occurredAt]);
|
|
const eventActivityTimes = factArray(events).flatMap((event) => [event?.createdAt, event?.occurredAt, event?.updatedAt]);
|
|
const startedAt = firstTimestampIso(checkpointTiming.startedAt, ...eventStartTimes);
|
|
const lastEventAt = latestTimestampIso(checkpointTiming.lastEventAt, ...eventActivityTimes);
|
|
const finishedAt = terminal ? latestTimestampIso(checkpointTiming.finishedAt, lastEventAt) : null;
|
|
const durationMs = elapsedFactMs(startedAt, terminal ? finishedAt : observedAt);
|
|
const lastEventAgeMs = terminal ? null : elapsedFactMs(lastEventAt, observedAt);
|
|
return {
|
|
...checkpointTiming,
|
|
startedAt,
|
|
lastEventAt,
|
|
finishedAt,
|
|
durationMs,
|
|
observedAt: terminal ? null : observedAt,
|
|
lastEventAgeMs,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function factTraceEventDto(event, index) {
|
|
const seq = factProjectedSeq(event);
|
|
if (!seq) return null;
|
|
const sourceSeq = Number.isFinite(Number(event?.sourceSeq)) && Number(event.sourceSeq) >= 0 ? Math.trunc(Number(event.sourceSeq)) : null;
|
|
return {
|
|
...event,
|
|
id: textValue(event?.id) || textValue(event?.sourceEventId) || null,
|
|
seq,
|
|
projectedSeq: seq,
|
|
sourceSeq,
|
|
type: textValue(event?.type ?? event?.eventType) || "event",
|
|
label: textValue(event?.label ?? event?.eventType ?? event?.type) || null,
|
|
status: normalizeStatus(event?.status),
|
|
createdAt: textValue(event?.createdAt ?? event?.occurredAt) || null,
|
|
updatedAt: factUpdatedAt(event),
|
|
valuesRedacted: event?.valuesRedacted !== false
|
|
};
|
|
}
|
|
|
|
function factProjectionForTrace(facts = {}, traceId) {
|
|
const checkpoint = factCheckpointForTrace(facts, traceId);
|
|
if (!checkpoint) {
|
|
return {
|
|
projectionStatus: "unknown",
|
|
projectionHealth: "unknown",
|
|
lastProjectedSeq: null,
|
|
sourceRunId: null,
|
|
sourceCommandId: null,
|
|
staleMs: null,
|
|
blocker: null,
|
|
updatedAt: null,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
const projectionStatus = normalizeProjectionStatus(checkpoint.projectionStatus);
|
|
const projectionHealth = normalizeProjectionHealth(checkpoint.projectionHealth, projectionStatus);
|
|
const diagnostic = objectValue(checkpoint.diagnostic);
|
|
const timing = factTimingProjection(checkpoint, normalizeStatus(checkpoint.status));
|
|
return {
|
|
projectionStatus,
|
|
projectionHealth,
|
|
lastProjectedSeq: factSeq(checkpoint),
|
|
sourceRunId: textValue(checkpoint.runId ?? checkpoint.sourceRunId) || null,
|
|
sourceCommandId: textValue(checkpoint.commandId ?? checkpoint.sourceCommandId) || null,
|
|
staleMs: null,
|
|
blocker: diagnostic.blocker ?? checkpoint.blocker ?? null,
|
|
timing,
|
|
startedAt: timing.startedAt,
|
|
lastEventAt: timing.lastEventAt,
|
|
finishedAt: timing.finishedAt,
|
|
durationMs: timing.durationMs,
|
|
updatedAt: factUpdatedAt(checkpoint),
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function factTimingProjection(record = null, status = null) {
|
|
const source = objectValue(record?.timing);
|
|
const observedAt = new Date().toISOString();
|
|
const startedAt = timestampIso(source?.startedAt ?? record?.startedAt ?? record?.createdAt);
|
|
const lastEventAt = timestampIso(source?.lastEventAt ?? record?.lastEventAt ?? record?.updatedAt ?? record?.createdAt);
|
|
const normalizedStatus = normalizeStatus(status ?? record?.status);
|
|
const terminal = record?.terminal === true || TERMINAL_STATUSES.has(normalizedStatus) && !RUNNING_STATUSES.has(normalizedStatus);
|
|
const finishedAt = terminal ? timestampIso(source?.finishedAt ?? record?.finishedAt ?? record?.completedAt) : null;
|
|
const durationMs = elapsedFactMs(startedAt, terminal ? finishedAt : observedAt);
|
|
const lastEventAgeMs = terminal ? null : elapsedFactMs(lastEventAt, observedAt);
|
|
return { startedAt, lastEventAt, finishedAt, durationMs, observedAt: terminal ? null : observedAt, lastEventAgeMs, valuesRedacted: source?.valuesRedacted !== false };
|
|
}
|
|
|
|
function factCombinedTimingProjection(status = null, ...records) {
|
|
const normalizedStatus = normalizeStatus(status ?? records.find((record) => record?.status)?.status);
|
|
const terminal = TERMINAL_STATUSES.has(normalizedStatus) && !RUNNING_STATUSES.has(normalizedStatus);
|
|
const observedAt = new Date().toISOString();
|
|
const timings = records.map((record) => factTimingSource(record)).filter(Boolean);
|
|
const startedAt = firstTimestampIso(...timings.map((timing) => timing.startedAt));
|
|
const lastEventAt = latestTimestampIso(...timings.map((timing) => timing.lastEventAt));
|
|
const finishedAt = terminal ? latestTimestampIso(...timings.map((timing) => timing.finishedAt)) : null;
|
|
const durationMs = elapsedFactMs(startedAt, terminal ? finishedAt : observedAt);
|
|
const lastEventAgeMs = terminal ? null : elapsedFactMs(lastEventAt, observedAt);
|
|
return { startedAt, lastEventAt, finishedAt, durationMs, observedAt: terminal ? null : observedAt, lastEventAgeMs, valuesRedacted: true };
|
|
}
|
|
|
|
function factTimingSource(record = null) {
|
|
if (!record) return null;
|
|
const source = objectValue(record?.timing);
|
|
return {
|
|
startedAt: timestampIso(source?.startedAt ?? record?.startedAt ?? record?.createdAt),
|
|
lastEventAt: timestampIso(source?.lastEventAt ?? record?.lastEventAt ?? record?.updatedAt ?? record?.createdAt),
|
|
finishedAt: timestampIso(source?.finishedAt ?? record?.finishedAt ?? record?.completedAt)
|
|
};
|
|
}
|
|
|
|
function firstTimestampIso(...values) {
|
|
for (const value of values) {
|
|
const timestamp = timestampIso(value);
|
|
if (timestamp) return timestamp;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function latestTimestampIso(...values) {
|
|
let latest = null;
|
|
let latestMs = Number.NEGATIVE_INFINITY;
|
|
for (const value of values) {
|
|
const timestamp = timestampIso(value);
|
|
if (!timestamp) continue;
|
|
const ms = Date.parse(timestamp);
|
|
if (!Number.isFinite(ms) || ms < latestMs) continue;
|
|
latest = timestamp;
|
|
latestMs = ms;
|
|
}
|
|
return latest;
|
|
}
|
|
|
|
function elapsedFactMs(startedAt, endedAt) {
|
|
const start = Date.parse(String(startedAt ?? ""));
|
|
const end = Date.parse(String(endedAt ?? ""));
|
|
if (!Number.isFinite(start) || !Number.isFinite(end) || end < start) return null;
|
|
return Math.trunc(end - start);
|
|
}
|
|
|
|
function timestampIso(value) {
|
|
const ms = Date.parse(String(value ?? ""));
|
|
return Number.isFinite(ms) ? new Date(ms).toISOString() : null;
|
|
}
|
|
|
|
function factCheckpointForTrace(facts = {}, traceId) {
|
|
const safeTrace = safeTraceId(traceId);
|
|
if (!safeTrace) return null;
|
|
return factArray(facts.checkpoints)
|
|
.filter((checkpoint) => checkpoint.traceId === safeTrace)
|
|
.sort(compareFactRecordsDesc)[0] ?? null;
|
|
}
|
|
|
|
function workbenchProjectionStoreError(error) {
|
|
const data = objectValue(error?.data) ?? {};
|
|
const retryAfterNumber = Number(data.retryAfterMs);
|
|
const retryAfterMs = Number.isFinite(retryAfterNumber) && retryAfterNumber >= 0 ? Math.trunc(retryAfterNumber) : null;
|
|
const retryAttempt = nonNegativeIntegerOrNull(data.retryAttempt ?? data.retryAttempts);
|
|
const retryMax = nonNegativeIntegerOrNull(data.retryMax);
|
|
const retryDelaysMs = Array.isArray(data.retryDelaysMs)
|
|
? data.retryDelaysMs.map(nonNegativeIntegerOrNull).filter((value) => value !== null)
|
|
: [];
|
|
return workbenchError("projection_store_unavailable", "Workbench durable projection store is unavailable.", {
|
|
causeCode: error?.code ?? null,
|
|
queryResult: textValue(data.queryResult) || null,
|
|
blockedLayer: textValue(data.blockedLayer) || null,
|
|
retryable: data.retryable !== false,
|
|
transient: data.transient === true || data.retryable === true,
|
|
retryAfterMs: retryAfterMs ?? null,
|
|
retryAttempt,
|
|
retryMax,
|
|
retryAttempts: retryAttempt,
|
|
retryExhausted: data.retryExhausted === true,
|
|
retryDelaysMs: retryDelaysMs.length > 0 ? retryDelaysMs : null
|
|
});
|
|
}
|
|
|
|
function nonNegativeIntegerOrNull(value) {
|
|
const parsed = Number(value);
|
|
return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : null;
|
|
}
|
|
|
|
function factSessionId(session) {
|
|
return safeSessionId(session?.sessionId ?? session?.id) ?? null;
|
|
}
|
|
|
|
function factLastTraceId(session) {
|
|
return safeTraceId(session?.lastTraceId ?? session?.traceId) ?? null;
|
|
}
|
|
|
|
function factLatestTraceIdForSession(facts = {}, sessionId) {
|
|
const turn = factArray(facts.turns).filter((item) => item.sessionId === sessionId).sort(compareFactRecordsDesc)[0] ?? null;
|
|
return safeTraceId(turn?.traceId) ?? null;
|
|
}
|
|
|
|
function factTurnId(turn, fallbackTraceId = null) {
|
|
return safeTurnId(turn?.turnId) ?? safeTraceId(turn?.turnId) ?? safeTraceId(fallbackTraceId) ?? null;
|
|
}
|
|
|
|
function factUpdatedAt(record) {
|
|
return textValue(record?.updatedAt ?? record?.occurredAt ?? record?.createdAt) || null;
|
|
}
|
|
|
|
function factSeq(record) {
|
|
for (const value of [record?.projectedSeq, record?.sourceSeq, record?.seq]) {
|
|
const parsed = Number(value);
|
|
if (Number.isFinite(parsed) && parsed >= 0) return Math.trunc(parsed);
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function factProjectedSeq(record) {
|
|
const parsed = Number(record?.projectedSeq);
|
|
return Number.isFinite(parsed) && parsed > 0 ? Math.trunc(parsed) : null;
|
|
}
|
|
|
|
function normalizeProjectionStatus(value) {
|
|
const status = normalizeStatus(value);
|
|
if (status === "caught-up") return "caught-up";
|
|
if (status === "caughtup") return "caught-up";
|
|
return ["projecting", "blocked", "stalled", "unknown"].includes(status) ? status : "unknown";
|
|
}
|
|
|
|
function normalizeProjectionHealth(value, projectionStatus = "unknown") {
|
|
const health = normalizeStatus(value);
|
|
if (health === "healthy" && projectionStatus === "caught-up") return "caught-up";
|
|
if (health === "healthy" && projectionStatus === "projecting") return "projecting";
|
|
if (["caught-up", "projecting", "degraded", "stalled", "unavailable", "unknown"].includes(health)) return health;
|
|
return projectionStatus === "caught-up" || projectionStatus === "projecting" ? projectionStatus : "unknown";
|
|
}
|
|
|
|
function compareFactSessionsDesc(left, right) {
|
|
return compareTimestampDesc(factUpdatedAt(left), factUpdatedAt(right)) || compareText(factSessionId(left), factSessionId(right));
|
|
}
|
|
|
|
function compareFactMessagesAsc(left, right) {
|
|
return compareNumberAsc(factSeq(left), factSeq(right)) || compareTimestampAsc(left?.createdAt ?? left?.updatedAt, right?.createdAt ?? right?.updatedAt) || compareText(left?.messageId, right?.messageId);
|
|
}
|
|
|
|
function compareFactPartsAsc(left, right) {
|
|
return compareNumberAsc(left?.partIndex, right?.partIndex) || compareNumberAsc(factSeq(left), factSeq(right)) || compareText(left?.partId, right?.partId);
|
|
}
|
|
|
|
function compareFactTraceEventsAsc(left, right) {
|
|
return compareNumberAsc(factProjectedSeq(left), factProjectedSeq(right));
|
|
}
|
|
|
|
function compareFactRecordsDesc(left, right) {
|
|
return compareNumberDesc(factSeq(left), factSeq(right)) || compareTimestampDesc(factUpdatedAt(left), factUpdatedAt(right));
|
|
}
|
|
|
|
function compareTimestampAsc(left, right) {
|
|
return timestampMs(left) - timestampMs(right);
|
|
}
|
|
|
|
function compareTimestampDesc(left, right) {
|
|
return timestampMs(right) - timestampMs(left);
|
|
}
|
|
|
|
function compareNumberAsc(left, right) {
|
|
return numericValue(left, Number.MAX_SAFE_INTEGER) - numericValue(right, Number.MAX_SAFE_INTEGER);
|
|
}
|
|
|
|
function compareNumberDesc(left, right) {
|
|
return numericValue(right, -1) - numericValue(left, -1);
|
|
}
|
|
|
|
function compareText(left, right) {
|
|
return textValue(left).localeCompare(textValue(right));
|
|
}
|
|
|
|
function timestampMs(value) {
|
|
const parsed = Date.parse(String(value ?? ""));
|
|
return Number.isFinite(parsed) ? parsed : 0;
|
|
}
|
|
|
|
function numericValue(value, fallback) {
|
|
const parsed = Number(value);
|
|
return Number.isFinite(parsed) ? parsed : fallback;
|
|
}
|
|
|
|
function mergeFactSets(...sets) {
|
|
const merged = emptyFactSet();
|
|
const keys = {
|
|
sessions: "sessionId",
|
|
messages: "messageId",
|
|
parts: "partId",
|
|
turns: "turnId",
|
|
traceEvents: "id",
|
|
checkpoints: "traceId"
|
|
};
|
|
for (const set of sets) {
|
|
for (const [key, idKey] of Object.entries(keys)) {
|
|
const seen = new Set(merged[key].map((item) => textValue(item?.[idKey])));
|
|
for (const item of factArray(set?.[key])) {
|
|
const id = textValue(item?.[idKey]);
|
|
if (id && seen.has(id)) continue;
|
|
merged[key].push(item);
|
|
if (id) seen.add(id);
|
|
}
|
|
}
|
|
}
|
|
return merged;
|
|
}
|
|
|
|
function emptyFactSet() {
|
|
return {
|
|
sessions: [],
|
|
messages: [],
|
|
parts: [],
|
|
turns: [],
|
|
traceEvents: [],
|
|
checkpoints: []
|
|
};
|
|
}
|
|
|
|
function factArray(value) {
|
|
return Array.isArray(value) ? value : [];
|
|
}
|
|
|
|
function uniqueText(values = []) {
|
|
return [...new Set(values.map((value) => textValue(value)).filter(Boolean))];
|
|
}
|
|
|
|
async function handleWorkbenchSessionDetail(request, response, options, actor, sessionId) {
|
|
const readModel = workbenchRuntimeFactsReadModel(request, options, actor);
|
|
const result = await queryFactsByRouteId(readModel, sessionId, { families: WORKBENCH_SESSION_DETAIL_FAMILIES });
|
|
if (result.error) return sendJson(response, 503, workbenchProjectionStoreError(result.error));
|
|
const session = visibleFactSessions(result.facts, actor)[0] ?? null;
|
|
if (!session) {
|
|
const body = workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId });
|
|
attachWorkbenchReadModelDiagnosticOtel(request, { route: "/v1/workbench/sessions/:id", code: "workbench_session_not_found", sessionId, count: result.count });
|
|
return sendJson(response, 404, body);
|
|
}
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "found",
|
|
contractVersion: "workbench-session-detail-v1",
|
|
session: factSessionDetail(session, result.facts),
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
});
|
|
}
|
|
|
|
async function handleWorkbenchMessagePage(request, response, url, options, actor, sessionId) {
|
|
const readModel = workbenchRuntimeFactsReadModel(request, options, actor);
|
|
const limit = boundedLimit(url.searchParams.get("limit"));
|
|
const cursorParam = url.searchParams.get("cursor") ?? url.searchParams.get("after");
|
|
const hasCursor = Boolean(textValue(cursorParam));
|
|
const result = await queryFactsByRouteId(readModel, sessionId, {
|
|
families: WORKBENCH_SESSION_MESSAGE_PAGE_FAMILIES,
|
|
...(hasCursor ? {} : { limit, messagesOrder: "updated_desc", partsOrder: "updated_desc", turnsOrder: "updated_desc", checkpointsOrder: "updated_desc" })
|
|
});
|
|
if (result.error) return sendJson(response, 503, workbenchProjectionStoreError(result.error));
|
|
const session = visibleFactSessions(result.facts, actor)[0] ?? null;
|
|
if (!session) {
|
|
const body = workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId });
|
|
attachWorkbenchReadModelDiagnosticOtel(request, { route: "/v1/workbench/sessions/:id/messages", code: "workbench_session_not_found", sessionId, count: result.count });
|
|
return sendJson(response, 404, body);
|
|
}
|
|
const messages = factMessagesForSession(session, result.facts);
|
|
const offset = hasCursor ? cursorOffset(cursorParam) : Math.max(0, messages.length - limit);
|
|
const page = messages.slice(offset, offset + limit);
|
|
const nextOffset = offset + page.length;
|
|
const resolvedSessionId = factSessionId(session);
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "succeeded",
|
|
contractVersion: "workbench-message-page-v1",
|
|
sessionId: resolvedSessionId,
|
|
messages: page,
|
|
count: page.length,
|
|
total: messages.length,
|
|
cursor: offset > 0 ? cursorFromOffset(offset) : null,
|
|
nextCursor: nextOffset < messages.length ? cursorFromOffset(nextOffset) : null,
|
|
hasMore: nextOffset < messages.length,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
});
|
|
}
|
|
|
|
async function handleWorkbenchTurnSnapshot(request, response, url, options, actor, rawTurnId) {
|
|
const startedAt = nowMs();
|
|
const queryTraceId = safeTraceId(url.searchParams.get("traceId"));
|
|
const traceId = safeTraceId(rawTurnId) ?? queryTraceId;
|
|
const turnId = safeTurnId(rawTurnId) || traceId || rawTurnId;
|
|
if (!traceId) {
|
|
const body = workbenchError("invalid_turn_id", "turnId must currently be a trace id or include traceId query.", { turnId });
|
|
recordWorkbenchTurnReadMetric(options, url, 400, body, startedAt);
|
|
return sendJson(response, 400, body);
|
|
}
|
|
const readModel = workbenchRuntimeFactsReadModel(request, options, actor);
|
|
const result = await readModel.queryFacts({ traceId, limit: MAX_PAGE_LIMIT, families: WORKBENCH_TRACE_EVENT_METADATA_FAMILIES });
|
|
if (result.error) {
|
|
const body = workbenchProjectionStoreError(result.error);
|
|
recordWorkbenchTurnReadMetric(options, url, 503, body, startedAt);
|
|
return sendJson(response, 503, body);
|
|
}
|
|
const session = visibleFactSessions(result.facts, actor).find((item) => item.lastTraceId === traceId) ?? visibleFactSessions(result.facts, actor)[0] ?? null;
|
|
const turn = factTurnForTrace(result.facts, traceId, turnId);
|
|
const found = Boolean(session || turn);
|
|
if (!found) {
|
|
const body = workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId });
|
|
attachWorkbenchReadModelDiagnosticOtel(request, { route: "/v1/workbench/turns/:id", code: "workbench_turn_not_found", traceId, turnId, count: result.count });
|
|
recordWorkbenchTurnReadMetric(options, url, 404, body, startedAt);
|
|
return sendJson(response, 404, body);
|
|
}
|
|
const projection = factProjectionForTrace(result.facts, traceId);
|
|
const snapshot = factTurnSnapshot({ turn, session, facts: result.facts, traceId, turnId });
|
|
const status = snapshot.status;
|
|
const body = {
|
|
ok: true,
|
|
status,
|
|
contractVersion: "workbench-turn-snapshot-v1",
|
|
turn: snapshot,
|
|
projection,
|
|
cache: result.cache ?? null,
|
|
projectionStatus: projection.projectionStatus,
|
|
projectionHealth: projection.projectionHealth,
|
|
lastProjectedSeq: projection.lastProjectedSeq,
|
|
sourceRunId: projection.sourceRunId,
|
|
sourceCommandId: projection.sourceCommandId,
|
|
staleMs: projection.staleMs,
|
|
blocker: projection.blocker,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
recordWorkbenchProjectionMetric(options, {
|
|
projection: { ...projection, status, eventCount: snapshot.trace?.eventCount ?? projection.lastProjectedSeq ?? 0 },
|
|
diagnostic: projection
|
|
});
|
|
recordWorkbenchTurnReadMetric(options, url, 200, body, startedAt);
|
|
recordWorkbenchCacheMetric(options, url, body, startedAt);
|
|
sendJson(response, 200, body);
|
|
}
|
|
|
|
async function handleWorkbenchTraceEventPage(request, response, url, options, actor, rawTraceId) {
|
|
const startedAt = nowMs();
|
|
const traceId = safeTraceId(rawTraceId);
|
|
if (!traceId) return sendJson(response, 400, workbenchError("invalid_trace_id", "traceId must start with trc_.", { traceId: rawTraceId }));
|
|
const readModel = workbenchRuntimeFactsReadModel(request, options, actor);
|
|
const pageOptions = tracePageOptions(url);
|
|
const metadata = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_METADATA_FAMILIES, limit: 1 });
|
|
if (metadata.error) return sendJson(response, 503, workbenchProjectionStoreError(metadata.error));
|
|
const resolved = await visibleFactSessionForTrace(readModel, metadata.facts, actor, traceId);
|
|
if (resolved.error) return sendJson(response, 503, workbenchProjectionStoreError(resolved.error));
|
|
const session = resolved.session;
|
|
const metadataFacts = resolved.facts ?? metadata.facts;
|
|
if (!session) {
|
|
const context = await readModel.queryFacts({ traceId, families: ["sessions", "turns", "checkpoints"], limit: MAX_PAGE_LIMIT });
|
|
if (context.error) return sendJson(response, 503, workbenchProjectionStoreError(context.error));
|
|
const contextSessions = visibleFactSessions(context.facts, actor);
|
|
const contextSession = contextSessions.find((item) => item.lastTraceId === traceId) ?? contextSessions[0] ?? null;
|
|
const contextTurn = factTurnForTrace(context.facts, traceId, traceId);
|
|
if (contextSession || contextTurn) {
|
|
const projection = factProjectionForTrace(context.facts, traceId);
|
|
const blocker = traceEventsReadModelBlocker("workbench_trace_metadata_missing", "Workbench trace metadata is missing from the trace events read model.", { traceId, projection, session: contextSession, turn: contextTurn, route: url.pathname });
|
|
attachWorkbenchReadModelDiagnosticOtel(request, { route: "/v1/workbench/traces/:id/events", code: "workbench_trace_metadata_missing", sessionId: factSessionId(contextSession), traceId, count: context.count });
|
|
return sendJson(response, 404, workbenchTraceEventsReadModelError(blocker, { traceId, projection, session: contextSession, turn: contextTurn }));
|
|
}
|
|
attachWorkbenchReadModelDiagnosticOtel(request, { route: "/v1/workbench/traces/:id/events", code: "workbench_trace_not_found", traceId, count: context.count });
|
|
return sendJson(response, 404, workbenchError("workbench_trace_not_found", "Workbench trace is not visible to the current actor.", { traceId }));
|
|
}
|
|
const projection = factProjectionForTrace(metadataFacts, traceId);
|
|
const pageResult = await readModel.queryFacts({ traceId, families: WORKBENCH_TRACE_EVENT_PAGE_FAMILIES, afterProjectedSeq: pageOptions.afterProjectedSeq, limit: pageOptions.limit + 1 });
|
|
if (pageResult.error) return sendJson(response, 503, workbenchProjectionStoreError(pageResult.error));
|
|
const page = traceEventPageFromFacts(pageResult.facts.traceEvents, pageOptions, { total: projection.lastProjectedSeq, traceStatus: session.status });
|
|
const missingTraceEvents = traceEventPageMissing(page, projection, pageOptions);
|
|
if (missingTraceEvents) {
|
|
const blocker = traceEventsReadModelBlocker("workbench_trace_events_missing", "Workbench trace events are missing from the durable read model.", { traceId, projection, session, route: url.pathname });
|
|
page.blocker = blocker;
|
|
}
|
|
const responseProjection = page.blocker ? blockedTraceEventProjection(projection, page.blocker) : projection;
|
|
const body = {
|
|
ok: true,
|
|
status: page.blocker ? "blocked" : "succeeded",
|
|
contractVersion: "workbench-trace-events-v1",
|
|
traceId,
|
|
sessionId: factSessionId(session),
|
|
threadId: safeOpaqueId(session?.threadId) ?? (textValue(session?.threadId) || null),
|
|
...page,
|
|
projection: responseProjection,
|
|
cache: pageResult.cache ?? metadata.cache ?? null,
|
|
projectionStatus: responseProjection.projectionStatus,
|
|
projectionHealth: responseProjection.projectionHealth,
|
|
lastProjectedSeq: responseProjection.lastProjectedSeq,
|
|
sourceRunId: responseProjection.sourceRunId,
|
|
sourceCommandId: responseProjection.sourceCommandId,
|
|
staleMs: responseProjection.staleMs,
|
|
blocker: responseProjection.blocker,
|
|
timing: responseProjection.timing,
|
|
startedAt: responseProjection.startedAt,
|
|
lastEventAt: responseProjection.lastEventAt,
|
|
finishedAt: responseProjection.finishedAt,
|
|
durationMs: responseProjection.durationMs,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
recordWorkbenchCacheMetric(options, url, body, startedAt);
|
|
sendJson(response, 200, body);
|
|
}
|
|
|
|
function traceEventPageFromFacts(sourceEvents, options, metadata = {}) {
|
|
const rows = factArray(sourceEvents).filter((event) => factProjectedSeq(event)).sort(compareFactTraceEventsAsc);
|
|
const collision = projectedSeqCollision(rows);
|
|
if (collision) return blockedTraceEventPage(options, metadata, collision);
|
|
const pageRows = rows.slice(0, options.limit);
|
|
const events = pageRows.map(factTraceEventDto).filter(Boolean);
|
|
const toProjectedSeq = events.length ? events.at(-1).projectedSeq : null;
|
|
const hasMore = rows.length > options.limit;
|
|
const total = Number.isFinite(Number(metadata.total)) ? Math.trunc(Number(metadata.total)) : hasMore ? null : Math.max(options.afterProjectedSeq, toProjectedSeq ?? options.afterProjectedSeq);
|
|
return {
|
|
events,
|
|
eventCount: total ?? Math.max(options.afterProjectedSeq, toProjectedSeq ?? options.afterProjectedSeq),
|
|
range: {
|
|
afterProjectedSeq: options.afterProjectedSeq,
|
|
fromProjectedSeq: events.length ? events[0].projectedSeq : null,
|
|
toProjectedSeq,
|
|
limit: options.limit,
|
|
returned: events.length,
|
|
total
|
|
},
|
|
hasMore,
|
|
nextProjectedSeq: events.length ? toProjectedSeq : options.afterProjectedSeq,
|
|
nextCursor: hasMore && toProjectedSeq ? `projected:${toProjectedSeq}` : null,
|
|
traceStatus: metadata.traceStatus ?? "unknown",
|
|
updatedAt: events.at(-1)?.updatedAt ?? events.at(-1)?.createdAt ?? null
|
|
};
|
|
}
|
|
|
|
function projectedSeqCollision(rows = []) {
|
|
const seen = new Map();
|
|
for (const row of rows) {
|
|
const projectedSeq = factProjectedSeq(row);
|
|
if (!projectedSeq) continue;
|
|
const id = textValue(row?.id ?? row?.sourceEventId) || null;
|
|
const existing = seen.get(projectedSeq);
|
|
if (existing && existing.id !== id) {
|
|
return {
|
|
code: "projected_seq_collision",
|
|
category: "workbench-projection",
|
|
layer: "workbench-trace-events",
|
|
message: "Trace projection contains duplicate projectedSeq values; canonical event order is blocked until projection is rebuilt.",
|
|
projectedSeq,
|
|
eventIds: [existing.id, id].filter(Boolean),
|
|
retryable: false,
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
if (!existing) seen.set(projectedSeq, { id });
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function blockedTraceEventPage(options, metadata = {}, blocker) {
|
|
const total = Number.isFinite(Number(metadata.total)) ? Math.trunc(Number(metadata.total)) : null;
|
|
return {
|
|
events: [],
|
|
eventCount: total ?? options.afterProjectedSeq,
|
|
range: {
|
|
afterProjectedSeq: options.afterProjectedSeq,
|
|
fromProjectedSeq: null,
|
|
toProjectedSeq: null,
|
|
limit: options.limit,
|
|
returned: 0,
|
|
total
|
|
},
|
|
hasMore: false,
|
|
nextProjectedSeq: options.afterProjectedSeq,
|
|
nextCursor: null,
|
|
traceStatus: metadata.traceStatus ?? "unknown",
|
|
updatedAt: null,
|
|
blocker
|
|
};
|
|
}
|
|
|
|
function blockedTraceEventProjection(projection = {}, blocker) {
|
|
return {
|
|
...projection,
|
|
projectionStatus: "blocked",
|
|
projectionHealth: "degraded",
|
|
blocker,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function traceEventPageMissing(page = {}, projection = {}, options = {}) {
|
|
if (page.blocker) return false;
|
|
const expectedSeq = Number(projection?.lastProjectedSeq);
|
|
if (!Number.isFinite(expectedSeq) || expectedSeq <= 0) return false;
|
|
return factArray(page.events).length === 0 && expectedSeq > Number(options.afterProjectedSeq ?? 0);
|
|
}
|
|
|
|
function traceEventsReadModelBlocker(code, message, { traceId, projection = {}, session = null, turn = null, route = null } = {}) {
|
|
return {
|
|
code,
|
|
message,
|
|
userMessage: message,
|
|
layer: "workbench-read-model",
|
|
category: "trace-events",
|
|
retryable: true,
|
|
route,
|
|
traceId,
|
|
sessionId: factSessionId(session) ?? turn?.sessionId ?? null,
|
|
turnId: factTurnId(turn, traceId),
|
|
projectionStatus: projection.projectionStatus ?? null,
|
|
projectionHealth: "degraded",
|
|
lastProjectedSeq: projection.lastProjectedSeq ?? null,
|
|
sourceRunId: projection.sourceRunId ?? null,
|
|
sourceCommandId: projection.sourceCommandId ?? null,
|
|
retryAfterMs: 5000,
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
function workbenchTraceEventsReadModelError(blocker, { traceId, projection = {}, session = null, turn = null } = {}) {
|
|
const responseProjection = blockedTraceEventProjection(projection, blocker);
|
|
return workbenchError(blocker.code, blocker.message, {
|
|
traceId,
|
|
sessionId: blocker.sessionId ?? factSessionId(session) ?? turn?.sessionId ?? null,
|
|
turnId: blocker.turnId ?? factTurnId(turn, traceId),
|
|
route: blocker.route ?? null,
|
|
layer: blocker.layer,
|
|
category: blocker.category,
|
|
retryable: blocker.retryable,
|
|
projection: responseProjection,
|
|
projectionStatus: responseProjection.projectionStatus,
|
|
projectionHealth: responseProjection.projectionHealth,
|
|
lastProjectedSeq: responseProjection.lastProjectedSeq,
|
|
sourceRunId: responseProjection.sourceRunId,
|
|
sourceCommandId: responseProjection.sourceCommandId,
|
|
blocker,
|
|
diagnostic: {
|
|
contractVersion: "hwlab-error-diagnostic-v1",
|
|
route: blocker.route ?? null,
|
|
layer: blocker.layer,
|
|
category: blocker.category,
|
|
code: blocker.code,
|
|
httpStatus: 404,
|
|
source: "server",
|
|
retryable: blocker.retryable,
|
|
valuesPrinted: false
|
|
}
|
|
});
|
|
}
|
|
|
|
async function visibleSessionById(store, sessionId, actor) {
|
|
const safeId = safeSessionId(sessionId);
|
|
if (!safeId) return null;
|
|
const session = await store?.getAgentSession?.(safeId) ?? null;
|
|
return canActorReadSession(session, actor) ? session : null;
|
|
}
|
|
|
|
async function visibleSessionByRouteId(store, routeId, actor) {
|
|
const sessionId = safeSessionId(routeId);
|
|
if (sessionId) return visibleSessionById(store, sessionId, actor);
|
|
const conversationId = safeConversationId(routeId);
|
|
if (!conversationId) return null;
|
|
const listed = await store?.listAgentSessionsForUser?.({ ownerUserId: actor.id, actorRole: actor.role, ownerScoped: true, conversationId, limit: 1, includeArchived: false }) ?? [];
|
|
const matched = listed.find((session) => canActorReadSession(session, actor)) ?? null;
|
|
if (matched) return matched;
|
|
const suffix = conversationId.slice("cnv_".length);
|
|
const deterministicSessionId = safeSessionId(`ses_${suffix}`);
|
|
return deterministicSessionId ? visibleSessionById(store, deterministicSessionId, actor) : null;
|
|
}
|
|
|
|
async function visibleSessionByTrace(store, traceId, actor) {
|
|
const safeId = safeTraceId(traceId);
|
|
if (!safeId) return null;
|
|
const session = await store?.getAgentSessionByTraceId?.(safeId) ?? null;
|
|
return canActorReadSession(session, actor) ? session : null;
|
|
}
|
|
|
|
async function sessionProjectionOptions(readModel, session, options) {
|
|
const snapshot = objectValue(session?.session);
|
|
const traceId = safeTraceId(session?.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null;
|
|
if (!traceId) return options;
|
|
const trace = await readModel.traceSnapshot(traceId);
|
|
const result = readModel.resultForTrace(traceId);
|
|
const projection = createWorkbenchTurnProjection({ traceId, result, session, trace });
|
|
const projectionDiagnostic = readModel.projectionDiagnostics({ traceId, result, trace, projection });
|
|
return { ...options, trace, result, projection, projectionDiagnostic };
|
|
}
|
|
|
|
function sessionListSummaries(readModel, sessions, options) {
|
|
return sessions
|
|
.map((session) => sessionSummary(session, sessionListProjectionOptions(readModel, session, options)))
|
|
.filter(Boolean);
|
|
}
|
|
|
|
function sessionListProjectionOptions(readModel, session, options) {
|
|
const snapshot = objectValue(session?.session);
|
|
const traceId = safeTraceId(session?.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId ?? snapshot.traceId) ?? null;
|
|
if (!traceId) return options;
|
|
const memoryResult = readModel.resultForTrace(traceId);
|
|
const hasDurableProjection = Boolean(snapshot.projectionStatus || snapshot.terminal !== undefined || snapshot.sealedAt);
|
|
const result = hasDurableProjection ? null : (memoryResult ?? compactSessionTraceResult(snapshot, session, traceId));
|
|
const memoryTrace = hasDurableProjection ? null : traceSnapshotSync(options, traceId);
|
|
const trace = hasDurableProjection ? null : compactSessionTraceSnapshot(snapshot, session, traceId, result, memoryTrace);
|
|
if (hasDurableProjection) {
|
|
const projection = createWorkbenchTurnProjection({
|
|
traceId,
|
|
turnId: traceId,
|
|
result: { status: snapshot.projectionStatus ?? snapshot.status, terminal: snapshot.terminal, timing: snapshot.timing, finalResponse: snapshot.finalResponse, valuesRedacted: true },
|
|
session: null,
|
|
trace: null
|
|
});
|
|
const projectionDiagnostic = readModel.projectionDiagnostics({ traceId, result: null, trace: null, projection });
|
|
return { ...options, trace: null, result: null, projection, projectionDiagnostic };
|
|
}
|
|
const projection = createWorkbenchTurnProjection({ traceId, result, session, trace });
|
|
const projectionDiagnostic = readModel.projectionDiagnostics({ traceId, result, trace, projection });
|
|
return { ...options, trace, result, projection, projectionDiagnostic };
|
|
}
|
|
|
|
function compactSessionTraceResult(snapshot, session, traceId) {
|
|
const record = compactSessionTraceRecord(snapshot, traceId);
|
|
if (record) return traceResultFromCompactRecord(record, traceId);
|
|
const snapshotTraceId = safeTraceId(snapshot.lastTraceId ?? snapshot.currentTraceId ?? snapshot.traceId ?? session?.lastTraceId) ?? null;
|
|
if (snapshotTraceId !== traceId) return null;
|
|
const traceSummary = optionalObject(snapshot.traceSummary);
|
|
const finalResponse = optionalObject(snapshot.finalResponse) ?? messageAuthorityTextValue(snapshot.finalResponse);
|
|
const agentRun = optionalObject(snapshot.agentRun);
|
|
if (!traceSummary && !finalResponse && !agentRun) return null;
|
|
return compactTraceResult({ traceId, status: snapshot.status, finalResponse, traceSummary, agentRun, session });
|
|
}
|
|
|
|
function compactSessionTraceRecord(snapshot, traceId) {
|
|
for (const source of [snapshot.traceResults, snapshot.traceResult]) {
|
|
const container = objectValue(source);
|
|
if (safeTraceId(container.traceId) === traceId) return container;
|
|
const byTraceId = objectValue(container[traceId]);
|
|
if (Object.keys(byTraceId).length > 0) return { ...byTraceId, traceId };
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function traceResultFromCompactRecord(record, traceId) {
|
|
const traceSummary = optionalObject(record.traceSummary);
|
|
const finalResponse = optionalObject(record.finalResponse) ?? messageAuthorityTextValue(record.finalResponse);
|
|
const agentRun = optionalObject(record.agentRun) ?? optionalObject(traceSummary?.agentRun);
|
|
if (!traceSummary && !finalResponse && !agentRun) return null;
|
|
return compactTraceResult({ traceId, status: record.status ?? record.terminalStatus, finalResponse, traceSummary, agentRun, session: record });
|
|
}
|
|
|
|
function compactTraceResult({ traceId, status, finalResponse, traceSummary, agentRun, session }) {
|
|
return {
|
|
traceId,
|
|
status: normalizeStatus(firstTextValue(status, traceSummary?.terminalStatus, agentRun?.terminalStatus, agentRun?.status)),
|
|
ownerUserId: session?.ownerUserId ?? null,
|
|
conversationId: session?.conversationId ?? null,
|
|
sessionId: session?.sessionId ?? session?.id ?? null,
|
|
threadId: session?.threadId ?? null,
|
|
finalResponse: finalResponse || null,
|
|
traceSummary: traceSummary ?? null,
|
|
agentRun: agentRun ?? null,
|
|
updatedAt: session?.updatedAt ?? traceSummary?.updatedAt ?? agentRun?.updatedAt ?? null,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
}
|
|
|
|
function compactSessionTraceSnapshot(snapshot, session, traceId, result, memoryTrace) {
|
|
const traceSummary = optionalObject(result?.traceSummary) ?? optionalObject(snapshot.traceSummary);
|
|
const agentRun = optionalObject(result?.agentRun) ?? optionalObject(traceSummary?.agentRun);
|
|
const finalResponse = optionalObject(result?.finalResponse) ?? optionalObject(snapshot.finalResponse) ?? messageAuthorityTextValue(result?.finalResponse ?? snapshot.finalResponse);
|
|
const status = normalizeStatus(firstTextValue(result?.status, traceSummary?.terminalStatus, agentRun?.terminalStatus, memoryTrace?.status));
|
|
if (!traceSummary && !agentRun && !finalResponse && memoryTrace) return memoryTrace;
|
|
const eventCount = compactTraceEventCount(traceSummary, agentRun, memoryTrace);
|
|
const terminal = TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status);
|
|
const compactTrace = {
|
|
...(memoryTrace ?? {}),
|
|
traceId,
|
|
status,
|
|
eventCount,
|
|
events: Array.isArray(memoryTrace?.events) ? memoryTrace.events : [],
|
|
lastEvent: memoryTrace?.lastEvent ?? null,
|
|
updatedAt: result?.updatedAt ?? traceSummary?.updatedAt ?? agentRun?.updatedAt ?? memoryTrace?.updatedAt ?? session?.updatedAt ?? null,
|
|
finalResponse: finalResponse || null,
|
|
terminalEvidence: terminal && traceSummary ? { status, terminalStatus: traceSummary.terminalStatus ?? status, traceSummary, valuesRedacted: true } : memoryTrace?.terminalEvidence ?? null,
|
|
agentRun: agentRun ?? memoryTrace?.agentRun ?? null,
|
|
valuesPrinted: false
|
|
};
|
|
return compactTrace;
|
|
}
|
|
|
|
function compactTraceEventCount(traceSummary, agentRun, memoryTrace) {
|
|
const memoryEvents = Array.isArray(memoryTrace?.events) ? memoryTrace.events : [];
|
|
const memoryHasProjection = memoryTrace?.lastEvent || memoryEvents.length > 0;
|
|
const candidates = memoryHasProjection
|
|
? [memoryTrace?.eventCount, traceSummary?.eventCount, traceSummary?.sourceEventCount, traceSummary?.lastSeq, agentRun?.lastSeq]
|
|
: [traceSummary?.eventCount, traceSummary?.sourceEventCount, traceSummary?.lastSeq, agentRun?.lastSeq, memoryTrace?.eventCount];
|
|
for (const value of candidates) {
|
|
const count = Number(value);
|
|
if (Number.isFinite(count) && count >= 0) return Math.trunc(count);
|
|
}
|
|
return Array.isArray(memoryTrace?.events) ? memoryTrace.events.length : 0;
|
|
}
|
|
|
|
function firstTextValue(...values) {
|
|
for (const value of values) {
|
|
const text = textValue(value);
|
|
if (text) return text;
|
|
}
|
|
return "";
|
|
}
|
|
|
|
function optionalObject(value) {
|
|
const object = objectValue(value);
|
|
return Object.keys(object).length > 0 ? object : null;
|
|
}
|
|
|
|
function sessionSummary(session, options) {
|
|
if (!session?.id) return null;
|
|
const snapshot = objectValue(session.session);
|
|
const currentTurn = currentWorkbenchTurnProjection(session, options);
|
|
const projectionDiagnostic = options.projectionDiagnostic ?? null;
|
|
const messages = sessionMessages(session, { ...options, currentTurn });
|
|
const status = currentTurn ? currentTurn.status : sessionLifecycleProjectionStatus(session);
|
|
return {
|
|
sessionId: session.id,
|
|
threadId: safeOpaqueId(session.threadId) ?? (textValue(session.threadId) || null),
|
|
agentId: session.agentId ?? "hwlab-code-agent",
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
|
lastTraceId: currentTurn ? currentTurn.traceId : null,
|
|
projection: projectionDiagnostic,
|
|
projectionStatus: projectionDiagnostic?.projectionStatus ?? null,
|
|
projectionHealth: projectionDiagnostic?.projectionHealth ?? null,
|
|
staleMs: projectionDiagnostic?.staleMs ?? null,
|
|
blocker: projectionDiagnostic?.blocker ?? null,
|
|
providerProfile: textValue(snapshot.providerProfile) || null,
|
|
messageCount: messages.length,
|
|
firstUserMessagePreview: firstUserPreview(messages),
|
|
updatedAt: session.updatedAt ?? snapshot.updatedAt ?? null,
|
|
turnSummary: currentTurn ? {
|
|
turnId: currentTurn.turnId,
|
|
traceId: currentTurn.traceId,
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
|
eventCount: currentTurn.eventCount,
|
|
projection: projectionDiagnostic,
|
|
projectionStatus: projectionDiagnostic?.projectionStatus ?? null,
|
|
projectionHealth: projectionDiagnostic?.projectionHealth ?? null,
|
|
staleMs: projectionDiagnostic?.staleMs ?? null,
|
|
blocker: projectionDiagnostic?.blocker ?? null,
|
|
updatedAt: currentTurn.updatedAt
|
|
} : null,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function sessionDetail(session, options) {
|
|
return {
|
|
...sessionSummary(session, options),
|
|
metadata: {
|
|
startedAt: session.startedAt ?? null,
|
|
endedAt: session.endedAt ?? null,
|
|
ownerUserId: session.ownerUserId ?? null
|
|
},
|
|
messagePageUrl: `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages`,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
}
|
|
|
|
function sessionMessages(session, options = {}) {
|
|
const snapshot = objectValue(session.session);
|
|
const raw = Array.isArray(snapshot.messages) ? snapshot.messages : Array.isArray(snapshot.chatMessages) ? snapshot.chatMessages : [];
|
|
const currentTurn = options.currentTurn ? options.currentTurn : currentWorkbenchTurnProjection(session, options);
|
|
const finalTraceId = currentTurn ? currentTurn.traceId : null;
|
|
const terminalProjection = terminalMessageProjection(snapshot, options, currentTurn);
|
|
const messages = raw
|
|
.map((message, index) => messageFact(message, index, session, snapshot))
|
|
.filter(Boolean)
|
|
.map((message) => applyTerminalMessageProjection(message, terminalProjection))
|
|
.map((message) => applyProjectionMessageDiagnostic(message, options.projectionDiagnostic, currentTurn));
|
|
const hasAssistantLikeFinal = messages.some((message) => isAssistantLikeRole(message.role) && (!finalTraceId || message.traceId === finalTraceId));
|
|
const finalText = terminalProjection?.text ?? null;
|
|
if (!hasAssistantLikeFinal && finalText) {
|
|
const finalMessageStatus = terminalProjection ? terminalProjection.status : sessionLifecycleProjectionStatus(session);
|
|
const finalMessageSource = terminalProjection ? terminalProjection.source : "finalResponse";
|
|
messages.push(messageFact({ role: "assistant", text: finalText, status: finalMessageStatus, traceId: finalTraceId, source: finalMessageSource }, messages.length, session, snapshot));
|
|
}
|
|
return messages;
|
|
}
|
|
|
|
function applyProjectionMessageDiagnostic(message, diagnostic, currentTurn) {
|
|
if (!diagnostic || !isAssistantLikeRole(message.role)) return message;
|
|
if (currentTurn?.traceId && message.traceId && message.traceId !== currentTurn.traceId) return message;
|
|
const trace = message.runnerTrace && typeof message.runnerTrace === "object" ? message.runnerTrace : null;
|
|
return {
|
|
...message,
|
|
projection: diagnostic,
|
|
projectionStatus: diagnostic.projectionStatus ?? null,
|
|
projectionHealth: diagnostic.projectionHealth ?? null,
|
|
staleMs: diagnostic.staleMs ?? null,
|
|
blocker: diagnostic.blocker ?? null,
|
|
runnerTrace: trace ? { ...trace, projection: diagnostic, projectionStatus: diagnostic.projectionStatus ?? null, projectionHealth: diagnostic.projectionHealth ?? null, staleMs: diagnostic.staleMs ?? null, blocker: diagnostic.blocker ?? null } : trace
|
|
};
|
|
}
|
|
|
|
function terminalMessageProjection(snapshot, options, currentTurn) {
|
|
if (!currentTurn) return null;
|
|
const traceId = currentTurn.traceId;
|
|
const status = currentTurn.status;
|
|
if (!TERMINAL_STATUSES.has(status) || RUNNING_STATUSES.has(status)) return null;
|
|
const text = projectionText(currentTurn.finalResponse);
|
|
return { traceId, status, text, source: "turn-projection" };
|
|
}
|
|
|
|
function applyTerminalMessageProjection(message, projection) {
|
|
if (!projection || !isAssistantLikeRole(message.role)) return message;
|
|
if (projection.traceId && message.traceId && message.traceId !== projection.traceId) return message;
|
|
const text = projection.text || "";
|
|
const parts = projectTerminalMessageParts(message.parts, message.messageId, projection.traceId ?? message.traceId, projection, text);
|
|
return {
|
|
...message,
|
|
traceId: projection.traceId ?? message.traceId,
|
|
turnId: message.turnId ?? projection.traceId ?? null,
|
|
status: projection.status,
|
|
parts,
|
|
text,
|
|
textPreview: text ? text.slice(0, 240) : message.textPreview ?? null,
|
|
updatedAt: message.updatedAt ?? null
|
|
};
|
|
}
|
|
|
|
function projectTerminalMessageParts(parts, messageId, traceId, projection, text) {
|
|
const sourceParts = Array.isArray(parts) ? parts : [];
|
|
const textIndex = sourceParts.findIndex((part) => part?.type === "final_response" || part?.partType === "final_response");
|
|
if (textIndex >= 0) {
|
|
return sourceParts.map((part, index) => index === textIndex ? { ...part, traceId, type: "final_response", text: text || part.text || null, status: projection.status } : part);
|
|
}
|
|
if (!text) return sourceParts;
|
|
return [partFact({ type: "final_response", text, status: projection.status }, 0, messageId, traceId), ...sourceParts];
|
|
}
|
|
|
|
function projectionText(...values) {
|
|
for (const value of values) {
|
|
if (value && typeof value === "object") {
|
|
const nested = messageAuthorityTextValue(value.text ?? value.content ?? value.message ?? value.summary ?? value.preview ?? value.title);
|
|
if (nested) return nested;
|
|
continue;
|
|
}
|
|
const text = messageAuthorityTextValue(value);
|
|
if (text) return text;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function messageFact(message, index, session, snapshot) {
|
|
const role = textValue(message?.role) || (index % 2 === 0 ? "user" : "assistant");
|
|
const traceId = safeTraceId(message?.traceId ?? message?.turnTraceId ?? session.lastTraceId ?? snapshot.lastTraceId) ?? null;
|
|
const messageId = safeMessageId(message?.messageId ?? message?.id) || `msg_${hash(`${session.id}:${index}:${role}:${traceId ?? "none"}`).slice(0, 24)}`;
|
|
const status = normalizeStatus(message?.status ?? (isAssistantLikeRole(role) ? session.status : "completed"));
|
|
const legacyText = projectionText(message?.text, message?.content, message?.message, message?.finalResponse);
|
|
const rawParts = Array.isArray(message?.parts) ? message.parts : [];
|
|
const parts = rawParts.length > 0
|
|
? rawParts.map((part, partIndex) => partFact(part, partIndex, messageId, traceId)).filter(Boolean)
|
|
: !isAssistantLikeRole(role) && legacyText ? [partFact({ type: "text", text: legacyText }, 0, messageId, traceId)] : [];
|
|
const text = factMessageAuthorityText({ role, status, parts, legacyText });
|
|
return {
|
|
messageId,
|
|
role,
|
|
sessionId: session.id,
|
|
traceId,
|
|
turnId: safeTurnId(message?.turnId) || traceId,
|
|
status,
|
|
parts,
|
|
text: text || "",
|
|
textPreview: text ? text.slice(0, 240) : null,
|
|
createdAt: textValue(message?.createdAt ?? message?.timestamp ?? message?.at) || session.startedAt || session.updatedAt || null,
|
|
updatedAt: textValue(message?.updatedAt) || session.updatedAt || null,
|
|
valuesRedacted: message?.valuesRedacted !== false
|
|
};
|
|
}
|
|
|
|
function partFact(part, index, messageId, traceId) {
|
|
const type = textValue(part?.type) || "text";
|
|
const text = messageAuthorityTextValue(part?.text ?? part?.content ?? part?.message);
|
|
return {
|
|
partId: safePartId(part?.partId ?? part?.id) || `prt_${hash(`${messageId}:${index}:${type}`).slice(0, 24)}`,
|
|
messageId,
|
|
traceId,
|
|
type,
|
|
text: text || null,
|
|
status: normalizeStatus(part?.status ?? "completed"),
|
|
toolName: textValue(part?.toolName ?? part?.name) || null,
|
|
createdAt: textValue(part?.createdAt ?? part?.timestamp) || null,
|
|
valuesRedacted: part?.valuesRedacted !== false
|
|
};
|
|
}
|
|
|
|
function turnSnapshot({ projection, result, session, trace }) {
|
|
const turnId = projection.turnId;
|
|
const traceId = projection.traceId;
|
|
const status = projection.status;
|
|
const messages = session ? sessionMessages(session, { result, trace }) : [];
|
|
const userMessage = messages.find((message) => message.role === "user") ?? null;
|
|
const assistantMessage = [...messages].reverse().find((message) => isAssistantLikeRole(message.role)) ?? null;
|
|
const finalText = projectionText(projection.finalResponse);
|
|
return {
|
|
turnId,
|
|
traceId,
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
|
sessionId: safeSessionId(result?.sessionId ?? result?.session?.sessionId ?? session?.id) ?? null,
|
|
threadId: safeOpaqueId(result?.threadId ?? result?.session?.threadId ?? session?.threadId) ?? (textValue(session?.threadId) || null),
|
|
userMessageId: userMessage?.messageId ?? null,
|
|
assistantMessageId: assistantMessage?.messageId ?? null,
|
|
assistantText: finalText ?? null,
|
|
finalResponse: projection.finalResponse ?? null,
|
|
timing: projection.timing ?? null,
|
|
startedAt: projection.timing?.startedAt ?? null,
|
|
lastEventAt: projection.timing?.lastEventAt ?? null,
|
|
finishedAt: projection.timing?.finishedAt ?? null,
|
|
durationMs: projection.timing?.durationMs ?? null,
|
|
agentRun: result?.agentRun ?? null,
|
|
trace: {
|
|
traceId,
|
|
status: trace?.status ?? "missing",
|
|
eventCount: trace?.eventCount ?? 0,
|
|
updatedAt: trace?.updatedAt ?? null
|
|
},
|
|
urls: {
|
|
self: `/v1/workbench/turns/${encodeURIComponent(turnId)}`,
|
|
traceEvents: `/v1/workbench/traces/${encodeURIComponent(traceId)}/events`
|
|
}
|
|
};
|
|
}
|
|
|
|
function tracePageOptions(url) {
|
|
const cursor = textValue(url.searchParams.get("cursor"));
|
|
const cursorProjectedSeq = cursor.startsWith("projected:") ? Number.parseInt(cursor.slice("projected:".length), 10) : NaN;
|
|
const afterProjectedSeq = Number.isInteger(cursorProjectedSeq) && cursorProjectedSeq >= 0
|
|
? cursorProjectedSeq
|
|
: nonNegativeInteger(url.searchParams.get("afterProjectedSeq"), 0);
|
|
return { afterProjectedSeq, limit: boundedLimit(url.searchParams.get("limit")) };
|
|
}
|
|
|
|
function traceSnapshotSync(options, traceId) {
|
|
return (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(traceId);
|
|
}
|
|
|
|
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 = safeSessionId(context.result?.sessionId ?? context.result?.session?.sessionId ?? context.session?.id) ?? null;
|
|
const threadId = safeOpaqueId(context.result?.threadId ?? context.result?.session?.threadId ?? context.session?.threadId) ?? (textValue(context.session?.threadId) || null);
|
|
writeEvent("workbench.trace.snapshot", {
|
|
type: "trace.snapshot",
|
|
reason,
|
|
sessionId,
|
|
threadId,
|
|
traceId,
|
|
snapshot: { ...traceSnapshotForRealtime(context.trace), sessionId, threadId },
|
|
cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
|
|
});
|
|
writeEvent("workbench.turn.snapshot", {
|
|
type: "turn.snapshot",
|
|
reason,
|
|
sessionId,
|
|
threadId,
|
|
traceId,
|
|
turn: realtimeTurnSnapshot(context),
|
|
cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
|
|
});
|
|
}
|
|
|
|
function realtimeTurnSnapshot(context) {
|
|
if (context?.factsReadBlocker) {
|
|
return {
|
|
...turnSnapshot(context),
|
|
projectionStatus: "blocked",
|
|
projectionHealth: "degraded",
|
|
blocker: context.factsReadBlocker,
|
|
projection: blockedTraceEventProjection(context.projection ?? {}, context.factsReadBlocker),
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
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;
|
|
}
|
|
return turnSnapshot(context);
|
|
}
|
|
|
|
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
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
async function visibleTraceContext(options, actor, traceId) {
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
const result = readModel.resultForTrace(traceId);
|
|
const session = await readModel.getSessionByTraceId(traceId);
|
|
if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) return { visible: false };
|
|
const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
|
|
if (!resultVisible && !session) return { visible: false };
|
|
const trace = await readModel.traceSnapshot(traceId);
|
|
const projection = createWorkbenchTurnProjection({ turnId: traceId, traceId, result, session, trace });
|
|
const factSessionIdValue = safeSessionId(session?.id ?? result?.sessionId ?? result?.session?.sessionId) ?? null;
|
|
let facts = emptyFactSet();
|
|
let factsReadBlocker = null;
|
|
if (factSessionIdValue && typeof readModel.queryFacts === "function") {
|
|
try {
|
|
const factResult = await readModel.queryFacts({ sessionId: factSessionIdValue, families: WORKBENCH_SESSION_DETAIL_FAMILIES, limit: 100 });
|
|
facts = factResult?.facts ?? facts;
|
|
} catch (error) {
|
|
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, session, 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 factSession = factArray(facts.sessions).find((item) => factSessionId(item) === factSessionIdValue) ?? null;
|
|
return { visible: true, turnId: projection.turnId, traceId, status: projection.status, projection, result, session, trace, facts, factSession, factsReadBlocker };
|
|
}
|
|
|
|
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 currentWorkbenchTurnProjection(session, options = {}) {
|
|
const traceId = safeTraceId(session?.lastTraceId) ?? null;
|
|
if (!traceId) return null;
|
|
const trace = options.trace?.traceId === traceId ? options.trace : traceSnapshotSync(options, traceId);
|
|
const result = traceScopedResult(options, traceId);
|
|
return createWorkbenchTurnProjection({ turnId: traceId, traceId, result, session, trace });
|
|
}
|
|
|
|
function traceScopedResult(options, traceId) {
|
|
const result = options.result?.traceId === traceId || (!options.result?.traceId && options.result) ? options.result : options.codeAgentChatResults?.get?.(traceId) ?? null;
|
|
return result && typeof result === "object" ? result : null;
|
|
}
|
|
|
|
function sessionLifecycleProjectionStatus(session) {
|
|
return normalizeStatus(session?.status);
|
|
}
|
|
|
|
function canActorReadSession(session, actor) {
|
|
if (!session || !actor) return false;
|
|
if (normalizeStatus(session.status) === "archived") return false;
|
|
if (actor.role === "admin") return true;
|
|
return session.ownerUserId === actor.id;
|
|
}
|
|
|
|
function canActorReadOwner(ownerUserId, actor) {
|
|
if (!ownerUserId || !actor) return true;
|
|
if (actor.role === "admin") return true;
|
|
return ownerUserId === actor.id;
|
|
}
|
|
|
|
function handleWorkbenchReadModelFailure(response, url, options = {}, error) {
|
|
const classified = classifyWorkbenchReadModelFailure(error);
|
|
const route = workbenchReadRouteTemplate(url?.pathname ?? "/v1/workbench");
|
|
logWorkbenchReadModelFailure(options.logger ?? console, {
|
|
event: "workbench_read_model_route_failed",
|
|
ok: false,
|
|
route,
|
|
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) return;
|
|
sendJson(response, classified.statusCode, workbenchError(classified.code, classified.message, {
|
|
route,
|
|
errorName: error?.name ?? "Error",
|
|
retryable: classified.retryable === true,
|
|
transient: classified.transient === true,
|
|
serviceId: classified.serviceId ?? null,
|
|
runtimeStatus: classified.runtimeStatus ?? null
|
|
}));
|
|
}
|
|
|
|
export function classifyWorkbenchReadModelFailure(error) {
|
|
const message = String(error?.message ?? error ?? "");
|
|
const code = String(error?.code ?? "").toLowerCase();
|
|
const runtimeStatus = Number(error?.data?.status ?? 0);
|
|
const runtimePayloadError = error?.data?.payload?.error ?? null;
|
|
if (error?.name === "WorkbenchRuntimeDependencyError" || code.startsWith("workbench_runtime_")) {
|
|
const retryable = error?.data?.retryable === true || error?.data?.transient === true || runtimePayloadError?.retryable === true;
|
|
return {
|
|
statusCode: retryable || (Number.isFinite(runtimeStatus) && runtimeStatus >= 500) ? 503 : 500,
|
|
code: String(runtimePayloadError?.code ?? error?.code ?? "workbench_runtime_dependency_failed"),
|
|
message: String(runtimePayloadError?.message ?? error?.message ?? "Workbench runtime dependency failed."),
|
|
retryable,
|
|
transient: retryable,
|
|
serviceId: String(error?.data?.serviceId ?? "hwlab-workbench-runtime"),
|
|
runtimeStatus: Number.isFinite(runtimeStatus) && runtimeStatus > 0 ? runtimeStatus : null
|
|
};
|
|
}
|
|
if (/timeout|connection terminated|terminating connection|econnreset|econnrefused|etimedout/iu.test(message) || /timeout|econnreset|econnrefused|etimedout/u.test(code)) {
|
|
return {
|
|
statusCode: 503,
|
|
code: "workbench_read_model_store_unavailable",
|
|
message: "Workbench read model store is temporarily unavailable."
|
|
};
|
|
}
|
|
return {
|
|
statusCode: 500,
|
|
code: "workbench_read_model_failed",
|
|
message: "Workbench read model failed."
|
|
};
|
|
}
|
|
|
|
function workbenchReadRouteTemplate(pathname) {
|
|
const path = String(pathname ?? "");
|
|
if (path === "/v1/workbench/sessions") return path;
|
|
if (/^\/v1\/workbench\/sessions\/[^/]+\/messages$/u.test(path)) return "/v1/workbench/sessions/:id/messages";
|
|
if (/^\/v1\/workbench\/sessions\/[^/]+$/u.test(path)) return "/v1/workbench/sessions/:id";
|
|
if (/^\/v1\/workbench\/turns\/[^/]+$/u.test(path)) return "/v1/workbench/turns/:id";
|
|
if (/^\/v1\/workbench\/traces\/[^/]+\/events$/u.test(path)) return "/v1/workbench/traces/:id/events";
|
|
return path || "/v1/workbench";
|
|
}
|
|
|
|
function logWorkbenchReadModelFailure(logger, payload) {
|
|
try {
|
|
const line = JSON.stringify(payload);
|
|
if (typeof logger?.error === "function") logger.error(line);
|
|
else if (typeof logger?.warn === "function") logger.warn(line);
|
|
} catch {
|
|
// Read-model failure logging must not affect response generation.
|
|
}
|
|
}
|
|
|
|
function methodNotAllowed(response, allowed) {
|
|
sendJson(response, 405, workbenchError("method_not_allowed", `Use ${allowed} for this Workbench read model route.`));
|
|
}
|
|
|
|
function workbenchError(code, message, extra = {}) {
|
|
return {
|
|
ok: false,
|
|
status: "failed",
|
|
error: { code, message, ...extra },
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
};
|
|
}
|
|
|
|
function boundedLimit(value) {
|
|
return Math.min(MAX_PAGE_LIMIT, parsePositiveInteger(value, DEFAULT_PAGE_LIMIT));
|
|
}
|
|
|
|
function boundedSessionListLimit(value) {
|
|
return Math.min(MAX_PAGE_LIMIT, parsePositiveInteger(value, DEFAULT_SESSION_LIST_LIMIT));
|
|
}
|
|
|
|
function cursorOffset(value) {
|
|
const text = textValue(value);
|
|
if (!text) return 0;
|
|
if (text.startsWith("idx:")) return nonNegativeInteger(text.slice(4), 0);
|
|
return nonNegativeInteger(text, 0);
|
|
}
|
|
|
|
function cursorFromOffset(offset) {
|
|
return `idx:${Math.max(0, Math.trunc(Number(offset) || 0))}`;
|
|
}
|
|
|
|
function nonNegativeInteger(value, fallback) {
|
|
const parsed = Number.parseInt(value ?? "", 10);
|
|
return Number.isInteger(parsed) && parsed >= 0 ? parsed : fallback;
|
|
}
|
|
|
|
function eventSeq(event, index) {
|
|
const seq = Number(event?.seq);
|
|
return Number.isFinite(seq) && seq > 0 ? Math.trunc(seq) : index + 1;
|
|
}
|
|
|
|
function normalizeStatus(value) {
|
|
const text = textValue(value).toLowerCase().replace(/_/gu, "-");
|
|
if (text === "cancelled") return "canceled";
|
|
return text || "unknown";
|
|
}
|
|
|
|
function normalizeTerminalStatus(value) {
|
|
const status = normalizeStatus(value);
|
|
return TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status) ? status : null;
|
|
}
|
|
|
|
function firstUserPreview(messages) {
|
|
return messages.find((message) => message.role === "user")?.textPreview ?? null;
|
|
}
|
|
|
|
function isAssistantLikeRole(role) {
|
|
return role === "assistant" || role === "agent";
|
|
}
|
|
|
|
function objectValue(value) {
|
|
return value && typeof value === "object" && !Array.isArray(value) ? value : {};
|
|
}
|
|
|
|
function textValue(value) {
|
|
return String(value ?? "").trim();
|
|
}
|
|
|
|
function messageAuthorityTextValue(value) {
|
|
const text = String(value ?? "").replace(/\r\n?/gu, "\n");
|
|
if (!text.trim() || text.trim() === "[object Object]") return "";
|
|
return text;
|
|
}
|
|
|
|
function safeTurnId(value) {
|
|
const text = textValue(value);
|
|
return /^turn_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
|
|
}
|
|
|
|
function safeMessageId(value) {
|
|
const text = textValue(value);
|
|
return /^msg_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
|
|
}
|
|
|
|
function safePartId(value) {
|
|
const text = textValue(value);
|
|
return /^prt_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
|
|
}
|
|
|
|
function hash(value) {
|
|
return createHash("sha256").update(String(value)).digest("hex");
|
|
}
|