fix(server): unblock persistent terminal input in Bun builds (#45287)

2026-11-15 20:52:01 +03:00 · 2026-08-26 10:47:16 -04:00
parent 3d7ba38965
commit ab2d251155
2 changed files with 119 additions and 70 deletions
@@ -95,13 +95,6 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experiment
      .handleRaw(
        "persistentPty.connect",
        Effect.fn("PersistentPtyHandler.connect")(function* (ctx) {
-          const exists = yield* pty.get(ctx.params.ptyID).pipe(
-            Effect.as(true),
-            Effect.catchTag("PersistentPty.NotFoundError", () => Effect.succeed(false)),
-            Effect.catchTag("PersistentPty.UnavailableError", () => Effect.succeed(false)),
-          )
-          if (!exists) return HttpServerResponse.empty({ status: 404 })
-
          const url = new URL(ctx.request.url, "http://localhost")
          const ticket = url.searchParams.get(PTY_CONNECT_TICKET_QUERY)
          if (ticket) {
@@ -121,52 +114,62 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experiment
          const write = yield* socket.writer
          const outbox = yield* Queue.unbounded<string | Uint8Array | Socket.CloseEvent>()
          const input = yield* Semaphore.make(1)
-          const attachment = yield* pty
-            .attach(ctx.params.ptyID, {
-              cursor,
-              attachmentID,
-              role,
-              takeover: url.searchParams.get("takeover") === "true",
-              onEvent: (event) => {
-                if (event.type === "output") Queue.offerUnsafe(outbox, event.data)
-                if (event.type === "resized")
-                  Queue.offerUnsafe(
-                    outbox,
-                    JSON.stringify({ ...event, checkpoint: Buffer.from(event.checkpoint).toString("base64") }),
-                  )
-                if (event.type !== "output" && event.type !== "resized")
-                  Queue.offerUnsafe(outbox, JSON.stringify(event))
-              },
-              onEnd: () => Queue.offerUnsafe(outbox, new Socket.CloseEvent(1000)),
-            })
-            .pipe(
-              Effect.catchTags({
-                "PersistentPty.NotFoundError": () => Effect.succeed(undefined),
-                "PersistentPty.UnavailableError": () => Effect.succeed(undefined),
+          let attachment: PersistentPty.Attachment | undefined
+          // Bun's native ws upgrade must start before asynchronous daemon I/O.
+          const onOpen = Effect.gen(function* () {
+            attachment = yield* pty
+              .attach(ctx.params.ptyID, {
+                cursor,
+                attachmentID,
+                role,
+                takeover: url.searchParams.get("takeover") === "true",
+                onEvent: (event) => {
+                  if (event.type === "output") Queue.offerUnsafe(outbox, event.data)
+                  if (event.type === "resized")
+                    Queue.offerUnsafe(
+                      outbox,
+                      JSON.stringify({ ...event, checkpoint: Buffer.from(event.checkpoint).toString("base64") }),
+                    )
+                  if (event.type !== "output" && event.type !== "resized")
+                    Queue.offerUnsafe(outbox, JSON.stringify(event))
+                },
+                onEnd: () => Queue.offerUnsafe(outbox, new Socket.CloseEvent(1000)),
+              })
+              .pipe(
+                Effect.catchTags({
+                  "PersistentPty.NotFoundError": () => Effect.succeed(undefined),
+                  "PersistentPty.UnavailableError": () => Effect.succeed(undefined),
+                }),
+              )
+            if (!attachment) {
+              Queue.offerUnsafe(outbox, new Socket.CloseEvent(4404, "terminal unavailable"))
+              return
+            }
+
+            Queue.offerUnsafe(
+              outbox,
+              JSON.stringify({
+                type: "attached",
+                attachmentID,
+                inputProtocol: framedInput ? 1 : 0,
+                info: attachment.info,
+                role: attachment.role,
+                generation: attachment.generation,
+                replay: {
+                  requestedOffset: attachment.replay.requestedOffset,
+                  availableOffset: attachment.replay.availableOffset,
+                  endOffset: attachment.replay.endOffset,
+                  truncated: attachment.replay.truncated,
+                },
              }),
            )
-          if (!attachment) return HttpServerResponse.empty({ status: 404 })
-
-          Queue.offerUnsafe(
-            outbox,
-            JSON.stringify({
-              type: "attached",
-              attachmentID,
-              inputProtocol: framedInput ? 1 : 0,
-              info: attachment.info,
-              role: attachment.role,
-              generation: attachment.generation,
-              replay: {
-                requestedOffset: attachment.replay.requestedOffset,
-                availableOffset: attachment.replay.availableOffset,
-                endOffset: attachment.replay.endOffset,
-                truncated: attachment.replay.truncated,
-              },
-            }),
-          )
-          if (attachment.replay.data.length > 0) Queue.offerUnsafe(outbox, attachment.replay.data)
-          Queue.offerUnsafe(outbox, JSON.stringify({ type: "replay_complete", endOffset: attachment.replay.endOffset }))
-          attachment.activate()
+            if (attachment.replay.data.length > 0) Queue.offerUnsafe(outbox, attachment.replay.data)
+            Queue.offerUnsafe(
+              outbox,
+              JSON.stringify({ type: "replay_complete", endOffset: attachment.replay.endOffset }),
+            )
+            attachment.activate()
+          })

          const drain = Effect.gen(function* () {
            while (true) {
@@ -178,28 +181,37 @@ export const PersistentPtyHandler = HttpApiBuilder.group(Api, "server.experiment

          yield* Effect.race(
            drain,
-            socket.runRaw((message) =>
-              input.withPermit(
-                Effect.suspend(() => {
-                  const data = typeof message === "string" ? Buffer.from(message) : message
-                  if (!framedInput)
-                    return pty
-                      .input(ctx.params.ptyID, attachmentID, attachment.info.size.cols, attachment.info.size.rows, data)
-                      .pipe(Effect.ignore)
-                  if (data.byteLength < 5) return Effect.void
-                  const view = new DataView(data.buffer, data.byteOffset, data.byteLength)
-                  const type = data[0]
-                  const cols = view.getUint16(1)
-                  const rows = view.getUint16(3)
-                  if ((type !== 0 && type !== 1) || cols === 0 || rows === 0) return Effect.void
-                  if (type === 0) return pty.control(ctx.params.ptyID, attachmentID, cols, rows).pipe(Effect.ignore)
-                  return pty.input(ctx.params.ptyID, attachmentID, cols, rows, data.subarray(5)).pipe(Effect.ignore)
-                }),
-              ),
+            socket.runRaw(
+              (message) =>
+                input.withPermit(
+                  Effect.suspend(() => {
+                    if (!attachment) return Effect.void
+                    const data = typeof message === "string" ? Buffer.from(message) : message
+                    if (!framedInput)
+                      return pty
+                        .input(
+                          ctx.params.ptyID,
+                          attachmentID,
+                          attachment.info.size.cols,
+                          attachment.info.size.rows,
+                          data,
+                        )
+                        .pipe(Effect.ignore)
+                    if (data.byteLength < 5) return Effect.void
+                    const view = new DataView(data.buffer, data.byteOffset, data.byteLength)
+                    const type = data[0]
+                    const cols = view.getUint16(1)
+                    const rows = view.getUint16(3)
+                    if ((type !== 0 && type !== 1) || cols === 0 || rows === 0) return Effect.void
+                    if (type === 0) return pty.control(ctx.params.ptyID, attachmentID, cols, rows).pipe(Effect.ignore)
+                    return pty.input(ctx.params.ptyID, attachmentID, cols, rows, data.subarray(5)).pipe(Effect.ignore)
+                  }),
+                ),
+              { onOpen },
            ),
          ).pipe(
            Effect.catchReason("SocketError", "SocketCloseError", () => Effect.void),
-            Effect.ensuring(Effect.sync(() => attachment.detach())),
+            Effect.ensuring(Effect.sync(() => attachment?.detach())),
            Effect.orDie,
          )
          return HttpServerResponse.empty()
@@ -100,7 +100,44 @@ smoke(
          expect(Buffer.from(snapshot.data.checkpoint, "base64").byteLength).toBeGreaterThan(0)
          expect(snapshot.data.info.output.tail).toBeGreaterThan(0)

+          const ticket = yield* request(
+            base,
+            "POST",
+            `/api/experimental/persistent-pty/${first.id}/connect-token`,
+            undefined,
+            {
+              "x-opencode-ticket": "1",
+            },
+          )
+          if (!isRecord(ticket.data) || typeof ticket.data.ticket !== "string")
+            throw new Error("Invalid connect ticket")
+          const connectTicket = ticket.data.ticket
          yield* request(base, "DELETE", `/api/experimental/persistent-pty/${first.id}`)
+          yield* Effect.promise(async () => {
+            const url = new URL(`/api/experimental/persistent-pty/${first.id}/connect`, base)
+            url.searchParams.set("ticket", "invalid")
+            expect((await fetch(url)).status).toBe(403)
+            url.protocol = "ws:"
+            url.searchParams.set("ticket", connectTicket)
+            const socket = new WebSocket(url)
+            try {
+              const closed = await new Promise<CloseEvent>((resolve, reject) => {
+                const timeout = setTimeout(() => reject(new Error("Removed terminal socket did not close")), 5_000)
+                socket.addEventListener("close", (event) => {
+                  clearTimeout(timeout)
+                  resolve(event)
+                })
+                socket.addEventListener("error", () => {
+                  clearTimeout(timeout)
+                  reject(new Error("Valid ticket should upgrade before the missing terminal is reported"))
+                })
+              })
+              expect(closed.code).toBe(4404)
+              expect(closed.reason).toBe("terminal unavailable")
+            } finally {
+              socket.close()
+            }
+          })
          expect(yield* Effect.promise(() => events.next("persistent-pty.removed"))).toMatchObject({
            data: { sessionID, ptyID: first.id },
          })