fix(server): detach PTYs when sockets close (#46068)

packages/server/src/handlers/persistent-pty.ts

@@ -12,6 +12,7 @@ import { HttpApiBuilder, HttpApiSchema } from "effect/unstable/httpapi"
 import { Socket } from "effect/unstable/socket"
 import { Api } from "../api"
 import { CorsConfig, isAllowedRequestOrigin } from "../cors"
+import { runPtySocket } from "./pty-socket"
 
 export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experimental", (handlers) =>
   Effect.gen(function* () {
@@ -191,7 +192,7 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experiment
             }
           })
 
-          yield* Effect.race(
+          yield* runPtySocket(
             drain,
             socket.runRaw(
               (message) =>
@@ -221,9 +222,9 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experiment
                 ),
               { onOpen },
             ),
+            () => attachment?.detach(),
           ).pipe(
             Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void),
-            Effect.ensuring(Effect.sync(() => attachment?.detach())),
             Effect.orDie,
           )
           return HttpServerResponse.empty()

packages/server/src/handlers/pty-socket.ts

@@ -0,0 +1,9 @@
+import { Effect } from "effect"
+
+export function runPtySocket<A, E, R, A2, E2, R2>(
+  drain: Effect.Effect<A, E, R>,
+  socket: Effect.Effect<A2, E2, R2>,
+  detach: () => void,
+) {
+  return Effect.raceFirst(drain, socket).pipe(Effect.ensuring(Effect.sync(detach)))
+}

packages/server/src/handlers/pty.ts

@@ -17,6 +17,7 @@ import {
 } from "@opencode-ai/protocol/groups/pty"
 import { response } from "../location"
 import { PtyEnvironment } from "../pty-environment"
+import { runPtySocket } from "./pty-socket"
 
 const ticketScope = Effect.gen(function* () {
   const location = yield* Location.Service
@@ -209,15 +210,15 @@ export const PtyHandler = HttpApiBuilder.group(Api, "server.pty", (handlers) =>
             }
           })
 
-          yield* Effect.race(
+          yield* runPtySocket(
             drain,
             socket.runRaw((message) => {
               const decoded = PtyProtocol.decodeInput(message)
               if (decoded !== undefined) attachment.write(decoded)
             }),
+            attachment.detach,
           ).pipe(
             Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void),
-            Effect.ensuring(Effect.sync(() => attachment.detach())),
             Effect.orDie,
           )
           return HttpServerResponse.empty()

packages/server/test/pty-socket.test.ts

@@ -0,0 +1,16 @@
+import { expect } from "bun:test"
+import { Effect, Option, Result } from "effect"
+import { it } from "../../core/test/lib/effect"
+import { runPtySocket } from "../src/handlers/pty-socket"
+
+it.live("detaches when the socket fails while the outbox drain is blocked", () =>
+  Effect.gen(function* () {
+    const state = { detached: false }
+    const result = yield* runPtySocket(Effect.never, Effect.fail("socket closed"), () => {
+      state.detached = true
+    }).pipe(Effect.result, Effect.timeoutOption("100 millis"))
+
+    expect(Option.isSome(result) && Result.isFailure(result.value)).toBeTrue()
+    expect(state.detached).toBeTrue()
+  }),
+)