diff --git a/internal/cloud/server-workbench-http.test.ts b/internal/cloud/server-workbench-http.test.ts index 34691cf9..27ffbb74 100644 --- a/internal/cloud/server-workbench-http.test.ts +++ b/internal/cloud/server-workbench-http.test.ts @@ -1522,7 +1522,7 @@ test("workbench trace events API blocks historical projectedSeq collisions inste } }); -test("workbench realtime stream accepts session authority without project or workspace", async () => { +test("workbench realtime stream surfaces facts blocker instead of legacy trace fallback", async () => { const traceStore = createCodeAgentTraceStore(); const results = createCodeAgentChatResultStore(); const traceId = "trc_workbench_realtime"; @@ -1538,8 +1538,8 @@ test("workbench realtime stream accepts session authority without project or wor updatedAt: "2026-06-17T02:00:00.000Z", session: { sessionStatus: "running", lastTraceId: traceId, messages: [{ role: "user", text: "stream", traceId }] } }; - traceStore.append(traceId, { type: "request", status: "accepted", label: "request:accepted" }); - results.set(traceId, { status: "running", traceId, ownerUserId: ACTOR.id, sessionId: session.id, threadId: session.threadId }); + traceStore.append(traceId, { type: "assistant", status: "completed", label: "assistant:completed", terminal: true, message: "legacy trace answer" }); + results.set(traceId, { status: "completed", traceId, ownerUserId: ACTOR.id, sessionId: session.id, threadId: session.threadId, finalResponse: { text: "legacy result answer" } }); const accessController = { store: { async getAgentSession(sessionId) { return sessionId === session.id ? session : null; }, @@ -1554,18 +1554,24 @@ test("workbench realtime stream accepts session authority without project or wor try { const { port } = server.address(); const eventsPromise = getSseEvents(port, `/v1/workbench/events?sessionId=${encodeURIComponent(session.id)}`, 4); - setTimeout(() => traceStore.append(traceId, { type: "backend", status: "running", label: "backend:live" }), 50); const events = await eventsPromise; - assert.deepEqual(events.slice(0, 3).map((event) => event.event), [ + assert.deepEqual(events.map((event) => event.event), [ "workbench.connected", + "workbench.error", "workbench.trace.snapshot", "workbench.turn.snapshot" ]); assert.equal(events[0].data.filters.sessionId, session.id); - assert.equal(events[1].data.traceId, traceId); - assert.equal(events[2].data.turn.traceId, traceId); - assert.equal(events[3].event, "workbench.trace.event"); - assert.equal(events[3].data.event.label, "backend:live"); + assert.equal(events[1].data.error.code, "workbench_facts_session_missing"); + assert.equal(events[2].data.traceId, traceId); + assert.equal(events[2].data.snapshot.status, "unknown"); + assert.equal(events[2].data.snapshot.eventCount, 0); + assert.equal(events[3].data.turn.traceId, traceId); + assert.equal(events[3].data.turn.status, "unknown"); + assert.equal(events[3].data.turn.terminal, false); + assert.equal(events[3].data.turn.finalResponse, null); + assert.equal(JSON.stringify(events).includes("legacy result answer"), false); + assert.equal(JSON.stringify(events).includes("legacy trace answer"), false); } finally { await new Promise((resolve, reject) => server.close((error) => error ? reject(error) : resolve())); } diff --git a/internal/cloud/server-workbench-http.ts b/internal/cloud/server-workbench-http.ts index a8150893..b845017f 100644 --- a/internal/cloud/server-workbench-http.ts +++ b/internal/cloud/server-workbench-http.ts @@ -2221,14 +2221,16 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, } const sessionId = safeSessionId(context.result?.sessionId ?? context.result?.session?.sessionId ?? context.session?.id) ?? null; const threadId = safeOpaqueId(context.result?.threadId ?? context.result?.session?.threadId ?? context.session?.threadId) ?? (textValue(context.session?.threadId) || null); + const realtimeTraceSnapshot = context.factsReadBlocker ? blockedRealtimeTraceSnapshot(context) : traceSnapshotForRealtime(context.trace); + const traceSeq = context.factsReadBlocker ? 0 : traceSnapshotLastSeq(context.trace); writeEvent("workbench.trace.snapshot", { type: "trace.snapshot", reason, sessionId, threadId, traceId, - snapshot: { ...traceSnapshotForRealtime(context.trace), sessionId, threadId }, - cursor: { traceSeq: traceSnapshotLastSeq(context.trace) } + snapshot: { ...realtimeTraceSnapshot, sessionId, threadId }, + cursor: { traceSeq } }); writeEvent("workbench.turn.snapshot", { type: "turn.snapshot", @@ -2237,21 +2239,12 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId, threadId, traceId, turn: realtimeTurnSnapshot(context), - cursor: { traceSeq: traceSnapshotLastSeq(context.trace) } + cursor: { traceSeq } }); } function realtimeTurnSnapshot(context) { - if (context?.factsReadBlocker) { - return { - ...turnSnapshot(context), - projectionStatus: "blocked", - projectionHealth: "degraded", - blocker: context.factsReadBlocker, - projection: blockedTraceEventProjection(context.projection ?? {}, context.factsReadBlocker), - valuesRedacted: true - }; - } + if (context?.factsReadBlocker) return blockedRealtimeTurnSnapshot(context, context.factsReadBlocker); const facts = context?.facts; const traceId = safeTraceId(context?.traceId); if (traceId && facts && context?.factSession) { @@ -2259,7 +2252,76 @@ function realtimeTurnSnapshot(context) { const projected = factTurnSnapshot({ turn: factTurn, session: context.factSession, facts, traceId, turnId: context.turnId }); if (projected) return projected; } - return turnSnapshot(context); + const blocker = traceEventsReadModelBlocker("workbench_facts_turn_missing", "Workbench durable facts did not contain a turn snapshot; realtime must surface the read-model gap instead of falling back to legacy trace projection.", { traceId, session: context?.factSession, route: "/v1/workbench/events" }); + return blockedRealtimeTurnSnapshot(context, blocker); +} + +function blockedRealtimeTraceSnapshot(context) { + const traceId = safeTraceId(context?.traceId ?? context?.factsReadBlocker?.traceId) ?? null; + const blocker = context?.factsReadBlocker ?? traceEventsReadModelBlocker("workbench_facts_unavailable", "Workbench durable facts are unavailable for realtime trace snapshot.", { traceId, route: "/v1/workbench/events" }); + const projection = blockedTraceEventProjection({}, blocker); + return { + traceId, + status: "unknown", + traceStatus: "unknown", + events: [], + eventCount: 0, + fullTraceLoaded: true, + hasMore: false, + nextProjectedSeq: 0, + updatedAt: null, + projection, + projectionStatus: projection.projectionStatus, + projectionHealth: projection.projectionHealth, + staleMs: projection.staleMs ?? null, + blocker, + valuesRedacted: true, + secretMaterialStored: false + }; +} + +function blockedRealtimeTurnSnapshot(context, blocker) { + const traceId = safeTraceId(context?.traceId ?? blocker?.traceId) ?? null; + const turnId = safeTurnId(context?.turnId ?? blocker?.turnId) || traceId; + const sessionId = safeSessionId(blocker?.sessionId ?? factSessionId(context?.factSession) ?? context?.session?.id) ?? null; + const threadId = safeOpaqueId(context?.factSession?.threadId ?? context?.session?.threadId) ?? (textValue(context?.session?.threadId) || null); + const projection = blockedTraceEventProjection({}, blocker); + return { + turnId, + traceId, + status: "unknown", + running: false, + terminal: false, + sessionId, + threadId, + userMessageId: null, + assistantMessageId: null, + assistantText: null, + finalResponse: null, + timing: null, + startedAt: null, + lastEventAt: null, + finishedAt: null, + durationMs: null, + agentRun: null, + trace: { + traceId, + status: "unknown", + eventCount: 0, + updatedAt: null + }, + urls: { + self: turnId ? `/v1/workbench/turns/${encodeURIComponent(turnId)}` : null, + traceEvents: traceId ? `/v1/workbench/traces/${encodeURIComponent(traceId)}/events` : null + }, + projectionStatus: projection.projectionStatus, + projectionHealth: projection.projectionHealth, + staleMs: projection.staleMs ?? null, + blocker, + projection, + valuesRedacted: true, + secretMaterialStored: false + }; } async function writeTraceRealtimeSnapshotSafe(input) { @@ -2305,6 +2367,15 @@ async function visibleTraceContext(options, actor, traceId) { } } const factSession = factArray(facts.sessions).find((item) => factSessionId(item) === factSessionIdValue) ?? null; + if (!factsReadBlocker) { + if (!factSessionIdValue) { + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_id_missing", "Workbench realtime could not resolve a durable session id; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" }); + } else if (typeof readModel.queryFacts !== "function") { + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_query_unavailable", "Workbench durable facts query is unavailable; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" }); + } else if (!factSession) { + factsReadBlocker = traceEventsReadModelBlocker("workbench_facts_session_missing", "Workbench durable facts did not contain the trace session; realtime must surface the read-model gap instead of using legacy trace projection.", { traceId, projection, session, route: "/v1/workbench/events" }); + } + } return { visible: true, turnId: projection.turnId, traceId, status: projection.status, projection, result, session, trace, facts, factSession, factsReadBlocker }; } diff --git a/web/hwlab-cloud-web/scripts/check.ts b/web/hwlab-cloud-web/scripts/check.ts index 4285c2c6..c005e1b0 100644 --- a/web/hwlab-cloud-web/scripts/check.ts +++ b/web/hwlab-cloud-web/scripts/check.ts @@ -64,6 +64,8 @@ const appSource = readCloudWebAppSource(rootDir); const workbenchStoreSource = readWeb("src/stores/workbench.ts"); const traceTimelineSource = readWeb("src/components/agent/TraceTimeline.vue"); const messageTraceDebugSource = readWeb("src/components/workbench/MessageTraceDebugPanel.vue"); +const conversationPanelSource = readWeb("src/components/workbench/ConversationPanel.vue"); +const serverWorkbenchHttpSource = fs.readFileSync(path.resolve(rootDir, "..", "..", "internal/cloud/server-workbench-http.ts"), "utf8"); assert.match(html, /
<\/div>/u, "index.html must expose Vue mount root"); assert.match(html, /src="\/src\/main\.ts"/u, "index.html must load Vue TS entry"); @@ -102,6 +104,11 @@ assertIncludes(appSource, "activityRef", "Code Agent inactivity-timeout activity assertIncludes(workbenchStoreSource, "connectWorkbenchEvents", "Workbench must use the unified SSE realtime entry"); assert.doesNotMatch(workbenchStoreSource, /scheduleRealtimeGapHydration|hydrateRealtimeGap/u, "Workbench realtime consumer must not repair projection through REST gap fill"); assert.doesNotMatch(workbenchStoreSource, /subscribeToTrace|TRACE_POLL_INTERVAL_MS/u, "Workbench store must not reintroduce active trace polling"); +assert.doesNotMatch(workbenchStoreSource, /message\.runnerTrace(?:\?\.|\.)status/u, "Workbench message lifecycle must not be inferred from runnerTrace.status"); +assert.doesNotMatch(conversationPanelSource, /message\.runnerTrace(?:\?\.|\.)status/u, "ConversationPanel must not override message completion from runnerTrace.status"); +assert.doesNotMatch(serverWorkbenchHttpSource, /return\s+turnSnapshot\(context\)|\.\.\.turnSnapshot\(context\)|blockedTraceEventProjection\(context\.projection/u, "Realtime turn snapshots must not fall back to legacy trace/result projection"); +assertIncludes(serverWorkbenchHttpSource, "blockedRealtimeTurnSnapshot", "Realtime read-model gaps must produce blocked diagnostic turn snapshots"); +assertIncludes(serverWorkbenchHttpSource, "workbench_facts_session_missing", "Realtime read-model gaps must expose a facts-missing blocker"); assertIncludes(appSource, "/v1/workbench/events", "Workbench realtime client must use the RESTful same-origin events endpoint"); assertIncludes(appSource, "/v1/workbench/traces/", "trace hydration must use Workbench read-model trace API"); assert.doesNotMatch(appSource, /\/v1\/agent\/(?:turns|traces|chat\/result)\//u, "Cloud Web must not call legacy Code Agent read-through turn/trace/result APIs"); diff --git a/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue b/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue index ac67a294..19e46963 100644 --- a/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue +++ b/web/hwlab-cloud-web/src/components/workbench/ConversationPanel.vue @@ -36,8 +36,7 @@ const scrollSignature = computed(() => workbench.activeMessages.map((message) => message.status, visibleMessageText(message).length, message.updatedAt ?? "", - message.runnerTrace?.eventCount ?? message.runnerTrace?.events?.length ?? 0, - message.runnerTrace?.status ?? "" + message.runnerTrace?.eventCount ?? message.runnerTrace?.events?.length ?? 0 ].join(":"))); watch(scrollSignature, async () => { @@ -71,7 +70,7 @@ function isTerminalAgentMessage(message: ChatMessage): boolean { function isCompletedAgentMessage(message: ChatMessage): boolean { if (message.role !== "agent") return false; - return [message.status, message.runnerTrace?.status].some((value) => String(value ?? "").trim().toLowerCase().replace(/_/gu, "-") === "completed"); + return String(message.status ?? "").trim().toLowerCase().replace(/_/gu, "-") === "completed"; } function hasSealedCompletedText(message: ChatMessage): boolean { @@ -238,10 +237,7 @@ function traceAutoExpanded(message: ChatMessage): boolean | null { } function traceForDisplay(message: ChatMessage): ChatMessage["runnerTrace"] { - if (!message.runnerTrace) return message.runnerTrace; - const status = firstNonEmptyString(message.status, message.runnerTrace.status); - if (!status || status === message.runnerTrace.status) return message.runnerTrace; - return { ...message.runnerTrace, status }; + return message.runnerTrace; } function messageDurationMeta(message: ChatMessage): { text: string; label: string } | null { diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index caaeed44..551fbfa7 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -582,8 +582,8 @@ export const useWorkbenchStore = defineStore("workbench", () => { const turn = turnStatusAuthority.value[id]; if (turn?.terminal === true || isTerminalMessageStatus(turn?.status)) return false; const message = [...messages.value].reverse().find((item) => firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === id) ?? null; - if (isTerminalMessageStatus(message?.status) || isTerminalMessageStatus(message?.runnerTrace?.status)) return false; - return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status) || isTraceActiveStatus(message?.runnerTrace?.status); + if (isTerminalMessageStatus(message?.status)) return false; + return isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status); } async function refreshSessionMessageProjectionPage(sessionId: string | null | undefined): Promise { @@ -665,7 +665,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { function messageHasActiveTurnStatusEvidence(message: ChatMessage, traceId: string): boolean { if (currentRequest.value?.traceId === traceId) return true; - return isTraceActiveStatus(message.status) || isTraceActiveStatus(message.runnerTrace?.status); + return isTraceActiveStatus(message.status); } function traceNeedsTurnStatusRefresh(traceId: string | null | undefined, source: ChatMessage[] = messages.value): boolean { @@ -674,8 +674,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const turn = turnStatusAuthority.value[id]; if (turn?.terminal === true || isTerminalMessageStatus(turn?.status)) return false; const message = [...source].reverse().find((item) => firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === id) ?? null; - if (isTerminalMessageStatus(message?.status) || isTerminalMessageStatus(message?.runnerTrace?.status)) return false; - if (message?.runnerTrace && traceSnapshotHasTerminalEvidence(message.runnerTrace)) return false; + if (isTerminalMessageStatus(message?.status)) return false; return true; } @@ -986,12 +985,12 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project markWorkbenchTraceEventsReceived({ traceId, events, transport: "rest_gap" }); updateSessionMessages(ownerSessionId, (source) => source.map((message) => { if (!messageMatchesTraceAuthority(message, traceId, authoritySessionId, ownerSessionId)) return message; - const displayStatus = firstNonEmptyString(message.status, message.runnerTrace?.status, result.status, result.runnerTrace?.status); + const traceDetailStatus = firstNonEmptyString(result.traceStatus, result.runnerTrace?.traceStatus, result.runnerTrace?.status); const { traceId: _resultTraceId, sessionId: _resultSessionId, threadId: _resultThreadId, ...resultTraceRest } = result.runnerTrace ?? {}; const nextTrace = { ...resultTraceRest, traceId: optionalString(result.traceId, traceId), - status: displayStatus ?? undefined, + status: traceDetailStatus ?? undefined, sessionId: optionalString(authoritySessionId, message.runnerTrace?.sessionId), threadId: optionalString(result.threadId, message.runnerTrace?.threadId), events, @@ -1422,8 +1421,8 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project function messageHasSealedTerminalResult(message: ChatMessage | null): boolean { if (!message) return false; - if (isTerminalMessageStatus(message.status) || isTerminalMessageStatus(message.runnerTrace?.status)) return true; - return message.runnerTrace ? traceSnapshotHasTerminalEvidence(message.runnerTrace) : false; + if (!isTerminalMessageStatus(message.status)) return false; + return Boolean(firstNonEmptyString(message.text, finalResponseText((message as Record).finalResponse))); } function failTrace(traceId: string, message: string): void { @@ -1499,7 +1498,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project function isRunningAgentMessageForTiming(message: ChatMessage): boolean { if (message.role !== "agent") return false; - const status = normalizedStatusText(firstNonEmptyString(message.status, message.runnerTrace?.status)); + const status = normalizedStatusText(message.status); if (status && isTerminalMessageStatus(status)) return false; return status === null || isTraceActiveStatus(status) || message.traceAutoLifecycle === "running"; } @@ -1713,7 +1712,7 @@ function activeTraceIdFromMessages(messages: ChatMessage[], turnStatusAuthority: for (const message of [...messages].reverse()) { if (message.role !== "agent") continue; if (messageHasCompletedFinalResponse(message)) continue; - if (isTerminalMessageStatus(message.status) || isTerminalMessageStatus(message.runnerTrace?.status)) continue; + if (isTerminalMessageStatus(message.status)) continue; const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId); if (!traceId) continue; const turn = turnStatusAuthority[traceId]; @@ -2095,7 +2094,7 @@ function traceHasCompletedFinalResponse(traceId: string | null | undefined, sour function messageHasCompletedFinalResponse(message: ChatMessage | null | undefined): boolean { if (!message || message.role !== "agent") return false; - if (normalizedStatusText(message.status) !== "completed" && normalizedStatusText(message.runnerTrace?.status) !== "completed") return false; + if (normalizedStatusText(message.status) !== "completed") return false; return Boolean(firstNonEmptyString(message.text, finalResponseText((message as Record).finalResponse))); }