From cb7d0739389a602d81000bfb5aeb6ebea9e853bb Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 10 Oct 2026 22:35:03 -0700 Subject: [PATCH] fix(server): OpenCode 2 continuations end after a reconnect took a Stop's end (#14744) Co-authored-by: Claude Opus 5.5 (1M context) --- .../Adapters/OpenCode2AdapterV2.test.ts | 356 +++++++++++++++++- .../Adapters/OpenCode2AdapterV2.testkit.ts | 6 +- .../src/server/v2/adapter.ts | 63 +++- 3 files changed, 409 insertions(+), 16 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts index 5c7161decd4497faa4ef8e2db712259c4be4e3df..9cb06619c08e7c9192f006551ecb26425367b4bb 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.test.ts @@ -48,6 +48,10 @@ import { t3McpServerName, } from "@t3tools/provider-opencode/testing"; import { openCode2ReplayRuntime } from "./OpenCode2AdapterV2.testkit.ts"; +import { + makeProviderReplayGate, + type ProviderReplayGate, +} from "@t3tools/provider-testing/replayGate"; const SESSION = "ses_f148ca2deffeJcwCnRQtb0YFNX"; const WORK = "/work/opencode2"; @@ -64,12 +68,18 @@ const reply = (operation: string, data: unknown): ProviderReplayEntry => ({ frame: { type: "sdk.response", operation, data }, }); const replyData = (operation: string, data: unknown) => reply(operation, { data }); -const event = (type: string, data: Record): ProviderReplayEntry => ({ +const event = ( + type: string, + data: Record, + /** Holds the event until a test's replay gate releases this label. */ + label?: string, +): ProviderReplayEntry => ({ type: "emit_inbound", frame: { type: "sdk.event", event: { id: `evt_${type.replaceAll(".", "")}0000`, created: 1, type, data, ...durable }, }, + ...(label === undefined ? {} : { label }), }); const durable = { durable: { aggregateID: SESSION, seq: 1, version: 1 } }; @@ -316,7 +326,11 @@ const colorForm = { */ const resumed = ( entries: ReadonlyArray, - options?: { readonly external?: boolean; readonly supervised?: boolean }, + options?: { + readonly external?: boolean; + readonly supervised?: boolean; + readonly replayGate?: ProviderReplayGate; + }, ) => Effect.gen(function* () { const runtime = yield* openCode2ReplayRuntime( @@ -335,7 +349,10 @@ const resumed = ( : []), ...entries, ]), - options?.external === undefined ? undefined : { external: options.external }, + { + ...(options?.external === undefined ? {} : { external: options.external }), + ...(options?.replayGate === undefined ? {} : { replayGate: options.replayGate }), + }, ); const thread = yield* runtime.resumeThread({ providerThread: providerThread(yield* DateTime.now), @@ -1724,6 +1741,88 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), ); + it.effect("still checks the server when a failed prompt's execution starts late", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const { runtime, thread } = yield* resumed([ + out("session.prompt", { sessionID: SESSION, text: "" }), + // The request failed, but the server took the prompt: its execution + // starts only after the turn already failed. + reply("session.prompt", { + status: 502, + body: { _tag: "UnknownError", message: "bad gateway" }, + }), + event("session.execution.started", { sessionID: SESSION }), + // That run is still going, so the next turn stops it and fails. + out("session.active"), + reply("session.active", { data: { [SESSION]: { type: "running" } } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + ]).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const ended = yield* terminals(runtime, 2); + yield* runtime.startTurn(turnInput(thread)).pipe(Effect.ignore); + // The late start has been read. + yield* Deferred.await(offered); + yield* runtime.startTurn(secondTurn(thread)); + const [first, second] = yield* Fiber.join(ended); + assert.equal(first?.status, "failed"); + assert.equal(second?.failure?.message, OPENCODE_2_STILL_STOPPING); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + + it.effect("still checks the server when a failed prompt starts after a reconnect", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const gate = makeProviderReplayGate(["drop"]); + const { runtime, thread } = yield* resumed( + [ + out("session.prompt", { sessionID: SESSION, text: "" }), + reply("session.prompt", { + status: 502, + body: { _tag: "UnknownError", message: "bad gateway" }, + }), + // The stream drops before the prompt's execution starts, and the + // reconnect finds the session active: the prompt's run is still to come. + event("session.usage.updated", { sessionID: SESSION }, "drop"), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [SESSION]: { type: "running" } }), + out("permission.list", { sessionID: SESSION }), + replyData("permission.list", []), + out("session.form.list", { sessionID: SESSION }), + replyData("session.form.list", []), + // Its own start: that run is still going, so the next turn stops it and fails. + event("session.execution.started", { sessionID: SESSION }), + out("session.active"), + reply("session.active", { data: { [SESSION]: { type: "running" } } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + const ended = yield* terminals(runtime, 2); + yield* runtime.startTurn(turnInput(thread)).pipe(Effect.ignore); + gate.release("drop"); + // The late start has been read. + yield* Deferred.await(offered); + yield* runtime.startTurn(secondTurn(thread)); + const [first, second] = yield* Fiber.join(ended); + assert.equal(first?.status, "failed"); + assert.equal(second?.failure?.message, OPENCODE_2_STILL_STOPPING); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + it.effect("prompts again without a check after the server refused a prompt", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ @@ -1764,6 +1863,257 @@ it.layer(McpProviderSessions.layer)("OpenCode2 adapter", (it) => { }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), ); + it.effect("ends a background reply's continuation after a lost stream took a Stop's end", () => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const gate = makeProviderReplayGate(["drop", "reply"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + // Restarting the turn stops only the parent, and the server never answers. + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", ""), + // The restarted turn finds the parent still stopping and fails. + out("session.active"), + replyData("session.active", { [SESSION]: { type: "running" } }), + out("session.interrupt", { sessionID: SESSION }), + reply("session.interrupt", { interrupted: true }), + // The stream drops before the parent's end; the reconnect finds it idle. + event("session.usage.updated", { sessionID: SESSION }, "drop"), + { type: "runtime_exit", status: "success" } as const, + out("event.subscribe"), + out("session.active"), + replyData("session.active", { [CHILD]: { type: "running" } }), + out("permission.list", { sessionID: CHILD }), + replyData("permission.list", []), + out("session.form.list", { sessionID: CHILD }), + replyData("session.form.list", []), + // The subagent ends and OpenCode answers its report on its own. + event("session.execution.succeeded", { sessionID: CHILD }), + event("session.inbox.enqueued", { + inboxID: "msg_report", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nCHILD_OK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: CHILD, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + event( + "session.text.ended", + { + sessionID: SESSION, + assistantMessageID: "msg_followup", + ordinal: 0, + text: "CHILD_OK", + }, + "reply", + ), + event("session.execution.succeeded", { sessionID: SESSION }), + // The subagent's session runs again: its turn marks that the reply's end was handled. + event("session.execution.started", { sessionID: CHILD }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + // The single reader of the runtime's events. + const attached = yield* Deferred.make(); + const collected = yield* runtime.events.pipe( + Stream.tap((event) => + event.type === "subagent.updated" && event.subagent.childThreadId !== null + ? Deferred.succeed(attached, undefined) + : Effect.void, + ), + Stream.takeUntil( + (event) => + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`) === true, + ), + Stream.runCollect, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(attached); + const interrupt = yield* runtime + .interruptTurn({ providerThread: thread, providerTurnId: yield* providerTurnId }) + .pipe(Effect.forkScoped); + yield* TestClock.adjust("11 seconds"); + yield* Fiber.join(interrupt); + yield* runtime.startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread }); + gate.release("drop"); + yield* Deferred.await(offered); + yield* runtime.startTurn({ + ...withLineage(thread), + runId: RunId.make("run:opencode2-adapter:wake"), + runOrdinal: 3, + providerTurnOrdinal: 3, + attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"), + message: { + ...turnInput(thread).message, + messageId: MessageId.make("message:opencode2-adapter:wake"), + createdBy: "agent" as const, + creationSource: "provider" as const, + }, + }); + gate.release("reply"); + // The reply's own end finishes its continuation: the lost Stop's end does not take it. + assert.deepEqual( + [...(yield* Fiber.join(collected))].flatMap((event) => + event.type === "turn.terminal" ? [event.status] : [], + ), + ["interrupted", "failed", "completed"], + ); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())), + ); + + /** + * A prompt fails without a clear rejection, OpenCode never took it, and a + * background subagent's report then wakes the parent into a reply. With + * `endedFirst` that reply ends before its continuation takes it. + */ + const replyAfterUnclearFailure = (endedFirst: boolean) => + Effect.gen(function* () { + const offered = yield* Deferred.make(); + const gate = makeProviderReplayGate(endedFirst ? ["wake"] : ["wake", "reply"]); + const { runtime, thread } = yield* resumed( + [ + ...backgroundLaunch(CHILD), + event("session.execution.succeeded", { sessionID: SESSION }), + out("session.update", { sessionID: CHILD, permissions: t3Rules }), + reply("session.update", null), + out("session.prompt", { sessionID: SESSION, text: "" }), + reply("session.prompt", { + status: 502, + body: { _tag: "UnknownError", message: "bad gateway" }, + }), + // The subagent ends and OpenCode answers its report on its own. + event("session.execution.succeeded", { sessionID: CHILD }, "wake"), + event("session.inbox.enqueued", { + inboxID: "msg_report", + sessionID: SESSION, + item: { + type: "synthetic", + payload: { + text: `\nCHILD_OK\n`, + description: "Sleep", + metadata: { + source: "subagent", + childID: CHILD, + agent: "General", + state: "completed", + }, + }, + delivery: "steer", + }, + }), + event("session.execution.started", { sessionID: SESSION }), + event( + "session.text.ended", + { + sessionID: SESSION, + assistantMessageID: "msg_followup", + ordinal: 0, + text: "CHILD_OK", + }, + endedFirst ? undefined : "reply", + ), + event("session.execution.succeeded", { sessionID: SESSION }), + // The subagent's session runs again: a turn of its own, after the reply's end. + event("session.execution.started", { sessionID: CHILD }), + ], + { replayGate: gate }, + ).pipe( + Effect.provideService(ProviderContinuationRequests.ProviderContinuationRequests, { + offer: () => Deferred.succeed(offered, undefined).pipe(Effect.asVoid), + take: Effect.never, + }), + ); + // The single reader of the runtime's events. + const launched = yield* Deferred.make(); + const childRan = yield* Deferred.make(); + const ended: Array = []; + const allEnded = yield* Deferred.make(); + yield* runtime.events.pipe( + Stream.tap((event) => + Effect.gen(function* () { + if ( + event.type === "provider_turn.updated" && + event.providerTurn.nativeTurnRef?.nativeId?.startsWith(`${CHILD}:turn:`) === true + ) { + yield* Deferred.succeed(childRan, undefined); + } + if (event.type !== "turn_item.updated" && event.type !== "turn.terminal") return; + ended.push(event); + if (event.type === "turn.terminal") yield* Deferred.succeed(launched, undefined); + if (ended.filter((entry) => entry.type === "turn.terminal").length === 3) { + yield* Deferred.succeed(allEnded, undefined); + } + }), + ), + Stream.runDrain, + Effect.forkScoped, + ); + yield* runtime.startTurn(withLineage(thread)); + yield* Deferred.await(launched); + yield* runtime + .startTurn({ ...secondTurn(thread), appThread: withLineage(thread).appThread }) + .pipe(Effect.ignore); + gate.release("wake"); + yield* Deferred.await(offered); + // Everything up to the subagent's next run has been read: the reply ended. + if (endedFirst) yield* Deferred.await(childRan); + yield* runtime.startTurn({ + ...withLineage(thread), + runId: RunId.make("run:opencode2-adapter:wake"), + runOrdinal: 3, + providerTurnOrdinal: 3, + attemptId: RunAttemptId.make("attempt:opencode2-adapter:wake"), + message: { + ...turnInput(thread).message, + messageId: MessageId.make("message:opencode2-adapter:wake"), + createdBy: "agent" as const, + creationSource: "provider" as const, + }, + }); + if (!endedFirst) gate.release("reply"); + // The reply's own end finishes its continuation: the failed prompt's mark does not take it. + yield* Deferred.await(allEnded); + assert.deepEqual( + ended.flatMap((event) => (event.type === "turn.terminal" ? [event.status] : [])), + ["completed", "failed", "completed"], + ); + assert.include( + ended.flatMap((event) => + event.type === "turn_item.updated" && event.turnItem.type === "assistant_message" + ? [event.turnItem.text] + : [], + ), + "CHILD_OK", + ); + }).pipe(Effect.scoped, Effect.provide(TestClock.layer())); + + it.effect("ends a background reply's continuation after a prompt failed unclearly", () => + replyAfterUnclearFailure(false), + ); + + it.effect("ends a finished background reply's continuation after a prompt failed unclearly", () => + replyAfterUnclearFailure(true), + ); + it.effect("does not report a Stop the server says did nothing", () => Effect.gen(function* () { const { runtime, thread } = yield* resumed([ diff --git a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts index 7eb2959974838ed3c6abe649dfb4c25d52ddc26d..db251e005d79762480f9625a03cbb6a0960a7715 100644 --- a/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts +++ b/apps/server/src/orchestration-v2/Adapters/OpenCode2AdapterV2.testkit.ts @@ -286,7 +286,11 @@ function layerRegistry( */ export const openCode2ReplayRuntime = ( entries: ReadonlyArray, - options?: { readonly external?: boolean; readonly borrowers?: { current: number } }, + options?: { + readonly external?: boolean; + readonly borrowers?: { current: number }; + readonly replayGate?: ProviderReplayGate; + }, ) => Effect.gen(function* () { const adapter = yield* makeReplayAdapter( diff --git a/packages/provider-opencode/src/server/v2/adapter.ts b/packages/provider-opencode/src/server/v2/adapter.ts index b23762fcc6e77c47a6548ff9ab6451d4e02d40b3..bf4cdceebae3019534c09e04fba12df611cb1daf 100644 --- a/packages/provider-opencode/src/server/v2/adapter.ts +++ b/packages/provider-opencode/src/server/v2/adapter.ts @@ -340,6 +340,12 @@ interface ThreadState { * so the next execution end belongs to that run; it clears this and ends no turn. */ unsettled: boolean; + /** + * Whether the run `unsettled` waits on had started when it was marked. A + * prompt whose request failed may start its run late; that start is the + * run's own, and only a start after it means its end was lost. + */ + unsettledStarted: boolean; /** The session's location, where its agents' path rules are read. */ directory: string; /** @@ -923,6 +929,11 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const emit = (event: ProviderAdapter.ProviderAdapterV2Event) => Queue.offer(events, event).pipe(Effect.asVoid); const ownerOf = (sessionId: string) => childOwners.get(sessionId) ?? threads.get(sessionId); + /** Marks a run T3 ended while OpenCode may still run it (see `unsettled`). */ + const markUnsettled = (state: ThreadState) => { + state.unsettled = true; + state.unsettledStarted = busy.has(state.sessionId); + }; const newThreadState = ( sessionId: string, @@ -936,6 +947,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr active: undefined, model: undefined, unsettled: false, + unsettledStarted: false, directory, agent: "build", title: undefined, @@ -1838,7 +1850,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr reason, }); if (state.active === undefined) return; - state.unsettled = true; + markUnsettled(state); yield* finishTurn(state, { status: "failed", failure: makeProviderFailure({ @@ -1935,7 +1947,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const [permissions, forms] = listed.value; if (permissions.length === 0 && forms.length === 0) return; // The stopped run's end is not the next turn's. - state.unsettled = true; + markUnsettled(state); for (const request of permissions) { yield* stopStaleRequest(state.sessionId, { type: "permission", id: request.id }); } @@ -2120,7 +2132,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr // goes on. A subagent's session has no Stop end to skip. const asker = threads.get(form.sessionID) ?? state; if (!cancelled) return yield* abandonRequest(asker, "form cancel failed"); - if (asker.subagent === undefined) asker.unsettled = true; + if (asker.subagent === undefined) markUnsettled(asker); yield* finishTurn(asker, { status: "failed", failure: makeProviderFailure({ @@ -2358,7 +2370,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr delivered.every((entry) => state.stoppedChildren.has(entry.childId)); for (const entry of delivered) state.stoppedChildren.delete(entry.childId); if (stopped) { - state.unsettled = true; + markUnsettled(state); yield* client.session .interrupt({ sessionID: Session.ID.make(state.sessionId) }) .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true })); @@ -2484,6 +2496,16 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr event.type === "session.execution.started" || event.type === "unreadable.execution.started" ) { + // A session runs one execution at a time, each opening with `started` + // on this ordered stream. Once the run `unsettled` waits on has + // started, a later start means that run is over: an end not seen by + // now was lost with a dropped stream, and the next end is this + // execution's own. A run not yet started when it was marked starts here. + const state = threads.get(sessionId); + if (state?.unsettled === true) { + if (state.unsettledStarted) state.unsettled = false; + else state.unsettledStarted = true; + } busy.add(sessionId); } else if (event.type === "unreadable.execution.ended" || executionEnd(event.type)) { busy.delete(sessionId); @@ -2515,14 +2537,20 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr return yield* route(event, sessionId); }); - /** Handles one event for its session, as the turn running there sees it. */ + /** + * Handles one event for its session, as the turn running there sees it. + * `held` is set for a held execution's events as its turn replays them. + */ const route = Effect.fnUntraced(function* ( event: OpenCode2Client.OpenCode2StreamEvent, sessionId: string | undefined, + held = false, ) { // The end of the run a timed-out Stop left behind; no turn is its own. + // A held execution's end is its own: any such run was over, or was this + // execution, before it started. const ended = - event.type === "unreadable.execution.ended" || executionEnd(event.type) + !held && (event.type === "unreadable.execution.ended" || executionEnd(event.type)) ? threads.get(sessionId ?? "") : undefined; if (ended?.unsettled === true) { @@ -2777,10 +2805,18 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const background = [...threads.values()].filter( (state) => state.subagent === undefined && hasBackground(state), ); - if (running.length === 0 && background.length === 0 && busy.size === 0) return; + const marked = [...threads.values()].some((state) => state.unsettled); + if (running.length === 0 && background.length === 0 && busy.size === 0 && !marked) return; const active = yield* client.session.active(); // An execution that ended while the stream was down never said so. for (const sessionId of busy) if (!(sessionId in active)) busy.delete(sessionId); + // A run `unsettled` waits on is over once its session is idle: any start + // from here is a new execution's, even if the run's own start was lost in + // the gap. A session still active may not have started that run yet, so + // its next start may still be the run's own; its end clears the mark. + for (const state of threads.values()) { + if (state.unsettled && !(state.sessionId in active)) state.unsettledStarted = true; + } for (const [sessionId, state] of running) { const turn = state.active; if (turn === undefined) continue; @@ -3403,7 +3439,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr /** Feeds a held execution's events to the turn now running on its session. */ const replay = Effect.fnUntraced(function* (wake: Wake) { wake.dropped = true; - for (const event of wake.events) yield* route(event, sessionOfEvent(event)); + for (const event of wake.events) yield* route(event, sessionOfEvent(event), true); }); /** @@ -3438,6 +3474,9 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr if (turn !== undefined && wake.first !== undefined) { turn.providerTurn = { ...turn.providerTurn, nativeTurnRef: ref(wake.first, "weak") }; } + // The execution still runs, so the run a mark waits on is this one or + // was over before it started: its end, still to come, is this turn's. + if (wake.running) state.unsettled = false; yield* replay(wake); }), ); @@ -3520,7 +3559,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr const running = state.wakes.some((wake) => wake.running); for (const wake of state.wakes.splice(0)) wake.dropped = true; if (running) { - state.unsettled = true; + markUnsettled(state); yield* client.session .interrupt({ sessionID: Session.ID.make(state.sessionId) }) .pipe(Effect.timeout(INTERRUPT_TIMEOUT), Effect.ignore({ log: true })); @@ -3896,7 +3935,7 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ? Effect.gen(function* () { // Without a clear rejection the server may have taken the // prompt, so the next turn checks before it prompts again. - if (!CLEAR_PROMPT_REJECTIONS.has(cause._tag)) state.unsettled = true; + if (!CLEAR_PROMPT_REJECTIONS.has(cause._tag)) markUnsettled(state); yield* finishTurn(state, { status: "failed", failure: makeProviderFailure({ cause, class: "provider_error" }), @@ -4004,13 +4043,13 @@ export const make = Effect.fn("OpenCode2Adapter.make")(function* (instanceId: Pr ), ); if (reply._tag === "None") { - state.unsettled = true; + markUnsettled(state); return yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); } // Between the executions a late steer spans nothing runs yet. The // steers are taken back, and a run they already started is skipped. if (!reply.value.interrupted && state.active === turn && turn.heldEnd !== undefined) { - state.unsettled = true; + markUnsettled(state); yield* lock.withPermit(finishTurn(state, { status: "interrupted" })); return yield* cancelStrandedSteers(state); } -- 2.54.0