fix(gate): end a sleep's wake at dispose(), and arm it again at start()

changelog.d/sleep-timer-ends-at-stop.fixed.md

@@ -0,0 +1,3 @@
+- A sleeping agent's wake no longer outlives its framework's `stop()`. The event gate's `dispose()` left the sleep's timer armed, so the stopped framework kept the process running until the sleep's end, and then queued a wake there. `dispose()` now clears that timer and keeps the sleep, and until the next `start()` nothing arms a sleep's or a self-wake's timer, even for a turn a failed `stop()` left running. The next `start()` on the same framework arms the wake again for the sleep's end, or wakes the agent at once if the sleep ended while it was stopped, as the queued wake did before.
+- A sleep now wakes the agent at its end by the gate's clock, whatever its length. Its timer could fire before the end, and the wake was then dropped, so the agent slept on until something else woke it. On Node a timer can fire a millisecond early: in a probe on Linux with Node 24.22.0, 7 to 10 of every 1,000 timers of 1 to 40 ms, armed the way a sleep arms them, did, and with Bun 1.4.3 none of 2,000 did. A clock set back did the same on any runtime, and a sleep longer than about 24.8 days fired its timer at once.
+- A sleep a conversation fork or an ephemeral run set ends when it does, at a fork's idle close or at `stop()`, or at the run's end, and so does a `skip_reply` self-wake it left pending. The gate's sleep holds back every agent's wakes but the heartbeat's and a privileged user's, so they were held back until its end, and its wake, like the self-wake's, then went to a name that no longer existed.

src/deadline-timer.ts

@@ -1,6 +1,6 @@
 /** The longest delay one setTimeout holds (2^31 - 1 ms, about 24.8 days).
  *  Node fires a longer one almost at once, with a TimeoutOverflowWarning. */
-const MAX_TIMER_MS = 2_147_483_647;
+export const MAX_TIMER_MS = 2_147_483_647;
 
 type Hold = (timer: ReturnType<typeof setTimeout>) => void;
 

src/framework.ts

@@ -3554,6 +3554,10 @@ export class AgentFramework {
     // and withheld spills, closed for the generation a stop ended, are open
     // for this one.
     this.lifecycleGeneration++;
+    // A sleep kept through a stop() wakes this lifecycle at its end, or now
+    // if it ended while stopped. Before the loop's first pass, so that a wake
+    // owed now joins the requests already pending in one turn.
+    this.eventGate?.resumeSleep();
     this.loopPromise = this.runLoop();
 
     // Start periodic sync timer (if enabled)
@@ -8454,6 +8458,8 @@ export class AgentFramework {
         // deregistered, so disposal must release EventGate liveness itself.
         this.eventGate?.onInferenceEnded(agent.name);
         this.agents.delete(agent.name);
+        // Its sleep and self-wake end with it, as a fork's do.
+        this.eventGate?.agentGone(agent.name);
         this.releaseWindowNowOrAfterResume(agent);
       }
       this.toolImageLedgers.delete(agent.name);
@@ -12645,6 +12651,9 @@ export class AgentFramework {
     this.activeStreams.delete(agentName);
     this.pendingAssistantBlocks.delete(agentName);
     this.agents.delete(agentName);
+    // Its sleep and self-wake end with it (the gate's sleep holds back every
+    // agent's wakes), and neither wake goes to a name that's gone.
+    this.eventGate?.agentGone(agentName);
     if (agent) this.releaseWindowNowOrAfterResume(agent);
     this.turnRouting.delete(agentName);
     this.proseSpeechChains.delete(agentName);
@@ -20966,8 +20975,10 @@ export class AgentFramework {
         }
         const wake = readSkipReplyWake(enrichedCall.input);
         if (wake.kind === 'wake' && this.eventGate) {
-          const { inMs } = this.eventGate.armSelfWake(agentName, wake.seconds, 'skip_reply');
-          console.error(`[self-wake] agent=${agentName} armed in ${Math.round(inMs / 1000)}s (skip_reply)`);
+          const { inMs, armed } = this.eventGate.armSelfWake(agentName, wake.seconds, 'skip_reply');
+          console.error(armed
+            ? `[self-wake] agent=${agentName} armed in ${Math.round(inMs / 1000)}s (skip_reply)`
+            : `[self-wake] agent=${agentName} not armed: the framework has stopped (skip_reply)`);
         }
       }
       this.dispatchChannelToolCall(agentName, enrichedCall);

src/gate/event-gate.ts

@@ -14,6 +14,7 @@
 import { existsSync, readFileSync, statSync, writeFileSync, mkdirSync } from 'node:fs';
 import { dirname, join } from 'node:path';
 import { GateScript } from './gate-script.js';
+import { MAX_TIMER_MS } from '../deadline-timer.js';
 import { safeSlice } from '../safe-slice.js';
 
 import type { ToolDefinition } from '../types/events.js';
@@ -602,6 +603,9 @@ export class EventGate {
   private sleepSuppressed = 0;
   private sleepTimer: ReturnType<typeof setTimeout> | null = null;
   private sleepAgent: string | undefined;
+  // From dispose() to the next resumeSleep(): no sleep or self-wake timer is
+  // armed, though a turn a failed stop() left running can still ask for one.
+  private disposed = false;
 
   // Self-wake timers per agent (skip_reply's wake_in_seconds). Unlike sleep,
   // a self-wake suppresses NOTHING — it means "if nothing else wakes me by
@@ -1735,9 +1739,9 @@ export class EventGate {
     this.sleepSuppressed = 0;
     // Arm the self-wake: when the window elapses, clear sleep and request an
     // inference so the agent resumes on its own (not dependent on a heartbeat
-    // tick or an incoming message). Cancelled by clearSleep()/`wake`.
-    if (this.sleepTimer) clearTimeout(this.sleepTimer);
-    this.sleepTimer = setTimeout(() => this.wakeFromSleep('sleep-expired'), ms);
+    // tick or an incoming message). Cancelled by clearSleep()/`wake`, and by
+    // dispose(), after which start() arms it again (resumeSleep).
+    this.armSleepWake();
     this.reloadPrivilegedIfChanged();
     this.emitTrace({
       type: 'gate:decision',
@@ -1750,6 +1754,48 @@ export class EventGate {
     return { until: this.sleepUntil };
   }
 
+  /**
+   * Arm timers again for a lifecycle that starts after a stop(), and the
+   * sleep's wake first: dispose() clears its timer and keeps the sleep. A
+   * sleep still ahead wakes the agent at its end; one that ended while
+   * stopped wakes it now, since that wake is what the agent asked for. With
+   * no sleep, wakeFromSleep does nothing.
+   */
+  resumeSleep(): void {
+    this.disposed = false;
+    if (this.sleepUntil > this.now()) this.armSleepWake();
+    else this.wakeFromSleep('sleep-expired');
+  }
+
+  /**
+   * Arm the sleep's wake for its end by the gate's own clock. One timer holds
+   * at most MAX_TIMER_MS, about 24.8 days, so a longer sleep waits in steps,
+   * and wakeFromSleep reads the clock as each one fires. Nothing is armed
+   * while the gate is disposed: resumeSleep() arms it at the next start().
+   */
+  private armSleepWake(): void {
+    if (this.sleepTimer) { clearTimeout(this.sleepTimer); this.sleepTimer = null; }
+    if (this.disposed) return;
+    const remaining = Math.max(0, this.sleepUntil - this.now());
+    this.sleepTimer = setTimeout(() => this.wakeFromSleep('sleep-expired'), Math.min(remaining, MAX_TIMER_MS));
+  }
+
+  /**
+   * `agentName` is gone, as a conversation fork's end or an ephemeral run's
+   * removes one: its pending self-wake is cancelled, and a sleep it set ends.
+   * The gate's sleep holds back every agent's wakes, so nobody is asleep once
+   * its agent is gone, and neither wake would reach a name that no longer
+   * exists. A sleep another agent set, or one set with no agent, stays.
+   */
+  agentGone(agentName: string): void {
+    const selfWake = this.selfWakeTimers.get(agentName);
+    if (selfWake) {
+      clearTimeout(selfWake);
+      this.selfWakeTimers.delete(agentName);
+    }
+    if (this.sleepUntil !== 0 && this.sleepAgent === agentName) this.clearSleep();
+  }
+
   // =========================================================================
   // Self-wake (skip_reply wake_in_seconds)
   // =========================================================================
@@ -1761,8 +1807,11 @@ export class EventGate {
    * Cancelled by the agent's next turn start (see onInferenceStarted):
    * the semantics are "if nothing else wakes me by then, wake me".
    */
-  armSelfWake(agentName: string, seconds: number, source = 'skip_reply'): { inMs: number } {
+  armSelfWake(agentName: string, seconds: number, source = 'skip_reply'): { inMs: number; armed: boolean } {
     const ms = Math.max(1_000, Math.min(3_600_000, Math.floor(seconds * 1000)));
+    // A stopped framework's late turn arms no timer (disposed). A self-wake
+    // doesn't outlive a stop, as dispose() clearing the pending ones says.
+    if (this.disposed) return { inMs: ms, armed: false };
     const prev = this.selfWakeTimers.get(agentName);
     if (prev) clearTimeout(prev);
     const timer = setTimeout(() => {
@@ -1803,7 +1852,7 @@ export class EventGate {
       behavior: 'skip',
       timestamp: this.now(),
     });
-    return { inMs: ms };
+    return { inMs: ms, armed: true };
   }
 
   /** End sleep immediately. Returns true if the agent was asleep. */
@@ -1821,11 +1870,20 @@ export class EventGate {
    * Clears the sleep state and requests an inference so the agent wakes on its
    * own — this is what makes `sleep(N)` a real "wake me in N seconds" rather
    * than just a suppression window. No-op if sleep was already cleared (via
-   * `wake`) or re-armed to a later time by a fresh `sleep` call.
+   * `wake`). A timer that fires before the sleep's end by the gate's clock
+   * arms the rest: a step of a sleep longer than one timer holds, a timer that
+   * fires early (Node's can, by a millisecond), or a clock set back.
    */
   private wakeFromSleep(reason: string): void {
-    this.sleepTimer = null;
-    if (this.sleepUntil === 0 || this.sleepUntil > this.now()) return;
+    // The timer that called it has fired; from resumeSleep(), for a sleep
+    // already over, one may still be armed, and it goes, so no stale step of
+    // it can fire later.
+    if (this.sleepTimer) { clearTimeout(this.sleepTimer); this.sleepTimer = null; }
+    if (this.sleepUntil === 0) return;
+    if (this.sleepUntil > this.now()) {
+      this.armSleepWake();
+      return;
+    }
     const note = this.sleepNote;
     const agent = this.sleepAgent;
     this.sleepUntil = 0;
@@ -2003,6 +2061,13 @@ export class EventGate {
       clearTimeout(timer);
     }
     this.selfWakeTimers.clear();
+    // The sleep stays, with its note and agent, and its timer goes: fired
+    // after a stop(), its wake reached a framework that had stopped, and the
+    // timer held the process until the sleep's end. The next start() arms it
+    // again (resumeSleep). Until then nothing arms one, nor a self-wake: a
+    // turn a failed stop() left running can still call sleep or skip_reply.
+    if (this.sleepTimer) { clearTimeout(this.sleepTimer); this.sleepTimer = null; }
+    this.disposed = true;
     this.rateLimitBuckets.clear();
     this.passiveSampleCounters.clear();
     this.rateLimitDenied.clear();

test/conversation-routing.test.ts

@@ -15,6 +15,8 @@ import { Membrane, OpenAIResponsesAPIAdapter, OpenAIResponsesFormatter } from '@
 import type { ContentBlock } from '@animalabs/membrane';
 import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js';
 import { forkAgentName } from '../src/mcpl/conversation-router.js';
+import type { EventGate } from '../src/gate/event-gate.js';
+import { goneNames } from './helpers/gone-names.js';
 
 /** The name the router gives a channel's fork, for the `srv` server every
  *  event here comes from. */
@@ -678,6 +680,68 @@ describe('Conversation routing', () => {
     }
   });
 
+  // The gate's sleep is one for every agent, and its wake goes to the agent
+  // that set it: a fork's sleep ends with the fork.
+  async function makeGatedFramework(conversations: Partial<ConversationRouterConfig> = {}) {
+    return AgentFramework.create({
+      storePath: join(tempDir, 'test.chronicle'),
+      membrane: membrane.asMembrane(),
+      agents: [{ name: 'trunk', model: 'test-model', systemPrompt: 'You are the trunk.' }],
+      modules: [],
+      conversations: { templateAgent: 'trunk', ...conversations },
+      gate: { config: { policies: [], default: 'always' }, configPath: join(tempDir, 'gate.json') },
+    });
+  }
+  const gateOf = (framework: AgentFramework) => (framework as unknown as { eventGate: EventGate }).eventGate;
+
+  it("a fork's sleep ends when the idle sweep closes the fork", async () => {
+    const IDLE_TTL_MS = 60_000;
+    membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }]));
+    const framework = await makeGatedFramework({ idleTtlMs: IDLE_TTL_MS });
+    try {
+      framework.pushEvent(incomingEvent({ channelId: 'slack:D7', text: 'hi', channelType: 'im' }));
+      await framework.runUntilIdle();
+      const fork = forkName('slack:D7', 1);
+      assert.ok(framework.getAgent(fork), 'the fork exists');
+      gateOf(framework).setSleep(3600, 'resting', fork);
+      const gone = goneNames(gateOf(framework));
+
+      forceExpiry(framework, 'slack:D7', IDLE_TTL_MS);
+      membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }]));
+      framework.pushEvent({ type: 'external-message', source: 'test', content: 'tick', metadata: {} } as unknown as ProcessEvent);
+      await framework.runUntilIdle();
+
+      assert.equal(framework.getAgent(fork), null, 'the fork is disposed');
+      assert.equal(gateOf(framework).getSleepState(), null, "and its sleep ends with it, so the other agents' wakes aren't held back");
+      assert.deepStrictEqual(gone, [fork], "the fork's end told the gate, which ends its self-wake too");
+    } finally {
+      // So that a run where the fork's end leaves the sleep doesn't hold the
+      // file's process for the sleep's hour.
+      gateOf(framework).clearSleep();
+      await framework.stop();
+    }
+  });
+
+  it("a fork's sleep ends when stop() ends the fork", async () => {
+    membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }]));
+    const framework = await makeGatedFramework();
+    framework.pushEvent(incomingEvent({ channelId: 'slack:D8', text: 'hi', channelType: 'im' }));
+    await framework.runUntilIdle();
+    const fork = forkName('slack:D8', 1);
+    assert.ok(framework.getAgent(fork), 'the fork exists');
+    const gate = gateOf(framework);
+    const gone = goneNames(gate);
+    try {
+      gate.setSleep(3600, 'resting', fork);
+      await framework.stop();
+      assert.equal(framework.getAgent(fork), null, 'stop() ends the fork');
+      assert.equal(gate.inferenceDiagnostics().sleepUntil, 0, 'and its sleep with it');
+      assert.deepStrictEqual(gone, [fork], "the fork's end told the gate, which ends its self-wake too");
+    } finally {
+      gate.clearSleep();
+    }
+  });
+
   it('a write deferred for a fork during its closure turn is stored in the fork before it is disposed', async () => {
     const IDLE_TTL_MS = 60_000;
     membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }]));