Compare commits

..

2 Commits

Author SHA1 Message Date
Aiden Cline 868a303b31 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.
2026-08-10 17:46:45 -05:00
Aiden Cline e5b576ce3a 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.
2026-08-10 17:35:50 -05:00
17 changed files with 186 additions and 451 deletions
@@ -41,7 +41,6 @@ export type SessionMessageAgentSelected = {
time: { created: number }
type: "agent-switched"
agent: string
previous?: string
}
export type PromptBase64 = string
@@ -2536,7 +2535,6 @@ export type SessionImportInput = {
readonly time: { readonly created: number }
readonly type: "agent-switched"
readonly agent: string
readonly previous?: string
}
| {
readonly id: string
@@ -2788,7 +2786,6 @@ export type SessionImportInput = {
readonly time: { readonly created: number }
readonly type: "agent-switched"
readonly agent: string
readonly previous?: string
}
| {
readonly id: string
@@ -3040,7 +3037,6 @@ export type SessionImportInput = {
readonly time: { readonly created: number }
readonly type: "agent-switched"
readonly agent: string
readonly previous?: string
}
| {
readonly id: string
@@ -1,4 +1,3 @@
import { readFile } from "node:fs/promises"
import path from "node:path"
import { sql } from "drizzle-orm"
import { Effect, Option, Schema } from "effect"
@@ -42,9 +41,9 @@ export default migration
export function importLegacyCredentials(tx: Parameters<DatabaseMigration.Migration["up"]>[0], filepath: string) {
return Effect.gen(function* () {
const content = yield* Effect.promise(() => readFile(filepath, "utf8").catch(() => undefined))
if (content === undefined) return
const input = Option.getOrUndefined(decodeJson(content))
const file = Bun.file(filepath)
if (!(yield* Effect.promise(() => file.exists()))) return
const input = Option.getOrUndefined(decodeJson(yield* Effect.promise(() => file.text())))
if (typeof input !== "object" || input === null || Array.isArray(input)) {
return yield* Effect.fail(new Error("Legacy credential file must contain an object"))
}
@@ -1,71 +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 path from "path"
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<Service, Interface>()("@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 ?? [])
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)
}
}
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)
}
}
}).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],
})
-3
View File
@@ -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<unknown, unknown>[]
export const locationServices = LayerNode.group<typeof locationServiceNodes>(locationServiceNodes)
+9 -14
View File
@@ -4,7 +4,6 @@ import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
export interface Adapter {
readonly getAgent: () => Effect.Effect<SessionMessage.AgentSelected["agent"] | undefined, never, never>
readonly getModel: () => Effect.Effect<SessionMessage.ModelSelected["model"] | undefined, never, never>
readonly getCurrentAssistant: () => Effect.Effect<SessionMessage.Assistant | undefined, never, never>
readonly getAssistant: (
@@ -60,19 +59,15 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) {
"session.created": () => Effect.void,
"session.usage.recorded": () => Effect.void,
"session.agent.selected": (event) => {
return Effect.gen(function* () {
const previous = yield* adapter.getAgent()
yield* adapter.appendMessage(
SessionMessage.AgentSelected.make({
id: SessionMessage.ID.fromEvent(event.id),
type: "agent-switched",
metadata: event.metadata,
agent: event.data.agent,
previous,
time: { created: event.created },
}),
)
})
return adapter.appendMessage(
SessionMessage.AgentSelected.make({
id: SessionMessage.ID.fromEvent(event.id),
type: "agent-switched",
metadata: event.metadata,
agent: event.data.agent,
time: { created: event.created },
}),
)
},
"session.model.selected": (event) => {
return Effect.gen(function* () {
+6 -21
View File
@@ -5,7 +5,6 @@ import { DateTime, Effect, Layer, Schema, Stream } from "effect"
import { Database } from "../database/database"
import { Bus } from "../bus"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { Agent } from "../agent"
import { Model } from "../model"
import { SessionEvent } from "./event"
import { SessionMessage } from "./message"
@@ -231,17 +230,6 @@ function run(db: DatabaseService, event: MessageEvent) {
}
const appendMessage = (message: SessionMessage.Info) => insertMessage(db, event, message)
const adapter: SessionMessageUpdater.Adapter = {
getAgent() {
return db
.select({ agent: SessionTable.agent })
.from(SessionTable)
.where(eq(SessionTable.id, event.data.sessionID))
.get()
.pipe(
Effect.orDie,
Effect.map((row) => (row?.agent ? Agent.ID.make(row.agent) : undefined)),
)
},
getModel() {
return db
.select({ model: SessionTable.model })
@@ -410,15 +398,12 @@ const layer = Layer.effectDiscard(
db.delete(SessionTable).where(eq(SessionTable.id, event.data.sessionID)).run().pipe(Effect.orDie),
)
yield* bus.project(SessionEvent.AgentSelected, (event) =>
Effect.gen(function* () {
yield* run(db, event)
yield* db
.update(SessionTable)
.set({ agent: event.data.agent, time_updated: DateTime.toEpochMillis(event.created) })
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie)
}),
db
.update(SessionTable)
.set({ agent: event.data.agent, time_updated: DateTime.toEpochMillis(event.created) })
.where(eq(SessionTable.id, event.data.sessionID))
.run()
.pipe(Effect.orDie, Effect.andThen(run(db, event))),
)
yield* bus.project(SessionEvent.ModelSelected, (event) =>
Effect.gen(function* () {
+28 -20
View File
@@ -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,29 +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) => path.basename(file) === "HEAD" && 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({
@@ -88,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],
})
@@ -164,20 +164,6 @@ describe("DatabaseMigration", () => {
expect(await Bun.file(source).text()).toBe(content)
})
test("skips legacy credential import when the source file is absent", async () => {
await using tmp = await tmpdir()
await run(
Effect.gen(function* () {
const db = yield* makeDb
yield* DatabaseMigration.apply(db)
yield* db.transaction((tx) => importLegacyCredentials(tx, path.join(tmp.path, "missing-auth.json")))
expect(yield* db.all(sql`SELECT id FROM credential`)).toEqual([])
}),
)
})
test("rolls back a failed migration without recording it", async () => {
await run(
Effect.gen(function* () {
+7 -232
View File
@@ -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,167 +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<Watcher.Service>) {
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<A, E, R>(
f: (directory: string, vcs?: Location.Interface["vcs"]) => Effect.Effect<A, E, R>,
options?: {
vcs?: "git" | "hg"
init?: (directory: string) => Promise<void>
watcher?: Layer.Layer<Watcher.Service>
},
) {
function withTmp<A, E, R>(f: (directory: string) => Effect.Effect<A, E, R>) {
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))))
Effect.promise(() => tmpdir()),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
).pipe(Effect.flatMap((tmp) => f(tmp.path)))
}
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).toEqual([{ path: path.join(directory, ".git", "HEAD"), type: "file" }])
}),
{ 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).toEqual([{ path: path.join(directory, ".hg", "branch"), type: "file" }])
}),
{ vcs: "hg", watcher },
)
})
})
function wait(check: (event: WatcherEvent) => boolean) {
return Effect.gen(function* () {
const bus = yield* Bus.Service
const deferred = yield* Deferred.make<WatcherEvent>()
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<E>(
check: (event: WatcherEvent) => boolean,
trigger: Effect.Effect<void, E>,
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<E>(check: (event: WatcherEvent) => boolean, trigger: Effect.Effect<void, E>) {
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<E>(check: (event: WatcherEvent) => boolean, trigger: () => Effect.Effect<void, E>) {
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)
})
}
describeNative("LocationWatcher", () => {
describeNative("Watcher", () => {
it.live("limits file watches to the exact target", () =>
withTmp((directory) =>
Effect.gen(function* () {
@@ -333,68 +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" },
),
)
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" },
),
)
})
+1 -4
View File
@@ -648,7 +648,7 @@ describe("Session.create", () => {
it.effect("switches the selected agent through the durable Session event", () =>
Effect.gen(function* () {
const session = yield* Session.Service
const created = yield* session.create({ location, agent: Agent.ID.make("build") })
const created = yield* session.create({ location })
yield* session.switchAgent({ sessionID: created.id, agent: Agent.ID.make("plan") })
@@ -656,9 +656,6 @@ describe("Session.create", () => {
expect(
Array.from(yield* logEvents(session, created.id, true).pipe(Stream.drop(1), Stream.take(1), Stream.runCollect)),
).toMatchObject([{ type: "session.agent.selected", data: { agent: "plan" } }])
expect(yield* session.messages({ sessionID: created.id, order: "asc" })).toMatchObject([
{ type: "agent-switched", agent: "plan", previous: "build" },
])
}),
)
@@ -358,7 +358,6 @@ describe("SessionProjector", () => {
directory: "/project",
title: "test",
version: "test",
agent: "plan",
model: previousModel,
})
.run()
@@ -460,10 +459,6 @@ describe("SessionProjector", () => {
text: "synthetic context",
metadata: { source: "projector-test" },
})
expect(messages.find((message) => message.type === "agent-switched")).toMatchObject({
agent: build,
previous: "plan",
})
expect(messages.find((message) => message.type === "model-switched")).toMatchObject({ previous: previousModel })
expect(messages.find((message) => message.type === "shell")).toMatchObject({
command: "pwd",
+4 -9
View File
@@ -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 = <A, E, R>(f: (directory: string) => Effect.Effect<A, E, R>) =>
const withHg = <A, E, R>(f: (directory: string) => Effect.Effect<A, E, R>) =>
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" } },
})
+128 -40
View File
@@ -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<typeof fakeWatcher>, 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,35 +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 } })
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())
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" })
expect(yield* Fiber.join(updated)).toMatchObject({
_tag: "Some",
value: { location: { directory }, data: { branch: "feature" } },
})
expect(yield* vcs.info()).toEqual({ branch: { current: "feature", default: "main" } })
}),
),
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)),
),
)
}),
)
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" } })
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" } })
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* () {
-3
View File
@@ -12733,9 +12733,6 @@
},
"agent": {
"type": "string"
},
"previous": {
"type": "string"
}
},
"required": ["id", "time", "type", "agent"],
-1
View File
@@ -42,7 +42,6 @@ export const AgentSelected = Schema.Struct({
...Base,
type: Schema.tag("agent-switched"),
agent: Agent.ID,
previous: Agent.ID.pipe(optional),
}).annotate({ identifier: "Session.Message.AgentSelected" })
export interface ModelSelected extends Schema.Schema.Type<typeof ModelSelected> {}
-3
View File
@@ -12733,9 +12733,6 @@
},
"agent": {
"type": "string"
},
"previous": {
"type": "string"
}
},
"required": ["id", "time", "type", "agent"],
-3
View File
@@ -12733,9 +12733,6 @@
},
"agent": {
"type": "string"
},
"previous": {
"type": "string"
}
},
"required": ["id", "time", "type", "agent"],