test(pty): migrate output isolation to Effect runner (#27235)

packages/opencode/test/pty/pty-output-isolation.test.ts

@@ -1,147 +1,162 @@
-import { describe, expect, test } from "bun:test"
+import { describe, expect } from "bun:test"
-import { AppRuntime } from "../../src/effect/app-runtime"
+import { Bus } from "../../src/bus"
-import { Effect } from "effect"
+import { Config } from "../../src/config/config"
-import { Instance } from "../../src/project/instance"
+import { Plugin } from "../../src/plugin"
-import { WithInstance } from "../../src/project/with-instance"
 import { Pty } from "../../src/pty"
-import { tmpdir } from "../fixture/fixture"
+import { Duration, Effect, Layer, Queue } from "effect"
-import { setTimeout as sleep } from "node:timers/promises"
+import { testEffect } from "../lib/effect"
+
+type Socket = Parameters<Pty.Interface["connect"]>[1]
+
+const it = testEffect(
+  Pty.layer.pipe(
+    Layer.provideMerge(Bus.layer),
+    Layer.provideMerge(Config.defaultLayer),
+    Layer.provideMerge(Plugin.defaultLayer),
+  ),
+)
+const ptyTest = process.platform === "win32" ? it.instance.skip : it.instance
+
+const createPty = Effect.fn("PtyOutputIsolationTest.createPty")(function* (input: Pty.CreateInput) {
+  const pty = yield* Pty.Service
+  return yield* Effect.acquireRelease(pty.create(input), (info) => pty.remove(info.id).pipe(Effect.ignore))
+})
+
+const decodeOutput = (data: string | Uint8Array | ArrayBuffer) =>
+  typeof data === "string"
+    ? data
+    : Buffer.from(data instanceof Uint8Array ? data : new Uint8Array(data)).toString("utf8")
+
+const makeSocket = Effect.fn("PtyOutputIsolationTest.makeSocket")(function* (data: unknown) {
+  const output = yield* Queue.unbounded<string>()
+  const chunks: string[] = []
+  const socket: Socket = {
+    readyState: 1,
+    data,
+    send: (data) => {
+      const text = decodeOutput(data)
+      chunks.push(text)
+      Queue.offerUnsafe(output, text)
+    },
+    close: () => {
+      // no-op (simulate abrupt drop)
+    },
+  }
+
+  return { socket, output, chunks }
+})
+
+const waitForOutput = (output: Queue.Queue<string>, text: string, duration: Duration.Input = "5 seconds") =>
+  Effect.gen(function* () {
+    let received = ""
+    while (!received.includes(text)) {
+      received += yield* Queue.take(output)
+    }
+    return received
+  }).pipe(
+    Effect.timeoutOrElse({
+      duration,
+      orElse: () => Effect.fail(new Error(`timeout waiting for output containing ${JSON.stringify(text)}`)),
+    }),
+  )
+
+const waitForLeakedOutput = (output: Queue.Queue<string>, text: string) =>
+  Effect.gen(function* () {
+    let received = ""
+    while (!received.includes(text)) {
+      received += yield* Queue.take(output)
+    }
+    return received
+  }).pipe(
+    Effect.timeoutOrElse({
+      duration: "100 millis",
+      orElse: () => Effect.succeed(undefined),
+    }),
+  )
 
 describe("pty", () => {
-  test("does not leak output when websocket objects are reused", async () => {
+  ptyTest(
-    await using dir = await tmpdir({ git: true })
+    "does not leak output when websocket objects are reused",
+    () =>
+      Effect.gen(function* () {
+        const pty = yield* Pty.Service
+        const a = yield* createPty({ command: "cat", title: "a" })
+        const b = yield* createPty({ command: "cat", title: "b" })
+        const connectionA = yield* makeSocket({ events: { connection: "a" } })
+        const connectionB = { events: { connection: "b" } }
 
-    await WithInstance.provide({
+        yield* pty.connect(a.id, connectionA.socket)
-      directory: dir.path,
-      fn: () =>
-        AppRuntime.runPromise(
-          Effect.gen(function* () {
-            const pty = yield* Pty.Service
-            const a = yield* pty.create({ command: "cat", title: "a" })
-            const b = yield* pty.create({ command: "cat", title: "b" })
-            try {
-              const outA: string[] = []
-              const outB: string[] = []
 
-              const ws = {
+        const outBQueue = yield* Queue.unbounded<string>()
-                readyState: 1,
+        const outB: string[] = []
-                data: { events: { connection: "a" } },
+        connectionA.socket.data = connectionB
-                send: (data: unknown) => {
+        connectionA.socket.send = (data) => {
-                  outA.push(typeof data === "string" ? data : Buffer.from(data as Uint8Array).toString("utf8"))
+          const text = decodeOutput(data)
-                },
+          outB.push(text)
-                close: () => {
+          Queue.offerUnsafe(outBQueue, text)
-                  // no-op (simulate abrupt drop)
+        }
-                },
+        yield* pty.connect(b.id, connectionA.socket)
-              }
 
-              yield* pty.connect(a.id, ws as any)
+        connectionA.chunks.length = 0
+        outB.length = 0
 
-              ws.data = { events: { connection: "b" } }
+        yield* pty.write(a.id, "AAA\n")
-              ws.send = (data: unknown) => {
+        const verifyA = yield* makeSocket({ events: { connection: "verify-a" } })
-                outB.push(typeof data === "string" ? data : Buffer.from(data as Uint8Array).toString("utf8"))
+        yield* pty.connect(a.id, verifyA.socket)
-              }
+        yield* waitForOutput(verifyA.output, "AAA")
-              yield* pty.connect(b.id, ws as any)
 
-              outA.length = 0
+        expect(outB.join("")).not.toContain("AAA")
-              outB.length = 0
+        expect(yield* waitForLeakedOutput(outBQueue, "AAA")).toBeUndefined()
+      }),
+    { git: true },
+  )
 
-              yield* pty.write(a.id, "AAA\n")
+  ptyTest(
-              yield* Effect.promise(() => sleep(100))
+    "does not leak output when Bun recycles websocket objects before re-connect",
+    () =>
+      Effect.gen(function* () {
+        const pty = yield* Pty.Service
+        const a = yield* createPty({ command: "cat", title: "a" })
+        const outA = yield* makeSocket({ events: { connection: "a" } })
+        const outB = yield* Queue.unbounded<string>()
 
-              expect(outB.join("")).not.toContain("AAA")
+        yield* pty.connect(a.id, outA.socket)
-            } finally {
+        outA.chunks.length = 0
-              yield* pty.remove(a.id)
-              yield* pty.remove(b.id)
-            }
-          }),
-        ),
-    })
-  })
 
-  test("does not leak output when Bun recycles websocket objects before re-connect", async () => {
+        const connectionB = { events: { connection: "b" } }
-    await using dir = await tmpdir({ git: true })
+        outA.socket.data = connectionB
+        outA.socket.send = (data) => {
+          Queue.offerUnsafe(outB, decodeOutput(data))
+        }
 
-    await WithInstance.provide({
+        yield* pty.write(a.id, "AAA\n")
-      directory: dir.path,
+        const verifyA = yield* makeSocket({ events: { connection: "verify-a" } })
-      fn: () =>
+        yield* pty.connect(a.id, verifyA.socket)
-        AppRuntime.runPromise(
+        yield* waitForOutput(verifyA.output, "AAA")
-          Effect.gen(function* () {
-            const pty = yield* Pty.Service
-            const a = yield* pty.create({ command: "cat", title: "a" })
-            try {
-              const outA: string[] = []
-              const outB: string[] = []
 
-              const ws = {
+        expect(yield* waitForLeakedOutput(outB, "AAA")).toBeUndefined()
-                readyState: 1,
+      }),
-                data: { events: { connection: "a" } },
+    { git: true },
-                send: (data: unknown) => {
+  )
-                  outA.push(typeof data === "string" ? data : Buffer.from(data as Uint8Array).toString("utf8"))
-                },
-                close: () => {
-                  // no-op (simulate abrupt drop)
-                },
-              }
 
-              yield* pty.connect(a.id, ws as any)
+  ptyTest(
-              outA.length = 0
+    "treats in-place socket data mutation as the same connection",
+    () =>
+      Effect.gen(function* () {
+        const pty = yield* Pty.Service
+        const a = yield* createPty({ command: "cat", title: "a" })
+        const ctx = { connId: 1 }
+        const out = yield* makeSocket(ctx)
 
-              ws.data = { events: { connection: "b" } }
+        yield* pty.connect(a.id, out.socket)
-              ws.send = (data: unknown) => {
+        out.chunks.length = 0
-                outB.push(typeof data === "string" ? data : Buffer.from(data as Uint8Array).toString("utf8"))
-              }
 
-              yield* pty.write(a.id, "AAA\n")
+        ctx.connId = 2
-              yield* Effect.promise(() => sleep(100))
 
-              expect(outB.join("")).not.toContain("AAA")
+        yield* pty.write(a.id, "AAA\n")
-            } finally {
-              yield* pty.remove(a.id)
-            }
-          }),
-        ),
-    })
-  })
 
-  test("treats in-place socket data mutation as the same connection", async () => {
+        expect(yield* waitForOutput(out.output, "AAA")).toContain("AAA")
-    await using dir = await tmpdir({ git: true })
+      }),
-
+    { git: true },
-    await WithInstance.provide({
+  )
-      directory: dir.path,
-      fn: () =>
-        AppRuntime.runPromise(
-          Effect.gen(function* () {
-            const pty = yield* Pty.Service
-            const a = yield* pty.create({ command: "cat", title: "a" })
-            try {
-              const out: string[] = []
-
-              const ctx = { connId: 1 }
-              const ws = {
-                readyState: 1,
-                data: ctx,
-                send: (data: unknown) => {
-                  out.push(typeof data === "string" ? data : Buffer.from(data as Uint8Array).toString("utf8"))
-                },
-                close: () => {
-                  // no-op
-                },
-              }
-
-              yield* pty.connect(a.id, ws as any)
-              out.length = 0
-
-              ctx.connId = 2
-
-              yield* pty.write(a.id, "AAA\n")
-              yield* Effect.promise(() => sleep(100))
-
-              expect(out.join("")).toContain("AAA")
-            } finally {
-              yield* pty.remove(a.id)
-            }
-          }),
-        ),
-    })
-  })
 })