From e5994d8af104a55098066554ac26b6f4956e1348 Mon Sep 17 00:00:00 2001 From: lyon Date: Fri, 19 Jun 2026 04:17:28 +0800 Subject: [PATCH] fix(workbench): use single turn projection authority --- internal/cloud/server-workbench-http.test.ts | 91 ++++++++ internal/cloud/server-workbench-http.ts | 58 ++--- internal/cloud/workbench-facts-store.ts | 22 +- internal/cloud/workbench-read-model.ts | 46 +--- internal/cloud/workbench-turn-projection.ts | 203 ++++++++++++++++++ .../scripts/workbench-e2e-server.ts | 45 +++- web/hwlab-cloud-web/src/api/workbench.ts | 10 +- web/hwlab-cloud-web/src/stores/workbench.ts | 29 ++- .../workbench-e2e/specs/event-replay.spec.ts | 44 ++++ 9 files changed, 433 insertions(+), 115 deletions(-) create mode 100644 internal/cloud/workbench-turn-projection.ts diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index cb5e74e6..145ffdc3 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -611,6 +611,97 @@ test("workbench read model recovers terminal session status from durable trace", } }); +test("workbench read model keeps non-terminal durable tool completed events out of turn status", async () => { + const traceStore = createCodeAgentTraceStore(); + const results = createCodeAgentChatResultStore(); + const traceId = "trc_workbench_durable_tool_completed_nonterminal"; + const session = { + id: "ses_workbench_durable_tool_completed_nonterminal", + projectId: "prj_hwpod_workbench", + agentId: "hwlab-code-agent", + status: "running", + ownerUserId: ACTOR.id, + conversationId: "cnv_workbench_durable_tool_completed_nonterminal", + threadId: "thread-workbench-durable-tool-completed-nonterminal", + lastTraceId: traceId, + updatedAt: "2026-06-18T19:35:51.787Z", + session: { + sessionStatus: "running", + lastTraceId: traceId, + messages: [ + { role: "user", text: "write benchmark", traceId, status: "sent" }, + { role: "agent", text: "", traceId, status: "running" } + ], + valuesRedacted: true, + secretMaterialStored: false + } + }; + traceStore.append(traceId, { seq: 1, type: "backend_status", status: "running", label: "runner:created", terminal: false }); + const durableEvents = [ + { traceId, seq: 1, type: "assistant_message", status: "running", label: "agentrun:assistant:message", terminal: false, message: "正在安装 Python。", createdAt: "2026-06-18T19:35:40.000Z", valuesPrinted: false }, + { traceId, seq: 2, type: "commandExecution", status: "completed", label: "item/commandExecution:completed", terminal: false, command: "ls -ld .", stdout: "OK\n", createdAt: "2026-06-18T19:35:51.787Z", valuesPrinted: false } + ]; + results.set(traceId, { + status: "running", + traceId, + ownerUserId: ACTOR.id, + conversationId: session.conversationId, + sessionId: session.id, + threadId: session.threadId, + finalResponse: null, + assistantText: null, + agentRun: { runId: "run_workbench_nonterminal_tool_completed", commandId: "cmd_workbench_nonterminal_tool_completed", status: "runner-job-created", runStatus: "pending", commandState: "pending" } + }); + 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: results }); + 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, "running"); + assert.equal(sessions.body.sessions[0].running, true); + assert.equal(sessions.body.sessions[0].terminal, false); + + 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, "running"); + assert.equal(messages.body.messages[1].text, ""); + + const turn = await getJson(port, `/v1/workbench/turns/${encodeURIComponent(traceId)}`); + assert.equal(turn.status, 200); + assert.equal(turn.body.status, "running"); + assert.equal(turn.body.turn.status, "running"); + assert.equal(turn.body.turn.running, true); + assert.equal(turn.body.turn.terminal, false); + assert.equal(turn.body.turn.finalResponse, null); + assert.equal(turn.body.projectionStatus, "projecting"); + + const trace = await getJson(port, `/v1/workbench/traces/${encodeURIComponent(traceId)}/events?limit=10`); + assert.equal(trace.status, 200); + assert.equal(trace.body.events.at(-1).status, "completed"); + assert.equal(trace.body.events.at(-1).terminal, false); + assert.equal(trace.body.projectionStatus, "projecting"); + } 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 cf792c34..2a1938c8 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -14,13 +14,11 @@ import { sendJson } from "./server-http-utils.ts"; import { createWorkbenchReadModel } from "./workbench-read-model.ts"; +import { createWorkbenchTurnProjection, RUNNING_STATUSES, TERMINAL_STATUSES } from "./workbench-turn-projection.ts"; const DEFAULT_PAGE_LIMIT = 50; const MAX_PAGE_LIMIT = 100; const DEFAULT_WORKBENCH_SSE_HEARTBEAT_MS = 15000; -const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]); -const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active"]); - export async function handleWorkbenchReadModelHttp(request, response, url, options = {}) { const perf = options.backendPerformance; const auth = perf ? await perf.measure("workbench_auth", () => authenticateWorkbenchRead(request, response, options)) : await authenticateWorkbenchRead(request, response, options); @@ -296,15 +294,16 @@ async function handleWorkbenchTurnSnapshot(response, url, options, actor, rawTur } const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId)); const trace = await readModel.traceSnapshot(traceId); - const status = workbenchTurnStatusFromFacts({ trace, result }); + const turnProjection = createWorkbenchTurnProjection({ turnId, traceId, result, session, trace }); + const status = turnProjection.status; const found = Boolean(resultVisible || session); if (!found) return sendJson(response, 404, workbenchError("workbench_turn_not_found", "Workbench turn is not visible to the current actor.", { turnId, traceId })); - const projection = readModel.projectionDiagnostics({ traceId, result, trace }); + const projection = readModel.projectionDiagnostics({ traceId, result, trace, projection: turnProjection }); sendJson(response, 200, { ok: true, status, contractVersion: "workbench-turn-snapshot-v1", - turn: turnSnapshot({ turnId, traceId, status, result, session, trace }), + turn: turnSnapshot({ projection: turnProjection, result, session, trace }), projection, projectionStatus: projection.projectionStatus, lastProjectedSeq: projection.lastProjectedSeq, @@ -331,7 +330,8 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT } const trace = await readModel.traceSnapshot(traceId); const page = traceEventPage(trace, tracePageOptions(url)); - const projection = readModel.projectionDiagnostics({ traceId, result, trace }); + const turnProjection = createWorkbenchTurnProjection({ traceId, result, session, trace }); + const projection = readModel.projectionDiagnostics({ traceId, result, trace, projection: turnProjection }); sendJson(response, 200, { ok: true, status: "succeeded", @@ -384,7 +384,8 @@ async function sessionProjectionOptions(readModel, session, options) { if (!traceId) return options; const trace = await readModel.traceSnapshot(traceId); const result = readModel.resultForTrace(traceId); - return { ...options, trace, result }; + const projection = createWorkbenchTurnProjection({ traceId, result, session, trace }); + return { ...options, trace, result, projection }; } function sessionSummary(session, options) { @@ -459,7 +460,7 @@ function terminalMessageProjection(snapshot, options, currentTurn) { const trace = currentTurn.trace; const status = currentTurn.status; if (!TERMINAL_STATUSES.has(status) || RUNNING_STATUSES.has(status)) return null; - const text = projectionText(result?.finalResponse, result?.assistantText, result?.reply, result?.text, result?.summary, trace?.finalResponse, trace?.terminalEvidence?.finalResponse, snapshot.finalResponse); + const text = projectionText(currentTurn.finalResponse, result?.finalResponse, result?.assistantText, result?.reply, result?.text, result?.summary, trace?.finalResponse, trace?.terminalEvidence?.finalResponse, snapshot.finalResponse); return { traceId, status, text, source: "turn-projection" }; } @@ -544,11 +545,14 @@ function partFact(part, index, messageId, traceId) { }; } -function turnSnapshot({ turnId, traceId, status, result, session, trace }) { +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(result?.finalResponse, result?.assistantText, result?.reply, result?.text, trace?.finalResponse, trace?.terminalEvidence?.finalResponse, assistantMessage?.text); + const finalText = projectionText(projection.finalResponse); return { turnId, traceId, @@ -560,7 +564,7 @@ function turnSnapshot({ turnId, traceId, status, result, session, trace }) { userMessageId: userMessage?.messageId ?? null, assistantMessageId: assistantMessage?.messageId ?? null, assistantText: finalText ?? null, - finalResponse: finalText ? { text: finalText, status, traceId, valuesPrinted: false } : null, + finalResponse: projection.finalResponse ?? null, agentRun: result?.agentRun ?? null, trace: { traceId, @@ -657,8 +661,8 @@ async function visibleTraceContext(options, actor, traceId) { const resultVisible = result && (session || actor.role === "admin" || Boolean(result.ownerUserId)); if (!resultVisible && !session) return { visible: false }; const trace = await readModel.traceSnapshot(traceId); - const status = workbenchTurnStatusFromFacts({ trace, result }); - return { visible: true, turnId: traceId, traceId, status, result, session, trace }; + const projection = createWorkbenchTurnProjection({ turnId: traceId, traceId, result, session, trace }); + return { visible: true, turnId: projection.turnId, traceId, status: projection.status, projection, result, session, trace }; } function traceSnapshotForRealtime(snapshot) { @@ -695,15 +699,7 @@ function currentWorkbenchTurnProjection(session, options = {}) { if (!traceId) return null; const trace = options.trace?.traceId === traceId ? options.trace : traceSnapshotSync(options, traceId); const result = traceScopedResult(options, traceId); - const status = workbenchTurnStatusFromFacts({ trace, result }); - return { - turnId: traceId, - traceId, - status, - trace, - eventCount: Number.isFinite(Number(trace?.eventCount)) ? Number(trace.eventCount) : null, - updatedAt: trace?.updatedAt ?? session?.updatedAt ?? null - }; + return createWorkbenchTurnProjection({ turnId: traceId, traceId, result, session, trace }); } function traceScopedResult(options, traceId) { @@ -711,22 +707,6 @@ function traceScopedResult(options, traceId) { return result && typeof result === "object" ? result : null; } -function workbenchTurnStatusFromFacts({ trace = null, result = null } = {}) { - const resultStatus = normalizeStatus(result?.status ?? result?.terminalStatus); - if (TERMINAL_STATUSES.has(resultStatus) && !RUNNING_STATUSES.has(resultStatus)) return resultStatus; - const traceStatus = workbenchTurnProjectionStatus(trace); - if (traceStatus !== "unknown") return traceStatus; - if (RUNNING_STATUSES.has(resultStatus)) return resultStatus; - return "unknown"; -} - -function workbenchTurnProjectionStatus(trace) { - const status = normalizeStatus(trace?.status); - if (status === "missing") return "unknown"; - if (RUNNING_STATUSES.has(status) || TERMINAL_STATUSES.has(status)) return status; - return "unknown"; -} - function sessionLifecycleProjectionStatus(session) { return normalizeStatus(session?.status); } diff --git a/internal/cloud/workbench-facts-store.ts b/internal/cloud/workbench-facts-store.ts index e2b9932c..d4cd6a0c 100644 --- a/internal/cloud/workbench-facts-store.ts +++ b/internal/cloud/workbench-facts-store.ts @@ -4,6 +4,7 @@ */ import { defaultCodeAgentTraceStore } from "./code-agent-trace-store.ts"; import { safeConversationId, safeSessionId, safeTraceId } from "./server-http-utils.ts"; +import { durableTraceStatus, normalizeWorkbenchStatus, TERMINAL_STATUSES, traceTerminalEvidence } from "./workbench-turn-projection.ts"; export function createWorkbenchFactsStore(options = {}, actor = null) { const accessStore = options.accessController?.store ?? options.accessController ?? null; @@ -111,6 +112,7 @@ async function durableTraceSnapshot(runtimeStore, traceId) { const firstEvent = normalizedEvents[0] ?? null; const lastEvent = normalizedEvents.at(-1) ?? null; const status = durableTraceStatus(normalizedEvents); + const terminalEvidence = traceTerminalEvidence({ events: normalizedEvents, status }); return { traceId, status, @@ -123,14 +125,12 @@ async function durableTraceSnapshot(runtimeStore, traceId) { assistantStreams: [], eventLabels: normalizedEvents.map((event) => event.label).filter(Boolean), lastEvent, + terminalEvidence, outputTruncated: normalizedEvents.some((event) => event.outputTruncated === true), valuesPrinted: false }; } -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)); } @@ -138,8 +138,8 @@ function hasTraceProjection(snapshot) { function shouldPreferDurableTrace(memory, durable) { if (!durable) return false; if (!hasTraceProjection(memory)) return true; - const memoryStatus = normalizeStatus(memory?.status); - const durableStatus = normalizeStatus(durable?.status); + const memoryStatus = normalizeWorkbenchStatus(memory?.status); + const durableStatus = normalizeWorkbenchStatus(durable?.status); if (TERMINAL_STATUSES.has(durableStatus) && !TERMINAL_STATUSES.has(memoryStatus)) return true; return traceLastSeq(durable) > traceLastSeq(memory); } @@ -151,18 +151,6 @@ function traceLastSeq(snapshot) { 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); - if (lastEvent?.terminal === true) { - if (normalized === "observed" || normalized === "event" || normalized === "unknown") return "completed"; - return normalized; - } - if (TERMINAL_STATUSES.has(normalized) || RUNNING_STATUSES.has(normalized)) return normalized; - if (["failed", "error", "timeout"].includes(normalized)) return normalized; - return "running"; -} - function canActorReadSession(session, actor) { if (!session || !actor) return false; if (normalizeStatus(session.status) === "archived") return false; diff --git a/internal/cloud/workbench-read-model.ts b/internal/cloud/workbench-read-model.ts index b1f45410..c2b3f2f2 100644 --- a/internal/cloud/workbench-read-model.ts +++ b/internal/cloud/workbench-read-model.ts @@ -3,6 +3,7 @@ * 职责: WorkbenchReadModel 组件入口。只通过 WorkbenchFactsStore 读取 durable projection facts，不执行 AgentRun sync/finalize/repair。 */ import { createWorkbenchFactsStore } from "./workbench-facts-store.ts"; +import { projectionDiagnostics as workbenchProjectionDiagnostics } from "./workbench-turn-projection.ts"; export function createWorkbenchReadModel(options = {}, actor = null) { const facts = createWorkbenchFactsStore(options, actor); @@ -17,49 +18,6 @@ export function createWorkbenchReadModel(options = {}, actor = null) { traceSnapshotSync: (traceId) => facts.traceSnapshotSync(traceId), subscribeTrace: (traceId, listener) => facts.subscribeTrace(traceId, listener), canReadOwner: (ownerUserId) => facts.canReadOwner(ownerUserId), - projectionDiagnostics: ({ traceId, result = null, trace = null } = {}) => projectionDiagnostics({ traceId, result, trace }) + projectionDiagnostics: ({ traceId, result = null, trace = null, projection = null } = {}) => workbenchProjectionDiagnostics({ traceId, result, trace, projection }) }; } - -export function projectionDiagnostics({ traceId, result = null, trace = null } = {}) { - const agentRun = result?.agentRun && typeof result.agentRun === "object" ? result.agentRun : null; - const events = Array.isArray(trace?.events) ? trace.events : []; - const lastEvent = trace?.lastEvent ?? events.at(-1) ?? null; - const lastProjectedSeq = Number(lastEvent?.seq ?? trace?.lastProjectedSeq ?? 0); - const hasTraceProjection = Boolean(trace && trace.status !== "missing" && (events.length > 0 || lastEvent)); - const sourceRunId = agentRun?.runId ?? lastEvent?.runId ?? null; - const sourceCommandId = agentRun?.commandId ?? lastEvent?.commandId ?? null; - const blocker = result?.blocker ?? result?.error ?? null; - const status = blocker - ? "blocked" - : !hasTraceProjection && (result || agentRun) - ? "projecting" - : isTerminalStatus(result?.status ?? trace?.status) - ? "caught-up" - : result || trace?.status !== "missing" - ? "projecting" - : "unknown"; - return { - projectionStatus: status, - lastProjectedSeq: Number.isFinite(lastProjectedSeq) && lastProjectedSeq > 0 ? Math.trunc(lastProjectedSeq) : null, - sourceRunId, - sourceCommandId, - blocker: blocker ? diagnosticBlocker(blocker) : null, - updatedAt: trace?.updatedAt ?? result?.updatedAt ?? null, - valuesRedacted: true - }; -} - -function diagnosticBlocker(value = {}) { - if (!value || typeof value !== "object") return { code: "projection_blocked", summary: String(value), valuesRedacted: true }; - return { - code: String(value.code ?? value.errorCode ?? "projection_blocked"), - summary: String(value.summary ?? value.userMessage ?? value.message ?? value.code ?? "Projection is blocked."), - retryable: typeof value.retryable === "boolean" ? value.retryable : null, - valuesRedacted: true - }; -} - -function isTerminalStatus(value) { - return ["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"].includes(String(value ?? "").trim().toLowerCase()); -} diff --git a/internal/cloud/workbench-turn-projection.ts b/internal/cloud/workbench-turn-projection.ts new file mode 100644 index 00000000..c60a92ac --- /dev/null +++ b/internal/cloud/workbench-turn-projection.ts @@ -0,0 +1,203 @@ +/* + * SPEC: PJ2026-0104010803 Workbench唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-010403 API契约 draft-2026-06-18-r1. + * 职责: 生成唯一 Workbench turn projection；trace event 状态只作为输入证据，不直接成为 turn lifecycle。 + */ + +export const TERMINAL_STATUSES = new Set(["completed", "failed", "blocked", "timeout", "cancelled", "canceled", "idle"]); +export const RUNNING_STATUSES = new Set(["running", "pending", "queued", "accepted", "dispatching", "streaming", "active", "processing", "busy", "creating"]); + +export function createWorkbenchTurnProjection({ turnId = null, traceId = null, result = null, session = null, trace = null } = {}) { + const projectionTraceId = textValue(traceId ?? trace?.traceId ?? result?.traceId ?? session?.lastTraceId) || null; + const projectionTurnId = textValue(turnId) || projectionTraceId; + const terminalEvidence = terminalTurnEvidence({ result, trace }); + const activeEvidence = activeTurnEvidence({ result, session, trace }); + const status = terminalEvidence?.status ?? activeEvidence?.status ?? "unknown"; + const running = RUNNING_STATUSES.has(status); + const terminal = Boolean(terminalEvidence && TERMINAL_STATUSES.has(status) && !running); + const finalText = terminal ? projectionText(result?.finalResponse, result?.assistantText, result?.reply, result?.text, result?.summary, trace?.finalResponse, trace?.terminalEvidence?.finalResponse) : null; + const agentRun = objectValue(result?.agentRun ?? trace?.agentRun); + const lastEvent = traceLastEvent(trace); + return { + turnId: projectionTurnId, + traceId: projectionTraceId, + status, + running, + terminal, + source: terminalEvidence?.source ?? activeEvidence?.source ?? null, + terminalEvidence, + finalResponse: finalText ? { text: finalText, status, traceId: projectionTraceId, valuesPrinted: false } : null, + assistantText: finalText, + lastProjectedSeq: traceLastSeq(trace), + sourceRunId: agentRun?.runId ?? lastEvent?.runId ?? lastEvent?.payload?.runId ?? null, + sourceCommandId: agentRun?.commandId ?? lastEvent?.commandId ?? lastEvent?.payload?.commandId ?? null, + eventCount: normalizedEventCount(trace), + updatedAt: trace?.updatedAt ?? result?.updatedAt ?? session?.updatedAt ?? null, + valuesRedacted: true + }; +} + +export function projectionDiagnostics({ traceId = null, projection = null, result = null, trace = null } = {}) { + const turn = projection ?? createWorkbenchTurnProjection({ traceId, result, trace }); + const blocker = result?.blocker ?? result?.error ?? null; + const hasProjectionInput = hasTraceProjection(trace) || Boolean(result || result?.agentRun); + const status = blocker ? "blocked" : turn.terminal ? "caught-up" : hasProjectionInput ? "projecting" : "unknown"; + return { + projectionStatus: status, + lastProjectedSeq: turn.lastProjectedSeq ?? null, + sourceRunId: turn.sourceRunId ?? null, + sourceCommandId: turn.sourceCommandId ?? null, + blocker: blocker ? diagnosticBlocker(blocker) : null, + updatedAt: turn.updatedAt ?? null, + valuesRedacted: true + }; +} + +export function durableTraceStatus(events = []) { + const terminal = terminalTraceEventEvidence(events); + if (terminal) return terminal.status; + const activeStatus = activeTraceEventStatus(events); + return activeStatus ?? (events.length > 0 ? "running" : "missing"); +} + +export function traceTerminalEvidence(trace = null) { + const direct = objectValue(trace?.terminalEvidence); + if (direct) { + const status = terminalStatusFromValue(direct.status ?? direct.terminalStatus ?? trace?.status) ?? "completed"; + return { source: "trace-terminal-evidence", status, evidence: direct, valuesRedacted: true }; + } + const finalResponse = objectValue(trace?.finalResponse); + if (finalResponse) { + const status = terminalStatusFromValue(finalResponse.status ?? trace?.status) ?? "completed"; + return { source: "trace-final-response", status, evidence: { textPresent: Boolean(projectionText(finalResponse)), valuesRedacted: true }, valuesRedacted: true }; + } + const events = Array.isArray(trace?.events) ? trace.events : []; + return terminalTraceEventEvidence(events); +} + +export function normalizeWorkbenchStatus(value) { + const text = String(value ?? "").trim().toLowerCase().replace(/_/gu, "-"); + if (text === "cancelled") return "canceled"; + return text || "unknown"; +} + +function terminalTurnEvidence({ result = null, trace = null } = {}) { + const resultStatus = terminalStatusFromValue(result?.status ?? result?.terminalStatus ?? result?.agentRun?.terminalStatus); + if (resultStatus) return { source: "result", status: resultStatus, valuesRedacted: true }; + return traceTerminalEvidence(trace); +} + +function activeTurnEvidence({ result = null, session = null, trace = null } = {}) { + const resultStatus = firstStatus([ + result?.status, + result?.agentRun?.status, + result?.agentRun?.commandState, + result?.agentRun?.runStatus + ], RUNNING_STATUSES); + if (resultStatus) return { source: "result", status: normalizeActiveStatus(resultStatus), valuesRedacted: true }; + const traceStatus = firstStatus([trace?.status, trace?.traceStatus], RUNNING_STATUSES); + if (traceStatus) return { source: "trace", status: normalizeActiveStatus(traceStatus), valuesRedacted: true }; + if (hasTraceProjection(trace)) return { source: "trace-events", status: "running", valuesRedacted: true }; + const sessionStatus = firstStatus([session?.status, session?.session?.sessionStatus], RUNNING_STATUSES); + if (sessionStatus) return { source: "session", status: normalizeActiveStatus(sessionStatus), valuesRedacted: true }; + return null; +} + +function firstStatus(values, accepted) { + for (const value of values) { + const status = normalizeWorkbenchStatus(value); + if (accepted.has(status)) return status; + } + return null; +} + +function terminalStatusFromValue(value) { + const status = normalizeWorkbenchStatus(value); + return TERMINAL_STATUSES.has(status) && !RUNNING_STATUSES.has(status) ? status : null; +} + +function normalizeActiveStatus(status) { + return status === "active" || status === "busy" || status === "processing" || status === "creating" ? "running" : status; +} + +function terminalTraceEventEvidence(events = []) { + for (const event of [...events].reverse()) { + if (!event || typeof event !== "object") continue; + const terminal = event.terminal === true || event.final === true || event.replyAuthority === true; + if (!terminal) continue; + const status = terminalStatusFromValue(event.status ?? event.terminalStatus) ?? (event.error || event.errorCode ? "failed" : "completed"); + return { + source: "trace-terminal-event", + status, + seq: eventSeq(event, events.indexOf(event)), + eventType: textValue(event.type ?? event.label) || null, + valuesRedacted: true + }; + } + return null; +} + +function activeTraceEventStatus(events = []) { + for (const event of [...events].reverse()) { + const status = normalizeWorkbenchStatus(event?.status ?? event?.type); + if (RUNNING_STATUSES.has(status)) return normalizeActiveStatus(status); + } + return null; +} + +function hasTraceProjection(trace) { + return Boolean(trace && trace.status !== "missing" && (normalizedEventCount(trace) > 0 || traceLastEvent(trace))); +} + +function normalizedEventCount(trace) { + const events = Array.isArray(trace?.events) ? trace.events : []; + const count = Number(trace?.eventCount ?? events.length); + return Number.isFinite(count) && count >= 0 ? Math.trunc(count) : events.length; +} + +function traceLastSeq(trace) { + const events = Array.isArray(trace?.events) ? trace.events : []; + const lastEvent = traceLastEvent(trace); + 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) || null; +} + +function traceLastEvent(trace) { + const events = Array.isArray(trace?.events) ? trace.events : []; + return trace?.lastEvent ?? events.at(-1) ?? null; +} + +function eventSeq(event, index) { + const seq = Number(event?.seq); + return Number.isFinite(seq) && seq > 0 ? Math.trunc(seq) : index + 1; +} + +function projectionText(...values) { + for (const value of values) { + if (value && typeof value === "object") { + const nested = textValue(value.text ?? value.content ?? value.message ?? value.summary ?? value.preview ?? value.title); + if (nested) return nested; + continue; + } + const text = textValue(value); + if (text) return text; + } + return null; +} + +function objectValue(value) { + return value && typeof value === "object" && !Array.isArray(value) ? value : null; +} + +function textValue(value) { + return String(value ?? "").trim(); +} + +function diagnosticBlocker(value = {}) { + if (!value || typeof value !== "object") return { code: "projection_blocked", summary: String(value), valuesRedacted: true }; + return { + code: String(value.code ?? value.errorCode ?? "projection_blocked"), + summary: String(value.summary ?? value.userMessage ?? value.message ?? value.code ?? "Projection is blocked."), + retryable: typeof value.retryable === "boolean" ? value.retryable : null, + valuesRedacted: true + }; +} diff --git a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts index 1e46d241..bc2eea41 100644 --- a/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts +++ b/web/hwlab-cloud-web/scripts/workbench-e2e-server.ts @@ -280,6 +280,7 @@ function createScenarioState(scenarioId: string): ScenarioState { } if (id === "progress-only-final-response") markRunningProgressOnly(sessions, traces); if (id === "terminal-completed-no-final-response") markRunningNoFinalResponse(sessions, traces); + if (id === "tool-completed-projection-running") markToolCompletedProjectionRunning(sessions, traces); if (id === "scroll-follow-long-trace") { const session = scrollFollowSession(); sessions.unshift(session); @@ -302,7 +303,7 @@ function createScenarioState(scenarioId: string): ScenarioState { ? "ses_completed" : id === "terminal-empty-trace" ? "ses_terminal_empty" - : id === "progress-only-final-response" || id === "terminal-completed-no-final-response" + : id === "progress-only-final-response" || id === "terminal-completed-no-final-response" || id === "tool-completed-projection-running" ? "ses_running" : base.selectedSessionId; const staleTraceId = id === "stale-nested-trace" || id === "stale-submit-restore" ? "trc_stale_502" : null; @@ -370,6 +371,44 @@ function markRunningNoFinalResponse(sessions: SessionRecord[], traces: Record): void { + const trace = toolCompletedProjectionRunningTrace(); + traces.trc_running = trace; + const session = sessions.find((item) => item.sessionId === "ses_running"); + if (!session) return; + session.status = "running"; + session.lastTraceId = "trc_running"; + session.firstUserMessagePreview = "tool completed but turn still running"; + session.messages = (session.messages ?? []).map((message) => { + if (message.role === "user") return { ...message, text: "tool completed but turn still running" }; + if (message.role !== "agent" || message.traceId !== "trc_running") return message; + return { ...message, text: "", status: "running", runnerTrace: trace }; + }); +} + +function toolCompletedProjectionRunningTrace(): JsonRecord { + const createdAt = new Date().toISOString(); + const events = [ + { seq: 1, createdAt, label: "agentrun:assistant:message", type: "assistant_message", status: "running", terminal: false, message: "正在安装 Python。" }, + { seq: 2, createdAt, label: "item/commandExecution:completed", type: "commandExecution", status: "completed", terminal: false, command: "ls -ld .", stdout: "OK\n" } + ]; + return { + traceId: "trc_running", + status: "completed", + turnStatus: "running", + projectionStatus: "projecting", + sessionId: "ses_running", + threadId: "thr_running", + turnId: "turn_running", + events, + eventCount: events.length, + fullTraceLoaded: true, + hasMore: false, + finalResponse: null, + assistantText: null + }; +} + function noFinalRunningTrace(): JsonRecord { const createdAt = new Date().toISOString(); return { @@ -862,6 +901,7 @@ function turnPayload(traceId: string): JsonRecord { return state.traces[traceId] ?? liveBackfillEarlyTrace(sessionId, threadId, traceId); } const trace = state.traces[traceId] ?? { traceId, status: "unknown", events: [] }; + if (state.scenarioId === "tool-completed-projection-running" && traceId === "trc_running") return { ...trace, traceId, status: "running", running: true, terminal: false, projectionStatus: "projecting", trace: { status: trace.status, eventCount: trace.eventCount ?? 0 } }; if (state.scenarioId === "terminal-turn-stale-session-active" && traceId === "trc_completed") return { ...trace, traceId, status: "active", running: true, terminal: false, trace: { status: "completed", eventCount: trace.eventCount ?? 0 } }; const status = String(trace.status ?? "unknown"); return { ...trace, traceId, status, running: ["running", "pending", "accepted"].includes(status), terminal: ["completed", "failed", "blocked", "timeout", "canceled"].includes(status) }; @@ -882,6 +922,9 @@ function tracePayload(traceId: string, url: URL): JsonRecord { function workbenchTracePayload(traceId: string, url: URL): JsonRecord { const payload = tracePayload(traceId, url); + if (state.scenarioId === "tool-completed-projection-running" && traceId === "trc_running") { + return { ok: true, status: "ok", contractVersion: "workbench-read-model-v1", traceId, sessionId: payload.sessionId ?? null, threadId: payload.threadId ?? null, traceStatus: "completed", events: payload.events, eventCount: payload.eventCount, hasMore: payload.hasMore, nextSeq: payload.nextSinceSeq, range: payload.range, fullTraceLoaded: payload.fullTraceLoaded, projectionStatus: "projecting", terminalEvidence: null, finalResponse: null, traceSummary: payload.traceSummary, retention: payload.retention }; + } return { ok: true, status: "ok", contractVersion: "workbench-read-model-v1", traceId, sessionId: payload.sessionId ?? null, threadId: payload.threadId ?? null, traceStatus: payload.status, events: payload.events, eventCount: payload.eventCount, hasMore: payload.hasMore, nextSeq: payload.nextSinceSeq, range: payload.range, fullTraceLoaded: payload.fullTraceLoaded, terminalEvidence: payload.terminalEvidence, finalResponse: payload.finalResponse, traceSummary: payload.traceSummary, retention: payload.retention }; } diff --git a/web/hwlab-cloud-web/src/api/workbench.ts b/web/hwlab-cloud-web/src/api/workbench.ts index cfa5b523..b80c0fcd 100644 --- a/web/hwlab-cloud-web/src/api/workbench.ts +++ b/web/hwlab-cloud-web/src/api/workbench.ts @@ -91,7 +91,8 @@ function normalizeWorkbenchTurnResult(result: ApiResult> if (!result.ok || !result.data) return result as ApiResult; const turn = recordValue(result.data.turn) ?? result.data; const trace = recordValue(turn.trace); - const status = canonicalWorkbenchStatus(textValue(turn.status) ?? textValue(result.data.status) ?? "unknown", textValue(trace?.status, result.data.traceStatus)); + const envelopeStatus = textValue(result.data.status); + const status = normalizeStatus(textValue(turn.status) ?? (envelopeStatus && !["ok", "succeeded", "found"].includes(envelopeStatus) ? envelopeStatus : null)) ?? "unknown"; const terminal = turn.terminal === true || isTerminalStatus(status); return { ...result, @@ -148,13 +149,6 @@ function normalizeWorkbenchTraceResult(result: ApiResult }; } -function canonicalWorkbenchStatus(primary: string | null, traceStatus: string | null): string { - const primaryStatus = normalizeStatus(primary); - const trace = normalizeStatus(traceStatus); - if (trace && isTerminalStatus(trace) && (!primaryStatus || isActiveStatus(primaryStatus) || !isTerminalStatus(primaryStatus))) return trace; - return primaryStatus ?? trace ?? "unknown"; -} - function normalizeStatus(value: unknown): string | null { const text = textValue(value); if (!text) return null; diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index f3b2b130..fdc880b8 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -729,7 +729,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { if (!id || !snapshot) return; if (!shouldApplyActiveTraceAuthority(id, traceResultSessionId(snapshot))) return; applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, snapshot)); - if (isTerminalMessageStatus(snapshot.status)) void refreshTerminalTraceFromRest(id, "realtime-trace-snapshot"); + if (traceSnapshotHasTerminalEvidence(snapshot)) void refreshTerminalTraceFromRest(id, "realtime-trace-snapshot"); } function applyRealtimeTraceEvent(traceId: string | null | undefined, event: WorkbenchRealtimeEvent["event"], snapshot: WorkbenchRealtimeEvent["snapshot"], realtimeEvent?: WorkbenchRealtimeEvent | null): void { @@ -740,7 +740,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const events = event ? [event] : Array.isArray(snapshot?.events) ? snapshot.events : []; markWorkbenchTraceEventsReceived({ traceId: id, events, transport: "sse", serverSentAt: realtimeEvent?.serverSentAt, eventCreatedAt: realtimeEvent?.eventCreatedAt, traceSeq: realtimeEvent?.traceSeq ?? realtimeEvent?.cursor?.traceSeq }); applyTraceSnapshot(id, realtimeSnapshotToTraceSnapshot(id, snapshot ?? { traceId: id, status: event?.status, events }, events)); - if (event?.terminal === true || isTerminalMessageStatus(event?.status) || isTerminalMessageStatus(snapshot?.status)) void refreshTerminalTraceFromRest(id, "realtime-trace-event"); + if (traceEventHasTerminalEvidence(event) || traceSnapshotHasTerminalEvidence(snapshot)) void refreshTerminalTraceFromRest(id, "realtime-trace-event"); } function applyRealtimeTurnSnapshot(turn: Record): void { @@ -836,16 +836,20 @@ export const useWorkbenchStore = defineStore("workbench", () => { } function applyTraceSnapshot(traceId: string, snapshot: TraceSnapshot): void { - const trace = snapshotToRunnerTrace(snapshot); + const turn = turnStatusAuthority.value[traceId]; + const trace = snapshotToRunnerTrace({ + ...snapshot, + status: firstNonEmptyString(turn?.status, snapshot.status) ?? snapshot.status, + traceStatus: firstNonEmptyString(snapshot.traceStatus, snapshot.status) ?? undefined + }); const authoritySessionId = traceResultSessionId(trace) ?? traceResultSessionId(snapshot); if (!shouldApplyActiveTraceAuthority(traceId, authoritySessionId)) return; - rememberTurnStatus(traceId, snapshot); - const status = statusFromResult(snapshot.status); - const terminal = isTerminalMessageStatus(status); updateActiveMessages((source) => source.map((message) => { if (!shouldApplyTraceToMessage(message, traceId, authoritySessionId)) return message; const runnerTrace = mergeRunnerTrace(message.runnerTrace, trace); rememberTraceAuthority(runnerTrace); + const status = statusFromResult(firstNonEmptyString(turn?.status, message.status) ?? message.status); + const terminal = turn?.terminal === true || isTerminalMessageStatus(turn?.status); const traceAssistantText = message.role === "agent" ? assistantTextFromTraceEvents(Array.isArray(runnerTrace.events) ? runnerTrace.events : []) : null; const error = message.role === "agent" ? normalizeAgentError(runnerTrace.error ?? message.error) : normalizeAgentError(message.error); const errorText = message.role === "agent" ? agentErrorDisplayText(error) : null; @@ -1257,6 +1261,19 @@ function isTerminalMessageStatus(value: unknown): boolean { return ["completed", "failed", "blocked", "timeout", "canceled", "cancelled", "stale", "thread-resume-failed"].includes(String(value ?? "").trim().toLowerCase().replace(/_/gu, "-")); } +function traceEventHasTerminalEvidence(event: unknown): boolean { + const record = recordValue(event); + return record?.terminal === true || record?.final === true || record?.replyAuthority === true; +} + +function traceSnapshotHasTerminalEvidence(snapshot: unknown): boolean { + const record = recordValue(snapshot); + if (!record) return false; + if (record.terminal === true || record.terminalEvidence || record.finalResponse) return true; + const events = Array.isArray(record.events) ? record.events : []; + return events.some((event) => traceEventHasTerminalEvidence(event)); +} + function recordValue(value: unknown): Record | null { return value && typeof value === "object" ? value as Record : null; } diff --git a/web/hwlab-cloud-web/tests/workbench-e2e/specs/event-replay.spec.ts b/web/hwlab-cloud-web/tests/workbench-e2e/specs/event-replay.spec.ts index 76f929a6..c6915a5d 100644 --- a/web/hwlab-cloud-web/tests/workbench-e2e/specs/event-replay.spec.ts +++ b/web/hwlab-cloud-web/tests/workbench-e2e/specs/event-replay.spec.ts @@ -43,6 +43,50 @@ test.describe("terminal completed without final response", () => { }); }); +test.describe("non-terminal tool completed event", () => { + test.use({ scenarioId: "tool-completed-projection-running" }); + + test("keeps turn, session card, and trace header running", async ({ page }, testInfo) => { + await gotoWorkbench(page, "/workbench/sessions/ses_running"); + + const turn = await page.request.get("/v1/workbench/turns/trc_running"); + expect(turn.status()).toBe(200); + const turnBody = await turn.json(); + expect(turnBody.turn.status).toBe("running"); + expect(turnBody.turn.trace.status).toBe("completed"); + + const trace = await page.request.get("/v1/workbench/traces/trc_running/events"); + expect(trace.status()).toBe(200); + const traceBody = await trace.json(); + expect(traceBody.traceStatus).toBe("completed"); + expect(traceBody.events.at(-1).status).toBe("completed"); + expect(traceBody.events.at(-1).terminal).toBe(false); + + const tab = page.locator(sessionTab("ses_running")); + await expect(tab).toHaveAttribute("data-running", "true"); + await expect(tab).toHaveAttribute("data-status", "running"); + + const agentCard = page.locator(`${selectors.messageCard}[data-role="agent"]`).last(); + await expect(agentCard).toHaveAttribute("data-status", "running"); + const timeline = agentCard.locator(`${selectors.traceTimeline}[data-status="running"]`); + await expect(timeline).toBeVisible(); + await timeline.locator("summary.trace-disclosure-summary").click(); + const toolRow = timeline.locator(`${selectors.traceRow}[data-row-kind="tool"]`).filter({ hasText: "commandExecution" }); + await expect(toolRow).toHaveAttribute("data-terminal", "false"); + await expect(toolRow).toContainText("ls -ld ."); + + const snapshot = await agentCard.evaluate((element) => ({ + status: element.getAttribute("data-status"), + messageTexts: Array.from(element.querySelectorAll(".message-markdown.message-text")).map((item) => (item.textContent ?? "").trim()).filter(Boolean), + bodyText: document.body.textContent ?? "" + })); + expect(snapshot.status).toBe("running"); + expect(snapshot.messageTexts).toEqual([]); + expect(snapshot.bodyText).not.toContain("Code Agent 已完成，但没有返回可展示的 final response。"); + await saveScreenshot(page, testInfo, "tool-completed-projection-running"); + }); +}); + test("fresh deep link replays completed turn from the session authority", async ({ page }, testInfo) => { const detail = await page.request.get("/v1/workbench/sessions/ses_completed"); expect(detail.status()).toBe(200);