diff --git a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts index 62a1755d..06a9338d 100644 --- a/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts +++ b/web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts @@ -20,7 +20,7 @@ import { checkWorkbenchHealth, createWorkbenchHealthProbeCache } from "../src/ut import { messageDiagnosticView } from "../src/utils/workbench-error-runtime.ts"; import { WORKBENCH_TIMELINE_OPENCODE_PARITY, buildWorkbenchTimelineRows, normalizeWorkbenchTimelineMessages, workbenchTimelineSignature } from "../src/stores/workbench-timeline-model.ts"; import { reduceWorkbenchRealtimeEvent, workbenchRealtimeEventIsBusinessActivity } from "../src/stores/workbench-event-reducer.ts"; -import { reduceWorkbenchLiveKafkaMessageState, workbenchLiveKafkaMessageId } from "../src/stores/workbench-live-kafka-event.ts"; +import { reduceWorkbenchLiveKafkaMessageState, workbenchAgentMessageIdForTrace } from "../src/stores/workbench-live-kafka-event.ts"; import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery } from "../src/stores/workbench-realtime-plan.ts"; import { WORKBENCH_REALTIME_AUTHORITY_VERSION, workbenchRealtimePrimaryAuthorityDecision, workbenchSyncReplayEvents } from "../src/stores/workbench-realtime-authority.ts"; import { cleanupWorkbenchServerStateDroppedSessions, cleanupWorkbenchServerStateSessions, createWorkbenchServerState, reduceWorkbenchServerState } from "../src/stores/workbench-server-state.ts"; @@ -476,7 +476,7 @@ test("live hwlab envelope projects assistant, tool/output, and terminal without const terminalEnvelope = envelope({ type: "result", eventType: "terminal", sourceEventId: "evt_terminal", traceId: "trc_live_web", sessionId: "ses_live_web", status: "completed", terminal: true }); visible = reduceWorkbenchLiveKafkaMessageState(visible, terminalEnvelope.event); assert.deepEqual(visible, { text: "running increment", status: "completed", terminal: true }); - assert.equal(workbenchLiveKafkaMessageId("trc_live_web"), "msg_live_trc_live_web"); + assert.equal(workbenchAgentMessageIdForTrace("trc_live_web"), "msg_live_web_agent"); const failed = reduceWorkbenchLiveKafkaMessageState( { text: "partial response", status: "running", terminal: false }, diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts index 2e6d386b..f3a2e022 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.test.ts @@ -2,7 +2,7 @@ import assert from "node:assert/strict"; import { test } from "bun:test"; import type { ChatMessage, TraceEvent } from "../types"; -import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, workbenchLiveKafkaEnvelope, workbenchLiveKafkaEventIsUserMessage, workbenchLiveKafkaProjectionTarget, workbenchUserMessageIdForTrace } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, workbenchAgentMessageIdForTrace, workbenchLiveKafkaEnvelope, workbenchLiveKafkaEventIsUserMessage, workbenchLiveKafkaProjectionTarget, workbenchUserMessageIdForTrace } from "./workbench-live-kafka-event"; import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, type WorkbenchServerState } from "./workbench-server-state"; test("only canonical HWLAB Kafka envelopes bypass terminal priority", () => { @@ -97,6 +97,61 @@ test("HTTP-first and Kafka-first admission converge on one stable optimistic use } }); +test("optimistic live and retained Kafka assistant rows share the trace-derived identity", () => { + const sessionId = "ses_2947a007-1823-41bf-aa94-d89b468f74a7"; + const traceId = "trc_mrgpndrwsk0edt"; + const messageId = workbenchAgentMessageIdForTrace(traceId); + assert.equal(messageId, "msg_mrgpndrwsk0edt_agent"); + const optimistic = { + id: messageId, + messageId, + role: "agent", + title: "Code Agent", + text: "", + status: "running", + traceId, + sessionId, + createdAt: "2026-07-11T18:51:00.000Z", + updatedAt: "2026-07-11T18:51:00.000Z" + } as ChatMessage; + const event = traceEvent(1, "assistant", { traceId, sessionId, assistantText: "retained reply" }); + const live = projectWorkbenchLiveKafkaMessage({ previous: optimistic, traceId, sessionId, event, receivedAt: "2026-07-11T18:51:01.000Z" }); + const replay = projectWorkbenchLiveKafkaMessage({ previous: null, traceId, sessionId, event, receivedAt: "2026-07-11T18:52:01.000Z" }); + + assert.equal(live.id, messageId); + assert.equal(live.messageId, messageId); + assert.equal(replay.id, messageId); + assert.equal(replay.messageId, messageId); + assert.equal(live.id, replay.id); +}); + +test("Kafka assistant projection replaces a pre-contract placeholder with the trace identity", () => { + const sessionId = "ses_live_agent_identity"; + const traceId = "trc_live_agent_identity"; + const previous = { + id: "msg_random_placeholder", + messageId: "msg_random_placeholder", + role: "agent", + title: "Code Agent", + text: "", + status: "running", + traceId, + sessionId, + createdAt: "2026-07-11T18:51:00.000Z", + updatedAt: "2026-07-11T18:51:00.000Z" + } as ChatMessage; + const projected = projectWorkbenchLiveKafkaMessage({ + previous, + traceId, + sessionId, + event: traceEvent(1, "tool", { traceId, sessionId }), + receivedAt: "2026-07-11T18:51:01.000Z" + }); + + assert.equal(projected.id, "msg_live_agent_identity_agent"); + assert.equal(projected.messageId, "msg_live_agent_identity_agent"); +}); + test("session replay keeps steer user input separate from the target agent lifecycle", () => { const sessionId = "ses_live_session_sequence"; const targetTraceId = "trc_live_target"; diff --git a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts index e5e2090d..679ecd43 100644 --- a/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts +++ b/web/hwlab-cloud-web/src/stores/workbench-live-kafka-event.ts @@ -40,17 +40,21 @@ export function reduceWorkbenchLiveKafkaMessageState(previous: WorkbenchLiveMess }; } -export function workbenchLiveKafkaMessageId(traceId: string): string { - return `msg_live_${traceId}`; +export function workbenchUserMessageIdForTrace(traceId: string): string { + return workbenchMessageIdForTrace(traceId, "user"); } -export function workbenchUserMessageIdForTrace(traceId: string): string { +export function workbenchAgentMessageIdForTrace(traceId: string): string { + return workbenchMessageIdForTrace(traceId, "agent"); +} + +function workbenchMessageIdForTrace(traceId: string, role: "user" | "agent"): string { const suffix = firstNonEmptyString(traceId) ?.replace(/^trc_/u, "") .replace(/[^A-Za-z0-9_.:-]/gu, "_") .slice(0, 48); - if (!suffix) throw new Error("traceId must produce a stable Workbench user message identity"); - return `msg_${suffix}_user`; + if (!suffix) throw new Error("traceId must produce a stable Workbench message identity"); + return `msg_${suffix}_${role}`; } export function workbenchLiveKafkaEnvelope(schema: unknown): boolean { @@ -165,10 +169,10 @@ export function projectWorkbenchLiveKafkaMessage(input: WorkbenchLiveKafkaProjec durationMs, updatedAt: receivedAt }; - const messageId = previous?.messageId ?? previous?.id ?? workbenchLiveKafkaMessageId(input.traceId); + const messageId = workbenchAgentMessageIdForTrace(input.traceId); return { ...(previous ?? {}), - id: previous?.id ?? messageId, + id: messageId, messageId, role: "agent", title: input.title ?? previous?.title ?? "Code Agent", diff --git a/web/hwlab-cloud-web/src/stores/workbench.ts b/web/hwlab-cloud-web/src/stores/workbench.ts index 9dcb5e33..b1fb746d 100644 --- a/web/hwlab-cloud-web/src/stores/workbench.ts +++ b/web/hwlab-cloud-web/src/stores/workbench.ts @@ -22,7 +22,7 @@ import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection"; import { cleanupWorkbenchServerStateSessions, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state"; import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache"; import { reduceWorkbenchRealtimeEvent, workbenchRealtimeEventIsBusinessActivity, type WorkbenchRealtimeAction } from "./workbench-event-reducer"; -import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, workbenchLiveKafkaEnvelope, workbenchLiveKafkaProjectionTarget, workbenchUserMessageIdForTrace } from "./workbench-live-kafka-event"; +import { projectWorkbenchLiveKafkaMessage, projectWorkbenchLiveKafkaUserMessage, workbenchAgentMessageIdForTrace, workbenchLiveKafkaEnvelope, workbenchLiveKafkaProjectionTarget, workbenchUserMessageIdForTrace } from "./workbench-live-kafka-event"; import { workbenchHistoryAuthorityPolicy } from "./workbench-kafka-refresh-policy"; import { messageHasSealedTerminalResult, messageIsSealedTerminal, traceAuthorityIsSealed } from "./workbench-terminal-authority"; import { boundedProjectionMessageLimit, mergeBoundedProjectionMessages, selectProjectionMessageWindow, traceProjectionIsTerminalSealed } from "./workbench-message-projection-budget"; @@ -764,6 +764,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const traceId = steerMode && composer.value.targetTraceId ? composer.value.targetTraceId : nextProtocolId("trc"); const steerTraceId = steerMode ? nextProtocolId("trc_steer") : null; const userMessageId = workbenchUserMessageIdForTrace(steerTraceId ?? traceId); + const agentMessageId = workbenchAgentMessageIdForTrace(traceId); const sessionId = composer.value.sessionId; if (realtimeCapabilities.liveKafkaSse && liveRealtimeReadySessionId.value !== sessionId) { error.value = "realtime_connecting"; @@ -776,7 +777,7 @@ export const useWorkbenchStore = defineStore("workbench", () => { const previousChatPending = chatPending.value; if (!steerMode) { const user = makeMessage("user", value, "sent", { id: userMessageId, messageId: userMessageId, traceId, sessionId, threadId, title: "用户", createdAt: submittedAt, updatedAt: submittedAt }); - const pending = makeMessage("agent", "", "running", { traceId, sessionId, threadId, title: "Code Agent", retryInput: value, traceAutoLifecycle: "running", createdAt: submittedAt, updatedAt: submittedAt, ...optimisticRunningTimingPatch(submittedAt) }); + const pending = makeMessage("agent", "", "running", { id: agentMessageId, messageId: agentMessageId, traceId, sessionId, threadId, title: "Code Agent", retryInput: value, traceAutoLifecycle: "running", createdAt: submittedAt, updatedAt: submittedAt, ...optimisticRunningTimingPatch(submittedAt) }); appendActiveMessages(user, pending); projectOptimisticRunningTurn({ sessionId, threadId, traceId, userText: value }); currentRequest.value = { traceId, sessionId, threadId, status: "running" };