Merge pull request #1551 from pikasTech/fix/1519-durable-session-status

fix(workbench): recover session status from durable trace

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();

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);

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);