lyon
805 lines
34 KiB
TypeScript
805 lines
34 KiB
TypeScript
/*
|
|
* SPEC: PJ2026-010403 API契约 draft-2026-06-18-r1; PJ2026-010401 Web工作台 draft-2026-06-18-r1; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
|
|
* 职责: 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";
|
|
|
|
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;
|
|
|
|
const projectId = textValue(url.searchParams.get("projectId")) || null;
|
|
const requestedSessionId = safeSessionId(url.searchParams.get("sessionId") ?? url.searchParams.get("includeSessionId"));
|
|
const requestedWorkspaceId = safeWorkspaceId(url.searchParams.get("workspaceId"));
|
|
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;
|
|
let closed = false;
|
|
let latestWorkspaceRevision = 0;
|
|
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", projectId, 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: {
|
|
workspaceId: requestedWorkspaceId,
|
|
sessionId: requestedSessionId,
|
|
traceId: requestedTraceId
|
|
}
|
|
});
|
|
|
|
const initialSession = requestedSessionId ? await visibleSessionById(options.accessController?.store, requestedSessionId, auth.actor) : null;
|
|
let initialWorkspace = null;
|
|
if (requestedWorkspaceId || projectId) {
|
|
initialWorkspace = perf
|
|
? await perf.measure("workbench_initial_workspace", () => visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: requestedWorkspaceId, projectId }))
|
|
: await visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: requestedWorkspaceId, projectId });
|
|
}
|
|
if (initialWorkspace) {
|
|
latestWorkspaceRevision = Number(initialWorkspace.revision ?? 0) || 0;
|
|
writeEvent("workbench.workspace.snapshot", {
|
|
type: "workspace.snapshot",
|
|
reason: "initial",
|
|
workspace: await publicWorkbenchWorkspace(options, initialWorkspace, auth.actor),
|
|
cursor: { workspaceRevision: latestWorkspaceRevision }
|
|
});
|
|
}
|
|
|
|
const initialSessionSnapshot = objectValue(initialSession?.session);
|
|
const activeTraceId = requestedTraceId
|
|
?? safeTraceId(initialSession?.lastTraceId ?? initialSessionSnapshot.lastTraceId ?? initialSessionSnapshot.currentTraceId)
|
|
?? safeTraceId(initialWorkspace?.activeTraceId ?? initialWorkspace?.workspace?.activeTraceId ?? initialWorkspace?.workspace?.lastTraceId);
|
|
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 {
|
|
const heartbeatWorkspaceId = requestedWorkspaceId ?? initialWorkspace?.workspaceId;
|
|
const workspace = heartbeatWorkspaceId || projectId
|
|
? await visibleWorkbenchWorkspace(options, auth.actor, { workspaceId: heartbeatWorkspaceId, projectId })
|
|
: null;
|
|
const revision = Number(workspace?.revision ?? 0) || 0;
|
|
if (workspace && revision > latestWorkspaceRevision) {
|
|
latestWorkspaceRevision = revision;
|
|
writeEvent("workbench.workspace.snapshot", {
|
|
type: "workspace.snapshot",
|
|
reason: "revision-change",
|
|
workspace: await publicWorkbenchWorkspace(options, workspace, auth.actor),
|
|
cursor: { workspaceRevision: revision }
|
|
});
|
|
}
|
|
writeEvent("workbench.heartbeat", {
|
|
type: "heartbeat",
|
|
workspaceId: workspace?.workspaceId ?? requestedWorkspaceId ?? null,
|
|
workspaceRevision: revision || latestWorkspaceRevision,
|
|
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, payload?.workspace?.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) {
|
|
const projectId = textValue(url.searchParams.get("projectId")) || null;
|
|
const limit = boundedLimit(url.searchParams.get("limit"));
|
|
const includeSessionId = safeSessionId(url.searchParams.get("includeSessionId"));
|
|
const store = options.accessController?.store;
|
|
const listInput = {
|
|
ownerUserId: actor.id,
|
|
actorRole: actor.role,
|
|
ownerScoped: true,
|
|
limit,
|
|
includeArchived: false
|
|
};
|
|
if (projectId) listInput.projectId = projectId;
|
|
let sessions = await store?.listAgentSessionsForUser?.(listInput) ?? [];
|
|
if (includeSessionId && !sessions.some((session) => session.id === includeSessionId)) {
|
|
const included = await visibleSessionById(store, includeSessionId, actor);
|
|
if (included) sessions = [included, ...sessions];
|
|
}
|
|
const summaries = sessions.map((session) => sessionSummary(session, options)).filter(Boolean);
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "succeeded",
|
|
contractVersion: "workbench-sessions-v1",
|
|
projectId,
|
|
sessions: summaries,
|
|
conversations: summaries.map(sessionConversationSeed),
|
|
count: summaries.length,
|
|
nextCursor: null,
|
|
valuesRedacted: true,
|
|
secretMaterialStored: false
|
|
});
|
|
}
|
|
|
|
async function handleWorkbenchSessionDetail(response, options, actor, sessionId) {
|
|
const session = await visibleSessionById(options.accessController?.store, sessionId, actor);
|
|
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 session = await visibleSessionById(options.accessController?.store, sessionId, actor);
|
|
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);
|
|
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,
|
|
conversationId: safeConversationId(session.conversationId) ?? null,
|
|
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 result = options.codeAgentChatResults?.get?.(traceId) ?? null;
|
|
const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor);
|
|
if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) {
|
|
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 traceSnapshot(options, traceId);
|
|
const status = turnStatus(result, session, 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 }));
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status,
|
|
contractVersion: "workbench-turn-snapshot-v1",
|
|
turn: turnSnapshot({ turnId, traceId, status, result, session, trace }),
|
|
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 result = options.codeAgentChatResults?.get?.(traceId) ?? null;
|
|
const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor);
|
|
if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) {
|
|
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 traceSnapshot(options, traceId);
|
|
const page = traceEventPage(trace, tracePageOptions(url));
|
|
sendJson(response, 200, {
|
|
ok: true,
|
|
status: "succeeded",
|
|
contractVersion: "workbench-trace-events-v1",
|
|
traceId,
|
|
...page,
|
|
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 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 traceId = safeTraceId(session.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null;
|
|
const trace = traceId ? traceSnapshotSync(options, traceId) : null;
|
|
const messages = sessionMessages(session);
|
|
const status = normalizeStatus(snapshot.sessionStatus ?? session.status ?? trace?.status);
|
|
return {
|
|
sessionId: session.id,
|
|
conversationId: safeConversationId(session.conversationId) ?? null,
|
|
threadId: safeOpaqueId(session.threadId) ?? (textValue(session.threadId) || null),
|
|
projectId: session.projectId ?? null,
|
|
agentId: session.agentId ?? "hwlab-code-agent",
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
|
lastTraceId: traceId,
|
|
providerProfile: textValue(snapshot.providerProfile) || null,
|
|
messageCount: messages.length,
|
|
firstUserMessagePreview: firstUserPreview(messages),
|
|
updatedAt: session.updatedAt ?? snapshot.updatedAt ?? null,
|
|
turnSummary: traceId ? {
|
|
turnId: traceId,
|
|
traceId,
|
|
status,
|
|
running: RUNNING_STATUSES.has(status),
|
|
terminal: TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status),
|
|
eventCount: trace?.eventCount ?? null,
|
|
updatedAt: trace?.updatedAt ?? session.updatedAt ?? null
|
|
} : 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 sessionConversationSeed(summary) {
|
|
return {
|
|
conversationId: summary.conversationId,
|
|
sessionId: summary.sessionId,
|
|
threadId: summary.threadId,
|
|
status: summary.status,
|
|
lastTraceId: summary.lastTraceId,
|
|
messageCount: summary.messageCount,
|
|
firstUserMessagePreview: summary.firstUserMessagePreview,
|
|
updatedAt: summary.updatedAt,
|
|
valuesRedacted: true
|
|
};
|
|
}
|
|
|
|
function sessionMessages(session) {
|
|
const snapshot = objectValue(session.session);
|
|
const raw = Array.isArray(snapshot.messages) ? snapshot.messages : Array.isArray(snapshot.chatMessages) ? snapshot.chatMessages : [];
|
|
const messages = raw.map((message, index) => messageFact(message, index, session, snapshot)).filter(Boolean);
|
|
const finalTraceId = safeTraceId(session.lastTraceId ?? snapshot.lastTraceId ?? snapshot.currentTraceId) ?? null;
|
|
const hasAssistantLikeFinal = messages.some((message) => isAssistantLikeRole(message.role) && (!finalTraceId || message.traceId === finalTraceId));
|
|
if (!hasAssistantLikeFinal && textValue(snapshot.finalResponse)) {
|
|
messages.push(messageFact({ role: "assistant", text: snapshot.finalResponse, traceId: session.lastTraceId ?? snapshot.lastTraceId, source: "finalResponse" }, messages.length, session, snapshot));
|
|
}
|
|
return messages;
|
|
}
|
|
|
|
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 = textValue(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,
|
|
conversationId: safeConversationId(session.conversationId) ?? null,
|
|
traceId,
|
|
turnId: safeTurnId(message?.turnId) || traceId,
|
|
status: normalizeStatus(message?.status ?? (isAssistantLikeRole(role) ? session.status : "completed")),
|
|
parts,
|
|
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) : [];
|
|
const userMessage = messages.find((message) => message.role === "user") ?? null;
|
|
const assistantMessage = [...messages].reverse().find((message) => isAssistantLikeRole(message.role)) ?? null;
|
|
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,
|
|
conversationId: safeConversationId(result?.conversationId ?? session?.conversationId) ?? null,
|
|
threadId: safeOpaqueId(result?.threadId ?? result?.session?.threadId ?? session?.threadId) ?? (textValue(session?.threadId) || null),
|
|
userMessageId: userMessage?.messageId ?? null,
|
|
assistantMessageId: assistantMessage?.messageId ?? 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`,
|
|
compatResult: `/v1/agent/chat/result/${encodeURIComponent(traceId)}`,
|
|
compatTurn: `/v1/agent/turns/${encodeURIComponent(traceId)}`,
|
|
compatTrace: `/v1/agent/traces/${encodeURIComponent(traceId)}`
|
|
}
|
|
};
|
|
}
|
|
|
|
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")) };
|
|
}
|
|
|
|
async function traceSnapshot(options, traceId) {
|
|
const memory = traceSnapshotSync(options, traceId);
|
|
if (memory?.status !== "missing" && (memory.eventCount > 0 || memory.lastEvent)) return memory;
|
|
const durable = await durableTraceSnapshot(options, traceId);
|
|
return durable ?? memory;
|
|
}
|
|
|
|
function traceSnapshotSync(options, traceId) {
|
|
return (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(traceId);
|
|
}
|
|
|
|
async function durableTraceSnapshot(options, traceId) {
|
|
const store = options.runtimeStore;
|
|
if (!store || typeof store.queryAgentTraceEvents !== "function") return null;
|
|
let result;
|
|
try {
|
|
result = await store.queryAgentTraceEvents({ traceId });
|
|
} catch {
|
|
return null;
|
|
}
|
|
const events = Array.isArray(result?.events) ? result.events : [];
|
|
if (events.length === 0) return null;
|
|
const normalizedEvents = events.map((event, index) => ({ ...event, traceId, seq: eventSeq(event, index) }));
|
|
const firstEvent = normalizedEvents[0] ?? null;
|
|
const lastEvent = normalizedEvents.at(-1) ?? null;
|
|
const status = durableTraceStatus(normalizedEvents);
|
|
return {
|
|
traceId,
|
|
status,
|
|
createdAt: firstEvent?.createdAt ?? null,
|
|
updatedAt: lastEvent?.createdAt ?? firstEvent?.createdAt ?? null,
|
|
startedAt: firstEvent?.createdAt ?? null,
|
|
finishedAt: TERMINAL_STATUSES.has(status) ? lastEvent?.createdAt ?? null : null,
|
|
eventCount: normalizedEvents.length,
|
|
events: normalizedEvents,
|
|
assistantStreams: [],
|
|
eventLabels: normalizedEvents.map((event) => event.label).filter(Boolean),
|
|
lastEvent,
|
|
outputTruncated: normalizedEvents.some((event) => event.outputTruncated === true),
|
|
valuesPrinted: false
|
|
};
|
|
}
|
|
|
|
function durableTraceStatus(events) {
|
|
const lastEvent = events.at(-1) ?? null;
|
|
const normalized = normalizeStatus(lastEvent?.status ?? lastEvent?.type);
|
|
if (lastEvent?.terminal === true) {
|
|
if (normalized === "observed" || normalized === "event" || normalized === "unknown") return "completed";
|
|
return normalized;
|
|
}
|
|
if (TERMINAL_STATUSES.has(normalized) || RUNNING_STATUSES.has(normalized)) return normalized;
|
|
if (["failed", "error", "timeout"].includes(normalized)) return normalized;
|
|
return "running";
|
|
}
|
|
|
|
async function visibleWorkbenchWorkspace(options, actor, input = {}) {
|
|
const store = options.accessController?.store;
|
|
if (!store) return null;
|
|
const workspaceId = safeWorkspaceId(input.workspaceId);
|
|
if (workspaceId) {
|
|
return await store.getWorkspaceForUser?.({ workspaceId, ownerUserId: actor.id, actorRole: actor.role }) ?? null;
|
|
}
|
|
const projectId = textValue(input.projectId);
|
|
if (!projectId) return null;
|
|
return await store.getOrCreateDefaultWorkspace?.({
|
|
ownerUserId: actor.id,
|
|
projectId,
|
|
now: new Date().toISOString()
|
|
}) ?? null;
|
|
}
|
|
|
|
async function publicWorkbenchWorkspace(options, workspace, actor) {
|
|
if (!workspace) return null;
|
|
if (typeof options.accessController?.publicWorkbenchWorkspace === "function") {
|
|
return await options.accessController.publicWorkbenchWorkspace(workspace, actor);
|
|
}
|
|
return workspace;
|
|
}
|
|
|
|
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 result = options.codeAgentChatResults?.get?.(traceId) ?? null;
|
|
const session = await visibleSessionByTrace(options.accessController?.store, traceId, actor);
|
|
if (result?.ownerUserId && !canActorReadOwner(result.ownerUserId, actor)) return { visible: false };
|
|
const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId));
|
|
if (!resultVisible && !session) return { visible: false };
|
|
const trace = await traceSnapshot(options, traceId);
|
|
const status = turnStatus(result, session, 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 turnStatus(result, session, trace) {
|
|
return normalizeStatus(
|
|
result?.status ??
|
|
result?.agentRun?.terminalStatus ??
|
|
result?.agentRun?.commandState ??
|
|
result?.agentRun?.status ??
|
|
session?.status ??
|
|
objectValue(session?.session).sessionStatus ??
|
|
trace?.status ??
|
|
"unknown"
|
|
);
|
|
}
|
|
|
|
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 safeWorkspaceId(value) {
|
|
const text = textValue(value);
|
|
return /^wsp_[A-Za-z0-9_.:-]+$/u.test(text) ? text : null;
|
|
}
|
|
|
|
function hash(value) {
|
|
return createHash("sha256").update(String(value)).digest("hex");
|
|
}
|