diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index bb3a6784..cb5e74e6 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -535,6 +535,82 @@ test("workbench read model lets terminal result override stale running session s } }); +test("workbench read model recovers terminal session status from durable trace", async () => { + const traceStore = createCodeAgentTraceStore(); + const traceId = "trc_workbench_durable_terminal_after_memory_running"; + const session = { + id: "ses_workbench_durable_terminal_after_memory_running", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "running", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_durable_terminal_after_memory_running", + threadId: "thread-workbench-durable-terminal-after-memory-running", + lastTraceId: traceId, + updatedAt: "2026-06-18T17:22:00.000Z", + session: { + sessionStatus: "running", + lastTraceId: traceId, + messages: [ + { role: "user", text: "reply OK", traceId, status: "sent" }, + { role: "agent", text: "", traceId, status: "running" } + ] + } + }; + traceStore.append(traceId, { type: "backend", status: "running", label: "runner:created", terminal: false, seq: 1 }); + const durableEvents = [ + { traceId, seq: 1, type: "request", status: "accepted", label: "request:accepted", createdAt: "2026-06-18T17:21:40.000Z", valuesPrinted: false }, + { traceId, seq: 2, type: "result", status: "completed", label: "result:completed", terminal: true, createdAt: "2026-06-18T17:22:00.000Z", valuesPrinted: false } + ]; + const accessController = { + store: { + async listAgentSessionsForUser() { return [session]; }, + async getAgentSession(sessionId) { return sessionId === session.id ? session : null; }, + async getAgentSessionByTraceId(requestTraceId) { return requestTraceId === traceId ? session : null; } + }, + async ensureBootstrap() {}, + async authenticate() { return { ok: true, actor: ACTOR, session: { id: "uss_workbench_reader" } }; } + }; + const runtimeStore = { + async queryAgentTraceEvents(params = {}) { + assert.equal(params.traceId, traceId); + return { events: durableEvents, count: durableEvents.length }; + } + }; + 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); + + 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); + + const messages = await getJson(port, `/v1/workbench/sessions/${encodeURIComponent(session.id)}/messages?limit=10`); + assert.equal(messages.status, 200); + assert.equal(messages.body.messages[1].status, "completed"); + + const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`); + assert.equal(turn.status, 200); + assert.equal(turn.body.status, "completed"); + assert.equal(turn.body.turn.status, "completed"); + assert.equal(turn.body.turn.running, false); + assert.equal(turn.body.turn.terminal, true); + assert.equal(turn.body.turn.trace.eventCount, 2); + assert.equal(turn.body.projectionStatus, "caught-up"); + } finally { + await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); + } +}); + test("workbench realtime stream accepts session authority without project or workspace", async () => { const traceStore = createCodeAgentTraceStore(); const results = createCodeAgentChatResultStore(); diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index 9817f0a3..cf792c34 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -219,7 +219,7 @@ async function handleWorkbenchSessionList(response, url, options, actor) { const included = await readModel.getSessionByRouteId(includeRouteId); if (included) responseSessions = [included, ...pageSessions]; } - const summaries = responseSessions.map((session) => sessionSummary(session, options)).filter(Boolean); + const summaries = (await Promise.all(responseSessions.map(async (session) => sessionSummary(session, await sessionProjectionOptions(readModel, session, options))))).filter(Boolean); const hasMore = naturalPage.length > limit; sendJson(response, 200, { ok: true, @@ -246,11 +246,12 @@ async function handleWorkbenchSessionDetail(response, options, actor, sessionId) const readModel = createWorkbenchReadModel(options, actor); const session = await readModel.getSessionByRouteId(sessionId); if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId })); + const projectionOptions = await sessionProjectionOptions(readModel, session, options); sendJson(response, 200, { ok: true, status: "found", contractVersion: "workbench-session-detail-v1", - session: sessionDetail(session, options), + session: sessionDetail(session, projectionOptions), valuesRedacted: true, secretMaterialStored: false }); @@ -260,7 +261,8 @@ async function handleWorkbenchMessagePage(response, url, options, actor, session const readModel = createWorkbenchReadModel(options, actor); const session = await readModel.getSessionByRouteId(sessionId); if (!session) return sendJson(response, 404, workbenchError("workbench_session_not_found", "Workbench session is not visible to the current actor.", { sessionId })); - const messages = sessionMessages(session, options); + const projectionOptions = await sessionProjectionOptions(readModel, session, options); + const messages = sessionMessages(session, projectionOptions); 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); @@ -376,6 +378,15 @@ async function visibleSessionByTrace(store, traceId, actor) { 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); + return { ...options, trace, result }; +} + function sessionSummary(session, options) { if (!session?.id) return null; const snapshot = objectValue(session.session); diff --git a/internal/cloud/workbench-facts-store.ts b/internal/cloud/workbench-facts-store.ts index 08fc603d..e2b9932c 100644 --- a/internal/cloud/workbench-facts-store.ts +++ b/internal/cloud/workbench-facts-store.ts @@ -73,8 +73,10 @@ export function createWorkbenchFactsStore(options = {}, actor = null) { async function traceSnapshot(traceId) { const memory = traceSnapshotSync(traceId); - if (memory?.status !== "missing" && (memory.eventCount > 0 || memory.lastEvent)) return memory; + if (hasTraceProjection(memory) && TERMINAL_STATUSES.has(normalizeStatus(memory.status))) return memory; const durable = await durableTraceSnapshot(runtimeStore, traceId); + if (durable && shouldPreferDurableTrace(memory, durable)) return durable; + if (hasTraceProjection(memory)) return memory; return durable ?? memory; } @@ -129,6 +131,26 @@ async function durableTraceSnapshot(runtimeStore, traceId) { const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]); const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active"]); +function hasTraceProjection(snapshot) { + return Boolean(snapshot?.status !== "missing" && (Number(snapshot?.eventCount ?? 0) > 0 || snapshot?.lastEvent)); +} + +function shouldPreferDurableTrace(memory, durable) { + if (!durable) return false; + if (!hasTraceProjection(memory)) return true; + const memoryStatus = normalizeStatus(memory?.status); + const durableStatus = normalizeStatus(durable?.status); + if (TERMINAL_STATUSES.has(durableStatus) && !TERMINAL_STATUSES.has(memoryStatus)) return true; + return traceLastSeq(durable) > traceLastSeq(memory); +} + +function traceLastSeq(snapshot) { + const events = Array.isArray(snapshot?.events) ? snapshot.events : []; + const lastEvent = snapshot?.lastEvent ?? events.at(-1) ?? null; + const indexedMax = events.reduce((max, event, index) => Math.max(max, eventSeq(event, index)), 0); + return Math.max(indexedMax, lastEvent ? eventSeq(lastEvent, Math.max(0, events.length - 1)) : 0); +} + function durableTraceStatus(events) { const lastEvent = events.at(-1) ?? null; const normalized = normalizeStatus(lastEvent?.status ?? lastEvent?.type);