diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts index baded8094ba65bdc9ca68eb9e5e30b5c2a27c013..40d034089ae95f23e89aee92bcc045f5c5d7cbbd 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.test.ts @@ -2,6 +2,7 @@ import * as NodeAssert from "node:assert/strict"; import * as NodeServices from "@effect/platform-node/NodeServices"; import { it } from "@effect/vitest"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as Deferred from "effect/Deferred"; @@ -781,6 +782,56 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => { }), ); + it.effect( + "stopSession completes after sendTurn is interrupted before the prompt is submitted", + () => + Effect.gen(function* () { + const adapter = yield* OpenCodeAdapter; + const threadId = asThreadId("thread-opencode-interrupted-before-submit"); + runtimeMock.state.createdSessionIds.push("ses_interrupted_before_submit"); + yield* adapter.startSession({ + provider: ProviderDriverKind.make("opencode"), + threadId, + runtimeMode: "full-access", + }); + + const baseClock = yield* Clock.clockWith(Effect.succeed); + const reached = yield* Deferred.make(); + const release = yield* Deferred.make(); + let calls = 0; + const gatedClock: Clock.Clock = { + ...baseClock, + currentTimeMillisUnsafe: () => baseClock.currentTimeMillisUnsafe(), + currentTimeMillis: Effect.suspend(() => { + calls += 1; + return calls === 2 + ? Deferred.succeed(reached, undefined).pipe( + Effect.andThen(Deferred.await(release)), + Effect.andThen(baseClock.currentTimeMillis), + ) + : baseClock.currentTimeMillis; + }), + }; + + const sendFiber = yield* adapter + .sendTurn({ + threadId, + input: "This prompt must not be submitted", + modelSelection: createModelSelection( + ProviderInstanceId.make("opencode"), + "opencode/kimi-k3", + ), + }) + .pipe(Effect.provideService(Clock.Clock, gatedClock), Effect.exit, Effect.forkChild); + yield* Deferred.await(reached); + yield* Fiber.interrupt(sendFiber); + + const stopExit = yield* adapter.stopSession(threadId).pipe(Effect.exit); + NodeAssert.equal(Exit.isSuccess(stopExit), true); + NodeAssert.equal(yield* adapter.hasSession(threadId), false); + }), + ); + it.effect("aborts a held teardown request before closing the session scope", () => Effect.gen(function* () { const adapter = yield* OpenCodeAdapter; diff --git a/apps/server/src/provider/Layers/OpenCodeAdapter.ts b/apps/server/src/provider/Layers/OpenCodeAdapter.ts index 04535edbd3afb22d1e54a66e5daf75a692dd6b34..34ebd1b544fb46dc3da13530bfd55da6e3f2785c 100644 --- a/apps/server/src/provider/Layers/OpenCodeAdapter.ts +++ b/apps/server/src/provider/Layers/OpenCodeAdapter.ts @@ -3531,7 +3531,18 @@ export function makeOpenCodeAdapter( ? { resumeCursor: context.session.resumeCursor } : {}), }; - }), + }).pipe( + // stopSession waits on submissionSettled; an interrupted sendTurn + // must not leave it pending. + Effect.ensuring( + Effect.suspend(() => { + const admission = context.promptAdmission; + return admission + ? Deferred.succeed(admission.submissionSettled, undefined).pipe(Effect.ignore) + : Effect.void; + }), + ), + ), ); });