diff --git a/changelog.d/555-repin-traces-follow-locus.fixed.md b/changelog.d/555-repin-traces-follow-locus.fixed.md new file mode 100644 index 000000000..22c2abe4a --- /dev/null +++ b/changelog.d/555-repin-traces-follow-locus.fixed.md @@ -0,0 +1,11 @@ +- After an addressed re-pin, or the agent's own `channel_open`, moves a + `locus` or `hybrid` turn's locus between rounds, the `inference:tokens` and + `inference:content_block` traces now name the new locus, and the typing + indicator moves there, from the next round on (#555). Before, both stayed + on the channel the stream started in for the rest of the stream, while the + round's prose was delivered to the new locus, and after a budget restart + or a guard retry the old channel's indicator kept running beside the new + one until the turn ended. Both name the conversation the turn is + attending: in `hybrid` mode they follow the locus whether or not a `>>>` + envelope sends the text elsewhere. An `explicit` turn's typing and traces + still show its trigger channel. diff --git a/changelog.d/555-trace-channel-follows-locus.changed.md b/changelog.d/555-trace-channel-follows-locus.changed.md new file mode 100644 index 000000000..7bc8e9a47 --- /dev/null +++ b/changelog.d/555-trace-channel-follows-locus.changed.md @@ -0,0 +1,17 @@ +- **Hosts that read the `channelId` of `inference:tokens` or + `inference:content_block` traces:** in `locus` and `hybrid` mode it now + follows the turn's locus. It was documented as "the turn's locus as the + stream started", so after an addressed re-pin, or the agent's own + `channel_open`, moved the locus between rounds, it kept naming the old + channel for the rest of the stream. It now names the new locus from the + next round on: from the model's first content in that round, or from its + retry after a refusal. It is still the channel the stream shows typing in, + and in `hybrid` mode still the locus whether or not a `>>>` envelope sends + the text elsewhere. + - **Migration:** a consumer that keeps the first `channelId` it sees for an + activation, as a voice relay may, works as before, and stays on the old + channel after a re-pin; to follow the move, read each trace's + `channelId`. A consumer that groups one stream's chunks by `channelId` + now finds a stream whose locus moved under two channels. + - **Unchanged:** `explicit` mode's traces, which name the trigger channel, + and the live outgoing stream (`channels/outgoing/chunk` and `complete`). diff --git a/src/framework.ts b/src/framework.ts index bc909dfa1..fd9e671a6 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -807,6 +807,14 @@ interface TurnRouting { * nowhere to speak. */ locus?: string; + /** + * The channel the turn's typing indicator was last shown in, by any of its + * streams (showTurnTyping). A stream that hands the turn to a successor (a + * budget restart, a guard retry) leaves its indicator running, so a later + * start, or a move, that shows it elsewhere stops this one first, as after + * a re-pin. + */ + typing?: string; /** * Channels the turn sent into with a send tool (SEND_ENGAGEMENT_TOOLS): a * conversational follow-up in one of them is addressed by context, so it @@ -14312,6 +14320,17 @@ export class AgentFramework { } } + /** + * Show the turn's typing indicator in `channelId`, stopping the one it last + * showed elsewhere: this stream's, or one an earlier stream of the turn + * left running when it handed the turn on. + */ + private showTurnTyping(routing: TurnRouting, channelId: string): void { + if (routing.typing !== undefined && routing.typing !== channelId) this.channelRegistry?.stopTyping(routing.typing); + this.channelRegistry?.startTyping(channelId); + routing.typing = channelId; + } + /** * The channel of the turn's newest arrival whose message is still in the * agent's store (TurnRouting.arrivals). A message removed since it was @@ -15673,14 +15692,16 @@ export class AgentFramework { // turn, which falls back to the global default rather than inheriting // 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, - // 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 + // Resolve the turn's outbound locus now: the home channel for forks, + // else the trigger channel, else the global default, which an + // ephemeral run never takes (takesGlobalFallback). Ambient mid-turn + // injections must not redirect the agent's plain prose (ambient + // chatter and reactions hijacked it before, 2026-07-21); only an + // addressed message, a follow-up in a channel the turn engaged + // (repinTurnLocus) or the agent's own channel_open moves it, between + // rounds. The eager snapshot protects heartbeat turns from a moving // defaultPublishChannel. Explicit mode uses a turn-scoped `>>` target - // and disabled mode has no prose target, so neither freezes or + // and disabled mode has no prose target, so neither resolves or // announces a locus. If the effective locus differs from the last one // announced, a durable `[routing]` notice goes into the window BEFORE // this turn compiles, so the agent always knows where its voice goes @@ -15743,7 +15764,7 @@ export class AgentFramework { turnProseRouting === 'explicit' ? trigger?.channelId ?? null : turnLocus ?? null; - if (earlyTypingChannel) this.channelRegistry?.startTyping(earlyTypingChannel); + if (earlyTypingChannel) this.showTurnTyping(routing, earlyTypingChannel); try { const requestSnapshot = this.captureInferenceToolSnapshot(agent); @@ -16294,22 +16315,40 @@ export class AgentFramework { let silenceBeforeRefusal: 'send' | 'private' | null | undefined; // Typing indicator: show " is typing…" for the whole duration of - // this turn. Started here (paired with the finally below, so it can never - // leak) and refreshed on a 7s interval by the ChannelRegistry until - // stopped on any exit path. Never moves mid-turn. - // - locus mode: the frozen routing pin (where bare prose will land). + // this turn. Started at the turn's start, before the compile, and here, + // and refreshed on a 7s interval by the ChannelRegistry until the finally + // below stops it on an exit path that ends the turn. A stream that hands + // the turn to a successor leaves it running, and the successor's start + // stops it first if it shows elsewhere (showTurnTyping). Typing is + // presence, not delivery: it says "attending to the conversation here", + // whichever channel a `>>>` or `>>` envelope then sends the words to. + // - locus and hybrid mode: the turn's locus, the conversation the turn + // is attending. An addressed re-pin or the agent's own channel_open + // can move it between rounds, and followTurnLocus (below) moves the + // indicator with it at the next round's first content, or at its + // retry after a refusal. // - explicit mode: there is no locus — indicate on the TRIGGER channel - // (where the wake came from). Typing is presence, not delivery, so - // this doesn't violate never-guess: it says "attending to what you - // sent here", which is true regardless of where the reply goes. - // Heartbeat/no-trigger explicit turns show no indicator. - const typingChannel = + // (where the wake came from). This doesn't violate never-guess: it + // says "attending to what you sent here", which is true regardless + // of where the reply goes. Heartbeat/no-trigger explicit turns show + // no indicator. + let typingChannel = turnProseRouting === 'disabled' ? null : turnProseRouting === 'explicit' ? trigger?.channelId ?? null : resolveTurnLocus(); - if (typingChannel) this.channelRegistry!.startTyping(typingChannel); + if (typingChannel) this.showTurnTyping(streamRouting, typingChannel); + // Locus and hybrid mode: move the indicator to the turn's locus when it + // has moved since the indicator last showed. Once a turn has a locus it + // always has one. + const followTurnLocus = (): void => { + if (turnProseRouting !== 'locus' && turnProseRouting !== 'hybrid') return; + const locus = resolveTurnLocus(); + if (!locus || locus === typingChannel) return; + this.showTurnTyping(streamRouting, locus); + typingChannel = locus; + }; // MCPL Spec 14.3 outgoing streaming: route text deltas to their // destination server AS THEY GENERATE (voice synthesis, live message @@ -16501,6 +16540,16 @@ export class AgentFramework { break; } this.touchEphemeralRun(agent.name, true); + // The model's content in a round, or its retry after a refusal, comes + // after the boundary that may have moved the turn's locus: show the + // round where it is, before its traces name the channel. The results + // a tool boundary reports (XML tool mode's, or a result with an + // image) belong to the round before, and a response with nothing in + // it moves nothing, so a channel_open just before an empty final + // response doesn't start typing in a turn that is ending. + const modelContent = (event.type === 'tokens' && event.meta.type !== 'tool_result') + || (event.type === 'block' && event.event.block.type !== 'tool_result'); + if (modelContent || event.type === 'retrying') followTurnLocus(); switch (event.type) { case 'tokens': { const trace: Parameters[0] = { @@ -16509,9 +16558,9 @@ export class AgentFramework { content: event.content, blockType: event.meta.type, blockIndex: event.meta.blockIndex, - // Turn-frozen locus (same value as the typing indicator above): - // lets trace consumers tag every chunk with the channel this - // turn's prose is bound for, without re-deriving routing. + // The channel the typing indicator above shows, the conversation + // the turn is attending: lets trace consumers tag every chunk + // with it, without re-deriving routing. channelId: typingChannel ?? undefined, }; // Delivery joins a round's text blocks with a newline diff --git a/src/types/trace.ts b/src/types/trace.ts index a91f2322b..f0d44ba61 100644 --- a/src/types/trace.ts +++ b/src/types/trace.ts @@ -124,9 +124,15 @@ export type TraceEvent = blockType: 'text' | 'thinking' | 'tool_call' | 'tool_result'; /** 0-indexed block position in the current assistant turn. */ blockIndex: number; - /** The channel this stream shows typing in: the turn's locus as the - * stream started (see inference:started), or in explicit mode the - * channel the start's wake came from. */ + /** The channel this stream shows typing in, the conversation the turn + * is attending. Locus and hybrid mode: the turn's locus, which an + * addressed re-pin or the agent's own `channel_open` can move between + * rounds, so it can change within one stream; the indicator and these + * traces follow it from the next round on, at the model's first + * content in it or its retry after a refusal, and a tool result the + * boundary reports carries the channel of the round it answers. In + * hybrid mode a `>>>` envelope can send the text itself elsewhere. + * Explicit mode: the channel the start's wake came from. */ channelId?: string; }) | (TraceEventBase & { @@ -141,9 +147,15 @@ export type TraceEvent = phase: 'block_start' | 'block_complete'; blockType: 'text' | 'thinking' | 'tool_call' | 'tool_result'; blockIndex: number; - /** The channel this stream shows typing in: the turn's locus as the - * stream started (see inference:started), or in explicit mode the - * channel the start's wake came from. */ + /** The channel this stream shows typing in, the conversation the turn + * is attending. Locus and hybrid mode: the turn's locus, which an + * addressed re-pin or the agent's own `channel_open` can move between + * rounds, so it can change within one stream; the indicator and these + * traces follow it from the next round on, at the model's first + * content in it or its retry after a refusal, and a tool result the + * boundary reports carries the channel of the round it answers. In + * hybrid mode a `>>>` envelope can send the text itself elsewhere. + * Explicit mode: the channel the start's wake came from. */ channelId?: string; }) | (TraceEventBase & { diff --git a/test/explicit-prose-routing.test.ts b/test/explicit-prose-routing.test.ts index 262fb958f..f5b7f7045 100644 --- a/test/explicit-prose-routing.test.ts +++ b/test/explicit-prose-routing.test.ts @@ -1637,6 +1637,48 @@ describe('explicit prose routing', () => { await framework.stop(); }); + it("explicit mode's typing and traces stay on its trigger channel across the turn's rounds", async () => { + // Explicit mode has no locus: its typing and traces show where the wake + // came from, and only a locus or hybrid turn's follow a moved locus. + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 't1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([{ type: 'text', text: '>>#alpha done' }] as ContentBlock[])); + + const framework = await createFramework(); + const routed = stubRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const typing: string[] = []; + registry.startTyping = (channelId: string) => { typing.push(`start ${channelId}`); }; + registry.stopTyping = (channelId?: string) => { typing.push(`stop ${channelId ?? '(all)'}`); }; + const traced = new Set(); + framework.onTrace((event) => { + const e = event as { type: string; blockType?: string; channelId?: string }; + if ((e.type === 'inference:tokens' || e.type === 'inference:content_block') && e.blockType === 'text') traced.add(e.channelId); + }); + const internals = framework as unknown as { + pendingRequests: InferenceRequest[]; + processInferenceRequests(): Promise; + }; + try { + internals.pendingRequests.push({ + agentName: 'assistant', reason: 'mcpl:channel-incoming', source: 'discord', timestamp: Date.now(), + channelId: 'chan-alpha', addressed: true, + }); + await internals.processInferenceRequests(); + await framework.runUntilIdle(); + + assert.equal(module.calls.length, 1, 'control: round 1 ran its tool'); + assert.deepEqual(routed, [{ text: 'done', locus: 'chan-alpha' }], 'control: round 2 ran and was delivered'); + assert.ok(typing.includes('start chan-alpha'), `typing showed on the trigger channel: ${JSON.stringify(typing)}`); + assert.equal(typing.includes('stop chan-alpha'), false, `and stayed there until the turn ended: ${JSON.stringify(typing)}`); + assert.ok(typing.includes('stop (all)'), 'control: the turn\'s end stopped it'); + assert.deepEqual([...traced], ['chan-alpha'], 'round 2 is traced in the trigger channel'); + } finally { + await framework.stop(); + } + }); + it('explicit mode streams nothing before a destination: unprefixed text never reaches the trigger channel', async () => { membrane.pushResponse(createMockResponse([ { type: 'text', text: 'Unprefixed musing.\n>>#alpha streamed after the prefix' }, diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index 7e4c6becb..2fb9234f2 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -188,7 +188,7 @@ describe('present while acting', () => { rmSync(tempDir, { recursive: true, force: true }); }); - async function createFramework(agentOptions: { maxStreamTokens?: number } = {}): Promise { + async function createFramework(agentOptions: { maxStreamTokens?: number; proseRouting?: 'locus' | 'hybrid' } = {}): Promise { const framework = await AgentFramework.create({ storePath: join(tempDir, 'test.chronicle'), membrane: membrane.asMembrane(), @@ -405,6 +405,156 @@ describe('present while acting', () => { await framework.stop(); }); + /** + * What the turn shows, beside what is delivered: the registry stub + * resolves `#cafe` and delivers to it, and records typing and the text + * traces with the round each belongs to. + */ + function watchShown(framework: AgentFramework) { + const routed = stubChannelRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + registry.resolveProseTarget = (spec: string) => spec === '#cafe' + ? { channelId: 'discord:g:cafe', label: '#cafe' } + : { error: `no channel matches "${spec}"` }; + registry.deliverSpeech = async (_agent: string, text: string, channelId: string) => { + routed.push({ text, locus: channelId }); + return { status: 'delivered' as const, destination: { serverId: 'stub', channelId }, at: Date.now() }; + }; + const shown: string[] = []; + registry.startTyping = (channelId: string) => { shown.push(`start ${channelId}`); }; + registry.stopTyping = (channelId?: string) => { shown.push(`stop ${channelId ?? '(all)'}`); }; + // The text traces of the turn's first stream, by round. + const traced: Array<{ round: number; channelId?: string }> = []; + let streams = 0; + let round = 1; + framework.onTrace((event) => { + const e = event as { type: string; blockType?: string; channelId?: string }; + if (e.type === 'inference:started') streams++; + if (streams !== 1) return; + if (e.type === 'inference:tool_calls_yielded') round++; + if ((e.type === 'inference:tokens' || e.type === 'inference:content_block') && e.blockType === 'text') traced.push({ round, channelId: e.channelId }); + }); + const tracedIn = (n: number) => [...new Set(traced.filter((t) => t.round === n).map((t) => t.channelId))]; + /** Typing stopped on the first channel and, next, started on `to`, once. */ + const typingMovedTo = (to: string) => { + const moved = shown.indexOf('stop chan-live-1'); + return moved > 0 && shown[moved + 1] === `start ${to}` && shown.filter((x) => x === `start ${to}`).length === 1; + }; + return { routed, shown, tracedIn, typingMovedTo }; + } + + function addressedFromDm(): void { + module.toolDelayMs = 25; + module.interjection = 'hey, quick question over here?'; + module.interjectionMetadata = { channelId: 'discord:dm:antra', tags: ['chat:addressed'] }; + } + + for (const proseRouting of ['locus', 'hybrid'] as const) { + it(`${proseRouting} mode: after a re-pin, the next round's traces and typing follow the turn's locus`, async () => { + // The re-pin moves where round 2's prose is delivered, and the + // conversation the turn attends. The traces that tag each chunk with + // its channel, and the typing indicator, move with it. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Station four, proceeding.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Answering the person who addressed me.' }, + ] as ContentBlock[])); + + const framework = await createFramework({ proseRouting }); + const watched = watchShown(framework); + addressedFromDm(); + + trigger(framework); + await framework.runUntilIdle(); + + assert.deepEqual(watched.routed, [ + { text: 'Station four, proceeding.', locus: 'chan-live-1' }, + { text: 'Answering the person who addressed me.', locus: 'discord:dm:antra' }, + ], 'control: delivery follows the re-pin'); + assert.deepEqual(watched.tracedIn(1), ['chan-live-1'], 'round 1 is traced in the first channel'); + assert.deepEqual(watched.tracedIn(2), ['discord:dm:antra'], 'round 2 is traced in the DM'); + assert.ok(watched.typingMovedTo('discord:dm:antra'), `typing moves to the DM, once: ${JSON.stringify(watched.shown)}`); + + await framework.stop(); + }); + } + + it("hybrid mode: while a `>>>` destination holds, a re-pin moves the traces and typing to the conversation the turn now attends", async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>>#cafe Station four, proceeding.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Answering after the move.' }, + ] as ContentBlock[])); + + const framework = await createFramework({ proseRouting: 'hybrid' }); + const watched = watchShown(framework); + addressedFromDm(); + + trigger(framework); + await framework.runUntilIdle(); + + assert.deepEqual(watched.routed, [ + { text: 'Station four, proceeding.', locus: 'discord:g:cafe' }, + { text: 'Answering after the move.', locus: 'discord:g:cafe' }, + ], 'control: delivery keeps the `>>>` destination after the re-pin'); + assert.deepEqual(watched.tracedIn(1), ['chan-live-1'], "round 1's traces name the turn's locus, not the destination"); + assert.deepEqual(watched.tracedIn(2), ['discord:dm:antra'], "round 2's traces name the locus the re-pin moved to"); + assert.ok(watched.typingMovedTo('discord:dm:antra'), `typing moves to the DM: ${JSON.stringify(watched.shown)}`); + + await framework.stop(); + }); + + it('a provider that reports text before any block event still moves the traces and typing at the round after a re-pin', async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Station four, proceeding.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Answering after the move.' }, + ] as ContentBlock[])); + + const framework = await createFramework(); + const watched = watchShown(framework); + addressedFromDm(); + dropBlockEvents(); + + trigger(framework); + await framework.runUntilIdle(); + + assert.deepEqual(watched.routed.map((r) => r.locus), ['chan-live-1', 'discord:dm:antra'], 'control: the re-pin moved round 2'); + assert.deepEqual(watched.tracedIn(2), ['discord:dm:antra'], 'round 2 is traced in the DM'); + assert.ok(watched.typingMovedTo('discord:dm:antra'), `typing moves to the DM: ${JSON.stringify(watched.shown)}`); + + await framework.stop(); + }); + + it("locus mode: a refusal before the retried round's first content traces the round's new locus", async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Station four, proceeding.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Answering after the retry.' }, + ] as ContentBlock[])); + + const framework = await createFramework(); + const watched = watchShown(framework); + addressedFromDm(); + refuseRoundOnce(2); + + trigger(framework); + await framework.runUntilIdle(); + + assert.deepEqual(watched.routed.map((r) => r.locus), ['chan-live-1', 'discord:dm:antra'], 'control: the re-pin moved round 2'); + assert.deepEqual(watched.tracedIn(2), ['discord:dm:antra'], "the retry's trace and round 2's name the DM"); + + await framework.stop(); + }); + it('a follow-up in a channel the agent explicitly engaged this turn re-pins', async () => { // 2026-07-31 n=7 (q's #portables reply): the agent explicitly sent into // a channel this turn; someone replies there WITHOUT a mention (ambient @@ -580,6 +730,75 @@ describe('present while acting', () => { }; } + /** + * The first stream's Nth round (1-based) is refused once, before it + * streams anything: membrane's retrying event comes first. + */ + function refuseRoundOnce(round: number): void { + const makeStream = membrane.streamYielding.bind(membrane); + let streams = 0; + membrane.streamYielding = (request, options) => { + const stream = makeStream(request, options); + if (++streams !== 1) return stream; + const iterate = stream[Symbol.asyncIterator].bind(stream); + stream[Symbol.asyncIterator] = async function* () { + let rounds = 1; + let refused = false; + for await (const event of { [Symbol.asyncIterator]: iterate }) { + if (rounds === round && !refused) { + refused = true; + yield { type: 'retrying', attempt: 1, maxAttempts: 1, reason: 'refusal' } as never; + } + yield event; + if (event.type === 'tool-calls') rounds++; + } + }; + return stream; + }; + } + + /** + * Streams whose providers report text before any block callback: every + * stream yields its tokens with no block events. + */ + function dropBlockEvents(): void { + const makeStream = membrane.streamYielding.bind(membrane); + membrane.streamYielding = (request, options) => { + const stream = makeStream(request, options); + const iterate = stream[Symbol.asyncIterator].bind(stream); + stream[Symbol.asyncIterator] = async function* () { + for await (const event of { [Symbol.asyncIterator]: iterate }) { + if (event.type !== 'block') yield event; + } + }; + return stream; + }; + } + + /** + * Streams that report each round's tool results as content after its tool + * boundary, as membrane does in XML tool mode or for a result with an + * image: a tool_result block with its tokens. + */ + function reportToolResults(): void { + const makeStream = membrane.streamYielding.bind(membrane); + membrane.streamYielding = (request, options) => { + const stream = makeStream(request, options); + const iterate = stream[Symbol.asyncIterator].bind(stream); + stream[Symbol.asyncIterator] = async function* () { + for await (const event of { [Symbol.asyncIterator]: iterate }) { + yield event; + if (event.type === 'tool-calls') { + yield { type: 'block', event: { event: 'block_start', index: 9, block: { type: 'tool_result' } } } as never; + yield { type: 'tokens', content: '{"ok":true}', meta: { type: 'tool_result', visible: true, blockIndex: 9 } } as never; + yield { type: 'block', event: { event: 'block_complete', index: 9, block: { type: 'tool_result' } } } as never; + } + } + }; + return stream; + }; + } + /** Feed a response to the Nth turn start (1-based) the moment it begins. */ function feedStart(framework: AgentFramework, start: number, ...responses: ContentBlock[][]): void { let starts = 0; @@ -1069,6 +1288,40 @@ describe('present while acting', () => { }); } + it("a budget restart that re-pins its turn stops the typing its earlier stream left in the old channel", async () => { + // The over-budget stream hands the turn on at a tool boundary, before any + // content of the next round, so it never shows the move itself: the + // restart's start, which takes the addressed message as an arrival, + // moves the indicator. + const overBudget = createMockResponse([ + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use'); + overBudget.usage = { inputTokens: 100, outputTokens: 5 }; + membrane.pushResponse(overBudget); + + const framework = await createFramework({ maxStreamTokens: 20 }); + const routed = stubChannelRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const shown: string[] = []; + registry.startTyping = (channelId: string) => { shown.push(`start ${channelId}`); }; + registry.stopTyping = (channelId?: string) => { shown.push(`stop ${channelId ?? '(all)'}`); }; + module.toolDelayMs = 25; + module.interjection = 'quick question for you'; + module.interjectionMetadata = addressed('discord:dm:antra'); + feedStart(framework, 2, [{ type: 'text', text: 'After the restart.' }] as ContentBlock[]); + + trigger(framework); + await framework.runUntilIdle(); + + assert.equal(membrane.calls.length, 2, 'control: the turn restarted for its budget'); + assert.deepEqual(routed, [{ text: 'After the restart.', locus: 'discord:dm:antra' }], 'control: the restart re-pinned to the DM'); + const started = shown.indexOf('start discord:dm:antra'); + const stopped = shown.indexOf('stop chan-live-1'); + assert.ok(started > 0 && stopped >= 0 && stopped < started, `typing stops in the old channel before it shows in the DM: ${JSON.stringify(shown)}`); + + await framework.stop(); + }); + for (const [label, messages, expected] of [ ["keeps the turn's locus", [], 'chan-turn'], ['moves it to the newest addressed arrival', @@ -1812,6 +2065,14 @@ describe('present while acting', () => { const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; (registry as { handleChannelToolCall?: unknown }).handleChannelToolCall = async () => ({ success: true, data: { channelId: 'discord:guild:observatory', opened: true } }); + const shown: string[] = []; + registry.startTyping = (channelId: string) => { shown.push(`start ${channelId}`); }; + registry.stopTyping = (channelId?: string) => { shown.push(`stop ${channelId ?? '(all)'}`); }; + const traced: Array<{ channelId?: string }> = []; + framework.onTrace((event) => { + const e = event as { type: string; blockType?: string; channelId?: string }; + if ((e.type === 'inference:tokens' || e.type === 'inference:content_block') && e.blockType === 'text') traced.push(e); + }); trigger(framework); await framework.runUntilIdle(); @@ -1819,6 +2080,11 @@ describe('present while acting', () => { assert.deepEqual(routed, [ { text: 'Hello, observatory.', locus: 'discord:guild:observatory' }, ], 'prose right after channel_open lands in the opened channel'); + // What the turn shows follows the open, as it follows a re-pin. + assert.ok(traced.length > 0, 'control: the next round was traced'); + assert.deepEqual([...new Set(traced.map((e) => e.channelId))], ['discord:guild:observatory'], 'the next round, the only one with text, is traced in the opened channel'); + const moved = shown.indexOf('stop chan-live-1'); + assert.ok(moved > 0 && shown[moved + 1] === 'start discord:guild:observatory', `typing moves to it: ${JSON.stringify(shown)}`); // The announcement rode the tool result the model saw. const results = membrane.lastStream!.receivedToolResults.flat() as Array<{ content?: unknown }>; assert.ok( @@ -1829,6 +2095,66 @@ describe('present while acting', () => { await framework.stop(); }); + for (const reported of [false, true]) { + it(`a channel_open just before an empty final response doesn't start typing in the opened channel${reported ? ', when its result is reported as content (XML tool mode)' : ''}`, async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Opening the observatory.' }, + { type: 'tool_use', id: 'c1', name: 'channel_open', input: { channelId: 'discord:guild:observatory' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([] as ContentBlock[])); + if (reported) reportToolResults(); + + const framework = await createFramework(); + stubChannelRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + (registry as { handleChannelToolCall?: unknown }).handleChannelToolCall = + async () => ({ success: true, data: { channelId: 'discord:guild:observatory', opened: true } }); + const shown: string[] = []; + registry.startTyping = (channelId: string) => { shown.push(`start ${channelId}`); }; + registry.stopTyping = (channelId?: string) => { shown.push(`stop ${channelId ?? '(all)'}`); }; + + trigger(framework); + await framework.runUntilIdle(); + + assert.ok(shown.includes('start chan-live-1'), `control: typing showed while round 1 spoke: ${JSON.stringify(shown)}`); + const results = membrane.lastStream!.receivedToolResults.flat() as Array<{ content?: unknown }>; + assert.ok(JSON.stringify(results).includes('plain speech now lands in this channel'), 'control: the open succeeded and moved the turn there'); + assert.equal(membrane.lastStream!.toolDepth, 1, 'control: the empty final response was the stream\'s last'); + assert.equal(shown.includes('start discord:guild:observatory'), false, `the empty last round moves nothing: ${JSON.stringify(shown)}`); + + await framework.stop(); + }); + } + + it('a locus turn that starts with nowhere to speak shows typing where its channel_open moves it', async () => { + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'c1', name: 'channel_open', input: { channelId: 'discord:guild:observatory' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Hello, observatory.' }, + ] as ContentBlock[])); + + const framework = await createFramework(); + const routed = stubChannelRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + registry.resolveLocus = () => null; + (registry as { handleChannelToolCall?: unknown }).handleChannelToolCall = + async () => ({ success: true, data: { channelId: 'discord:guild:observatory', opened: true } }); + const shown: string[] = []; + registry.startTyping = (channelId: string) => { shown.push(`start ${channelId}`); }; + registry.stopTyping = (...args: unknown[]) => { shown.push(args.length === 0 ? 'stop (all)' : `stop ${String(args[0])}`); }; + + trigger(framework); + await framework.runUntilIdle(); + + assert.deepEqual(routed, [ + { text: 'Hello, observatory.', locus: 'discord:guild:observatory' }, + ], 'control: the open gave the turn somewhere to speak'); + assert.deepEqual(shown.filter((x) => x !== 'stop (all)'), ['start discord:guild:observatory'], `typing shows only in the opened channel, and nothing else is stopped: ${JSON.stringify(shown)}`); + + await framework.stop(); + }); + // ------------------------------------------------------------------------- // A channel its server declared inbound-only takes no plain speech (the // registry refuses it), so every notice that says where plain speech goes