diff --git a/changelog.d/turn-routing-per-turn.fixed.md b/changelog.d/turn-routing-per-turn.fixed.md new file mode 100644 index 000000000..f3e16bbc4 --- /dev/null +++ b/changelog.d/turn-routing-per-turn.fixed.md @@ -0,0 +1,44 @@ +- In explicit mode, a turn that a tool ends, such as with `skip_reply`, no + longer leaves the prose it queued to run under the next turn's routing + state. Such a turn settles before its earlier rounds' queued deliveries + have posted, and in explicit mode the next turn can start at once. That + start cleared the sticky destination those deliveries read when they + post, so a queued unprefixed segment bounced into the next turn. Each + turn now has its own routing state (the sticky destination, the turn's + destination, hybrid's hold) and its logical turn's ` !` continuation and + receipt tallies (the destinations and suppressed segments its receipt + reports), and a queued delivery keeps the one of the turn that queued + it. In hybrid and + locus mode the ended turn already keeps the next one waiting until its + receipt is written, as before. +- The agent's prose deliveries go out in the order they were written, + across turns: every stream queues its deliveries behind the agent's + earlier ones, a fresh turn's included (apart from a stream that a later + one replaced). A turn's reply, and its completion, can therefore wait for + an earlier turn's posts: up to their request timeouts, one after another, + or for good when `requestTimeoutMs` is 0 and a post is never answered. +- In explicit and hybrid mode, a ` !` continuation that an earlier round + asked for now starts when a tool ends the turn, once that turn's + deliveries have run, unless the framework has stopped or the agent has + been disposed by then (a dropped one is logged). In explicit mode a turn + that another event started meanwhile runs first. Before, hybrid mode + started it only after the next turn, and explicit mode dropped it, or, + when its delivery ran late, started it after the next turn. +- A logical turn now stays open from the start that opens it until the turn + settles (it completes, a tool ends it, or it fails or is abandoned), and + every start that routes prose while it is open continues it: a framework + retry, the start that consumes unstreamed tool results (whichever wake + the scheduler starts it with), the later start a provider-capacity hold + defers a turn to. Each begins the sticky destination and hybrid's hold + afresh, as before, but a ` !` asked for earlier in the turn still stands, + in explicit mode too, and in hybrid and locus mode the receipt at the + turn's end reports every stream's deliveries, including ones that land + after a new stream began. Before, explicit mode dropped that ` !`, and the + receipt counted only the deliveries that landed after the new start. +- A turn that completes now waits for its queued deliveries before its + receipt and its continuation check, in every mode, whatever its last + response. Before, only a completion with final text waited, so in hybrid + and locus mode a stream whose last response was empty (a budget + restart's, say) wrote its receipt first: those deliveries were never + receipted, or were counted in the next turn's receipt. A ` !` they carried + was missed too. diff --git a/src/framework.ts b/src/framework.ts index dd5ac6c5e..dbb2e4033 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -525,6 +525,53 @@ function resumesLogicalTurn(reason: string | undefined): boolean { return reason === 'context_budget_restart' || reason === 'tool_result_guard_retry'; } +/** + * One turn's routing and delivery receipts (AgentFramework.turnRouting). + * A fresh turn gets a new object; a queued delivery keeps its own turn's. + */ +interface TurnRouting { + /** The sticky destination, set by each `>>` (hybrid: `>>>`) prefix that resolves. */ + pin?: string; + /** + * Explicit mode: the destination the turn's delivered rounds have set, + * decided at each round boundary (roundDestination), where a stream that + * resumes the turn starts. + */ + destination: string | null; + /** Hybrid: held back after a malformed or unresolved envelope, until a later `>>>` names a target. */ + hybridHold: boolean; + /** What the logical turn keeps across a new stream (LogicalTurn). */ + logical: LogicalTurn; +} + +/** + * What every stream of one logical turn shares, from the fresh start that + * opens it until it settles (AgentFramework.openLogical): a framework retry, + * a re-inference after unstreamed tool results, a capacity-held turn's later + * compile. beginAgentTurn starts each with fresh routing, but a ` !` asked + * for in any of them stands, and the receipt at the turn's end reports every + * stream's deliveries, even ones that land after a new stream began. + */ +interface LogicalTurn { + /** The turn asked for a ` !` continuation; taken once. */ + continuation: boolean; + /** Successful plain-prose destinations in delivery order; deduped by the receipt. */ + deliveries: string[]; + /** Plain-prose segments the turn suppressed, reported by its receipt. */ + suppressed: number; +} + +function freshTurnRouting(): TurnRouting { + return { destination: null, hybridHold: false, logical: { continuation: false, deliveries: [], suppressed: 0 } }; +} + +/** How a round's prose is delivered: kept private, and under which turn's routing. */ +interface ProseRound { + keepPrivate?: boolean; + /** The routing state of the turn that queued the round. */ + routing: TurnRouting; +} + function isTurnContinuation(reason: string): boolean { return reason === 'context_budget_restart' || reason === 'tool_results_ready' || reason === 'tool_result_guard_retry'; @@ -1172,26 +1219,6 @@ export class AgentFramework { * keeps unrelated channels from moving the pin. Cleared at every fresh * turn's start; budget restarts keep it. */ private turnEngagedChannels: Map> = new Map(); - /** Channels this turn's PLAIN PROSE was actually delivered to, in delivery - * order (deduped at render). Feeds the `[delivered]` receipt appended at - * logical turn end: explicit sends receipt themselves via their - * tool_result, but auto-routed prose previously vanished into the router - * with no in-window record — the agent could never see where its own - * words landed (2026-07-31 misroute series; antra: minimal receipts). - * Cleared at every fresh turn's start; budget restarts keep it (same - * logical turn, receipt covers the whole turn). */ - private turnProseDeliveries: Map = new Map(); - /** Count of plain-prose segments SUPPRESSED this turn by sticky - * explicit-send silencing. The suppression rule itself is untouched - * (antra, 2026-07-31: visibility is enough) — but it must never be a - * silent black hole: Mythos multiplexed two threads in one round (prose - * for the lane + explicit send for another channel, textbook per the - * routing doc), the silencing ate the prose reply, and his own record - * ("scene delivered to #mythos via plain speech — routing confirmed") - * believed it delivered. The receipt now reports suppression, so the - * author sees the segment's fate one turn later. Cleared each fresh - * turn; budget restarts keep it. */ - private turnProseSuppressed: Map = new Map(); /** A tool boundary injected fresh CONVERSATIONAL input (a real message — * not a reaction or a system marker) into the live stream. Tells * driveStream to clear sticky explicit-send suppression before handling @@ -1206,28 +1233,43 @@ export class AgentFramework { // ---- Explicit prose routing (docs/explicit-prose-routing.md) ---------- /** Latest bounced (undelivered) prose per agent — the `{{unsent}}` source. */ private proseClipboards: Map = new Map(); - /** Sticky per-TURN delivery target, set by each `>>` prefix that resolves. - * Cleared at every fresh turn start, and on a silent wake; a stream that - * resumes the turn (resumesLogicalTurn) keeps it. */ - private proseTargetPins: Map = new Map(); /** - * Explicit mode: the destination the turn's delivered rounds have set, - * decided at each round boundary (roundDestination). A stream that resumes - * the turn (a tool-result-guard retry, a budget restart) starts there, - * however far the gateway's queued deliveries have got. Turn-scoped, like - * the sticky destination. + * Each agent's routing state for its current turn (TurnRouting): the + * sticky destination, the turn's destination for a resumed stream, + * hybrid's hold, a `!` continuation and receipt tallies. A fresh turn starts a new object + * instead of clearing the old one, and a stream that resumes the turn + * (resumesLogicalTurn) keeps it. A queued delivery holds the object of the + * turn that queued it, so prose a turn leaves queued when it ends (a + * skip_reply ends it before its posts have run) still follows that turn's + * destination, and its late writes can't reach the next turn. */ - private proseTurnDestinations: Map = new Map(); + private turnRouting: Map = new Map(); /** - * The latest link of each agent's chain of queued prose deliveries, so a - * stream that resumes the turn queues behind the earlier streams' ones. + * The latest link of each agent's chain of queued prose deliveries. Every + * stream queues its own behind it, a fresh turn's included, so the agent's + * posts go out in the order they were written, across streams and turns. */ private proseSpeechChains: Map> = new Map(); - /** Hybrid router is fail-closed after a malformed/unresolved envelope until a new valid target. */ - private proseHybridSuppressed: Set = new Set(); - /** Agents whose current turn requested `!` continuation — re-woken when the - * turn completes instead of pausing until the next external event. */ - private proseContinuations: Set = new Set(); + /** + * Each agent's open logical turn (LogicalTurn). It stays open from the + * fresh start that opens it until the turn settles (settleAgent: it + * completes, a tool ends it, or it fails or is abandoned), and every fresh + * start that routes prose while it is open continues it: a framework + * retry, the start that consumes tool results the non-streaming fallback + * left (whichever wake the scheduler starts it with), the later compile a + * provider-capacity hold defers a turn to. A silent start neither + * continues nor opens one. A stream that ends without settling needs to do + * nothing to carry its turn. Settling only detaches the turn from future + * starts: the settling stream keeps its own object, which its remaining + * deliveries, its receipt and its ` !` check go on using (a completion + * settles before it awaits its chain and writes its receipt). So every + * logical end must close inheritance once, before the next fresh start: + * a turn left open would hand its object, late writes and all, to an + * unrelated turn. A tool-ended turn settles before the next turn can + * start. Only the stream that holds the open turn closes it; disposal + * drops it, and stop() clears every agent's. + */ + private openLogical: Map = new Map(); /** Consecutive bounce-wakes per agent; capped so an agent that keeps * emitting unprefixed prose can't wake-loop (notices still append). */ private proseBounceStreaks: Map = new Map(); @@ -2152,6 +2194,10 @@ export class AgentFramework { // A stopped host must never hang behind its own cooldown. this.cancelProviderAdmission(); + // An open logical turn doesn't survive a stop: a start after a restart + // on this instance opens a new one rather than carrying its ` !` and + // receipt tallies across (openLogical). + this.openLogical.clear(); // Cancel all active streams for (const agent of this.agents.values()) { @@ -4463,13 +4509,22 @@ export class AgentFramework { run.lastActivity = Date.now(); } - /** Settle an active ephemeral run (no-op for resident agents). The tool-call - * count comes from the run's own counter — the single source of truth, - * which unlike a per-stream local survives budget restarts. */ + /** A logical turn's end, for every agent: it closes the agent's open + * logical turn when it is the settling stream's own (openLogical), and + * settles an active ephemeral run. The run's tool-call count comes from + * its own counter — the single source of truth, which unlike a + * per-stream local survives budget restarts. */ private settleAgent( agentName: string, outcome: { stopReason: AgentSettleResult['stopReason']; speech: string; error?: string }, + /** The settling stream's logical turn, when that stream still owns the + * agent: it ends here, so the agent's next fresh start opens a new one. + * A settle from a stream a newer one replaced closes nothing. */ + closes: LogicalTurn | undefined, ): void { + // This identity check is the only guard against a stale settle closing + // a newer turn (no test reaches one); keep it when editing call sites. + if (closes && this.openLogical.get(agentName) === closes) this.openLogical.delete(agentName); const run = this.ephemeralRuns.get(agentName); if (!run) return; run.lastActivity = Date.now(); @@ -4611,6 +4666,9 @@ export class AgentFramework { // generation mismatch and leave any future owner untouched. this.activeStreams.delete(agent.name); this.pendingAssistantBlocks.delete(agent.name); + this.turnRouting.delete(agent.name); + this.proseSpeechChains.delete(agent.name); + this.openLogical.delete(agent.name); // The late stream finalizer intentionally refuses name-keyed cleanup once // deregistered, so disposal must release EventGate liveness itself. this.eventGate?.onInferenceEnded(agent.name); @@ -7162,7 +7220,7 @@ export class AgentFramework { // agent must be wakeable the moment it goes idle. (Idempotent; // driveStream's finally is the backstop.) this.eventGate?.onInferenceEnded(agent.name); - this.settleAgent(agent.name, { stopReason: 'turn_ended', speech: '' }); + this.settleAgent(agent.name, { stopReason: 'turn_ended', speech: '' }, this.turnRouting.get(agent.name)?.logical); this.emitTrace({ type: 'inference:turn_ended', agentName: agent.name }); } else if (overBudget || overPhysical) { // Context budget exceeded: break the stream, let compile() compress. @@ -7221,7 +7279,8 @@ export class AgentFramework { ...(midTurnInjections.length > 0 ? { injectedMessages: midTurnInjections.length } : {}), }); } else { - // Non-streaming fallback: schedule re-inference + // Non-streaming fallback: schedule re-inference, which continues + // this stream's logical turn (openLogical). this.pendingRequests.push({ agentName: agent.name, reason: 'tool_results_ready', @@ -7717,6 +7776,9 @@ export class AgentFramework { const channelId = this.conversationAgentHomes.get(agentName); const agent = this.agents.get(agentName); this.agents.delete(agentName); + this.turnRouting.delete(agentName); + this.proseSpeechChains.delete(agentName); + this.openLogical.delete(agentName); this.toolImageLedgers.delete(agentName); this.agentConfigs.delete(agentName); this.conversationAgentHomes.delete(agentName); @@ -8826,24 +8888,19 @@ export class AgentFramework { return stamp ? { silentHeartbeat: stamp } : undefined; } - /** Record prose segments suppressed by explicit-send silencing. */ - private recordProseSuppression(agentName: string, count: number): void { + /** Record plain-prose segments this turn suppressed. */ + private recordProseSuppression(routing: TurnRouting, count: number): void { if (count <= 0) return; - this.turnProseSuppressed.set(agentName, (this.turnProseSuppressed.get(agentName) ?? 0) + count); + routing.logical.suppressed += count; } /** Record a successful plain-prose delivery for this turn's receipt. */ private recordProseDelivery( - agentName: string, + routing: TurnRouting, outcome: { delivered: boolean; channelId: string } | null | undefined, ): void { if (!outcome?.delivered) return; - let list = this.turnProseDeliveries.get(agentName); - if (!list) { - list = []; - this.turnProseDeliveries.set(agentName, list); - } - list.push(outcome.channelId); + routing.logical.deliveries.push(outcome.channelId); } /** @@ -8859,12 +8916,12 @@ export class AgentFramework { * delivered nothing. Failures are already marked separately * ([discord-send-failed]); this is the success half. */ - private appendProseDeliveryReceipt(agent: Agent): void { - const list = this.turnProseDeliveries.get(agent.name); - const suppressed = this.turnProseSuppressed.get(agent.name) ?? 0; - if ((!list || list.length === 0) && suppressed === 0) return; - this.turnProseDeliveries.delete(agent.name); - this.turnProseSuppressed.delete(agent.name); + private appendProseDeliveryReceipt(agent: Agent, routing: TurnRouting): void { + const list = routing.logical.deliveries; + const suppressed = routing.logical.suppressed; + if (list.length === 0 && suppressed === 0) return; + routing.logical.deliveries = []; + routing.logical.suppressed = 0; const seen = new Set(); const shown: string[] = []; for (const id of list ?? []) { @@ -8992,6 +9049,47 @@ export class AgentFramework { } } + /** + * A `!` continuation the turn asked for (`>>#x !`, `>>>#x !`): start another + * turn now instead of pausing until the next external event. It reads the + * logical turn of the stream that asked (shared with a retry or a + * re-inference that continues it), once its deliveries have run, when the + * turn completes or a tool ends it. A stream that ends in an error asks for + * none itself, but a framework retry that follows it carries the request; + * a turn abandoned by a quiesce asks for none. + */ + private requestProseContinuation(agent: Agent, routing: TurnRouting): void { + // The flag may be shared with an earlier stream of the same logical + // turn (LogicalTurn); its deliveries are on the agent's chain ahead of + // this stream's, so by the time this check runs, they have run. + if (!routing.logical.continuation) return; + routing.logical.continuation = false; + // An explicit turn's check can run after its stream has returned. A + // stopped framework, or an agent no longer registered under its name (a + // disposed fork or ephemeral run), takes no new turn from it. + if (this.providerAdmissionClosed || this.agents.get(agent.name) !== agent) { + console.error(`[prose] ${agent.name}: '!' continuation dropped — ${this.providerAdmissionClosed ? 'the framework is stopping' : 'the agent is no longer registered'}`); + return; + } + console.error(`[prose] ${agent.name}: '!' continuation — requesting immediate next turn`); + this.pendingRequests.push({ + agentName: agent.name, + reason: 'prose-continuation', + source: 'framework', + timestamp: Date.now(), + }); + } + + /** The agent's routing state for its current turn, made when it has none. */ + private routingOf(agentName: string): TurnRouting { + let routing = this.turnRouting.get(agentName); + if (!routing) { + routing = freshTurnRouting(); + this.turnRouting.set(agentName, routing); + } + return routing; + } + /** * Explicit-prose-routing delivery gateway (docs/explicit-prose-routing.md). * Every prose segment of an `explicit`-mode agent flows through here: @@ -9001,7 +9099,11 @@ export class AgentFramework { * or via the turn's sticky target). */ private async deliverProse(agent: Agent, rawText: string, round?: { keepPrivate?: boolean }): Promise { - await this.deliverAdmittedProse(agent, this.admitProse(rawText), round); + // The direct entry delivers under the live turn's routing. + await this.deliverAdmittedProse(agent, this.admitProse(rawText), { + keepPrivate: round?.keepPrivate, + routing: this.routingOf(agent.name), + }); } /** @@ -9027,7 +9129,7 @@ export class AgentFramework { private async deliverAdmittedProse( agent: Agent, envelopes: AdmittedEnvelope[], - round?: { keepPrivate?: boolean }, + round: ProseRound, ): Promise { // Multi-envelope: a single prose block may address SEVERAL destinations, // each envelope routed independently (sticky/bounce as usual). @@ -9056,7 +9158,9 @@ export class AgentFramework { rawText: string, locus: string | null, allowLocus: boolean, + turnRouting: TurnRouting, ): Promise { + const routing = turnRouting; const lines = rawText.split('\n'); const envelopes: string[] = []; let current: string[] = []; @@ -9074,67 +9178,67 @@ export class AgentFramework { const normalized = envelope.replace(/^\s+(?=>>>)/, ''); const attempted = normalized.startsWith('>>>'); const parsed = parseHybridProsePrefix(normalized); - if (parsed.continueTurn) this.proseContinuations.add(agent.name); + if (parsed.continueTurn) routing.logical.continuation = true; if (parsed.kind === 'private') { - this.proseTargetPins.delete(agent.name); - this.proseHybridSuppressed.add(agent.name); + delete routing.pin; + routing.hybridHold = true; console.error(`[prose] ${agent.name}: >>>skip_reply — ${parsed.body.length} chars kept in context (not sent)`); continue; } if (parsed.kind === 'target') { const resolved = this.channelRegistry!.resolveProseTarget(parsed.target!); if ('error' in resolved) { - this.proseTargetPins.delete(agent.name); - this.proseHybridSuppressed.add(agent.name); + delete routing.pin; + routing.hybridHold = true; this.bounceProse(agent, parsed.body, resolved.error, resolved.candidates, '>>>'); continue; } let body = parsed.body; const usedClipboard = body.includes('{{unsent}}'); if (usedClipboard) body = body.replaceAll('{{unsent}}', this.proseClipboards.get(agent.name) ?? ''); - this.proseTargetPins.set(agent.name, resolved.channelId); - this.proseHybridSuppressed.delete(agent.name); + routing.pin = resolved.channelId; + routing.hybridHold = false; if (!body.trim()) { console.error(`[prose] ${agent.name}: >>>${parsed.target} established target; empty body — nothing sent yet`); continue; } try { const outcome = await this.channelRegistry!.routeSpeech(agent.name, body, resolved.channelId); - this.recordProseDelivery(agent.name, outcome); + this.recordProseDelivery(routing, outcome); if (outcome?.delivered) { this.proseBounceStreaks.delete(agent.name); if (usedClipboard) this.proseClipboards.delete(agent.name); } else { - this.proseTargetPins.delete(agent.name); - this.proseHybridSuppressed.add(agent.name); + delete routing.pin; + routing.hybridHold = true; this.bounceProse(agent, parsed.body, `delivery to ${resolved.channelId} was not confirmed`, undefined, '>>>'); } } catch (err) { - this.proseTargetPins.delete(agent.name); - this.proseHybridSuppressed.add(agent.name); + delete routing.pin; + routing.hybridHold = true; this.bounceProse(agent, parsed.body, `delivery to ${resolved.channelId} failed: ${err instanceof Error ? err.message : String(err)}`, undefined, '>>>'); } continue; } if (attempted) { - this.proseTargetPins.delete(agent.name); - this.proseHybridSuppressed.add(agent.name); + delete routing.pin; + routing.hybridHold = true; this.bounceProse(agent, normalized, 'malformed >>> routing envelope: destination is missing', undefined, '>>>'); continue; } - if (this.proseHybridSuppressed.has(agent.name)) { - this.recordProseSuppression(agent.name, 1); + if (routing.hybridHold) { + this.recordProseSuppression(routing, 1); continue; } - const sticky = this.proseTargetPins.get(agent.name); + const sticky = routing.pin; if (!allowLocus && !sticky) { - this.recordProseSuppression(agent.name, 1); + this.recordProseSuppression(routing, 1); continue; } const target = sticky ?? locus; try { const outcome = await this.channelRegistry!.routeSpeech(agent.name, envelope, target); - this.recordProseDelivery(agent.name, outcome); + this.recordProseDelivery(routing, outcome); } catch (err) { console.error('hybrid locus delivery failed:', err); } @@ -9144,16 +9248,15 @@ export class AgentFramework { private async deliverProseEnvelope( agent: Agent, envelope: AdmittedEnvelope, - round?: { keepPrivate?: boolean }, + round: ProseRound, ): Promise { const name = agent.name; + const routing = round.routing; const rawText = envelope.text; const parsed = parseProsePrefix(rawText); - // A private round (one that called skip_reply) asks for no continuation. - // Its turn ends at the round's tool results, and explicit mode doesn't - // await the speech chain there, so this delivery can run after the next - // turn has begun: a flag set here would reach that turn. - if (parsed.continueTurn && !round?.keepPrivate) this.proseContinuations.add(name); + // A private round (one that called skip_reply) asks for no publication + // or continuation, even when delivery runs after that turn has ended. + if (parsed.continueTurn && !round.keepPrivate) routing.logical.continuation = true; if (parsed.kind === 'private') { // The text already lives in the assistant message (window/chronicle); @@ -9164,7 +9267,7 @@ export class AgentFramework { return; } - if (round?.keepPrivate) { + if (round.keepPrivate) { // skip_reply's promise: "Any plain text you wrote this turn stays // private and is NOT posted". Every envelope of the round stays in the // context, as a >>skip_reply envelope does: it isn't sent, doesn't @@ -9189,10 +9292,10 @@ export class AgentFramework { } targetChannel = res.channelId; // Sticky within the turn: later unprefixed segments follow this target. - this.proseTargetPins.set(name, targetChannel); + routing.pin = targetChannel; body = parsed.body; } else { - const sticky = this.proseTargetPins.get(name); + const sticky = routing.pin; if (!sticky) { this.bounceProse(agent, rawText, 'no destination prefix and no destination set yet this turn'); return; @@ -9567,23 +9670,30 @@ export class AgentFramework { // A framework retry is a new physical stream in the same logical turn. this.logicalTurnToolCalls.set(agent, { turnToken, count: previousLogicalToolState?.count ?? 0 }); } - // Fresh turn: forget the previous turn's explicit-send engagements and - // prose deliveries — both are strictly turn-scoped (a restart continues - // the same logical turn and keeps them). + // Fresh turn: forget the previous turn's explicit-send engagements. + // A restart continues the same logical turn and keeps them. this.turnEngagedChannels.delete(agent.name); - this.turnProseDeliveries.delete(agent.name); - this.turnProseSuppressed.delete(agent.name); - this.proseHybridSuppressed.delete(agent.name); - // The sticky destination and the turn's destination are turn-scoped in - // every mode, so a turn whose mode changes before it resumes can't - // start from an earlier turn's. - this.proseTargetPins.delete(agent.name); - this.proseTurnDestinations.delete(agent.name); + // A new routing state for the new turn, in every mode: the sticky + // destination, the turn's destination, hybrid's hold and the receipt + // tallies start empty. The previous turn's object isn't cleared: prose + // it left queued still reads and writes its own. + const fresh = freshTurnRouting(); + // A start while the agent's logical turn is open continues it: it + // shares the earlier stream's ` !` and receipt tallies, which a + // delivery still queued there can write after this stream begins + // (openLogical). Otherwise this start opens one. A silent start does + // neither: it routes no prose, so it keeps a logical turn of its own, + // and an open turn waits for the next start that routes prose. + if (!trigger?.suppressProse) { + const open = this.openLogical.get(agent.name); + if (open) fresh.logical = open; + else this.openLogical.set(agent.name, fresh.logical); + } + this.turnRouting.set(agent.name, fresh); if (turnProseRouting === 'explicit' || turnProseRouting === 'disabled') { // Explicit and disabled prose routing have no inferred locus. // Explicit mode uses a turn-scoped `>>` target; disabled mode has no // prose target at all. Neither mode freezes or announces a locus. - this.proseContinuations.delete(agent.name); this.midTurnInputSignals.delete(agent.name); this.turnLocusPins.delete(agent.name); } else { @@ -9600,10 +9710,16 @@ export class AgentFramework { // opened/pinned a channel mid-turn. if (trigger?.suppressProse) { this.turnLocusPins.delete(agent.name); - this.proseTargetPins.delete(agent.name); - this.proseTurnDestinations.delete(agent.name); - this.proseContinuations.delete(agent.name); + // Silent: no routing state. A continuing silent start keeps its turn's + // logical turn; a fresh one keeps the one of its own it was given. + const silent = freshTurnRouting(); + const logical = this.turnRouting.get(agent.name)?.logical; + if (logical) silent.logical = logical; + this.turnRouting.set(agent.name, silent); } + // This start's own logical turn: a failure before its stream exists + // settles it, and no other start's. + const startLogical = this.turnRouting.get(agent.name)?.logical; this.touchEphemeralRun(agent.name, true); this.emitTrace({ @@ -9782,7 +9898,7 @@ export class AgentFramework { stopReason: 'exhausted', speech: '', error: err.message, - }); + }, startLogical); this.emitTrace({ type: 'inference:exhausted', agentName: agent.name, @@ -9846,6 +9962,9 @@ export class AgentFramework { // from the map here: the turn-alive busy check in processInferenceRequests // means no successor turn can have replaced it while we compiled. const myTurnToken = this.activeTurnTokens.get(agent.name); + // The turn-alive guard also keeps this turn's routing current at entry. + // Retain it for queued deliveries and receipts after a tool ends the turn. + const streamRouting = this.routingOf(agent.name); let hadToolCalls = false; let generationLost = false; @@ -9868,18 +9987,14 @@ export class AgentFramework { // is preserved end-to-end. Each link catches its own error: one failed // post must not silence the rest of the turn. // - // A stream that resumes its turn (as beginAgentTurn's continuingTurn: a - // budget restart, a tool-result-guard retry) keeps the turn's routing - // state, and queues its deliveries behind the earlier streams' ones. So - // the turn's deliveries run in the order its text was written, and the - // sticky destination moves in that order too, whichever stream wrote the - // text and however long an earlier post takes. + // Every stream queues its deliveries behind the agent's latest link, so + // the agent's posts go out in the order they were written, across a + // turn's streams (a budget restart, a tool-result-guard retry) and across + // turns. A turn that a tool ends (skip_reply) settles before its queued + // prose has posted; that prose keeps its own turn's routing state + // (turnRouting), and the next turn's posts wait behind it. const resumesTurn = resumesLogicalTurn(trigger?.reason); - let turnSpeechChain: Promise = resumesTurn - ? this.proseSpeechChains.get(agent.name) ?? Promise.resolve() - : Promise.resolve(); - // Kept at once, so a fresh turn that queues nothing still replaces the - // previous turn's chain for a stream that resumes it. + let turnSpeechChain: Promise = this.proseSpeechChains.get(agent.name) ?? Promise.resolve(); this.proseSpeechChains.set(agent.name, turnSpeechChain); const chainSpeech = (link: () => Promise, failure: string): void => { turnSpeechChain = turnSpeechChain.then(link).catch((err) => console.error(failure, err)); @@ -9889,7 +10004,7 @@ export class AgentFramework { const enqueueSpeech = (text: string, locus: string | null): void => { chainSpeech(async () => { const outcome = await this.channelRegistry!.routeSpeech(agent.name, text, locus); - this.recordProseDelivery(agent.name, outcome); + this.recordProseDelivery(streamRouting, outcome); }, 'mid-turn speech routing failed:'); }; @@ -9977,7 +10092,7 @@ export class AgentFramework { // deliveries needn't have run. The trigger channel is only where // explicit mode's typing indicator shows. initialTarget: turnProseRouting === 'explicit' - ? (resumesTurn ? this.proseTurnDestinations.get(agent.name) ?? null : null) + ? (resumesTurn ? streamRouting.destination : null) : typingChannel, resolve: (spec) => { const r = this.channelRegistry!.resolveProseTarget(spec); @@ -10093,7 +10208,7 @@ export class AgentFramework { // Explicit mode's retried attempt starts where the turn's // delivered rounds have set the destination, not at the stream's // seed: rounds this stream delivered before the refusal stand. - if (turnProseRouting === 'explicit') proseStream?.retarget(this.proseTurnDestinations.get(agent.name) ?? null); + if (turnProseRouting === 'explicit') proseStream?.retarget(streamRouting.destination); this.emitTrace({ type: 'inference:tokens', agentName: agent.name, @@ -10235,14 +10350,16 @@ export class AgentFramework { console.error( `[routing] ${agent.name}: mid-turn prose NOT routed (proseRouting=disabled)`, ); - this.recordProseSuppression(agent.name, roundSegments.length); + this.recordProseSuppression(streamRouting, roundSegments.length); } else if (turnProseRouting === 'hybrid') { if (turnSilenced) { - this.recordProseSuppression(agent.name, roundSegments.length); + this.recordProseSuppression(streamRouting, roundSegments.length); } else if (!hasSameRoundPrivateThink) { const locus = resolveTurnLocus(); + // Queued under this turn's routing, whenever it posts. + const routing = streamRouting; for (const seg of roundSegments) { - chainSpeech(() => this.deliverHybridProse(agent, seg, locus, true), 'mid-turn hybrid prose delivery failed:'); + chainSpeech(() => this.deliverHybridProse(agent, seg, locus, true, routing), 'mid-turn hybrid prose delivery failed:'); } } } else if (turnProseRouting === 'explicit') { @@ -10260,17 +10377,16 @@ export class AgentFramework { // once, for the turn's destination and for its queued // delivery alike. const admitted = roundSegments.map((seg) => this.admitProse(seg)); + // Queued under this turn's routing, whenever it posts. + const routing = streamRouting; if (!keepPrivate) { - this.proseTurnDestinations.set( - agent.name, - this.roundDestination(this.proseTurnDestinations.get(agent.name) ?? null, admitted.flat()), - ); + routing.destination = this.roundDestination(routing.destination, admitted.flat()); } console.error( `[prose] ${agent.name}: mid-turn round [${roundToolNames.join(', ')}] -> ${roundSegments.length} segment(s) via prose gateway${keepPrivate ? ', kept private (skip_reply)' : ''}`, ); for (const envelopes of admitted) { - chainSpeech(() => this.deliverAdmittedProse(agent, envelopes, { keepPrivate }), 'mid-turn prose delivery failed:'); + chainSpeech(() => this.deliverAdmittedProse(agent, envelopes, { keepPrivate, routing }), 'mid-turn prose delivery failed:'); } } // The stream's next round starts where delivery sends @@ -10278,14 +10394,14 @@ export class AgentFramework { // followed this round's prefixes as it was written, but a // private round's, a >>skip_reply envelope's or a target // that didn't resolve leave the destination as it was. - proseStream?.retarget(this.proseTurnDestinations.get(agent.name) ?? null); + proseStream?.retarget(streamRouting.destination); } else if (turnSilenced) { console.error( `[routing] ${agent.name}: mid-turn round [${roundToolNames.join(', ')}] -> prose NOT routed (turn silenced)`, ); // Visible in the turn-end receipt — silencing must never // be a silent black hole (n=8: the flying-scene reply). - this.recordProseSuppression(agent.name, roundSegments.length); + this.recordProseSuppression(streamRouting, roundSegments.length); } else if (hasSameRoundPrivateThink) { console.error( `[routing] ${agent.name}: mid-turn round [${roundToolNames.join(', ')}] -> prose NOT routed (same_round_think_text_policy=private)`, @@ -10327,7 +10443,7 @@ export class AgentFramework { // persisted as assistant speech, including a terminal refusal // after the single recovery retry. proseStream?.reset(); - if (turnProseRouting === 'explicit') proseStream?.retarget(this.proseTurnDestinations.get(agent.name) ?? null); + if (turnProseRouting === 'explicit') proseStream?.retarget(streamRouting.destination); if (withheld) { const usage = response.details?.usage ?? response.usage; const tokenUsage = usage ? { @@ -10516,6 +10632,9 @@ export class AgentFramework { ? splitProseSegments(liveProseRouting ? terminalContent : response.content) : []; const admitFinal = turnProseRouting === 'explicit' && !!this.channelRegistry; + // The final text is delivered under this turn's routing, as its + // queued rounds are. + const finalRouting = streamRouting; const admittedSpeech = admitFinal && finalSpeechText ? this.admitProse(finalSpeechText) : null; const admittedSegments = admitFinal ? finalSegments.map((seg) => this.admitProse(seg)) : []; @@ -10553,7 +10672,7 @@ export class AgentFramework { .filter((block: ContentBlock): block is ContentBlock & { type: 'text' } => block.type === 'text') .map((block) => block.text) .join('\n'), - }); + }, streamRouting.logical); this.emitTrace({ type: 'inference:completed', @@ -10733,31 +10852,33 @@ export class AgentFramework { if (speechContent.length > 0 && this.channelRegistry) { const speechText = finalSpeechText; if (speechText) { - // A turn resumed after its earlier rounds delivers after - // them: their queued deliveries are this chain's. + // After the agent's earlier deliveries: every stream's chain + // starts from the agent's latest link, so this waits for the + // turn's earlier rounds and for an earlier turn's posts still + // queued. await turnSpeechChain; if (turnProseRouting === 'disabled') { console.error(`[routing] ${agent.name}: text-only prose NOT routed (proseRouting=disabled)`); - this.recordProseSuppression(agent.name, 1); + this.recordProseSuppression(streamRouting, 1); } else if (turnSilenced && turnProseRouting !== 'explicit') { // Only reachable on a guard recovery that carried the // same-turn send suppression, or a suppressProse trigger // (a fresh text-only turn starts unsilenced). Explicit // mode is exempt, as mid-turn. console.error(`[routing] ${agent.name}: text-only prose NOT routed (turn silenced)`); - this.recordProseSuppression(agent.name, 1); + this.recordProseSuppression(streamRouting, 1); } else if (turnProseRouting === 'hybrid') { const locus = resolveTurnLocus(); console.error(`[prose] ${agent.name}: text-only turn -> hybrid prose gateway`); try { - await this.deliverHybridProse(agent, speechText, locus, true); + await this.deliverHybridProse(agent, speechText, locus, true, finalRouting); } catch (err) { console.error('text-only hybrid prose delivery failed:', err); } } else if (turnProseRouting === 'explicit') { console.error(`[prose] ${agent.name}: text-only turn -> prose gateway`); try { - await this.deliverAdmittedProse(agent, admittedSpeech!); + await this.deliverAdmittedProse(agent, admittedSpeech!, { routing: finalRouting }); } catch (err) { console.error('text-only prose delivery failed:', err); } @@ -10774,7 +10895,7 @@ export class AgentFramework { ); try { const outcome = await this.channelRegistry.routeSpeech(agent.name, speechText, locus); - this.recordProseDelivery(agent.name, outcome); + this.recordProseDelivery(streamRouting, outcome); } catch (err) { console.error('speech routing failed:', err); } @@ -10815,16 +10936,16 @@ export class AgentFramework { console.error( `[routing] ${agent.name}: tool-call trailing prose NOT routed (proseRouting=disabled)`, ); - this.recordProseSuppression(agent.name, segments.length); + this.recordProseSuppression(streamRouting, segments.length); } } else if (turnProseRouting === 'hybrid') { if (silenced && segments.length > 0) { - this.recordProseSuppression(agent.name, segments.length); + this.recordProseSuppression(streamRouting, segments.length); } else if (segments.length > 0) { const locus = resolveTurnLocus(); for (const seg of segments) { try { - await this.deliverHybridProse(agent, seg, locus, true); + await this.deliverHybridProse(agent, seg, locus, true, finalRouting); } catch (err) { console.error('trailing hybrid prose delivery failed:', err); } @@ -10837,7 +10958,7 @@ export class AgentFramework { ); for (const envelopes of admittedSegments) { try { - await this.deliverAdmittedProse(agent, envelopes); + await this.deliverAdmittedProse(agent, envelopes, { routing: finalRouting }); } catch (err) { console.error('trailing prose delivery failed:', err); } @@ -10849,7 +10970,7 @@ export class AgentFramework { `(${silenced ? 'silencing tool / explicit send' : 'no trailing prose'})`, ); if (silenced && segments.length > 0) { - this.recordProseSuppression(agent.name, segments.length); + this.recordProseSuppression(streamRouting, segments.length); } } else { // Reuse the locus pinned at the turn's first live-routed @@ -10867,7 +10988,7 @@ export class AgentFramework { for (const seg of segments) { try { const outcome = await this.channelRegistry.routeSpeech(agent.name, seg, locus); - this.recordProseDelivery(agent.name, outcome); + this.recordProseDelivery(streamRouting, outcome); } catch (err) { console.error('speech routing failed:', err); } @@ -10879,28 +11000,21 @@ export class AgentFramework { // doesn't depend on the status field, so it correctly runs after. // Logical turn end (normal completion): drop the `[delivered]` - // receipt. All deliveries are settled — the live chain was - // awaited before trailing dispatch, and trailing/text-only - // segments were awaited in-loop. Locus mode only; explicit-mode - // envelopes acknowledge themselves through the prose gateway. + // receipt, once every delivery has settled. The text-only and + // trailing paths awaited the chain when there was final text; a + // turn whose last response is empty awaits it here, so the + // receipt and the continuation see its last rounds' deliveries. + // Locus and hybrid mode; explicit-mode envelopes acknowledge + // themselves through the prose gateway. + await turnSpeechChain; if (!trigger?.suppressProse && turnProseRouting !== 'explicit') { - this.appendProseDeliveryReceipt(agent); + this.appendProseDeliveryReceipt(agent, streamRouting); } - // Explicit-prose `!` continuation: a prose segment this turn asked - // to keep going (`>>#x !` / `>>private !`) — start another turn - // now instead of pausing until the next external event. This gives - // prose the same continuation ability tool rounds have (wanted for + // A `!` continuation this turn asked for. This gives prose the + // same continuation ability tool rounds have (wanted for // robotics-style loops where an end-of-turn pause is harmful). - if (this.proseContinuations.delete(agent.name)) { - console.error(`[prose] ${agent.name}: '!' continuation — requesting immediate next turn`); - this.pendingRequests.push({ - agentName: agent.name, - reason: 'prose-continuation', - source: 'framework', - timestamp: Date.now(), - }); - } + this.requestProseContinuation(agent, finalRouting); break; } @@ -10952,7 +11066,7 @@ export class AgentFramework { stopReason: 'exhausted', speech: '', error: 'Turn abandoned by operator quiesce', - }); + }, this.agents.get(agent.name) === agent && agent.streamId === myStreamId ? streamRouting.logical : undefined); this.emitTrace({ type: 'inference:aborted', agentName: agent.name, @@ -10981,7 +11095,7 @@ export class AgentFramework { stopReason: 'exhausted', speech: '', error: err.message, - }); + }, this.agents.get(agent.name) === agent && agent.streamId === myStreamId ? streamRouting.logical : undefined); this.emitTrace({ type: 'inference:exhausted', agentName: agent.name, @@ -11035,7 +11149,7 @@ export class AgentFramework { stopReason: 'exhausted', speech: '', error: 'Turn abandoned by operator quiesce', - }); + }, streamRouting.logical); } this.emitTrace({ type: 'inference:aborted', @@ -11048,13 +11162,22 @@ export class AgentFramework { // endTurn IS a logical turn end — earlier rounds may have // live-routed prose (narrate → skip_reply is a real shape), // so settle the delivery chain and drop the receipt. A - // budget restart continues the same logical turn: the - // deliveries map persists (cleared only at fresh turn - // start) and the continuation stream's end writes ONE - // receipt for the whole turn. + // budget restart keeps the same turn's routing and tallies; + // the continuation stream's end writes one receipt for it. if (cancelKind === 'turn_ended' && !trigger?.suppressProse && turnProseRouting !== 'explicit') { await turnSpeechChain; - this.appendProseDeliveryReceipt(agent); + this.appendProseDeliveryReceipt(agent, streamRouting); + } + // A `!` an earlier round asked for still starts its + // continuation when a tool ends the turn, once the turn's + // deliveries have run. Explicit mode doesn't hold the turn + // for them, so it checks at the end of its chain. + if (cancelKind === 'turn_ended') { + if (turnProseRouting === 'explicit') { + void turnSpeechChain.then(() => this.requestProseContinuation(agent, streamRouting)); + } else { + this.requestProseContinuation(agent, streamRouting); + } } return; } @@ -11070,7 +11193,7 @@ export class AgentFramework { stopReason: 'exhausted', speech: '', error: `Stream aborted: ${reason}`, - }); + }, streamRouting.logical); this.emitTrace({ type: 'inference:exhausted', agentName: agent.name, @@ -11216,7 +11339,7 @@ export class AgentFramework { stopReason: 'exhausted', speech: '', error: err.message, - }); + }, streamRouting.logical); this.emitTrace({ type: 'inference:exhausted', agentName: agent.name, @@ -15821,7 +15944,7 @@ export class AgentFramework { } else if (agent?.proseRouting === 'explicit') { // Explicit mode: announce to the turn's sticky prose target if the // model has set one; otherwise stay quiet — never guess a channel. - const target = this.proseTargetPins.get(agentName); + const target = this.turnRouting.get(agentName)?.pin; if (target) { this.channelRegistry.routeSpeech(agentName, text, target).catch((err) => { console.error('[sleep] announce failed:', err instanceof Error ? err.message : err); diff --git a/test/conversation-routing.test.ts b/test/conversation-routing.test.ts index cd55d3c4d..489a94e87 100644 --- a/test/conversation-routing.test.ts +++ b/test/conversation-routing.test.ts @@ -361,6 +361,11 @@ describe('Conversation routing', () => { framework.getAgent('conversation-slack-D3-g1'), null, 'closed fork should be disposed once its closure turn finished', ); + const internals = framework as unknown as { + turnRouting: Map; proseSpeechChains: Map; + }; + assert.equal(internals.turnRouting.has('conversation-slack-D3-g1'), false, 'the closed fork leaves no routing'); + assert.equal(internals.proseSpeechChains.has('conversation-slack-D3-g1'), false, 'the closed fork leaves no speech chain'); // Next DM spawns a fresh generation. membrane.pushResponse(createMockResponse([{ type: 'text', text: 'hi again' }])); diff --git a/test/ephemeral-settle.test.ts b/test/ephemeral-settle.test.ts index 7fe53756c..ce029a56c 100644 --- a/test/ephemeral-settle.test.ts +++ b/test/ephemeral-settle.test.ts @@ -144,6 +144,11 @@ describe('runEphemeralToCompletion settle signal', () => { toolCallsCount: 0, }); assert.strictEqual(framework.getAgent(agent.name), null); + const internals = framework as unknown as { + turnRouting: Map; proseSpeechChains: Map; + }; + assert.equal(internals.turnRouting.has(agent.name), false, 'disposal removes turn routing'); + assert.equal(internals.proseSpeechChains.has(agent.name), false, 'disposal removes the speech chain'); } finally { await stopAndRemove(framework, tempDir); } diff --git a/test/explicit-prose-routing.test.ts b/test/explicit-prose-routing.test.ts index a7cc06210..b5b325520 100644 --- a/test/explicit-prose-routing.test.ts +++ b/test/explicit-prose-routing.test.ts @@ -352,12 +352,741 @@ describe('explicit prose routing', () => { await framework.stop(); }); + + type RoutingState = { + pin?: string; destination: string | null; hybridHold: boolean; + logical: { continuation: boolean; deliveries: string[]; suppressed: number }; + }; + + async function within(promise: Promise, label: string, ms = 5000): Promise { + let timer: NodeJS.Timeout | undefined; + try { + return await Promise.race([promise, new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error('Timed out waiting for ' + label)), ms); + })]); + } finally { + clearTimeout(timer); + } + } + + function nextTrace(framework: AgentFramework, type: string) { + let resolve!: () => void; + const promise = new Promise((done) => { resolve = done; }); + const off = framework.onTrace((event) => { + if (event.type === type && (event as { agentName?: string }).agentName === 'assistant') { + off(); + resolve(); + } + }); + return { promise, dispose: off }; + } + + async function withQueuedTurns( + mode: 'explicit' | 'hybrid', + lateText: string, + nextResponses: ReturnType[], + inspect: (fixture: { + framework: AgentFramework; routed: Array<{ text: string; locus: string | null }>; + earlier: RoutingState; next: RoutingState; turns: () => number; + release: () => void; drain: () => Promise; + }) => Promise, + ): Promise { + const arrows = mode === 'explicit' ? '>>' : '>>>'; + membrane.pushResponse(createMockResponse([ + { type: 'text', text: arrows + '#alpha first' }, + { type: 'tool_use', id: 'q1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: lateText }, + { type: 'tool_use', id: 'q2', name: 'robot--move', input: { dir: 'south' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'q3', name: 'skip_reply', input: {} }, + ] as ContentBlock[], 'tool_use')); + + const framework = await createFramework(mode); + const routed = stubRegistry(framework); + const skip = answerSkipReply(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const route = registry.routeSpeech as (agent: string, text: string, locus?: string | null) => Promise; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + registry.routeSpeech = async (agent: string, text: string, locus?: string | null) => { + const outcome = await route(agent, text, locus); + if (text.trim() === 'first') await held; + return outcome; + }; + let turns = 0; + framework.onTrace((event) => { + if (event.type !== 'inference:started') return; + turns++; + if (turns === 2) for (const response of nextResponses) membrane.pushResponse(response); + // Serve a bounce wake or an unexpected continuation without hanging. + if (turns === 3) membrane.pushResponse(createMockResponse([{ type: 'text', text: arrows + 'skip_reply' }])); + }); + const internals = framework as unknown as { + turnRouting: Map; pendingRequests: InferenceRequest[]; + processInferenceRequests(): Promise; + }; + const processRequests = internals.processInferenceRequests.bind(framework); + let nextWakeQueued = false; + let resolveBlocked!: () => void; + const blocked = new Promise((resolve) => { resolveBlocked = resolve; }); + internals.processInferenceRequests = async () => { + await processRequests(); + if (nextWakeQueued && turns === 1 && internals.pendingRequests.some((r) => r.agentName === 'assistant')) { + resolveBlocked(); + } + }; + const ended = nextTrace(framework, 'inference:turn_ended'); + let completed: ReturnType | undefined; + let started: ReturnType | undefined; + let stopped = false; + let driving: Promise | undefined; + const drain = async () => { + stopped = true; + if (driving) await within(driving, 'the queued turns to drain', 10000); + }; + try { + trigger(framework); + driving = (async () => { + while (!stopped) { + await framework.runUntilIdle(); + if (!stopped) await new Promise((resolve) => setImmediate(resolve)); + } + })(); + void driving.catch(() => {}); + await within(ended.promise, 'the first turn to end'); + assert.equal(skip.calls, 1, 'skip_reply ended the first turn'); + const earlier = internals.turnRouting.get('assistant')!; + completed = nextTrace(framework, 'inference:completed'); + started = nextTrace(framework, 'inference:started'); + nextWakeQueued = true; + trigger(framework); + if (mode === 'hybrid') { + const disposition = await within(Promise.race([ + blocked.then(() => 'blocked'), started.promise.then(() => 'started'), + ]), 'the next wake to reach scheduling'); + assert.equal(disposition, 'blocked', 'hybrid retains the earlier turn token until its receipt tail drains'); + assert.equal(internals.turnRouting.get('assistant'), earlier, 'the earlier turn still owns routing'); + assert.deepEqual(routed.map((r) => r.text.trim()), ['first']); + release(); + await within(completed.promise, 'the next inference after release'); + assert.equal(turns, 2, 'hybrid starts the next turn after the earlier teardown'); + } else { + await within(completed.promise, 'the next inference to complete'); + assert.equal(turns, 2, 'explicit starts the next turn without waiting'); + assert.deepEqual(routed.map((r) => r.text.trim()), ['first'], + "only the first post has entered routing while its confirmation is held"); + } + await inspect({ framework, routed, earlier, next: internals.turnRouting.get('assistant')!, + turns: () => turns, release, drain }); + } finally { + ended.dispose(); + completed?.dispose(); + started?.dispose(); + internals.processInferenceRequests = processRequests; + release(); + try { await drain(); } finally { await framework.stop(); } + } + } + + + async function withHeldHybridCallback( + lateText: string | null, + inspect: (fixture: { + framework: AgentFramework; routed: Array<{ text: string; locus: string | null }>; + gateway: { + deliverHybridProse(agent: unknown, text: string, locus: string, allowLocus: boolean, routing: RoutingState): Promise; + appendProseDeliveryReceipt(agent: unknown, routing: RoutingState): void; + }; + earlier: RoutingState; next: RoutingState; agent: unknown; + pending: Promise; release: () => void; + }) => Promise, + ): Promise { + const framework = await createFramework('hybrid'); + const routed = stubRegistry(framework); + const gateway = framework as unknown as { + turnRouting: Map; + deliverHybridProse(agent: unknown, text: string, locus: string, allowLocus: boolean, routing: RoutingState): Promise; + appendProseDeliveryReceipt(agent: unknown, routing: RoutingState): void; + }; + const agent = framework.getAgent('assistant')!; + const earlier: RoutingState = { destination: null, hybridHold: false, logical: { continuation: false, deliveries: [], suppressed: 0 } }; + const next: RoutingState = { ...earlier, logical: { continuation: false, deliveries: [], suppressed: 0 } }; + gateway.turnRouting.set('assistant', earlier); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const route = registry.routeSpeech as (agent: string, text: string, locus?: string | null) => Promise; + let release!: () => void; + let resolveEntered!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const entered = new Promise((resolve) => { resolveEntered = resolve; }); + registry.routeSpeech = async (name: string, text: string, locus?: string | null) => { + const outcome = await route(name, text, locus); + resolveEntered(); + await held; + return outcome; + }; + const first = gateway.deliverHybridProse(agent, '>>>#alpha first', 'world:commons', true, earlier); + const pending = lateText === null ? first + : first.then(() => gateway.deliverHybridProse(agent, lateText, 'world:commons', true, earlier)); + void pending.catch(() => {}); + try { + await within(entered, 'the earlier callback to enter routing'); + // Exercise callback ownership directly. Normal hybrid scheduling keeps + // its turn token until the receipt tail drains, so fresh turns serialize. + gateway.turnRouting.set('assistant', next); + await inspect({ framework, routed, gateway, earlier, next, agent, pending, release }); + } finally { + release(); + try { await within(pending, 'the held callback to drain'); } finally { await framework.stop(); } + } + } + + for (const mode of ['explicit', 'hybrid'] as const) { + const title = mode === 'explicit' + ? "explicit: prose a turn leaves queued follows that turn's destination, and the next turn's posts wait behind it" + : "hybrid: the next turn waits for the ended turn's receipt tail, and posts keep their destinations and order"; + it(title, async () => { + const arrows = mode === 'explicit' ? '>>' : '>>>'; + await withQueuedTurns(mode, 'and then this', [ + createMockResponse([{ type: 'text', text: arrows + '#cafe next' }]), + ], async ({ framework, routed, release, drain }) => { + release(); + await drain(); + assert.deepEqual(routed.map((r) => ({ text: r.text.trim(), locus: r.locus })), [ + { text: 'first', locus: 'chan-alpha' }, + { text: 'and then this', locus: 'chan-alpha' }, + { text: 'next', locus: 'discord:g:cafe' }, + ], "each post follows its own turn's destination, in delivery order"); + assert.equal(context(framework).bounces, 0, 'nothing bounces into the next turn'); + }); + }); + } + + it("a queued round's late continuation starts once its tool-ended turn's deliveries have run", async () => { + await withQueuedTurns('explicit', '>>#alpha ! late', [ + createMockResponse([{ type: 'text', text: '>>#cafe next' }]), + ], async ({ earlier, next, turns, release, drain }) => { + assert.equal(earlier.logical.continuation, false, 'the continuation has not run before release'); + release(); + for (const end = Date.now() + 5000; turns() < 3 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + await drain(); + assert.equal(turns(), 3, 'the continuation starts its own turn, once'); + assert.equal(earlier.logical.continuation, false, "the earlier turn's request was taken from its own object"); + assert.equal(next.logical.continuation, false, 'the next turn never receives that continuation'); + }); + }); + + for (const mode of ['explicit', 'hybrid'] as const) { + it(`${mode}: a ' !' an earlier round asked for starts its continuation when skip_reply ends the turn`, async () => { + const arrows = mode === 'explicit' ? '>>' : '>>>'; + membrane.pushResponse(createMockResponse([ + { type: 'text', text: arrows + '#alpha ! keep going' }, + { type: 'tool_use', id: 'k1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'k2', name: 'skip_reply', input: {} }, + ] as ContentBlock[], 'tool_use')); + + const framework = await createFramework(mode); + const routed = stubRegistry(framework); + const skip = answerSkipReply(framework); + // Count turns before feedTurn subscribes: a listener that unsubscribes + // during an event hides that event from the listener after it. + let turns = 0; + framework.onTrace((e) => { + if ((e as { type: string }).type === 'inference:started') turns++; + }); + feedTurn(framework, 2, [{ type: 'text', text: arrows + 'skip_reply' }] as ContentBlock[]); + const reasons: string[] = []; + const internals = framework as unknown as { pendingRequests: InferenceRequest[]; processInferenceRequests(): Promise }; + const process = internals.processInferenceRequests.bind(framework); + internals.processInferenceRequests = async () => { + for (const request of internals.pendingRequests) reasons.push(request.reason); + await process(); + }; + + trigger(framework); + await framework.runUntilIdle(); + for (const end = Date.now() + 5000; turns < 2 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + await framework.runUntilIdle(); + } + + assert.equal(skip.calls, 1, 'skip_reply ended the first turn'); + assert.deepEqual(routed.map((r) => r.text.trim()), ['keep going']); + assert.ok(reasons.includes('prose-continuation'), 'the continuation was requested'); + assert.equal(turns, 2, 'and it started one more turn'); + + await framework.stop(); + }); + } + + for (const mode of ['explicit', 'hybrid'] as const) { + it(`${mode}: a tool-ended turn's ' !' queued behind a held post starts its continuation once that post settles`, async () => { + const arrows = mode === 'explicit' ? '>>' : '>>>'; + membrane.pushResponse(createMockResponse([ + { type: 'text', text: `${arrows}#alpha first\n${arrows}#alpha ! second` }, + { type: 'tool_use', id: 'w1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'w2', name: 'skip_reply', input: {} }, + ] as ContentBlock[], 'tool_use')); + const framework = await createFramework(mode); + const routed = stubRegistry(framework); + answerSkipReply(framework); + let turns = 0; + framework.onTrace((e) => { + if ((e as { type: string }).type === 'inference:started') turns++; + }); + feedTurn(framework, 2, [{ type: 'text', text: arrows + 'skip_reply' }] as ContentBlock[]); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const route = registry.routeSpeech as (agent: string, text: string, locus?: string | null) => Promise; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + let entered!: () => void; + const reached = new Promise((resolve) => { entered = resolve; }); + registry.routeSpeech = async (agent: string, text: string, locus?: string | null) => { + const outcome = await route(agent, text, locus); + if (text.trim() === 'first') { entered(); await held; } + return outcome; + }; + const internals = framework as unknown as { pendingRequests: InferenceRequest[] }; + const ended = nextTrace(framework, 'inference:turn_ended'); + trigger(framework); + const running = framework.runUntilIdle(); + try { + await within(reached, "the first envelope's post to start"); + await within(ended.promise, 'skip_reply to end the turn'); + assert.equal(turns, 1); + assert.ok(!internals.pendingRequests.some((r) => r.reason === 'prose-continuation'), + "no continuation while the ' !' envelope waits behind the held post"); + release(); + await within(running, 'the turn to settle'); + for (const end = Date.now() + 5000; turns < 2 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + await framework.runUntilIdle(); + } + assert.deepEqual(routed.map((r) => r.text.trim()), ['first', 'second']); + assert.equal(turns, 2, 'the continuation starts once the queued envelope has run'); + } finally { + ended.dispose(); + release(); + await framework.stop(); + } + }); + } + + for (const when of ['already set', 'set late'] as const) { + it(`hybrid: a ' !' an earlier stream asked for (${when}) still starts its continuation after a re-inference on unstreamed tool results`, async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'carrying on' }] as ContentBlock[])); + const framework = await createFramework('hybrid'); + const routed = stubRegistry(framework); + let turns = 0; + framework.onTrace((e) => { + if ((e as { type: string }).type === 'inference:started') turns++; + }); + feedTurn(framework, 2, [{ type: 'text', text: '>>>skip_reply' }] as ContentBlock[]); + const internals = framework as unknown as { + turnRouting: Map; pendingRequests: InferenceRequest[]; + proseSpeechChains: Map>; + openLogical: Map; + }; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + try { + // The turn's earlier stream asked for a continuation, or will once + // its queued delivery runs; the turn goes on in a re-inference, which + // starts a new stream with a fresh routing state while the earlier + // stream's logical turn is still open. + const earlier: RoutingState = { destination: null, hybridHold: false, logical: { continuation: when === 'already set', deliveries: [], suppressed: 0 } }; + internals.turnRouting.set('assistant', earlier); + internals.openLogical.set('assistant', earlier.logical); + if (when === 'set late') { + internals.proseSpeechChains.set('assistant', held.then(() => { earlier.logical.continuation = true; })); + } + internals.pendingRequests.push({ agentName: 'assistant', reason: 'tool_results_ready', source: 'framework', timestamp: Date.now() } as InferenceRequest); + const running = framework.runUntilIdle(); + if (when === 'set late') { + for (const end = Date.now() + 5000; turns < 1 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + assert.equal(turns, 1, 'the re-inference has started'); + assert.equal(earlier.logical.continuation, false, 'the earlier delivery has not run as the re-inference starts'); + release(); + } + await within(running, 'the re-inference to settle'); + for (const end = Date.now() + 5000; turns < 2 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + await framework.runUntilIdle(); + } + assert.deepEqual(routed.map((r) => r.text.trim()), ['carrying on']); + assert.equal(turns, 2, 'the re-inference completes, then the continuation it carried starts'); + assert.equal(earlier.logical.continuation, false, "the check took the earlier stream's flag"); + } finally { + release(); + await framework.stop(); + } + }); + } + + it("hybrid: a re-inference on unstreamed tool results receipts the earlier stream's late delivery with its own", async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'carrying on' }] as ContentBlock[])); + const framework = await createFramework('hybrid'); + const routed = stubRegistry(framework); + let turns = 0; + framework.onTrace((e) => { + if ((e as { type: string }).type === 'inference:started') turns++; + }); + const internals = framework as unknown as { + turnRouting: Map; pendingRequests: InferenceRequest[]; + proseSpeechChains: Map>; + openLogical: Map; + }; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + try { + // The earlier stream left its tool results pending, its logical turn + // still open; its delivery lands only after the re-inference has + // begun, into the logical turn the two share. + const earlier: RoutingState = { destination: null, hybridHold: false, logical: { continuation: false, deliveries: [], suppressed: 0 } }; + internals.turnRouting.set('assistant', earlier); + internals.openLogical.set('assistant', earlier.logical); + internals.proseSpeechChains.set('assistant', held.then(() => { earlier.logical.deliveries.push('chan-alpha'); })); + internals.pendingRequests.push({ agentName: 'assistant', reason: 'tool_results_ready', source: 'framework', timestamp: Date.now() } as InferenceRequest); + const running = framework.runUntilIdle(); + for (const end = Date.now() + 5000; turns < 1 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + assert.equal(turns, 1, 'the re-inference has started'); + assert.deepEqual(earlier.logical.deliveries, [], 'the earlier delivery has not landed yet'); + release(); + await within(running, 'the re-inference to settle'); + assert.deepEqual(routed.map((r) => r.text.trim()), ['carrying on']); + const receipt = context(framework).texts.filter((t) => t.startsWith('[delivered]')); + assert.equal(receipt.length, 1, `one receipt for the logical turn: ${receipt.join(' | ')}`); + assert.match(receipt[0]!, /chan-alpha/, "it reports the earlier stream's late delivery"); + assert.match(receipt[0]!, /world:commons/, "and the re-inference's own"); + } finally { + release(); + await framework.stop(); + } + }); + + for (const mode of ['explicit', 'hybrid'] as const) { + const p = mode === 'explicit' ? '>>' : '>>>'; + it(`${mode}: an error retry keeps a ' !' whose envelope is delivered after the retry starts`, async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: `${p}#alpha first\n${p}#alpha ! second` }, + { type: 'tool_use', id: 'retry-held', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([] as ContentBlock[])); + const framework = await createFramework(mode); + const routed = stubRegistry(framework); + const internals = framework as unknown as { + proseSpeechChains: Map>; + channelRegistry: { routeSpeech: (...args: unknown[]) => Promise }; + }; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + let reached!: () => void; + const entered = new Promise((resolve) => { reached = resolve; }); + const route = internals.channelRegistry.routeSpeech; + internals.channelRegistry.routeSpeech = async (...args: unknown[]) => { + const result = await route(...args); + if (String(args[1]).trim() === 'first') { reached(); await held; } + return result; + }; + let firstChain: Promise | undefined; + let restarted!: () => void; + const retryStarted = new Promise((resolve) => { restarted = resolve; }); + let starts = 0; + framework.onTrace((event) => { + if (event.type === 'inference:failed') firstChain = internals.proseSpeechChains.get('assistant'); + if (event.type === 'inference:started') { + starts++; + if (starts === 2) { + membrane.pushResponse(createMockResponse([{ type: 'text', text: `${p}skip_reply retry done` }])); + restarted(); + } + if (starts === 3) membrane.pushResponse(createMockResponse([{ type: 'text', text: `${p}skip_reply continuation done` }])); + } + }); + // The first stream ends in a retryable error instead of completing; the + // retry's own end waits for the first stream's queued deliveries, so the + // held ' !' envelope runs after the retry has started. + const makeStream = membrane.streamYielding.bind(membrane); + let streams = 0; + membrane.streamYielding = (request, options) => { + const stream = makeStream(request, options); + const index = ++streams; + const iterate = stream[Symbol.asyncIterator].bind(stream); + stream[Symbol.asyncIterator] = async function* () { + for await (const event of { [Symbol.asyncIterator]: iterate }) { + if (event.type === 'complete' && index === 1) { + yield { type: 'error' as const, error: new Error('synthetic transient stream error') }; + return; + } + if (event.type === 'complete' && index === 2) { + assert.ok(firstChain, 'the first stream queued a delivery chain'); + await firstChain; + } + yield event; + } + }; + return stream; + }; + let running: Promise | undefined; + try { + trigger(framework); + running = framework.runUntilIdle(); + void running.catch(() => {}); + await within(entered, 'the held first post'); + await within(retryStarted, 'the error retry to start', 10000); + assert.equal(starts, 2); + release(); + await within(running, 'the retry and continuation to finish', 10000); + assert.deepEqual(routed.map((r) => r.text.trim()), ['first', 'second']); + assert.equal(membrane.calls.length, 3, "the late ' !' survives the error retry and starts a continuation"); + } finally { + release(); + if (running) await running; + await framework.stop(); + } + }); + } + + it("the non-streaming fallback leaves the scheduling turn's logical turn open for the re-inference", async () => { + // Adapted from a reviewer's producer test: the agent's direct inference + // API leaves tools pending with no yielding stream, and the tool result + // goes through the framework's real event handling. + const framework = await createFramework('hybrid'); + stubRegistry(framework); + const agent = framework.getAgent('assistant')!; + const internals = framework as unknown as { + routingOf(name: string): RoutingState; + openLogical: Map; + pendingRequests: InferenceRequest[]; + handleProcessEvent(event: ProcessEvent): Promise; + }; + const origin = internals.routingOf('assistant').logical; + origin.continuation = true; + origin.deliveries.push('chan-alpha'); + // The turn that ran the tools is open, as its fresh start left it. + internals.openLogical.set('assistant', origin); + const response = createMockResponse([ + { type: 'tool_use', id: 'unstreamed-one', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use'); + (membrane as unknown as { stream: () => Promise }).stream = async () => response; + try { + agent.getContextManager().addMessage('user', [{ type: 'text', text: 'Take the direct inference path.' }]); + await agent.runInference(framework.getAllTools()); + assert.equal(agent.state.status, 'waiting_for_tools'); + assert.equal((agent.state as { stream?: unknown }).stream, undefined, 'the direct API has no yielding stream'); + agent.addAssistantResponse(response.content); + await internals.handleProcessEvent({ + type: 'tool-result', source: 'robot', moduleName: 'robot', agentName: 'assistant', callId: 'unstreamed-one', + result: { success: true, data: { ok: true } }, + } as ProcessEvent); + assert.ok(internals.pendingRequests.some((r) => r.reason === 'tool_results_ready'), 'the fallback scheduled the re-inference'); + assert.equal(internals.openLogical.get('assistant'), origin, 'without settling the turn it continues'); + } finally { + await framework.stop(); + } + }); + + it("a settle closes the open logical turn only when it is the settling stream's own, and stop() clears open turns", async () => { + const framework = await createFramework('hybrid'); + const internals = framework as unknown as { + openLogical: Map; + settleAgent(name: string, outcome: { stopReason: 'completed'; speech: string }, closes: RoutingState['logical'] | undefined): void; + }; + const open: RoutingState['logical'] = { continuation: true, deliveries: [], suppressed: 0 }; + const stale: RoutingState['logical'] = { continuation: false, deliveries: [], suppressed: 0 }; + internals.openLogical.set('assistant', open); + // A settle from a stream a newer one replaced passes its own logical + // turn, or none; neither closes the open one. + internals.settleAgent('assistant', { stopReason: 'completed', speech: '' }, stale); + internals.settleAgent('assistant', { stopReason: 'completed', speech: '' }, undefined); + assert.equal(internals.openLogical.get('assistant'), open, "a stale settle leaves the newer turn open"); + internals.settleAgent('assistant', { stopReason: 'completed', speech: '' }, open); + assert.equal(internals.openLogical.has('assistant'), false, "the turn's own settle closes it"); + internals.openLogical.set('assistant', open); + await framework.stop(); + assert.equal(internals.openLogical.size, 0, 'stop() clears every open logical turn'); + }); + + it('a silent start leaves an open logical turn to the next start that routes prose', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'quiet' }] as ContentBlock[])); + const framework = await createFramework('hybrid'); + stubRegistry(framework); + let turns = 0; + framework.onTrace((e) => { + if ((e as { type: string }).type === 'inference:started') turns++; + }); + const internals = framework as unknown as { + turnRouting: Map; pendingRequests: InferenceRequest[]; + openLogical: Map; + }; + try { + const open: RoutingState['logical'] = { continuation: true, deliveries: [], suppressed: 0 }; + internals.openLogical.set('assistant', open); + internals.pendingRequests.push({ agentName: 'assistant', reason: 'heartbeat', source: 'framework', timestamp: Date.now(), suppressProse: true } as InferenceRequest); + // Bounded: a silent start that took the open turn's ' !' would start a + // continuation with no response queued, and this would wait for good. + await within(framework.runUntilIdle(), 'the silent turn to settle', 3000); + assert.equal(turns, 1); + assert.notEqual(internals.turnRouting.get('assistant')?.logical, open, 'the silent start kept a logical turn of its own'); + assert.equal(internals.openLogical.get('assistant'), open, 'and neither took nor closed the open one'); + assert.equal(open.continuation, true, "the open turn's ' !' is untouched"); + // The next start that routes prose continues it, and its ' !' runs. + // (feedTurn counts starts from here: the prose turn is 1, the + // continuation 2.) + membrane.pushResponse(createMockResponse([{ type: 'text', text: '>>>skip_reply' }] as ContentBlock[])); + feedTurn(framework, 2, [{ type: 'text', text: '>>>skip_reply' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + for (const end = Date.now() + 5000; turns < 3 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + await framework.runUntilIdle(); + } + assert.equal(turns, 3, "the next prose turn continued the open turn and took its ' !'"); + } finally { + await framework.stop(); + } + }); + + for (const mode of ['explicit', 'hybrid'] as const) { + it(`${mode}: the wake that consumes unstreamed tool results continues the turn that scheduled them`, async () => { + const p = mode === 'explicit' ? '>>' : '>>>'; + membrane.pushResponse(createMockResponse([{ type: 'text', text: `${p}skip_reply carrying on` }] as ContentBlock[])); + const framework = await createFramework(mode); + stubRegistry(framework); + let turns = 0; + framework.onTrace((e) => { + if ((e as { type: string }).type === 'inference:started') turns++; + }); + feedTurn(framework, 2, [{ type: 'text', text: `${p}skip_reply` }] as ContentBlock[]); + const internals = framework as unknown as { + turnRouting: Map; + openLogical: Map; + }; + try { + // The scheduling turn asked for a continuation and left its tool + // results pending; the scheduler then starts the agent with another + // wake, which consumes them. + const scheduling: RoutingState = { destination: null, hybridHold: false, logical: { continuation: true, deliveries: [], suppressed: 0 } }; + internals.openLogical.set('assistant', scheduling.logical); + internals.turnRouting.set('assistant', { destination: null, hybridHold: false, logical: { continuation: false, deliveries: [], suppressed: 0 } }); + trigger(framework); + await framework.runUntilIdle(); + for (const end = Date.now() + 5000; turns < 2 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + await framework.runUntilIdle(); + } + assert.equal(turns, 2, "the consuming turn took the scheduling turn's ' !'"); + assert.equal(scheduling.logical.continuation, false); + assert.equal(internals.openLogical.has('assistant'), false, 'and the turns it ran have settled'); + } finally { + await framework.stop(); + } + }); + } + + for (const boundary of ['stop', 'dispose'] as const) { + it(`a tool-ended explicit turn's late ' !' takes no new turn after ${boundary === 'stop' ? 'stop()' : 'its agent is disposed'}`, async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#alpha ! first' }, + { type: 'tool_use', id: 'p1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'p2', name: 'skip_reply', input: {} }, + ] as ContentBlock[], 'tool_use')); + const framework = await createFramework(); + stubRegistry(framework); + answerSkipReply(framework); + const internals = framework as unknown as { + channelRegistry: { routeSpeech: (...args: unknown[]) => Promise }; + proseSpeechChains: Map>; + turnRouting: Map; + pendingRequests: InferenceRequest[]; + disposeConversationAgent(name: string): void; + }; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + let entered!: () => void; + const reached = new Promise((resolve) => { entered = resolve; }); + internals.channelRegistry.routeSpeech = async () => { + entered(); + await held; + return { delivered: true, channelId: 'chan-alpha' }; + }; + try { + trigger(framework); + await within(framework.runUntilIdle(), 'the tool-ended turn to settle'); + await within(reached, 'the queued delivery to reach routing'); + const chain = internals.proseSpeechChains.get('assistant')!; + const routing = internals.turnRouting.get('assistant')!; + assert.equal(routing.logical.continuation, true, "the held delivery's ' !' is on its turn's object"); + assert.equal(internals.pendingRequests.length, 0); + if (boundary === 'stop') await framework.stop(); + else internals.disposeConversationAgent('assistant'); + release(); + await within(chain, 'the queued delivery'); + await new Promise((resolve) => setImmediate(resolve)); + assert.equal(internals.pendingRequests.filter((r) => r.reason === 'prose-continuation').length, 0, + 'the late check adds no request'); + assert.equal(routing.logical.continuation, false, 'it took the flag and dropped it'); + } finally { + release(); + await framework.stop(); + } + }); + } + + it("hybrid: a held callback's late malformed envelope holds only its supplied turn", async () => { + await withHeldHybridCallback('>>>', async ({ earlier, next, routed, gateway, agent, pending, release }) => { + assert.equal(earlier.hybridHold, false, 'the malformed envelope has not run before release'); + release(); + await within(pending, 'the late malformed envelope'); + assert.equal(earlier.hybridHold, true, 'the late hold belongs to the supplied earlier turn'); + assert.equal(next.hybridHold, false, 'the newer object stays unheld'); + await gateway.deliverHybridProse(agent, 'next', 'world:commons', true, next); + assert.deepEqual(routed.map((r) => ({ text: r.text.trim(), locus: r.locus })), [ + { text: 'first', locus: 'chan-alpha' }, + { text: 'next', locus: 'world:commons' }, + ], 'the newer callback still publishes to its own locus'); + }); + }); + + it("hybrid callbacks and receipts keep each turn's counts while a post is held", async () => { + await withHeldHybridCallback(null, async ({ framework, earlier, next, gateway, agent, pending, release }) => { + await gateway.deliverHybridProse(agent, 'quiet without a destination', 'world:commons', false, next); + release(); + await within(pending, 'the earlier delivery confirmation'); + gateway.appendProseDeliveryReceipt(agent, earlier); + gateway.appendProseDeliveryReceipt(agent, next); + const receipts = context(framework).texts.filter((text) => text.startsWith('[delivered]')); + assert.equal(receipts.length, 2, 'each supplied turn gets its own receipt'); + const delivered = receipts.find((text) => text.includes('chan-alpha')); + const suppressed = receipts.find((text) => text.includes('suppressed')); + assert.ok(delivered, "the earlier turn's delivery is acknowledged"); + assert.doesNotMatch(delivered, /suppressed/, "the earlier receipt excludes the newer suppression"); + assert.ok(suppressed, "the newer turn's suppression is acknowledged"); + assert.match(suppressed, /1 plain-speech segment.*suppressed/); + assert.doesNotMatch(suppressed, /chan-alpha/, "the newer receipt excludes the earlier delivery"); + }); + }); + it("a private round's ' !' leaves no continuation request behind", async () => { const framework = await createFramework(); const routed = stubRegistry(framework); const internals = framework as unknown as { deliverProse(agent: unknown, text: string, round?: { keepPrivate?: boolean }): Promise; - proseContinuations: Set; + turnRouting: Map; }; const agent = framework.getAgent('assistant')!; @@ -365,12 +1094,12 @@ describe('explicit prose routing', () => { // so the round's delivery can run once the next turn has begun. await internals.deliverProse(agent, '>>#alpha ! kept private', { keepPrivate: true }); assert.equal(routed.length, 0, 'nothing is delivered'); - assert.equal(internals.proseContinuations.has('assistant'), false, 'and no continuation is asked for'); + assert.equal(internals.turnRouting.get('assistant')?.logical.continuation ?? false, false, 'and no continuation is asked for'); // The same envelope outside a private round asks for one. await internals.deliverProse(agent, '>>#alpha ! sent'); assert.equal(routed.length, 1); - assert.equal(internals.proseContinuations.has('assistant'), true); + assert.equal(internals.turnRouting.get('assistant')?.logical.continuation, true); await framework.stop(); }); @@ -459,7 +1188,7 @@ describe('explicit prose routing', () => { deliverProse(agent: unknown, text: string): Promise; admitProse(text: string): unknown[]; roundDestination(current: string | null, envelopes: unknown[]): string | null; - proseTargetPins: Map; + turnRouting: Map; }; const agent = framework.getAgent('assistant')!; const rounds: string[][] = [ @@ -472,11 +1201,11 @@ describe('explicit prose routing', () => { try { for (const segments of rounds) { for (const start of [null, 'chan-alpha']) { - internals.proseTargetPins.clear(); - if (start) internals.proseTargetPins.set('assistant', start); + internals.turnRouting.set('assistant', { pin: start ?? undefined, destination: null, hybridHold: false, + logical: { continuation: false, deliveries: [], suppressed: 0 } }); for (const segment of segments) await internals.deliverProse(agent, segment); const admitted = segments.flatMap((segment) => internals.admitProse(segment)); - assert.equal(internals.roundDestination(start, admitted), internals.proseTargetPins.get('assistant') ?? null, + assert.equal(internals.roundDestination(start, admitted), internals.turnRouting.get('assistant')?.pin ?? null, `${JSON.stringify(segments)} from ${start}`); } } @@ -489,7 +1218,8 @@ describe('explicit prose routing', () => { const framework = await createFramework('hybrid'); const routed = stubRegistry(framework); const gateway = framework as unknown as { - deliverHybridProse(agent: unknown, text: string, locus: string, allowLocus: boolean): Promise; + deliverHybridProse(agent: unknown, text: string, locus: string, allowLocus: boolean, routing: RoutingState): Promise; + turnRouting: Map; }; const agent = framework.getAgent('assistant')!; // The registry's resolver trims its argument, as the router's must here. @@ -498,7 +1228,13 @@ describe('explicit prose routing', () => { try { for (const text of cases) { routed.length = 0; - await gateway.deliverHybridProse(agent, text, 'chan-locus', true); + const queued: RoutingState = { destination: null, hybridHold: false, + logical: { continuation: false, deliveries: [], suppressed: 0 } }; + const live: RoutingState = { ...queued, logical: { continuation: false, deliveries: [], suppressed: 0 } }; + gateway.turnRouting.set('assistant', live); + await gateway.deliverHybridProse(agent, text, 'chan-locus', true, queued); + assert.equal(queued.hybridHold, true, 'the supplied turn receives the hold'); + assert.equal(live.hybridHold, false, 'the live turn stays unheld'); const router = new ProseStreamRouter({ mode: 'hybrid', initialTarget: 'chan-locus', resolve }); const streamed = [...router.feed(text), ...router.finish()]; assert.equal(routed.length, 0, `delivery bounces ${JSON.stringify(text)}`); @@ -513,16 +1249,25 @@ describe('explicit prose routing', () => { const framework = await createFramework('hybrid'); const routed = stubRegistry(framework); const gateway = framework as unknown as { - deliverHybridProse(agent: unknown, text: string, locus: string, allowLocus: boolean): Promise; + deliverHybridProse(agent: unknown, text: string, locus: string, allowLocus: boolean, routing: RoutingState): Promise; + turnRouting: Map; }; const agent = framework.getAgent('assistant')!; const resolve = (spec: string) => (spec.trim() === '#alpha' ? 'chan-alpha' : null); try { for (const prefix of ['>>>', '>>> ', '>>>\t']) { routed.length = 0; - await gateway.deliverHybridProse(agent, '>>>#alpha first', 'chan-locus', true); - await gateway.deliverHybridProse(agent, prefix, 'chan-locus', true); - await gateway.deliverHybridProse(agent, 'secret', 'chan-locus', true); + const queued: RoutingState = { destination: null, hybridHold: false, + logical: { continuation: false, deliveries: [], suppressed: 0 } }; + const live: RoutingState = { ...queued, logical: { continuation: false, deliveries: [], suppressed: 0 } }; + gateway.turnRouting.set('assistant', live); + await gateway.deliverHybridProse(agent, '>>>#alpha first', 'chan-locus', true, queued); + await gateway.deliverHybridProse(agent, prefix, 'chan-locus', true, queued); + await gateway.deliverHybridProse(agent, 'secret', 'chan-locus', true, queued); + assert.deepEqual(queued.logical.deliveries, ['chan-alpha'], 'the supplied turn records its delivery'); + assert.equal(queued.logical.suppressed, 1, 'the supplied turn records its suppression'); + assert.deepEqual(live.logical.deliveries, [], 'the live turn receives no delivery'); + assert.equal(live.logical.suppressed, 0, 'the live turn receives no suppression'); const router = new ProseStreamRouter({ mode: 'hybrid', initialTarget: 'chan-locus', resolve }); const streamed = [ ...router.feed('>>>#alpha first'), ...router.endSegment(), @@ -544,7 +1289,8 @@ describe('explicit prose routing', () => { const internals = framework as unknown as { channelRegistry: { resolveProseTarget: (spec: string) => unknown }; admitProse(text: string): Array<{ text: string; resolved?: { channelId?: string; error?: string } }>; - deliverAdmittedProse(agent: unknown, envelopes: unknown[]): Promise; + deliverAdmittedProse(agent: unknown, envelopes: unknown[], round: { routing: RoutingState }): Promise; + routingOf(name: string): RoutingState; }; const resolve = internals.channelRegistry.resolveProseTarget.bind(internals.channelRegistry); internals.channelRegistry.resolveProseTarget = (spec: string) => { @@ -554,7 +1300,7 @@ describe('explicit prose routing', () => { try { const admitted = internals.admitProse('>>#boom one\n>>#alpha two'); assert.match(admitted[0]!.resolved?.error ?? '', /directory unavailable/); - await internals.deliverAdmittedProse(framework.getAgent('assistant')!, admitted); + await internals.deliverAdmittedProse(framework.getAgent('assistant')!, admitted, { routing: internals.routingOf('assistant') }); assert.deepEqual(routed.map((r) => r.text), ['two'], 'the other envelope still goes out'); } finally { await framework.stop(); diff --git a/test/tool-result-guard.test.ts b/test/tool-result-guard.test.ts index a9599a66e..f92bdb6a7 100644 --- a/test/tool-result-guard.test.ts +++ b/test/tool-result-guard.test.ts @@ -811,58 +811,196 @@ test("explicit mode: a round's targets are resolved once, when it ends, for its } }); -test("a fresh turn that delivers nothing still replaces an earlier turn's speech chain for its restart", { timeout: 15000 }, async () => { - // An earlier turn's delivery that never settles: a stream resuming this - // turn must not queue behind it. + +test("a fresh turn and its restart wait behind the earlier turn's speech chain", { timeout: 15000 }, async () => { + const fresh = createMockResponse([ + { type: 'text', text: '>>#alpha fresh' }, + { type: 'tool_use', id: 'fresh', name: 'test--read', input: {} }, + ] as ContentBlock[], 'tool_use'); + fresh.usage = { inputTokens: 1, outputTokens: 1 }; const quiet = createMockResponse([ - { type: 'tool_use', id: 'one', name: 'test--read', input: {} }, + { type: 'tool_use', id: 'restart', name: 'test--read', input: {} }, ] as ContentBlock[], 'tool_use'); quiet.usage = { inputTokens: 100, outputTokens: 5 }; - // The reply names its destination, so nothing bounces into another turn. - const reply = createMockResponse([{ type: 'text', text: '>>#alpha continued' }] as ContentBlock[], 'end_turn'); - const h = await harness([[quiet], [reply]], { proseRouting: 'explicit', maxStreamTokens: 20 }); + const reply = createMockResponse([{ type: 'text', text: 'continued' }], 'end_turn'); + const h = await harness([[fresh, quiet], [reply]], { proseRouting: 'explicit', maxStreamTokens: 20 }); + const routed: string[] = []; + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const routeSpeech = async (_agent: string, text: string, channelId: string) => { + routed.push(text.trim()); + if (text.trim() === 'earlier') await held; + return { delivered: true, channelId }; + }; + (h.framework as unknown as { channelRegistry: unknown }).channelRegistry = new Proxy({ + resolveProseTarget: (spec: string) => spec === '#alpha' ? { channelId: 'chan-alpha' } : { error: 'unknown channel' }, + routeSpeech, + getDefaultPublishChannel: () => null, isChannelOpen: () => true, + getDescriptor: () => undefined, getChannelTools: () => [], + }, { get: (target, key: string) => key in target ? (target as Record)[key] : () => undefined }); + const prior = routeSpeech('assistant', 'earlier', 'chan-alpha').then(() => {}); (h.framework as unknown as { proseSpeechChains: Map> }) - .proseSpeechChains.set('assistant', new Promise(() => {})); + .proseSpeechChains.set('assistant', prior); + let resolveCompleted!: () => void; + const completed = new Promise((resolve) => { resolveCompleted = resolve; }); + const off = h.framework.onTrace((event) => { + if (event.type === 'inference:completed') resolveCompleted(); + }); + const bounded = async (promise: Promise, label: string) => { + let timer: NodeJS.Timeout | undefined; + try { + await Promise.race([promise, new Promise((_, reject) => { + timer = setTimeout(() => reject(new Error('Timed out waiting for ' + label)), 5000); + })]); + } finally { clearTimeout(timer); } + }; + let run: Promise | undefined; + try { + run = h.run(); + void run.catch(() => {}); + await bounded(completed, 'the resumed inference'); + assert.equal(h.membrane.requests.length, 2, 'a budget restart followed the fresh stream'); + assert.deepEqual(routed, ['earlier'], 'both fresh-turn and resumed posts await the prior delivery'); + release(); + await bounded(run, 'the resumed turn to drain'); + assert.deepEqual(routed, ['earlier', 'fresh', 'continued'], 'all three posts keep their order'); + } finally { + off(); + release(); + try { if (run) await bounded(run, 'cleanup'); } finally { await h.framework.stop(); } + } +}); + +test("explicit: a restarted turn whose last response is empty checks for its ' !' after the earlier stream's queued delivery", { timeout: 15000 }, async () => { + // Stream one's round holds its first envelope's post, with a ' !' envelope + // queued behind it. The round crosses the budget, the turn restarts, and + // the restart's response is empty, so nothing but the completion's own + // wait orders the continuation check after that envelope. + const quiet = createMockResponse([ + { type: 'text', text: '>>#alpha first\n>>#alpha ! second' }, + { type: 'tool_use', id: 'one', name: 'test--read', input: {} }, + ] as ContentBlock[], 'tool_use'); + quiet.usage = { inputTokens: 100, outputTokens: 5 }; + const empty = createMockResponse([] as ContentBlock[], 'end_turn'); + const after = createMockResponse([{ type: 'text', text: '>>skip_reply' }] as ContentBlock[], 'end_turn'); + const h = await harness([[quiet], [empty], [after]], { proseRouting: 'explicit', maxStreamTokens: 20 }); + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); const routed: string[] = []; (h.framework as unknown as { channelRegistry: unknown }).channelRegistry = new Proxy({ resolveProseTarget: (spec: string) => spec === '#alpha' ? { channelId: 'chan-alpha' } : { error: `no channel ${spec}` }, + resolveLocus: () => 'chan-locus', routeSpeech: async (_agent: string, text: string, channelId: string) => { routed.push(text); + if (text.trim() === 'first') await held; return { delivered: true, channelId }; }, getDefaultPublishChannel: () => null, isChannelOpen: () => true, getDescriptor: () => undefined, getChannelTools: () => [], }, { get: (target, key: string) => key in target ? (target as Record)[key] : () => undefined }); + const internals = h.framework as unknown as { + requestProseContinuation(agent: unknown, routing: { logical: { continuation: boolean } }): void; + }; + const check = internals.requestProseContinuation.bind(h.framework); + const seen: boolean[] = []; + internals.requestProseContinuation = (agent, routing) => { + seen.push(routing.logical.continuation); + check(agent, routing); + }; try { const run = h.run(); - let timer: NodeJS.Timeout | undefined; - const timeout = new Promise<'timeout'>((resolve) => { timer = setTimeout(() => resolve('timeout'), 10000); }); - try { - assert.notEqual(await Promise.race([run.then(() => 'done' as const), timeout]), 'timeout', - 'the resumed turn finishes'); - } finally { - clearTimeout(timer); + for (const end = Date.now() + 5000; h.membrane.requests.length < 2 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); } - assert.equal(h.membrane.requests.length, 2); - assert.deepEqual(routed.map((text) => text.trim()), ['continued']); + assert.equal(h.membrane.requests.length, 2, 'the turn restarted for its budget'); + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.deepEqual(seen, [], 'no continuation check while the earlier post is held'); + release(); + await run; + for (const end = Date.now() + 5000; h.membrane.requests.length < 3 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + await h.framework.runUntilIdle(); + } + assert.deepEqual(routed.map((text) => text.trim()), ['first', 'second']); + assert.equal(seen[0], true, "the completion's check saw the ' !' the queued envelope set"); + assert.equal(h.membrane.requests.length, 3, 'and the continuation ran'); } finally { + release(); await h.framework.stop(); } }); -test("a fresh turn clears the sticky destination and the turn's destination in every mode", async () => { +test("hybrid: a restarted turn whose last response is empty writes its receipt after the earlier stream's queued delivery", { timeout: 15000 }, async () => { + // Stream one names #alpha, and its post is held. Its round crosses the + // budget, so the turn restarts, and the restart's response is empty. + const quiet = createMockResponse([ + { type: 'text', text: '>>>#alpha first' }, + { type: 'tool_use', id: 'one', name: 'test--read', input: {} }, + ] as ContentBlock[], 'tool_use'); + quiet.usage = { inputTokens: 100, outputTokens: 5 }; + const empty = createMockResponse([] as ContentBlock[], 'end_turn'); + const h = await harness([[quiet], [empty]], { proseRouting: 'hybrid', maxStreamTokens: 20 }); + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const routed: string[] = []; + (h.framework as unknown as { channelRegistry: unknown }).channelRegistry = new Proxy({ + resolveProseTarget: (spec: string) => spec === '#alpha' ? { channelId: 'chan-alpha' } : { error: `no channel ${spec}` }, + resolveLocus: () => 'chan-locus', + routeSpeech: async (_agent: string, text: string, channelId: string) => { + routed.push(text); + if (text.trim() === 'first') await held; + return { delivered: true, channelId }; + }, + getDefaultPublishChannel: () => null, isChannelOpen: () => true, + getDescriptor: () => undefined, getChannelTools: () => [], + }, { get: (target, key: string) => key in target ? (target as Record)[key] : () => undefined }); + const receipts = () => h.framework.getAgent('assistant')!.getContextManager().getAllMessages() + .flatMap((message) => message.content) + .filter((block): block is ContentBlock & { type: 'text' } => block.type === 'text' && (block as { text: string }).text.startsWith('[delivered]')) + .map((block) => block.text); + try { + const run = h.run(); + for (const end = Date.now() + 5000; h.membrane.requests.length < 2 && Date.now() < end;) { + await new Promise((resolve) => setTimeout(resolve, 5)); + } + assert.equal(h.membrane.requests.length, 2, 'the turn restarted for its budget'); + await new Promise((resolve) => setTimeout(resolve, 100)); + assert.deepEqual(receipts(), [], 'no receipt while the earlier post is still held'); + release(); + await run; + assert.deepEqual(routed.map((text) => text.trim()), ['first']); + const written = receipts(); + assert.equal(written.length, 1, 'one receipt for the turn'); + assert.match(written[0]!, /chan-alpha/, 'and it names the post that was still queued'); + } finally { + release(); + await h.framework.stop(); + } +}); + +test("a fresh turn starts new routing state in every mode, and leaves the earlier turn's as it was", async () => { for (const mode of ['explicit', 'hybrid', 'locus'] as const) { const h = await harness([[answer()]], { proseRouting: mode }); - const internals = h.framework as unknown as { - proseTurnDestinations: Map; - proseTargetPins: Map; - }; - internals.proseTurnDestinations.set('assistant', 'chan-earlier'); - internals.proseTargetPins.set('assistant', 'chan-earlier'); + type Routing = { pin?: string; destination: string | null; hybridHold: boolean; + logical: { continuation: boolean; deliveries: string[]; suppressed: number } }; + const internals = h.framework as unknown as { turnRouting: Map }; + const earlier: Routing = { pin: 'chan-earlier', destination: 'chan-earlier', hybridHold: true, + logical: { continuation: true, deliveries: ['chan-earlier'], suppressed: 2 } }; + internals.turnRouting.set('assistant', earlier); try { await h.run(); - assert.equal(internals.proseTurnDestinations.has('assistant'), false, `${mode}: the turn's destination`); - assert.equal(internals.proseTargetPins.has('assistant'), false, `${mode}: the sticky destination`); + const now = internals.turnRouting.get('assistant'); + assert.notEqual(now, earlier, `${mode}: the new turn has its own routing state`); + assert.equal(now?.pin, undefined, `${mode}: no sticky destination`); + assert.equal(now?.destination, null, `${mode}: no turn destination`); + assert.equal(now?.hybridHold, false, `${mode}: no hybrid hold`); + assert.equal(now?.logical.continuation, false, `${mode}: no continuation`); + assert.notEqual(now?.logical, earlier.logical, `${mode}: a fresh turn's logical state isn't shared with the earlier turn's`); + assert.deepEqual(now?.logical.deliveries, [], `${mode}: no deliveries carried into the new turn`); + assert.equal(now?.logical.suppressed, 0, `${mode}: no suppression carried into the new turn`); + assert.deepEqual(earlier, { pin: 'chan-earlier', destination: 'chan-earlier', hybridHold: true, + logical: { continuation: true, deliveries: ['chan-earlier'], suppressed: 2 } }, + `${mode}: all earlier-turn routing and receipt state stays with its queued prose`); } finally { await h.framework.stop(); }