Merge branch 'dev' into kit/effectify-pty

packages/opencode/specs/effect-migration.md

@@ -128,7 +128,7 @@ Still open and likely worth migrating:
 
 - [ ] `Plugin`
 - [ ] `ToolRegistry`
-- [ ] `Pty`
+- [x] `Pty`
 - [ ] `Worktree`
 - [ ] `Installation`
 - [ ] `Bus`

packages/opencode/src/effect/instances.ts

@@ -4,6 +4,7 @@ import { FileTime } from "@/file/time-service"
 import { FileWatcher } from "@/file/watcher"
 import { Format } from "@/format/service"
 import { Permission } from "@/permission/service"
+import { Pty } from "@/pty"
 import { Instance } from "@/project/instance"
 import { Vcs } from "@/project/vcs"
 import { ProviderAuth } from "@/provider/auth-service"
@@ -24,6 +25,7 @@ export type InstanceServices =
   | FileTime.Service
   | Format.Service
   | File.Service
+  | Pty.Service
   | Skill.Service
   | Snapshot.Service
 
@@ -44,6 +46,7 @@ function lookup(_key: string) {
     FileTime.layer,
     Format.layer,
     File.layer,
+    Pty.layer,
     Skill.defaultLayer,
     Snapshot.defaultLayer,
   ).pipe(Layer.provide(ctx))

packages/opencode/src/effect/runtime.ts

@@ -20,6 +20,10 @@ export function runPromiseInstance<A, E>(effect: Effect.Effect<A, E, InstanceSer
   return runtime.runPromise(effect.pipe(Effect.provide(Instances.get(Instance.directory))))
 }
 
+export function runSyncInstance<A, E>(effect: Effect.Effect<A, E, InstanceServices>) {
+  return runtime.runSync(effect.pipe(Effect.provide(Instances.get(Instance.directory))))
+}
+
 export function disposeRuntime() {
   return runtime.dispose()
 }

packages/opencode/src/pty/index.ts

@@ -1,13 +1,12 @@
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
+import { InstanceContext } from "@/effect/instance-context"
 import { type IPty } from "bun-pty"
 import z from "zod"
 import { Log } from "../util/log"
-import { Instance } from "../project/instance"
 import { lazy } from "@opencode-ai/util/lazy"
-import { Shell } from "@/shell/shell"
-import { Plugin } from "@/plugin"
 import { PtyID } from "./schema"
+import { Effect, Layer, ServiceMap } from "effect"
 
 export namespace Pty {
   const log = Log.create({ service: "pty" })
@@ -90,9 +89,31 @@ export namespace Pty {
     subscribers: Map<unknown, Socket>
   }
 
-  const state = Instance.state(
+  export interface Interface {
-    () => new Map<PtyID, ActiveSession>(),
+    readonly list: () => Effect.Effect<Info[]>
-    async (sessions) => {
+    readonly get: (id: PtyID) => Effect.Effect<Info | undefined>
+    readonly create: (input: CreateInput) => Effect.Effect<Info>
+    readonly update: (id: PtyID, input: UpdateInput) => Effect.Effect<Info | undefined>
+    readonly remove: (id: PtyID) => Effect.Effect<void>
+    readonly resize: (id: PtyID, cols: number, rows: number) => Effect.Effect<void>
+    readonly write: (id: PtyID, data: string) => Effect.Effect<void>
+    readonly connect: (
+      id: PtyID,
+      ws: Socket,
+      cursor?: number,
+    ) => Effect.Effect<{ onMessage: (message: string | ArrayBuffer) => void; onClose: () => void } | undefined>
+  }
+
+  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Pty") {}
+
+  export const layer = Layer.effect(
+    Service,
+    Effect.gen(function* () {
+      const instance = yield* InstanceContext
+      const sessions = new Map<PtyID, ActiveSession>()
+
+      yield* Effect.addFinalizer(() =>
+        Effect.sync(() => {
           for (const session of sessions.values()) {
             try {
               session.process.kill()
@@ -100,24 +121,41 @@ export namespace Pty {
             for (const [key, ws] of session.subscribers.entries()) {
               try {
                 if (ws.data === key) ws.close()
-          } catch {
+              } catch {}
-            // ignore
-          }
             }
           }
           sessions.clear()
-    },
+        }),
       )
 
-  export function list() {
+      const removeSession = (id: PtyID) => {
-    return Array.from(state().values()).map((s) => s.info)
+        const session = sessions.get(id)
+        if (!session) return
+        sessions.delete(id)
+        log.info("removing session", { id })
+        try {
+          session.process.kill()
+        } catch {}
+        for (const [key, ws] of session.subscribers.entries()) {
+          try {
+            if (ws.data === key) ws.close()
+          } catch {}
+        }
+        session.subscribers.clear()
+        Bus.publish(Event.Deleted, { id: session.info.id })
       }
 
-  export function get(id: PtyID) {
+      const list = Effect.fn("Pty.list")(function* () {
-    return state().get(id)?.info
+        return Array.from(sessions.values()).map((s) => s.info)
-  }
+      })
 
-  export async function create(input: CreateInput) {
+      const get = Effect.fn("Pty.get")(function* (id: PtyID) {
+        return sessions.get(id)?.info
+      })
+
+      const create = Effect.fn("Pty.create")(function* (input: CreateInput) {
+        return yield* Effect.promise(async () => {
+          const [{ Shell }, { Plugin }] = await Promise.all([import("@/shell/shell"), import("@/plugin")])
           const id = PtyID.ascending()
           const command = input.command || Shell.preferred()
           const args = input.args || []
@@ -125,7 +163,7 @@ export namespace Pty {
             args.push("-l")
           }
 
-    const cwd = input.cwd || Instance.directory
+          const cwd = input.cwd || instance.directory
           const shellEnv = await Plugin.trigger("shell.env", { cwd }, { env: {} })
           const env = {
             ...process.env,
@@ -166,9 +204,8 @@ export namespace Pty {
             cursor: 0,
             subscribers: new Map(),
           }
-    state().set(id, session)
+          sessions.set(id, session)
-    ptyProcess.onData(
+          ptyProcess.onData((chunk) => {
-      Instance.bind((chunk) => {
             session.cursor += chunk.length
 
             for (const [key, ws] of session.subscribers.entries()) {
@@ -176,12 +213,10 @@ export namespace Pty {
                 session.subscribers.delete(key)
                 continue
               }
-
               if (ws.data !== key) {
                 session.subscribers.delete(key)
                 continue
               }
-
               try {
                 ws.send(chunk)
               } catch {
@@ -194,23 +229,21 @@ export namespace Pty {
             const excess = session.buffer.length - BUFFER_LIMIT
             session.buffer = session.buffer.slice(excess)
             session.bufferCursor += excess
-      }),
+          })
-    )
+          ptyProcess.onExit(({ exitCode }) => {
-    ptyProcess.onExit(
-      Instance.bind(({ exitCode }) => {
             if (session.info.status === "exited") return
             log.info("session exited", { id, exitCode })
             session.info.status = "exited"
             Bus.publish(Event.Exited, { id, exitCode })
-        remove(id)
+            removeSession(id)
-      }),
+          })
-    )
           Bus.publish(Event.Created, { info })
           return info
-  }
+        })
+      })
 
-  export async function update(id: PtyID, input: UpdateInput) {
+      const update = Effect.fn("Pty.update")(function* (id: PtyID, input: UpdateInput) {
-    const session = state().get(id)
+        const session = sessions.get(id)
         if (!session) return
         if (input.title) {
           session.info.title = input.title
@@ -220,54 +253,35 @@ export namespace Pty {
         }
         Bus.publish(Event.Updated, { info: session.info })
         return session.info
-  }
+      })
 
-  export async function remove(id: PtyID) {
+      const remove = Effect.fn("Pty.remove")(function* (id: PtyID) {
-    const session = state().get(id)
+        removeSession(id)
-    if (!session) return
+      })
-    state().delete(id)
-    log.info("removing session", { id })
-    try {
-      session.process.kill()
-    } catch {}
-    for (const [key, ws] of session.subscribers.entries()) {
-      try {
-        if (ws.data === key) ws.close()
-      } catch {
-        // ignore
-      }
-    }
-    session.subscribers.clear()
-    Bus.publish(Event.Deleted, { id: session.info.id })
-  }
 
-  export function resize(id: PtyID, cols: number, rows: number) {
+      const resize = Effect.fn("Pty.resize")(function* (id: PtyID, cols: number, rows: number) {
-    const session = state().get(id)
+        const session = sessions.get(id)
         if (session && session.info.status === "running") {
           session.process.resize(cols, rows)
         }
-  }
+      })
 
-  export function write(id: PtyID, data: string) {
+      const write = Effect.fn("Pty.write")(function* (id: PtyID, data: string) {
-    const session = state().get(id)
+        const session = sessions.get(id)
         if (session && session.info.status === "running") {
           session.process.write(data)
         }
-  }
+      })
 
-  export function connect(id: PtyID, ws: Socket, cursor?: number) {
+      const connect = Effect.fn("Pty.connect")(function* (id: PtyID, ws: Socket, cursor?: number) {
-    const session = state().get(id)
+        const session = sessions.get(id)
         if (!session) {
           ws.close()
           return
         }
         log.info("client connected to session", { id })
 
-    // Use ws.data as the unique key for this connection lifecycle.
-    // If ws.data is undefined, fallback to ws object.
         const connectionKey = ws.data && typeof ws.data === "object" ? ws.data : ws
-
-    // Optionally cleanup if the key somehow exists
         session.subscribers.delete(connectionKey)
         session.subscribers.set(connectionKey, ws)
 
@@ -317,5 +331,55 @@ export namespace Pty {
             cleanup()
           },
         }
+      })
+
+      return Service.of({ list, get, create, update, remove, resize, write, connect })
+    }),
+  )
+
+  function runtime() {
+    return require("@/effect/runtime") as typeof import("@/effect/runtime")
+  }
+
+  function run<A, E>(effect: Effect.Effect<A, E, Service>) {
+    return runtime().runPromiseInstance(effect)
+  }
+
+  function runSync<A, E>(effect: Effect.Effect<A, E, Service>) {
+    return runtime().runSyncInstance(effect)
+  }
+
+  // Sync facades
+  export function list() {
+    return runSync(Service.use((svc) => svc.list()))
+  }
+
+  export function get(id: PtyID) {
+    return runSync(Service.use((svc) => svc.get(id)))
+  }
+
+  export function resize(id: PtyID, cols: number, rows: number) {
+    runSync(Service.use((svc) => svc.resize(id, cols, rows)))
+  }
+
+  export function write(id: PtyID, data: string) {
+    runSync(Service.use((svc) => svc.write(id, data)))
+  }
+
+  export function connect(id: PtyID, ws: Socket, cursor?: number) {
+    return runSync(Service.use((svc) => svc.connect(id, ws, cursor)))
+  }
+
+  // Async facades
+  export async function create(input: CreateInput) {
+    return run(Service.use((svc) => svc.create(input)))
+  }
+
+  export async function update(id: PtyID, input: UpdateInput) {
+    return run(Service.use((svc) => svc.update(id, input)))
+  }
+
+  export async function remove(id: PtyID) {
+    return run(Service.use((svc) => svc.remove(id)))
   }
 }