From fed0e12fd6de751811b0133f485fbfa93f8909ba Mon Sep 17 00:00:00 2001 From: Petra-1012 Date: Mon, 19 Oct 2026 02:51:24 -0400 Subject: [PATCH 1/6] fix(mcpl): an open by speech or a reply that fails puts back the decision it replaced MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit openChannelNow records the decision to open before the round-trip, so the intent outlives a failed subscribe and reconciliation retries it. That is right for channel_open, the agent's own decision, whose failure says the channel stays selected. For an open that speech or a reply made it isn't. Speech fails with "channel is closed and open failed", but the decision to open stays. The channel's next open, by the agent's next speech or by reconciliation when the server announces the channel again, then opens it with no notice, since openChannelNow reports `opened: false` for a channel already decided open. An agent last told the channel is closed starts getting its traffic, and was never told it's open or how to close it. The put-back: - An open by speech or a reply whose round-trip fails puts back the decision it replaced, with that decision's source (such as `default-closed`, nobody having decided), through setDesiredState, so the lifecycle log records it and a restart replays it. A channel with no decision on record, as one registered from a push, goes back to closed by default. - It puts it back only while the record is still the one it wrote, so a decision made during the round-trip, such as the agent closing or opening the channel, stands. The failure then comes as a ChannelOpenOvertakenError, with the open's own message. - A failed request may have reached the server, as when its answer is lost to a timeout. So the put-back's close is carried out through the queue, as a decline's is, unless other lifecycle work for the channel is queued, which carries it out anyway (closeThroughQueue, shared with the close of a substituted channel). - If the put-back can't be recorded, the failure says the channel stays selected for opening. - Where the open fails before any request leaves, on a server without channels.lifecycle (§14.1) or one that's gone, it writes nothing, where it used to write a decision only to put it back. - channel_open's own decision still outlives a failed round-trip. channels/close gets the deadline channels/open has when requestTimeoutMs is 0, and requestTimeoutMs's doc says so. A delivery waits for a channel's queued lifecycle work, closes included, so a close never answered would hold it for good, and the put-back's close follows an open that may have failed for that very silence. The agent's decisions are recorded as the agent's: - A decision that records the state already on record writes nothing, unless it's the agent's own (agent-tool, or invitation-declined from 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. - So channel_open's "already open" answer and channel_close's "already closed" answer record the agent's decision, and so does a decline over a default. If either answer can't record it, it says so. - A channel the agent opened explicitly, over an open by speech, a reply, the server or a policy, now refuses a module's housekeeping close. A channel the agent closed or declined over a default isn't lifted open by a later registration. - channel_close's provenance check now runs before its "already closed" answer, which records closeSource, so a module close with an unknown machine source on an already-closed channel is refused for its source. - The put-back's identity check depends on this rule: the agent's channel_open is the only decision of the same state that can be made during the round-trip, and it gets a new record. A reply's failed open is told to the agent: - A reply whose open would have opened a channel not selected open, and failed, tells the agent, on a server whose lifecycle the host directs: the reply went out, opening the channel failed, and channel_open tries again. It sits beside the opened-by-reply notice of the success path. Before, the failure reached only the console and a trace. - openIfClosedForSend's open-failed answer says whether the open was such an opening (`opening`). A failed reconfirm of a channel already selected open or tuned out isn't one, and neither is an open that a decision made during its round-trip replaced, such as the agent's close in the same turn as its reply. A put-back that couldn't be recorded still is. - Without channels.lifecycle the server opens and closes its own channels and channel_open can't open one, so there is no notice there. ChannelRegistry.directsLifecycle is public for the framework to ask. - It names channel_open as the way to try again only to an agent that can call it (canAgentUseTool, dispatch's own gate): a conversation fork's channel_open is refused outright, and an agent's allowedTools can leave it out. The notice goes to the agent whose reply it was, or to the primary when none is named, as the opened notice does. Tests, in test/channel-lifecycle-order.test.ts: - speech whose open fails leaves the decision as it was, and the next speech's open gives its notice; - the server announcing the channel again afterwards opens nothing; - the agent's close, and the agent's channel_open, made while speech's open is out stand when that open fails; - a reply whose open fails puts back the agent's close; - a channel registered from a push is put back as closed by default; - the put-back survives a rebuild from the store; - when the server announces the channel again while speech's open is out, the queued reconcile carries out the put-back with one close; - speech waiting on a failed open into a channel the agent closed is refused; - an open whose answer is lost after the server opened the channel is closed on the wire too, and the agent's channel_open after it waits behind that close in the channel's queue; - speech or a reply on a server without channels.lifecycle, and a reply whose server is gone, writes no decision and queues no close; - channel_open on a server without channels.lifecycle still records the agent's decision; - the agent's channel_open over a channel speech or the server opened is recorded as the agent's, so a housekeeping close is refused; - the agent's channel_close, and its decline, over a default are recorded, and a later registration's initiallyOpen doesn't open the channel; - a housekeeping close over a default records nothing, the agent's channel_open twice writes one record, and its channel_close over its own decline writes nothing; - a module close outside the machine-close vocabulary is refused on a channel already closed too, where it got "already closed"; - a put-back that can't be recorded, and either "already" answer that can't record the agent's decision, say what failed; - a reply's failed open that would have opened the channel tells the agent (with the notice's system metadata), and a failed reconfirm, or one on a server without channels.lifecycle, tells it nothing; - the notice names channel_open only to an agent that can call it, so not to a fork or to an agent whose allowedTools leave it out, and an agent that has ended is told nothing; a notice with no agent named, or an empty name, goes to the primary and names channel_open as the primary may call it; - a reply's failed reconfirm of a channel selected open or tuned out isn't an opening; a reply that waits on a queued close leaving the channel closed sends no open; a reply's open that the agent's close replaced during its round-trip, or that failed after the agent's channel_open during it, isn't an opening that failed; one whose put-back can't be recorded is. In test/mcpl-request-outcome.test.ts: channels/close keeps a deadline when requestTimeoutMs is 0, and keeps a configured requestTimeoutMs otherwise. Co-Authored-By: Claude Opus 5.5 --- .../failed-speech-open-puts-back.fixed.md | 31 + src/framework.ts | 70 +- src/mcpl/channel-registry.ts | 170 ++++- src/mcpl/server-connection.ts | 8 +- src/mcpl/types.ts | 7 +- test/channel-lifecycle-order.test.ts | 659 +++++++++++++++++- test/mcpl-request-outcome.test.ts | 14 + 7 files changed, 922 insertions(+), 37 deletions(-) create mode 100644 changelog.d/failed-speech-open-puts-back.fixed.md 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..9d2c1287d --- /dev/null +++ b/changelog.d/failed-speech-open-puts-back.fixed.md @@ -0,0 +1,31 @@ +- 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. `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. + 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 d72da2323..dcdad4c2b 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -14407,6 +14407,47 @@ 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 to the primary + * when no agent is named, as the opened notice does. An agent that has + * ended is told nothing, and nor is anyone in its place: the notice is + * about what that agent can do next. 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. System message, as the opened + * notice is. + */ + private recordReplyOpenFailedNotice( + agentName: string | undefined, + channel: { channelId: string; label?: string }, + ): void { + const shown = channel.label && channel.label !== channel.channelId + ? `${channel.label} (${channel.channelId})` + : channel.channelId; + if (agentName && !this.agents.has(agentName)) { + console.error(`[channel] ${agentName} has ended; not telling it that ${shown} couldn't be opened`); + return; + } + const who = agentName || this.deliveryAgent()?.name; + const canOpen = who !== undefined && !this.conversationAgentHomes.has(who) && this.canAgentUseTool(who, 'channel_open'); + const text = `[channels] Not open: ${shown}. Your reply there went out, but opening the channel failed.` + + (canOpen ? ' channel_open tries again.' : ''); + try { + this.addMessage( + 'user', + [{ type: 'text', text }], + { system: true, kind: 'channel-notice' } as MessageMetadata, + agentName ? { forAgent: agentName } : undefined, + ); + console.error(`[channel] reply open-failed notice -> ${agentName || 'primary'}: ${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 @@ -24858,12 +24899,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 ( @@ -24877,7 +24919,7 @@ export class AgentFramework { if (target) { this.channelRegistry .openIfClosedForSend(target, serverId) - .then(({ status, channelId, label }) => { + .then(({ status, channelId, label, opening }) => { // 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. @@ -24902,10 +24944,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 && this.channelRegistry?.directsLifecycle(serverId) === true; 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 }); } }) .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 5a0bb585c..1659643ce 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -60,8 +60,16 @@ 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 {} 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 @@ -825,7 +833,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'); } @@ -2175,7 +2183,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, @@ -2583,6 +2604,17 @@ 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.closeThroughQueue(serverId, channelId); + } + + /** + * Carry out a channel's closed 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). A failure is + * reported as a trace; the decision stands. + */ + private closeThroughQueue(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. @@ -2828,10 +2860,20 @@ 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. And since a failed request may have reached + * the server, the put-back's close 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 @@ -2843,17 +2885,63 @@ 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 Error( + `${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)}`, + ); + } + // The decision put back is a closed one: an open that turned the + // channel open replaced no open decision. The failed request may have + // reached the server, so the close is carried out through the queue, + // as a decline's is, unless other lifecycle work for the channel is + // queued, which carries it out anyway. (An incidental open gets here + // only where the host directs the lifecycle: see the grant check + // above.) + const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); + if (current && !this.lifecycleOutstanding.has(key)) { + this.closeThroughQueue(entry.serverId, entry.descriptor.id); + } + } + throw err; } - return { result: applied.result!, opened }; } /** @@ -3042,7 +3130,13 @@ 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. */ async openIfClosedForSend( rawChannelId: string, @@ -3051,6 +3145,7 @@ export class ChannelRegistry { status: 'opened' | 'already-open' | 'unknown-channel' | 'ambiguous' | 'open-failed'; channelId?: string; label?: string; + opening?: boolean; }> { let matches = [...this.channels.values()].filter( (e) => e.descriptor.id === rawChannelId && (!serverId || e.serverId === serverId), @@ -3065,6 +3160,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); @@ -3073,6 +3169,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) { @@ -3084,6 +3181,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, @@ -3091,7 +3192,7 @@ 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, ...(current ? { label: current.descriptor.label } : {}) }; } } @@ -3132,6 +3233,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' }, @@ -3202,15 +3316,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 @@ -3231,6 +3336,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 4b4d39337..98a0157ae 100644 --- a/src/mcpl/server-connection.ts +++ b/src/mcpl/server-connection.ts @@ -1485,9 +1485,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 4ef2a3963..a3168a9b9 100644 --- a/src/mcpl/types.ts +++ b/src/mcpl/types.ts @@ -414,9 +414,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 b6c80e178..17a19b1cb 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'; @@ -1238,3 +1238,660 @@ 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", 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 { answer } = await replyWhoseOpenFails(f); + assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true }); + assert.deepEqual(decision(f.registry), { state: 'open', source: 'opened-by-reply' }, 'the channel stays selected for opening'); +}); + +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('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 { 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' }); +}); + +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[]; +}) { + 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.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'; + }; + framework.channelRegistry = { + openIfClosedForSend: async () => ({ status: 'open-failed', channelId: 'x', label: '#x', opening: opts.opening }), + directsLifecycle: () => opts.directs, + }; + 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"); +}); + +test("the reply notice names channel_open only to an agent that can call it, and an agent that has ended is told nothing", async () => { + const facts = '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.'; + 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 ended = await replyOpenFailedNotices({ opening: true, directs: true, live: [] }); + assert.deepEqual(ended.told, []); + assert.ok(ended.logged.some((l) => l.includes("agent has ended; not telling it that #x (x) couldn't be opened"))); +}); + +test("a reply notice with no agent named, or an empty name, goes to the primary, and names channel_open as the primary may call it", () => { + for (const [name, denied] of [[undefined, false], ['', false], [undefined, true]] as const) { + const told: Array<{ text: string; forAgent?: string }> = []; + const framework = Object.create(AgentFramework.prototype) as any; + framework.agents = new Map([['primary', { name: 'primary', canUseTool: (tool: string) => !(denied && tool === 'channel_open') }]]); + framework.primaryAgentName = '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 = () => {}; + try { + framework.recordReplyOpenFailedNotice(name, { channelId: 'x', label: '#x' }); + } finally { + console.error = error; + } + assert.deepEqual(told, [{ + text: '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.' + (denied ? '' : ' channel_open tries again.'), + }], `name ${JSON.stringify(name)}${denied ? ', a primary without channel_open' : ''}: no forAgent, so the primary reads it`); + } +}); + +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 43be23748..f4c34e322 100644 --- a/test/mcpl-request-outcome.test.ts +++ b/test/mcpl-request-outcome.test.ts @@ -94,6 +94,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(); From 079f19810f5bb54802101cd05ff53453bcf9632a Mon Sep 17 00:00:00 2001 From: Petra-1012 Date: Mon, 19 Oct 2026 06:34:04 -0400 Subject: [PATCH 2/6] fix(mcpl): a reply whose put-back can't be recorded says the channel stays selected When the store refuses the put-back's record, openChannelNow keeps the open decision, and speech's failure says the channel stays selected for opening. The reply path dropped that: openIfClosedForSend answered `open-failed` alone, so the reply notice said only that the channel wasn't open, though a later registration may still open it. - openChannelNow throws ChannelOpenPutBackUnrecordedError there, with the same message as before, so speech's failure text is unchanged. - openIfClosedForSend's `open-failed` answer carries `staysSelected: true` for it. - The reply notice then adds that the channel stays selected for opening, as the host couldn't record undoing that, and, in channel_open's own failure's words for a decision that outlives a failed open, that a later delivery or reconnect tries again and channel_close withdraws that. channel_close is named only where it would run: an agent whose allowedTools hold it, and a conversation fork only for its home channel, as dispatchChannelToolCall reads it. - The notice's log line names the agent its advice was judged for, which is where addMessage delivers it, rather than 'primary'. - The changelog's first bullet says that where the put-back can't be recorded, the decision stays and no close is sent. Tests, in test/channel-lifecycle-order.test.ts: a reply's failed open whose put-back can't be recorded answers `staysSelected: true`; the notice for such an answer says the channel stays selected, names channel_close to an agent that can call it, and leaves it out for an agent without it and for a fork away from its home; and the notice with no agent named, or an empty name, is judged for, and logged as read by, the agent addMessage delivers it to: the primary, else the first registered agent. Co-Authored-By: Claude Opus 5.5 --- .../failed-speech-open-puts-back.fixed.md | 12 +++- src/framework.ts | 31 ++++++--- src/mcpl/channel-registry.ts | 15 ++++- test/channel-lifecycle-order.test.ts | 63 +++++++++++++++---- 4 files changed, 96 insertions(+), 25 deletions(-) diff --git a/changelog.d/failed-speech-open-puts-back.fixed.md b/changelog.d/failed-speech-open-puts-back.fixed.md index 9d2c1287d..16a8e9e3b 100644 --- a/changelog.d/failed-speech-open-puts-back.fixed.md +++ b/changelog.d/failed-speech-open-puts-back.fixed.md @@ -6,8 +6,10 @@ 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. `channel_open`'s own decision still outlives a failed open, as - its failure says. + the server. 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 @@ -16,7 +18,11 @@ (`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. - Before, the failure reached only the logs. + Where the host couldn't record putting the channel's decision back, the + notice says the channel stays selected for opening, as speech's failure + does, and, in `channel_open`'s own failure's words, that a later delivery + or reconnect tries again and `channel_close` withdraws that, where the + agent can call it. 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 diff --git a/src/framework.ts b/src/framework.ts index dcdad4c2b..58cbf0862 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -14411,18 +14411,26 @@ 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 to the primary - * when no agent is named, as the opened notice does. An agent that has + * 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. An agent that has * ended is told nothing, and nor is anyone in its place: the notice is * about what that agent can do next. 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. System message, as the opened - * notice is. + * agent's allowedTools can leave it out. Where the host couldn't record + * putting the channel's decision back, it says the channel stays selected + * for opening, as speech's failure does, and what that means, in + * channel_open's own failure's words: a later delivery or reconnect tries + * again, and channel_close withdraws that, named only to an agent that can + * call it here (a fork may close only its home channel). System message, + * as the opened notice is. */ private recordReplyOpenFailedNotice( agentName: string | undefined, channel: { channelId: string; label?: string }, + staysSelected = false, ): void { const shown = channel.label && channel.label !== channel.channelId ? `${channel.label} (${channel.channelId})` @@ -14433,7 +14441,16 @@ export class AgentFramework { } const who = 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 text = `[channels] Not open: ${shown}. Your reply there went out, but opening the channel failed.` + + (staysSelected + ? " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again" + + (canClose ? ', and channel_close withdraws that.' : '.') + : '') + (canOpen ? ' channel_open tries again.' : ''); try { this.addMessage( @@ -14442,7 +14459,7 @@ export class AgentFramework { { system: true, kind: 'channel-notice' } as MessageMetadata, agentName ? { forAgent: agentName } : undefined, ); - console.error(`[channel] reply open-failed notice -> ${agentName || 'primary'}: ${shown}`); + console.error(`[channel] reply open-failed notice -> ${who ?? 'no agent'}: ${shown}`); } catch (err) { console.error('recordReplyOpenFailedNotice failed:', err); } @@ -24919,7 +24936,7 @@ export class AgentFramework { if (target) { this.channelRegistry .openIfClosedForSend(target, serverId) - .then(({ status, channelId, label, opening }) => { + .then(({ status, channelId, label, opening, staysSelected }) => { // 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. @@ -24957,7 +24974,7 @@ export class AgentFramework { `[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 }); + if (tell) this.recordReplyOpenFailedNotice(agentName, { channelId: channelId ?? target, label }, staysSelected === true); } }) .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 1659643ce..ca10bf5c7 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -64,6 +64,10 @@ class ChannelLifecycleCancelledError extends Error {} * 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 {} const CHANNEL_LIFECYCLE_LOG_ID = 'mcpl/channel-lifecycle'; /** The sources of a decision the agent itself made: channel_open and @@ -2923,9 +2927,10 @@ export class ChannelRegistry { } 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 Error( + 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 @@ -3136,7 +3141,9 @@ export class ChannelRegistry { * 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. + * channel closed before any open was sent. `staysSelected` says the + * put-back couldn't be recorded, so the channel stays selected for opening + * (ChannelOpenPutBackUnrecordedError). */ async openIfClosedForSend( rawChannelId: string, @@ -3146,6 +3153,7 @@ export class ChannelRegistry { channelId?: string; label?: string; opening?: boolean; + staysSelected?: true; }> { let matches = [...this.channels.values()].filter( (e) => e.descriptor.id === rawChannelId && (!serverId || e.serverId === serverId), @@ -3193,6 +3201,9 @@ export class ChannelRegistry { }); const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); return { status: 'open-failed', channelId: entry.descriptor.id, opening, + // Where the put-back couldn't be recorded, the channel stays selected + // for opening, and the agent is told so, as speech's failure says. + ...(err instanceof ChannelOpenPutBackUnrecordedError ? { staysSelected: true as const } : {}), ...(current ? { label: current.descriptor.label } : {}) }; } } diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index 17a19b1cb..aa6b60c0c 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -1403,12 +1403,12 @@ test("a reply's failed open after the agent's channel_open during its round-trip assert.equal(f.isOpen(), true); }); -test("a reply's failed open whose put-back can't be recorded is still an opening that failed", async () => { +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 { answer } = await replyWhoseOpenFails(f); - assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true }); + assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true, staysSelected: true }); assert.deepEqual(decision(f.registry), { state: 'open', source: 'opened-by-reply' }, 'the channel stays selected for opening'); }); @@ -1657,6 +1657,7 @@ test("channel_open and channel_close on a channel already so say what failed whe // bare framework, as in checkpoint-attribution. async function replyOpenFailedNotices(opts: { opening: boolean; directs: boolean; agentName?: string; live?: string[]; homes?: Array<[string, string]>; denied?: string[]; + staysSelected?: true; }) { const agentName = opts.agentName ?? 'agent'; const told: Array<{ text: string; metadata: unknown; forAgent?: string }> = []; @@ -1683,7 +1684,8 @@ async function replyOpenFailedNotices(opts: { return 'm'; }; framework.channelRegistry = { - openIfClosedForSend: async () => ({ status: 'open-failed', channelId: 'x', label: '#x', opening: opts.opening }), + openIfClosedForSend: async () => ({ status: 'open-failed', channelId: 'x', label: '#x', opening: opts.opening, + ...(opts.staysSelected ? { staysSelected: true } : {}) }), directsLifecycle: () => opts.directs, }; const error = console.error; @@ -1716,27 +1718,63 @@ test("a reply whose open would have opened the channel, and failed, tells the ag '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, staysSelected: true })).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"); }); -test("the reply notice names channel_open only to an agent that can call it, and an agent that has ended is told nothing", async () => { +test("the reply notice names channel_open and channel_close only to an agent that can call them, and an agent that has ended is told nothing", 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']], staysSelected: true }); + 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']], staysSelected: true }); + 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'], staysSelected: true }); + 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'], staysSelected: true }); + assert.deepEqual(noClose.told.map((t) => t.text), [facts + stays + '. channel_open tries again.'], + 'an agent whose allowedTools leave channel_close out'); + const ended = await replyOpenFailedNotices({ opening: true, directs: true, live: [] }); assert.deepEqual(ended.told, []); assert.ok(ended.logged.some((l) => l.includes("agent has ended; not telling it that #x (x) couldn't be opened"))); }); -test("a reply notice with no agent named, or an empty name, goes to the primary, and names channel_open as the primary may call it", () => { - for (const [name, denied] of [[undefined, false], ['', false], [undefined, true]] as const) { +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; staysSelected?: boolean; + 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', staysSelected: true, + 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', staysSelected: true, + reader: 'primary', text: facts + stays + '. channel_open tries again.' }, + { 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([['primary', { name: 'primary', canUseTool: (tool: string) => !(denied && tool === 'channel_open') }]]); - framework.primaryAgentName = 'primary'; + 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 }) => { @@ -1744,15 +1782,14 @@ test("a reply notice with no agent named, or an empty name, goes to the primary, return 'm'; }; const error = console.error; - console.error = () => {}; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; try { - framework.recordReplyOpenFailedNotice(name, { channelId: 'x', label: '#x' }); + framework.recordReplyOpenFailedNotice(c.name, { channelId: 'x', label: '#x' }, c.staysSelected === true); } finally { console.error = error; } - assert.deepEqual(told, [{ - text: '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.' + (denied ? '' : ' channel_open tries again.'), - }], `name ${JSON.stringify(name)}${denied ? ', a primary without channel_open' : ''}: no forAgent, so the primary reads it`); + 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}`); } }); From d1df7b76206a47c1ca2e4eeabb3f753aed3577ad Mon Sep 17 00:00:00 2001 From: Petra-1012 Date: Mon, 19 Oct 2026 07:55:49 -0400 Subject: [PATCH 3/6] test(mcpl): a put-back that can't be recorded sends no close The changelog says that where the host can't record putting a failed open's decision back, the decision stays and no close is sent. The two tests of that case, for speech and for a reply, asserted the failure's words and the decision, and now also assert that no `channels/close` goes out. Co-Authored-By: Claude Opus 5.5 --- test/channel-lifecycle-order.test.ts | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index aa6b60c0c..0a2ecf032 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -1407,9 +1407,12 @@ test("a reply's failed open whose put-back can't be recorded is still an opening 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, staysSelected: true }); 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 () => { @@ -1625,10 +1628,13 @@ test("when putting back the decision a failed open replaced can't be recorded, s 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 () => { From c55b32dcf4aef9bcf530b6d0ae59a7d96136b15a Mon Sep 17 00:00:00 2001 From: Petra-1012 Date: Mon, 19 Oct 2026 12:43:42 -0400 Subject: [PATCH 4/6] fix(mcpl): a failed open's put-back of a default follows the server's latest announcement slimepriestess's review of d1df7b76 found that an initially-open re-announcement during a failed incidental open is lost to the put-back: the lift (ensureInitialDesiredState) finds the open's own record and writes nothing, and the put-back then restores `default-closed`, so the server's bootstrap preference waits for its next registration. - After a put-back, the registry runs the lift against the channel's current descriptor. A default-closed record means nobody decided, so it follows the server's latest announcement and the subscription policy; every real decision is left alone, as the lift already does. A lift that can't be recorded leaves the default, closed, and the open's own failure is reported. - Where the lift selects the channel for opening, the queued lifecycle work opens it (closeThroughQueue becomes applyThroughQueue, and its trace names the decision it carries out), and the failure says why: ChannelOpenSelectedAgainError for speech, and `selectedForOpening` on openIfClosedForSend's answer, which replaces `staysSelected` with the reason: 'unrecorded', 'server-bootstrap' or 'subscription-policy'. - The reply notice says the channel is selected for opening again and why, and that channel_close withdraws that; channel_open isn't named there, since the host tries again itself. - From the review's second note: where the replying agent has ended and the channel is selected for opening, the primary is told by name, as recordChannelAutoOpenNotice tells it of an open; where the channel stays closed, no one is told, as before. - By the same rule, where the reply's 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, isn't the reader and can close it. Tests: the reviewer's sequence for speech and a reply; the server removed and added again as initially open; the subscription policy; an agent's close put back and left alone; a lift that can't be recorded; and the notice's new wording, routing and advice, including a reader who can't close the channel. Co-Authored-By: Claude Opus 5.5 --- .../failed-speech-open-puts-back.fixed.md | 26 ++- src/framework.ts | 98 ++++++--- src/mcpl/channel-registry.ts | 104 ++++++--- test/channel-lifecycle-order.test.ts | 201 ++++++++++++++++-- 4 files changed, 354 insertions(+), 75 deletions(-) diff --git a/changelog.d/failed-speech-open-puts-back.fixed.md b/changelog.d/failed-speech-open-puts-back.fixed.md index 16a8e9e3b..07d186a18 100644 --- a/changelog.d/failed-speech-open-puts-back.fixed.md +++ b/changelog.d/failed-speech-open-puts-back.fixed.md @@ -6,10 +6,14 @@ 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 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. + 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 @@ -18,11 +22,15 @@ (`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 host couldn't record putting the channel's decision back, the - notice says the channel stays selected for opening, as speech's failure - does, and, in `channel_open`'s own failure's words, that a later delivery - or reconnect tries again and `channel_close` withdraws that, where the - agent can call it. Before, the failure reached only the logs. + 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. If the replying agent has ended by then, or can't + close the channel, as a fork away from its home can't, the primary is + told by name. 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 diff --git a/src/framework.ts b/src/framework.ts index 58cbf0862..26e9d8f5c 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -102,7 +102,7 @@ import { parseProsePrefix, parseHybridProsePrefix } from './mcpl/prose-grammar.j import { ProseStreamRouter } from './mcpl/prose-stream-router.js'; import { detectKnownToolWrapperProse } from './tool-wrapper-prose-guard.js'; import { InferenceRouter } from './mcpl/inference-router.js'; -import { ChannelRegistry, connectorText, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome } from './mcpl/channel-registry.js'; +import { ChannelRegistry, connectorText, 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'; @@ -14414,55 +14414,103 @@ export class AgentFramework { * 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. An agent that has - * ended is told nothing, and nor is anyone in its place: the notice is - * about what that agent can do next. It names channel_open as the way to + * 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, as the opened notice + * tells it of an open by an agent that has ended: no channel opens without + * someone who can close it being told. 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 host couldn't record - * putting the channel's decision back, it says the channel stays selected - * for opening, as speech's failure does, and what that means, in - * channel_open's own failure's words: a later delivery or reconnect tries - * again, and channel_close withdraws that, named only to an agent that can - * call it here (a fork may close only its home channel). System message, - * as the opened notice is. + * 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, by the + * same rule. System message, as the opened notice is. */ private recordReplyOpenFailedNotice( agentName: string | undefined, channel: { channelId: string; label?: string }, - staysSelected = false, + selectedForOpening?: SelectedForOpening, ): void { const shown = channel.label && channel.label !== channel.channelId ? `${channel.label} (${channel.channelId})` : channel.channelId; - if (agentName && !this.agents.has(agentName)) { + 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 who = agentName || this.deliveryAgent()?.name; + 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 text = `[channels] Not open: ${shown}. Your reply there went out, but opening the channel failed.` + - (staysSelected - ? " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again" + - (canClose ? ', and channel_close withdraws that.' : '.') - : '') + - (canOpen ? ' channel_open tries again.' : ''); + 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 }], - { system: true, kind: 'channel-notice' } as MessageMetadata, - agentName ? { forAgent: agentName } : undefined, + (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}`, ); - console.error(`[channel] reply open-failed notice -> ${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 can close + // it: no channel opens without someone who can close it being told. + 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); + } + } } /** @@ -24936,7 +24984,7 @@ export class AgentFramework { if (target) { this.channelRegistry .openIfClosedForSend(target, serverId) - .then(({ status, channelId, label, opening, staysSelected }) => { + .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. @@ -24974,7 +25022,7 @@ export class AgentFramework { `[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 }, staysSelected === true); + 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 ca10bf5c7..d600cf7b0 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -68,6 +68,19 @@ class ChannelOpenOvertakenError extends Error {} * 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 @@ -2608,17 +2621,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.closeThroughQueue(serverId, channelId); + this.applyThroughQueue(serverId, channelId); } /** - * Carry out a channel's closed decision through its lifecycle queue, for a + * 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). A failure is - * reported as a trace; the decision stands. + * 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 closeThroughQueue(serverId: string, channelId: string): void { + 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. @@ -2627,7 +2642,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, }); }); } @@ -2872,12 +2887,17 @@ export class ChannelRegistry { * 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. And since a failed request may have reached - * the server, the put-back's close 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. + * 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 @@ -2934,15 +2954,44 @@ export class ChannelRegistry { ); } // The decision put back is a closed one: an open that turned the - // channel open replaced no open decision. The failed request may have - // reached the server, so the close is carried out through the queue, - // as a decline's is, unless other lifecycle work for the channel is - // queued, which carries it out anyway. (An incidental open gets here - // only where the host directs the lifecycle: see the grant check - // above.) + // 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, which + // carries it out anyway. (An incidental open gets here only where the + // host directs the lifecycle: see the grant check above.) if (current && !this.lifecycleOutstanding.has(key)) { - this.closeThroughQueue(entry.serverId, entry.descriptor.id); + 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; @@ -3141,9 +3190,11 @@ export class ChannelRegistry { * 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. `staysSelected` says the - * put-back couldn't be recorded, so the channel stays selected for opening - * (ChannelOpenPutBackUnrecordedError). + * 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, @@ -3153,7 +3204,7 @@ export class ChannelRegistry { channelId?: string; label?: string; opening?: boolean; - staysSelected?: true; + selectedForOpening?: SelectedForOpening; }> { let matches = [...this.channels.values()].filter( (e) => e.descriptor.id === rawChannelId && (!serverId || e.serverId === serverId), @@ -3201,9 +3252,10 @@ export class ChannelRegistry { }); const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); return { status: 'open-failed', channelId: entry.descriptor.id, opening, - // Where the put-back couldn't be recorded, the channel stays selected - // for opening, and the agent is told so, as speech's failure says. - ...(err instanceof ChannelOpenPutBackUnrecordedError ? { staysSelected: true as const } : {}), + // 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 } : {}) }; } } diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index 0a2ecf032..de1db8fea 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -1409,7 +1409,7 @@ test("a reply's failed open whose put-back can't be recorded is still an opening 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, staysSelected: true }); + 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'); @@ -1459,6 +1459,89 @@ test('when speech\'s open fails while the server announces the channel again, th 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.equal(f.isOpen(), true); +}); + +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)] }); @@ -1663,7 +1746,7 @@ test("channel_open and channel_close on a channel already so say what failed whe // bare framework, as in checkpoint-attribution. async function replyOpenFailedNotices(opts: { opening: boolean; directs: boolean; agentName?: string; live?: string[]; homes?: Array<[string, string]>; denied?: string[]; - staysSelected?: true; + selectedForOpening?: 'unrecorded' | 'server-bootstrap' | 'subscription-policy'; primary?: string; }) { const agentName = opts.agentName ?? 'agent'; const told: Array<{ text: string; metadata: unknown; forAgent?: string }> = []; @@ -1684,6 +1767,7 @@ async function replyOpenFailedNotices(opts: { 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 } : {}) }); @@ -1691,7 +1775,7 @@ async function replyOpenFailedNotices(opts: { }; framework.channelRegistry = { openIfClosedForSend: async () => ({ status: 'open-failed', channelId: 'x', label: '#x', opening: opts.opening, - ...(opts.staysSelected ? { staysSelected: true } : {}) }), + ...(opts.selectedForOpening ? { selectedForOpening: opts.selectedForOpening } : {}) }), directsLifecycle: () => opts.directs, }; const error = console.error; @@ -1724,12 +1808,18 @@ test("a reply whose open would have opened the channel, and failed, tells the ag '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, staysSelected: true })).told.map((t) => t.text), [ + 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, and an agent that has ended is told nothing", async () => { +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']] }); @@ -1738,38 +1828,60 @@ test("the reply notice names channel_open and channel_close only to an agent tha 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']], staysSelected: true }); + 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']], staysSelected: true }); + 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'], staysSelected: true }); + 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'], staysSelected: true }); + 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 ended = await replyOpenFailedNotices({ opening: true, directs: true, live: [] }); - assert.deepEqual(ended.told, []); + 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 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; staysSelected?: boolean; + 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', staysSelected: true, + { 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', staysSelected: true, + { 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', @@ -1790,7 +1902,7 @@ test("a reply notice with no agent named, or an empty name, goes out unrouted, w const error = console.error; console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; try { - framework.recordReplyOpenFailedNotice(c.name, { channelId: 'x', label: '#x' }, c.staysSelected === true); + framework.recordReplyOpenFailedNotice(c.name, { channelId: 'x', label: '#x' }, c.selectedForOpening); } finally { console.error = error; } @@ -1799,6 +1911,65 @@ test("a reply notice with no agent named, or an empty name, goes out unrouted, w } }); +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 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')] }); From 42ee16770701d601679923e65c5d753f211d64a5 Mon Sep 17 00:00:00 2001 From: Petra-1012 Date: Mon, 19 Oct 2026 13:23:55 -0400 Subject: [PATCH 5/6] fix(mcpl): the put-back's queued work counts only work that will run, and the reply notice reads the grant before the open - The decision a failed open's put-back leaves is carried out through the queue unless other lifecycle work for the channel will carry it out. Work queued before the server's registrations were removed cancels when it would run, so it no longer counts: lifecycleQueuedRemovals keeps the removal count the newest outstanding operation was queued under. Before, a reconcile queued behind the open, a removal and the channel back from a push left neither the lifted open nor the put-back's close sent. - The framework reads whether the host directs the server's lifecycle before the reply's open. On transport loss the connection clears its grant before it rejects the pending open, so read after the failure it said no, and no reader was told the open failed. The send-engagement test's stub registry gains directsLifecycle, which the path now reads on every send into a channel. - Text: the reply notice's comments, the changelog and the description say when the primary is told (registered; for a reader who can't close, able to close it), rather than stating the rule as a guarantee. Tests: the removal sequence, with and without a policy that lifts the default; a lifted open that fails traced as an open; a grant dropped while the open is out; and notice cases that pin the primary's own advice, a reader who can close, a primary that can't open, and a primary named but not registered. Co-Authored-By: Claude Opus 5.5 --- .../failed-speech-open-puts-back.fixed.md | 8 +- src/framework.ts | 45 ++++++----- src/mcpl/channel-registry.ts | 28 +++++-- test/channel-lifecycle-order.test.ts | 80 ++++++++++++++++++- test/present-while-acting.test.ts | 1 + 5 files changed, 131 insertions(+), 31 deletions(-) diff --git a/changelog.d/failed-speech-open-puts-back.fixed.md b/changelog.d/failed-speech-open-puts-back.fixed.md index 07d186a18..84c4b2ab0 100644 --- a/changelog.d/failed-speech-open-puts-back.fixed.md +++ b/changelog.d/failed-speech-open-puts-back.fixed.md @@ -28,9 +28,11 @@ 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. If the replying agent has ended by then, or can't - close the channel, as a fork away from its home can't, the primary is - told by name. Before, the failure reached only the logs. + 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 diff --git a/src/framework.ts b/src/framework.ts index 26e9d8f5c..c5ff0f974 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -14417,22 +14417,23 @@ export class AgentFramework { * 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, as the opened notice - * tells it of an open by an agent that has ended: no channel opens without - * someone who can close it being told. 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, by the - * same rule. System message, as the opened notice is. + * 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, @@ -14488,8 +14489,9 @@ export class AgentFramework { } // 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 can close - // it: no channel opens without someone who can close it being told. + // 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 && @@ -24982,6 +24984,11 @@ 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, opening, selectedForOpening }) => { @@ -25017,7 +25024,7 @@ export class AgentFramework { // 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 && this.channelRegistry?.directsLifecycle(serverId) === true; + const tell = opening === true && directs; console.error( `[channel] ${agentName}: ${toolName} delivered to ${target} but the open-on-send failed` + (tell ? '; the agent is told' : ''), diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index d600cf7b0..7860e74e9 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -838,6 +838,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(); @@ -2461,6 +2465,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), ); @@ -2605,10 +2610,22 @@ 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. */ + 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 @@ -2979,10 +2996,11 @@ export class ChannelRegistry { } // 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, which - // carries it out anyway. (An incidental open gets here only where the - // host directs the lifecycle: see the grant check above.) - if (current && !this.lifecycleOutstanding.has(key)) { + // 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 directs the lifecycle: see the grant check above.) + if (current && !this.outstandingWorkCarriesOut(entry.serverId, key)) { this.applyThroughQueue(entry.serverId, entry.descriptor.id); } if (selectedAgain) { diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index de1db8fea..a4de09c81 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -1509,6 +1509,52 @@ test("when a reply's open fails while its server is removed and added again as i 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)] }); @@ -1746,7 +1792,7 @@ test("channel_open and channel_close on a channel already so say what failed whe // 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; + selectedForOpening?: 'unrecorded' | 'server-bootstrap' | 'subscription-policy'; primary?: string; dropsDuring?: true; }) { const agentName = opts.agentName ?? 'agent'; const told: Array<{ text: string; metadata: unknown; forAgent?: string }> = []; @@ -1773,10 +1819,16 @@ async function replyOpenFailedNotices(opts: { 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 () => ({ status: 'open-failed', channelId: 'x', label: '#x', opening: opts.opening, - ...(opts.selectedForOpening ? { selectedForOpening: opts.selectedForOpening } : {}) }), - directsLifecycle: () => opts.directs, + 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(' ')); }; @@ -1808,6 +1860,9 @@ test("a reply whose open would have opened the channel, and failed, tells the ag '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"); @@ -1859,6 +1914,14 @@ test("the reply notice names channel_open and channel_close only to an agent tha 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"))); @@ -1933,6 +1996,15 @@ test("where a reply's reader can't close a channel selected for opening, the pri { 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' }] }, diff --git a/test/present-while-acting.test.ts b/test/present-while-acting.test.ts index 7e5d7c8ad..009bc50a3 100644 --- a/test/present-while-acting.test.ts +++ b/test/present-while-acting.test.ts @@ -1524,6 +1524,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[] = []; From dc9a23429939c6e4e3184f3e43c76acbea019ddd Mon Sep 17 00:00:00 2001 From: Petra-1012 Date: Mon, 19 Oct 2026 14:09:59 -0400 Subject: [PATCH 6/6] docs(mcpl): the queued-work check and the put-back's comment say what holds after a removal or a lost grant; tests pin the queue's counts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit From the review of 1cd1d524, and Cordelia-2434's read of this round: - outstandingWorkCarriesOut's doc says it answers for now: a removal after it cancels that work too and takes the channel's entry, so the next registration carries the decision out, where the host directs the lifecycle then (reconcileChannels returns at its §14.1 check otherwise). - The put-back's comment says an incidental open got there where the host directed the lifecycle when it opened, and that where the grant has lost channels.lifecycle since (a transport drop clears it), the queued work fails for it and the next registration carries the decision out, on the same condition. - The queue-release test asserts lifecycleQueuedRemovals empties with the queue, and the re-registration test asserts the one open after the failed open, so the newest operation's count decides. Co-Authored-By: Claude Opus 5.5 --- src/mcpl/channel-registry.ts | 11 +++++++++-- test/channel-lifecycle-order.test.ts | 7 ++++++- 2 files changed, 15 insertions(+), 3 deletions(-) diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 7860e74e9..c703f366d 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -2620,7 +2620,10 @@ export class ChannelRegistry { /** 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. */ + * 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); @@ -2999,7 +3002,11 @@ export class ChannelRegistry { // 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 directs the lifecycle: see the grant check above.) + // 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); } diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index a4de09c81..8318a78c1 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -1106,10 +1106,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 () => { @@ -1506,6 +1509,8 @@ test("when a reply's open fails while its server is removed and added again as i 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); });