diff --git a/changelog.d/551-ephemeral-no-global-fallback.fixed.md b/changelog.d/551-ephemeral-no-global-fallback.fixed.md new file mode 100644 index 000000000..87c1d839a --- /dev/null +++ b/changelog.d/551-ephemeral-no-global-fallback.fixed.md @@ -0,0 +1,23 @@ +- An ephemeral run (a subagent) no longer posts its plain prose to whichever + channel last received a message (#551). That channel, the process-global + fallback, was the locus of any `locus` or `hybrid` run with no channel of its + own, so delegated work, possibly drawn from a DM, could land in an unrelated, + possibly public channel. No ephemeral run takes the global fallback now, + whatever its prose routing: + - a `locus` or `hybrid` run's plain prose with no channel of its own is + returned to its caller as before and published nowhere; a sleep + announcement it makes is published nowhere too. Neither counts as a failed + delivery, so neither writes a `[send-undeliverable]` notice, an + `mcpl:speech-route-failed` trace or a log line; + - `channel_publish` without a `channelId` is refused, with words saying to + name the channel; + - its channel context has no `incoming` message, and no default outgoing + channel unless it opened one itself; + - a refusal of its inference adds no reaction anywhere, and its + `failures.log` record names no `locus`. + + A channel the run opens itself, and a destination it names in its prose + (`hybrid`'s `>>>`, `explicit`'s `>>`), still route as before. Residents and + conversation forks keep the fallback. `channel_publish`'s description now + says where an omitted `channelId` goes: the conversation's bound channel, + else the most recent incoming one, and nowhere for a subagent. diff --git a/src/framework.ts b/src/framework.ts index 5898ae2e6..1a7eac698 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -2160,8 +2160,10 @@ export class AgentFramework { private ephemeralRuns: Map = new Map(); /** Ephemeral namespaces/names are single-generation for this framework * lifetime. Reusing one while an old stream drains would let name-keyed - * liveness/settlement state cross generations. */ - private usedEphemeralAgentNames: Set = new Set(); + * liveness/settlement state cross generations. Each name records what + * reserved it: an ephemeral run (createEphemeralAgent), which takes no + * global fallback even after its release, or a conversation fork. */ + private usedEphemeralAgentNames: Map = new Map(); /** One-shot generation tickets minted by createEphemeralAgent. */ private ephemeralCandidates: WeakMap = new WeakMap(); /** Warn only once per Agent instance after utilities are registered. */ @@ -7684,7 +7686,7 @@ export class AgentFramework { if (this.usedEphemeralAgentNames.has(config.name) || this.agents.has(config.name)) { throw new Error(`Ephemeral agent name \"${config.name}\" is already registered or has been used in this framework`); } - this.usedEphemeralAgentNames.add(config.name); + this.usedEphemeralAgentNames.set(config.name, 'run'); try { const namespace = `subagent/${config.name}`; @@ -8459,13 +8461,17 @@ export class AgentFramework { /** * Mark an inference refusal visibly: react on the message that holds the - * conversational locus (the most recent incoming channel message) with a - * category-specific emoji. Best-effort — failures are logged, never thrown, - * and non-Discord loci are silently skipped. + * refused agent's conversational locus (the most recent incoming channel + * message; an ephemeral run has none, so a subagent's refusal marks no + * message) with a category-specific emoji. Best-effort — failures are + * logged, never thrown, and non-Discord loci are silently skipped. */ private async reactToRefusal(agentName: string, category: string): Promise { try { - const incoming = this.channelRegistry?.buildChannelContext()?.incoming; + // The refused agent's own view of the latest inbound message: an + // ephemeral run has none (takesGlobalFallback), so a subagent's refusal + // marks no one else's channel. + const incoming = this.channelRegistry?.buildChannelContext(agentName)?.incoming; if (!incoming) return; // incoming.channelId is the MCPL composite id ("discord::"); // the reaction tool wants the raw Discord channel (or thread) id — the @@ -11601,7 +11607,7 @@ export class AgentFramework { if (this.usedEphemeralAgentNames.has(name) || this.agents.has(name)) { throw new Error(`Conversation agent name "${name}" is already registered or reserved`); } - this.usedEphemeralAgentNames.add(name); + this.usedEphemeralAgentNames.set(name, 'conversation'); try { const contextManager = await ContextManager.open({ @@ -13184,7 +13190,8 @@ export class AgentFramework { // Ambient chatter must not outrank an addressed message just by being // newest (2026-07-21 Cairn lounge misroute, turn-start variant). // Non-channel wakes (heartbeats, module events, reactions — which never - // carry channelId) leave both undefined → global fallback. + // carry channelId) leave both undefined → global fallback, which an + // ephemeral run never takes (takesGlobalFallback). // Track the winning REQUEST, not just its channel: channel, addressed // and counterparty must come from the same message, or a batch of // "ambient from A, then addressed from B" would report B's channel @@ -15604,7 +15611,8 @@ export class AgentFramework { // a previous turn's channel. routing.triggerChannel = trigger?.channelId; // FREEZE the turn's outbound locus now (turn-frozen routing): the home - // channel for forks, else the trigger channel, else the global default. + // channel for forks, else the trigger channel, else the global default, + // which an ephemeral run never takes (takesGlobalFallback). // Mid-turn injections must not redirect the agent's plain prose // (ambient chatter and reactions hijacked it before, 2026-07-21), and // the eager snapshot protects heartbeat turns from a moving @@ -17320,7 +17328,7 @@ export class AgentFramework { // one channel must not land its postscript in another. const locus = resolveTurnLocus(); console.error( - `[routing] ${agent.name}: tool-call turn [${toolNames.join(', ')}] -> routing ${segments.length} ${liveProseRouting ? 'trailing ' : ''}prose segment(s) -> ${locus ?? '(default)'}`, + `[routing] ${agent.name}: tool-call turn [${toolNames.join(', ')}] -> routing ${segments.length} ${liveProseRouting ? 'trailing ' : ''}prose segment(s) -> ${locus ?? '(none)'}`, ); // Deliver sequentially (await each) so the segments land in order. for (const seg of segments) { @@ -21031,7 +21039,8 @@ export class AgentFramework { kind: 'refusal', category, streak, - locus: this.channelRegistry?.buildChannelContext()?.incoming?.channelId ?? null, + // The refused agent's own view: a subagent's refusal names no channel. + locus: this.channelRegistry?.buildChannelContext(agentName)?.incoming?.channelId ?? null, tokens: tokens ?? null, }); if (streak >= 2) { @@ -21492,11 +21501,17 @@ export class AgentFramework { // a later start of the turn is told where its unprefixed speech lands // unless a hybrid `>>>` destination takes it. Empty for a heartbeat / // no-trigger turn until its locus is resolved, so that resolution - // takes the global fallback; the turn then keeps that channel. + // takes the global fallback (none for an ephemeral run); the turn then + // keeps that channel. activeChannelResolver: (agentName) => { const routing = this.turnRouting.get(agentName); return routing?.locus ?? routing?.triggerChannel; }, + // An ephemeral run (a subagent) never takes the process-global + // fallback: its prose is its caller's result, and the latest inbound + // message's channel is nobody's choice for it (#551). By name, so a + // speech dispatch that comes after the run's release still counts. + takesGlobalFallback: (conversationId) => this.usedEphemeralAgentNames.get(conversationId) !== 'run', eventGatePresent: this.eventGate !== null, // A text-only turn whose speech couldn't be delivered must not vanish // silently: record a `[discord-send-failed]` marker in chronicle so the diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 521e4d50e..5a0bb585c 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -370,11 +370,11 @@ const CHANNEL_TOOL_DEFINITIONS: ToolDefinition[] = [ }, { name: 'channel_publish', - description: 'Publish a message to a channel. If channelId is omitted, publishes to the most recent incoming channel.', + description: 'Publish a message to a channel. If channelId is omitted, publishes to the channel this conversation is bound to, if any, else to the most recent incoming channel. A subagent has neither, and must name one.', inputSchema: { type: 'object' as const, properties: { - channelId: { type: 'string', description: 'ID of the channel to publish to (defaults to the most recent incoming channel)' }, + channelId: { type: 'string', description: 'ID of the channel to publish to (defaults to the conversation\'s bound channel, else the most recent incoming channel; required of a subagent)' }, content: { type: 'string', description: 'Text content to publish' }, text: { type: 'string', description: 'Alias for content' }, }, @@ -591,8 +591,10 @@ interface ChannelRegistryOptions { /** * Called when a text-only turn's speech could NOT be delivered to its * conversational locus — no locus, an unregistered channel, a missing - * server, or the server reporting `delivered: false`. The host wires this - * to drop a `[discord-send-failed]` marker into chronicle so the failure is + * server, or the server reporting `delivered: false`. Not called for an + * ephemeral run's speech with no locus, which is its caller's result + * (`takesGlobalFallback`). The host wires this to drop a + * `[discord-send-failed]` marker into chronicle so the failure is * visible to the agent (and operator) rather than silently lost. Must not * itself trigger inference (avoid wake loops). */ @@ -652,6 +654,18 @@ interface ChannelRegistryOptions { * `defaultPublishChannel`. */ activeChannelResolver?: (agentName: string) => string | undefined; + /** + * Whether a conversation takes the process-global fallback, the channel of + * the most recent delivered inbound message: as where its plain speech lands + * when it has no home and no channel of its own this turn, as its channel + * context's `incoming` and default outgoing channel, and as + * `channel_publish`'s target when it names none. False for an ephemeral run + * (a subagent): its prose is its caller's result, so a channel that only the + * latest inbound message picked is never its destination, and its plain + * speech with no channel of its own is undelivered rather than a failed + * delivery. Absent, every conversation takes it. + */ + takesGlobalFallback?: (conversationId: string) => boolean; /** Whether this framework has EventGate and can honor skip_reply self-wakes. */ eventGatePresent?: boolean; } @@ -696,6 +710,7 @@ export class ChannelRegistry { }) => void; private homeChannelResolver?: (agentName: string) => string | undefined; private activeChannelResolver?: (agentName: string) => string | undefined; + private takesGlobalFallback?: (conversationId: string) => boolean; private store?: JsStore; /** Registered channels, keyed by `{serverId}:{channelId}`. */ @@ -872,6 +887,7 @@ export class ChannelRegistry { this.onChannelAutoOpened = options?.onChannelAutoOpened; this.homeChannelResolver = options?.homeChannelResolver; this.activeChannelResolver = options?.activeChannelResolver; + this.takesGlobalFallback = options?.takesGlobalFallback; this.store = options?.store; this.initializeLifecycleStore(); this.initializeLabelHistoryStore(); @@ -1631,7 +1647,7 @@ export class ChannelRegistry { }); case 'channel_publish': - return this.handleToolPublish(input as { channelId: string; content: string }); + return this.handleToolPublish(input as { channelId: string; content: string }, origin); case 'think': return this.handleToolThink(input as { content?: string }); @@ -1666,9 +1682,15 @@ export class ChannelRegistry { // (heartbeats). Without this, a fork // was advertised the global locus but published somewhere else; and a trunk // agent was told the wrong channel under concurrency. - const home = agentName ? this.homeChannelResolver?.(agentName) : undefined; - const active = agentName ? this.activeChannelResolver?.(agentName) : undefined; - const outgoing = home ?? active ?? this.defaultPublishChannel; + // Each lookup is keyed on whether a name was given, not its truthiness: + // '' is a name an ephemeral run can have, and resolveLocus reads it too. + const home = agentName !== undefined ? this.homeChannelResolver?.(agentName) : undefined; + const active = agentName !== undefined ? this.activeChannelResolver?.(agentName) : undefined; + // An ephemeral run takes no global fallback (takesGlobalFallback): neither + // the latest inbound message's channel as its outgoing one, nor that + // message as the one it is replying to. + const fallback = agentName !== undefined ? this.globalFallbackFor(agentName) : this.defaultPublishChannel; + const outgoing = home ?? active ?? fallback; if (openChannels.length === 0 && !outgoing) { return undefined; @@ -1679,9 +1701,9 @@ export class ChannelRegistry { // Incoming: the most-recent inbound message (what the agent is replying to). // Left process-global — per-channel inbound tracking (the right messageId for // a fork's own channel) is a separate concern from the outbound routing fix. - if (this.defaultPublishChannel && this.defaultPublishMessageId) { + if (fallback && this.defaultPublishMessageId) { context.incoming = { - channelId: this.defaultPublishChannel, + channelId: fallback, messageId: this.defaultPublishMessageId, threadId: this.defaultPublishThreadId, }; @@ -3512,7 +3534,12 @@ export class ChannelRegistry { resolveLocus(conversationId: string): string | null { const home = this.homeChannelResolver?.(conversationId); - return home ?? this.activeChannelResolver?.(conversationId) ?? this.defaultPublishChannel ?? null; + return home ?? this.activeChannelResolver?.(conversationId) ?? this.globalFallbackFor(conversationId); + } + + /** The process-global fallback channel for this conversation, or null when it takes none (takesGlobalFallback). */ + private globalFallbackFor(conversationId: string): string | null { + return this.takesGlobalFallback?.(conversationId) === false ? null : this.defaultPublishChannel; } /** @@ -3716,7 +3743,9 @@ export class ChannelRegistry { * Plain speech for one turn segment: publish `text` to the turn's frozen * locus and report the PublishOutcome. A failure (or an uncertain outcome) * is surfaced as a trace and through onRouteFailure, so the agent learns - * its words did not (or may not have) reached anyone. + * its words did not (or may not have) reached anyone. An ephemeral run's + * speech with no locus is the exception: it is its caller's result, so it + * returns as not delivered and surfaces nothing. */ async deliverSpeech( conversationId: string, @@ -3727,7 +3756,9 @@ export class ChannelRegistry { * dispatch resolving live can read the NEXT turn's trigger state or a * stale global default and land the reply in the wrong channel (item-3, * PR #32, and the 2026-07-22 Sol DM misroute). Explicit `null` means - * "this turn is pinned to no locus": fail loudly rather than guess. */ + * "this turn is pinned to no locus": fail loudly rather than guess, + * except for an ephemeral run, whose speech is then its caller's + * result. */ locusChannelId: string | null, /** `heldOnFailure`: the caller holds a `failed` delivery's words as a * draft with its own notice, so no failure marker is written for it. @@ -3762,6 +3793,12 @@ export class ChannelRegistry { return fail(null, reason, { status: 'failed', reason, at: Date.now() }); } if (!locusChannelId) { + // An ephemeral run with no channel of its own: its plain speech is its + // caller's result (takesGlobalFallback). It returns as not delivered, + // with its own reason, and writes no line, trace or notice. + if (this.takesGlobalFallback?.(conversationId) === false) { + return { status: 'failed', reason: "an ephemeral run's plain speech with no channel of its own is its caller's result", at: Date.now() }; + } // The turn froze with no locus (no home, no triggering channel, no // global inbound ever seen) — the agent was told its prose stays in // the archive; honor that. @@ -3807,7 +3844,10 @@ export class ChannelRegistry { }; } - private async handleToolPublish(input: { channelId?: string; content?: string; text?: string }): Promise { + private async handleToolPublish( + input: { channelId?: string; content?: string; text?: string }, + origin?: ChannelToolOrigin, + ): Promise { // Resolve content: accept both `content` and `text` (backward compat) const messageText = input.content ?? input.text; if (!messageText) { @@ -3818,12 +3858,17 @@ export class ChannelRegistry { }; } - // Resolve channelId: default to most recent incoming channel - const channelId = input.channelId ?? this.defaultPublishChannel; + // Resolve channelId: default to the most recent incoming channel, which an + // ephemeral run never takes (takesGlobalFallback), so it names one. + const caller = origin?.kind === 'agent' ? origin.agentName : undefined; + const channelId = input.channelId ?? (caller !== undefined ? this.globalFallbackFor(caller) : this.defaultPublishChannel); if (!channelId) { return { success: false, - error: 'No channelId specified and no default channel available', + error: caller !== undefined && this.takesGlobalFallback?.(caller) === false + ? 'Nothing was published: name the channel with channelId. A subagent has no default channel, since its plain ' + + 'speech is returned to whoever called it, so channel_publish sends only to a channel it names.' + : 'No channelId specified and no default channel available', isError: true, }; } diff --git a/src/types/agent.ts b/src/types/agent.ts index 1e224c81c..58c2bebac 100644 --- a/src/types/agent.ts +++ b/src/types/agent.ts @@ -382,7 +382,8 @@ export interface InferenceRequest { * replies in the channel it is answering — not the process-global most-recent- * inbound locus, which a concurrent message elsewhere can hijack (item-3 * redux). Undefined for non-channel wakes (heartbeat, timers, module events), - * which correctly fall back to the global default channel. + * which correctly fall back to the global default channel, except an + * ephemeral run's (a subagent's), which takes no global fallback. */ channelId?: string; /** diff --git a/test/ephemeral-no-global-fallback.test.ts b/test/ephemeral-no-global-fallback.test.ts new file mode 100644 index 000000000..ce70c2b1d --- /dev/null +++ b/test/ephemeral-no-global-fallback.test.ts @@ -0,0 +1,286 @@ +/** + * An ephemeral run (a subagent) takes no process-global fallback (#551). The + * fallback is the channel of the most recent delivered inbound message: a + * resident's turn with no channel of its own speaks there. A subagent's plain + * prose is its caller's result, so that channel is never where it lands, + * never its channel context's default outgoing channel or `incoming` message, + * never `channel_publish`'s default, and never where its refusal is marked or + * logged. With no channel of its own, its prose is not delivered, and nothing + * reports a failure. A channel the run opens itself is still its own. + * + * Harness: the RFC-006 loopback MCPL server and a mocked model; the event + * loop is driven explicitly. + */ +import { describe, it } from 'node:test'; +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'; + +type Fixture = Awaited>; +type Registry = { + getDefaultPublishChannel(): string | null; + getDesiredState(serverId: string, channelId: string): string | undefined; + resolveLocus(conversationId: string): string | null; + buildChannelContext(agentName?: string): { incoming?: { channelId: string }; defaultOutgoing?: { channelId: string } } | undefined; + routeSpeech(conversationId: string, text: string, locus: string | null): Promise; + deliverSpeech(conversationId: string, text: string, locus: string | null, opts?: unknown): Promise; + handleChannelToolCall(toolName: string, input: unknown, origin?: unknown): Promise<{ success: boolean; error?: string }>; +}; +const registry = (f: Fixture) => (f.framework as unknown as { channelRegistry: Registry }).channelRegistry; + +const BUSY = 'discord:g1:busy'; +const SIDE = 'discord:g1:side'; +const text = (t: string) => createMockResponse([{ type: 'text', text: t }]); +const publishedText = (f: Fixture) => JSON.stringify(f.published); + +/** Open `busy`, and let a resident's reply there make it the latest inbound channel: the global fallback. */ +async function busyFallback(f: Fixture): Promise { + await f.send('channels/register', { + channels: [ + { id: BUSY, type: 'discord', label: 'busy', initiallyOpen: true, metadata: { channelType: 'guild_text' } }, + { id: SIDE, type: 'discord', label: 'side', metadata: { channelType: 'guild_text' } }, + ], + }); + await eventually(() => registry(f).getDesiredState('editor', BUSY) === 'open', 'busy is open'); + await f.send('channels/incoming', { + messages: [{ + channelId: BUSY, messageId: 'b1', timestamp: TS, author: { id: 'U1', name: 'antra' }, + tags: ['chat:mention', 'chat:addressed', 'chat:from-human'], content: [{ type: 'text', text: 'hello agent' }], + }], + }); + await f.framework.runUntilIdle(); // the resident answers with the fixture's scripted 'ok' + assert.equal(registry(f).getDefaultPublishChannel(), BUSY, 'control: busy is the global fallback'); +} + +/** + * Run one subagent to completion and return its result. A run's final speech + * is routed after the run settles, and may be in flight after its release, + * so with `speech: true` this waits for the run's speech to reach + * deliverSpeech and for each delivery to settle: a post the run made is then + * at the server, and one it didn't make can't still arrive. + */ +async function runSubagent( + f: Fixture, name: string, proseRouting: 'locus' | 'hybrid', speech: boolean, ...responses: ReturnType[] +) { + const reg = registry(f); + const deliver = reg.deliverSpeech.bind(reg); + const deliveries: Promise[] = []; + reg.deliverSpeech = (...args) => { + const done = deliver(...args); + if (args[0] === name) deliveries.push(done); + return done; + }; + try { + const created = await f.framework.createEphemeralAgent({ + name, model: 'test', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting, + }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Find it and report back.' }]); + for (const r of responses) f.membrane.pushResponse(r); + const run = f.framework.runEphemeralToCompletion(created.agent, created.contextManager); + await f.framework.runUntilIdle(); + const result = await run; + await f.framework.runUntilIdle(); + if (speech) { + await eventually(() => deliveries.length > 0, `${name}'s speech reached deliverSpeech`); + await settles(Promise.all(deliveries), `${name}'s speech deliveries`); + } + return result; + } finally { + reg.deliverSpeech = deliver; + } +} + +/** What the run left in its own namespace. */ +async function archived(f: Fixture, name: string): Promise { + const archive = await ContextManager.open({ + store: f.framework.getStore(), namespace: `subagent/${name}`, isolate: true, strategy: new PassthroughStrategy(), + }); + return JSON.stringify(archive.getAllMessages()); +} + +function speechTraces(f: Fixture) { + const seen: Array<{ type: string; conversationId?: string; channelId?: string }> = []; + const off = f.framework.onTrace((t) => { + const type = String((t as { type: string }).type); + if (type === 'mcpl:speech-routed' || type === 'mcpl:speech-route-failed') seen.push(t as never); + }); + return { seen, off }; +} + +describe('an ephemeral run takes no global fallback (#551)', () => { + for (const mode of ['locus', 'hybrid'] as const) { + it(`a ${mode}-mode subagent's final prose is returned to its caller, not posted to the latest inbound channel`, async () => { + const f = await fixture(); + const t = speechTraces(f); + try { + await busyFallback(f); + const result = await runSubagent(f, `sub-${mode}`, mode, true, text('subagent private findings')); + + assert.match(result.speech, /subagent private findings/, 'the caller gets the prose'); + assert.ok(!publishedText(f).includes('subagent private findings'), `not published: ${publishedText(f)}`); + const own = t.seen.filter((x) => x.conversationId === `sub-${mode}`); + assert.deepEqual(own, [], 'neither routed nor counted as a failed route'); + // Control: a name the framework didn't reserve for a run, with no + // turn of its own (a resident's heartbeat, say), still resolves the + // fallback. + assert.equal(registry(f).resolveLocus('resident-with-no-turn'), BUSY, 'control: the fallback is kept for others'); + } finally { t.off(); await f.close(); } + }); + } + + it('channel_publish with no channelId is refused for a subagent, and publishes to a channel it names; a resident keeps the default', async () => { + const f = await fixture(); + try { + await busyFallback(f); + const tool = (id: string, input: Record) => + createMockResponse([{ type: 'tool_use', id, name: 'channel_publish', input }], 'tool_use'); + await runSubagent(f, 'sub-publish', 'locus', false, + tool('p1', { content: 'unnamed post' }), + tool('p2', { channelId: BUSY, content: 'named post' }), + text('done')); + + // Both tool calls finished inside the run's turn, before it settled. + const posted = publishedText(f); + assert.ok(!posted.includes('unnamed post'), `the unnamed post went nowhere: ${posted}`); + assert.ok(posted.includes('named post'), 'a channel the subagent names is published to'); + const archive = await archived(f, 'sub-publish'); + assert.ok(archive.includes('name the channel with channelId. A subagent has no default channel'), 'the refusal says to name the channel'); + + // Control: a resident's channel_publish with no channelId still goes to the latest inbound channel. + const resident = await registry(f).handleChannelToolCall('channel_publish', { content: 'resident unnamed post' }, { kind: 'agent', agentName: 'agent' }); + assert.equal(resident.success, true, resident.error); + const post = f.published.find((p) => JSON.stringify(p).includes('resident unnamed post')) as { channelId?: string } | undefined; + assert.equal(post?.channelId, BUSY, 'control: the resident\'s unnamed post went to busy'); + } finally { await f.close(); } + }); + + it('a subagent\'s channel context names no default outgoing channel and no incoming message, a subagent named \'\' included; a resident\'s does', async () => { + const f = await fixture(); + try { + await busyFallback(f); + for (const name of ['sub-context', '']) { + await f.framework.createEphemeralAgent({ name, model: 'test', systemPrompt: 'Do the task.', proseRouting: 'locus' }); + const sub = registry(f).buildChannelContext(name); + assert.ok(sub, `${JSON.stringify(name)} has a channel context: busy is open`); + assert.equal(sub.defaultOutgoing, undefined, JSON.stringify(name)); + assert.equal(sub.incoming, undefined, JSON.stringify(name)); + } + const resident = registry(f).buildChannelContext('agent'); + assert.equal(resident?.defaultOutgoing?.channelId, BUSY, 'control'); + assert.equal(resident?.incoming?.channelId, BUSY, 'control'); + } finally { await f.close(); } + }); + + it('a subagent named \'\' that opened a channel is told that channel, where its speech goes', async () => { + const f = await fixture(); + try { + await busyFallback(f); + await f.framework.createEphemeralAgent({ name: '', model: 'test', systemPrompt: 'Do the task.', proseRouting: 'locus' }); + // The run's turn opened side: its routing state names it, as + // channel_open's result leaves it. + (f.framework as unknown as { turnRouting: Map }).turnRouting.set('', { locus: SIDE }); + + assert.equal(registry(f).resolveLocus(''), SIDE, 'control: its speech goes to side'); + assert.equal(registry(f).buildChannelContext('')?.defaultOutgoing?.channelId, SIDE, 'and its channel context says so'); + } finally { await f.close(); } + }); + + it('a subagent\'s refusal is marked on no one\'s message; a resident\'s still is', async () => { + const calls: Array> = []; + const f = await fixture({ toolsCall: (params) => { calls.push(params); return { content: [] }; } }); + try { + await busyFallback(f); + await f.framework.createEphemeralAgent({ name: 'sub-refused', model: 'test', systemPrompt: 'Do the task.', proseRouting: 'locus' }); + const react = (name: string) => + (f.framework as unknown as { reactToRefusal(name: string, category: string): Promise }).reactToRefusal(name, 'other'); + + await react('sub-refused'); + assert.equal(calls.length, 0, 'no reaction for the subagent\'s refusal'); + await react('agent'); + await eventually(() => calls.length === 1, 'control: the resident\'s refusal is marked'); + assert.equal(calls[0]!.name, 'add_reaction'); + } finally { await f.close(); } + }); + + it('a subagent\'s refusal is logged with no locus; a resident\'s with the latest inbound channel', async (t) => { + const f = await fixture(); + try { + await busyFallback(f); + await f.framework.createEphemeralAgent({ name: 'sub-logged', model: 'test', systemPrompt: 'Do the task.', proseRouting: 'locus' }); + const fw = f.framework as unknown as { + logFailure(record: Record): void; + noteRefusal(name: string, category: string): number; + }; + const records: Array> = []; + t.mock.method(fw, 'logFailure', (record: Record) => { records.push(record); }); + + fw.noteRefusal('sub-logged', 'other'); + fw.noteRefusal('agent', 'other'); + assert.deepEqual(records.map((r) => [r.agent, r.kind, r.locus]), [['sub-logged', 'refusal', null], ['agent', 'refusal', BUSY]]); + } finally { await f.close(); } + }); + + it('after the run is released, its name still takes no fallback, and a dispatch with no locus stays quiet', async () => { + const f = await fixture(); + const t = speechTraces(f); + try { + await busyFallback(f); + await runSubagent(f, 'sub-late', 'locus', true, text('first findings')); + assert.equal(f.framework.getAgent('sub-late'), null, 'released'); + + assert.equal(registry(f).resolveLocus('sub-late'), null); + assert.equal(await registry(f).routeSpeech('sub-late', 'late words', null), null); + assert.deepEqual(t.seen.filter((x) => x.conversationId === 'sub-late'), []); + } finally { t.off(); await f.close(); } + }); + + it('a conversation fork, whose name is reserved like a run\'s, keeps the fallback', async () => { + const f = await fixture({ framework: { conversations: { templateAgent: 'agent', bind: { channel: 'always' } } } as never }); + try { + await f.send('channels/register', { + channels: [{ id: BUSY, type: 'discord', label: 'busy', initiallyOpen: true, metadata: { channelType: 'guild_text' } }], + }); + await eventually(() => registry(f).getDesiredState('editor', BUSY) === 'open', 'busy is open'); + await f.send('channels/incoming', { + messages: [{ + channelId: BUSY, messageId: 'b1', timestamp: TS, author: { id: 'U1', name: 'antra' }, + tags: ['chat:ambient', 'chat:from-human'], content: [{ type: 'text', text: 'ambient words' }], + }], + }); + await f.framework.runUntilIdle(); + const fork = (f.framework as unknown as { getConversationRouter(): { getBinding(id: string): { agentName: string } | undefined } }) + .getConversationRouter().getBinding(BUSY)?.agentName; + assert.ok(fork, 'a conversation fork was spawned for busy'); + + assert.equal(registry(f).buildChannelContext(fork)?.incoming?.channelId, BUSY, 'its channel context still names the latest inbound message'); + } finally { await f.close(); } + }); + + it('a subagent\'s sleep announcement is not posted to the latest inbound channel', async () => { + const f = await fixture({ framework: { gate: { config: { default: 'always' } } } as never }); + const t = speechTraces(f); + 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')); + + 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(); } + }); + + it('a channel the subagent opens itself is still where its prose lands', async () => { + const f = await fixture(); + try { + await busyFallback(f); + await runSubagent(f, 'sub-side', 'locus', true, + createMockResponse([{ type: 'tool_use', id: 'o1', name: 'channel_open', input: { channelId: SIDE } }], 'tool_use'), + text('said in side')); + + const sideposts = f.published.filter((p) => JSON.stringify(p).includes('said in side')); + assert.equal(sideposts.length, 1, `published once: ${publishedText(f)}`); + assert.equal((sideposts[0] as { channelId?: string }).channelId, SIDE); + } finally { await f.close(); } + }); +}); diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index 7809a1dc2..d84be3ea5 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -83,6 +83,8 @@ class SenderModule implements Module { } const SEND_FAILED = '[discord-send-failed]'; +// What a turn with no locus is told instead. +const UNDELIVERABLE = '[send-undeliverable]'; const BOUNCED = 'was not delivered —'; const count = (messages: Array<{ content: unknown }>, needle: string) => messages.filter((m) => JSON.stringify(m.content).includes(needle)).length; @@ -302,35 +304,48 @@ describe('notices about an agent\'s own action', () => { assert.ok(errors.some((e) => e.includes('[route-failure] released-conversation has ended, and its reply had no channel to reach'))); }); - it('tells no one, logged, when an ephemeral run\'s final reply had no channel to reach, before or after its release', async (t) => { + it('treats an ephemeral run\'s final reply with no channel of its own as its caller\'s result: no failure, logged or told', async (t) => { const errors: string[] = []; t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); const speech = 'Worker speech with no channel to go to.'; const archived = await runEphemeral('locus', speech); - // The run settles before its final reply is routed, so its release and - // the reply's failure can come in either order: the notice lands in its - // window and is read at its release, or the failure finds it gone. Either - // way, a reply that had no channel to reach is passed on to no one. + // The run's prose is its caller's result (#551): with no channel of its + // own it is undelivered, not a failed delivery, before or after its + // release, so nothing is written, logged or passed on. assert.ok(archived.some((m) => m.participant === 'worker' && JSON.stringify(m.content).includes(speech)), 'the run\'s reply is archived'); + assert.equal(count(archived, UNDELIVERABLE), 0, 'no no-locus notice in its own window'); const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); assert.equal(count(messages, 'Your previous reply'), 0, 'the primary did not write that reply'); assert.equal(count(messages, 'A reply by worker'), 0); assert.ok(!errors.some((e) => /not a registered agent/.test(e))); - assert.ok(errors.some((e) => e.includes('[route-failure] worker has ended, and its reply had no channel to reach'))); + assert.ok(!errors.some((e) => e.includes('[route-failure] worker')), 'no route failure is logged'); + assert.ok(!errors.some((e) => e.includes('[routeSpeech] worker')), 'no "NOT routed" line'); assert.equal(internals(framework).deferredMessages.length, 0); }); - it('passes an ephemeral run\'s failed delivery to a channel on to the primary, by name', async () => { + it('never sends an ephemeral run\'s reply to the default channel, so nothing fails there', async (t) => { (internals(framework).channelRegistry as unknown as { defaultPublishChannel: string | null }).defaultPublishChannel = 'missing-channel'; - const speech = 'Worker speech for a channel that is not registered.'; + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + + await runEphemeral('locus', 'Worker speech that the default channel never gets.'); + + // The latest inbound channel is nobody's choice for a subagent (#551). + const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); + assert.equal(count(messages, SEND_FAILED), 0, 'the primary is told of no failure'); + assert.ok(!errors.some((e) => e.includes('missing-channel')), 'no delivery to the default channel was attempted'); + assert.equal(internals(framework).deferredMessages.length, 0); + }); - await runEphemeral('locus', speech); + it('passes an ended ephemeral run\'s failed delivery to a channel it was given on to the primary, by name', async () => { + await runEphemeral('locus', 'Worker speech.'); + // After the run's release, its speech is routed to a channel given + // directly, as a channel the run opened or named would be; the channel + // isn't registered, so the delivery fails. + await internals(framework).channelRegistry.routeSpeech('worker', 'Worker speech for a channel that is not registered.', 'missing-channel'); - // Whether the delivery fails before the run's release (its window's - // notice is passed on at release) or after it (the failure finds it - // gone), the primary is told once, by name. const { messages } = await framework.getAgent('primary')!.getContextManager().compile(); const named = messages.filter((m) => JSON.stringify(m.content).includes(SEND_FAILED)); assert.equal(named.length, 1); @@ -338,8 +353,7 @@ describe('notices about an agent\'s own action', () => { assert.equal(internals(framework).deferredMessages.length, 0); }); - it('still releases an ephemeral run when the primary cannot be told, logged', async (t) => { - (internals(framework).channelRegistry as unknown as { defaultPublishChannel: string | null }).defaultPublishChannel = 'missing-channel'; + it('logs it when the primary cannot be told of an ended ephemeral run\'s failed delivery', async (t) => { const primaryCm = framework.getAgent('primary')!.getContextManager(); const store = primaryCm.addMessage.bind(primaryCm); t.mock.method(primaryCm, 'addMessage', (...args: Parameters) => { @@ -349,9 +363,10 @@ describe('notices about an agent\'s own action', () => { const errors: string[] = []; t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); - await runEphemeral('locus', 'Worker speech for a channel that is not registered.'); - + await runEphemeral('locus', 'Worker speech.'); assert.equal(framework.getAgent('worker'), null, 'the run is released'); + await internals(framework).channelRegistry.routeSpeech('worker', 'Worker speech for a channel that is not registered.', 'missing-channel'); + assert.ok(errors.some((e) => e.includes('[route-failure] failed to tell the primary that worker\'s reply to missing-channel failed'))); }); @@ -362,7 +377,6 @@ describe('notices about an agent\'s own action', () => { }); try { await internals(bare).initializeMcpl([]); - (internals(bare).channelRegistry as unknown as { defaultPublishChannel: string | null }).defaultPublishChannel = 'missing-channel'; const errors: string[] = []; t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); const created = await bare.createEphemeralAgent({ @@ -374,6 +388,9 @@ describe('notices about an agent\'s own action', () => { bare.start(); await run; await bare.runUntilIdle(); + // After the run's release, its speech is routed to a channel given + // directly, as a channel of its own would be, and the delivery fails. + await internals(bare).channelRegistry.routeSpeech('worker', 'Worker speech with nobody to tell.', 'missing-channel'); const archive = await ContextManager.open({ store: bare.getStore(), namespace: 'subagent/worker', isolate: true, strategy: new PassthroughStrategy(),