lyon
769 lines
33 KiB
TypeScript
769 lines
33 KiB
TypeScript
/*
|
|
* SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
|
|
* 职责: 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";
|
|
|
|
const DEFAULT_PAGE_LIMIT = 50;
|
|
const MAX_PAGE_LIMIT = 100;
|
|
const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000;
|
|
const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]);
|
|
const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active"]);
|
|
|
|
export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) {
|
|
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(response, url, options, auth.actor)) : handleWorkbenchSessionList(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(response, url, options, auth.actor, decodeURIComponent(sessionMessagesMatch[1]))) : handleWorkbenchMessagePage(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(response, options, auth.actor, decodeURIComponent(sessionMatch[1]))) : handleWorkbenchSessionDetail(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(response, url, options, auth.actor, decodeURIComponent(turnMatch[1]))) : handleWorkbenchTurnSnapshot(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(response, url, options, auth.actor, decodeURIComponent(traceEventsMatch[1]))) : handleWorkbenchTraceEventPage(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 }));
|
|
}
|
|
|
|
export async function handleWorkbenchRealtimeHttp(request, response, url, options = {}) {
|
|
if (request.method !== "GET") return methodNotAllowed(response, "GET");
|
|
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 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 traceStore = options.traceStore ?? defaultCodeAgentTraceStore;
|
|
const readModel = createWorkbenchReadModel(options, auth.actor);
|
|
let closed = false;
|
|
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"
|
|
});
|
|
|
|
const writeEvent = (name, payload = {}) => {
|
|
if (closed || response.destroyed) return;
|
|
const startedAt = nowMs();
|
|
const eventCreatedAt = realtimeEventCreatedAt(payload);
|
|
const traceSeq = realtimeTraceSeq(payload);
|
|
response.write(`event: ${name}\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" });
|
|
};
|
|
|
|
writeEvent("workbench.connected", {
|
|
type: "connected",
|
|
status: "connected",
|
|
heartbeatMs,
|
|
filters: {
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId
|
|
}
|
|
});
|
|
|
|
const initialSession = requestedSessionId ? await readModel.getSessionById(requestedSessionId) : null;
|
|
const initialSessionSnapshot = objectValue(initialSession?.session);
|
|
const activeTraceId = requestedTraceId
|
|
?? safeTraceId(initialSession?.lastTraceId ?? initialSessionSnapshot.lastTraceId ?? initialSessionSnapshot.currentTraceId)
|
|
?? null;
|
|
if (activeTraceId) {
|
|
await (perf ? perf.measure("workbench_initial_trace", () => writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" })) : writeTraceRealtimeSnapshot({ writeEvent, options, actor: auth.actor, traceId: activeTraceId, reason: "initial" }));
|
|
const unsubscribe = traceStore.subscribe(activeTraceId, (event, snapshot) => {
|
|
if (event) {
|
|
writeEvent("workbench.trace.event", {
|
|
type: "trace.event",
|
|
traceId: activeTraceId,
|
|
event,
|
|
snapshot: traceSnapshotSummary(snapshot),
|
|
cursor: { traceSeq: eventSeq(event, Number(snapshot?.eventCount ?? 1) - 1) }
|
|
});
|
|
} else {
|
|
writeEvent("workbench.trace.snapshot", {
|
|
type: "trace.snapshot",
|
|
traceId: activeTraceId,
|
|
reason: "trace-store-update",
|
|
snapshot: traceSnapshotForRealtime(snapshot),
|
|
cursor: { traceSeq: traceSnapshotLastSeq(snapshot) }
|
|
});
|
|
}
|
|
if (event?.terminal === true || TERMINAL_STATUSES.has(normalizeStatus(event?.status))) {
|
|
void writeTraceRealtimeSnapshot({ 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", () => {
|
|
closed = true;
|
|
for (const item of cleanup.splice(0)) item();
|
|
});
|
|
}
|
|
|
|
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 nowMs() {
|
|
if (typeof performance !== "undefined" && typeof performance.now === "function") return performance.now();
|
|
return Date.now();
|
|
}
|
|
|
|
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(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 limit = boundedLimit(url.searchParams.get("limit"));
|
|
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
|
|
const includeRouteId = includeSessionId ?? safeConversationId(url.searchParams.get("includeSessionId"));
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
const sessions = await readModel.listSessions({ limit, includeRouteId });
|
|
const summaries = sessions.map((session) => sessionSummary(session, options)).filter(Boolean);
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "succeeded",
|
|
contractVersion: "workbench-sessions-v1",
|
|
sessions: summaries,
|
|
count: summaries.length,
|
|
nextCursor: null,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
});
|
|
}
|
|
|
|
async function handleWorkbenchSessionDetail(response, options, actor, sessionId) {
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
const session = await readModel.getSessionByRouteId(sessionId);
|
|
if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId }));
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "found",
|
|
contractVersion: "workbench-session-detail-v1",
|
|
session: sessionDetail(session, options),
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
});
|
|
}
|
|
|
|
async function handleWorkbenchMessagePage(response, url, options, actor, sessionId) {
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
const session = await readModel.getSessionByRouteId(sessionId);
|
|
if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId }));
|
|
const messages = sessionMessages(session, options);
|
|
const limit = boundedLimit(url.searchParams.get("limit"));
|
|
const offset = cursorOffset(url.searchParams.get("cursor") ?? url.searchParams.get("after"));
|
|
const page = messages.slice(offset, offset + limit);
|
|
const nextOffset = offset + page.length;
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "succeeded",
|
|
contractVersion: "workbench-message-page-v1",
|
|
sessionId: session.id,
|
|
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(response, url, options, actor, rawTurnId) {
|
|
const queryTraceId = safeTraceId(url.searchParams.get("traceId"));
|
|
const traceId = safeTraceId(rawTurnId) ?? queryTraceId;
|
|
const turnId = safeTurnId(rawTurnId) || traceId || rawTurnId;
|
|
if (!traceId) return sendJson(response, 400, workbenchError("invalid_turn_id", "turnId must currently be a trace id or include traceId query.", { turnId }));
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
const result = readModel.resultForTrace(traceId);
|
|
const session = await readModel.getSessionByTraceId(traceId);
|
|
if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) {
|
|
return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench turn.", { traceId }));
|
|
}
|
|
const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
|
|
const trace = await readModel.traceSnapshot(traceId);
|
|
const status = workbenchTurnProjectionStatus(trace);
|
|
const found = Boolean(resultVisible || session);
|
|
if (!found) return sendJson(response, 404, workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId }));
|
|
const projection = readModel.projectionDiagnostics({ traceId, result, trace });
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status,
|
|
contractVersion: "workbench-turn-snapshot-v1",
|
|
turn: turnSnapshot({ turnId, traceId, status, result, session, trace }),
|
|
projection,
|
|
projectionStatus: projection.projectionStatus,
|
|
lastProjectedSeq: projection.lastProjectedSeq,
|
|
sourceRunId: projection.sourceRunId,
|
|
sourceCommandId: projection.sourceCommandId,
|
|
blocker: projection.blocker,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
});
|
|
}
|
|
|
|
async function handleWorkbenchTraceEventPage(response, url, options, actor, rawTraceId) {
|
|
const traceId = safeTraceId(rawTraceId);
|
|
if (!traceId) return sendJson(response, 400, workbenchError("invalid_trace_id", "traceId must start with trc_.", { traceId: rawTraceId }));
|
|
const readModel = createWorkbenchReadModel(options, actor);
|
|
const result = readModel.resultForTrace(traceId);
|
|
const session = await readModel.getSessionByTraceId(traceId);
|
|
if (result?.ownerUserId && !readModel.canReadOwner(result.ownerUserId)) {
|
|
return sendJson(response, 403, workbenchError("agent_session_owner_required", "Only the session owner or admin can read this Workbench trace.", { traceId }));
|
|
}
|
|
const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
|
|
if (!resultVisible && !session) {
|
|
return sendJson(response, 404, workbenchError("workbench_trace_not_found", "Workbench trace is not visible to the current actor.", { traceId }));
|
|
}
|
|
const trace = await readModel.traceSnapshot(traceId);
|
|
const page = traceEventPage(trace, tracePageOptions(url));
|
|
const projection = readModel.projectionDiagnostics({ traceId, result, trace });
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "succeeded",
|
|
contractVersion: "workbench-trace-events-v1",
|
|
traceId,
|
|
...page,
|
|
projection,
|
|
projectionStatus: projection.projectionStatus,
|
|
lastProjectedSeq: projection.lastProjectedSeq,
|
|
sourceRunId: projection.sourceRunId,
|
|
sourceCommandId: projection.sourceCommandId,
|
|
blocker: projection.blocker,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: 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;
|
|
}
|
|
|
|
function sessionSummary(session, options) {
|
|
if (!session?.id) return null;
|
|
const snapshot = objectValue(session.session);
|
|
const currentTurn = currentWorkbenchTurnProjection(session, options);
|
|
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,
|
|
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,
|
|
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));
|
|
const hasAssistantLikeFinal = messages.some((message) => isAssistantLikeRole(message.role) && (!finalTraceId || message.traceId === finalTraceId));
|
|
const finalText = terminalProjection?.text ?? projectionText(snapshot.finalResponse);
|
|
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 terminalMessageProjection(snapshot, options, currentTurn) {
|
|
if (!currentTurn) return null;
|
|
const traceId = currentTurn.traceId;
|
|
const result = options.result ?? options.codeAgentChatResults?.get?.(traceId) ?? null;
|
|
const trace = currentTurn.trace;
|
|
const status = currentTurn.status;
|
|
if (!TERMINAL_STATUSES.has(status) || RUNNING_STATUSES.has(status)) return null;
|
|
const text = projectionText(result?.finalResponse, result?.assistantText, result?.reply, result?.text, result?.summary, trace?.finalResponse, trace?.terminalEvidence?.finalResponse, snapshot.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 || message.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 === "text" || part?.text);
|
|
if (textIndex >= 0) {
|
|
return sourceParts.map((part, index) => index === textIndex ? { ...part, traceId, text: text || part.text || null, status: projection.status } : part);
|
|
}
|
|
if (!text) return sourceParts;
|
|
return [partFact({ type: "text", text, status: projection.status }, 0, messageId, traceId), ...sourceParts];
|
|
}
|
|
|
|
function projectionText(...values) {
|
|
for (const value of values) {
|
|
if (value && typeof value === "object") {
|
|
const nested = textValue(value.text ?? value.content ?? value.message ?? value.summary ?? value.preview ?? value.title);
|
|
if (nested) return nested;
|
|
continue;
|
|
}
|
|
const text = textValue(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 text = 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)
|
|
: text ? [partFact({ type: "text", text }, 0, messageId, traceId)] : [];
|
|
return {
|
|
messageId,
|
|
role,
|
|
sessionId: session.id,
|
|
traceId,
|
|
turnId: safeTurnId(message?.turnId) || traceId,
|
|
status: normalizeStatus(message?.status ?? (isAssistantLikeRole(role) ? session.status : "completed")),
|
|
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 = textValue(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({ turnId, traceId, status, result, session, trace }) {
|
|
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(result?.finalResponse, result?.assistantText, result?.reply, result?.text, trace?.finalResponse, trace?.terminalEvidence?.finalResponse, assistantMessage?.text);
|
|
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: finalText ? { text: finalText, status, traceId, valuesPrinted: false } : 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 traceEventPage(snapshot, options) {
|
|
const sourceEvents = Array.isArray(snapshot?.events) ? snapshot.events : [];
|
|
const indexed = sourceEvents.map((event, index) => ({ event, seq: eventSeq(event, index) }));
|
|
const startIndex = indexed.findIndex((item) => item.seq > options.afterSeq);
|
|
const offset = startIndex >= 0 ? startIndex : indexed.length;
|
|
const page = indexed.slice(offset, offset + options.limit);
|
|
const events = page.map((item) => ({ ...item.event, seq: item.seq }));
|
|
const toSeq = page.length ? page[page.length - 1].seq : options.afterSeq;
|
|
const hasMore = offset + page.length < indexed.length;
|
|
return {
|
|
events,
|
|
eventCount: indexed.length,
|
|
range: {
|
|
afterSeq: options.afterSeq,
|
|
fromSeq: page.length ? page[0].seq : null,
|
|
toSeq,
|
|
limit: options.limit,
|
|
returned: events.length,
|
|
total: indexed.length
|
|
},
|
|
hasMore,
|
|
nextSeq: toSeq,
|
|
nextCursor: hasMore ? `seq:${toSeq}` : null,
|
|
traceStatus: snapshot?.status ?? "missing",
|
|
updatedAt: snapshot?.updatedAt ?? null
|
|
};
|
|
}
|
|
|
|
function tracePageOptions(url) {
|
|
const cursor = textValue(url.searchParams.get("cursor"));
|
|
const cursorSeq = cursor.startsWith("seq:") ? Number.parseInt(cursor.slice(4), 10) : NaN;
|
|
const afterSeq = Number.isInteger(cursorSeq)
|
|
? cursorSeq
|
|
: nonNegativeInteger(url.searchParams.get("afterSeq") ?? url.searchParams.get("sinceSeq") ?? url.searchParams.get("since"), 0);
|
|
return { afterSeq, 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;
|
|
}
|
|
writeEvent("workbench.trace.snapshot", {
|
|
type: "trace.snapshot",
|
|
reason,
|
|
traceId,
|
|
snapshot: traceSnapshotForRealtime(context.trace),
|
|
cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
|
|
});
|
|
writeEvent("workbench.turn.snapshot", {
|
|
type: "turn.snapshot",
|
|
reason,
|
|
traceId,
|
|
turn: turnSnapshot(context),
|
|
cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
|
|
});
|
|
}
|
|
|
|
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 status = workbenchTurnProjectionStatus(trace);
|
|
return { visible: true, turnId: traceId, traceId, status, result, session, trace };
|
|
}
|
|
|
|
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 status = workbenchTurnProjectionStatus(trace);
|
|
return {
|
|
turnId: traceId,
|
|
traceId,
|
|
status,
|
|
trace,
|
|
eventCount: Number.isFinite(Number(trace?.eventCount)) ? Number(trace.eventCount) : null,
|
|
updatedAt: trace?.updatedAt ?? session?.updatedAt ?? null
|
|
};
|
|
}
|
|
|
|
function workbenchTurnProjectionStatus(trace) {
|
|
const status = normalizeStatus(trace?.status);
|
|
if (status === "missing") return "unknown";
|
|
if (RUNNING_STATUSES.has(status) || TERMINAL_STATUSES.has(status)) return status;
|
|
return "unknown";
|
|
}
|
|
|
|
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 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 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 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 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");
|
|
}
|