From 4dfb2670fd99e13235d3d49573aa9ea6a19e8ff6 Mon Sep 17 00:00:00 2001 From: Hilda-2681 Date: Fri, 16 Oct 2026 16:59:29 -0400 Subject: [PATCH] fix(conversations): a message during its channel's fork spawn waits for it With conversation routing on, the first qualifying message on a channel spawns its fork: createConversationAgent reserves the fork's name, opens and seeds its context, and only then is the channel bound to it. The event queue handles its messages one at a time, and so does the coalescer (its deliveries and deferred batches' audience lookups run under one lock), but the two run beside each other, and a spawn can wait: the fork's seed compiles the template through afterAnnotations, which waits while the template's withheld tool results are being saved to the workspace. Meanwhile a second message for the same new channel was lost one of two ways: - it decided on the same spawn (a plain message and a coalesced one, or a message and a deferred batch whose audience lookup spawns the fork), and the second spawn of the name failed ("Conversation agent name ... is already registered or reserved"; traced as mcpl:conversation-spawn-failed on an incoming message's path, logged on a batch's); - it wouldn't bind the channel on its own, such as an unmentioned message where only a mention binds, so it decided `unbound` and was dropped as unrouted, though a moment later it would have gone into the fork. Either way it wasn't delivered to any agent: the trunk never holds channel traffic. - routeConversation decides a routing once no spawn is in flight for its channel, and starts the spawn it decides on, in one synchronous step with the check, so at most one spawn per channel is in flight. A routing that finds its channel's fork being spawned waits for that spawn to settle, then decides with the settled state: the fork once it is bound, or, if the spawn failed, what a message arriving next would get, which proposes the spawn again if the message qualifies. A failed spawn's own message is lost as before; the failure isn't shared with the waiters. - spawnConversationAgent creates the fork, binds the channel and persists the counters, registered in conversationSpawns under the channel until it settles. routeConversation registers a spawn only when none is in flight for the channel, and only that spawn's settlement removes its entry, so the entry removed is always its own; it is removed before anyone waiting on it resumes, so a waiter deciding again doesn't find it. - routeConversationIncoming and coalescedAudience, the two spawn sites, go through them. The waiting routing holds the path it runs on for as long as the spawn's seed waits: on the event queue, every channel's messages; under the coalescer's lock (a coalesced delivery, or a batch's wake from admit or freeze), every subject. The recovery timer's and boot's wakes run outside the lock and hold nothing else. That's bounded by the template's withheld-result deadline (WITHHELD_SPILL_DEADLINE_MS, 30 s per batch), which the spawner's own path already waited. By reading, nothing the spawn awaits waits on the event queue or takes the coalescer's lock: the guard's spill (spillWithheld -> writeToolResultFile -> the workspace module's writeBinary, raced against the batch deadline), the fork's open and dry compile and the template's dry compile (context-manager over the store), and bind and persist (synchronous). Two channel ids that forkAgentName sanitizes to one name still take one fork name at the same generation, and the second's spawn fails as on main; that's separate. Tests: test/conversation-spawn-in-flight.test.ts holds the template's aborted batch mid-annotation (its spill waits on a gate), so the first routing's spawn waits in its seed: - a coalesced message while a plain message's spawn waits: one spawn, no failure, the fork holds both; - an unmentioned message while a mention's spawn waits, bind 'mention': nothing unrouted, the fork holds both; - a plain message while a deferred batch's audience spawn waits: no failure, the fork holds the plain message; - a deferred batch whose audience is looked up while a plain message's spawn waits: one spawn, and the batch wakes that fork, which renders it; - a coalesced message waiting on a spawn that fails: the first failure is traced and the waiter spawns the fork, holding its message; - a plain message waiting on a batch's spawn that fails: it spawns the fork itself, holding its message. mcpl-rejection-log-lines' two spawn-failure tests run the spawn paths on a stub framework, which now needs the framework's own routeConversation, spawnConversationAgent and conversationSpawns (spawnStep()). Co-Authored-By: Claude Opus 5.5 --- changelog.d/conversation-spawn-once.fixed.md | 1 + src/framework.ts | 68 ++++- test/conversation-spawn-in-flight.test.ts | 259 +++++++++++++++++++ test/mcpl-rejection-log-lines.test.ts | 9 + 4 files changed, 329 insertions(+), 8 deletions(-) create mode 100644 changelog.d/conversation-spawn-once.fixed.md create mode 100644 test/conversation-spawn-in-flight.test.ts diff --git a/changelog.d/conversation-spawn-once.fixed.md b/changelog.d/conversation-spawn-once.fixed.md new file mode 100644 index 000000000..2d6033992 --- /dev/null +++ b/changelog.d/conversation-spawn-once.fixed.md @@ -0,0 +1 @@ +- A message that arrives on a channel while its conversation fork is being spawned now goes where it would have gone just after the spawn: into the fork once it is bound. The event queue and the coalescer each deliver their messages one at a time, but they run beside each other, and a fork's spawn waits while its template's withheld tool results are still being saved to the workspace. Meanwhile a second message for the same new channel, plain or coalesced, or a deferred batch that spawns the fork to wake it, could decide on the same spawn, and that second spawn failed ("already registered or reserved"): traced as `mcpl:conversation-spawn-failed` on an incoming message's path, logged on a batch's. A message that wouldn't bind the channel on its own, such as an unmentioned one where only a mention binds, was dropped as unrouted. Either way the message wasn't delivered to any agent. Now a routing on a channel whose fork is being spawned waits for that spawn and decides once it settles. If the spawn failed, its own message is lost as before, and a waiting routing decides as a message arriving next would, proposing the spawn again if its message qualifies. While it waits, the path it runs on waits too, for as long as the spawn's seed waits (the withheld results' 30-second deadline per batch): on the event queue, every channel's messages; under the coalescer's lock, every subject's. diff --git a/src/framework.ts b/src/framework.ts index 94f1035ff..e9257e7bd 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -96,6 +96,7 @@ import { detectKnownToolWrapperProse } from './tool-wrapper-prose-guard.js'; import { InferenceRouter } from './mcpl/inference-router.js'; import { ChannelRegistry, type ChannelToolOrigin } from './mcpl/channel-registry.js'; import { ConversationRouter, CONVERSATION_ROUTING_DEPRECATION_NOTICE } from './mcpl/conversation-router.js'; +import type { IncomingMessageFacts, RoutingDecision } from './mcpl/conversation-router.js'; import { safeSlice } from './safe-slice.js'; import { ProviderWaits, EVERY_MODEL, isReleaseModel, withoutControls, type ProviderWait } from './provider-waits.js'; import type { WorkspaceModule } from './modules/workspace/index.js'; @@ -1938,6 +1939,9 @@ export class AgentFramework { * turn still lands in the right channel. */ private conversationAgentHomes: Map = new Map(); + /** Forks being spawned, by channel id, each until its spawn settles (see + * routeConversation). The promise never rejects. */ + private readonly conversationSpawns = new Map>(); /** Last idle-TTL sweep timestamp. */ private lastConversationSweep = 0; @@ -9433,7 +9437,7 @@ export class AgentFramework { const router = this.conversationRouter!; const descriptor = this.channelRegistry?.getDescriptor(event.channelId); - const decision = router.route({ + const decision = await this.routeConversation({ channelId: event.channelId, mentioned: event.metadata?.mentioned === true, kind: ConversationRouter.classifyChannel(descriptor, event.metadata), @@ -9451,9 +9455,7 @@ export class AgentFramework { let agent: Agent | undefined; if (decision.kind === 'spawn') { try { - agent = await this.createConversationAgent(decision.agentName, event.channelId); - router.bind(event.channelId, decision.agentName, decision.generation); - this.persistConversationRouterState(); + agent = await decision.spawned; this.emitTrace({ type: 'mcpl:conversation-spawned', channelId: event.channelId, @@ -9620,6 +9622,58 @@ export class AgentFramework { } } + /** + * The router's decision for a message or batch on `facts.channelId`, made + * once no spawn is in flight for that channel, with a spawn it decides on + * already started. A routing that finds its channel's fork being spawned + * waits for that spawn to settle and decides then, whatever it would have + * decided meanwhile: once the fork is bound, the decision is the fork, and + * if the spawn failed, it decides as a message arriving next would, which + * proposes the spawn again if the message qualifies. The check, the + * decision and the spawn's start are one synchronous step, so at most one + * spawn per channel is in flight. Without the wait, a second spawn of the + * fork's name failed ("already registered or reserved"), and a message + * that didn't qualify on its own was dropped as unrouted; either way its + * message was lost. Each path that delivers messages handles them one at + * a time, but the event queue and the coalescer run beside each other, and + * a spawn can wait, as on its template's annotations + * (createConversationAgent); the waiting routing holds its own path for as + * long, which none of the spawn's awaits needs. + */ + private async routeConversation(facts: IncomingMessageFacts): Promise< + Exclude | (Extract & { spawned: Promise }) + > { + for (;;) { + const inFlight = this.conversationSpawns.get(facts.channelId); + if (inFlight) { + await inFlight; + continue; + } + const decision = this.conversationRouter!.route(facts); + if (decision.kind !== 'spawn') return decision; + return { ...decision, spawned: this.spawnConversationAgent(facts.channelId, decision) }; + } + } + + /** + * Create the fork a spawn decision names, bind its channel to it and + * persist the counters, registered in conversationSpawns under the channel + * until it settles. routeConversation registers a spawn only when none is + * in flight for the channel, and only this spawn's settlement removes it, + * so the entry it removes is its own. It's removed before anyone waiting + * on it resumes, so a waiter that decides again doesn't find it. + */ + private spawnConversationAgent(channelId: string, decision: { agentName: string; generation: number }): Promise { + const spawn = this.createConversationAgent(decision.agentName, channelId).then((agent) => { + this.conversationRouter!.bind(channelId, decision.agentName, decision.generation); + this.persistConversationRouterState(); + return agent; + }); + const done = (): void => { this.conversationSpawns.delete(channelId); }; + this.conversationSpawns.set(channelId, spawn.then(done, done)); + return spawn; + } + /** Persist the router's generation counters (see hydration in create()). */ private persistConversationRouterState(): void { if (!this.conversationRouter) return; @@ -10104,16 +10158,14 @@ export class AgentFramework { // No fork yet: the same bind decision an incoming message would get. A // batch that qualifies spawns its fork now so it has someone to wake. const metadata = (occ.event.lane === 'channel' ? occ.event.event.metadata : occ.event.event.origin) ?? {}; - const decision = router.route({ + const decision = await this.routeConversation({ channelId: occ.scope.id, mentioned: metadata.mentioned === true, kind: ConversationRouter.classifyChannel(this.channelRegistry?.getDescriptor(occ.scope.id), metadata), }); if (decision.kind !== 'spawn') return decision.kind === 'unbound' ? [] : (this.agents.has(decision.agentName) ? [decision.agentName] : []); try { - await this.createConversationAgent(decision.agentName, occ.scope.id); - router.bind(occ.scope.id, decision.agentName, decision.generation); - this.persistConversationRouterState(); + await decision.spawned; return [decision.agentName]; } catch (err) { console.error(`[coalescing] could not spawn a conversation agent for ${logValue(occ.scope.id)}:`, err); diff --git a/test/conversation-spawn-in-flight.test.ts b/test/conversation-spawn-in-flight.test.ts new file mode 100644 index 000000000..7b688608c --- /dev/null +++ b/test/conversation-spawn-in-flight.test.ts @@ -0,0 +1,259 @@ +/** + * A channel's fork is spawned once, and every message that arrives on the + * channel while it is being spawned goes where it would have gone just after. + * The event queue and the coalescer each handle their messages one at a time, + * but they run beside each other, and a spawn can wait, as a fork's seed waits + * for its template's annotations. A routing that decided on the same spawn + * meanwhile used to spawn the same name again, which failed ("already + * registered or reserved"), and a message that didn't qualify on its own was + * dropped as unrouted; either way its message was lost. Now a routing on that + * channel waits for the spawn in flight and decides then: into the fork once + * it is bound, or, if the spawn failed, as a message arriving next would. + * + * Each test holds the template's aborted batch mid-annotation (its spill + * waits on a gate), so the first routing's spawn waits in its seed until the + * test opens the gate. Where a fork's turn runs (a mention, or a batch's + * wake), the fixture's membrane answers it. + */ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import type { TraceEvent } from '../src/index.js'; +import { createMockResponse } from './helpers/mock-membrane.js'; +import { fixture } from './helpers/coalescing-fixture.js'; + +const FORK = 'conversation-chat-g1'; + +/** A gate a test opens itself, and a wait for a condition the code under + * test makes true (no fixed timers standing in for events). */ +function gate() { + let open!: () => void; + const opened = new Promise((resolve) => { open = resolve; }); + return { opened, open }; +} +async function until(check: () => boolean, what: string) { + for (let i = 0; i < 400 && !check(); i++) await new Promise((resolve) => setTimeout(resolve, 5)); + assert.ok(check(), `timed out waiting until ${what}`); +} + +/** A plain channels/incoming message for `chat`: no coalesce, so the event + * queue delivers it. Not a mention unless `metadata` says so. */ +const plain = (messageId: string, text: string, metadata?: Record) => ({ + channelId: 'chat', messageId, timestamp: new Date().toISOString(), author: { id: 'u', name: 'User' }, + content: [{ type: 'text', text }], ...(metadata ? { metadata } : {}), +}); + +/** A deferred channel-scoped batch for `chat`, whose wake looks up its + * audience, spawning the fork if none is bound. */ +const BATCH_FLAGS = { deferred: true, channelId: 'chat', key: 'activity' }; + +/** + * The coalescing fixture with conversation routing (by default on every + * channel message, triggering only on a mention) and the template's aborted + * batch held mid-annotation. `seeds` counts the template's compiles that wait + * for its annotations, a fork's seed among them; `routings` counts routings + * of incoming channel messages and `audiences` a batch's audience lookups; + * `failures`, `spawned` and `unrouted` collect the routing traces. + */ +async function heldTemplate(options: { server?: Record; conversations?: Record } = {}) { + const f = await fixture({ + agents: [{ name: 'agent', model: 'test', systemPrompt: 'system', toolResultGuard: true }], + framework: { + conversations: options.conversations ?? { templateAgent: 'agent', bind: { channel: 'always' }, trigger: { channel: 'mention' } }, + } as never, + server: options.server ?? {}, + }); + const write = gate(); + let spills = 0; + (f.framework as unknown as { spillWithheld: () => Promise }).spillWithheld = async () => { + spills++; + await write.opened; + return null; + }; + const responses: unknown[] = [ + createMockResponse([{ type: 'tool_use', id: 'one', name: 'test--read', input: {} }], 'tool_use'), + { aborted: true, reason: 'test-abort' }, + ]; + (f.membrane as unknown as { stream: () => Promise }).stream = async () => { + const response = responses.shift(); + assert.ok(response, 'unexpected inference'); + return response; + }; + const template = f.framework.getAgent('agent')!; + template.getContextManager().addMessage('user', [{ type: 'text', text: 'read' }]); + await template.runInference(f.framework.getAllTools()); + template.provideToolResult('one', { success: true, data: 'original' }); + assert.equal((await template.runInference(f.framework.getAllTools())).aborted, true); + await until(() => spills === 1, "the template's aborted batch is being annotated"); + + const seeds = { calls: 0 }; + const guard = template.toolResultGuard; + const afterAnnotations = guard.afterAnnotations.bind(guard); + guard.afterAnnotations = (read) => { seeds.calls++; return afterAnnotations(read); }; + const internals = f.framework as unknown as { + routeConversationIncoming: (...args: unknown[]) => Promise; + coalescedAudience: (...args: unknown[]) => Promise; + }; + const routings = { calls: 0 }; + const route = internals.routeConversationIncoming.bind(f.framework); + internals.routeConversationIncoming = (...args) => { routings.calls++; return route(...args); }; + const audiences = { calls: 0 }; + const audience = internals.coalescedAudience.bind(f.framework); + internals.coalescedAudience = (...args) => { audiences.calls++; return audience(...args); }; + const failures: string[] = []; + const spawned: string[] = []; + const unrouted: string[] = []; + f.framework.onTrace((e: TraceEvent) => { + if (e.type === 'mcpl:conversation-spawn-failed') failures.push((e as { error: string }).error); + if (e.type === 'mcpl:conversation-spawned') spawned.push((e as { agentName: string }).agentName); + if (e.type === 'mcpl:conversation-unrouted') unrouted.push((e as { messageId: string }).messageId); + }); + await f.register('chat'); + const holds = (text: string) => JSON.stringify(f.framework.getAgent(FORK)?.getContextManager().getAllMessages() ?? []).includes(text); + return { f, write, seeds, routings, audiences, failures, spawned, unrouted, holds }; +} + +/** The template's next compile, the first spawn's seed, fails once. */ +function failNextSeed(t: Awaited>) { + const manager = t.f.framework.getAgent('agent')!.getContextManager(); + const compile = manager.compile.bind(manager); + let failNext = true; + manager.compile = ((...args: Parameters) => { + if (!failNext) return compile(...args); + failNext = false; + return Promise.reject(new Error('the template could not be compiled (test)')); + }) as typeof compile; +} + +describe('messages on a channel whose fork is being spawned', () => { + it('a coalesced message routed while a plain message spawns the fork goes into that fork', async () => { + const t = await heldTemplate(); + try { + await t.f.send('channels/incoming', { messages: [plain('p1', 'first-plain')] }); + const queue = t.f.framework.runUntilIdle(); + await until(() => t.seeds.calls > 0, "the plain message's spawn waits for its seed"); + const coalesced = t.f.send('channels/incoming', { messages: [t.f.channel('e2', 'second-coalesced', { initial: true }, 'chat', 'm2')] }); + await until(() => t.routings.calls === 2, 'the coalesced message is being routed'); + t.write.open(); + await Promise.all([queue, coalesced]); + assert.deepEqual(t.failures, []); + assert.deepEqual(t.spawned, [FORK], 'one spawn'); + assert.equal(t.f.framework.getAgent('conversation-chat-g2'), null); + assert.ok(t.holds('first-plain'), 'the fork holds the plain message'); + assert.ok(t.holds('second-coalesced'), 'and the coalesced one'); + } finally { + t.write.open(); + await t.f.close(); + } + }); + + it('an unmentioned message routed while a mention spawns the fork goes into that fork', async () => { + const t = await heldTemplate({ conversations: { templateAgent: 'agent', bind: { channel: 'mention' }, trigger: { channel: 'mention' } } }); + t.f.alwaysRespond(); + try { + await t.f.send('channels/incoming', { messages: [plain('p1', 'first-mention', { mentioned: true })] }); + const queue = t.f.framework.runUntilIdle(); + await until(() => t.seeds.calls > 0, "the mention's spawn waits for its seed"); + // On its own it wouldn't bind the channel, so it decides `unbound` + // unless it waits for the fork. + const coalesced = t.f.send('channels/incoming', { messages: [t.f.channel('e2', 'second-unmentioned', { initial: true }, 'chat', 'm2')] }); + await until(() => t.routings.calls === 2, 'the unmentioned message is being routed'); + t.write.open(); + await Promise.all([queue, coalesced]); + await t.f.framework.runUntilIdle(); + assert.deepEqual(t.unrouted, [], 'nothing was dropped as unrouted'); + assert.deepEqual(t.spawned, [FORK]); + assert.ok(t.holds('first-mention'), 'the fork holds the mention'); + assert.ok(t.holds('second-unmentioned'), 'and the unmentioned message'); + } finally { + t.write.open(); + await t.f.close(); + } + }); + + it("a plain message routed while a batch's audience spawns the fork goes into that fork", async () => { + const t = await heldTemplate({ server: { shouldTriggerInference: () => true } }); + t.f.alwaysRespond(); + try { + const pushed = t.f.send('push/event', t.f.params('b1', 'batch', BATCH_FLAGS)); + await until(() => t.seeds.calls > 0, "the batch's spawn waits for its seed"); + await t.f.send('channels/incoming', { messages: [plain('p1', 'plain-during-batch')] }); + const queue = t.f.framework.runUntilIdle(); + await until(() => t.routings.calls === 1, 'the plain message is being routed'); + t.write.open(); + await Promise.all([pushed, queue]); + await t.f.framework.runUntilIdle(); + assert.deepEqual(t.failures, []); + assert.ok(t.f.framework.getAgent(FORK), "the batch's audience spawned the fork"); + assert.equal(t.f.framework.getAgent('conversation-chat-g2'), null); + assert.ok(t.holds('plain-during-batch'), 'the fork holds the plain message'); + } finally { + t.write.open(); + await t.f.close(); + } + }); + + it("a batch whose audience is looked up while a plain message spawns the fork wakes that fork", async () => { + const t = await heldTemplate({ server: { shouldTriggerInference: () => true } }); + t.f.alwaysRespond(); + try { + await t.f.send('channels/incoming', { messages: [plain('p1', 'first-plain')] }); + const queue = t.f.framework.runUntilIdle(); + await until(() => t.seeds.calls > 0, "the plain message's spawn waits for its seed"); + const pushed = t.f.send('push/event', t.f.params('b1', 'batch', BATCH_FLAGS)); + await until(() => t.audiences.calls > 0, "the batch's audience is being looked up"); + t.write.open(); + await Promise.all([queue, pushed]); + await t.f.framework.runUntilIdle(); + assert.deepEqual(t.spawned, [FORK], 'one spawn'); + assert.equal(t.f.framework.getAgent('conversation-chat-g2'), null); + assert.ok(t.holds('first-plain'), 'the fork holds the plain message'); + assert.ok(t.holds('document_diff'), "and the batch it was woken for, as rendered at the fork's turn"); + } finally { + t.write.open(); + await t.f.close(); + } + }); + + it('a coalesced message waiting on a spawn that fails proposes the spawn itself', async () => { + const t = await heldTemplate(); + failNextSeed(t); + try { + await t.f.send('channels/incoming', { messages: [plain('p1', 'first-plain')] }); + const queue = t.f.framework.runUntilIdle(); + await until(() => t.seeds.calls > 0, "the plain message's spawn waits for its seed"); + const coalesced = t.f.send('channels/incoming', { messages: [t.f.channel('e2', 'second-coalesced', { initial: true }, 'chat', 'm2')] }); + await until(() => t.routings.calls === 2, 'the coalesced message is being routed'); + t.write.open(); + await Promise.all([queue, coalesced]); + assert.deepEqual(t.failures, ['the template could not be compiled (test)'], "the first spawn's failure, traced"); + assert.deepEqual(t.spawned, [FORK], "the waiting routing's own spawn"); + assert.ok(t.holds('second-coalesced'), 'the fork holds the waiting message'); + assert.equal(t.holds('first-plain'), false, "the failed spawn's message isn't in it, as before"); + } finally { + t.write.open(); + await t.f.close(); + } + }); + + it("a plain message waiting on a batch's spawn that fails proposes the spawn itself", async () => { + const t = await heldTemplate({ server: { shouldTriggerInference: () => true } }); + t.f.alwaysRespond(); + failNextSeed(t); + try { + const pushed = t.f.send('push/event', t.f.params('b1', 'batch', BATCH_FLAGS)); + await until(() => t.seeds.calls > 0, "the batch's spawn waits for its seed"); + await t.f.send('channels/incoming', { messages: [plain('p1', 'plain-after-batch')] }); + const queue = t.f.framework.runUntilIdle(); + await until(() => t.routings.calls === 1, 'the plain message is being routed'); + t.write.open(); + await Promise.all([pushed, queue]); + // The batch's failed spawn is logged on its path, not traced. + assert.deepEqual(t.failures, []); + assert.deepEqual(t.spawned, [FORK], "the waiting message's own spawn"); + assert.ok(t.holds('plain-after-batch'), 'the fork holds the waiting message'); + } finally { + t.write.open(); + await t.f.close(); + } + }); +}); diff --git a/test/mcpl-rejection-log-lines.test.ts b/test/mcpl-rejection-log-lines.test.ts index 31e94cede..68ca8fe42 100644 --- a/test/mcpl-rejection-log-lines.test.ts +++ b/test/mcpl-rejection-log-lines.test.ts @@ -270,12 +270,20 @@ test('a refusal reaction with no server for its locus names the locus as one fie assert.deepEqual(lines, [`[inference-refusal] reaction skipped: no MCPL server for locus ${field(locus)} (agent=agent)`]); }); +/** The framework's own spawn step, which both spawn paths decide through, + * for a stub framework: it calls the stub's createConversationAgent. */ +function spawnStep() { + const proto = AgentFramework.prototype as unknown as Record<'routeConversation' | 'spawnConversationAgent', unknown>; + return { conversationSpawns: new Map(), routeConversation: proto.routeConversation, spawnConversationAgent: proto.spawnConversationAgent }; +} + test('a conversation agent that fails to spawn names its channel as one field', async (t) => { const lines = captureConsoleError(t); const traces: Array<{ type: string }> = []; const fakeThis = { conversationRouter: { route: () => ({ kind: 'spawn', agentName: 'conversation-chat-g1', generation: 1, trigger: true }), templateAgent: 'agent' }, channelRegistry: { getDescriptor: () => undefined }, + ...spawnStep(), createConversationAgent: async () => { throw new Error('no room'); }, emitTrace: (e: { type: string }) => { traces.push(e); }, }; @@ -294,6 +302,7 @@ test('a coalesced batch whose conversation agent fails to spawn names its channe route: () => ({ kind: 'spawn', agentName: 'conversation-chat-g1', generation: 1, trigger: true }), }, channelRegistry: { getDescriptor: () => undefined }, + ...spawnStep(), createConversationAgent: async () => { throw new Error('no room'); }, }; const audience = (AgentFramework.prototype as unknown as { coalescedAudience: (occ: unknown) => Promise }).coalescedAudience;