fix: bound workbench realtime recovery reads
This commit is contained in:
UniDesk Codex
@@ -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`，不要在业务组件里各自手写跳转判断。
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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" });
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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: "schedule-session-list"; sessionId: string }
|
||||
| { type: "refresh-turn-status"; traceId: string; force: true }
|
||||
| { type: "hydrate-trace-events"; traceId: string; force: true };
|
||||
| { type: "refresh-session-messages"; sessionId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority }
|
||||
| { type: "schedule-session-list"; sessionId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority }
|
||||
| { type: "refresh-turn-status"; traceId: string; reason: string; force: false; authority: WorkbenchRealtimeRecoveryAuthority }
|
||||
| { 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("schedule-session-list") && sessionId) steps.push({ type: "schedule-session-list", sessionId });
|
||||
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(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("hydrate-trace-events")) steps.push({ type: "hydrate-trace-events", 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(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":
|
||||
|
||||
@@ -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 {
|
||||
|
||||
+1
@@ -28,6 +28,7 @@ declare global {
|
||||
workbenchReadFailureCooldownMs?: number;
|
||||
workbenchTurnStatusMinRefreshMs?: number;
|
||||
workbenchTraceEventsMinRefreshMs?: number;
|
||||
workbenchSessionDetailMinRefreshMs?: number;
|
||||
workbenchSessionMessagesMinRefreshMs?: number;
|
||||
workbenchRealtimeSessionMessagesMinRefreshMs?: number;
|
||||
workbenchTraceEventsTimeoutMs?: number;
|
||||
|
||||
@@ -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 };
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user