fix(pty): pty session handle leak (anomalyco#15599)

Author: kikuchan

packages/opencode/src/pty/index.ts

@@ -239,6 +239,7 @@ export namespace Pty {
       session.bufferCursor += excess
     })
     ptyProcess.onExit(({ exitCode }) => {
+      if (session.info.status === "exited") return
       log.info("session exited", { id, exitCode })
       const wasKilled = session.info.status === "killed"
       if (!wasKilled) {
@@ -271,15 +272,8 @@ export namespace Pty {
         Bus.publish(Event.Output, { id, chunk: msg, cursor: session.cursor })
       }
 
-      for (const [key, ws] of session.subscribers.entries()) {
-        try {
-          if (ws.data === key) ws.close()
-        } catch {
-          // ignore
-        }
-      }
-      session.subscribers.clear()
       Bus.publish(Event.Exited, { id, exitCode })
+      remove(id)
     })
     Bus.publish(Event.Created, { info })
     return info
@@ -460,6 +454,7 @@ export namespace Pty {
       session.bufferCursor += excess
     })
     ptyProcess.onExit(({ exitCode }) => {
+      if (session.info.status === "exited") return
       log.info("session exited after restart", { id, exitCode })
       const wasKilled = session.info.status === "killed"
       if (!wasKilled) {
@@ -492,21 +487,15 @@ export namespace Pty {
         Bus.publish(Event.Output, { id, chunk: msg, cursor: session.cursor })
       }
 
-      for (const [key, ws] of session.subscribers.entries()) {
-        try {
-          if (ws.data === key) ws.close()
-        } catch {
-          // ignore
-        }
-      }
-      session.subscribers.clear()
       Bus.publish(Event.Exited, { id, exitCode })
+      remove(id)
     })
   }
 
   export async function remove(id: string) {
     const session = state().get(id)
     if (!session) return
+    state().delete(id)
     log.info("removing session", { id })
     try {
       session.process.kill()
@@ -519,7 +508,6 @@ export namespace Pty {
       }
     }
     session.subscribers.clear()
-    state().delete(id)
     Bus.publish(Event.Deleted, { id })
   }
 

packages/opencode/test/pty/pty-session.test.ts

@@ -0,0 +1,87 @@
+import { describe, expect, test } from "bun:test"
+import { Bus } from "../../src/bus"
+import { Instance } from "../../src/project/instance"
+import { Pty } from "../../src/pty"
+import { tmpdir } from "../fixture/fixture"
+import { setTimeout as sleep } from "node:timers/promises"
+
+const wait = async (fn: () => boolean, ms = 2000) => {
+  const end = Date.now() + ms
+  while (Date.now() < end) {
+    if (fn()) return
+    await sleep(25)
+  }
+  throw new Error("timeout waiting for pty events")
+}
+
+const pick = (log: Array<{ type: "created" | "exited" | "deleted"; id: string }>, id: string) => {
+  return log.filter((evt) => evt.id === id).map((evt) => evt.type)
+}
+
+describe("pty", () => {
+  test("publishes created, exited, deleted in order for /bin/ls + remove", async () => {
+    if (process.platform === "win32") return
+
+    await using dir = await tmpdir({ git: true })
+
+    await Instance.provide({
+      directory: dir.path,
+      fn: async () => {
+        const log: Array<{ type: "created" | "exited" | "deleted"; id: string }> = []
+        const off = [
+          Bus.subscribe(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
+          Bus.subscribe(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
+          Bus.subscribe(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
+        ]
+
+        let id = ""
+        try {
+          const info = await Pty.create({ command: "/bin/ls", title: "ls" })
+          id = info.id
+
+          await wait(() => pick(log, id).includes("exited"))
+
+          await Pty.remove(id)
+          await wait(() => pick(log, id).length >= 3)
+          expect(pick(log, id)).toEqual(["created", "exited", "deleted"])
+        } finally {
+          off.forEach((x) => x())
+          if (id) await Pty.remove(id)
+        }
+      },
+    })
+  })
+
+  test("publishes created, exited, deleted in order for /bin/sh + remove", async () => {
+    if (process.platform === "win32") return
+
+    await using dir = await tmpdir({ git: true })
+
+    await Instance.provide({
+      directory: dir.path,
+      fn: async () => {
+        const log: Array<{ type: "created" | "exited" | "deleted"; id: string }> = []
+        const off = [
+          Bus.subscribe(Pty.Event.Created, (evt) => log.push({ type: "created", id: evt.properties.info.id })),
+          Bus.subscribe(Pty.Event.Exited, (evt) => log.push({ type: "exited", id: evt.properties.id })),
+          Bus.subscribe(Pty.Event.Deleted, (evt) => log.push({ type: "deleted", id: evt.properties.id })),
+        ]
+
+        let id = ""
+        try {
+          const info = await Pty.create({ command: "/bin/sh", title: "sh" })
+          id = info.id
+
+          await sleep(100)
+
+          await Pty.remove(id)
+          await wait(() => pick(log, id).length >= 3)
+          expect(pick(log, id)).toEqual(["created", "exited", "deleted"])
+        } finally {
+          off.forEach((x) => x())
+          if (id) await Pty.remove(id)
+        }
+      },
+    })
+  })
+})