diff --git a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client-worker-lifecycle.test.ts b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client-worker-lifecycle.test.ts index 69cb7d18652..c2d6c856a47 100644 --- a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client-worker-lifecycle.test.ts +++ b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client-worker-lifecycle.test.ts @@ -90,6 +90,25 @@ describe('ForeignSqliteReaderClient worker lifecycle', () => { client.readOpenCodeBinderSessions('/synthetic/opencode.db', { ms: 0, id: '' }), failure: [], answer: [{ id: 'ses_a', directory: '/w', createdAtMs: 1, parentId: null }] + }, + { + reader: 'openCodePromptFailure', + read: (client: ForeignSqliteReaderClient) => + client.readOpenCodePromptFailure({ + dbPath: '/synthetic/opencode.db', + sessionId: 'ses-a', + startedAt: 100, + failedAt: 200 + }), + failure: { failure: null }, + answer: { + failure: { + inputId: 'input', + type: 'provider.quota', + message: 'Quota exhausted', + status: 429 + } + } } ])( '$reader refuses new native workers until the retired worker positively terminates', diff --git a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.test.ts b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.test.ts index 318fc2b8c69..668fdf20282 100644 --- a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.test.ts +++ b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.test.ts @@ -70,6 +70,52 @@ function settle(): Promise { } describe('ForeignSqliteReaderClient', () => { + it('includes queue time in the native failure deadline and removes expired work', async () => { + vi.useFakeTimers() + const workers: FakeWorker[] = [] + const client = new ForeignSqliteReaderClient({ workerFactory: fakeFactory(workers), log() {} }) + try { + const query = { dbPath: '/a/opencode.db', sessionId: 'a', startedAt: 100, failedAt: 200 } + const first = client.readOpenCodePromptFailure(query) + const second = client.readOpenCodePromptFailure({ ...query, sessionId: 'b' }) + let queued: unknown + void second.then((value) => { + queued = value + }) + await vi.advanceTimersByTimeAsync(1_999) + workers[0]?.reply({ failure: null }) + await first + await vi.advanceTimersByTimeAsync(1) + expect(queued).toEqual({ failure: null }) + expect(workers[0]?.terminated).toBe(true) + } finally { + client.dispose() + vi.useRealTimers() + } + }) + + it('does not share native prompt evidence across different turns in the same session', async () => { + const workers: FakeWorker[] = [] + const client = new ForeignSqliteReaderClient({ workerFactory: fakeFactory(workers), log() {} }) + const query = { dbPath: '/a/opencode.db', sessionId: 'ses-own', startedAt: 100, failedAt: 200 } + const first = client.readOpenCodePromptFailure(query) + expect(client.readOpenCodePromptFailure(query)).toBe(first) + const next = client.readOpenCodePromptFailure({ ...query, startedAt: 300, failedAt: 400 }) + const quota = { + failure: { inputId: 'input', type: 'provider.quota', message: 'Quota exhausted', status: 429 } + } + workers[0]?.reply(quota) + await expect(first).resolves.toEqual(quota) + await settle() + workers[0]?.reply({ failure: null }) + await expect(next).resolves.toEqual({ failure: null }) + expect(workers[0]?.posted.map((request) => request.kind)).toEqual([ + 'openCodePromptFailure', + 'openCodePromptFailure' + ]) + client.dispose() + }) + it('returns the reader result the worker posts', async () => { const workers: FakeWorker[] = [] const client = new ForeignSqliteReaderClient({ workerFactory: fakeFactory(workers), log() {} }) diff --git a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.ts b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.ts index ad25bd300cd..223c474b1c7 100644 --- a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.ts +++ b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-client.ts @@ -16,19 +16,27 @@ import type { ForeignSqliteReaderRequest, ForeignSqliteReaderResponse } from './foreign-sqlite-reader-protocol' +import { + openCodePromptFailureUnavailable, + parseOpenCodePromptFailureResult, + type OpenCodePromptFailureQuery, + type OpenCodePromptFailureResult +} from './opencode-prompt-failure-result' // Foreign SQLite opens can block on WAL rebuilds; isolate each reader so slow databases cannot stall main or peers. const READ_TIMEOUT_MS: Record = { // Allow margin beyond observed 4–6s WAL-index rebuilds. cursorProfile: 10_000, - openCodeBinderSessions: 60_000 + openCodeBinderSessions: 60_000, + openCodePromptFailure: 2_000 } // Keep workers alive across poll intervals to avoid spawning on every round. const DEFAULT_IDLE_TEARDOWN_MS: Record = { cursorProfile: 30_000, // The binder polls once per minute. - openCodeBinderSessions: 120_000 + openCodeBinderSessions: 120_000, + openCodePromptFailure: 30_000 } const MAX_CONSECUTIVE_DEATHS = 2 const MAX_QUEUED_READS = 8 @@ -38,6 +46,7 @@ type LaneSettings = { log: (message: string) => void timeoutMs: number idleTeardownMs: number + deadlineIncludesQueue?: boolean } export class ForeignSqliteReaderLane { @@ -93,8 +102,11 @@ export class ForeignSqliteReaderLane { private async dispatch(buildRequest: (id: number) => ForeignSqliteReaderRequest): Promise { const { kind, settings } = this + const deadline = settings.deadlineIncludesQueue ? new AbortController() : undefined + const timer = deadline ? setTimeout(() => deadline.abort(), settings.timeoutMs) : undefined + timer?.unref() try { - const response = await this.queue.dispatch(buildRequest, settings.timeoutMs) + const response = await this.queue.dispatch(buildRequest, settings.timeoutMs, deadline?.signal) if (!response.ok) { settings.log(`Foreign SQLite reader ${kind} failed: ${response.error}`) return this.failure() @@ -109,6 +121,8 @@ export class ForeignSqliteReaderLane { // A failed worker read must never fall back to blocking the main thread. settings.log(`Foreign SQLite reader ${kind} did not answer: ${errorText(err)}`) return this.failure() + } finally { + clearTimeout(timer) } } } @@ -116,6 +130,7 @@ export class ForeignSqliteReaderLane { export class ForeignSqliteReaderClient { private readonly cursorProfile: ForeignSqliteReaderLane private readonly openCodeBinderSessions: ForeignSqliteReaderLane + private readonly openCodePromptFailure: ForeignSqliteReaderLane constructor(options: { workerFactory: WorkerThreadFactory @@ -141,6 +156,12 @@ export class ForeignSqliteReaderClient { openCodeBinderSessionsFailure, settings('openCodeBinderSessions') ) + this.openCodePromptFailure = new ForeignSqliteReaderLane( + 'openCodePromptFailure', + parseOpenCodePromptFailureResult, + openCodePromptFailureUnavailable, + { ...settings('openCodePromptFailure'), deadlineIncludesQueue: true } + ) } readCursorProfile(dbPath: string): Promise { @@ -165,6 +186,18 @@ export class ForeignSqliteReaderClient { dispose(): void { this.cursorProfile.dispose() this.openCodeBinderSessions.dispose() + this.openCodePromptFailure.dispose() + } + + readOpenCodePromptFailure( + query: OpenCodePromptFailureQuery + ): Promise { + const key = JSON.stringify([query.dbPath, query.sessionId, query.startedAt, query.failedAt]) + return this.openCodePromptFailure.read(key, (id) => ({ + id, + kind: 'openCodePromptFailure', + ...query + })) } } diff --git a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-dispatch.ts b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-dispatch.ts index 3e7bd6abb55..a43d340b048 100644 --- a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-dispatch.ts +++ b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-dispatch.ts @@ -4,6 +4,7 @@ import type { } from './foreign-sqlite-reader-protocol' import { readCursorProfile } from './readers/cursor-profile' import { readOpenCodeBinderSessions } from './readers/opencode-binder-sessions' +import { readOpenCodePromptFailure } from './readers/opencode-prompt-failure' /** Structured-clone requests may carry unknown kinds despite the static type. */ export function handleForeignSqliteReaderRequest( @@ -17,6 +18,8 @@ export function handleForeignSqliteReaderRequest( return { id, ok: true, value: readCursorProfile(request.dbPath) } case 'openCodeBinderSessions': return { id, ok: true, value: readOpenCodeBinderSessions(request.dbPath, request.cursor) } + case 'openCodePromptFailure': + return { id, ok: true, value: readOpenCodePromptFailure(request) } } return unknownKind(id, kind) } catch (err) { diff --git a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-protocol.ts b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-protocol.ts index f74d029858a..9d5ff579887 100644 --- a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-protocol.ts +++ b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-protocol.ts @@ -1,6 +1,7 @@ // Keep this Electron-free so both worker and main-process clients can import it. import type { OpenCodeSessionCursor } from './opencode-binder-sessions-result' +import type { OpenCodePromptFailureQuery } from './opencode-prompt-failure-result' type CursorProfileRequest = { id: number @@ -15,7 +16,15 @@ type OpenCodeBinderSessionsRequest = { cursor: OpenCodeSessionCursor } -export type ForeignSqliteReaderRequest = CursorProfileRequest | OpenCodeBinderSessionsRequest +type OpenCodePromptFailureRequest = OpenCodePromptFailureQuery & { + id: number + kind: 'openCodePromptFailure' +} + +export type ForeignSqliteReaderRequest = + | CursorProfileRequest + | OpenCodeBinderSessionsRequest + | OpenCodePromptFailureRequest export type ForeignSqliteReaderKind = ForeignSqliteReaderRequest['kind'] diff --git a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-spawn.ts b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-spawn.ts index 5ee4184205a..7ad4dd43427 100644 --- a/src/main/foreign-sqlite-readers/foreign-sqlite-reader-spawn.ts +++ b/src/main/foreign-sqlite-readers/foreign-sqlite-reader-spawn.ts @@ -6,6 +6,10 @@ import type { CursorDesktopProfileReadResult } from './cursor-profile-result' import { ForeignSqliteReaderClient } from './foreign-sqlite-reader-client' import { resolveForeignSqliteReaderEntryPath } from './foreign-sqlite-reader-entry-path' import type { BinderSessionRow, OpenCodeSessionCursor } from './opencode-binder-sessions-result' +import type { + OpenCodePromptFailureQuery, + OpenCodePromptFailureResult +} from './opencode-prompt-failure-result' function defaultWorkerFactory(): Worker { const workerPath = resolveForeignSqliteReaderEntryPath(currentWorkerEntryLayout(__dirname)) @@ -26,6 +30,12 @@ export function readCursorDesktopProfile(dbPath: string): Promise { + return getSharedClient().readOpenCodePromptFailure(query) +} + /** Return oldest-first rows after cursor; unreadable or unavailable readers answer []. */ export async function readOpenCodeBinderSessions( dbPath: string, diff --git a/src/main/foreign-sqlite-readers/opencode-prompt-failure-result.ts b/src/main/foreign-sqlite-readers/opencode-prompt-failure-result.ts new file mode 100644 index 00000000000..b9ed2441f48 --- /dev/null +++ b/src/main/foreign-sqlite-readers/opencode-prompt-failure-result.ts @@ -0,0 +1,32 @@ +import { z } from 'zod' + +export type OpenCodePromptFailureQuery = { + dbPath: string + sessionId: string + startedAt: number + failedAt: number +} + +export const openCodePromptFailureSchema = z.object({ + inputId: z.string().min(1).max(256), + type: z.string().min(1).max(128), + message: z.string().min(1).max(4096), + status: z.number().int().min(100).max(599).optional(), + resetsAt: z.number().int().positive().optional(), + responseType: z.string().max(128).optional(), + responseMessage: z.string().max(4096).optional() +}) + +export type OpenCodePromptFailure = z.infer +export type OpenCodePromptFailureResult = { failure: OpenCodePromptFailure | null } + +export function parseOpenCodePromptFailureResult( + value: unknown +): OpenCodePromptFailureResult | null { + const parsed = z.object({ failure: openCodePromptFailureSchema.nullable() }).safeParse(value) + return parsed.success ? parsed.data : null +} + +export function openCodePromptFailureUnavailable(): OpenCodePromptFailureResult { + return { failure: null } +} diff --git a/src/main/foreign-sqlite-readers/readers/opencode-prompt-failure.test.ts b/src/main/foreign-sqlite-readers/readers/opencode-prompt-failure.test.ts new file mode 100644 index 00000000000..8c7b0985537 --- /dev/null +++ b/src/main/foreign-sqlite-readers/readers/opencode-prompt-failure.test.ts @@ -0,0 +1,141 @@ +import { mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import SyncDatabase from '../../sqlite/sync-database' +import { readOpenCodePromptFailure } from './opencode-prompt-failure' + +let directory: string +let db: SyncDatabase +let dbPath: string + +beforeEach(() => { + directory = mkdtempSync(join(tmpdir(), 'orca-opencode-failure-')) + dbPath = join(directory, 'opencode.db') + db = new SyncDatabase(dbPath) + db.exec(`CREATE TABLE session_message ( + id TEXT PRIMARY KEY, session_id TEXT, type TEXT, seq INTEGER, time_created INTEGER, data TEXT + ); CREATE INDEX session_message_session_seq ON session_message(session_id, seq);`) + db.prepare('INSERT INTO session_message VALUES (?, ?, ?, ?, ?, ?)').run( + 'input', + 'ses-own', + 'user', + 0, + 100, + '{}' + ) +}) + +afterEach(() => { + db.close() + rmSync(directory, { recursive: true, force: true }) +}) + +function message(seq: number, error: unknown, options: { session?: string; at?: number } = {}) { + db.prepare('INSERT INTO session_message VALUES (?, ?, ?, ?, ?, ?)').run( + `msg-${seq}`, + options.session ?? 'ses-own', + 'idle', + seq, + options.at ?? 150, + JSON.stringify({ error }) + ) +} + +const quota = { type: 'provider.quota', message: 'Free usage limit reached.', status: 429 } +const query = { sessionId: 'ses-own', startedAt: 100, failedAt: 200 } + +describe('OpenCode native prompt failure', () => { + it('recovers the native type and HTTP status from the completed turn', () => { + message(1, quota) + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ + failure: { ...quota, inputId: 'input' } + }) + expect(db.prepare('SELECT COUNT(*) AS count FROM session_message').get()).toEqual({ count: 2 }) + }) + + it('does not reuse an outcome when a newer input arrives in the same millisecond', () => { + message(1, quota, { at: 100 }) + db.prepare('INSERT INTO session_message VALUES (?, ?, ?, ?, ?, ?)').run( + 'next-input', + 'ses-own', + 'user', + 2, + 100, + '{}' + ) + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ failure: null }) + }) + + it('requires an input from this prompt even when an older execution fails during its window', () => { + db.prepare('UPDATE session_message SET time_created = 99 WHERE id = ?').run('input') + message(1, quota) + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ failure: null }) + }) + + it('ignores other sessions and failures before or after this prompt', () => { + message(1, quota, { at: 99 }) + message(2, quota, { at: 201 }) + message(3, quota, { session: 'ses-other' }) + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ failure: null }) + }) + + it('does not reuse a failed step after a newer successful outcome', () => { + message(1, quota) + message(2, undefined) + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ failure: null }) + }) + + it('keeps retry timing anchored to the native failure rather than the eventual read', () => { + message(1, { + type: 'provider.rate-limit', + message: 'Slow down', + status: 429, + retryAfterMs: 60_000 + }) + expect(readOpenCodePromptFailure({ dbPath, ...query }).failure).toMatchObject({ + resetsAt: 60_150 + }) + }) + + it.each([-1, 'invalid', 1e99])( + 'ignores malformed retry timing without losing the failure: %j', + (retryAfterMs) => { + message(1, { ...quota, retryAfterMs }) + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ + failure: { ...quota, inputId: 'input' } + }) + } + ) + + it('preserves a typed gateway refusal hidden behind native authentication classification', () => { + message(1, { + type: 'provider.auth', + message: 'Provider authentication required', + status: 403, + response: { + body: JSON.stringify({ error: { type: 'RegionError', message: 'Region blocked' } }) + } + }) + expect(readOpenCodePromptFailure({ dbPath, ...query }).failure).toMatchObject({ + status: 403, + responseType: 'RegionError', + responseMessage: 'Region blocked' + }) + }) + + it('bounds provider response bodies without losing the native failure', () => { + message(1, { ...quota, response: { body: 'x'.repeat(100_000) } }) + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ + failure: { ...quota, inputId: 'input' } + }) + }) + + it('leaves unknown schemas and malformed latest messages unresolved', () => { + message(1, quota) + db.prepare('UPDATE session_message SET data = ?').run('{broken') + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ failure: null }) + db.exec('DROP TABLE session_message') + expect(readOpenCodePromptFailure({ dbPath, ...query })).toEqual({ failure: null }) + }) +}) diff --git a/src/main/foreign-sqlite-readers/readers/opencode-prompt-failure.ts b/src/main/foreign-sqlite-readers/readers/opencode-prompt-failure.ts new file mode 100644 index 00000000000..527b3048897 --- /dev/null +++ b/src/main/foreign-sqlite-readers/readers/opencode-prompt-failure.ts @@ -0,0 +1,104 @@ +import { z } from 'zod' +import { usageLimitResetsAt } from '../../../shared/agent-session-usage-limit' +import { readOpenCodeDatabase } from '../../ai-vault/session-scanner-opencode-sqlite-open' +import { columnExists, tableExists } from '../../opencode-usage/schema-helpers' +import { + openCodePromptFailureSchema, + type OpenCodePromptFailureQuery, + type OpenCodePromptFailureResult +} from '../opencode-prompt-failure-result' + +const rowSchema = z.object({ + inputId: z.string(), + type: z.string(), + message: z.string(), + status: z.number().nullable(), + createdAt: z.number().int(), + retryAfterMs: z.number().int().nonnegative().nullable().catch(null), + response: z.string().nullable() +}) +const responseSchema = z.object({ + error: z.object({ type: z.string().max(128), message: z.string().max(4096).optional() }) +}) + +export function readOpenCodePromptFailure( + query: OpenCodePromptFailureQuery +): OpenCodePromptFailureResult { + return readOpenCodeDatabase({ + dbPath: query.dbPath, + read: (db) => { + const table = 'session_message' + if ( + !tableExists(db, table) || + !['id', 'session_id', 'seq', 'type', 'data', 'time_created'].every((column) => + columnExists(db, table, column) + ) + ) { + return { failure: null } + } + // Require fresh input before the latest outcome; never borrow an earlier turn's failure. + const row: unknown = db + .prepare( + `WITH input AS ( + SELECT id AS inputId, seq AS input_seq, time_created AS input_created FROM session_message + WHERE session_id = ? AND type = 'user' ORDER BY seq DESC LIMIT 1 + ), outcome AS ( + SELECT seq, data, time_created FROM session_message + WHERE session_id = ? AND type IN ('assistant', 'idle') + ORDER BY seq DESC LIMIT 1 + ) SELECT + inputId, + CASE WHEN json_valid(data) THEN substr(json_extract(data, '$.error.type'), 1, 128) END AS type, + CASE WHEN json_valid(data) THEN substr(json_extract(data, '$.error.message'), 1, 4096) END AS message, + CASE WHEN json_valid(data) THEN json_extract(data, '$.error.status') END AS status, + time_created AS createdAt, + CASE WHEN json_valid(data) THEN json_extract(data, '$.error.retryAfterMs') END AS retryAfterMs, + CASE WHEN json_valid(data) THEN substr(json_extract(data, '$.error.response.body'), 1, 16384) END AS response + FROM outcome, input + WHERE input_seq < seq AND input_created >= ? AND input_created <= ? + AND time_created >= ? AND time_created <= ?` + ) + .get( + query.sessionId, + query.sessionId, + query.startedAt, + query.failedAt, + query.startedAt, + query.failedAt + ) + const parsed = rowSchema.safeParse(row) + if (!parsed.success) { + return { failure: null } + } + let response: z.infer | undefined + if (parsed.data.response) { + try { + const decoded = responseSchema.safeParse(JSON.parse(parsed.data.response)) + if (decoded.success) { + response = decoded.data + } + } catch { + // An invalid response body does not invalidate the native error itself. + } + } + const failure = openCodePromptFailureSchema.safeParse({ + inputId: parsed.data.inputId, + type: parsed.data.type, + message: parsed.data.message, + ...(parsed.data.status === null ? {} : { status: parsed.data.status }), + ...(parsed.data.retryAfterMs === null + ? {} + : { + resetsAt: usageLimitResetsAt( + parsed.data.createdAt + parsed.data.retryAfterMs, + query.failedAt + ) + }), + ...(response + ? { responseType: response.error.type, responseMessage: response.error.message } + : {}) + }) + return { failure: failure.success ? failure.data : null } + } + }) +} diff --git a/src/main/opencode/opencode-acp-connection.test.ts b/src/main/opencode/opencode-acp-connection.test.ts new file mode 100644 index 00000000000..7f62cca6de7 --- /dev/null +++ b/src/main/opencode/opencode-acp-connection.test.ts @@ -0,0 +1,351 @@ +import { mkdirSync, mkdtempSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { PassThrough } from 'node:stream' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { createFakePipedChild } from '../../shared/__fixtures__/fake-spawned-child' +import { AcpScriptedAgent, deferred } from '../acp/acp-scripted-agent.test-support' +import { AcpAgentError, AcpAuthRequiredError } from '../acp/acp-errors' +import { AcpTimelineTranslator } from '../acp/acp-timeline-translator' +import { acpLaunchSpecFor } from '../acp/acp-launch-specs' +import { openAcpAdapterRig, sendHello } from '../acp/acp-structured-adapter.test-support' +import { + closeProviderTimelineRigs, + SESSION +} from '../native-chat/agent-session-timeline/provider-timeline-assembler-test-support' +import { OPENCODE_ACP_DIALECT } from '../acp/acp-dialects/opencode-dialect' +import { readOpenCodePromptFailure } from '../foreign-sqlite-readers/readers/opencode-prompt-failure' +import type { OpenCodePromptFailureQuery } from '../foreign-sqlite-readers/opencode-prompt-failure-result' +import SyncDatabase from '../sqlite/sync-database' +import { createOpenCodeAcpConnection } from './opencode-acp-connection' + +const cleanups: (() => Promise)[] = [] +afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup() + } + vi.unstubAllEnvs() + await closeProviderTimelineRigs() +}) + +async function fixture( + native: unknown, + options: { + stale?: boolean + unavailable?: boolean + readGate?: Promise + readStarted?: () => void + acpCode?: number + } = {} +) { + const directory = mkdtempSync(join(tmpdir(), 'orca-opencode-acp-failure-')) + const dataHome = join(directory, 'account-data') + mkdirSync(join(dataHome, 'opencode'), { recursive: true }) + const dbPath = join(dataHome, 'opencode', 'opencode.db') + const db = new SyncDatabase(dbPath) + db.exec(`CREATE TABLE session_message ( + id TEXT PRIMARY KEY, session_id TEXT, type TEXT, seq INTEGER, time_created INTEGER, data TEXT + )`) + const agent = new AcpScriptedAgent() + const child = Object.assign(createFakePipedChild(), { + pid: 9_999_999, + stdout: agent.stdout, + stdin: agent.stdin, + stderr: new PassThrough() + }) + agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 1 })) + agent.on('session/new', (frame) => agent.reply(frame, { sessionId: 'ses-own' })) + let turn = 0 + agent.on('session/prompt', (frame) => { + turn++ + db.prepare('INSERT OR REPLACE INTO session_message VALUES (?, ?, ?, ?, ?, ?)').run( + `input-${turn}`, + 'ses-own', + 'user', + turn * 10, + Date.now(), + '{}' + ) + db.prepare('INSERT OR REPLACE INTO session_message VALUES (?, ?, ?, ?, ?, ?)').run( + `outcome-${turn}`, + 'ses-own', + 'idle', + turn * 10 + 1, + options.stale ? 1 : Date.now(), + JSON.stringify({ error: native }) + ) + agent.fail( + frame, + options.acpCode ?? -32000, + 'Authentication required: provider authentication required', + { + details: 'provider authentication required' + } + ) + }) + const readFailure = vi.fn(async (query: OpenCodePromptFailureQuery) => { + if (options.unavailable) { + throw new Error('database unavailable') + } + const result = readOpenCodePromptFailure(query) + options.readStarted?.() + await options.readGate + return result + }) + const connection = createOpenCodeAcpConnection( + { + command: 'opencode2', + args: ['acp'], + cwd: directory, + env: { XDG_DATA_HOME: dataHome }, + envToDelete: ['OPENCODE_DB'] + }, + {}, + () => child, + readFailure + ) + cleanups.push(async () => { + child.emit('exit', 0, null) + await connection.close() + db.close() + rmSync(directory, { recursive: true, force: true }) + }) + await connection.start({ cwd: directory, mcpServers: [] }) + return { connection, readFailure, dbPath, agent, child, cwd: directory } +} + +const quota = { type: 'provider.quota', message: 'Free usage limit reached.', status: 429 } + +describe('OpenCode ACP native failure recovery', () => { + it('leaves successful turns off the native failure reader', async () => { + const { connection, readFailure, agent } = await fixture(quota) + agent.on('session/prompt', (frame) => agent.reply(frame, { stopReason: 'end_turn' })) + await expect(connection.prompt([{ type: 'text', text: 'hello' }])).resolves.toEqual({ + stopReason: 'end_turn' + }) + expect(readFailure).not.toHaveBeenCalled() + }) + + it('reads the selected account after stripping an inherited database override', async () => { + vi.stubEnv('OPENCODE_DB', join(tmpdir(), 'other-account', 'opencode.db')) + const { connection, readFailure, dbPath } = await fixture(quota) + await expect(connection.prompt([{ type: 'text', text: 'hello' }])).rejects.toMatchObject({ + message: quota.message + }) + expect(readFailure).toHaveBeenCalledWith(expect.objectContaining({ dbPath })) + }) + + it.each([-32000, -32603])( + 'reports native quota correctly when ACP returns %i', + async (acpCode) => { + const { connection, readFailure, dbPath } = await fixture(quota, { acpCode }) + const error: unknown = await connection + .prompt([{ type: 'text', text: 'hello' }]) + .catch((e) => e) + expect(error).toBeInstanceOf(AcpAgentError) + expect(error).not.toBeInstanceOf(AcpAuthRequiredError) + if (!(error instanceof AcpAgentError)) { + throw error + } + expect(error.message).toBe(quota.message) + expect(error.data).toMatchObject({ errorName: 'provider.quota', status: 429 }) + expect(readFailure).toHaveBeenCalledWith( + expect.objectContaining({ dbPath, sessionId: 'ses-own' }) + ) + const translator = new AcpTimelineTranslator({ + sessionId: 'ses-own', + agentName: 'OpenCode 2', + dialect: OPENCODE_ACP_DIALECT + }) + translator.openPrompt('send', 100) + const rows = translator.promptFailed('send', error, 200) + expect(rows).toContainEqual( + expect.objectContaining({ + type: 'item.update', + body: expect.objectContaining({ + failure: expect.objectContaining({ + kind: 'usageLimitReached', + detail: { text: quota.message, audience: 'person' } + }) + }) + }) + ) + } + ) + + it('journals the recovered failure through the full chat adapter and accepts the next send', async () => { + const spec = acpLaunchSpecFor('opencode2') + if (!spec) { + throw new Error('OpenCode 2 launch specification missing') + } + const { connection, agent, cwd } = await fixture({ ...quota, retryAfterMs: 60_000 }) + const rig = await openAcpAdapterRig({ + spec, + launch: { cwd }, + deps: { connect: () => connection, now: Date.now } + }) + cleanups.push(async () => { + await rig.adapter.closeSession(SESSION) + }) + await rig.acquire() + await sendHello(rig, 'first') + await rig.settle() + expect((await rig.rig.rows()).map((row) => row.body)).toContainEqual( + expect.objectContaining({ + kind: 'status', + tone: 'warning', + failure: expect.objectContaining({ + kind: 'usageLimitReached', + detail: { text: quota.message, audience: 'person' }, + usageLimit: { resetsAt: expect.any(Number) } + }) + }) + ) + expect((await rig.rig.turns())[0]?.outcome).toBe('failure') + agent.on('session/prompt', (frame) => agent.reply(frame, { stopReason: 'end_turn' })) + await sendHello(rig, 'second') + await rig.settle() + expect((await rig.rig.turns()).map((turn) => turn.outcome)).toEqual(['failure', 'success']) + }) + + it.each([ + { type: 'provider.quota', message: 'Insufficient account funds', status: 402 }, + { + type: 'provider.auth', + message: 'Provider authentication required', + status: 403, + response: { + body: JSON.stringify({ error: { type: 'FreeTierError', message: 'Free tier restricted' } }) + } + } + ])('does not turn billing or access refusal into a sign-in failure: $status', async (native) => { + const { connection } = await fixture(native) + const error: unknown = await connection + .prompt([{ type: 'text', text: 'hello' }]) + .catch((e) => e) + expect(error).not.toBeInstanceOf(AcpAuthRequiredError) + if (!(error instanceof AcpAgentError)) { + throw error + } + expect(OPENCODE_ACP_DIALECT.promptFailure?.(error, 100)).toBeUndefined() + }) + + it('keeps the same session usable after a quota failure', async () => { + const { connection, agent } = await fixture(quota) + await expect(connection.prompt([{ type: 'text', text: 'hello' }])).rejects.toMatchObject({ + message: quota.message + }) + agent.on('session/prompt', (frame) => agent.reply(frame, { stopReason: 'end_turn' })) + await expect(connection.prompt([{ type: 'text', text: 'try another model' }])).resolves.toEqual( + { + stopReason: 'end_turn' + } + ) + expect(connection.closed).toBe(false) + }) + + it('keeps another prompt out until native failure recovery settles', async () => { + const started = deferred() + const gate = deferred() + const { connection, agent } = await fixture(quota, { + readGate: gate.promise, + readStarted: () => started.resolve() + }) + const pending = connection.prompt([{ type: 'text', text: 'first' }]).catch((error) => error) + await started.promise + const overlapping = connection + .prompt([{ type: 'text', text: 'second' }]) + .catch((error) => error) + gate.resolve() + await expect(overlapping).resolves.toMatchObject({ message: 'ACP prompt already in progress' }) + await expect(pending).resolves.toMatchObject({ message: quota.message }) + expect(agent.frames.filter((frame) => frame.method === 'session/prompt')).toHaveLength(1) + }) + + it('preserves newer native failure classifications instead of interpreting their raw body again', async () => { + const { connection } = await fixture({ + type: 'provider.free-limit', + message: 'Connect a provider or switch models.', + status: 429, + response: { + body: JSON.stringify({ error: { type: 'FreeUsageLimitError', message: 'Legacy message' } }) + } + }) + await expect(connection.prompt([{ type: 'text', text: 'hello' }])).rejects.toMatchObject({ + message: 'Connect a provider or switch models.', + data: { errorName: 'provider.free-limit', status: 429 } + }) + }) + + it('does not consume the same native failure twice when successive prompts share a clock tick', async () => { + const clock = vi.spyOn(Date, 'now').mockReturnValue(1_800_000_000_000) + try { + const { connection, agent } = await fixture(quota) + await expect(connection.prompt([{ type: 'text', text: 'first' }])).rejects.toMatchObject({ + message: quota.message + }) + agent.on('session/prompt', (frame) => agent.fail(frame, -32000, 'Authentication required')) + await expect(connection.prompt([{ type: 'text', text: 'second' }])).rejects.toMatchObject({ + name: 'AcpAuthRequiredError', + message: 'Authentication required' + }) + } finally { + clock.mockRestore() + } + }) + + it('keeps HTTP 401 classified as authentication even with an access error in the body', async () => { + const { connection } = await fixture({ + type: 'provider.auth', + message: 'Invalid API key', + status: 401, + response: { + body: JSON.stringify({ error: { type: 'RegionError', message: 'Region blocked' } }) + } + }) + await expect(connection.prompt([{ type: 'text', text: 'hello' }])).rejects.toMatchObject({ + name: 'AcpAuthRequiredError', + message: 'Invalid API key' + }) + }) + + it('does not revise a closed connection with a late native read', async () => { + const started = deferred() + const gate = deferred() + const { connection, child } = await fixture(quota, { + readGate: gate.promise, + readStarted: () => started.resolve() + }) + const pending = connection.prompt([{ type: 'text', text: 'hello' }]) + await started.promise + child.emit('exit', 0, null) + gate.resolve() + await expect(pending).rejects.toMatchObject({ + name: 'AcpAuthRequiredError', + message: 'Authentication required: provider authentication required' + }) + }) + + it('keeps genuine authentication failures classified as signed out', async () => { + const { connection } = await fixture({ + type: 'provider.auth', + message: 'Invalid API key', + status: 401 + }) + await expect(connection.prompt([{ type: 'text', text: 'hello' }])).rejects.toMatchObject({ + name: 'AcpAuthRequiredError', + message: 'Invalid API key' + }) + }) + + it.each([{ stale: true }, { unavailable: true }])( + 'retains the ACP error without fresh native evidence: %j', + async (options) => { + const { connection } = await fixture(quota, options) + await expect(connection.prompt([{ type: 'text', text: 'hello' }])).rejects.toMatchObject({ + name: 'AcpAuthRequiredError', + message: 'Authentication required: provider authentication required' + }) + } + ) +}) diff --git a/src/main/opencode/opencode-acp-connection.ts b/src/main/opencode/opencode-acp-connection.ts new file mode 100644 index 00000000000..de4e6a64b7b --- /dev/null +++ b/src/main/opencode/opencode-acp-connection.ts @@ -0,0 +1,121 @@ +import { homedir } from 'node:os' +import { resolve } from 'node:path' +import { spawnProcess } from '@orca/process-host' +import type { PipedProcessSpawner } from '@orca/process-host/process-spec' +import { isRecord } from '../../shared/agent-status-child-work-value-guards' +import { AcpAgentConnection, type AcpAgentConnectionOptions } from '../acp/acp-agent-connection' +import { AcpAgentError, AcpAuthRequiredError, AcpSignInPromptError } from '../acp/acp-errors' +import type { AcpSessionStartOptions } from '../acp/acp-session-setup' +import type { PromptRequest } from '../acp/generated/acp-protocol.generated' +import { readOpenCodePromptFailure } from '../foreign-sqlite-readers/foreign-sqlite-reader-spawn' +import type { OpenCodePromptFailure } from '../foreign-sqlite-readers/opencode-prompt-failure-result' +import { + resolveProviderChildEnv, + type ProviderProcessLaunch +} from '../provider-process/provider-process-launch' +import { resolveOpenCodeDatabasePath } from './opencode-data-directory' + +const ACCESS_REFUSALS = new Set(['FreeTierError', 'RegionError']) + +function nativePromptError(failure: OpenCodePromptFailure, original: AcpAgentError): AcpAgentError { + const access = + failure.type === 'provider.auth' && + failure.status === 403 && + ACCESS_REFUSALS.has(failure.responseType ?? '') + const errorName = access ? 'provider.access' : failure.type + const message = (access ? failure.responseMessage : undefined) ?? failure.message + const data = { + ...(isRecord(original.data) ? original.data : {}), + details: message, + errorName, + ...(failure.status === undefined ? {} : { status: failure.status }), + ...(failure.resetsAt === undefined ? {} : { resetsAt: failure.resetsAt }) + } + return !access && + (failure.status === 401 || ['provider.auth', 'provider.authentication'].includes(errorName)) + ? new AcpAuthRequiredError( + message, + data, + original instanceof AcpAuthRequiredError ? original.authMethods : undefined + ) + : new AcpAgentError(-32603, message, data) +} + +export function createOpenCodeAcpConnection( + launch: ProviderProcessLaunch, + options: AcpAgentConnectionOptions = {}, + spawnImpl: PipedProcessSpawner = spawnProcess, + readFailure: typeof readOpenCodePromptFailure = readOpenCodePromptFailure +): AcpAgentConnection { + return new OpenCodeAcpConnection(launch, options, spawnImpl, readFailure) +} + +class OpenCodeAcpConnection extends AcpAgentConnection { + private readonly dbPath: string | null + private providerSessionId: string | undefined + private lastRecoveredInput: string | undefined + private prompting = false + + constructor( + launch: ProviderProcessLaunch, + options: AcpAgentConnectionOptions, + spawnImpl: PipedProcessSpawner, + private readonly readFailure: typeof readOpenCodePromptFailure + ) { + const env = resolveProviderChildEnv(launch, process.env) + const home = (process.platform === 'win32' ? env.USERPROFILE : env.HOME) || homedir() + const selected = resolveOpenCodeDatabasePath(env, home) + const dbPath = selected === null ? null : resolve(launch.cwd ?? process.cwd(), selected) + super(launch, options, spawnImpl) + this.dbPath = dbPath + } + + override async start(options: AcpSessionStartOptions) { + const started = await super.start(options) + this.providerSessionId = started.sessionId + return started + } + + override async prompt(prompt: PromptRequest['prompt'], meta?: PromptRequest['_meta']) { + if (this.prompting) { + throw new Error('ACP prompt already in progress') + } + this.prompting = true + const startedAt = Date.now() + try { + return await super.prompt(prompt, meta) + } catch (error) { + if ( + !(error instanceof AcpAgentError) || + error instanceof AcpSignInPromptError || + ![-32000, -32603].includes(error.code) || + !this.dbPath || + !this.providerSessionId || + this.closed + ) { + throw error + } + const failedAt = Date.now() + let failure: OpenCodePromptFailure | null = null + try { + failure = ( + await this.readFailure({ + dbPath: this.dbPath, + sessionId: this.providerSessionId, + startedAt, + failedAt + }) + ).failure + } catch { + // Native history is best effort; never hide the ACP failure when it cannot be read. + } + if (!failure || this.closed || failure.inputId === this.lastRecoveredInput) { + throw error + } + this.lastRecoveredInput = failure.inputId + throw nativePromptError(failure, error) + } finally { + this.prompting = false + } + } +} diff --git a/src/main/runtime/structured-agent-runtime-registrations.ts b/src/main/runtime/structured-agent-runtime-registrations.ts index d2207587d5c..ed570e9c2cd 100644 --- a/src/main/runtime/structured-agent-runtime-registrations.ts +++ b/src/main/runtime/structured-agent-runtime-registrations.ts @@ -15,6 +15,7 @@ import { createStructuredClaudeRuntimeAdapter } from './structured-claude-runtim import type { AcpLaunchSpec } from '../acp/acp-launch-specs' import { acpStructuredAgentDefinition } from '../acp/acp-structured-agent-definitions' import { createAcpAgentConnection } from '../acp/acp-agent-connection' +import { createOpenCodeAcpConnection } from '../opencode/opencode-acp-connection' import { acpStructuredLaunchPart } from '../acp/acp-structured-launch-resolution' import { AcpStructuredSessionAdapter } from '../acp/acp-structured-session-adapter' import { PI_RPC_RUNTIME_REGISTRATION } from '../pi/rpc-runtime-registration' @@ -173,7 +174,11 @@ function createAcpAdapter( readLaunch: composeStructuredLaunchRead(definition, sources, part), dataDirectory: deps.stateDirectory, resolveLaunchArgs: () => deps.resolveLaunchArgs(spec.agent), - connect: deps.openAcpConnection ?? createAcpAgentConnection, + connect: + deps.openAcpConnection ?? + (spec.agent === 'opencode' || spec.agent === 'opencode2' + ? createOpenCodeAcpConnection + : createAcpAgentConnection), onChildWorkEvidence: (sessionId, evidence, producer) => context.host()?.publishChildWorkEvidence(sessionId, evidence, producer), ...(deps.readProcessStartTime ? { readProcessStartTime: deps.readProcessStartTime } : {}),