From 49dd2ca29d110b08d007b3bf8acf79ecd47159b4 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Sat, 24 Oct 2026 08:35:38 -0500 Subject: [PATCH 1/8] fix(opencode): a failed steer no longer leaves the turn running forever A steer whose prompt request failed kept the turn's admission pending, so every later idle was held for a prompt OpenCode never received. Clear the admission on failure and end the turn if the session already went idle. Co-Authored-By: Claude Opus 5.5 --- .../src/server/adapter.test.ts | 77 +++++++++++++++++++ .../provider-opencode/src/server/adapter.ts | 9 +++ 2 files changed, 86 insertions(+) diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index fef2c2ca1f96..a4cd160d42d5 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -1630,6 +1630,83 @@ describe("OpenCodeAdapterV2", () => { ), ); + it.effect.each(["during", "after"] as const)( + "completes the turn when the session goes idle %s a rejected steer", + (idleTiming) => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const sessionId = `native-steer-rejected-${idleTiming}`; + const idle = { + type: "session.status", + properties: { sessionID: sessionId, status: { type: "idle" } }, + }; + const promptIds: Array = []; + const harness = yield* makeOpenCodeRuntimeHarness( + `steer-rejected-${idleTiming}`, + sessionId, + { + event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, + session: { + create: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), + get: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), + messages: async () => ({ data: [] }), + promptAsync: async (input: { messageID: string }) => { + promptIds.push(input.messageID); + if (promptIds.length === 1) return { data: true }; + if (idleTiming === "during") await nativeEvents.push(idle); + throw new Error("steer rejected"); + }, + abort: async () => ({ data: true }), + children: async () => ({ data: [] }), + }, + }, + ); + yield* harness.startTurn(); + yield* Effect.promise(() => + nativeEvents.push({ + type: "message.updated", + properties: { + sessionID: sessionId, + info: { + id: promptIds[0], + sessionID: sessionId, + role: "user", + time: { created: DateTime.toEpochMillis(harness.now) }, + }, + }, + }), + ); + const running = yield* harness.runtime.readThreadSnapshot({ + providerThread: harness.providerThread, + }); + const steer = yield* Effect.exit( + harness.runtime.steerTurn({ + threadId: harness.threadId, + runId: harness.runId, + providerThread: harness.providerThread, + providerTurnId: running.providerTurns.at(-1)!.id, + message: { + messageId: MessageId.make(`message-steer-rejected-${idleTiming}`), + text: "also cover the README", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + }), + ); + assert.isTrue(Exit.isFailure(steer)); + if (idleTiming === "after") yield* Effect.promise(() => nativeEvents.push(idle)); + + const settled = yield* harness.runtime.readThreadSnapshot({ + providerThread: harness.providerThread, + }); + assert.equal(settled.providerTurns.at(-1)?.status, "completed"); + }).pipe( + Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer)), + Effect.scoped, + ), + ); + it.effect("interrupts an initial prompt while its SDK request is pending", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index 9a6a6e265b20..f54f711e8241 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -3356,6 +3356,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function turn.usage.promptMessageIds.add(turn.admissionMessageId); const admissionSettled = turn.admissionSettled; const admissionAbortController = turn.admissionAbortController; + const generation = turn.admissionGeneration; yield* submitPrompt( state, turn, @@ -3367,6 +3368,14 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function }, admissionAbortController!, ).pipe( + // A rejected steer never runs, so idle no longer waits for it. The + // turn ends now if idle already arrived, or on the next idle. + Effect.tapError(() => { + if (!promptAdmissionIsCurrent(state, turn, generation)) return Effect.void; + const idleDuringAdmission = turn.idleDuringAdmission; + cancelOpenCodePromptAdmission(turn, state.nextAdmissionGeneration++); + return idleDuringAdmission ? finalizeTurn(state, turn, "completed") : Effect.void; + }), Effect.ensuring( Effect.all([ Deferred.succeed(admissionSettled, undefined).pipe(Effect.ignore), From 784e36bf0e2e2809f8b4f4d359b4b708dbd29ea9 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 13:30:16 +0000 Subject: [PATCH 2/8] fix(opencode): reconcile current status after a rejected steer --- packages/provider-opencode/src/server/adapter.ts | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index 31b417ded71c..88fb0d307c10 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -3349,14 +3349,11 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function }, admissionAbortController!, ).pipe( - // A rejected steer never runs, so idle no longer waits for it. The - // turn ends now if idle already arrived, or on the next idle. - Effect.tapError(() => { - if (!promptAdmissionIsCurrent(state, turn, generation)) return Effect.void; - const idleDuringAdmission = turn.idleDuringAdmission; - cancelOpenCodePromptAdmission(turn, state.nextAdmissionGeneration++); - return idleDuringAdmission ? finalizeTurn(state, turn, "completed") : Effect.void; - }), + Effect.tapError(() => + promptAdmissionIsCurrent(state, turn, generation) + ? reconcilePromptAdmission(state, turn) + : Effect.void, + ), Effect.ensuring( Effect.all([ Deferred.succeed(admissionSettled, undefined).pipe(Effect.ignore), From a80aa2c12594f1b0e7af2a4753de599cb0708ce4 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 13:35:16 +0000 Subject: [PATCH 3/8] fix(opencode): recheck idle received during admission reconciliation --- .../src/server/adapter.test.ts | 231 +++++++++--------- .../provider-opencode/src/server/adapter.ts | 7 +- 2 files changed, 126 insertions(+), 112 deletions(-) diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index dac708a19db9..0bb2329c786e 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -1721,125 +1721,134 @@ describe("OpenCodeAdapterV2", () => { ), ); - it.effect.each(["during", "after", "stale-idle-then-busy", "no-idle"] as const)( - "completes the turn when the session goes idle %s a rejected steer", - (idleTiming) => - Effect.gen(function* () { - const nativeEvents = asyncEventStream(); - const sessionId = `native-steer-rejected-${idleTiming}`; - const idle = { - type: "session.status", - properties: { sessionID: sessionId, status: { type: "idle" } }, - }; - const promptIds: Array = []; - const terminal = yield* Deferred.make(); - const statusRead = promiseGate(); - let statusCalls = 0; - const harness = yield* makeOpenCodeRuntimeHarness( - `steer-rejected-${idleTiming}`, - sessionId, - { - event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, - session: { - create: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), - get: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), - messages: async () => ({ data: [] }), - status: async () => { - statusCalls++; - statusRead.resolve(); - return { - data: { - [sessionId]: { - type: idleTiming === "during" || idleTiming === "no-idle" ? "idle" : "busy", - }, - }, - }; - }, - promptAsync: async (input: { messageID: string }) => { - promptIds.push(input.messageID); - if (promptIds.length === 1) return { data: true }; - if (idleTiming === "during" || idleTiming === "stale-idle-then-busy") { - await nativeEvents.push(idle); - } - if (idleTiming === "stale-idle-then-busy") { - await nativeEvents.push({ - type: "session.status", - properties: { sessionID: sessionId, status: { type: "busy" } }, - }); - } - throw new Error("steer rejected"); + it.effect.each([ + "during", + "after", + "stale-idle-then-busy", + "no-idle", + "idle-during-status", + ] as const)("completes the turn when the session goes idle %s a rejected steer", (idleTiming) => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const sessionId = `native-steer-rejected-${idleTiming}`; + const idle = { + type: "session.status", + properties: { sessionID: sessionId, status: { type: "idle" } }, + }; + const promptIds: Array = []; + const terminal = yield* Deferred.make(); + const statusRead = promiseGate(); + let statusCalls = 0; + const harness = yield* makeOpenCodeRuntimeHarness(`steer-rejected-${idleTiming}`, sessionId, { + event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, + session: { + create: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), + get: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), + messages: async () => ({ data: [] }), + status: async () => { + statusCalls++; + statusRead.resolve(); + if (idleTiming === "idle-during-status" && statusCalls === 1) { + await nativeEvents.push(idle); + return { data: { [sessionId]: { type: "busy" } } }; + } + return { + data: { + [sessionId]: { + type: + idleTiming === "during" || + idleTiming === "no-idle" || + idleTiming === "idle-during-status" + ? "idle" + : "busy", + }, }, - abort: async () => ({ data: true }), - children: async () => ({ data: [] }), - }, + }; }, - ); - yield* harness.startTurn(); - yield* Effect.promise(() => - nativeEvents.push({ - type: "message.updated", - properties: { + promptAsync: async (input: { messageID: string }) => { + promptIds.push(input.messageID); + if (promptIds.length === 1) return { data: true }; + if (idleTiming === "during" || idleTiming === "stale-idle-then-busy") { + await nativeEvents.push(idle); + } + if (idleTiming === "stale-idle-then-busy") { + await nativeEvents.push({ + type: "session.status", + properties: { sessionID: sessionId, status: { type: "busy" } }, + }); + } + throw new Error("steer rejected"); + }, + abort: async () => ({ data: true }), + children: async () => ({ data: [] }), + }, + }); + yield* harness.startTurn(); + yield* Effect.promise(() => + nativeEvents.push({ + type: "message.updated", + properties: { + sessionID: sessionId, + info: { + id: promptIds[0], sessionID: sessionId, - info: { - id: promptIds[0], - sessionID: sessionId, - role: "user", - time: { created: DateTime.toEpochMillis(harness.now) }, - }, + role: "user", + time: { created: DateTime.toEpochMillis(harness.now) }, }, - }), - ); - const running = yield* harness.runtime.readThreadSnapshot({ + }, + }), + ); + const running = yield* harness.runtime.readThreadSnapshot({ + providerThread: harness.providerThread, + }); + yield* harness.runtime.events.pipe( + Stream.tap((event) => + event.type === "turn.terminal" ? Deferred.succeed(terminal, undefined) : Effect.void, + ), + Stream.runDrain, + Effect.forkScoped, + ); + const steer = yield* Effect.exit( + harness.runtime.steerTurn({ + threadId: harness.threadId, + runId: harness.runId, + providerThread: harness.providerThread, + providerTurnId: running.providerTurns.at(-1)!.id, + message: { + messageId: MessageId.make(`message-steer-rejected-${idleTiming}`), + text: "also cover the README", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + }), + ); + assert.isTrue(Exit.isFailure(steer)); + if (idleTiming === "stale-idle-then-busy") { + const active = yield* harness.runtime.readThreadSnapshot({ providerThread: harness.providerThread, }); - yield* harness.runtime.events.pipe( - Stream.tap((event) => - event.type === "turn.terminal" ? Deferred.succeed(terminal, undefined) : Effect.void, - ), - Stream.runDrain, - Effect.forkScoped, - ); - const steer = yield* Effect.exit( - harness.runtime.steerTurn({ - threadId: harness.threadId, - runId: harness.runId, - providerThread: harness.providerThread, - providerTurnId: running.providerTurns.at(-1)!.id, - message: { - messageId: MessageId.make(`message-steer-rejected-${idleTiming}`), - text: "also cover the README", - attachments: [], - createdBy: "user", - creationSource: "web", - }, - }), - ); - assert.isTrue(Exit.isFailure(steer)); - if (idleTiming === "stale-idle-then-busy") { - const active = yield* harness.runtime.readThreadSnapshot({ - providerThread: harness.providerThread, - }); - assert.equal(active.providerTurns.at(-1)?.status, "running"); - } - yield* Effect.promise(() => statusRead.promise); - if (idleTiming === "after" || idleTiming === "stale-idle-then-busy") { - const active = yield* harness.runtime.readThreadSnapshot({ - providerThread: harness.providerThread, - }); - assert.equal(active.providerTurns.at(-1)?.status, "running"); - yield* Effect.promise(() => nativeEvents.push(idle)); - } - yield* Deferred.await(terminal); - assert.equal(statusCalls, 1); - - const settled = yield* harness.runtime.readThreadSnapshot({ + assert.equal(active.providerTurns.at(-1)?.status, "running"); + } + yield* Effect.promise(() => statusRead.promise); + if (idleTiming === "after" || idleTiming === "stale-idle-then-busy") { + const active = yield* harness.runtime.readThreadSnapshot({ providerThread: harness.providerThread, }); - assert.equal(settled.providerTurns.at(-1)?.status, "completed"); - }).pipe( - Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer)), - Effect.scoped, - ), + assert.equal(active.providerTurns.at(-1)?.status, "running"); + yield* Effect.promise(() => nativeEvents.push(idle)); + } + yield* Deferred.await(terminal); + assert.equal(statusCalls, idleTiming === "idle-during-status" ? 2 : 1); + + const settled = yield* harness.runtime.readThreadSnapshot({ + providerThread: harness.providerThread, + }); + assert.equal(settled.providerTurns.at(-1)?.status, "completed"); + }).pipe( + Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer)), + Effect.scoped, + ), ); it.effect("interrupts an initial prompt while its SDK request is pending", () => diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index 88fb0d307c10..60ed7bc9ccc4 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -2115,6 +2115,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function generation: number, ) { while (promptAdmissionIsCurrent(state, turn, generation)) { + turn.idleDuringAdmission = false; const status = yield* reconcileOpenCodePromptAdmissionStatus( turn, generation, @@ -2146,7 +2147,11 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function yield* finalizeTurn(state, turn, "completed"); return; } - if (status === "busy") return; + if (status === "busy") { + if (!turn.idleDuringAdmission) return; + turn.admissionPending = true; + continue; + } yield* Effect.sleep("250 millis"); } }); From 51d8f1a955800e1bb6f659d0881cc83406ef9948 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Mon, 26 Oct 2026 13:42:44 +0000 Subject: [PATCH 4/8] fix(opencode): preserve admission idle across repeated user updates --- .../src/server/adapter.test.ts | 34 ++++++++++++++++++- .../provider-opencode/src/server/adapter.ts | 9 +++-- 2 files changed, 40 insertions(+), 3 deletions(-) diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index 0bb2329c786e..26e2dad19682 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -1727,6 +1727,7 @@ describe("OpenCodeAdapterV2", () => { "stale-idle-then-busy", "no-idle", "idle-during-status", + "repeated-user-during-status", ] as const)("completes the turn when the session goes idle %s a rejected steer", (idleTiming) => Effect.gen(function* () { const nativeEvents = asyncEventStream(); @@ -1748,6 +1749,20 @@ describe("OpenCodeAdapterV2", () => { status: async () => { statusCalls++; statusRead.resolve(); + if (idleTiming === "repeated-user-during-status") { + await nativeEvents.push({ + type: "message.updated", + properties: { + sessionID: sessionId, + info: { + id: promptIds[1], + sessionID: sessionId, + role: "user", + time: { created: 1 }, + }, + }, + }); + } if (idleTiming === "idle-during-status" && statusCalls === 1) { await nativeEvents.push(idle); return { data: { [sessionId]: { type: "busy" } } }; @@ -1758,6 +1773,7 @@ describe("OpenCodeAdapterV2", () => { type: idleTiming === "during" || idleTiming === "no-idle" || + idleTiming === "repeated-user-during-status" || idleTiming === "idle-during-status" ? "idle" : "busy", @@ -1768,6 +1784,22 @@ describe("OpenCodeAdapterV2", () => { promptAsync: async (input: { messageID: string }) => { promptIds.push(input.messageID); if (promptIds.length === 1) return { data: true }; + if (idleTiming === "repeated-user-during-status") { + await nativeEvents.push(idle); + await nativeEvents.push({ + type: "message.updated", + properties: { + sessionID: sessionId, + info: { + id: input.messageID, + sessionID: sessionId, + role: "user", + time: { created: 1 }, + }, + }, + }); + return { data: true }; + } if (idleTiming === "during" || idleTiming === "stale-idle-then-busy") { await nativeEvents.push(idle); } @@ -1823,7 +1855,7 @@ describe("OpenCodeAdapterV2", () => { }, }), ); - assert.isTrue(Exit.isFailure(steer)); + assert.equal(Exit.isFailure(steer), idleTiming !== "repeated-user-during-status"); if (idleTiming === "stale-idle-then-busy") { const active = yield* harness.runtime.readThreadSnapshot({ providerThread: harness.providerThread, diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index 60ed7bc9ccc4..13eada979705 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -288,6 +288,7 @@ interface ActiveOpenCodeTurn { admissionAccepted: boolean; admissionMessageObserved: boolean; idleDuringAdmission: boolean; + admissionIdleSequence: number; admissionSettled: Deferred.Deferred; admissionAbortController: AbortController | null; } @@ -2115,7 +2116,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function generation: number, ) { while (promptAdmissionIsCurrent(state, turn, generation)) { - turn.idleDuringAdmission = false; + const idleSequence = turn.admissionIdleSequence; const status = yield* reconcileOpenCodePromptAdmissionStatus( turn, generation, @@ -2148,7 +2149,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function return; } if (status === "busy") { - if (!turn.idleDuringAdmission) return; + if (turn.admissionIdleSequence === idleSequence) return; turn.admissionPending = true; continue; } @@ -2238,6 +2239,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function admissionAccepted: true, admissionMessageObserved: true, idleDuringAdmission: false, + admissionIdleSequence: 0, admissionSettled: Deferred.makeUnsafe(), admissionAbortController: null, }; @@ -2602,6 +2604,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function } if (event.properties.status.type === "idle" && state.activeTurn !== null) { if (state.activeTurn.admissionPending) { + state.activeTurn.admissionIdleSequence++; advanceOpenCodePromptAdmission(state.activeTurn, "idle"); return; } @@ -2618,6 +2621,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function const state = threads.get(event.properties.sessionID); if (state?.activeTurn !== null && state?.activeTurn !== undefined) { if (state.activeTurn.admissionPending) { + state.activeTurn.admissionIdleSequence++; advanceOpenCodePromptAdmission(state.activeTurn, "idle"); return; } @@ -3168,6 +3172,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function admissionAccepted: isCompaction, admissionMessageObserved: false, idleDuringAdmission: false, + admissionIdleSequence: 0, admissionSettled: Deferred.makeUnsafe(), // @effect-diagnostics-next-line abortControllerInEffect:off - Turn interruption owns prompt admission cancellation. admissionAbortController: new AbortController(), From 6dd5df611defb08cbecd5a66c546703680a4a30b Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 06:05:29 +0000 Subject: [PATCH 5/8] fix(opencode): recheck admission after busy activity --- .../src/server/adapter.test.ts | 29 ++++++++++++++++++- .../provider-opencode/src/server/adapter.ts | 24 ++++++++------- 2 files changed, 41 insertions(+), 12 deletions(-) diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index 6d9a099dd808..b3f13631cd65 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -1941,6 +1941,7 @@ describe("OpenCodeAdapterV2", () => { "stale-idle-then-busy", "no-idle", "idle-during-status", + "busy-during-status", "repeated-user-during-status", ] as const)("completes the turn when the session goes idle %s a rejected steer", (idleTiming) => Effect.gen(function* () { @@ -1953,6 +1954,7 @@ describe("OpenCodeAdapterV2", () => { const promptIds: Array = []; const terminal = yield* Deferred.make(); const statusRead = promiseGate(); + const statusRechecked = promiseGate(); let statusCalls = 0; const harness = yield* makeOpenCodeRuntimeHarness(`steer-rejected-${idleTiming}`, sessionId, { event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, @@ -1963,6 +1965,17 @@ describe("OpenCodeAdapterV2", () => { status: async () => { statusCalls++; statusRead.resolve(); + if (idleTiming === "busy-during-status") { + if (statusCalls === 1) { + await nativeEvents.push({ + type: "session.status", + properties: { sessionID: sessionId, status: { type: "busy" } }, + }); + return { data: { [sessionId]: { type: "idle" } } }; + } + statusRechecked.resolve(); + return { data: { [sessionId]: { type: statusCalls === 2 ? "busy" : "idle" } } }; + } if (idleTiming === "repeated-user-during-status") { await nativeEvents.push({ type: "message.updated", @@ -2077,6 +2090,18 @@ describe("OpenCodeAdapterV2", () => { assert.equal(active.providerTurns.at(-1)?.status, "running"); } yield* Effect.promise(() => statusRead.promise); + if (idleTiming === "busy-during-status") { + const outcome = yield* Effect.raceFirst( + Effect.promise(() => statusRechecked.promise).pipe(Effect.as("rechecked")), + Deferred.await(terminal).pipe(Effect.as("completed")), + ); + assert.equal(outcome, "rechecked"); + const active = yield* harness.runtime.readThreadSnapshot({ + providerThread: harness.providerThread, + }); + assert.equal(active.providerTurns.at(-1)?.status, "running"); + yield* Effect.promise(() => nativeEvents.push(idle)); + } if (idleTiming === "after" || idleTiming === "stale-idle-then-busy") { const active = yield* harness.runtime.readThreadSnapshot({ providerThread: harness.providerThread, @@ -2085,7 +2110,9 @@ describe("OpenCodeAdapterV2", () => { yield* Effect.promise(() => nativeEvents.push(idle)); } yield* Deferred.await(terminal); - assert.equal(statusCalls, idleTiming === "idle-during-status" ? 2 : 1); + if (idleTiming !== "busy-during-status") { + assert.equal(statusCalls, idleTiming === "idle-during-status" ? 2 : 1); + } const settled = yield* harness.runtime.readThreadSnapshot({ providerThread: harness.providerThread, diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index 3bdb4a9612bf..a92c64aeacd9 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -286,7 +286,7 @@ interface ActiveOpenCodeTurn { admissionAccepted: boolean; admissionMessageObserved: boolean; idleDuringAdmission: boolean; - admissionIdleSequence: number; + admissionStatusSequence: number; admissionSettled: Deferred.Deferred; admissionAbortController: AbortController | null; } @@ -2132,7 +2132,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function generation: number, ) { while (promptAdmissionIsCurrent(state, turn, generation)) { - const idleSequence = turn.admissionIdleSequence; + const statusSequence = turn.admissionStatusSequence; const status = yield* reconcileOpenCodePromptAdmissionStatus( turn, generation, @@ -2160,15 +2160,15 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function ) { return; } + if (turn.admissionStatusSequence !== statusSequence) { + turn.admissionPending = true; + continue; + } if (status === "idle") { yield* finalizeTurn(state, turn, "completed"); return; } - if (status === "busy") { - if (turn.admissionIdleSequence === idleSequence) return; - turn.admissionPending = true; - continue; - } + if (status === "busy") return; yield* Effect.sleep("250 millis"); } }); @@ -2253,7 +2253,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function admissionAccepted: true, admissionMessageObserved: true, idleDuringAdmission: false, - admissionIdleSequence: 0, + admissionStatusSequence: 0, admissionSettled: Deferred.makeUnsafe(), admissionAbortController: null, }; @@ -2612,13 +2612,15 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function } const state = threads.get(sessionId); if (state === undefined) return; + if (state.activeTurn?.admissionPending) { + state.activeTurn.admissionStatusSequence++; + } if (event.properties.status.type === "busy") { yield* updateProviderThread(state, { status: "active" }); return; } if (event.properties.status.type === "idle" && state.activeTurn !== null) { if (state.activeTurn.admissionPending) { - state.activeTurn.admissionIdleSequence++; advanceOpenCodePromptAdmission(state.activeTurn, "idle"); return; } @@ -2635,7 +2637,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function const state = threads.get(event.properties.sessionID); if (state?.activeTurn !== null && state?.activeTurn !== undefined) { if (state.activeTurn.admissionPending) { - state.activeTurn.admissionIdleSequence++; + state.activeTurn.admissionStatusSequence++; advanceOpenCodePromptAdmission(state.activeTurn, "idle"); return; } @@ -3221,7 +3223,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function admissionAccepted: isCompaction, admissionMessageObserved: false, idleDuringAdmission: false, - admissionIdleSequence: 0, + admissionStatusSequence: 0, admissionSettled: Deferred.makeUnsafe(), // @effect-diagnostics-next-line abortControllerInEffect:off - Turn interruption owns prompt admission cancellation. admissionAbortController: new AbortController(), From 1915c8ce887fdc9409dcf60a5044cbbe7b8325d7 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Fri, 30 Oct 2026 15:36:24 +0000 Subject: [PATCH 6/8] fix(opencode): retry stalled admission status reads --- packages/provider-opencode/src/server/adapter.test.ts | 10 ++++++++++ packages/provider-opencode/src/server/adapter.ts | 11 ++++++++--- 2 files changed, 18 insertions(+), 3 deletions(-) diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index b3f13631cd65..849818dd37cc 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -33,6 +33,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import * as ProviderHost from "@t3tools/provider-core/server/ProviderHost"; import * as TestProviderHost from "@t3tools/provider-testing/TestProviderHost"; @@ -2516,6 +2517,15 @@ describe("OpenCodeAdapterV2", () => { "unknown", ); assert.isTrue(admission.admissionPending); + const stalled = yield* reconcileOpenCodePromptAdmissionStatus( + admission, + 4, + Effect.never, + ).pipe(Effect.forkChild({ startImmediately: true })); + yield* TestClock.adjust("10 seconds"); + assert.isDefined(stalled.pollUnsafe()); + assert.equal(yield* Fiber.join(stalled), "unknown"); + assert.isTrue(admission.admissionPending); assert.equal( yield* reconcileOpenCodePromptAdmissionStatus(admission, 4, Effect.succeed("idle")), "idle", diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index a92c64aeacd9..fc9d866bc3e0 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -343,7 +343,10 @@ export const reconcileOpenCodePromptAdmissionStatus = Effect.fn( if (admission.admissionGeneration !== generation || !admission.admissionPending) { return "stale" as const; } - const status = yield* readStatus; + const status = yield* readStatus.pipe( + Effect.timeoutOption("10 seconds"), + Effect.map(Option.getOrElse(() => "unknown" as const)), + ); if (admission.admissionGeneration !== generation || !admission.admissionPending) { return "stale" as const; } @@ -2136,8 +2139,10 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function const status = yield* reconcileOpenCodePromptAdmissionStatus( turn, generation, - sdkCall("session.status", { sessionID: state.nativeSessionId, generation }, () => - client.session.status(), + sdkCall( + "session.status", + { sessionID: state.nativeSessionId, generation }, + (signal) => client.session.status(undefined, { signal }), ).pipe( Effect.match({ onFailure: () => "unknown" as const, From e9fc9f7fb6a35dbbe7aa00e44a6d01fa32f094d1 Mon Sep 17 00:00:00 2001 From: Victor Vogel Date: Thu, 12 Nov 2026 08:21:32 +0000 Subject: [PATCH 7/8] fix(opencode): a steer whose reply was lost keeps its turn until OpenCode answers Only an HTTP error reply or a connection that never opened proves OpenCode rejected a steer. Any other failure may follow a steer OpenCode took, so the turn now waits for that steer's message, or Stop, instead of completing on the next idle and dropping its output. Co-Authored-By: Claude Opus 5.5 --- .../src/server/adapter.test.ts | 137 +++++++++++++++++- .../provider-opencode/src/server/adapter.ts | 56 +++++-- 2 files changed, 180 insertions(+), 13 deletions(-) diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index 787d7656863c..9049bb67d68f 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -2665,6 +2665,7 @@ describe("OpenCodeAdapterV2", () => { "idle-during-status", "busy-during-status", "repeated-user-during-status", + "after-refused-connection", ] as const)("completes the turn when the session goes idle %s a rejected steer", (idleTiming) => Effect.gen(function* () { const nativeEvents = asyncEventStream(); @@ -2758,7 +2759,13 @@ describe("OpenCodeAdapterV2", () => { properties: { sessionID: sessionId, status: { type: "busy" } }, }); } - throw new Error("steer rejected"); + if (idleTiming === "after-refused-connection") { + throw new TypeError("fetch failed", { + cause: Object.assign(new Error("connect ECONNREFUSED"), { code: "ECONNREFUSED" }), + }); + } + // The SDK wraps an HTTP error reply in an Error with its status on `cause`. + throw new Error("steer rejected", { cause: { body: "steer rejected", status: 503 } }); }, abort: async () => ({ data: true }), children: async () => ({ data: [] }), @@ -2824,7 +2831,11 @@ describe("OpenCodeAdapterV2", () => { assert.equal(active.providerTurns.at(-1)?.status, "running"); yield* Effect.promise(() => nativeEvents.push(idle)); } - if (idleTiming === "after" || idleTiming === "stale-idle-then-busy") { + if ( + idleTiming === "after" || + idleTiming === "stale-idle-then-busy" || + idleTiming === "after-refused-connection" + ) { const active = yield* harness.runtime.readThreadSnapshot({ providerThread: harness.providerThread, }); @@ -2846,6 +2857,128 @@ describe("OpenCodeAdapterV2", () => { ), ); + it.effect.each(["", ", even after a retry is rejected"] as const)( + "keeps the turn for a steer whose reply was lost until its output lands%s", + (retry) => + Effect.gen(function* () { + const nativeEvents = asyncEventStream(); + const sessionId = `native-steer-reply-lost${retry === "" ? "" : "-retry"}`; + const promptIds: Array = []; + const harness = yield* makeOpenCodeRuntimeHarness(sessionId, sessionId, { + event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, + session: { + create: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), + get: async () => ({ data: { id: sessionId, time: { created: 1, updated: 1 } } }), + messages: async () => ({ data: [] }), + // OpenCode took the steer but has not saved its message yet. + status: async () => ({ data: { [sessionId]: { type: "idle" } } }), + promptAsync: async (input: { messageID: string }) => { + promptIds.push(input.messageID); + if (promptIds.length === 1) return { data: true }; + if (promptIds.length === 3) { + throw new Error("Service Unavailable", { cause: { status: 503 } }); + } + throw new TypeError("fetch failed", { + cause: Object.assign(new Error("other side closed"), { code: "UND_ERR_SOCKET" }), + }); + }, + abort: async () => ({ data: true }), + children: async () => ({ data: [] }), + }, + }); + yield* harness.startTurn(); + const userMessage = (id: string | undefined) => + Effect.promise(() => + nativeEvents.push({ + type: "message.updated", + properties: { + sessionID: sessionId, + info: { id, sessionID: sessionId, role: "user", time: { created: 1 } }, + }, + }), + ); + yield* userMessage(promptIds[0]); + const running = yield* harness.runtime.readThreadSnapshot({ + providerThread: harness.providerThread, + }); + const received = yield* harness.runtime.events.pipe( + Stream.takeUntil((event) => event.type === "turn.terminal"), + Stream.runCollect, + Effect.forkScoped, + ); + const steer = harness.runtime.steerTurn({ + threadId: harness.threadId, + runId: harness.runId, + providerThread: harness.providerThread, + providerTurnId: running.providerTurns.at(-1)!.id, + message: { + messageId: MessageId.make("message-steer-reply-lost"), + text: "also cover the README", + attachments: [], + createdBy: "user", + creationSource: "web", + }, + }); + assert.isTrue(Exit.isFailure(yield* Effect.exit(steer))); + // T3 retries a failed steer; this retry's request is refused outright. + if (retry !== "") assert.isTrue(Exit.isFailure(yield* Effect.exit(steer))); + // A real status read answers before OpenCode saves the steer's message. + yield* Effect.yieldNow; + yield* userMessage(promptIds[1]); + const assistant = (completed?: number) => + Effect.promise(() => + nativeEvents.push({ + type: "message.updated", + properties: { + sessionID: sessionId, + info: { + id: "steer-answer", + sessionID: sessionId, + role: "assistant", + time: completed === undefined ? { created: 2 } : { created: 2, completed }, + }, + }, + }), + ); + yield* assistant(); + yield* Effect.promise(() => + nativeEvents.push({ + type: "message.part.updated", + properties: { + sessionID: sessionId, + part: { + id: "steer-answer-text", + sessionID: sessionId, + messageID: "steer-answer", + type: "text", + text: "README covered", + time: { start: 2 }, + }, + }, + }), + ); + yield* assistant(3); + yield* Effect.promise(() => + nativeEvents.push({ + type: "session.status", + properties: { sessionID: sessionId, status: { type: "idle" } }, + }), + ); + const events = yield* Fiber.join(received); + const answers = events.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem.text] + : [], + ); + assert.include(answers, "README covered"); + const terminal = events.at(-1); + assert.equal(terminal?.type === "turn.terminal" ? terminal.status : undefined, "completed"); + }).pipe( + Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer)), + Effect.scoped, + ), + ); + it.effect("interrupts an initial prompt while its SDK request is pending", () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); diff --git a/packages/provider-opencode/src/server/adapter.ts b/packages/provider-opencode/src/server/adapter.ts index 7a1e5b0d83fb..55b4e4106848 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -289,6 +289,7 @@ interface ActiveOpenCodeTurn { admissionAccepted: boolean; admissionMessageObserved: boolean; idleDuringAdmission: boolean; + admissionUncertain: boolean; admissionStatusSequence: number; admissionSettled: Deferred.Deferred; admissionAbortController: AbortController | null; @@ -882,6 +883,28 @@ function isMessageAbortedError(event: Extract(); const queue: Array = [cause]; @@ -2285,6 +2308,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function admissionAccepted: true, admissionMessageObserved: true, idleDuringAdmission: false, + admissionUncertain: false, admissionStatusSequence: 0, admissionSettled: Deferred.makeUnsafe(), admissionAbortController: null, @@ -3304,6 +3328,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function admissionAccepted: isCompaction, admissionMessageObserved: false, idleDuringAdmission: false, + admissionUncertain: false, admissionStatusSequence: 0, admissionSettled: Deferred.makeUnsafe(), // @effect-diagnostics-next-line abortControllerInEffect:off - Turn interruption owns prompt admission cancellation. @@ -3463,6 +3488,8 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function `OpenCode turn ${steerInput.providerTurnId} is not active`, ); } + // A retry of a steer whose earlier attempt may have reached OpenCode stays uncertain. + if (!turn.admissionPending) turn.admissionUncertain = false; turn.admissionGeneration = state.nextAdmissionGeneration++; turn.admissionMessageId = admissionMessageId; turn.admissionPending = true; @@ -3477,6 +3504,14 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function const admissionSettled = turn.admissionSettled; const admissionAbortController = turn.admissionAbortController; const generation = turn.admissionGeneration; + const acceptAdmission = Effect.suspend(() => + state.activeTurn === turn && + !turn.finalized && + !turn.interrupted && + advanceOpenCodePromptAdmission(turn, "accepted") === "reconcile-idle" + ? reconcilePromptAdmission(state, turn) + : Effect.void, + ); yield* submitPrompt( state, turn, @@ -3489,11 +3524,15 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function }, admissionAbortController!, ).pipe( - Effect.tapError(() => - promptAdmissionIsCurrent(state, turn, generation) - ? reconcilePromptAdmission(state, turn) - : Effect.void, - ), + Effect.tapError((error) => { + if (!promptAdmissionIsCurrent(state, turn, generation)) return Effect.void; + if (isOpenCodePromptRejected(error) && !turn.admissionUncertain) { + return reconcilePromptAdmission(state, turn); + } + // OpenCode may have taken this or an earlier attempt: wait for its output or Stop. + turn.admissionUncertain = true; + return acceptAdmission; + }), Effect.ensuring( Effect.all([ Deferred.succeed(admissionSettled, undefined).pipe(Effect.ignore), @@ -3505,12 +3544,7 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function ]).pipe(Effect.asVoid), ), ); - if (state.activeTurn === turn && !turn.finalized && !turn.interrupted) { - const admissionAction = advanceOpenCodePromptAdmission(turn, "accepted"); - if (admissionAction === "reconcile-idle") { - yield* reconcilePromptAdmission(state, turn); - } - } + yield* acceptAdmission; }, (effect, steerInput) => effect.pipe( From 368f13d6c8efa975ed7baa8180c9857802f554d3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 07:06:45 +0000 Subject: [PATCH 8/8] test(opencode): keep only failed-steer cases that catch a regression Drop variants that pass on main or repeat another case's path; each remaining case fails on main or when one piece of the fix is removed. Co-Authored-By: Claude Opus 5.5 --- .../src/server/adapter.test.ts | 78 +++---------------- 1 file changed, 9 insertions(+), 69 deletions(-) diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index 52cacc1f7cc4..218fd9cda3fe 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -2724,12 +2724,8 @@ describe("OpenCodeAdapterV2", () => { it.effect.each([ "during", - "after", - "stale-idle-then-busy", - "no-idle", "idle-during-status", "busy-during-status", - "repeated-user-during-status", "after-refused-connection", ] as const)("completes the turn when the session goes idle %s a rejected steer", (idleTiming) => Effect.gen(function* () { @@ -2764,66 +2760,20 @@ describe("OpenCodeAdapterV2", () => { statusRechecked.resolve(); return { data: { [sessionId]: { type: statusCalls === 2 ? "busy" : "idle" } } }; } - if (idleTiming === "repeated-user-during-status") { - await nativeEvents.push({ - type: "message.updated", - properties: { - sessionID: sessionId, - info: { - id: promptIds[1], - sessionID: sessionId, - role: "user", - time: { created: 1 }, - }, - }, - }); - } if (idleTiming === "idle-during-status" && statusCalls === 1) { await nativeEvents.push(idle); return { data: { [sessionId]: { type: "busy" } } }; } return { data: { - [sessionId]: { - type: - idleTiming === "during" || - idleTiming === "no-idle" || - idleTiming === "repeated-user-during-status" || - idleTiming === "idle-during-status" - ? "idle" - : "busy", - }, + [sessionId]: { type: idleTiming === "after-refused-connection" ? "busy" : "idle" }, }, }; }, promptAsync: async (input: { messageID: string }) => { promptIds.push(input.messageID); if (promptIds.length === 1) return { data: true }; - if (idleTiming === "repeated-user-during-status") { - await nativeEvents.push(idle); - await nativeEvents.push({ - type: "message.updated", - properties: { - sessionID: sessionId, - info: { - id: input.messageID, - sessionID: sessionId, - role: "user", - time: { created: 1 }, - }, - }, - }); - return { data: true }; - } - if (idleTiming === "during" || idleTiming === "stale-idle-then-busy") { - await nativeEvents.push(idle); - } - if (idleTiming === "stale-idle-then-busy") { - await nativeEvents.push({ - type: "session.status", - properties: { sessionID: sessionId, status: { type: "busy" } }, - }); - } + if (idleTiming === "during") await nativeEvents.push(idle); if (idleTiming === "after-refused-connection") { throw new TypeError("fetch failed", { cause: Object.assign(new Error("connect ECONNREFUSED"), { code: "ECONNREFUSED" }), @@ -2876,13 +2826,7 @@ describe("OpenCodeAdapterV2", () => { }, }), ); - assert.equal(Exit.isFailure(steer), idleTiming !== "repeated-user-during-status"); - if (idleTiming === "stale-idle-then-busy") { - const active = yield* harness.runtime.readThreadSnapshot({ - providerThread: harness.providerThread, - }); - assert.equal(active.providerTurns.at(-1)?.status, "running"); - } + assert.isTrue(Exit.isFailure(steer)); yield* Effect.promise(() => statusRead.promise); if (idleTiming === "busy-during-status") { const outcome = yield* Effect.raceFirst( @@ -2896,11 +2840,7 @@ describe("OpenCodeAdapterV2", () => { assert.equal(active.providerTurns.at(-1)?.status, "running"); yield* Effect.promise(() => nativeEvents.push(idle)); } - if ( - idleTiming === "after" || - idleTiming === "stale-idle-then-busy" || - idleTiming === "after-refused-connection" - ) { + if (idleTiming === "after-refused-connection") { const active = yield* harness.runtime.readThreadSnapshot({ providerThread: harness.providerThread, }); @@ -2922,12 +2862,12 @@ describe("OpenCodeAdapterV2", () => { ), ); - it.effect.each(["", ", even after a retry is rejected"] as const)( - "keeps the turn for a steer whose reply was lost until its output lands%s", - (retry) => + it.effect( + "keeps the turn for a steer whose reply was lost until its output lands, even after a retry is rejected", + () => Effect.gen(function* () { const nativeEvents = asyncEventStream(); - const sessionId = `native-steer-reply-lost${retry === "" ? "" : "-retry"}`; + const sessionId = "native-steer-reply-lost"; const promptIds: Array = []; const harness = yield* makeOpenCodeRuntimeHarness(sessionId, sessionId, { event: { subscribe: async () => ({ stream: nativeEvents.stream }) }, @@ -2986,7 +2926,7 @@ describe("OpenCodeAdapterV2", () => { }); assert.isTrue(Exit.isFailure(yield* Effect.exit(steer))); // T3 retries a failed steer; this retry's request is refused outright. - if (retry !== "") assert.isTrue(Exit.isFailure(yield* Effect.exit(steer))); + assert.isTrue(Exit.isFailure(yield* Effect.exit(steer))); // A real status read answers before OpenCode saves the steer's message. yield* Effect.yieldNow; yield* userMessage(promptIds[1]);