fix(opencode): settle interrupted prompt admissions

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";
@@ -833,6 +834,54 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => {
     }),
   );
 
+  it.effect("stops after sendTurn is interrupted before prompt submission", () =>
+    Effect.gen(function* () {
+      const adapter = yield* OpenCodeAdapter;
+      const threadId = asThreadId("thread-opencode-interrupted-before-submit");
+      yield* adapter.startSession({
+        provider: ProviderDriverKind.make("opencode"),
+        threadId,
+        runtimeMode: "full-access",
+      });
+
+      const baseClock = yield* Clock.clockWith(Effect.succeed);
+      const admissionPublished = yield* Deferred.make<void>();
+      const release = yield* Deferred.make<void>();
+      let calls = 0;
+      const gatedClock: Clock.Clock = {
+        ...baseClock,
+        currentTimeMillisUnsafe: () => baseClock.currentTimeMillisUnsafe(),
+        currentTimeMillis: Effect.suspend(() => {
+          calls += 1;
+          return calls === 2
+            ? Deferred.succeed(admissionPublished, 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(admissionPublished);
+      yield* Fiber.interrupt(sendFiber);
+      NodeAssert.equal(runtimeMock.state.promptCalls.length, 0);
+
+      yield* adapter.stopSession(threadId);
+      NodeAssert.equal(yield* adapter.hasSession(threadId), false);
+      NodeAssert.deepEqual(runtimeMock.state.closeCalls, ["http://127.0.0.1:9999"]);
+    }),
+  );
+
   it.effect("aborts a held teardown request before closing the session scope", () =>
     Effect.gen(function* () {
       const adapter = yield* OpenCodeAdapter;

apps/server/src/provider/Layers/OpenCodeAdapter.ts

@@ -3597,6 +3597,7 @@ export function makeOpenCodeAdapter(
         });
       }
 
+      let submissionSettled: Deferred.Deferred<void> | undefined;
       return yield* context.promptSemaphore.withPermit(
         Effect.gen(function* () {
           yield* requireNotQuarantined(context);
@@ -3658,6 +3659,7 @@ export function makeOpenCodeAdapter(
             },
           };
           context.promptGeneration = promptGeneration;
+          submissionSettled = promptAdmission.submissionSettled;
           context.promptAdmission = promptAdmission;
 
           context.activeTurnId = turnId;
@@ -3937,7 +3939,18 @@ export function makeOpenCodeAdapter(
               ? { resumeCursor: context.session.resumeCursor }
               : {}),
           };
-        }),
+        }).pipe(
+          // Stop waits for submission to settle even when the caller is
+          // interrupted before the prompt fiber exists. Complete only this
+          // send's admission, while it still owns the prompt permit.
+          Effect.ensuring(
+            Effect.suspend(() =>
+              submissionSettled
+                ? Deferred.succeed(submissionSettled, undefined).pipe(Effect.ignore)
+                : Effect.void,
+            ),
+          ),
+        ),
       );
     });