Merge pull request #1541 from pikasTech/fix/1529-workbench-session-authority

fix(workbench): guard cross-session realtime projection

internal/cloud/server-workbench-http.ts

@@ -112,6 +112,8 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
 
   const initialSession = requestedSessionId ? await readModel.getSessionById(requestedSessionId) : null;
   const initialSessionSnapshot = objectValue(initialSession?.session);
+  const streamSessionId = safeSessionId(initialSession?.id ?? requestedSessionId) ?? null;
+  const streamThreadId = safeOpaqueId(initialSession?.threadId ?? initialSessionSnapshot.threadId) ?? (textValue(initialSession?.threadId ?? initialSessionSnapshot.threadId) || null);
   const activeTraceId = requestedTraceId
     ?? safeTraceId(initialSession?.lastTraceId ?? initialSessionSnapshot.lastTraceId ?? initialSessionSnapshot.currentTraceId)
     ?? null;
@@ -121,17 +123,21 @@ export async function handleWorkbenchRealtimeHttp(request, response, url, option
       if (event) {
         writeEvent("workbench.trace.event", {
           type: "trace.event",
+          sessionId: streamSessionId,
+          threadId: streamThreadId,
           traceId: activeTraceId,
           event,
-          snapshot: traceSnapshotSummary(snapshot),
+          snapshot: { ...traceSnapshotSummary(snapshot), sessionId: streamSessionId, threadId: streamThreadId },
           cursor: { traceSeq: eventSeq(event, Number(snapshot?.eventCount ?? 1) - 1) }
         });
       } else {
         writeEvent("workbench.trace.snapshot", {
           type: "trace.snapshot",
+          sessionId: streamSessionId,
+          threadId: streamThreadId,
           traceId: activeTraceId,
           reason: "trace-store-update",
-          snapshot: traceSnapshotForRealtime(snapshot),
+          snapshot: { ...traceSnapshotForRealtime(snapshot), sessionId: streamSessionId, threadId: streamThreadId },
           cursor: { traceSeq: traceSnapshotLastSeq(snapshot) }
         });
       }
@@ -312,6 +318,8 @@ async function handleWorkbenchTraceEventPage(response, url, options, actor, rawT
     status: "succeeded",
     contractVersion: "workbench-trace-events-v1",
     traceId,
+    sessionId: safeSessionId(result?.sessionId ?? result?.session?.sessionId ?? session?.id) ?? null,
+    threadId: safeOpaqueId(result?.threadId ?? result?.session?.threadId ?? session?.threadId) ?? (textValue(session?.threadId) || null),
     ...page,
     projection,
     projectionStatus: projection.projectionStatus,
@@ -591,16 +599,22 @@ async function writeTraceRealtimeSnapshot({ writeEvent, options, actor, traceId,
     });
     return;
   }
+  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);
   writeEvent("workbench.trace.snapshot", {
     type: "trace.snapshot",
     reason,
+    sessionId,
+    threadId,
     traceId,
-    snapshot: traceSnapshotForRealtime(context.trace),
+    snapshot: { ...traceSnapshotForRealtime(context.trace), sessionId, threadId },
     cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }
   });
   writeEvent("workbench.turn.snapshot", {
     type: "turn.snapshot",
     reason,
+    sessionId,
+    threadId,
     traceId,
     turn: turnSnapshot(context),
     cursor: { traceSeq: traceSnapshotLastSeq(context.trace) }

web/hwlab-cloud-web/scripts/workbench-e2e-server.ts

@@ -108,16 +108,16 @@ async function handleRequest(request: IncomingMessage, response: ServerResponse)
 
   const sessionMessagesMatch = path.match(/^\/v1\/workbench\/sessions\/([^/]+)\/messages$/u);
   if (sessionMessagesMatch && method === "GET") {
-    await delay(state.sessionDetailDelayMs);
     const sessionId = canonicalSessionId(decodeURIComponent(sessionMessagesMatch[1] ?? ""));
+    await delay(sessionDetailDelayMs(sessionId));
     const session = visibleSessionById(sessionId);
     return session ? json(response, 200, workbenchSessionMessagesPayload(session, url)) : json(response, 404, { ok: false, status: 404, error: { code: "session_not_found" } });
   }
 
   const workbenchSessionMatch = path.match(/^\/v1\/workbench\/sessions\/([^/]+)$/u);
   if (workbenchSessionMatch && method === "GET") {
-    await delay(state.sessionDetailDelayMs);
     const sessionId = canonicalSessionId(decodeURIComponent(workbenchSessionMatch[1] ?? ""));
+    await delay(sessionDetailDelayMs(sessionId));
     if (state.scenarioId === "session-switch-detail-404-isolated" && sessionId === "ses_stale_404") return json(response, 404, { ok: false, status: 404, error: { code: "session_not_found" } });
     if (state.scenarioId === "completed-replay-detail-404" && sessionId === "ses_completed") return json(response, 404, { ok: false, status: 404, error: { code: "session_replay_unavailable" } });
     const session = visibleSessionById(sessionId, { includeArchived: true });
@@ -258,6 +258,10 @@ function createScenarioState(scenarioId: string): ScenarioState {
     sessions.push(terminalEmptyTraceSession());
     traces.trc_terminal_empty = terminalEmptyTrace();
   }
+  if (id === "cross-session-late-events") {
+    sessions.unshift(crossSessionLateB(), crossSessionLateA());
+    traces.trc_late_A = crossSessionLateTraceA();
+  }
   if (id === "deleted-session-deeplink") sessions.unshift(archivedDeletedSession());
   if (id === "session-switch-empty-reload") sessions.push(emptySession());
   const selectedSessionId = id === "deep-link" || id === "stale-nested-trace"
@@ -402,6 +406,54 @@ function submitSplitTrace(): JsonRecord {
   return { traceId: "trc_submit_split", status: "running", sessionId: "ses_submit_split", threadId: "thr_submit_split", turnId: "turn_submit_split", events: [], eventCount: 0, fullTraceLoaded: false, hasMore: false };
 }
 
+function crossSessionLateA(): SessionRecord {
+  const now = new Date().toISOString();
+  return {
+    sessionId: "ses_late_A",
+    threadId: "thr_late_A",
+    status: "completed",
+    lastTraceId: "trc_late_A",
+    startedAt: now,
+    updatedAt: now,
+    messageCount: 2,
+    firstUserMessagePreview: "写一个python脚本测试性能",
+    messages: [
+      { id: "msg_late_A_user", messageId: "msg_late_A_user", role: "user", title: "用户", text: "写一个python脚本测试性能", status: "sent", createdAt: now, sessionId: "ses_late_A", threadId: "thr_late_A", traceId: "trc_late_A", turnId: "turn_late_A" },
+      { id: "msg_late_A_agent", messageId: "msg_late_A_agent", role: "agent", title: "Code Agent", text: "已写好 perf_test.py，脚本可用。", status: "completed", createdAt: now, sessionId: "ses_late_A", threadId: "thr_late_A", traceId: "trc_late_A", turnId: "turn_late_A", runnerTrace: crossSessionLateTraceA() }
+    ]
+  };
+}
+
+function crossSessionLateB(): SessionRecord {
+  const now = new Date().toISOString();
+  return {
+    sessionId: "ses_late_B",
+    threadId: "thr_late_B",
+    status: "active",
+    lastTraceId: "trc_late_A",
+    startedAt: now,
+    updatedAt: now,
+    messageCount: 2,
+    firstUserMessagePreview: "B session waiting",
+    messages: [
+      { id: "msg_late_B_user", messageId: "msg_late_B_user", role: "user", title: "用户", text: "B session waiting", status: "sent", createdAt: now, sessionId: "ses_late_B", threadId: "thr_late_B", traceId: "trc_late_A", turnId: "turn_late_B" },
+      { id: "msg_late_B_agent", messageId: "msg_late_B_agent", role: "agent", title: "Code Agent", text: "", status: "running", createdAt: now, sessionId: "ses_late_B", threadId: "thr_late_B", traceId: "trc_late_A", turnId: "turn_late_B", runnerTrace: { traceId: "trc_late_A", status: "running", sessionId: "ses_late_B", threadId: "thr_late_B", events: [], eventCount: 0, fullTraceLoaded: false, hasMore: false } }
+    ]
+  };
+}
+
+function crossSessionLateTraceA(): JsonRecord {
+  const now = new Date().toISOString();
+  const finalText = "已写好 perf_test.py，脚本可用。";
+  const event = { seq: 1, createdAt: now, label: "agentrun:assistant:message", type: "assistant_message", status: "completed", replyAuthority: true, final: true, message: finalText, terminal: true };
+  return { traceId: "trc_late_A", status: "completed", sessionId: "ses_late_A", threadId: "thr_late_A", turnId: "turn_late_A", events: [event], eventCount: 1, fullTraceLoaded: true, hasMore: false, finalResponse: { text: finalText, status: "completed" }, assistantText: finalText };
+}
+
+function sessionDetailDelayMs(sessionId: string): number {
+  if (state.scenarioId === "cross-session-late-events" && sessionId === "ses_late_A") return 900;
+  return state.sessionDetailDelayMs;
+}
+
 function acceptChatTurn(body: JsonRecord): JsonRecord {
   const sessionId = typeof body.sessionId === "string" && body.sessionId.trim() ? body.sessionId : state.selectedSessionId;
   const session = sessionById(sessionId) ?? createManualSession({ sessionId });
@@ -569,12 +621,23 @@ function tracePayload(traceId: string, url: URL): JsonRecord {
 
 function workbenchTracePayload(traceId: string, url: URL): JsonRecord {
   const payload = tracePayload(traceId, url);
-  return { ok: true, status: "ok", contractVersion: "workbench-read-model-v1", traceId, 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 };
+  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 };
 }
 
 function sse(response: ServerResponse, url: URL): void {
   response.writeHead(200, { "content-type": "text/event-stream; charset=utf-8", "cache-control": "no-cache", connection: "keep-alive" });
   writeSse(response, "workbench.connected", { type: "connected", sessionId: url.searchParams.get("sessionId") });
+  if (state.scenarioId === "cross-session-late-events") {
+    const scenarioId = state.scenarioId;
+    setTimeout(() => {
+      if (state.scenarioId !== scenarioId) return;
+      const trace = crossSessionLateTraceA();
+      const event = (trace.events as JsonRecord[])[0];
+      writeSse(response, "workbench.trace.event", { type: "trace.event", sessionId: "ses_late_A", threadId: "thr_late_A", traceId: "trc_late_A", event, snapshot: trace });
+      writeSse(response, "workbench.turn.snapshot", { type: "turn.snapshot", sessionId: "ses_late_A", threadId: "thr_late_A", traceId: "trc_late_A", turn: trace });
+    }, 650);
+    return;
+  }
   if (!state.terminalScript) return;
   const scenarioId = state.scenarioId;
   setTimeout(() => {
@@ -582,9 +645,9 @@ function sse(response: ServerResponse, url: URL): void {
     const terminalStatus = state.terminalFailureScript ? "failed" : "completed";
     const terminalText = state.terminalFailureScript ? "恢复后失败：缺少受控依赖。" : "事件重放后完成。";
     const event = { seq: 3, createdAt: new Date().toISOString(), label: "agentrun:assistant:message", type: "assistant_message", status: terminalStatus, replyAuthority: true, final: true, message: terminalText, terminal: true };
-    writeSse(response, "workbench.trace.event", { type: "trace.event", traceId: "trc_running", event, snapshot: { traceId: "trc_running", status: terminalStatus, events: [event], eventCount: 3, fullTraceLoaded: true, finalResponse: { text: terminalText, status: terminalStatus } } });
+    writeSse(response, "workbench.trace.event", { type: "trace.event", sessionId: "ses_running", threadId: "thr_running", traceId: "trc_running", event, snapshot: { traceId: "trc_running", sessionId: "ses_running", threadId: "thr_running", status: terminalStatus, events: [event], eventCount: 3, fullTraceLoaded: true, finalResponse: { text: terminalText, status: terminalStatus } } });
     finishRunningSession(terminalStatus, terminalText);
-    writeSse(response, "workbench.turn.snapshot", { type: "turn.snapshot", traceId: "trc_running", turn: turnPayload("trc_running") });
+    writeSse(response, "workbench.turn.snapshot", { type: "turn.snapshot", sessionId: "ses_running", threadId: "thr_running", traceId: "trc_running", turn: turnPayload("trc_running") });
   }, 350);
 }
 

web/hwlab-cloud-web/src/stores/workbench.ts

@@ -96,7 +96,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     const selected = targetSessionId ? await loadWorkbenchSession(targetSessionId, listedSessions.find((item) => item.sessionId === targetSessionId) ?? null) : null;
     clearSessionDetailLoading(targetSessionId);
     if (selected && !isArchivedSession(selected) && isCurrentSessionSelection(requestEpoch, selected.sessionId)) applySelectedSessionDetail(selected, messages.value);
-    if (selected && !isArchivedSession(selected) && routeRequestId && !isCurrentSessionSelection(requestEpoch, selected.sessionId)) applySelectedSessionDetail(selected, messages.value);
+    if (selected && !isArchivedSession(selected) && routeRequestId && !isCurrentSessionSelection(requestEpoch, selected.sessionId)) rememberSessionDetail(selected);
     if (selected && isArchivedSession(selected)) isolateSessionLoadFailure(selected.sessionId, null, [], "session archived", { restorePrevious: false });
     if (options.invalidRouteId || (targetSessionId && !selected)) isolateSessionLoadFailure(targetSessionId ?? options.invalidRouteId ?? "invalid-session", null, [], "session URL not found", { restorePrevious: false });
     const nextSessions = stableSessionList(sessions.value, listedSessions, selected?.sessionId ?? routeSessionId ?? includeSessionId, selected);
@@ -258,6 +258,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   function applyTurnStatusSnapshot(traceId: string, result: AgentChatResultResponse | TraceSnapshot): void {
+    if (!shouldApplyActiveTraceAuthority(traceId, traceResultSessionId(result))) return;
     rememberTurnStatus(traceId, result);
     syncTurnStatusToMessage(traceId, result);
   }
@@ -286,8 +287,9 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   function syncTurnStatusToMessage(traceId: string, result: AgentChatResultResponse | TraceSnapshot): void {
     const status = statusFromResult(result.status);
     const terminal = (result as AgentChatResultResponse).terminal === true || isTerminalMessageStatus(status);
+    const authoritySessionId = traceResultSessionId(result);
     messages.value = messages.value.map((message) => {
-      if (message.traceId !== traceId || message.role !== "agent") return message;
+      if (!shouldApplyTraceToMessage(message, traceId, authoritySessionId)) return message;
       const runnerTrace = mergeTerminalResultTrace(message.runnerTrace, result as AgentChatResultResponse);
       rememberTraceAuthority(runnerTrace);
       const error = normalizeAgentError((result as AgentChatResultResponse).error ?? runnerTrace?.error ?? message.error);
@@ -457,10 +459,12 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   function applyTraceHydrationResult(traceId: string, result: AgentChatResultResponse): void {
+    const authoritySessionId = traceResultSessionId(result);
+    if (!shouldApplyActiveTraceAuthority(traceId, authoritySessionId)) return;
     const events = Array.isArray(result.events) ? result.events : Array.isArray(result.traceEvents) ? result.traceEvents : [];
     markWorkbenchTraceEventsReceived({ traceId, events, transport: "rest_gap" });
     messages.value = messages.value.map((message) => {
-      if (firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) !== traceId || message.role !== "agent") return message;
+      if (!shouldApplyTraceToMessage(message, traceId, authoritySessionId)) return message;
       const runnerTrace = mergeRunnerTrace(message.runnerTrace, {
         ...(result.runnerTrace ?? {}),
         traceId: result.traceId ?? traceId,
@@ -581,6 +585,52 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     return firstNonEmptyString(currentRequest.value?.traceId, activeTraceIdFromMessages(messages.value, turnStatusAuthority.value));
   }
 
+  function realtimeEventSessionId(event: WorkbenchRealtimeEvent): string | null {
+    const turn = recordValue(event.turn);
+    const snapshot = recordValue(event.snapshot);
+    const traceEvent = recordValue(event.event);
+    return normalizeWorkbenchSessionId(firstNonEmptyString(event.sessionId, turn?.sessionId, snapshot?.sessionId, traceEvent?.sessionId));
+  }
+
+  function traceResultSessionId(result: AgentChatResultResponse | TraceSnapshot | Record<string, unknown> | null | undefined): string | null {
+    const value = recordValue(result);
+    const runnerTrace = recordValue(value?.runnerTrace);
+    const session = recordValue(value?.session);
+    return normalizeWorkbenchSessionId(firstNonEmptyString(value?.sessionId, runnerTrace?.sessionId, session?.sessionId));
+  }
+
+  function messageSessionAuthority(message: ChatMessage | null | undefined): string | null {
+    return normalizeWorkbenchSessionId(firstNonEmptyString(message?.sessionId, message?.runnerTrace?.sessionId));
+  }
+
+  function shouldApplyActiveTraceAuthority(traceId: string | null | undefined, authoritySessionId: string | null | undefined): boolean {
+    const id = firstNonEmptyString(traceId);
+    const activeId = activeSessionId.value;
+    const sessionId = normalizeWorkbenchSessionId(authoritySessionId);
+    if (!activeId) return false;
+    if (sessionId && sessionId !== activeId) return false;
+    if (!id) return Boolean(sessionId && sessionId === activeId);
+    const message = messages.value.find((item) => firstNonEmptyString(item.traceId, item.runnerTrace?.traceId) === id) ?? null;
+    const messageSessionId = messageSessionAuthority(message);
+    if (messageSessionId && messageSessionId !== activeId) return false;
+    if (sessionId && messageSessionId && sessionId !== messageSessionId) return false;
+    const knownSessionId = normalizeWorkbenchSessionId(firstNonEmptyString(turnStatusAuthority.value[id]?.sessionId, traceAuthorityById.value[id]?.sessionId));
+    if (knownSessionId && knownSessionId !== activeId) return false;
+    if (sessionId && knownSessionId && sessionId !== knownSessionId) return false;
+    return Boolean(sessionId || messageSessionId || knownSessionId || message);
+  }
+
+  function shouldApplyTraceToMessage(message: ChatMessage, traceId: string, authoritySessionId: string | null | undefined): boolean {
+    if (message.role !== "agent" || firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) !== traceId) return false;
+    const activeId = activeSessionId.value;
+    const sessionId = normalizeWorkbenchSessionId(authoritySessionId);
+    const messageSessionId = messageSessionAuthority(message);
+    if (activeId && messageSessionId && messageSessionId !== activeId) return false;
+    if (sessionId && activeId && sessionId !== activeId) return false;
+    if (sessionId && messageSessionId && sessionId !== messageSessionId) return false;
+    return true;
+  }
+
   function applyRealtimeEvent(event: WorkbenchRealtimeEvent, eventName: string): void {
     recordActivity(event.type ? `realtime:${event.type}` : `realtime:${eventName}`);
     if (event.type === "trace.snapshot") {
@@ -604,6 +654,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   function applyRealtimeTraceSnapshot(traceId: string | null | undefined, snapshot: WorkbenchRealtimeEvent["snapshot"]): void {
     const id = firstNonEmptyString(traceId, snapshot?.traceId);
     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");
   }
@@ -611,6 +662,8 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   function applyRealtimeTraceEvent(traceId: string | null | undefined, event: WorkbenchRealtimeEvent["event"], snapshot: WorkbenchRealtimeEvent["snapshot"], realtimeEvent?: WorkbenchRealtimeEvent | null): void {
     const id = firstNonEmptyString(traceId, event?.traceId, snapshot?.traceId);
     if (!id) return;
+    const sessionId = realtimeEvent ? realtimeEventSessionId(realtimeEvent) : traceResultSessionId(snapshot ?? event ?? null);
+    if (!shouldApplyActiveTraceAuthority(id, sessionId)) return;
     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));
@@ -620,6 +673,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   function applyRealtimeTurnSnapshot(turn: Record<string, unknown>): void {
     const traceId = firstNonEmptyString(turn.traceId);
     if (!traceId) return;
+    if (!shouldApplyActiveTraceAuthority(traceId, traceResultSessionId(turn))) return;
     const status = firstNonEmptyString(turn.status) ?? undefined;
     applyTurnStatusSnapshot(traceId, { traceId, status, running: turn.running === true, terminal: turn.terminal === true, sessionId: firstNonEmptyString(turn.sessionId) ?? undefined, threadId: firstNonEmptyString(turn.threadId) ?? undefined, agentRun: turn.agentRun as AgentRunProvenance | undefined } as AgentChatResultResponse);
     if (turn.terminal === true || isTerminalMessageStatus(status)) void refreshTerminalTraceFromRest(traceId, "realtime-turn-snapshot");
@@ -710,11 +764,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
 
   function applyTraceSnapshot(traceId: string, snapshot: TraceSnapshot): void {
     const trace = snapshotToRunnerTrace(snapshot);
+    const authoritySessionId = traceResultSessionId(trace) ?? traceResultSessionId(snapshot);
+    if (!shouldApplyActiveTraceAuthority(traceId, authoritySessionId)) return;
     rememberTurnStatus(traceId, snapshot);
     const status = statusFromResult(snapshot.status);
     const terminal = isTerminalMessageStatus(status);
     messages.value = messages.value.map((message) => {
-      if (message.traceId !== traceId) return message;
+      if (!shouldApplyTraceToMessage(message, traceId, authoritySessionId)) return message;
       const runnerTrace = mergeRunnerTrace(message.runnerTrace, trace);
       rememberTraceAuthority(runnerTrace);
       const traceAssistantText = message.role === "agent" ? assistantTextFromTraceEvents(Array.isArray(runnerTrace.events) ? runnerTrace.events : []) : null;
@@ -732,13 +788,15 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   function completeTrace(traceId: string, result: AgentChatResultResponse): void {
+    const authoritySessionId = traceResultSessionId(result);
+    if (!shouldApplyActiveTraceAuthority(traceId, authoritySessionId)) return;
     clearProjectionCatchup(traceId);
     const resultTrace = recordValue(result.runnerTrace);
     const traceAssistantText = assistantTextFromTraceEvents(firstArray(result.events, result.traceEvents, resultTrace?.events));
     const text = firstNonEmptyString(result.assistantText, finalResponseText(result.finalResponse), typeof result.reply === "string" ? result.reply : result.reply?.content, agentErrorDisplayText(result.error), result.text, result.summary, traceAssistantText) ?? "Code Agent 已完成，但没有返回可展示的 final response。";
     const terminalStatus = result.status === "completed" ? "completed" : statusFromResult(result.status);
     messages.value = messages.value.map((message) => {
-      if (message.traceId !== traceId || message.role !== "agent") return message;
+      if (!shouldApplyTraceToMessage(message, traceId, authoritySessionId)) return message;
       const runnerTrace = mergeTerminalResultTrace(message.runnerTrace, result);
       rememberTraceAuthority(runnerTrace);
       const error = normalizeAgentError(result.error ?? runnerTrace?.error ?? message.error);
@@ -768,8 +826,10 @@ export const useWorkbenchStore = defineStore("workbench", () => {
   }
 
   function applyTerminalResultDiagnostics(traceId: string, result: AgentChatResultResponse): void {
+    const authoritySessionId = traceResultSessionId(result);
+    if (!shouldApplyActiveTraceAuthority(traceId, authoritySessionId)) return;
     messages.value = messages.value.map((message) => {
-      if (message.traceId !== traceId || message.role !== "agent") return message;
+      if (!shouldApplyTraceToMessage(message, traceId, authoritySessionId)) return message;
       const runnerTrace = mergeTerminalResultTrace(message.runnerTrace, result);
       rememberTraceAuthority(runnerTrace);
       const error = normalizeAgentError(result.error ?? runnerTrace?.error ?? message.error);

web/hwlab-cloud-web/src/views/workbench/CodeWorkbenchView.vue

@@ -1,5 +1,5 @@
 <script setup lang="ts">
-// SPEC: PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
+// SPEC: PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection; PJ2026-01060505 Workbench Performance draft-2026-06-17-p0
 // Starts Workbench first-screen journey timing around route hydrate.
 
 import { computed, onBeforeUnmount, onMounted, ref, watch } from "vue";

web/hwlab-cloud-web/tests/workbench-e2e/specs/cross-session-authority.spec.ts

@@ -0,0 +1,24 @@
+import { expect, saveScreenshot, test } from "../fixtures/test";
+import { selectors, sessionTab } from "../fixtures/selectors";
+
+test.describe("cross-session late authority", () => {
+  test.use({ scenarioId: "cross-session-late-events" });
+
+  test("late A realtime events do not pollute B active session", async ({ page }, testInfo) => {
+    await page.goto("/workbench/sessions/ses_late_B");
+    await expect(page.locator(selectors.workspace)).toBeVisible();
+    await expect(page.locator(sessionTab("ses_late_B"))).toBeVisible();
+    await expect(page.locator(sessionTab("ses_late_B"))).toHaveAttribute("data-active", "true");
+    await page.waitForTimeout(1_200);
+
+    expect(page.url()).toMatch(/\/workbench\/sessions\/ses_late_B$/u);
+    expect(await page.locator(sessionTab("ses_late_B")).getAttribute("data-active")).toBe("true");
+    const conversationText = await page.locator(selectors.conversationList).textContent();
+    expect(conversationText ?? "").not.toContain("写一个python脚本测试性能");
+    expect(conversationText ?? "").not.toContain("perf_test.py");
+    expect(await page.locator(`${selectors.messageCard}[data-role="user"]`).count()).toBe(1);
+
+    await expect(page.locator(selectors.commandInput)).toBeVisible();
+    await saveScreenshot(page, testInfo, "cross-session-late-events-b-active");
+  });
+});