From 638509bcf16e0729d3458b1fe6df4588696d5126 Mon Sep 17 00:00:00 2001 From: lyon Date: Fri, 19 Jun 2026 23:39:54 +0800 Subject: [PATCH] fix: use compact workbench session summaries --- internal/cloud/server-workbench-http.test.ts | 88 ++++++++++++- internal/cloud/server-workbench-http.ts | 125 ++++++++++++++++++- web/hwlab-cloud-web/src/stores/workbench.ts | 2 +- 3 files changed, 207 insertions(+), 8 deletions(-) diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 4ff6aeb0..13ea81a5 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -201,6 +201,7 @@ test("workbench read model exposes session, messages, turn, and trace without wr assert.equal(sessions.body.sessions[0].sessionId, session.id); assert.equal(sessions.body.sessions[0].turnSummary.traceId, traceId); assert.equal(listInputs[0].projectId, undefined); + assert.equal(listInputs[0].limit, 21); const staleProject = await getJson(port, `/v1/workbench/sessions?projectId=prj_stale_filter&includeSessionId=${encodeURIComponent(session.id)}`); assert.equal(staleProject.status, 400); @@ -674,6 +675,83 @@ test("workbench read model lets terminal result override stale running session s } }); +test("workbench session list uses compact trace result without durable trace hydration", async () => { + const traceStore = createCodeAgentTraceStore(); + const traceId = "trc_workbench_compact_list_summary"; + const finalText = "compact list summary OK"; + const session = { + id: "ses_workbench_compact_list_summary", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "running", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_compact_list_summary", + threadId: "thread-workbench-compact-list-summary", + lastTraceId: traceId, + updatedAt: "2026-06-19T15:40:00.000Z", + session: { + sessionStatus: "running", + lastTraceId: traceId, + messages: [ + { role: "user", text: "compact summary", traceId, status: "sent" }, + { role: "agent", text: "", traceId, status: "running" } + ], + traceResults: { + [traceId]: { + traceId, + status: "completed", + finalResponse: { text: finalText, status: "completed", traceId, valuesPrinted: false }, + traceSummary: { + traceId, + source: "agent-session-compact-summary", + sourceEventCount: 42, + terminalStatus: "completed", + agentRun: { runId: "run_workbench_compact_list_summary", commandId: "cmd_workbench_compact_list_summary", lastSeq: 42, valuesPrinted: false }, + valuesPrinted: false + }, + agentRun: { runId: "run_workbench_compact_list_summary", commandId: "cmd_workbench_compact_list_summary", status: "completed", terminalStatus: "completed", lastSeq: 42, valuesPrinted: false }, + updatedAt: "2026-06-19T15:40:00.000Z", + valuesRedacted: true, + secretMaterialStored: false + } + }, + valuesRedacted: true, + secretMaterialStored: false + } + }; + const accessController = { + store: { + async listAgentSessionsForUser() { return [session]; } + }, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + let durableQueryCount = 0; + const runtimeStore = { + async queryAgentTraceEvents() { + durableQueryCount += 1; + throw new Error("session list must not hydrate durable trace events"); + } + }; + const server = createCloudApiServer({ accessController, traceStore, runtimeStore, codeAgentChatResults: createCodeAgentChatResultStore() }); + await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve)); + + try { + const { port } = server.address(); + const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`); + assert.equal(sessions.status, 200); + assert.equal(sessions.body.sessions[0].status, "completed"); + assert.equal(sessions.body.sessions[0].running, false); + assert.equal(sessions.body.sessions[0].terminal, true); + assert.equal(sessions.body.sessions[0].turnSummary.status, "completed"); + assert.equal(sessions.body.sessions[0].turnSummary.eventCount, 42); + assert.equal(sessions.body.sessions[0].projectionStatus, "caught-up"); + assert.equal(durableQueryCount, 0); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench read model recovers terminal session status from durable trace", async () => { const traceStore = createCodeAgentTraceStore(); const traceId = "trc_workbench_durable_terminal_after_memory_running"; @@ -710,8 +788,10 @@ test("workbench read model recovers terminal session status from durable trace", async ensureBootstrap() {}, async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } }; + let durableQueryCount = 0; const runtimeStore = { async queryAgentTraceEvents(params = {}) { + durableQueryCount += 1; assert.equal(params.traceId, traceId); return { events: durableEvents, count: durableEvents.length }; } @@ -723,15 +803,17 @@ test("workbench read model recovers terminal session status from durable trace", const { port } = server.address(); const sessions = await getJson(port, `/v1/workbench/sessions?includeSessionId=${encodeURIComponent(session.id)}`); assert.equal(sessions.status, 200); - assert.equal(sessions.body.sessions[0].status, "completed"); - assert.equal(sessions.body.sessions[0].running, false); - assert.equal(sessions.body.sessions[0].terminal, true); + assert.equal(sessions.body.sessions[0].status, "running"); + assert.equal(sessions.body.sessions[0].running, true); + assert.equal(sessions.body.sessions[0].terminal, false); + assert.equal(durableQueryCount, 0); const detail = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}`); assert.equal(detail.status, 200); assert.equal(detail.body.session.status, "completed"); assert.equal(detail.body.session.running, false); assert.equal(detail.body.session.terminal, true); + assert.equal(durableQueryCount, 1); const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`); assert.equal(messages.status, 200); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 93f8657f..a7664aab 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -17,7 +17,7 @@ import { createWorkbenchReadModel } from "./workbench-read-model.ts"; import { createWorkbenchTurnProjection, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; const DEFAULT_PAGE_LIMIT = 50; -const DEFAULT_SESSION_LIST_LIMIT = 8; +const DEFAULT_SESSION_LIST_LIMIT = 20; const MAX_PAGE_LIMIT = 100; const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000; export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) { @@ -308,19 +308,20 @@ async function authenticateWorkbenchRead(request, response, options) { 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 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 naturalPage = await readModel.listSessions({ limit: limit + 1, offset }); + const naturalPage = await (perf ? perf.measure("workbench_session_page_query", () => readModel.listSessions({ limit: limit + 1, offset })) : readModel.listSessions({ limit: limit + 1, offset })); const pageSessions = naturalPage.slice(0, limit); let responseSessions = pageSessions; if (includeRouteId && !pageSessions.some((session) => sessionMatchesRouteId(session, includeRouteId))) { - const included = await readModel.getSessionByRouteId(includeRouteId); + const included = await (perf ? perf.measure("workbench_session_include_query", () => readModel.getSessionByRouteId(includeRouteId)) : readModel.getSessionByRouteId(includeRouteId)); if (included) responseSessions = [included, ...pageSessions]; } - const summaries = (await Promise.all(responseSessions.map(async (session) => sessionSummary(session, await sessionProjectionOptions(readModel, session, options))))).filter(Boolean); + const summaries = await (perf ? perf.measure("workbench_session_summary_projection", () => sessionListSummaries(readModel, responseSessions, options)) : sessionListSummaries(readModel, responseSessions, options)); const hasMore = naturalPage.length > limit; sendJson(response, 200, { ok: true, @@ -511,6 +512,122 @@ async function sessionProjectionOptions(readModel, session, options) { 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 result = memoryResult ?? compactSessionTraceResult(snapshot, session, traceId); + const memoryTrace = traceSnapshotSync(options, traceId); + const trace = compactSessionTraceSnapshot(snapshot, session, traceId, result, memoryTrace); + 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); diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 272033ac..0aa2f98a 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -21,7 +21,7 @@ const TRACE_HYDRATION_MAX_ATTEMPTS = 3; const TRACE_HYDRATION_RETRY_DELAY_MS = 700; const ACTIVE_TURN_GAP_DELAY_MS = 1_000; const ACTIVE_TURN_GAP_MAX_ATTEMPTS = 12; -const SESSION_LIST_PAGE_LIMIT = 8; +const SESSION_LIST_PAGE_LIMIT = 20; interface HydrateOptions { sessionId?: string | null;