diff --git a/packages/provider-opencode/src/server/adapter.test.ts b/packages/provider-opencode/src/server/adapter.test.ts index c1c89b2b96263..218fd9cda3fe6 100644 --- a/packages/provider-opencode/src/server/adapter.test.ts +++ b/packages/provider-opencode/src/server/adapter.test.ts @@ -2722,6 +2722,268 @@ describe("OpenCodeAdapterV2", () => { }).pipe(Effect.provide(Layer.merge(IdAllocator.layer, McpProviderSessions.layer))), ); + it.effect.each([ + "during", + "idle-during-status", + "busy-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(); + 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(); + const statusRechecked = 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 === "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 === "idle-during-status" && statusCalls === 1) { + await nativeEvents.push(idle); + return { data: { [sessionId]: { type: "busy" } } }; + } + return { + data: { + [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 === "during") await nativeEvents.push(idle); + 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: [] }), + }, + }); + 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, + }); + 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)); + 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-refused-connection") { + 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); + if (idleTiming !== "busy-during-status") { + 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( + "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"; + 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. + 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(); @@ -3102,6 +3364,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 4d265ab9dc460..bfde592ee29f0 100644 --- a/packages/provider-opencode/src/server/adapter.ts +++ b/packages/provider-opencode/src/server/adapter.ts @@ -314,6 +314,8 @@ interface ActiveOpenCodeTurn { admissionAccepted: boolean; admissionMessageObserved: boolean; idleDuringAdmission: boolean; + admissionUncertain: boolean; + admissionStatusSequence: number; admissionSettled: Deferred.Deferred; admissionAbortController: AbortController | null; } @@ -370,7 +372,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; } @@ -904,6 +909,28 @@ function isMessageAbortedError(event: Extract(); const queue: Array = [cause]; @@ -2223,11 +2250,14 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function generation: number, ) { while (promptAdmissionIsCurrent(state, turn, generation)) { + const statusSequence = turn.admissionStatusSequence; 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, @@ -2250,6 +2280,10 @@ 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; @@ -2344,6 +2378,8 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function admissionAccepted: true, admissionMessageObserved: true, idleDuringAdmission: false, + admissionUncertain: false, + admissionStatusSequence: 0, admissionSettled: Deferred.makeUnsafe(), admissionAbortController: null, }; @@ -2703,6 +2739,9 @@ 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; @@ -2725,6 +2764,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.admissionStatusSequence++; advanceOpenCodePromptAdmission(state.activeTurn, "idle"); return; } @@ -3358,6 +3398,8 @@ 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. admissionAbortController: new AbortController(), @@ -3516,6 +3558,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; @@ -3529,6 +3573,15 @@ 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; + 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, @@ -3541,6 +3594,15 @@ export const makeOpenCodeAdapterV2 = Effect.fn("makeOpenCodeAdapterV2")(function }, admissionAbortController!, ).pipe( + 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), @@ -3552,12 +3614,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(