Merge branch 'dev' into kit/effectify-command

Constrain BusEvent.define to ZodObject instead of ZodType so TS knows
event properties are always a record. Type GlobalBus payload as
{ type: string; properties: Record<string, unknown> } instead of any.

Refactor watcher test to use Bus.subscribe instead of raw GlobalBus
listener, removing hand-rolled event types and unnecessary casts.

packages/opencode/specs/effect-migration.md

@@ -128,11 +128,11 @@ Still open and likely worth migrating:
 
 - [ ] `Plugin`
 - [ ] `ToolRegistry`
-- [x] `Pty`
+- [ ] `Pty`
 - [ ] `Worktree`
 - [ ] `Installation`
 - [ ] `Bus`
-- [ ] `Command`
+- [x] `Command`
 - [ ] `Config`
 - [ ] `Session`
 - [ ] `SessionProcessor`

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

@@ -1,5 +1,5 @@
 import z from "zod"
-import type { ZodType } from "zod"
+import type { ZodObject, ZodRawShape } 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 ZodType>(type: Type, properties: Properties) {
+  export function define<Type extends string, Properties extends ZodObject<ZodRawShape>>(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: any
+      payload: { type: string; properties: Record<string, unknown> }
     },
   ]
 }>()

packages/opencode/src/command/index.ts

@@ -1,15 +1,18 @@
 import { BusEvent } from "@/bus/bus-event"
+import { InstanceContext } from "@/effect/instance-context"
+import { runPromiseInstance } from "@/effect/runtime"
 import { SessionID, MessageID } from "@/session/schema"
+import { Effect, Fiber, Layer, ServiceMap } from "effect"
 import z from "zod"
 import { Config } from "../config/config"
-import { Instance } from "../project/instance"
-import { Identifier } from "../id/id"
 import PROMPT_INITIALIZE from "./template/initialize.txt"
 import PROMPT_REVIEW from "./template/review.txt"
 import { MCP } from "../mcp"
 import { Skill } from "../skill"
+import { Log } from "../util/log"
 
 export namespace Command {
+  const log = Log.create({ service: "command" })
   export const Event = {
     Executed: BusEvent.define(
       "command.executed",
@@ -57,95 +60,126 @@ export namespace Command {
     REVIEW: "review",
   } as const
 
-  const state = Instance.state(async () => {
-    const cfg = await Config.get()
+  export interface Interface {
+    readonly get: (name: string) => Effect.Effect<Info | undefined>
+    readonly list: () => Effect.Effect<Info[]>
+  }
 
-    const result: Record<string, Info> = {
-      [Default.INIT]: {
-        name: Default.INIT,
-        description: "create/update AGENTS.md",
-        source: "command",
-        get template() {
-          return PROMPT_INITIALIZE.replace("${path}", Instance.worktree)
-        },
-        hints: hints(PROMPT_INITIALIZE),
-      },
-      [Default.REVIEW]: {
-        name: Default.REVIEW,
-        description: "review changes [commit|branch|pr], defaults to uncommitted",
-        source: "command",
-        get template() {
-          return PROMPT_REVIEW.replace("${path}", Instance.worktree)
-        },
-        subtask: true,
-        hints: hints(PROMPT_REVIEW),
-      },
-    }
+  export class Service extends ServiceMap.Service<Service, Interface>()("@opencode/Command") {}
 
-    for (const [name, command] of Object.entries(cfg.command ?? {})) {
-      result[name] = {
-        name,
-        agent: command.agent,
-        model: command.model,
-        description: command.description,
-        source: "command",
-        get template() {
-          return command.template
-        },
-        subtask: command.subtask,
-        hints: hints(command.template),
-      }
-    }
-    for (const [name, prompt] of Object.entries(await MCP.prompts())) {
-      result[name] = {
-        name,
-        source: "mcp",
-        description: prompt.description,
-        get template() {
-          // since a getter can't be async we need to manually return a promise here
-          return new Promise<string>(async (resolve, reject) => {
-            const template = await MCP.getPrompt(
-              prompt.client,
-              prompt.name,
-              prompt.arguments
-                ? // substitute each argument with $1, $2, etc.
-                  Object.fromEntries(prompt.arguments?.map((argument, i) => [argument.name, `$${i + 1}`]))
-                : {},
-            ).catch(reject)
-            resolve(
-              template?.messages
-                .map((message) => (message.content.type === "text" ? message.content.text : ""))
-                .join("\n") || "",
-            )
-          })
-        },
-        hints: prompt.arguments?.map((_, i) => `$${i + 1}`) ?? [],
-      }
-    }
+  export const layer = Layer.effect(
+    Service,
+    Effect.gen(function* () {
+      const instance = yield* InstanceContext
 
-    // Add skills as invokable commands
-    for (const skill of await Skill.all()) {
-      // Skip if a command with this name already exists
-      if (result[skill.name]) continue
-      result[skill.name] = {
-        name: skill.name,
-        description: skill.description,
-        source: "skill",
-        get template() {
-          return skill.content
-        },
-        hints: [],
-      }
-    }
+      const commands: Record<string, Info> = {}
 
-    return result
-  })
+      const load = Effect.fn("Command.load")(function* () {
+        yield* Effect.promise(async () => {
+          const cfg = await Config.get()
+
+          commands[Default.INIT] = {
+            name: Default.INIT,
+            description: "create/update AGENTS.md",
+            source: "command",
+            get template() {
+              return PROMPT_INITIALIZE.replace("${path}", instance.worktree)
+            },
+            hints: hints(PROMPT_INITIALIZE),
+          }
+          commands[Default.REVIEW] = {
+            name: Default.REVIEW,
+            description: "review changes [commit|branch|pr], defaults to uncommitted",
+            source: "command",
+            get template() {
+              return PROMPT_REVIEW.replace("${path}", instance.worktree)
+            },
+            subtask: true,
+            hints: hints(PROMPT_REVIEW),
+          }
+
+          for (const [name, command] of Object.entries(cfg.command ?? {})) {
+            commands[name] = {
+              name,
+              agent: command.agent,
+              model: command.model,
+              description: command.description,
+              source: "command",
+              get template() {
+                return command.template
+              },
+              subtask: command.subtask,
+              hints: hints(command.template),
+            }
+          }
+          for (const [name, prompt] of Object.entries(await MCP.prompts())) {
+            commands[name] = {
+              name,
+              source: "mcp",
+              description: prompt.description,
+              get template() {
+                // since a getter can't be async we need to manually return a promise here
+                return new Promise<string>(async (resolve, reject) => {
+                  const template = await MCP.getPrompt(
+                    prompt.client,
+                    prompt.name,
+                    prompt.arguments
+                      ? // substitute each argument with $1, $2, etc.
+                        Object.fromEntries(prompt.arguments?.map((argument, i) => [argument.name, `$${i + 1}`]))
+                      : {},
+                  ).catch(reject)
+                  resolve(
+                    template?.messages
+                      .map((message) => (message.content.type === "text" ? message.content.text : ""))
+                      .join("\n") || "",
+                  )
+                })
+              },
+              hints: prompt.arguments?.map((_, i) => `$${i + 1}`) ?? [],
+            }
+          }
+
+          // Add skills as invokable commands
+          for (const skill of await Skill.all()) {
+            // Skip if a command with this name already exists
+            if (commands[skill.name]) continue
+            commands[skill.name] = {
+              name: skill.name,
+              description: skill.description,
+              source: "skill",
+              get template() {
+                return skill.content
+              },
+              hints: [],
+            }
+          }
+        })
+      })
+
+      const loadFiber = yield* load().pipe(
+        Effect.catchCause((cause) => Effect.sync(() => log.error("init failed", { cause }))),
+        Effect.forkScoped,
+      )
+
+      const get = Effect.fn("Command.get")(function* (name: string) {
+        yield* Fiber.join(loadFiber)
+        return commands[name]
+      })
+
+      const list = Effect.fn("Command.list")(function* () {
+        yield* Fiber.join(loadFiber)
+        return Object.values(commands)
+      })
+
+      return Service.of({ get, list })
+    }),
+  )
 
   export async function get(name: string) {
-    return state().then((x) => x[name])
+    return runPromiseInstance(Service.use((svc) => svc.get(name)))
   }
 
   export async function list() {
-    return state().then((x) => Object.values(x))
+    return runPromiseInstance(Service.use((svc) => svc.list()))
   }
 }

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,
+          payload: event as { type: string; properties: Record<string, unknown> },
         })
       })
       // Wait 250ms and retry if SSE connection fails

packages/opencode/src/effect/instances.ts

@@ -1,10 +1,10 @@
 import { Effect, Layer, LayerMap, ServiceMap } from "effect"
+import { Command } from "@/command"
 import { File } from "@/file/service"
 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"
@@ -17,6 +17,7 @@ import { registerDisposer } from "./instance-registry"
 export { InstanceContext } from "./instance-context"
 
 export type InstanceServices =
+  | Command.Service
   | Question.Service
   | Permission.Service
   | ProviderAuth.Service
@@ -25,7 +26,6 @@ export type InstanceServices =
   | FileTime.Service
   | Format.Service
   | File.Service
-  | Pty.Service
   | Skill.Service
   | Snapshot.Service
 
@@ -38,6 +38,7 @@ export type InstanceServices =
 function lookup(_key: string) {
   const ctx = Layer.sync(InstanceContext, () => InstanceContext.of(Instance.current))
   return Layer.mergeAll(
+    Command.layer,
     Question.layer,
     Permission.layer,
     ProviderAuth.defaultLayer,
@@ -46,7 +47,6 @@ 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,10 +20,6 @@ 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,12 +1,13 @@
 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" })
@@ -89,297 +90,232 @@ export namespace Pty {
     subscribers: Map<unknown, Socket>
   }
 
-  export interface Interface {
-    readonly list: () => Effect.Effect<Info[]>
-    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()
-            } 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 })
+  const state = Instance.state(
+    () => new Map<PtyID, ActiveSession>(),
+    async (sessions) => {
+      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 {}
-        }
-        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 {
-            cleanup()
-            ws.close()
-            return
+            // ignore
           }
         }
-
-        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 })
-    }),
+      }
+      sessions.clear()
+    },
   )
 
-  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()))
+    return Array.from(state().values()).map((s) => s.info)
   }
 
   export function get(id: PtyID) {
-    return runSync(Service.use((svc) => svc.get(id)))
+    return state().get(id)?.info
   }
 
-  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)))
+    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) {
-    return run(Service.use((svc) => svc.update(id, input)))
+    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) {
-    return run(Service.use((svc) => svc.remove(id)))
+    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)
+    if (session && session.info.status === "running") {
+      session.process.resize(cols, rows)
+    }
+  }
+
+  export function write(id: PtyID, data: string) {
+    const session = state().get(id)
+    if (session && session.info.status === "running") {
+      session.process.write(data)
+    }
+  }
+
+  export function connect(id: PtyID, ws: Socket, cursor?: number) {
+    const session = state().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)
+
+    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 {
+        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/src/session/prompt.ts

@@ -1782,6 +1782,9 @@ NOTE: At any point in time through this workflow you should feel free to ask the
   export async function command(input: CommandInput) {
     log.info("command", input)
     const command = await Command.get(input.command)
+    if (!command) {
+      throw new NamedError.Unknown({ message: `Command not found: "${input.command}"` })
+    }
     const agentName = command.agent ?? input.agent ?? (await Agent.defaultAgent())
 
     const raw = input.arguments.match(argsRegex) ?? []

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: WatcherEvent } }
+type BusUpdate = { directory?: string; payload: { type: string; properties: Record<string, unknown> } }
 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
-    if (!check(evt.payload.properties)) return
-    hit(evt.payload.properties)
+    const props = evt.payload.properties as WatcherEvent
+    if (!check(props)) return
+    hit(props)
   }
 
-  function cleanup() {
+  GlobalBus.on("event", on)
+
+  return () => {
     if (done) return
     done = true
     GlobalBus.off("event", on)
   }
-
-  GlobalBus.on("event", on)
-  return cleanup
 }
 
 function wait(directory: string, check: (evt: WatcherEvent) => boolean) {