mirror of
https://github.com/Kilo-Org/kilocode.git
synced 2026-08-31 01:37:28 +08:00
refactor(cli): scope watcher state per instance
This commit is contained in:
@@ -1,10 +1,12 @@
|
||||
import { InstanceState } from "@/effect/instance-state"
|
||||
import { registerDisposer } from "@/effect/instance-registry"
|
||||
import type { InstanceContext } from "@/project/instance-context"
|
||||
import * as Log from "@opencode-ai/core/util/log"
|
||||
import { Flag } from "@opencode-ai/core/flag/flag"
|
||||
import { Location } from "@opencode-ai/core/location"
|
||||
import { LocationServiceMap, locationServiceMapLayer } from "@opencode-ai/core/location-services"
|
||||
import { AbsolutePath } from "@opencode-ai/core/schema"
|
||||
import { Cause, Context, Effect, Layer, Scope } from "effect"
|
||||
import { Cause, Context, Effect, Exit, Layer, Scope } from "effect"
|
||||
|
||||
const log = Log.create({ service: "kilocode-watcher" })
|
||||
|
||||
@@ -17,13 +19,8 @@ export namespace KilocodeWatcher {
|
||||
|
||||
// Embedded editor clients (VS Code, JetBrains) have their own file watching
|
||||
// and git integration and do not consume the CLI's vcs.branch.updated event,
|
||||
// so they must not eagerly warm the location stack — that starts a native
|
||||
// @parcel/watcher subscription per instance that lives for the whole session.
|
||||
// On macOS FSEvents watches the entire subtree recursively (the ignore list
|
||||
// is only a userspace filter), so an always-on, consumer-less watcher on a
|
||||
// churny workspace burns CPU and leaks native memory while idle. The
|
||||
// standalone CLI/TUI stays eager because its sidebar branch label is the only
|
||||
// consumer and no request-driven route would otherwise build the stack.
|
||||
// so they must not eagerly warm the location stack. The standalone CLI/TUI
|
||||
// keeps this subscription for live branch-label updates.
|
||||
export function eager(client = Flag.KILO_CLIENT) {
|
||||
return client !== "vscode" && client !== "jetbrains"
|
||||
}
|
||||
@@ -33,31 +30,47 @@ export namespace KilocodeWatcher {
|
||||
Effect.gen(function* () {
|
||||
const locations = yield* LocationServiceMap.Service
|
||||
const scope = yield* Scope.Scope
|
||||
const active = new Map<string, Scope.Closeable>()
|
||||
const ref = (directory: string) => Location.Ref.make({ directory: AbsolutePath.make(directory) })
|
||||
|
||||
const state = yield* InstanceState.make(
|
||||
Effect.fn("KilocodeWatcher.state")(function* (ctx) {
|
||||
if (ctx.project.vcs !== "git") return
|
||||
// Warm the v2 location stack for this instance and hold it for the
|
||||
// instance lifetime. Its Watcher subscribes to .git so Vcs sees HEAD
|
||||
// changes and publishes vcs.branch.updated in the CLI, where no v2
|
||||
// route would otherwise build the stack. The ref must be built the
|
||||
// same way the file/pty handlers build theirs (Location.Ref.make) so
|
||||
// the LayerMap shares a single build per directory.
|
||||
const ref = Location.Ref.make({ directory: AbsolutePath.make(ctx.directory) })
|
||||
yield* locations.contextEffect(ref)
|
||||
// Tear the stack down with the instance instead of letting it idle
|
||||
// in the LayerMap; same pattern as the pty handlers' disposer.
|
||||
yield* Effect.addFinalizer(() => locations.invalidate(ref).pipe(Effect.ignore))
|
||||
}),
|
||||
const off = registerDisposer((directory) =>
|
||||
Effect.runPromise(
|
||||
Effect.gen(function* () {
|
||||
const child = active.get(directory)
|
||||
if (child) {
|
||||
active.delete(directory)
|
||||
yield* Scope.close(child, Exit.void)
|
||||
}
|
||||
yield* locations.invalidate(ref(directory))
|
||||
}).pipe(Effect.ignore),
|
||||
),
|
||||
)
|
||||
yield* Effect.addFinalizer(() => Effect.sync(off))
|
||||
yield* Effect.addFinalizer(() =>
|
||||
Effect.forEach(active.values(), (child) => Scope.close(child, Exit.void), { discard: true }).pipe(
|
||||
Effect.andThen(Effect.sync(() => active.clear())),
|
||||
),
|
||||
)
|
||||
|
||||
const warm = (ctx: InstanceContext, child: Scope.Closeable) =>
|
||||
Scope.provide(child)(locations.contextEffect(ref(ctx.directory)))
|
||||
|
||||
return Service.of({
|
||||
init: Effect.fn("KilocodeWatcher.init")(function* () {
|
||||
yield* InstanceState.get(state).pipe(
|
||||
const ctx = yield* InstanceState.context
|
||||
if (ctx.project.vcs !== "git" || active.has(ctx.directory)) return
|
||||
|
||||
const child = yield* Scope.make()
|
||||
active.set(ctx.directory, child)
|
||||
yield* warm(ctx, child).pipe(
|
||||
Effect.catchCause((cause) =>
|
||||
Effect.sync(() => log.warn("instance watcher init failed", { err: Cause.squash(cause) })),
|
||||
Effect.gen(function* () {
|
||||
if (active.get(ctx.directory) === child) active.delete(ctx.directory)
|
||||
yield* Scope.close(child, Exit.void).pipe(Effect.ignore)
|
||||
yield* Effect.sync(() => log.warn("instance watcher init failed", { err: Cause.squash(cause) }))
|
||||
}),
|
||||
),
|
||||
Effect.forkIn(scope),
|
||||
Effect.forkIn(scope, { startImmediately: true }),
|
||||
)
|
||||
}),
|
||||
})
|
||||
|
||||
@@ -1,14 +1,20 @@
|
||||
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
|
||||
import { afterAll, beforeAll, describe, expect, test } from "bun:test"
|
||||
import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner"
|
||||
import { Deferred, Effect, Fiber, Layer } from "effect"
|
||||
import { Context, Deferred, Effect, Fiber, Layer, LayerMap } from "effect"
|
||||
import * as TestConsole from "effect/testing/TestConsole"
|
||||
import { GlobalBus, type GlobalEvent } from "../../src/bus/global"
|
||||
import { InstanceRef } from "../../src/effect/instance-ref"
|
||||
import { disposeInstance } from "../../src/effect/instance-registry"
|
||||
import { Git } from "../../src/git"
|
||||
import { InstanceBootstrap } from "../../src/project/bootstrap"
|
||||
import { InstanceStore } from "../../src/project/instance-store"
|
||||
import { KilocodeWatcher } from "../../src/kilocode/watcher"
|
||||
import type { InstanceContext } from "../../src/project/instance-context"
|
||||
import { tmpdirScoped } from "../fixture/fixture"
|
||||
import { awaitWithTimeout, testEffect } from "../lib/effect"
|
||||
import { Location } from "@opencode-ai/core/location"
|
||||
import { LocationServiceMap, type LocationServices } from "@opencode-ai/core/location-services"
|
||||
|
||||
const layer = Layer.mergeAll(
|
||||
AppNodeBuilder.build(InstanceStore.node, [[InstanceStore.bootstrapNode, InstanceBootstrap.node]]),
|
||||
@@ -44,47 +50,104 @@ describe("KilocodeWatcher.eager", () => {
|
||||
})
|
||||
})
|
||||
|
||||
live("instances publish branch updates after git switch", () =>
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped({ git: true })
|
||||
const git = yield* Git.Service
|
||||
const store = yield* InstanceStore.Service
|
||||
const current = yield* git.branch(dir)
|
||||
if (!current) return yield* Effect.die("missing initial branch")
|
||||
live(
|
||||
"instances publish branch updates after git switch",
|
||||
() =>
|
||||
Effect.gen(function* () {
|
||||
const dir = yield* tmpdirScoped({ git: true })
|
||||
const git = yield* Git.Service
|
||||
const store = yield* InstanceStore.Service
|
||||
const current = yield* git.branch(dir)
|
||||
if (!current) return yield* Effect.die("missing initial branch")
|
||||
|
||||
const branch = `watch-${Math.random().toString(36).slice(2)}`
|
||||
const created = yield* git.run(["branch", branch], { cwd: dir })
|
||||
expect(created.exitCode).toBe(0)
|
||||
yield* store.load({ directory: dir })
|
||||
const branch = `watch-${Math.random().toString(36).slice(2)}`
|
||||
const created = yield* git.run(["branch", branch], { cwd: dir })
|
||||
expect(created.exitCode).toBe(0)
|
||||
yield* store.load({ directory: dir })
|
||||
|
||||
const pending = yield* Deferred.make<string | undefined>()
|
||||
const handler = (event: GlobalEvent) => {
|
||||
if (event.directory !== dir || event.payload.type !== "vcs.branch.updated") return
|
||||
if (event.payload.properties.branch !== branch) return
|
||||
Deferred.doneUnsafe(pending, Effect.succeed(event.payload.properties.branch))
|
||||
}
|
||||
GlobalBus.on("event", handler)
|
||||
yield* Effect.addFinalizer(() => Effect.sync(() => GlobalBus.off("event", handler)))
|
||||
|
||||
// The watcher exposes no readiness signal (its .git subscription is forked
|
||||
// during instance warm-up), so keep generating HEAD churn in the background
|
||||
// and synchronize on the event itself with the full test budget.
|
||||
const churn = yield* Effect.gen(function* () {
|
||||
while (true) {
|
||||
yield* git.run(["switch", current], { cwd: dir })
|
||||
yield* Effect.sleep("50 millis")
|
||||
yield* git.run(["switch", branch], { cwd: dir })
|
||||
yield* Effect.sleep("100 millis")
|
||||
const pending = yield* Deferred.make<string | undefined>()
|
||||
const handler = (event: GlobalEvent) => {
|
||||
if (event.directory !== dir || event.payload.type !== "vcs.branch.updated") return
|
||||
if (event.payload.properties.branch !== branch) return
|
||||
Deferred.doneUnsafe(pending, Effect.succeed(event.payload.properties.branch))
|
||||
}
|
||||
}).pipe(Effect.forkScoped)
|
||||
GlobalBus.on("event", handler)
|
||||
yield* Effect.addFinalizer(() => Effect.sync(() => GlobalBus.off("event", handler)))
|
||||
|
||||
const updated = yield* awaitWithTimeout(
|
||||
Deferred.await(pending),
|
||||
"timed out waiting for vcs.branch.updated",
|
||||
"15 seconds",
|
||||
)
|
||||
yield* Fiber.interrupt(churn)
|
||||
expect(updated).toBe(branch)
|
||||
}),
|
||||
// The watcher exposes no readiness signal (its .git subscription is forked
|
||||
// during instance warm-up), so keep generating HEAD churn in the background
|
||||
// and synchronize on the event itself with the full test budget.
|
||||
const churn = yield* Effect.gen(function* () {
|
||||
while (true) {
|
||||
yield* git.run(["switch", current], { cwd: dir })
|
||||
yield* Effect.sleep("50 millis")
|
||||
yield* git.run(["switch", branch], { cwd: dir })
|
||||
yield* Effect.sleep("100 millis")
|
||||
}
|
||||
}).pipe(Effect.forkScoped)
|
||||
|
||||
const updated = yield* awaitWithTimeout(
|
||||
Deferred.await(pending),
|
||||
"timed out waiting for vcs.branch.updated",
|
||||
"15 seconds",
|
||||
)
|
||||
yield* Fiber.interrupt(churn)
|
||||
expect(updated).toBe(branch)
|
||||
}),
|
||||
20_000,
|
||||
)
|
||||
|
||||
test.serial(
|
||||
"isolates location lifetimes between instances",
|
||||
async () => {
|
||||
await Effect.runPromise(
|
||||
Effect.gen(function* () {
|
||||
const one = yield* tmpdirScoped()
|
||||
const two = yield* tmpdirScoped()
|
||||
const warmed = new Map<string, number>()
|
||||
const invalidated: string[] = []
|
||||
const map = yield* LayerMap.make((ref: Location.Ref) =>
|
||||
Layer.effectContext(
|
||||
Effect.acquireRelease(
|
||||
Effect.sync(() => {
|
||||
warmed.set(ref.directory, (warmed.get(ref.directory) ?? 0) + 1)
|
||||
return Context.empty() as Context.Context<LocationServices>
|
||||
}),
|
||||
() => Effect.sync(() => invalidated.push(ref.directory)),
|
||||
),
|
||||
),
|
||||
)
|
||||
const watcher = KilocodeWatcher.layer.pipe(Layer.provide(Layer.succeed(LocationServiceMap.Service, map)))
|
||||
const services = yield* Layer.build(watcher)
|
||||
const init = (directory: string) =>
|
||||
KilocodeWatcher.Service.use((service) => service.init()).pipe(
|
||||
Effect.provide(services),
|
||||
Effect.provideService(InstanceRef, {
|
||||
directory,
|
||||
worktree: directory,
|
||||
project: { vcs: "git" },
|
||||
} as InstanceContext),
|
||||
)
|
||||
|
||||
yield* init(one)
|
||||
yield* init(one)
|
||||
yield* init(two)
|
||||
yield* Effect.yieldNow
|
||||
expect(warmed).toEqual(
|
||||
new Map([
|
||||
[one, 1],
|
||||
[two, 1],
|
||||
]),
|
||||
)
|
||||
yield* Effect.promise(() => disposeInstance(one))
|
||||
expect(invalidated).toEqual([one])
|
||||
yield* Effect.promise(() => disposeInstance(two))
|
||||
expect(invalidated).toEqual([one, two])
|
||||
}).pipe(
|
||||
Effect.scoped,
|
||||
Effect.provide(Layer.mergeAll(AppNodeBuilder.build(CrossSpawnSpawner.node), TestConsole.layer)),
|
||||
),
|
||||
)
|
||||
},
|
||||
20_000,
|
||||
)
|
||||
|
||||
@@ -7,8 +7,9 @@
|
||||
"description": "Legacy InstanceState.make usages in Kilo-owned code (packages/opencode/src/kilocode/**, packages/opencode/src/kilo-*/**). Target: encapsulate in scoped Effect Services in packages/core.",
|
||||
"allowed": {
|
||||
"packages/opencode/src/kilo-sessions/kilo-sessions.ts": { "count": 1, "owner": "session-runtime", "reason": "Kilo session coordination state" },
|
||||
"packages/opencode/src/kilocode/background-process/index.ts": { "count": 1, "owner": "process-runtime", "reason": "Directory-keyed background process registry" },
|
||||
"packages/opencode/src/kilocode/watcher.ts": { "count": 1, "owner": "watcher-runtime", "reason": "Eager location watcher subscription" }
|
||||
"packages/opencode/src/kilocode/interactive-terminal/index.ts": { "count": 1, "owner": "terminal-runtime", "reason": "Interactive terminal manager state" },
|
||||
"packages/opencode/src/kilocode/notebook/service.ts": { "count": 1, "owner": "notebook-runtime", "reason": "Notebook cell execution service state" },
|
||||
"packages/opencode/src/kilocode/project-id.ts": { "count": 1, "owner": "project-runtime", "reason": "Cached project identifier resolution" }
|
||||
}
|
||||
},
|
||||
"kilo-database-constructors": {
|
||||
|
||||
Reference in New Issue
Block a user