feat: complete workbench opencode runtime migration

2026-06-30 22:45:17 +08:00
parent 6568e60f72
commit c306fc39a6
14 changed files with 1341 additions and 412 deletions
@@ -38,6 +38,7 @@ const requiredFiles = Object.freeze([
  "src/stores/app.ts",
  "src/stores/auth.ts",
  "src/stores/workbench.ts",
+  "src/stores/workbench-event-reducer.ts",
  "src/stores/workbench-timeline-model.ts",
  "src/stores/workbench-session-cache.ts",
  "src/composables/useWorkbenchScrollRuntime.ts",
@@ -78,6 +79,12 @@ const pkg = JSON.parse(readWeb("package.json")) as { dependencies?: Record<strin
 const appSource = readCloudWebAppSource(rootDir);
 const workbenchStoreSource = readWeb("src/stores/workbench.ts");
 const workbenchRealtimeRuntimeSource = `${readWeb("src/utils/workbench-realtime-runtime.ts")}\n${readWeb("src/utils/workbench-stream-transport.ts")}`;
+const workbenchRefreshRuntimeSource = readWeb("src/utils/workbench-refresh-runtime.ts");
+const workbenchEventReducerSource = readWeb("src/stores/workbench-event-reducer.ts");
+const workbenchTimelineRuntimeSource = readWeb("src/stores/workbench-timeline-model.ts");
+const workbenchScrollRuntimeSource = readWeb("src/composables/useWorkbenchScrollRuntime.ts");
+const workbenchErrorRuntimeSource = readWeb("src/utils/workbench-error-runtime.ts");
+const workbenchHealthRuntimeSource = readWeb("src/utils/workbench-health.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");
@@ -119,9 +126,32 @@ assertIncludes(appSource, "hwlab_session", "auth comments/code must preserve Web
 assertIncludes(appSource, "activityRef", "Code Agent inactivity-timeout activityRef must be preserved");
 assertIncludes(workbenchStoreSource, "createWorkbenchStreamTransportRuntime", "Workbench store must enter SSE through the realtime runtime boundary");
 assertIncludes(workbenchRealtimeRuntimeSource, "connectWorkbenchEvents", "Workbench realtime runtime must own the unified SSE EventSource entry");
+assertIncludes(workbenchRealtimeRuntimeSource, "WorkbenchStreamTransportRecovery", "SSE transport must own recovery actions, not just wrap EventSource");
+assertIncludes(workbenchRealtimeRuntimeSource, "cursorByKey", "SSE transport runtime must own cursor replay state");
+assertIncludes(workbenchRefreshRuntimeSource, "WorkbenchTraceHydrationQueueRuntime", "Trace hydration queue must live in refresh runtime");
+assertIncludes(workbenchStoreSource, "createWorkbenchTraceHydrationQueueRuntime", "Workbench store must use the migrated trace hydration runtime");
+assertIncludes(workbenchStoreSource, "createWorkbenchScheduledTaskRuntime", "Workbench store must use the migrated scheduled task runtime");
+assertIncludes(workbenchEventReducerSource, "reduceWorkbenchRealtimeEvent", "Realtime event reducer must own SSE event classification");
+assertIncludes(workbenchEventReducerSource, "workbench-event-reducer", "Realtime event reducer must emit module diagnostics for monitor root cause");
+assertIncludes(workbenchStoreSource, "reduceWorkbenchRealtimeEvent", "Workbench store must consume migrated realtime reducer actions");
+assert.doesNotMatch(workbenchStoreSource, /function applyRealtimeEvent[\s\S]{0,900}event\.type ===/u, "Workbench store applyRealtimeEvent must not branch directly on raw SSE event.type");
+assertIncludes(workbenchTimelineRuntimeSource, "TurnGap", "Timeline runtime must preserve OpenCode TurnGap row parity");
+assertIncludes(workbenchTimelineRuntimeSource, "AssistantPart", "Timeline runtime must preserve OpenCode AssistantPart row parity");
+assertIncludes(workbenchTimelineRuntimeSource, "unwrapErrorMessage", "Timeline runtime must preserve OpenCode readable error row behavior");
+assertIncludes(workbenchScrollRuntimeSource, "createWorkbenchScrollPersistenceRuntime", "Scroll runtime must preserve OpenCode session/tab scroll persistence");
+assertIncludes(conversationPanelSource, "sessionKey: scrollSessionKey", "ConversationPanel must use session-scoped scroll persistence in production");
+assertIncludes(workbenchErrorRuntimeSource, "formatServerError", "Error runtime must preserve OpenCode readable server error formatting");
+assertIncludes(workbenchErrorRuntimeSource, "projectionDiagnosticFromFailure", "Error runtime must own projection diagnostic envelopes");
+assertIncludes(workbenchHealthRuntimeSource, "createScopedCache", "ScopedCache must be used by a production Workbench runtime, not only by tests/checks");
+assert.doesNotMatch(workbenchStoreSource, /function (?:normalizeProjectionDiagnostic|projectionDiagnosticFromApiFailure|projectionDiagnosticFromFailure|normalizeProjectionBlocker|normalizeErrorDiagnostic|normalizeApiErrorRecord|agentErrorFromProjection)\b/u, "Workbench store must call ErrorRuntime directly instead of reintroducing diagnostic wrapper ownership");
+assertIncludes(workbenchStoreSource, "cleanupDroppedWorkbenchSessionCaches", "Workbench store must run OpenCode-style dropped session cache cleanup after trim");
+assertIncludes(workbenchStoreSource, "cleanupWorkbenchServerStateSessions", "Workbench store must clean dropped session trace/turn authority after trim");
+assert.doesNotMatch(workbenchStoreSource, /case "session\.forget"[\s\S]{0,400}sessionStatusById: withoutKey\(state\.sessionStatusById/u, "Session forget must not leave trace/turn authority cleanup out of reducer helpers");
+assert.doesNotMatch(workbenchStoreSource, /traceHydration(?:InFlight|Queued|Queue)|traceHydrationPumpActive|terminalRealtimeRefreshInFlight|realtimeSessionMessagesInFlight|realtimeErrorGapFillLastAtByKey|activeTraceRestGapFillTimers|realtimeOutboxSeqByKey|sessionListRefreshInFlight|sessionListRefreshTimers/u, "Workbench store must not reintroduce migrated in-flight/timer/cursor state");
 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");
-assertIncludes(workbenchStoreSource, "if (existing) return;", "Session list scheduled refreshes must coalesce instead of resetting timers under SSE error storms");
+assertIncludes(workbenchRefreshRuntimeSource, "if (existing)", "Scheduled refresh runtime must coalesce instead of resetting timers under SSE error storms");
+assertIncludes(workbenchRefreshRuntimeSource, "replaceTimer", "Scheduled refresh runtime must make replacement explicit instead of ad hoc timer resets");
 assertIncludes(workbenchStoreSource, "if (activeId) scheduleSessionListRefresh(activeId, SESSION_LIST_REALTIME_REFRESH_DELAY_MS);", "Realtime stream errors must schedule bounded session list refreshes");
 assert.doesNotMatch(workbenchStoreSource, /handleRealtimeStreamError[\s\S]{0,1200}refreshSessions\([^;]+force:\s*true/u, "Realtime stream errors must not force-refresh the full session list");
 assert.doesNotMatch(workbenchStoreSource, /refreshActiveTraceFromRest[\s\S]{0,1200}refreshSessions\([^;]+force:\s*true/u, "Active trace REST gap-fill must not force-refresh the full session list");
@@ -12,7 +12,9 @@ import { composeWorkbenchScopedKey, splitWorkbenchScopedKey, workbenchPathKey, w
 import { createSafeStorageRuntime, isStorageQuotaError, migrateLegacyStorage, normalizePersistedValue, readJsonStorage, removePersistedTarget, removeStorageKey, writeJsonStorage, type StorageLike } from "../src/utils/safe-storage.ts";
 import { checkWorkbenchHealth, createWorkbenchHealthProbeCache } from "../src/utils/workbench-health.ts";
 import { buildWorkbenchTimelineRows, normalizeWorkbenchTimelineMessages, workbenchTimelineSignature } from "../src/stores/workbench-timeline-model.ts";
-import { trimWorkbenchSessionCache } from "../src/stores/workbench-session-cache.ts";
+import { reduceWorkbenchRealtimeEvent } from "../src/stores/workbench-event-reducer.ts";
+import { cleanupWorkbenchServerStateDroppedSessions, cleanupWorkbenchServerStateSessions, createWorkbenchServerState, reduceWorkbenchServerState } from "../src/stores/workbench-server-state.ts";
+import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "../src/stores/workbench-session-cache.ts";

 test("Workbench scoped keys encode delimiter characters", () => {
  assert.deepEqual(splitWorkbenchScopedKey(workbenchRealtimeScopeKey("ses|one", "trc/two")), ["workbench.realtime", "ses|one", "trc/two"]);
@@ -152,9 +154,24 @@ test("timeline model deduplicates authoritative messages with stable row keys",
  ]);

  assert.equal(rows.length, 2);
-  assert.deepEqual(rows.map((row) => `${row.role}:${row.message.text}`), ["user:hi", "agent:done"]);
+  const messageRows = rows.flatMap((row) => row.message ? [row] : []);
+  assert.deepEqual(messageRows.map((row) => `${row.role}:${row.message.text}`), ["user:hi", "agent:done"]);
  assert.match(workbenchTimelineSignature(rows), /completed:4/u);
-  assert.deepEqual(normalizeWorkbenchTimelineMessages(rows.map((row) => row.message)).map((message) => message.messageId), ["msg_user", "msg_agent"]);
+  assert.deepEqual(normalizeWorkbenchTimelineMessages(messageRows.map((row) => row.message)).map((message) => message.messageId), ["msg_user", "msg_agent"]);
+});
+
+test("timeline model preserves OpenCode turn gap, thinking, retry and error row parity", () => {
+  const rows = buildWorkbenchTimelineRows([
+    { id: "msg_user_1", messageId: "msg_user_1", role: "user", title: "用户", text: "hi", status: "sent", createdAt: "2026-06-30T00:00:00.000Z", sessionId: "ses_a", traceId: "trc_a" },
+    { id: "msg_agent_1", messageId: "msg_agent_1", role: "agent", title: "Code Agent", text: "", status: "running", createdAt: "2026-06-30T00:00:01.000Z", sessionId: "ses_a", traceId: "trc_a" },
+    { id: "msg_user_2", messageId: "msg_user_2", role: "user", title: "用户", text: "again", status: "sent", createdAt: "2026-06-30T00:00:02.000Z", sessionId: "ses_a", traceId: "trc_b" },
+    { id: "msg_agent_2", messageId: "msg_agent_2", role: "agent", title: "Code Agent", text: "", status: "retry" as never, createdAt: "2026-06-30T00:00:03.000Z", sessionId: "ses_a", traceId: "trc_b", error: { message: "Error: {\"error\":{\"type\":\"Provider\",\"message\":\"offline\"}}" } }
+  ]);
+
+  assert.ok(rows.some((row) => row.type === "TurnGap"));
+  assert.ok(rows.some((row) => row.type === "Thinking"));
+  assert.ok(rows.some((row) => row.type === "Retry"));
+  assert.ok(rows.some((row) => row.type === "Error" && row.text === "Provider: offline"));
 });

 test("session cache trim keeps retained sessions and trims messages by explicit policy", () => {
@@ -180,6 +197,49 @@ test("session cache trim keeps retained sessions and trims messages by explicit
  assert.deepEqual(Object.keys(result.state.sessionsById).sort(), ["ses_1", "ses_3"]);
 });

+test("session cache cleanup drops stale session/message caches with OpenCode cleanup parity", () => {
+  const result = cleanupDroppedWorkbenchSessionCaches({
+    sessionOrder: ["ses_1", "ses_2"],
+    sessionsById: { ses_1: { sessionId: "ses_1" }, ses_2: { sessionId: "ses_2" } },
+    messagesBySessionId: { ses_1: [], ses_2: [] }
+  }, ["ses_2"]);
+
+  assert.deepEqual(result.droppedSessionIds, ["ses_1"]);
+  assert.deepEqual(result.state.sessionOrder, ["ses_2"]);
+  assert.deepEqual(Object.keys(result.state.sessionsById), ["ses_2"]);
+});
+
+test("server-state cleanup removes dropped session trace and turn authority", () => {
+  let state = createWorkbenchServerState();
+  state = reduceWorkbenchServerState(state, { type: "session.detail", session: { sessionId: "ses_1", messages: [{ id: "m1", role: "agent", title: "Code Agent", text: "", status: "running", createdAt: "2026-06-30T00:00:00.000Z", traceId: "trc_1", runnerTrace: { traceId: "trc_1", sessionId: "ses_1", events: [] } }] } });
+  state = reduceWorkbenchServerState(state, { type: "turn.status", turn: { traceId: "trc_1", status: "running", running: true, terminal: false, sessionId: "ses_1" } });
+  state = reduceWorkbenchServerState(state, { type: "trace.snapshot", traceId: "trc_1", trace: { traceId: "trc_1", sessionId: "ses_1", events: [] } });
+
+  const cleaned = cleanupWorkbenchServerStateDroppedSessions(state, []);
+  assert.deepEqual(Object.keys(cleaned.messagesBySessionId), []);
+  assert.deepEqual(Object.keys(cleaned.turnStatusByTraceId), []);
+  assert.deepEqual(Object.keys(cleaned.traceById), []);
+
+  const cleanedById = cleanupWorkbenchServerStateSessions(state, ["ses_1"]);
+  assert.deepEqual(Object.keys(cleanedById.messagesBySessionId), []);
+  assert.deepEqual(Object.keys(cleanedById.turnStatusByTraceId), []);
+  assert.deepEqual(Object.keys(cleanedById.traceById), []);
+});
+
+test("realtime event reducer classifies SSE payloads before store side effects", () => {
+  const trace = reduceWorkbenchRealtimeEvent({ type: "trace.event", traceId: "trc_1", event: { traceId: "trc_1", label: "delta" }, snapshot: { traceId: "trc_1", status: "running" } }, "workbench.trace.event");
+  assert.equal(trace.activityLabel, "realtime:trace.event");
+  assert.equal(trace.action.type, "trace.event");
+  assert.equal(trace.diagnostic.module, "workbench-event-reducer");
+
+  const emptyMessage = reduceWorkbenchRealtimeEvent({ type: "message.snapshot" }, "workbench.message.snapshot");
+  assert.deepEqual(emptyMessage.action, { type: "ignore", reason: "message.snapshot.missing-message" });
+
+  const error = reduceWorkbenchRealtimeEvent({ type: "error", traceId: "trc_2", error: { message: "offline" } }, "workbench.error");
+  assert.equal(error.action.type, "projection.error");
+  assert.equal(error.diagnostic.traceId, "trc_2");
+});
+
 test("health probe cache records ok and unavailable states", async () => {
  const cache = createWorkbenchHealthProbeCache({ cacheMs: 100 });
  const ok = await cache.probe({ key: "workbench", fetcher: async () => ({ ready: true }), classify: (value) => value.ready ? "ok" : "degraded" });
@@ -26,7 +26,8 @@ import { traceLifecycleExpanded } from "./trace-lifecycle";
 const workbench = useWorkbenchStore();
 const clipboard = useClipboard();
 const panelRef = ref<HTMLElement | null>(null);
-const { following, keepBottomAfterUpdate, onScroll: onPanelScroll, scrollToBottom } = useWorkbenchBottomFollowScroll(panelRef, { threshold: 56 });
+const scrollSessionKey = computed(() => workbench.activeSessionId ?? "workbench");
+const { following, keepBottomAfterUpdate, onScroll: onPanelScroll, scrollToBottom } = useWorkbenchBottomFollowScroll(panelRef, { threshold: 56, sessionKey: scrollSessionKey, tabKey: "conversation" });
 const detailMessageId = ref<string | null>(null);
 const visibleMessages = computed(() => workbench.sessionDetailLoading ? [] : workbench.activeMessages);
 const visibleTimelineRows = computed(() => buildWorkbenchTimelineRows(visibleMessages.value));
@@ -282,34 +283,41 @@ function formatDuration(ms: number): string {
 <template>
  <section id="conversation-list" ref="panelRef" class="conversation-panel" :data-following="following ? 'true' : 'false'" @scroll="onPanelScroll">
    <LoadingState v-if="workbench.sessionDetailLoading" class="conversation-detail-loading" label="加载中" />
+    <template v-for="row in visibleTimelineRows" :key="row.key">
+      <div v-if="row.type === 'TurnGap'" class="timeline-row timeline-turn-gap" aria-hidden="true" />
+      <div v-else-if="row.type === 'Thinking'" class="timeline-row timeline-thinking" :data-session-id="row.sessionId || undefined" :data-trace-id="row.traceId || undefined">
+        <LoadingState class="message-loading" :label="row.reasoningHeading || '思考中...'" compact />
+      </div>
+      <div v-else-if="row.type === 'Retry'" class="timeline-row timeline-retry" :data-session-id="row.sessionId || undefined" :data-trace-id="row.traceId || undefined">等待重试</div>
+      <div v-else-if="row.type === 'Error'" class="timeline-row timeline-error" :data-session-id="row.sessionId || undefined" :data-trace-id="row.traceId || undefined">{{ row.text }}</div>
        <article
-          v-for="{ key, message } in visibleTimelineRows"
-          :key="key"
+          v-else-if="row.message"
          class="message-card"
-          :data-role="message.role"
-          :data-status="message.status"
-          :data-session-id="message.sessionId || undefined"
-          :data-message-id="message.id || undefined"
-          :data-trace-id="message.traceId || message.runnerTrace?.traceId || undefined"
-          :data-turn-id="message.turnId || message.traceId || message.runnerTrace?.traceId || undefined"
+          :data-role="row.message.role"
+          :data-status="row.message.status"
+          :data-session-id="row.message.sessionId || undefined"
+          :data-message-id="row.message.id || undefined"
+          :data-trace-id="row.message.traceId || row.message.runnerTrace?.traceId || undefined"
+          :data-turn-id="row.message.turnId || row.message.traceId || row.message.runnerTrace?.traceId || undefined"
        >
-      <header v-if="message.role !== 'user'">
+      <header v-if="row.message.role !== 'user'">
        <div class="message-header-main">
-          <strong>{{ message.title }}</strong>
-          <span v-if="messageDurationMeta(message)" class="message-timing-meta message-duration-meta" :aria-label="messageDurationMeta(message)?.label" :title="messageDurationMeta(message)?.label">{{ messageDurationMeta(message)?.text }}</span>
+          <strong>{{ row.message.title }}</strong>
+          <span v-if="messageDurationMeta(row.message)" class="message-timing-meta message-duration-meta" :aria-label="messageDurationMeta(row.message)?.label" :title="messageDurationMeta(row.message)?.label">{{ messageDurationMeta(row.message)?.text }}</span>
        </div>
        <div class="message-header-actions">
-          <span v-if="messageActivityMeta(message)" class="message-activity-meta" :aria-label="messageActivityMeta(message)?.label" :title="messageActivityMeta(message)?.label">{{ messageActivityMeta(message)?.text }}</span>
-          <span v-if="isRunningMessage(message)" class="agent-running-indicator" aria-label="运行中" role="status"><span aria-hidden="true" /></span>
-          <StatusBadge v-if="showStatusBadge(message)" :status="message.status" />
-          <button v-if="message.role === 'agent'" class="message-detail-button" type="button" aria-label="运行详情" title="运行详情" @click="openDetails(message)">!</button>
+          <span v-if="messageActivityMeta(row.message)" class="message-activity-meta" :aria-label="messageActivityMeta(row.message)?.label" :title="messageActivityMeta(row.message)?.label">{{ messageActivityMeta(row.message)?.text }}</span>
+          <span v-if="isRunningMessage(row.message)" class="agent-running-indicator" aria-label="运行中" role="status"><span aria-hidden="true" /></span>
+          <StatusBadge v-if="showStatusBadge(row.message)" :status="row.message.status" />
+          <button v-if="row.message.role === 'agent'" class="message-detail-button" type="button" aria-label="运行详情" title="运行详情" @click="openDetails(row.message)">!</button>
        </div>
      </header>
-      <TraceTimeline v-if="message.role === 'agent' && message.runnerTrace" :trace="traceForDisplay(message)" :auto-expanded="traceAutoExpanded(message)" :storage-key="traceStorageKey(message)" />
-      <LoadingState v-if="isAwaitingAgentBody(message)" class="message-loading" label="思考中..." compact />
-      <MessageMarkdown v-if="showMessageText(message)" class="message-text" :source="visibleMessageText(message)" />
-      <ApiErrorDiagnostic v-if="messageHasDiagnostic(message)" class="message-diagnostic projection-diagnostic" :error="messageDiagnosticText(message)" :api-error="messageApiError(message)" :diagnostic="messageErrorDiagnostic(message)" compact />
+      <TraceTimeline v-if="row.message.role === 'agent' && row.message.runnerTrace" :trace="traceForDisplay(row.message)" :auto-expanded="traceAutoExpanded(row.message)" :storage-key="traceStorageKey(row.message)" />
+      <LoadingState v-if="isAwaitingAgentBody(row.message)" class="message-loading" label="思考中..." compact />
+      <MessageMarkdown v-if="showMessageText(row.message)" class="message-text" :source="visibleMessageText(row.message)" />
+      <ApiErrorDiagnostic v-if="messageHasDiagnostic(row.message)" class="message-diagnostic projection-diagnostic" :error="messageDiagnosticText(row.message)" :api-error="messageApiError(row.message)" :diagnostic="messageErrorDiagnostic(row.message)" compact />
    </article>
+    </template>
    <div v-if="!workbench.sessionDetailLoading && workbench.activeMessages.length === 0 && workbench.error" class="conversation-empty-hint conversation-error-hint">加载失败：{{ workbench.error }}</div>
    <div v-else-if="!workbench.sessionDetailLoading && workbench.activeMessages.length === 0" class="conversation-empty-hint">发起对话，或从左侧选择 session。</div>
    <div v-if="detailMessage" class="workbench-dialog-backdrop" role="presentation" @click.self="detailMessageId = null">
@@ -1,9 +1,182 @@
 // SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first.
-// Responsibility: Workbench scroll/follow runtime boundary over the shared bottom-follow primitive.
+// Responsibility: Workbench bottom-follow plus session/tab scroll persistence.
+// Mechanical source references:
+// - OpenCode context/layout-scroll.ts:16-21 persistence state.
+// - OpenCode context/layout-scroll.ts:35-51 seed/read.
+// - OpenCode context/layout-scroll.ts:53-81 debounce flush.
+// - OpenCode context/layout-scroll.ts:93-110 drop cleanup.

-import type { Ref } from "vue";
+import { computed, nextTick, onBeforeUnmount, watch, type ComputedRef, type Ref } from "vue";
 import { useBottomFollowScroll, type BottomFollowScrollOptions } from "@/composables/useBottomFollowScroll";
+import { readWorkbenchJson, writeWorkbenchJson } from "@/utils/workbench-storage-runtime";

-export function useWorkbenchBottomFollowScroll<T extends HTMLElement>(containerRef: Ref<T | null>, options: BottomFollowScrollOptions = {}) {
-  return useBottomFollowScroll(containerRef, options);
+export interface SessionScroll {
+  x: number;
+  y: number;
+}
+
+export type WorkbenchScrollMap = Record<string, SessionScroll>;
+
+export interface WorkbenchScrollPersistenceOptions {
+  debounceMs?: number;
+  getSnapshot: (sessionKey: string) => WorkbenchScrollMap | undefined;
+  onFlush: (sessionKey: string, scroll: WorkbenchScrollMap) => void;
+}
+
+export interface WorkbenchBottomFollowScrollOptions extends BottomFollowScrollOptions {
+  sessionKey?: string | Ref<string | null | undefined> | ComputedRef<string | null | undefined> | (() => string | null | undefined);
+  tabKey?: string;
+  persist?: boolean;
+}
+
+const WORKBENCH_SCROLL_STORAGE_KEY = "hwlab.workbench.scrollPersistence.v1";
+
+export function createWorkbenchScrollPersistenceRuntime(opts: WorkbenchScrollPersistenceOptions) {
+  const wait = Math.max(0, Math.trunc(opts.debounceMs ?? 200));
+  const cache: Record<string, WorkbenchScrollMap> = {};
+  const dirty = new Set<string>();
+  const timers = new Map<string, ReturnType<typeof setTimeout>>();
+
+  function clone(input?: WorkbenchScrollMap): WorkbenchScrollMap {
+    const out: WorkbenchScrollMap = {};
+    if (!input) return out;
+    for (const key of Object.keys(input)) {
+      const pos = input[key];
+      if (!pos) continue;
+      out[key] = { x: pos.x, y: pos.y };
+    }
+    return out;
+  }
+
+  function seed(sessionKey: string): void {
+    const next = clone(opts.getSnapshot(sessionKey));
+    const current = cache[sessionKey];
+    if (!current) {
+      cache[sessionKey] = next;
+      return;
+    }
+    if (Object.keys(current).length > 0) return;
+    if (Object.keys(next).length === 0) return;
+    cache[sessionKey] = next;
+  }
+
+  function scroll(sessionKey: string, tab: string): SessionScroll | undefined {
+    seed(sessionKey);
+    return cache[sessionKey]?.[tab] ?? opts.getSnapshot(sessionKey)?.[tab];
+  }
+
+  function schedule(sessionKey: string): void {
+    const prev = timers.get(sessionKey);
+    if (prev) clearTimeout(prev);
+    timers.set(sessionKey, setTimeout(() => flush(sessionKey), wait));
+  }
+
+  function setScroll(sessionKey: string, tab: string, pos: SessionScroll): void {
+    seed(sessionKey);
+    const prev = cache[sessionKey]?.[tab];
+    if (prev?.x === pos.x && prev?.y === pos.y) return;
+    cache[sessionKey] = { ...(cache[sessionKey] ?? {}), [tab]: { x: pos.x, y: pos.y } };
+    dirty.add(sessionKey);
+    schedule(sessionKey);
+  }
+
+  function flush(sessionKey: string): void {
+    const timer = timers.get(sessionKey);
+    if (timer) clearTimeout(timer);
+    timers.delete(sessionKey);
+    if (!dirty.has(sessionKey)) return;
+    dirty.delete(sessionKey);
+    opts.onFlush(sessionKey, clone(cache[sessionKey]));
+  }
+
+  function flushAll(): void {
+    for (const key of Array.from(dirty)) flush(key);
+  }
+
+  function drop(keys: string[]): void {
+    if (keys.length === 0) return;
+    for (const key of keys) {
+      const timer = timers.get(key);
+      if (timer) clearTimeout(timer);
+      timers.delete(key);
+      dirty.delete(key);
+      delete cache[key];
+    }
+  }
+
+  function dispose(): void {
+    flushAll();
+    drop(Array.from(timers.keys()));
+  }
+
+  return { cache, drop, flush, flushAll, scroll, seed, setScroll, dispose };
+}
+
+const scrollPersistence = createWorkbenchScrollPersistenceRuntime({
+  getSnapshot: (sessionKey) => readWorkbenchJson<Record<string, WorkbenchScrollMap>>(WORKBENCH_SCROLL_STORAGE_KEY, {}).value[sessionKey],
+  onFlush: (sessionKey, scroll) => {
+    const current = readWorkbenchJson<Record<string, WorkbenchScrollMap>>(WORKBENCH_SCROLL_STORAGE_KEY, {}).value;
+    writeWorkbenchJson(WORKBENCH_SCROLL_STORAGE_KEY, { ...current, [sessionKey]: scroll });
+  }
+});
+
+export function useWorkbenchBottomFollowScroll<T extends HTMLElement>(containerRef: Ref<T | null>, options: WorkbenchBottomFollowScrollOptions = {}) {
+  const bottom = useBottomFollowScroll(containerRef, options);
+  const tabKey = options.tabKey ?? "main";
+  const persist = options.persist !== false;
+  const sessionKey = computed(() => normalizeSessionKey(resolveSessionKey(options.sessionKey)));
+
+  watch([containerRef, sessionKey], async () => {
+    if (!persist) return;
+    await nextTick();
+    restoreScroll();
+  }, { flush: "post", immediate: true });
+
+  onBeforeUnmount(() => {
+    persistScroll();
+    scrollPersistence.flushAll();
+  });
+
+  function onScroll(): void {
+    bottom.onScroll();
+    persistScroll();
+  }
+
+  async function keepBottomAfterUpdate(): Promise<void> {
+    await bottom.keepBottomAfterUpdate();
+    if (!bottom.following.value) restoreScroll();
+  }
+
+  function scrollToBottom(): void {
+    bottom.scrollToBottom();
+    void nextTick().then(persistScroll);
+  }
+
+  function restoreScroll(): void {
+    const element = containerRef.value;
+    if (!element) return;
+    const pos = scrollPersistence.scroll(sessionKey.value, tabKey);
+    if (!pos) return;
+    element.scrollTo({ left: pos.x, top: pos.y, behavior: "auto" });
+  }
+
+  function persistScroll(): void {
+    if (!persist) return;
+    const element = containerRef.value;
+    if (!element) return;
+    scrollPersistence.setScroll(sessionKey.value, tabKey, { x: element.scrollLeft, y: element.scrollTop });
+  }
+
+  return { ...bottom, keepBottomAfterUpdate, onScroll, restoreScroll, scrollToBottom };
+}
+
+function resolveSessionKey(value: WorkbenchBottomFollowScrollOptions["sessionKey"]): string | null | undefined {
+  if (typeof value === "function") return value();
+  if (value && typeof value === "object" && "value" in value) return value.value;
+  return value;
+}
+
+function normalizeSessionKey(value: string | null | undefined): string {
+  const text = String(value ?? "").trim();
+  return text || "workbench";
 }
@@ -0,0 +1,75 @@
+// SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-0104010803 唯一投影 draft-2026-06-18-p0-unique-projection.
+// Responsibility: OpenCode-style Workbench realtime event reducer; store code consumes actions instead of branching on raw SSE events.
+// Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:21-48 global event classification, :93-270 session/message reducer action boundary.
+// Mechanical source: /root/opencode/packages/opencode/src/cli/cmd/run/session-data.ts:1-17 reducer side-effect boundary and replay notes.
+
+import type { WorkbenchRealtimeEvent } from "@/api/workbench-events";
+import { firstNonEmptyString } from "@/utils";
+
+export type WorkbenchRealtimeAction =
+  | { type: "trace.snapshot"; traceId: string | null; snapshot: WorkbenchRealtimeEvent["snapshot"] }
+  | { type: "trace.event"; traceId: string | null; event: WorkbenchRealtimeEvent["event"]; snapshot: WorkbenchRealtimeEvent["snapshot"]; realtimeEvent: WorkbenchRealtimeEvent }
+  | { type: "message.snapshot"; realtimeEvent: WorkbenchRealtimeEvent }
+  | { type: "turn.snapshot"; turn: NonNullable<WorkbenchRealtimeEvent["turn"]> }
+  | { type: "projection.error"; realtimeEvent: WorkbenchRealtimeEvent }
+  | { type: "trace.unavailable"; traceId: string | null; reason: string }
+  | { type: "ignore"; reason: string };
+
+export interface WorkbenchRealtimeReducerOutput {
+  activityLabel: string;
+  action: WorkbenchRealtimeAction;
+  diagnostic: {
+    module: "workbench-event-reducer";
+    eventName: string;
+    eventType: string | null;
+    actionType: WorkbenchRealtimeAction["type"];
+    traceId: string | null;
+    sessionId: string | null;
+    valuesRedacted: true;
+  };
+}
+
+export function reduceWorkbenchRealtimeEvent(event: WorkbenchRealtimeEvent, eventName: string): WorkbenchRealtimeReducerOutput {
+  const eventType = firstNonEmptyString(event.type);
+  const action = reduceRealtimeAction(event, eventName);
+  return {
+    activityLabel: eventType ? `realtime:${eventType}` : `realtime:${eventName}`,
+    action,
+    diagnostic: {
+      module: "workbench-event-reducer",
+      eventName,
+      eventType: eventType ?? null,
+      actionType: action.type,
+      traceId: realtimeTraceId(event),
+      sessionId: realtimeSessionId(event),
+      valuesRedacted: true
+    }
+  };
+}
+
+function reduceRealtimeAction(event: WorkbenchRealtimeEvent, eventName: string): WorkbenchRealtimeAction {
+  switch (event.type) {
+    case "trace.snapshot":
+      return { type: "trace.snapshot", traceId: realtimeTraceId(event), snapshot: event.snapshot ?? null };
+    case "trace.event":
+      return { type: "trace.event", traceId: realtimeTraceId(event), event: event.event ?? null, snapshot: event.snapshot ?? null, realtimeEvent: event };
+    case "message.snapshot":
+      return event.message ? { type: "message.snapshot", realtimeEvent: event } : { type: "ignore", reason: "message.snapshot.missing-message" };
+    case "turn.snapshot":
+      return event.turn ? { type: "turn.snapshot", turn: event.turn } : { type: "ignore", reason: "turn.snapshot.missing-turn" };
+    case "trace.unavailable":
+      return { type: "trace.unavailable", traceId: realtimeTraceId(event), reason: firstNonEmptyString(event.reason, "realtime-trace-unavailable") ?? "realtime-trace-unavailable" };
+    case "error":
+      return { type: "projection.error", realtimeEvent: event };
+  }
+  if (eventName === "workbench.error") return { type: "projection.error", realtimeEvent: event };
+  return { type: "ignore", reason: firstNonEmptyString(event.type, eventName, "unsupported") ?? "unsupported" };
+}
+
+function realtimeTraceId(event: WorkbenchRealtimeEvent): string | null {
+  return firstNonEmptyString(event.traceId, event.snapshot?.traceId, event.event?.traceId, event.message?.traceId) ?? null;
+}
+
+function realtimeSessionId(event: WorkbenchRealtimeEvent): string | null {
+  return firstNonEmptyString(event.sessionId, event.message?.sessionId, event.snapshot?.sessionId, event.event?.sessionId) ?? null;
+}
@@ -20,6 +20,7 @@ export type WorkbenchServerAction =
  | { type: "message.snapshot"; sessionId: string | null | undefined; message: ChatMessage | null | undefined }
  | { type: "session.status"; session: SessionStatusAuthority }
  | { type: "session.forget"; sessionId: string }
+  | { type: "session.cleanupDropped"; keepSessionIds: string[] }
  | { type: "turn.status"; turn: TurnStatusAuthority }
  | { type: "turn.forget"; traceId: string }
  | { type: "trace.snapshot"; traceId: string; trace: NonNullable<ChatMessage["runnerTrace"]> };
@@ -48,13 +49,9 @@ export function reduceWorkbenchServerState(state: WorkbenchServerState, action:
    case "session.status":
      return { ...state, sessionStatusById: { ...state.sessionStatusById, [action.session.sessionId]: action.session } };
    case "session.forget":
-      return {
-        ...state,
-        sessionOrder: state.sessionOrder.filter((id) => id !== action.sessionId),
-        sessionsById: withoutKey(state.sessionsById, action.sessionId),
-        messagesBySessionId: withoutKey(state.messagesBySessionId, action.sessionId),
-        sessionStatusById: withoutKey(state.sessionStatusById, action.sessionId)
-      };
+      return cleanupWorkbenchServerStateSessions(state, [action.sessionId]);
+    case "session.cleanupDropped":
+      return cleanupWorkbenchServerStateSessions(state, droppedSessionIds(state, new Set(action.keepSessionIds)));
    case "turn.status":
      return { ...state, turnStatusByTraceId: { ...state.turnStatusByTraceId, [action.turn.traceId]: action.turn } };
    case "turn.forget":
@@ -64,6 +61,25 @@ export function reduceWorkbenchServerState(state: WorkbenchServerState, action:
  }
 }

+// Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:50-91 cleanupDroppedSessionCaches.
+export function cleanupWorkbenchServerStateDroppedSessions(state: WorkbenchServerState, keepSessionIds: string[]): WorkbenchServerState {
+  return reduceWorkbenchServerState(state, { type: "session.cleanupDropped", keepSessionIds });
+}
+
+export function cleanupWorkbenchServerStateSessions(state: WorkbenchServerState, sessionIds: string[]): WorkbenchServerState {
+  const ids = unique(sessionIds.filter(Boolean));
+  if (ids.length === 0) return state;
+  const dropped = new Set(ids);
+  const scoped = cleanupSessionScopedAuthority(state, ids);
+  return {
+    ...scoped,
+    sessionOrder: scoped.sessionOrder.filter((sessionId) => !dropped.has(sessionId)),
+    sessionsById: omitRecordKeys(scoped.sessionsById, dropped),
+    messagesBySessionId: omitRecordKeys(scoped.messagesBySessionId, dropped),
+    sessionStatusById: omitRecordKeys(scoped.sessionStatusById, dropped)
+  };
+}
+
 export function selectSessionStatusAuthority(state: WorkbenchServerState): Readonly<Record<string, SessionStatusAuthority>> {
  return state.sessionStatusById;
 }
@@ -366,6 +382,52 @@ function withoutKey<T>(source: Record<string, T>, key: string): Record<string, T
  return next;
 }

+function cleanupSessionScopedAuthority(state: WorkbenchServerState, sessionIds: string[]): WorkbenchServerState {
+  if (sessionIds.length === 0) return state;
+  const dropped = new Set(sessionIds);
+  const traceIds = traceIdsForSessions(state, dropped);
+  return {
+    ...state,
+    sessionStatusById: omitRecordKeys(state.sessionStatusById, dropped),
+    turnStatusByTraceId: omitRecordKeysByValue(state.turnStatusByTraceId, (turn, traceId) => dropped.has(String(turn.sessionId ?? "")) || traceIds.has(traceId)),
+    traceById: omitRecordKeysByValue(state.traceById, (trace, traceId) => dropped.has(String(trace.sessionId ?? "")) || traceIds.has(traceId))
+  };
+}
+
+function droppedSessionIds(state: WorkbenchServerState, keep: Set<string>): string[] {
+  return [
+    ...Object.keys(state.sessionsById),
+    ...Object.keys(state.messagesBySessionId),
+    ...Object.keys(state.sessionStatusById),
+    ...state.sessionOrder
+  ].filter((sessionId, index, list) => Boolean(sessionId) && !keep.has(sessionId) && list.indexOf(sessionId) === index);
+}
+
+function traceIdsForSessions(state: WorkbenchServerState, sessionIds: Set<string>): Set<string> {
+  const traces = new Set<string>();
+  for (const sessionId of sessionIds) {
+    for (const message of state.messagesBySessionId[sessionId] ?? []) {
+      const traceId = message.traceId ?? message.runnerTrace?.traceId;
+      if (traceId) traces.add(traceId);
+    }
+  }
+  for (const [traceId, turn] of Object.entries(state.turnStatusByTraceId)) if (sessionIds.has(String(turn.sessionId ?? ""))) traces.add(traceId);
+  for (const [traceId, trace] of Object.entries(state.traceById)) if (sessionIds.has(String(trace.sessionId ?? ""))) traces.add(traceId);
+  return traces;
+}
+
+function omitRecordKeys<T>(record: Record<string, T>, keys: Set<string>): Record<string, T> {
+  const next: Record<string, T> = {};
+  for (const key of Object.keys(record)) if (!keys.has(key)) next[key] = record[key] as T;
+  return next;
+}
+
+function omitRecordKeysByValue<T>(record: Record<string, T>, shouldDrop: (value: T, key: string) => boolean): Record<string, T> {
+  const next: Record<string, T> = {};
+  for (const [key, value] of Object.entries(record)) if (!shouldDrop(value as T, key)) next[key] = value as T;
+  return next;
+}
+
 function unique(values: string[]): string[] {
  return [...new Set(values)];
 }
@@ -21,6 +21,11 @@ export interface WorkbenchSessionCacheTrimResult<T extends WorkbenchSessionCache
  trimmedMessageSessionIds: string[];
 }

+export interface WorkbenchDroppedSessionCacheCleanup<T extends WorkbenchSessionCacheState> {
+  state: T;
+  droppedSessionIds: string[];
+}
+
 export function trimWorkbenchSessionCache<T extends WorkbenchSessionCacheState>(state: T, policy: WorkbenchSessionCachePolicy): WorkbenchSessionCacheTrimResult<T> {
  const retain = new Set(normalizeIds(policy.retainSessionIds));
  const maxSessions = positiveIntegerOrNull(policy.maxSessions);
@@ -48,6 +53,27 @@ export function trimWorkbenchSessionCache<T extends WorkbenchSessionCacheState>(
  };
 }

+// Mechanical source: /root/opencode/packages/app/src/context/global-sync/event-reducer.ts:50-91 cleanupDroppedSessionCaches.
+export function cleanupDroppedWorkbenchSessionCaches<T extends WorkbenchSessionCacheState>(state: T, keepSessionIds: Iterable<string | null | undefined>): WorkbenchDroppedSessionCacheCleanup<T> {
+  const keep = new Set(normalizeIds(keepSessionIds));
+  const stale = [
+    ...Object.keys(state.sessionsById),
+    ...Object.keys(state.messagesBySessionId),
+    ...state.sessionOrder
+  ].filter((sessionId, index, list) => Boolean(sessionId) && !keep.has(sessionId) && list.indexOf(sessionId) === index);
+  if (stale.length === 0) return { state, droppedSessionIds: [] };
+  const dropped = new Set(stale);
+  return {
+    state: {
+      ...state,
+      sessionOrder: state.sessionOrder.filter((sessionId) => !dropped.has(sessionId)),
+      sessionsById: omitRecordKeys(state.sessionsById, dropped),
+      messagesBySessionId: omitRecordKeys(state.messagesBySessionId, dropped)
+    },
+    droppedSessionIds: stale
+  };
+}
+
 function keepSessionOrder(order: string[], retain: Set<string>, maxSessions: number): string[] {
  const kept: string[] = [];
  for (const sessionId of order) {
@@ -69,6 +95,12 @@ function pickRecordKeys<T>(record: Record<string, T>, keys: Set<string>): Record
  return next;
 }

+function omitRecordKeys<T>(record: Record<string, T>, keys: Set<string>): Record<string, T> {
+  const next: Record<string, T> = {};
+  for (const key of Object.keys(record)) if (!keys.has(key)) next[key] = record[key] as T;
+  return next;
+}
+
 function normalizeIds(values: Iterable<string | null | undefined> | null | undefined): string[] {
  if (!values) return [];
  return [...values].map((value) => String(value ?? "").trim()).filter(Boolean);
@@ -1,54 +1,80 @@
 // SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-0104010803 Workbench唯一投影 draft-2026-06-28-p0-d518-session-timeline-consistency.
 // Responsibility: Stable Workbench timeline row identities derived from authoritative message projections.
+// Mechanical source references:
+// - OpenCode pages/session/timeline/rows.ts:8-30 TimelineRowMap.
+// - OpenCode pages/session/timeline/rows.ts:32-105 tagged row/key/equality.
+// - OpenCode pages/session/timeline/rows.ts:107-242 constructMessageRows.
+// - OpenCode pages/session/timeline/rows.ts:284-329 unwrapErrorMessage.

 import type { ChatMessage } from "@/types";
 import { firstScopePart, workbenchTimelineRowScopeKey } from "@/utils/workbench-key";

-export interface WorkbenchMessageTimelineRow {
-  type: "message";
+export type WorkbenchTimelineRowType = "TurnGap" | "CommentStrip" | "UserMessage" | "TurnDivider" | "AssistantPart" | "Thinking" | "Retry" | "DiffSummary" | "Error";
+
+interface WorkbenchTimelineRowBase {
+  type: WorkbenchTimelineRowType;
  key: string;
  identity: string;
-  message: ChatMessage;
-  role: ChatMessage["role"];
  sessionId: string | null;
  traceId: string | null;
+  userMessageId: string | null;
+  message: ChatMessage | null;
+  role?: ChatMessage["role"];
 }

-export type WorkbenchTimelineRow = WorkbenchMessageTimelineRow;
+export interface WorkbenchMessageTimelineRow extends WorkbenchTimelineRowBase {
+  type: "UserMessage" | "AssistantPart";
+  message: ChatMessage;
+  role: ChatMessage["role"];
+  anchor?: boolean;
+  previousAssistantPart?: boolean;
+}
+
+export interface WorkbenchSyntheticTimelineRow extends WorkbenchTimelineRowBase {
+  type: "TurnGap" | "CommentStrip" | "TurnDivider" | "Thinking" | "Retry" | "DiffSummary" | "Error";
+  message: null;
+  label?: "compaction" | "interrupted";
+  text?: string | null;
+  reasoningHeading?: string | null;
+  diffs?: Array<Record<string, unknown>>;
+}
+
+export type WorkbenchTimelineRow = WorkbenchMessageTimelineRow | WorkbenchSyntheticTimelineRow;

 export function buildWorkbenchTimelineRows(messages: ChatMessage[]): WorkbenchTimelineRow[] {
  const rows: WorkbenchTimelineRow[] = [];
-  const indexByIdentity = new Map<string, number>();
-  messages.forEach((message, index) => {
+  let currentUserMessageId: string | null = null;
+  let assistantPartCount = 0;
+  const deduped = dedupeTimelineMessages(messages);
+  deduped.forEach((message, index) => {
    const identity = workbenchMessageIdentity(message, index);
-    const row: WorkbenchMessageTimelineRow = {
-      type: "message",
-      key: workbenchTimelineRowScopeKey({
-        sessionId: message.sessionId,
-        traceId: message.traceId,
-        type: "message",
-        role: message.role,
-        rowId: identity
-      }),
-      identity,
-      message,
-      role: message.role,
-      sessionId: firstScopePart(message.sessionId),
-      traceId: firstScopePart(message.traceId)
-    };
-    const existing = indexByIdentity.get(identity);
-    if (existing === undefined) {
-      indexByIdentity.set(identity, rows.length);
-      rows.push(row);
-    } else {
-      rows[existing] = row;
+    const sessionId = firstScopePart(message.sessionId);
+    const traceId = firstScopePart(message.traceId, message.runnerTrace?.traceId);
+    if (message.role === "user") {
+      if (currentUserMessageId) rows.push(syntheticRow("TurnGap", currentUserMessageId, sessionId, traceId));
+      currentUserMessageId = identity;
+      assistantPartCount = 0;
+      rows.push(messageRow("UserMessage", message, identity, sessionId, traceId, { anchor: true }));
+      if (messageHasCommentStrip(message)) rows.push(syntheticRow("CommentStrip", identity, sessionId, traceId));
+      return;
    }
+
+    const userMessageId = currentUserMessageId ?? previousUserMessageIdentity(deduped, index) ?? identity;
+    rows.push(messageRow("AssistantPart", message, identity, sessionId, traceId, { userMessageId, previousAssistantPart: assistantPartCount > 0 }));
+    assistantPartCount += 1;
+
+    if (messageIsThinking(message)) rows.push(syntheticRow("Thinking", userMessageId, sessionId, traceId, { reasoningHeading: reasoningHeading(message) }));
+    if (messageIsRetry(message)) rows.push(syntheticRow("Retry", userMessageId, sessionId, traceId));
+    const errorText = messageErrorText(message);
+    if (errorText) rows.push(syntheticRow("Error", userMessageId, sessionId, traceId, { text: errorText }));
+    const diffs = summaryDiffs(message);
+    if (diffs.length > 0 && !messageIsActive(message)) rows.push(syntheticRow("DiffSummary", userMessageId, sessionId, traceId, { diffs }));
  });
  return rows;
 }

 export function normalizeWorkbenchTimelineMessages(messages: ChatMessage[]): ChatMessage[] {
-  return buildWorkbenchTimelineRows(messages).map((row) => row.message);
+  return buildWorkbenchTimelineRows(messages).flatMap((row) => row.message ? [row.message] : []);
 }

 export function workbenchMessageIdentity(message: ChatMessage, fallbackIndex: number | null = null): string {
@@ -56,12 +82,188 @@ export function workbenchMessageIdentity(message: ChatMessage, fallbackIndex: nu
  if (authorityId) return authorityId;
  const turnId = firstScopePart(message.turnId);
  if (turnId) return `turn:${turnId}`;
-  const traceId = firstScopePart(message.traceId);
+  const traceId = firstScopePart(message.traceId, message.runnerTrace?.traceId);
  if (traceId && message.role === "agent") return `agent-trace:${traceId}`;
  return `message-index:${fallbackIndex ?? message.createdAt}`;
 }

 export function workbenchTimelineSignature(rows: WorkbenchTimelineRow[]): string {
-  return rows.map((row) => [row.key, row.message.status, String(row.message.text ?? "").length].join(":")).join("|");
+  return rows.map((row) => {
+    const state = row.message ? row.message.status : row.text ?? row.label ?? row.reasoningHeading ?? "";
+    const length = row.message ? String(row.message.text ?? "").length : String(state ?? "").length;
+    return [row.type, row.key, state, length].join(":");
+  }).join("|");
 }

+export function timelineRowMessage(row: WorkbenchTimelineRow): ChatMessage | null {
+  return row.message;
+}
+
+function dedupeTimelineMessages(messages: ChatMessage[]): ChatMessage[] {
+  const rows: ChatMessage[] = [];
+  const indexByIdentity = new Map<string, number>();
+  messages.forEach((message, index) => {
+    const identity = workbenchMessageIdentity(message, index);
+    const existing = indexByIdentity.get(identity);
+    if (existing === undefined) {
+      indexByIdentity.set(identity, rows.length);
+      rows.push(message);
+    } else {
+      rows[existing] = message;
+    }
+  });
+  return rows;
+}
+
+function messageRow(type: "UserMessage" | "AssistantPart", message: ChatMessage, identity: string, sessionId: string | null, traceId: string | null, extra: { userMessageId?: string | null; anchor?: boolean; previousAssistantPart?: boolean } = {}): WorkbenchMessageTimelineRow {
+  const userMessageId = extra.userMessageId ?? (type === "UserMessage" ? identity : null);
+  return {
+    type,
+    key: rowKey(type, sessionId, traceId, userMessageId, message.role, identity),
+    identity,
+    message,
+    role: message.role,
+    sessionId,
+    traceId,
+    userMessageId,
+    anchor: extra.anchor,
+    previousAssistantPart: extra.previousAssistantPart
+  };
+}
+
+function syntheticRow(type: WorkbenchSyntheticTimelineRow["type"], userMessageId: string | null, sessionId: string | null, traceId: string | null, extra: Partial<WorkbenchSyntheticTimelineRow> = {}): WorkbenchSyntheticTimelineRow {
+  const identity = `${type}:${userMessageId ?? "pending"}:${extra.label ?? extra.text ?? extra.reasoningHeading ?? ""}`;
+  return {
+    type,
+    key: rowKey(type, sessionId, traceId, userMessageId, null, identity),
+    identity,
+    sessionId,
+    traceId,
+    userMessageId,
+    message: null,
+    label: extra.label,
+    text: extra.text ?? null,
+    reasoningHeading: extra.reasoningHeading ?? null,
+    diffs: extra.diffs
+  };
+}
+
+function rowKey(type: WorkbenchTimelineRowType, sessionId: string | null, traceId: string | null, userMessageId: string | null, role: ChatMessage["role"] | null, rowId: string): string {
+  return workbenchTimelineRowScopeKey({ sessionId, traceId, type, role, rowId: userMessageId ? `${userMessageId}:${rowId}` : rowId });
+}
+
+function previousUserMessageIdentity(messages: ChatMessage[], beforeIndex: number): string | null {
+  for (let index = beforeIndex - 1; index >= 0; index -= 1) {
+    const message = messages[index];
+    if (message?.role === "user") return workbenchMessageIdentity(message, index);
+  }
+  return null;
+}
+
+function messageHasCommentStrip(message: ChatMessage): boolean {
+  return Array.isArray((message as Record<string, unknown>).comments) && ((message as Record<string, unknown>).comments as unknown[]).length > 0;
+}
+
+function messageIsThinking(message: ChatMessage): boolean {
+  return message.role === "agent" && messageIsActive(message) && !String(message.text ?? "").trim();
+}
+
+function messageIsRetry(message: ChatMessage): boolean {
+  return normalizeStatus(message.status) === "retry";
+}
+
+function messageIsActive(message: ChatMessage): boolean {
+  const status = normalizeStatus(message.status);
+  return status === "pending" || status === "running" || status === "busy" || status === "retry";
+}
+
+function messageErrorText(message: ChatMessage): string | null {
+  const source = recordValue(message.error) ?? recordValue(message.projection?.blocker) ?? recordValue(message.runnerTrace?.blocker);
+  const text = firstNonEmptyString(source?.userMessage, source?.message, source?.summary, source?.code ? String(source.code) : null);
+  return text ? unwrapErrorMessage(text) : null;
+}
+
+function summaryDiffs(message: ChatMessage): Array<Record<string, unknown>> {
+  const summary = recordValue((message as Record<string, unknown>).summary);
+  const diffs = Array.isArray(summary?.diffs) ? summary.diffs : [];
+  const seen = new Set<string>();
+  return diffs.reduceRight<Array<Record<string, unknown>>>((result, item) => {
+    const diff = recordValue(item);
+    const file = firstNonEmptyString(diff?.file);
+    if (!diff || !file || seen.has(file)) return result;
+    seen.add(file);
+    result.push(diff);
+    return result;
+  }, []).reverse();
+}
+
+function reasoningHeading(message: ChatMessage): string | null {
+  const traceEvents = Array.isArray(message.runnerTrace?.events) ? message.runnerTrace.events : [];
+  const reasoning = traceEvents.map((event) => firstNonEmptyString(event.text, event.message, event.outputSummary)).find(Boolean) ?? null;
+  return reasoning ? extractHeading(reasoning) : null;
+}
+
+function extractHeading(text: string): string | null {
+  const markdown = text.replace(/\r\n?/g, "\n");
+  const html = markdown.match(/<h[1-6][^>]*>([\s\S]*?)<\/h[1-6]>/i);
+  if (html?.[1]) return cleanHeading(html[1].replace(/<[^>]+>/g, " "));
+  const atx = markdown.match(/^\s{0,3}#{1,6}[ \t]+(.+?)(?:[ \t]+#+[ \t]*)?$/m);
+  if (atx?.[1]) return cleanHeading(atx[1]);
+  const setext = markdown.match(/^([^\n]+)\n(?:=+|-)\s*$/m);
+  if (setext?.[1]) return cleanHeading(setext[1]);
+  const strong = markdown.match(/^\s*(?:\*\*|__)(.+?)(?:\*\*|__)\s*$/m);
+  if (strong?.[1]) return cleanHeading(strong[1]);
+  return null;
+}
+
+function cleanHeading(value: string): string | null {
+  const text = value.replace(/`([^`]+)`/g, "$1").replace(/\[([^\]]+)\]\([^)]+\)/g, "$1").replace(/[*_~]+/g, "").trim();
+  return text || null;
+}
+
+function unwrapErrorMessage(message: string): string {
+  const text = message.replace(/^Error:\s*/, "").trim();
+  const parse = (value: string): unknown => {
+    try { return JSON.parse(value) as unknown; } catch { return undefined; }
+  };
+  const read = (value: string): unknown => {
+    const first = parse(value);
+    return typeof first === "string" ? parse(first.trim()) : first;
+  };
+  let json = read(text);
+  if (json === undefined) {
+    const start = text.indexOf("{");
+    const end = text.lastIndexOf("}");
+    if (start !== -1 && end > start) json = read(text.slice(start, end + 1));
+  }
+  const source = recordValue(json);
+  if (!source) return message;
+  const error = recordValue(source.error);
+  if (error) {
+    const type = firstNonEmptyString(error.type);
+    const msg = firstNonEmptyString(error.message);
+    if (type && msg) return `${type}: ${msg}`;
+    if (msg) return msg;
+    if (type) return type;
+    const code = firstNonEmptyString(error.code);
+    if (code) return code;
+  }
+  return firstNonEmptyString(source.message, source.error) ?? message;
+}
+
+function normalizeStatus(status: unknown): string {
+  return String(status ?? "").trim().toLowerCase().replace(/_/gu, "-");
+}
+
+function recordValue(value: unknown): Record<string, unknown> | null {
+  return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
+}
+
+function firstNonEmptyString(...values: unknown[]): string | null {
+  for (const value of values) {
+    if (typeof value !== "string" && typeof value !== "number") continue;
+    const text = String(value).trim();
+    if (text) return text;
+  }
+  return null;
+}
@@ -5,19 +5,20 @@ import { computed, nextTick, ref } from "vue";
 import { defineStore } from "pinia";
 import { api } from "@/api";
 import { createWorkbenchHealthProbeCache } from "@/utils/workbench-health";
-import { normalizeApiErrorRecord as normalizeRuntimeApiErrorRecord, normalizeErrorDiagnostic as normalizeRuntimeErrorDiagnostic } from "@/utils/workbench-error-runtime";
-import { createWorkbenchReadHydrationRuntime, shouldCooldownWorkbenchReadFailure as shouldCooldownWorkbenchReadRuntimeFailure } from "@/utils/workbench-refresh-runtime";
+import { agentErrorFromProjection, normalizeApiErrorRecord, normalizeErrorDiagnostic, normalizeProjectionBlocker, normalizeProjectionDiagnostic, projectionDiagnosticFromApiFailure, projectionDiagnosticFromFailure } from "@/utils/workbench-error-runtime";
+import { createWorkbenchReadHydrationRuntime, createWorkbenchScheduledTaskRuntime, createWorkbenchTraceHydrationQueueRuntime, shouldCooldownWorkbenchReadFailure as shouldCooldownWorkbenchReadRuntimeFailure } from "@/utils/workbench-refresh-runtime";
 import { readWorkbenchJson, readWorkbenchNumber, readWorkbenchString, removeWorkbenchStorageKey, writeWorkbenchJson, writeWorkbenchString } from "@/utils/workbench-storage-runtime";
-import { createWorkbenchStreamTransportRuntime, type WorkbenchRealtimeEvent } from "@/utils/workbench-realtime-runtime";
+import { createWorkbenchStreamTransportRuntime, type WorkbenchRealtimeEvent, type WorkbenchStreamTransportRecovery } from "@/utils/workbench-realtime-runtime";
 import { mergeRunnerTrace, snapshotToRunnerTrace, type TraceSnapshot } from "@/composables/useTraceSubscription";
 import type { AgentChatResponse, AgentChatResultResponse, AgentRunProvenance, ApiError, ApiResult, ChatMessage, ErrorDiagnostic, LiveSurface, ProjectionBlocker, ProjectionDiagnostic, ProviderProfile, TraceEvent, WorkbenchSessionRecord, WorkbenchTurnTimingProjection } from "@/types";
 import { firstNonEmptyString, nextProtocolId, normalizeWorkbenchSessionId, normalizeWorkbenchSessionRouteId } from "@/utils";
-import { workbenchRealtimeScopeKey } from "@/utils/workbench-key";
+import { composeWorkbenchScopedKey } from "@/utils/workbench-key";
 import { failWorkbenchSessionSwitch, failWorkbenchSubmitJourney, finishWorkbenchSessionSwitchFullLoad, markWorkbenchSubmitApiAccepted, markWorkbenchTraceEventsReceived, markWorkbenchTraceProjected, recordWorkbenchLoadingState, startWorkbenchSessionSwitch, startWorkbenchSubmitJourney } from "@/utils/workbench-performance";
 import { RECENT_DRAFTS_STORAGE_KEY, appendSessionPage, defaultProviderProfileOptions, isArchivedSession, mergeSessionIntoList, normalizeChatMessageStatus, normalizeRecentDrafts, normalizeWorkbenchMessageTitle, providerProfileOptionsFromPayload, recordRecentDraft, resolveCancelableAgentMessage, resolveComposerState, shouldShowSessionListLoading, sortSessionTabs, stableSessionList, type DraftEntry, type ProviderProfileOption, type TurnStatusAuthority } from "./workbench-session";
 import { initialWorkbenchSessionIdFromLocation } from "./workbench-projection";
-import { createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state";
-import { trimWorkbenchSessionCache } from "./workbench-session-cache";
+import { cleanupWorkbenchServerStateSessions, createWorkbenchServerState, reduceWorkbenchServerState, selectActiveMessages, selectActiveSession, selectSessionList, selectSessionStatusAuthority, selectTraceAuthorityById, selectTurnStatusAuthority, type WorkbenchServerAction } from "./workbench-server-state";
+import { cleanupDroppedWorkbenchSessionCaches, trimWorkbenchSessionCache } from "./workbench-session-cache";
+import { reduceWorkbenchRealtimeEvent, type WorkbenchRealtimeAction } from "./workbench-event-reducer";

 const DEFAULT_CODE_AGENT_TIMEOUT_MS = 1_800_000;
 const DEFAULT_GATEWAY_TIMEOUT_MS = 120_000;
@@ -83,22 +84,14 @@ export const useWorkbenchStore = defineStore("workbench", () => {
  const sessionStatusAuthority = computed(() => selectSessionStatusAuthority(serverState.value));
  const turnStatusAuthority = computed(() => selectTurnStatusAuthority(serverState.value));
  const traceAuthorityById = computed(() => selectTraceAuthorityById(serverState.value));
-  const traceHydrationInFlight = new Set<string>();
-  const traceHydrationQueued = new Set<string>();
-  const traceHydrationQueue: ChatMessage[] = [];
-  const forcedTraceHydrationRetryTimers = new Map<string, number>();
-  let traceHydrationPumpActive = false;
-  const terminalRealtimeRefreshInFlight = new Set<string>();
-  const realtimeSessionMessagesInFlight = new Set<string>();
-  const realtimeSessionMessagesLastRefreshAtById = new Map<string, number>();
-  const realtimeSessionRefreshInFlight = new Set<string>();
-  const realtimeErrorGapFillLastAtByKey = new Map<string, number>();
-  const activeTraceRestGapFillTimers = new Map<string, number>();
  const realtimeTransport = createWorkbenchStreamTransportRuntime();
-  const realtimeOutboxSeqByKey = new Map<string, number>();
-  const sessionListRefreshInFlight = new Map<string, Promise<void>>();
-  const sessionListRefreshTimers = new Map<string, number>();
-  const sessionListLastRefreshAtByKey = new Map<string, number>();
+  const traceHydrationRuntime = createWorkbenchTraceHydrationQueueRuntime<ChatMessage>({ concurrency: TRACE_HYDRATION_BACKGROUND_CONCURRENCY, delayMs: TRACE_HYDRATION_BACKGROUND_DELAY_MS });
+  const forcedTraceHydrationRetryRuntime = createWorkbenchScheduledTaskRuntime();
+  const terminalRealtimeRefreshRuntime = createWorkbenchScheduledTaskRuntime();
+  const realtimeSessionMessagesRuntime = createWorkbenchScheduledTaskRuntime();
+  const realtimeSessionRefreshRuntime = createWorkbenchScheduledTaskRuntime();
+  const activeTraceRestGapFillRuntime = createWorkbenchScheduledTaskRuntime();
+  const sessionListRefreshRuntime = createWorkbenchScheduledTaskRuntime();
  const workbenchProjectionSignalSourceId = nextProtocolId("wbtab");
  let workbenchProjectionSignalChannel: BroadcastChannel | null = null;
  const workbenchReadHydrationRuntime = createWorkbenchReadHydrationRuntime({ concurrency: WORKBENCH_READ_HYDRATION_CONCURRENCY, failureCooldownMs: WORKBENCH_READ_FAILURE_COOLDOWN_MS });
@@ -312,17 +305,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
    const requestIncludeSessionId = firstNonEmptyString(includeSessionId);
    const requestLimit = currentSessionListLimit();
    const requestKey = sessionListRefreshKey(requestIncludeSessionId, requestLimit);
-    const existing = sessionListRefreshInFlight.get(requestKey);
-    if (existing) return existing;
-    const now = Date.now();
-    const lastRefreshAt = sessionListLastRefreshAtByKey.get(requestKey) ?? 0;
-    if (options.force !== true && lastRefreshAt > 0 && now - lastRefreshAt < SESSION_LIST_MIN_REFRESH_INTERVAL_MS) return;
-    sessionListLastRefreshAtByKey.set(requestKey, now);
-    const refresh = refreshSessionsNow(requestIncludeSessionId, requestLimit).finally(() => {
-      if (sessionListRefreshInFlight.get(requestKey) === refresh) sessionListRefreshInFlight.delete(requestKey);
-    });
-    sessionListRefreshInFlight.set(requestKey, refresh);
-    return refresh;
+    await sessionListRefreshRuntime.run(requestKey, () => refreshSessionsNow(requestIncludeSessionId, requestLimit), { force: options.force, minIntervalMs: SESSION_LIST_MIN_REFRESH_INTERVAL_MS, reason: "session-list" });
  }

  async function refreshSessionsNow(includeSessionId: string | null, limit: number): Promise<void> {
@@ -350,20 +333,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
      void refreshSessions(requestIncludeSessionId);
      return;
    }
-    const existing = sessionListRefreshTimers.get(requestKey);
-    if (existing) return;
-    const lastRefreshAt = sessionListLastRefreshAtByKey.get(requestKey) ?? 0;
-    const cooldownMs = lastRefreshAt > 0 ? Math.max(0, SESSION_LIST_MIN_REFRESH_INTERVAL_MS - (Date.now() - lastRefreshAt)) : 0;
-    const boundedDelayMs = Math.max(0, Math.trunc(delayMs), cooldownMs);
-    const timer = window.setTimeout(() => {
-      sessionListRefreshTimers.delete(requestKey);
-      void refreshSessions(requestIncludeSessionId);
-    }, boundedDelayMs);
-    sessionListRefreshTimers.set(requestKey, timer);
+    sessionListRefreshRuntime.schedule(requestKey, () => refreshSessions(requestIncludeSessionId), { delayMs, minIntervalMs: SESSION_LIST_MIN_REFRESH_INTERVAL_MS, reason: "session-list" });
  }

  function sessionListRefreshKey(includeSessionId: string | null | undefined, limit: number): string {
-    return `${firstNonEmptyString(includeSessionId) ?? ""}|${limit}`;
+    return composeWorkbenchScopedKey("workbench.session-list", firstNonEmptyString(includeSessionId), limit);
  }

  async function loadMoreSessions(): Promise<void> {
@@ -409,8 +383,16 @@ export const useWorkbenchStore = defineStore("workbench", () => {

  function rememberSessionList(source: WorkbenchSessionRecord[]): void {
    reduceServerState({ type: "session.list", sessions: source });
-    const trimmed = trimWorkbenchSessionCache(serverState.value, { retainSessionIds: [activeSessionId.value, explicitSessionId.value], maxSessions: currentSessionListLimit() });
-    serverState.value = trimmed.state;
+    const beforeTrim = serverState.value;
+    const trimmed = trimWorkbenchSessionCache(beforeTrim, { retainSessionIds: [activeSessionId.value, explicitSessionId.value], maxSessions: currentSessionListLimit() });
+    const cleaned = cleanupDroppedWorkbenchSessionCaches(trimmed.state, trimmed.state.sessionOrder);
+    const authorityCleaned = cleanupWorkbenchServerStateSessions(beforeTrim, [...trimmed.evictedSessionIds, ...cleaned.droppedSessionIds]);
+    serverState.value = {
+      ...authorityCleaned,
+      sessionOrder: cleaned.state.sessionOrder,
+      sessionsById: cleaned.state.sessionsById,
+      messagesBySessionId: cleaned.state.messagesBySessionId
+    };
  }

  function forgetSession(sessionId: string): void {
@@ -482,61 +464,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
  }

  function workbenchReadCooldownKey(kind: string, id: string): string {
-    return `${kind}:${id}`;
-  }
-
-  function createWorkbenchReadCooldownResult(key: string, cooldownUntilMs: number): ApiResult<unknown> {
-    const retryAfterMs = Math.max(0, cooldownUntilMs - Date.now());
-    const message = `Workbench read hydration is cooling down after a transient failure (${key}); retry after ${retryAfterMs}ms.`;
-    const diagnostic = {
-      code: "workbench_read_hydration_cooldown",
-      category: "network",
-      source: "workbench-web",
-      message,
-      retryable: true,
-    };
-    return {
-      ok: false,
-      status: 503,
-      data: null,
-      error: message,
-      apiError: {
-        code: "workbench_read_hydration_cooldown",
-        category: "network",
-        source: "workbench-web",
-        message,
-        retryable: true,
-        diagnostic,
-      },
-      diagnostic,
-    } as ApiResult<unknown>;
-  }
-
-  function createWorkbenchReadThrottledResult(key: string, nextAllowedAtMs: number): ApiResult<unknown> {
-    const retryAfterMs = Math.max(0, nextAllowedAtMs - Date.now());
-    const message = `Workbench read hydration is throttled (${key}); retry after ${retryAfterMs}ms.`;
-    const diagnostic = {
-      code: "workbench_read_hydration_throttled",
-      category: "throttle",
-      source: "workbench-web",
-      message,
-      retryable: true,
-    };
-    return {
-      ok: false,
-      status: 0,
-      data: null,
-      error: message,
-      apiError: {
-        code: "workbench_read_hydration_throttled",
-        category: "throttle",
-        source: "workbench-web",
-        message,
-        retryable: true,
-        diagnostic,
-      },
-      diagnostic,
-    } as ApiResult<unknown>;
+    return composeWorkbenchScopedKey("workbench.read", kind, id);
  }

  function isApiResultLike(value: unknown): value is ApiResult<unknown> {
@@ -593,17 +521,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {

  async function refreshRealtimeSessionMessages(sessionId: string | null | undefined, reason: string, options: { force?: boolean } = {}): Promise<void> {
    const id = normalizeWorkbenchSessionId(sessionId);
-    if (!id || id !== activeSessionId.value || realtimeSessionMessagesInFlight.has(id)) return;
-    const lastRefreshAt = realtimeSessionMessagesLastRefreshAtById.get(id) ?? 0;
-    if (!options.force && lastRefreshAt > 0 && Date.now() - lastRefreshAt < WORKBENCH_REALTIME_SESSION_MESSAGES_MIN_REFRESH_MS) return;
-    realtimeSessionMessagesLastRefreshAtById.set(id, Date.now());
-    realtimeSessionMessagesInFlight.add(id);
-    try {
+    if (!id || id !== activeSessionId.value) return;
+    await realtimeSessionMessagesRuntime.run(composeWorkbenchScopedKey("workbench.realtime.session-messages", id), async () => {
      recordActivity(reason);
      await refreshSessionMessageProjectionPage(id, { force: true });
-    } finally {
-      realtimeSessionMessagesInFlight.delete(id);
-    }
+    }, { force: options.force, minIntervalMs: WORKBENCH_REALTIME_SESSION_MESSAGES_MIN_REFRESH_MS, reason });
  }

  async function refreshMessageProjectionForTrace(sessionId: string | null | undefined, traceId: string, options: { force?: boolean } = {}): Promise<void> {
@@ -890,28 +812,27 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
  async function hydrateTraceEventsForMessage(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
    const traceId = message.traceId ?? message.runnerTrace?.traceId;
    if (!traceId) return;
-    if (traceHydrationInFlight.has(traceId)) {
-      if (options.force) scheduleForcedTraceHydrationRetry(message);
-      return;
-    }
-    traceHydrationInFlight.add(traceId);
-    try {
-      let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace);
-      for (let page = 0; page < TRACE_HYDRATION_MAX_PAGES; page += 1) {
-        const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force });
-        if (!result.ok || !result.data) {
-          if (shouldSuppressTransientWorkbenchReadFailure(result)) return;
-          if (messageHasCompletedFinalResponse(message)) return;
-          applyProjectionDiagnostic(traceId, projectionDiagnosticFromApiFailure(result, { code: "trace_hydration_failed", message: result.error ?? "Trace 更新超时，运行记录暂不可见。", health: result.status === 0 ? "unavailable" : "degraded" }));
-          return;
-        }
-        applyTraceHydrationResult(traceId, result.data);
-        const nextProjectedSeq = traceNextProjectedSeq(result.data, afterProjectedSeq);
-        if (result.data.hasMore !== true || nextProjectedSeq <= afterProjectedSeq) return;
-        afterProjectedSeq = nextProjectedSeq;
+    await traceHydrationRuntime.run(traceId, () => hydrateTraceEventsForMessageNow(message, options), {
+      onAlreadyRunning: options.force ? () => scheduleForcedTraceHydrationRetry(message) : undefined
+    });
+  }
+
+  async function hydrateTraceEventsForMessageNow(message: ChatMessage, options: { force?: boolean } = {}): Promise<void> {
+    const traceId = message.traceId ?? message.runnerTrace?.traceId;
+    if (!traceId) return;
+    let afterProjectedSeq = traceHydrationProjectedSeq(message.runnerTrace);
+    for (let page = 0; page < TRACE_HYDRATION_MAX_PAGES; page += 1) {
+      const result = await fetchTraceHydrationPage(traceId, afterProjectedSeq, { force: options.force });
+      if (!result.ok || !result.data) {
+        if (shouldSuppressTransientWorkbenchReadFailure(result)) return;
+        if (messageHasCompletedFinalResponse(message)) return;
+        applyProjectionDiagnostic(traceId, projectionDiagnosticFromApiFailure(result, { code: "trace_hydration_failed", message: result.error ?? "Trace 更新超时，运行记录暂不可见。", health: result.status === 0 ? "unavailable" : "degraded" }));
+        return;
      }
-    } finally {
-      traceHydrationInFlight.delete(traceId);
+      applyTraceHydrationResult(traceId, result.data);
+      const nextProjectedSeq = traceNextProjectedSeq(result.data, afterProjectedSeq);
+      if (result.data.hasMore !== true || nextProjectedSeq <= afterProjectedSeq) return;
+      afterProjectedSeq = nextProjectedSeq;
    }
  }

@@ -939,20 +860,17 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project

  function scheduleForcedTraceHydrationRetry(message: ChatMessage): void {
    const traceId = firstNonEmptyString(message.traceId, message.runnerTrace?.traceId);
-    if (!traceId || typeof window === "undefined" || forcedTraceHydrationRetryTimers.has(traceId)) return;
-    const timer = window.setTimeout(() => {
-      forcedTraceHydrationRetryTimers.delete(traceId);
+    if (!traceId || typeof window === "undefined") return;
+    forcedTraceHydrationRetryRuntime.schedule(composeWorkbenchScopedKey("workbench.trace-hydration.retry", traceId), async () => {
      const ownerSessionId = traceOwnerSessionId(traceId, messageSessionAuthority(message));
      const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
      const latest = latestMessageForTrace(traceId, ownerMessages) ?? message;
-      void hydrateTraceEventsForMessage(latest, { force: true });
-    }, TRACE_HYDRATION_RETRY_DELAY_MS);
-    forcedTraceHydrationRetryTimers.set(traceId, timer);
+      await hydrateTraceEventsForMessage(latest, { force: true });
+    }, { delayMs: TRACE_HYDRATION_RETRY_DELAY_MS, reason: "trace-hydration-retry" });
  }

  async function hydrateTraceEvents(source: ChatMessage[] = messages.value): Promise<void> {
    for (const message of traceHydrationCandidates(source)) queueTraceHydration(message);
-    void pumpTraceHydrationQueue();
  }

  function traceHydrationCandidates(source: ChatMessage[]): ChatMessage[] {
@@ -962,30 +880,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
  function queueTraceHydration(message: ChatMessage): void {
    const traceId = message.traceId ?? message.runnerTrace?.traceId;
    if (!traceId) return;
-    if (traceHydrationInFlight.has(traceId) || traceHydrationQueued.has(traceId)) return;
-    traceHydrationQueued.add(traceId);
-    traceHydrationQueue.push(message);
-  }
-
-  async function pumpTraceHydrationQueue(): Promise<void> {
-    if (traceHydrationPumpActive) return;
-    traceHydrationPumpActive = true;
-    try {
-      while (traceHydrationQueue.length > 0) {
-        const batch = traceHydrationQueue.splice(0, TRACE_HYDRATION_BACKGROUND_CONCURRENCY);
-        await Promise.all(batch.map(async (message) => {
-          const traceId = message.traceId ?? message.runnerTrace?.traceId;
-          try {
-            await hydrateTraceEventsForMessage(message);
-          } finally {
-            if (traceId) traceHydrationQueued.delete(traceId);
-          }
-        }));
-        if (traceHydrationQueue.length > 0) await delayTraceHydrationRetry(TRACE_HYDRATION_BACKGROUND_DELAY_MS);
-      }
-    } finally {
-      traceHydrationPumpActive = false;
-    }
+    traceHydrationRuntime.enqueue(traceId, () => hydrateTraceEventsForMessageNow(message));
  }

  function messagesWithTraceAuthority(source: ChatMessage[]): ChatMessage[] {
@@ -1135,26 +1030,21 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
  function restartRealtime(reason: string): void {
    const traceId = realtimeTraceId();
    const sessionId = selectedSessionId.value ?? null;
-    const key = workbenchRealtimeScopeKey(sessionId, traceId);
-    const afterSeq = realtimeOutboxSeqByKey.get(key) ?? null;
+    void reason;
    realtimeTransport.restart({
      sessionId,
      traceId,
-      afterSeq,
+      errorRecoveryMinMs: WORKBENCH_REALTIME_ERROR_GAP_FILL_MIN_MS,
      onOpen: () => undefined,
-      onError: () => handleRealtimeStreamError(sessionId, traceId),
+      onRecovery: (recovery) => handleRealtimeRecovery(recovery),
      onEvent: (event, eventName) => applyRealtimeEvent(event, eventName)
    });
  }

-  function handleRealtimeStreamError(sessionId: string | null, traceId: string | null): void {
-    const activeId = normalizeWorkbenchSessionId(sessionId ?? selectedSessionId.value);
-    const activeTraceId = firstNonEmptyString(traceId, realtimeTraceId());
+  function handleRealtimeRecovery(recovery: WorkbenchStreamTransportRecovery): void {
+    const activeId = normalizeWorkbenchSessionId(recovery.sessionId ?? selectedSessionId.value);
+    const activeTraceId = firstNonEmptyString(recovery.traceId, realtimeTraceId());
    if (!activeId && !activeTraceId) return;
-    const key = workbenchRealtimeScopeKey(activeId, activeTraceId);
-    const lastAt = realtimeErrorGapFillLastAtByKey.get(key) ?? 0;
-    if (lastAt > 0 && Date.now() - lastAt < WORKBENCH_REALTIME_ERROR_GAP_FILL_MIN_MS) return;
-    realtimeErrorGapFillLastAtByKey.set(key, Date.now());
    if (activeId === activeSessionId.value) void refreshRealtimeSessionMessages(activeId, "realtime-error:messages", { force: true });
    if (activeId) scheduleSessionListRefresh(activeId, SESSION_LIST_REALTIME_REFRESH_DELAY_MS);
    if (!activeTraceId || !shouldApplyActiveTraceAuthority(activeTraceId, activeId)) return;
@@ -1166,20 +1056,13 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
  function scheduleActiveTraceRestGapFill(traceId: string | null | undefined, reason: string, delayMs = WORKBENCH_ACTIVE_TRACE_REST_GAP_FILL_INITIAL_MS): void {
    const id = firstNonEmptyString(traceId);
    if (!id || typeof window === "undefined") return;
-    clearActiveTraceRestGapFill(id);
-    const timer = window.setTimeout(() => {
-      activeTraceRestGapFillTimers.delete(id);
-      void refreshActiveTraceFromRest(id, reason);
-    }, Math.max(0, Math.trunc(delayMs)));
-    activeTraceRestGapFillTimers.set(id, timer);
+    activeTraceRestGapFillRuntime.schedule(composeWorkbenchScopedKey("workbench.active-trace-gap", id), () => refreshActiveTraceFromRest(id, reason), { delayMs, replaceTimer: true, reason });
  }

  function clearActiveTraceRestGapFill(traceId: string | null | undefined): void {
    const id = firstNonEmptyString(traceId);
    if (!id || typeof window === "undefined") return;
-    const timer = activeTraceRestGapFillTimers.get(id);
-    if (timer) window.clearTimeout(timer);
-    activeTraceRestGapFillTimers.delete(id);
+    activeTraceRestGapFillRuntime.cancelTimer(composeWorkbenchScopedKey("workbench.active-trace-gap", id));
  }

  async function refreshActiveTraceFromRest(traceId: string, reason: string): Promise<void> {
@@ -1221,22 +1104,6 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
    return normalizeWorkbenchSessionId(firstNonEmptyString(event.sessionId, turn?.sessionId, snapshot?.sessionId, traceEvent?.sessionId));
  }

-  function rememberRealtimeOutboxSeq(event: WorkbenchRealtimeEvent): void {
-    const outboxSeq = realtimeOutboxSeq(event);
-    if (outboxSeq === null) return;
-    const sessionId = realtimeEventSessionId(event) ?? selectedSessionId.value ?? null;
-    const traceId = firstNonEmptyString(event.traceId, event.event?.traceId, event.snapshot?.traceId, currentRequest.value?.traceId, activeTraceIdFromMessages(messages.value, turnStatusAuthority.value));
-    if (!sessionId && !traceId) return;
-    const key = workbenchRealtimeScopeKey(sessionId, traceId);
-    const previous = realtimeOutboxSeqByKey.get(key) ?? 0;
-    if (outboxSeq > previous) realtimeOutboxSeqByKey.set(key, outboxSeq);
-  }
-
-  function realtimeOutboxSeq(event: WorkbenchRealtimeEvent): number | null {
-    const parsed = Number(event.cursor?.outboxSeq ?? event.outboxSeq);
-    return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : null;
-  }
-
  function traceResultSessionId(result: AgentChatResultResponse | TraceSnapshot | Record<string, unknown> | null | undefined): string | null {
    const value = recordValue(result);
    const runnerTrace = recordValue(value?.runnerTrace);
@@ -1299,31 +1166,33 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
  }

  function applyRealtimeEvent(event: WorkbenchRealtimeEvent, eventName: string): void {
-    recordActivity(event.type ? `realtime:${event.type}` : `realtime:${eventName}`);
-    rememberRealtimeOutboxSeq(event);
-    if (event.type === "trace.snapshot") {
-      applyRealtimeTraceSnapshot(event.traceId, event.snapshot);
-      return;
-    }
-    if (event.type === "trace.event") {
-      applyRealtimeTraceEvent(event.traceId, event.event, event.snapshot, event);
-      return;
-    }
-    if (event.type === "message.snapshot" && event.message) {
-      applyRealtimeMessageSnapshot(event);
-      return;
-    }
-    if (event.type === "turn.snapshot" && event.turn) {
-      applyRealtimeTurnSnapshot(event.turn);
-      return;
-    }
-    if (eventName === "workbench.error" || event.type === "error") {
-      applyRealtimeProjectionError(event);
-      return;
-    }
-    if (event.type === "trace.unavailable") {
-      const traceId = firstNonEmptyString(event.traceId);
-      if (traceId) void clearActiveTrace(traceId, "realtime-trace-unavailable");
+    const reduced = reduceWorkbenchRealtimeEvent(event, eventName);
+    recordActivity(reduced.activityLabel);
+    applyWorkbenchRealtimeAction(reduced.action);
+  }
+
+  function applyWorkbenchRealtimeAction(action: WorkbenchRealtimeAction): void {
+    switch (action.type) {
+      case "trace.snapshot":
+        applyRealtimeTraceSnapshot(action.traceId, action.snapshot);
+        return;
+      case "trace.event":
+        applyRealtimeTraceEvent(action.traceId, action.event, action.snapshot, action.realtimeEvent);
+        return;
+      case "message.snapshot":
+        applyRealtimeMessageSnapshot(action.realtimeEvent);
+        return;
+      case "turn.snapshot":
+        applyRealtimeTurnSnapshot(action.turn);
+        return;
+      case "projection.error":
+        applyRealtimeProjectionError(action.realtimeEvent);
+        return;
+      case "trace.unavailable":
+        if (action.traceId) void clearActiveTrace(action.traceId, action.reason);
+        return;
+      case "ignore":
+        return;
    }
  }

@@ -1366,9 +1235,8 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project

  async function refreshTerminalTraceFromRest(traceId: string, reason: string): Promise<void> {
    const id = firstNonEmptyString(traceId);
-    if (!id || terminalRealtimeRefreshInFlight.has(id)) return;
-    terminalRealtimeRefreshInFlight.add(id);
-    try {
+    if (!id) return;
+    await terminalRealtimeRefreshRuntime.run(composeWorkbenchScopedKey("workbench.terminal-trace-refresh", id), async () => {
      const ownerBefore = traceOwnerSessionId(id, null);
      if (ownerBefore === activeSessionId.value) recordActivity(reason);
      await refreshTurnStatusByTraceId(id, { force: true });
@@ -1382,9 +1250,7 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project
      if (!message && ownerSessionId) {
        await refreshRealtimeSessionMessages(ownerSessionId, `${reason}:message-gap`, { force: true });
      }
-    } finally {
-      terminalRealtimeRefreshInFlight.delete(id);
-    }
+    }, { force: true, reason });
  }

  function installRealtimeVisibilityHandler(): void {
@@ -1463,16 +1329,13 @@ function nonBlockingProjection(projection: ProjectionDiagnostic | null): Project

  async function refreshRealtimeSessionFromRest(sessionId: string, reason: string): Promise<void> {
    const id = normalizeWorkbenchSessionId(sessionId);
-    if (!id || id !== activeSessionId.value || realtimeSessionRefreshInFlight.has(id)) return;
-    realtimeSessionRefreshInFlight.add(id);
-    try {
+    if (!id || id !== activeSessionId.value) return;
+    await realtimeSessionRefreshRuntime.run(composeWorkbenchScopedKey("workbench.realtime.session-refresh", id), async () => {
      recordActivity(reason);
      const existing = sessions.value.find((item) => item.sessionId === id) ?? null;
      const selected = await loadWorkbenchSession(id, existing);
      if (selected && activeSessionId.value === id) applySelectedSessionDetail(selected, "system");
-    } finally {
-      realtimeSessionRefreshInFlight.delete(id);
-    }
+    }, { force: true, reason });
  }

  function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void {
@@ -2271,29 +2134,6 @@ function projectionFromMessage(message: ChatMessage, trace: ChatMessage["runnerT
  return normalizeProjectionDiagnostic(message.projection ?? message) ?? normalizeProjectionDiagnostic(trace?.projection ?? trace) ?? null;
 }

-function normalizeProjectionDiagnostic(value: unknown): ProjectionDiagnostic | null {
-  const record = recordValue(value);
-  if (!record) return null;
-  const nested = recordValue(record.projection);
-  const source = nested ?? record;
-  const rawBlocker = recordValue(source.blocker ?? record.blocker);
-  const apiError = normalizeApiErrorRecord(source.apiError ?? record.apiError, null);
-  const diagnostic = normalizeErrorDiagnostic(source.diagnostic, record.diagnostic, rawBlocker?.diagnostic, apiError?.diagnostic);
-  const blocker = normalizeProjectionBlocker(rawBlocker, { diagnostic, apiError });
-  const projectionStatus = firstNonEmptyString(source.projectionStatus, record.projectionStatus);
-  const projectionHealth = firstNonEmptyString(source.projectionHealth, record.projectionHealth);
-  const lastProjectedSeq = firstFiniteNumber(source.lastProjectedSeq, record.lastProjectedSeq);
-  const staleMs = firstFiniteNumber(source.staleMs, record.staleMs);
-  const sourceRunId = firstNonEmptyString(source.sourceRunId, record.sourceRunId);
-  const sourceCommandId = firstNonEmptyString(source.sourceCommandId, record.sourceCommandId);
-  if (!projectionStatus && !projectionHealth && !blocker && lastProjectedSeq == null && staleMs == null && !sourceRunId && !sourceCommandId) return null;
-  return { ...source, projectionStatus, projectionHealth, lastProjectedSeq: lastProjectedSeq ?? null, staleMs: staleMs ?? null, sourceRunId: sourceRunId ?? null, sourceCommandId: sourceCommandId ?? null, blocker, diagnostic, apiError, valuesRedacted: source.valuesRedacted !== false } as ProjectionDiagnostic;
-}
-
-function projectionDiagnosticFromApiFailure(result: ApiResult<unknown>, input: { code: string; message: string; health: string }): ProjectionDiagnostic {
-  return projectionDiagnosticFromFailure({ ...input, status: result.status, apiError: result.apiError, diagnostic: result.diagnostic });
-}
-
 function shouldSuppressTransientWorkbenchReadFailure(result: ApiResult<unknown>): boolean {
  const diagnostic = normalizeErrorDiagnostic(result.diagnostic, result.apiError?.diagnostic);
  const code = firstStringOrNumber(result.apiError?.code, diagnostic?.code);
@@ -2303,87 +2143,10 @@ function shouldSuppressTransientWorkbenchReadFailure(result: ApiResult<unknown>)
  return result.status === 0 && source === "browser" && code === "browser_network_error" && category === "network";
 }

-function projectionDiagnosticFromFailure(input: { code: string; message: string; health: string; status?: number | null; apiError?: ApiError | null; diagnostic?: ErrorDiagnostic | null; route?: string | null; source?: string | null }): ProjectionDiagnostic {
-  const apiError = input.apiError ?? null;
-  const diagnostic = normalizeErrorDiagnostic(input.diagnostic, apiError?.diagnostic, {
-    contractVersion: "hwlab-error-diagnostic-v1",
-    traceId: apiError?.traceId ?? null,
-    requestId: apiError?.requestId ?? null,
-    route: firstNonEmptyString(apiError?.route, input.route),
-    layer: apiError?.layer ?? "workbench",
-    category: apiError?.category ?? "workbench-projection",
-    code: firstStringOrNumber(apiError?.code, input.code),
-    httpStatus: input.status ?? null,
-    source: firstNonEmptyString(apiError?.source, input.source, "browser"),
-    retryable: apiError?.retryable ?? true,
-    valuesPrinted: false
-  });
-  const message = firstNonEmptyString(apiError?.userMessage, apiError?.message, input.message) ?? input.message;
-  const blocker = normalizeProjectionBlocker({
-    code: firstStringOrNumber(apiError?.code, diagnostic?.code, input.code),
-    message,
-    userMessage: firstNonEmptyString(apiError?.userMessage, message),
-    layer: firstNonEmptyString(apiError?.layer, diagnostic?.layer, "workbench"),
-    category: firstNonEmptyString(apiError?.category, diagnostic?.category, "workbench-projection"),
-    retryable: firstBoolean(apiError?.retryable, diagnostic?.retryable, true),
-    route: firstNonEmptyString(apiError?.route, diagnostic?.route, input.route),
-    traceId: firstNonEmptyString(apiError?.traceId, diagnostic?.traceId),
-    requestId: firstNonEmptyString(apiError?.requestId, diagnostic?.requestId),
-    source: firstNonEmptyString(apiError?.source, diagnostic?.source, input.source),
-    httpStatus: firstFiniteNumber(diagnostic?.httpStatus, input.status),
-    diagnostic,
-    valuesPrinted: false
-  }, { diagnostic, apiError, fallbackCode: input.code, fallbackMessage: message });
-  return { projectionStatus: "blocked", projectionHealth: input.health, blocker, diagnostic, apiError, valuesRedacted: true };
-}
-
-function normalizeProjectionBlocker(value: unknown, context: { diagnostic?: ErrorDiagnostic | null; apiError?: ApiError | null; fallbackCode?: string | null; fallbackMessage?: string | null } = {}): ProjectionBlocker | null {
-  const record = recordValue(value);
-  if (!record && !context.diagnostic && !context.apiError && !context.fallbackMessage && !context.fallbackCode) return null;
-  const source = record ?? {};
-  const diagnostic = context.diagnostic ?? normalizeErrorDiagnostic(source.diagnostic, context.apiError?.diagnostic);
-  const code = firstStringOrNumber(source.code, context.apiError?.code, diagnostic?.code, context.fallbackCode);
-  const message = firstNonEmptyString(source.message, source.userMessage, source.summary, context.apiError?.userMessage, context.apiError?.message, context.fallbackMessage);
-  const retryable = firstBoolean(source.retryable, context.apiError?.retryable, diagnostic?.retryable);
-  return {
-    ...source,
-    code: code === null ? undefined : String(code),
-    message: message ?? null,
-    userMessage: firstNonEmptyString(source.userMessage, context.apiError?.userMessage, message),
-    summary: firstNonEmptyString(source.summary),
-    layer: firstNonEmptyString(source.layer, context.apiError?.layer, diagnostic?.layer),
-    category: firstNonEmptyString(source.category, context.apiError?.category, diagnostic?.category, "workbench-projection"),
-    retryable: retryable ?? true,
-    route: firstNonEmptyString(source.route, context.apiError?.route, diagnostic?.route),
-    traceId: firstNonEmptyString(source.traceId, context.apiError?.traceId, diagnostic?.traceId),
-    requestId: firstNonEmptyString(source.requestId, context.apiError?.requestId, diagnostic?.requestId),
-    source: firstNonEmptyString(source.source, context.apiError?.source, diagnostic?.source),
-    httpStatus: firstFiniteNumber(source.httpStatus, source.http_status, diagnostic?.httpStatus) ?? null,
-    diagnostic,
-    valuesPrinted: source.valuesPrinted === true
-  } as ProjectionBlocker;
-}
-
-function normalizeErrorDiagnostic(...values: unknown[]): ErrorDiagnostic | null {
-  return normalizeRuntimeErrorDiagnostic(...values);
-}
-
-function normalizeApiErrorRecord(value: unknown, fallback: string | null = null, diagnostic: ErrorDiagnostic | null = null): ApiError | null {
-  return normalizeRuntimeApiErrorRecord(value, fallback, diagnostic);
-}
-
 function agentErrorFromApiFailure(result: ApiResult<unknown>, projection: ProjectionDiagnostic, fallback: string): ChatMessage["error"] {
  return normalizeAgentError(result.apiError ?? projection.apiError ?? projection.blocker ?? { message: result.error ?? fallback, diagnostic: result.diagnostic ?? projection.diagnostic }) ?? { message: result.error ?? fallback, diagnostic: result.diagnostic ?? projection.diagnostic };
 }

-function agentErrorFromProjection(projection: ProjectionDiagnostic): ChatMessage["error"] | null {
-  const blocker = projection.blocker ?? null;
-  const diagnostic = projection.diagnostic ?? blocker?.diagnostic ?? projection.apiError?.diagnostic ?? null;
-  const message = firstNonEmptyString(blocker?.userMessage, blocker?.message, blocker?.summary, projection.apiError?.userMessage, projection.apiError?.message, diagnostic?.code ? String(diagnostic.code) : null);
-  if (!blocker && !projection.apiError && !diagnostic && !message) return null;
-  return normalizeAgentError({ ...(projection.apiError ?? {}), ...(blocker ?? {}), message, code: firstStringOrNumber(blocker?.code, projection.apiError?.code, diagnostic?.code), diagnostic, traceId: firstNonEmptyString(blocker?.traceId, projection.apiError?.traceId, diagnostic?.traceId), requestId: firstNonEmptyString(blocker?.requestId, projection.apiError?.requestId, diagnostic?.requestId) });
-}
-
 function messageNeedsTerminalDiagnostics(message: ChatMessage): boolean {
  if (message.role !== "agent") return false;
  if (!isTerminalMessageStatus(message.status)) return false;
@@ -1,7 +1,36 @@
 // SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first.
-// Responsibility: Workbench typed error and diagnostic normalization. Mechanically migrated from OpenCode error-envelope semantics, then adapted to HWLAB API contracts.
+// Responsibility: Workbench typed error, diagnostic normalization, projection blockers and readable server errors.
+// Mechanical source references:
+// - OpenCode utils/server-errors.ts:1-17 named error shapes.
+// - OpenCode utils/server-errors.ts:28-35 formatServerError.
+// - OpenCode utils/server-errors.ts:38-55 unwrap/type guards.
+// - OpenCode utils/server-errors.ts:57-88 readable config/provider errors.
+// - OpenCode stream.transport.ts:250-271 formatUnknownError.
+// - OpenCode session-data.ts:163-183 formatError boundary.

-import type { ApiError, ErrorDiagnostic } from "@/types";
+import type { ApiError, ChatMessage, ErrorDiagnostic, ProjectionBlocker, ProjectionDiagnostic } from "@/types";
+
+export type ConfigInvalidError = {
+  name: "ConfigInvalidError";
+  data: { path?: string; message?: string; issues?: Array<{ message: string; path: string[] }> };
+};
+
+export type ProviderModelNotFoundError = {
+  name: "ProviderModelNotFoundError";
+  data: { providerID: string; modelID: string; suggestions?: string[] };
+};
+
+type Translator = (key: string, vars?: Record<string, string | number>) => string;
+
+export function formatServerError(error: unknown, translate?: Translator, fallback?: string): string {
+  const unwrapped = unwrapNamedError(error);
+  if (isConfigInvalidErrorLike(unwrapped)) return parseReadableConfigInvalidError(unwrapped, translate);
+  if (isProviderModelNotFoundErrorLike(unwrapped)) return parseReadableProviderModelNotFoundError(unwrapped, translate);
+  if (error instanceof Error && error.message) return error.message;
+  if (typeof error === "string" && error) return error;
+  if (fallback) return fallback;
+  return tr(translate, "error.chain.unknown", "Unknown error");
+}

 export function normalizeErrorDiagnostic(...values: unknown[]): ErrorDiagnostic | null {
  for (const value of values) {
@@ -51,6 +80,146 @@ export function normalizeApiErrorRecord(value: unknown, fallback: string | null
  } as ApiError;
 }

+export function normalizeProjectionDiagnostic(value: unknown): ProjectionDiagnostic | null {
+  const record = recordValue(value);
+  if (!record) return null;
+  const nested = recordValue(record.projection);
+  const source = nested ?? record;
+  const rawBlocker = recordValue(source.blocker ?? record.blocker);
+  const apiError = normalizeApiErrorRecord(source.apiError ?? record.apiError, null);
+  const diagnostic = normalizeErrorDiagnostic(source.diagnostic, record.diagnostic, rawBlocker?.diagnostic, apiError?.diagnostic);
+  const blocker = normalizeProjectionBlocker(rawBlocker, { diagnostic, apiError });
+  const projectionStatus = firstNonEmptyString(source.projectionStatus, record.projectionStatus);
+  const projectionHealth = firstNonEmptyString(source.projectionHealth, record.projectionHealth);
+  const lastProjectedSeq = firstFiniteNumber(source.lastProjectedSeq, record.lastProjectedSeq);
+  const staleMs = firstFiniteNumber(source.staleMs, record.staleMs);
+  const sourceRunId = firstNonEmptyString(source.sourceRunId, record.sourceRunId);
+  const sourceCommandId = firstNonEmptyString(source.sourceCommandId, record.sourceCommandId);
+  if (!projectionStatus && !projectionHealth && !blocker && lastProjectedSeq == null && staleMs == null && !sourceRunId && !sourceCommandId) return null;
+  return { ...source, projectionStatus, projectionHealth, lastProjectedSeq: lastProjectedSeq ?? null, staleMs: staleMs ?? null, sourceRunId: sourceRunId ?? null, sourceCommandId: sourceCommandId ?? null, blocker, diagnostic, apiError, valuesRedacted: source.valuesRedacted !== false } as ProjectionDiagnostic;
+}
+
+export function projectionDiagnosticFromApiFailure(result: { status: number; error?: string | null; apiError?: ApiError | null; diagnostic?: ErrorDiagnostic | null }, input: { code: string; message: string; health: string }): ProjectionDiagnostic {
+  return projectionDiagnosticFromFailure({ ...input, status: result.status, apiError: result.apiError, diagnostic: result.diagnostic });
+}
+
+export function projectionDiagnosticFromFailure(input: { code: string; message: string; health: string; status?: number | null; apiError?: ApiError | null; diagnostic?: ErrorDiagnostic | null; route?: string | null; source?: string | null }): ProjectionDiagnostic {
+  const apiError = input.apiError ?? null;
+  const diagnostic = normalizeErrorDiagnostic(input.diagnostic, apiError?.diagnostic, {
+    contractVersion: "hwlab-error-diagnostic-v1",
+    traceId: apiError?.traceId ?? null,
+    requestId: apiError?.requestId ?? null,
+    route: firstNonEmptyString(apiError?.route, input.route),
+    layer: apiError?.layer ?? "workbench",
+    category: apiError?.category ?? "workbench-projection",
+    code: firstStringOrNumber(apiError?.code, input.code),
+    httpStatus: input.status ?? null,
+    source: firstNonEmptyString(apiError?.source, input.source, "browser"),
+    retryable: apiError?.retryable ?? true,
+    valuesPrinted: false
+  });
+  const message = firstNonEmptyString(apiError?.userMessage, apiError?.message, input.message) ?? input.message;
+  const blocker = normalizeProjectionBlocker({
+    code: firstStringOrNumber(apiError?.code, diagnostic?.code, input.code),
+    message,
+    userMessage: firstNonEmptyString(apiError?.userMessage, message),
+    layer: firstNonEmptyString(apiError?.layer, diagnostic?.layer, "workbench"),
+    category: firstNonEmptyString(apiError?.category, diagnostic?.category, "workbench-projection"),
+    retryable: firstBoolean(apiError?.retryable, diagnostic?.retryable, true),
+    route: firstNonEmptyString(apiError?.route, diagnostic?.route, input.route),
+    traceId: firstNonEmptyString(apiError?.traceId, diagnostic?.traceId),
+    requestId: firstNonEmptyString(apiError?.requestId, diagnostic?.requestId),
+    source: firstNonEmptyString(apiError?.source, diagnostic?.source, input.source),
+    httpStatus: firstFiniteNumber(diagnostic?.httpStatus, input.status),
+    diagnostic,
+    valuesPrinted: false
+  }, { diagnostic, apiError, fallbackCode: input.code, fallbackMessage: message });
+  return { projectionStatus: "blocked", projectionHealth: input.health, blocker, diagnostic, apiError, valuesRedacted: true };
+}
+
+export function normalizeProjectionBlocker(value: unknown, context: { diagnostic?: ErrorDiagnostic | null; apiError?: ApiError | null; fallbackCode?: string | null; fallbackMessage?: string | null } = {}): ProjectionBlocker | null {
+  const record = recordValue(value);
+  if (!record && !context.diagnostic && !context.apiError && !context.fallbackMessage && !context.fallbackCode) return null;
+  const source = record ?? {};
+  const diagnostic = context.diagnostic ?? normalizeErrorDiagnostic(source.diagnostic, context.apiError?.diagnostic);
+  const code = firstStringOrNumber(source.code, context.apiError?.code, diagnostic?.code, context.fallbackCode);
+  const message = firstNonEmptyString(source.message, source.userMessage, source.summary, context.apiError?.userMessage, context.apiError?.message, context.fallbackMessage);
+  const retryable = firstBoolean(source.retryable, context.apiError?.retryable, diagnostic?.retryable);
+  return {
+    ...source,
+    code: code === null ? undefined : String(code),
+    message: message ?? null,
+    userMessage: firstNonEmptyString(source.userMessage, context.apiError?.userMessage, message),
+    summary: firstNonEmptyString(source.summary),
+    layer: firstNonEmptyString(source.layer, context.apiError?.layer, diagnostic?.layer),
+    category: firstNonEmptyString(source.category, context.apiError?.category, diagnostic?.category, "workbench-projection"),
+    retryable: retryable ?? true,
+    route: firstNonEmptyString(source.route, context.apiError?.route, diagnostic?.route),
+    traceId: firstNonEmptyString(source.traceId, context.apiError?.traceId, diagnostic?.traceId),
+    requestId: firstNonEmptyString(source.requestId, context.apiError?.requestId, diagnostic?.requestId),
+    source: firstNonEmptyString(source.source, context.apiError?.source, diagnostic?.source),
+    httpStatus: firstFiniteNumber(source.httpStatus, source.http_status, diagnostic?.httpStatus) ?? null,
+    diagnostic,
+    valuesPrinted: source.valuesPrinted === true
+  } as ProjectionBlocker;
+}
+
+export function agentErrorFromProjection(projection: ProjectionDiagnostic): ChatMessage["error"] | null {
+  const blocker = projection.blocker ?? null;
+  const diagnostic = projection.diagnostic ?? blocker?.diagnostic ?? projection.apiError?.diagnostic ?? null;
+  const message = firstNonEmptyString(blocker?.userMessage, blocker?.message, blocker?.summary, projection.apiError?.userMessage, projection.apiError?.message, diagnostic?.code ? String(diagnostic.code) : null);
+  if (!blocker && !projection.apiError && !diagnostic && !message) return null;
+  return normalizeApiErrorRecord({ ...(projection.apiError ?? {}), ...(blocker ?? {}), message, code: firstStringOrNumber(blocker?.code, projection.apiError?.code, diagnostic?.code), diagnostic, traceId: firstNonEmptyString(blocker?.traceId, projection.apiError?.traceId, diagnostic?.traceId), requestId: firstNonEmptyString(blocker?.requestId, projection.apiError?.requestId, diagnostic?.requestId) }, message) as ChatMessage["error"];
+}
+
+function tr(translator: Translator | undefined, key: string, text: string, vars?: Record<string, string | number>): string {
+  if (!translator) return text;
+  const out = translator(key, vars);
+  if (!out || out === key) return text;
+  return out;
+}
+
+function unwrapNamedError(error: unknown): unknown {
+  if (error instanceof Error && error.cause && typeof error.cause === "object" && "body" in error.cause) return (error.cause as Record<string, unknown>).body;
+  return error;
+}
+
+function isConfigInvalidErrorLike(error: unknown): error is ConfigInvalidError {
+  const record = recordValue(error);
+  return record?.name === "ConfigInvalidError" && typeof record.data === "object" && record.data !== null;
+}
+
+function isProviderModelNotFoundErrorLike(error: unknown): error is ProviderModelNotFoundError {
+  const record = recordValue(error);
+  return record?.name === "ProviderModelNotFoundError" && typeof record.data === "object" && record.data !== null;
+}
+
+export function parseReadableConfigInvalidError(errorInput: ConfigInvalidError, translator?: Translator): string {
+  const file = errorInput.data.path && errorInput.data.path !== "config" ? errorInput.data.path : "config";
+  const detail = errorInput.data.message?.trim() ?? "";
+  const issues = (errorInput.data.issues ?? []).map((issue) => {
+    const msg = issue.message.trim();
+    if (!issue.path.length) return msg;
+    return `${issue.path.join(".")}: ${msg}`;
+  }).filter(Boolean);
+  const msg = issues.length ? issues.join("\n") : detail;
+  if (!msg) return tr(translator, "error.chain.configInvalid", `Config file at ${file} is invalid`, { path: file });
+  return tr(translator, "error.chain.configInvalidWithMessage", `Config file at ${file} is invalid: ${msg}`, { path: file, message: msg });
+}
+
+function parseReadableProviderModelNotFoundError(errorInput: ProviderModelNotFoundError, translator?: Translator): string {
+  const provider = errorInput.data.providerID.trim();
+  const model = errorInput.data.modelID.trim();
+  const list = (errorInput.data.suggestions ?? []).map((value) => value.trim()).filter(Boolean);
+  const body = tr(translator, "error.chain.modelNotFound", `Model not found: ${provider}/${model}`, { provider, model });
+  const tail = tr(translator, "error.chain.checkConfig", "Check your config (opencode.json) provider/model names");
+  if (list.length) {
+    const suggestions = list.slice(0, 5).join(", ");
+    return [body, tr(translator, "error.chain.didYouMean", `Did you mean: ${suggestions}`, { suggestions }), tail].join("\n");
+  }
+  return [body, tail].join("\n");
+}
+
 function recordValue(value: unknown): Record<string, unknown> | null {
  return value && typeof value === "object" && !Array.isArray(value) ? value as Record<string, unknown> : null;
 }
@@ -1,6 +1,8 @@
 // SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first; PJ2026-01060505 Workbench性能 draft-2026-06-19-p0.
 // Responsibility: Health probe retry/cache helpers. Mechanically migrated from OpenCode app/utils/server-health.ts with all timing supplied by runtime policy.

+import { createScopedCache } from "@/utils/scoped-cache";
+
 export type WorkbenchHealthState = "ok" | "degraded" | "unavailable" | "blocked";

 export interface WorkbenchHealthSnapshot<T = unknown> {
@@ -32,8 +34,12 @@ interface HealthCacheEntry<T> {
  promise: Promise<WorkbenchHealthSnapshot<T>>;
 }

+interface HealthLatestEntry {
+  snapshot: WorkbenchHealthSnapshot | null;
+}
+
 export class WorkbenchHealthProbeCache {
-  private readonly latest = new Map<string, WorkbenchHealthSnapshot>();
+  private readonly latest: ReturnType<typeof createScopedCache<HealthLatestEntry>>;
  private readonly inflight = new Map<string, HealthCacheEntry<unknown>>();
  private readonly now: () => number;
  private readonly cacheMs: number | null;
@@ -41,10 +47,11 @@ export class WorkbenchHealthProbeCache {
  constructor(options: WorkbenchHealthCacheOptions = {}) {
    this.now = options.now ?? Date.now;
    this.cacheMs = positiveIntegerOrNull(options.cacheMs);
+    this.latest = createScopedCache<HealthLatestEntry>(() => ({ snapshot: null }), { ttlMs: this.cacheMs, now: () => this.now() });
  }

  get(key: string): WorkbenchHealthSnapshot | null {
-    return this.latest.get(normalizeKey(key)) ?? null;
+    return this.latest.peek(normalizeKey(key))?.snapshot ?? null;
  }

  probe<T>(input: CheckWorkbenchHealthOptions<T>): Promise<WorkbenchHealthSnapshot<T>> {
@@ -53,7 +60,7 @@ export class WorkbenchHealthProbeCache {
    const now = this.now();
    if (hit && (!hit.done || (this.cacheMs !== null && now - hit.at < this.cacheMs))) return hit.promise;
    const promise = checkWorkbenchHealth(input).then((snapshot) => {
-      this.latest.set(key, snapshot);
+      this.latest.get(key).snapshot = snapshot;
      return snapshot;
    }).finally(() => {
      const current = this.inflight.get(key);
@@ -1,4 +1,4 @@
 // SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first.
 // Responsibility: Production Workbench realtime runtime entry point for SSE transport, event coalescing, and scoped cursor ownership.

-export { createWorkbenchStreamTransportRuntime, WorkbenchStreamTransportRuntime, type WorkbenchStreamTransportRestartInput, type WorkbenchStreamTransportRestartResult, type WorkbenchRealtimeEvent } from "@/utils/workbench-stream-transport";
+export { createWorkbenchStreamTransportRuntime, WorkbenchStreamTransportRuntime, type WorkbenchStreamTransportRecovery, type WorkbenchStreamTransportRestartInput, type WorkbenchStreamTransportRestartResult, type WorkbenchStreamTransportState, type WorkbenchRealtimeEvent } from "@/utils/workbench-stream-transport";
@@ -1,5 +1,10 @@
 // SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first.
-// Responsibility: Workbench REST refresh queue, keyed single-flight, cooldown, and request-storm suppression.
+// Responsibility: Workbench REST refresh queue, keyed single-flight, cooldown, scheduled gap-fill and request-storm suppression.
+// Mechanical source references:
+// - OpenCode util/queue.ts:1-32 AsyncQueue and bounded work.
+// - OpenCode runtime.queue.ts:54-113 queue state/control.
+// - OpenCode runtime.queue.ts:113-230 drain/send/abort flow.
+// - OpenCode stream.transport.ts:83-130 per-turn wait/tick state consumed by recovery scheduling.

 import { AsyncQueue } from "@/utils/scheduler/async-queue";
 import { createKeyedSingleflight } from "@/utils/scheduler/keyed-singleflight";
@@ -29,6 +34,33 @@ export interface WorkbenchReadHydrationRunOptions {

 type QueuedTask = () => void;

+type TimerHandle = ReturnType<typeof setTimeout>;
+
+export interface WorkbenchKeyedTaskRunOptions {
+  minIntervalMs?: number;
+  force?: boolean;
+  replace?: boolean;
+  reason?: string | null;
+}
+
+export interface WorkbenchScheduledTaskOptions extends WorkbenchKeyedTaskRunOptions {
+  delayMs?: number;
+  replaceTimer?: boolean;
+}
+
+export interface WorkbenchScheduledTaskRuntimeOptions {
+  now?: () => number;
+  setTimeout?: (callback: () => void, delayMs: number) => TimerHandle;
+  clearTimeout?: (handle: TimerHandle) => void;
+}
+
+export interface WorkbenchTraceHydrationQueueRuntimeOptions {
+  concurrency: number;
+  delayMs: number;
+  setTimeout?: (callback: () => void, delayMs: number) => TimerHandle;
+  clearTimeout?: (handle: TimerHandle) => void;
+}
+
 export class WorkbenchReadHydrationRuntime {
  private readonly queue = new AsyncQueue<QueuedTask>();
  private readonly singleflight = createKeyedSingleflight<unknown>();
@@ -83,10 +115,142 @@ export class WorkbenchReadHydrationRuntime {
  }
 }

+export class WorkbenchKeyedTaskRuntime {
+  private readonly singleflight = createKeyedSingleflight<unknown>();
+  private readonly lastStartedAtByKey = new Map<string, number>();
+  private readonly now: () => number;
+
+  constructor(options: { now?: () => number } = {}) {
+    this.now = options.now ?? Date.now;
+  }
+
+  run<T>(key: string, task: () => Promise<T>, options: WorkbenchKeyedTaskRunOptions = {}): Promise<T | undefined> {
+    const normalized = normalizeRequiredKey(key);
+    const now = this.now();
+    const minIntervalMs = Math.max(0, Math.trunc(options.minIntervalMs ?? 0));
+    const lastStartedAtMs = this.lastStartedAtByKey.get(normalized) ?? 0;
+    if (options.force !== true && minIntervalMs > 0 && lastStartedAtMs > 0 && now - lastStartedAtMs < minIntervalMs) return Promise.resolve(undefined);
+    this.lastStartedAtByKey.set(normalized, now);
+    return this.singleflight.run(normalized, task, { reason: options.reason ?? null, replace: options.replace }) as Promise<T>;
+  }
+
+  cancel(key: string, reason = "workbench-keyed-task-canceled"): void {
+    this.singleflight.cancel(normalizeRequiredKey(key), reason);
+  }
+
+  clear(reason = "workbench-keyed-task-cleared"): void {
+    this.singleflight.clear(reason);
+    this.lastStartedAtByKey.clear();
+  }
+}
+
+export class WorkbenchScheduledTaskRuntime extends WorkbenchKeyedTaskRuntime {
+  private readonly timers = new Map<string, TimerHandle>();
+  private readonly setTimer: (callback: () => void, delayMs: number) => TimerHandle;
+  private readonly clearTimer: (handle: TimerHandle) => void;
+
+  constructor(options: WorkbenchScheduledTaskRuntimeOptions = {}) {
+    super({ now: options.now });
+    this.setTimer = options.setTimeout ?? ((callback, delayMs) => setTimeout(callback, delayMs));
+    this.clearTimer = options.clearTimeout ?? ((handle) => clearTimeout(handle));
+  }
+
+  schedule<T>(key: string, task: () => Promise<T>, options: WorkbenchScheduledTaskOptions = {}): boolean {
+    const normalized = normalizeRequiredKey(key);
+    const existing = this.timers.get(normalized);
+    if (existing) {
+      if (options.replaceTimer !== true) return false;
+      this.clearTimer(existing);
+      this.timers.delete(normalized);
+    }
+    const delayMs = Math.max(0, Math.trunc(options.delayMs ?? 0));
+    const timer = this.setTimer(() => {
+      this.timers.delete(normalized);
+      void this.run(normalized, task, options);
+    }, delayMs);
+    this.timers.set(normalized, timer);
+    return true;
+  }
+
+  cancelTimer(key: string): void {
+    const normalized = normalizeRequiredKey(key);
+    const timer = this.timers.get(normalized);
+    if (timer) this.clearTimer(timer);
+    this.timers.delete(normalized);
+  }
+
+  clearAll(reason = "workbench-scheduled-task-cleared"): void {
+    for (const timer of this.timers.values()) this.clearTimer(timer);
+    this.timers.clear();
+    this.clear(reason);
+  }
+}
+
+export class WorkbenchTraceHydrationQueueRuntime<T> {
+  private readonly queue = new AsyncQueue<QueuedTask>();
+  private readonly inFlight = new Set<string>();
+  private readonly queued = new Set<string>();
+  private readonly delayMs: number;
+  private readonly setTimer: (callback: () => void, delayMs: number) => TimerHandle;
+
+  constructor(options: WorkbenchTraceHydrationQueueRuntimeOptions) {
+    const concurrency = Math.max(1, Math.trunc(options.concurrency));
+    this.delayMs = Math.max(0, Math.trunc(options.delayMs));
+    this.setTimer = options.setTimeout ?? ((callback, delayMs) => setTimeout(callback, delayMs));
+    for (let index = 0; index < concurrency; index += 1) void this.workerLoop();
+  }
+
+  run(key: string, task: () => Promise<void>, options: { onAlreadyRunning?: () => void } = {}): Promise<boolean> {
+    const normalized = normalizeRequiredKey(key);
+    if (this.inFlight.has(normalized)) {
+      options.onAlreadyRunning?.();
+      return Promise.resolve(false);
+    }
+    this.inFlight.add(normalized);
+    return task().then(() => true, (error) => {
+      throw error;
+    }).finally(() => {
+      this.inFlight.delete(normalized);
+    });
+  }
+
+  enqueue(key: string, task: () => Promise<void>): boolean {
+    const normalized = normalizeRequiredKey(key);
+    if (this.inFlight.has(normalized) || this.queued.has(normalized)) return false;
+    this.queued.add(normalized);
+    this.queue.push(() => {
+      void this.run(normalized, task).finally(() => {
+        this.queued.delete(normalized);
+      });
+    });
+    return true;
+  }
+
+  private async workerLoop(): Promise<void> {
+    for (;;) {
+      const run = await this.queue.next();
+      run();
+      if (this.delayMs > 0) await new Promise<void>((resolve) => this.setTimer(resolve, this.delayMs));
+    }
+  }
+}
+
 export function createWorkbenchReadHydrationRuntime(options: WorkbenchReadHydrationRuntimeOptions): WorkbenchReadHydrationRuntime {
  return new WorkbenchReadHydrationRuntime(options);
 }

+export function createWorkbenchKeyedTaskRuntime(options: { now?: () => number } = {}): WorkbenchKeyedTaskRuntime {
+  return new WorkbenchKeyedTaskRuntime(options);
+}
+
+export function createWorkbenchScheduledTaskRuntime(options: WorkbenchScheduledTaskRuntimeOptions = {}): WorkbenchScheduledTaskRuntime {
+  return new WorkbenchScheduledTaskRuntime(options);
+}
+
+export function createWorkbenchTraceHydrationQueueRuntime<T>(options: WorkbenchTraceHydrationQueueRuntimeOptions): WorkbenchTraceHydrationQueueRuntime<T> {
+  return new WorkbenchTraceHydrationQueueRuntime<T>(options);
+}
+
 export function shouldCooldownWorkbenchReadFailure(value: unknown): boolean {
  const result = value as ApiResultLike | null;
  if (!result || typeof result !== "object" || typeof result.status !== "number") return false;
@@ -136,6 +300,12 @@ function normalizeOptionalKey(value: string | null | undefined): string | null {
  return text || null;
 }

+function normalizeRequiredKey(value: string): string {
+  const text = String(value ?? "").trim();
+  if (!text) throw new Error("workbench runtime key is required");
+  return text;
+}
+
 function firstNonEmptyString(...values: unknown[]): string | null {
  for (const value of values) {
    if (typeof value !== "string") continue;
@@ -1,5 +1,11 @@
 // SPEC: PJ2026-0106050514 Workbench实时运行面 draft-2026-06-30-p0-1297-spec-first.
-// Responsibility: Workbench SSE transport lifecycle and scope-key ownership.
+// Responsibility: Workbench SSE transport lifecycle, cursor ownership, turn-style tick state and recovery actions.
+// Mechanical source references:
+// - OpenCode stream.transport.ts:1-17 transport/turn coordination intent.
+// - OpenCode stream.transport.ts:83-130 Wait/State shape.
+// - OpenCode stream.transport.ts:229-247 abort/stale wait race.
+// - OpenCode stream.transport.ts:391-443 stream acquire/subscribe.
+// - OpenCode stream.transport.ts:504-530 reduce output/trace.

 import { connectWorkbenchEvents, type WorkbenchEventStream, type WorkbenchRealtimeEvent } from "@/api/workbench-events";
 import { workbenchRealtimeScopeKey } from "@/utils/workbench-key";
@@ -10,8 +16,11 @@ export interface WorkbenchStreamTransportRestartInput {
  sessionId?: string | null;
  traceId?: string | null;
  afterSeq?: number | null;
+  errorRecoveryMinMs?: number | null;
  onOpen?: () => void;
  onError?: (event: Event) => void;
+  onState?: (state: WorkbenchStreamTransportState) => void;
+  onRecovery?: (recovery: WorkbenchStreamTransportRecovery) => void;
  onEvent: (event: WorkbenchRealtimeEvent, eventName: string) => void;
 }

@@ -21,38 +30,207 @@ export interface WorkbenchStreamTransportRestartResult {
  streamStarted: boolean;
 }

+export type WorkbenchStreamTransportPhase = "idle" | "connecting" | "open" | "event" | "error" | "closed" | "blocked";
+
+export type WorkbenchStreamTransportRecoveryAction = "refresh-session-messages" | "schedule-session-list" | "refresh-turn-status" | "hydrate-trace-events";
+
+export interface WorkbenchStreamTransportState {
+  key: string;
+  sessionId: string | null;
+  traceId: string | null;
+  phase: WorkbenchStreamTransportPhase;
+  tick: number;
+  live: boolean;
+  outboxSeq: number | null;
+  traceSeq: number | null;
+  reason?: string | null;
+  errorName?: string | null;
+  diagnostic?: Record<string, unknown> | null;
+}
+
+export interface WorkbenchStreamTransportRecovery {
+  key: string;
+  sessionId: string | null;
+  traceId: string | null;
+  tick: number;
+  reason: string;
+  actions: WorkbenchStreamTransportRecoveryAction[];
+  diagnostic: Record<string, unknown>;
+}
+
+interface WorkbenchTransportWait {
+  tick: number;
+  armed: boolean;
+  live: boolean;
+}
+
+interface WorkbenchTransportCursor {
+  outboxSeq: number | null;
+  traceSeq: number | null;
+}
+
 export class WorkbenchStreamTransportRuntime {
  private stream: WorkbenchEventStream | null = null;
  private key = "";
+  private tick = 0;
+  private wait: WorkbenchTransportWait | null = null;
+  private readonly cursorByKey = new Map<string, WorkbenchTransportCursor>();
+  private readonly lastRecoveryAtByKey = new Map<string, number>();

  restart(input: WorkbenchStreamTransportRestartInput): WorkbenchStreamTransportRestartResult {
    const key = workbenchRealtimeScopeKey(input.sessionId ?? null, input.traceId ?? null);
    if (key === this.key && this.stream) return { key, changed: false, streamStarted: true };
    this.stop();
    this.key = key;
+    this.armWait(key);
    if (!input.sessionId && !input.traceId) return { key, changed: true, streamStarted: false };
+    this.emitState(input, "connecting", null);
    this.stream = connectWorkbenchEvents({
      sessionId: input.sessionId ?? null,
      traceId: input.traceId ?? null,
-      afterSeq: input.afterSeq ?? null,
-      onOpen: input.onOpen,
-      onError: input.onError,
-      onEvent: input.onEvent
+      afterSeq: input.afterSeq ?? this.cursorByKey.get(key)?.outboxSeq ?? null,
+      onOpen: () => {
+        this.markLive(key);
+        this.emitState(input, "open", null);
+        input.onOpen?.();
+      },
+      onError: (event) => {
+        this.markBlocked(key);
+        this.emitState(input, "error", null, event.type);
+        input.onError?.(event);
+        this.requestRecovery(input, event.type || "eventsource-error");
+      },
+      onEvent: (event, eventName) => {
+        this.rememberCursor(key, event);
+        this.markLive(key);
+        this.emitState(input, "event", eventName);
+        input.onEvent(event, eventName);
+      }
    });
    return { key, changed: true, streamStarted: Boolean(this.stream) };
  }

  stop(): void {
+    const previousKey = this.key;
    this.stream?.close();
    this.stream = null;
    this.key = "";
+    if (previousKey) this.wait = null;
  }

  currentKey(): string {
    return this.key;
  }
+
+  currentCursor(key: string = this.key): WorkbenchTransportCursor {
+    return this.cursorByKey.get(key) ?? { outboxSeq: null, traceSeq: null };
+  }
+
+  clear(key?: string | null): void {
+    const normalized = normalizeKey(key);
+    if (normalized) {
+      this.cursorByKey.delete(normalized);
+      this.lastRecoveryAtByKey.delete(normalized);
+      if (this.key === normalized) this.stop();
+      return;
+    }
+    this.stop();
+    this.cursorByKey.clear();
+    this.lastRecoveryAtByKey.clear();
+  }
+
+  private armWait(key: string): void {
+    this.tick += 1;
+    this.wait = { tick: this.tick, armed: Boolean(key), live: false };
+  }
+
+  private markLive(key: string): void {
+    if (!this.wait || this.key !== key) return;
+    this.wait = { ...this.wait, live: true };
+  }
+
+  private markBlocked(key: string): void {
+    if (!this.wait || this.key !== key) return;
+    this.wait = { ...this.wait, live: false };
+  }
+
+  private rememberCursor(key: string, event: WorkbenchRealtimeEvent): void {
+    const previous = this.cursorByKey.get(key) ?? { outboxSeq: null, traceSeq: null };
+    const outboxSeq = maxCursor(previous.outboxSeq, numericCursor(event.cursor?.outboxSeq ?? event.outboxSeq));
+    const traceSeq = maxCursor(previous.traceSeq, numericCursor(event.cursor?.traceSeq ?? event.traceSeq ?? event.event?.seq ?? event.event?.projectedSeq));
+    this.cursorByKey.set(key, { outboxSeq, traceSeq });
+  }
+
+  private requestRecovery(input: WorkbenchStreamTransportRestartInput, reason: string): void {
+    const key = this.key;
+    if (!key) return;
+    const now = Date.now();
+    const minMs = Math.max(0, Math.trunc(input.errorRecoveryMinMs ?? 0));
+    const lastAt = this.lastRecoveryAtByKey.get(key) ?? 0;
+    if (lastAt > 0 && now - lastAt < minMs) return;
+    this.lastRecoveryAtByKey.set(key, now);
+    const sessionId = input.sessionId ?? null;
+    const traceId = input.traceId ?? null;
+    const actions: WorkbenchStreamTransportRecoveryAction[] = [];
+    if (sessionId) actions.push("refresh-session-messages", "schedule-session-list");
+    if (traceId) actions.push("refresh-turn-status", "hydrate-trace-events");
+    const diagnostic = this.diagnosticEnvelope("workbench_sse_recovery", reason, key, actions);
+    input.onRecovery?.({ key, sessionId, traceId, tick: this.wait?.tick ?? this.tick, reason, actions, diagnostic });
+  }
+
+  private emitState(input: WorkbenchStreamTransportRestartInput, phase: WorkbenchStreamTransportPhase, reason: string | null, errorName: string | null = null): void {
+    if (!input.onState) return;
+    const key = this.key || workbenchRealtimeScopeKey(input.sessionId ?? null, input.traceId ?? null);
+    const cursor = this.currentCursor(key);
+    input.onState({
+      key,
+      sessionId: input.sessionId ?? null,
+      traceId: input.traceId ?? null,
+      phase,
+      tick: this.wait?.tick ?? this.tick,
+      live: this.wait?.live === true,
+      outboxSeq: cursor.outboxSeq,
+      traceSeq: cursor.traceSeq,
+      reason,
+      errorName,
+      diagnostic: phase === "error" || phase === "blocked" ? this.diagnosticEnvelope("workbench_sse_transport_error", errorName ?? reason ?? phase, key, []) : null
+    });
+  }
+
+  private diagnosticEnvelope(code: string, reason: string, key: string, actions: WorkbenchStreamTransportRecoveryAction[]): Record<string, unknown> {
+    return {
+      contractVersion: "hwlab-error-diagnostic-v1",
+      code,
+      category: "transport",
+      source: "workbench-web",
+      layer: "workbench-stream-transport",
+      rootCause: reason,
+      recoveryAction: actions.join(",") || "observe-transport-state",
+      transportState: this.wait?.live === true ? "live" : "degraded",
+      scopedKey: key,
+      tick: this.wait?.tick ?? this.tick,
+      retryable: true,
+      valuesPrinted: false,
+      valuesRedacted: true
+    };
+  }
 }

 export function createWorkbenchStreamTransportRuntime(): WorkbenchStreamTransportRuntime {
  return new WorkbenchStreamTransportRuntime();
 }
+
+function normalizeKey(value: string | null | undefined): string | null {
+  const text = String(value ?? "").trim();
+  return text || null;
+}
+
+function numericCursor(value: unknown): number | null {
+  const parsed = Number(value);
+  return Number.isFinite(parsed) && parsed >= 0 ? Math.trunc(parsed) : null;
+}
+
+function maxCursor(previous: number | null, next: number | null): number | null {
+  if (next === null) return previous;
+  return previous === null ? next : Math.max(previous, next);
+}