🤖 refactor: convert periodic-worker scheduling to Effect Schedule + Scope

src/node/services/heartbeatService.test.ts

@@ -18,6 +18,7 @@ import type { BackgroundProcessManager } from "./backgroundProcessManager";
 import type { ExtensionMetadataService } from "./ExtensionMetadataService";
 import { advanceAnchoredDeadline, HeartbeatService } from "./heartbeatService";
 import type { HistoryService } from "./historyService";
+import { IdleDispatcher } from "./idleDispatcher";
 import type { InitStateManager } from "./initStateManager";
 import type { TaskService } from "./taskService";
 import { makeAgentTaskIntegrationFake } from "./taskWorkspaceSeam.testUtils";
@@ -584,6 +585,60 @@ describe("HeartbeatService", () => {
       expect(internals.queuedWorkspaceIds.size).toBe(0);
       expect(executeHeartbeatMock).not.toHaveBeenCalled();
     });
+
+    test("start failure releases earlier acquisitions and leaves the service restartable", () => {
+      const dispatcher = new IdleDispatcher();
+      const emitter = new EventEmitter();
+      let failListenerRegistration = true;
+      const realOn = emitter.on.bind(emitter);
+      // Fail only the SECOND listener registration ("metadata") so the test
+      // also covers partial-acquisition rollback: the already-registered
+      // "activity" listener must be released, not leaked across retries.
+      emitter.on = ((event: string, listener: (...args: unknown[]) => void) => {
+        if (failListenerRegistration && event === "metadata") {
+          throw new Error("listener registration failed");
+        }
+        return realOn(event, listener);
+      }) as typeof emitter.on;
+      const failingWorkspaceService = Object.assign(emitter, {
+        getChatHistory: getChatHistoryMock,
+        executeHeartbeat: executeHeartbeatMock,
+        isBusyForMessage: isBusyForMessageMock,
+      }) as unknown as WorkspaceService;
+
+      const failingService = new HeartbeatService(
+        mockConfig,
+        mockExtensionMetadata,
+        failingWorkspaceService,
+        mockTaskService,
+        dispatcher
+      );
+
+      expect(() => failingService.start()).toThrow();
+
+      // The "activity" listener registered before the failing "metadata"
+      // registration must have been rolled back — a leak here would double up
+      // event handling after a successful retry.
+      expect(emitter.listenerCount("activity")).toBe(0);
+
+      // The idle-consumer registration acquired before the failing step must
+      // have been released: re-registering the same consumer name would
+      // otherwise trip the dispatcher's duplicate-registration assert.
+      const disposeProbe = dispatcher.registerConsumer({
+        name: "heartbeat",
+        priority: 50,
+        buildPayload: () => Promise.resolve(null),
+      });
+      disposeProbe();
+
+      // The rollback restores the stopped state, so start() succeeds once the
+      // failure cause is fixed.
+      failListenerRegistration = false;
+      failingService.start();
+      expect(emitter.listenerCount("activity")).toBe(1);
+      failingService.stop();
+      expect(emitter.listenerCount("activity")).toBe(0);
+    });
   });
 
   describe("event handling", () => {

src/node/services/heartbeatService.ts

@@ -1,3 +1,4 @@
+import { Duration, Effect, Exit, Schedule, Scope, type Fiber } from "effect";
 import assert from "@/common/utils/assert";
 import type { MuxMessage } from "@/common/types/message";
 import type { ProjectsConfig, Workspace } from "@/common/types/project";
@@ -61,8 +62,22 @@ export class HeartbeatService {
 
   private timelineRecorder: TimelineRecorder = NOOP_TIMELINE_RECORDER;
 
-  private startupTimeout: ReturnType<typeof setTimeout> | null = null;
-  private checkInterval: ReturnType<typeof setInterval> | null = null;
+  // The scheduler runs as a single Effect fiber forked into `lifecycleScope`:
+  // sleep(STARTUP_DELAY_MS), then tick immediately and every CHECK_INTERVAL_MS.
+  // The legacy two-field shape is preserved because it is the observable
+  // lifecycle contract (tests pin the null/non-null progression):
+  // `startupTimeout` holds the fiber while the startup delay is pending and
+  // `checkInterval` holds it once the periodic ticker is live.
+  private startupTimeout: Fiber.Fiber<void> | null = null;
+  private checkInterval: Fiber.Fiber<void> | null = null;
+  /**
+   * Owns every resource start() acquires — idle-consumer registration,
+   * workspace event listeners, and the scheduler fiber. Closing it releases
+   * them in reverse acquisition order (fiber interrupt, listeners off,
+   * consumer dispose — the same order the hand-rolled stop() used) and is
+   * guaranteed to run them even when a later startup step throws.
+   */
+  private lifecycleScope: Scope.Closeable | null = null;
   private stopped = true;
 
   private readonly nextEligibleAtByWorkspaceId = new Map<string, number>();
@@ -114,25 +129,90 @@ export class HeartbeatService {
     this.stopped = false;
     this.lifecycleVersion += 1;
 
-    this.heartbeatConsumerDisposer = this.idleDispatcher.registerConsumer({
-      name: HEARTBEAT_IDLE_CONSUMER_NAME,
-      priority: HEARTBEAT_IDLE_CONSUMER_PRIORITY,
-      buildPayload: (workspaceId) => this.buildHeartbeatDispatchPayload(workspaceId),
-    });
-    this.workspaceService.on("activity", this.onActivity);
-    this.workspaceService.on("metadata", this.onMetadata);
-
-    this.startupTimeout = setTimeout(() => {
-      if (this.stopped) {
+    // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this`
+    const self = this;
+    const scope = Scope.makeUnsafe();
+    this.lifecycleScope = scope;
+
+    const scheduler = Effect.gen(function* () {
+      yield* Effect.sleep(Duration.millis(STARTUP_DELAY_MS));
+      // Defensive parity with the legacy setTimeout callback: interruption via
+      // stop() already prevents this resumption, so a stopped service must
+      // never transition into the ticking phase even if a wake-up raced it.
+      if (self.stopped) {
         return;
       }
 
+      // Startup delay elapsed: the same fiber now becomes the periodic ticker.
+      self.checkInterval = self.startupTimeout;
+      self.startupTimeout = null;
+      // Effect.repeat runs the first tick immediately (matching the legacy
+      // direct tick() call when the startup timer fired), then Schedule.fixed
+      // reproduces setInterval cadence: wall-clock anchored, no burst catch-up.
+      // tick() is synchronous fire-and-forget, so the body never delays a slot.
+      yield* Effect.sync(() => self.tick()).pipe(
+        Effect.repeat(Schedule.fixed(Duration.millis(CHECK_INTERVAL_MS)))
+      );
+    });
+
+    const acquireResources = Effect.gen(function* () {
+      yield* Effect.acquireRelease(
+        Effect.sync(() => {
+          self.heartbeatConsumerDisposer = self.idleDispatcher.registerConsumer({
+            name: HEARTBEAT_IDLE_CONSUMER_NAME,
+            priority: HEARTBEAT_IDLE_CONSUMER_PRIORITY,
+            buildPayload: (workspaceId) => self.buildHeartbeatDispatchPayload(workspaceId),
+          });
+        }),
+        () =>
+          Effect.sync(() => {
+            self.heartbeatConsumerDisposer?.();
+            self.heartbeatConsumerDisposer = null;
+          })
+      );
+      // One acquireRelease per listener: a combined acquisition would install
+      // its finalizer only after BOTH .on() calls succeed, so a throw from the
+      // second registration (e.g. a `newListener` hook) would leak the first
+      // listener across start() retries (Codex P2 on #4031).
+      yield* Effect.acquireRelease(
+        Effect.sync(() => {
+          self.workspaceService.on("activity", self.onActivity);
+        }),
+        () =>
+          Effect.sync(() => {
+            self.workspaceService.off("activity", self.onActivity);
+          })
+      );
+      yield* Effect.acquireRelease(
+        Effect.sync(() => {
+          self.workspaceService.on("metadata", self.onMetadata);
+        }),
+        () =>
+          Effect.sync(() => {
+            self.workspaceService.off("metadata", self.onMetadata);
+          })
+      );
+      self.startupTimeout = yield* Effect.forkIn(scheduler, scope);
+    });
+
+    try {
+      // Runs synchronously: acquisitions are Effect.sync and forkIn executes
+      // 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));
+    } catch (error) {
+      // Guaranteed cleanup on partial startup failure: close the scope so the
+      // finalizers registered before the failing step run (the hand-rolled
+      // version leaked earlier acquisitions here), restore the stopped state
+      // so a later start() retry is possible, then surface the error.
+      this.lifecycleScope = null;
       this.startupTimeout = null;
-      this.tick();
-      this.checkInterval = setInterval(() => {
-        this.tick();
-      }, CHECK_INTERVAL_MS);
-    }, STARTUP_DELAY_MS);
+      this.checkInterval = null;
+      this.stopped = true;
+      Effect.runSync(Scope.close(scope, Exit.void));
+      throw error;
+    }
 
     log.info("HeartbeatService started", {
       startupDelayMs: STARTUP_DELAY_MS,
@@ -144,19 +224,25 @@ export class HeartbeatService {
     this.stopped = true;
     this.lifecycleVersion += 1;
 
-    if (this.startupTimeout) {
-      clearTimeout(this.startupTimeout);
-      this.startupTimeout = null;
-    }
-    if (this.checkInterval) {
-      clearInterval(this.checkInterval);
-      this.checkInterval = null;
-    }
-
-    this.workspaceService.off("activity", this.onActivity);
-    this.workspaceService.off("metadata", this.onMetadata);
-    this.heartbeatConsumerDisposer?.();
-    this.heartbeatConsumerDisposer = null;
+    // Captured before teardown for the shutdown log below.
+    const schedulerPhase =
+      this.checkInterval != null
+        ? "ticking"
+        : this.startupTimeout != null
+          ? "startup_delay"
+          : "not_started";
+
+    if (this.lifecycleScope) {
+      const scope = this.lifecycleScope;
+      this.lifecycleScope = null;
+      // Releases everything start() acquired, in reverse acquisition order:
+      // 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.startupTimeout = null;
+    this.checkInterval = null;
 
     this.nextEligibleAtByWorkspaceId.clear();
     this.trackedIntervalMsByWorkspaceId.clear();
@@ -166,7 +252,7 @@ export class HeartbeatService {
     this.isProcessingQueue = false;
     this.tickInFlight = false;
 
-    log.info("HeartbeatService stopped");
+    log.info("HeartbeatService stopped", { schedulerPhase });
   }
 
   private tick(): void {

src/node/services/idleCompactionService.ts

@@ -1,3 +1,4 @@
+import { Duration, Effect, Exit, Schedule, Scope } from "effect";
 import assert from "@/common/utils/assert";
 import type { Config } from "@/node/config";
 import type { HistoryService } from "./historyService";
@@ -44,8 +45,12 @@ export class IdleCompactionService {
   private readonly historyService: HistoryService;
   private readonly extensionMetadata: ExtensionMetadataService;
   private readonly executeIdleCompaction: (workspaceId: string) => Promise<void>;
-  private initialTimeout: ReturnType<typeof setTimeout> | null = null;
-  private checkInterval: ReturnType<typeof setInterval> | null = null;
+  /**
+   * Owns the checker fiber forked by start(): sleep(INITIAL_CHECK_DELAY_MS),
+   * then check immediately and every CHECK_INTERVAL_MS. Closing the scope in
+   * stop() interrupts the fiber, synchronously clearing its pending timer.
+   */
+  private lifecycleScope: Scope.Closeable | null = null;
   private readonly queue: QueuedIdleCompaction[] = [];
   private readonly queuedWorkspaceIds = new Set<string>();
   private readonly activeWorkspaceIds = new Set<string>();
@@ -76,14 +81,27 @@ export class IdleCompactionService {
   start(): void {
     this.stopped = false;
 
-    // First check after delay to let startup settle.
-    this.initialTimeout = setTimeout(() => {
-      void this.checkAllWorkspaces();
-      // Then periodically.
-      this.checkInterval = setInterval(() => {
-        void this.checkAllWorkspaces();
-      }, CHECK_INTERVAL_MS);
-    }, INITIAL_CHECK_DELAY_MS);
+    // eslint-disable-next-line @typescript-eslint/no-this-alias -- Effect.gen generator bodies do not inherit `this`
+    const self = this;
+    const scope = Scope.makeUnsafe();
+    this.lifecycleScope = scope;
+
+    const checker = Effect.gen(function* () {
+      // First check after delay to let startup settle.
+      yield* Effect.sleep(Duration.millis(INITIAL_CHECK_DELAY_MS));
+      // Effect.repeat runs the first check immediately (matching the legacy
+      // direct call when the initial timer fired), then Schedule.fixed
+      // reproduces setInterval cadence. The check stays fire-and-forget so a
+      // slow sweep never delays the next cadence slot (same as setInterval).
+      yield* Effect.sync(() => {
+        void self.checkAllWorkspaces();
+      }).pipe(Effect.repeat(Schedule.fixed(Duration.millis(CHECK_INTERVAL_MS))));
+    });
+    // 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));
+
     log.info("IdleCompactionService started", {
       initialDelayMs: INITIAL_CHECK_DELAY_MS,
       intervalMs: CHECK_INTERVAL_MS,
@@ -96,13 +114,12 @@ export class IdleCompactionService {
   stop(): void {
     this.stopped = true;
 
-    if (this.initialTimeout) {
-      clearTimeout(this.initialTimeout);
-      this.initialTimeout = null;
-    }
-    if (this.checkInterval) {
-      clearInterval(this.checkInterval);
-      this.checkInterval = null;
+    if (this.lifecycleScope) {
+      const scope = this.lifecycleScope;
+      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));
     }
 
     // Best-effort queue reset: do not start new compactions after stop().

src/node/services/idleDispatcher.ts

@@ -1,3 +1,4 @@
+import { Duration, Effect, type Fiber } from "effect";
 import assert from "@/common/utils/assert";
 import { log } from "./log";
 
@@ -29,7 +30,13 @@ interface IdleDispatcherOptions {
 interface PendingDispatchRequest {
   readonly sources: Set<string>;
   readonly resolvers: Array<() => void>;
-  debounceTimer: ReturnType<typeof setTimeout> | null;
+  /**
+   * Debounce fiber: sleeps for `debounceMs` (Effect's clock registers a plain
+   * `setTimeout` under the hood), then marks the workspace ready. Non-null
+   * exactly while the debounce window is open — later requests for the same
+   * workspace coalesce into this pending entry instead of re-arming it.
+   */
+  debounceFiber: Fiber.Fiber<void> | null;
 }
 
 export class IdleDispatcher {
@@ -89,14 +96,24 @@ export class IdleDispatcher {
       pending.sources.add(source);
       pending.resolvers.push(resolve);
 
-      if (pending.debounceTimer != null) {
+      if (pending.debounceFiber != null) {
         return;
       }
 
-      pending.debounceTimer = setTimeout(() => {
-        pending.debounceTimer = null;
-        this.markWorkspaceReady(workspaceId);
-      }, this.debounceMs);
+      // Effect.runFork executes synchronously up to the sleep, so the debounce
+      // timer is registered before this callback returns (same observable
+      // ordering as the previous setTimeout call); a zero-duration sleep still
+      // defers to a timer tick rather than firing inline.
+      pending.debounceFiber = Effect.runFork(
+        Effect.sleep(Duration.millis(this.debounceMs)).pipe(
+          Effect.flatMap(() =>
+            Effect.sync(() => {
+              pending.debounceFiber = null;
+              this.markWorkspaceReady(workspaceId);
+            })
+          )
+        )
+      );
     });
   }
 
@@ -109,7 +126,7 @@ export class IdleDispatcher {
     const pending: PendingDispatchRequest = {
       sources: new Set<string>(),
       resolvers: [],
-      debounceTimer: null,
+      debounceFiber: null,
     };
     this.pendingByWorkspaceId.set(workspaceId, pending);
     return pending;