diff --git a/changelog.d/failed-speech-open-puts-back.fixed.md b/changelog.d/failed-speech-open-puts-back.fixed.md new file mode 100644 index 000000000..84c4b2ab0 --- /dev/null +++ b/changelog.d/failed-speech-open-puts-back.fixed.md @@ -0,0 +1,47 @@ +- When an open that speech or a reply made fails, the channel's decision goes + back to what it was. So an agent told the channel is closed and the open + failed isn't later put into it by that open's decision. Before, the + decision to open stayed: the agent's next speech there, or reconciliation + when the server announced the channel again, opened it without the notice + that says it's open and how to close it. A decision made while the open + was out stands. Since the failed request may have reached the server, as + when its answer is lost to a timeout, the host also closes the channel on + the server. Where the decision put back means nobody had decided, the + channel then follows its server's latest announcement and the + subscription policy, which a re-announcement during the open couldn't + apply: if they select it for opening, the host tries the open again, and + the failure says why. Where the host can't record putting the decision + back, the decision stays, no close is sent, and the failure says the + channel stays selected for opening. `channel_open`'s own decision still + outlives a failed open, as its failure says. +- A reply into a closed channel whose open fails now tells the agent that + the reply went out and opening the channel failed, and that + `channel_open` tries again, where the agent can call it: a conversation + fork can't, and an agent's `allowedTools` can leave it out. The notice + comes where the host directs the server's lifecycle + (`channels.lifecycle`), and only when no decision about the channel, such + as the agent closing it, was made while the open was out. A reply whose + open only reconfirms a channel the agent already has open gets none. + Where the channel is selected for opening after the failure, the notice + says so and why, as speech's failure does, and that `channel_close` + withdraws that, where the agent can call it: where the host couldn't + record putting the channel's decision back, in `channel_open`'s own + failure's words, a later delivery or reconnect tries again; where the + server's latest announcement or the policy selected it again, the host + tries again itself. Where the channel is selected for opening and the + replying agent has ended, the primary is told by name instead, if one is + registered; where the replying agent can't close the channel, as a fork + away from its home can't, the primary is told by name too, if it can. + Before, the failure reached only the logs. +- On a server without `channels.lifecycle`, an open by speech or a reply, + which fails there before any request is sent, no longer records a decision. +- `channel_open` on a channel that's already open, and `channel_close` on one + that's already closed, now record the agent's decision, and so does + `channel_decline` on a channel closed by default. So a module's + housekeeping close is refused for a channel the agent opened itself, even + where speech, a reply, the server or a policy had opened it first. And a + channel the agent closed or declined isn't opened later because its server + registers it as initially open, or a policy admits it. +- `channels/close` now has the deadline `channels/open` has when + `requestTimeoutMs` is 0. A close the server never answers no longer holds, + for good, the deliveries that wait for the channel's lifecycle work. diff --git a/src/framework.ts b/src/framework.ts index e81191947..47ae0b67d 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -102,7 +102,7 @@ import { isSilentHeartbeatMarker, isVisiblyEmptyContent, NON_STRING_TEXT_PLACEHO import { parseProsePrefix, parseHybridProsePrefix } from './mcpl/prose-grammar.js'; import { detectKnownToolWrapperProse } from './tool-wrapper-prose-guard.js'; import { InferenceRouter } from './mcpl/inference-router.js'; -import { ChannelRegistry, connectorText, destinationKey, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome } from './mcpl/channel-registry.js'; +import { ChannelRegistry, connectorText, destinationKey, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome, type SelectedForOpening } from './mcpl/channel-registry.js'; import { KEEP_IDLE_MS, KEEP_NEWEST, ProseDraftStore, draftState, uncertainAttempt, type Draft, type DraftAttempt, type DraftReason, type DraftState, type InheritedRisk } from './prose-drafts.js'; import { ConversationRouter, CONVERSATION_ROUTING_DEPRECATION_NOTICE } from './mcpl/conversation-router.js'; import { INBOUND_SOURCE_KEY, type InboundAcceptanceObserver, type InboundChannelSource, type InboundLane, type InboundSource, type InboundUnscopedSource } from './mcpl/inbound-source.js'; @@ -15045,6 +15045,114 @@ export class AgentFramework { } } + /** + * Window notice for a reply whose channel couldn't be opened: the reply + * went out through its send tool, but the open that engaging the channel + * asks for failed. Without it the agent would wait for a conversation it + * won't see. It goes to the agent whose reply it was, or, when no agent + * is named, where addMessage delivers an unrouted message (deliveryAgent: + * the primary, else the first registered agent), as the opened notice + * does, and its advice is judged for that agent. Where the channel stays + * closed, an agent that has ended is told nothing, and nor is anyone in + * its place: nothing follows from the failure. Where it's selected for + * opening, the primary is told by name in its place, if one is registered, + * as the opened notice tells it of an open by an agent that has ended, so + * an agent that could close the channel hears it may open. It names + * channel_open as the way to try again only to an agent that can call it: + * a conversation fork's channel_open is refused outright + * (dispatchChannelToolCall), and an agent's allowedTools can leave it out. + * Where the channel is selected for opening, it says so and why, as + * speech's failure does, and what that means: where the host couldn't + * record putting the channel's decision back, in channel_open's own + * failure's words, a later delivery or reconnect tries again; where the + * server's latest announcement or the subscription policy lifted the + * default put back, the host tries again itself, so channel_open isn't + * named. Either way channel_close withdraws that, named only to an agent + * that can call it here (a fork may close only its home channel). Where + * the reader can't, the primary is told by name too, if it's registered, + * isn't the reader and can close it. System message, as the opened notice + * is. + */ + private recordReplyOpenFailedNotice( + agentName: string | undefined, + channel: { channelId: string; label?: string }, + selectedForOpening?: SelectedForOpening, + ): void { + const shown = channel.label && channel.label !== channel.channelId + ? `${channel.label} (${channel.channelId})` + : channel.channelId; + const ended = !!agentName && !this.agents.has(agentName); + if (ended && !selectedForOpening) { + console.error(`[channel] ${agentName} has ended; not telling it that ${shown} couldn't be opened`); + return; + } + const primary = this.primaryAgentName ?? undefined; + if (ended && (!primary || !this.agents.has(primary))) { + console.error(`[channel] ${agentName} has ended, and there is no primary to tell that ${shown} couldn't be opened and is selected for opening`); + return; + } + const who = ended ? primary : agentName || this.deliveryAgent()?.name; + const canOpen = who !== undefined && !this.conversationAgentHomes.has(who) && this.canAgentUseTool(who, 'channel_open'); + // A fork's channel_close is refused for any channel but its home, as + // dispatchChannelToolCall reads it. + const home = who !== undefined ? this.conversationAgentHomes.get(who) : undefined; + const canClose = who !== undefined && (!home || home === channel.channelId) && + this.canAgentUseTool(who, 'channel_close'); + const selected = selectedForOpening === 'unrecorded' + ? " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again" + : selectedForOpening !== undefined + ? ` It's selected for opening again, as ${selectedForOpening === 'server-bootstrap' ? 'its server announces it open' : 'the subscription policy admits it'}: the host tries again` + : ''; + const hostRetries = selectedForOpening === 'server-bootstrap' || selectedForOpening === 'subscription-policy'; + const text = `[channels] Not open: ${shown}. ` + + (ended + ? `${agentName}'s reply there went out, but opening the channel failed; ${agentName} has ended.` + : 'Your reply there went out, but opening the channel failed.') + + (selected ? selected + (canClose ? ', and channel_close withdraws that.' : '.') : '') + + (canOpen && !hostRetries ? ' channel_open tries again.' : ''); + try { + this.addMessage( + 'user', + [{ type: 'text', text }], + (ended + ? { system: true, kind: 'channel-notice', speaker: agentName } + : { system: true, kind: 'channel-notice' }) as MessageMetadata, + agentName && !ended ? { forAgent: agentName } : undefined, + ); + console.error( + `[channel] reply open-failed notice -> ${ended ? `the primary, by name (${agentName} has ended)` : who ?? 'no agent'}: ${shown}`, + ); + } catch (err) { + console.error('recordReplyOpenFailedNotice failed:', err); + } + // Where the reader can't close a channel selected for opening, as a fork + // away from its home can't, or an agent whose allowedTools leave + // channel_close out, the primary is told by name too, where it's + // registered and can close it, so an agent that could close the channel + // hears it may open. + const primaryHome = primary !== undefined ? this.conversationAgentHomes.get(primary) : undefined; + if ( + !ended && selectedForOpening !== undefined && who !== undefined && !canClose && + primary !== undefined && primary !== who && this.agents.has(primary) && + (!primaryHome || primaryHome === channel.channelId) && this.canAgentUseTool(primary, 'channel_close') + ) { + const primaryCanOpen = !this.conversationAgentHomes.has(primary) && this.canAgentUseTool(primary, 'channel_open'); + const told = `[channels] Not open: ${shown}. ${who}'s reply there went out, but opening the channel failed, and ${who} can't close it.` + + selected + ', and channel_close withdraws that.' + + (primaryCanOpen && !hostRetries ? ' channel_open tries again.' : ''); + try { + this.addMessage( + 'user', + [{ type: 'text', text: told }], + { system: true, kind: 'channel-notice', speaker: who } as MessageMetadata, + ); + console.error(`[channel] reply open-failed notice -> the primary, by name (${who} can't close it): ${shown}`); + } catch (err) { + console.error('recordReplyOpenFailedNotice failed:', err); + } + } + } + /** * A `!` continuation the turn asked for (`>>#x !`, `>>>#x !`): start another * turn now instead of pausing until the next external event. It reads the @@ -25688,12 +25796,13 @@ export class AgentFramework { } } - // INVARIANT: engaging a channel opens it. A successful explicit send - // into a channel the registry knows as closed drives the same open - // flow as channel_open (durable desired state, the server's - // channels/open), so there is no such thing as posting into a channel - // that stays half-alive (no typing, no reactions, no inbound - // forwarding). + // Engaging a channel opens it. A successful explicit send into a + // channel the registry knows as closed drives the open flow + // channel_open uses (the server's channels/open), so posting there + // doesn't leave the channel half-alive (no typing, no reactions, no + // inbound forwarding). If that open fails, the channel stays as it + // was, and where the host directs the server's lifecycle the agent + // is told, with how to try again. // Fire-and-forget: the send already succeeded; the open must not // delay or fail the tool result. if ( @@ -25705,9 +25814,14 @@ export class AgentFramework { extractChannelIdFromToolResult(result.content) ?? (typeof args.channelId === 'string' ? (args.channelId as string) : undefined); if (target) { + // Whether the host directs the server's lifecycle is read before + // the open, not after its failure: a transport that drops while the + // open is out clears the grant before the failure arrives, and the + // notice below is for that failure too. + const directs = this.channelRegistry.directsLifecycle(serverId); this.channelRegistry .openIfClosedForSend(target, serverId) - .then(({ status, channelId, label }) => { + .then(({ status, channelId, label, opening, selectedForOpening }) => { // Turn-scoped engagement record (mid-turn re-pin, 2026-07-31 // n=7): a human follow-up in a channel the agent explicitly // sent into this turn counts as conversationally addressed. @@ -25732,10 +25846,20 @@ export class AgentFramework { 'opened-by-reply', ); } else if (status === 'open-failed') { + // A reply's open that would have opened a channel not + // selected open, and failed, is told to the agent, with how + // to try again, where the host directs the server's + // lifecycle. The opened notice would have told it of the + // open. A failed reconfirm of a channel already selected + // open changes nothing the agent was told. Without + // channels.lifecycle the server opens and closes its own + // channels, and channel_open can't open one either. + const tell = opening === true && directs; console.error( - `[channel] ${agentName}: ${toolName} delivered to ${target} but the open-on-send failed; ` + - 'a later registration or delivery tries the open again', + `[channel] ${agentName}: ${toolName} delivered to ${target} but the open-on-send failed` + + (tell ? '; the agent is told' : ''), ); + if (tell) this.recordReplyOpenFailedNotice(agentName, { channelId: channelId ?? target, label }, selectedForOpening); } }) .catch((err) => console.error('[channel] open-on-send error:', err)); diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index f23e367a0..6840609d5 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -60,8 +60,33 @@ class ChannelLifecycleConvergenceError extends Error {} class ChannelOpenSubstitutionError extends Error {} /** An operation queued for a server's registrations that were removed. */ class ChannelLifecycleCancelledError extends Error {} +/** The failure of an open by speech or a reply after a decision made during + * its round-trip, such as the agent's own close or channel_open, replaced + * the one the open wrote. That decision stands, and nothing is put back. */ +class ChannelOpenOvertakenError extends Error {} +/** The failure of an open by speech or a reply whose put-back couldn't be + * recorded: the decision to open stays, so the channel stays selected for + * opening, and reconciliation may open it later. */ +class ChannelOpenPutBackUnrecordedError extends Error {} +/** The failure of an open by speech or a reply whose put-back restored a + * pure default that the channel's current descriptor, or the subscription + * policy, lifts: the channel is selected for opening again, and `by` says + * which. */ +class ChannelOpenSelectedAgainError extends Error { + constructor(message: string, readonly by: Exclude, options?: ErrorOptions) { + super(message, options); + } +} +/** Why a channel is selected for opening after a failed open by speech or a + * reply: its put-back couldn't be recorded, or the server's latest + * announcement or the subscription policy lifted the default it restored. */ +export type SelectedForOpening = 'unrecorded' | 'server-bootstrap' | 'subscription-policy'; const CHANNEL_LIFECYCLE_LOG_ID = 'mcpl/channel-lifecycle'; +/** The sources of a decision the agent itself made: channel_open and + * channel_close, and channel_decline. */ +const AGENT_DECISION_SOURCES: ReadonlySet = new Set(['agent-tool', 'invitation-declined']); + /** * What a channels/open answer says about which channel the server opened: * its `channel.id`, or else the `channelId` of a history item. Each is @@ -866,6 +891,10 @@ export class ChannelRegistry { // target before confirming anything. Unlike the tail, which settles a few // microtasks after its operation's last check, this drops at that check. private lifecycleOutstanding = new Map(); + // The removal count each channel's newest outstanding operation was queued + // under. Where it's older than its server's, every operation still + // outstanding for the channel cancels when it would run. + private lifecycleQueuedRemovals = new Map(); // Losing an open receipt does not confirm closure. Keep that distinction // private while preserving the public open:boolean availability projection. private unconfirmedChannelTargets = new WeakSet(); @@ -878,7 +907,7 @@ export class ChannelRegistry { /** Whether the host directs this server's channel lifecycle (§14.1: its * `channels.lifecycle` grant). Where it doesn't, the server opens and * closes its own channels, and sends no receipt the host could wait for. */ - private directsLifecycle(serverId: string): boolean { + directsLifecycle(serverId: string): boolean { return CapabilityGrant.of(this.serverRegistry.getServer(serverId)).has('channels.lifecycle'); } @@ -2254,7 +2283,20 @@ export class ChannelRegistry { source: string, ): void { const key = this.lifecycleKey(serverId, channelId); - if (this.desiredStates.get(key)?.state === desired) return; + const current = this.desiredStates.get(key); + // A decision that records the state already on record changes nothing, + // so it writes nothing, unless it's the agent's own (channel_open, + // channel_close or channel_decline) and the record isn't. The record + // then says the agent decided, which is all handleToolClose's + // explicit-open guard and ensureInitialDesiredState's lift of a default + // ask of it. openChannelNow depends on it too: a failed open by speech + // or a reply tells a decision made during its round-trip by the record + // no longer being the object it wrote, and the agent's channel_open, + // the only decision of the same state that can be made then, gets a new + // record here. + if (current?.state === desired && !(AGENT_DECISION_SOURCES.has(source) && !AGENT_DECISION_SOURCES.has(current.source))) { + return; + } this.appendLifecycleEvent({ kind: 'desired-state', serverId, @@ -2536,6 +2578,7 @@ export class ChannelRegistry { const removals = this.serverRemovals.get(serverId) ?? 0; const previous = this.channelLifecycleTails.get(lifecycleKey) ?? Promise.resolve(); this.lifecycleOutstanding.set(lifecycleKey, (this.lifecycleOutstanding.get(lifecycleKey) ?? 0) + 1); + this.lifecycleQueuedRemovals.set(lifecycleKey, removals); const operation = previous.catch(() => {}).then( () => this.runLifecycleOperation(serverId, channelId, removals, options), ); @@ -2681,10 +2724,25 @@ export class ChannelRegistry { // on, nothing it does will open a new target. const left = (this.lifecycleOutstanding.get(lifecycleKey) ?? 1) - 1; if (left > 0) this.lifecycleOutstanding.set(lifecycleKey, left); - else this.lifecycleOutstanding.delete(lifecycleKey); + else { + this.lifecycleOutstanding.delete(lifecycleKey); + this.lifecycleQueuedRemovals.delete(lifecycleKey); + } } } + /** Whether lifecycle work already outstanding for the channel will carry out + * its current decision: some is, and the newest was queued since its + * server's registrations were last removed, so it won't cancel when it + * would run. Each operation reads the decision when it runs. It answers for + * now: a removal after it cancels that work too, and takes the channel's + * entry with it, so its next registration carries the decision out, where + * this host directs the lifecycle then. */ + private outstandingWorkCarriesOut(serverId: string, key: string): boolean { + return this.lifecycleOutstanding.has(key) && + this.lifecycleQueuedRemovals.get(key) === (this.serverRemovals.get(serverId) ?? 0); + } + /** * A server that answered an open with another channel really opened that * one. When it's registered here and the host wants it closed, close it @@ -2697,6 +2755,19 @@ export class ChannelRegistry { private closeSubstitutedChannel(serverId: string, channelId: string): void { if (!this.channels.has(`${serverId}:${channelId}`)) return; if (this.getDesiredState(serverId, channelId) !== 'closed') return; + this.applyThroughQueue(serverId, channelId); + } + + /** + * Carry out a channel's decision through its lifecycle queue, for a + * transport the server may have opened without the host's knowing: one it + * opened for another channel's request (closeSubstitutedChannel), or one a + * failed open's request may have reached (openChannelNow). The decision is + * a closed one, except after a failed open whose put-back the server's + * latest announcement or the policy lifted. A failure is reported as a + * trace; the decision stands. + */ + private applyThroughQueue(serverId: string, channelId: string): void { this.applyDesiredChannelState(serverId, channelId).catch((err) => { // Removal while queued is cancellation, as in reconcileChannels, and // exhausted convergence emitted its own diagnostic. @@ -2705,7 +2776,7 @@ export class ChannelRegistry { err instanceof ChannelLifecycleConvergenceError) return; this.emitTraceFn({ type: 'mcpl:channel-reconcile-failed', - serverId, channelId, desired: 'closed', error: (err as Error).message, + serverId, channelId, desired: this.getDesiredState(serverId, channelId), error: (err as Error).message, }); }); } @@ -2955,10 +3026,25 @@ export class ChannelRegistry { * 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). - * Desired state is recorded BEFORE the server round-trip: intent sticks - * even if the subscribe fails, and reconciliation retries later. - * Automatic delivery needs open transport, not a change of attention: - * preserve tune-out epochs both before and during the round-trip. + * Desired state is recorded BEFORE the server round-trip. The agent's own + * open is a decision about the channel: it sticks even if the subscribe + * fails, and reconciliation retries later. An open by speech or a reply is + * incidental to its words: when its round-trip fails, the decision it + * wrote goes back to the one it replaced, so the agent, told the channel + * is closed and the open failed, isn't later put into it by that decision. + * That holds only while the record is still the one it wrote: a decision + * made meanwhile stands, and the failure then comes as a + * ChannelOpenOvertakenError. Where the decision put back is a pure default, + * nobody had decided, so the channel then follows its server's latest + * announcement and the policy: a re-announcement during the round-trip + * found the open's own record and lifted nothing. If they select it for + * opening, the failure says so (ChannelOpenSelectedAgainError). And since + * a failed request may have reached the server, the decision that stands + * is carried out on the wire too. An open by speech or a reply that can't + * send its request, on a server without channels.lifecycle, writes + * nothing. Automatic delivery needs open transport, not a change of + * attention: preserve tune-out epochs both before and during the + * round-trip. * * `opened` says whether this call turned a closed channel open, which is * what the delivery and reply notices tell the resident. Reconfirming the @@ -2970,17 +3056,98 @@ export class ChannelRegistry { history?: ChannelHistoryRequest, ): Promise<{ result: ChannelsOpenResult; opened: boolean }> { const transportOnly = source !== 'agent-tool'; + const key = this.lifecycleKey(entry.serverId, entry.descriptor.id); + if (transportOnly) { + // Without channels.lifecycle the open fails before any request leaves + // (§14.1), as it does without the server, so an incidental open would + // only write a decision to put back. It writes none. + const server = this.serverRegistry.getServer(entry.serverId); + if (!server) throw new Error(`Server not found: ${entry.serverId}`); + if (!CapabilityGrant.of(server).has('channels.lifecycle')) { + throw new Error(`channels.lifecycle not in "${entry.serverId}"'s effective grant (§14.1)`); + } + } + // The decision this open replaces. A channel with none on record, as one + // registered from a push, was closed by default: nobody had decided. + const replaced = this.desiredStates.get(key) ?? { state: 'closed' as const, source: 'default-closed' }; const before = this.getDesiredState(entry.serverId, entry.descriptor.id); const opened = before !== 'open' && before !== 'tuned-out'; if (!transportOnly || before !== 'tuned-out') { this.setDesiredState(entry.serverId, entry.descriptor.id, 'open', source); } - const applied = await this.applyDesiredChannelState(entry.serverId, entry.descriptor.id, { history }); - const desired = this.getDesiredState(entry.serverId, entry.descriptor.id); - if (!applied.open || (desired !== 'open' && !(transportOnly && desired === 'tuned-out'))) { - throw new Error('Channel open was superseded by a newer lifecycle decision'); + // The record an incidental open wrote, to put back if its round-trip fails. + const wrote = transportOnly && opened ? this.desiredStates.get(key) : undefined; + try { + const applied = await this.applyDesiredChannelState(entry.serverId, entry.descriptor.id, { history }); + const desired = this.getDesiredState(entry.serverId, entry.descriptor.id); + if (!applied.open || (desired !== 'open' && !(transportOnly && desired === 'tuned-out'))) { + throw new Error('Channel open was superseded by a newer lifecycle decision'); + } + return { result: applied.result!, opened }; + } catch (err) { + if (wrote && this.desiredStates.get(key) !== wrote) { + throw new ChannelOpenOvertakenError(err instanceof Error ? err.message : String(err), { cause: err }); + } + if (wrote) { + try { + this.setDesiredState(entry.serverId, entry.descriptor.id, replaced.state, replaced.source); + } catch (putBackErr) { + // The caller reports both: the channel isn't open, but it stays + // selected for opening, so reconciliation may open it later. + throw new ChannelOpenPutBackUnrecordedError( + `${err instanceof Error ? err.message : String(err)}; the channel stays selected for opening, ` + + `as the host couldn't record undoing that: ${putBackErr instanceof Error ? putBackErr.message : String(putBackErr)}`, + { cause: err }, + ); + } + // The decision put back is a closed one: an open that turned the + // channel open replaced no open decision. Where it's a pure default, + // nobody had decided, so the channel follows its server's latest + // announcement and the policy, not the ones from before the open: a + // re-announcement during the round-trip found the open's own record + // and lifted nothing. The lift is ensureInitialDesiredState's own, + // which leaves every real decision alone. One that can't be recorded + // leaves the default, closed. + const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); + let selectedAgain: Exclude | undefined; + if (current) { + try { + this.ensureInitialDesiredState(entry.serverId, current.descriptor); + } catch (liftErr) { + console.error( + `[channel] ${entry.serverId}:${logValue(entry.descriptor.id)}: the lift after a failed open couldn't be recorded:`, + liftErr, + ); + } + const now = this.desiredStates.get(key); + if (now?.state === 'open' && (now.source === 'server-bootstrap' || now.source === 'subscription-policy')) { + selectedAgain = now.source; + } + } + // The failed request may have reached the server, so the decision + // that stands is carried out through the queue, as a decline's close + // is, unless other lifecycle work for the channel is queued that will + // carry it out anyway: work queued before its server's registrations + // were removed cancels instead. (An incidental open gets here only + // where the host directed the lifecycle when it opened: see the grant + // check above. Where the grant has lost channels.lifecycle since, as a + // transport drop clears it, the queued work fails for it, and the + // server's next registration carries the decision out, where this host + // directs the lifecycle then.) + if (current && !this.outstandingWorkCarriesOut(entry.serverId, key)) { + this.applyThroughQueue(entry.serverId, entry.descriptor.id); + } + if (selectedAgain) { + throw new ChannelOpenSelectedAgainError( + `${err instanceof Error ? err.message : String(err)}; the channel is selected for opening again, as ` + + (selectedAgain === 'server-bootstrap' ? 'its server announces it open' : 'the subscription policy admits it'), + selectedAgain, + { cause: err }, + ); + } + } + throw err; } - return { result: applied.result!, opened }; } /** @@ -3213,7 +3380,17 @@ export class ChannelRegistry { * engaging a channel opens it, so typing indicators, reaction machinery, * and inbound forwarding all come alive with the first outbound message. * Accepts composite MCPL ids or raw server-internal ids (explicit send - * tools receive whatever the agent typed). + * tools receive whatever the agent typed). An `open-failed` answer says + * whether the open that failed was one that would have opened a channel + * not selected open (`opening`), as opposed to one reconfirming the + * transport of a channel already selected open or tuned out. It isn't one + * either where a decision made during its round-trip replaced it + * (ChannelOpenOvertakenError), or where queued lifecycle work left the + * channel closed before any open was sent. `selectedForOpening` says the + * channel is selected for opening after the failure, and why: the put-back + * couldn't be recorded (ChannelOpenPutBackUnrecordedError), or the server's + * latest announcement or the policy lifted the default it restored + * (ChannelOpenSelectedAgainError). */ async openIfClosedForSend( rawChannelId: string, @@ -3222,6 +3399,8 @@ export class ChannelRegistry { status: 'opened' | 'already-open' | 'unknown-channel' | 'ambiguous' | 'open-failed'; channelId?: string; label?: string; + opening?: boolean; + selectedForOpening?: SelectedForOpening; }> { let matches = [...this.channels.values()].filter( (e) => e.descriptor.id === rawChannelId && (!serverId || e.serverId === serverId), @@ -3236,6 +3415,7 @@ export class ChannelRegistry { if (matches.length === 0) return { status: 'unknown-channel' }; if (matches.length > 1) return { status: 'ambiguous' }; let entry = matches[0]!; + let opening = false; try { if (this.hasPendingLifecycle(entry)) { entry = await this.waitForOpenChannel(entry); @@ -3244,6 +3424,7 @@ export class ChannelRegistry { if (entry.open && (desired === 'open' || desired === 'tuned-out')) { return { status: 'already-open', channelId: entry.descriptor.id, label: entry.descriptor.label }; } + opening = desired !== 'open' && desired !== 'tuned-out'; const { opened } = await this.openChannelNow(entry, 'opened-by-reply'); entry = await this.waitForOpenChannel(entry); if (opened) { @@ -3255,6 +3436,10 @@ export class ChannelRegistry { } return { status: opened ? 'opened' : 'already-open', channelId: entry.descriptor.id, label: entry.descriptor.label }; } catch (err) { + // Where a decision made during the open's round-trip replaced it, such + // as the agent's own close or channel_open, the channel is as that + // decision left it, and the agent isn't told the open failed. + if (err instanceof ChannelOpenOvertakenError) opening = false; this.emitTraceFn({ type: 'mcpl:channel-open-failed', serverId: entry.serverId, @@ -3262,7 +3447,11 @@ export class ChannelRegistry { error: (err as Error).message, }); const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); - return { status: 'open-failed', channelId: entry.descriptor.id, + return { status: 'open-failed', channelId: entry.descriptor.id, opening, + // Where the channel is selected for opening after the failure, the + // agent is told so and why, as speech's failure says. + ...(err instanceof ChannelOpenPutBackUnrecordedError ? { selectedForOpening: 'unrecorded' as const } : {}), + ...(err instanceof ChannelOpenSelectedAgainError ? { selectedForOpening: err.by } : {}), ...(current ? { label: current.descriptor.label } : {}) }; } } @@ -3303,6 +3492,19 @@ export class ChannelRegistry { const alreadyDesiredOpen = this.getDesiredState(entry.serverId, input.channelId) === 'open'; if (entry.open && alreadyDesiredOpen && !input.backscroll && !this.hasPendingLifecycle(entry)) { + // Open already, perhaps because speech, a reply, the server or a + // policy opened it: the agent's open is still its decision, and is + // recorded as one (setDesiredState), so the explicit-open guard of + // handleToolClose reads it. + try { + this.setDesiredState(entry.serverId, input.channelId, 'open', 'agent-tool'); + } catch (err) { + return { + success: false, + error: `Channel ${input.channelId} is already open, but recording your decision to open it failed: ${(err as Error).message}`, + isError: true, + }; + } return { success: true, data: { channelId: input.channelId, status: 'already open' }, @@ -3373,15 +3575,6 @@ export class ChannelRegistry { }; } - const alreadyDesiredClosed = this.getDesiredState(entry.serverId, input.channelId) === 'closed'; - if (!entry.open && alreadyDesiredClosed && !this.hasPendingLifecycle(entry) && - !this.unconfirmedChannelTargets.has(entry)) { - return { - success: true, - data: { channelId: input.channelId, status: 'already closed' }, - }; - } - // Provenance comes from trusted dispatch, not from self-description: // only module-origin calls may record a machine source, and only from // the closed vocabulary. Agent-origin (or origin-less — the safe @@ -3402,6 +3595,27 @@ export class ChannelRegistry { closeSource = input.source; } + const alreadyDesiredClosed = this.getDesiredState(entry.serverId, input.channelId) === 'closed'; + if (!entry.open && alreadyDesiredClosed && !this.hasPendingLifecycle(entry) && + !this.unconfirmedChannelTargets.has(entry)) { + // Closed already, perhaps only by default: the agent's close is still + // its decision, and is recorded as one (setDesiredState), so a later + // registration's lift of a default leaves the channel closed. + try { + this.setDesiredState(entry.serverId, input.channelId, 'closed', closeSource); + } catch (err) { + return { + success: false, + error: `Channel ${input.channelId} is already closed, but recording the decision to close it failed: ${(err as Error).message}`, + isError: true, + }; + } + return { + success: true, + data: { channelId: input.channelId, status: 'already closed' }, + }; + } + // Housekeeping must not override stated intent (issue #5: GC closes wore // the agent's badge and reset explicitly-opened doors). A machine-sourced // close of a channel whose current desired state is an agent/operator diff --git a/src/mcpl/server-connection.ts b/src/mcpl/server-connection.ts index 059f10779..a3a802c45 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -1534,9 +1534,13 @@ export class McplServerConnection extends EventEmitter { return this.sendRequest(McplMethod.ChannelsOpen, params as unknown as Record, { timeoutMs }) as Promise; } - /** Send `channels/close` request and await result. */ + /** Send `channels/close` request and await result. It keeps a deadline as + * channels/open does: a delivery waits for a channel's queued lifecycle + * work, closes included, so a close that is never answered would hold the + * delivery, and the turn that waits for it, for good. */ sendChannelsClose(params: ChannelsCloseParams): Promise { - return this.sendRequest(McplMethod.ChannelsClose, params as unknown as Record) as Promise; + const timeoutMs = this.requestTimeoutMs > 0 ? this.requestTimeoutMs : this.publishDeadlineMs; + return this.sendRequest(McplMethod.ChannelsClose, params as unknown as Record, { timeoutMs }) as Promise; } /** Send `channels/acknowledge` request and await result. */ diff --git a/src/mcpl/types.ts b/src/mcpl/types.ts index 977fce5df..f443c19c7 100644 --- a/src/mcpl/types.ts +++ b/src/mcpl/types.ts @@ -430,9 +430,10 @@ export interface McplServerConfig { * are also the only bound for one that closes its stdin, since Bun emits no * 'error' event for the writes to it. requestTimeoutMs is the timeout of * every request that names none; the handshake's initialize has its own. - * Set 0 to disable, except for channels/publish and the channels/open a - * delivery may send first, which keep the 60 s default: a publish's outcome - * is reported to the resident and recorded for its draft, so it must end. + * Set 0 to disable, except for channels/publish, channels/open and + * channels/close, which keep the 60 s default: a publish's outcome is + * reported to the resident and recorded for its draft, so it must end, and + * a delivery waits for a channel's queued opens and closes. * Default: 60000 (60 seconds). A value above 2^31−1 is refused: Node runs a * longer timer after about 1 ms, so every request would time out at once. * diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index bab5fbe42..81b624266 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -2,7 +2,7 @@ import { test } from 'node:test'; import { mkdtempSync, rmSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import type { JsStore } from '@animalabs/chronicle'; +import { JsStore } from '@animalabs/chronicle'; import { AgentFramework } from '../src/index.js'; import { MockMembrane } from './helpers/mock-membrane.js'; import assert from 'node:assert/strict'; @@ -1443,10 +1443,13 @@ test('declining an invitation closes an unconfirmed transport for real, and wait test("a settled operation releases its channel's queue key", async () => { const f = fixture(); await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); - const internals = f.registry as unknown as { channelLifecycleTails: Map; lifecycleOutstanding: Map }; + const internals = f.registry as unknown as { + channelLifecycleTails: Map; lifecycleOutstanding: Map; lifecycleQueuedRemovals: Map; + }; for (let i = 0; i < 3; i++) await tick(); assert.equal(internals.channelLifecycleTails.size, 0); assert.equal(internals.lifecycleOutstanding.size, 0); + assert.equal(internals.lifecycleQueuedRemovals.size, 0, 'the removal count kept for the queue goes with it'); }); test('the notices say what happened, name the opt-out, and state no rule about later sends', async () => { @@ -1575,3 +1578,948 @@ test('a channel its server registers while a disconnect closes it is removed wit rmSync(dir, { recursive: true, force: true }); } }); + +// An open by speech or a reply is incidental to its words. When its +// round-trip fails, the decision it wrote goes back to the one it replaced, +// while the record is still its own. The agent's own channel_open keeps its +// decision through a failed round-trip (#210). + +/** The decision on record for a channel, with its source. */ +function decision(registry: ChannelRegistry, id = 'x') { + const internals = registry as unknown as { + desiredStates: Map; + lifecycleKey(serverId: string, channelId: string): string; + }; + const record = internals.desiredStates.get(internals.lifecycleKey('test', id)); + return record && { state: record.state, source: record.source }; +} + +/** The lifecycle records the registry appends from now on, with their state and source. */ +function recordLog(registry: ChannelRegistry) { + const log: Array<{ desired?: string; source?: string }> = []; + const r = registry as unknown as { appendLifecycleEvent(e: { desired?: string; source?: string }): void }; + const original = r.appendLifecycleEvent.bind(registry); + r.appendLifecycleEvent = (e) => { original(e); log.push({ desired: e.desired, source: e.source }); }; + return log; +} + +/** Makes the registry's next lifecycle append, or the one `skip` appends later, fail as a full store would. */ +function failAppend(registry: ChannelRegistry, skip = 0) { + const r = registry as unknown as { appendLifecycleEvent(e: unknown): void }; + const original = r.appendLifecycleEvent.bind(registry); + let left = skip; + r.appendLifecycleEvent = (e) => { + if (left-- === 0) { + r.appendLifecycleEvent = original; + throw new Error('ENOSPC: no space left on device'); + } + original(e); + }; +} + +/** Starts speech into `x` while the server holds its open, and fails that open. */ +async function speechWhoseOpenFails(f: ReturnType, others: () => Promise[] = () => []) { + f.held.add('x'); + const speech = f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' }); + await tick(); + const meanwhile = others(); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + await f.drain(speech, ...meanwhile); + return { outcome: await speech, meanwhile }; +} + +async function replyWhoseOpenFails(f: ReturnType, others: () => Promise[] = () => []) { + f.held.add('x'); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + const meanwhile = others(); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + await f.drain(reply, ...meanwhile); + return { answer: await reply, meanwhile }; +} + +test("speech whose open fails puts back the decision it replaced, and the next speech's open is a fresh one with its notice", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); + + const { outcome } = await speechWhoseOpenFails(f); + assert.equal(outcome.status, 'failed'); + assert.match(outcome.reason ?? '', /^channel is closed and open failed/); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }, 'what the agent was told stays true'); + assert.equal(f.notices.length, 0); + + const again = await f.registry.publish('agent', 'more words', { serverId: 'test', channelId: 'x' }); + assert.equal(again.status, 'delivered'); + assert.equal(f.isOpen(), true); + assert.deepEqual(f.notices.map((n) => n.source), ['opened-by-delivery'], 'the open that succeeds tells the agent the channel is open'); +}); + +test('after speech whose open failed, the server announcing the channel again opens nothing', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await speechWhoseOpenFails(f); + const opens = f.calls.filter((c) => c.kind === 'open').length; + + const before = f.calls.length; + await f.drain(f.registry.handleChanged('test', { removed: ['x'], added: [descriptor('x', false)] })); + assert.deepEqual(f.calls.slice(before).map((c) => c.kind), ['close'], "the registration's reconcile ran, and carried out the closed decision"); + assert.equal(f.calls.filter((c) => c.kind === 'open').length, opens, 'no open is sent'); + assert.equal(f.isOpen(), false); + assert.deepEqual(f.notices, []); +}); + +test("the agent's close made while speech's open is out stands when that open fails", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const { outcome } = await speechWhoseOpenFails(f, () => [f.tool('close')]); + assert.equal(outcome.status, 'failed'); + assert.equal(outcome.reason, 'channel is closed and open failed: fixture lifecycle failure', "the open's own failure, carried by ChannelOpenOvertakenError"); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); +}); + +test("the agent's channel_open made while speech's open is out stands when that open fails", async () => { + // channel_open records the state speech's open already recorded. It still + // gets its own record, so the failed open sees a decision made meanwhile + // and puts nothing back. + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const { meanwhile } = await speechWhoseOpenFails(f, () => [f.tool('open')]); + const opened = (await meanwhile[0]) as { success: boolean }; + assert.equal(opened.success, true, "the agent's open goes through"); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'agent-tool' }); + assert.equal(f.isOpen(), true); +}); + +test("a reply whose open fails puts back the agent's close", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await f.drain(f.tool('close')); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); + + f.held.add('x'); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + assert.deepEqual(await reply, { status: 'open-failed', channelId: 'x', label: 'x', opening: true }, 'an open that would have opened the channel'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }, "the agent's close is on record again"); +}); + +test("a reply's open that the agent's own close replaced during its round-trip isn't an opening that failed", async () => { + // The agent replies and leaves in one turn: the open-on-send is + // fire-and-forget, so its channel_close can land while the reply's open is + // out. The open's operation then carries out the close (Cordelia-2434's + // PROBE 8). + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + f.held.add('x'); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + assert.equal(f.pending.length, 1, "the reply's open is out when the agent's close lands"); + const closed = f.tool('close'); + await tick(); + f.held.delete('x'); + await f.drain(reply, closed); + assert.equal(((await closed) as { success: boolean }).success, true); + assert.deepEqual(await reply, { status: 'open-failed', channelId: 'x', label: 'x', opening: false }, 'nothing to tell an agent that just closed it'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }, "the agent's close stands"); + assert.equal(f.isOpen(), false); +}); + +test("a reply's failed open after the agent's channel_open during its round-trip isn't an opening that failed", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const { answer, meanwhile } = await replyWhoseOpenFails(f, () => [f.tool('open')]); + assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: false }, "the agent's own open answers for itself"); + assert.deepEqual(f.traces.filter((t) => t.type === 'mcpl:channel-open-failed').map((t) => t.error), ['fixture lifecycle failure'], + "the open's own failure, carried by ChannelOpenOvertakenError"); + assert.equal(((await meanwhile[0]) as { success: boolean }).success, true, "the agent's open goes through"); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'agent-tool' }); + assert.equal(f.isOpen(), true); +}); + +test("a reply's failed open whose put-back can't be recorded is still an opening that failed, and says the channel stays selected", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + failAppend(f.registry, 1); // the open's own record goes through; the put-back's doesn't + const closes = f.calls.filter((c) => c.kind === 'close').length; + const { answer } = await replyWhoseOpenFails(f); + assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true, selectedForOpening: 'unrecorded' }); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'opened-by-reply' }, 'the channel stays selected for opening'); + await tick(); await tick(); + assert.equal(f.calls.filter((c) => c.kind === 'close').length, closes, 'no close is sent: the decision stays open'); +}); + +test("a channel registered from a push, with no decision on record, is put back as closed by default when speech's open fails", async () => { + const f = fixture(); + f.registry.ensureChannelRegistered('test', 'x', 'x'); + assert.equal(decision(f.registry), undefined, 'nobody has decided'); + const { outcome } = await speechWhoseOpenFails(f); + assert.equal(outcome.status, 'failed'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); +}); + +test("the put-back is durable: a registry rebuilt from the store doesn't hold the failed open's decision", async () => { + const dir = mkdtempSync(join(tmpdir(), 'failed-speech-open-')); + const store = JsStore.openOrCreate({ path: join(dir, 'store') }); + try { + const f = fixture(undefined, store); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await speechWhoseOpenFails(f); + + const rebuilt = fixture(undefined, store); + assert.deepEqual(decision(rebuilt.registry), { state: 'closed', source: 'default-closed' }); + } finally { + store.close(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('when speech\'s open fails while the server announces the channel again, the queued reconcile carries out the put-back: one close, no notice', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + f.held.add('x'); + const speech = f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' }); + await tick(); + assert.equal(f.pending.length, 1, "speech's open is out"); + const reannounce = f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); + const before = f.calls.length; + f.pending.shift()!.fail(); + f.held.delete('x'); + await f.drain(speech, reannounce); + assert.equal((await speech).status, 'failed'); + assert.deepEqual(f.calls.slice(before).map((c) => c.kind), ['close'], 'the queued reconcile closes x, and nothing else does'); + assert.equal(f.isOpen(), false); + assert.deepEqual(f.notices, []); +}); + +test("when an open by speech or a reply fails while the server announces the channel again as initially open, the default put back follows that announcement: the channel is selected for opening again, the queued work opens it, and the failure says why", async () => { + for (const via of ['speech', 'reply'] as const) { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const reannounce = () => [f.registry.handleChanged('test', { added: [descriptor('x', true)] })]; + if (via === 'speech') { + const { outcome } = await speechWhoseOpenFails(f, reannounce); + assert.equal(outcome.status, 'failed'); + assert.equal(outcome.reason, + 'channel is closed and open failed: fixture lifecycle failure; the channel is selected for opening again, as its server announces it open'); + } else { + const { answer } = await replyWhoseOpenFails(f, reannounce); + assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true, selectedForOpening: 'server-bootstrap' }); + } + await tick(); await tick(); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'server-bootstrap' }, + `${via}: the announcement the open's own record hid applies to the default put back`); + assert.deepEqual(f.calls.map((c) => c.kind), ['close', 'open', 'open'], + `${via}: the registration's close, the open that failed, then the queued work's open; nothing closes x`); + assert.equal(f.isOpen(), true, via); + assert.deepEqual(f.notices, [], `${via}: the failure itself says why, so no other notice`); + } +}); + +test("when a reply's open fails after the subscription policy came to admit the channel, the default put back follows the policy, and the failure says so", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const { answer } = await replyWhoseOpenFails(f, () => { f.registry.setSubscriptionPolicy('test', 'auto'); return []; }); + assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true, selectedForOpening: 'subscription-policy' }); + await tick(); await tick(); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'subscription-policy' }); + assert.equal(f.isOpen(), true); + assert.deepEqual(f.notices, [], 'the failure says it, so no separate admission notice'); +}); + +test("when a reply's open fails while its server is removed and added again as initially open, the default put back follows the new registration", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const { answer, meanwhile } = await replyWhoseOpenFails(f, () => { + f.registry.removeServer('test'); + return [f.registry.handleChanged('test', { added: [descriptor('x', true)] })]; + }); + void meanwhile; + assert.equal(answer.status, 'open-failed'); + assert.equal(answer.selectedForOpening, 'server-bootstrap'); + await tick(); await tick(); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'server-bootstrap' }); + assert.deepEqual(f.calls.slice(f.calls.findIndex((c) => c.kind === 'open') + 1).map((c) => c.kind), ['open'], + "the new registration's queued work carries the decision out, so the put-back queues nothing of its own"); + assert.equal(f.isOpen(), true); +}); + +test("when an open's server is removed and its channel comes back from a push while the open is out, the decision the put-back leaves is still carried out: work queued before the removal cancels instead", async () => { + for (const policy of [true, false]) { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const { answer } = await replyWhoseOpenFails(f, () => { + // A reconcile queued behind the open, then the removal that will cancel it. + const reannounce = f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + f.registry.removeServer('test'); + f.registry.ensureChannelRegistered('test', 'x'); + if (policy) f.registry.setSubscriptionPolicy('test', 'auto'); + return [reannounce]; + }); + await tick(); await tick(); + const after = f.calls.slice(f.calls.findIndex((c) => c.kind === 'open') + 1).map((c) => c.kind); + if (policy) { + assert.equal(answer.selectedForOpening, 'subscription-policy'); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'subscription-policy' }); + assert.deepEqual(after, ['open'], 'the lifted decision is carried out after the cancelled work'); + assert.equal(f.isOpen(), true); + } else { + assert.equal(answer.selectedForOpening, undefined); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); + assert.deepEqual(after, ['close'], "the put-back's close goes out, since the failed request may have reached the server"); + } + } +}); + +test("a lifted put-back's queued open that fails is traced as an open that failed, and the decision stands", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + f.held.add('x'); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + f.registry.setSubscriptionPolicy('test', 'auto'); + f.pending.shift()!.fail(); + for (let i = 0; i < 20 && f.pending.length === 0; i++) await tick(); + assert.equal(f.pending.length, 1, "the queued work's open is out"); + f.pending.shift()!.fail(); + f.held.delete('x'); + await f.drain(reply); + for (let i = 0; i < 10; i++) await tick(); + assert.equal((await reply).selectedForOpening, 'subscription-policy'); + assert.deepEqual(f.traces.filter((t) => t.type === 'mcpl:channel-reconcile-failed').map((t) => t.desired), ['open']); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'subscription-policy' }); +}); + +test("a failed open's put-back whose lift can't be recorded leaves the default, closed, and reports the open's own failure", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + const { outcome } = await speechWhoseOpenFails(f, () => { + failAppend(f.registry, 1); // the put-back's record goes through; the lift's doesn't + return [f.registry.handleChanged('test', { added: [descriptor('x', true)] })]; + }); + assert.equal(outcome.reason, 'channel is closed and open failed: fixture lifecycle failure'); + await tick(); await tick(); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); + assert.deepEqual(f.calls.map((c) => c.kind), ['close', 'open', 'close'], 'the queued work carries out the close'); +}); + +test("a failed open by speech that put back a decision the agent made isn't lifted by the server's announcement: only a pure default follows it", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); await tick(); + // The agent's close fails at the server, so x is decided closed with its + // transport still open, and speech engages it. + f.held.add('x'); + const close = f.tool('close'); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + await f.drain(close); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); + const { outcome } = await speechWhoseOpenFails(f, () => [f.registry.handleChanged('test', { added: [descriptor('x', true)] })]); + assert.equal(outcome.status, 'failed'); + assert.equal(outcome.reason, 'channel is closed and open failed: fixture lifecycle failure', 'nothing selects it for opening again'); + await tick(); await tick(); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }, "the agent's close stands"); +}); + +test('speech that waits on a failed open into a channel the agent closed is refused, as the close is on record again', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); await tick(); + assert.equal(f.actual.get('x'), true); + // The agent's close fails at the server, so x is decided closed with its + // transport still open. + f.held.add('x'); + const close = f.tool('close'); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + await f.drain(close); + assert.equal((await close).success, false); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); + + f.held.add('x'); + const first = f.registry.publish('agent', 'first', { serverId: 'test', channelId: 'x' }); + await tick(); + assert.equal(f.pending.length, 1, "the first speech's open is out"); + const second = f.registry.publish('agent', 'second', { serverId: 'test', channelId: 'x' }); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + await f.drain(first, second); + assert.equal((await first).status, 'failed'); + assert.equal((await second).status, 'failed', 'the second speech waited on that open and is refused'); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing is published into the channel the agent closed'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); +}); + +test("an open whose answer is lost after the server opened the channel is closed on the wire too, since its transport wasn't confirmed", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + f.held.add('x'); + const speech = f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' }); + await tick(); + const open = f.pending.shift()!; + f.actual.set('x', true); // the server opened x ... + open.fail(); // ... and its answer was lost, as on a timeout + f.held.delete('x'); + const before = f.calls.length; + await f.drain(speech); + for (let i = 0; i < 5; i++) await tick(); + assert.equal((await speech).status, 'failed'); + assert.deepEqual(f.calls.slice(before).map((c) => c.kind), ['close'], 'the put-back is carried out through the queue'); + assert.equal(f.actual.get('x'), false, "the server's transport is closed"); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); + assert.deepEqual(f.notices, []); +}); + +test("a reply on a server without channels.lifecycle writes no decision, so the agent's close stays the only record", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); await tick(); + f.setGrant(ALL_CAPABILITY_PATHS.filter((p) => p !== 'channels.lifecycle')); + assert.equal((await f.tool('close')).success, false, 'the close itself is refused for the grant'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }, "the agent's close is recorded"); + await f.registry.handleIncoming('test', { messages: [message()] }); + const log = recordLog(f.registry); + for (let i = 0; i < 3; i++) { + assert.equal((await f.registry.openIfClosedForSend('x', 'test')).status, 'open-failed'); + } + await tick(); + assert.deepEqual(log, [], 'no record is written'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); +}); + +test("speech into a channel that isn't open, on a server without channels.lifecycle, writes no decision", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + f.setGrant(ALL_CAPABILITY_PATHS.filter((p) => p !== 'channels.lifecycle')); + const log = recordLog(f.registry); + const outcome = await f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' }); + await tick(); + assert.equal(outcome.status, 'failed'); + assert.match(outcome.reason ?? '', /^channel is closed and open failed: channels\.lifecycle not in "test"'s effective grant/); + assert.deepEqual(log, [], 'no record is written'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); +}); + +test("the agent's channel_open over a channel its server opened at registration is the agent's decision, which a housekeeping close doesn't override", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); await tick(); + assert.equal(f.isOpen(), true); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'server-bootstrap' }); + + assert.equal(dataOf(await f.tool('open')).status, 'already open'); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'agent-tool' }); + const housekeeping = await f.registry.handleChannelToolCall( + 'channel_close', { serverId: 'test', channelId: 'x', source: 'housekeeping' }, { kind: 'module' }, + ); + assert.equal(housekeeping.success, false, "the agent's explicit open stands against housekeeping"); + assert.equal(f.isOpen(), true); +}); + +test("the agent's channel_close of a channel closed by default is recorded, so a later registration's initiallyOpen doesn't open it", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); + + assert.equal(dataOf(await f.tool('close')).status, 'already closed'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); + await f.drain(f.registry.handleChanged('test', { removed: ['x'], added: [descriptor('x', true)] })); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }, "the agent's close stands"); + assert.equal(f.isOpen(), false); +}); + +test("the agent's decline of a channel closed by default is recorded, so a later registration's initiallyOpen doesn't open it", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); + + const declined = await f.registry.handleChannelToolCall('channel_decline', { serverId: 'test', channelId: 'x', messageId: 'm1' }); + assert.equal(declined.success, true); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'invitation-declined' }); + await f.drain(f.registry.handleChanged('test', { removed: ['x'], added: [descriptor('x', true)] })); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'invitation-declined' }, "the agent's decline stands"); + assert.equal(f.isOpen(), false); +}); + +test('a housekeeping close of a channel closed by default records nothing, and the channel stays undecided', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + const log = recordLog(f.registry); + const housekeeping = await f.registry.handleChannelToolCall( + 'channel_close', { serverId: 'test', channelId: 'x', source: 'housekeeping' }, { kind: 'module' }, + ); + assert.equal(dataOf(housekeeping).status, 'already closed'); + assert.deepEqual(log, []); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); +}); + +test("a module close outside the machine-close vocabulary is refused on a channel already closed too, where it got 'already closed'", async () => { + // The source is checked before the "already closed" answer, since that + // answer now records the decision with it. + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + const log = recordLog(f.registry); + const gremlin = await f.registry.handleChannelToolCall( + 'channel_close', { serverId: 'test', channelId: 'x', source: 'gremlin-module' }, { kind: 'module' }, + ); + assert.equal(gremlin.success, false); + assert.equal(gremlin.isError, true); + assert.match(gremlin.error ?? '', /^Unknown machine close source 'gremlin-module'/); + assert.deepEqual(log, []); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); +}); + +test("the agent's channel_open twice over a channel speech opened writes one record", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + assert.equal((await f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' })).status, 'delivered'); + const log = recordLog(f.registry); + assert.equal(dataOf(await f.tool('open')).status, 'already open'); + assert.equal(dataOf(await f.tool('open')).status, 'already open'); + assert.deepEqual(log, [{ desired: 'open', source: 'agent-tool' }]); +}); + +test("when putting back the decision a failed open replaced can't be recorded, speech's failure says the channel stays selected for opening", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + failAppend(f.registry, 1); // the open's own record goes through; the put-back's doesn't + const closes = f.calls.filter((c) => c.kind === 'close').length; + const { outcome } = await speechWhoseOpenFails(f); + assert.equal(outcome.status, 'failed'); + assert.match(outcome.reason ?? '', /^channel is closed and open failed: fixture lifecycle failure; the channel stays selected for opening, as the host couldn't record undoing that: ENOSPC/); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'opened-by-delivery' }); + await tick(); await tick(); + assert.equal(f.calls.filter((c) => c.kind === 'close').length, closes, 'no close is sent: the decision stays open'); +}); + +test("channel_open and channel_close on a channel already so say what failed when the agent's decision can't be recorded", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + assert.equal((await f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' })).status, 'delivered'); + failAppend(f.registry); + const opened = await f.tool('open'); + assert.equal(opened.success, false); + assert.equal(opened.isError, true); + assert.match(opened.error ?? '', /^Channel x is already open, but recording your decision to open it failed: ENOSPC/); + + const g = fixture(); + await g.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + failAppend(g.registry); + const closed = await g.tool('close'); + assert.equal(closed.success, false); + assert.equal(closed.isError, true); + assert.match(closed.error ?? '', /^Channel x is already closed, but recording the decision to close it failed: ENOSPC/); +}); + +// The framework's half of a reply's failed open: what it tells the agent +// once the registry has answered a send's open. The registry's half, the +// put-back and `opening`, is pinned by the registry tests. The harness is a +// bare framework, as in checkpoint-attribution. +async function replyOpenFailedNotices(opts: { + opening: boolean; directs: boolean; agentName?: string; live?: string[]; homes?: Array<[string, string]>; denied?: string[]; + selectedForOpening?: 'unrecorded' | 'server-bootstrap' | 'subscription-policy'; primary?: string; dropsDuring?: true; +}) { + const agentName = opts.agentName ?? 'agent'; + const told: Array<{ text: string; metadata: unknown; forAgent?: string }> = []; + const logged: string[] = []; + const server = { + id: 'srv', + isConnected: true, + async sendToolsList() { return { tools: [{ name: 'send_message', inputSchema: { type: 'object' } }] }; }, + async sendToolsCall() { return { content: [{ type: 'text', text: 'sent' }] }; }, + }; + const framework = Object.create(AgentFramework.prototype) as any; + framework.mcplTools = []; + framework.mcplToolFeatureSets = new Map(); + framework.mcplServerConfigs = new Map([['srv', { id: 'srv', toolPrefix: 'srv' }]]); + framework.mcplServerRegistry = { getAllServers: () => [server], getServer: (id: string) => (id === 'srv' ? server : null) }; + framework.checkpointManager = null; + framework.turnRouting = new Map(); + framework.agents = new Map((opts.live ?? [agentName]).map((name) => [name, { canUseTool: (tool: string) => !(opts.denied ?? []).includes(tool) }])); + framework.usedEphemeralAgentNames = new Set(); + framework.conversationAgentHomes = new Map(opts.homes ?? []); + framework.primaryAgentName = opts.primary; + framework.emitTrace = () => {}; + framework.addMessage = (_participant: string, content: Array<{ text: string }>, metadata: unknown, routed?: { forAgent?: string }) => { + told.push({ text: content[0]!.text, metadata, ...(routed ? { forAgent: routed.forAgent } : {}) }); + return 'm'; + }; + // A transport that drops while the open is out clears the server's grant + // before the open's failure arrives (dropsDuring). + let opened = false; + framework.channelRegistry = { + openIfClosedForSend: async () => { + opened = true; + return { status: 'open-failed', channelId: 'x', label: '#x', opening: opts.opening, + ...(opts.selectedForOpening ? { selectedForOpening: opts.selectedForOpening } : {}) }; + }, + directsLifecycle: () => opts.directs && !(opts.dropsDuring && opened), + }; + const error = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + await framework.refreshMcplTools(); + const sent = await new Promise<{ success?: boolean; error?: string }>((resolve) => { + framework.pushEvent = (event: { type: string; result: { success?: boolean; error?: string } }) => { + if (event.type === 'tool-result') resolve(event.result); + }; + framework.dispatchMcplToolCall(agentName, { id: 'c1', name: 'srv--send_message', input: { channelId: 'x' } }, 'srv', 'srv'); + }); + assert.equal(sent.success, true, `the send went out: ${sent.error ?? ''}`); + // The open's answer is handled once its console line is written. + for (let i = 0; i < 50 && !logged.some((l) => l.includes('open-on-send failed')); i++) await tick(); + } finally { + console.error = error; + } + assert.ok(logged.some((l) => l.includes('open-on-send failed')), 'the failed open was handled'); + return { told, logged }; +} + +test("a reply whose open would have opened the channel, and failed, tells the agent; a failed reconfirm, or a server without channels.lifecycle, tells it nothing", async () => { + assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true })).told, [{ + text: '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. channel_open tries again.', + metadata: { system: true, kind: 'channel-notice' }, + forAgent: 'agent', + }]); + assert.deepEqual((await replyOpenFailedNotices({ opening: false, directs: true })).told, [], + 'a failed reconfirm of a channel already selected open changes nothing the agent was told'); + assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: false })).told, [], + "where the server opens and closes its own channels, channel_open can't open one either"); + assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true, dropsDuring: true })).told.map((t) => t.text), [ + '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. channel_open tries again.', + ], "a transport that drops while the open is out clears the grant before the failure arrives: whether the host directs the lifecycle is read before the open"); + assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true, selectedForOpening: 'unrecorded' })).told.map((t) => t.text), [ + "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again, and channel_close withdraws that. channel_open tries again.", + ], "where the put-back couldn't be recorded, as speech's failure says, in channel_open's own failure's words"); + assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true, selectedForOpening: 'server-bootstrap' })).told.map((t) => t.text), [ + "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It's selected for opening again, as its server announces it open: the host tries again, and channel_close withdraws that.", + ], "where the server's announcement lifted the default put back, the host tries again itself, so channel_open isn't named"); + assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true, selectedForOpening: 'subscription-policy' })).told.map((t) => t.text), [ + "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It's selected for opening again, as the subscription policy admits it: the host tries again, and channel_close withdraws that.", + ], 'where the subscription policy lifted it'); +}); + +test("the reply notice names channel_open and channel_close only to an agent that can call them; where the replying agent has ended, no one is told unless the channel is selected for opening, and then the primary is, by name", async () => { + const facts = '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.'; + const stays = " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again"; + const fork = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'x']] }); + assert.deepEqual(fork.told.map((t) => t.text), [facts], "a conversation fork's channel_open is refused outright"); + + const sendOnly = await replyOpenFailedNotices({ opening: true, directs: true, denied: ['channel_open'] }); + assert.deepEqual(sendOnly.told.map((t) => t.text), [facts], "an agent whose allowedTools leave channel_open out"); + + const forkHome = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'x']], selectedForOpening: 'unrecorded' }); + assert.deepEqual(forkHome.told.map((t) => t.text), [facts + stays + ', and channel_close withdraws that.'], + 'a fork may close its home channel, though not open it'); + const forkElsewhere = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'y']], selectedForOpening: 'unrecorded' }); + assert.deepEqual(forkElsewhere.told.map((t) => t.text), [facts + stays + '.'], + "a fork's channel_close is refused for a channel that isn't its home"); + const noOpen = await replyOpenFailedNotices({ opening: true, directs: true, denied: ['channel_open'], selectedForOpening: 'unrecorded' }); + assert.deepEqual(noOpen.told.map((t) => t.text), [facts + stays + ', and channel_close withdraws that.'], + 'an agent without channel_open is still told how to withdraw the selection'); + const noClose = await replyOpenFailedNotices({ opening: true, directs: true, denied: ['channel_close'], selectedForOpening: 'unrecorded' }); + assert.deepEqual(noClose.told.map((t) => t.text), [facts + stays + '. channel_open tries again.'], + 'an agent whose allowedTools leave channel_close out'); + + const again = " It's selected for opening again, as its server announces it open: the host tries again"; + const forkAgain = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'x']], selectedForOpening: 'server-bootstrap' }); + assert.deepEqual(forkAgain.told.map((t) => t.text), [facts + again + ', and channel_close withdraws that.'], + 'a fork told the server lifted its home channel may close it'); + + const ended = await replyOpenFailedNotices({ opening: true, directs: true, live: ['primary'], primary: 'primary' }); + assert.deepEqual(ended.told, [], 'the channel stays closed, so nothing follows from the failure'); + assert.ok(ended.logged.some((l) => l.includes("agent has ended; not telling it that #x (x) couldn't be opened"))); + + const endedFacts = "[channels] Not open: #x (x). agent's reply there went out, but opening the channel failed; agent has ended."; + const endedAgain = await replyOpenFailedNotices({ opening: true, directs: true, live: ['primary'], primary: 'primary', selectedForOpening: 'server-bootstrap' }); + assert.deepEqual(endedAgain.told, [{ + text: endedFacts + again + ', and channel_close withdraws that.', + metadata: { system: true, kind: 'channel-notice', speaker: 'agent' }, + }], 'selected for opening, so the primary is told by name, unrouted, with advice judged for it'); + assert.ok(endedAgain.logged.some((l) => l === '[channel] reply open-failed notice -> the primary, by name (agent has ended): #x (x)')); + const endedStays = await replyOpenFailedNotices({ opening: true, directs: true, live: ['primary'], primary: 'primary', selectedForOpening: 'unrecorded' }); + assert.deepEqual(endedStays.told.map((t) => t.text), [endedFacts + stays + ', and channel_close withdraws that. channel_open tries again.']); + const endedPrimaryNoClose = await replyOpenFailedNotices({ opening: true, directs: true, live: ['primary'], primary: 'primary', denied: ['channel_close'], selectedForOpening: 'server-bootstrap' }); + assert.deepEqual(endedPrimaryNoClose.told.map((t) => t.text), [endedFacts + again + '.'], + "the advice is judged for the primary, which can't close it, not for the agent that ended"); + const endedPrimaryNoOpen = await replyOpenFailedNotices({ opening: true, directs: true, live: ['primary'], primary: 'primary', denied: ['channel_open'], selectedForOpening: 'unrecorded' }); + assert.deepEqual(endedPrimaryNoOpen.told.map((t) => t.text), [endedFacts + stays + ', and channel_close withdraws that.'], + "nor is channel_open named to a primary that can't call it"); + const endedPrimaryGone = await replyOpenFailedNotices({ opening: true, directs: true, live: ['other'], primary: 'gone', selectedForOpening: 'server-bootstrap' }); + assert.deepEqual(endedPrimaryGone.told, [], 'a primary named but not registered is no one to tell'); + const endedNoPrimary = await replyOpenFailedNotices({ opening: true, directs: true, live: [], selectedForOpening: 'server-bootstrap' }); + assert.deepEqual(endedNoPrimary.told, [], 'with no primary there is no one to tell'); + assert.ok(endedNoPrimary.logged.some((l) => l.includes("agent has ended, and there is no primary to tell that #x (x) couldn't be opened and is selected for opening"))); +}); + +test("a reply notice with no agent named, or an empty name, goes out unrouted, with its advice judged for, and its log naming, the agent addMessage delivers to", () => { + const facts = '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.'; + const stays = " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again"; + const cases: Array<{ + why: string; name?: string; agents: Array<[string, string[]]>; primary: string | null; + selectedForOpening?: 'unrecorded' | 'server-bootstrap' | 'subscription-policy'; + reader: string; text: string; + }> = [ + { why: 'no name, the primary', agents: [['primary', []]], primary: 'primary', reader: 'primary', text: facts + ' channel_open tries again.' }, + { why: 'an empty name, the primary', name: '', agents: [['primary', []]], primary: 'primary', reader: 'primary', text: facts + ' channel_open tries again.' }, + { why: 'a primary without channel_open', agents: [['primary', ['channel_open']]], primary: 'primary', reader: 'primary', text: facts }, + { why: 'stays selected, a primary that can close', agents: [['primary', []]], primary: 'primary', selectedForOpening: 'unrecorded', + reader: 'primary', text: facts + stays + ', and channel_close withdraws that. channel_open tries again.' }, + { why: 'stays selected, a primary without channel_close', agents: [['primary', ['channel_close']]], primary: 'primary', selectedForOpening: 'unrecorded', + reader: 'primary', text: facts + stays + '. channel_open tries again.' }, + { why: 'selected again by the server, a primary that can open and close', agents: [['primary', []]], primary: 'primary', + selectedForOpening: 'server-bootstrap', reader: 'primary', + text: facts + " It's selected for opening again, as its server announces it open: the host tries again, and channel_close withdraws that." }, + { why: 'no primary: the first registered agent', agents: [['first', ['channel_open']], ['second', []]], primary: null, + reader: 'first', text: facts }, + { why: 'a primary that is not the first registered agent', agents: [['first', ['channel_open']], ['second', []]], primary: 'second', + reader: 'second', text: facts + ' channel_open tries again.' }, + ]; + for (const c of cases) { + const told: Array<{ text: string; forAgent?: string }> = []; + const logged: string[] = []; + const framework = Object.create(AgentFramework.prototype) as any; + framework.agents = new Map(c.agents.map(([name, denied]) => [name, { name, canUseTool: (tool: string) => !denied.includes(tool) }])); + framework.primaryAgentName = c.primary; + framework.usedEphemeralAgentNames = new Set(); + framework.conversationAgentHomes = new Map(); + framework.addMessage = (_participant: string, content: Array<{ text: string }>, _metadata: unknown, routed?: { forAgent?: string }) => { + told.push({ text: content[0]!.text, ...(routed ? { forAgent: routed.forAgent } : {}) }); + return 'm'; + }; + const error = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + framework.recordReplyOpenFailedNotice(c.name, { channelId: 'x', label: '#x' }, c.selectedForOpening); + } finally { + console.error = error; + } + assert.deepEqual(told, [{ text: c.text }], `${c.why}: no forAgent, so addMessage delivers it to ${c.reader}`); + assert.deepEqual(logged, [`[channel] reply open-failed notice -> ${c.reader}: #x (x)`], `${c.why}: the log names ${c.reader}`); + } +}); + +test("where a reply's reader can't close a channel selected for opening, the primary is told by name too, if it can close it", () => { + const facts = '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.'; + const stays = " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again"; + const again = " It's selected for opening again, as its server announces it open: the host tries again"; + const byName = (who: string) => `[channels] Not open: #x (x). ${who}'s reply there went out, but opening the channel failed, and ${who} can't close it.`; + const cases: Array<{ + why: string; name: string; agents: Array<[string, string[]]>; homes?: Array<[string, string]>; primary: string | null; + selectedForOpening?: 'unrecorded' | 'server-bootstrap'; + told: Array<{ text: string; metadata: unknown; forAgent?: string }>; + }> = [ + { why: 'a fork away from its home, selected again by its server', name: 'fork', agents: [['fork', []], ['primary', []]], homes: [['fork', 'y']], + primary: 'primary', selectedForOpening: 'server-bootstrap', + told: [ + { text: facts + again + '.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, + { text: byName('fork') + again + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + ] }, + { why: "an agent whose allowedTools leave channel_close out, the put-back unrecorded", name: 'helper', agents: [['helper', ['channel_close']], ['primary', []]], + primary: 'primary', selectedForOpening: 'unrecorded', + told: [ + { text: facts + stays + '. channel_open tries again.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'helper' }, + { text: byName('helper') + stays + ', and channel_close withdraws that. channel_open tries again.', metadata: { system: true, kind: 'channel-notice', speaker: 'helper' } }, + ] }, + { why: 'a reader who can close it, beside a primary that can too', name: 'helper', agents: [['helper', []], ['primary', []]], + primary: 'primary', selectedForOpening: 'server-bootstrap', + told: [{ text: facts + again + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'helper' }] }, + { why: "a primary that can close it but not open it, the put-back unrecorded", name: 'fork', agents: [['fork', []], ['primary', ['channel_open']]], + homes: [['fork', 'y']], primary: 'primary', selectedForOpening: 'unrecorded', + told: [ + { text: facts + stays + '.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, + { text: byName('fork') + stays + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + ] }, + { why: 'a fork away from its home, the channel left closed', name: 'fork', agents: [['fork', []], ['primary', []]], homes: [['fork', 'y']], + primary: 'primary', + told: [{ text: facts, metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }] }, + { why: "a primary that can't close it either", name: 'fork', agents: [['fork', []], ['primary', ['channel_close']]], homes: [['fork', 'y']], + primary: 'primary', selectedForOpening: 'server-bootstrap', + told: [{ text: facts + again + '.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }] }, + { why: 'no primary', name: 'fork', agents: [['fork', []]], homes: [['fork', 'y']], primary: null, selectedForOpening: 'server-bootstrap', + told: [{ text: facts + again + '.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }] }, + { why: "the primary itself, without channel_close", name: 'primary', agents: [['primary', ['channel_close']]], primary: 'primary', selectedForOpening: 'unrecorded', + told: [{ text: facts + stays + '. channel_open tries again.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'primary' }] }, + ]; + for (const c of cases) { + const told: Array<{ text: string; metadata: unknown; forAgent?: string }> = []; + const logged: string[] = []; + const framework = Object.create(AgentFramework.prototype) as any; + framework.agents = new Map(c.agents.map(([name, denied]) => [name, { name, canUseTool: (tool: string) => !denied.includes(tool) }])); + framework.primaryAgentName = c.primary; + framework.usedEphemeralAgentNames = new Set(); + framework.conversationAgentHomes = new Map(c.homes ?? []); + framework.addMessage = (_participant: string, content: Array<{ text: string }>, metadata: unknown, routed?: { forAgent?: string }) => { + told.push({ text: content[0]!.text, metadata, ...(routed ? { forAgent: routed.forAgent } : {}) }); + return 'm'; + }; + const error = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + framework.recordReplyOpenFailedNotice(c.name, { channelId: 'x', label: '#x' }, c.selectedForOpening); + } finally { + console.error = error; + } + assert.deepEqual(told, c.told, c.why); + if (c.told.length === 2) { + assert.ok(logged.includes(`[channel] reply open-failed notice -> the primary, by name (${c.name} can't close it): #x (x)`), `${c.why}: the log says why`); + } + } +}); + +test("a reply's open that only reconfirms a tuned-out channel's transport, and fails, isn't one that would have opened it", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + f.registry.enterTuneOut('test', 'x', { + epochId: 'kept-epoch', cadenceSeconds: 60, backlogCap: 20, maxWakes: 3, startedAtSequence: 1, + }, 'agent-tool'); + f.held.add('x'); + const retarget = f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B')] }); + await tick(); + f.pending.shift()!.fail(); + await retarget; + assert.equal(f.isOpen(), false); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + assert.deepEqual(await reply, { status: 'open-failed', channelId: 'x', label: 'x', opening: false }); + assert.equal(f.registry.getDesiredState('test', 'x'), 'tuned-out'); +}); + +test("a reply's open that only reconfirms a channel selected open, and fails, isn't one that would have opened it", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + f.held.add('x'); + const retarget = f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B')] }); + await tick(); + f.pending.shift()!.fail(); + await retarget; + assert.equal(f.isOpen(), false); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open'); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + f.pending.shift()!.fail(); + f.held.delete('x'); + assert.deepEqual(await reply, { status: 'open-failed', channelId: 'x', label: 'x', opening: false }); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open', 'the decision stands'); +}); + +test("a reply that waits on a queued close, which leaves the channel closed, isn't an opening that failed: nothing was opened", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); await tick(); + assert.equal(f.isOpen(), true); + f.held.add('x'); + const closed = f.tool('close'); + await tick(); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + f.held.delete('x'); + const before = f.calls.length; + await f.drain(closed, reply); + assert.deepEqual(await reply, { status: 'open-failed', channelId: 'x', label: 'x', opening: false }); + assert.deepEqual(f.calls.slice(before).filter((c) => c.kind === 'open'), [], 'the reply sent no open'); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'agent-tool' }); +}); + +test("a reply into a channel whose server is gone writes no decision and queues no close", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + const log = recordLog(f.registry); + const calls = f.calls.length; + // The server is gone while its channel is still registered. + (f.registry as unknown as { serverRegistry: { getServer: () => undefined } }).serverRegistry = { getServer: () => undefined }; + const answer = await f.registry.openIfClosedForSend('x', 'test'); + await tick(); await tick(); + assert.equal(answer.status, 'open-failed'); + assert.deepEqual(f.traces.filter((t) => t.type === 'mcpl:channel-open-failed').map((t) => t.error), ['Server not found: test'], + 'the failure names the missing server, not a missing grant'); + assert.deepEqual(log, [], 'nothing to put back, and no close to queue'); + assert.equal(f.calls.length, calls); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'default-closed' }); +}); + +test("the agent's channel_close over its own decline writes nothing: both are the agent's", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + const declined = await f.registry.handleChannelToolCall('channel_decline', { serverId: 'test', channelId: 'x', messageId: 'm1' }); + assert.equal(declined.success, true); + const log = recordLog(f.registry); + const closed = await f.tool('close'); + assert.equal(dataOf(closed).status, 'already closed'); + assert.deepEqual(log, []); + assert.deepEqual(decision(f.registry), { state: 'closed', source: 'invitation-declined' }); +}); + +test("channel_open on a server without channels.lifecycle still records the agent's decision, though its open fails", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + await tick(); await tick(); + f.setGrant(ALL_CAPABILITY_PATHS.filter((p) => p !== 'channels.lifecycle')); + const opened = await f.tool('open'); + assert.equal(opened.success, false); + assert.match(opened.error ?? '', /channels\.lifecycle not in "test"'s effective grant/); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'agent-tool' }, "the agent's own open stays selected (#210), unlike speech's"); +}); + +test("the agent's channel_open after a lost answer waits behind the put-back's close in the channel's queue", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + f.held.add('x'); + const speech = f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' }); + await tick(); + const open = f.pending.shift()!; + const before = f.calls.length; + f.actual.set('x', true); // the server opened x, and its answer was lost + open.fail(); + await tick(); await tick(); + assert.equal((await speech).status, 'failed'); + assert.equal(f.pending.length, 1, "the put-back's close is out"); + const reopened = f.tool('open'); + await tick(); await tick(); + assert.equal(f.pending.length, 1, "the agent's open waits for that close: one request per channel at a time"); + f.held.delete('x'); + await f.drain(reopened); + assert.equal((await reopened).success, true); + const after = f.calls.slice(before).map((c) => c.kind); + assert.equal(after[0], 'close', `the put-back's close goes first: ${JSON.stringify(after)}`); + assert.equal(after.at(-1), 'open', `and the agent's open after it: ${JSON.stringify(after)}`); + assert.equal(f.actual.get('x'), true); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'agent-tool' }); +}); + +test("the agent's channel_open over a channel speech opened is the agent's decision, which a housekeeping close doesn't override", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', false)] }); + assert.equal((await f.registry.publish('agent', 'words', { serverId: 'test', channelId: 'x' })).status, 'delivered'); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'opened-by-delivery' }); + + assert.equal((await f.tool('open')).success, true); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'agent-tool' }); + const housekeeping = await f.registry.handleChannelToolCall( + 'channel_close', { serverId: 'test', channelId: 'x', source: 'housekeeping' }, { kind: 'module' }, + ); + assert.equal(housekeeping.success, false); + assert.match(housekeeping.error ?? '', /was explicitly opened \(source: agent-tool\)/); + assert.equal(f.isOpen(), true); +}); diff --git a/test/mcpl-request-outcome.test.ts b/test/mcpl-request-outcome.test.ts index b6443fb4a..46813c983 100644 --- a/test/mcpl-request-outcome.test.ts +++ b/test/mcpl-request-outcome.test.ts @@ -103,6 +103,20 @@ test('channels/open keeps a deadline when requestTimeoutMs is 0, as channels/pub assert.match(err.message, /^The "outcome" server didn't answer within 0\.2 seconds/); }); +test('channels/close keeps a deadline when requestTimeoutMs is 0, as channels/open does', { timeout: 10_000 }, async () => { + connection = await McplServerConnection.connect(config({ requestTimeoutMs: 0 }), HOST_CAPS); + (connection as unknown as { publishDeadlineMs: number }).publishDeadlineMs = 200; + const err = await rejectsWith(connection.sendChannelsClose({ channelId: 'nowhere' } as never), 'no-response'); + assert.match(err.message, /^The "outcome" server didn't answer within 0\.2 seconds/); +}); + +test('channels/close keeps a configured requestTimeoutMs rather than the deadline for 0', { timeout: 10_000 }, async () => { + connection = await McplServerConnection.connect(config({ requestTimeoutMs: 200 }), HOST_CAPS); + (connection as unknown as { publishDeadlineMs: number }).publishDeadlineMs = 60_000; + const err = await rejectsWith(connection.sendChannelsClose({ channelId: 'nowhere' } as never), 'no-response'); + assert.match(err.message, /^The "outcome" server didn't answer within 0\.2 seconds/); +}); + test('a request on a closed connection is not-sent', async () => { connection = await McplServerConnection.connect(config(), HOST_CAPS); await connection.close(); diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index c8ccfc2ec..2f7c043d9 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -1791,6 +1791,7 @@ describe('present while acting', () => { framework.checkpointManager = null; framework.channelRegistry = { openIfClosedForSend: () => new Promise((resolve) => { resolveOpen = resolve; }), + directsLifecycle: () => true, }; framework.turnRouting = new Map([['assistant', sending]]); const results: unknown[] = [];