diff --git a/.agents/references/scars/full.md b/.agents/references/scars/full.md index de53edc3a224..25f19857bfe3 100644 --- a/.agents/references/scars/full.md +++ b/.agents/references/scars/full.md @@ -69,6 +69,7 @@ ## Provider tools and workspace ownership - Agent CLI stdout is a data boundary: send logs and failures to stderr, use the server-matched launcher rather than ambient PATH, and never infer matching protocol contracts from equal package versions. +- OpenCode SSE connection success is not semantic progress: admit prompts only after the subscription is established, correlate each prompt with a client message ID, and reconcile transcript plus native status until correlated work is idle before projecting one terminal turn event. Replayed evidence must be idempotent, and interrupt/session replacement remains authoritative over delayed recovery results. - Pending CLI approvals/questions must combine `activityContext` with the recent activity window and honor terminal lifecycle events; a request outside the window is not resolved. - Thread history reads must filter and limit in SQL before decoding, omit unrelated checkpoints, and bind pagination cursors to thread/view. Unary RPC deadlines must not cap stream lifetime or imply that timed-out mutations were rejected. - Preserve typed thread-read input failures through HTTP and RPC; missing/ambiguous threads and invalid cursors are client errors, not error-logged repository failures. diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index 686aca7072f0..e43ae8a1f078 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -48,6 +48,8 @@ type MessageEntry = { info: { id: string; role: "user" | "assistant"; + parentID?: string; + error?: unknown; }; parts: Array; }; @@ -65,6 +67,9 @@ const runtimeMock = { promptAsyncError: null as Error | null, closeError: null as Error | null, messages: [] as MessageEntry[], + sessionStatus: "idle" as "busy" | "idle", + subscribeFailures: 0, + subscribeCalls: 0, subscribedEvents: [] as unknown[], subscribedEventDelayMs: 0, sessionParents: new Map(), @@ -99,6 +104,9 @@ const runtimeMock = { this.state.promptAsyncError = null; this.state.closeError = null; this.state.messages = []; + this.state.sessionStatus = "idle"; + this.state.subscribeFailures = 0; + this.state.subscribeCalls = 0; this.state.subscribedEvents = []; this.state.subscribedEventDelayMs = 0; this.state.sessionParents.clear(); @@ -215,6 +223,7 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { }, abort: async ({ sessionID }: { sessionID: string }) => { runtimeMock.state.abortCalls.push(sessionID); + runtimeMock.state.sessionStatus = "idle"; if (runtimeMock.state.abortError) { throw runtimeMock.state.abortError; } @@ -228,8 +237,14 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { if (runtimeMock.state.promptAsyncError) { throw runtimeMock.state.promptAsyncError; } + runtimeMock.state.sessionStatus = "busy"; }, messages: async () => ({ data: runtimeMock.state.messages }), + status: async () => ({ + data: { + "http://127.0.0.1:9999/session": { type: runtimeMock.state.sessionStatus }, + }, + }), revert: async ({ sessionID, messageID }: { sessionID: string; messageID?: string }) => { runtimeMock.state.revertCalls.push({ sessionID, @@ -250,18 +265,25 @@ const OpenCodeRuntimeTestDouble: OpenCodeRuntimeShape = { }, }, event: { - subscribe: async () => ({ - stream: (async function* () { - for (const event of runtimeMock.state.subscribedEvents) { - if (runtimeMock.state.subscribedEventDelayMs > 0) { - await new Promise((resolve) => - setTimeout(resolve, runtimeMock.state.subscribedEventDelayMs), - ); + subscribe: async () => { + runtimeMock.state.subscribeCalls += 1; + if (runtimeMock.state.subscribeFailures > 0) { + runtimeMock.state.subscribeFailures -= 1; + throw new Error("event stream disconnected"); + } + return { + stream: (async function* () { + for (const event of runtimeMock.state.subscribedEvents) { + if (runtimeMock.state.subscribedEventDelayMs > 0) { + await new Promise((resolve) => + setTimeout(resolve, runtimeMock.state.subscribedEventDelayMs), + ); + } + yield event; } - yield event; - } - })(), - }), + })(), + }; + }, }, permission: { list: async () => { @@ -1575,7 +1597,10 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { ), }); - assert.deepEqual(runtimeMock.state.promptCalls.at(-1), { + const promptCall = { ...(runtimeMock.state.promptCalls.at(-1) as Record) }; + assert.equal(typeof promptCall.messageID, "string"); + delete promptCall.messageID; + assert.deepEqual(promptCall, { sessionID: "http://127.0.0.1:9999/session", model: { providerID: "anthropic", @@ -1607,7 +1632,10 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { ]), }); - assert.deepEqual(runtimeMock.state.promptCalls.at(-1), { + const promptCall = { ...(runtimeMock.state.promptCalls.at(-1) as Record) }; + assert.equal(typeof promptCall.messageID, "string"); + delete promptCall.messageID; + assert.deepEqual(promptCall, { sessionID: "http://127.0.0.1:9999/session", model: { providerID: "openai", @@ -1653,7 +1681,10 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { input: "Fix it", }); - assert.deepEqual(runtimeMock.state.promptCalls.at(-1), { + const promptCall = { ...(runtimeMock.state.promptCalls.at(-1) as Record) }; + assert.equal(typeof promptCall.messageID, "string"); + delete promptCall.messageID; + assert.deepEqual(promptCall, { sessionID: "http://127.0.0.1:9999/session", model: { providerID: "anthropic", @@ -1664,6 +1695,131 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }).pipe(Effect.provide(adapterLayer)); }); + it.effect("reconciles a completed prompt when SSE has no semantic events", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-silent-sse"); + const observed = yield* adapter.streamEvents.pipe( + Stream.filter((event) => event.threadId === threadId), + Stream.takeUntil((event) => event.type === "turn.completed"), + Stream.runCollect, + Effect.forkChild, + ); + + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + const turn = yield* adapter.sendTurn({ + threadId, + input: "Recover this silent turn", + modelSelection: createModelSelection(ProviderInstanceId.make("opencode"), "openai/gpt-5"), + }); + const prompt = runtimeMock.state.promptCalls.at(-1) as { messageID: string }; + + yield* sleep(150); + runtimeMock.state.messages = [ + { + info: { id: prompt.messageID, role: "user" }, + parts: [{ id: "user-part", type: "text", messageID: prompt.messageID, text: "Recover" }], + }, + { + info: { + id: "assistant-silent", + role: "assistant", + parentID: prompt.messageID, + }, + parts: [ + { + id: "assistant-part", + messageID: "assistant-silent", + type: "text", + text: "Recovered output", + time: { start: 1, end: 2 }, + }, + ], + }, + ]; + runtimeMock.state.sessionStatus = "idle"; + + const events = Array.from(yield* Fiber.join(observed).pipe(Effect.timeout("2 seconds"))); + const completed = events.filter( + (event) => event.type === "turn.completed" && event.turnId === turn.turnId, + ); + assert.equal(completed.length, 1); + assert.deepEqual( + events + .filter((event) => event.type === "content.delta") + .map((event) => (event.type === "content.delta" ? event.payload.delta : "")), + ["Recovered output"], + ); + }), + ); + + it.effect("waits for an event reconnect before admitting a prompt", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-reconnect"); + runtimeMock.state.subscribeFailures = 1; + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + assert.ok(runtimeMock.state.subscribeCalls >= 2); + yield* adapter.stopSession(threadId); + }), + ); + + it.effect("bounds initial event connection retries", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-connection-timeout"); + runtimeMock.state.subscribeFailures = 100; + + const error = yield* adapter + .startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }) + .pipe(Effect.flip); + + assert.equal(error._tag, "ProviderAdapterProcessError"); + assert.ok(runtimeMock.state.subscribeCalls >= 5); + assert.equal(runtimeMock.state.abortCalls.length, 1); + }), + ); + + it.effect("rejects a follow-up when the native session remains busy", () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-native-busy-follow-up"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + yield* adapter.sendTurn({ + threadId, + input: "First turn", + modelSelection: createModelSelection(ProviderInstanceId.make("opencode"), "openai/gpt-5"), + }); + const error = yield* adapter + .sendTurn({ + threadId, + input: "Follow-up", + modelSelection: createModelSelection(ProviderInstanceId.make("opencode"), "openai/gpt-5"), + }) + .pipe(Effect.flip); + assert.equal(error._tag, "ProviderAdapterRequestError"); + if (error._tag === "ProviderAdapterRequestError") { + assert.match(error.detail, /active turn|busy/); + } + }), + ); + it.effect("rejects sendTurn model selections for another instance id", () => { const customInstanceId = ProviderInstanceId.make("opencode_zen"); const adapterLayer = Layer.effect( diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 1c2a7a9845c5..7b5ff9222909 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -14,9 +14,11 @@ import { } from "@t3tools/contracts"; import { Cause, + Deferred, Effect, Exit, FileSystem, + Fiber, Path, Queue, Ref, @@ -55,6 +57,20 @@ import { } from "../opencodeRuntime.ts"; const PROVIDER = ProviderDriverKind.make("opencode"); +const OPENCODE_CONNECTION_TIMEOUT = "5 seconds"; +const OPENCODE_INITIAL_SUBSCRIBE_ATTEMPTS = 5; +const OPENCODE_RECOVERY_ATTEMPTS = 24; +const OPENCODE_RECOVERY_DELAY_MS = 100; +const OPENCODE_ADMISSION_ATTEMPTS = 5; +const OPENCODE_ADMISSION_DELAY_MS = 100; + +const sleepOpenCode = (milliseconds: number) => + Effect.promise( + () => + new Promise((resolve) => { + setTimeout(resolve, milliseconds); + }), + ); /** * Version tag stamped into the OpenCode resume cursor. Bump if the cursor @@ -197,8 +213,14 @@ interface OpenCodeSessionContext { // until native removal or session teardown, but do not retain other part payloads. readonly textPartsByMessageId: Map>; activeTurnId: TurnId | undefined; + activePromptMessageId: string | undefined; activeAgent: string | undefined; activeVariant: string | undefined; + readonly promptAdmission: Semaphore.Semaphore; + readonly connectionReady: Deferred.Deferred; + readonly emittedToolStates: Map; + readonly terminalTurnIds: Set; + recoveryFiber: Fiber.Fiber | undefined; /** * One-shot guard flipped by `stopOpenCodeContext` / `emitUnexpectedExit`. * The session lifecycle is owned by `sessionScope`; this Ref exists only @@ -851,6 +873,12 @@ export function makeOpenCodeAdapter( return true; }); + let reconcileTurn: ( + context: OpenCodeSessionContext, + turnId: TurnId, + promptMessageId: string, + ) => Effect.Effect; + const handleSubscribedEvent = Effect.fn("handleSubscribedEvent")(function* ( context: OpenCodeSessionContext, event: OpenCodeSubscribedEvent, @@ -970,6 +998,10 @@ export function makeOpenCodeAdapter( } if (part.type === "tool") { + if (context.emittedToolStates.get(part.id) === part.state.status) { + break; + } + context.emittedToolStates.set(part.id, part.state.status); const itemType = toToolLifecycleItemType(part.tool); const title = part.state.status === "running" ? (part.state.title ?? part.tool) : part.tool; @@ -1184,20 +1216,12 @@ export function makeOpenCodeAdapter( turnId && !context.awaitingBusyAfterInterruption ) { - context.activeTurnId = undefined; - context.interruptedTurnId = undefined; - updateProviderSession(context, { status: "ready" }, { clearActiveTurnId: true }); - yield* emit({ - ...(yield* buildEventBase({ - threadId: context.session.threadId, - turnId, - raw: event, - })), - type: "turn.completed", - payload: { - state: "completed", - }, - }); + const promptMessageId = context.activePromptMessageId; + if (promptMessageId !== undefined && context.recoveryFiber === undefined) { + context.recoveryFiber = yield* reconcileTurn(context, turnId, promptMessageId).pipe( + Effect.forkIn(context.sessionScope), + ); + } } break; } @@ -1212,28 +1236,8 @@ export function makeOpenCodeAdapter( } const message = sessionErrorMessage(event.properties.error); const activeTurnId = context.activeTurnId; - context.activeTurnId = undefined; - updateProviderSession( - context, - { - status: "error", - lastError: message, - }, - { clearActiveTurnId: true }, - ); if (activeTurnId) { - yield* emit({ - ...(yield* buildEventBase({ - threadId: context.session.threadId, - turnId: activeTurnId, - raw: event, - })), - type: "turn.completed", - payload: { - state: "failed", - errorMessage: message, - }, - }); + yield* finishTurn(context, activeTurnId, "failed", message); } yield* emit({ ...(yield* buildEventBase({ @@ -1255,6 +1259,184 @@ export function makeOpenCodeAdapter( } }); + const finishTurn = Effect.fn("finishTurn")(function* ( + context: OpenCodeSessionContext, + turnId: TurnId, + state: "completed" | "failed", + errorMessage?: string, + ) { + if (context.terminalTurnIds.has(turnId)) { + return; + } + context.terminalTurnIds.add(turnId); + if (context.activeTurnId !== turnId) { + return; + } + context.activeTurnId = undefined; + context.activePromptMessageId = undefined; + context.activeAgent = undefined; + context.activeVariant = undefined; + updateProviderSession( + context, + { + status: state === "completed" ? "ready" : "error", + ...(errorMessage ? { lastError: errorMessage } : {}), + }, + { clearActiveTurnId: true, ...(state === "completed" ? { clearLastError: true } : {}) }, + ); + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + })), + type: "turn.completed", + payload: { + state, + ...(errorMessage ? { errorMessage } : {}), + }, + }); + }); + + reconcileTurn = Effect.fn("reconcileTurn")(function* ( + context: OpenCodeSessionContext, + turnId: TurnId, + promptMessageId: string, + ) { + let lastFailure: string | undefined; + for (let attempt = 0; attempt < OPENCODE_RECOVERY_ATTEMPTS; attempt += 1) { + if ( + context.activeTurnId !== turnId || + context.interruptedTurnId === turnId || + (yield* Ref.get(context.stopped)) + ) { + return; + } + + const result = yield* Effect.all( + { + status: runOpenCodeSdk("session.status", () => context.client.session.status()), + messages: runOpenCodeSdk("session.messages", () => + context.client.session.messages({ sessionID: context.openCodeSessionId }), + ), + }, + { concurrency: 2 }, + ).pipe(Effect.result); + + if (result._tag === "Failure") { + lastFailure = openCodeRuntimeErrorDetail(result.failure); + yield* sleepOpenCode(OPENCODE_RECOVERY_DELAY_MS); + continue; + } + + const status = result.success.status.data?.[context.openCodeSessionId]; + const messages = result.success.messages.data ?? []; + for (const entry of messages) { + const info = entry.info as { + readonly id: string; + readonly role: "user" | "assistant"; + readonly parentID?: string; + readonly error?: unknown; + }; + context.messageRoleById.set(info.id, info.role); + yield* handleSubscribedEvent(context, { + type: "message.updated", + properties: { + sessionID: context.openCodeSessionId, + info: entry.info, + }, + } as OpenCodeSubscribedEvent); + for (const part of entry.parts) { + yield* handleSubscribedEvent(context, { + type: "message.part.updated", + properties: { + sessionID: context.openCodeSessionId, + part, + time: Date.now(), + }, + } as OpenCodeSubscribedEvent); + } + } + + if (status?.type === "retry") { + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + })), + type: "runtime.warning", + payload: { + message: status.message, + detail: status, + }, + }); + } + + const promptIndex = messages.findIndex( + (candidate) => candidate.info.id === promptMessageId, + ); + const assistant = + promptIndex >= 0 + ? messages.find( + (entry) => + entry.info.role === "assistant" && + ((entry.info as { readonly parentID?: string }).parentID === promptMessageId || + messages.findIndex((candidate) => candidate.info.id === entry.info.id) > + promptIndex), + ) + : undefined; + if (assistant?.info.role === "assistant" && assistant.info.error !== undefined) { + yield* finishTurn(context, turnId, "failed", sessionErrorMessage(assistant.info.error)); + return; + } + if (status?.type === "idle" && assistant !== undefined) { + yield* finishTurn(context, turnId, "completed"); + return; + } + + yield* sleepOpenCode(OPENCODE_RECOVERY_DELAY_MS); + } + + const message = + lastFailure ?? + "OpenCode accepted the prompt but did not provide correlated transcript and idle evidence."; + yield* emit({ + ...(yield* buildEventBase({ + threadId: context.session.threadId, + turnId, + })), + type: "runtime.warning", + payload: { + message: "OpenCode turn recovery is degraded.", + detail: message, + }, + }); + yield* finishTurn(context, turnId, "failed", message); + }); + + const recoverPromptAdmission = Effect.fn("recoverPromptAdmission")(function* ( + context: OpenCodeSessionContext, + promptMessageId: string, + ) { + for (let attempt = 0; attempt < OPENCODE_ADMISSION_ATTEMPTS; attempt += 1) { + const messages = yield* runOpenCodeSdk("session.messages", () => + context.client.session.messages({ sessionID: context.openCodeSessionId }), + ).pipe(Effect.result); + if ( + messages._tag === "Success" && + (messages.success.data ?? []).some((entry) => entry.info.id === promptMessageId) + ) { + return true; + } + yield* Effect.promise( + () => + new Promise((resolve) => { + setTimeout(resolve, OPENCODE_ADMISSION_DELAY_MS); + }), + ); + } + return false; + }); + const startEventPump = Effect.fn("startEventPump")(function* (context: OpenCodeSessionContext) { // One AbortController per session scope. The finalizer fires when // the scope closes (explicit stop, unexpected exit, or layer @@ -1266,43 +1448,130 @@ export function makeOpenCodeAdapter( Effect.sync(() => eventsAbortController.abort()), ); - // Fibers forked into `context.sessionScope` are interrupted - // automatically when the scope closes — no bookkeeping required. - yield* Effect.flatMap( - runOpenCodeSdk("event.subscribe", () => - context.client.event.subscribe(undefined, { - signal: eventsAbortController.signal, - }), - ), - (subscription) => - Stream.fromAsyncIterable( - subscription.stream, - (cause) => - new OpenCodeRuntimeError({ - operation: "event.subscribe", - detail: openCodeRuntimeErrorDetail(cause), - cause, - }), - ).pipe(Stream.runForEach((event) => handleSubscribedEvent(context, event))), - ).pipe( - Effect.exit, - Effect.flatMap((exit) => - Effect.gen(function* () { - // Expected paths: caller aborted the fetch or the session - // has already been marked stopped. Treat as a clean exit. - if (eventsAbortController.signal.aborted || (yield* Ref.get(context.stopped))) { + // Keep the native subscription alive across transport loss. The first + // successful subscribe is the admission barrier; later reconnects are + // followed by transcript/status reconciliation by the active-turn + // watchdog. + let initialSubscription: + | Awaited> + | undefined; + let initialDelayMs = 100; + for ( + let attempt = 0; + attempt < OPENCODE_INITIAL_SUBSCRIBE_ATTEMPTS && initialSubscription === undefined; + attempt += 1 + ) { + if (yield* Ref.get(context.stopped) || eventsAbortController.signal.aborted) { + return yield* new OpenCodeRuntimeError({ + operation: "event.subscribe", + detail: "OpenCode event connection admission was cancelled.", + }); + } + const initialExit = yield* Effect.exit( + runOpenCodeSdk("event.subscribe", () => + context.client.event.subscribe(undefined, { + signal: eventsAbortController.signal, + }), + ), + ); + if (Exit.isSuccess(initialExit)) { + initialSubscription = initialExit.value; + break; + } + if (attempt + 1 >= OPENCODE_INITIAL_SUBSCRIBE_ATTEMPTS) { + return yield* new OpenCodeRuntimeError({ + operation: "event.subscribe", + detail: `OpenCode event connection was not established after ${OPENCODE_INITIAL_SUBSCRIBE_ATTEMPTS} attempts: ${openCodeRuntimeErrorDetail(Cause.squash(initialExit.cause))}`, + cause: Cause.squash(initialExit.cause), + }); + } + yield* sleepOpenCode(initialDelayMs); + initialDelayMs = Math.min(initialDelayMs * 2, 2_000); + } + if (initialSubscription === undefined) { + return yield* new OpenCodeRuntimeError({ + operation: "event.subscribe", + detail: "OpenCode event connection admission did not produce a subscription.", + }); + } + yield* Deferred.succeed(context.connectionReady, undefined); + yield* Effect.gen(function* () { + let subscription = initialSubscription; + let reconnectDelayMs = 100; + while (!(yield* Ref.get(context.stopped)) && !eventsAbortController.signal.aborted) { + const exit = yield* Effect.exit( + Stream.fromAsyncIterable( + subscription.stream, + (cause) => + new OpenCodeRuntimeError({ + operation: "event.subscribe", + detail: openCodeRuntimeErrorDetail(cause), + cause, + }), + ).pipe( + Stream.runForEach((event) => + handleSubscribedEvent(context, event).pipe( + Effect.catchCause((cause) => + emitUnexpectedExit( + context, + openCodeRuntimeErrorDetail(Cause.squash(cause)), + ).pipe(Effect.ignore), + ), + ), + ), + ), + ); + if (yield* Ref.get(context.stopped) || eventsAbortController.signal.aborted) { + return; + } + yield* Effect.logWarning( + Exit.isSuccess(exit) + ? "OpenCode event stream ended; reconnecting." + : "OpenCode event stream disconnected; reconnecting.", + { + threadId: context.session.threadId, + ...(Exit.isFailure(exit) + ? { detail: openCodeRuntimeErrorDetail(Cause.squash(exit.cause)) } + : {}), + }, + ); + + let nextSubscription: + | Awaited> + | undefined; + while ( + nextSubscription === undefined && + !(yield* Ref.get(context.stopped)) && + !eventsAbortController.signal.aborted + ) { + yield* sleepOpenCode(reconnectDelayMs); + if (yield* Ref.get(context.stopped) || eventsAbortController.signal.aborted) { return; } - if (Exit.isFailure(exit)) { - yield* emitUnexpectedExit( - context, - openCodeRuntimeErrorDetail(Cause.squash(exit.cause)), - ); + const nextExit = yield* Effect.exit( + runOpenCodeSdk("event.subscribe", () => + context.client.event.subscribe(undefined, { + signal: eventsAbortController.signal, + }), + ), + ); + if (Exit.isSuccess(nextExit)) { + nextSubscription = nextExit.value; + reconnectDelayMs = 100; + break; } - }), - ), - Effect.forkIn(context.sessionScope), - ); + yield* Effect.logWarning("OpenCode event resubscribe failed; retrying.", { + threadId: context.session.threadId, + detail: openCodeRuntimeErrorDetail(Cause.squash(nextExit.cause)), + }); + reconnectDelayMs = Math.min(reconnectDelayMs * 2, 2_000); + } + if (nextSubscription === undefined) { + return; + } + subscription = nextSubscription; + } + }).pipe(Effect.forkIn(context.sessionScope)); if (!context.server.external && context.server.exitCode !== null) { yield* context.server.exitCode.pipe( @@ -1594,13 +1863,44 @@ export function makeOpenCodeAdapter( textPartsByMessageId: new Map(), messageRoleById: new Map(), activeTurnId: undefined, + activePromptMessageId: undefined, activeAgent: undefined, activeVariant: undefined, + promptAdmission: Semaphore.makeUnsafe(1), + connectionReady: yield* Deferred.make(), + emittedToolStates: new Map(), + terminalTurnIds: new Set(), + recoveryFiber: undefined, stopped: yield* Ref.make(false), sessionScope: started.sessionScope, }; sessions.set(input.threadId, context); - yield* startEventPump(context); + yield* startEventPump(context).pipe( + Effect.catchTag("OpenCodeRuntimeError", (cause) => + Effect.gen(function* () { + sessions.delete(input.threadId); + yield* stopOpenCodeContext(context); + return yield* new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: `OpenCode event connection was not established: ${cause.detail}`, + cause, + }); + }), + ), + ); + yield* Deferred.await(context.connectionReady).pipe( + Effect.timeout(OPENCODE_CONNECTION_TIMEOUT), + Effect.mapError( + (cause) => + new ProviderAdapterProcessError({ + provider: PROVIDER, + threadId: input.threadId, + detail: `OpenCode event connection was not established: ${openCodeRuntimeErrorDetail(cause)}`, + cause, + }), + ), + ); yield* recoverPendingRequests(context); yield* emit({ @@ -1624,7 +1924,15 @@ export function makeOpenCodeAdapter( const sendTurn: OpenCodeAdapterShape["sendTurn"] = Effect.fn("sendTurn")(function* (input) { const context = ensureSessionContext(sessions, input.threadId); + if (context.activeTurnId !== undefined) { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session.promptAsync", + detail: "OpenCode session is still busy with an active turn.", + }); + } const turnId = TurnId.make(`opencode-turn-${crypto.randomUUID()}`); + const promptMessageId = `t3-${crypto.randomUUID()}`; const modelSelection = input.modelSelection ?? (context.session.model @@ -1667,45 +1975,73 @@ export function makeOpenCodeAdapter( const agent = selectedAgent === "plan" ? undefined : selectedAgent; const variant = getModelSelectionStringOptionValue(modelSelection, "variant"); - context.activeTurnId = turnId; - context.activeAgent = agent; - context.activeVariant = variant; - updateProviderSession( - context, - { - status: "running", - activeTurnId: turnId, - model: modelSelection?.model ?? context.session.model, - }, - { clearLastError: true }, - ); + let lifecycleStarted = false; + const promptRequest = Effect.gen(function* () { + const status = yield* runOpenCodeSdk("session.status", () => + context.client.session.status(), + ); + const nativeStatus = status.data?.[context.openCodeSessionId]; + if (nativeStatus?.type === "busy") { + return yield* new ProviderAdapterRequestError({ + provider: PROVIDER, + method: "session.promptAsync", + detail: "OpenCode session is still busy with an active native turn.", + }); + } + context.activeTurnId = turnId; + context.activePromptMessageId = promptMessageId; + context.awaitingBusyAfterInterruption = false; + context.activeAgent = agent; + context.activeVariant = variant; + updateProviderSession( + context, + { + status: "running", + activeTurnId: turnId, + model: modelSelection?.model ?? context.session.model, + }, + { clearLastError: true }, + ); - yield* emit({ - ...(yield* buildEventBase({ threadId: input.threadId, turnId })), - type: "turn.started", - payload: { - model: modelSelection?.model ?? context.session.model, - ...(variant ? { effort: variant } : {}), - }, + yield* emit({ + ...(yield* buildEventBase({ threadId: input.threadId, turnId })), + type: "turn.started", + payload: { + model: modelSelection?.model ?? context.session.model, + ...(variant ? { effort: variant } : {}), + }, + }); + lifecycleStarted = true; + yield* runOpenCodeSdk("session.promptAsync", () => + context.client.session.promptAsync({ + sessionID: context.openCodeSessionId, + messageID: promptMessageId, + model: parsedModel, + ...(context.activeAgent ? { agent: context.activeAgent } : {}), + ...(context.activeVariant ? { variant: context.activeVariant } : {}), + parts: [...(text ? [{ type: "text" as const, text }] : []), ...fileParts], + }), + ).pipe( + Effect.catchTag("OpenCodeRuntimeError", (error) => + recoverPromptAdmission(context, promptMessageId).pipe( + Effect.flatMap((admitted) => (admitted ? Effect.void : Effect.fail(error))), + ), + ), + ); }); - - yield* runOpenCodeSdk("session.promptAsync", () => - context.client.session.promptAsync({ - sessionID: context.openCodeSessionId, - model: parsedModel, - ...(context.activeAgent ? { agent: context.activeAgent } : {}), - ...(context.activeVariant ? { variant: context.activeVariant } : {}), - parts: [...(text ? [{ type: "text" as const, text }] : []), ...fileParts], - }), - ).pipe( - Effect.mapError(toRequestError), + yield* context.promptAdmission.withPermit(promptRequest).pipe( + Effect.catchTag("OpenCodeRuntimeError", (cause) => Effect.fail(toRequestError(cause))), // On failure: clear active-turn state, flip the session back to ready // with lastError set, emit turn.aborted, then let the typed error // propagate. We don't need to rebuild the error here — `toRequestError` // already produced the right shape. Effect.tapError((requestError) => Effect.gen(function* () { + if (!lifecycleStarted) { + return; + } context.activeTurnId = undefined; + context.activePromptMessageId = undefined; context.activeAgent = undefined; context.activeVariant = undefined; updateProviderSession( @@ -1730,6 +2066,9 @@ export function makeOpenCodeAdapter( }), ), ); + context.recoveryFiber = yield* reconcileTurn(context, turnId, promptMessageId).pipe( + Effect.forkIn(context.sessionScope), + ); return { threadId: input.threadId, @@ -1762,6 +2101,7 @@ export function makeOpenCodeAdapter( ); yield* abortOpenCodeDescendants(context).pipe(Effect.mapError(toRequestError)); context.activeTurnId = undefined; + context.activePromptMessageId = undefined; context.interruptedTurnId = interruptedTurnId; updateProviderSession(context, { status: "ready" }, { clearActiveTurnId: true }); if (interruptedTurnId) { diff --git a/apps/server/src/provider/opencodeRuntime.ts b/apps/server/src/provider/opencodeRuntime.ts index 36678b110f27..3e39756017f8 100644 --- a/apps/server/src/provider/opencodeRuntime.ts +++ b/apps/server/src/provider/opencodeRuntime.ts @@ -79,7 +79,11 @@ function nestedErrorCause(cause: unknown): unknown { export function openCodeRuntimeErrorDetail(cause: unknown): string { if (OpenCodeRuntimeError.is(cause)) return cause.detail; - if (cause instanceof Error && cause.message.trim().length > 0) { + if ( + cause instanceof Error && + typeof cause.message === "string" && + cause.message.trim().length > 0 + ) { const message = cause.message.trim(); if (message.includes("ChildProcess.exitCode")) { const nestedCause = nestedErrorCause(cause);