From 608357b5dd414a0f5e652fa1fabd51848ec7e5c4 Mon Sep 17 00:00:00 2001 From: Milo-2653 Date: Mon, 19 Oct 2026 04:59:20 -0400 Subject: [PATCH] fix(routing): after a re-pin, the token traces and typing follow the turn's locus An addressed re-pin at a tool boundary, or the agent's own channel_open, moves a locus or hybrid turn's locus between rounds, and the next round's prose is delivered there. driveStream computed the channel it shows once, when the stream started, so for the rest of the stream the inference:tokens and inference:content_block traces named the old channel, and the typing indicator stayed there (#555). The turn-start comment still said the locus is frozen. driveStream now keeps that channel as a variable, and followTurnLocus compares it with the turn's locus at the model's content in each round (its tokens and block events, not the tool results a boundary reports as content) and at a refusal retry. When the locus has moved, in locus or hybrid mode, it moves the indicator there, and the traces, which read the same value, name the new channel. 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. Explicit mode keeps its trigger channel. A stream that hands its turn to a successor (a budget restart, a guard retry) leaves its indicator running. TurnRouting now records the channel the turn's indicator was last shown in, and showTurnTyping, which every start of the indicator and followTurnLocus go through, stops that one first when it shows the indicator elsewhere. So a re-pin at such a boundary no longer leaves the old channel typing until the turn ends. Both name the conversation the turn is attending: in hybrid mode they follow the locus whether or not a >>> envelope sends the text elsewhere. The live outgoing stream is unchanged. Tests: the re-pin rig's locus and hybrid re-pins, a pinned hybrid turn, a stream with no block events, a refusal before the retried round's content, a budget restart that re-pins, channel_open, an empty final response with and without its tool's result reported as content, and a locus turn with nowhere to speak at its start; an explicit turn's typing and traces across its rounds. Co-Authored-By: Claude Opus 5.5 --- .../555-repin-traces-follow-locus.fixed.md | 11 + ...555-trace-channel-follows-locus.changed.md | 17 + src/framework.ts | 91 +++-- src/types/trace.ts | 24 +- test/explicit-prose-routing.test.ts | 42 +++ test/present-while-acting.test.ts | 328 +++++++++++++++++- 6 files changed, 485 insertions(+), 28 deletions(-) create mode 100644 changelog.d/555-repin-traces-follow-locus.fixed.md create mode 100644 changelog.d/555-trace-channel-follows-locus.changed.md 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 d72da2323..61c53f376 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -790,6 +790,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 @@ -14280,6 +14288,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 @@ -15611,14 +15630,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 @@ -15681,7 +15702,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); @@ -16232,22 +16253,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 @@ -16433,6 +16472,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] = { @@ -16441,9 +16490,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 dca24323c..584d058e2 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 4b01e0a78..002e396c0 100644 --- a/test/explicit-prose-routing.test.ts +++ b/test/explicit-prose-routing.test.ts @@ -1631,6 +1631,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 7e5d7c8ad..63c885ab8 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', @@ -1810,6 +2063,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(); @@ -1817,6 +2078,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( @@ -1827,6 +2093,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(); + }); + it('reactions and system markers injected mid-turn do not clear send suppression', async () => { // A reaction (`chat:reaction` tag) or a `system: true` marker is not // conversational input: prose following an explicit send stays suppressed,