test(server): verify real PTY websocket transports

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

@@ -188,14 +188,15 @@ smoke(
       const first = Schema.decodeUnknownSync(PersistentPty.Info)(
         (yield* request(base, "POST", `/api/experimental/session/${sessionID}/terminal`, {
           command: "/usr/bin/env",
-          args: ["/bin/sh", "-c", "stty -echo; printf terminal-one; cat"],
+          // Exercise transport bursts without macOS's 1024-byte canonical line buffer limit.
+          args: ["/bin/sh", "-c", "stty -echo -icanon min 1 time 0; printf terminal-one; cat"],
           cwd: process.cwd(),
           title: "first",
           env: {},
         })).data,
       )
       expect(first.command).toBe("/usr/bin/env")
-      expect(first.args).toEqual(["/bin/sh", "-c", "stty -echo; printf terminal-one; cat"])
+      expect(first.args).toEqual(["/bin/sh", "-c", "stty -echo -icanon min 1 time 0; printf terminal-one; cat"])
       expect(first.cwd).toBe(process.cwd())
       expect(yield* Effect.promise(() => events.next("persistent-pty.added"))).toMatchObject({
         data: { sessionID, terminal: { id: first.id } },
@@ -570,7 +571,13 @@ async function verifySharedControl(base: string, ptyID: string) {
     second.socket.send(inputFrame(70, 20, "from-second\n"))
     await waitForSocketOutput([first, second], "from-second")
 
-    await waitForForegroundProcess([first, second], "cat")
+    // opencode-pty v0.1.13 uses /proc on Linux and returns None on other platforms:
+    // https://github.com/anomalyco/opencode-pty/blob/v0.1.13/src/service.rs#L1091-L1128
+    if (process.platform === "linux") await waitForForegroundProcess([first, second], "cat")
+    if (process.platform !== "linux") {
+      expect(first.foregroundProcess).toBeNull()
+      expect(second.foregroundProcess).toBeNull()
+    }
 
     for (const character of "printf abc | rev\n") second.socket.send(inputFrame(70, 20, character))
     await waitForSocketOutput([first, second], "printf abc | rev")
@@ -578,12 +585,36 @@ async function verifySharedControl(base: string, ptyID: string) {
     second.socket.send(inputFrame(70, 20, "x".repeat(1024)))
     second.socket.send(inputFrame(70, 20, "after-burst\n"))
     await waitForSocketOutput([first, second], "after-burst")
+    expect(first.output).toContain("x".repeat(1024) + "after-burst")
+    expect(second.output).toContain("x".repeat(1024) + "after-burst")
     expect(first.closed).toBeFalse()
     expect(second.closed).toBeFalse()
     expect(first.resizes).toBeGreaterThan(0)
     expect(second.resizes).toBeGreaterThan(0)
     expect(first.output).not.toContain("\0")
     expect(second.output).not.toContain("\0")
+    expect(first.size).toEqual({ cols: 70, rows: 20 })
+    expect(second.size).toEqual({ cols: 70, rows: 20 })
+
+    await waitForController(first, "second")
+    second.socket.close()
+    // Promotion is emitted by the daemon only after the server detaches the closed socket.
+    await waitForController(first, "first")
+    expect(second.closed).toBeTrue()
+    first.socket.send(inputFrame(70, 20, "after-detach\n"))
+    await waitForSocketOutput([first], "after-detach")
+
+    const replay = await openTerminalSocket(base, ptyID, "replay", "observer")
+    try {
+      // openTerminalSocket resolves at replay_complete, before any new input is sent.
+      expect(replay.output).toContain("from-first")
+      expect(replay.output).toContain("from-second")
+      expect(replay.output).toContain("after-detach")
+      first.socket.send(inputFrame(70, 20, "after-replay\n"))
+      await waitForSocketOutput([first, replay], "after-replay")
+    } finally {
+      replay.socket.close()
+    }
   } finally {
     first.socket.close()
     second.socket.close()
@@ -615,6 +646,8 @@ async function openTerminalSocket(
     output: "",
     closed: false,
     resizes: 0,
+    size: undefined as { cols: number; rows: number } | undefined,
+    controller: undefined as string | undefined,
     foregroundProcess: null as string | null,
   }
   state.socket.binaryType = "arraybuffer"
@@ -636,6 +669,12 @@ async function openTerminalSocket(
           return
         }
         state.resizes++
+        if (typeof message.cols === "number" && typeof message.rows === "number")
+          state.size = { cols: message.cols, rows: message.rows }
+        return
+      }
+      if (message.type === "controller_changed") {
+        state.controller = typeof message.attachmentID === "string" ? message.attachmentID : undefined
         return
       }
       if (message.type === "foreground_process_changed") {
@@ -668,6 +707,15 @@ async function openTerminalSocket(
   return state
 }
 
+async function waitForController(socket: { controller: string | undefined; closed: boolean }, expected: string) {
+  for (let attempt = 0; attempt < 100; attempt++) {
+    if (socket.controller === expected) return
+    if (socket.closed) throw new Error("Persistent PTY observer disconnected")
+    await Bun.sleep(20)
+  }
+  throw new Error(`Persistent PTY controller did not become ${expected}: ${socket.controller}`)
+}
+
 async function waitForForegroundProcess(
   sockets: Array<{ foregroundProcess: string | null; closed: boolean }>,
   expected: string,

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

@@ -0,0 +1,125 @@
+import { expect } from "bun:test"
+import { Effect } from "effect"
+import { HttpServer } from "effect/unstable/http"
+import fs from "node:fs/promises"
+import os from "node:os"
+import path from "node:path"
+import { OpenCode } from "../../client/src/promise/index"
+import { it } from "../../core/test/lib/effect"
+import { ServerProcess } from "../src/process"
+
+const live = process.platform === "win32" ? it.live.skip : it.live
+
+live(
+  "PTY websocket preserves replay, text and binary input, resize, reconnect, and exit",
+  () =>
+    Effect.gen(function* () {
+      const directory = yield* Effect.acquireRelease(
+        Effect.promise(() => fs.mkdtemp(path.join(os.tmpdir(), "opencode-pty-transport-"))),
+        (directory) => Effect.promise(() => fs.rm(directory, { recursive: true, force: true })),
+      )
+      const server = yield* ServerProcess.start<never, never>({
+        hostname: "127.0.0.1",
+        port: 0,
+        password: "secret",
+        app: { version: "test-version" },
+        database: { path: ":memory:" },
+        models: { fetch: false },
+        config: { directory, project: false, content: "{}" },
+        fs: { filewatcher: false },
+      })
+      const base = HttpServer.formatAddress(server.address)
+      const client = OpenCode.make({ baseUrl: base, headers: { authorization: `Basic ${btoa("opencode:secret")}` } })
+      const location = { directory }
+      const terminal = yield* Effect.acquireRelease(
+        Effect.promise(() =>
+          client.pty.create({
+            location,
+            command: "/bin/sh",
+            args: [
+              "-c",
+              "stty -echo; printf 'ready\\n'; while IFS= read -r line; do case \"$line\" in size) stty size;; quit) exit 0;; *) printf 'reply:%s\\n' \"$line\";; esac; done",
+            ],
+            cwd: directory,
+          }),
+        ),
+        (terminal) => Effect.promise(() => client.pty.remove({ location, ptyID: terminal.data.id })),
+      )
+      yield* Effect.promise(async () => {
+        const connect = async () => {
+          const token = await client.pty.connect.token({ location, ptyID: terminal.data.id, "x-opencode-ticket": "1" })
+          const url = new URL(`/api/pty/${terminal.data.id}/connect`, base)
+          url.protocol = "ws:"
+          url.searchParams.set("location[directory]", directory)
+          url.searchParams.set("ticket", token.data.ticket)
+          const socket = new WebSocket(url)
+          socket.binaryType = "arraybuffer"
+          const state = { output: "", frames: [] as string[], closed: false, code: 0, error: false }
+          socket.addEventListener("message", (event) => {
+            if (typeof event.data === "string") {
+              state.output += event.data
+              state.frames.push("output")
+              return
+            }
+            const bytes = new Uint8Array(event.data)
+            if (bytes[0] === 0) {
+              state.frames.push("cursor")
+              return
+            }
+            state.output += new TextDecoder().decode(bytes)
+            state.frames.push("output")
+          })
+          socket.addEventListener("close", (event) => {
+            state.closed = true
+            state.code = event.code
+          })
+          socket.addEventListener("error", () => {
+            state.error = true
+          })
+          await waitFor(() => state.frames.includes("cursor") || state.closed || state.error)
+          expect(state.error).toBeFalse()
+          expect(state.closed).toBeFalse()
+          return { socket, state }
+        }
+        const first = await connect()
+        try {
+          await waitFor(() => first.state.output.includes("ready"))
+          first.socket.send("text\n")
+          first.socket.send(new TextEncoder().encode("binary\n"))
+          await waitFor(() => first.state.output.includes("reply:binary"))
+          expect(first.state.output).toContain("reply:text")
+          await client.pty.update({ location, ptyID: terminal.data.id, size: { rows: 31, cols: 93 } })
+          first.socket.send("size\n")
+          await waitFor(() => first.state.output.includes("31 93"))
+          first.socket.close()
+          await waitFor(() => first.state.closed)
+          const second = await connect()
+          try {
+            expect(second.state.output).toContain("reply:text")
+            expect(second.state.output).toContain("reply:binary")
+            expect(second.state.frames.at(-1)).toBe("cursor")
+            second.socket.send("reconnected\n")
+            await waitFor(() => second.state.output.includes("reply:reconnected"))
+            expect(first.state.output).not.toContain("reply:reconnected")
+            second.socket.send("quit\n")
+            await waitFor(() => second.state.closed)
+            expect(second.state.code).toBe(1000)
+            expect((await client.pty.get({ location, ptyID: terminal.data.id })).data.status).toBe("exited")
+          } finally {
+            second.socket.close()
+          }
+        } finally {
+          first.socket.close()
+        }
+      })
+    }),
+  20_000,
+)
+
+async function waitFor(predicate: () => boolean) {
+  for (let attempt = 0; attempt < 200; attempt++) {
+    if (predicate()) return
+    await Bun.sleep(20)
+  }
+  throw new Error("PTY transport did not reach the expected state")
+}