refactor(watcher): effectify FileWatcher as scoped service

Convert FileWatcher from Instance.state namespace to an Effect
ServiceMap.Service with proper scope-based lifecycle management.

packages/opencode/src/effect/instances.ts

@@ -3,6 +3,7 @@ import { registerDisposer } from "./instance-registry"
 import { ProviderAuthService } from "@/provider/auth-service"
 import { QuestionService } from "@/question/service"
 import { PermissionService } from "@/permission/service"
+import { FileWatcherService } from "@/file/watcher"
 import { Instance } from "@/project/instance"
 import type { Project } from "@/project/project"
 
@@ -17,7 +18,7 @@ export class InstanceContext extends ServiceMap.Service<InstanceContext, Instanc
   "opencode/InstanceContext",
 ) {}
 
-export type InstanceServices = QuestionService | PermissionService | ProviderAuthService
+export type InstanceServices = QuestionService | PermissionService | ProviderAuthService | FileWatcherService
 
 function lookup(directory: string) {
   const project = Instance.project
@@ -26,6 +27,7 @@ function lookup(directory: string) {
     Layer.fresh(QuestionService.layer),
     Layer.fresh(PermissionService.layer),
     Layer.fresh(ProviderAuthService.layer),
+    Layer.fresh(FileWatcherService.layer),
   ).pipe(Layer.provide(ctx))
 }
 

packages/opencode/src/file/watcher.ts

@@ -1,7 +1,7 @@
 import { BusEvent } from "@/bus/bus-event"
 import { Bus } from "@/bus"
+import { InstanceContext } from "@/effect/instances"
 import z from "zod"
-import { Instance } from "../project/instance"
 import { Log } from "../util/log"
 import { FileIgnore } from "./ignore"
 import { Config } from "../config/config"
@@ -9,118 +9,134 @@ import path from "path"
 // @ts-ignore
 import { createWrapper } from "@parcel/watcher/wrapper"
 import { lazy } from "@/util/lazy"
-import { withTimeout } from "@/util/timeout"
 import type ParcelWatcher from "@parcel/watcher"
 import { Flag } from "@/flag/flag"
 import { readdir } from "fs/promises"
 import { git } from "@/util/git"
 import { Protected } from "./protected"
+import { Cause, Effect, Layer, ServiceMap } from "effect"
 
 const SUBSCRIBE_TIMEOUT_MS = 10_000
 
 declare const OPENCODE_LIBC: string | undefined
 
-export namespace FileWatcher {
-  const log = Log.create({ service: "file.watcher" })
+const log = Log.create({ service: "file.watcher" })
 
-  export const Event = {
-    Updated: BusEvent.define(
-      "file.watcher.updated",
-      z.object({
-        file: z.string(),
-        event: z.union([z.literal("add"), z.literal("change"), z.literal("unlink")]),
-      }),
-    ),
+const event = {
+  Updated: BusEvent.define(
+    "file.watcher.updated",
+    z.object({
+      file: z.string(),
+      event: z.union([z.literal("add"), z.literal("change"), z.literal("unlink")]),
+    }),
+  ),
+}
+
+const watcher = lazy((): typeof import("@parcel/watcher") | undefined => {
+  try {
+    const binding = require(
+      `@parcel/watcher-${process.platform}-${process.arch}${process.platform === "linux" ? `-${OPENCODE_LIBC || "glibc"}` : ""}`,
+    )
+    return createWrapper(binding) as typeof import("@parcel/watcher")
+  } catch (error) {
+    log.error("failed to load watcher binding", { error })
+    return
   }
+})
 
-  const watcher = lazy((): typeof import("@parcel/watcher") | undefined => {
-    try {
-      const binding = require(
-        `@parcel/watcher-${process.platform}-${process.arch}${process.platform === "linux" ? `-${OPENCODE_LIBC || "glibc"}` : ""}`,
-      )
-      return createWrapper(binding) as typeof import("@parcel/watcher")
-    } catch (error) {
-      log.error("failed to load watcher binding", { error })
-      return
-    }
-  })
+function getBackend() {
+  if (process.platform === "win32") return "windows"
+  if (process.platform === "darwin") return "fs-events"
+  if (process.platform === "linux") return "inotify"
+}
 
-  const state = Instance.state(
-    async () => {
-      log.info("init")
-      const cfg = await Config.get()
-      const backend = (() => {
-        if (process.platform === "win32") return "windows"
-        if (process.platform === "darwin") return "fs-events"
-        if (process.platform === "linux") return "inotify"
-      })()
+export namespace FileWatcher {
+  export const Event = event
+}
+
+const init = Effect.fn("FileWatcherService.init")(function* () {})
+
+export namespace FileWatcherService {
+  export interface Service {
+    readonly init: () => Effect.Effect<void>
+  }
+}
+
+export class FileWatcherService extends ServiceMap.Service<FileWatcherService, FileWatcherService.Service>()(
+  "@opencode/FileWatcher",
+) {
+  static readonly layer = Layer.effect(
+    FileWatcherService,
+    Effect.gen(function* () {
+      const instance = yield* InstanceContext
+
+      if (Flag.OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER) return FileWatcherService.of({ init })
+
+      log.info("init", { directory: instance.directory })
+
+      const backend = getBackend()
       if (!backend) {
-        log.error("watcher backend not supported", { platform: process.platform })
-        return {}
+        log.error("watcher backend not supported", { directory: instance.directory, platform: process.platform })
+        return FileWatcherService.of({ init })
       }
-      log.info("watcher backend", { platform: process.platform, backend })
 
       const w = watcher()
-      if (!w) return {}
+      if (!w) return FileWatcherService.of({ init })
 
-      const subscribe: ParcelWatcher.SubscribeCallback = (err, evts) => {
+      log.info("watcher backend", { directory: instance.directory, platform: process.platform, backend })
+
+      const subs: ParcelWatcher.AsyncSubscription[] = []
+      yield* Effect.addFinalizer(() => Effect.promise(() => Promise.allSettled(subs.map((sub) => sub.unsubscribe()))))
+
+      const cb: ParcelWatcher.SubscribeCallback = (err, evts) => {
         if (err) return
         for (const evt of evts) {
-          if (evt.type === "create") Bus.publish(Event.Updated, { file: evt.path, event: "add" })
-          if (evt.type === "update") Bus.publish(Event.Updated, { file: evt.path, event: "change" })
-          if (evt.type === "delete") Bus.publish(Event.Updated, { file: evt.path, event: "unlink" })
+          if (evt.type === "create") Bus.publish(event.Updated, { file: evt.path, event: "add" })
+          if (evt.type === "update") Bus.publish(event.Updated, { file: evt.path, event: "change" })
+          if (evt.type === "delete") Bus.publish(event.Updated, { file: evt.path, event: "unlink" })
         }
       }
 
-      const subs: ParcelWatcher.AsyncSubscription[] = []
+      const subscribe = (dir: string, ignore: string[]) =>
+        Effect.gen(function* () {
+          const sub = yield* Effect.promise(() => w.subscribe(dir, cb, { ignore, backend }))
+          subs.push(sub)
+        }).pipe(
+          Effect.timeout(SUBSCRIBE_TIMEOUT_MS),
+          Effect.catchCause((cause) => {
+            log.error("failed to subscribe", { dir, cause: Cause.pretty(cause) })
+            return Effect.void
+          }),
+        )
+
+      const cfg = yield* Effect.promise(() => Config.get())
       const cfgIgnores = cfg.watcher?.ignore ?? []
 
       if (Flag.OPENCODE_EXPERIMENTAL_FILEWATCHER) {
-        const pending = w.subscribe(Instance.directory, subscribe, {
-          ignore: [...FileIgnore.PATTERNS, ...cfgIgnores, ...Protected.paths()],
-          backend,
-        })
-        const sub = await withTimeout(pending, SUBSCRIBE_TIMEOUT_MS).catch((err) => {
-          log.error("failed to subscribe to Instance.directory", { error: err })
-          pending.then((s) => s.unsubscribe()).catch(() => {})
-          return undefined
-        })
-        if (sub) subs.push(sub)
+        yield* subscribe(instance.directory, [...FileIgnore.PATTERNS, ...cfgIgnores, ...Protected.paths()])
       }
 
-      if (Instance.project.vcs === "git") {
-        const result = await git(["rev-parse", "--git-dir"], {
-          cwd: Instance.worktree,
-        })
-        const vcsDir = result.exitCode === 0 ? path.resolve(Instance.worktree, result.text().trim()) : undefined
+      if (instance.project.vcs === "git") {
+        const result = yield* Effect.promise(() =>
+          git(["rev-parse", "--git-dir"], {
+            cwd: instance.project.worktree,
+          }),
+        )
+        const vcsDir = result.exitCode === 0 ? path.resolve(instance.project.worktree, result.text().trim()) : undefined
         if (vcsDir && !cfgIgnores.includes(".git") && !cfgIgnores.includes(vcsDir)) {
-          const gitDirContents = await readdir(vcsDir).catch(() => [])
-          const ignoreList = gitDirContents.filter((entry) => entry !== "HEAD")
-          const pending = w.subscribe(vcsDir, subscribe, {
-            ignore: ignoreList,
-            backend,
-          })
-          const sub = await withTimeout(pending, SUBSCRIBE_TIMEOUT_MS).catch((err) => {
-            log.error("failed to subscribe to vcsDir", { error: err })
-            pending.then((s) => s.unsubscribe()).catch(() => {})
-            return undefined
-          })
-          if (sub) subs.push(sub)
+          const ignore = (yield* Effect.promise(() => readdir(vcsDir).catch(() => []))).filter(
+            (entry) => entry !== "HEAD",
+          )
+          yield* subscribe(vcsDir, ignore)
         }
       }
 
-      return { subs }
-    },
-    async (state) => {
-      if (!state.subs) return
-      await Promise.all(state.subs.map((sub) => sub?.unsubscribe()))
-    },
+      return FileWatcherService.of({ init })
+    }).pipe(
+      Effect.catchCause((cause) => {
+        log.error("failed to init watcher service", { cause: Cause.pretty(cause) })
+        return Effect.succeed(FileWatcherService.of({ init: init }))
+      }),
+    ),
   )
-
-  export function init() {
-    if (Flag.OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER) {
-      return
-    }
-    state()
-  }
 }

packages/opencode/src/project/bootstrap.ts

@@ -1,7 +1,7 @@
 import { Plugin } from "../plugin"
 import { Format } from "../format"
 import { LSP } from "../lsp"
-import { FileWatcher } from "../file/watcher"
+import { FileWatcherService } from "../file/watcher"
 import { File } from "../file"
 import { Project } from "./project"
 import { Bus } from "../bus"
@@ -12,6 +12,7 @@ import { Log } from "@/util/log"
 import { ShareNext } from "@/share/share-next"
 import { Snapshot } from "../snapshot"
 import { Truncate } from "../tool/truncation"
+import { runPromiseInstance } from "@/effect/runtime"
 
 export async function InstanceBootstrap() {
   Log.Default.info("bootstrapping", { directory: Instance.directory })
@@ -19,7 +20,7 @@ export async function InstanceBootstrap() {
   ShareNext.init()
   Format.init()
   await LSP.init()
-  FileWatcher.init()
+  await runPromiseInstance(FileWatcherService.use((service) => service.init()))
   File.init()
   Vcs.init()
   Snapshot.init()

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

@@ -0,0 +1,205 @@
+import { $ } from "bun"
+import { afterEach, expect, mock, test } from "bun:test"
+import fs from "fs/promises"
+import path from "path"
+import { tmpdir } from "../fixture/fixture"
+
+const flags = await import("../../src/flag/flag")
+
+mock.module("@/flag/flag", () => ({
+  Flag: {
+    ...flags.Flag,
+    OPENCODE_EXPERIMENTAL_FILEWATCHER: true,
+    OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: false,
+  },
+}))
+
+async function load() {
+  const { runPromiseInstance } = await import("../../src/effect/runtime")
+  const watcher = await import("../../src/file/watcher")
+  const { GlobalBus } = await import("../../src/bus/global")
+  const { Instance } = await import("../../src/project/instance")
+
+  return {
+    GlobalBus,
+    FileWatcher: watcher.FileWatcher,
+    FileWatcherService: watcher.FileWatcherService,
+    Instance,
+    runPromiseInstance,
+  }
+}
+
+async function start(directory: string) {
+  const { FileWatcherService, Instance, runPromiseInstance } = await load()
+  await Instance.provide({
+    directory,
+    fn: () => runPromiseInstance(FileWatcherService.use((service) => service.init())),
+  })
+  await Bun.sleep(100)
+}
+
+async function stop(directory: string) {
+  const { Instance } = await load()
+  await Instance.provide({
+    directory,
+    fn: () => Instance.dispose(),
+  })
+  await Bun.sleep(100)
+}
+
+async function nextUpdate(
+  directory: string,
+  check: (evt: { file: string; event: "add" | "change" | "unlink" }) => boolean,
+  run: () => Promise<void>,
+) {
+  const { FileWatcher, GlobalBus } = await load()
+
+  return await new Promise<{ file: string; event: "add" | "change" | "unlink" }>((resolve, reject) => {
+    const on = (evt: {
+      directory?: string
+      payload: {
+        type: string
+        properties: {
+          file: string
+          event: "add" | "change" | "unlink"
+        }
+      }
+    }) => {
+      if (evt.directory !== directory) return
+      if (evt.payload.type !== FileWatcher.Event.Updated.type) return
+      if (!check(evt.payload.properties)) return
+      clearTimeout(timeout)
+      GlobalBus.off("event", on)
+      resolve(evt.payload.properties)
+    }
+
+    const timeout = setTimeout(() => {
+      GlobalBus.off("event", on)
+      reject(new Error("timed out waiting for file watcher event"))
+    }, 5000)
+
+    GlobalBus.on("event", on)
+
+    run().catch((err) => {
+      clearTimeout(timeout)
+      GlobalBus.off("event", on)
+      reject(err)
+    })
+  })
+}
+
+afterEach(async () => {
+  const { Instance } = await load()
+  await Instance.disposeAll()
+})
+
+test("FileWatcherService publishes root create, update, and delete events", async () => {
+  await using tmp = await tmpdir({ git: true })
+  const file = path.join(tmp.path, "watch.txt")
+
+  await start(tmp.path)
+
+  await expect(
+    nextUpdate(
+      tmp.path,
+      (evt) => evt.file === file && evt.event === "add",
+      () => fs.writeFile(file, "a"),
+    ),
+  ).resolves.toEqual({
+    file,
+    event: "add",
+  })
+
+  await expect(
+    nextUpdate(
+      tmp.path,
+      (evt) => evt.file === file && evt.event === "change",
+      () => fs.writeFile(file, "b"),
+    ),
+  ).resolves.toEqual({
+    file,
+    event: "change",
+  })
+
+  await expect(
+    nextUpdate(
+      tmp.path,
+      (evt) => evt.file === file && evt.event === "unlink",
+      () => fs.unlink(file),
+    ),
+  ).resolves.toEqual({
+    file,
+    event: "unlink",
+  })
+})
+
+test("FileWatcherService watches non-git roots", async () => {
+  await using tmp = await tmpdir()
+  const file = path.join(tmp.path, "plain.txt")
+
+  await start(tmp.path)
+
+  await expect(
+    nextUpdate(
+      tmp.path,
+      (evt) => evt.file === file && evt.event === "add",
+      () => fs.writeFile(file, "plain"),
+    ),
+  ).resolves.toEqual({
+    file,
+    event: "add",
+  })
+})
+
+test("FileWatcherService cleanup stops publishing events", async () => {
+  await using tmp = await tmpdir({ git: true })
+  const file = path.join(tmp.path, "after-dispose.txt")
+  const { FileWatcher, GlobalBus } = await load()
+  let seen = false
+
+  await start(tmp.path)
+  await stop(tmp.path)
+
+  const on = (evt: { directory?: string; payload: { type: string; properties: { file: string } } }) => {
+    if (evt.directory !== tmp.path) return
+    if (evt.payload.type !== FileWatcher.Event.Updated.type) return
+    if (evt.payload.properties.file === file) seen = true
+  }
+
+  GlobalBus.on("event", on)
+
+  try {
+    await fs.writeFile(file, "gone")
+    await Bun.sleep(500)
+    expect(seen).toBe(false)
+  } finally {
+    GlobalBus.off("event", on)
+  }
+})
+
+test("FileWatcherService ignores non-HEAD git metadata changes", async () => {
+  await using tmp = await tmpdir({ git: true })
+  const file = path.join(tmp.path, ".git", "index")
+  const edit = path.join(tmp.path, "tracked.txt")
+  const { FileWatcher, GlobalBus } = await load()
+  let seen = false
+
+  await start(tmp.path)
+
+  const on = (evt: { directory?: string; payload: { type: string; properties: { file: string } } }) => {
+    if (evt.directory !== tmp.path) return
+    if (evt.payload.type !== FileWatcher.Event.Updated.type) return
+    if (evt.payload.properties.file === file) seen = true
+  }
+
+  GlobalBus.on("event", on)
+
+  try {
+    await fs.writeFile(edit, "a")
+    await $`git add .`.cwd(tmp.path).quiet().nothrow()
+    await Bun.sleep(500)
+    expect(seen).toBe(false)
+  } finally {
+    GlobalBus.off("event", on)
+  }
+})