Merge pull request #2331 from pikasTech/issue-2322-opencode-runtime-unify

fix: 收敛 Workbench realtime 恢复读预算

docs/reference/cloud-workbench.md

@@ -31,6 +31,8 @@ Session rail 是该规则的高频区域。`/v1/agent/conversations` 还未返
 
 Session rail 的后台恢复刷新必须有硬边界。显式用户动作或强一致操作（例如选择会话、删除当前会话）可以立即刷新会话列表；SSE error、active trace REST gap-fill、terminal refresh、trace hydration 等后台补偿路径不得绕过 session list 的冷却/合并机制去强制刷新完整列表。后台路径应优先补当前 trace、turn status、message projection 和必要的 trace events；需要刷新 session rail 时走统一的 scheduled refresh，并按 session/list key 合并已有 timer，避免网络抖动或 EventSource error storm 把 `/v1/workbench/sessions` 放大成浏览器内存和 CDP responsiveness 红灯。
 
+Workbench realtime 恢复与 REST gap-fill 必须统一接入已经迁移的 OpenCode-style runtime 模块。`workbench-stream-transport` 只拥有 SSE lifecycle、cursor 和 recovery reason；`workbench-realtime-plan` 只把 transport action 转成纯 plan；`workbench-refresh-runtime`、keyed singleflight、scheduled task runtime、trace hydration queue、server-state reducer 和 session cache 是恢复读取的唯一 substrate。`workbench.ts` 不得再持有新的 in-flight map、timer map、cursor map、REST gap-fill queue 或第二套 recovery coordinator。`force` 是用户显式操作和恢复优先级语义，不得绕过同 key 的 singleflight、cooldown、min interval 或 storm budget；这些预算、退避、并发、页数、重试和窗口参数只由 node/lane YAML-backed runtime policy 注入，SPEC 只声明字段族、责任边界和验收读取方式，不写死数值。
+
 Workbench 只能维护一条会话恢复与提交路径。首次打开、新建后继续、从左侧 session rail 切换、直接进入 `/workbench/sessions/<conversationId>` 恢复时，都必须以当前 route/active conversation id 作为会话真相，并通过同一条 conversation detail hydration 路径得到 messages、turn state、trace/status 和 markdown 渲染输入；不得另写只消费列表 snapshot、workspace stub 或 localStorage selected id 的恢复分支。
 
 Workbench 的 URL 反射必须服从用户当前导航和组件生命周期。`activeConversationId`、hydrate、select conversation 或列表刷新等异步状态只能在当前 route 仍属于 Workbench section、路径仍是 `/workbench`/`/workspace` 系列且 Workbench 组件仍 active 时，才允许把 URL 反射到 `/workbench/sessions/<conversationId>`；用户已经点击 Dashboard、API Keys、Admin、Settings 或其他非 Workbench 导航后，晚到的 Workbench 响应只能更新 store，不得再调用 `router.replace`/`router.push` 把全局 route 拉回 Workbench。新增 session 恢复或 URL 反射入口时必须复用共享路由守卫，例如 `web/hwlab-cloud-web/src/router/workbench-navigation.ts`，不要在业务组件里各自手写跳转判断。

web/hwlab-cloud-web/scripts/check.ts

@@ -185,7 +185,14 @@ assertIncludes(workbenchPerformanceSource, "clearResourceTimings", "Workbench pe
 assertIncludes(workbenchStoreSource, "recordWorkbenchRuntimeDiagnostic", "Workbench store must surface SSE recovery diagnostics to the performance probe");
 assertIncludes(workbenchRealtimePlanSource, "new Set(recovery.actions)", "Realtime recovery planner must consume transport-owned actions explicitly");
 assertIncludes(workbenchRealtimePlanSource, "actions.has(\"schedule-session-list\")", "Realtime stream errors must schedule bounded session list refreshes only when transport requests that action");
+assertIncludes(workbenchRealtimePlanSource, "authority: \"automatic-recovery\"", "Realtime recovery planner must classify transport recovery as automatic recovery authority");
+assert.doesNotMatch(workbenchRealtimePlanSource, /force:\s*true/u, "Realtime recovery planner must not turn transport recovery into force-refresh work");
+assert.doesNotMatch(workbenchRefreshRuntimeSource, /!options\.force\s*&&\s*minIntervalMs/u, "Workbench refresh runtime must not let force bypass minInterval storm budget");
+assertIncludes(workbenchRefreshRuntimeSource, "forceBudgetBypass = false", "Workbench refresh runtime diagnostics must make force budget suppression visible");
 assertIncludes(workbenchStoreSource, "runtimePolicy.sessionListRealtimeRefreshDelayMs", "Realtime recovery delay must come from runtime policy instead of store constants");
+assertIncludes(workbenchStoreSource, "workbenchReadCooldownKey(\"session-detail\"", "Realtime session detail recovery must enter Workbench read hydration runtime");
+assertIncludes(workbenchStoreSource, "runtimePolicy.workbenchSessionDetailMinRefreshMs", "Realtime session detail recovery budget must come from runtime policy");
+assert.doesNotMatch(workbenchStoreSource, /refreshRealtimeSessionMessages[\s\S]{0,900}refreshSessionMessageProjectionPage\(id, \{ force: true \}\)/u, "Realtime session message recovery must not force-bypass the message projection refresh budget");
 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");
 assert.doesNotMatch(workbenchStoreSource, /refreshTerminalTraceFromRest[\s\S]{0,1200}refreshSessions\([^;]+force:\s*true/u, "Terminal trace REST refresh must not force-refresh the full session list");

web/hwlab-cloud-web/scripts/workbench-realtime-runtime.test.ts

@@ -15,6 +15,7 @@ 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 { messageDiagnosticView } from "../src/utils/workbench-error-runtime.ts";
+import { createWorkbenchReadHydrationRuntime } from "../src/utils/workbench-refresh-runtime.ts";
 import { WORKBENCH_TIMELINE_OPENCODE_PARITY, buildWorkbenchTimelineRows, normalizeWorkbenchTimelineMessages, workbenchTimelineSignature } from "../src/stores/workbench-timeline-model.ts";
 import { reduceWorkbenchRealtimeEvent } from "../src/stores/workbench-event-reducer.ts";
 import { planWorkbenchRealtimeApply, planWorkbenchRealtimeRecovery } from "../src/stores/workbench-realtime-plan.ts";
@@ -32,12 +33,14 @@ test("Workbench runtime policy reads injected config while preserving defaults",
   const policy = workbenchRuntimePolicy({
     sessionListPageLimit: 7,
     traceHydrationMaxPages: 2,
+    workbenchSessionDetailMinRefreshMs: 1234,
     workbenchRealtimeErrorGapFillMinMs: 0,
     defaultGatewayTimeoutMs: "bad"
   });
 
   assert.equal(policy.sessionListPageLimit, 7);
   assert.equal(policy.traceHydrationMaxPages, 2);
+  assert.equal(policy.workbenchSessionDetailMinRefreshMs, 1234);
   assert.equal(policy.workbenchRealtimeErrorGapFillMinMs, 0);
   assert.equal(policy.defaultGatewayTimeoutMs, 120_000);
 });
@@ -313,6 +316,8 @@ test("realtime recovery planner gates refresh steps by transport actions and aut
   const recovery = recoveryEvent(["refresh-session-messages", "schedule-session-list", "refresh-turn-status", "hydrate-trace-events"]);
   const authorized = planWorkbenchRealtimeRecovery(recovery, { selectedSessionId: "ses_1", activeSessionId: "ses_1", fallbackTraceId: "trc_1", activeTraceAuthorized: true });
   assert.deepEqual(authorized.steps.map((step) => step.type), ["refresh-session-messages", "schedule-session-list", "refresh-turn-status", "hydrate-trace-events"]);
+  assert.deepEqual(authorized.steps.map((step) => step.authority), ["automatic-recovery", "automatic-recovery", "automatic-recovery", "automatic-recovery"]);
+  assert.deepEqual(authorized.steps.map((step) => step.force), [false, false, false, false]);
   assert.equal(authorized.sessionId, "ses_1");
   assert.equal(authorized.traceId, "trc_1");
 
@@ -323,6 +328,25 @@ test("realtime recovery planner gates refresh steps by transport actions and aut
   assert.deepEqual(inactive.steps.map((step) => step.type), ["schedule-session-list", "refresh-turn-status", "hydrate-trace-events"]);
 });
 
+test("read hydration runtime keeps force inside the same storm budget", async () => {
+  let clock = 1_000;
+  let calls = 0;
+  const runtime = createWorkbenchReadHydrationRuntime({ concurrency: 1, failureCooldownMs: 1_000, now: () => clock });
+  const task = async () => ({ ok: true, status: 200, data: ++calls });
+
+  assert.deepEqual(await runtime.run(task, "turn:trc_1", { minIntervalMs: 500, force: true }), { ok: true, status: 200, data: 1 });
+  const throttled = await runtime.run(task, "turn:trc_1", { minIntervalMs: 500, force: true }) as { ok: boolean; status: number; diagnostic?: Record<string, unknown> };
+  assert.equal(throttled.ok, false);
+  assert.equal(throttled.status, 0);
+  assert.equal(throttled.diagnostic?.code, "workbench_read_hydration_throttled");
+  assert.equal(throttled.diagnostic?.forceRequested, true);
+  assert.equal(throttled.diagnostic?.forceBudgetBypass, false);
+  assert.equal(calls, 1);
+
+  clock = 1_501;
+  assert.deepEqual(await runtime.run(task, "turn:trc_1", { minIntervalMs: 500, force: true }), { ok: true, status: 200, data: 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" });

web/hwlab-cloud-web/src/config/workbench-runtime-policy.ts

@@ -16,6 +16,7 @@ export interface WorkbenchRuntimePolicy {
   workbenchReadFailureCooldownMs: number;
   workbenchTurnStatusMinRefreshMs: number;
   workbenchTraceEventsMinRefreshMs: number;
+  workbenchSessionDetailMinRefreshMs: number;
   workbenchSessionMessagesMinRefreshMs: number;
   workbenchRealtimeSessionMessagesMinRefreshMs: number;
   workbenchTraceEventsTimeoutMs: number;
@@ -42,6 +43,7 @@ const DEFAULT_WORKBENCH_RUNTIME_POLICY: WorkbenchRuntimePolicy = Object.freeze({
   workbenchReadFailureCooldownMs: 5_000,
   workbenchTurnStatusMinRefreshMs: 2_000,
   workbenchTraceEventsMinRefreshMs: 4_000,
+  workbenchSessionDetailMinRefreshMs: 5_000,
   workbenchSessionMessagesMinRefreshMs: 5_000,
   workbenchRealtimeSessionMessagesMinRefreshMs: 1_000,
   workbenchTraceEventsTimeoutMs: 5_000,
@@ -70,6 +72,7 @@ export function workbenchRuntimePolicy(input: unknown = runtimePolicyConfig()):
     workbenchReadFailureCooldownMs: positiveNumber(source.workbenchReadFailureCooldownMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchReadFailureCooldownMs),
     workbenchTurnStatusMinRefreshMs: nonNegativeNumber(source.workbenchTurnStatusMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchTurnStatusMinRefreshMs),
     workbenchTraceEventsMinRefreshMs: nonNegativeNumber(source.workbenchTraceEventsMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchTraceEventsMinRefreshMs),
+    workbenchSessionDetailMinRefreshMs: nonNegativeNumber(source.workbenchSessionDetailMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchSessionDetailMinRefreshMs),
     workbenchSessionMessagesMinRefreshMs: nonNegativeNumber(source.workbenchSessionMessagesMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchSessionMessagesMinRefreshMs),
     workbenchRealtimeSessionMessagesMinRefreshMs: nonNegativeNumber(source.workbenchRealtimeSessionMessagesMinRefreshMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchRealtimeSessionMessagesMinRefreshMs),
     workbenchTraceEventsTimeoutMs: positiveNumber(source.workbenchTraceEventsTimeoutMs, DEFAULT_WORKBENCH_RUNTIME_POLICY.workbenchTraceEventsTimeoutMs),

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

@@ -34,11 +34,13 @@ export interface WorkbenchRealtimeRecoveryContext {
   activeTraceAuthorized?: boolean;
 }
 
+export type WorkbenchRealtimeRecoveryAuthority = "automatic-recovery";
+
 export type WorkbenchRealtimeRecoveryStep =
-  | { type: "refresh-session-messages"; sessionId: string; reason: string; force: true }
+  | { type: "refresh-session-messages"; sessionId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority }
-  | { type: "schedule-session-list"; sessionId: string }
+  | { type: "schedule-session-list"; sessionId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority }
-  | { type: "refresh-turn-status"; traceId: string; force: true }
+  | { type: "refresh-turn-status"; traceId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority }
-  | { type: "hydrate-trace-events"; traceId: string; force: true };
+  | { type: "hydrate-trace-events"; traceId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority };
 
 export interface WorkbenchRealtimeRecoveryPlan {
   sessionId: string | null;
@@ -72,11 +74,11 @@ export function planWorkbenchRealtimeRecovery(recovery: WorkbenchStreamTransport
   const traceId = firstNonEmptyString(recovery.traceId, context.fallbackTraceId);
   const actions = new Set(recovery.actions);
   const steps: WorkbenchRealtimeRecoveryStep[] = [];
-  if (actions.has("refresh-session-messages") && sessionId && sessionId === context.activeSessionId) steps.push({ type: "refresh-session-messages", sessionId, reason: "realtime-error:messages", force: true });
+  if (actions.has("refresh-session-messages") && sessionId && sessionId === context.activeSessionId) steps.push(automaticRecoveryStep({ type: "refresh-session-messages", sessionId, reason: "realtime-error:messages" }));
-  if (actions.has("schedule-session-list") && sessionId) steps.push({ type: "schedule-session-list", sessionId });
+  if (actions.has("schedule-session-list") && sessionId) steps.push(automaticRecoveryStep({ type: "schedule-session-list", sessionId, reason: "realtime-error:session-list" }));
   if (traceId && context.activeTraceAuthorized === true) {
-    if (actions.has("refresh-turn-status")) steps.push({ type: "refresh-turn-status", traceId, force: true });
+    if (actions.has("refresh-turn-status")) steps.push(automaticRecoveryStep({ type: "refresh-turn-status", traceId, reason: "realtime-error:turn-status" }));
-    if (actions.has("hydrate-trace-events")) steps.push({ type: "hydrate-trace-events", traceId, force: true });
+    if (actions.has("hydrate-trace-events")) steps.push(automaticRecoveryStep({ type: "hydrate-trace-events", traceId, reason: "realtime-error:trace-events" }));
   }
   return {
     sessionId: sessionId ?? null,
@@ -92,6 +94,10 @@ export function planWorkbenchRealtimeRecovery(recovery: WorkbenchStreamTransport
   };
 }
 
+function automaticRecoveryStep(step: Omit<WorkbenchRealtimeRecoveryStep, "authority" | "force">): WorkbenchRealtimeRecoveryStep {
+  return { ...step, authority: "automatic-recovery", force: false } as WorkbenchRealtimeRecoveryStep;
+}
+
 function applySteps(action: WorkbenchRealtimeAction): WorkbenchRealtimeApplyStep[] {
   switch (action.type) {
     case "trace.snapshot":

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

@@ -544,7 +544,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     if (!id || id !== activeSessionId.value) return;
     await realtimeSessionMessagesRuntime.run(composeWorkbenchScopedKey("workbench.realtime.session-messages", id), async () => {
       recordActivity(reason);
-      await refreshSessionMessageProjectionPage(id, { force: true });
+      await refreshSessionMessageProjectionPage(id, { force: options.force });
     }, { force: options.force, minIntervalMs: runtimePolicy.workbenchRealtimeSessionMessagesMinRefreshMs, reason });
   }
 
@@ -1093,11 +1093,11 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     if (!id) return;
     const ownerBefore = traceOwnerSessionId(id, null);
     if (ownerBefore === activeSessionId.value) recordActivity(reason);
-    await refreshTurnStatusByTraceId(id, { force: true });
+    await refreshTurnStatusByTraceId(id);
     const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore;
     if (ownerSessionId) {
       scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListRealtimeRefreshDelayMs);
-      await refreshMessageProjectionForTrace(ownerSessionId, id, { force: true });
+      await refreshMessageProjectionForTrace(ownerSessionId, id);
     }
     const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
     const message = latestMessageForTrace(id, ownerMessages);
@@ -1107,7 +1107,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
       clearActiveTraceRestGapFill(id);
       return;
     }
-    if (message) await hydrateTraceEventsForMessage(message, { force: true });
+    if (message) await hydrateTraceEventsForMessage(message);
     const stillActive = currentRequest.value?.traceId === id || isTraceActiveStatus(turn?.status) || isTraceActiveStatus(message?.status);
     if (stillActive) scheduleActiveTraceRestGapFill(id, "active-rest-gap:repeat", runtimePolicy.workbenchActiveTraceRestGapFillRepeatMs);
   }
@@ -1265,18 +1265,18 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     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 });
+      await refreshTurnStatusByTraceId(id);
       const ownerSessionId = traceOwnerSessionId(id, turnStatusAuthority.value[id]?.sessionId ?? null) ?? ownerBefore;
       if (ownerSessionId) {
         scheduleSessionListRefresh(ownerSessionId, runtimePolicy.sessionListTerminalRefreshDelayMs);
-        await refreshMessageProjectionForTrace(ownerSessionId, id, { force: true });
+        await refreshMessageProjectionForTrace(ownerSessionId, id);
       }
       const ownerMessages = ownerSessionId ? serverState.value.messagesBySessionId[ownerSessionId] ?? [] : messages.value;
       const message = latestMessageForTrace(id, ownerMessages);
       if (!message && ownerSessionId) {
-        await refreshRealtimeSessionMessages(ownerSessionId, `${reason}:message-gap`, { force: true });
+        await refreshRealtimeSessionMessages(ownerSessionId, `${reason}:message-gap`);
       }
-    }, { force: true, reason });
+    }, { reason });
   }
 
   function installRealtimeVisibilityHandler(): void {
@@ -1320,7 +1320,7 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     if (!sessionId || sessionId !== activeSessionId.value) return;
     const traceId = firstNonEmptyString(record.traceId);
     if (traceId && messages.value.some((message) => firstNonEmptyString(message.traceId, message.runnerTrace?.traceId) === traceId)) return;
-    void refreshRealtimeSessionMessages(sessionId, `cross-tab-session-projection:${firstNonEmptyString(record.reason, traceId, "session") ?? "session"}`, { force: true });
+    void refreshRealtimeSessionMessages(sessionId, `cross-tab-session-projection:${firstNonEmptyString(record.reason, traceId, "session") ?? "session"}`);
   }
 
   function applyTraceSnapshot(traceId: string, snapshot: TraceSnapshot): void {
@@ -1359,9 +1359,13 @@ export const useWorkbenchStore = defineStore("workbench", () => {
     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);
+      const selected = await runWorkbenchReadHydration(
+        () => loadWorkbenchSession(id, existing),
+        workbenchReadCooldownKey("session-detail", id),
+        { minIntervalMs: runtimePolicy.workbenchSessionDetailMinRefreshMs }
+      );
       if (selected && activeSessionId.value === id) applySelectedSessionDetail(selected, "system");
-    }, { force: true, reason });
+    }, { reason });
   }
 
   function completeTrace(traceId: string, result: AgentChatResultResponse, options: { forceRead?: boolean } = {}): void {

web/hwlab-cloud-web/src/types/global.d.ts

@@ -28,6 +28,7 @@ declare global {
           workbenchReadFailureCooldownMs?: number;
           workbenchTurnStatusMinRefreshMs?: number;
           workbenchTraceEventsMinRefreshMs?: number;
+          workbenchSessionDetailMinRefreshMs?: number;
           workbenchSessionMessagesMinRefreshMs?: number;
           workbenchRealtimeSessionMessagesMinRefreshMs?: number;
           workbenchTraceEventsTimeoutMs?: number;

web/hwlab-cloud-web/src/utils/workbench-refresh-runtime.ts

@@ -27,6 +27,7 @@ export interface WorkbenchReadHydrationRuntimeOptions {
 
 export interface WorkbenchReadHydrationRunOptions {
   minIntervalMs?: number;
+  /** Priority marker only. Runtime budgets still apply to force requests. */
   force?: boolean;
   shouldCooldownFailure?: (value: unknown) => boolean;
   reason?: string | null;
@@ -38,6 +39,7 @@ type TimerHandle = ReturnType<typeof setTimeout>;
 
 export interface WorkbenchKeyedTaskRunOptions {
   minIntervalMs?: number;
+  /** Priority marker only. Runtime budgets still apply to force requests. */
   force?: boolean;
   replace?: boolean;
   reason?: string | null;
@@ -99,7 +101,7 @@ export class WorkbenchReadHydrationRuntime {
       if (cooldownUntilMs > 0) this.cooldownUntilByKey.delete(key);
       const minIntervalMs = Math.max(0, Math.trunc(options.minIntervalMs ?? 0));
       const lastStartedAtMs = this.lastStartedAtByKey.get(key) ?? 0;
-      if (!options.force && minIntervalMs > 0 && lastStartedAtMs > 0 && now - lastStartedAtMs < minIntervalMs) return createWorkbenchReadThrottledResult(key, lastStartedAtMs + minIntervalMs, now) as T;
+      if (minIntervalMs > 0 && lastStartedAtMs > 0 && now - lastStartedAtMs < minIntervalMs) return createWorkbenchReadThrottledResult(key, lastStartedAtMs + minIntervalMs, now, options.force === true) as T;
       if (minIntervalMs > 0) this.lastStartedAtByKey.set(key, now);
     }
     const value = await task();
@@ -129,7 +131,7 @@ export class WorkbenchKeyedTaskRuntime {
     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);
+    if (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>;
   }
@@ -270,10 +272,12 @@ function createWorkbenchReadCooldownResult(key: string, cooldownUntilMs: number,
   return { ok: false, status: 503, data: null, error: message, apiError: { ...diagnostic, message, diagnostic }, diagnostic };
 }
 
-function createWorkbenchReadThrottledResult(key: string, nextAllowedAtMs: number, now: number): ApiResultLike {
+function createWorkbenchReadThrottledResult(key: string, nextAllowedAtMs: number, now: number, forceRequested = false): ApiResultLike {
   const retryAfterMs = Math.max(0, nextAllowedAtMs - now);
   const message = `Workbench read hydration is throttled (${key}); retry after ${retryAfterMs}ms.`;
   const diagnostic = diagnosticEnvelope("workbench_read_hydration_throttled", "throttle", message, key, retryAfterMs);
+  diagnostic.forceRequested = forceRequested;
+  diagnostic.forceBudgetBypass = false;
   return { ok: false, status: 0, data: null, error: message, apiError: { ...diagnostic, message, diagnostic }, diagnostic };
 }