From e5b576ce3a6e4510aa981c3018d8d7b98f14ea00 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Mon, 10 Aug 2026 17:35:50 -0500 Subject: [PATCH 1/2] fix(core): restore parcel watch for git HEAD Bun fs.watch misses git's atomic HEAD.lock rename, so checkout never refreshed the cached branch. Watch the git dir with Parcel like dev, keeping only HEAD and HEAD.lock. --- .../core/src/filesystem/location-watcher.ts | 16 ++++++++---- packages/core/src/vcs.ts | 5 +++- packages/core/test/filesystem/watcher.test.ts | 26 +++++++++++++++++-- packages/core/test/vcs.test.ts | 5 +++- 4 files changed, 43 insertions(+), 9 deletions(-) diff --git a/packages/core/src/filesystem/location-watcher.ts b/packages/core/src/filesystem/location-watcher.ts index 6848933fee8a..a2f7fa6de245 100644 --- a/packages/core/src/filesystem/location-watcher.ts +++ b/packages/core/src/filesystem/location-watcher.ts @@ -4,7 +4,6 @@ import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { Context, Effect, Layer, Stream } from "effect" import { FileSystem } from "@opencode-ai/schema/filesystem" import { Document } from "@opencode-ai/schema/config" -import path from "path" import { Config } from "../config" import { Bus } from "../bus" import { FSUtil } from "@opencode-ai/util/fs-util" @@ -36,22 +35,29 @@ const layer = Layer.effect( .filter((entry): entry is Document => entry.type === "document") .flatMap((item) => item.info.watcher?.ignore ?? []) + const watch = (dir: string, keep: readonly string[]) => + Effect.gen(function* () { + const ignore = (yield* fs.readDirectoryEntries(dir).pipe(Effect.catch(() => Effect.succeed([])))).flatMap( + (entry) => (keep.includes(entry.name) ? [] : [entry.name]), + ) + const updates = yield* watcher.subscribe({ path: dir, type: "directory", ignore }) + yield* updates.pipe(Stream.runForEach(publish), Effect.forkScoped) + }) + if (location.vcs?.type === "git") { const resolved = (yield* git.repo.discover(location.directory))?.gitDirectory const vcs = resolved ? yield* fs.realPath(resolved).pipe(Effect.catch(() => Effect.succeed(resolved))) : undefined if (vcs && !config.includes(".git") && !config.includes(vcs) && (!resolved || !config.includes(resolved))) { - const updates = yield* watcher.subscribe({ path: path.join(vcs, "HEAD"), type: "file" }) - yield* updates.pipe(Stream.runForEach(publish), Effect.forkScoped) + yield* watch(vcs, ["HEAD", "HEAD.lock"]) } } if (location.vcs?.type === "hg") { const store = location.vcs.store const vcs = yield* fs.realPath(store).pipe(Effect.catch(() => Effect.succeed(store))) if (!config.includes(".hg") && !config.includes(vcs)) { - const updates = yield* watcher.subscribe({ path: path.join(vcs, "branch"), type: "file" }) - yield* updates.pipe(Stream.runForEach(publish), Effect.forkScoped) + yield* watch(vcs, ["branch"]) } } }).pipe( diff --git a/packages/core/src/vcs.ts b/packages/core/src/vcs.ts index db1f95d66124..47daa6792742 100644 --- a/packages/core/src/vcs.ts +++ b/packages/core/src/vcs.ts @@ -52,7 +52,10 @@ const layer = Layer.effect( const store = yield* fs.realPath(vcs.store).pipe(Effect.catch(() => Effect.succeed(vcs.store))) const isBranchMetadata = vcs.type === "git" - ? (file: string) => path.basename(file) === "HEAD" && FSUtil.contains(store, file) + ? (file: string) => { + const name = path.basename(file) + return (name === "HEAD" || name === "HEAD.lock") && FSUtil.contains(store, file) + } : (file: string) => path.resolve(file) === path.join(store, "branch") yield* bus.subscribe(FileSystem.Event.Changed).pipe( Stream.filter((event) => isBranchMetadata(event.data.file)), diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index 7b60f7dce7d9..a450323cf3c8 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -195,7 +195,12 @@ describe("LocationWatcher subscriptions", () => { Effect.retry(Schedule.spaced("10 millis")), ) yield* Effect.sleep("10 millis") - expect(subscriptions).toEqual([{ path: path.join(directory, ".git", "HEAD"), type: "file" }]) + expect(subscriptions).toHaveLength(1) + const git = subscriptions[0] + if (git?.type !== "directory") throw new Error("expected a directory watch") + expect(git.path).toBe(path.join(directory, ".git")) + expect(git.ignore ?? []).not.toContain("HEAD") + expect(git.ignore ?? []).toContain("objects") }), { vcs: "git", watcher }, ) @@ -218,7 +223,7 @@ describe("LocationWatcher subscriptions", () => { Effect.retry(Schedule.spaced("10 millis")), ) yield* Effect.sleep("10 millis") - expect(subscriptions).toEqual([{ path: path.join(directory, ".hg", "branch"), type: "file" }]) + expect(subscriptions).toMatchObject([{ path: path.join(directory, ".hg"), type: "directory" }]) }), { vcs: "hg", watcher }, ) @@ -351,6 +356,23 @@ describeNative("LocationWatcher", () => { ), ) + it.live("publishes .git/HEAD events from git checkout", () => + withTmp( + (directory) => + Effect.gen(function* () { + const head = path.join(directory, ".git", "HEAD") + const branch = `watch-${Math.random().toString(36).slice(2)}` + yield* ready(head) + const event = yield* nextUpdate( + (item) => path.basename(item.file) === "HEAD" || path.basename(item.file) === "HEAD.lock", + Effect.promise(() => $`git checkout -q -b ${branch}`.cwd(directory).quiet()), + ) + expect(["HEAD", "HEAD.lock"]).toContain(path.basename(event.file)) + }), + { vcs: "git" }, + ), + ) + const describeSymlink = process.platform !== "win32" ? describe : describe.skip describeSymlink("symlinked .git", () => { it.live("publishes .git/HEAD events through a symlinked .git directory", () => diff --git a/packages/core/test/vcs.test.ts b/packages/core/test/vcs.test.ts index 677fb4f156ad..aa87a8ee8b30 100644 --- a/packages/core/test/vcs.test.ts +++ b/packages/core/test/vcs.test.ts @@ -112,7 +112,10 @@ describe("Vcs", () => { yield* bus.publish(FileSystem.Event.Changed, { file: path.join(directory, "HEAD"), event: "change" }) expect(yield* vcs.info()).toEqual({ branch: { current: "main", default: undefined } }) - yield* bus.publish(FileSystem.Event.Changed, { file: path.join(directory, ".git", "HEAD"), event: "change" }) + yield* bus.publish(FileSystem.Event.Changed, { + file: path.join(directory, ".git", "HEAD.lock"), + event: "change", + }) expect(yield* Fiber.join(updated)).toMatchObject({ _tag: "Some", value: { location: { directory }, data: { branch: "feature" } }, From 868a303b3182a3aa3b8b042b9a73458da87edf29 Mon Sep 17 00:00:00 2001 From: Aiden Cline Date: Mon, 10 Aug 2026 17:46:45 -0500 Subject: [PATCH 2/2] fix(core): move branch watch into Vcs LocationWatcher only existed to republish git/hg metadata as filesystem.changed for Vcs. Nothing else consumed those events. Vcs now owns the Parcel watch, same as skills and config. --- .../core/src/filesystem/location-watcher.ts | 77 ------ packages/core/src/location-services.ts | 3 - packages/core/src/vcs.ts | 51 ++-- packages/core/test/filesystem/watcher.test.ts | 261 +----------------- packages/core/test/vcs-hg.test.ts | 13 +- packages/core/test/vcs.test.ts | 165 ++++++++--- 6 files changed, 164 insertions(+), 406 deletions(-) delete mode 100644 packages/core/src/filesystem/location-watcher.ts diff --git a/packages/core/src/filesystem/location-watcher.ts b/packages/core/src/filesystem/location-watcher.ts deleted file mode 100644 index a2f7fa6de245..000000000000 --- a/packages/core/src/filesystem/location-watcher.ts +++ /dev/null @@ -1,77 +0,0 @@ -export * as LocationWatcher from "./location-watcher" - -import { makeLocationNode } from "@opencode-ai/util/effect/app-node" -import { Context, Effect, Layer, Stream } from "effect" -import { FileSystem } from "@opencode-ai/schema/filesystem" -import { Document } from "@opencode-ai/schema/config" -import { Config } from "../config" -import { Bus } from "../bus" -import { FSUtil } from "@opencode-ai/util/fs-util" -import { Git } from "../git" -import { Location } from "../location" -import { Watcher } from "./watcher" - -export interface Interface {} - -export class Service extends Context.Service()("@opencode/LocationWatcher") {} - -const layer = Layer.effect( - Service, - Effect.gen(function* () { - const location = yield* Location.Service - const watcher = yield* Watcher.Service - const bus = yield* Bus.Service - const fs = yield* FSUtil.Service - const git = yield* Git.Service - const configService = yield* Config.Service - const publish = (update: { type: "create" | "update" | "delete"; path: string }) => - bus.publish(FileSystem.Event.Changed, { - file: update.path, - event: update.type === "create" ? "add" : update.type === "update" ? "change" : "unlink", - }) - - yield* Effect.gen(function* () { - const config = (yield* configService.entries()) - .filter((entry): entry is Document => entry.type === "document") - .flatMap((item) => item.info.watcher?.ignore ?? []) - - const watch = (dir: string, keep: readonly string[]) => - Effect.gen(function* () { - const ignore = (yield* fs.readDirectoryEntries(dir).pipe(Effect.catch(() => Effect.succeed([])))).flatMap( - (entry) => (keep.includes(entry.name) ? [] : [entry.name]), - ) - const updates = yield* watcher.subscribe({ path: dir, type: "directory", ignore }) - yield* updates.pipe(Stream.runForEach(publish), Effect.forkScoped) - }) - - if (location.vcs?.type === "git") { - const resolved = (yield* git.repo.discover(location.directory))?.gitDirectory - const vcs = resolved - ? yield* fs.realPath(resolved).pipe(Effect.catch(() => Effect.succeed(resolved))) - : undefined - if (vcs && !config.includes(".git") && !config.includes(vcs) && (!resolved || !config.includes(resolved))) { - yield* watch(vcs, ["HEAD", "HEAD.lock"]) - } - } - if (location.vcs?.type === "hg") { - const store = location.vcs.store - const vcs = yield* fs.realPath(store).pipe(Effect.catch(() => Effect.succeed(store))) - if (!config.includes(".hg") && !config.includes(vcs)) { - yield* watch(vcs, ["branch"]) - } - } - }).pipe( - Effect.withSpan("LocationWatcher.start", { attributes: { directory: location.directory } }), - Effect.catchCause((cause) => Effect.logError("failed to init location watcher service", { cause })), - Effect.forkScoped, - ) - - return Service.of({}) - }), -) - -export const node = makeLocationNode({ - service: Service, - layer, - deps: [Watcher.node, FSUtil.node, Location.node, Config.node, Git.node, Bus.node], -}) diff --git a/packages/core/src/location-services.ts b/packages/core/src/location-services.ts index 53bdaf363386..cb2ba2c3a2be 100644 --- a/packages/core/src/location-services.ts +++ b/packages/core/src/location-services.ts @@ -15,7 +15,6 @@ import { FileSystemSearch } from "./filesystem/search" import { Generate } from "./generate" import { Form } from "./form" import { Image } from "./image" -import { LocationWatcher } from "./filesystem/location-watcher" import { Integration } from "./integration" import { Location } from "./location" import { LocationMutation } from "./location-mutation" @@ -99,8 +98,6 @@ const locationServiceNodes = [ Snapshot.node, SessionRunnerLLM.node, Vcs.node, - // Start repository watches only after boot-critical filesystem and Git work. - LocationWatcher.node, ] as const satisfies readonly Node.LocationNode[] export const locationServices = LayerNode.group(locationServiceNodes) diff --git a/packages/core/src/vcs.ts b/packages/core/src/vcs.ts index 47daa6792742..b9dbdcd92f78 100644 --- a/packages/core/src/vcs.ts +++ b/packages/core/src/vcs.ts @@ -3,7 +3,6 @@ export * as Vcs from "./vcs" import path from "path" import { Context, Effect, Layer, Stream } from "effect" import { FileDiff } from "@opencode-ai/schema/file-diff" -import { FileSystem } from "@opencode-ai/schema/filesystem" import { FileStatus, Info, Mode } from "@opencode-ai/schema/vcs" import { VcsEvent } from "@opencode-ai/schema/vcs-event" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" @@ -11,6 +10,8 @@ import { FSUtil } from "@opencode-ai/util/fs-util" import { Location } from "./location" import { AppProcess } from "@opencode-ai/util/process" import { Bus } from "./bus" +import { Git } from "./git" +import { Watcher } from "./filesystem/watcher" import { VcsGit } from "./vcs/git" import { VcsHg } from "./vcs/hg" @@ -44,32 +45,36 @@ const layer = Layer.effect( const fs = yield* FSUtil.Service const location = yield* Location.Service const bus = yield* Bus.Service + const git = yield* Git.Service + const watcher = yield* Watcher.Service const impl = adapter(proc, fs, location) const vcs = location.vcs const state = { info: impl ? yield* impl.info() : ({ branch: {} } satisfies Info) } if (vcs && impl) { - const store = yield* fs.realPath(vcs.store).pipe(Effect.catch(() => Effect.succeed(vcs.store))) - const isBranchMetadata = - vcs.type === "git" - ? (file: string) => { - const name = path.basename(file) - return (name === "HEAD" || name === "HEAD.lock") && FSUtil.contains(store, file) - } - : (file: string) => path.resolve(file) === path.join(store, "branch") - yield* bus.subscribe(FileSystem.Event.Changed).pipe( - Stream.filter((event) => isBranchMetadata(event.data.file)), - Stream.runForEach((event) => - Effect.gen(function* () { - const next = yield* impl.info() - const changed = state.info.branch.current !== next.branch.current - state.info = next - if (!changed) return - yield* bus.publish(VcsEvent.BranchUpdated, { branch: next.branch.current }) - }).pipe(Effect.withSpan("Vcs.refreshBranch", { attributes: { file: event.data.file } })), - ), - Effect.forkScoped({ startImmediately: true }), - ) + yield* Effect.gen(function* () { + const discovered = vcs.type === "git" ? (yield* git.repo.discover(location.directory))?.gitDirectory : undefined + const target = discovered ?? vcs.store + const dir = yield* fs.realPath(target).pipe(Effect.catch(() => Effect.succeed(target))) + const keep = vcs.type === "git" ? ["HEAD", "HEAD.lock"] : ["branch"] + const ignore = (yield* fs.readDirectoryEntries(dir).pipe(Effect.catch(() => Effect.succeed([])))).flatMap( + (entry) => (keep.includes(entry.name) ? [] : [entry.name]), + ) + const updates = yield* watcher.subscribe({ path: dir, type: "directory", ignore }) + yield* updates.pipe( + Stream.filter((update) => keep.includes(path.basename(update.path))), + Stream.runForEach((update) => + Effect.gen(function* () { + const next = yield* impl.info() + const changed = state.info.branch.current !== next.branch.current + state.info = next + if (!changed) return + yield* bus.publish(VcsEvent.BranchUpdated, { branch: next.branch.current }) + }).pipe(Effect.withSpan("Vcs.refreshBranch", { attributes: { file: update.path } })), + ), + Effect.forkScoped({ startImmediately: true }), + ) + }).pipe(Effect.catchCause((cause) => Effect.logError("failed to watch vcs metadata", { cause }))) } return Service.of({ @@ -91,5 +96,5 @@ const layer = Layer.effect( export const node = makeLocationNode({ service: Service, layer: layer, - deps: [AppProcess.node, FSUtil.node, Location.node, Bus.node], + deps: [AppProcess.node, FSUtil.node, Location.node, Bus.node, Git.node, Watcher.node], }) diff --git a/packages/core/test/filesystem/watcher.test.ts b/packages/core/test/filesystem/watcher.test.ts index a450323cf3c8..a567360d4de4 100644 --- a/packages/core/test/filesystem/watcher.test.ts +++ b/packages/core/test/filesystem/watcher.test.ts @@ -1,28 +1,15 @@ -import { $ } from "bun" import { describe, expect } from "bun:test" -import fs from "fs/promises" import path from "path" -import { Deferred, Duration, Effect, Fiber, Layer, Option, Schedule, Stream } from "effect" -import { Config } from "@opencode-ai/core/config" +import { Deferred, Effect, Fiber, Layer, Schedule, Stream } from "effect" import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" -import { LayerNode } from "@opencode-ai/util/effect/layer-node" -import { Bus } from "@opencode-ai/core/bus" import { FSUtil } from "@opencode-ai/util/fs-util" -import { LocationWatcher } from "@opencode-ai/core/filesystem/location-watcher" import { Watcher } from "@opencode-ai/core/filesystem/watcher" -import { FileSystem } from "@opencode-ai/schema/filesystem" -import { Location } from "@opencode-ai/core/location" -import { AbsolutePath } from "@opencode-ai/core/schema" -import { location } from "../fixture/location" import { tmpdir } from "../fixture/tmpdir" import { testEffect } from "../lib/effect" -type WatcherEvent = { file: string; event: "add" | "change" | "unlink" } const describeNative = process.env.CI ? describe.skip : describe -const it = testEffect(AppNodeBuilder.build(LayerNode.group([FSUtil.node, Bus.node]))) - -const configLayer = Config.testLayer() +const it = testEffect(AppNodeBuilder.build(FSUtil.node)) describe("Watcher.testLayer", () => { it.effect("records subscriptions and broadcasts emitted updates through the service", () => @@ -40,7 +27,6 @@ describe("Watcher.testLayer", () => { yield* test.emit({ type: "update", path: "/root/file.md" }) expect(Array.from(yield* Fiber.join(received))).toEqual([{ type: "update", path: "/root/file.md" }]) - // subscriptions() reports acquired watches, so paths come back resolved. expect(yield* test.subscriptions()).toEqual([{ path: path.resolve("/root"), type: "directory" }]) }).pipe(Effect.provide(Watcher.testLayer)), ) @@ -126,172 +112,20 @@ describe("Watcher lifecycle", () => { expect(counts.unsubscribes).toBe(0) return consumer }).pipe(withNative(native)) - // Closing the layer scope tears the native subscription down while the - // consumer still holds a reference; the consumer's own release as its - // stream ends must not tear it down a second time. yield* Fiber.join(consumer) expect(counts.unsubscribes).toBe(1) }) }) }) -function provide(directory: string, vcs?: Location.Interface["vcs"], watcher?: Layer.Layer) { - const locationLayer = Layer.succeed( - Location.Service, - Location.Service.of(location({ directory: AbsolutePath.make(directory) }, { vcs })), - ) - const built = AppNodeBuilder.build(LocationWatcher.node, [ - [Config.node, configLayer], - [Location.node, locationLayer], - ...(watcher ? ([[Watcher.node, watcher]] as const) : []), - ]) - return Effect.provide(built) -} - -function withTmp( - f: (directory: string, vcs?: Location.Interface["vcs"]) => Effect.Effect, - options?: { - vcs?: "git" | "hg" - init?: (directory: string) => Promise - watcher?: Layer.Layer - }, -) { +function withTmp(f: (directory: string) => Effect.Effect) { return Effect.acquireRelease( - Effect.promise(async () => { - const tmp = await tmpdir() - if (options?.vcs === "hg") { - await fs.mkdir(path.join(tmp.path, ".hg")) - return { tmp, vcs: { type: "hg" as const, store: AbsolutePath.make(path.join(tmp.path, ".hg")) } } - } - if (options?.vcs !== "git") return { tmp, vcs: undefined } - await $`git init`.cwd(tmp.path).quiet() - await $`git config core.fsmonitor false`.cwd(tmp.path).quiet() - await $`git config commit.gpgsign false`.cwd(tmp.path).quiet() - await $`git config user.email test@opencode.test`.cwd(tmp.path).quiet() - await $`git config user.name Test`.cwd(tmp.path).quiet() - await $`git commit --allow-empty -m root`.cwd(tmp.path).quiet() - await options.init?.(tmp.path) - return { tmp, vcs: { type: "git" as const, store: AbsolutePath.make(path.join(tmp.path, ".git")) } } - }), - ({ tmp }) => Effect.promise(() => tmp[Symbol.asyncDispose]()), - ).pipe(Effect.flatMap(({ tmp, vcs }) => f(tmp.path, vcs).pipe(provide(tmp.path, vcs, options?.watcher)))) -} - -describe("LocationWatcher subscriptions", () => { - it.live("watches only exact Git branch metadata", () => { - const subscriptions: Watcher.WatchInput[] = [] - const watcher = Layer.succeed( - Watcher.Service, - Watcher.Service.of({ - subscribe: (input) => Effect.sync(() => subscriptions.push(input)).pipe(Effect.as(Stream.empty)), - }), - ) - return withTmp( - (directory) => - Effect.gen(function* () { - yield* LocationWatcher.Service - yield* Effect.sync(() => subscriptions.length).pipe( - Effect.filterOrFail((count) => count > 0), - Effect.retry(Schedule.spaced("10 millis")), - ) - yield* Effect.sleep("10 millis") - expect(subscriptions).toHaveLength(1) - const git = subscriptions[0] - if (git?.type !== "directory") throw new Error("expected a directory watch") - expect(git.path).toBe(path.join(directory, ".git")) - expect(git.ignore ?? []).not.toContain("HEAD") - expect(git.ignore ?? []).toContain("objects") - }), - { vcs: "git", watcher }, - ) - }) - - it.live("watches only exact Hg branch metadata", () => { - const subscriptions: Watcher.WatchInput[] = [] - const watcher = Layer.succeed( - Watcher.Service, - Watcher.Service.of({ - subscribe: (input) => Effect.sync(() => subscriptions.push(input)).pipe(Effect.as(Stream.empty)), - }), - ) - return withTmp( - (directory) => - Effect.gen(function* () { - yield* LocationWatcher.Service - yield* Effect.sync(() => subscriptions.length).pipe( - Effect.filterOrFail((count) => count > 0), - Effect.retry(Schedule.spaced("10 millis")), - ) - yield* Effect.sleep("10 millis") - expect(subscriptions).toMatchObject([{ path: path.join(directory, ".hg"), type: "directory" }]) - }), - { vcs: "hg", watcher }, - ) - }) -}) - -function wait(check: (event: WatcherEvent) => boolean) { - return Effect.gen(function* () { - const bus = yield* Bus.Service - const deferred = yield* Deferred.make() - const fiber = yield* bus.subscribe(FileSystem.Event.Changed).pipe( - Stream.runForEach((event) => { - if (!check(event.data)) return Effect.void - return Deferred.succeed(deferred, event.data).pipe(Effect.asVoid) - }), - Effect.forkScoped, - ) - yield* Effect.yieldNow - return { deferred, fiber } - }) -} - -function maybeNextUpdate( - check: (event: WatcherEvent) => boolean, - trigger: Effect.Effect, - timeout: Duration.Input = "5 seconds", -) { - return Effect.acquireUseRelease( - wait(check), - ({ deferred }) => trigger.pipe(Effect.andThen(Deferred.await(deferred)), Effect.timeoutOption(timeout)), - ({ fiber }) => Fiber.interrupt(fiber), - ) -} - -function nextUpdate(check: (event: WatcherEvent) => boolean, trigger: Effect.Effect) { - return Effect.gen(function* () { - const result = yield* maybeNextUpdate(check, trigger) - if (Option.isSome(result)) return result.value - return yield* Effect.fail(new Error("timed out waiting for file watcher update")) - }) -} - -function eventuallyUpdate(check: (event: WatcherEvent) => boolean, trigger: () => Effect.Effect) { - return Effect.gen(function* () { - while (true) { - const result = yield* maybeNextUpdate(check, trigger(), "250 millis") - if (Option.isSome(result)) return result.value - } - }).pipe( - Effect.timeoutOrElse({ - duration: "5 seconds", - orElse: () => Effect.fail(new Error("timed out waiting for file watcher readiness")), - }), - ) -} - -function ready(file: string, eventFile = file) { - return Effect.gen(function* () { - const fs = yield* FSUtil.Service - const content = (yield* fs.readFileStringSafe(file)) ?? `ready-${Math.random()}` - yield* eventuallyUpdate( - (event) => event.file === eventFile, - () => fs.writeFileString(file, content), - ).pipe(Effect.asVoid) - }) + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ).pipe(Effect.flatMap((tmp) => f(tmp.path))) } -describeNative("LocationWatcher", () => { +describeNative("Watcher", () => { it.live("limits file watches to the exact target", () => withTmp((directory) => Effect.gen(function* () { @@ -338,85 +172,4 @@ describeNative("LocationWatcher", () => { }).pipe(Effect.provide(AppNodeBuilder.build(Watcher.node))), ), ) - - it.live("publishes .git/HEAD events", () => - withTmp( - (directory) => - Effect.gen(function* () { - const fs = yield* FSUtil.Service - const head = path.join(directory, ".git", "HEAD") - const branch = `watch-${Math.random().toString(36).slice(2)}` - yield* ready(head) - yield* Effect.promise(() => $`git branch ${branch}`.cwd(directory).quiet()) - expect( - yield* nextUpdate((event) => event.file === head, fs.writeFileString(head, `ref: refs/heads/${branch}\n`)), - ).toEqual({ file: head, event: "change" }) - }), - { vcs: "git" }, - ), - ) - - it.live("publishes .git/HEAD events from git checkout", () => - withTmp( - (directory) => - Effect.gen(function* () { - const head = path.join(directory, ".git", "HEAD") - const branch = `watch-${Math.random().toString(36).slice(2)}` - yield* ready(head) - const event = yield* nextUpdate( - (item) => path.basename(item.file) === "HEAD" || path.basename(item.file) === "HEAD.lock", - Effect.promise(() => $`git checkout -q -b ${branch}`.cwd(directory).quiet()), - ) - expect(["HEAD", "HEAD.lock"]).toContain(path.basename(event.file)) - }), - { vcs: "git" }, - ), - ) - - const describeSymlink = process.platform !== "win32" ? describe : describe.skip - describeSymlink("symlinked .git", () => { - it.live("publishes .git/HEAD events through a symlinked .git directory", () => - withTmp( - (directory) => - Effect.gen(function* () { - const afs = yield* FSUtil.Service - const actual = path.join(directory, "..", `actual_${path.basename(directory)}`) - yield* Effect.addFinalizer(() => Effect.promise(() => fs.rm(actual, { recursive: true, force: true }))) - const head = path.join(directory, ".git", "HEAD") - yield* ready(head, path.join(actual, "HEAD")) - const branch = `watch-${Math.random().toString(36).slice(2)}` - yield* Effect.promise(() => $`git branch ${branch}`.cwd(directory).quiet()) - expect( - yield* nextUpdate( - (event) => event.file === path.join(actual, "HEAD"), - afs.writeFileString(head, `ref: refs/heads/${branch}\n`), - ), - ).toEqual({ file: path.join(actual, "HEAD"), event: "change" }) - }), - { - vcs: "git", - init: async (directory) => { - const actual = path.join(directory, "..", `actual_${path.basename(directory)}`) - await fs.rename(path.join(directory, ".git"), actual) - await fs.symlink(actual, path.join(directory, ".git")) - }, - }, - ), - ) - }) - - it.live("publishes .hg/branch events", () => - withTmp( - (directory) => - Effect.gen(function* () { - const fs = yield* FSUtil.Service - const branch = path.join(directory, ".hg", "branch") - yield* ready(branch) - expect( - yield* nextUpdate((event) => event.file === branch, fs.writeFileString(branch, "feature\n")), - ).toMatchObject({ file: branch }) - }), - { vcs: "hg" }, - ), - ) }) diff --git a/packages/core/test/vcs-hg.test.ts b/packages/core/test/vcs-hg.test.ts index 38b43f885bfc..76b4969ecbf5 100644 --- a/packages/core/test/vcs-hg.test.ts +++ b/packages/core/test/vcs-hg.test.ts @@ -8,7 +8,6 @@ import { Bus } from "@opencode-ai/core/bus" import { Location } from "@opencode-ai/core/location" import { AbsolutePath } from "@opencode-ai/core/schema" import { Vcs } from "@opencode-ai/core/vcs" -import { FileSystem } from "@opencode-ai/schema/filesystem" import { VcsEvent } from "@opencode-ai/schema/vcs-event" import { location } from "./fixture/location" import { tmpdir } from "./fixture/tmpdir" @@ -42,7 +41,9 @@ const withTmp = (f: (directory: string) => Effect.Effect) => const withHg = (f: (directory: string) => Effect.Effect) => withTmp((directory) => - Effect.promise(() => hg(directory, "init")).pipe(Effect.andThen(f(directory).pipe(provide(directory)))), + Effect.promise(() => hg(directory, "init")).pipe( + Effect.andThen(f(directory).pipe(provide(directory))), + ), ) async function hg(directory: string, ...args: string[]) { @@ -124,13 +125,7 @@ describeHg("Vcs mercurial", () => { .subscribe(VcsEvent.BranchUpdated) .pipe(Stream.take(1), Stream.runHead, Effect.forkScoped({ startImmediately: true })) yield* Effect.promise(() => hg(directory, "branch", "-q", "feature")) - expect(yield* vcs.info()).toEqual({ branch: { current: "default", default: "default" } }) - - yield* bus.publish(FileSystem.Event.Changed, { - file: path.join(directory, ".hg", "branch"), - event: "change", - }) - expect(yield* Fiber.join(updated)).toMatchObject({ + expect(yield* Fiber.join(updated).pipe(Effect.timeout("5 seconds"))).toMatchObject({ _tag: "Some", value: { location: { directory }, data: { branch: "feature" } }, }) diff --git a/packages/core/test/vcs.test.ts b/packages/core/test/vcs.test.ts index aa87a8ee8b30..9104845f0fa0 100644 --- a/packages/core/test/vcs.test.ts +++ b/packages/core/test/vcs.test.ts @@ -8,27 +8,66 @@ import { Bus } from "@opencode-ai/core/bus" import { Location } from "@opencode-ai/core/location" import { AbsolutePath } from "@opencode-ai/core/schema" import { Vcs } from "@opencode-ai/core/vcs" -import { FileSystem } from "@opencode-ai/schema/filesystem" +import { Watcher } from "@opencode-ai/core/filesystem/watcher" import { VcsEvent } from "@opencode-ai/schema/vcs-event" import { location } from "./fixture/location" import { tmpdir } from "./fixture/tmpdir" import { it } from "./lib/effect" +const describeNative = process.env.CI ? describe.skip : describe + +const locationLayer = (directory: string, git?: boolean) => + Layer.succeed( + Location.Service, + Location.Service.of( + location( + { directory: AbsolutePath.make(directory) }, + git ? { vcs: { type: "git", store: AbsolutePath.make(path.join(directory, ".git")) } } : {}, + ), + ), + ) + const provide = (directory: string, input: { git?: boolean } = {}) => + Effect.provide( + LayerNode.compile(LayerNode.group([Vcs.node, Bus.node]), [[Location.node, locationLayer(directory, input.git)]]), + ) + +function fakeWatcher() { + const subscriptions: Watcher.WatchInput[] = [] + const active = new Set<(update: Watcher.Update) => void>() + const native = Watcher.Native.of({ + subscribe: (input) => + Effect.sync(() => { + subscriptions.push( + input.type === "file" + ? { path: input.target, type: "file" } + : input.ignore.length > 0 + ? { path: input.target, type: "directory", ignore: input.ignore } + : { path: input.target, type: "directory" }, + ) + active.add(input.publish) + return { + unsubscribe: () => { + active.delete(input.publish) + return Promise.resolve() + }, + } + }), + }) + return { + subscriptions: () => [...subscriptions], + emit: (update: Watcher.Update) => { + for (const publish of active) publish(update) + }, + layer: Watcher.layer().pipe(Layer.provide(Layer.succeed(Watcher.Native, native))), + } +} + +const provideFake = (directory: string, fake: ReturnType, git = true) => Effect.provide( LayerNode.compile(LayerNode.group([Vcs.node, Bus.node]), [ - [ - Location.node, - Layer.succeed( - Location.Service, - Location.Service.of( - location( - { directory: AbsolutePath.make(directory) }, - input.git ? { vcs: { type: "git", store: AbsolutePath.make(path.join(directory, ".git")) } } : {}, - ), - ), - ), - ], + [Location.node, locationLayer(directory, git)], + [Watcher.node, fake.layer], ]), ) @@ -93,38 +132,84 @@ describe("Vcs", () => { ), ) - it.live("caches branch info and publishes HEAD changes", () => - withGit((directory) => - Effect.gen(function* () { - yield* Effect.promise(async () => { - await fs.writeFile(path.join(directory, "file.txt"), "one\n") - await commitAll(directory, "initial") - }) - const vcs = yield* Vcs.Service - const bus = yield* Bus.Service - expect(yield* vcs.info()).toEqual({ branch: { current: "main", default: undefined } }) + it.live("watches git branch metadata", () => + withTmp((directory) => { + const fake = fakeWatcher() + return Effect.promise(() => initRepo(directory)).pipe( + Effect.andThen( + Effect.gen(function* () { + yield* Vcs.Service + expect(fake.subscriptions()).toHaveLength(1) + const git = fake.subscriptions()[0] + if (git?.type !== "directory") throw new Error("expected a directory watch") + expect(git.path).toBe(path.join(directory, ".git")) + expect(git.ignore ?? []).not.toContain("HEAD") + expect(git.ignore ?? []).toContain("objects") + }).pipe(provideFake(directory, fake)), + ), + ) + }), + ) - const updated = yield* bus - .subscribe(VcsEvent.BranchUpdated) - .pipe(Stream.take(1), Stream.runHead, Effect.forkScoped({ startImmediately: true })) - yield* Effect.promise(() => $`git checkout -q -b feature`.cwd(directory).quiet()) + it.live("caches branch info and publishes HEAD changes", () => + withTmp((directory) => { + const fake = fakeWatcher() + return Effect.promise(async () => { + await initRepo(directory) + await fs.writeFile(path.join(directory, "file.txt"), "one\n") + await commitAll(directory, "initial") + }).pipe( + Effect.andThen( + Effect.gen(function* () { + const vcs = yield* Vcs.Service + const bus = yield* Bus.Service + expect(yield* vcs.info()).toMatchObject({ branch: { current: "main" } }) - yield* bus.publish(FileSystem.Event.Changed, { file: path.join(directory, "HEAD"), event: "change" }) - expect(yield* vcs.info()).toEqual({ branch: { current: "main", default: undefined } }) + const updated = yield* bus + .subscribe(VcsEvent.BranchUpdated) + .pipe(Stream.take(1), Stream.runHead, Effect.forkScoped({ startImmediately: true })) + yield* Effect.promise(() => $`git checkout -q -b feature`.cwd(directory).quiet()) + fake.emit({ type: "update", path: path.join(directory, ".git", "index.lock") }) + expect(yield* vcs.info()).toMatchObject({ branch: { current: "main" } }) - yield* bus.publish(FileSystem.Event.Changed, { - file: path.join(directory, ".git", "HEAD.lock"), - event: "change", - }) - expect(yield* Fiber.join(updated)).toMatchObject({ - _tag: "Some", - value: { location: { directory }, data: { branch: "feature" } }, - }) - expect(yield* vcs.info()).toEqual({ branch: { current: "feature", default: "main" } }) - }), - ), + fake.emit({ type: "update", path: path.join(directory, ".git", "HEAD.lock") }) + expect(yield* Fiber.join(updated)).toMatchObject({ + _tag: "Some", + value: { location: { directory }, data: { branch: "feature" } }, + }) + expect(yield* vcs.info()).toMatchObject({ branch: { current: "feature" } }) + }).pipe(provideFake(directory, fake)), + ), + ) + }), ) + describeNative("native watches", () => { + it.live("publishes branch updates on git checkout", () => + withGit((directory) => + Effect.gen(function* () { + yield* Effect.promise(async () => { + await fs.writeFile(path.join(directory, "file.txt"), "one\n") + await commitAll(directory, "initial") + }) + const vcs = yield* Vcs.Service + const bus = yield* Bus.Service + expect(yield* vcs.info()).toMatchObject({ branch: { current: "main" } }) + const updated = yield* bus + .subscribe(VcsEvent.BranchUpdated) + .pipe(Stream.take(1), Stream.runHead, Effect.forkScoped({ startImmediately: true })) + yield* Effect.promise(() => $`git checkout -q -b feature`.cwd(directory).quiet()) + expect(yield* Fiber.join(updated).pipe(Effect.timeout("5 seconds"))).toMatchObject({ + _tag: "Some", + value: { data: { branch: "feature" } }, + }) + expect(yield* vcs.info()).toMatchObject({ branch: { current: "feature" } }) + }), + ), + { timeout: 15_000 }, + ) + }) + it.live("diffs the working copy against HEAD with patches", () => withGit((directory) => Effect.gen(function* () {