diff --git a/changelog.d/inbound-only-channels.fixed.md b/changelog.d/inbound-only-channels.fixed.md new file mode 100644 index 000000000..37283f069 --- /dev/null +++ b/changelog.d/inbound-only-channels.fixed.md @@ -0,0 +1,38 @@ +- The host no longer publishes into a channel its server declared inbound-only + (`direction: "inbound"`, SPEC §14.2). A server's live-transcript channel can + be one, where an agent speaking in the room arrives addressed. Until now + that moved a resident's plain speech there, the delivery attempt opened the + channel if she had closed it, and the server refused the publish. + - Every publish of plain speech into such a channel is refused before the + channel is opened or published to, a drafts resend and `{{unsent}}` + included, with a reason that names the declaration and says to use that + server's own tools. A channel declared inbound-only while the publish + waits, on the channel's lifecycle work, its open or anything else, is + refused too, and so is one whose registration is gone by the time the + words would be published. Nothing is published to either. A declaration + that comes with the channel's own opening leaves the channel open: the + agent gets the failure for its words, and the notice that the channel is + open if this open is what opened it. + - A `>>` or `>>>` target naming the channel bounces with the same reason, + and the closed-channel invitation offers no reply into it. In explicit + mode the refused target also clears the turn's destination, so the turn's + later text without a prefix bounces too, rather than going to the + destination named before it. A delivery refused because the channel was + declared inbound-only after its target resolved ends the destination + too, unless a later round has named another channel by then. + - `channel_publish` and the subconscious's `speak_in_channel` return an + error that names it. The outgoing stream skips the channel, and the + channel context names no default outgoing channel for it. + - A channel the host knows only from a push is refused once the server's + answer to opening it, wherever the host opens it, declares it + inbound-only. + - The routing notices say plain speech is refused there, and to use that + server's own tools. The locus and hybrid mode primers, shown when an + agent's mode changes, say plain speech goes to the locus unless its server + declared that channel inbound-only. +- `speak_in_channel`'s `mcpl:speech-routed` trace now follows a confirmed + delivery, with `serverId`, `delivered` and `textLen` as plain speech's trace + has them, and no `messageId`, as before. Any other result is traced as + `mcpl:speech-route-failed`, with the publish's reason, the connector's + detail when it sent some, and `outcome: "unknown"` when the words may have + been posted or `"failed"` when nothing was, a refusal included. diff --git a/docs/explicit-prose-routing.md b/docs/explicit-prose-routing.md index 25ad17af9..8db5f6f99 100644 --- a/docs/explicit-prose-routing.md +++ b/docs/explicit-prose-routing.md @@ -26,7 +26,9 @@ A prose segment's FIRST line may carry a routing prefix: unprefixed segments of the SAME turn, including those a later stream of it writes, such as a framework retry's, a budget restart's or the compile a provider-capacity hold defers the turn to. A round kept private (one that - calls `skip_reply`, or has a same-round private think) sets no target. + calls `skip_reply`, or has a same-round private think) sets no target. A + target its server declared inbound-only is refused and clears the turn's + target, so later unprefixed segments bounce until a new `>>` target. Cross-turn, nothing is implicit. - **Continuation modifier:** ` !` immediately after the target (`>>#ops !` / `>>skip_reply !`) requests an immediate re-wake after this turn @@ -81,7 +83,7 @@ depends on the route: Explicit send tools (`send_message` etc.) are unchanged. Channel lifecycle invariants (send-opens-channel, subscribed⇒open) are unchanged — a `>>` send -into a closed channel opens it, with the usual `[channels]` notice. +into a channel decided closed opens it, with the usual `[channels]` notice. ## Teaching the grammar (teach-by-bounce) diff --git a/src/framework.ts b/src/framework.ts index d72da2323..bc909dfa1 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -334,6 +334,19 @@ function estimateAppendedRoundTokens( return Math.ceil(chars / 4) + images * 1600 + blockCount * 20; } +/** + * What a notice says in place of "your plain speech lands there" when the + * channel's server declared it inbound-only (isInboundOnly in the channel + * registry, which publishes nothing there). ChannelRegistry.inboundOnlyServer + * names the declaring server. + */ +function inboundOnlySpeechNote(serverId: string): string { + return ( + `${serverId} declared this channel inbound-only, so plain speech there is refused ` + + `and stays in your archive. To speak there, use ${serverId}'s own tools.` + ); +} + /** * Explicit delivery tools whose successful use into a closed channel OPENS * it (send implies engagement — see openIfClosedForSend). Bare tool names, @@ -421,11 +434,13 @@ const isAddressedMessage = ( function proseModePrimer(mode: 'explicit' | 'hybrid' | 'locus' | 'disabled', helpAvailable: boolean): string { if (mode === 'locus') { return '[prose-routing] Mode change: your plain text auto-routes to the ' + - 'conversational locus again. `>>` destination prefixes are no longer needed.'; + 'conversational locus again, unless its server declared that channel inbound-only. ' + + '`>>` destination prefixes are no longer needed.'; } if (mode === 'hybrid') { return '[prose-routing] Hybrid routing enabled: unprefixed text still lands in the ' + - 'current conversational locus. A leading `>>>destination` envelope instead routes ' + + 'current conversational locus, unless its server declared that channel inbound-only. ' + + 'A leading `>>>destination` envelope instead routes ' + 'that prose to one uniquely resolved authorized Discord or Eidoverse channel. The ' + 'envelope remains in your memory but recipients see only its body; delivery or failure ' + 'is reported back to you.'; @@ -453,7 +468,9 @@ const PROSE_ROUTING_HELP = ' >>#channel-name … or >>@person … (DM) or >>service:guild:id …\n' + ' One message may hold several sections: each line starting with ">>" begins\n' + ' a new envelope, routed to its own destination independently.\n' + - ' The first destination in a turn applies to the rest of that turn.\n' + + ' A destination applies to the rest of that turn, until another is named.\n' + + ' A name that matches no channel, or several, bounces and leaves it as it\n' + + ' was. A channel its server declared inbound-only is refused, and clears it.\n' + ' Append " !" after the destination (e.g. ">>#ops !") to start your next turn\n' + ' immediately when this one ends, instead of pausing until the next event.\n' + ' >>skip_reply — text stays in your context only.\n' + @@ -722,7 +739,7 @@ function splitExplicitEnvelopes(rawText: string): string[] { */ interface AdmittedEnvelope { text: string; - resolved?: { channelId: string } | { error: string; candidates?: string[] }; + resolved?: { channelId: string } | { error: string; candidates?: string[]; refusedChannelId?: string }; } /** The domains an mcpl/manifestChanged announcement names, for its receipt: @@ -800,9 +817,15 @@ 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 each of the - * turn's streams starts. + * Explicit mode: where each of the turn's streams starts. It changes as a + * round is admitted, before its delivery (roundDestination): a `>>` target + * that resolves sets it, and one refused for speech clears it. A target + * that matches no channel, or several, leaves it, and so does a round that + * names none, or one kept private by skip_reply. A failed delivery to a channel that is + * inbound-only by then clears it too, while it still names that channel + * (deliverProseEnvelope). The turn's final text doesn't set it: the stream + * settles the turn before delivering that text, so no later stream of the + * turn starts after it, and a later start opens a turn of its own. */ destination: string | null; /** @@ -12004,9 +12027,13 @@ export class AgentFramework { const target = typeof namer.proseTargetFor === 'function' ? namer.proseTargetFor(opts.channelId, opts.serverId) : [hashLabel, opts.channelId].find((t): t is string => !!t && !/\s/.test(t)); - const replyOption = target - ? `1. Reply — write your reply this turn prefixed with ">>${target}"; it will be delivered there, which opens the channel.\n` - : `1. Replying with a ">>" prefix isn't available here: no single-word target reaches this channel unambiguously. Open it (option 2) to reply.\n`; + const inboundOnlyServer = this.channelRegistry.inboundOnlyServer(opts.channelId, opts.serverId); + const replyOption = inboundOnlyServer + ? `1. Replying there isn't available: ${inboundOnlyServer} declared this channel inbound-only, so nothing ` + + `you write is posted into it. To speak there, use ${inboundOnlyServer}'s own tools.\n` + : target + ? `1. Reply — write your reply this turn prefixed with ">>${target}"; it will be delivered there, which opens the channel.\n` + : `1. Replying with a ">>" prefix isn't available here: no single-word target reaches this channel unambiguously. Open it (option 2) to reply.\n`; const missedCount = opts.missedMessages ?? 0; const missedNote = missedCount > 0 @@ -13368,7 +13395,9 @@ export class AgentFramework { if (parsed.body.includes('{{unsent}}')) this.dismissLatestBounce(agent); privateCount++; } else if (parsed.kind === 'target') { - const resolved = this.channelRegistry?.resolveProseTarget(parsed.target!); + // As deliverHybridProse resolves it: a channel its server declared + // inbound-only is refused as a target for speech. + const resolved = this.channelRegistry?.resolveProseTarget(parsed.target!, { forSpeech: true }); if (resolved && !('error' in resolved)) { routing.pin = resolved.channelId; delete routing.hybridHold; @@ -14201,9 +14230,12 @@ export class AgentFramework { label && label !== locus ? `${label.startsWith('#') ? label : `#${label}`} (${locus})` : locus; - text = - `[routing] Your plain speech now lands in ${shown}. ` + - 'Other channels need an explicit send tool.'; + const inboundOnlyServer = this.channelRegistry?.inboundOnlyServer(locus); + text = inboundOnlyServer + ? `[routing] The conversation is now in ${shown}. ${inboundOnlySpeechNote(inboundOnlyServer)} ` + + 'Other channels need an explicit send tool.' + : `[routing] Your plain speech now lands in ${shown}. ` + + 'Other channels need an explicit send tool.'; } try { @@ -14325,12 +14357,15 @@ export class AgentFramework { const shown = label && label !== channelId ? `${label.startsWith('#') ? label : `#${label}`} (${channelId})` : channelId; + const inboundOnlyServer = this.channelRegistry?.inboundOnlyServer(channelId); const content: ContentBlock[] = [{ type: 'text', - text: - `[routing] The conversation moved to ${shown} — your plain ` + - 'speech now lands there for the rest of this turn. Other ' + - 'channels need an explicit send tool.', + text: inboundOnlyServer + ? `[routing] The conversation moved to ${shown}. ${inboundOnlySpeechNote(inboundOnlyServer)} ` + + 'Other channels need an explicit send tool.' + : `[routing] The conversation moved to ${shown} — your plain ` + + 'speech now lands there for the rest of this turn. Other ' + + 'channels need an explicit send tool.', }]; const metadata = { system: true, kind: 'routing-notice' } as MessageMetadata; try { @@ -14481,7 +14516,7 @@ export class AgentFramework { const parsed = parseProsePrefix(text); if (parsed.kind !== 'target') return { text }; try { - return { text, resolved: this.channelRegistry!.resolveProseTarget(parsed.target!) }; + return { text, resolved: this.channelRegistry!.resolveProseTarget(parsed.target!, { forSpeech: true }) }; } catch (err) { // Admission runs in the stream's event loop, so a resolver that // throws makes the envelope bounce rather than stop the round. @@ -14670,15 +14705,21 @@ export class AgentFramework { } /** - * Where a delivered round leaves its turn's sticky destination, as the - * explicit gateway sets it once the round's envelopes run: each `>>` - * envelope whose target resolved becomes the destination. One that didn't + * Where an admitted round leaves its turn's sticky destination, set as + * the round is admitted, before its delivery: each `>>` envelope whose + * target resolved becomes the destination. One that didn't * resolve bounces and leaves it, as do `>>skip_reply` and unprefixed text. + * One refused for speech (its server declared the channel inbound-only) + * bounces and clears it: the text after it was meant for that channel, so + * it bounces too rather than going to the turn's earlier destination. */ private roundDestination(current: string | null, envelopes: AdmittedEnvelope[]): string | null { let destination = current; for (const envelope of envelopes) { - if (envelope.resolved && 'channelId' in envelope.resolved) destination = envelope.resolved.channelId; + const resolved = envelope.resolved; + if (!resolved) continue; + if ('channelId' in resolved) destination = resolved.channelId; + else if (resolved.refusedChannelId) destination = null; } return destination; } @@ -14706,7 +14747,7 @@ export class AgentFramework { continue; } if (parsed.kind === 'target') { - const resolved = this.channelRegistry!.resolveProseTarget(parsed.target!); + const resolved = this.channelRegistry!.resolveProseTarget(parsed.target!, { forSpeech: true }); if ('error' in resolved) { delete routing.pin; routing.hybridHold = 'failed'; @@ -14864,6 +14905,13 @@ export class AgentFramework { const res = envelope.resolved; if (!res) throw new Error(`[prose] ${name}: a >> envelope reached delivery without the target admitted with it`); if ('error' in res) { + // A target refused for speech ends the turn's destination. This drops + // the pin, which later text follows. roundDestination cleared the + // destination later streams start at when it admitted the round, for + // any round but the turn's final text, after which nothing of the turn + // streams. The text after the target was meant for that channel, not + // the earlier one, so it bounces too. + if (res.refusedChannelId) delete routing.pin; this.bounceProse(agent, parsed.body, res.error, res.candidates, '>>', round); return; } @@ -14874,7 +14922,7 @@ export class AgentFramework { } else { const sticky = routing.pin; if (!sticky) { - this.bounceProse(agent, rawText, 'no destination prefix and no destination set yet this turn', undefined, '>>', round); + this.bounceProse(agent, rawText, 'no destination prefix, and this turn has no destination now', undefined, '>>', round); return; } targetChannel = sticky; @@ -14894,9 +14942,11 @@ export class AgentFramework { return; } + let delivered = false; try { const outcome = await this.deliverWithUnsent(agent, body, targetChannel, unsent.draft); - if (outcome.status === 'delivered') { + delivered = outcome.status === 'delivered'; + if (delivered) { this.proseBounceStreaks.delete(name); } else if (outcome.status === 'failed') { // Unknown deliveries are marker-only: offering a fresh draft would @@ -14906,6 +14956,18 @@ export class AgentFramework { } catch (err) { this.bounceProse(agent, envelopeBody, `sending to ${targetChannel} threw: ${err instanceof Error ? err.message : String(err)}`, undefined, '>>', round, { substituted: unsent.draft, outcome: 'unknown' }); } + if (!delivered && this.channelRegistry!.inboundOnlyServer(targetChannel) !== undefined) { + // The channel is inbound-only by now: its server declared it after the + // round was admitted, or in its answer to opening it. Whatever made this + // delivery fail, a refusal or a publish whose outcome is unknown, later + // text there would be refused, so what still names the channel ends, as + // a refusal at admission ends it: the pin, unless a later delivery has + // moved it, and the destination later streams start at, unless a later + // round has named another. A failed delivery to a channel that isn't + // inbound-only keeps them, as before. + if (routing.pin === targetChannel) delete routing.pin; + if (routing.destination === targetChannel) routing.destination = null; + } } /** Cap on consecutive bounce-triggered wakes (notices still append after). */ @@ -16252,9 +16314,11 @@ export class AgentFramework { // MCPL Spec 14.3 outgoing streaming: route text deltas to their // destination server AS THEY GENERATE (voice synthesis, live message // rendering). Pure observer surface — suppression is fail-closed (nothing - // streams that delivery would refuse), emission is capability-gated in - // the registry (`channels.streaming`), and delivery via deliverProse - // remains the sole authoritative send. + // streams whose destination delivery wouldn't choose, or to a channel + // the router's `speakable` refuses), emission is capability-gated in the + // registry (`channels.streaming`), and delivery via deliverProse remains + // the sole authoritative send. Delivery can still refuse text that has + // streamed, for what it learns later (ProseStreamRouter's header). const newOutgoingInferenceId = (): string => `inf_${Date.now().toString(36)}_${Math.random().toString(36).slice(2, 8)}`; // NOT const: a membrane refusal retry abandons everything streamed so @@ -16298,7 +16362,7 @@ export class AgentFramework { // Explicit text has no destination until a `>>` prefix names one, // so a fresh turn's stream starts suppressed. A later stream of the // turn (a resume, a retry, a re-inference, a capacity-deferred - // compile) starts at the destination the turn's delivered rounds + // compile) starts at the destination the turn's admitted rounds // have set: a private round's prefixes don't count, and the // gateway's queued deliveries needn't have run. The trigger channel // is only where explicit mode's typing indicator shows. @@ -16309,6 +16373,10 @@ export class AgentFramework { const r = this.channelRegistry!.resolveProseTarget(spec); return 'channelId' in r ? r.channelId : null; }, + // Nothing is published into a channel its server declared + // inbound-only, so text routed there while it is one is never + // streamed or kept for the stream's end. + speakable: (channelId) => !this.channelRegistry!.inboundOnlyServer(channelId), }) : null; const emitOutgoing = (deltas: { channelId: string; delta: string }[]): void => { @@ -16491,8 +16559,8 @@ export class AgentFramework { proseStream?.reset(); lastTextBlock = undefined; // 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. + // admitted rounds have set the destination, not at the stream's + // seed: rounds this stream admitted before the refusal stand. if (turnProseRouting === 'explicit') proseStream?.retarget(streamRouting.destination); this.emitTrace({ type: 'inference:tokens', @@ -16728,7 +16796,8 @@ export class AgentFramework { // unprefixed text: the turn's destination. The stream // 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. + // that didn't resolve leave the destination as it was, + // and a target refused for speech clears it. proseStream?.retarget(streamRouting.destination); } else if (turnSilenced) { console.error( @@ -21512,10 +21581,10 @@ export class AgentFramework { eventGatePresent: this.eventGate !== null, // A text-only turn whose speech couldn't be delivered must not vanish // silently: record a `[discord-send-failed]` marker in chronicle so the - // agent sees, on her next turn, that her reply never reached the human. - // addMessage() alone does not request inference, so this never wakes - // her (matching the `discord-send-failed-skip` gate intent: context - // yes, wake no). + // agent sees, on her next turn, that her reply didn't reach the human, + // or wasn't confirmed to. addMessage() alone does not request + // inference, so this never wakes her (matching the + // `discord-send-failed-skip` gate intent: context yes, wake no). onRouteFailure: ({ conversationId, channelId, reason, textLen, outcome, detail }) => { try { // The marker is `system: true`, so it is never conversational and @@ -25116,11 +25185,14 @@ export class AgentFramework { && this.openLogical.get(agentName) === callingTurn) { callingTurn.locus = opened; this.lastAnnouncedLocus.set(agentName, opened); + const inboundOnlyServer = this.channelRegistry?.inboundOnlyServer(opened); result = { ...result, data: { ...(result.data as Record | undefined), - routing: 'Your plain speech now lands in this channel. Other channels need an explicit send tool.', + routing: inboundOnlyServer + ? `${inboundOnlySpeechNote(inboundOnlyServer)} Other channels need an explicit send tool.` + : 'Your plain speech now lands in this channel. Other channels need an explicit send tool.', }, }; console.error( diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 5a0bb585c..4e558591f 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -252,6 +252,27 @@ interface ChannelEntry { open: boolean; } +/** + * A channel its server declared inbound-only (SPEC §14.2 `direction`): its + * messages come in, and the host publishes nothing into it. No publish, from + * plain speech, `channel_publish` or `publishForAgent`, and no outgoing + * stream. Typing and acknowledgements, which carry no content, still go + * (SPEC §14.5: "Delivery occurs only via `channels/publish`"). A channel + * declared `outbound` or `bidirectional`, or with no direction, isn't one. + */ +function isInboundOnly(entry: ChannelEntry): boolean { + return entry.descriptor.direction === 'inbound'; +} + +/** Why the host sends nothing into an inbound-only channel, as the resident + * reads it in a send-failure marker or a tool error. */ +function inboundOnlyReason(entry: ChannelEntry): string { + return ( + `inbound-only: server "${entry.serverId}" declared this channel's direction "inbound", ` + + 'so the host publishes nothing into it; to speak there, use that server\'s own tools' + ); +} + function sameChannelTarget(a: ChannelDescriptor, b: ChannelDescriptor): boolean { return a.type === b.type && isDeepStrictEqual(a.address, b.address); } @@ -589,14 +610,19 @@ interface ChannelRegistryOptions { /** Callback to determine whether an incoming message should trigger inference. */ shouldTriggerInference?: (content: string, metadata: Record) => boolean; /** - * Called when a text-only turn's speech could NOT be delivered to its - * conversational locus — no locus, an unregistered channel, a missing - * server, or the server reporting `delivered: false`. Not called for an - * ephemeral run's speech with no locus, which is its caller's result - * (`takesGlobalFallback`). The host wires this to drop a - * `[discord-send-failed]` marker into chronicle so the failure is - * visible to the agent (and operator) rather than silently lost. Must not - * itself trigger inference (avoid wake loops). + * Called when a text-only turn's speech was not delivered to its + * conversational locus (`failed`), or may not have been (`unknown`), for + * whatever reason deliverSpeech gives, such as no locus, a channel that + * isn't registered or refuses speech, a missing server or grant, a failed + * open, the server reporting `delivered: false`, or a publish answered + * with an error, a timeout or a lost connection, after which whether it + * was posted is unknown. It isn't called for an ephemeral run's speech + * whose turn has no locus (a null locus), which is its caller's result + * (`takesGlobalFallback`), nor for a `failed` delivery whose caller holds + * its words with a notice of its own (deliverSpeech's `heldOnFailure`). + * The host wires this to drop a `[discord-send-failed]` marker into + * chronicle so the failure is visible to the agent (and operator) rather + * than silently lost. Must not itself trigger inference (avoid wake loops). */ onRouteFailure?: (info: { conversationId: string; @@ -1676,8 +1702,9 @@ export class ChannelRegistry { const openChannels = this.getOpenChannels(); // Resolve the outbound locus the SAME way routeSpeech does, so the agent is - // told the channel its speech will actually land in: a conversation fork's - // home channel, else this turn's channel (single trunk agent — item-3 + // told the channel its speech will actually land in (none, when its server + // declared it inbound-only, since speech isn't posted there): a conversation + // fork's home channel, else this turn's channel (single trunk agent — item-3 // redux: its locus, else its triggering channel), else the global default // (heartbeats). Without this, a fork // was advertised the global locus but published somewhere else; and a trunk @@ -1709,8 +1736,10 @@ export class ChannelRegistry { }; } - // Default outgoing: the resolved outbound locus (home channel for forks). - if (outgoing) { + // Default outgoing: the resolved outbound locus (home channel for forks), + // unless its server declared it inbound-only: routeSpeech publishes nothing + // there, so it isn't where the agent's speech goes. + if (outgoing && !this.inboundOnlyServer(outgoing)) { context.defaultOutgoing = { channelId: outgoing, }; @@ -2313,16 +2342,50 @@ export class ChannelRegistry { /** * Publish into a channel on behalf of a named non-resident agent (the - * subconscious's speak_in_channel). Same delivery path as the - * channel_publish tool; the agent name rides the speech-routed trace. + * subconscious's speak_in_channel). Same delivery path, and the same result, + * as the channel_publish tool. Its traces carry the agent's name and follow + * the publish's outcome as deliverSpeech's do: speech-routed after a + * confirmed delivery, and speech-route-failed otherwise, `unknown` when the + * words may have been posted and `failed` when nothing was, a refusal before + * the publish included. The routed trace carries no message id, as before: + * nothing here reads one, and a consumer that edits the newest traced + * message in a channel would have this post to edit when an interruption of + * the resident matches none. */ async publishForAgent( channelId: string, text: string, agentName: string, ): Promise<{ success: boolean; data?: unknown; error?: string; isError?: boolean }> { - this.emitTraceFn({ type: 'mcpl:speech-routed', conversationId: agentName, channelId, text }); - return this.handleToolPublish({ channelId, content: text }); + // The server the publish goes to, as handleToolPublish finds it before + // it awaits: the channel may be removed while the reply is pending. + const serverId = this.findChannelEntry(channelId)?.serverId ?? ''; + const seen: { outcome?: PublishOutcome } = {}; + const result = await this.handleToolPublish({ channelId, content: text }, undefined, seen); + if (seen.outcome?.status === 'delivered') { + this.emitTraceFn({ + type: 'mcpl:speech-routed', + conversationId: agentName, + serverId, + channelId, + delivered: true, + textLen: text.length, + text, + }); + } else { + this.emitTraceFn({ + type: 'mcpl:speech-route-failed', + conversationId: agentName, + channelId, + // The publish's own reason and the connector's detail, as + // deliverSpeech's trace gives them, or the refusal before any publish. + reason: seen.outcome?.reason ?? result.error ?? 'the publish failed', + textLen: text.length, + outcome: seen.outcome?.status === 'unknown' ? 'unknown' : 'failed', + ...(seen.outcome?.detail !== undefined ? { detail: connectorText(seen.outcome.detail) } : {}), + }); + } + return result; } /** @@ -2556,6 +2619,7 @@ export class ChannelRegistry { `; ${channelId} is not marked open, and nothing in the answer is used`, ); } + if (open) this.applyOpenDeclaration(entry, result); entry.open = open; this.unconfirmedChannelTargets.delete(entry); const now = this.getDesiredState(serverId, channelId); @@ -2740,6 +2804,17 @@ export class ChannelRegistry { return this.findChannelEntry(channelId)?.serverId ?? null; } + /** + * The server that declared this channel inbound-only (isInboundOnly), or + * undefined when it isn't one. The notices name that server. A caller that + * knows the channel's server passes it; otherwise the channel is found by + * its id alone (findChannelEntry). + */ + inboundOnlyServer(channelId: string, serverId?: string): string | undefined { + const entry = serverId ? this.channels.get(`${serverId}:${channelId}`) : this.findChannelEntry(channelId); + return entry && isInboundOnly(entry) ? entry.serverId : undefined; + } + /** * Find a channel entry by channelId (searches across all servers). * Returns the first match. @@ -2823,8 +2898,9 @@ export class ChannelRegistry { /** * Record durable desired-open intent, then queue the channel's lifecycle * work (applyDesiredChannelState), which tells the server to subscribe and - * marks the live entry open on a receipt for its current target. Every - * open the agent or its speech asks for — the channel_open tool, delivery + * marks the live entry open on a receipt for its current target, applying + * an inbound-only declaration the answer carries (applyOpenDeclaration). + * Every open the agent or its speech asks for — the channel_open tool, delivery * into a closed locus, an explicit send into a closed channel — comes * here. Reconciliation queues that work directly, as the closes do * (channel_close, a decline, and the close of a substituted channel). @@ -2856,6 +2932,31 @@ export class ChannelRegistry { return { result: applied.result!, opened }; } + /** + * An open's answer carries the server's descriptor for the channel + * (§14.3), the first place a lazily registered channel's own declaration + * can arrive. runLifecycleOperation hands it here, wherever the host opens + * a channel, once the answer is known to be for the channel's current + * entry and target. An answer that names another channel never arrives + * here: it's a substitution, which fails the open. One whose channel names + * no id is this channel's, as runLifecycleOperation takes it, so its + * declaration applies too. + * + * Only an inbound-only declaration is applied. The answer isn't always the + * server's latest word: a `channels/changed` that arrives in the same read + * as the answer is handled before the open's operation resumes, so the + * answer is applied over it. Applying only a refusal keeps that race + * fail-closed: a stale answer can keep a channel refused until the + * server's next update, but never lift a newer refusal. A server lifts its + * declaration by changing or registering the channel again. + */ + private applyOpenDeclaration(entry: ChannelEntry, answer: unknown): void { + const declared = (answer as { channel?: { direction?: unknown } } | undefined)?.channel; + if (declared?.direction === 'inbound') { + entry.descriptor = { ...entry.descriptor, direction: 'inbound' }; + } + } + /** * Resolve a `>>` prose-routing target (explicit prose routing — * docs/explicit-prose-routing.md) to a registered channel. @@ -2869,9 +2970,28 @@ export class ChannelRegistry { * * Ambiguity is an error carrying candidates — never a guess: this is the * mechanism that makes prose misdelivery structurally impossible. + * + * `forSpeech` is for a target the agent's speech will be posted to: a + * channel its server declared inbound-only (isInboundOnly) then resolves + * to an error that says so, since nothing is published there, and that + * names the channel as `refusedChannelId`, so a caller can tell a refused + * channel from a name that resolved to nothing. Without it, as for reading + * a channel's history, any registered channel resolves. */ resolveProseTarget( spec: string, + opts?: { forSpeech?: boolean }, + ): { channelId: string; label?: string } | { error: string; candidates?: string[]; refusedChannelId?: string } { + const resolved = this.resolveTargetByName(spec); + if (!opts?.forSpeech || 'error' in resolved) return resolved; + const entry = this.findChannelEntry(resolved.channelId); + return entry && isInboundOnly(entry) + ? { error: `${resolved.channelId} is ${inboundOnlyReason(entry)}`, refusedChannelId: resolved.channelId } + : resolved; + } + + private resolveTargetByName( + spec: string, ): { channelId: string; label?: string } | { error: string; candidates?: string[] } { const entries = [...this.channels.values()]; const trimmed = spec.trim(); @@ -3030,7 +3150,7 @@ export class ChannelRegistry { ]; for (const c of candidates) { if (/\s/.test(c) || c === '#') continue; - const r = this.resolveProseTarget(c); + const r = this.resolveProseTarget(c, { forSpeech: true }); if ('channelId' in r && r.channelId === d.id) return c; } return undefined; @@ -3462,18 +3582,6 @@ export class ChannelRegistry { }; } - /** - * Host-owned output routing (see forking-knowledge-miner LOCUS-ROUTING-DESIGN). - * Publish the agent's plain-text speech to the current conversational locus - * (the most recent incoming channel, tracked cross-surface here in the host). - * Called by the framework on a text-only turn — replaces the per-surface - * sticky auto-post that used to live in discord-mcpl. Returns null when there - * is no locus / the channel or its server can't be resolved (in which case - * the speech simply stays in chronicle + module surfaces). - */ - /** Resolve the outbound locus (fork HOME → this-turn's TRIGGERING channel → - * process-global default). Public so a multi-segment caller can snapshot it - * ONCE and pin every segment to it via routeSpeech's `overrideChannelId`. */ /** * MCPL Spec 14.3 outgoing streaming: forward a moderated text delta to the * server owning the channel, AS THE MODEL GENERATES. Emitted only when that @@ -3522,6 +3630,9 @@ export class ChannelRegistry { // never-guess rule that governs delivery. const matches = [...this.channels.values()].filter((e) => e.descriptor.id === channelId); if (matches.length !== 1) return null; + // Nothing is published into an inbound-only channel, so nothing is + // streamed there either. + if (isInboundOnly(matches[0]!)) return null; const server = this.serverRegistry.getServer(matches[0]!.serverId); // §5.4: the GRANT gates streaming, not the raw advertisement. The old // `capabilities?.channels?.streaming` check was doubly wrong: undefined @@ -3532,6 +3643,9 @@ export class ChannelRegistry { return server; } + /** Resolve the outbound locus (fork HOME → this-turn's TRIGGERING channel → + * process-global default). Public so a multi-segment caller can snapshot it + * ONCE and pin every segment to it via routeSpeech's `overrideChannelId`. */ resolveLocus(conversationId: string): string | null { const home = this.homeChannelResolver?.(conversationId); return home ?? this.activeChannelResolver?.(conversationId) ?? this.globalFallbackFor(conversationId); @@ -3608,7 +3722,27 @@ export class ChannelRegistry { * possible to send into a closed channel: a closed but registered * destination is opened first (`openSource` names why), once its server * and that server's channels.publish grant are known to take the send, so - * a refused send opens nothing; and a failed open sends nothing. + * a send refused before the open opens nothing; and a failed open sends + * nothing. + * + * A channel its server declared inbound-only (isInboundOnly) takes nothing + * the host would post, whichever path the words took, a drafts resend + * included: it's refused before anything opens. The channel's queued + * lifecycle work and the open await the server, and a channels/changed + * update, a re-register or the open's own answer (applyOpenDeclaration) + * may declare the channel inbound-only meanwhile, or a removal take it + * away. So the check just before the publish, after every await, reads + * the channel's registration as it is then, and refuses one that is gone + * or inbound-only. A removal during a wait already fails it there + * (waitForOpenChannel); the check also catches one made after the last + * wait, as the opening's notice can. Words refused after an open for a + * declaration leave the channel open, and the agent gets the failure for + * the words. An open by delivery (`openSource` 'opened-by-delivery', the + * default) that turned a closed channel open also gives the notice that + * it's open, which says how to close it. One that only + * reconfirmed the transport of a channel already decided open or tuned out + * changes no decision, and gives none. When the registration is gone, the + * failure says the channel was removed. * * Only the connector's `delivered: true` confirms a post. `failed` means * nothing was posted: the attempt stopped before the request was written, @@ -3645,6 +3779,9 @@ export class ChannelRegistry { // A send that would be refused is refused before anything opens: a // closed channel opens only for words that can go out. + if (isInboundOnly(entry)) { + return { status: 'failed', destination, reason: inboundOnlyReason(entry), at: at() }; + } const server = this.serverRegistry.getServer(entry.serverId); if (!server) { return { status: 'failed', destination, reason: `server "${entry.serverId}" not found`, at: at() }; @@ -3726,6 +3863,19 @@ export class ChannelRegistry { } } + // Checked again here, after every await above, the lifecycle work's, the + // open's or any other: a registration that is gone, or inbound-only, at + // this moment takes nothing. The entry is read from the map rather than + // taken from the last wait: the opening's notice runs in between, and + // publish() resumes a microtask after the wait. + const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); + if (!current) { + return { status: 'failed', destination, reason: 'the channel was removed before the words were published', at: at() }; + } + if (isInboundOnly(current)) { + return { status: 'failed', destination, reason: inboundOnlyReason(current), at: at() }; + } + let result: unknown; try { result = await server.sendChannelsPublish({ @@ -3743,9 +3893,13 @@ export class ChannelRegistry { * Plain speech for one turn segment: publish `text` to the turn's frozen * locus and report the PublishOutcome. A failure (or an uncertain outcome) * is surfaced as a trace and through onRouteFailure, so the agent learns - * its words did not (or may not have) reached anyone. An ephemeral run's - * speech with no locus is the exception: it is its caller's result, so it - * returns as not delivered and surfaces nothing. + * its words did not (or may not have) reached anyone. A `failed` one whose + * caller holds its words with a notice of its own (`heldOnFailure`) gets + * the trace but not onRouteFailure. An ephemeral run's speech with no locus + * (null) is its caller's result: it returns as not delivered and surfaces + * nothing. This is host-owned output routing (see forking-knowledge-miner + * LOCUS-ROUTING-DESIGN), which replaced the per-surface sticky auto-post + * that used to live in discord-mcpl. */ async deliverSpeech( conversationId: string, @@ -3847,6 +4001,9 @@ export class ChannelRegistry { private async handleToolPublish( input: { channelId?: string; content?: string; text?: string }, origin?: ChannelToolOrigin, + /** Filled with the publish's outcome once a request is attempted, for + * publishForAgent's trace. */ + seen?: { outcome?: PublishOutcome }, ): Promise { // Resolve content: accept both `content` and `text` (backward compat) const messageText = input.content ?? input.text; @@ -3881,6 +4038,13 @@ export class ChannelRegistry { isError: true, }; } + if (isInboundOnly(entry)) { + return { + success: false, + error: `Channel ${channelId} is ${inboundOnlyReason(entry)}`, + isError: true, + }; + } const server = this.serverRegistry.getServer(entry.serverId); if (!server) { @@ -3919,6 +4083,7 @@ export class ChannelRegistry { } catch (err) { outcome = publishErrorOutcome(err, destination, Date.now()); } + if (seen) seen.outcome = outcome; if (outcome.status === 'delivered') return { success: true, data: result }; // The connector's data may say what a multi-part send already posted. const detail = outcome.status === 'unknown' && outcome.detail !== undefined ? ` Connector detail: ${connectorText(outcome.detail)}` : ''; diff --git a/src/mcpl/prose-stream-router.ts b/src/mcpl/prose-stream-router.ts index 308f554ff..e8337e7d7 100644 --- a/src/mcpl/prose-stream-router.ts +++ b/src/mcpl/prose-stream-router.ts @@ -16,8 +16,13 @@ * * Suppression is fail-closed: text whose destination is unknown (explicit * mode before the first `>>`, unresolvable/ambiguous targets, `>>skip_reply`) - * emits NOTHING — the delivery path will bounce/retain it; a stream consumer - * must never see text that delivery would refuse to send. + * emits NOTHING — the delivery path will bounce/retain it. Text for a channel + * the `speakable` option refuses emits nothing either, and which servers see + * what is emitted is the registry's (sendOutgoingChunk). The stream can't + * know what delivery decides later: `speakable` is asked as the text is + * routed, so a channel declared inbound-only after its text was emitted, or + * one whose open then fails, is refused at delivery, though a consumer has + * seen the text. * * Three modes, mirroring proseRouting: * - explicit: parse `>>` prefixes. A fresh turn's stream starts suppressed; @@ -58,6 +63,14 @@ export class ProseStreamRouter { initialTarget: string | null; /** Resolve a `>>spec` to a channelId, or null (unresolved/ambiguous). */ resolve: (spec: string) => string | null; + /** Whether speech may be published to this channel now. Text for a + * channel it refuses is neither emitted nor kept for outgoing/complete, + * so the stream's end can't carry it there should that change before + * the stream ends. Asked as the text is routed, while delivery decides + * when its round is admitted: text refused here stays out of the + * stream even if the channel accepts speech again by then and the + * round posts it. Every channel, when left out. */ + speakable?: (channelId: string) => boolean; }, ) { this.target = opts.initialTarget; @@ -74,7 +87,8 @@ export class ProseStreamRouter { * this at each round's end, with the turn's destination: a private round's * prefixes, a `>>skip_reply` envelope and a target that didn't resolve all * leave that destination as it was, though the stream followed them while - * the round was written. + * the round was written. A target refused for its server's inbound-only + * declaration clears it. */ retarget(target: string | null): void { this.target = target; @@ -100,7 +114,7 @@ export class ProseStreamRouter { feed(delta: string): RoutedDelta[] { const out: RoutedDelta[] = []; for (const ch of delta) this.step(ch, out); - return coalesce(out); + return this.keep(out); } /** @@ -127,7 +141,7 @@ export class ProseStreamRouter { this.held = ''; this.token = ''; if (this.opts.mode !== 'locus') this.state = 'line-start'; - return coalesce(out); + return this.keep(out); } /** End of stream: the last segment ends. */ @@ -253,7 +267,17 @@ export class ProseStreamRouter { private emit(text: string, out: RoutedDelta[]): void { if (this.target === null || !text) return; out.push({ channelId: this.target, delta: text }); - this.accumulated.set(this.target, (this.accumulated.get(this.target) ?? '') + text); + } + + /** + * What one feed or segment end routed, less any channel `speakable` + * refuses, kept for outgoing/complete. A feed runs without yielding, so + * one answer per channel holds for all of it. + */ + private keep(out: RoutedDelta[]): RoutedDelta[] { + const kept = coalesce(out).filter((d) => this.opts.speakable?.(d.channelId) ?? true); + for (const d of kept) this.accumulated.set(d.channelId, (this.accumulated.get(d.channelId) ?? '') + d.delta); + return kept; } } diff --git a/src/types/trace.ts b/src/types/trace.ts index dca24323c..a91f2322b 100644 --- a/src/types/trace.ts +++ b/src/types/trace.ts @@ -442,13 +442,16 @@ export type TraceEvent = }) // Host-owned speech routing (ChannelRegistry.routeSpeech). One trace per - // delivered prose segment. `text` is the segment verbatim — the same content + // delivered prose segment, and one per publish the subconscious's + // speak_in_channel delivers (ChannelRegistry.publishForAgent): that one's + // conversationId is the speaking agent's name, and it carries no + // messageId. `text` is the segment verbatim — the same content // already on the bus chunk-by-chunk via inference:tokens, and mcpl:* traces // are 'ops'-scoped on the webui wire — and `messageId` is the surface's id - // for the posted message (when the MCPL server reports one). Together they - // give external taps the (channelId, messageId, text) tuple needed to edit - // a just-posted message, e.g. a TTS interruption truncating a reply to the - // words actually voiced. + // for the posted message (on routeSpeech's traces, when the MCPL server + // reports one). Together they give external taps the (channelId, + // messageId, text) tuple needed to edit a just-posted message, e.g. a TTS + // interruption truncating a reply to the words actually voiced. | (TraceEventBase & { type: 'mcpl:speech-routed'; /** Agent / conversation whose speech was routed. */ @@ -459,7 +462,8 @@ export type TraceEvent = textLen: number; /** The routed prose segment, verbatim. */ text: string; - /** Surface message id (e.g. Discord message id), when reported. */ + /** Surface message id (e.g. Discord message id), when reported, on + * routeSpeech's traces. speak_in_channel's carry none. */ messageId?: string; }) | (TraceEventBase & { @@ -473,6 +477,9 @@ export type TraceEvent = /** `failed`: nothing was posted; `unknown`: the request was dispatched * and no valid receipt came back, so it may have been. */ outcome?: 'failed' | 'unknown'; + /** The connector error's structured data, bounded (connectorText), + * when it sent some: it may say what a multi-part send posted. */ + detail?: string; }) // Admin puppet: an operator executed a tool AS an agent and stored the diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index b6c80e178..3cd3b782a 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -40,6 +40,9 @@ function fixture(onNotice?: NoticeHandler, store?: JsStore, coalesce?: RegistryO const routeFailures: Array<{ reason: string }> = []; // A server that answers an open for another channel, by requested id. const answeredFor = new Map(); + // The direction an open's answer declares, by the address the open was + // sent to; an address not here declares the descriptor's own. + const declares = new Map(); const serve = (kind: 'open' | 'close', params: { channelId?: string; address?: unknown }) => { const id = params.channelId!; calls.push({ kind, id, address: params.address }); @@ -47,7 +50,10 @@ function fixture(onNotice?: NoticeHandler, store?: JsStore, coalesce?: RegistryO const finish = () => { const target = kind === 'open' ? answeredFor.get(id) ?? id : id; actual.set(target, kind === 'open'); - resolve(kind === 'open' ? { channel: descriptor(target) } : { closed: true }); + const declared = declares.get(String((params.address as { value?: unknown } | undefined)?.value)); + resolve(kind === 'open' + ? { channel: { ...descriptor(target), ...(declared ? { direction: declared } : {}) } } + : { closed: true }); }; if (held.has(id)) pending.push({ finish, fail: () => reject(new Error('fixture lifecycle failure')) }); else finish(); @@ -87,7 +93,7 @@ function fixture(onNotice?: NoticeHandler, store?: JsStore, coalesce?: RegistryO assert.equal(settled, true, 'lifecycle operations must settle'); await result; }; - return { registry, actual, calls, pending, held, traces, isOpen, tool, drain, publishedWhileOpen, notices, routeFailures, replaceServer, answeredFor, setGrant }; + return { registry, actual, calls, pending, held, traces, isOpen, tool, drain, publishedWhileOpen, notices, routeFailures, replaceServer, answeredFor, declares, setGrant }; } test('remove/re-add during reconcile converges on the replacement desired state, with ACK first (#185)', async () => { @@ -317,6 +323,108 @@ test('a draft published into a channel with a close pending waits for it and pos assert.equal(f.notices.length, 0); }); +test('a channel declared inbound-only while a publish waits for its lifecycle work is refused, and nothing is published', async () => { + // A publish that finds the channel's lifecycle work queued waits for it. + // A channels/changed during that wait can declare the channel + // inbound-only, so the publish checks the channel again after the wait, + // just before it publishes, and refuses it there. + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + f.held.add('x'); + const open = f.tool('open'); + await tick(); + const draft = f.registry.publish('resident', 'a draft', { serverId: 'test', channelId: 'x' }); + await tick(); + await f.registry.handleChanged('test', { updated: [{ ...descriptor('x'), direction: 'inbound' }] }); + await f.drain(open, draft); + assert.equal((await open).success, true); + const outcome = await draft; + assert.equal(outcome.status, 'failed'); + assert.match(outcome.reason ?? '', /^inbound-only: server "test" declared this channel's direction "inbound"/); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + assert.equal(f.isOpen(), true, 'the open the agent asked for stands'); +}); + +for (const change of ['announced again as inbound-only', 'removed'] as const) { + test(`a channel ${change} while its opening is announced is refused, and nothing is published`, async () => { + // The notice of an opening runs after the open and before the publish. A + // channels/changed there can replace the channel's entry or remove it, + // so the check before the publish reads the channel's current entry, not + // the one the open returned. + let announced: Promise | undefined; + let f!: ReturnType; + f = fixture(() => { + announced = f.registry.handleChanged('test', change === 'removed' + ? { removed: ['x'] } + : { added: [{ ...descriptor('x'), direction: 'inbound' }] }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + assert.equal(await f.registry.routeSpeech('resident', 'hello', 'x'), null); + assert.equal(f.notices.length, 1, 'the opening was announced'); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + assert.match(f.routeFailures[0]?.reason ?? '', change === 'removed' + ? /^the channel was removed before the words were published$/ + : /^inbound-only: server "test" declared this channel's direction "inbound"/); + await announced; + }); +} + +for (const opener of ['tool', 'reconcile'] as const) { + test(`an open answered after the agent closed the channel during it leaves it closed, and its inbound-only declaration applies (${opener})`, async () => { + // The close is the newer decision, so the channel ends closed. The answer + // is still the server's word about the channel, so a later publish is + // refused before anything opens. + const f = fixture(); + f.declares.set('original', 'inbound'); + if (opener === 'tool') await f.registry.handleChanged('test', { added: [descriptor('x')] }); + f.held.add('x'); + const opening = opener === 'tool' + ? f.tool('open') + : f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); + assert.equal(f.pending.length, 1, 'the open is in flight'); + const close = f.tool('close'); + await f.drain(opening, close); + assert.equal((await close).success, true); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.equal(f.isOpen(), false); + assert.equal(f.actual.get('x'), false); + assert.equal(f.registry.inboundOnlyServer('x', 'test'), 'test', "the late answer's declaration applies"); + const opens = f.calls.filter((c) => c.kind === 'open').length; + const outcome = await f.registry.publish('resident', 'words', { serverId: 'test', channelId: 'x' }); + assert.equal(outcome.status, 'failed'); + assert.match(outcome.reason ?? '', /^inbound-only: /); + assert.equal(f.calls.filter((c) => c.kind === 'open').length, opens, 'refused before anything opens'); + }); +} + +test("an open answered for another channel applies nothing of its answer to the channel opened", async () => { + // A substitution fails the open before the answer's declaration is read. + // That is what lets an answer whose channel names no id stand for the + // channel opened. + const f = fixture(); + f.answeredFor.set('x', 'y'); + f.declares.set('original', 'inbound'); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + assert.equal((await f.tool('open')).success, false); + assert.ok(f.traces.some((e) => e.type === 'mcpl:channel-open-substituted' && e.channelId === 'x'), + 'the open reached the server and was answered for another channel'); + assert.equal(f.registry.inboundOnlyServer('x', 'test'), undefined, "the other channel's declaration isn't applied"); +}); + +test("an open's answer for a target the channel has since left applies nothing; the new target's answer decides", async () => { + const f = fixture(); + f.declares.set('original', 'inbound'); + f.held.add('x'); + const opening = f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'new-target')] }); + await f.drain(opening); + assert.deepEqual(f.calls.map((c) => c.address), [{ value: 'original' }, { value: 'new-target' }]); + assert.equal(f.isOpen(), true); + assert.equal(f.registry.inboundOnlyServer('x', 'test'), undefined, "the old target's declaration isn't applied"); +}); + test('waiting for a backscroll does not issue new opens or auto-open notices for each speech', async () => { const f = fixture(); await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); diff --git a/test/channel-registry-routing.test.ts b/test/channel-registry-routing.test.ts index 80a825658..abbcf4fe8 100644 --- a/test/channel-registry-routing.test.ts +++ b/test/channel-registry-routing.test.ts @@ -2,11 +2,12 @@ import { CapabilityGrant, ALL_CAPABILITY_PATHS } from '../src/mcpl/capability-gr import { test } from 'node:test'; import assert from 'node:assert/strict'; import { ChannelRegistry } from '../src/mcpl/channel-registry.js'; +import { McplRequestError, McplServerConnection } from '../src/mcpl/server-connection.js'; import { parseProsePrefix } from '../src/mcpl/prose-grammar.js'; import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; -type RouteFailure = { conversationId: string; channelId: string | null; reason: string; textLen: number }; +type RouteFailure = { conversationId: string; channelId: string | null; reason: string; textLen: number; outcome?: 'failed' | 'unknown'; detail?: string }; /** * Build a registry with a mock server whose publish result is configurable, @@ -29,24 +30,43 @@ function makeRegistry( const openCalls: Array<{ channelId?: string }> = []; const closeCalls: Array<{ channelId?: string }> = []; + const streamCalls: Array<{ kind: 'chunk' | 'complete'; channelId?: string }> = []; let failOpens = false; + let onOpen: (() => Promise) | undefined; + let onPublish: (() => Promise) | undefined; + let onClose: (() => Promise) | undefined; + let onAcknowledge: (() => Promise) | undefined; + let openResult: unknown = {}; const mockServer = { // Post-policy state: full grant, so tests exercise delivery, not §5.3 denial. grant: new CapabilityGrant(new Set(ALL_CAPABILITY_PATHS), []), sendChannelsPublish: async (params: { channelId?: string; conversationId?: string }) => { publishCalls.push(params); + await onPublish?.(); return publishResult; }, + sendChannelsOutgoingChunk: (params: { channelId?: string }) => { + streamCalls.push({ kind: 'chunk', channelId: params.channelId }); + }, + sendChannelsOutgoingComplete: (params: { channelId?: string }) => { + streamCalls.push({ kind: 'complete', channelId: params.channelId }); + }, sendChannelsOpen: async (params: { channelId?: string }) => { if (failOpens) throw new Error('open refused by server'); openCalls.push(params); - return {}; + await onOpen?.(); + return openResult; }, sendChannelsClose: async (params: { channelId?: string }) => { closeCalls.push(params); + await onClose?.(); return {}; }, + sendChannelsAcknowledge: async () => { + await onAcknowledge?.(); + return { acknowledged: true }; + }, }; const serverRegistry = { getServer: (_id: string) => mockServer, @@ -73,8 +93,17 @@ function makeRegistry( }).findChannelEntry(channelId); return { - registry, failures, traces, publishCalls, openCalls, closeCalls, lookup, + registry, failures, traces, publishCalls, openCalls, closeCalls, streamCalls, lookup, setFailOpens: (v: boolean) => { failOpens = v; }, + /** Runs while the server is answering an open, before it returns. */ + setOnOpen: (fn: (() => Promise) | undefined) => { onOpen = fn; }, + /** Runs while the server is answering a publish, before it returns. */ + setOnPublish: (fn: (() => Promise) | undefined) => { onPublish = fn; }, + /** What the server answers channels/open with. */ + setOpenResult: (result: unknown) => { openResult = result; }, + /** Runs while the server is answering a close, or an acknowledgement. */ + setOnClose: (fn: (() => Promise) | undefined) => { onClose = fn; }, + setOnAcknowledge: (fn: (() => Promise) | undefined) => { onAcknowledge = fn; }, }; } @@ -133,7 +162,8 @@ test('routeSpeech surfaces a failure when the server reports delivered:false', a assert.equal(failures.length, 1, 'onRouteFailure should fire'); assert.equal(failures[0].channelId, 'ch-x'); assert.match(failures[0].reason, /delivered:false/); - assert.ok(traces.some(t => t.type === 'mcpl:speech-route-failed')); + assert.equal(failures[0].outcome, 'failed', 'the server said nothing was delivered'); + assert.ok(traces.some(t => t.type === 'mcpl:speech-route-failed' && t.outcome === 'failed')); }); test('routeSpeech routes a conversation fork to its HOME channel, not the global last-inbound (item 3)', async () => { @@ -354,6 +384,7 @@ test('routeSpeech does NOT deliver when the open-on-delivery fails', async () => assert.equal(publishCalls.length, publishesBefore, 'publish must not be attempted'); assert.equal(failures.length, 1); assert.match(failures[0]!.reason, /closed and open failed/); + assert.equal(failures[0]!.outcome, 'failed', 'nothing was published'); }); test('routeSpeech with an omitted locus fails loudly as a routing bug', async () => { @@ -590,3 +621,527 @@ test('proseTargetFor: no token when none is safe (shared id across servers, whit assert.equal(registry.proseTargetFor('chan:8', 'discord-a'), '#only-a'); assert.equal(registry.proseTargetFor('chan:8', 'discord-b'), undefined); }); + +// --------------------------------------------------------------------------- +// Inbound-only channels (SPEC §14.2 `direction: 'inbound'`): the host posts +// nothing into them. pocket-body declares its live transcript this way and +// refuses publishes there, and another agent speaking in the room arrives on +// it addressed, so a resident's plain speech can be aimed at it. +// --------------------------------------------------------------------------- + +async function registerInboundOnly(registry: ChannelRegistry): Promise { + await registry.handleChanged('body', { + added: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'inbound' }], + } as never); +} + +test('plain speech into a channel its server declared inbound-only is refused before the channel is opened or published to', async () => { + const { registry, failures, traces, publishCalls, openCalls, lookup } = makeRegistry({ delivered: true }); + await registerInboundOnly(registry); + assert.equal(lookup('body:ears')?.open, false, 'precondition: registered closed'); + + const result = await registry.routeSpeech('linn', 'Hello, room.', 'body:ears'); + + assert.equal(result, null, 'nothing was delivered'); + assert.deepEqual(publishCalls, [], 'nothing was published'); + assert.deepEqual(openCalls, [], 'the delivery attempt did not open the channel'); + assert.equal(lookup('body:ears')?.open, false); + assert.equal(failures.length, 1, 'the refusal reaches the resident as a send failure'); + assert.equal(failures[0]!.channelId, 'body:ears'); + assert.match(failures[0]!.reason, /^inbound-only: server "body" declared this channel's direction "inbound"/); + assert.match(failures[0]!.reason, /use that server's own tools$/); + assert.ok(traces.some((t) => t.type === 'mcpl:speech-route-failed' && t.channelId === 'body:ears')); +}); + +test('publish() itself refuses an inbound-only channel, so a drafts resend, which names the server, is refused before anything opens', async () => { + const { registry, failures, publishCalls, openCalls } = makeRegistry({ delivered: true }); + await registerInboundOnly(registry); + + // As a drafts resend calls it: an exact destination, no speech routing. + const outcome = await registry.publish('linn', 'held words', { serverId: 'body', channelId: 'body:ears' }); + + assert.equal(outcome.status, 'failed', 'nothing was posted'); + assert.deepEqual(outcome.destination, { serverId: 'body', channelId: 'body:ears', label: 'Body · ears' }); + assert.match(outcome.reason ?? '', /^inbound-only: server "body" declared this channel's direction "inbound"/); + assert.deepEqual(openCalls, [], 'the closed channel was not opened'); + assert.deepEqual(publishCalls, []); + assert.deepEqual(failures, [], "the caller reports it: publish() writes no marker of its own"); +}); + +test('channel_publish into an inbound-only channel is refused, naming the declaration, and publishes nothing', async () => { + const { registry, publishCalls } = makeRegistry({ delivered: true }); + await registerInboundOnly(registry); + + const result = await registry.handleChannelToolCall('channel_publish', { + channelId: 'body:ears', + content: 'Hello, room.', + }); + + assert.equal(result.success, false); + assert.equal(result.isError, true); + assert.match(String(result.error), /^Channel body:ears is inbound-only: server "body" declared this channel's direction "inbound"/); + assert.deepEqual(publishCalls, [], 'nothing was published'); +}); + +test('an inbound-only channel gets no outgoing stream', async () => { + const { registry, streamCalls } = makeRegistry({ delivered: true }); + await registerInboundOnly(registry); + + registry.sendOutgoingChunk('body:ears', 'linn', 'inf_1', 0, 'Hello'); + registry.sendOutgoingComplete('body:ears', 'linn', 'inf_1', 'Hello, room.'); + + assert.deepEqual(streamCalls, []); +}); + +test('a channel declared outbound or bidirectional, or with no direction, still takes speech, channel_publish and the stream', async () => { + const { registry, failures, publishCalls, streamCalls } = makeRegistry({ delivered: true }); + await registry.handleChanged('body', { + added: [ + { id: 'body:out', type: 'chat', label: 'out', direction: 'outbound' }, + { id: 'body:both', type: 'chat', label: 'both', direction: 'bidirectional' }, + { id: 'body:unstated', type: 'chat', label: 'unstated' }, + ], + } as never); + + for (const id of ['body:out', 'body:both', 'body:unstated']) { + assert.ok(await registry.routeSpeech('linn', `speech to ${id}`, id), `speech delivered to ${id}`); + const published = await registry.handleChannelToolCall('channel_publish', { channelId: id, content: `publish to ${id}` }); + assert.equal(published.success, true, `channel_publish delivered to ${id}`); + registry.sendOutgoingChunk(id, 'linn', 'inf_1', 0, 'x'); + } + + assert.deepEqual(failures, []); + assert.deepEqual(publishCalls.map((c) => c.channelId), [ + 'body:out', 'body:out', 'body:both', 'body:both', 'body:unstated', 'body:unstated', + ]); + assert.deepEqual(streamCalls.map((c) => c.channelId), ['body:out', 'body:both', 'body:unstated']); +}); + +test('inboundOnlyServer names the declaring server for an inbound-only channel, and nothing for any other', async () => { + // The routing notices ask the registry this, so they say what routeSpeech does. + const { registry } = makeRegistry({ delivered: true }); + await registerInboundOnly(registry); + await registry.handleChanged('body', { + added: [ + { id: 'body:out', type: 'chat', label: 'out', direction: 'outbound' }, + { id: 'body:both', type: 'chat', label: 'both', direction: 'bidirectional' }, + { id: 'body:unstated', type: 'chat', label: 'unstated' }, + ], + } as never); + + assert.equal(registry.inboundOnlyServer('body:ears'), 'body'); + for (const id of ['body:out', 'body:both', 'body:unstated', 'body:unregistered']) { + assert.equal(registry.inboundOnlyServer(id), undefined, id); + } +}); + +test('the refusal follows the descriptor its server declared last: channels/register, and channels/changed updates either way', async () => { + const { registry, failures, publishCalls } = makeRegistry({ delivered: true }); + + // channels/register declares it inbound-only. + await registry.handleRegister('body', { + channels: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'inbound' }], + } as never); + assert.equal(await registry.routeSpeech('linn', 'one', 'body:ears'), null); + assert.equal(registry.inboundOnlyServer('body:ears'), 'body'); + + // An update declares it bidirectional: speech goes out. + await registry.handleChanged('body', { + updated: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'bidirectional' }], + } as never); + assert.ok(await registry.routeSpeech('linn', 'two', 'body:ears')); + assert.equal(registry.inboundOnlyServer('body:ears'), undefined); + + // Another declares it inbound-only again: refused again. + await registry.handleChanged('body', { + updated: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'inbound' }], + } as never); + assert.equal(await registry.routeSpeech('linn', 'three', 'body:ears'), null); + + assert.deepEqual(publishCalls.map((c) => c.channelId), ['body:ears'], 'only the bidirectional turn published'); + assert.equal(failures.length, 2); + for (const f of failures) { + assert.match(f.reason, /^inbound-only: server "body" declared this channel's direction "inbound"/); + } +}); + +test('publishForAgent into an inbound-only channel is refused and traced as failed; into any other channel it traces once as routed, after the publish', async () => { + const { registry, publishCalls, traces } = makeRegistry({ delivered: true, messageId: 'm-77' } as { delivered: boolean }); + await registerInboundOnly(registry); + await registry.handleChanged('body', { + added: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + const routed = () => traces.filter((t) => t.type === 'mcpl:speech-routed'); + + const refused = await registry.publishForAgent('body:ears', 'a thought', 'Subconscious'); + assert.equal(refused.success, false); + assert.match(String(refused.error), /inbound-only: server "body" declared this channel's direction "inbound"/); + assert.equal(publishCalls.length, 0, 'nothing was published'); + assert.equal(routed().length, 0, 'speech that was never posted is not traced as routed'); + const failed = traces.filter((t) => t.type === 'mcpl:speech-route-failed'); + assert.equal(failed.length, 1, 'the refusal is traced as a failure'); + assert.equal(failed[0]!.conversationId, 'Subconscious'); + assert.equal(failed[0]!.channelId, 'body:ears'); + assert.match(String(failed[0]!.reason), /inbound-only: server "body" declared/); + assert.equal(failed[0]!.outcome, 'failed', 'nothing was published'); + + const posted = await registry.publishForAgent('body:home', 'a thought', 'Subconscious'); + assert.equal(posted.success, true); + assert.deepEqual(publishCalls.map((c) => c.channelId), ['body:home']); + // Shaped as routeSpeech's trace is, but without the surface's message id, + // though the server named one ('m-77'). + assert.deepEqual(routed(), [{ + type: 'mcpl:speech-routed', conversationId: 'Subconscious', serverId: 'body', channelId: 'body:home', + delivered: true, textLen: 'a thought'.length, text: 'a thought', + }], "no messageId: connectome-host's TTS relay keeps every routed trace without reading conversationId, " + + 'and when an interruption of the resident matches no entry it edits the newest traced message in the ' + + "channel, so an id here would let it trim the subconscious's post"); +}); + +test('publishForAgent traces a publish the server reports undelivered as failed, not routed', async () => { + const { registry, publishCalls, traces } = makeRegistry({ delivered: false }); + await registry.handleChanged('body', { + added: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + + await registry.publishForAgent('body:home', 'a thought', 'Subconscious'); + + assert.deepEqual(publishCalls.map((c) => c.channelId), ['body:home']); + assert.deepEqual(traces.filter((t) => t.type === 'mcpl:speech-routed'), []); + assert.deepEqual( + traces.filter((t) => t.type === 'mcpl:speech-route-failed').map((t) => [t.conversationId, t.channelId, t.reason, t.outcome]), + [['Subconscious', 'body:home', 'server "body" reported delivered:false', 'failed']], + ); +}); + +test("publishForAgent traces what a rejected publish may have done: unknown after an error response or a timeout, failed when the request never left", async () => { + const { registry, traces, setOnPublish } = makeRegistry({ delivered: true }); + await registry.handleChanged('body', { + added: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + const failed = () => traces.filter((t) => t.type === 'mcpl:speech-route-failed'); + + // An error response may follow a partial post. + setOnPublish(async () => { + throw new McplRequestError('The "body" server answered with an error: Rate limited after part 1', 'error-response', { posted: ['m-1'] }); + }); + const unconfirmed = await registry.publishForAgent('body:home', 'a long thought', 'Subconscious'); + assert.equal(unconfirmed.error, 'Publish to body:home wasn\'t confirmed: The "body" server answered with an error: Rate limited after part 1 (delivery uncertain). It may or may not have been posted, so sending it again could post it twice. Connector detail: {"posted":["m-1"]}'); + assert.deepEqual(failed().map((t) => [t.reason, t.outcome, t.detail]), + [['The "body" server answered with an error: Rate limited after part 1 (delivery uncertain)', 'unknown', '{"posted":["m-1"]}']], + "the trace carries the connector's detail, as plain speech's does"); + + // So may a request that went out and got no answer. + setOnPublish(async () => { + throw new McplRequestError('MCPL server "body" did not respond to channels/publish within 30000ms', 'no-response'); + }); + await registry.publishForAgent('body:home', 'a thought', 'Subconscious'); + assert.equal(failed()[1]!.outcome, 'unknown'); + + // Only a request that never left proves nothing was posted. + setOnPublish(async () => { + throw new McplRequestError('Cannot send request: connection to "body" is closed', 'not-sent'); + }); + const notSent = await registry.publishForAgent('body:home', 'another', 'Subconscious'); + assert.match(notSent.error ?? '', /^Publish to body:home failed: .*is closed\. Nothing was posted\./); + assert.equal(failed()[2]!.outcome, 'failed'); + assert.equal('detail' in failed()[2]!, false, 'no detail when the connector sent none'); + assert.equal(traces.filter((t) => t.type === 'mcpl:speech-routed').length, 0); +}); + +test('buildChannelContext names no defaultOutgoing for a locus its server declared inbound-only', async () => { + const active: Record = { linn: 'body:ears', nell: 'body:home' }; + const { registry } = makeRegistry({ delivered: true }, () => undefined, (n) => active[n]); + await registerInboundOnly(registry); + await registry.handleChanged('body', { + added: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + + assert.equal(registry.buildChannelContext('linn')?.defaultOutgoing, undefined, 'speech goes nowhere, so no channel is named'); + assert.equal(registry.buildChannelContext('nell')?.defaultOutgoing?.channelId, 'body:home'); +}); + +test('an inbound-only channel resolves as a speech target to an error naming the declaration, and still resolves for reading', async () => { + const { registry } = makeRegistry({ delivered: true }); + await registerInboundOnly(registry); + await registry.handleChanged('body', { + added: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + + const speech = registry.resolveProseTarget('body:ears', { forSpeech: true }); + assert.ok('error' in speech, 'refused as a target for speech'); + assert.match( + (speech as { error: string }).error, + /^body:ears is inbound-only: server "body" declared this channel's direction "inbound"/, + ); + // It names the refused channel, which a name that resolves to nothing doesn't. + assert.equal((speech as { refusedChannelId?: string }).refusedChannelId, 'body:ears'); + const nowhere = registry.resolveProseTarget('#nowhere', { forSpeech: true }); + assert.ok('error' in nowhere && !('refusedChannelId' in nowhere), 'an unknown name is no refusal'); + // Reading its history still finds it. + assert.deepEqual(registry.resolveProseTarget('body:ears'), { channelId: 'body:ears', label: 'Body · ears' }); + assert.deepEqual(registry.resolveProseTargetDurable('body:ears'), { channelId: 'body:ears', label: 'Body · ears' }); + // No reply token is offered for it; one is for the bidirectional channel. + assert.equal(registry.proseTargetFor('body:ears', 'body'), undefined); + assert.ok(registry.proseTargetFor('body:home', 'body')); + // A bidirectional channel resolves either way. + assert.deepEqual(registry.resolveProseTarget('body:home', { forSpeech: true }), { channelId: 'body:home', label: 'Body · home' }); +}); + +test('a channel declared inbound-only while the host opens it for speech is refused before the publish, and stays open with the open\'s notice', async () => { + const autoOpened: Array<{ source: string; channels: Array<{ channelId: string }> }> = []; + const { registry, publishCalls, openCalls, closeCalls, failures, setOnOpen, lookup } = + makeRegistry({ delivered: true }, undefined, undefined, (info) => autoOpened.push(info)); + await registry.handleRegister('body', { + channels: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'bidirectional' }], + } as never); + // Registering reconciles the channel to its default, closed, which sends + // one channels/close; the claim below is about what happens after that. + const closesAtRegistration = closeCalls.length; + setOnOpen(async () => { + await registry.handleChanged('body', { + updated: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'inbound' }], + } as never); + }); + + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:ears'), null); + assert.deepEqual(openCalls.map((c) => c.channelId), ['body:ears'], 'the closed channel was opened'); + assert.equal(publishCalls.length, 0, 'nothing was published after the declaration'); + assert.equal(failures.length, 1); + assert.match(failures[0]!.reason, /^inbound-only: server "body" declared this channel's direction "inbound"/); + // The open happened, whatever the recheck decided: the channel stays open, + // and the open's notice, which says how to close it, went out with the marker. + assert.equal(lookup('body:ears')?.open, true, 'the opened channel stays open'); + assert.equal(closeCalls.length, closesAtRegistration, 'nothing closes it after the open'); + assert.deepEqual(autoOpened.map((n) => [n.source, n.channels.map((c) => c.channelId)]), [['opened-by-delivery', ['body:ears']]]); +}); + +test("publishForAgent's routed trace names the server it published through, though the channel is removed while the reply is pending", async () => { + const { registry, traces, setOnPublish } = makeRegistry({ delivered: true }); + await registry.handleChanged('body', { + added: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + setOnPublish(async () => { + await registry.handleChanged('body', { removed: ['body:home'] } as never); + }); + + const posted = await registry.publishForAgent('body:home', 'a thought', 'Subconscious'); + assert.equal(posted.success, true); + const routed = traces.filter((t) => t.type === 'mcpl:speech-routed'); + assert.equal(routed.length, 1); + assert.equal(routed[0]!.serverId, 'body'); +}); + +test("a server's error response to the publish goes through fail() as unconfirmed: a trace and the send-failure marker (R1-2)", async () => { + const { registry, failures, traces, publishCalls, setOnPublish } = makeRegistry({ delivered: true }); + await registry.handleRegister('body', { + channels: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + setOnPublish(async () => { + throw new McplRequestError('The "body" server answered with an error: Channel not permitted', 'error-response'); + }); + + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:home'), null); + assert.equal(publishCalls.length, 1, 'the publish was sent, and the server answered with an error'); + assert.deepEqual(failures.map((f) => f.reason), ['The "body" server answered with an error: Channel not permitted (delivery uncertain)'], + "the marker gives the server's own words, and says delivery is uncertain"); + // An error response may follow a partial post, so whether it was posted is unknown. + assert.deepEqual(failures.map((f) => f.outcome), ['unknown']); + const failed = traces.filter((t) => t.type === 'mcpl:speech-route-failed'); + assert.equal(failed.length, 1); + assert.equal(failed[0]!.channelId, 'body:home'); + assert.equal(failed[0]!.outcome, 'unknown'); + assert.equal(traces.filter((t) => t.type === 'mcpl:speech-routed').length, 0); +}); + +test("a lazily registered channel whose open answers inbound-only is refused before the publish, and stays refused (R1-3)", async () => { + const announced: Array<{ source: string; channels: Array<{ channelId: string }> }> = []; + const { registry, publishCalls, openCalls, failures, traces, setOpenResult } = + makeRegistry({ delivered: true }, undefined, undefined, (info) => { announced.push(info); }); + registry.ensureChannelRegistered('body', 'body:ears', 'Body · ears'); + setOpenResult({ channel: { id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'inbound' } }); + + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:ears'), null); + assert.deepEqual(openCalls.map((c) => c.channelId), ['body:ears'], 'speech opened the channel, as for any closed locus'); + assert.equal(publishCalls.length, 0, 'nothing was published after the server declared it'); + assert.match(failures[0]!.reason, /^inbound-only: server "body" declared this channel's direction "inbound"/); + // The open happened, and the channel stays open: the notice says so, and + // how to close it, while the marker reports the refused speech. + assert.deepEqual(announced.map((a) => [a.source, a.channels.map((c) => c.channelId)]), [['opened-by-delivery', ['body:ears']]]); + assert.equal(traces.filter((t) => t.type === 'mcpl:channel-opened-by-send').length, 1); + assert.equal(registry.isChannelOpen('body:ears'), true); + + // The next speech is refused before anything goes to the server. + assert.equal(await registry.routeSpeech('linn', 'again', 'body:ears'), null); + assert.equal(openCalls.length, 1); + assert.equal(publishCalls.length, 0); + assert.equal(failures.length, 2); +}); + +test("an open's answer whose channel names no id is this channel's, so its inbound-only declaration applies (R1-3)", async () => { + // The lifecycle work takes an answer whose channel has no id as the + // requested channel's, not a substitution, so the declaration it carries + // applies as well, and the speech is refused rather than posted. + const { registry, publishCalls, openCalls, failures, setOpenResult } = makeRegistry({ delivered: true }); + registry.ensureChannelRegistered('body', 'body:ears', 'Body · ears'); + setOpenResult({ channel: { direction: 'inbound' } }); + + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:ears'), null); + assert.deepEqual(openCalls.map((c) => c.channelId), ['body:ears'], 'speech opened the channel'); + assert.equal(publishCalls.length, 0, 'nothing was published'); + assert.match(failures[0]!.reason, /^inbound-only: server "body" declared this channel's direction "inbound"/); + assert.equal(registry.isChannelOpen('body:ears'), true, 'the open stands'); +}); + +test("an open's answer that declares no direction, or one that isn't inbound, leaves speech published (R1-3)", async () => { + // An answer for another channel is a substitution, which fails the open + // whatever it declares (channel-open-substitution.test.ts). + for (const answer of [ + { channel: { id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'bidirectional' } }, + {}, + ]) { + const { registry, publishCalls, failures, setOpenResult } = makeRegistry({ delivered: true }); + registry.ensureChannelRegistered('body', 'body:ears', 'Body · ears'); + setOpenResult(answer); + assert.deepEqual(await registry.routeSpeech('linn', 'hello', 'body:ears'), { delivered: true, channelId: 'body:ears' }, + `published after the answer ${JSON.stringify(answer)}`); + assert.equal(publishCalls.length, 1); + assert.equal(failures.length, 0); + } +}); + +test('a channel removed while the host opens it for speech is refused, and nothing is published (R1-3)', async () => { + // Whatever the open's answer declares: there's no entry left to apply it to. + for (const answer of [{}, { channel: { id: 'body:home', type: 'chat', label: 'Body · home', direction: 'inbound' } }]) { + const { registry, publishCalls, openCalls, failures, setOnOpen, setOpenResult } = makeRegistry({ delivered: true }); + await registry.handleRegister('body', { + channels: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + setOpenResult(answer); + setOnOpen(async () => { + await registry.handleChanged('body', { removed: ['body:home'] } as never); + }); + + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:home'), null); + assert.deepEqual(openCalls.map((c) => c.channelId), ['body:home']); + assert.equal(publishCalls.length, 0, 'nothing was published to the removed channel'); + assert.deepEqual(failures.map((f) => f.reason), ['channel is closed and open failed: Channel is no longer registered: body:home'], + `refused as removed, after the answer ${JSON.stringify(answer)}`); + } +}); + +test("a server's refusal of an open or a tool's publish reaches the agent in its own words", async () => { + const refusal = () => new McplRequestError('The "body" server answered with an error: Channel unavailable', 'error-response'); + const { registry, failures, setOnOpen, setOnPublish } = makeRegistry({ delivered: true }); + await registry.handleRegister('body', { + channels: [{ id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }], + } as never); + setOnOpen(async () => { throw refusal(); }); + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:home'), null); + assert.deepEqual(failures.map((f) => f.reason), ['channel is closed and open failed: The "body" server answered with an error: Channel unavailable']); + const opened = await registry.handleChannelToolCall('channel_open', { channelId: 'body:home' }, { kind: 'agent', agentName: 'linn' }); + assert.equal(opened.error, 'Failed to open channel: The "body" server answered with an error: Channel unavailable (it stays selected for opening: a later delivery or reconnect tries again, and channel_close withdraws that)'); + + setOnOpen(undefined); + setOnPublish(async () => { throw new McplRequestError('The "body" server answered with an error: Channel not permitted', 'error-response'); }); + const published = await registry.handleChannelToolCall('channel_publish', { channelId: 'body:home', content: 'hi' }, { kind: 'agent', agentName: 'linn' }); + assert.equal(published.error, 'Publish to body:home wasn\'t confirmed: The "body" server answered with an error: Channel not permitted (delivery uncertain). It may or may not have been posted, so sending it again could post it twice.'); +}); + +test("an open's answer can't lift an inbound-only declaration: a server lifts it by changing or registering the channel (R1-3)", async () => { + const { registry, publishCalls, failures, setOpenResult } = makeRegistry({ delivered: true }); + await registry.handleRegister('body', { + channels: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'inbound' }], + } as never); + setOpenResult({ channel: { id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'bidirectional' } }); + const opened = await registry.handleChannelToolCall('channel_open', { channelId: 'body:ears' }, { kind: 'agent', agentName: 'linn' }); + assert.equal(opened.success, true); + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:ears'), null, 'still refused after the answer'); + assert.equal(publishCalls.length, 0); + assert.match(failures[0]!.reason, /^inbound-only: server "body" declared/); + + // The server's own change lifts it. + await registry.handleChanged('body', { + updated: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'bidirectional' }], + } as never); + assert.deepEqual(await registry.routeSpeech('linn', 'now', 'body:ears'), { delivered: true, channelId: 'body:ears' }); + assert.equal(publishCalls.length, 1); +}); + +test("an open's answer that arrives in the same read as a newer channels/changed can't lift its refusal (R1-3)", async () => { + // Greta-2425's probe: the connection resolves the open's answer, then + // handles the next line at once, so the answer is applied after, and + // over, the channels/changed the server sent after it. + let answerOpen!: (answer: unknown) => void; + let openSent!: () => void; + const opening = new Promise((resolve) => { openSent = resolve; }); + const publishCalls: string[] = []; + const failures: string[] = []; + const mockServer = { + grant: new CapabilityGrant(new Set(ALL_CAPABILITY_PATHS), []), + sendChannelsPublish: async (p: { channelId?: string }) => { publishCalls.push(String(p.channelId)); return { delivered: true }; }, + sendChannelsOutgoingChunk: () => {}, + sendChannelsOutgoingComplete: () => {}, + sendChannelsOpen: () => new Promise((resolve) => { answerOpen = resolve; openSent(); }), + sendChannelsClose: async () => ({}), + }; + const registry = new ChannelRegistry( + { getServer: () => mockServer } as unknown as McplServerRegistry, + {} as FeatureSetManager, + () => {}, + () => {}, + { onRouteFailure: (info) => { failures.push(info.reason); } }, + ); + await registry.handleChanged('body', { + added: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'bidirectional' }], + } as never); + const delivery = registry.routeSpeech('linn', 'hello', 'body:ears'); + await opening; + answerOpen({ channel: { id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'bidirectional' } }); + void registry.handleChanged('body', { + updated: [{ id: 'body:ears', type: 'chat', label: 'Body · ears', direction: 'inbound' }], + } as never); + + assert.equal(await delivery, null); + assert.deepEqual(publishCalls, [], 'nothing was published into the channel declared inbound-only'); + assert.match(failures[0]!, /^inbound-only: server "body" declared/); +}); + +test("an open by reconcile applies its answer too: a channel the answer declares inbound-only is refused (R1-3)", async () => { + const { registry, publishCalls, openCalls, failures, setOpenResult } = makeRegistry({ delivered: true }); + const ears = { id: 'body:ears', type: 'chat', label: 'Body · ears' }; + await registry.handleRegister('body', { channels: [ears] } as never); + const opened = await registry.handleChannelToolCall('channel_open', { channelId: 'body:ears' }, { kind: 'agent', agentName: 'linn' }); + assert.equal(opened.success, true); + // The server registers again, as after a reconnect, and reconcile reopens + // the channel the agent opened: this time the answer declares it. + setOpenResult({ channel: { ...ears, direction: 'inbound' } }); + await registry.handleRegister('body', { channels: [ears] } as never); + assert.equal(openCalls.length, 2, 'reconcile opened it again'); + + assert.equal(await registry.routeSpeech('linn', 'hello', 'body:ears'), null); + assert.equal(publishCalls.length, 0); + assert.match(failures[0]!.reason, /^inbound-only: server "body" declared/); +}); + +// A registry with body:home registered, counting the server's opens. An open +// raced by a close or by a re-registration is the lifecycle queue's +// (channel-lifecycle-order.test.ts). +async function withHome() { + const made = makeRegistry({ delivered: true }); + const home = { id: 'body:home', type: 'chat', label: 'Body · home', direction: 'bidirectional' }; + await made.registry.handleRegister('body', { channels: [home] } as never); + let opens = 0; + made.setOnOpen(async () => { opens++; }); + return { ...made, home, opens: () => opens }; +} + +test("a reconcile's open answered while the channel is tuned out is recorded: a tune-out keeps the transport open", async () => { + const { registry, home, opens } = await withHome(); + assert.equal(registry.isChannelOpen('body:home'), false); + (registry as unknown as { setDesiredState(s: string, c: string, state: string, source: string): void }) + .setDesiredState('body', 'body:home', 'tuned-out', 'agent-tool'); + await registry.handleChanged('body', { added: [home] } as never); + assert.equal(opens(), 1, "the announcement's reconcile opened it"); + assert.equal(registry.isChannelOpen('body:home'), true); +}); diff --git a/test/explicit-prose-routing.test.ts b/test/explicit-prose-routing.test.ts index 4b01e0a78..262fb958f 100644 --- a/test/explicit-prose-routing.test.ts +++ b/test/explicit-prose-routing.test.ts @@ -93,8 +93,14 @@ function stubRegistry(framework: AgentFramework, plan: Array<'delivered' | 'fals : { status: 'failed' as const, destination, reason: 'stub reported delivered:false', at: Date.now() }; }; const explicit: Record = { - resolveProseTarget: (spec: string) => - known[spec] ?? { error: 'no channel matches "' + spec + '"' }, + // As the registry does, a channel its server declared inbound-only + // resolves for reading but not as a target for speech. + resolveProseTarget: (spec: string, opts?: { forSpeech?: boolean }) => + spec === '#ears' + ? opts?.forSpeech + ? { error: 'body:ears is inbound-only: server "body" declared this channel\'s direction "inbound", so the host publishes nothing into it; to speak there, use that server\'s own tools', refusedChannelId: 'body:ears' } + : { channelId: 'body:ears', label: 'Body · ears' } + : known[spec] ?? { error: 'no channel matches "' + spec + '"' }, deliverSpeech, routeSpeech: async (agent: string, text: string, locus?: string | null) => { const outcome = await deliverSpeech(agent, text, locus); @@ -1702,6 +1708,11 @@ describe('explicit prose routing', () => { ['>>#nowhere bounced'], ['>>skip_reply kept\nand unprefixed'], ['unprefixed', '>>#alpha second segment'], + // A target refused for speech clears the destination (R1-1), and a + // later one that resolves sets it again. + ['>>#ears for the room'], + ['>>#alpha first\n>>#ears for the room'], + ['>>#ears for the room', '>>#alpha after'], ]; try { for (const segments of rounds) { @@ -1785,6 +1796,28 @@ describe('explicit prose routing', () => { } }); + it('the stream neither sends nor keeps text for a channel `speakable` refuses, and keeps what it routes there once `speakable` accepts it', () => { + let earsSpeakable = false; + const router = new ProseStreamRouter({ + mode: 'explicit', + initialTarget: null, + resolve: (spec) => (spec === '#ears' ? 'body:ears' : spec === '#alpha' ? 'chan-alpha' : null), + speakable: (channelId) => channelId !== 'body:ears' || earsSpeakable, + }); + const streamed = [ + ...router.feed('>>#ears refused words'), ...router.endSegment(), + ...router.feed('>>#alpha for alpha'), ...router.endSegment(), + ]; + earsSpeakable = true; + streamed.push(...router.feed('>>#ears spoken words'), ...router.finish()); + assert.deepEqual(streamed, [ + { channelId: 'chan-alpha', delta: 'for alpha' }, + { channelId: 'body:ears', delta: 'spoken words' }, + ]); + assert.deepEqual([...router.byChannel()], [['chan-alpha', 'for alpha'], ['body:ears', 'spoken words']], + 'outgoing/complete keeps only what was streamed'); + }); + it('hybrid: a hold beside a send writes the routing and drafts of its own round, not the live turn\'s', async () => { const framework = await createFramework('hybrid'); stubRegistry(framework); @@ -1809,6 +1842,25 @@ describe('explicit prose routing', () => { } }); + it('hybrid: a hold beside a send refuses a >>> target its server declared inbound-only, as delivery does, and drops the pin before it', async () => { + const framework = await createFramework('hybrid'); + stubRegistry(framework); + const gateway = framework as unknown as { + holdSilencedProse(agent: unknown, segments: string[], cause: 'send', privateThink: boolean, round: { routing: RoutingState; notice: 'later' }): void; + }; + const agent = framework.getAgent('assistant')!; + try { + // An earlier round of the turn pinned #alpha. + const routing: RoutingState = routingState({ pin: 'chan-alpha' }); + gateway.holdSilencedProse(agent, ['>>>#ears for the room', 'after it'], 'send', false, { routing, notice: 'later' }); + assert.equal(routing.pin, undefined, 'the inbound-only channel is never pinned, and the earlier pin is dropped'); + assert.equal(routing.hybridHold, 'failed', 'the envelope fails as a refused target does at delivery'); + assert.deepEqual(routing.drafts.map((d) => d.text), ['for the room', 'after it'], 'and the words are held as drafts'); + } finally { + await framework.stop(); + } + }); + it('hybrid: after >>>skip_reply, the next unprefixed envelope stays in the context and counts as kept private', async () => { const framework = await createFramework('hybrid'); const routed = stubRegistry(framework); @@ -1989,6 +2041,24 @@ describe('explicit prose routing', () => { await framework.stop(); }); + it('the hybrid and locus primers say plain text goes to the locus unless its server declared that channel inbound-only', async () => { + const framework = await createFramework('hybrid'); + const agent = framework.getAgent('assistant')!; + const prime = (framework as unknown as { maybePrimeProseMode(agent: unknown): void }).maybePrimeProseMode.bind(framework); + prime(agent); + // The agent's mode is set from its config; a later config change is what + // brings the locus primer. + (agent as unknown as { proseRouting: string }).proseRouting = 'locus'; + prime(agent); + const primers = agent.getContextManager().getAllMessages().flatMap(m => m.content) + .filter(b => b.type === 'text').map(b => (b as { text: string }).text) + .filter(t => t.startsWith('[prose-routing]')); + assert.equal(primers.length, 2, String(primers)); + assert.match(primers[0]!, /^\[prose-routing\] Hybrid routing enabled: unprefixed text still lands in the current conversational locus, unless its server declared that channel inbound-only\. /); + assert.match(primers[1]!, /^\[prose-routing\] Mode change: your plain text auto-routes to the conversational locus again, unless its server declared that channel inbound-only\. /); + await framework.stop(); + }); + it('prose_help tool returns the routing reference on demand', async () => { membrane.pushResponse(createMockResponse([ { type: 'tool_use', id: 'c1', name: 'prose_help', input: {} }, @@ -2005,8 +2075,30 @@ describe('explicit prose routing', () => { const stream = membrane.lastStream!; assert.equal(stream.receivedToolResults.length, 1); - const result = JSON.stringify(stream.receivedToolResults[0]); - assert.ok(result.includes('destination'), 'help text served as tool result'); + const strings: string[] = []; + JSON.stringify(stream.receivedToolResults[0], (_key, value) => { + if (typeof value === 'string') strings.push(value); + return value; + }); + // The help is the agent's only full reference for a destination's rule, + // so each of its clauses is pinned, read across the text's line breaks. + // The result's content is the help as a JSON string, so it's decoded first. + const decoded = strings.map((s) => { + try { + const value: unknown = JSON.parse(s); + return typeof value === 'string' ? value : s; + } catch { + return s; + } + }); + const help = decoded.join(' ').replace(/\s+/g, ' '); + for (const clause of [ + 'A destination applies to the rest of that turn, until another is named.', + 'A name that matches no channel, or several, bounces and leaves it as it was.', + 'A channel its server declared inbound-only is refused, and clears it.', + ]) { + assert.ok(help.includes(clause), `the help says: ${clause}`); + } assert.deepEqual(routed, [{ text: 'got it', locus: 'chan-alpha' }]); await framework.stop(); @@ -2048,6 +2140,271 @@ describe('explicit prose routing', () => { await framework.stop(); }); + for (const [mode, authored] of [ + ['hybrid', '>>>#ears\nHello, room.'], + ['explicit', '>>#ears Hello, room.'], + ] as const) { + it(`${mode}: a target its server declared inbound-only bounces with that reason and publishes nothing`, async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: authored }] as ContentBlock[])); + const framework = await createFramework(mode); + const routed = stubRegistry(framework); + feedTurn(framework, 2, [{ type: 'text', text: (mode === 'hybrid' ? '>>>' : '>>') + 'skip_reply {{unsent}}' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + assert.deepEqual(routed, [], 'nothing was published'); + const texts = framework.getAgent('assistant')!.getContextManager().getAllMessages().flatMap(m => m.content) + .filter(b => b.type === 'text').map(b => (b as { text: string }).text); + const bounce = texts.find(t => t.startsWith('[prose-routing]') && t.includes('not delivered')); + assert.ok(bounce, 'the text bounced'); + assert.ok(bounce!.includes('body:ears is inbound-only: server "body" declared'), `the bounce gives the declaration: ${bounce}`); + assert.ok(!bounce!.includes('was not confirmed'), `the bounce doesn't call it an unconfirmed delivery: ${bounce}`); + await framework.stop(); + }); + } + + it("explicit: a target refused for speech clears the turn's destination, so later unprefixed text bounces rather than going to the earlier one", async () => { + // Anarchid's probe (agent-framework#350, R1-1): the turn names #alpha, + // then #ears, which its server declared inbound-only, then writes on + // without a prefix. That text was meant for the room, not #alpha. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#alpha first' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#ears for the room' }, + { type: 'tool_use', id: 'c2', name: 'robot--move', input: { dir: 'south' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'and more for the room' }, + ] as ContentBlock[])); + const framework = await createFramework('explicit'); + const routed = stubRegistry(framework); + // The wakes the bounces ask for set the retained text aside. + feedTurn(framework, 2, [{ type: 'text', text: '>>skip_reply' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + assert.deepEqual(routed, [{ text: 'first', locus: 'chan-alpha' }], 'only the text before the refusal was published'); + const bounces = framework.getAgent('assistant')!.getContextManager().getAllMessages().flatMap(m => m.content) + .filter(b => b.type === 'text').map(b => (b as { text: string }).text) + .filter(t => t.startsWith('[prose-routing]') && t.includes('not delivered')); + assert.ok(bounces.some(t => t.includes('body:ears is inbound-only')), `the refused target bounced with its reason: ${bounces.join(' | ')}`); + assert.ok(bounces.some(t => t.includes('this turn has no destination now')), `the later unprefixed text bounced: ${bounces.join(' | ')}`); + await framework.stop(); + }); + + it("explicit: text routed to a channel its server declared inbound-only isn't streamed there, or sent at the stream's end", async () => { + // The round's delivery refuses #ears and bounces its text. The stream + // took it as it was written, and its end would send it should the server + // lift the declaration before the stream ends. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#ears for the room' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#alpha done' }, + ] as ContentBlock[])); + const framework = await createFramework('explicit'); + stubRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + registry.inboundOnlyServer = (channelId: string) => (channelId === 'body:ears' ? 'body' : undefined); + const chunks: Array<{ channelId: string; delta: string }> = []; + const completes: Array<{ channelId: string; text: string }> = []; + registry.sendOutgoingChunk = (channelId: string, _agent: string, _inference: string, _index: number, delta: string) => { + chunks.push({ channelId, delta }); + }; + registry.sendOutgoingComplete = (channelId: string, _agent: string, _inference: string, text: string) => { + completes.push({ channelId, text }); + }; + feedTurn(framework, 2, [{ type: 'text', text: '>>skip_reply' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + assert.deepEqual(chunks.filter((c) => c.channelId === 'body:ears'), [], 'nothing is streamed to the inbound-only channel'); + assert.deepEqual(completes, [{ channelId: 'chan-alpha', text: 'done' }], "the stream's end sends only what the stream routed and kept"); + await framework.stop(); + }); + + for (const refusal of ['inbound-only', 'inbound-only, unconfirmed', 'another failure', 'another failure, unconfirmed', 'delivered, then declared'] as const) { + const inboundOnly = refusal.startsWith('inbound-only'); + it(`explicit: a delivery that ${refusal === 'delivered, then declared' ? 'is delivered while the channel turns inbound-only' : `fails (${refusal})`} ${inboundOnly ? "clears the turn's destination, as a refusal at admission does" : 'keeps the destination, as before'}`, async () => { + // #alpha resolves when the round is admitted. By its delivery, its + // server has declared it inbound-only (as a channels/changed, or the + // answer to opening it, can), or the publish fails some other way. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#alpha first' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'and more' }, + ] as ContentBlock[])); + const framework = await createFramework('explicit'); + const routed = stubRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const declared = new Set(); + registry.inboundOnlyServer = (channelId: string) => (declared.has(channelId) ? 'body' : undefined); + // The turn's routing, as its first delivery sees it. + let turn: { pin?: string; destination: string | null } | undefined; + registry.deliverSpeech = async (_agent: string, text: string, locus?: string | null) => { + turn ??= (framework as unknown as { turnRouting: Map }).turnRouting.get('assistant'); + routed.push({ text, locus: locus ?? null }); + if (inboundOnly || refusal === 'delivered, then declared') declared.add(locus ?? ''); + const destination = { serverId: 'stub', channelId: locus ?? '' }; + if (refusal === 'delivered, then declared') return { status: 'delivered', destination, at: Date.now() }; + // A publish that may have been posted, as one that timed out may. + if (refusal.endsWith('unconfirmed')) return { status: 'unknown', destination, reason: 'channels/publish timed out (delivery uncertain)', at: Date.now() }; + return { status: 'failed', destination, reason: inboundOnly ? 'inbound-only: server "body" declared it' : 'stub refused', at: Date.now() }; + }; + feedTurn(framework, 2, [{ type: 'text', text: '>>skip_reply' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + const bounces = context(framework).texts.filter((t) => t.startsWith('[prose-routing]') && t.includes('not delivered')); + if (inboundOnly) { + assert.deepEqual(routed, [{ text: 'first', locus: 'chan-alpha' }], 'the later text never went to the refused channel'); + assert.ok(bounces.some((t) => t.includes('this turn has no destination now')), `the later text bounced: ${bounces.join(' | ')}`); + // Both halves of the destination end: where delivery sends later + // text, and where a later stream of the turn would start. + assert.equal(turn?.pin, undefined); + assert.equal(turn?.destination, null); + } else { + assert.deepEqual(routed, [{ text: 'first', locus: 'chan-alpha' }, { text: 'and more', locus: 'chan-alpha' }]); + // Each failed delivery bounces its own words; none bounces for want of a destination. + assert.equal(bounces.filter((t) => t.includes('delivery to chan-alpha failed')).length, refusal === 'another failure' ? 2 : 0); + assert.equal(bounces.some((t) => t.includes('this turn has no destination now')), false); + assert.equal(turn?.pin, 'chan-alpha'); + assert.equal(turn?.destination, 'chan-alpha'); + } + await framework.stop(); + }); + } + + it('explicit: a refusal at delivery ends a destination that names the refused channel even when the pin has moved off it', async () => { + // Each half ends while it still names the refused channel, on its own. + // Here another delivery (as a replaced stream's queued one can) moves the + // pin to #cafe during #alpha's delivery, while the destination still + // names #alpha: the pin stays, and the destination ends. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#alpha first' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'and more' }, + ] as ContentBlock[])); + const framework = await createFramework('explicit'); + const routed = stubRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const declared = new Set(); + registry.inboundOnlyServer = (channelId: string) => (declared.has(channelId) ? 'body' : undefined); + let turn: { pin?: string; destination: string | null } | undefined; + let atFailure: { pin?: string; destination: string | null } | undefined; + registry.deliverSpeech = async (_agent: string, text: string, locus?: string | null) => { + turn ??= (framework as unknown as { turnRouting: Map }).turnRouting.get('assistant'); + routed.push({ text, locus: locus ?? null }); + const destination = { serverId: 'stub', channelId: locus ?? '' }; + if (locus === 'chan-alpha') { + turn!.pin = 'discord:g:cafe'; + atFailure = { pin: turn!.pin, destination: turn!.destination }; + declared.add('chan-alpha'); + return { status: 'failed', destination, reason: 'inbound-only: server "body" declared it', at: Date.now() }; + } + return { status: 'delivered', destination, at: Date.now() }; + }; + feedTurn(framework, 2, [{ type: 'text', text: '>>skip_reply' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + assert.deepEqual(atFailure, { pin: 'discord:g:cafe', destination: 'chan-alpha' }, 'the state the clear met'); + assert.equal(turn?.pin, 'discord:g:cafe', 'the moved pin stays'); + assert.equal(turn?.destination, null, 'the destination that named the refused channel ends'); + assert.deepEqual(routed, [{ text: 'first', locus: 'chan-alpha' }, { text: 'and more', locus: 'discord:g:cafe' }]); + await framework.stop(); + }); + + it("explicit: a delivery refused after a later round named another channel leaves that round's destination", async () => { + // Greta-2425's probe. Deliveries queue behind each other, and the turn + // goes on without waiting, so round 2's >>#cafe is admitted, setting the + // destination, before round 1's delivery to #alpha fails. The clear ends + // only a destination that still names #alpha. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#alpha first' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#cafe second' }, + { type: 'tool_use', id: 'c2', name: 'robot--move', input: { dir: 'south' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'and more' }, + ] as ContentBlock[])); + const framework = await createFramework('explicit'); + const routed = stubRegistry(framework); + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + const declared = new Set(); + registry.inboundOnlyServer = (channelId: string) => (declared.has(channelId) ? 'body' : undefined); + const turnOf = () => (framework as unknown as { turnRouting: Map }).turnRouting.get('assistant'); + let turn: { pin?: string; destination: string | null } | undefined; + let destinationAtFailure: unknown = 'never reached'; + registry.deliverSpeech = async (_agent: string, text: string, locus?: string | null) => { + turn ??= turnOf(); + routed.push({ text, locus: locus ?? null }); + const destination = { serverId: 'stub', channelId: locus ?? '' }; + if (locus === 'chan-alpha') { + // Hold #alpha's delivery until round 2 is admitted, then refuse it: + // its server has declared it inbound-only by now. + const deadline = Date.now() + 3000; + while (turn?.destination !== 'discord:g:cafe' && Date.now() < deadline) await new Promise((r) => setTimeout(r, 5)); + destinationAtFailure = turn?.destination ?? null; + declared.add('chan-alpha'); + return { status: 'failed', destination, reason: 'inbound-only: server "body" declared it', at: Date.now() }; + } + return { status: 'delivered', destination, at: Date.now() }; + }; + feedTurn(framework, 2, [{ type: 'text', text: '>>skip_reply' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + assert.equal(destinationAtFailure, 'discord:g:cafe', 'round 2 was admitted before #alpha refused'); + assert.deepEqual(routed, [ + { text: 'first', locus: 'chan-alpha' }, + { text: 'second', locus: 'discord:g:cafe' }, + { text: 'and more', locus: 'discord:g:cafe' }, + ]); + // Delivery and a later stream of the turn agree on #cafe. + assert.equal(turn?.pin, 'discord:g:cafe'); + assert.equal(turn?.destination, 'discord:g:cafe'); + await framework.stop(); + }); + + for (const [final, kept] of [['>>#ears ! for the room', false], ['>>#cafe ! for the room', true]] as const) { + it(`explicit: a turn's final text (${kept ? 'its target kept' : 'its target refused'}) reaches nothing of its ' !' continuation, which starts a turn of its own`, async () => { + // The turn's final text is delivered after the turn settles, so neither + // its pin nor the destination a later stream would start at carries + // into the continuation: that's a new turn, with no destination. + membrane.pushResponse(createMockResponse([ + { type: 'text', text: '>>#alpha first' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: final }, + ] as ContentBlock[])); + const framework = await createFramework('explicit'); + const routed = stubRegistry(framework); + const stream = captureStream(framework); + feedTurn(framework, 2, [{ type: 'text', text: 'and more for the room' }] as ContentBlock[]); + feedTurn(framework, 3, [{ type: 'text', text: '>>skip_reply' }] as ContentBlock[]); + feedTurn(framework, 4, [{ type: 'text', text: '>>skip_reply' }] as ContentBlock[]); + trigger(framework); + await framework.runUntilIdle(); + // The first turn makes two calls, its tool round and its final text. + assert.ok(membrane.calls.length >= 3, `the continuation ran (${membrane.calls.length} calls)`); + assert.deepEqual(routed, kept + ? [{ text: 'first', locus: 'chan-alpha' }, { text: 'for the room', locus: 'discord:g:cafe' }] + : [{ text: 'first', locus: 'chan-alpha' }]); + const continuation = stream.chunks.filter((c) => c.delta.includes('and more')).map((c) => c.channelId); + assert.deepEqual(continuation, [], 'the continuation streamed nothing anywhere'); + assert.equal(stream.completes.some((c) => c.text.includes('and more')), false); + const bounces = context(framework).texts.filter((t) => t.startsWith('[prose-routing]') && t.includes('not delivered')); + assert.ok(bounces.some((t) => t.includes('this turn has no destination now')), `the continuation's text bounced: ${bounces.join(' | ')}`); + await framework.stop(); + }); + } + it('hybrid leaves unprefixed prose on the frozen locus', async () => { membrane.pushResponse(createMockResponse([{ type: 'text', text: 'Ordinary field speech.' }] as ContentBlock[])); const framework = await createFramework('hybrid'); diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index 7e5d7c8ad..7e4c6becb 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -1561,6 +1561,7 @@ describe('present while acting', () => { framework.agents = new Map([['assistant', { proseRouting: 'locus' }]]); framework.channelRegistry = { handleChannelToolCall: () => new Promise((resolve) => { resolveOpen = resolve; }), + inboundOnlyServer: () => undefined, }; framework.turnRouting = new Map([['assistant', calling]]); framework.openLogical = new Map([['assistant', calling]]); @@ -1600,6 +1601,7 @@ describe('present while acting', () => { framework.agents = new Map([['assistant', { proseRouting: 'locus' }]]); framework.channelRegistry = { handleChannelToolCall: async () => ({ success: true, data: { channelId: 'discord:guild:observatory' } }), + inboundOnlyServer: () => undefined, }; framework.turnRouting = new Map([['assistant', calling]]); if (!silent) framework.openLogical = new Map([['assistant', calling]]); @@ -1827,6 +1829,117 @@ describe('present while acting', () => { await framework.stop(); }); + // ------------------------------------------------------------------------- + // A channel its server declared inbound-only takes no plain speech (the + // registry refuses it), so every notice that says where plain speech goes + // says that instead. pocket-body's live transcript is one: another agent + // speaking in the room arrives there addressed. + // ------------------------------------------------------------------------- + + /** The stub registry answers that body:ears is inbound-only, declared by pocket-body. */ + function declareEarsInboundOnly(framework: AgentFramework): Record { + const registry = (framework as unknown as { channelRegistry: Record }).channelRegistry; + registry.inboundOnlyServer = (channelId: string) => (channelId === 'body:ears' ? 'pocket-body' : undefined); + return registry; + } + + function assertSaysRefused(text: string | undefined, what: string): void { + assert.ok(text, `${what} was written`); + assert.ok( + text!.includes( + 'pocket-body declared this channel inbound-only, so plain speech there is refused and stays in your archive. ' + + "To speak there, use pocket-body's own tools.", + ), + `${what} says plain speech there is refused, where it stays, and what to use instead: ${text}`, + ); + assert.ok(!text!.includes('lands'), `${what} doesn't say the speech lands there: ${text}`); + } + + it('the routing notice for a locus its server declared inbound-only says plain speech there is refused', async () => { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'one' }] as ContentBlock[])); + + const framework = await createFramework(); + stubChannelRegistry(framework); + declareEarsInboundOnly(framework); + + trigger(framework); + await framework.runUntilIdle(); + + (framework as unknown as { announceLocusIfChanged(agentName: string, locus: string | null): void }) + .announceLocusIfChanged('assistant', 'body:ears'); + + const cm = (framework as unknown as { + agents: Map }> } }>; + }).agents.get('assistant')!.getContextManager(); + const notices = cm.getAllMessages() + .flatMap((m) => m.content) + .map((b) => b.text ?? '') + .filter((t) => t.startsWith('[routing]')); + const notice = notices.at(-1); + assert.ok(notice?.startsWith('[routing] The conversation is now in body:ears.'), String(notice)); + assertSaysRefused(notice, 'the routing notice'); + + await framework.stop(); + }); + + it('an addressed arrival on an inbound-only channel re-pins there, and its notice says plain speech there is refused', async () => { + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Station four, proceeding.' }, + { type: 'tool_use', id: 'c1', name: 'robot--move', input: { dir: 'north' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Answering the room.' }, + ] as ContentBlock[])); + + const framework = await createFramework(); + const routed = stubChannelRegistry(framework); + declareEarsInboundOnly(framework); + module.toolDelayMs = 25; + module.interjection = 'Linn, can you hear me?'; + module.interjectionMetadata = { channelId: 'body:ears', tags: ['chat:addressed'] }; + + trigger(framework); + await framework.runUntilIdle(); + + // The locus follows the conversation as before; the registry, not the + // pin, is what keeps the speech out of the channel. + assert.deepEqual(routed, [ + { text: 'Station four, proceeding.', locus: 'chan-live-1' }, + { text: 'Answering the room.', locus: 'body:ears' }, + ]); + const wired = membrane.lastStream!.receivedToolResultOptions + .flatMap((o) => o?.injectedMessages ?? []) + .flatMap((m) => m.content as Array<{ type: string; text?: string }>) + .map((b) => b.text ?? ''); + const notice = wired.find((t) => t.startsWith('[routing]')); + assert.ok(notice?.startsWith('[routing] The conversation moved to body:ears.'), String(notice)); + assertSaysRefused(notice, 'the re-pin notice'); + + await framework.stop(); + }); + + it('channel_open of an inbound-only channel says in its result that plain speech there is refused', async () => { + membrane.pushResponse(createMockResponse([ + { type: 'tool_use', id: 'c1', name: 'channel_open', input: { channelId: 'body:ears' } }, + ] as ContentBlock[], 'tool_use')); + membrane.pushResponse(createMockResponse([ + { type: 'text', text: 'Listening.' }, + ] as ContentBlock[])); + + const framework = await createFramework(); + stubChannelRegistry(framework); + const registry = declareEarsInboundOnly(framework); + registry.handleChannelToolCall = async () => ({ success: true, data: { channelId: 'body:ears', opened: true } }); + + trigger(framework); + await framework.runUntilIdle(); + + const results = JSON.stringify(membrane.lastStream!.receivedToolResults.flat()); + assertSaysRefused(results, 'the channel_open result'); + + await framework.stop(); + }); + it('reactions and system markers injected mid-turn do not clear send suppression', async () => { // A reaction (`chat:reaction` tag) or a `system: true` marker is not // conversational input: prose following an explicit send stays suppressed, diff --git a/test/trunk-channel-routing.test.ts b/test/trunk-channel-routing.test.ts index 2da99d131..26fc8fa3d 100644 --- a/test/trunk-channel-routing.test.ts +++ b/test/trunk-channel-routing.test.ts @@ -162,6 +162,7 @@ describe('Trunk channel routing (item-3 redux)', () => { isChannelOpen: () => false, getChannelLabel: () => undefined, getDescriptor: () => ({ capabilities: { history: { maxMessages: 80 } } }), + inboundOnlyServer: () => undefined, stopAll: () => {}, }; @@ -220,6 +221,7 @@ describe('Trunk channel routing (item-3 redux)', () => { isChannelOpen: () => false, getChannelLabel: () => undefined, getDescriptor: () => ({ capabilities: { history: { maxMessages: 80 } } }), + inboundOnlyServer: () => undefined, stopAll: () => {}, }; @@ -287,6 +289,7 @@ describe('Trunk channel routing (item-3 redux)', () => { getChannelLabel: () => undefined, getDescriptor: () => ({ label: 'DM: _reim0n', capabilities: { history: { maxMessages: 80 } } }), proseTargetFor: (id: string, serverId?: string) => { asked.push(`${serverId}/${id}`); return '@_reim0n'; }, + inboundOnlyServer: () => undefined, stopAll: () => {}, }; i.handleMcplPushEvent(dmEvent('m3')); @@ -306,6 +309,7 @@ describe('Trunk channel routing (item-3 redux)', () => { getChannelLabel: () => undefined, getDescriptor: () => ({ label: 'DM: _reim0n', capabilities: { history: { maxMessages: 80 } } }), proseTargetFor: () => undefined, + inboundOnlyServer: () => undefined, stopAll: () => {}, }; i2.handleMcplPushEvent(dmEvent('m4')); @@ -323,6 +327,7 @@ describe('Trunk channel routing (item-3 redux)', () => { isChannelOpen: () => false, getChannelLabel: () => undefined, getDescriptor: () => undefined, + inboundOnlyServer: () => undefined, stopAll: () => {}, }; i3.handleMcplPushEvent(dmEvent('m5')); @@ -330,6 +335,48 @@ describe('Trunk channel routing (item-3 redux)', () => { await fw3.stop(); }); + it('the invitation offers no reply into a channel its server declared inbound-only, and says why', async () => { + // pocket-body's live transcript: another agent speaking in the room arrives + // on it addressed, and the host publishes nothing into it, so a ">>" reply would + // only be refused. + const framework = await makeFramework(); + const i = internals(framework); + i.channelRegistry = { + ensureChannelRegistered: () => {}, + isChannelOpen: () => false, + getDescriptor: () => ({ label: 'Body · ears', direction: 'inbound' }), + getChannelLabel: () => 'Body · ears', + proseTargetFor: () => '#ears', + inboundOnlyServer: (id: string) => (id === 'body:ears' ? 'pocket-body' : undefined), + stopAll: () => {}, + }; + i.handleMcplPushEvent({ + type: 'mcpl:push-event', + serverId: 'pocket-body', + featureSet: 'body', + eventId: 'ears_1', + content: [{ type: 'text', text: 'Fable: Linn, can you hear me?' }], + origin: { source: 'pocket-body', messageId: 'e1', mcplChannelId: 'body:ears', channelName: 'Body · ears' }, + tags: ['chat:addressed'], + timestamp: new Date().toISOString(), + inferenceId: 'i-e1', + triggerInference: false, + }); + + const text = framework.getAgent('scout')!.getContextManager().queryMessages({}).messages.at(-1)?.content + .filter((block): block is { type: 'text'; text: string } => block.type === 'text') + .map((block) => block.text) + .join('\n') ?? ''; + assert.match(text, /Channel invitation/); + assert.match( + text, + /1\. Replying there isn't available: pocket-body declared this channel inbound-only, so nothing you write is posted into it\. To speak there, use pocket-body's own tools\./, + ); + assert.doesNotMatch(text, /prefixed with ">>/); + assert.match(text, /channel_open with channelId "body:ears"/, 'opening it, to hear the room, stays available'); + await framework.stop(); + }); + it('the invitation adds a # only where the channel\'s name lacks one, and names an unlabelled channel by its id', async () => { // #406's run: slack-mcpl registers "#general (acme)" and its push carries no // channelName, so the invitation named the place "##general (acme)". @@ -346,6 +393,7 @@ describe('Trunk channel routing (item-3 redux)', () => { isChannelOpen: () => false, getDescriptor: (id: string) => ({ label: labels[id] }), getChannelLabel: (_serverId: string, id: string) => labels[id], + inboundOnlyServer: () => undefined, stopAll: () => {}, }; const textOf = () =>