diff --git a/packages/ui/src/lib/opencode/client.ts b/packages/ui/src/lib/opencode/client.ts index 394e0c42..29a4c278 100644 --- a/packages/ui/src/lib/opencode/client.ts +++ b/packages/ui/src/lib/opencode/client.ts @@ -15,12 +15,25 @@ import type { import type { PermissionRequest } from "@/types/permission"; import type { QuestionRequest } from "@/types/question"; import { waitForWorktreeBootstrap } from "@/lib/worktrees/worktreeBootstrap"; +import { + assertProviderCircuitClosed, + recordProviderSuccess, + recordProviderError, + shouldRetry, + getRetryDelayMs, +} from "./provider-tracker"; // Use relative path by default (works with both dev and nginx proxy server) // Can be overridden with VITE_OPENCODE_URL for absolute URLs in special deployments const DEFAULT_BASE_URL = import.meta.env.VITE_OPENCODE_URL || "/api"; const ABSOLUTE_URL_PATTERN = /^[a-zA-Z][a-zA-Z\d+\-.]*:\/\//; +const isRetryableFetchError = (error: unknown): boolean => { + if (error instanceof DOMException && error.name === 'AbortError') return true; + if (error instanceof TypeError) return true; + return false; +}; + const ensureAbsoluteBaseUrl = (candidate: string): string => { const normalized = typeof candidate === "string" && candidate.trim().length > 0 ? candidate.trim() : "/api"; @@ -725,39 +738,58 @@ class OpencodeService { }); } - let response: Response; - try { - response = await fetch(url.toString(), { - method: 'POST', - headers: { - 'content-type': 'application/json', - accept: 'application/json', - }, - body: JSON.stringify({ - model: { - providerID: params.providerID, - modelID: params.modelID, - }, - agent: params.agent, - variant: params.variant, - ...(params.messageId ? { messageID: params.messageId } : {}), - ...(params.format ? { format: params.format } : {}), - parts, - }), - }); - } catch (error) { - console.error('[git-generation][browser] prompt_async request failed before response', { - sessionId: params.id, - url: url.toString(), - directory: this.currentDirectory, - hasFormat: Boolean(params.format), - message: error instanceof Error ? error.message : String(error), - error, - }); - throw error; - } + assertProviderCircuitClosed(params.providerID); + + let response!: Response; + + for (let attempt = 0; attempt < 3; attempt++) { + try { + response = await fetch(url.toString(), { + method: 'POST', + headers: { + 'content-type': 'application/json', + accept: 'application/json', + }, + body: JSON.stringify({ + model: { + providerID: params.providerID, + modelID: params.modelID, + }, + agent: params.agent, + variant: params.variant, + ...(params.messageId ? { messageID: params.messageId } : {}), + ...(params.format ? { format: params.format } : {}), + parts, + }), + }); + } catch (error) { + if (attempt < 2 && isRetryableFetchError(error)) { + const delay = getRetryDelayMs(attempt); + console.warn( + `[prompt] fetch failed for ${params.providerID}/${params.modelID} (attempt ${attempt + 1}/3), retrying in ${delay}ms`, + (error as Error)?.message + ); + await new Promise((resolve) => setTimeout(resolve, delay)); + continue; + } + recordProviderError(params.providerID); + throw error; + } + + if (response.ok) { + recordProviderSuccess(params.providerID); + return tempMessageId; + } + + if (shouldRetry(params.providerID, response.status, attempt)) { + const delay = getRetryDelayMs(attempt); + console.warn( + `[prompt] ${response.status} for ${params.providerID}/${params.modelID} (attempt ${attempt + 1}/3), retrying in ${delay}ms` + ); + await new Promise((resolve) => setTimeout(resolve, delay)); + continue; + } - if (!response.ok) { let detail = ''; try { detail = await response.text(); @@ -765,12 +797,13 @@ class OpencodeService { // ignore } const suffix = detail && detail.trim().length > 0 ? `: ${detail.trim()}` : ''; - throw new Error(`Failed to send message (${response.status})${suffix}`); + const error = new Error(`Failed to send message (${response.status})${suffix}`); + recordProviderError(params.providerID, response.status); + throw error; } - - // Return temporary ID for optimistic UI - // Real messageID will come from server via SSE events - return tempMessageId; + // Defensive fallback — all loop paths return/throw, but TypeScript + // control flow analysis cannot prove exhaustiveness without this. + throw new Error('Failed to send message after retries'); } async sendCommand(params: { diff --git a/packages/ui/src/lib/opencode/provider-tracker.ts b/packages/ui/src/lib/opencode/provider-tracker.ts new file mode 100644 index 00000000..41ee8352 --- /dev/null +++ b/packages/ui/src/lib/opencode/provider-tracker.ts @@ -0,0 +1,133 @@ +/** + * Provider Circuit-Breaker & Retry Tracker + * + * Tracks per-provider error state to enable: + * - Transparent retry with exponential backoff for transient errors + * - Circuit breaking (pause requests to a provider during error storms) + * + * Inspired by HiveMind (arXiv:2604.17111) OS-inspired scheduling primitives. + */ + +const DEFAULT_CIRCUIT_BREAK_THRESHOLD = 3 +const DEFAULT_CIRCUIT_COOLDOWN_MS = 30_000 +const DEFAULT_RETRY_BASE_DELAY_MS = 1000 +const DEFAULT_RETRY_MAX_DELAY_MS = 32_000 +const DEFAULT_RETRY_MAX_ATTEMPTS = 3 +const PROVIDER_EVICTION_TTL_MS = 60 * 60 * 1000 +const PROVIDER_EVICTION_INTERVAL_MS = 10 * 60 * 1000 + +const RETRYABLE_STATUS_CODES = new Set([429, 502, 503, 504]) + +type ProviderState = { + consecutiveErrors: number + lastErrorAt: number + circuitOpen: boolean + circuitOpenAt: number + circuitCooldownMs: number +} + +const providers = new Map() + +function evictStaleProviders(): void { + const now = Date.now() + for (const [providerID, state] of providers) { + if ( + state.consecutiveErrors === 0 && + now - state.lastErrorAt > PROVIDER_EVICTION_TTL_MS + ) { + providers.delete(providerID) + } + } +} + +if (typeof setInterval !== 'undefined') { + setInterval(evictStaleProviders, PROVIDER_EVICTION_INTERVAL_MS) +} + +function getOrCreateProvider(providerID: string): ProviderState { + let state = providers.get(providerID) + if (!state) { + state = { + consecutiveErrors: 0, + lastErrorAt: 0, + circuitOpen: false, + circuitOpenAt: 0, + circuitCooldownMs: DEFAULT_CIRCUIT_COOLDOWN_MS, + } + providers.set(providerID, state) + } + return state +} + +export function recordProviderSuccess(providerID: string): void { + if (!providerID) return + const state = providers.get(providerID) + if (!state) return + state.consecutiveErrors = 0 + state.lastErrorAt = 0 +} + +export function recordProviderError(providerID: string, status?: number): void { + if (!providerID) return + const state = getOrCreateProvider(providerID) + state.consecutiveErrors += 1 + state.lastErrorAt = Date.now() + + if ( + isCircuitBreakerStatus(status) && + state.consecutiveErrors >= DEFAULT_CIRCUIT_BREAK_THRESHOLD + ) { + state.circuitOpen = true + state.circuitOpenAt = Date.now() + console.warn( + `[provider-tracker] Circuit opened for ${providerID} after ${state.consecutiveErrors} consecutive errors` + ) + } +} + +function isCircuitBreakerStatus(status?: number): boolean { + return status !== undefined && RETRYABLE_STATUS_CODES.has(status) +} + +export function isCircuitOpen(providerID: string): boolean { + const state = providers.get(providerID) + if (!state?.circuitOpen) return false + + const elapsed = Date.now() - state.circuitOpenAt + if (elapsed >= state.circuitCooldownMs) { + state.circuitOpen = false + state.consecutiveErrors = 0 + state.circuitCooldownMs = Math.min( + state.circuitCooldownMs * 2, + DEFAULT_RETRY_MAX_DELAY_MS * 4 + ) + return false + } + + return true +} + +export function shouldRetry(providerID: string, status: number, attempt: number): boolean { + if (!RETRYABLE_STATUS_CODES.has(status)) return false + if (attempt >= DEFAULT_RETRY_MAX_ATTEMPTS - 1) return false + if (isCircuitOpen(providerID)) return false + return true +} + +export function assertProviderCircuitClosed(providerID: string): void { + if (!providerID || !isCircuitOpen(providerID)) return + throw new Error(`Provider ${providerID} is temporarily unavailable after repeated errors. Please retry shortly.`) +} + +export function getRetryDelayMs(attempt: number): number { + const delay = DEFAULT_RETRY_BASE_DELAY_MS * 2 ** attempt + return Math.min(delay, DEFAULT_RETRY_MAX_DELAY_MS) +} + +export function resetCircuit(providerID: string): void { + const state = providers.get(providerID) + if (!state) return + state.consecutiveErrors = 0 + state.circuitOpen = false + state.circuitCooldownMs = DEFAULT_CIRCUIT_COOLDOWN_MS +} diff --git a/packages/web/server/index.js b/packages/web/server/index.js index 5612e7f5..b830f25f 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -35,6 +35,8 @@ import { createGlobalUiEventBroadcaster, createGlobalMessageStreamHub, createMessageStreamWsRuntime, + DEFAULT_UPSTREAM_STALL_TIMEOUT_MS, + UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS, } from './lib/event-stream/index.js'; import { createFsSearchRuntime as createFsSearchRuntimeFactory } from './lib/fs/search.js'; import { createOpenCodeLifecycleRuntime } from './lib/opencode/lifecycle.js'; @@ -381,6 +383,17 @@ const sessionRuntime = createSessionRuntime({ broadcastEvent: broadcastGlobalUiEvent, }); +const getActiveSessionCount = () => { + const snapshot = sessionRuntime.getSessionActivitySnapshot(); + return Object.values(snapshot).filter((entry) => entry.type === 'busy').length; +}; + +const getUpstreamStallTimeoutMs = () => ( + getActiveSessionCount() > 1 + ? UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS + : DEFAULT_UPSTREAM_STALL_TIMEOUT_MS +); + const projectConfigRuntime = createProjectConfigRuntime({ fsPromises, path, @@ -655,6 +668,7 @@ const setAutoAcceptSession = (...args) => notificationTriggerRuntime.setAutoAcce const globalMessageStreamHub = createGlobalMessageStreamHub({ buildOpenCodeUrl, getOpenCodeAuthHeaders, + upstreamStallTimeoutMs: getUpstreamStallTimeoutMs, }); const openCodeWatcherRuntime = createOpenCodeWatcherRuntime({ @@ -881,6 +895,7 @@ const openCodeLifecycleRuntime = createOpenCodeLifecycleRuntime({ buildAugmentedPath, buildManagedOpenCodePath, getManagedOpenCodeShellEnvSnapshot: getLoginShellEnvSnapshot, + getActiveSessionCount, }); const restartOpenCode = (...args) => openCodeLifecycleRuntime.restartOpenCode(...args); @@ -1188,6 +1203,7 @@ async function main(options = {}) { globalEventHub: globalMessageStreamHub, processForwardedEventPayload, messageStreamWsClients: uiNotificationWsClients, + upstreamStallTimeoutMs: getUpstreamStallTimeoutMs, terminalHeartbeatIntervalMs: TERMINAL_INPUT_WS_HEARTBEAT_INTERVAL_MS, terminalRebindWindowMs: TERMINAL_INPUT_WS_REBIND_WINDOW_MS, terminalMaxRebindsPerWindow: TERMINAL_INPUT_WS_MAX_REBINDS_PER_WINDOW, diff --git a/packages/web/server/lib/event-stream/index.js b/packages/web/server/lib/event-stream/index.js index e55309f0..06dae6ef 100644 --- a/packages/web/server/lib/event-stream/index.js +++ b/packages/web/server/lib/event-stream/index.js @@ -20,5 +20,6 @@ export { export { DEFAULT_UPSTREAM_RECONNECT_DELAY_MS, DEFAULT_UPSTREAM_STALL_TIMEOUT_MS, + UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS, createUpstreamSseReader, } from './upstream-reader.js'; diff --git a/packages/web/server/lib/event-stream/upstream-reader.js b/packages/web/server/lib/event-stream/upstream-reader.js index 253a1e9c..0acff519 100644 --- a/packages/web/server/lib/event-stream/upstream-reader.js +++ b/packages/web/server/lib/event-stream/upstream-reader.js @@ -1,8 +1,14 @@ import { parseSseEventEnvelope } from './protocol.js'; export const DEFAULT_UPSTREAM_STALL_TIMEOUT_MS = 20_000; +export const UPSTREAM_STALL_TIMEOUT_CONCURRENT_MS = DEFAULT_UPSTREAM_STALL_TIMEOUT_MS * 3; export const DEFAULT_UPSTREAM_RECONNECT_DELAY_MS = 250; +function resolveTimeoutMs(value, fallback) { + const resolved = typeof value === 'function' ? value() : value; + return Number.isFinite(resolved) ? resolved : fallback; +} + function waitForReconnectDelay(ms, signal) { if (signal?.aborted) { return Promise.resolve(); @@ -76,14 +82,15 @@ export function createUpstreamSseReader({ }; const resetStallTimer = () => { clearStallTimer(); - if (stallTimeoutMs <= 0) { + const currentStallTimeoutMs = resolveTimeoutMs(stallTimeoutMs, DEFAULT_UPSTREAM_STALL_TIMEOUT_MS); + if (currentStallTimeoutMs <= 0) { return; } stallTimer = setTimeout(() => { abortReason = 'upstream_stalled'; controller.abort(); - }, stallTimeoutMs); + }, currentStallTimeoutMs); }; try { diff --git a/packages/web/server/lib/event-stream/upstream-reader.test.js b/packages/web/server/lib/event-stream/upstream-reader.test.js index 652dd340..616b5b96 100644 --- a/packages/web/server/lib/event-stream/upstream-reader.test.js +++ b/packages/web/server/lib/event-stream/upstream-reader.test.js @@ -117,6 +117,51 @@ describe('createUpstreamSseReader', () => { expect(reader.getLastEventId()).toBe('evt-2'); }); + it('resolves the stall timeout for each upstream read window', async () => { + const events = []; + let attempt = 0; + let currentTimeout = 10; + let reader; + + reader = createUpstreamSseReader({ + buildUrl: () => 'http://127.0.0.1:4096/global/event', + stallTimeoutMs: () => currentTimeout, + reconnectDelayMs: 0, + fetchImpl: async (_url, options) => { + attempt += 1; + + if (attempt === 1) { + currentTimeout = 60; + return createSseResponse({ + signal: options.signal, + holdOpen: true, + blocks: [ + 'id: evt-1\ndata: {"type":"server.connected","properties":{}}\n\n', + ], + }); + } + + return createSseResponse({ + signal: options.signal, + blocks: [ + 'id: evt-2\ndata: {"type":"session.updated","properties":{}}\n\n', + ], + }); + }, + onEvent(event) { + events.push(event.eventId); + if (event.eventId === 'evt-2') { + reader.stop(); + } + }, + }); + + await reader.start(); + + expect(events).toEqual(['evt-1', 'evt-2']); + expect(attempt).toBe(2); + }); + it('reports unavailable upstream responses and continues reconnecting until stopped', async () => { const errors = []; let attempt = 0; diff --git a/packages/web/server/lib/opencode/lifecycle.js b/packages/web/server/lib/opencode/lifecycle.js index 9b2519ac..536680ba 100644 --- a/packages/web/server/lib/opencode/lifecycle.js +++ b/packages/web/server/lib/opencode/lifecycle.js @@ -25,6 +25,7 @@ export const createOpenCodeLifecycleRuntime = (deps) => { buildAugmentedPath, buildManagedOpenCodePath, getManagedOpenCodeShellEnvSnapshot, + getActiveSessionCount = () => 0, } = deps; const killProcessOnPort = (port) => { @@ -796,15 +797,51 @@ export const createOpenCodeLifecycleRuntime = (deps) => { * Perform an immediate (one-shot) health check and restart OpenCode if it's * not healthy. Callers on the SSE / WS proxy path use this to trigger * recovery without waiting for the next periodic interval (up to 15 s). + * + * Skips restart when sessions are actively busy — a busy server under + * concurrent load can fail the health check timeout without actually + * being dead (the health endpoint competes with LLM work). + * Forces restart if sessions stay "busy" and the server stays unhealthy + * for over 2 minutes (staleness guard against stuck session state). */ + const STALE_BUSY_GRACE_MS = 2 * 60 * 1000; + let lastUnhealthyWithBusySessionsAt = 0; + + const shouldSkipRestartForBusySessions = () => { + const activeCount = getActiveSessionCount(); + if (activeCount === 0) { + lastUnhealthyWithBusySessionsAt = 0; + return false; + } + + const now = Date.now(); + if (!lastUnhealthyWithBusySessionsAt) { + lastUnhealthyWithBusySessionsAt = now; + return true; + } + + if (now - lastUnhealthyWithBusySessionsAt >= STALE_BUSY_GRACE_MS) { + console.warn( + `[lifecycle] OpenCode unhealthy with ${activeCount} busy session(s) for > 2 min — forcing restart` + ); + lastUnhealthyWithBusySessionsAt = 0; + return false; + } + + return true; + }; + const triggerHealthCheck = async () => { if (!state.openCodeProcess || state.isShuttingDown || state.isRestartingOpenCode) return; try { const healthy = await isOpenCodeProcessHealthy(); if (!healthy) { + if (shouldSkipRestartForBusySessions()) return; console.log('[lifecycle] immediate health check: OpenCode not healthy, restarting...'); await restartOpenCode(); + } else { + lastUnhealthyWithBusySessionsAt = 0; } } catch (error) { console.error(`[lifecycle] immediate health check error: ${error.message}`); @@ -822,8 +859,11 @@ export const createOpenCodeLifecycleRuntime = (deps) => { try { const healthy = await isOpenCodeProcessHealthy(); if (!healthy) { + if (shouldSkipRestartForBusySessions()) return; console.log('OpenCode process not running, restarting...'); await restartOpenCode(); + } else { + lastUnhealthyWithBusySessionsAt = 0; } } catch (error) { console.error(`Health check error: ${error.message}`); diff --git a/packages/web/server/lib/opencode/startup-pipeline-runtime.js b/packages/web/server/lib/opencode/startup-pipeline-runtime.js index c17c5be9..6011df09 100644 --- a/packages/web/server/lib/opencode/startup-pipeline-runtime.js +++ b/packages/web/server/lib/opencode/startup-pipeline-runtime.js @@ -24,6 +24,7 @@ export const createStartupPipelineRuntime = (dependencies) => { processForwardedEventPayload, messageStreamWsClients, triggerHealthCheck, + upstreamStallTimeoutMs, terminalHeartbeatIntervalMs, terminalRebindWindowMs, terminalMaxRebindsPerWindow, @@ -80,6 +81,7 @@ export const createStartupPipelineRuntime = (dependencies) => { processForwardedEventPayload, wsClients: messageStreamWsClients, triggerHealthCheck, + upstreamStallTimeoutMs, }); setupProxy(app);