fix(workbench): recover session status from durable trace

This commit is contained in:
lyon
2026-06-19 01:34:53 +08:00
parent 587396e513
commit 111b2b468a
3 changed files with 113 additions and 4 deletions
@@ -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();
+14 -3
View File
@@ -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);
+23 -1
View File
@@ -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);