diff --git a/changelog.d/fork-sleep-refused.changed.md b/changelog.d/fork-sleep-refused.changed.md new file mode 100644 index 000000000..0a8c598f3 --- /dev/null +++ b/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. diff --git a/src/framework.ts b/src/framework.ts index 188498cf3..c93b9375d 100644 --- a/src/framework.ts +++ b/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 }; diff --git a/src/gate/event-gate.ts b/src/gate/event-gate.ts index 32500b7ff..f21b940d3 100644 --- a/src/gate/event-gate.ts +++ b/src/gate/event-gate.ts @@ -3488,10 +3488,9 @@ 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); @@ -3499,7 +3498,6 @@ export class EventGate { clearTimeout(selfWake); this.selfWakeTimers.delete(agentName); } - if (this.sleepUntil !== 0 && this.sleepAgent === agentName) this.clearSleep(); } // ========================================================================= diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index 5efb897aa..39618bc97 100644 --- a/test/conversation-routing.test.ts +++ b/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 = {}) { 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 diff --git a/test/ephemeral-no-global-fallback.test.ts b/test/ephemeral-no-global-fallback.test.ts index 02395db45..44aad3b4d 100644 --- a/test/ephemeral-no-global-fallback.test.ts +++ b/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>; 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 () => { diff --git a/test/event-gate-sleep.test.ts b/test/event-gate-sleep.test.ts index c2c7c15a0..9b4ca6f3c 100644 --- a/test/event-gate-sleep.test.ts +++ b/test/event-gate-sleep.test.ts @@ -24,9 +24,37 @@ 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'; +import { gateToolCalls } from './helpers/gate-tool-calls.js'; +import { forceExpiry } from './helpers/force-expiry.js'; +import type { ChannelRef } from '../src/mcpl/channel-registry.js'; +import { fixture, eventually, TS, ok } from './helpers/coalescing-fixture.js'; const TMP_DIR = join(import.meta.dirname, '../.test-tmp-gate-sleep'); +/** A framework whose agent forks a conversation for each message on the open + * channel `channelId`, with a gate; `incoming` sends that channel a message. */ +async function forkFixture(channelId: string, options: { agents?: unknown[]; idleTtlMs?: number } = {}) { + const f = await fixture({ + ...(options.agents ? { agents: options.agents } : {}), + framework: { + gate: { config: { policies: [], default: 'always' }, configPath: join(TMP_DIR, `gate-${channelId.replace(/\W/g, '-')}.json`) }, + conversations: { templateAgent: 'agent', bind: { channel: 'always' }, trigger: { channel: 'always' }, ...(options.idleTtlMs ? { idleTtlMs: options.idleTtlMs } : {}) }, + } as never, + }); + const gate = (f.framework as unknown as { eventGate: EventGate }).eventGate; + const incoming = (messageId: string) => f.send('channels/incoming', { + messages: [{ channelId, messageId, timestamp: TS, author: { id: 'U1', name: 'antra' }, tags: ['chat:from-human'], content: [{ type: 'text', text: 'hello' }] }], + }); + try { + await f.send('channels/register', { channels: [{ id: channelId, type: 'discord', label: channelId.split(':').pop(), initiallyOpen: true, metadata: { channelType: 'guild_text' } }] }); + await eventually(() => (f.framework as unknown as { channelRegistry: { getDesiredState(s: string, c: string): string | undefined } }).channelRegistry.getDesiredState('editor', channelId) === 'open', 'the channel is open'); + } catch (err) { + await f.close(); + throw err; + } + return { f, gate, incoming }; +} + function makeGate(now?: () => number) { const inferenceRequests: Array<{ agentName: string; reason: string; source: string }> = []; const traces: string[] = []; @@ -171,17 +199,15 @@ describe("EventGate sleep: its wake, a stop and a fork's end", () => { } }); - it("agentGone() ends that agent's sleep and self-wake together, and leaves another agent's", () => { + it("agentGone() cancels that agent's self-wake, 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('fork'), "the gone agent's 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 { @@ -189,20 +215,6 @@ describe("EventGate sleep: its wake, a stop and a fork's end", () => { } }); - 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(); @@ -233,9 +245,10 @@ describe("EventGate sleep: its wake, a stop and a fork's end", () => { } }); - it("an ephemeral run's sleep ends when the run does", async () => { + it("an ephemeral run's sleep and wake are refused, and its end tells the gate", 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: 'tool_use', id: 'w1', name: 'wake', input: {} }], 'tool_use')); membrane.pushResponse(createMockResponse([{ type: 'text', text: 'done' }])); const framework = await AgentFramework.create({ storePath: join(TMP_DIR, 'store-ephemeral'), @@ -246,26 +259,98 @@ describe("EventGate sleep: its wake, a stop and a fork's end", () => { }); 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'); - }); + const calls = gateToolCalls(framework); try { framework.start(); + // The primary sleeps, so a run's sleep would replace its note and a + // run's wake would end it. + gate.setSleep(3600, 'resting', 'agent'); 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"); + assert.deepStrictEqual(calls.refused, [ + 'sleep: Nothing was set: a sleep quiets the whole agent, not one run.', + "wake: Nothing was changed: a sleep is the agent's, so this run can't end one.", + ]); + assert.deepStrictEqual(calls.started, [], 'refused before the tools run'); + assert.strictEqual(gate.getSleepState()?.note, 'resting', "the primary's sleep stands, neither replaced nor ended"); + assert.deepStrictEqual(gone, ['worker'], "the run's end told the gate, which cancels its self-wake"); } finally { - // So that a run where the sleep outlives the run doesn't hold the - // file's process for the sleep's hour. + // So that a run where the sleep takes doesn't hold the file's process + // for its hour. gate.clearSleep(); await framework.stop(); } }); + it("the primary's sleep still silences a message to its conversation fork", async () => { + // The gate decides a channel message's wake before it's routed to a fork, + // so the primary's sleep quiets every conversation of the agent. + const CH = 'discord:g1:dm'; + const { f, gate, incoming } = await forkFixture(CH); + const router = () => (f.framework as unknown as { getConversationRouter(): { getBinding(channel: ChannelRef): { agentName: string } | undefined } }).getConversationRouter(); + try { + gate.setSleep(3600, 'resting', 'agent'); + const before = f.membrane.calls.length; + await incoming('m1'); + await f.framework.runUntilIdle(); + assert.ok(router().getBinding({ serverId: 'editor', channelId: CH }) !== undefined, 'the message reached a conversation fork'); + assert.strictEqual(f.membrane.calls.length, before, 'no turn while the primary sleeps'); + assert.ok((gate.getSleepState()?.suppressed ?? 0) >= 1, 'the gate held its wake'); + + // Control: awake, the next message wakes the fork. + gate.clearSleep(); + f.membrane.pushResponse(ok()); + await incoming('m2'); + await f.framework.runUntilIdle(); + assert.strictEqual(f.membrane.calls.length, before + 1, 'the fork took a turn'); + } finally { + gate.clearSleep(); + await f.close(); + } + }); + + for (const [what, allowedTools, remedy] of [ + ['can call skip_reply is told its wake_in_seconds is the way back', 'all', ' To come back to this conversation later, call skip_reply with wake_in_seconds.'], + ["can't call skip_reply isn't pointed at it", ['sleep', 'wake'], ''], + ] as const) it(`a fork that ${what}, when its sleep is refused`, async () => { + const CH = 'discord:g1:dm2'; + const { f, gate, incoming } = await forkFixture(CH, { agents: [{ name: 'agent', model: 'test', systemPrompt: 'test', allowedTools }] }); + const calls = gateToolCalls(f.framework); + try { + await incoming('m1'); // the fork's first turn answers with the fixture's 'ok' + await f.framework.runUntilIdle(); + f.membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 's1', name: 'sleep', input: { seconds: 3600, announce: false } }], 'tool_use')); + f.membrane.pushResponse(ok()); + await incoming('m2'); + await f.framework.runUntilIdle(); + assert.deepStrictEqual(calls.refused, [`sleep: Nothing was set: a sleep quiets the whole agent, not one conversation.${remedy}`]); + } finally { + gate.clearSleep(); + await f.close(); + } + }); + + it("a closing fork isn't pointed at skip_reply when its sleep is refused, since its self-wake would end with it", async () => { + const CH = 'discord:g1:dm3'; + const IDLE_TTL_MS = 60_000; + const { f, gate, incoming } = await forkFixture(CH, { idleTtlMs: IDLE_TTL_MS }); + const calls = gateToolCalls(f.framework); + try { + await incoming('m1'); + await f.framework.runUntilIdle(); // the fork's first turn answers with the fixture's 'ok' + forceExpiry(f.framework, { serverId: 'editor', channelId: CH }, IDLE_TTL_MS); // idle past the TTL, and the sweep due + f.membrane.pushResponse(createMockResponse([{ type: 'tool_use', id: 's1', name: 'sleep', input: { seconds: 3600, announce: false } }], 'tool_use')); + f.membrane.pushResponse(ok()); + f.framework.pushEvent({ type: 'external-message', source: 'test', content: 'tick', metadata: {} } as never); + await f.framework.runUntilIdle(); + assert.deepStrictEqual(calls.refused, ['sleep: Nothing was set: a sleep quiets the whole agent, not one conversation.'], 'its closure turn is told no way back'); + } finally { + gate.clearSleep(); + await f.close(); + } + }); + 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. diff --git a/test/helpers/force-expiry.ts b/test/helpers/force-expiry.ts new file mode 100644 index 000000000..dbc5d686d --- /dev/null +++ b/test/helpers/force-expiry.ts @@ -0,0 +1,12 @@ +import assert from 'node:assert/strict'; +import type { AgentFramework } from '../../src/index.js'; +import type { ChannelRef } from '../../src/mcpl/channel-registry.js'; + +/** 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. */ +export function forceExpiry(framework: AgentFramework, channel: ChannelRef, idleTtlMs: number): void { + const binding = framework.getConversationRouter()!.getBinding(channel); + assert.ok(binding, `binding for ${channel.channelId} should exist before forcing expiry`); + binding!.lastActivity = Date.now() - idleTtlMs - 1_000; + (framework as unknown as { lastConversationSweep: number }).lastConversationSweep = 0; +} diff --git a/test/helpers/gate-tool-calls.ts b/test/helpers/gate-tool-calls.ts new file mode 100644 index 000000000..621240530 --- /dev/null +++ b/test/helpers/gate-tool-calls.ts @@ -0,0 +1,16 @@ +/** What the gate's sleep and wake tools answered, as the framework queued + * each result: `tool: error` for a refused call, and the calls that started + * (a refusal starts none). */ +export function gateToolCalls(framework: { onTrace(listener: (event: unknown) => void): unknown }): { refused: string[]; succeeded: string[]; started: string[] } { + const calls = { refused: [] as string[], succeeded: [] as string[], started: [] as string[] }; + framework.onTrace((e) => { + const traced = e as { type?: string; module?: string; tool?: string; processEvent?: { type?: string; toolName?: string; moduleName?: string; result?: { refused?: boolean; success?: boolean; error?: string } } }; + if (traced.type === 'tool:started' && traced.module === 'gate') calls.started.push(String(traced.tool)); + const ev = traced.processEvent; + if (traced.type === 'process:received' && ev?.type === 'tool-result' && ev.moduleName === 'gate' && (ev.toolName === 'sleep' || ev.toolName === 'wake')) { + if (ev.result?.refused === true) calls.refused.push(`${ev.toolName}: ${ev.result.error}`); + else if (ev.result?.success === true) calls.succeeded.push(String(ev.toolName)); + } + }); + return calls; +} diff --git a/test/helpers/gone-names.ts b/test/helpers/gone-names.ts index 8c9c53c3d..6b373efbc 100644 --- a/test/helpers/gone-names.ts +++ b/test/helpers/gone-names.ts @@ -1,7 +1,7 @@ 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. */ + * whose self-wake rule the gate's own tests check. */ export function goneNames(gate: EventGate): string[] { const gone: string[] = []; const agentGone = gate.agentGone.bind(gate); diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index eb256bf5f..0d4f4d6c5 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -1790,6 +1790,7 @@ describe('present while acting', () => { framework.emitTrace = () => {}; framework.pushEvent = () => {}; framework.conversationAgentHomes = new Map(); + framework.ephemeralRuns = new Map(); framework.lastAnnouncedLocus = new Map(); framework.activeTurnTriggers = new Map(); framework.openLogical = new Map(); diff --git a/test/send-into-closed-channel.test.ts b/test/send-into-closed-channel.test.ts index ad1e8d309..e46e4f412 100644 --- a/test/send-into-closed-channel.test.ts +++ b/test/send-into-closed-channel.test.ts @@ -145,8 +145,10 @@ async function harness(opts: { sharedIdFirst?: boolean } = {}) { setStateJson: (_id: string, value: unknown) => { frameworkState = value; }, }; framework.agents = new Map([['sol', agentStub()]]); - // Conversation agents by their home channel; sol isn't one. + // Conversation agents by their home channel, and ephemeral runs by name; + // sol is neither. framework.conversationAgentHomes = new Map(); + framework.ephemeralRuns = new Map(); const windowMessages: string[] = []; /** Whose window each of those was addressed to: addMessage's forAgent, or * undefined for the primary agent's. */ @@ -857,11 +859,8 @@ test("a sleep announced into a channel the agent or a module closed says the age assert.deepEqual(published.splice(0), [{ channelId: 'discord:g1:343434', text: quiet }]); routing.locus = ref(channelId); - // A conversation agent's announcement reads the same: the words follow the channel. - framework.conversationAgentHomes.set('sol', channelId); - slept = await sleep({ seconds: 600 }); - framework.conversationAgentHomes.delete('sol'); - assert.deepEqual(published.splice(0), [{ channelId, text: addressed }]); + // A conversation agent announces nothing: its sleep is refused before the + // announcement (dispatchSleepToolCall; event-gate-sleep.test.ts). // Hybrid routing announces to the locus as locus mode does. framework.agents.set('sol', agentStub({ proseRouting: 'hybrid' }));