diff --git a/cli/src/opencode/opencodeRemoteLauncher.test.ts b/cli/src/opencode/opencodeRemoteLauncher.test.ts index 48f180c1..92235905 100644 --- a/cli/src/opencode/opencodeRemoteLauncher.test.ts +++ b/cli/src/opencode/opencodeRemoteLauncher.test.ts @@ -37,7 +37,14 @@ const harness = vi.hoisted(() => ({ newSessionImpl: null as null | (() => Promise), disconnectImpl: null as null | (() => Promise), permissionCancelError: null as Error | null, - serverStopError: null as Error | null + serverStopError: null as Error | null, + eventStreamOptions: [] as Array<{ + baseUrl: string; + directory: string; + sessionId: string; + onRetry: (retry: { attempt: number; message: string }) => void; + }>, + eventStreamCloseCount: 0 })); // Captures the RemoteLauncherDisplayContext (including onExit/ @@ -188,6 +195,25 @@ const compactHarness = vi.hoisted(() => ({ >) })); +// Partial: only the subscription is stubbed. The retry formatter is pure and +// its output is part of what this launcher is asserted to forward. +vi.mock('./utils/opencodeEventStream', async (importOriginal) => ({ + ...(await importOriginal()), + subscribeToOpencodeEvents: vi.fn((options: { + baseUrl: string; + directory: string; + sessionId: string; + onRetry: (retry: { attempt: number; message: string }) => void; + }) => { + harness.eventStreamOptions.push(options); + return { + close: () => { + harness.eventStreamCloseCount++; + } + }; + }) +})); + vi.mock('./utils/opencodeCompactBridge', () => ({ splitProviderModel: (combined: string | null | undefined) => { if (!combined) return null; @@ -281,6 +307,7 @@ function createSessionStub( const sessionEvents: Array<{ type: string; [key: string]: unknown }> = []; const sentAgentMessages: unknown[] = []; + const claudeSessionMessages: unknown[] = []; const rpcHandlers = new Map unknown>(); const setModelReasoningEffort = vi.fn(); const pushKeepAlive = vi.fn(); @@ -294,7 +321,9 @@ function createSessionStub( } }, sendAgentMessage(_message: unknown) {}, - sendClaudeSessionMessage(_message: unknown) {}, + sendClaudeSessionMessage(message: unknown) { + claudeSessionMessages.push(message); + }, sendUserMessage(_text: string) {}, sendSessionEvent(event: { type: string; [key: string]: unknown }) { sessionEvents.push(event); @@ -333,7 +362,7 @@ function createSessionStub( sendUserMessage(_text: string) {} }; - return { session, sessionEvents, sentAgentMessages, agentMessages: sentAgentMessages, rpcHandlers, setModelReasoningEffort, pushKeepAlive, emitMessagesConsumedCalls, thinkingChangeCalls }; + return { session, sessionEvents, sentAgentMessages, agentMessages: sentAgentMessages, claudeSessionMessages, rpcHandlers, setModelReasoningEffort, pushKeepAlive, emitMessagesConsumedCalls, thinkingChangeCalls }; } function createCompactMode(model?: string): OpencodeMode { @@ -385,6 +414,8 @@ describe('opencodeRemoteLauncher inline model switch', () => { harness.disconnectImpl = null; harness.permissionCancelError = null; harness.serverStopError = null; + harness.eventStreamOptions = []; + harness.eventStreamCloseCount = 0; inkHarness.lastRenderProps = null; }); @@ -2035,6 +2066,138 @@ describe('opencodeRemoteLauncher inline model switch', () => { await launchPromise; }); + it('subscribes to the agent event stream for this session and directory, and closes it on teardown', async () => { + const { session } = createSessionStub([ + { message: 'first', mode: createMode() } + ]); + + await opencodeRemoteLauncher(session as never); + + expect(harness.eventStreamOptions).toHaveLength(1); + expect(harness.eventStreamOptions[0]).toMatchObject({ + baseUrl: 'http://127.0.0.1:48273', + // Without the directory the endpoint reports nothing at all, and + // says so with neither an error nor a 404. + directory: '/tmp/hapi-opencode-test', + sessionId: 'acp-session-1' + }); + // A stream left open would keep reconnecting to a process that is gone. + expect(harness.eventStreamCloseCount).toBe(1); + }); + + it('reports an upstream retry as progress carrying the provider text, not as a failure', async () => { + harness.hangPrompt = true; + const { session, agentMessages, claudeSessionMessages, thinkingChangeCalls } = createSessionStub([ + { message: 'first', mode: createMode() } + ]); + + const launchPromise = opencodeRemoteLauncher(session as never); + await vi.waitFor(() => expect(harness.eventStreamOptions).toHaveLength(1)); + await vi.waitFor(() => expect(harness.events).toContain('prompt:start')); + + harness.eventStreamOptions[0].onRetry({ + attempt: 2, + message: 'Rate limit exceeded: free-models-per-day.' + }); + + // Sent on the api_error channel the web timeline folds, not the error + // channel — the agent is still working. + expect(agentMessages).toEqual([]); + expect(claudeSessionMessages).toHaveLength(1); + expect(claudeSessionMessages[0]).toMatchObject({ + type: 'system', + subtype: 'api_error', + retryAttempt: 2, + maxRetries: 0 + }); + // The provider's own words survive; without them the timeline says + // only "Retrying...". + const error = (claudeSessionMessages[0] as { error: { message: string } }).error; + expect(error.message).toContain('Rate limit exceeded: free-models-per-day.'); + expect(error.message).toContain('attempt 2'); + // A retry is not a turn boundary: the session really is still busy, + // and the hub owns how that composes with its queue grace. + expect(session.thinking).toBe(true); + expect(thinkingChangeCalls).toEqual([true]); + + harness.resolvePrompt!(); + await launchPromise; + }); + + it('tells the user why the prompt failed instead of pointing at logs they cannot read', async () => { + // The agent's own explanation already reaches this process: the ACP + // transport rejects `session/prompt` with the JSON-RPC error message + // verbatim (AcpStdioTransport's response handler). Discarding it left + // a remote user — the only user this launcher has — staring at a + // session that stopped for no stated reason, with "check the logs" + // pointing at a machine they are not sitting in front of. + harness.promptImpl = async () => { + throw new Error('Internal error: Rate limit exceeded: free-models-per-day. Add credits to unlock 1000 free models.'); + }; + const { session, agentMessages } = createSessionStub([ + { message: 'first', mode: createMode() } + ]); + + await opencodeRemoteLauncher(session as never); + + expect(agentMessages).toEqual([{ + type: 'error', + message: 'OpenCode prompt failed: Rate limit exceeded: free-models-per-day. Add credits to unlock 1000 free models.' + }]); + }); + + it('delivers both the raw stderr dump and the readable sentence on a hard error', async () => { + // Measured (isolated E2E, stub provider answering 400): OpenCode + // dumps the JSON-RPC error onto stderr, the shared ACP reader splits + // it line by line, and each line is reported — then the rejected + // session/prompt reaches the catch. Both arrive, on purpose. The + // fragments are upstream's existing behaviour and the sentence is + // the only line of the set a person can read, so it follows them as + // a summary rather than as a repeat. An earlier revision suppressed + // it as a duplicate; see reportPromptFailure's doc comment for why + // that is not coming back. + harness.promptImpl = async () => { + harness.stderrHandler!({ + type: 'unknown', + message: 'Error handling request {', + raw: 'Error handling request {' + }); + harness.stderrHandler!({ + type: 'unknown', + message: 'message: "Internal error: SENTINEL stub rejected the request",', + raw: 'message: "Internal error: SENTINEL stub rejected the request",' + }); + throw new Error('Internal error: SENTINEL stub rejected the request'); + }; + const { session, agentMessages } = createSessionStub([ + { message: 'first', mode: createMode() } + ]); + + await opencodeRemoteLauncher(session as never); + + expect(agentMessages).toEqual([ + { type: 'error', message: 'Error handling request {' }, + { type: 'error', message: 'message: "Internal error: SENTINEL stub rejected the request",' }, + { type: 'error', message: 'OpenCode prompt failed: SENTINEL stub rejected the request' } + ]); + }); + + it('falls back to the original wording when the failure carries no message at all', async () => { + harness.promptImpl = async () => { + throw new Error(''); + }; + const { session, agentMessages } = createSessionStub([ + { message: 'first', mode: createMode() } + ]); + + await opencodeRemoteLauncher(session as never); + + expect(agentMessages).toEqual([{ + type: 'error', + message: 'OpenCode prompt failed. Check logs for details.' + }]); + }); + it('serializes setModel after the previous prompt resolves', async () => { const { session } = createSessionStub([ { message: 'first', mode: createMode('ollama/a') }, diff --git a/cli/src/opencode/opencodeRemoteLauncher.ts b/cli/src/opencode/opencodeRemoteLauncher.ts index 05cf4673..d67aee49 100644 --- a/cli/src/opencode/opencodeRemoteLauncher.ts +++ b/cli/src/opencode/opencodeRemoteLauncher.ts @@ -14,6 +14,13 @@ import type { OpencodeMode, PermissionMode } from './types'; import { RPC_METHODS } from '@hapi/protocol/rpcMethods'; import { allocateFreePort, createOpencodeBackend } from './utils/opencodeBackend'; import { captureCompactionMarkerSnapshot, fetchCompactionResult, splitProviderModel, triggerOpencodeCompact } from './utils/opencodeCompactBridge'; +import { formatOpencodePromptError } from './utils/opencodeErrorText'; +import { + formatOpencodeRetryStatus, + subscribeToOpencodeEvents, + type OpencodeEventSubscription, + type OpencodeRetryStatus +} from './utils/opencodeEventStream'; import { OpencodePermissionHandler } from './utils/permissionHandler'; import { getOpencodeNativeToolInstruction, PLAN_MODE_INSTRUCTION } from './utils/systemPrompt'; import { resolveThoughtLevelEffort } from './thoughtLevelEffort'; @@ -124,6 +131,8 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { private setModelSupported: boolean | undefined = undefined; private setEffortSupported: boolean | undefined = undefined; private activeAcpSessionId: string | null = null; + /** Subscription to the agent's own server event stream; null until the ACP session id is known, closed in cleanup(). */ + private eventStream: OpencodeEventSubscription | null = null; private stallErrorReportedForPrompt = false; constructor( @@ -163,7 +172,8 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { // opencodeCompactBridge.ts). const hostname = '127.0.0.1'; const port = await allocateFreePort(hostname); - this.baseUrl = `http://${hostname}:${port}`; + const baseUrl = `http://${hostname}:${port}`; + this.baseUrl = baseUrl; const backend = createOpencodeBackend({ cwd: session.path, @@ -209,6 +219,28 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { session.onSessionFound(acpSessionId); this.activeAcpSessionId = acpSessionId; + // Upstream retries are announced only on the agent's own event + // stream — not as ACP notifications and not on stderr. Measured + // against a provider stubbed to answer 429: 40 minutes, 85 retries, + // zero ACP updates, zero stderr bytes, and a prompt that never + // settled. Without this the user watches a session that looks like + // it is thinking and never learns it is rate limited. + // + // Attached here rather than at spawn time because the stream is + // filtered by ACP session id, which only exists once new/loadSession + // has answered. Failing to subscribe degrades that visibility and + // nothing else, so it is deliberately neither awaited nor + // error-checked; the subscription reports its own failures at debug + // level and keeps the session running. + this.eventStream = subscribeToOpencodeEvents({ + baseUrl, + // Required. Omitting it yields heartbeats and no session events + // at all, with no error to notice — see the subscription's doc. + directory: session.path, + sessionId: acpSessionId, + onRetry: (retry) => this.surfaceUpstreamRetry(retry) + }); + // Seed currentBackendModel from the ACP session metadata so the first // batch — whose model the hub mirrors from the just-discovered session — // does not trigger a redundant setModel on the very first turn. @@ -584,7 +616,7 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { void backend.refreshSessionInfo(acpSessionId, session.path); } catch (error) { logger.warn('[opencode-remote] prompt failed', error); - this.surfaceAgentError('OpenCode prompt failed. Check logs for details.'); + this.reportPromptFailure(error); } finally { session.onThinkingChange(false); await this.permissionHandler?.cancelAll('Prompt finished'); @@ -614,6 +646,10 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { } protected async cleanup(): Promise { + // Before anything that can fail: a stream left open would keep + // reconnecting to a subprocess that is on its way out. + this.eventStream?.close(); + this.eventStream = null; this.clearAbortHandlers(this.session.client.rpcHandlerManager); const failures: unknown[] = []; if (this.permissionHandler) { @@ -672,6 +708,77 @@ class OpencodeRemoteLauncher extends RemoteLauncherBase { } } + /** + * Reports an upstream retry as progress, not as a failure: the agent is + * still working and will try again on its own. + * + * Sent as the `api_error` system message Claude sessions already use for + * the same situation, rather than as a new message type, so the web + * timeline folds consecutive retries into a single block instead of + * stacking one warning per attempt (`foldApiErrorEvents` in + * web/src/chat/reducerEvents.ts). `maxRetries` is 0 because OpenCode + * announces no ceiling — it retries until the provider lets it through + * (its own issue tracker has the missing circuit breaker open) — and the + * presentation already has a branch for exactly that. The provider's own + * text rides in `error` so the reader learns *why* the session is + * waiting, which is the entire point; without it the timeline says only + * "Retrying...". + * + * `thinking` is deliberately untouched. During a retry the session + * really is busy, and the hub composes the flag it shows from what this + * session reports plus its own queue grace + * (`hub/src/sync/sessionCache.ts`) — this side reports what is true and + * does not reach across that seam. + */ + private surfaceUpstreamRetry(retry: OpencodeRetryStatus): void { + const text = formatOpencodeRetryStatus(retry); + this.session.client.sendClaudeSessionMessage({ + type: 'system', + uuid: randomUUID(), + subtype: 'api_error', + retryAttempt: retry.attempt, + maxRetries: 0, + error: { message: text } + }); + this.messageBuffer.addMessage(text, 'status'); + } + + /** + * Reports a rejected `session/prompt` to the user in the provider's own + * words. + * + * Nothing had to be detected to make this possible: `AcpStdioTransport` + * already rejects the pending request with the JSON-RPC `error.message` + * verbatim, and this catch used to log it and hand the user a fixed + * "check logs for details" line instead. A remote user is by definition + * not at the machine holding those logs. + * + * This reports unconditionally, and that is a decision rather than an + * omission. On a hard error OpenCode also dumps the same JSON-RPC error + * onto stderr, which the shared ACP reader reports a line at a time, so + * the user sees three raw fragments and then this one sentence. An + * earlier revision tried to suppress the sentence as a duplicate; do + * not add that back: + * + * - The fragments are a JSON dump split at newlines. This sentence is + * the only line of the four a person can read. It follows them as a + * summary, which is not the same thing as a repeat. + * - Suppressing on containment deleted it in every measured hard error, + * because a fragment embeds it as a substring; suppressing on exact + * equivalence never fired at all, because the fragment and this + * sentence are differently worded. There was no observed case where + * the check both fired and was right. + * - Doing it at all needs a per-turn ledger of what has been shown, and + * the stderr path has no turn boundary of its own to reset it on: a + * /compact batch never reaches this loop's per-prompt reset, so the + * ledger leaked across batches and silently muted the stderr channel + * for the rest of such a session. That channel discarded nothing + * before this branch existed and should keep discarding nothing. + */ + private reportPromptFailure(error: unknown): void { + this.surfaceAgentError(formatOpencodePromptError(error)); + } + private surfaceAgentError(message: string): void { this.session.sendAgentMessage({ type: 'error', message }); this.messageBuffer.addMessage(message, 'status'); diff --git a/cli/src/opencode/utils/opencodeErrorText.test.ts b/cli/src/opencode/utils/opencodeErrorText.test.ts new file mode 100644 index 00000000..4ef5388c --- /dev/null +++ b/cli/src/opencode/utils/opencodeErrorText.test.ts @@ -0,0 +1,110 @@ +import { describe, expect, it } from 'vitest'; +import { + OPENCODE_PROVIDER_TEXT_MAX_LENGTH, + formatOpencodePromptError, + truncateOpencodeProviderText +} from './opencodeErrorText'; + +/** The wording this path rendered before the reason was wired through. */ +const FALLBACK = 'OpenCode prompt failed. Check logs for details.'; + +describe('truncateOpencodeProviderText', () => { + it('leaves text that already fits untouched', () => { + const text = 'Rate limit exceeded: free-models-per-day.'; + expect(truncateOpencodeProviderText(text)).toBe(text); + }); + + it('caps text no provider bounds, marking that it was cut', () => { + // Measured: a provider answering a rate limit with a 20KB body + // produces a 20,210-character message, forwarded verbatim. + const capped = truncateOpencodeProviderText('x'.repeat(20_210)); + + expect(capped).toHaveLength(OPENCODE_PROVIDER_TEXT_MAX_LENGTH); + expect(capped.endsWith('…')).toBe(true); + }); +}); + +describe('formatOpencodePromptError', () => { + it('states the provider reason the ACP transport already delivered', () => { + const error = new Error('Internal error: Rate limit exceeded: free-models-per-day. Add credits to unlock 1000 free models.'); + + expect(formatOpencodePromptError(error)).toBe( + 'OpenCode prompt failed: Rate limit exceeded: free-models-per-day. Add credits to unlock 1000 free models.' + ); + }); + + it('does not relay a rejection that arrives without the JSON-RPC wrapper', () => { + // The wrapper is the only evidence the text came back over the wire + // from OpenCode rather than being built locally by the transport. + expect(formatOpencodePromptError(new Error('Session not found'))).toBe(FALLBACK); + }); + + it('does not mistake a message that merely mentions an internal error for the wrapper', () => { + // Anchored: only a message that starts with it is agent-authored. + expect(formatOpencodePromptError(new Error('The provider reported an Internal error: upstream'))) + .toBe(FALLBACK); + }); + + it('reads a rejection thrown as a bare string', () => { + expect(formatOpencodePromptError('Internal error: Overloaded')) + .toBe('OpenCode prompt failed: Overloaded'); + }); + + it('reads a rejection thrown as a plain JSON-RPC error object', () => { + expect(formatOpencodePromptError({ + code: -32603, + message: 'Internal error: Overloaded', + data: { service: 'session', errorName: 'APIError' } + })).toBe('OpenCode prompt failed: Overloaded'); + }); + + it('collapses embedded newlines so one failure stays one timeline line', () => { + expect(formatOpencodePromptError(new Error('Internal error: Rate limit exceeded.\n\n Try again later.'))) + .toBe('OpenCode prompt failed: Rate limit exceeded. Try again later.'); + }); + + it('caps a message no provider bounds', () => { + const formatted = formatOpencodePromptError(new Error(`Internal error: ${'y'.repeat(20_210)}`)); + + expect(formatted).toBe(`OpenCode prompt failed: ${'y'.repeat(OPENCODE_PROVIDER_TEXT_MAX_LENGTH - 1)}…`); + }); + + it('keeps the original wording when there is no message', () => { + expect(formatOpencodePromptError(new Error(''))).toBe(FALLBACK); + expect(formatOpencodePromptError(new Error(' '))).toBe(FALLBACK); + }); + + it('keeps the original wording when the wrapper is all there was', () => { + // A bare "Internal error:" says nothing the fallback does not. + expect(formatOpencodePromptError(new Error('Internal error: '))).toBe(FALLBACK); + }); + + + it('does not relay a rejection OpenCode did not author', () => { + // AcpStdioTransport builds this one itself on process close and + // appends up to 4KB of raw subprocess stderr to the message. Relaying + // it would publish local stderr, credentials included, to the remote + // timeline. + const closeError = new Error( + 'ACP process exited (code=1, signal=null). stderr: AWS_SECRET_ACCESS_KEY=wJalrXUt shutting down' + ); + + const formatted = formatOpencodePromptError(closeError); + + expect(formatted).toBe(FALLBACK); + expect(formatted).not.toContain('AWS_SECRET_ACCESS_KEY'); + expect(formatted).not.toContain('stderr'); + }); + + it('does not relay the transport\'s own timeout', () => { + expect(formatOpencodePromptError(new Error("ACP request 'session/prompt' timed out after 1000ms"))) + .toBe(FALLBACK); + }); + + it('keeps the original wording for a rejection that is not an error at all', () => { + expect(formatOpencodePromptError(null)).toBe(FALLBACK); + expect(formatOpencodePromptError(undefined)).toBe(FALLBACK); + expect(formatOpencodePromptError(42)).toBe(FALLBACK); + expect(formatOpencodePromptError({ code: -32603 })).toBe(FALLBACK); + }); +}); diff --git a/cli/src/opencode/utils/opencodeErrorText.ts b/cli/src/opencode/utils/opencodeErrorText.ts new file mode 100644 index 00000000..4317ab73 --- /dev/null +++ b/cli/src/opencode/utils/opencodeErrorText.ts @@ -0,0 +1,131 @@ +import { asString, isObject } from '@hapi/protocol'; + +/** + * Longest run of provider-authored text this session puts in front of a user. + * + * OpenCode does not bound what it forwards: a provider that answers a rate + * limit with a 20KB body produces a `session/prompt` error message of 20,210 + * characters verbatim (measured against opencode 1.18.15). That is not an + * error report, it is a wall — it buries the one sentence that matters, and + * every copy of it is persisted in the hub's message store and replayed to + * every client that opens the session. + * + * 200 is not a fresh guess: it is the cap `safeErrorReason()` in + * opencodeCompactBridge.ts already applies to the same kind of value (the + * provider message OpenCode persists on a failed compaction), so the two + * places this session shows provider text agree. Real messages are far + * shorter — the rate-limit report this feature was built for is 79 + * characters. + */ +export const OPENCODE_PROVIDER_TEXT_MAX_LENGTH = 200; + +/** Shown when the failure carries nothing readable; preserved verbatim from before this text was wired through, so that path is a strict no-op. */ +const OPENCODE_PROMPT_FAILED_FALLBACK = 'OpenCode prompt failed. Check logs for details.'; + +/** + * The JSON-RPC layer's own wrapper, and the marker that a rejection carries + * agent-authored text at all. + * + * OpenCode maps every unhandled service failure to code -32603 and prefixes + * the provider's text with this, so it appears on messages that have nothing + * internal about them ("Internal error: Rate limit exceeded…"). It describes + * the transport, tells the reader nothing, and pushes the actual reason past + * the fold on a phone — so it is stripped before display. + * + * Requiring it is what keeps this an allowlist. `AcpStdioTransport` flattens + * a JSON-RPC error response to `new Error(response.error.message)`, losing + * `code` and `data`, so by the time the launcher catches a rejection it + * cannot tell one apart from an error the transport built itself — and those + * are not agent-authored. The process-close error appends up to 4KB of raw + * subprocess stderr to its message, which must never reach a remote + * timeline; the timeout error names an internal method and duration. Neither + * carries this wrapper. Anything without it falls back to + * {@link OPENCODE_PROMPT_FAILED_FALLBACK}, which is what this path rendered + * before, so an unrecognised shape degrades to the old behaviour rather than + * publishing whatever it happened to contain. + */ +const JSON_RPC_INTERNAL_ERROR_PREFIX = /^internal error:\s*/i; + +/** + * One displayable line: every whitespace run (including newlines a provider + * embedded) collapsed to a single space. + * + * Exported for the same reason {@link OPENCODE_PROVIDER_TEXT_MAX_LENGTH} is: + * the event stream relays provider text to the same surfaces under the same + * one-line contract, and two copies of this rule would drift. + */ +export function collapseWhitespace(text: string): string { + return text.replace(/\s+/g, ' ').trim(); +} + +/** + * Caps provider-authored text at {@link OPENCODE_PROVIDER_TEXT_MAX_LENGTH}. + * + * The ellipsis is load-bearing rather than decoration: a rate-limit message + * cut mid-sentence without one reads as a complete (and different) statement + * from the provider. + */ +export function truncateOpencodeProviderText(text: string): string { + if (text.length <= OPENCODE_PROVIDER_TEXT_MAX_LENGTH) { + return text; + } + return `${text.slice(0, OPENCODE_PROVIDER_TEXT_MAX_LENGTH - 1).trimEnd()}…`; +} + +/** + * The agent's own words behind a failed `session/prompt`, as one line, with + * the JSON-RPC wrapper stripped — or null when the rejection is not one + * OpenCode authored. + * + * Not truncated here; {@link formatOpencodePromptError} applies the cap at + * the point of display. + * + * Only `message` is read, and only when it carries + * {@link JSON_RPC_INTERNAL_ERROR_PREFIX}. That pairing is the whole safety + * property: the wrapper says the text came back over the wire from OpenCode, + * and that channel is sanitised — probed with a stub provider answering + * `Authorization: Bearer …`, `Set-Cookie: …` and a 4KB body, the ACP error + * carried none of them, only the provider's message text. (The same failure + * also appears on OpenCode's `/event` stream as `session.error`, where all of + * it *is* present verbatim — which is exactly why the event subscription does + * not read that event.) + */ +function extractOpencodePromptErrorText(error: unknown): string | null { + const raw = error instanceof Error + ? error.message + : typeof error === 'string' + ? error + : isObject(error) + ? asString(error.message) + : null; + if (raw === null) return null; + const collapsed = collapseWhitespace(raw); + if (!JSON_RPC_INTERNAL_ERROR_PREFIX.test(collapsed)) return null; + const text = collapsed.replace(JSON_RPC_INTERNAL_ERROR_PREFIX, '').trim(); + return text || null; +} + +/** Prefix {@link formatOpencodePromptError} puts in front of the provider's text. */ +const OPENCODE_PROMPT_FAILED_PREFIX = 'OpenCode prompt failed: '; + +/** + * Renders the failure of an OpenCode `session/prompt` for the user. + * + * The reason was never missing — `AcpStdioTransport` rejects the pending + * request with the JSON-RPC `error.message` verbatim, which is where the + * provider's own explanation lives. The launcher used to catch that, log it, + * and hand the user a fixed "check logs for details" string instead. A + * remote user is by definition not at the machine holding those logs, so + * that read as "it stopped, and you may not know why". + * + * The old wording is kept as the prefix rather than replaced, so an operator + * who has learned to grep for "OpenCode prompt failed" still finds it, and + * a failure with nothing readable to say renders exactly the sentence it + * always did. + */ +export function formatOpencodePromptError(error: unknown): string { + const message = extractOpencodePromptErrorText(error); + return message + ? `${OPENCODE_PROMPT_FAILED_PREFIX}${truncateOpencodeProviderText(message)}` + : OPENCODE_PROMPT_FAILED_FALLBACK; +} diff --git a/cli/src/opencode/utils/opencodeEventStream.test.ts b/cli/src/opencode/utils/opencodeEventStream.test.ts new file mode 100644 index 00000000..f591c002 --- /dev/null +++ b/cli/src/opencode/utils/opencodeEventStream.test.ts @@ -0,0 +1,758 @@ +import { describe, expect, it, vi } from 'vitest'; +import { logger } from '@/ui/logger'; +import { OPENCODE_PROVIDER_TEXT_MAX_LENGTH, truncateOpencodeProviderText } from './opencodeErrorText'; +import { + formatOpencodeRetryStatus, + subscribeToOpencodeEvents +} from './opencodeEventStream'; + +vi.mock('@/ui/logger', () => ({ + logger: { debug: vi.fn(), warn: vi.fn(), info: vi.fn() } +})); + +const NOW = 1_786_154_400_000; +const SESSION_CWD = '/home/user/projects/thing'; +const EXPECTED_URL = `http://127.0.0.1:48273/event?directory=${encodeURIComponent(SESSION_CWD)}`; + +function createSseStream() { + let controller: ReadableStreamDefaultController | null = null; + const stream = new ReadableStream({ + start(c) { + controller = c; + } + }); + const encoder = new TextEncoder(); + // Writes are tolerated after the stream is gone: closing the + // subscription cancels its reader, which closes this controller. A + // caller writing into that is exactly the "frame arrives after close" + // case, and it must not blow up the test. + const write = (text: string) => { + try { + controller?.enqueue(encoder.encode(text)); + } catch { + // Stream already closed. + } + }; + return { + stream, + /** Writes one SSE frame, the framing opencode's /event endpoint uses. */ + push(event: unknown) { + write(`data: ${JSON.stringify(event)}\n\n`); + }, + pushRaw(text: string) { + write(text); + }, + close() { + try { + controller?.close(); + } catch { + // Already closed — cancelling the reader closes it too. + } + } + }; +} + +function retryEvent(attempt: number, sessionId = 'ses_ours') { + return { + id: `evt_${attempt}`, + type: 'session.status', + properties: { + sessionID: sessionId, + status: { + type: 'retry', + attempt, + message: 'Rate limit exceeded: free-models-per-day.', + next: NOW + 30_000 + } + } + }; +} + +function statusEvent(type: 'busy' | 'idle', sessionId = 'ses_ours') { + return { + id: `evt_${type}`, + type: 'session.status', + properties: { sessionID: sessionId, status: { type } } + }; +} + +function subscribe(options: { + fetchImpl: (url: string, init?: RequestInit) => Promise; + onRetry?: (retry: { attempt: number; message: string }) => void; + /** Records the backoff schedule instead of waiting it out. */ + delays?: number[]; + /** + * Closes the subscription once this many delays are recorded. Reconnects + * are unbounded by design, so a test that stubs the wait to nothing must + * stop the loop itself. + */ + stopAfterDelays?: number; +}) { + let subscription: { close: () => void } | null = null; + subscription = subscribeToOpencodeEvents({ + baseUrl: 'http://127.0.0.1:48273', + directory: SESSION_CWD, + sessionId: 'ses_ours', + onRetry: options.onRetry ?? (() => {}), + fetchImpl: options.fetchImpl, + sleep: async (ms: number) => { + options.delays?.push(ms); + if (options.stopAfterDelays && (options.delays?.length ?? 0) >= options.stopAfterDelays) { + subscription?.close(); + } + // Yield a real macrotask so assertions can run between attempts. + await new Promise((resolve) => setTimeout(resolve, 0)); + } + }); + return subscription; +} + +describe('formatOpencodeRetryStatus', () => { + it('states the provider message and the attempt number', () => { + const text = formatOpencodeRetryStatus({ + attempt: 2, + message: 'Rate limit exceeded: free-models-per-day.' + }); + + expect(text).toBe('Rate limit exceeded: free-models-per-day. (attempt 2)'); + }); + + it('does not render a countdown, which would outlive the turn it describes', () => { + // This becomes a timeline block. Anything time-relative here would + // still be claiming a retry is due while sitting above a reply that + // arrived minutes ago. + const text = formatOpencodeRetryStatus({ + attempt: 4, + message: 'Overloaded.' + }); + + expect(text).toBe('Overloaded. (attempt 4)'); + expect(text).not.toMatch(/retry|retrying in|\ds\b|shortly/i); + }); + + it('stays on one line and within the provider-text budget', () => { + // It renders into a plain span with no `whitespace-pre-line`, and + // the cap this session promises for provider text is a per-message + // promise, not a per-field one. + const text = formatOpencodeRetryStatus({ + attempt: 7, + message: truncateOpencodeProviderText('m'.repeat(20_210)) + }); + + expect(text).not.toContain('\n'); + expect(text.length).toBeLessThanOrEqual(OPENCODE_PROVIDER_TEXT_MAX_LENGTH + ' (attempt 7)'.length); + }); +}); + + +describe('subscribeToOpencodeEvents', () => { + it('scopes the subscription to the session directory, without which it would receive nothing', async () => { + // Measured: the unscoped stream delivers heartbeats and nothing + // else — no error, no 404, just a subscription that never reports. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const subscription = subscribe({ fetchImpl }); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + expect(fetchImpl.mock.calls[0][0]).toBe(EXPECTED_URL); + + subscription.close(); + sse.close(); + }); + + it('reports every new retry attempt once, ignoring repeats of the same attempt', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push(retryEvent(1)); + sse.push(retryEvent(1)); + sse.push(retryEvent(2)); + sse.push(retryEvent(3)); + + await vi.waitFor(() => expect(retries).toHaveLength(3)); + expect(retries).toEqual([1, 2, 3]); + + subscription.close(); + sse.close(); + }); + + it('reports a request that restarts its attempt count inside the same turn', async () => { + // A turn is not one provider request: every tool-call round trip + // starts a fresh completion whose attempt counter begins at 1 + // again, with no idle/busy in between. Suppressing anything at or + // below the highest attempt seen would freeze the timeline on the + // first request's last number while the agent kept retrying — the + // exact silence this feature removes. A falling attempt number is + // news, not a duplicate. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push(retryEvent(1)); + sse.push(retryEvent(2)); + sse.push(retryEvent(1)); + + await vi.waitFor(() => expect(retries).toEqual([1, 2, 1])); + + subscription.close(); + sse.close(); + }); + + it('thins a long run of retries instead of writing one row a minute forever', async () => { + // Each report is persisted by the hub and rebroadcast to every + // client; folding happens only when one of them renders. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + for (let attempt = 1; attempt <= 20; attempt += 1) { + sse.push(retryEvent(attempt)); + } + + await vi.waitFor(() => expect(retries).toEqual([1, 2, 3, 4, 8, 16])); + + subscription.close(); + sse.close(); + }); + + it('drops an attempt number that is not a counting number', async () => { + // Infinity comes from JSON.parse on an overflowing literal; the + // rest satisfy the dense-reporting branch and would be rendered + // verbatim, and 0 collides with the turn-boundary sentinel. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.pushRaw( + 'data: {"type":"session.status","properties":{"sessionID":"ses_ours",' + + '"status":{"type":"retry","attempt":1e999,"message":"Rate limit exceeded."}}}\n\n' + ); + for (const attempt of [0, -1, 0.5, 2.5]) { + sse.push({ + id: `evt_${attempt}`, + type: 'session.status', + properties: { + sessionID: 'ses_ours', + status: { type: 'retry', attempt, message: 'Rate limit exceeded.', next: NOW } + } + }); + } + sse.push(retryEvent(1)); + + await vi.waitFor(() => expect(retries).toEqual([1])); + + subscription.close(); + sse.close(); + }); + + + it('drops a message that only looks like text, and flattens one that is not a single line', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const messages: string[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => messages.push(retry.message) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + // Whitespace only: truthy, but nothing a user could read. + sse.push({ + id: 'evt_blank', + type: 'session.status', + properties: { sessionID: 'ses_ours', status: { type: 'retry', attempt: 1, message: ' \n ', next: NOW } } + }); + sse.push({ + id: 'evt_multiline', + type: 'session.status', + properties: { + sessionID: 'ses_ours', + status: { type: 'retry', attempt: 2, message: 'Rate limit exceeded.\n\n Try again later.', next: NOW } + } + }); + + await vi.waitFor(() => expect(messages).toEqual(['Rate limit exceeded. Try again later.'])); + + subscription.close(); + sse.close(); + }); + + it('keeps reading when a consumer throws, and does not file that as a parse failure', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ + fetchImpl, + onRetry: (retry) => { + retries.push(retry.attempt); + if (retry.attempt === 1) throw new Error('hub client exploded'); + } + }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push(retryEvent(1)); + sse.push(retryEvent(2)); + + await vi.waitFor(() => expect(retries).toEqual([1, 2])); + + subscription.close(); + sse.close(); + }); + + it('stops dispatching the rest of a chunk once closed mid-frame', async () => { + // One read can carry several frames, and close() lands synchronously + // from the launcher's cleanup — so re-checking only once per chunk + // would drain the remainder into a session that is already gone. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + let subscription: { close: () => void } | null = null; + subscription = subscribeToOpencodeEvents({ + baseUrl: 'http://127.0.0.1:48273', + directory: SESSION_CWD, + sessionId: 'ses_ours', + fetchImpl, + sleep: async () => {}, + onRetry: (retry) => { + retries.push(retry.attempt); + subscription?.close(); + } + }); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + // Two frames arriving as one chunk; handling the first closes it. + sse.pushRaw(`data: ${JSON.stringify(retryEvent(1))}\n\ndata: ${JSON.stringify(retryEvent(2))}\n\n`); + + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(retries).toEqual([1]); + + sse.close(); + }); + + it('warns once when no connection attempt ever produces a usable stream', async () => { + // Every other failure here is debug-level, but a subscription that + // never works is indistinguishable from a session that simply never + // retried — and silence is the bug this exists to remove. + vi.mocked(logger.warn).mockClear(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit): Promise => { + throw new Error('connection refused'); + }); + const delays: number[] = []; + const subscription = subscribe({ fetchImpl, delays, stopAfterDelays: 9 }); + + await vi.waitFor(() => expect(delays.length).toBeGreaterThanOrEqual(8)); + expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(1); + expect(vi.mocked(logger.warn).mock.calls[0][0]).toContain('will not be reported'); + + subscription.close(); + }); + + it('warns again about a later outage after a connection that worked', async () => { + // "Once" means once per outage. A session that stumbles at startup + // and then loses the subprocess hours later must still report the + // second failure, which is the one that actually costs the user + // their retry reports. + vi.mocked(logger.warn).mockClear(); + const healthy = createSseStream(); + let call = 0; + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit): Promise => { + call += 1; + if (call === 6) return new Response(healthy.stream, { status: 200 }); + throw new Error('connection refused'); + }); + const delays: number[] = []; + const subscription = subscribe({ fetchImpl, delays, stopAfterDelays: 30 }); + + await vi.waitFor(() => expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(1)); + + // The sixth attempt delivers a frame and then ends, which is what + // clears both the counter and the warning latch. + healthy.push(retryEvent(1)); + healthy.close(); + + await vi.waitFor(() => expect(vi.mocked(logger.warn)).toHaveBeenCalledTimes(2)); + + subscription.close(); + }); + + it('treats the next turn as new information after the session goes idle and busy again', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push(retryEvent(1)); + await vi.waitFor(() => expect(retries).toHaveLength(1)); + + // Turn boundary. Measured: both are published twice within the same + // millisecond, so the reset must be idempotent. + sse.push(statusEvent('idle')); + sse.push(statusEvent('idle')); + sse.push(statusEvent('busy')); + sse.push(statusEvent('busy')); + sse.push(retryEvent(1)); + + await vi.waitFor(() => expect(retries).toHaveLength(2)); + expect(retries).toEqual([1, 1]); + + subscription.close(); + sse.close(); + }); + + it('ignores status events belonging to another session', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push(retryEvent(1, 'ses_someone_else')); + // Nor may another session's turn boundary clear our dedupe state. + sse.push(statusEvent('idle', 'ses_someone_else')); + // A frame for our own session proves the stream was being read at all. + sse.push(retryEvent(2)); + sse.push(retryEvent(2)); + + await vi.waitFor(() => expect(retries).toHaveLength(1)); + expect(retries).toEqual([2]); + + subscription.close(); + sse.close(); + }); + + it('ignores a status event that cannot be attributed to any session', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push({ + id: 'evt_orphan', + type: 'session.status', + properties: { status: { type: 'retry', attempt: 9, message: 'nope', next: NOW + 1_000 } } + }); + sse.push(retryEvent(1)); + + await vi.waitFor(() => expect(retries).toEqual([1])); + + subscription.close(); + sse.close(); + }); + + it('never reads session.error, which carries the provider credentials verbatim', async () => { + // Measured against a stub that answered with Authorization, + // Set-Cookie and a 4KB body: this payload reproduces all of it. The + // same failure reaches the user through the ACP prompt error, which + // carries only the message. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: unknown[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push({ + id: 'evt_err', + type: 'session.error', + properties: { + sessionID: 'ses_ours', + error: { + name: 'APIError', + data: { + message: 'Unauthorized', + responseHeaders: { + authorization: 'Bearer sk-live-do-not-relay', + 'set-cookie': 'session=do-not-relay' + }, + responseBody: 'do-not-relay' + } + } + } + }); + // Reaching a later frame proves the stream kept being read. + sse.push(retryEvent(1)); + + await vi.waitFor(() => expect(retries).toHaveLength(1)); + expect(JSON.stringify(retries)).not.toContain('do-not-relay'); + + subscription.close(); + sse.close(); + }); + + it('caps a retry message no provider bounds', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const messages: string[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => messages.push(retry.message) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push({ + id: 'evt_long', + type: 'session.status', + properties: { + sessionID: 'ses_ours', + status: { type: 'retry', attempt: 1, message: 'w'.repeat(20_210), next: NOW } + } + }); + + await vi.waitFor(() => expect(messages).toHaveLength(1)); + expect(messages[0]).toHaveLength(OPENCODE_PROVIDER_TEXT_MAX_LENGTH); + expect(messages[0].endsWith('…')).toBe(true); + + subscription.close(); + sse.close(); + }); + + it('still reports a retry whose next-attempt timestamp is missing or malformed', async () => { + // `next` is not rendered, so it must not gate anything. Dropping a + // retry over a field nobody reads would hide exactly the report this + // feature exists to deliver. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + const retryWithoutNext = (attempt: number, next?: unknown) => ({ + id: `evt_${attempt}`, + type: 'session.status', + properties: { + sessionID: 'ses_ours', + status: { type: 'retry', attempt, message: 'Rate limit exceeded.', ...(next === undefined ? {} : { next }) } + } + }); + + sse.push(retryWithoutNext(1)); + sse.push(retryWithoutNext(2, 'not-a-timestamp')); + sse.push(retryWithoutNext(3, null)); + + await vi.waitFor(() => expect(retries).toEqual([1, 2, 3])); + + subscription.close(); + sse.close(); + }); + + it('does not let an unreadable retry reset the duplicate suppression', async () => { + // A retry whose message is empty cannot be rendered, but it is still + // a retry — not a turn boundary. Treating it as one would clear the + // dedupe state and let the agent's next repeat of the same attempt + // through as if it were new. + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push(retryEvent(1)); + await vi.waitFor(() => expect(retries).toEqual([1])); + + sse.push({ + id: 'evt_blank', + type: 'session.status', + properties: { + sessionID: 'ses_ours', + status: { type: 'retry', attempt: 1, message: '', next: NOW + 30_000 } + } + }); + // OpenCode repeats the same attempt while a backoff is pending. + sse.push(retryEvent(1)); + sse.push(retryEvent(2)); + + await vi.waitFor(() => expect(retries).toEqual([1, 2])); + + subscription.close(); + sse.close(); + }); + + it('ignores heartbeats and unparsable frames without dropping the stream', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + sse.push({ id: 'evt_hb', type: 'server.heartbeat', properties: {} }); + sse.pushRaw('data: {not json\n\n'); + sse.pushRaw(': a comment line\n\n'); + sse.push(retryEvent(1)); + + await vi.waitFor(() => expect(retries).toHaveLength(1)); + + subscription.close(); + sse.close(); + }); + + it('resubscribes when the stream ends', async () => { + const streams = [createSseStream(), createSseStream()]; + let call = 0; + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(streams[call++]?.stream ?? null, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + streams[0].close(); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(2)); + streams[1].push(retryEvent(1)); + await vi.waitFor(() => expect(retries).toHaveLength(1)); + + subscription.close(); + streams[1].close(); + }); + + it('stops for good on a build whose HTTP API has no /event endpoint', async () => { + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(null, { status: 404 })); + const subscription = subscribe({ fetchImpl }); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + // A 404 is a permanent answer: retrying it would just fill the log. + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(fetchImpl).toHaveBeenCalledTimes(1); + + subscription.close(); + }); + + it('backs off instead of spinning when the server accepts and closes the body immediately', async () => { + // A 200 whose body ends without a single frame is not a working + // stream, however clean the end looks. Counting it as success is + // what would produce a silent twice-a-second reconnect loop. + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => { + const sse = createSseStream(); + sse.close(); + return new Response(sse.stream, { status: 200 }); + }); + const delays: number[] = []; + const subscription = subscribe({ fetchImpl, delays, stopAfterDelays: 6 }); + + await vi.waitFor(() => expect(delays.length).toBeGreaterThanOrEqual(4)); + expect(delays.slice(0, 4)).toEqual([500, 1_000, 2_000, 4_000]); + + subscription.close(); + }); + + it('reconnects promptly after a stream that did deliver events', async () => { + // The counter is about productivity, so a stream that worked and + // then dropped starts again from the base delay rather than + // inheriting an earlier backoff. + const streams = [createSseStream(), createSseStream(), createSseStream()]; + let call = 0; + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => { + const sse = streams[call++]; + if (!sse) return new Response(null, { status: 500 }); + if (call === 1) { + // First attempt: no frames at all, so the next wait backs off. + sse.close(); + } + return new Response(sse.stream, { status: 200 }); + }); + const delays: number[] = []; + const subscription = subscribe({ fetchImpl, delays, stopAfterDelays: 4 }); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(2)); + expect(delays).toEqual([500]); + + // Second attempt delivers a frame, then drops. + const retries: number[] = []; + streams[1].push(retryEvent(1)); + streams[1].close(); + + await vi.waitFor(() => expect(delays.length).toBeGreaterThanOrEqual(2)); + // Back to the base delay rather than continuing on to 1000ms. + expect(delays[1]).toBe(500); + expect(retries).toEqual([]); + + subscription.close(); + streams[2].close(); + }); + + it('keeps retrying a refused connection rather than giving up on the session', async () => { + // Capped, never abandoned: a session can outlive a transient outage + // by hours, and this is the only channel carrying retry reports. + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit): Promise => { + throw new Error('connection refused'); + }); + const delays: number[] = []; + const subscription = subscribe({ fetchImpl, delays, stopAfterDelays: 10 }); + + await vi.waitFor(() => expect(delays.length).toBeGreaterThanOrEqual(8)); + expect(delays.slice(0, 3)).toEqual([500, 1_000, 2_000]); + expect(Math.max(...delays)).toBeLessThanOrEqual(30_000); + + subscription.close(); + }); + + it('hands back the response body on the paths that will never read it', async () => { + const cancel = vi.fn(async () => {}); + const makeBody = () => ({ cancel, getReader: () => { throw new Error('never read'); } }); + let call = 0; + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => { + call += 1; + return { + status: call === 1 ? 503 : 404, + ok: false, + body: makeBody() + } as unknown as Response; + }); + const delays: number[] = []; + const subscription = subscribe({ fetchImpl, delays, stopAfterDelays: 5 }); + + // 503 -> body released, backs off, retries. 404 -> body released and + // the subscription stops for good. + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(2)); + await vi.waitFor(() => expect(cancel).toHaveBeenCalledTimes(2)); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(fetchImpl).toHaveBeenCalledTimes(2); + + subscription.close(); + }); + + it('replays the last event id when reconnecting', async () => { + const streams = [createSseStream(), createSseStream()]; + let call = 0; + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => { + const sse = streams[call++]; + return new Response(sse?.stream ?? null, { status: 200 }); + }); + const subscription = subscribe({ fetchImpl }); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + expect(fetchImpl.mock.calls[0][1]?.headers).not.toHaveProperty('last-event-id'); + + streams[0].pushRaw(`id: evt_42\ndata: ${JSON.stringify(retryEvent(1))}\n\n`); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + streams[0].close(); + + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(2)); + expect(fetchImpl.mock.calls[1][1]?.headers).toMatchObject({ 'last-event-id': 'evt_42' }); + + subscription.close(); + streams[1].close(); + }); + + it('stops reading once closed', async () => { + const sse = createSseStream(); + const fetchImpl = vi.fn(async (_url: string, _init?: RequestInit) => new Response(sse.stream, { status: 200 })); + const retries: number[] = []; + const subscription = subscribe({ fetchImpl, onRetry: (retry) => retries.push(retry.attempt) }); + await vi.waitFor(() => expect(fetchImpl).toHaveBeenCalledTimes(1)); + + subscription.close(); + subscription.close(); + sse.push(retryEvent(1)); + sse.close(); + + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(retries).toEqual([]); + // Closing must not trigger another subscription attempt. + expect(fetchImpl).toHaveBeenCalledTimes(1); + }); +}); diff --git a/cli/src/opencode/utils/opencodeEventStream.ts b/cli/src/opencode/utils/opencodeEventStream.ts new file mode 100644 index 00000000..751f79bb --- /dev/null +++ b/cli/src/opencode/utils/opencodeEventStream.ts @@ -0,0 +1,504 @@ +import { asNumber, asString, isObject } from '@hapi/protocol'; +import { logger } from '@/ui/logger'; +import type { FetchLike } from './opencodeCompactBridge'; +import { collapseWhitespace, truncateOpencodeProviderText } from './opencodeErrorText'; + +/** + * `session.status` with `type: "retry"` — OpenCode announcing that it hit an + * upstream failure and will try again. + * + * The payload also carries `next` (the absolute timestamp of the following + * attempt) and an optional `action` (a title/message/label/link hint). + * Neither is modelled. `next` would only invite a countdown, which must not + * be rendered — see formatOpencodeRetryStatus. `action` has never been + * observed on a live retry, and rendering it meant concatenating four more + * unbounded provider strings into one timeline line: past the length this + * session promises for provider text, with a `link` that truncation turns + * from a long URL into a wrong one presented as if it were clickable. + */ +export type OpencodeRetryStatus = { + attempt: number; + message: string; +}; + +export type OpencodeEventStreamOptions = { + baseUrl: string; + /** + * The session's working directory. Required, not optional: the endpoint + * scopes its events by it, and without it the stream is silently useless + * (see subscribeToOpencodeEvents). + */ + directory: string; + /** Only events for this ACP session id (`ses_…`) are reported. */ + sessionId: string; + /** + * A retry announcement. Structured rather than pre-rendered because the + * launcher reports retries as progress (an `api_error` system message the + * web timeline folds), not as a failure — see its call site. + */ + onRetry: (retry: OpencodeRetryStatus) => void; + fetchImpl?: FetchLike; + sleep?: (ms: number) => Promise; +}; + +export type OpencodeEventSubscription = { + /** Idempotent. Aborts the in-flight request and stops reconnecting. */ + close: () => void; +}; + +const RECONNECT_BASE_DELAY_MS = 500; +const RECONNECT_MAX_DELAY_MS = 30_000; + +/** Every attempt up to this number is reported; past it, only powers of two are. */ +const RETRY_REPORT_DENSE_THRESHOLD = 3; + +/** + * How many connection attempts may produce nothing before saying so out + * loud once. Below this the endpoint is merely being restarted or is not up + * yet, which is routine and belongs at debug level. + */ +const UNPRODUCTIVE_ATTEMPTS_BEFORE_WARNING = 5; + +/** + * Whether an attempt number is worth a row in the user's timeline. + * + * Each report becomes a message the hub persists to SQLite and rebroadcasts + * over SSE to every connected client, and it is kept in session exports. + * `foldApiErrorEvents` collapses a run of them when a client renders, but + * that is a view concern — the rows, the traffic and the export are not + * folded. OpenCode never gives up and its backoff tops out at 30 seconds, + * so an unfiltered subscription writes about two rows a minute for as long + * as the provider keeps refusing: the measured 40-minute stall alone would + * have been 85 of them, and a session left overnight is four figures. + * + * Sampling on the attempt number rather than on elapsed time keeps this a + * pure function of what the agent itself published — so it is decidable the + * moment an event arrives, needs no clock, and is testable without fake + * timers. A time-based sampler would also keep emitting rows through a long + * backoff during which nothing new has happened. + * + * Early attempts are what a watching user actually reads, so the first few + * come through unfiltered and the rest grow logarithmically: 1, 2, 3, 4, 8, + * 16, 32 … Progress stays legible without the row count tracking wall time. + */ +function isReportableRetryAttempt(attempt: number): boolean { + if (attempt <= RETRY_REPORT_DENSE_THRESHOLD) { + return true; + } + // log2 rather than a bitwise trick: attempt is provider-supplied and + // `x & (x - 1)` silently truncates anything past 32 bits. + return Number.isInteger(Math.log2(attempt)); +} + +/** + * Delay before the next connection attempt, keyed on how many attempts in a + * row produced nothing at all. + * + * A connection only counts as productive if it actually delivered an SSE + * frame — not merely if the server answered with headers. A server that + * accepts the request and closes the body immediately would otherwise look + * like a clean end of stream forever, reconnecting twice a second without + * even logging it. + * + * There is no attempt ceiling on purpose. This is the only channel carrying + * upstream retry reports, a reconnect costs a loopback request, and a + * session can outlive a transient outage by hours — giving up for good + * would silently leave the user back where this feature started. (Only a + * 404 stops it, since that answer will not change.) + */ +function reconnectDelayMs(unproductiveAttempts: number): number { + if (unproductiveAttempts <= 0) { + return RECONNECT_BASE_DELAY_MS; + } + return Math.min( + RECONNECT_BASE_DELAY_MS * 2 ** (unproductiveAttempts - 1), + RECONNECT_MAX_DELAY_MS + ); +} + +/** + * Renders a retry announcement: what went wrong and which attempt this is. + * + * Deliberately no countdown, though the payload carries the next attempt's + * timestamp. This becomes a timeline block that outlives the turn, so + * "retrying in 30s" would still be sitting above a long-since successful + * reply, asserting something that stopped being true a minute later. + * Progress reads from the attempt number climbing instead, and because + * consecutive retries fold into one block the user sees a single line whose + * count goes up — then simply stops when the agent gets through, with the + * assistant's reply as the recovery signal. + * + * No agent-name prefix either: this renders inside that agent's own session + * timeline, where saying so again is noise. One line, no newlines: this is + * rendered into a plain span with no `whitespace-pre-line`, so a line break + * here would collapse into a space anyway. + */ +export function formatOpencodeRetryStatus(status: OpencodeRetryStatus): string { + return `${status.message} (attempt ${status.attempt})`; +} + +/** + * Reads one `session.status` retry payload. + * + * Every string here is provider-authored and unbounded, exactly like the + * `session/prompt` error message, so it is capped at the parse boundary — + * the one place this text enters — rather than at each place it is + * rendered. See OPENCODE_PROVIDER_TEXT_MAX_LENGTH for the measurement + * behind that. + */ +function parseRetryStatus(status: unknown): OpencodeRetryStatus | null { + if (!isObject(status) || status.type !== 'retry') return null; + // A counting number or nothing. `typeof === 'number'` would let + // JSON.parse's Infinity (from an overflowing literal) through, and + // asNumber alone still admits 0, -1 and 0.5 — all of which satisfy the + // dense-reporting branch below and would be rendered verbatim as + // "(attempt 0.5)". Zero is worse than cosmetic: it is the same value + // the turn-boundary reset uses as its sentinel, so accepting it merges + // "saw a retry" with "this turn has had none". + const attempt = asNumber(status.attempt); + // Collapsed before it is judged, not after: a provider message of " \n " + // is truthy but says nothing, and embedded newlines would survive into a + // status line this module promises to keep to one line. Truncating alone + // does neither. + const message = collapseWhitespace(asString(status.message) ?? ''); + // Only what is actually rendered is required: dropping a retry because a + // field we never show is missing or malformed would hide the very report + // this exists to deliver. + if (attempt === null || !Number.isInteger(attempt) || attempt < 1 || !message) return null; + + return { attempt, message: truncateOpencodeProviderText(message) }; +} + +/** + * Subscribes to an `opencode acp` subprocess's server event stream + * (`GET {baseUrl}/event`, Server-Sent Events) and reports this session's + * upstream retries. + * + * This exists because that is the *only* channel that carries them. + * Measured against opencode 1.18.15 with a provider stubbed to always answer + * 429: over 40 minutes and 85 retries the ACP channel produced zero + * `session/update` notifications and zero stderr output while + * `session/prompt` stayed pending, yet this stream emitted a + * `session.status` retry event — carrying the provider's own message and + * attempt number — throughout. Without it the user sees a session that + * appears to be thinking and never learns that it is rate limited. + * + * **The `directory` query parameter is not optional.** Subscribing to the + * same server with and without it from one process and comparing: the + * unscoped stream delivered `server.connected` and `server.heartbeat` and + * nothing else for 100 seconds, while the scoped one delivered every + * `session.status`. Omitting it produces no error and no 404 — just a + * subscription that reports nothing, forever. + * + * The stream's `session.error` event is deliberately **not** read. It looks + * like the richer report and is the opposite: probed with a stub that + * answered with `Authorization: Bearer …`, `Set-Cookie: …`, an + * `x-request-id` and a 4KB body, that payload carried all of it verbatim in + * `responseHeaders` and `responseBody`. Relaying it would write the user's + * provider credentials into their session timeline and the hub's message + * database, permanently. The same failure arrives on the ACP channel already + * stripped to a message, which is what the launcher's prompt catch reports. + * + * Strictly supplementary: every failure path here is a debug log and a + * degraded-but-working session. A build whose HTTP API predates `/event` + * answers 404, which stops the subscription for good rather than retrying. + * + * Read as `fetch` + a hand-rolled frame reader rather than through the + * `EventSource` the runtime provides, because the endpoint is unauthenticated + * loopback but the tests are not: injecting `fetchImpl` and `sleep` is what + * lets the reconnect schedule and every failure branch be asserted + * deterministically, without a live server or fake timers. Everything + * `EventSource` would have handled — frame framing, reconnect backoff, + * `Last-Event-ID` replay — is implemented here and covered by those tests. + * + * **Known limitation: delegated turns are not covered.** An OpenCode + * task/subagent runs in its own child `ses_…`, and its retries carry that + * child's id, so the exact-match filter here drops them while the parent + * session simply stays `busy`. A rate limit hit inside a delegated turn is + * therefore still silent, and that silence is indistinguishable from the + * bug this subscription exists to fix. Deliberately not addressed here: + * following child sessions means tracking their lifetimes and deciding + * whose timeline they belong in, which is a larger change than this one. + */ +export function subscribeToOpencodeEvents(options: OpencodeEventStreamOptions): OpencodeEventSubscription { + const { + baseUrl, + directory, + sessionId, + onRetry, + fetchImpl = fetch as FetchLike, + sleep = (ms: number) => new Promise((resolve) => { + const timer = setTimeout(resolve, ms); + timer.unref?.(); + }) + } = options; + + const url = `${baseUrl}/event?directory=${encodeURIComponent(directory)}`; + const controller = new AbortController(); + let closed = false; + // Structurally typed to just what close() needs: the global and + // node:stream/web ReadableStreamDefaultReader types are not assignable + // to each other under this tsconfig. + let activeReader: { cancel: () => Promise } | null = null; + // The attempt number of the last retry seen, not of the last one + // reported: this exists purely to drop the republished copies OpenCode + // emits while a backoff is pending, and `isReportableRetryAttempt` + // separately decides which of the distinct ones are worth a row. + // Reset at a turn boundary so the next turn's attempt 1 is new again. + // + // The attempt number alone is the whole key, deliberately. Widening it + // with `next` would catch the one case this drops — a turn whose + // requests each retry exactly once, arriving as 1, 1, 1 — but the user + // sees no difference either way: `foldApiErrorEvents` collapses a run + // of consecutive api_error blocks into a single line showing the latest + // attempt, so publishing attempt 1 three times renders exactly as + // publishing it once. Reading `next` to buy nothing would also undo the + // decision not to model it at all (see OpencodeRetryStatus): once it is + // in hand, rendering a countdown from it is one line away. + let lastSeenAttempt = 0; + // Whether this subscription has ever been handed a status it could + // attribute to this session. See its one use: proving the subscription + // is actually live is otherwise impossible to distinguish from a + // healthy session that simply never retried. + let sawAttributedStatus = false; + // Last SSE `id:` seen, replayed as `last-event-id` on reconnect. Harmless + // if the server does not support resuming; if it does, it closes the gap + // where a retry announcement lands while we are reconnecting. + let lastEventId: string | null = null; + // Frames delivered by the connection currently being read. A connection + // that delivers nothing is not a healthy stream, however cleanly it ends. + let framesThisConnection = 0; + + const handleEvent = (event: unknown): void => { + if (!isObject(event)) return; + if (event.type !== 'session.status') return; + const properties = isObject(event.properties) ? event.properties : null; + if (!properties) return; + // Status is per-session bookkeeping and drives the dedupe state + // below, so an event that cannot be attributed must not touch it. + if (asString(properties.sessionID) !== sessionId) return; + + if (!sawAttributedStatus) { + sawAttributedStatus = true; + // The one positive confirmation that this subscription is wired + // to the right session and directory. Without it, a wrong + // baseUrl and a session that simply never retried look + // identical in the logs — and this whole feature exists + // because silence was the bug. + logger.debug('[opencode-events] receiving status for this session; retry reporting is live'); + } + + const isRetry = isObject(properties.status) && properties.status.type === 'retry'; + if (!isRetry) { + // idle or busy: a turn boundary. Deliberately not surfaced — + // turn state already has its own channel, and the hub layers + // its own queue rules on top of what this session reports — + // but it does mean the next turn's attempt 1 is new + // information again. Both are honoured rather than just idle: + // a reconnect that lands between the two would otherwise leave + // the following turn's retries suppressed as duplicates. + lastSeenAttempt = 0; + return; + } + + const retry = parseRetryStatus(properties.status); + if (!retry) { + // A retry we could not read (empty message, schema drift) is + // still a retry: treating it as a turn boundary would reset + // the dedupe state and let the agent's next repeat of the + // same attempt through as if it were new. + logger.debug('[opencode-events] ignoring unreadable retry status', properties.status); + return; + } + + // Equality, not a high-water mark. A turn is not one provider + // request: every tool-call round trip starts a fresh completion, + // and the attempt counter restarts at 1 with it. Suppressing + // anything at or below the highest number seen would mean a turn + // whose first request reached attempt 3 goes permanently quiet for + // every request after it — the agent still retrying, the timeline + // frozen on "attempt 3". So a *decreasing* attempt is not a + // duplicate at all: it is the news that a new request has started + // failing. Only an exact repeat is dropped, which is what OpenCode + // republishes while a backoff is pending. + if (retry.attempt === lastSeenAttempt) return; + lastSeenAttempt = retry.attempt; + + if (!isReportableRetryAttempt(retry.attempt)) return; + onRetry(retry); + }; + + /** Emits one complete SSE frame's `data:` payload, per the text/event-stream framing rules. */ + const handleFrame = (frame: string): void => { + framesThisConnection += 1; + const lines = frame.split(/\r?\n/); + const id = lines.find((line) => line.startsWith('id:')); + if (id) { + lastEventId = id.slice('id:'.length).trim(); + } + const data = lines + .filter((line) => line.startsWith('data:')) + .map((line) => line.slice('data:'.length).trim()) + .join('\n'); + if (!data) return; + let payload: unknown; + try { + payload = JSON.parse(data); + } catch (error) { + logger.debug('[opencode-events] ignoring unparsable event payload', error); + return; + } + // Dispatched outside the parse guard on purpose: everything + // downstream of handleEvent is HAPI's own code (the launcher's + // report, the hub client), and a throw from there is a bug in this + // process — not provider noise. Sharing one catch labelled + // "unparsable payload" would file it as the latter and lose it. + try { + handleEvent(payload); + } catch (error) { + logger.debug('[opencode-events] reporting a retry threw; continuing to read the stream', error); + } + }; + + const consumeStream = async (body: ReadableStream): Promise => { + const reader = body.getReader(); + activeReader = reader; + const decoder = new TextDecoder(); + let buffer = ''; + while (!closed) { + const { done, value } = await reader.read(); + // Re-check after awaiting: close() can land while a read is + // pending, and a frame that arrives with it belongs to a session + // that is already gone. + if (closed || done) return; + buffer += decoder.decode(value, { stream: true }); + for (;;) { + // Re-checked per frame, not merely per chunk: one read can + // deliver several frames, and close() lands synchronously + // from cleanup(). Without this the rest of a chunk would + // still be dispatched after the session is gone. + if (closed) return; + const separator = /\r?\n\r?\n/.exec(buffer); + if (!separator) break; + const frame = buffer.slice(0, separator.index); + buffer = buffer.slice(separator.index + separator[0].length); + handleFrame(frame); + } + } + }; + + const run = async (): Promise => { + let unproductiveAttempts = 0; + let warnedUnreachable = false; + + /** + * Counts another connection attempt that delivered nothing, and + * says so out loud once when they stop looking transient. + * + * Everything else in this module is debug-level on purpose, but a + * subscription that never works is invisible by construction: a + * wrong baseUrl, a firewalled port or an upstream that stopped + * serving this endpoint all present as a session that simply never + * retried. Since silence is the exact bug this feature exists to + * remove, that one case gets a warning. Once per outage, not once + * per attempt and not once per subscription: the flag clears again + * on the next connection that actually delivers something, so a + * session that stumbles at startup and then loses the subprocess + * hours later still reports the second, more serious failure. + */ + const noteUnproductiveAttempt = (): number => { + unproductiveAttempts += 1; + if (!warnedUnreachable && unproductiveAttempts >= UNPRODUCTIVE_ATTEMPTS_BEFORE_WARNING) { + warnedUnreachable = true; + logger.warn( + `[opencode-events] no usable event stream after ${unproductiveAttempts} attempts at ${url};` + + ' upstream retries and rate limits will not be reported for this session' + ); + } + return unproductiveAttempts; + }; + + while (!closed) { + let response: Response; + try { + response = await fetchImpl(url, { + signal: controller.signal, + headers: { + accept: 'text/event-stream', + ...(lastEventId ? { 'last-event-id': lastEventId } : {}) + } + }); + } catch (error) { + if (closed || controller.signal.aborted) return; + const delay = reconnectDelayMs(noteUnproductiveAttempt()); + logger.debug( + `[opencode-events] could not reach the event stream (attempt ${unproductiveAttempts}); retrying in ${delay}ms`, + error + ); + await sleep(delay); + continue; + } + + if (response.status === 404) { + // Permanent: this build's HTTP API predates /event. Release + // the connection and stop, so close() afterwards is a no-op. + void response.body?.cancel().catch(() => {}); + controller.abort(); + logger.debug('[opencode-events] this OpenCode build has no /event endpoint; retry reporting disabled'); + return; + } + + if (!response.ok || !response.body) { + // Nothing will read this body, so hand it back rather than + // leaving the connection open until the session tears down. + void response.body?.cancel().catch(() => {}); + const delay = reconnectDelayMs(noteUnproductiveAttempt()); + logger.debug( + `[opencode-events] event stream answered ${response.status} (attempt ${unproductiveAttempts}); retrying in ${delay}ms`, + null + ); + await sleep(delay); + continue; + } + + framesThisConnection = 0; + try { + await consumeStream(response.body); + } catch (error) { + if (closed || controller.signal.aborted) return; + logger.debug('[opencode-events] event stream dropped', error); + } finally { + activeReader = null; + } + if (closed) return; + + // Counted on what the connection actually delivered, not on the + // response status: a stream that produced frames and then ended + // reconnects promptly, one that produced none backs off. + if (framesThisConnection > 0) { + unproductiveAttempts = 0; + // Armed again: this connection worked, so a later outage is + // a new fact rather than a continuation of an old one. + warnedUnreachable = false; + await sleep(reconnectDelayMs(unproductiveAttempts)); + } else { + await sleep(reconnectDelayMs(noteUnproductiveAttempt())); + } + } + }; + + void run().catch((error) => { + logger.debug('[opencode-events] event stream subscription ended unexpectedly', error); + }); + + return { + close: () => { + if (closed) return; + closed = true; + controller.abort(); + // Releases a read that is parked waiting for the next frame, so + // no request is left dangling when the session goes away. + void activeReader?.cancel().catch(() => {}); + } + }; +} diff --git a/web/src/chat/presentation.test.ts b/web/src/chat/presentation.test.ts index da177426..b78d1d0e 100644 --- a/web/src/chat/presentation.test.ts +++ b/web/src/chat/presentation.test.ts @@ -52,6 +52,61 @@ describe('getEventPresentation — agent errors', () => { }) }) +describe('getEventPresentation — api-error', () => { + it('appends the reason to the retry wording rather than replacing it', () => { + // An agent that retries without announcing a ceiling. The reason is + // what tells a stuck-looking session apart from a rate-limited one, + // but "Retrying" is what says the agent has not given up — so both + // survive, and no producer has to encode the second one itself. + const result = getEventPresentation({ + type: 'api-error', + retryAttempt: 2, + maxRetries: 0, + error: { message: 'Rate limit exceeded: free-models-per-day. (attempt 2)' } + }) + + expect(result).toEqual({ + icon: '⏳', + text: 'API error: Retrying... Rate limit exceeded: free-models-per-day. (attempt 2)' + }) + }) + + it('reads a reason attached as a bare string', () => { + const result = getEventPresentation({ + type: 'api-error', + retryAttempt: 1, + maxRetries: 0, + error: 'Overloaded.' + }) + + expect(result.text).toBe('API error: Retrying... Overloaded.') + }) + + // Claude sessions reach this same branch set and must render exactly as + // they did before a reason was ever displayed here. + it('keeps the retry wording for an api error carrying no reason', () => { + expect(getEventPresentation({ type: 'api-error', retryAttempt: 1, maxRetries: 0, error: undefined })) + .toEqual({ icon: '⏳', text: 'API error: Retrying...' }) + }) + + it('keeps the retry wording when the reason is empty or unreadable', () => { + expect(getEventPresentation({ type: 'api-error', retryAttempt: 1, maxRetries: 0, error: { message: ' ' } }).text) + .toBe('API error: Retrying...') + expect(getEventPresentation({ type: 'api-error', retryAttempt: 1, maxRetries: 0, error: { code: 429 } }).text) + .toBe('API error: Retrying...') + }) + + it('keeps the counted, exhausted and bare renderings untouched', () => { + const error = { message: 'Rate limit exceeded.' } + expect(getEventPresentation({ type: 'api-error', retryAttempt: 2, maxRetries: 10, error })) + .toEqual({ icon: '⏳', text: 'API error: Retrying (2/10)' }) + expect(getEventPresentation({ type: 'api-error', retryAttempt: 10, maxRetries: 10, error })) + .toEqual({ icon: '⚠️', text: 'API error: Max retries reached' }) + expect(getEventPresentation({ type: 'api-error', retryAttempt: 0, maxRetries: 0, error })) + .toEqual({ icon: '⚠️', text: 'API error' }) + }) +}) + describe('getEventPresentation — limit-warning', () => { it('formats five_hour warning', () => { const result = getEventPresentation({ diff --git a/web/src/chat/presentation.ts b/web/src/chat/presentation.ts index a57a8d9d..86f92b95 100644 --- a/web/src/chat/presentation.ts +++ b/web/src/chat/presentation.ts @@ -168,6 +168,25 @@ export type EventPresentation = { text: string } +/** + * Pulls a human-readable reason out of an `api-error` payload, if it carries + * one. The field has always been normalized through (`normalizeAgent.ts` + * copies `data.error` verbatim) and never read — an agent that explained + * itself was rendered as "API error" all the same. + */ +function apiErrorDetail(error: unknown): string | null { + if (typeof error === 'string') { + return error.trim() || null + } + if (error && typeof error === 'object' && 'message' in error) { + const message = (error as { message: unknown }).message + if (typeof message === 'string') { + return message.trim() || null + } + } + return null +} + export function getEventPresentation(event: AgentEvent): EventPresentation { if (event.type === 'api-error') { const { retryAttempt, maxRetries } = event as { retryAttempt: number; maxRetries: number } @@ -178,7 +197,15 @@ export function getEventPresentation(event: AgentEvent): EventPresentation { return { icon: '⏳', text: `API error: Retrying (${retryAttempt}/${maxRetries})` } } if (retryAttempt > 0) { - return { icon: '⏳', text: 'API error: Retrying...' } + // An agent retrying without announcing a ceiling. "Retrying..." + // alone is what makes a rate-limited session look like one that + // is merely slow, so the reason is appended when there is one — + // appended, not substituted, because the existing wording is + // the part that says the agent has not given up. Replacing it + // would put that entirely in the hands of whatever text a + // producer happens to attach. + const detail = apiErrorDetail((event as { error?: unknown }).error) + return { icon: '⏳', text: detail ? `API error: Retrying... ${detail}` : 'API error: Retrying...' } } return { icon: '⚠️', text: 'API error' } }