diff --git a/changelog.d/sleep-timer-ends-at-stop.fixed.md b/changelog.d/sleep-timer-ends-at-stop.fixed.md new file mode 100644 index 000000000..dce43fb60 --- /dev/null +++ b/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. diff --git a/src/deadline-timer.ts b/src/deadline-timer.ts index 3c477afd1..4ecb20966 100644 --- a/src/deadline-timer.ts +++ b/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) => void; diff --git a/src/framework.ts b/src/framework.ts index 2df4807f1..cd628c4a9 100644 --- a/src/framework.ts +++ b/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); diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 6aee71f7e..9480007cd 100644 --- a/src/gate/event-gate.ts +++ b/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 | 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(); diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index 357aa418f..928047842 100644 --- a/test/conversation-routing.test.ts +++ b/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 = {}) { + 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!' }])); diff --git a/test/event-gate-sleep.test.ts b/test/event-gate-sleep.test.ts new file mode 100644 index 000000000..c2c7c15a0 --- /dev/null +++ b/test/event-gate-sleep.test.ts @@ -0,0 +1,349 @@ +/** + * A sleep's wake: EventGate's sleep timer, dispose() and resumeSleep(). + * + * Contract: + * - dispose() keeps the sleep (until, note, agent) and clears its timer, so + * nothing fires into a stopped framework and nothing holds the process. + * Until the next resumeSleep(), nothing arms a sleep's or a self-wake's + * timer, though a turn a failed stop() left running can still ask. + * - resumeSleep(), which the framework's start() calls before its loop's + * first pass, arms the wake again for a sleep still ahead, and wakes at + * once for one that ended while stopped: that wake is the one the agent + * asked for. + * - The wake comes at the sleep's end by the gate's own clock: a sleep + * longer than one timer holds waits in steps, and a timer that fires + * before the end waits out the rest. + */ + +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert'; +import { existsSync, mkdirSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { JsStore } from '@animalabs/chronicle'; +import { EventGate } from '../src/gate/event-gate.js'; +import { AgentFramework } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +import { goneNames } from './helpers/gone-names.js'; + +const TMP_DIR = join(import.meta.dirname, '../.test-tmp-gate-sleep'); + +function makeGate(now?: () => number) { + const inferenceRequests: Array<{ agentName: string; reason: string; source: string }> = []; + const traces: string[] = []; + const gate = new EventGate({ + now, + configPath: join(TMP_DIR, 'gate.json'), + initialConfig: { policies: [], default: 'always' }, + emitTrace: (e) => { if (typeof e.eventType === 'string') traces.push(e.eventType); }, + addMessage: () => '', + requestInference: (a, r, s) => inferenceRequests.push({ agentName: a, reason: r, source: s }), + getAgentNames: () => ['agent'], + }); + return { gate, inferenceRequests, traces }; +} + +const wait = (ms: number) => new Promise((r) => setTimeout(r, ms)); +const timeouts = () => process.getActiveResourcesInfo().filter((r) => r === 'Timeout').length; + +/** stop(), failing with why if the sleep's wake had already fired when + * stop() disposed the gate: a sleep that ends during stop()'s own teardown, + * before the gate is disposed, fires by design, so the test couldn't check + * a sleep kept through a stop. It reads the gate's own state at dispose(), + * which a wake that fired has cleared. */ +async function stopDisposingBefore(framework: AgentFramework, gate: EventGate): Promise { + const dispose = gate.dispose.bind(gate); + let keptAtDispose: boolean | undefined; + gate.dispose = () => { keptAtDispose ??= gate.inferenceDiagnostics().sleepUntil > 0; dispose(); }; + await framework.stop(); + assert.ok(keptAtDispose === true, + `${keptAtDispose === undefined ? 'stop() never disposed the gate' : "the sleep's wake fired before stop() disposed the gate"}, so the sleep ended during its teardown, which fires by design: this run can't check a sleep kept through a stop`); +} + +describe("EventGate sleep: its wake, a stop and a fork's end", () => { + beforeEach(() => { + if (existsSync(TMP_DIR)) rmSync(TMP_DIR, { recursive: true }); + mkdirSync(TMP_DIR, { recursive: true }); + }); + afterEach(() => { + if (existsSync(TMP_DIR)) rmSync(TMP_DIR, { recursive: true }); + }); + + it('dispose() keeps the sleep and lets nothing fire', async () => { + const { gate, inferenceRequests, traces } = makeGate(); + const before = timeouts(); + // The first sleep is short too: if setting a sleep again stopped + // replacing its timer, the first sleep's timer would hold this file's + // process until it fired. + gate.setSleep(2, 'resting', 'agent'); + gate.setSleep(0.2, 'resting', 'agent'); + assert.strictEqual(timeouts(), before + 1, 'a sleep set again replaces its timer'); + gate.dispose(); + assert.strictEqual(timeouts(), before, 'dispose() leaves no timer to hold the process'); + await wait(400); + assert.deepStrictEqual(inferenceRequests, [], 'no wake after dispose()'); + assert.ok(!traces.includes('sleep:expired'), 'no expiry after dispose()'); + assert.ok(gate.inferenceDiagnostics().sleepUntil > 0, 'the sleep is kept for the next start'); + }); + + // This test, the next, and the gate's-clock test run on node:test's mock + // timers and Date, so their verdicts follow the clock they set, not the + // host's scheduling. The tests that count a timer read it without waiting. + it('resumeSleep() wakes once at the end of a sleep still ahead', (t) => { + t.mock.timers.enable({ apis: ['setTimeout', 'Date'] }); + const { gate, inferenceRequests } = makeGate(); + gate.setSleep(0.4, 'resting', 'agent'); + gate.dispose(); + t.mock.timers.tick(150); + gate.resumeSleep(); + t.mock.timers.tick(249); + assert.strictEqual(inferenceRequests.length, 0, 'not before the sleep ends'); + t.mock.timers.tick(1); + assert.deepStrictEqual(inferenceRequests, [{ agentName: 'agent', reason: 'sleep-expired: resting', source: 'sleep' }]); + gate.resumeSleep(); + t.mock.timers.tick(1_000); + assert.strictEqual(inferenceRequests.length, 1, 'with the sleep over, resumeSleep() does nothing'); + gate.dispose(); + }); + + it('resumeSleep() wakes at once for a sleep that ended while disposed', (t) => { + t.mock.timers.enable({ apis: ['setTimeout', 'Date'] }); + const { gate, inferenceRequests } = makeGate(); + gate.setSleep(0.1, 'resting', 'agent'); + gate.dispose(); + t.mock.timers.tick(250); + assert.strictEqual(inferenceRequests.length, 0, 'nothing while disposed'); + gate.resumeSleep(); + assert.deepStrictEqual(inferenceRequests, [{ agentName: 'agent', reason: 'sleep-expired: resting', source: 'sleep' }]); + assert.strictEqual(gate.inferenceDiagnostics().sleepUntil, 0); + gate.dispose(); + }); + + it('a sleep longer than one timer holds keeps its wake armed', async () => { + // setTimeout holds at most 2^31-1 ms, about 24.8 days, and fires a longer + // delay after 1 ms. + const { gate, inferenceRequests } = makeGate(); + const timer = () => (gate as unknown as { sleepTimer: unknown }).sleepTimer; + try { + gate.setSleep(30 * 24 * 3600, 'resting', 'agent'); + const first = timer(); + await wait(50); + assert.ok(first !== null && timer() === first, + "one timer still waits for the sleep's end: a step one timer holds, not one that fires at once and is armed again"); + assert.deepStrictEqual(inferenceRequests, []); + } finally { + gate.clearSleep(); + } + }); + + it("a sleep's wake follows the gate's own clock, and waits out a clock set back", (t) => { + t.mock.timers.enable({ apis: ['setTimeout', 'Date'] }); + // The gate's clock runs ten days ahead of the wall clock, and is set back + // a second once the sleep is set, so the 100 ms timer fires 900 ms before + // the sleep's end by the gate's clock, and the rest falls due at 1,100 ms. + let offset = 10 * 24 * 3600 * 1000; + const { gate, inferenceRequests } = makeGate(() => Date.now() + offset); + try { + gate.setSleep(0.1, 'resting', 'agent'); + offset -= 1_000; + t.mock.timers.tick(100); + t.mock.timers.tick(999); + assert.deepStrictEqual(inferenceRequests, [], "not before the sleep's end by the gate's clock"); + t.mock.timers.tick(1); + assert.deepStrictEqual(inferenceRequests, [{ agentName: 'agent', reason: 'sleep-expired: resting', source: 'sleep' }]); + } finally { + gate.clearSleep(); + } + }); + + it('resumeSleep() for a sleep already over cancels the timer still armed for it', () => { + let offset = 0; + const { gate, inferenceRequests } = makeGate(() => Date.now() + offset); + const before = timeouts(); + try { + gate.setSleep(60, 'resting', 'agent'); + // The gate's clock passes the sleep's end before its timer fires. + offset += 120_000; + gate.resumeSleep(); + assert.deepStrictEqual(inferenceRequests, [{ agentName: 'agent', reason: 'sleep-expired: resting', source: 'sleep' }], 'it wakes the agent once'); + assert.strictEqual(timeouts(), before, 'and leaves no timer armed'); + } finally { + gate.clearSleep(); + } + }); + + it("agentGone() ends that agent's sleep and self-wake together, and leaves another agent's", () => { + const { gate } = makeGate(); + const selfWakes = (gate as unknown as { selfWakeTimers: Map }).selfWakeTimers; + const before = timeouts(); + try { + gate.setSleep(60, 'resting', 'fork'); + gate.armSelfWake('fork', 60); + gate.armSelfWake('agent', 60); + gate.agentGone('fork'); + assert.strictEqual(gate.inferenceDiagnostics().sleepUntil, 0, "the gone agent's sleep ends"); + assert.ok(!selfWakes.has('fork'), 'and its self-wake goes, entry and all'); + assert.ok(selfWakes.has('agent'), "another agent's self-wake stays"); + assert.strictEqual(timeouts(), before + 1, "and only that one's timer is left"); + } finally { + gate.dispose(); + } + }); + + it("agentGone() leaves a sleep another agent set, or one set with no agent", () => { + const { gate } = makeGate(); + try { + gate.setSleep(60, 'resting', 'agent'); + gate.agentGone('fork'); + assert.ok(gate.inferenceDiagnostics().sleepUntil > 0, "another agent's end leaves its sleep"); + gate.setSleep(60, 'resting'); + gate.agentGone('fork'); + assert.ok(gate.inferenceDiagnostics().sleepUntil > 0, 'a sleep with no agent stays'); + } finally { + gate.clearSleep(); + } + }); + + it('after dispose(), a sleep set again arms no timer until resumeSleep()', () => { + const { gate } = makeGate(); + const before = timeouts(); + try { + gate.dispose(); + gate.setSleep(60, 'resting', 'agent'); + assert.strictEqual(timeouts(), before, 'no timer while disposed'); + assert.ok(gate.inferenceDiagnostics().sleepUntil > 0, 'the sleep is kept for the next start'); + gate.resumeSleep(); + assert.strictEqual(timeouts(), before + 1, 'resumeSleep() arms it'); + } finally { + gate.clearSleep(); + } + }); + + it('after dispose(), a self-wake arms no timer, and one asked for after resumeSleep() does', () => { + const { gate } = makeGate(); + const before = timeouts(); + try { + gate.dispose(); + assert.strictEqual(gate.armSelfWake('agent', 60).armed, false, 'it says it armed nothing'); + assert.strictEqual(timeouts(), before, 'no timer while disposed'); + gate.resumeSleep(); + assert.strictEqual(gate.armSelfWake('agent', 60).armed, true); + assert.strictEqual(timeouts(), before + 1, 'armed once resumed'); + } finally { + gate.dispose(); + } + }); + + it("an ephemeral run's sleep ends when the run does", async () => { + const membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 's1', name: 'sleep', input: { seconds: 3600, announce: false } }], 'tool_use')); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'done' }])); + const framework = await AgentFramework.create({ + storePath: join(TMP_DIR, 'store-ephemeral'), + membrane: membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + gate: { config: { policies: [], default: 'always' }, configPath: join(TMP_DIR, 'gate-ephemeral.json') }, + modules: [], + }); + const gate = (framework as unknown as { eventGate: EventGate }).eventGate; + const gone = goneNames(gate); + const slept: string[] = []; + framework.onTrace((e) => { + if ((e as { eventType?: unknown }).eventType === 'sleep:set') slept.push('sleep:set'); + }); + try { + framework.start(); + const created = await framework.createEphemeralAgent({ name: 'worker', model: 'test-model', systemPrompt: 'Work.', allowedTools: 'all' }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Go.' }]); + await framework.runEphemeralToCompletion(created.agent, created.contextManager); + assert.deepStrictEqual(slept, ['sleep:set'], 'the run slept'); + assert.strictEqual(gate.getSleepState(), null, 'and its sleep ended with it'); + assert.deepStrictEqual(gone, ['worker'], "the run's end told the gate, which ends its self-wake too"); + } finally { + // So that a run where the sleep outlives the run doesn't hold the + // file's process for the sleep's hour. + gate.clearSleep(); + await framework.stop(); + } + }); + + it("a framework stopped while its agent sleeps wakes nothing until it's started again, then wakes the agent", async () => { + // A store the caller owns: an owned store closes at stop(), and a start() + // after that is refused. + const store = JsStore.openOrCreate({ path: join(TMP_DIR, 'store') }); + const membrane = new MockMembrane(); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'awake' }])); + const framework = await AgentFramework.create({ + store, + membrane: membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + gate: { config: { policies: [], default: 'always' }, configPath: join(TMP_DIR, 'gate-stopped.json') }, + modules: [], + }); + const internals = framework as unknown as { + eventGate: EventGate; + pendingRequests: Array<{ source: string }>; + }; + const expiries: string[] = []; + framework.onTrace((e) => { + if ((e as { eventType?: unknown }).eventType === 'sleep:expired') expiries.push('sleep:expired'); + }); + try { + framework.start(); + // Long enough for stop() to dispose the gate first even on a loaded + // host: a sleep that ends during stop()'s own teardown still fires. + internals.eventGate.setSleep(1, 'resting', 'agent'); + await stopDisposingBefore(framework, internals.eventGate); + await wait(1_500); + assert.deepStrictEqual(expiries, [], 'nothing expires while stopped'); + assert.ok(!internals.pendingRequests.some((r) => r.source === 'sleep'), 'no wake is queued while stopped'); + framework.start(); + const deadline = Date.now() + 3_000; + while (membrane.calls.length === 0 && Date.now() < deadline) await wait(25); + assert.strictEqual(membrane.calls.length, 1, 'the start wakes the agent once'); + assert.deepStrictEqual(expiries, ['sleep:expired']); + } finally { + await framework.stop(); + store.close(); + } + }); + + it('a sleep that ended while stopped wakes the agent in one turn with a request already pending at start', async () => { + const store = JsStore.openOrCreate({ path: join(TMP_DIR, 'store') }); + const membrane = new MockMembrane(); + for (let i = 0; i < 2; i++) membrane.pushResponse(createMockResponse([{ type: 'text', text: 'awake' }])); + const framework = await AgentFramework.create({ + store, + membrane: membrane.asMembrane(), + agents: [{ name: 'agent', model: 'test-model', systemPrompt: 'test' }], + gate: { config: { policies: [], default: 'always' }, configPath: join(TMP_DIR, 'gate-one-turn.json') }, + modules: [], + }); + const internals = framework as unknown as { eventGate: EventGate; pendingRequests: unknown[] }; + const gate = internals.eventGate; + const expiries: string[] = []; + framework.onTrace((e) => { + if ((e as { eventType?: unknown }).eventType === 'sleep:expired') expiries.push('sleep:expired'); + }); + try { + framework.start(); + // As above: the sleep ends after stop() has disposed the gate, so its + // wake is owed at start rather than queued during the teardown. + gate.setSleep(1, 'resting', 'agent'); + await stopDisposingBefore(framework, gate); + await wait(1_500); + assert.strictEqual(framework.nudgeAgent('agent', 'test').ok, true); + framework.start(); + // The loop's first pass takes every pending request at once: the owed + // wake joined the nudge if nothing is left for a second turn. + assert.deepStrictEqual(expiries, ['sleep:expired'], 'the start owed the sleep its wake'); + assert.deepStrictEqual(internals.pendingRequests, [], 'and the first pass took it with the nudge'); + const deadline = Date.now() + 3_000; + while (membrane.calls.length === 0 && Date.now() < deadline) await wait(25); + await wait(600); + assert.strictEqual(membrane.calls.length, 1, 'one turn, and no second one after it'); + } finally { + await framework.stop(); + store.close(); + } + }); +}); diff --git a/test/helpers/gone-names.ts b/test/helpers/gone-names.ts new file mode 100644 index 000000000..8c9c53c3d --- /dev/null +++ b/test/helpers/gone-names.ts @@ -0,0 +1,10 @@ +import type { EventGate } from '../../src/gate/event-gate.js'; + +/** The names the framework tells the gate are gone, through agentGone(), + * whose sleep and self-wake rule the gate's own tests check. */ +export function goneNames(gate: EventGate): string[] { + const gone: string[] = []; + const agentGone = gate.agentGone.bind(gate); + gate.agentGone = (name: string) => { gone.push(name); agentGone(name); }; + return gone; +}