Merge pull request #2507 from pikasTech/fix/2486-refresh-identity

fix: 统一纯 Kafka 刷新前后的消息身份

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 },

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

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",

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" };