Merge branch 'dev' into kit/effectify-pty

packages/opencode/specs/effect-migration.md

@@ -126,9 +126,9 @@ Done now:
 
 Still open and likely worth migrating:
 
-- [x] `Plugin`
+- [ ] `Plugin`
 - [ ] `ToolRegistry`
-- [ ] `Pty`
+- [x] `Pty`
 - [ ] `Worktree`
 - [ ] `Installation`
 - [ ] `Bus`

packages/opencode/src/bus/bus-event.ts

@@ -1,5 +1,5 @@
 import z from "zod"
-import type { ZodObject, ZodRawShape } from "zod"
+import type { ZodType } from "zod"
 import { Log } from "../util/log"
 
 export namespace BusEvent {
@@ -9,7 +9,7 @@ export namespace BusEvent {
 
   const registry = new Map<string, Definition>()
 
-  export function define<Type extends string, Properties extends ZodObject<ZodRawShape>>(type: Type, properties: Properties) {
+  export function define<Type extends string, Properties extends ZodType>(type: Type, properties: Properties) {
     const result = {
       type,
       properties,

packages/opencode/src/bus/global.ts

@@ -4,7 +4,7 @@ export const GlobalBus = new EventEmitter<{
   event: [
     {
       directory?: string
-      payload: { type: string; properties: Record<string, unknown> }
+      payload: any
     },
   ]
 }>()

packages/opencode/src/control-plane/workspace.ts

@@ -124,7 +124,7 @@ export namespace Workspace {
       await parseSSE(res.body, stop, (event) => {
         GlobalBus.emit("event", {
           directory: space.id,
-          payload: event as { type: string; properties: Record<string, unknown> },
+          payload: event,
         })
       })
       // Wait 250ms and retry if SSE connection fails

packages/opencode/src/effect/instances.ts

@@ -4,13 +4,13 @@ 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"
 import { Question } from "@/question/service"
 import { Skill } from "@/skill/service"
 import { Snapshot } from "@/snapshot/service"
-import { Plugin } from "@/plugin"
 import { InstanceContext } from "./instance-context"
 import { registerDisposer } from "./instance-registry"
 
@@ -25,9 +25,9 @@ export type InstanceServices =
   | FileTime.Service
   | Format.Service
   | File.Service
+  | Pty.Service
   | Skill.Service
   | Snapshot.Service
-  | Plugin.Service
 
 // NOTE: LayerMap only passes the key (directory string) to lookup, but we need
 // the full instance context (directory, worktree, project). We read from the
@@ -46,9 +46,9 @@ function lookup(_key: string) {
     FileTime.layer,
     Format.layer,
     File.layer,
+    Pty.layer,
     Skill.defaultLayer,
     Snapshot.defaultLayer,
-    Plugin.layer,
   ).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/plugin/index.ts

@@ -1,206 +1,144 @@
 import type { Hooks, PluginInput, Plugin as PluginInstance } from "@opencode-ai/plugin"
+import { Config } from "../config/config"
 import { Bus } from "../bus"
 import { Log } from "../util/log"
 import { createOpencodeClient } from "@opencode-ai/sdk"
+import { Server } from "../server/server"
 import { BunProc } from "../bun"
+import { Instance } from "../project/instance"
 import { Flag } from "../flag/flag"
+import { CodexAuthPlugin } from "./codex"
+import { Session } from "../session"
 import { NamedError } from "@opencode-ai/util/error"
-import { Effect, Layer, ServiceMap } from "effect"
+import { CopilotAuthPlugin } from "./copilot"
-import { InstanceContext } from "@/effect/instance-context"
+import { gitlabAuthPlugin as GitlabAuthPlugin } from "opencode-gitlab-auth"
 
 export namespace Plugin {
   const log = Log.create({ service: "plugin" })
 
-  export interface Interface {
+  // Built-in plugins that are directly imported (not installed from npm)
-    readonly trigger: <
+  const INTERNAL_PLUGINS: PluginInstance[] = [CodexAuthPlugin, CopilotAuthPlugin, GitlabAuthPlugin]
-      Name extends Exclude<keyof Required<Hooks>, "auth" | "event" | "tool">,
-      Input = Parameters<Required<Hooks>[Name]>[0],
-      Output = Parameters<Required<Hooks>[Name]>[1],
-    >(
-      name: Name,
-      input: Input,
-      output: Output,
-    ) => Effect.Effect<Output>
-    readonly list: () => Effect.Effect<Hooks[]>
-    readonly init: () => Effect.Effect<void>
-  }
 
-  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Plugin") {}
+  const state = Instance.state(async () => {
+    const client = createOpencodeClient({
+      baseUrl: "http://localhost:4096",
+      directory: Instance.directory,
+      headers: Flag.OPENCODE_SERVER_PASSWORD
+        ? {
+            Authorization: `Basic ${Buffer.from(`${Flag.OPENCODE_SERVER_USERNAME ?? "opencode"}:${Flag.OPENCODE_SERVER_PASSWORD}`).toString("base64")}`,
+          }
+        : undefined,
+      fetch: async (...args) => Server.Default().fetch(...args),
+    })
+    const config = await Config.get()
+    const hooks: Hooks[] = []
+    const input: PluginInput = {
+      client,
+      project: Instance.project,
+      worktree: Instance.worktree,
+      directory: Instance.directory,
+      get serverUrl(): URL {
+        return Server.url ?? new URL("http://localhost:4096")
+      },
+      $: Bun.$,
+    }
 
-  export const layer = Layer.effect(
+    for (const plugin of INTERNAL_PLUGINS) {
-    Service,
+      log.info("loading internal plugin", { name: plugin.name })
-    Effect.gen(function* () {
+      const init = await plugin(input).catch((err) => {
-      const instance = yield* InstanceContext
+        log.error("failed to load internal plugin", { name: plugin.name, error: err })
-      const hooks: Hooks[] = []
+      })
-      let task: Promise<void> | undefined
+      if (init) hooks.push(init)
+    }
 
-      const load = Effect.fn("Plugin.load")(function* () {
+    let plugins = config.plugin ?? []
-        yield* Effect.promise(async () => {
+    if (plugins.length) await Config.waitForDependencies()
-          const [{ Config }, { Server }, codex, copilot, gitlab] = await Promise.all([
+
-            import("../config/config"),
+    for (let plugin of plugins) {
-            import("../server/server"),
+      // ignore old codex plugin since it is supported first party now
-            import("./codex"),
+      if (plugin.includes("opencode-openai-codex-auth") || plugin.includes("opencode-copilot-auth")) continue
-            import("./copilot"),
+      log.info("loading plugin", { path: plugin })
-            import("opencode-gitlab-auth"),
+      if (!plugin.startsWith("file://")) {
-          ])
+        const lastAtIndex = plugin.lastIndexOf("@")
-          const internal: PluginInstance[] = [codex.CodexAuthPlugin, copilot.CopilotAuthPlugin, gitlab.gitlabAuthPlugin]
+        const pkg = lastAtIndex > 0 ? plugin.substring(0, lastAtIndex) : plugin
-          const client = createOpencodeClient({
+        const version = lastAtIndex > 0 ? plugin.substring(lastAtIndex + 1) : "latest"
-            baseUrl: "http://localhost:4096",
+        plugin = await BunProc.install(pkg, version).catch((err) => {
-            directory: instance.directory,
+          const cause = err instanceof Error ? err.cause : err
-            headers: Flag.OPENCODE_SERVER_PASSWORD
+          const detail = cause instanceof Error ? cause.message : String(cause ?? err)
-              ? {
+          log.error("failed to install plugin", { pkg, version, error: detail })
-                  Authorization: `Basic ${Buffer.from(`${Flag.OPENCODE_SERVER_USERNAME ?? "opencode"}:${Flag.OPENCODE_SERVER_PASSWORD}`).toString("base64")}`,
+          Bus.publish(Session.Event.Error, {
-                }
+            error: new NamedError.Unknown({
-              : undefined,
+              message: `Failed to install plugin ${pkg}@${version}: ${detail}`,
-            fetch: async (...args) => Server.Default().fetch(...args),
+            }).toObject(),
           })
-          const config = await Config.get()
+          return ""
-          const input: PluginInput = {
+        })
-            client,
+        if (!plugin) continue
-            project: instance.project,
+      }
-            worktree: instance.worktree,
+      // Prevent duplicate initialization when plugins export the same function
-            directory: instance.directory,
+      // as both a named export and default export (e.g., `export const X` and `export default X`).
-            get serverUrl(): URL {
+      // Object.entries(mod) would return both entries pointing to the same function reference.
-              return Server.url ?? new URL("http://localhost:4096")
+      await import(plugin)
-            },
+        .then(async (mod) => {
-            $: Bun.$,
+          const seen = new Set<PluginInstance>()
-          }
+          for (const [_name, fn] of Object.entries<PluginInstance>(mod)) {
-
+            if (seen.has(fn)) continue
-          for (const plugin of internal) {
+            seen.add(fn)
-            log.info("loading internal plugin", { name: plugin.name })
+            hooks.push(await fn(input))
-            const init = await plugin(input).catch((err) => {
-              log.error("failed to load internal plugin", { name: plugin.name, error: err })
-            })
-            if (init) hooks.push(init)
-          }
-
-          let plugins = config.plugin ?? []
-          if (plugins.length) await Config.waitForDependencies()
-
-          for (let plugin of plugins) {
-            // ignore old codex plugin since it is supported first party now
-            if (plugin.includes("opencode-openai-codex-auth") || plugin.includes("opencode-copilot-auth")) continue
-            log.info("loading plugin", { path: plugin })
-            if (!plugin.startsWith("file://")) {
-              const lastAtIndex = plugin.lastIndexOf("@")
-              const pkg = lastAtIndex > 0 ? plugin.substring(0, lastAtIndex) : plugin
-              const version = lastAtIndex > 0 ? plugin.substring(lastAtIndex + 1) : "latest"
-              plugin = await BunProc.install(pkg, version).catch((err) => {
-                const cause = err instanceof Error ? err.cause : err
-                const detail = cause instanceof Error ? cause.message : String(cause ?? err)
-                log.error("failed to install plugin", { pkg, version, error: detail })
-                void import("../session").then(({ Session }) =>
-                  Bus.publish(Session.Event.Error, {
-                    error: new NamedError.Unknown({
-                      message: `Failed to install plugin ${pkg}@${version}: ${detail}`,
-                    }).toObject(),
-                  }),
-                )
-                return ""
-              })
-              if (!plugin) continue
-            }
-            // Prevent duplicate initialization when plugins export the same function
-            // as both a named export and default export (e.g., `export const X` and `export default X`).
-            // Object.entries(mod) would return both entries pointing to the same function reference.
-            await import(plugin)
-              .then(async (mod) => {
-                const seen = new Set<PluginInstance>()
-                for (const [_name, fn] of Object.entries<PluginInstance>(mod)) {
-                  if (seen.has(fn)) continue
-                  seen.add(fn)
-                  hooks.push(await fn(input))
-                }
-              })
-              .catch((err) => {
-                const message = err instanceof Error ? err.message : String(err)
-                log.error("failed to load plugin", { path: plugin, error: message })
-                void import("../session").then(({ Session }) =>
-                  Bus.publish(Session.Event.Error, {
-                    error: new NamedError.Unknown({
-                      message: `Failed to load plugin ${plugin}: ${message}`,
-                    }).toObject(),
-                  }),
-                )
-              })
           }
         })
-      })
+        .catch((err) => {
-
+          const message = err instanceof Error ? err.message : String(err)
-      const ensure = Effect.fn("Plugin.ensure")(function* () {
+          log.error("failed to load plugin", { path: plugin, error: message })
-        yield* Effect.promise(() => {
+          Bus.publish(Session.Event.Error, {
-          task ??= Effect.runPromise(
+            error: new NamedError.Unknown({
-            load().pipe(Effect.catchCause((cause) => Effect.sync(() => log.error("init failed", { cause })))),
+              message: `Failed to load plugin ${plugin}: ${message}`,
-          )
+            }).toObject(),
-          return task
-        })
-      })
-
-      const trigger = Effect.fn("Plugin.trigger")(function* <
-        Name extends Exclude<keyof Required<Hooks>, "auth" | "event" | "tool">,
-        Input = Parameters<Required<Hooks>[Name]>[0],
-        Output = Parameters<Required<Hooks>[Name]>[1],
-      >(name: Name, input: Input, output: Output) {
-        if (!name) return output
-        yield* ensure()
-        yield* Effect.promise(async () => {
-          for (const hook of hooks) {
-            const fn = hook[name]
-            if (!fn) continue
-            // @ts-expect-error if you feel adventurous, please fix the typing, make sure to bump the try-counter if you
-            // give up.
-            // try-counter: 2
-            await fn(input, output)
-          }
-        })
-        return output
-      })
-
-      const list = Effect.fn("Plugin.list")(function* () {
-        yield* ensure()
-        return hooks
-      })
-
-      const init = Effect.fn("Plugin.init")(function* () {
-        yield* ensure()
-        yield* Effect.promise(async () => {
-          const { Config } = await import("../config/config")
-          const config = await Config.get()
-          for (const hook of hooks) {
-            await (hook as any).config?.(config)
-          }
-          Bus.subscribeAll(async (input) => {
-            for (const hook of hooks) {
-              hook["event"]?.({
-                event: input,
-              })
-            }
           })
         })
-      })
+    }
 
-      return Service.of({ trigger, list, init })
+    return {
-    }),
+      hooks,
-  ).pipe(Layer.fresh)
+      input,
-
+    }
-  async function run<A, E>(effect: Effect.Effect<A, E, Service>) {
+  })
-    const { runPromiseInstance } = await import("@/effect/runtime")
-    return runPromiseInstance(effect)
-  }
 
   export async function trigger<
     Name extends Exclude<keyof Required<Hooks>, "auth" | "event" | "tool">,
     Input = Parameters<Required<Hooks>[Name]>[0],
     Output = Parameters<Required<Hooks>[Name]>[1],
   >(name: Name, input: Input, output: Output): Promise<Output> {
-    return run(Service.use((svc) => svc.trigger(name, input, output)))
+    if (!name) return output
+    for (const hook of await state().then((x) => x.hooks)) {
+      const fn = hook[name]
+      if (!fn) continue
+      // @ts-expect-error if you feel adventurous, please fix the typing, make sure to bump the try-counter if you
+      // give up.
+      // try-counter: 2
+      await fn(input, output)
+    }
+    return output
   }
 
-  export async function list(): Promise<Hooks[]> {
+  export async function list() {
-    return run(Service.use((svc) => svc.list()))
+    return state().then((x) => x.hooks)
   }
 
   export async function init() {
-    return run(Service.use((svc) => svc.init()))
+    const hooks = await state().then((x) => x.hooks)
+    const config = await Config.get()
+    for (const hook of hooks) {
+      // @ts-expect-error this is because we haven't moved plugin to sdk v2
+      await hook.config?.(config)
+    }
+    Bus.subscribeAll(async (input) => {
+      const hooks = await state().then((x) => x.hooks)
+      for (const hook of hooks) {
+        hook["event"]?.({
+          event: input,
+        })
+      }
+    })
   }
 }

packages/opencode/src/provider/auth-service.ts

@@ -1,4 +1,4 @@
-import type { AuthOuathResult, Hooks } from "@opencode-ai/plugin"
+import type { AuthOuathResult } from "@opencode-ai/plugin"
 import { NamedError } from "@opencode-ai/util/error"
 import * as Auth from "@/auth/effect"
 import { ProviderID } from "./schema"
@@ -6,8 +6,6 @@ import { Array as Arr, Effect, Layer, Record, Result, ServiceMap, Struct } from
 import z from "zod"
 
 export namespace ProviderAuth {
-  type Hook = NonNullable<Hooks["auth"]>
-
   export const Method = z
     .object({
       type: z.union([z.literal("oauth"), z.literal("api")]),
@@ -107,26 +105,20 @@ export namespace ProviderAuth {
     Service,
     Effect.gen(function* () {
       const auth = yield* Auth.Auth.Service
-      let hooks: Record<ProviderID, Hook> | undefined
+      const hooks = yield* Effect.promise(async () => {
+        const mod = await import("../plugin")
+        const plugins = await mod.Plugin.list()
+        return Record.fromEntries(
+          Arr.filterMap(plugins, (x) =>
+            x.auth?.provider !== undefined
+              ? Result.succeed([ProviderID.make(x.auth.provider), x.auth] as const)
+              : Result.failVoid,
+          ),
+        )
+      })
       const pending = new Map<ProviderID, AuthOuathResult>()
 
-      const load = Effect.fn("ProviderAuth.load")(function* () {
-        if (hooks) return hooks
-        hooks = yield* Effect.promise(async () => {
-          const mod = await import("../plugin")
-          const plugins = await mod.Plugin.list()
-          const result = {} as Record<ProviderID, Hook>
-          for (const item of plugins) {
-            if (item.auth?.provider === undefined) continue
-            result[ProviderID.make(item.auth.provider)] = item.auth
-          }
-          return result
-        })
-        return hooks
-      })
-
       const methods = Effect.fn("ProviderAuth.methods")(function* () {
-        const hooks = yield* load()
         return Record.map(hooks, (item) =>
           item.methods.map(
             (method): Method => ({
@@ -160,7 +152,6 @@ export namespace ProviderAuth {
         method: number
         inputs?: Record<string, string>
       }) {
-        const hooks = yield* load()
         const method = hooks[input.providerID].methods[input.method]
         if (method.type !== "oauth") return
 
@@ -187,7 +178,6 @@ export namespace ProviderAuth {
         method: number
         code?: string
       }) {
-        yield* load()
         const match = pending.get(input.providerID)
         if (!match) return yield* Effect.fail(new OauthMissing({ providerID: input.providerID }))
         if (match.method === "code" && !input.code) {

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,232 +89,297 @@ 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>
-      for (const session of sessions.values()) {
+    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()
+            } catch {}
+            for (const [key, ws] of session.subscribers.entries()) {
+              try {
+                if (ws.data === key) ws.close()
+              } catch {}
+            }
+          }
+          sessions.clear()
+        }),
+      )
+
+      const removeSession = (id: PtyID) => {
+        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 })
+      }
+
+      const list = Effect.fn("Pty.list")(function* () {
+        return Array.from(sessions.values()).map((s) => s.info)
+      })
+
+      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 || []
+          if (command.endsWith("sh")) {
+            args.push("-l")
+          }
+
+          const cwd = input.cwd || instance.directory
+          const shellEnv = await Plugin.trigger("shell.env", { cwd }, { env: {} })
+          const env = {
+            ...process.env,
+            ...input.env,
+            ...shellEnv.env,
+            TERM: "xterm-256color",
+            OPENCODE_TERMINAL: "1",
+          } as Record<string, string>
+
+          if (process.platform === "win32") {
+            env.LC_ALL = "C.UTF-8"
+            env.LC_CTYPE = "C.UTF-8"
+            env.LANG = "C.UTF-8"
+          }
+          log.info("creating session", { id, cmd: command, args, cwd })
+
+          const spawn = await pty()
+          const ptyProcess = spawn(command, args, {
+            name: "xterm-256color",
+            cwd,
+            env,
+          })
+
+          const info = {
+            id,
+            title: input.title || `Terminal ${id.slice(-4)}`,
+            command,
+            args,
+            cwd,
+            status: "running",
+            pid: ptyProcess.pid,
+          } as const
+          const session: ActiveSession = {
+            info,
+            process: ptyProcess,
+            buffer: "",
+            bufferCursor: 0,
+            cursor: 0,
+            subscribers: new Map(),
+          }
+          sessions.set(id, session)
+          ptyProcess.onData((chunk) => {
+            session.cursor += chunk.length
+
+            for (const [key, ws] of session.subscribers.entries()) {
+              if (ws.readyState !== 1) {
+                session.subscribers.delete(key)
+                continue
+              }
+              if (ws.data !== key) {
+                session.subscribers.delete(key)
+                continue
+              }
+              try {
+                ws.send(chunk)
+              } catch {
+                session.subscribers.delete(key)
+              }
+            }
+
+            session.buffer += chunk
+            if (session.buffer.length <= BUFFER_LIMIT) return
+            const excess = session.buffer.length - BUFFER_LIMIT
+            session.buffer = session.buffer.slice(excess)
+            session.bufferCursor += excess
+          })
+          ptyProcess.onExit(({ exitCode }) => {
+            if (session.info.status === "exited") return
+            log.info("session exited", { id, exitCode })
+            session.info.status = "exited"
+            Bus.publish(Event.Exited, { id, exitCode })
+            removeSession(id)
+          })
+          Bus.publish(Event.Created, { info })
+          return info
+        })
+      })
+
+      const update = Effect.fn("Pty.update")(function* (id: PtyID, input: UpdateInput) {
+        const session = sessions.get(id)
+        if (!session) return
+        if (input.title) {
+          session.info.title = input.title
+        }
+        if (input.size) {
+          session.process.resize(input.size.cols, input.size.rows)
+        }
+        Bus.publish(Event.Updated, { info: session.info })
+        return session.info
+      })
+
+      const remove = Effect.fn("Pty.remove")(function* (id: PtyID) {
+        removeSession(id)
+      })
+
+      const resize = Effect.fn("Pty.resize")(function* (id: PtyID, cols: number, rows: number) {
+        const session = sessions.get(id)
+        if (session && session.info.status === "running") {
+          session.process.resize(cols, rows)
+        }
+      })
+
+      const write = Effect.fn("Pty.write")(function* (id: PtyID, data: string) {
+        const session = sessions.get(id)
+        if (session && session.info.status === "running") {
+          session.process.write(data)
+        }
+      })
+
+      const connect = Effect.fn("Pty.connect")(function* (id: PtyID, ws: Socket, cursor?: number) {
+        const session = sessions.get(id)
+        if (!session) {
+          ws.close()
+          return
+        }
+        log.info("client connected to session", { id })
+
+        const connectionKey = ws.data && typeof ws.data === "object" ? ws.data : ws
+        session.subscribers.delete(connectionKey)
+        session.subscribers.set(connectionKey, ws)
+
+        const cleanup = () => {
+          session.subscribers.delete(connectionKey)
+        }
+
+        const start = session.bufferCursor
+        const end = session.cursor
+
+        const from =
+          cursor === -1 ? end : typeof cursor === "number" && Number.isSafeInteger(cursor) ? Math.max(0, cursor) : 0
+
+        const data = (() => {
+          if (!session.buffer) return ""
+          if (from >= end) return ""
+          const offset = Math.max(0, from - start)
+          if (offset >= session.buffer.length) return ""
+          return session.buffer.slice(offset)
+        })()
+
+        if (data) {
+          try {
+            for (let i = 0; i < data.length; i += BUFFER_CHUNK) {
+              ws.send(data.slice(i, i + BUFFER_CHUNK))
+            }
           } catch {
-            // ignore
+            cleanup()
+            ws.close()
+            return
           }
         }
-      }
+
-      sessions.clear()
+        try {
-    },
+          ws.send(meta(end))
+        } catch {
+          cleanup()
+          ws.close()
+          return
+        }
+        return {
+          onMessage: (message: string | ArrayBuffer) => {
+            session.process.write(String(message))
+          },
+          onClose: () => {
+            log.info("client disconnected from session", { id })
+            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 Array.from(state().values()).map((s) => s.info)
+    return runSync(Service.use((svc) => svc.list()))
   }
 
   export function get(id: PtyID) {
-    return state().get(id)?.info
+    return runSync(Service.use((svc) => svc.get(id)))
-  }
-
-  export async function create(input: CreateInput) {
-    const id = PtyID.ascending()
-    const command = input.command || Shell.preferred()
-    const args = input.args || []
-    if (command.endsWith("sh")) {
-      args.push("-l")
-    }
-
-    const cwd = input.cwd || Instance.directory
-    const shellEnv = await Plugin.trigger("shell.env", { cwd }, { env: {} })
-    const env = {
-      ...process.env,
-      ...input.env,
-      ...shellEnv.env,
-      TERM: "xterm-256color",
-      OPENCODE_TERMINAL: "1",
-    } as Record<string, string>
-
-    if (process.platform === "win32") {
-      env.LC_ALL = "C.UTF-8"
-      env.LC_CTYPE = "C.UTF-8"
-      env.LANG = "C.UTF-8"
-    }
-    log.info("creating session", { id, cmd: command, args, cwd })
-
-    const spawn = await pty()
-    const ptyProcess = spawn(command, args, {
-      name: "xterm-256color",
-      cwd,
-      env,
-    })
-
-    const info = {
-      id,
-      title: input.title || `Terminal ${id.slice(-4)}`,
-      command,
-      args,
-      cwd,
-      status: "running",
-      pid: ptyProcess.pid,
-    } as const
-    const session: ActiveSession = {
-      info,
-      process: ptyProcess,
-      buffer: "",
-      bufferCursor: 0,
-      cursor: 0,
-      subscribers: new Map(),
-    }
-    state().set(id, session)
-    ptyProcess.onData(
-      Instance.bind((chunk) => {
-        session.cursor += chunk.length
-
-        for (const [key, ws] of session.subscribers.entries()) {
-          if (ws.readyState !== 1) {
-            session.subscribers.delete(key)
-            continue
-          }
-
-          if (ws.data !== key) {
-            session.subscribers.delete(key)
-            continue
-          }
-
-          try {
-            ws.send(chunk)
-          } catch {
-            session.subscribers.delete(key)
-          }
-        }
-
-        session.buffer += chunk
-        if (session.buffer.length <= BUFFER_LIMIT) return
-        const excess = session.buffer.length - BUFFER_LIMIT
-        session.buffer = session.buffer.slice(excess)
-        session.bufferCursor += excess
-      }),
-    )
-    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)
-      }),
-    )
-    Bus.publish(Event.Created, { info })
-    return info
-  }
-
-  export async function update(id: PtyID, input: UpdateInput) {
-    const session = state().get(id)
-    if (!session) return
-    if (input.title) {
-      session.info.title = input.title
-    }
-    if (input.size) {
-      session.process.resize(input.size.cols, input.size.rows)
-    }
-    Bus.publish(Event.Updated, { info: session.info })
-    return session.info
-  }
-
-  export async function remove(id: PtyID) {
-    const session = state().get(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 session = state().get(id)
+    runSync(Service.use((svc) => svc.resize(id, cols, rows)))
-    if (session && session.info.status === "running") {
-      session.process.resize(cols, rows)
-    }
   }
 
   export function write(id: PtyID, data: string) {
-    const session = state().get(id)
+    runSync(Service.use((svc) => svc.write(id, data)))
-    if (session && session.info.status === "running") {
-      session.process.write(data)
-    }
   }
 
   export function connect(id: PtyID, ws: Socket, cursor?: number) {
-    const session = state().get(id)
+    return runSync(Service.use((svc) => svc.connect(id, ws, cursor)))
-    if (!session) {
+  }
-      ws.close()
-      return
-    }
-    log.info("client connected to session", { id })
 
-    // Use ws.data as the unique key for this connection lifecycle.
+  // Async facades
-    // If ws.data is undefined, fallback to ws object.
+  export async function create(input: CreateInput) {
-    const connectionKey = ws.data && typeof ws.data === "object" ? ws.data : ws
+    return run(Service.use((svc) => svc.create(input)))
+  }
 
-    // Optionally cleanup if the key somehow exists
+  export async function update(id: PtyID, input: UpdateInput) {
-    session.subscribers.delete(connectionKey)
+    return run(Service.use((svc) => svc.update(id, input)))
-    session.subscribers.set(connectionKey, ws)
+  }
 
-    const cleanup = () => {
+  export async function remove(id: PtyID) {
-      session.subscribers.delete(connectionKey)
+    return run(Service.use((svc) => svc.remove(id)))
-    }
-
-    const start = session.bufferCursor
-    const end = session.cursor
-
-    const from =
-      cursor === -1 ? end : typeof cursor === "number" && Number.isSafeInteger(cursor) ? Math.max(0, cursor) : 0
-
-    const data = (() => {
-      if (!session.buffer) return ""
-      if (from >= end) return ""
-      const offset = Math.max(0, from - start)
-      if (offset >= session.buffer.length) return ""
-      return session.buffer.slice(offset)
-    })()
-
-    if (data) {
-      try {
-        for (let i = 0; i < data.length; i += BUFFER_CHUNK) {
-          ws.send(data.slice(i, i + BUFFER_CHUNK))
-        }
-      } catch {
-        cleanup()
-        ws.close()
-        return
-      }
-    }
-
-    try {
-      ws.send(meta(end))
-    } catch {
-      cleanup()
-      ws.close()
-      return
-    }
-    return {
-      onMessage: (message: string | ArrayBuffer) => {
-        session.process.write(String(message))
-      },
-      onClose: () => {
-        log.info("client disconnected from session", { id })
-        cleanup()
-      },
-    }
   }
 }

packages/opencode/test/file/watcher.test.ts

@@ -16,7 +16,7 @@ const describeWatcher = FileWatcher.hasNativeBinding() && !process.env.CI ? desc
 // Helpers
 // ---------------------------------------------------------------------------
 
-type BusUpdate = { directory?: string; payload: { type: string; properties: Record<string, unknown> } }
+type BusUpdate = { directory?: string; payload: { type: string; properties: WatcherEvent } }
 type WatcherEvent = { file: string; event: "add" | "change" | "unlink" }
 
 /** Run `body` with a live FileWatcher service. */
@@ -40,18 +40,18 @@ function listen(directory: string, check: (evt: WatcherEvent) => boolean, hit: (
     if (done) return
     if (evt.directory !== directory) return
     if (evt.payload.type !== FileWatcher.Event.Updated.type) return
-    const props = evt.payload.properties as WatcherEvent
+    if (!check(evt.payload.properties)) return
-    if (!check(props)) return
+    hit(evt.payload.properties)
-    hit(props)
   }
 
-  GlobalBus.on("event", on)
+  function cleanup() {
-
-  return () => {
     if (done) return
     done = true
     GlobalBus.off("event", on)
   }
+
+  GlobalBus.on("event", on)
+  return cleanup
 }
 
 function wait(directory: string, check: (evt: WatcherEvent) => boolean) {