diff --git a/src/constants/terminationTimeouts.ts b/src/constants/terminationTimeouts.ts index d8d35dc7b9..0f69546a8a 100644 --- a/src/constants/terminationTimeouts.ts +++ b/src/constants/terminationTimeouts.ts @@ -15,3 +15,10 @@ export const BACKUP_GIT_TIMEOUT_MS = 5 * 60 * 1000; * `desktop/main.ts` and `cli/server.ts` race the whole dispose against. */ export const APP_RUNTIME_DISPOSE_TIMEOUT_MS = 2 * 1000; + +/** + * Bounds the early `AppFiberScope` close in `ServiceContainer.dispose()` + * (interrupt + await of supervised fibers). Together with the runtime dispose + * bound this fits inside the same 5 s quit budgets. + */ +export const APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS = 2 * 1000; diff --git a/src/node/services/di/appFiberScope.test.ts b/src/node/services/di/appFiberScope.test.ts new file mode 100644 index 0000000000..94e852c642 --- /dev/null +++ b/src/node/services/di/appFiberScope.test.ts @@ -0,0 +1,128 @@ +import { describe, expect, it, spyOn } from "bun:test"; +import { Effect, Exit, Fiber, Layer, Scope } from "effect"; +import { log } from "@/node/services/log"; +import { AppFiberScopeLive, AppFiberScopeTag } from "./appFiberScope"; +import { closeScopeBounded, disposeAppRuntime, makeAppRuntime } from "./appRuntime"; +import { EffectRunnerLive, EffectRunnerTag } from "./effectRunner"; + +/** An interruptible I/O-style suspension that never resolves; records interruption. */ +function hungIo(onInterrupt: () => void): Effect.Effect { + return Effect.callback(() => + Effect.sync(() => { + onInterrupt(); + }) + ); +} + +function buildSeams() { + const app = makeAppRuntime(AppFiberScopeLive.pipe(Layer.provideMerge(EffectRunnerLive))); + return { + app, + appFiberScope: app.get(AppFiberScopeTag), + runner: app.get(EffectRunnerTag), + }; +} + +describe("AppFiberScope", () => { + it("closeScopeBounded interrupts and awaits an I/O-suspended fiber forked into it", async () => { + const { app, appFiberScope } = buildSeams(); + const steps: string[] = []; + const fiber = app.managed.runSync( + Effect.forkIn( + hungIo(() => steps.push("cancelled")).pipe( + Effect.ensuring(Effect.sync(() => steps.push("finalized"))) + ), + appFiberScope + ) + ); + + await closeScopeBounded(appFiberScope); + + // Interrupted (the cancel path ran) and awaited (its finalizer ran before + // the close resolved), so shutdown can rely on the fiber being gone. + expect(steps).toEqual(["cancelled", "finalized"]); + expect(fiber.pollUnsafe()).toBeDefined(); + expect(Exit.isFailure(fiber.pollUnsafe()!)).toBe(true); + await disposeAppRuntime(app.managed); + }); + + it("fibers forked through the EffectRunner are interrupted by neither close (unsupervised)", async () => { + const { app, appFiberScope, runner } = buildSeams(); + let interrupted = false; + const fiber = runner.runFork( + hungIo(() => { + interrupted = true; + }) + ); + + await closeScopeBounded(appFiberScope); + await disposeAppRuntime(app.managed); + + // The runner is not a supervisor: its fibers belong to whoever forked them + // (a worker's own scope with an explicit stop()), so dispose leaves them be. + expect(interrupted).toBe(false); + expect(fiber.pollUnsafe()).toBeUndefined(); + await Effect.runPromise(Fiber.interrupt(fiber)); + expect(interrupted).toBe(true); + }); + + it("disposeAppRuntime re-closes an already-closed AppFiberScope idempotently", async () => { + const { app, appFiberScope } = buildSeams(); + let finalized = 0; + app.managed.runSync( + Scope.addFinalizer( + appFiberScope, + Effect.sync(() => { + finalized += 1; + }) + ) + ); + + await closeScopeBounded(appFiberScope); + expect(finalized).toBe(1); + + // The runtime's layer scope owns the child; closing the runtime afterwards + // must neither throw nor run the child's finalizers a second time. + await disposeAppRuntime(app.managed); + expect(finalized).toBe(1); + expect(app.managed.cachedContext).toBeUndefined(); + }); + + it("runtime dispose alone closes the AppFiberScope (backstop for a missed explicit close)", async () => { + const { app, appFiberScope } = buildSeams(); + let interrupted = false; + app.managed.runSync( + Effect.forkIn( + hungIo(() => { + interrupted = true; + }), + appFiberScope + ) + ); + + await disposeAppRuntime(app.managed); + + expect(interrupted).toBe(true); + expect(appFiberScope.state._tag).toBe("Closed"); + }); + + it("closeScopeBounded returns at the timeout when a fiber cannot be interrupted, warning instead of rejecting", async () => { + const warnSpy = spyOn(log, "warn").mockImplementation(() => undefined); + try { + const { app, appFiberScope } = buildSeams(); + // Uninterruptible and never-resolving: the scope close can never finish. + app.managed.runSync(Effect.forkIn(Effect.uninterruptible(Effect.never), appFiberScope)); + + const startedAt = Date.now(); + await closeScopeBounded(appFiberScope, 50); + + expect(Date.now() - startedAt).toBeLessThan(2_000); + expect(warnSpy).toHaveBeenCalledTimes(1); + expect(String(warnSpy.mock.calls[0]?.[0])).toContain("timed out"); + // The runtime dispose then hits the same hung finalizer; keep it bounded too. + await disposeAppRuntime(app.managed, 50); + } finally { + warnSpy.mockRestore(); + } + }); +}); diff --git a/src/node/services/di/appFiberScope.ts b/src/node/services/di/appFiberScope.ts new file mode 100644 index 0000000000..2d3302cf51 --- /dev/null +++ b/src/node/services/di/appFiberScope.ts @@ -0,0 +1,36 @@ +/** + * `AppFiberScope` — the runtime-owned, *supervised* fiber scope (Effect + * migration Phase 11, PR 2). + * + * A child of the app runtime's layer scope. Fibers forked into it with + * `Effect.forkIn(effect, appFiberScope)` are interrupted **and awaited** when + * it closes, which `ServiceContainer.dispose()` does explicitly and early + * (right after `backgroundProcessManager.beginShutdown()`, before the + * hand-ordered teardown steps) via `closeScopeBounded`, so interrupted fibers + * can still use their dependencies while they finalize. `runtime.dispose()` + * later re-closes it idempotently as a backstop. + * + * This is the seam for I/O-suspended, long-lived work that shutdown must wait + * for (the streamManager engine core, in a later phase). It is the counterpart + * of `EffectRunner` (`./effectRunner.ts`), which is unsupervised: a fiber forked + * through the runner is interrupted by neither close. Anything forked here must + * tolerate interruption at any suspension point and must not depend on + * resources torn down before the close (see the dispose order in + * `ServiceContainer`). No production occupant yet; the contract is pinned by + * tests. + */ +import { Context, Effect, Layer, Scope } from "effect"; + +export class AppFiberScopeTag extends Context.Service()( + "xum/AppFiberScope" +) {} + +/** + * Synchronous body (DI contract: layers build without suspending). Parallel + * finalizer strategy: forked fibers are independent, so they are interrupted + * concurrently rather than one after another. + */ +export const AppFiberScopeLive: Layer.Layer = Layer.effect( + AppFiberScopeTag, + Effect.flatMap(Effect.scope, (parent) => Scope.fork(parent, "parallel")) +); diff --git a/src/node/services/di/appRuntime.ts b/src/node/services/di/appRuntime.ts index 77d15344a0..bba3552b19 100644 --- a/src/node/services/di/appRuntime.ts +++ b/src/node/services/di/appRuntime.ts @@ -29,11 +29,20 @@ * `disposeAppRuntime` (the last dispose step) reorders nothing. It is wired * now so that later phases (the streamManager engine core) have a fixed, * bounded slot for scope-owned resources. + * - **Two runtime seams, not one handle.** Workers hold an `EffectRunner` + * (`./effectRunner.ts`: context-bound, unsupervised, `R = never`) so they can + * run on the runtime's `Clock` without ever holding the runtime; work that + * shutdown must await forks into `AppFiberScope` (`./appFiberScope.ts`), + * the one supervised resource, closed explicitly early in `dispose()` via + * `closeScopeBounded` and re-closed idempotently by `disposeAppRuntime`. */ import assert from "@/common/utils/assert"; -import { Context, Duration, Effect, Fiber, ManagedRuntime } from "effect"; +import { Context, Duration, Effect, Exit, Fiber, ManagedRuntime, Scope } from "effect"; import type { Layer } from "effect"; -import { APP_RUNTIME_DISPOSE_TIMEOUT_MS } from "@/constants/terminationTimeouts"; +import { + APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS, + APP_RUNTIME_DISPOSE_TIMEOUT_MS, +} from "@/constants/terminationTimeouts"; import { log } from "@/node/services/log"; export interface AppRuntime { @@ -79,37 +88,63 @@ export function disposeAppRuntime( runtime: ManagedRuntime.ManagedRuntime, timeoutMs: number = APP_RUNTIME_DISPOSE_TIMEOUT_MS ): Promise { - return Effect.runPromise(disposeAppRuntimeEffect(runtime, timeoutMs)); + return Effect.runPromise( + boundedTeardown("AppRuntime", "disposed", runtime.disposeEffect, timeoutMs) + ); } -function disposeAppRuntimeEffect( - runtime: ManagedRuntime.ManagedRuntime, +/** + * Close a runtime-owned scope (`AppFiberScope`), interrupting and awaiting the + * fibers forked into it, bounded by `timeoutMs`. Same contract as + * `disposeAppRuntime`: never rejects, idempotent (`Scope.close` on a closed + * scope is a no-op), warns and returns at the bound if a fiber's finalization + * hangs. + */ +export function closeScopeBounded( + scope: Scope.Closeable, + timeoutMs: number = APP_FIBER_SCOPE_CLOSE_TIMEOUT_MS +): Promise { + return Effect.runPromise( + boundedTeardown("AppFiberScope", "closed", Scope.close(scope, Exit.void), timeoutMs) + ); +} + +/** + * Uninterruptible teardown shell around a bounded wait on `target`. The target + * runs in a detached fiber and the caller waits on its join, so a hung + * finalizer cannot pin shutdown past the bound — only the wait is + * interruptible (so the timeout can win the race; house shape from #4038), + * the target keeps running best-effort, and a warning is logged. Defects are + * folded into a warning so the returned effect never fails. + */ +function boundedTeardown( + subject: string, + doneVerb: string, + target: Effect.Effect, timeoutMs: number ): Effect.Effect { const startedAt = performance.now(); - // Uninterruptible teardown shell; only the bounded wait inside is interruptible - // so the timeout can win the race (house shape from #4038). return Effect.uninterruptible( Effect.gen(function* () { - const closing = yield* Effect.forkDetach(runtime.disposeEffect); + const closing = yield* Effect.forkDetach(target); yield* Effect.interruptible( Fiber.join(closing).pipe(Effect.timeout(Duration.millis(timeoutMs))) ).pipe( Effect.catchTag("TimeoutError", () => Effect.sync(() => { - log.warn("[shutdown] AppRuntime dispose timed out; finalizers continue best-effort", { + log.warn(`[shutdown] ${subject} teardown timed out; finalizers continue best-effort`, { timeoutMs, }); }) ) ); - log.debug("[shutdown] AppRuntime disposed", { + log.debug(`[shutdown] ${subject} ${doneVerb}`, { ms: Math.round(performance.now() - startedAt), }); }).pipe( Effect.catchDefect((defect) => Effect.sync(() => { - log.warn("[shutdown] AppRuntime dispose failed", { error: defect }); + log.warn(`[shutdown] ${subject} teardown failed`, { error: defect }); }) ) ) diff --git a/src/node/services/di/effectRunner.test.ts b/src/node/services/di/effectRunner.test.ts new file mode 100644 index 0000000000..4dae693182 --- /dev/null +++ b/src/node/services/di/effectRunner.test.ts @@ -0,0 +1,159 @@ +import { describe, expect, it } from "bun:test"; +import { Clock, Context, Duration, Effect, Exit, Fiber, Schedule, Scope } from "effect"; +import { makeAppRuntime } from "./appRuntime"; +import { defaultEffectRunner, EffectRunnerLive, EffectRunnerTag } from "./effectRunner"; +import { makeTestEffectRunner } from "./testEffectRunner"; + +/** + * Pins the runtime facts the clock-driven workers rely on when they run + * through an `EffectRunner` (Phase 11 PR 2 acceptance), so a later effect + * upgrade that changes them fails here rather than in a worker suite. + */ +describe("EffectRunner", () => { + it("defaultEffectRunner forks synchronously up to the first sleep", () => { + const steps: string[] = []; + + const fiber = defaultEffectRunner.runFork( + Effect.gen(function* () { + steps.push("before-sleep"); + yield* Effect.sleep(Duration.hours(1)); + steps.push("after-sleep"); + }) + ); + + expect(steps).toEqual(["before-sleep"]); + defaultEffectRunner.runFork(Fiber.interrupt(fiber)); + }); + + it("EffectRunnerLive captures the upstream Clock (a TestClock when provided)", async () => { + const clock = makeTestEffectRunner(); + try { + expect(clock.runner.runSync(Clock.currentTimeMillis)).toBe(0); + await clock.adjust(Duration.millis(1_500)); + expect(clock.runner.runSync(Clock.currentTimeMillis)).toBe(1_500); + } finally { + await clock.dispose(); + } + }); + + it("schedules forked fibers on the default scheduler, not the eager build's sync scheduler", async () => { + // makeAppRuntime builds with runSync, whose fiber carries a microtask + // scheduler. EffectRunnerLive strips it so runner.runFork behaves like + // Effect.runFork: a yield resumes on a macrotask. + const app = makeAppRuntime(EffectRunnerLive); + const runner = app.get(EffectRunnerTag); + let resumed = false; + + runner.runFork( + Effect.yieldNow.pipe( + Effect.andThen( + Effect.sync(() => { + resumed = true; + }) + ) + ) + ); + + expect(resumed).toBe(false); + await Promise.resolve(); + expect(resumed).toBe(false); + await new Promise((resolve) => setImmediate(resolve)); + expect(resumed).toBe(true); + }); + + it("rejects effects with service requirements at the type level", () => { + class Probe extends Context.Service()("test/Probe") {} + const needsService: Effect.Effect = Effect.void; + // @ts-expect-error -- EffectRunner is not a service locator: R must be never. + defaultEffectRunner.runSync(needsService); + }); +}); + +describe("makeTestEffectRunner", () => { + it("runFork reaches its first sleep synchronously and adjust resumes it with its continuation", async () => { + const clock = makeTestEffectRunner(); + try { + const steps: string[] = []; + clock.runner.runFork( + Effect.gen(function* () { + steps.push("before-sleep"); + yield* Effect.sleep(Duration.minutes(1)); + steps.push("after-sleep"); + }) + ); + expect(steps).toEqual(["before-sleep"]); + + await clock.adjust(Duration.seconds(59)); + expect(steps).toEqual(["before-sleep"]); + + // The due sleep and its synchronous continuation run before adjust resolves. + await clock.adjust(Duration.seconds(1)); + expect(steps).toEqual(["before-sleep", "after-sleep"]); + } finally { + await clock.dispose(); + } + }); + + it("drives a Schedule.fixed cadence one tick per interval, including nested due sleeps", async () => { + const clock = makeTestEffectRunner(); + try { + let ticks = 0; + const scope = Scope.makeUnsafe(); + clock.runner.runSync( + Effect.forkIn( + Effect.gen(function* () { + yield* Effect.sleep(Duration.seconds(60)); + yield* Effect.sync(() => { + ticks += 1; + }).pipe(Effect.repeat(Schedule.fixed(Duration.seconds(30)))); + }), + scope + ) + ); + + await clock.adjust(Duration.seconds(60)); + expect(ticks).toBe(1); + await clock.adjust(Duration.seconds(30)); + expect(ticks).toBe(2); + // One adjust spanning two intervals fires both slots (the second sleep is + // registered while the first resumes, still inside the adjust window). + await clock.adjust(Duration.seconds(60)); + expect(ticks).toBe(4); + + clock.runner.runSync(Scope.close(scope, Exit.void)); + await clock.adjust(Duration.seconds(90)); + expect(ticks).toBe(4); + } finally { + await clock.dispose(); + } + }); + + it("closes a scope owning a TestClock-suspended fiber synchronously", async () => { + const clock = makeTestEffectRunner(); + try { + const scope = Scope.makeUnsafe(); + let interrupted = false; + const fiber = clock.runner.runSync( + Effect.forkIn( + Effect.sleep(Duration.hours(1)).pipe( + Effect.onInterrupt(() => + Effect.sync(() => { + interrupted = true; + }) + ) + ), + scope + ) + ); + + // The worker stop() contract: the close completes synchronously because + // the fiber only suspends on its clock, TestClock included. + clock.runner.runSync(Scope.close(scope, Exit.void)); + + expect(interrupted).toBe(true); + expect(fiber.pollUnsafe()).toBeDefined(); + } finally { + await clock.dispose(); + } + }); +}); diff --git a/src/node/services/di/effectRunner.ts b/src/node/services/di/effectRunner.ts new file mode 100644 index 0000000000..28285f55f5 --- /dev/null +++ b/src/node/services/di/effectRunner.ts @@ -0,0 +1,82 @@ +/** + * `EffectRunner` — the context-bound, *unsupervised* runner seam (Effect + * migration Phase 11, PR 2). + * + * Clock-driven workers (heartbeat, idle compaction, retry backoff) run their + * lifecycle fibers through an `EffectRunner` instead of the global + * `Effect.runX`. The runner is a thin `Effect.run…With(context)` bundle, so: + * + * - **Same start semantics as `Effect.runX`.** `runSync` completes + * synchronously (a fiber that suspends is a defect, exactly like + * `Effect.runSync`), and `runFork` executes the fiber up to its first async + * boundary before returning — the ordering the workers' `start()`/`stop()` + * contracts rely on. + * - **Unsupervised.** Fibers forked through it belong to the worker's own + * `Scope` (explicit `start()`/`stop()`), not to the app runtime: + * `runtime.dispose()` neither interrupts nor awaits them, and a late worker + * callback after dispose cannot hit "ManagedRuntime disposed". Work that + * must be awaited on shutdown forks into `AppFiberScope` instead + * (`./appFiberScope.ts`). + * - **Not a service locator.** Every method accepts only `Effect`: + * defaulted references such as `Clock` do not appear in `R`, so a + * context-bound runner lets a worker run on a `TestClock`, while anything + * that needs a service must take it as an explicit constructor dependency. + * + * `defaultEffectRunner` is the global runtime (today's exact behavior) and is + * the default for every constructor parameter that accepts a runner, so + * direct construction in tests and CLI roots is unchanged. + */ +import { Context, Effect, Layer, Scheduler, Scope } from "effect"; +import type { Exit, Fiber } from "effect"; + +export interface EffectRunner { + runSync(effect: Effect.Effect): A; + runSyncExit(effect: Effect.Effect): Exit.Exit; + runFork(effect: Effect.Effect): Fiber.Fiber; + runPromise(effect: Effect.Effect): Promise; + runPromiseExit(effect: Effect.Effect): Promise>; +} + +/** The global Effect runtime — what every worker used before the seam existed. */ +export const defaultEffectRunner: EffectRunner = { + runSync: Effect.runSync, + runSyncExit: Effect.runSyncExit, + runFork: Effect.runFork, + runPromise: Effect.runPromise, + runPromiseExit: Effect.runPromiseExit, +}; + +/** A runner whose fibers start with `context` (its `Clock` and other refs). */ +export function effectRunnerFromContext(context: Context.Context): EffectRunner { + return { + runSync: Effect.runSyncWith(context), + runSyncExit: Effect.runSyncExitWith(context), + runFork: Effect.runForkWith(context), + runPromise: Effect.runPromiseWith(context), + runPromiseExit: Effect.runPromiseExitWith(context), + }; +} + +export class EffectRunnerTag extends Context.Service()( + "xum/EffectRunner" +) {} + +/** + * Captures the building fiber's context, so the runner carries whatever the + * layers beneath it provide (stores, and in tests a `TestClock` as the `Clock` + * reference). Place it at the base of the graph. + * + * Build-fiber artifacts are stripped because they are not services: the layer + * `Scope` (unsupervised fibers must not hold the runtime's scope), the layer + * memo map, and the eager `runSync` build's microtask scheduler — without the + * omit, every fiber forked through the runner would be scheduled on that + * scheduler instead of the default one `Effect.runFork` uses. + */ +export const EffectRunnerLive: Layer.Layer = Layer.effect( + EffectRunnerTag, + Effect.map(Effect.context(), (context) => + effectRunnerFromContext( + Context.omit(Scope.Scope, Layer.CurrentMemoMap, Scheduler.Scheduler)(context) + ) + ) +); diff --git a/src/node/services/di/layers/app.ts b/src/node/services/di/layers/app.ts index 0e24acae66..853b515fab 100644 --- a/src/node/services/di/layers/app.ts +++ b/src/node/services/di/layers/app.ts @@ -1,5 +1,7 @@ import { Layer } from "effect"; import type { ConfigStores } from "@/node/config"; +import { AppFiberScopeLive } from "@/node/services/di/appFiberScope"; +import { EffectRunnerLive } from "@/node/services/di/effectRunner"; import type { AppTags } from "@/node/services/di/tags"; import { MemoryMetaLive } from "./core"; import { StoresLive } from "./stores"; @@ -12,7 +14,14 @@ import { StoresLive } from "./stores"; * right-hand operand satisfies the left-hand operand's requirements and both * stay exposed in the final context. `Layer.mergeAll` is only for true * siblings; it does not satisfy one sibling's requirements from another. + * + * The runtime seams sit at the base, above the stores: `EffectRunnerLive` + * captures its building context, so placing it there keeps that context to the + * stores plus references (`Clock`, …). */ export function AppLive(stores: ConfigStores): Layer.Layer { - return MemoryMetaLive.pipe(Layer.provideMerge(StoresLive(stores))); + const runtimeSeams = AppFiberScopeLive.pipe( + Layer.provideMerge(EffectRunnerLive.pipe(Layer.provideMerge(StoresLive(stores)))) + ); + return MemoryMetaLive.pipe(Layer.provideMerge(runtimeSeams)); } diff --git a/src/node/services/di/tags.ts b/src/node/services/di/tags.ts index be2aead7b6..b92b8b2a25 100644 --- a/src/node/services/di/tags.ts +++ b/src/node/services/di/tags.ts @@ -20,6 +20,8 @@ import type { WorkspaceSessionLocator, } from "@/node/config"; import type { MemoryMetaService } from "@/node/services/memoryMeta"; +import type { AppFiberScopeTag } from "./appFiberScope"; +import type { EffectRunnerTag } from "./effectRunner"; export class ConfigTag extends Context.Service()("xum/Config") {} export class SessionLocatorTag extends Context.Service< @@ -50,5 +52,11 @@ export type StoreTags = | SecretsStoreTag | FileLeaseManagerTag; +/** + * The runtime seams provided at the base of every graph (`./effectRunner.ts`, + * `./appFiberScope.ts`); their tags live next to their layers. + */ +export type RuntimeSeamTags = EffectRunnerTag | AppFiberScopeTag; + /** Every service the desktop/server app graph (`AppLive`) provides. */ -export type AppTags = StoreTags | MemoryMeta; +export type AppTags = StoreTags | RuntimeSeamTags | MemoryMeta; diff --git a/src/node/services/di/testEffectRunner.ts b/src/node/services/di/testEffectRunner.ts new file mode 100644 index 0000000000..f124719ef9 --- /dev/null +++ b/src/node/services/di/testEffectRunner.ts @@ -0,0 +1,37 @@ +/** + * Test helper: an `EffectRunner` bound to a `TestClock`, so a worker + * constructed with `runner` sleeps on virtual time that the test advances with + * `adjust`/`setTime` (no real timers, no polling). + * + * Built exactly like production (`EffectRunnerLive` at the base of a + * `makeAppRuntime` graph) with `TestClock.layer()` as the provider, so the + * worker under test and the test's `adjust` share one clock. `adjust` resolves + * after every due sleep has been resumed and its synchronous continuation has + * run (pinned in `testEffectRunner.test.ts`). + */ +import { Layer } from "effect"; +import type { Duration } from "effect"; +import { TestClock } from "effect/testing"; +import { disposeAppRuntime, makeAppRuntime } from "./appRuntime"; +import { EffectRunnerLive, EffectRunnerTag, type EffectRunner } from "./effectRunner"; + +export interface TestEffectRunner { + readonly runner: EffectRunner; + /** Advance the test clock, running every sleep due on or before the new time. */ + adjust(duration: Duration.Input): Promise; + /** Set the test clock to an absolute time (ms since epoch); same firing rule. */ + setTime(timestampMs: number): Promise; + dispose(): Promise; +} + +export function makeTestEffectRunner(options?: TestClock.TestClock.Options): TestEffectRunner { + const app = makeAppRuntime(EffectRunnerLive.pipe(Layer.provideMerge(TestClock.layer(options)))); + const runner = app.get(EffectRunnerTag); + return { + runner, + // Run through the runner so `TestClock.adjust` reads the captured clock. + adjust: (duration) => runner.runPromise(TestClock.adjust(duration)), + setTime: (timestampMs) => runner.runPromise(TestClock.setTime(timestampMs)), + dispose: () => disposeAppRuntime(app.managed), + }; +} diff --git a/src/node/services/heartbeatService.testClock.test.ts b/src/node/services/heartbeatService.testClock.test.ts new file mode 100644 index 0000000000..14c670d339 --- /dev/null +++ b/src/node/services/heartbeatService.testClock.test.ts @@ -0,0 +1,89 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { EventEmitter } from "events"; +import { Duration } from "effect"; +import type { ProjectsConfig } from "@/common/types/project"; +import type { Config } from "@/node/config"; +import { makeTestEffectRunner, type TestEffectRunner } from "./di/testEffectRunner"; +import type { ExtensionMetadataService } from "./ExtensionMetadataService"; +import { CHECK_INTERVAL_MS, HeartbeatService, STARTUP_DELAY_MS } from "./heartbeatService"; +import type { TaskService } from "./taskService"; +import type { WorkspaceService } from "./workspaceService"; + +/** + * Cadence on virtual time. The real-timer suite (`heartbeatService.test.ts`) + * keeps exercising the default runner; this one drives the scheduler fiber + * through a TestClock-bound `EffectRunner`. + */ +describe("HeartbeatService on a TestClock", () => { + let clock: TestEffectRunner; + let service: HeartbeatService; + // Every tick starts with a synchronous getAllSnapshots() call, so its call + // count is the tick count. + let getAllSnapshotsMock: ReturnType Promise>>>; + + beforeEach(() => { + clock = makeTestEffectRunner(); + getAllSnapshotsMock = mock(() => Promise.resolve(new Map())); + const config = { + loadConfigOrDefault: mock((): ProjectsConfig => ({ projects: new Map() })), + } as unknown as Config; + const extensionMetadata = { + getAllSnapshots: getAllSnapshotsMock, + getSnapshot: mock(() => Promise.resolve(null)), + } as unknown as ExtensionMetadataService; + const workspaceService = Object.assign(new EventEmitter(), { + getChatHistory: mock(() => Promise.resolve([])), + executeHeartbeat: mock(() => Promise.resolve()), + isBusyForMessage: mock(() => false), + }) as unknown as WorkspaceService; + const taskService = { + hasActiveDescendantAgentTasksForWorkspace: mock(() => false), + } as unknown as TaskService; + + service = new HeartbeatService( + config, + extensionMetadata, + workspaceService, + taskService, + undefined, + clock.runner + ); + }); + + afterEach(async () => { + service.stop(); + await clock.dispose(); + }); + + test("first tick after the startup delay, then one per check interval, none after stop()", async () => { + service.start(); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(0); + + await clock.adjust(Duration.millis(STARTUP_DELAY_MS - 1)); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(0); + + await clock.adjust(Duration.millis(1)); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(1); + + await clock.adjust(Duration.millis(CHECK_INTERVAL_MS - 1)); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(1); + await clock.adjust(Duration.millis(1)); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(2); + await clock.adjust(Duration.millis(CHECK_INTERVAL_MS)); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(3); + + service.stop(); + await clock.adjust(Duration.millis(CHECK_INTERVAL_MS * 3)); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(3); + }); + + test("stop() during the startup delay cancels the first tick", async () => { + service.start(); + await clock.adjust(Duration.millis(STARTUP_DELAY_MS / 2)); + + service.stop(); + + await clock.adjust(Duration.millis(STARTUP_DELAY_MS + CHECK_INTERVAL_MS)); + expect(getAllSnapshotsMock).toHaveBeenCalledTimes(0); + }); +}); diff --git a/src/node/services/heartbeatService.ts b/src/node/services/heartbeatService.ts index 9172b47a33..227501a271 100644 --- a/src/node/services/heartbeatService.ts +++ b/src/node/services/heartbeatService.ts @@ -13,6 +13,7 @@ import { type HeartbeatTrigger, } from "@/constants/heartbeat"; import type { Config } from "@/node/config"; +import { defaultEffectRunner, type EffectRunner } from "./di/effectRunner"; import type { ExtensionMetadataService } from "./ExtensionMetadataService"; import { IdleDispatcher, type IdleDispatchPayload } from "./idleDispatcher"; import { log } from "./log"; @@ -20,8 +21,8 @@ import type { TaskService } from "./taskService"; import { NOOP_TIMELINE_RECORDER, type TimelineRecorder } from "./timelineRecorder"; import type { WorkspaceService } from "./workspaceService"; -const STARTUP_DELAY_MS = 60 * 1000; // 60s - let startup settle -const CHECK_INTERVAL_MS = 30 * 1000; // 30s tick +export const STARTUP_DELAY_MS = 60 * 1000; // 60s - let startup settle +export const CHECK_INTERVAL_MS = 30 * 1000; // 30s tick const MAX_CONCURRENT_HEARTBEATS = 1; const HEARTBEAT_IDLE_CONSUMER_NAME = "heartbeat"; const HEARTBEAT_IDLE_CONSUMER_PRIORITY = 50; @@ -59,6 +60,12 @@ export class HeartbeatService { private readonly workspaceService: WorkspaceService; private readonly taskService: TaskService; private readonly idleDispatcher: IdleDispatcher; + /** + * Runs the lifecycle effects below (scope acquisition, scheduler fork, scope + * close). Context-bound in the app (so the scheduler fiber reads the + * runtime's `Clock` — a `TestClock` in tests); the global runtime by default. + */ + private readonly runner: EffectRunner; private timelineRecorder: TimelineRecorder = NOOP_TIMELINE_RECORDER; @@ -104,13 +111,15 @@ export class HeartbeatService { extensionMetadata: ExtensionMetadataService, workspaceService: WorkspaceService, taskService: TaskService, - idleDispatcher?: IdleDispatcher + idleDispatcher?: IdleDispatcher, + runner: EffectRunner = defaultEffectRunner ) { this.config = config; this.extensionMetadata = extensionMetadata; this.workspaceService = workspaceService; this.taskService = taskService; this.idleDispatcher = idleDispatcher ?? new IdleDispatcher(); + this.runner = runner; this.onActivity = (event) => this.handleActivityEvent(event); this.onMetadata = (event) => this.handleMetadataEvent(event); @@ -200,7 +209,7 @@ export class HeartbeatService { // the scheduler up to its first sleep before returning, so the startup // timer is registered before start() returns (same observable ordering // as the previous setTimeout call). - Effect.runSync(Scope.provide(scope)(acquireResources)); + this.runner.runSync(Scope.provide(scope)(acquireResources)); } catch (error) { // Guaranteed cleanup on partial startup failure: close the scope so the // finalizers registered before the failing step run (the hand-rolled @@ -210,7 +219,7 @@ export class HeartbeatService { this.startupTimeout = null; this.checkInterval = null; this.stopped = true; - Effect.runSync(Scope.close(scope, Exit.void)); + this.runner.runSync(Scope.close(scope, Exit.void)); throw error; } @@ -239,7 +248,7 @@ export class HeartbeatService { // scheduler fiber interrupt (synchronously clearing its pending timer), // listeners off, consumer disposer — completing synchronously because // the fiber only ever suspends on its clock timer. - Effect.runSync(Scope.close(scope, Exit.void)); + this.runner.runSync(Scope.close(scope, Exit.void)); } this.startupTimeout = null; this.checkInterval = null; diff --git a/src/node/services/idleCompactionService.testClock.test.ts b/src/node/services/idleCompactionService.testClock.test.ts new file mode 100644 index 0000000000..0dcb439e8d --- /dev/null +++ b/src/node/services/idleCompactionService.testClock.test.ts @@ -0,0 +1,71 @@ +import { afterEach, beforeEach, describe, expect, mock, test } from "bun:test"; +import { Duration } from "effect"; +import type { ProjectsConfig } from "@/common/types/project"; +import type { Config } from "@/node/config"; +import { makeTestEffectRunner, type TestEffectRunner } from "./di/testEffectRunner"; +import type { ExtensionMetadataService } from "./ExtensionMetadataService"; +import type { HistoryService } from "./historyService"; +import { + CHECK_INTERVAL_MS, + IdleCompactionService, + INITIAL_CHECK_DELAY_MS, +} from "./idleCompactionService"; + +/** + * Checker cadence on virtual time (the real-timer suite in + * `idleCompactionService.test.ts` keeps covering the default runner). + */ +describe("IdleCompactionService on a TestClock", () => { + let clock: TestEffectRunner; + let service: IdleCompactionService; + // checkAllWorkspaces() reads the config synchronously first, so with no + // projects configured its call count is the check count. + let loadConfigMock: ReturnType ProjectsConfig>>; + + beforeEach(() => { + clock = makeTestEffectRunner(); + loadConfigMock = mock((): ProjectsConfig => ({ projects: new Map() })); + service = new IdleCompactionService( + { loadConfigOrDefault: loadConfigMock } as unknown as Config, + {} as HistoryService, + {} as ExtensionMetadataService, + () => Promise.resolve(), + clock.runner + ); + }); + + afterEach(async () => { + service.stop(); + await clock.dispose(); + }); + + test("first check after the initial delay, then one per interval, none after stop()", async () => { + service.start(); + expect(loadConfigMock).toHaveBeenCalledTimes(0); + + await clock.adjust(Duration.millis(INITIAL_CHECK_DELAY_MS - 1)); + expect(loadConfigMock).toHaveBeenCalledTimes(0); + await clock.adjust(Duration.millis(1)); + expect(loadConfigMock).toHaveBeenCalledTimes(1); + + await clock.adjust(Duration.millis(CHECK_INTERVAL_MS)); + expect(loadConfigMock).toHaveBeenCalledTimes(2); + // Sweeps are fire-and-forget, so an adjust spanning two slots fires both. + await clock.adjust(Duration.millis(CHECK_INTERVAL_MS * 2)); + expect(loadConfigMock).toHaveBeenCalledTimes(4); + + service.stop(); + await clock.adjust(Duration.millis(CHECK_INTERVAL_MS * 3)); + expect(loadConfigMock).toHaveBeenCalledTimes(4); + }); + + test("stop() during the initial delay cancels the first check", async () => { + service.start(); + await clock.adjust(Duration.millis(INITIAL_CHECK_DELAY_MS / 2)); + + service.stop(); + + await clock.adjust(Duration.millis(INITIAL_CHECK_DELAY_MS + CHECK_INTERVAL_MS)); + expect(loadConfigMock).toHaveBeenCalledTimes(0); + }); +}); diff --git a/src/node/services/idleCompactionService.ts b/src/node/services/idleCompactionService.ts index 75abaf5d34..9fd91007e7 100644 --- a/src/node/services/idleCompactionService.ts +++ b/src/node/services/idleCompactionService.ts @@ -1,13 +1,14 @@ import { Duration, Effect, Exit, Schedule, Scope } from "effect"; import assert from "@/common/utils/assert"; import type { Config } from "@/node/config"; +import { defaultEffectRunner, type EffectRunner } from "./di/effectRunner"; import type { HistoryService } from "./historyService"; import type { ExtensionMetadataService } from "./ExtensionMetadataService"; import { computeRecencyFromMessages } from "@/common/utils/recency"; import { log } from "./log"; -const INITIAL_CHECK_DELAY_MS = 60 * 1000; // 1 minute - let startup initialization settle -const CHECK_INTERVAL_MS = 60 * 60 * 1000; // 1 hour +export const INITIAL_CHECK_DELAY_MS = 60 * 1000; // 1 minute - let startup initialization settle +export const CHECK_INTERVAL_MS = 60 * 60 * 1000; // 1 hour const HOURS_TO_MS = 60 * 60 * 1000; /** @@ -45,6 +46,12 @@ export class IdleCompactionService { private readonly historyService: HistoryService; private readonly extensionMetadata: ExtensionMetadataService; private readonly executeIdleCompaction: (workspaceId: string) => Promise; + /** + * Runs the checker fork and the scope close. Context-bound in the app (the + * checker reads the runtime's `Clock` — a `TestClock` in tests); the global + * runtime by default. + */ + private readonly runner: EffectRunner; /** * Owns the checker fiber forked by start(): sleep(INITIAL_CHECK_DELAY_MS), * then check immediately and every CHECK_INTERVAL_MS. Closing the scope in @@ -66,12 +73,14 @@ export class IdleCompactionService { config: Config, historyService: HistoryService, extensionMetadata: ExtensionMetadataService, - executeIdleCompaction: (workspaceId: string) => Promise + executeIdleCompaction: (workspaceId: string) => Promise, + runner: EffectRunner = defaultEffectRunner ) { this.config = config; this.historyService = historyService; this.extensionMetadata = extensionMetadata; this.executeIdleCompaction = executeIdleCompaction; + this.runner = runner; } /** @@ -100,7 +109,7 @@ export class IdleCompactionService { // Runs synchronously up to the checker's first sleep, so the initial-delay // timer is registered before start() returns (same as the previous // setTimeout call). - Effect.runSync(Effect.forkIn(checker, scope)); + this.runner.runSync(Effect.forkIn(checker, scope)); log.info("IdleCompactionService started", { initialDelayMs: INITIAL_CHECK_DELAY_MS, @@ -119,7 +128,7 @@ export class IdleCompactionService { this.lifecycleScope = null; // Interrupts the checker fiber, synchronously clearing its pending // timer — the fiber only ever suspends on its clock timer. - Effect.runSync(Scope.close(scope, Exit.void)); + this.runner.runSync(Scope.close(scope, Exit.void)); } // Best-effort queue reset: do not start new compactions after stop(). diff --git a/src/node/services/retryManager.testClock.test.ts b/src/node/services/retryManager.testClock.test.ts new file mode 100644 index 0000000000..c8812fd92a --- /dev/null +++ b/src/node/services/retryManager.testClock.test.ts @@ -0,0 +1,79 @@ +import { afterEach, beforeEach, describe, expect, it, mock } from "bun:test"; +import { Duration } from "effect"; +import { calculateBackoffDelay } from "@/common/utils/messages/retryState"; +import { makeTestEffectRunner, type TestEffectRunner } from "./di/testEffectRunner"; +import { RetryManager, type RetryStatusEvent } from "./retryManager"; + +/** + * Backoff timing on virtual time (the real-timer suite in + * `retryManager.test.ts` keeps covering the default runner, which today's + * streamManager call site still uses). + */ +describe("RetryManager on a TestClock", () => { + let clock: TestEffectRunner; + let manager: RetryManager; + let onRetry: ReturnType Promise>>; + let events: RetryStatusEvent[]; + + beforeEach(() => { + clock = makeTestEffectRunner(); + onRetry = mock(() => Promise.resolve()); + events = []; + manager = new RetryManager( + "workspace-1", + onRetry, + (event) => { + events.push(event); + }, + clock.runner + ); + }); + + afterEach(async () => { + manager.dispose(); + await clock.dispose(); + }); + + it("fires exactly at the backoff delay", async () => { + manager.handleStreamFailure({ type: "unknown", message: "transient" }); + const delayMs = calculateBackoffDelay(1); + expect(events.map((event) => event.type)).toEqual(["auto-retry-scheduled"]); + expect(manager.isRetryPending).toBe(true); + + await clock.adjust(Duration.millis(delayMs - 1)); + expect(onRetry).not.toHaveBeenCalled(); + expect(manager.isRetryPending).toBe(true); + + await clock.adjust(Duration.millis(1)); + expect(onRetry).toHaveBeenCalledTimes(1); + expect(manager.isRetryPending).toBe(false); + expect(events.map((event) => event.type)).toEqual([ + "auto-retry-scheduled", + "auto-retry-starting", + ]); + }); + + it("cancel() before the delay elapses means the retry never fires", async () => { + manager.handleStreamFailure({ type: "unknown", message: "transient" }); + manager.cancel(); + expect(manager.isRetryPending).toBe(false); + + await clock.adjust(Duration.millis(calculateBackoffDelay(1) * 10)); + + expect(onRetry).not.toHaveBeenCalled(); + expect(events.map((event) => event.type)).toEqual(["auto-retry-scheduled"]); + }); + + it("a second failure before the delay reschedules with the next backoff", async () => { + manager.handleStreamFailure({ type: "unknown" }); + await clock.adjust(Duration.millis(calculateBackoffDelay(1) - 1)); + manager.handleStreamFailure({ type: "unknown" }); + const secondDelayMs = calculateBackoffDelay(2); + + // The superseded timer is gone: only the new one can fire. + await clock.adjust(Duration.millis(secondDelayMs - 1)); + expect(onRetry).not.toHaveBeenCalled(); + await clock.adjust(Duration.millis(1)); + expect(onRetry).toHaveBeenCalledTimes(1); + }); +}); diff --git a/src/node/services/retryManager.ts b/src/node/services/retryManager.ts index 0363affcaa..f736e217c2 100644 --- a/src/node/services/retryManager.ts +++ b/src/node/services/retryManager.ts @@ -10,6 +10,7 @@ import { isNonRetryableSendError, isNonRetryableStreamError, } from "@/common/utils/messages/retryEligibility"; +import { defaultEffectRunner, type EffectRunner } from "./di/effectRunner"; export interface RetryFailureError { type: string; @@ -63,7 +64,14 @@ export class RetryManager { constructor( private readonly workspaceId: string, private readonly onRetry: () => Promise, - private readonly onStatusChange: (event: RetryStatusEvent) => void + private readonly onStatusChange: (event: RetryStatusEvent) => void, + /** + * Runs the retry fiber fork and its interrupt. The global runtime by + * default (the streamManager call site keeps it until the runtime seam + * reaches StreamManager); a context-bound runner puts the backoff sleep on + * the runtime's `Clock` — a `TestClock` in tests. + */ + private readonly runner: EffectRunner = defaultEffectRunner ) { assert(this.workspaceId.trim().length > 0, "RetryManager: workspaceId must be non-empty"); assert(typeof this.onRetry === "function", "RetryManager: onRetry must be a function"); @@ -118,7 +126,7 @@ export class RetryManager { } /** - * Fork the retry fiber. `Effect.runFork` executes synchronously up to the + * Fork the retry fiber. `runner.runFork` executes synchronously up to the * sleep, so the backoff timer is registered before this method returns — * the same observable ordering as the previous `setTimeout` call. * @@ -131,7 +139,7 @@ export class RetryManager { // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this` const self = this; this.retryPending = true; - this.retryFiber = Effect.runFork( + this.retryFiber = this.runner.runFork( Effect.gen(function* () { yield* Effect.sleep(Duration.millis(delayMs)); self.retryPending = false; @@ -195,7 +203,7 @@ export class RetryManager { this.retryPending = false; // Fire-and-forget: the interrupt signal lands synchronously; awaiting // full fiber exit is unnecessary (and impossible from sync callers). - Effect.runFork(Fiber.interrupt(fiber)); + this.runner.runFork(Fiber.interrupt(fiber)); } } diff --git a/src/node/services/serviceContainer.test.ts b/src/node/services/serviceContainer.test.ts index 3b7a3f7fc5..a7fe39f249 100644 --- a/src/node/services/serviceContainer.test.ts +++ b/src/node/services/serviceContainer.test.ts @@ -2,9 +2,12 @@ import * as path from "path"; import * as fs from "fs"; import * as os from "os"; import { afterEach, beforeEach, describe, expect, it, spyOn } from "bun:test"; -import { Context, Layer } from "effect"; +import { Context, Duration, Effect, Layer } from "effect"; +import { TestClock } from "effect/testing"; import { MULTI_PROJECT_CONFIG_KEY } from "@/common/constants/multiProject"; import { createConfigStores, type Config, type ConfigStores } from "@/node/config"; +import { AppFiberScopeTag } from "@/node/services/di/appFiberScope"; +import { EffectRunnerTag } from "@/node/services/di/effectRunner"; import * as appLayers from "@/node/services/di/layers/app"; import { MemoryMeta } from "@/node/services/di/tags"; import { ServiceContainer } from "./serviceContainer"; @@ -143,6 +146,118 @@ describe("ServiceContainer", () => { // The afterEach dispose()+shutdown() pair then exercises the latched path. }); + it("exposes the runtime seams through the field and the Effect context", () => { + services = new ServiceContainer(stores); + + const effectContext = services.toORPCContext()["effect/context"]; + + expect(Context.get(effectContext, AppFiberScopeTag)).toBe(services.appFiberScope); + expect(services.appFiberScope.state._tag).not.toBe("Closed"); + expect(Context.get(effectContext, EffectRunnerTag)).toBe(services.runtime.get(EffectRunnerTag)); + }); + + it("closes the AppFiberScope (interrupt + await) before the explicit teardown steps", async () => { + services = new ServiceContainer(stores); + const steps: string[] = []; + // An I/O-suspended occupant: never resolves on its own, records its cancel + // path and finalizer. Supervised fibers must be gone before any explicit + // teardown step so they can still use their dependencies while finalizing. + services.runtime.managed.runSync( + Effect.forkIn( + Effect.callback(() => + Effect.sync(() => { + steps.push("occupant-cancelled"); + }) + ).pipe(Effect.ensuring(Effect.sync(() => steps.push("occupant-finalized")))), + services.appFiberScope + ) + ); + // desktopBridgeServer.stop() is the first explicit teardown step after the + // shutdown latch. + const bridgeStopSpy = spyOn(services.desktopBridgeServer, "stop").mockImplementation(() => { + steps.push("bridge-stop"); + return Promise.resolve(undefined); + }); + + await services.dispose(); + + expect(bridgeStopSpy).toHaveBeenCalledTimes(1); + expect(steps).toEqual(["occupant-cancelled", "occupant-finalized", "bridge-stop"]); + expect(services.appFiberScope.state._tag).toBe("Closed"); + }); + + it("shares one teardown across concurrent dispose() calls", async () => { + services = new ServiceContainer(stores); + const steps: string[] = []; + // An occupant whose finalization is asynchronous: the first dispose() is + // still awaiting it when the second dispose() arrives. Without a shared + // teardown the second call would find the scope already marked closed and + // proceed to the explicit steps while this finalizer is still running. + services.runtime.managed.runSync( + Effect.forkIn( + Effect.callback(() => Effect.void).pipe( + Effect.ensuring( + Effect.promise(() => new Promise((resolve) => setTimeout(resolve, 30))).pipe( + Effect.andThen(Effect.sync(() => steps.push("occupant-finalized"))) + ) + ) + ), + services.appFiberScope + ) + ); + const bridgeStopSpy = spyOn(services.desktopBridgeServer, "stop").mockImplementation(() => { + steps.push("bridge-stop"); + return Promise.resolve(undefined); + }); + + await Promise.all([services.dispose(), services.dispose()]); + + expect(bridgeStopSpy).toHaveBeenCalledTimes(1); + expect(steps).toEqual(["occupant-finalized", "bridge-stop"]); + }); + + it("runs the clock-driven workers on the runtime's clock", async () => { + // Inject a TestClock beneath the real graph: EffectRunnerLive captures it, + // so the workers' lifecycle fibers sleep on virtual time if (and only if) + // the container hands them the runtime's runner. + const realAppLive = appLayers.AppLive; + const appLiveSpy = spyOn(appLayers, "AppLive").mockImplementation((appStores) => + realAppLive(appStores).pipe(Layer.provideMerge(TestClock.layer())) + ); + try { + services = new ServiceContainer(stores); + } finally { + appLiveSpy.mockRestore(); + } + const runtime = services.runtime.managed; + // IdleCompactionService.checkAllWorkspaces reads the config synchronously + // at the start of every check. + const loadConfigSpy = spyOn(services.config, "loadConfigOrDefault"); + // HeartbeatService's observable lifecycle contract: the scheduler fiber is + // held in `startupTimeout` during the startup delay and moves to + // `checkInterval` once ticking (same shape heartbeatService.test.ts pins). + const heartbeatInternals = services.heartbeatService as unknown as { + startupTimeout: unknown; + checkInterval: unknown; + }; + + services.idleCompactionService.start(); + services.heartbeatService.start(); + expect(loadConfigSpy).not.toHaveBeenCalled(); + expect(heartbeatInternals.startupTimeout).not.toBeNull(); + expect(heartbeatInternals.checkInterval).toBeNull(); + + // Both workers wait one minute before their first tick. + await runtime.runPromise(TestClock.adjust(Duration.minutes(1))); + + expect(loadConfigSpy).toHaveBeenCalledTimes(1); + expect(heartbeatInternals.startupTimeout).toBeNull(); + expect(heartbeatInternals.checkInterval).not.toBeNull(); + + services.heartbeatService.stop(); + services.idleCompactionService.stop(); + }); + it("surfaces a throwing layer as a synchronous constructor throw", () => { const realAppLive = appLayers.AppLive; const appLiveSpy = spyOn(appLayers, "AppLive").mockImplementation((appStores) => diff --git a/src/node/services/serviceContainer.ts b/src/node/services/serviceContainer.ts index cbe9bdbf5d..867e82e8fc 100644 --- a/src/node/services/serviceContainer.ts +++ b/src/node/services/serviceContainer.ts @@ -81,7 +81,15 @@ import { DesktopBridgeServer } from "@/node/services/desktop/DesktopBridgeServer import { DesktopSessionManager } from "@/node/services/desktop/DesktopSessionManager"; import { DesktopTokenManager } from "@/node/services/desktop/DesktopTokenManager"; import type { ORPCContext } from "@/node/orpc/context"; -import { disposeAppRuntime, makeAppRuntime, type AppRuntime } from "@/node/services/di/appRuntime"; +import type { Scope } from "effect"; +import { AppFiberScopeTag } from "@/node/services/di/appFiberScope"; +import { + closeScopeBounded, + disposeAppRuntime, + makeAppRuntime, + type AppRuntime, +} from "@/node/services/di/appRuntime"; +import { EffectRunnerTag } from "@/node/services/di/effectRunner"; import { AppLive } from "@/node/services/di/layers/app"; import { MemoryMeta, type AppTags } from "@/node/services/di/tags"; /** @@ -97,6 +105,11 @@ import { MemoryMeta, type AppTags } from "@/node/services/di/tags"; */ export class ServiceContainer { public readonly runtime: AppRuntime; + /** + * Supervised fiber scope owned by the runtime (`di/appFiberScope.ts`). + * Closed early in `dispose()`; no production occupant yet. + */ + public readonly appFiberScope: Scope.Closeable; public readonly workflowRuntimeFactory = new QuickJSRuntimeFactory(); public readonly config: Config; public readonly sessionLocator: WorkspaceSessionLocator; @@ -165,13 +178,25 @@ export class ServiceContainer { public readonly idleDispatcher: IdleDispatcher; public readonly heartbeatService: HeartbeatService; public readonly agentStatusService: AgentStatusService; - private runtimeDisposed = false; + /** + * The in-flight (or completed) `dispose()` teardown. Every caller shares it, + * so a concurrent or repeated dispose() (the desktop's two before-quit + * paths, tests' dispose-then-shutdown) awaits the one sequence instead of + * re-running steps — in particular it cannot observe the AppFiberScope as + * already closed and start tearing down dependencies while the first call + * is still awaiting the scope's fibers. + */ + private disposePromise: Promise | null = null; constructor(stores: ConfigStores) { // Built eagerly and synchronously (a layer body that throws fails the // constructor, like any service constructor) before the constructor-wired // services so layer-provided instances can be passed into them. this.runtime = makeAppRuntime(AppLive(stores)); + this.appFiberScope = this.runtime.get(AppFiberScopeTag); + // Clock-driven workers run their lifecycle fibers through the runtime's + // context-bound runner (unsupervised; see di/effectRunner.ts). + const effectRunner = this.runtime.get(EffectRunnerTag); const config = stores.config; this.config = config; this.sessionLocator = stores.sessionLocator; @@ -294,7 +319,8 @@ export class ServiceContainer { config, this.historyService, this.extensionMetadata, - (workspaceId) => this.workspaceService.executeIdleCompaction(workspaceId) + (workspaceId) => this.workspaceService.executeIdleCompaction(workspaceId), + effectRunner ); // Forward terminal idle-compaction outcomes so the loop stops re-attempting a // persistently failing workspace (immediately on model_not_found, otherwise after @@ -312,7 +338,8 @@ export class ServiceContainer { this.extensionMetadata, this.workspaceService, this.taskService, - this.idleDispatcher + this.idleDispatcher, + effectRunner ); this.timelineService = new TimelineService( config, @@ -755,13 +782,24 @@ export class ServiceContainer { /** * Dispose all services. Called on app quit to clean up resources. - * Terminates all background processes to prevent orphans. + * Terminates all background processes to prevent orphans. Idempotent: + * concurrent and repeated calls share one teardown (see `disposePromise`). */ - async dispose(): Promise { + dispose(): Promise { + this.disposePromise ??= this.disposeOnce(); + return this.disposePromise; + } + + private async disposeOnce(): Promise { // Must run before any session teardown: AgentSession.dispose() triggers // backgroundProcessManager.cleanup(), which would otherwise erase the persisted // armed-monitor registry records that drive post-restart "monitor lost" wakes. this.backgroundProcessManager.beginShutdown(); + // Interrupt and await the runtime's supervised fibers while every dependency + // they might touch during finalization is still alive. Fixed here (before + // the explicit teardown) so later occupants do not re-derive the position; + // bounded and idempotent, and never rejects (di/appRuntime.ts). + await closeScopeBounded(this.appFiberScope); // Stop the bridge before closing sessions so desktop clients get a clean disconnect. await this.desktopBridgeServer.stop(); this.desktopTokenManager.dispose(); @@ -792,12 +830,7 @@ export class ServiceContainer { await this.timelineService.flush(); // Last: close the Effect runtime's scope. No layer owns finalizers yet, so // this only releases the runtime; the position (after every explicit - // teardown step) is fixed now for later scope-owned occupants. Latched so - // the concurrent before-quit paths and dispose-then-shutdown callers cannot - // race a second close into the bounded wait. - if (!this.runtimeDisposed) { - this.runtimeDisposed = true; - await disposeAppRuntime(this.runtime.managed); - } + // teardown step) is fixed now for later scope-owned occupants. + await disposeAppRuntime(this.runtime.managed); } }