diff --git a/packages/opencode/src/cli/tui/worker.ts b/packages/opencode/src/cli/tui/worker.ts index 4cf6b2d446b3..913e9372d3c8 100644 --- a/packages/opencode/src/cli/tui/worker.ts +++ b/packages/opencode/src/cli/tui/worker.ts @@ -2,14 +2,12 @@ import { Server } from "@/server/server" import { InstanceRuntime } from "@/project/instance-runtime" import { Rpc } from "@/util/rpc" import { upgrade } from "@/cli/upgrade" -import { Config } from "@/config/config" import { GlobalBus } from "@/bus/global" import { ServerAuth } from "@/server/auth" import { writeHeapSnapshot } from "node:v8" import { Heap } from "@/cli/heap" import { AppRuntime } from "@/effect/app-runtime" -import { Effect } from "effect" -import { disposeAllInstancesAndEmitGlobalDisposed } from "@/server/global-lifecycle" +import { reloadWhenSessionsIdle } from "@/server/global-lifecycle" Heap.start() @@ -26,6 +24,7 @@ GlobalBus.on("event", (event) => { }) let server: Awaited> | undefined +let reloading: Promise | undefined export const rpc = { async fetch(input: { url: string; method: string; headers: Record; body?: string }) { @@ -61,13 +60,14 @@ export const rpc = { await upgrade().catch(() => {}) }, async reload() { - await AppRuntime.runPromise( - Effect.gen(function* () { - const cfg = yield* Config.Service - yield* cfg.invalidate() - yield* disposeAllInstancesAndEmitGlobalDisposed({ swallowErrors: true }) - }), - ) + // SIGUSR2 arrives from desktop environments on theme changes, so a reload + // can land mid-run. Signals that arrive while one is pending join it. + if (!reloading) { + reloading = AppRuntime.runPromise(reloadWhenSessionsIdle()).finally(() => { + reloading = undefined + }) + } + await reloading }, async shutdown() { await InstanceRuntime.disposeAllInstances() diff --git a/packages/opencode/src/project/instance-store.ts b/packages/opencode/src/project/instance-store.ts index 720549ddaff7..4740e5df27cd 100644 --- a/packages/opencode/src/project/instance-store.ts +++ b/packages/opencode/src/project/instance-store.ts @@ -19,6 +19,8 @@ export interface LoadInput { export interface Interface { readonly load: (input: LoadInput) => Effect.Effect + /** Loaded instances only: entries still booting are awaited, entries whose boot failed are omitted. */ + readonly list: () => Effect.Effect readonly reload: (input: LoadInput) => Effect.Effect readonly dispose: (ctx: InstanceContext) => Effect.Effect readonly disposeDirectory: (directory: string) => Effect.Effect @@ -123,6 +125,19 @@ const layer: Layer.Layer Deferred.await(entry.deferred).pipe(Effect.exit)) + const current = [...cache.values()] + if (current.length === entries.length && current.every((entry, index) => entry === entries[index])) { + return exits.filter(Exit.isSuccess).map((exit) => exit.value) + } + } + }) + const reload = (input: LoadInput): Effect.Effect => { const directory = FSUtil.resolve(input.directory) return Effect.uninterruptibleMask((restore) => @@ -193,6 +208,7 @@ const layer: Layer.Layer + status.list().pipe(Effect.provideService(InstanceRef, ctx)), + ) + const sessions = active.reduce((count, item) => count + item.size, 0) + if (sessions === 0) break + if (!deferred) yield* Effect.logInfo("deferring reload until sessions are idle", { sessions }) + deferred = true + yield* Effect.sleep(IDLE_POLL_INTERVAL) + } + if (deferred) yield* Effect.logInfo("sessions idle, reloading") + yield* config.invalidate() + yield* disposeAllInstancesAndEmitGlobalDisposed({ swallowErrors: true }) +}) + +const IDLE_POLL_INTERVAL = "250 millis" + export * as GlobalLifecycle from "./global-lifecycle" diff --git a/packages/opencode/test/project/instance.test.ts b/packages/opencode/test/project/instance.test.ts index f78b99ef7d9b..90335908e325 100644 --- a/packages/opencode/test/project/instance.test.ts +++ b/packages/opencode/test/project/instance.test.ts @@ -50,6 +50,19 @@ describe("InstanceStore", () => { }), ) + it.live("lists loaded instance contexts", () => + Effect.gen(function* () { + const first = yield* tmpdirScoped({ git: true }) + const second = yield* tmpdirScoped({ git: true }) + const store = yield* InstanceStore.Service + + yield* store.load({ directory: first }) + yield* store.load({ directory: second }) + + expect((yield* store.list()).map((ctx) => ctx.directory).toSorted()).toEqual([first, second].toSorted()) + }), + ) + it.live("runs bootstrap with InstanceRef provided", () => Effect.gen(function* () { const dir = yield* tmpdirScoped({ git: true }) diff --git a/packages/opencode/test/server/global-lifecycle.test.ts b/packages/opencode/test/server/global-lifecycle.test.ts new file mode 100644 index 000000000000..8a880e52c79f --- /dev/null +++ b/packages/opencode/test/server/global-lifecycle.test.ts @@ -0,0 +1,153 @@ +import { describe, expect } from "bun:test" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { Npm } from "@opencode-ai/core/npm" +import { httpClient } from "@opencode-ai/core/effect/app-node-platform" +import { Deferred, Effect, Fiber, Layer } from "effect" +import { HttpClient } from "effect/unstable/http" +import { GlobalBus, type GlobalEvent } from "@/bus/global" +import { Account } from "@/account/account" +import { Auth } from "@/auth" +import { Config } from "@/config/config" +import { Env } from "@/env" +import { InstanceRef } from "@/effect/instance-ref" +import { InstanceBootstrap } from "@/project/bootstrap" +import { InstanceStore } from "@/project/instance-store" +import { reloadWhenSessionsIdle } from "@/server/global-lifecycle" +import { SessionID } from "@/session/schema" +import { SessionStatus } from "@/session/status" +import { AccountTest } from "../fake/account" +import { AuthTest } from "../fake/auth" +import { NpmTest } from "../fake/npm" +import { tmpdirScoped } from "../fixture/fixture" +import { awaitWithTimeout, testEffect } from "../lib/effect" + +let bootstrapRun: Effect.Effect = Effect.void +const noopBootstrap = Layer.succeed( + InstanceBootstrap.Service, + InstanceBootstrap.Service.of({ run: Effect.suspend(() => bootstrapRun) }), +) + +const setBootstrap = (run: Effect.Effect) => + Effect.acquireRelease( + Effect.sync(() => { + bootstrapRun = run + }), + () => + Effect.sync(() => { + bootstrapRun = Effect.void + }), + ) +const unexpectedHttp = HttpClient.make((request) => + Effect.die(`unexpected http request: ${request.method} ${request.url}`), +) + +const it = testEffect( + LayerNode.compile( + LayerNode.group([ + InstanceStore.node, + SessionStatus.node, + Config.node, + FSUtil.node, + Env.node, + CrossSpawnSpawner.node, + ]), + [ + [InstanceStore.bootstrapNode, noopBootstrap], + [Auth.node, AuthTest.empty], + [Account.node, AccountTest.empty], + [Npm.node, NpmTest.noop], + [httpClient, Layer.succeed(HttpClient.HttpClient, unexpectedHttp)], + ], + ), +) + +const sessionID = SessionID.make("ses_global_lifecycle") + +const collectGlobalDisposed = () => + Effect.acquireRelease( + Effect.sync(() => { + const events: GlobalEvent[] = [] + const handler = (event: GlobalEvent) => { + if (event.payload?.type === "global.disposed") events.push(event) + } + GlobalBus.on("event", handler) + return { events, handler } + }), + ({ handler }) => Effect.sync(() => GlobalBus.off("event", handler)), + ).pipe(Effect.map(({ events }) => events)) + +describe("reloadWhenSessionsIdle", () => { + it.live("disposes instances and emits global.disposed when no session is busy", () => + Effect.gen(function* () { + const dir = yield* tmpdirScoped({ git: true }) + const store = yield* InstanceStore.Service + const before = yield* store.load({ directory: dir }) + const disposed = yield* collectGlobalDisposed() + + yield* awaitWithTimeout(reloadWhenSessionsIdle(), "reload blocked while no session was busy") + + expect(disposed).toHaveLength(1) + expect(yield* store.load({ directory: dir })).not.toBe(before) + }), + ) + + it.live("defers disposal until the busy session goes idle", () => + Effect.gen(function* () { + const dir = yield* tmpdirScoped({ git: true }) + const store = yield* InstanceStore.Service + const status = yield* SessionStatus.Service + const ctx = yield* store.load({ directory: dir }) + const disposed = yield* collectGlobalDisposed() + yield* status.set(sessionID, { type: "busy" }).pipe(Effect.provideService(InstanceRef, ctx)) + + const reload = yield* reloadWhenSessionsIdle().pipe(Effect.forkScoped({ startImmediately: true })) + yield* Effect.sleep("600 millis") + expect(disposed).toHaveLength(0) + expect(yield* store.load({ directory: dir })).toBe(ctx) + + yield* status.set(sessionID, { type: "idle" }).pipe(Effect.provideService(InstanceRef, ctx)) + yield* awaitWithTimeout(Fiber.join(reload), "reload did not run after the session went idle") + + expect(disposed).toHaveLength(1) + expect(yield* store.load({ directory: dir })).not.toBe(ctx) + }), + ) + + it.live("waits for a session on an instance that loaded while another was still booting", () => + Effect.gen(function* () { + const slow = yield* tmpdirScoped({ git: true }) + const fast = yield* tmpdirScoped({ git: true }) + const store = yield* InstanceStore.Service + const status = yield* SessionStatus.Service + const started = yield* Deferred.make() + const release = yield* Deferred.make() + yield* setBootstrap( + Effect.gen(function* () { + if ((yield* InstanceRef)?.directory !== slow) return + yield* Deferred.succeed(started, undefined) + yield* Deferred.await(release) + }), + ) + const disposed = yield* collectGlobalDisposed() + + yield* store.load({ directory: slow }).pipe(Effect.forkScoped({ startImmediately: true })) + yield* Deferred.await(started) + const reload = yield* reloadWhenSessionsIdle().pipe(Effect.forkScoped({ startImmediately: true })) + const ctx = yield* store.load({ directory: fast }) + yield* status.set(sessionID, { type: "busy" }).pipe(Effect.provideService(InstanceRef, ctx)) + yield* Deferred.succeed(release, undefined) + + yield* Effect.sleep("600 millis") + expect(disposed).toHaveLength(0) + expect(yield* store.load({ directory: fast })).toBe(ctx) + + yield* status.set(sessionID, { type: "idle" }).pipe(Effect.provideService(InstanceRef, ctx)) + yield* awaitWithTimeout(Fiber.join(reload), "reload did not run after the session went idle") + + expect(disposed).toHaveLength(1) + expect(yield* store.load({ directory: fast })).not.toBe(ctx) + }), + ) +})