diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 739dff5d..fd9a2a2a 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1678,7 +1678,7 @@ test("workbench realtime stream surfaces facts blocker instead of legacy trace f try { const { port } = server.address(); - const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}`, 4); + const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}&traceId=${encodeURIComponent(traceId)}`, 4); const events = await eventsPromise; assert.deepEqual(events.map((event) => event.event), [ "workbench.connected", @@ -1687,7 +1687,7 @@ test("workbench realtime stream surfaces facts blocker instead of legacy trace f "workbench.turn.snapshot" ]); assert.equal(events[0].data.filters.sessionId, session.id); - assert.equal(events[1].data.error.code, "workbench_facts_session_missing"); + assert.equal(events[1].data.error.code, "workbench_facts_trace_missing"); assert.equal(events[2].data.traceId, traceId); assert.equal(events[2].data.snapshot.status, "unknown"); assert.equal(events[2].data.snapshot.eventCount, 0); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 6f79eaf5..7b25f59c 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -15,7 +15,7 @@ import { } 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 { durableTraceStatus, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; import { emitHttpServerRequestSpan } from "./otel-trace.ts"; const DEFAULT_PAGE_LIMIT = 50; @@ -26,7 +26,7 @@ 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_METADATA_FAMILIES = Object.freeze(["sessions", "messages", "parts", "turns", "checkpoints"]); const WORKBENCH_TRACE_EVENT_PAGE_FAMILIES = Object.freeze(["traceEvents"]); const activeWorkbenchRealtimeConnections = new Set(); @@ -203,12 +203,25 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option } }); - 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 resolveInitialSession = requestedSessionId && requestedAfterSeq <= 0; + const initialContext = resolveInitialSession + ? await realtimeInitialSessionContext(readModel, auth.actor, requestedSessionId) + : { facts: emptyFactSet(), session: null, blocker: null }; + if (initialContext.blocker) { + writeEvent("workbench.error", { + type: "error", + reason: "initial-session", + sessionId: requestedSessionId, + traceId: requestedTraceId, + error: initialContext.blocker + }); + } + const initialFactSession = initialContext.session; + const streamSessionId = factSessionId(initialFactSession) ?? requestedSessionId ?? null; + const streamThreadId = safeOpaqueId(initialFactSession?.threadId) ?? (textValue(initialFactSession?.threadId) || null); const activeTraceId = requestedTraceId - ?? safeTraceId(initialSession?.lastTraceId ?? initialSessionSnapshot.lastTraceId ?? initialSessionSnapshot.currentTraceId) + ?? factLastTraceId(initialFactSession) + ?? factLatestTraceIdForSession(initialContext.facts, streamSessionId) ?? null; realtimeSessionId = streamSessionId ?? requestedSessionId ?? null; realtimeTraceId = activeTraceId ?? requestedTraceId ?? null; @@ -240,7 +253,7 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option outboxCursor = row.outboxSeq; const rowTraceId = safeTraceId(row.traceId) ?? activeTraceId; const rowSessionId = safeSessionId(row.sessionId ?? streamSessionId) ?? streamSessionId; - if (rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) { + if (requestedAfterSeq <= 0 && rowTraceId && !outboxTraceSnapshots.has(rowTraceId)) { outboxTraceSnapshots.add(rowTraceId); await writeTraceRealtimeSnapshotSafe({ writeEvent, options, actor: auth.actor, traceId: rowTraceId, reason: "outbox-session" }); } @@ -701,9 +714,15 @@ async function handleWorkbenchSessionList(request, response, url, options, actor 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 }); + const runtime = workbenchRuntimeReader(request, options); if (typeof runtime?.listWorkbenchSessions !== "function") { - throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for session list.", false); + if (typeof runtime?.queryWorkbenchFacts !== "function") { + throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for session list.", false); + } + const readModel = workbenchRuntimeFactsReadModel(request, options, actor); + const payload = await handleWorkbenchSessionListFromFacts(readModel, url, options, actor, { limit, offset, includeRouteId }); + recordWorkbenchCacheMetric(options, url, payload, startedAt); + return sendJson(response, 200, payload); } const payload = await runtime.listWorkbenchSessions({ limit, @@ -716,48 +735,17 @@ async function handleWorkbenchSessionList(request, response, url, options, actor 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.")); +async function handleWorkbenchSessionListFromFacts(readModel, url, options, actor, { limit, offset, includeRouteId }) { 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)); + if (pageResult.error) throw 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)); + if (includeResult.error) throw includeResult.error; const included = visibleFactSessions(includeResult.facts, actor)[0] ?? null; if (included) { pageFacts = mergeFactSets(pageFacts, includeResult.facts); @@ -766,11 +754,11 @@ async function handleWorkbenchSessionListBunLegacy(response, url, options, actor } 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)); + if (summaryResult.error) throw 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, { + return { ok: true, status: "succeeded", contractVersion: "workbench-sessions-v1", @@ -779,9 +767,60 @@ async function handleWorkbenchSessionListBunLegacy(response, url, options, actor cursor: offset > 0 ? cursorFromOffset(offset) : null, hasMore, nextCursor: hasMore ? cursorFromOffset(offset + limit) : null, + cache: pageResult.cache ?? summaryResult.cache ?? null, valuesRedacted: true, secretMaterialStored: false - }); + }; +} + +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 workbenchRuntimeReader(request, options) { + return options.workbenchRuntime + ?? options.runtimeStore + ?? createWorkbenchRuntimeClient({ env: options.env ?? process.env, fetch: options.fetch, logger: options.logger, traceparent: request?.hwlabHttpRequestContext?.traceparent }); +} + +function workbenchRuntimeFactsReadModel(request, options, actor) { + const runtime = workbenchRuntimeReader(request, options); + if (typeof runtime?.queryWorkbenchFacts !== "function") { + throw runtimeDependencyError("workbench_runtime_unconfigured", "Workbench runtime service is required for Workbench facts reads.", false); + } + return { + async queryFacts(query = {}) { + try { + return await runtime.queryWorkbenchFacts({ + ...query, + actor: { id: actor.id, role: actor.role ?? "user", valuesRedacted: true } + }); + } catch (error) { + return { facts: emptyFactSet(), count: 0, durable: true, error: workbenchRuntimeFactsQueryError(error), persistence: null }; + } + } + }; +} + +function workbenchRuntimeFactsQueryError(error) { + const normalized = error instanceof Error ? error : new Error(String(error ?? "Workbench facts query failed.")); + const data = objectValue(normalized.data); + normalized.data = { + ...data, + retryable: data.retryable !== false, + transient: data.transient !== false, + retryAfterMs: data.retryAfterMs ?? 5000, + valuesRedacted: true + }; + return normalized; +} + +function workbenchRuntimeOutboxReader(request, options) { + return workbenchRuntimeReader(request, options); } function queryWorkbenchSessionInclude(readModel, includeRouteId, perf = null) { @@ -790,13 +829,6 @@ function queryWorkbenchSessionInclude(readModel, includeRouteId, perf = null) { 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); @@ -828,7 +860,7 @@ async function queryFactsForSessionSummaries(readModel, sessions = []) { // 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"] }) + ? await readModel.queryFacts({ sessionIds, families: ["messages", "parts", "turns"] }) : { facts: emptyFactSet(), count: 0, durable: false, persistence: null }; if (bySession.error) return bySession; const byTrace = traceIds.length > 0 @@ -1593,7 +1625,7 @@ async function handleWorkbenchMessagePage(request, response, url, options, actor 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" }) + ...(hasCursor ? {} : { limit: limit + 1, 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; @@ -1603,7 +1635,7 @@ async function handleWorkbenchMessagePage(request, response, url, options, actor return sendJson(response, 404, body); } const messages = factMessagesForSession(session, result.facts); - const offset = hasCursor ? cursorOffset(cursorParam) : Math.max(0, messages.length - limit); + const offset = hasCursor ? cursorOffset(cursorParam) : 0; const page = messages.slice(offset, offset + limit); const nextOffset = offset + page.length; const resolvedSessionId = factSessionId(session); @@ -1634,7 +1666,7 @@ async function handleWorkbenchTurnSnapshot(request, response, url, options, acto 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 }); + const result = await readModel.queryFacts({ traceId, limit: MAX_PAGE_LIMIT, families: [...WORKBENCH_TRACE_EVENT_METADATA_FAMILIES, ...WORKBENCH_TRACE_EVENT_PAGE_FAMILIES] }); if (result.error) { const body = workbenchProjectionStoreError(result.error); recordWorkbenchTurnReadMetric(options, url, 503, body, startedAt); @@ -1715,7 +1747,8 @@ async function handleWorkbenchTraceEventPage(request, response, url, options, ac 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; + attachWorkbenchReadModelDiagnosticOtel(request, { route: "/v1/workbench/traces/:id/events", code: "workbench_trace_events_missing", sessionId: factSessionId(session), traceId, count: pageResult.count }); + return sendJson(response, 404, workbenchTraceEventsReadModelError(blocker, { traceId, projection, session, turn: traceTurn })); } const responseProjection = page.blocker ? blockedTraceEventProjection(projection, page.blocker) : projection; const body = { @@ -1891,300 +1924,6 @@ function workbenchTraceEventsReadModelError(blocker, { traceId, projection = {}, }); } -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") { @@ -2198,33 +1937,6 @@ function projectionText(...values) { 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); @@ -2241,45 +1953,6 @@ function partFact(part, index, messageId, traceId) { }; } -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; @@ -2289,8 +1962,31 @@ function tracePageOptions(url) { return { afterProjectedSeq, limit: boundedLimit(url.searchParams.get("limit")) }; } -function traceSnapshotSync(options, traceId) { - return (options.traceStore ?? defaultCodeAgentTraceStore).snapshot(traceId); +async function realtimeInitialSessionContext(readModel, actor, sessionId) { + const safeId = safeSessionId(sessionId); + if (!safeId || typeof readModel?.queryFacts !== "function") { + const blocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench realtime initial session must be read from durable facts; the facts query interface is unavailable.", { session: { sessionId: safeId }, route: "/v1/workbench/events" }); + return { facts: emptyFactSet(), session: null, blocker }; + } + try { + const result = await readModel.queryFacts({ sessionId: safeId, families: WORKBENCH_SESSION_DETAIL_FAMILIES, limit: MAX_PAGE_LIMIT }); + if (result.error || result.durable === false) { + const blocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench realtime initial session facts query failed; realtime must surface the read-model error instead of falling back to legacy session snapshots.", { session: { sessionId: safeId }, route: "/v1/workbench/events" }); + blocker.causeCode = result.error?.code ?? null; + blocker.causeName = result.error?.name ?? null; + blocker.message = result.error?.message ? `${blocker.message} Cause: ${result.error.message}` : result.durable === false ? `${blocker.message} Cause: durable facts reader unavailable.` : blocker.message; + return { facts: emptyFactSet(), session: null, blocker }; + } + const facts = result.facts ?? emptyFactSet(); + const session = visibleFactSessions(facts, actor).find((item) => factSessionId(item) === safeId) ?? null; + return { facts, session, blocker: null }; + } catch (error) { + const blocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench realtime initial session facts query failed; realtime must surface the read-model error instead of falling back to legacy session snapshots.", { session: { sessionId: safeId }, route: "/v1/workbench/events" }); + blocker.causeCode = error?.code ?? null; + blocker.causeName = error?.name ?? null; + blocker.message = error?.message ? `${blocker.message} Cause: ${error.message}` : blocker.message; + return { facts: emptyFactSet(), session: null, blocker }; + } } async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, reason }) { @@ -2312,8 +2008,8 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, 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); + const sessionId = factSessionId(context.factSession) ?? safeSessionId(context.session?.id) ?? null; + const threadId = safeOpaqueId(context.factSession?.threadId ?? context.session?.threadId) ?? (textValue(context.factSession?.threadId ?? context.session?.threadId) || null); const realtimeTraceSnapshot = context.factsReadBlocker ? blockedRealtimeTraceSnapshot(context) : traceSnapshotForRealtime(context.trace); const traceSeq = context.factsReadBlocker ? 0 : traceSnapshotLastSeq(context.trace); writeEvent("workbench.trace.snapshot", { @@ -2438,38 +2134,59 @@ async function writeTraceRealtimeSnapshotSafe(input) { 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; + if (typeof readModel.queryFacts !== "function") { + const projection = factProjectionForTrace(emptyFactSet(), traceId); + const factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench durable facts query is unavailable; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" }); + return { visible: true, turnId: traceId, traceId, status: "unknown", projection, result: null, session: null, trace: factTraceSnapshot(emptyFactSet(), traceId), facts: emptyFactSet(), factSession: null, factsReadBlocker }; + } + let facts = emptyFactSet(); let factsReadBlocker = null; - if (factSessionIdValue && typeof readModel.queryFacts === "function") { - try { - const factResult = await readModel.queryFacts({ sessionId: factSessionIdValue, families: WORKBENCH_SESSION_DETAIL_FAMILIES, limit: 100 }); + try { + const factResult = await readModel.queryFacts({ traceId, families: [...WORKBENCH_SESSION_DETAIL_FAMILIES, ...WORKBENCH_TRACE_EVENT_PAGE_FAMILIES], limit: MAX_PAGE_LIMIT }); + if (factResult.error || factResult.durable === false) { + const projection = factProjectionForTrace(facts, traceId); + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts query failed; realtime must surface the read-model error instead of silently falling back to legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" }); + factsReadBlocker.causeCode = factResult.error?.code ?? null; + factsReadBlocker.causeName = factResult.error?.name ?? null; + factsReadBlocker.message = factResult.error?.message ? `${factsReadBlocker.message} Cause: ${factResult.error.message}` : factResult.durable === false ? `${factsReadBlocker.message} Cause: durable facts reader unavailable.` : factsReadBlocker.message; + } else { facts = factResult?.facts ?? facts; - } catch (error) { - 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; } + } catch (error) { + const projection = factProjectionForTrace(facts, traceId); + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts query failed; realtime must surface the read-model error instead of silently falling back to legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" }); + factsReadBlocker.causeCode = error?.code ?? null; + factsReadBlocker.causeName = error?.name ?? null; + factsReadBlocker.message = error?.message ? `${factsReadBlocker.message} Cause: ${error.message}` : factsReadBlocker.message; } - const factSession = factArray(facts.sessions).find((item) => factSessionId(item) === factSessionIdValue) ?? null; - if (!factsReadBlocker) { - if (!factSessionIdValue) { - factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_id_missing", "Workbench realtime could not resolve a durable session id; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" }); - } else if (typeof readModel.queryFacts !== "function") { - factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench durable facts query is unavailable; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" }); - } else if (!factSession) { - factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_missing", "Workbench durable facts did not contain the trace session; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" }); - } + + const resolved = factsReadBlocker ? { session: null, facts } : await visibleFactSessionForTrace(readModel, facts, actor, traceId); + if (resolved.error) { + const projection = factProjectionForTrace(facts, traceId); + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_failed", "Workbench durable facts session lookup failed; realtime must surface the read-model error instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" }); + factsReadBlocker.causeCode = resolved.error?.code ?? null; + factsReadBlocker.causeName = resolved.error?.name ?? null; + factsReadBlocker.message = resolved.error?.message ? `${factsReadBlocker.message} Cause: ${resolved.error.message}` : factsReadBlocker.message; + } else if (resolved.facts) { + facts = resolved.facts; } - return { visible: true, turnId: projection.turnId, traceId, status: projection.status, projection, result, session, trace, facts, factSession, factsReadBlocker }; + + const factSession = resolved.session ?? null; + const trace = factTraceSnapshot(facts, traceId); + const turn = factTurnForTrace(facts, traceId, traceId); + let projection = factsReadBlocker ? blockedTraceEventProjection(factProjectionForTrace(facts, traceId), factsReadBlocker) : factProjectionForTrace(facts, traceId); + const visible = Boolean(factsReadBlocker || factSession || turn || factCheckpointForTrace(facts, traceId) || trace.eventCount > 0); + if (!visible) { + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_trace_missing", "Workbench realtime could not find durable facts for the requested trace; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, route: "/v1/workbench/events" }); + projection = blockedTraceEventProjection(projection, factsReadBlocker); + } + if (!factsReadBlocker && !factSession) { + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_missing", "Workbench durable facts did not contain the trace session; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, turn, route: "/v1/workbench/events" }); + projection = blockedTraceEventProjection(projection, factsReadBlocker); + } + const snapshot = factTurnSnapshot({ turn, session: factSession, facts, traceId, turnId: traceId }); + return { visible: true, turnId: snapshot.turnId ?? traceId, traceId, status: snapshot.status, projection, result: null, session: null, trace, facts, factSession, factsReadBlocker }; } function traceSnapshotForRealtime(snapshot) { @@ -2501,36 +2218,6 @@ function traceSnapshotLastSeq(snapshot) { 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");