fix(gate): refuse a conversation fork's and an ephemeral run's sleep and wake

changelog.d/fork-sleep-refused.changed.md

@@ -0,0 +1 @@
+- A conversation fork's or an ephemeral run's `sleep` and `wake` are refused, with a result saying that nothing was set or changed, and why. The gate's sleep quiets the whole agent, so a fork's sleep held back the primary's wakes until the sleep or the fork ended, and a fork's `wake` ended the primary's own sleep. Where a fork can call `skip_reply`, the refusal names its `wake_in_seconds` as the way to come back to the conversation later. The primary's sleep is unchanged, and still quiets the messages its forks would wake on.

src/framework.ts

@@ -9362,7 +9362,7 @@ 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.
+        // Its pending self-wake ends with it, as a fork's does.
         this.eventGate?.agentGone(agent.name);
         this.releaseWindowNowOrAfterResume(agent);
       }
@@ -13777,8 +13777,7 @@ 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.
+    // Its pending self-wake ends with it, rather than wake a name that's gone.
     this.eventGate?.agentGone(agentName);
     if (agent) this.releaseWindowNowOrAfterResume(agent);
     this.turnRouting.delete(agentName);
@@ -30128,6 +30127,27 @@ export class AgentFramework {
   }
 
   private dispatchSleepToolCall(agentName: string, call: ToolCall): void {
+    // The gate's sleep quiets the whole agent, every conversation of it, so a
+    // conversation fork or an ephemeral run neither sets one nor ends the
+    // agent's own. Refused before the tool runs, as a fork's channel_open is.
+    const fork = this.conversationAgentHomes.has(agentName);
+    if (fork || this.ephemeralRuns.has(agentName)) {
+      const one = fork ? 'conversation' : 'run';
+      // Only where the fork can follow it: skip_reply needs the channel
+      // registry, which a fork made from an event pushed straight to the
+      // framework (pushEvent, with no MCPL server) doesn't have, and a
+      // closing fork's wake ends with it.
+      const comeBack = fork && this.channelRegistry && this.canAgentUseTool(agentName, 'skip_reply')
+        && !this.closingConversationAgents.has(agentName)
+        ? ' To come back to this conversation later, call skip_reply with wake_in_seconds.'
+        : '';
+      const error = call.name === 'sleep'
+        ? `Nothing was set: a sleep quiets the whole agent, not one ${one}.${comeBack}`
+        : `Nothing was changed: a sleep is the agent's, so this ${one} can't end one.`;
+      this.emitTrace({ type: 'tool:failed', module: 'gate', tool: call.name, callId: call.id, error: `${agentName}: ${error}` });
+      this.pushEvent({ type: 'tool-result', callId: call.id, toolName: call.name, agentName, moduleName: 'gate', result: this.refusedResult(agentName, call.id, error) });
+      return;
+    }
     this.emitTrace({ type: 'tool:started', module: 'gate', tool: call.name, callId: call.id, input: call.input });
     const gate = this.eventGate!;
     const input = (call.input ?? {}) as { seconds?: number; announce?: boolean; message?: string };

src/gate/event-gate.ts

@@ -3488,18 +3488,16 @@ export class EventGate {
 
   /**
    * `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.
+   * removes one: its pending self-wake is cancelled, rather than wake a name
+   * that no longer exists. It holds no sleep: neither can set one (the
+   * framework refuses their `sleep`).
    */
   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();
   }
 
   // =========================================================================

test/conversation-routing.test.ts

@@ -17,6 +17,8 @@ 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';
+import { gateToolCalls } from './helpers/gate-tool-calls.js';
+import { forceExpiry } from './helpers/force-expiry.js';
 
 /** The name the router gives a channel's fork, for the `srv` server every
  *  event here comes from. */
@@ -84,15 +86,6 @@ describe('Conversation routing', () => {
     });
   }
 
-  /** Make a binding look idle past the TTL and re-arm the sweep throttle —
-   * deterministic expiry instead of racing a 1ms TTL against the wall clock. */
-  function forceExpiry(framework: AgentFramework, channelId: string, idleTtlMs: number) {
-    const binding = framework.getConversationRouter()!.getBinding({ serverId: 'srv', channelId });
-    assert.ok(binding, `binding for ${channelId} should exist before forcing expiry`);
-    binding!.lastActivity = Date.now() - idleTtlMs - 1_000;
-    (framework as unknown as { lastConversationSweep: number }).lastConversationSweep = 0;
-  }
-
   it('rejects an unknown template agent at creation', async () => {
     await assert.rejects(
       () => AgentFramework.create({
@@ -752,7 +745,7 @@ describe('Conversation routing', () => {
     assert.ok(g1, 'g1 fork should exist after spawn');
 
     // Force expiry deterministically and provide the closure-turn response.
-    forceExpiry(framework, 'slack:D3', IDLE_TTL_MS);
+    forceExpiry(framework, { serverId: 'srv', channelId: 'slack:D3' }, IDLE_TTL_MS);
     membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }]));
 
     // Nudge the loop so the sweep runs.
@@ -819,8 +812,9 @@ 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.
+  // The gate's sleep quiets the whole agent, every conversation of it, so a
+  // fork's sleep and wake are refused (#646), and a fork's end tells the gate,
+  // which cancels the fork's pending self-wake.
   async function makeGatedFramework(conversations: Partial<ConversationRouterConfig> = {}) {
     return AgentFramework.create({
       storePath: join(tempDir, 'test.chronicle'),
@@ -832,55 +826,109 @@ describe('Conversation routing', () => {
     });
   }
   const gateOf = (framework: AgentFramework) => (framework as unknown as { eventGate: EventGate }).eventGate;
+  const ordinaryWake = { eventType: 'external-message', serverId: 'test', channelId: '', content: 'tick' };
 
-  it("a fork's sleep ends when the idle sweep closes the fork", async () => {
+  it("a fork's sleep is refused, so the primary's ordinary wakes aren't held back", async () => {
+    membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 's1', name: 'sleep', input: { seconds: 3600, announce: false } }], 'tool_use'));
+    membrane.pushResponse(createMockResponse([{ type: 'text', text: 'ok' }]));
+    const framework = await makeGatedFramework();
+    const gate = gateOf(framework);
+    const calls = gateToolCalls(framework);
+    try {
+      framework.pushEvent(incomingEvent({ channelId: 'slack:D9', text: 'hi', channelType: 'im' }));
+      await framework.runUntilIdle();
+      assert.deepStrictEqual(calls.refused, ['sleep: Nothing was set: a sleep quiets the whole agent, not one conversation.']);
+      assert.deepStrictEqual(calls.started, [], 'refused before the tool runs');
+      assert.equal(gate.getSleepState(), null, 'nothing is asleep');
+      assert.notEqual(gate.evaluate(ordinaryWake).policyName, 'asleep', "the primary's ordinary wake goes on to the policies");
+    } finally {
+      // So that a run where the sleep takes doesn't hold the file's process for its hour.
+      gate.clearSleep();
+      await framework.stop();
+    }
+  });
+
+  it("a fork's wake is refused, so the primary's sleep stays", async () => {
+    membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 'w1', name: 'wake', input: {} }], 'tool_use'));
+    membrane.pushResponse(createMockResponse([{ type: 'text', text: 'ok' }]));
+    const framework = await makeGatedFramework();
+    const gate = gateOf(framework);
+    const calls = gateToolCalls(framework);
+    try {
+      gate.setSleep(3600, 'resting', 'trunk');
+      framework.pushEvent(incomingEvent({ channelId: 'slack:D10', text: 'hi', channelType: 'im' }));
+      await framework.runUntilIdle();
+      assert.deepStrictEqual(calls.refused, ["wake: Nothing was changed: a sleep is the agent's, so this conversation can't end one."]);
+      assert.deepStrictEqual(calls.started, [], 'refused before the tool runs');
+      assert.ok(gate.getSleepState() !== null, "the primary's sleep stays");
+    } finally {
+      gate.clearSleep();
+      await framework.stop();
+    }
+  });
+
+  it("a fork's end at the idle sweep tells the gate", async () => {
     const IDLE_TTL_MS = 60_000;
     membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }]));
     const framework = await makeGatedFramework({ idleTtlMs: IDLE_TTL_MS });
+    const gone = goneNames(gateOf(framework));
     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);
+      forceExpiry(framework, { serverId: 'srv', channelId: '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");
+      assert.deepStrictEqual(gone, [fork], 'and its end told the gate, which cancels its self-wake');
     } 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 () => {
+  it("a fork's end at stop() tells the gate, and leaves the primary's sleep", async () => {
     membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hello!' }]));
     const framework = await makeGatedFramework();
+    const gate = gateOf(framework);
+    const gone = goneNames(gate);
     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);
+      gate.setSleep(3600, 'resting', 'trunk');
       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");
+      assert.deepStrictEqual(gone, [fork], 'and its end told the gate');
+      assert.ok(gate.inferenceDiagnostics().sleepUntil > 0, "the primary's sleep is kept through the fork's end and the stop");
     } finally {
       gate.clearSleep();
     }
   });
 
+  it("the primary's own wake and sleep still work", async () => {
+    // One turn: wake (with nothing asleep), then sleep, which ends the turn,
+    // since a second turn would first need something to rouse the sleeper.
+    membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 'w1', name: 'wake', input: {} }], 'tool_use'));
+    membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 's1', name: 'sleep', input: { seconds: 3600, announce: false } }], 'tool_use'));
+    const framework = await makeGatedFramework();
+    const gate = gateOf(framework);
+    const calls = gateToolCalls(framework);
+    try {
+      assert.strictEqual(framework.nudgeAgent('trunk', 'test').ok, true);
+      await framework.runUntilIdle();
+      assert.deepStrictEqual(calls.succeeded, ['wake', 'sleep']);
+      assert.deepStrictEqual(calls.refused, []);
+      assert.ok(gate.getSleepState() !== null, 'the primary slept');
+    } finally {
+      gate.clearSleep();
+      await framework.stop();
+    }
+  });
+
   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!' }]));
@@ -890,7 +938,7 @@ describe('Conversation routing', () => {
     const g1 = framework.getAgent(forkName('slack:D4', 1))!;
     assert.ok(g1, 'g1 fork should exist after spawn');
 
-    forceExpiry(framework, 'slack:D4', IDLE_TTL_MS);
+    forceExpiry(framework, { serverId: 'srv', channelId: 'slack:D4' }, IDLE_TTL_MS);
     membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }]));
     // While the closure turn is alive, a write addressed to the fork is
     // deferred to that turn's end.
@@ -926,7 +974,7 @@ describe('Conversation routing', () => {
     const g1 = framework.getAgent(fork)!;
     assert.ok(g1, 'g1 fork should exist after spawn');
 
-    forceExpiry(framework, 'slack:D6', IDLE_TTL_MS);
+    forceExpiry(framework, { serverId: 'srv', channelId: 'slack:D6' }, IDLE_TTL_MS);
     membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }]));
     const internals = framework as unknown as {
       addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string }): string;
@@ -979,7 +1027,7 @@ describe('Conversation routing', () => {
       return add(...args);
     };
 
-    forceExpiry(framework, 'slack:D5', IDLE_TTL_MS);
+    forceExpiry(framework, { serverId: 'srv', channelId: 'slack:D5' }, IDLE_TTL_MS);
     membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }]));
     const internals = framework as unknown as {
       addMessage(participant: string, content: ContentBlock[], metadata?: undefined, opts?: { forAgent?: string }): string;
@@ -1032,7 +1080,7 @@ describe('Conversation routing', () => {
       lastConversationSweep: number;
       sweepExpiredConversations(): void;
     };
-    forceExpiry(framework, channelId, IDLE_TTL_MS);
+    forceExpiry(framework, { serverId: 'srv', channelId }, IDLE_TTL_MS);
     internals.sweepExpiredConversations();
     assert.equal(internals.pendingRequests.filter((r) => r.agentName === fork).length, 1, 'the closure turn is queued');
     internals.pendingRequests.splice(0, internals.pendingRequests.length,
@@ -1091,7 +1139,7 @@ describe('Conversation routing', () => {
     const g1 = framework.getAgent(forkName('slack:D6', 1))!;
     assert.ok(g1, 'g1 fork should exist after spawn');
 
-    forceExpiry(framework, 'slack:D6', IDLE_TTL_MS);
+    forceExpiry(framework, { serverId: 'srv', channelId: 'slack:D6' }, IDLE_TTL_MS);
     membrane.pushResponse(createMockResponse([
       { type: 'text', text: 'Checking.' },
       { type: 'tool_use', id: 'e1', name: 'test--echo', input: {} },
@@ -1148,7 +1196,7 @@ describe('Conversation routing', () => {
       });
       failCheckpointRemoval(framework, fork);
 
-      forceExpiry(framework, 'slack:D9', IDLE_TTL_MS);
+      forceExpiry(framework, { serverId: 'srv', channelId: 'slack:D9' }, IDLE_TTL_MS);
       membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Final report posted.' }]));
       // The disposal runs in the closure turn's driveStream finally, whose
       // promise nothing holds: had it rejected, node:test would fail this

test/ephemeral-no-global-fallback.test.ts

@@ -16,6 +16,7 @@ import assert from 'node:assert/strict';
 import { ContextManager, PassthroughStrategy } from '@animalabs/context-manager';
 import { fixture, eventually, settles, TS } from './helpers/coalescing-fixture.js';
 import { createMockResponse } from './helpers/mock-membrane.js';
+import { gateToolCalls } from './helpers/gate-tool-calls.js';
 
 type Fixture = Awaited<ReturnType<typeof fixture>>;
 type Ref = { serverId: string; channelId: string };
@@ -260,17 +261,18 @@ describe('an ephemeral run takes no global fallback (#551)', () => {
     } finally { await f.close(); }
   });
 
-  it('a subagent\'s sleep announcement is not posted to the latest inbound channel', async () => {
+  it('a subagent\'s sleep is refused, and the gate stays awake', async () => {
     const f = await fixture({ framework: { gate: { config: { default: 'always' } } } as never });
-    const t = speechTraces(f);
+    const calls = gateToolCalls(f.framework);
     try {
       await busyFallback(f);
       await runSubagent(f, 'sub-sleep', 'locus', true,
-        createMockResponse([{ type: 'tool_use', id: 's1', name: 'sleep', input: { seconds: 60, message: 'subagent going quiet' } }], 'tool_use'));
+        createMockResponse([{ type: 'tool_use', id: 's1', name: 'sleep', input: { seconds: 60 } }], 'tool_use'),
+        text('done'));
 
-      assert.ok(!publishedText(f).includes('subagent going quiet'), `not published: ${publishedText(f)}`);
-      assert.deepEqual(t.seen.filter((x) => x.conversationId === 'sub-sleep'), [], 'neither routed nor counted as a failed route');
-    } finally { t.off(); await f.close(); }
+      assert.deepEqual(calls.refused, ['sleep: Nothing was set: a sleep quiets the whole agent, not one run.']);
+      assert.equal((f.framework as unknown as { eventGate: { getSleepState(): unknown } }).eventGate.getSleepState(), null, 'the gate stays awake');
+    } finally { await f.close(); }
   });
 
   it('a channel the subagent opens itself is still where its prose lands', async () => {