diff --git a/changelog.d/210-channel-lifecycle-order.fixed.md b/changelog.d/210-channel-lifecycle-order.fixed.md new file mode 100644 index 000000000..d7d2a1556 --- /dev/null +++ b/changelog.d/210-channel-lifecycle-order.fixed.md @@ -0,0 +1,9 @@ +- Failed backscroll requests preserve the registry's known-open state rather than claiming a close. Automatic delivery joins pending lifecycle work without reopening a channel closed during the wait. A delivery or reply is announced as opening a channel, with the current registration label, only when the channel wasn't already decided open or tuned out; reconfirming the transport of a channel that was isn't announced. Speech engages a channel decided closed as a reply does, even while a failed close has left its transport open, wherever the server's grant lets the host open it. When `channel_open` fails for a channel that isn't open and is still decided open, its error says the channel stays selected for opening, since that decision outlives the failed round-trip and a later delivery or reconnect tries again. + +- Channel lifecycle operations now run in order per channel and reconcile the latest desired state and registration after an in-flight operation completes. Successful reconciliation confirms the current target and desired state at its receipt boundary; a superseded tool request reports that its requested state no longer applies. Automatic delivery preserves active tune-out attention state while opening or awaiting transport. A registration batch reconciles its channels side by side, so one held RPC doesn't delay its siblings. Removing a server cancels its queued lifecycle operations, and the framework removes its channel entries before closing its connection, so nothing queued runs against a closing connection or a later re-registration. + +- A channel's transport state is known only from a receipt for its current registration, connection, type, and address. A channel a registration creates or replaces, a retarget, a substitution, or a failed open or close leaves it unconfirmed: failed corrective opens fail joined delivery, while a same-target backscroll failure keeps a channel that was open usable. Incoming traffic confirms nothing unconfirmed (after a retarget it may still come from the old subscription), and it marks a channel open only while that channel is decided open or tuned out, so a close receipt stands. Explicit close, and declining an invitation, close an unconfirmed target for real instead of reporting it already closed. A retarget of a channel decided open or tuned out opens the new target. + +- Lifecycle convergence makes at most five transport attempts per operation. Continued descriptor/intent supersession fails with a diagnostic, releases the queue, and gives joined speech or reply preparation a visible failure instead of hanging. Later stable operations can retry. + +- A `channels/open` answered for a different channel than the one requested, by the answer's `channel.id` or a history item's `channelId`, compared exactly as the server's traffic is, fails the open. The requested channel isn't marked open (its open state becomes unconfirmed), nothing in the answer is used, its history included, and the host logs `[channel-open-substituted]` and emits an `mcpl:channel-open-substituted` trace, whose `answeredFor` is the answered id as the log line prints it, escaped by `logValue`. When the channel the server opened instead is registered and its desired state is closed, the host closes it again, so its traffic doesn't reach the resident. A substituted answer on a receipt already made stale by retargeting is retried like any stale receipt. An answer that names no channel, by a null id or none, is accepted as before; one whose id isn't a string fails the open the same way, with nothing to close. Either rollout order is safe: on a server that still falls back, the resident sees a failed open for an id that server no longer knows (a deleted channel), rather than another channel's traffic. `ChannelsOpenParams.channelId` is documented as exact: a server that doesn't know the id should refuse. diff --git a/changelog.d/210-policy-admission-notice.breaking.md b/changelog.d/210-policy-admission-notice.breaking.md new file mode 100644 index 000000000..eb209f3ff --- /dev/null +++ b/changelog.d/210-policy-admission-notice.breaking.md @@ -0,0 +1 @@ +- `onChannelAutoOpened` with `source: 'subscription-policy'` now reports a durable admission decision before transport reconciliation. Treat it as intent; await the registration call and inspect `getOpenChannels()` when confirmed live state is needed. `opened-by-delivery` reports a delivery that opened a channel not already decided open or tuned out, once its transport is confirmed; a delivery that only reconfirms the transport of a channel that was isn't reported. The built-in host notice now distinguishes selected-for-opening from currently open. diff --git a/src/framework.ts b/src/framework.ts index 68a607d54..edb424eee 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -14305,13 +14305,15 @@ export class AgentFramework { } /** - * Durable window notice for a channel the agent did not ask to open — - * subscription-policy admission, or an open forced by the agent's own - * delivery into a closed channel. The agent must always learn that the - * channel opened, and that `channel_close` opts out (their decision - * outranks policy). What an open channel delivers is its server's to say - * (#406), so the notice doesn't promise its traffic. System message: never - * triggers inference, never conversational, never locus-eligible. + * Durable window notice for a channel the agent did not ask to open: a + * subscription-policy admission, announced as the decision before its + * transport RPC (a failed open keeps the decision, and channel_list shows + * transport state), or an open that the agent's own delivery or reply + * caused (`opened-by-delivery`, `opened-by-reply`), once its transport is + * confirmed. Either way it names `channel_close` as the opt-out (their + * decision outranks policy). What an open channel delivers is its server's + * to say (#406), so the notice doesn't promise its traffic. System message: + * never triggers inference, never conversational, never locus-eligible. */ private recordChannelAutoOpenNotice( agentName: string | undefined, @@ -14324,8 +14326,7 @@ export class AgentFramework { .join(', '); const plural = channels.length > 1; const optOut = ' What an open channel delivers to you depends on its server. ' + - `If you don't want ${plural ? 'one of them' : 'it'}, use channel_close — ` + - 'your choice sticks; neither policy nor a later send will silently reopen it.'; + `If you don't want ${plural ? 'one of them' : 'it'}, use channel_close; your choice outranks policy.`; // An open caused by an agent's own send is that agent's to hear about, in // the window it reads; a policy admission with no agent behind it goes to // the primary. An agent that has ended (a run whose final reply is routed @@ -14342,15 +14343,14 @@ export class AgentFramework { console.error(`[channel] ${agentName} has ended, and there is no primary to tell that ${shown} opened`); return; } - text = `[channels] Now open: ${shown} — opened by ${agentName}'s reply; ${agentName} has ended.` + optOut; + text = `[channels] Now open: ${shown}: opened by ${agentName}'s reply; ${agentName} has ended.` + optOut; metadata = { system: true, kind: 'channel-notice', speaker: agentName } as MessageMetadata; told = `the primary, by name (${agentName} has ended)`; } else { - const causeText = - cause === 'subscription-policy' - ? 'your subscription policy just admitted ' + (plural ? 'them' : 'it') - : 'your own message into the closed channel engaged it'; - text = `[channels] Now open: ${shown} — ${causeText}.` + optOut; + text = (cause === 'subscription-policy' + ? `[channels] Selected for opening by subscription policy: ${shown}. ` + + 'This records the admission decision; channel_list shows current transport state.' + : `[channels] Now open: ${shown}: your own message into the closed channel opened it.`) + optOut; if (agentName) target = { forAgent: agentName }; told = agentName ?? 'primary'; } @@ -22706,19 +22706,21 @@ export class AgentFramework { } /** - * Disconnect an MCPL server at runtime: close the connection, destroy its - * feature-set and checkpoint state (permanent removal — unlike a transient - * transport close, which preserves checkpoints for the reconnect), remove - * its channels from the registry, drop routing entries, and refresh tools. + * Disconnect an MCPL server at runtime: remove its channels from the + * registry (first, cancelling their queued lifecycle work), close the + * connection, destroy its feature-set and checkpoint state (permanent + * removal — unlike a transient transport close, which preserves checkpoints + * for the reconnect), drop routing entries, and refresh tools. * No-op-ish if the server is not connected (still clears routing state). * * It holds the id throughout (holdMcpServerId): a connect for it is * refused meanwhile, and a connect already in flight is ended (closed, or * displaced when its legacy handshake finishes) and settles before - * anything is removed. A close that can't confirm a child exited fails the - * disconnect before anything is removed: the server stays registered, - * closed, so nothing launches beside a child that may still be running, - * and disconnecting again retries the reap. + * anything but the channel entries the disconnect found when it began is + * removed. A close that can't confirm a child exited fails the disconnect + * before anything but those entries is removed: the server stays + * registered, closed, so nothing launches beside a child that may still be + * running, and disconnecting again retries the reap. */ async disconnectMcplServer(id: string): Promise { this.refuseOperatorAfterStop('disconnect MCPL server'); @@ -22736,6 +22738,9 @@ export class AgentFramework { const registry = this.mcplServerRegistry; const hold = this.holdMcpServerId(id, 'disconnect'); try { + // Channel entries go first: removing them cancels the server's queued + // lifecycle work, which must not run against the closing connection. + this.channelRegistry?.removeServer(id); // Close what the id has now, so a connect in flight ends rather than // being waited out: a modern connect is ended by its close, a legacy // one still in its handshake (or a WebSocket dial waiting for its @@ -22769,6 +22774,8 @@ export class AgentFramework { ); } if (modern && this.modernServers().get(id) === modern) this.modernServers().delete(id); + // Again, for the channels a connect that was still running registered + // after the first removal, and any lifecycle work they queued. this.channelRegistry?.removeServer(id); // Permanent removal: also destroy feature-set and checkpoint state @@ -24356,9 +24363,11 @@ export class AgentFramework { // Recorded here — not eagerly — because injection metadata // carries the CANONICAL channel id (descriptor form), while // send args often carry the raw provider id; this callback - // has the resolved form. The already-open path resolves on + // has the resolved form. A channel confirmed open resolves on // the microtask queue, i.e. before the tool-result event - // that evaluates injections. + // that evaluates injections. One decided open whose transport + // is unconfirmed is opened again first, so its engagement is + // recorded when that open returns. if (status !== 'unknown-channel' && status !== 'ambiguous') { sendingTurn?.engaged.add(channelId ?? target); } @@ -24373,7 +24382,8 @@ export class AgentFramework { ); } else if (status === 'open-failed') { console.error( - `[channel] ${agentName}: ${toolName} delivered to ${target} but the open-on-send failed — channel remains half-alive, will reconcile`, + `[channel] ${agentName}: ${toolName} delivered to ${target} but the open-on-send failed; ` + + 'a later registration or delivery tries the open again', ); } }) diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index 067fce3bf..521e4d50e 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -16,6 +16,7 @@ import type { ContentBlock } from '@animalabs/membrane'; import { convertBlock } from './convert-block.js'; import type { JsStore } from '@animalabs/chronicle'; +import { isDeepStrictEqual } from 'node:util'; import { ensureStateRegistered } from '../module-registry.js'; import type { @@ -51,8 +52,42 @@ import { readSkipReplyWake } from './skip-reply-wake.js'; // ============================================================================ const TYPING_INTERVAL_MS = 7_000; +const MAX_CHANNEL_LIFECYCLE_ATTEMPTS = 5; + +/** A successful RPC can still fail to establish the latest channel target. */ +class ChannelLifecycleConvergenceError extends Error {} +/** A channels/open answered for a channel other than the one it named. */ +class ChannelOpenSubstitutionError extends Error {} +/** An operation queued for a server's registrations that were removed. */ +class ChannelLifecycleCancelledError extends Error {} const CHANNEL_LIFECYCLE_LOG_ID = 'mcpl/channel-lifecycle'; +/** + * 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 + * evidence. A string other than the requested id names another channel (ids + * compare exactly, as the server's traffic is matched). Null or absence names + * no channel, and gives undefined, as does an answer that names none at all. + * Any other value is malformed: it names nothing the host could register. + */ +function substitutedChannel( + result: ChannelsOpenResult | undefined, + requested: string, +): { kind: 'other'; id: string } | { kind: 'malformed'; value: unknown } | undefined { + const named = (value: unknown) => { + if (value === undefined || value === null || value === requested) return undefined; + return typeof value === 'string' ? { kind: 'other' as const, id: value } : { kind: 'malformed' as const, value }; + }; + const answered = named(result?.channel?.id); + if (answered) return answered; + const history: unknown[] = Array.isArray(result?.history) ? result.history : []; + for (const item of history) { + const from = named((item as { channelId?: unknown } | null)?.channelId); + if (from) return from; + } + return undefined; +} + /** * Durable "which label did this channelId have, each time we saw it" log. * `resolveProseTarget()` resolves a label to an id only for channels @@ -213,9 +248,14 @@ function shallowEqualRecord( interface ChannelEntry { serverId: string; descriptor: ChannelDescriptor; + /** Whether the current target is confirmed open; false can be unconfirmed. */ open: boolean; } +function sameChannelTarget(a: ChannelDescriptor, b: ChannelDescriptor): boolean { + return a.type === b.type && isDeepStrictEqual(a.address, b.address); +} + /** Minimal responder interface for sending JSON-RPC results back. */ interface Responder { respond(result: unknown): void; @@ -568,14 +608,15 @@ interface ChannelRegistryOptions { detail?: string; }) => void; /** - * Called when channels were opened WITHOUT the agent asking (subscription - * policy admitting a newly discovered channel, or delivery into a closed - * locus). The host wires this to drop a durable notice into the agent's - * window — the agent must always learn that the channel opened, and that - * `channel_close` opts out (their decision outranks policy). What an open - * channel delivers is the server's to say (#406). Fires ONCE per channel - * ever: the desired-state decision is durable, so reboots do not - * re-announce. Must not trigger inference. + * Announces either a durable subscription-policy admission (before its + * transport RPC; not an open confirmation), or a delivery that turned a + * closed channel open, once its transport is confirmed. Reconfirming the + * transport of a channel already decided open or tuned out isn't one. + * The host records a durable, non-waking notice. What an open channel + * delivers is the server's to say (#406). + * Policy admissions fire once per channel: the persisted desired-state + * decision prevents re-announcement on retry or reboot. channel_close + * supersedes policy; a fresh delivery into a closed locus engages it again. */ onChannelAutoOpened?: (info: { /** Agent whose action caused the open (delivery); absent for policy opens. */ @@ -744,6 +785,46 @@ export class ChannelRegistry { /** One-time migration inputs from the retired recipe auto-open policy. */ private legacyPolicies = new Map(); private migratedLegacyPolicies = new Set(); + + // Keep one transport operation in flight per logical channel, including + // across descriptor replacement. Settled tails release themselves; a + // failed operation must not poison later decisions. + private channelLifecycleTails = new Map>(); + // Bumped when a server's registrations are removed: an operation queued + // before that is cancelled when it would run, rather than acting on a + // closing connection or on a later re-registration's entries. + private serverRemovals = new Map(); + // Operations queued or running per channel that will still read its + // 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(); + // Losing an open receipt does not confirm closure. Keep that distinction + // private while preserving the public open:boolean availability projection. + private unconfirmedChannelTargets = new WeakSet(); + + private invalidateTransportConfirmation(entry: ChannelEntry): void { + entry.open = false; + this.unconfirmedChannelTargets.add(entry); + } + + /** 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 { + return CapabilityGrant.of(this.serverRegistry.getServer(serverId)).has('channels.lifecycle'); + } + + /** A registration's entry: nothing about its transport is known until a + * receipt for its target confirms it open or closed. */ + private registerEntry(key: string, serverId: string, descriptor: ChannelDescriptor): void { + const entry: ChannelEntry = { serverId, descriptor, open: false }; + this.channels.set(key, entry); + this.unconfirmedChannelTargets.add(entry); + } + + private hasPendingLifecycle(entry: ChannelEntry): boolean { + return this.channelLifecycleTails.has(this.lifecycleKey(entry.serverId, entry.descriptor.id)); + } private handleCoalescedIncoming?: ChannelRegistryOptions['handleCoalescedIncoming']; private acceptInbound?: ChannelRegistryOptions['acceptInbound']; private channelTools: ToolDefinition[]; @@ -834,11 +915,7 @@ export class ChannelRegistry { continue; } const key = `${serverId}:${channel.id}`; - this.channels.set(key, { - serverId, - descriptor: channel, - open: false, - }); + this.registerEntry(key, serverId, channel); this.appendLabelSighting(channel.id, channel.label, extractDmMeta(channel.metadata)); registeredIds.push(channel.id); results.push({ id: channel.id, accepted: true }); @@ -879,6 +956,7 @@ export class ChannelRegistry { // channels/register — a changed-set cannot smuggle in what register // would have rejected (PR #79 review blocker 5). const addedResults: Array<{ id: string; accepted: boolean; reason?: string }> = []; + const retargeted: ChannelDescriptor[] = []; // Process removed channels — itemized truthfully: removal of a channel // we never had is reported, not silently accepted. if (params.removed) { @@ -908,6 +986,21 @@ export class ChannelRegistry { this.emitTraceFn({ type: 'mcpl:channel-descriptor-rejected', serverId, channelId: channel.id, reason: 'update for unregistered channel' }); continue; } + // A live subscription to the old target does not confirm the new + // type/address. Label-only updates preserve that confirmation. A + // channel decided open (or tuned out) is opened at its new target. + if (!sameChannelTarget(existing.descriptor, channel)) { + this.invalidateTransportConfirmation(existing); + // An operation queued or running for this channel reads its target + // again before it confirms anything, and after a failed RPC, so it + // opens the new one. With none, a channel decided open (or tuned + // out) is reconciled here. + const desired = this.getDesiredState(serverId, channel.id); + if ((desired === 'open' || desired === 'tuned-out') && + !this.lifecycleOutstanding.has(this.lifecycleKey(serverId, channel.id))) { + retargeted.push(channel); + } + } existing.descriptor = channel; this.appendLabelSighting(channel.id, channel.label, extractDmMeta(channel.metadata)); addedResults.push({ id: channel.id, accepted: true }); @@ -925,11 +1018,7 @@ export class ChannelRegistry { continue; } const key = `${serverId}:${channel.id}`; - this.channels.set(key, { - serverId, - descriptor: channel, - open: false, - }); + this.registerEntry(key, serverId, channel); this.appendLabelSighting(channel.id, channel.label, extractDmMeta(channel.metadata)); addedResults.push({ id: channel.id, accepted: true }); accepted.push(channel); @@ -944,7 +1033,8 @@ export class ChannelRegistry { // settled above, so nothing in the response depends on reconciling. responder?.respond({ results: addedResults }); - if (accepted.length > 0) await this.reconcileChannels(serverId, accepted); + const reconcile = [...accepted, ...retargeted]; + if (reconcile.length > 0) await this.reconcileChannels(serverId, reconcile); this.emitTraceFn({ type: 'mcpl:channels-changed', @@ -1128,15 +1218,31 @@ export class ChannelRegistry { // exactly the authority a self-attested channel would be stealing). // A coalesced item is "accepted" only once the coalescer admits it, so // for those this runs after the hook (below). + const incomingEntry = this.channels.get(incomingKey)!; const markAccepted = () => { this.defaultPublishChannel = message.channelId; this.defaultPublishMessageId = message.messageId; this.defaultPublishThreadId = message.threadId; - // A server sending channels/incoming is authoritative evidence that - // the transport is actually open. This repairs transient status only; - // durable desired state still changes exclusively through lifecycle - // operations. - this.channels.get(incomingKey)!.open = true; + // Where the host directs this server's lifecycle, only a receipt for + // the current target confirms it: traffic shows the transport open + // only for a confirmed target, and only for a channel still decided + // open or tuned out. After a retarget, traffic for this id may still + // come from the old subscription, and a close receipt that landed + // first (or during the coalescer's await) stands. Where the host + // doesn't direct it, the server opens its own channels and its + // traffic is the only evidence the host gets, so it marks the + // channel open, as it always has. The entry is the one the message + // was accepted for: one a re-registration put in its place meanwhile + // waits for its own evidence. Durable desired state still changes + // exclusively through lifecycle operations. + if (this.directsLifecycle(serverId)) { + if (this.unconfirmedChannelTargets.has(incomingEntry)) return; + const desired = this.getDesiredState(serverId, message.channelId); + if (desired !== 'open' && desired !== 'tuned-out') return; + } else { + this.unconfirmedChannelTargets.delete(incomingEntry); + } + incomingEntry.open = true; }; if (!coalesced) markAccepted(); @@ -1606,6 +1712,10 @@ export class ChannelRegistry { * server's channels don't linger and route speech into the void. */ removeServer(serverId: string): void { + // Every lifecycle operation queued for this server is cancelled when it + // would run (ChannelLifecycleCancelledError), so none acts on a closing + // connection or on a later re-registration's entries. + this.serverRemovals.set(serverId, (this.serverRemovals.get(serverId) ?? 0) + 1); for (const [key, entry] of this.channels) { if (entry.serverId !== serverId) continue; this.channels.delete(key); @@ -2277,6 +2387,225 @@ export class ChannelRegistry { return channel.initiallyOpen !== true && byPolicy; } + /** + * Apply current durable intent, serializing the transport rather than the + * server's registration ACK. Re-read after every await: registration, + * descriptor updates, or a newer decision may have replaced our snapshot. + */ + private applyDesiredChannelState( + serverId: string, + channelId: string, + options?: { history?: ChannelHistoryRequest }, + ): Promise<{ open: boolean; result?: ChannelsOpenResult }> { + const lifecycleKey = this.lifecycleKey(serverId, channelId); + 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); + const operation = previous.catch(() => {}).then( + () => this.runLifecycleOperation(serverId, channelId, removals, options), + ); + // Keep the rejection visible to existing joiners, while observing it + // immediately and letting each queued successor start after either result. + const tail = operation.then(() => {}); + this.channelLifecycleTails.set(lifecycleKey, tail); + const cleanup = () => { + if (this.channelLifecycleTails.get(lifecycleKey) === tail) { + this.channelLifecycleTails.delete(lifecycleKey); + } + }; + void tail.then(cleanup, cleanup); + return operation; + } + + /** One queued lifecycle operation, run when its turn in the queue comes. */ + private async runLifecycleOperation( + serverId: string, + channelId: string, + removals: number, + options?: { history?: ChannelHistoryRequest }, + ): Promise<{ open: boolean; result?: ChannelsOpenResult }> { + const channelKey = `${serverId}:${channelId}`; + const lifecycleKey = this.lifecycleKey(serverId, channelId); + try { + let attempts = 0; + for (;;) { + if ((this.serverRemovals.get(serverId) ?? 0) !== removals) { + throw new ChannelLifecycleCancelledError( + `Channel ${channelId}: its server's registrations were removed while this operation was queued`, + ); + } + const entry = this.channels.get(channelKey); + if (!entry) throw new Error(`Channel is no longer registered: ${channelId}`); + const server = this.serverRegistry.getServer(serverId); + if (!server) throw new Error(`Server not found: ${serverId}`); + if (!CapabilityGrant.of(server).has('channels.lifecycle')) { + throw new Error(`channels.lifecycle not in "${serverId}"'s effective grant (§14.1)`); + } + const descriptor = entry.descriptor; + const desired = this.getDesiredState(serverId, channelId); + const open = desired === 'open' || desired === 'tuned-out'; + // Per-RPC timeouts do not bound a stream of promptly successful but + // continually superseded receipts. Bound this whole operation, then + // release its queue slot and surface the failure to delivery joiners. + // Its open flag needs no change here: every retry follows a replaced + // registration, a retarget (already invalidated, whether the old + // target's RPC was answered or failed), or a receipt for this very + // target. + if (attempts >= MAX_CHANNEL_LIFECYCLE_ATTEMPTS) { + const error = new ChannelLifecycleConvergenceError( + `Channel ${channelId} did not converge after ${attempts} lifecycle attempts; its target or desired state kept changing`, + ); + this.emitTraceFn({ + type: 'mcpl:channel-reconcile-failed', + serverId, channelId, desired, attempts, error: error.message, + }); + throw error; + } + attempts++; + const sameTarget = () => this.serverRegistry.getServer(serverId) === server && + sameChannelTarget(entry.descriptor, descriptor); + let result: ChannelsOpenResult | undefined; + try { + if (open) { + result = await server.sendChannelsOpen({ + channelId, type: descriptor.type, address: descriptor.address, + ...(options?.history ? { history: options.history } : {}), + }); + } else { + await server.sendChannelsClose({ channelId }); + } + } catch (error) { + // An RPC can fail after the server acted, so a failure for the + // current target leaves its outcome unknown: it is marked + // unconfirmed with open unchanged, which keeps a confirmed-open + // channel usable after a failed backscroll (the mark is read only + // while not open). A failure from an older connection or target + // also withdraws the open flag. + if (this.channels.get(channelKey) === entry) { + if (sameTarget()) this.unconfirmedChannelTargets.add(entry); + else this.invalidateTransportConfirmation(entry); + } + // A retarget while this RPC was out queued nothing for the new + // target, since handleChanged leaves that to this operation. So a + // failure for the old target says nothing about the new one, and a + // channel still decided open or tuned out tries its new target, + // within the attempt limit. + if (this.channels.get(channelKey) === entry && !sameChannelTarget(entry.descriptor, descriptor)) { + const now = this.getDesiredState(serverId, channelId); + if (now === 'open' || now === 'tuned-out') continue; + } + throw error; + } + // A server can answer for a channel other than the one named: one + // that falls back to its first channel of the type for an id it no + // longer knows opens that channel and returns its history. Whatever + // becomes of this receipt, the server did that: say so, and undo the + // open it made if the host wants that channel closed. + const substitute = open ? substitutedChannel(result, channelId) : undefined; + if (substitute !== undefined) { + // Each id prints through logValue, as the rejection lines' ids do, + // so an odd one can't add a line or a field. + const answered = logValue(substitute.kind === 'other' ? substitute.id : substitute.value); + console.error( + `[channel-open-substituted] server=${serverId} requested=${logValue(channelId)} answered=${answered}`, + ); + this.emitTraceFn({ type: 'mcpl:channel-open-substituted', serverId, channelId, answeredFor: answered }); + // A malformed id names nothing registered here, so there is + // nothing to close. + if (substitute.kind === 'other') this.closeSubstitutedChannel(serverId, substitute.id); + } + // Never apply an old receipt to the current target, even transiently: + // a corrective open might fail before that stale bit is replaced. + if (this.channels.get(channelKey) !== entry) continue; + if (!sameTarget()) { + this.invalidateTransportConfirmation(entry); + continue; + } + // A current receipt for another channel confirms nothing about this + // one, and its history is another channel's. So the open fails, + // nothing in the answer is used, and this channel's open state is + // unconfirmed: a server that doesn't resolve this id may not resolve + // its sends either. + if (substitute !== undefined) { + this.invalidateTransportConfirmation(entry); + throw new ChannelOpenSubstitutionError( + `Server "${serverId}" answered channels/open for ${channelId} with ` + + (substitute.kind === 'other' ? 'a different channel' : 'a channel id that is not a string') + + `; ${channelId} is not marked open, and nothing in the answer is used`, + ); + } + entry.open = open; + this.unconfirmedChannelTargets.delete(entry); + const now = this.getDesiredState(serverId, channelId); + if (open !== (now === 'open' || now === 'tuned-out')) continue; + return { open, result }; + } + } finally { + // Synchronous with the operation's last look at its target: from here + // 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); + } + } + + /** + * 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 + * through its own lifecycle queue: otherwise its traffic would reach the + * resident, marked open by its first incoming message. A channel the host + * wants open is left alone, and a close can't be answered for another + * channel, so this never starts another substitution. An id the host + * doesn't know needs nothing: §14.5 refuses its traffic. + */ + private closeSubstitutedChannel(serverId: string, channelId: string): void { + if (!this.channels.has(`${serverId}:${channelId}`)) return; + if (this.getDesiredState(serverId, channelId) !== 'closed') return; + this.applyDesiredChannelState(serverId, channelId).catch((err) => { + // Removal while queued is cancellation, as in reconcileChannels, and + // exhausted convergence emitted its own diagnostic. + if (!this.channels.has(`${serverId}:${channelId}`) || + err instanceof ChannelLifecycleCancelledError || + err instanceof ChannelLifecycleConvergenceError) return; + this.emitTraceFn({ + type: 'mcpl:channel-reconcile-failed', + serverId, channelId, desired: 'closed', error: (err as Error).message, + }); + }); + } + + /** + * Join outstanding lifecycle work without recording new intent. Retargeting + * may have invalidated open confirmation while the decision remains pending. + * A close that settles during the wait remains authoritative; waiting for + * it is not an instruction to open the channel again. + */ + private async waitForOpenChannel(entry: ChannelEntry): Promise { + const lifecycleKey = this.lifecycleKey(entry.serverId, entry.descriptor.id); + for (;;) { + const pending = this.channelLifecycleTails.get(lifecycleKey); + if (!pending) break; + try { + await pending; + } catch (error) { + // A redundant backscroll failure can leave a known-open channel + // usable. Exhausted convergence is different: it never established + // the current target, so joined delivery must fail visibly. + if (error instanceof ChannelLifecycleConvergenceError) throw error; + } + // A settled tail releases its own key. One still current after it + // settled would be awaited again without ever yielding to the event + // loop, so stop waiting rather than spin. + if (this.channelLifecycleTails.get(lifecycleKey) === pending) break; + } + const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); + if (!current) throw new Error('Channel registration was removed while delivery waited'); + if (!current.open || this.getDesiredState(current.serverId, current.descriptor.id) === 'closed') { + throw new Error('Channel is not confirmed open for delivery, or was selected for closure while delivery waited'); + } + return current; + } + private async reconcileChannels( serverId: string, channels: ChannelDescriptor[], @@ -2302,54 +2631,18 @@ export class ChannelRegistry { // first boot on a new policy doesn't produce one notice per channel. const policyOpened: Array<{ channelId: string; label?: string }> = []; + // Record all admissions before waiting on any channel. A slow first RPC + // must not delay intent for the rest of this registration batch. for (const channel of channels) { - if (this.ensureInitialDesiredState(serverId, channel)) { - policyOpened.push({ channelId: channel.id, label: channel.label }); - } - const key = `${serverId}:${channel.id}`; - const desired = this.getDesiredState(serverId, channel.id); - const entry = this.channels.get(key); - // Tuned-out sits on the OPEN side of reconcile: traffic must keep - // arriving (the subconscious reads it); only main's wake/visibility - // is diverted, downstream at ingestion. - if (desired !== 'open' && desired !== 'tuned-out') { - try { - await server.sendChannelsClose({ channelId: channel.id }); - if (entry) entry.open = false; - } catch (err) { - this.emitTraceFn({ - type: 'mcpl:channel-reconcile-failed', - serverId, - channelId: channel.id, - desired, - error: (err as Error).message, - }); - } - continue; - } - - try { - await server.sendChannelsOpen({ - channelId: channel.id, - type: channel.type, - address: channel.address, - }); - if (entry) entry.open = true; - } catch (err) { - if (entry) entry.open = false; - this.emitTraceFn({ - type: 'mcpl:channel-reconcile-failed', - serverId, - channelId: channel.id, - desired, - error: (err as Error).message, - }); + const entry = this.channels.get(`${serverId}:${channel.id}`); + if (entry && this.ensureInitialDesiredState(serverId, entry.descriptor)) { + policyOpened.push({ channelId: channel.id, label: entry.descriptor.label }); } } - // Announce policy admissions to the agent — no channel may open without - // them being told, and told how to opt out (channel_close; their - // decision outranks policy). + // Announce the durable admission decision before awaiting transport. + // This is intent, not confirmation that an open RPC succeeded. It stays + // truthful if the first attempt fails, and a retry needs no new admission. if (policyOpened.length > 0) { try { this.onChannelAutoOpened?.({ @@ -2361,6 +2654,30 @@ export class ChannelRegistry { console.error('onChannelAutoOpened (policy) failed:', err); } } + + // Each channel has its own lifecycle queue, so a held RPC for one (up to + // its timeout, times its attempts) doesn't delay its siblings. + await Promise.allSettled(channels.map(async (channel) => { + if (!this.channels.has(`${serverId}:${channel.id}`)) return; + try { + await this.applyDesiredChannelState(serverId, channel.id); + } catch (err) { + // Removal while queued is cancellation, not a failed subscription. + // Exhausted convergence and a substituted answer already emitted + // their own diagnostics. + if (!this.channels.has(`${serverId}:${channel.id}`) || + err instanceof ChannelLifecycleCancelledError || + err instanceof ChannelLifecycleConvergenceError || + err instanceof ChannelOpenSubstitutionError) return; + this.emitTraceFn({ + type: 'mcpl:channel-reconcile-failed', + serverId, + channelId: channel.id, + desired: this.getDesiredState(serverId, channel.id), + error: (err as Error).message, + }); + } + })); } // ========================================================================== @@ -2482,37 +2799,39 @@ export class ChannelRegistry { } /** - * The single open executor: record durable desired-open intent, tell the - * server to subscribe, mark the live entry open. Every path that opens a - * channel — the agent's channel_open tool, delivery into a closed locus, - * an explicit send into a closed channel — funnels here, so lifecycle - * events, desired state, and live state can never diverge by path. + * Record durable desired-open intent, then queue the channel's lifecycle + * work (applyDesiredChannelState), which tells the server to subscribe and + * marks the live entry open on a receipt for its current target. Every + * open the agent or its speech asks for — the channel_open tool, delivery + * 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. + * + * `opened` says whether this call turned a closed channel open, which is + * what the delivery and reply notices tell the resident. Reconfirming the + * transport of a channel already decided open or tuned out isn't one. */ private async openChannelNow( entry: ChannelEntry, - source: string, + source: 'agent-tool' | 'opened-by-reply' | 'opened-by-delivery', history?: ChannelHistoryRequest, - ): Promise { - this.setDesiredState(entry.serverId, entry.descriptor.id, 'open', source); - const server = this.serverRegistry.getServer(entry.serverId); - if (!server) { - throw new Error(`Server not found: ${entry.serverId}`); + ): Promise<{ result: ChannelsOpenResult; opened: boolean }> { + const transportOnly = source !== 'agent-tool'; + 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); } - if (!CapabilityGrant.of(server).has('channels.lifecycle')) { - // Desired state is already recorded — intent sticks; reconciliation - // will retry if the grant later widens (§6.7 expansion-on-receipt). - throw new Error(`channels.lifecycle not in "${entry.serverId}"'s effective grant (§14.1)`); + 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'); } - const result: ChannelsOpenResult = await server.sendChannelsOpen({ - channelId: entry.descriptor.id, - type: entry.descriptor.type, - address: entry.descriptor.address, - ...(history ? { history } : {}), - }); - entry.open = true; - return result; + return { result: applied.result!, opened }; } /** @@ -2723,19 +3042,25 @@ export class ChannelRegistry { } if (matches.length === 0) return { status: 'unknown-channel' }; if (matches.length > 1) return { status: 'ambiguous' }; - const entry = matches[0]!; - const resolved = { channelId: entry.descriptor.id, label: entry.descriptor.label }; - if (entry.open && this.getDesiredState(entry.serverId, entry.descriptor.id) === 'open') { - return { status: 'already-open', ...resolved }; - } + let entry = matches[0]!; try { - await this.openChannelNow(entry, 'opened-by-reply'); - this.emitTraceFn({ - type: 'mcpl:channel-opened-by-send', - serverId: entry.serverId, - channelId: entry.descriptor.id, - }); - return { status: 'opened', ...resolved }; + if (this.hasPendingLifecycle(entry)) { + entry = await this.waitForOpenChannel(entry); + } + const desired = this.getDesiredState(entry.serverId, entry.descriptor.id); + if (entry.open && (desired === 'open' || desired === 'tuned-out')) { + return { status: 'already-open', channelId: entry.descriptor.id, label: entry.descriptor.label }; + } + const { opened } = await this.openChannelNow(entry, 'opened-by-reply'); + entry = await this.waitForOpenChannel(entry); + if (opened) { + this.emitTraceFn({ + type: 'mcpl:channel-opened-by-send', + serverId: entry.serverId, + channelId: entry.descriptor.id, + }); + } + return { status: opened ? 'opened' : 'already-open', channelId: entry.descriptor.id, label: entry.descriptor.label }; } catch (err) { this.emitTraceFn({ type: 'mcpl:channel-open-failed', @@ -2743,7 +3068,9 @@ export class ChannelRegistry { channelId: entry.descriptor.id, error: (err as Error).message, }); - return { status: 'open-failed', ...resolved }; + const current = this.channels.get(`${entry.serverId}:${entry.descriptor.id}`); + return { status: 'open-failed', channelId: entry.descriptor.id, + ...(current ? { label: current.descriptor.label } : {}) }; } } @@ -2782,7 +3109,7 @@ export class ChannelRegistry { } const alreadyDesiredOpen = this.getDesiredState(entry.serverId, input.channelId) === 'open'; - if (entry.open && alreadyDesiredOpen && !input.backscroll) { + if (entry.open && alreadyDesiredOpen && !input.backscroll && !this.hasPendingLifecycle(entry)) { return { success: true, data: { channelId: input.channelId, status: 'already open' }, @@ -2796,7 +3123,7 @@ export class ChannelRegistry { ...(input.beforeMessageId ? { beforeMessageId: input.beforeMessageId } : {}), } : undefined; - const result = await this.openChannelNow(entry, 'agent-tool', history); + const { result } = await this.openChannelNow(entry, 'agent-tool', history); return { success: true, data: { @@ -2807,9 +3134,18 @@ export class ChannelRegistry { }, }; } catch (err) { + // The decision was recorded before the round-trip and outlives its + // failure. Say so: otherwise a later delivery that opens the channel + // would be the resident's first sign that it was still wanted open. + const current = this.channels.get(`${entry.serverId}:${input.channelId}`); + const stillSelected = current !== undefined && !current.open && + this.getDesiredState(entry.serverId, input.channelId) === 'open'; return { success: false, - error: `Failed to open channel: ${(err as Error).message}`, + error: `Failed to open channel: ${(err as Error).message}` + + (stillSelected + ? ' (it stays selected for opening: a later delivery or reconnect tries again, and channel_close withdraws that)' + : ''), isError: true, }; } @@ -2845,7 +3181,8 @@ export class ChannelRegistry { } const alreadyDesiredClosed = this.getDesiredState(entry.serverId, input.channelId) === 'closed'; - if (!entry.open && alreadyDesiredClosed) { + if (!entry.open && alreadyDesiredClosed && !this.hasPendingLifecycle(entry) && + !this.unconfirmedChannelTargets.has(entry)) { return { success: true, data: { channelId: input.channelId, status: 'already closed' }, @@ -2914,8 +3251,10 @@ export class ChannelRegistry { isError: true, }; } - await server.sendChannelsClose({ channelId: input.channelId }); - entry.open = false; + const applied = await this.applyDesiredChannelState(entry.serverId, input.channelId); + if (applied.open || this.getDesiredState(entry.serverId, input.channelId) !== 'closed') { + throw new Error('Channel close was superseded by a newer lifecycle decision'); + } return { success: true, data: { channelId: input.channelId, status: 'closed' }, @@ -2981,6 +3320,23 @@ export class ChannelRegistry { } } + // Where the host directs the lifecycle, a decline goes through the + // lifecycle queue like any closed decision, unless a receipt confirmed + // the transport closed with nothing pending: an unconfirmed transport may + // be open, and pending work must settle on this decision. A failed close + // is reported; the decision stands. + let closeError: string | undefined; + const current = this.channels.get(`${entry.serverId}:${input.channelId}`); + if (current && this.directsLifecycle(entry.serverId) && + (this.unconfirmedChannelTargets.has(current) || this.hasPendingLifecycle(current))) { + try { + const applied = await this.applyDesiredChannelState(entry.serverId, input.channelId); + if (applied.open) closeError = 'the close was superseded by a newer lifecycle decision'; + } catch (error) { + closeError = (error as Error).message; + } + } + this.appendLifecycleEvent({ kind: 'invitation-declined', serverId: entry.serverId, @@ -2997,6 +3353,7 @@ export class ChannelRegistry { acknowledged, ...(representation ? { representation } : {}), ...(acknowledgmentError ? { acknowledgmentError } : {}), + ...(closeError ? { closeError: `Failed to close the channel's transport: ${closeError}` } : {}), }, }; } @@ -3252,7 +3609,7 @@ export class ChannelRegistry { const at = (): number => Date.now(); const found = this.findExactEntry(target); if ('error' in found) return { status: 'failed', reason: found.error, at: at() }; - const entry = found.entry; + let entry = found.entry; const destination: PublishDestination = { serverId: entry.serverId, channelId: entry.descriptor.id, @@ -3276,29 +3633,60 @@ export class ChannelRegistry { }; } - if (!entry.open) { + // Lifecycle work already queued for the channel settles first: words go + // out only into the channel it leaves, and fail visibly if it leaves + // none open. + if (this.hasPendingLifecycle(entry)) { try { - await this.openChannelNow(entry, openSource); - if (target.routed) { - console.error( - `[routeSpeech] ${conversationId}: locus ${logValue(entry.descriptor.id)} was closed — opened by delivery (speech implies engagement)`, - ); - } - this.emitTraceFn({ - type: 'mcpl:channel-opened-by-send', - serverId: entry.serverId, - channelId: entry.descriptor.id, - }); - if (openSource === 'opened-by-delivery') { - try { - this.onChannelAutoOpened?.({ - conversationId, - serverId: entry.serverId, - source: 'opened-by-delivery', - channels: [{ channelId: entry.descriptor.id, label: entry.descriptor.label }], - }); - } catch (err) { - console.error('onChannelAutoOpened (delivery) failed:', err); + entry = await this.waitForOpenChannel(entry); + } catch (err) { + return { + status: 'failed', + destination, + reason: `lifecycle work prevented delivery: ${connectorText(err instanceof Error ? err.message : String(err))}`, + at: at(), + }; + } + } + + // INVARIANT: it is not possible to send into a closed channel. A send + // into a closed-but-registered channel (the canonical case: a DM that + // woke the agent, since DMs register closed) OPENS it first, so typing + // indicators, reaction machinery and inbound forwarding come alive with + // the words. If the open fails, nothing goes out: a half-alive delivery + // is worse than a loud marker. A channel decided closed is engaged even + // while its transport is still open (a close that failed), as a reply + // would engage it, wherever an open can succeed. Without the lifecycle + // grant it can't, and a send opens before it publishes, so there a + // channel whose transport is open keeps publishing. + const engagesDecidedClosed = this.getDesiredState(entry.serverId, entry.descriptor.id) === 'closed' && + CapabilityGrant.of(server).has('channels.lifecycle'); + if (!entry.open || engagesDecidedClosed) { + try { + const { opened } = await this.openChannelNow(entry, openSource); + entry = await this.waitForOpenChannel(entry); + if (opened) { + if (target.routed) { + console.error( + `[routeSpeech] ${conversationId}: locus ${logValue(entry.descriptor.id)} was closed — opened by delivery (speech implies engagement)`, + ); + } + this.emitTraceFn({ + type: 'mcpl:channel-opened-by-send', + serverId: entry.serverId, + channelId: entry.descriptor.id, + }); + if (openSource === 'opened-by-delivery') { + try { + this.onChannelAutoOpened?.({ + conversationId, + serverId: entry.serverId, + source: 'opened-by-delivery', + channels: [{ channelId: entry.descriptor.id, label: entry.descriptor.label }], + }); + } catch (err) { + console.error('onChannelAutoOpened (delivery) failed:', err); + } } } } catch (err) { diff --git a/src/mcpl/types.ts b/src/mcpl/types.ts index fcc6ab6fc..4ef2a3963 100644 --- a/src/mcpl/types.ts +++ b/src/mcpl/types.ts @@ -1112,7 +1112,15 @@ export interface ChannelsListResult { * Spec Section 14.3. */ export interface ChannelsOpenParams { - /** Exact registered id. Preferred over type/address matching. */ + /** + * The exact registered id to open. Without it, a server matches by `type` + * and `address`, as before. A server that doesn't know the id should + * refuse rather than open another channel: the host fails an open whose + * answer names a different channel id, by its `channel.id` or a history + * item's `channelId`, compared exactly, as the server's traffic is. When + * that other channel is registered and decided closed, the host closes it + * again. + */ channelId?: string; type: string; address?: Record; diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts new file mode 100644 index 000000000..b6c80e178 --- /dev/null +++ b/test/channel-lifecycle-order.test.ts @@ -0,0 +1,1240 @@ +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 { AgentFramework } from '../src/index.js'; +import { MockMembrane } from './helpers/mock-membrane.js'; +import assert from 'node:assert/strict'; +import { ChannelRegistry } from '../src/mcpl/channel-registry.js'; +import { CapabilityGrant, ALL_CAPABILITY_PATHS } from '../src/mcpl/capability-grant.js'; +import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; +import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; +import type { ChannelDescriptor, ChannelIncomingMessage, ChannelsOpenResult } from '../src/mcpl/types.js'; + +const tick = () => new Promise((resolve) => setImmediate(resolve)); +function descriptor(id = 'x', initiallyOpen = false, address = 'original'): ChannelDescriptor { + return { id, type: 'test', label: id, direction: 'bidirectional', initiallyOpen, address: { value: address } }; +} + +type RegistryOptions = NonNullable[4]>; +type NoticeHandler = NonNullable; + +/** A tool result's data, as the channel tools shape it. */ +const dataOf = (result: { data?: unknown }) => (result.data ?? {}) as { status?: string; closeError?: string }; + +function message(channelId = 'x', messageId = 'm1'): ChannelIncomingMessage { + return { + channelId, messageId, author: { id: 'u1', name: 'someone' }, + timestamp: '2026-10-11T00:00:00Z', content: [{ type: 'text', text: 'hello' }], + }; +} + +function fixture(onNotice?: NoticeHandler, store?: JsStore, coalesce?: RegistryOptions['handleCoalescedIncoming']) { + const actual = new Map(); + const calls: Array<{ kind: 'open' | 'close'; id: string; address?: unknown }> = []; + const pending: Array<{ finish: () => void; fail: () => void }> = []; + const held = new Set(); + const publishedWhileOpen: boolean[] = []; + const notices: Array<{ source: string; channels: Array<{ channelId: string; label?: string }> }> = []; + const routeFailures: Array<{ reason: string }> = []; + // A server that answers an open for another channel, by requested id. + const answeredFor = new Map(); + const serve = (kind: 'open' | 'close', params: { channelId?: string; address?: unknown }) => { + const id = params.channelId!; + calls.push({ kind, id, address: params.address }); + return new Promise((resolve, reject) => { + const finish = () => { + const target = kind === 'open' ? answeredFor.get(id) ?? id : id; + actual.set(target, kind === 'open'); + resolve(kind === 'open' ? { channel: descriptor(target) } : { closed: true }); + }; + if (held.has(id)) pending.push({ finish, fail: () => reject(new Error('fixture lifecycle failure')) }); + else finish(); + }); + }; + const server = { + grant: new CapabilityGrant(new Set(ALL_CAPABILITY_PATHS), []), + sendChannelsOpen: (params: { channelId?: string; address?: unknown }) => serve('open', params) as Promise, + sendChannelsClose: (params: { channelId: string }) => serve('close', params), + sendChannelsPublish: async (params: { channelId: string }) => { + publishedWhileOpen.push(actual.get(params.channelId) === true); + return { delivered: true }; + }, + }; + let currentServer = server; + const replaceServer = () => { currentServer = { ...server }; }; + const setGrant = (paths: readonly string[]) => { currentServer.grant = new CapabilityGrant(new Set(paths), []); }; + const traces: Array<{ type: string; [key: string]: unknown }> = []; + const registry = new ChannelRegistry( + { getServer: () => currentServer } as unknown as McplServerRegistry, + {} as FeatureSetManager, () => {}, (event) => traces.push(event), + { store, onRouteFailure: (info) => routeFailures.push(info), + onChannelAutoOpened: (info) => { notices.push(info); onNotice?.(info); }, + ...(coalesce ? { handleCoalescedIncoming: coalesce } : {}) }, + ); + const isOpen = (id = 'x') => registry.getOpenChannels().some((e) => e.descriptor.id === id); + const tool = (kind: 'open' | 'close') => registry.handleChannelToolCall('channel_' + kind, { serverId: 'test', channelId: 'x' }); + // An out-of-order server serves the most recently received operation first. + // The fixed host should never have two in flight for this channel. + const drain = async (...work: Promise[]) => { + let settled = false; + const result = Promise.all(work).finally(() => { settled = true; }); + for (let i = 0; i < 100 && !settled; i++) { + await tick(); + pending.pop()?.finish(); + } + assert.equal(settled, true, 'lifecycle operations must settle'); + await result; + }; + return { registry, actual, calls, pending, held, traces, isOpen, tool, drain, publishedWhileOpen, notices, routeFailures, replaceServer, answeredFor, setGrant }; +} + +test('remove/re-add during reconcile converges on the replacement desired state, with ACK first (#185)', async () => { + const f = fixture(); + f.held.add('x'); + const first = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + assert.equal(f.pending.length, 1); + let acknowledged = false; + const replacement = f.registry.handleChanged('test', { + removed: ['x'], added: [descriptor('x', true, 'replacement')], + }, { respond: () => { acknowledged = true; } }); + assert.equal(acknowledged, true, 'registration ACK must not wait for the lifecycle queue'); + await tick(); + assert.equal(f.pending.length, 1, 'the replacement waits for the operation in flight: one RPC per channel at a time'); + await f.drain(first, replacement); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open'); + assert.equal(f.isOpen(), true); + assert.equal(f.actual.get('x'), true, 'older close must not be the last server-side operation'); + assert.deepEqual(f.calls.filter((c) => c.kind === 'open').at(-1)?.address, { value: 'replacement' }); +}); + +test('explicit open racing a default-close reconcile uses the latest intent', async () => { + const f = fixture(); + f.held.add('x'); + const reconcile = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + const opened = f.tool('open'); + await f.drain(reconcile, opened); + assert.equal((await opened).success, true); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open'); + assert.equal(f.isOpen(), true); + assert.equal(f.actual.get('x'), true); +}); + +for (const firstKind of ['open', 'close'] as const) { + test('overlapping explicit ' + firstKind + ' then opposite intent leaves the newest decision in force', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', firstKind === 'close')] }); + f.held.add('x'); + const first = f.tool(firstKind); + await tick(); + const lastKind = firstKind === 'open' ? 'close' : 'open'; + const last = f.tool(lastKind); + await f.drain(first, last); + assert.equal((await first).success, false); + assert.match(String((await first).error), /superseded by a newer lifecycle decision/); + assert.doesNotMatch(String((await first).error), /stays selected/, 'a newer decision replaced it'); + assert.equal((await last).success, true); + const desired = lastKind === 'open'; + assert.equal(f.registry.getDesiredState('test', 'x'), lastKind === 'open' ? 'open' : 'closed'); + assert.equal(f.isOpen(), desired); + assert.equal(f.actual.get('x'), desired); + }); +} + +test('an already-closed shortcut waits for pending transport work', async () => { + const f = fixture(); + f.actual.set('x', true); + f.held.add('x'); + const reconcile = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + let closeSettled = false; + const close = f.tool('close').then((result) => { closeSettled = true; return result; }); + await tick(); + assert.equal(closeSettled, false, 'entry.open is provisional while a lifecycle RPC is pending'); + await f.drain(reconcile, close); + assert.equal((await close).success, true); + assert.equal(f.actual.get('x'), false); +}); + +test('removal cancels queued work rather than recreating or mutating an unregistered channel', async () => { + const f = fixture(); + f.held.add('x'); + const reconcile = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + const open = f.tool('open'); + await f.registry.handleChanged('test', { removed: ['x'] }); + await f.drain(reconcile, open); + assert.equal((await open).success, false); + assert.equal(f.registry.listChannelsRaw().length, 0); + assert.equal(f.calls.filter((c) => c.kind === 'open').length, 0); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-reconcile-failed').length, 0, + 'removal while queued is cancellation, not a failed subscription'); +}); + +test('a failed operation releases the queue for a later decision', async () => { + const f = fixture(); + f.held.add('x'); + const reconcile = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + const open = f.tool('open'); + f.pending.shift()!.fail(); + await f.drain(reconcile, open); + assert.equal((await open).success, true); + assert.equal(f.actual.get('x'), true); + assert.equal(f.isOpen(), true); + assert.ok(f.traces.some((e) => e.type === 'mcpl:channel-reconcile-failed')); +}); + +test('a blocked channel does not delay another channel lifecycle or its ACK', async () => { + const f = fixture(); + f.held.add('x'); + const blocked = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + let acknowledged = false; + await f.registry.handleChanged('test', { added: [descriptor('y', true)] }, { + respond: () => { acknowledged = true; }, + }); + assert.equal(acknowledged, true); + assert.equal(f.actual.get('y'), true); + assert.equal(f.pending.length, 1); + await f.drain(blocked); +}); + +test('descriptor updates during an open refresh its target, but labels alone do not reopen', async () => { + for (const changesAddress of [false, true]) { + const f = fixture(); + f.held.add('x'); + const opening = f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); + await f.registry.handleChanged('test', { + updated: [{ ...descriptor('x', true, changesAddress ? 'new-target' : 'original'), label: 'new label' }], + }); + await f.drain(opening); + assert.equal(f.calls.length, changesAddress ? 2 : 1); + assert.deepEqual(f.calls.at(-1)?.address, { value: changesAddress ? 'new-target' : 'original' }); + assert.equal(f.isOpen(), true); + } +}); + +for (const delivery of ['speech', 'reply'] as const) { + test(delivery + ' during pending transport work preserves the newer tune-out epoch', async () => { + // Felix's review reproduction: merely waiting for transport must not + // turn a delivery into a decision to cancel attention diversion. + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.calls.length = 0; + f.held.add('x'); + const close = f.tool('close'); + await tick(); + f.registry.enterTuneOut('test', 'x', { + epochId: 'new-epoch', cadenceSeconds: 60, backlogCap: 20, maxWakes: 3, startedAtSequence: 1, + }, 'agent-tool'); + // The close in flight converges on the tune-out by opening again, and the + // delivery joins it: the channel was already decided open, so a reply + // reports it already open, and nothing sends a third RPC. + const sent = delivery === 'speech' + ? f.registry.routeSpeech('resident', 'hello', 'x').then((result) => result?.delivered === true) + : f.registry.openIfClosedForSend('x', 'test').then((result) => result.status === 'already-open'); + await f.drain(close, sent); + assert.equal(await sent, true); + assert.deepEqual(f.calls.map((c) => c.kind), ['close', 'open']); + assert.equal(f.registry.getTuneOutState('test', 'x')?.params.epochId, 'new-epoch'); + assert.equal(f.actual.get('x'), true); + if (delivery === 'speech') assert.deepEqual(f.publishedWhileOpen, [true]); + assert.equal((await f.tool('open')).success, false, 'explicit channel_open still requires tune-out cancellation'); + assert.equal(f.registry.getTuneOutState('test', 'x')?.params.epochId, 'new-epoch'); + }); +} + +test('automatic opening accepts a newer tune-out established during its RPC without clearing it', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + f.held.add('x'); + const speech = f.registry.routeSpeech('resident', 'hello', 'x'); + await tick(); + f.registry.enterTuneOut('test', 'x', { + epochId: 'mid-open-epoch', cadenceSeconds: 60, backlogCap: 20, maxWakes: 3, startedAtSequence: 1, + }, 'agent-tool'); + await f.drain(speech); + assert.equal((await speech)?.delivered, true); + assert.equal(f.registry.getTuneOutState('test', 'x')?.params.epochId, 'mid-open-epoch'); + assert.deepEqual(f.publishedWhileOpen, [true]); +}); + +test('a failed backscroll open preserves an already-live subscription', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.calls.length = 0; + f.held.add('x'); + const backscroll = f.registry.handleChannelToolCall('channel_open', { serverId: 'test', channelId: 'x', backscroll: 5 }); + await tick(); + f.pending.shift()!.fail(); + assert.equal((await backscroll).success, false); + assert.doesNotMatch(String((await backscroll).error), /stays selected/, 'the channel is still open'); + assert.equal(f.isOpen(), true, 'a failed redundant open is not a confirmed close'); + const speech = f.registry.routeSpeech('resident', 'hello', 'x'); + await f.drain(speech); + assert.equal((await speech)?.delivered, true); + assert.equal(f.calls.length, 1, 'delivery must not need another open after a failed backscroll'); + assert.deepEqual(f.publishedWhileOpen, [true]); +}); + +test('speech does not replace a pending explicit close with a new open decision', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.held.add('x'); + const close = f.tool('close'); + await tick(); + const speech = f.registry.routeSpeech('resident', 'tail of the turn', 'x'); + await f.drain(close, speech); + assert.equal((await close).success, true); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.equal(f.actual.get('x'), false); + assert.equal(f.isOpen(), false); + assert.equal(f.notices.length, 0); +}); + +test('a draft published into a channel with a close pending waits for it and posts nothing, as speech does', async () => { + // Drafts publish through the same open step as routed speech, so they meet + // the channel's queued lifecycle work as speech and replies do. + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.held.add('x'); + const close = f.tool('close'); + await tick(); + const draft = f.registry.publish('resident', 'a draft', { serverId: 'test', channelId: 'x' }); + await f.drain(close, draft); + assert.equal((await close).success, true); + const outcome = await draft; + assert.equal(outcome.status, 'failed'); + assert.match(outcome.reason ?? '', /^lifecycle work prevented delivery: /); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.equal(f.isOpen(), false); + assert.deepEqual(f.publishedWhileOpen, [], 'nothing was published'); + assert.equal(f.notices.length, 0); +}); + +test('waiting for a backscroll does not issue new opens or auto-open notices for each speech', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.calls.length = 0; + f.held.add('x'); + const backscroll = f.registry.handleChannelToolCall('channel_open', { serverId: 'test', channelId: 'x', backscroll: 5 }); + await tick(); + const speech1 = f.registry.routeSpeech('resident', 'one', 'x'); + const speech2 = f.registry.routeSpeech('resident', 'two', 'x'); + await f.drain(backscroll, speech1, speech2); + assert.equal((await speech1)?.delivered, true); + assert.equal((await speech2)?.delivered, true); + assert.equal(f.calls.length, 1, 'only the explicit backscroll needs an open RPC'); + assert.equal(f.notices.length, 0, 'waiting is not a closed-to-open transition'); +}); + +for (const delivery of ['speech', 'reply'] as const) { + test(delivery + ' opening reports the replacement registration label', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + f.held.add('x'); + const opening = delivery === 'speech' + ? f.registry.routeSpeech('resident', 'hello', 'x') + : f.registry.openIfClosedForSend('x', 'test'); + await tick(); + const replacement = f.registry.handleChanged('test', { + removed: ['x'], added: [{ ...descriptor('x', true, 'replacement'), label: 'Replacement label' }], + }); + await f.drain(opening, replacement); + if (delivery === 'speech') { + assert.equal(f.notices.length, 1); + assert.equal(f.notices[0].channels[0].label, 'Replacement label'); + } else { + const result = await opening; + assert.ok(result && 'label' in result); + assert.equal(result.label, 'Replacement label'); + } + }); +} + +test('policy announces intent before transport rather than a stale open after supersession', async () => { + const f = fixture(); + f.registry.setSubscriptionPolicy('test', 'auto'); + f.held.add('x'); + const batch = f.registry.handleChanged('test', { added: [descriptor(), descriptor('y')] }); + assert.equal(f.notices.length, 1, 'the decision is announced before transport can stall'); + assert.deepEqual(f.notices[0].channels, [{ channelId: 'x', label: 'x' }, { channelId: 'y', label: 'y' }]); + await tick(); + const closeY = await f.registry.handleChannelToolCall('channel_close', { serverId: 'test', channelId: 'y' }); + assert.equal(closeY.success, true); + await f.drain(batch); + assert.equal(f.registry.getDesiredState('test', 'y'), 'closed'); + assert.equal(f.notices.length, 1, 'no stale confirmation is emitted after the later close'); +}); + +test('a fresh delivery into a settled closed channel remains a new engagement', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + assert.equal((await f.tool('close')).success, true); + assert.equal(f.actual.get('x'), false); + const delivered = await f.registry.routeSpeech('resident', 'a new engagement', 'x'); + assert.equal(delivered?.delivered, true); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open'); + assert.equal(f.notices.length, 1); +}); + +test('a close that completes while delivery joins a backscroll remains authoritative', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.held.add('x'); + const backscroll = f.registry.handleChannelToolCall('channel_open', { serverId: 'test', channelId: 'x', backscroll: 5 }); + await tick(); + const speech = f.registry.routeSpeech('resident', 'tail', 'x'); + const close = f.tool('close'); + await f.drain(backscroll, speech, close); + assert.equal((await close).success, true); + assert.equal(await speech, null); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.deepEqual(f.publishedWhileOpen, []); + assert.equal(f.notices.length, 0); +}); + +test('concurrent delivery into a closed channel shares one opening and one notice', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + f.calls.length = 0; + f.held.add('x'); + const first = f.registry.routeSpeech('resident', 'one', 'x'); + const second = f.registry.routeSpeech('resident', 'two', 'x'); + await f.drain(first, second); + assert.equal((await first)?.delivered, true); + assert.equal((await second)?.delivered, true); + assert.equal(f.calls.length, 1); + assert.equal(f.notices.length, 1); + assert.deepEqual(f.publishedWhileOpen, [true, true]); +}); + +test('failed policy opening records intent once, with truthful resident text through a retry', async () => { + const dir = mkdtempSync(join(tmpdir(), 'policy-intent-notice-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'resident', model: 'test-model', systemPrompt: 'Test.' }], + modules: [], syncIntervalMs: 0, + }); + try { + const recordNotice = (framework as unknown as { + recordChannelAutoOpenNotice(agent: undefined, channels: Array<{ channelId: string; label?: string }>, cause: 'subscription-policy' | 'opened-by-delivery'): void; + }).recordChannelAutoOpenNotice.bind(framework); + const onNotice: NoticeHandler = (info) => recordNotice(undefined, info.channels, info.source); + const f = fixture(onNotice, framework.getStore()); + f.registry.setSubscriptionPolicy('test', 'auto'); + f.held.add('x'); + const first = f.registry.handleChanged('test', { added: [descriptor()] }); + assert.equal(f.notices.length, 1); + await tick(); + f.pending.shift()!.fail(); + await first; + assert.equal(f.isOpen(), false); + + const noticeTexts = () => framework.getAgent('resident')!.getContextManager().getAllMessages() + .flatMap((m) => m.content) + .filter((b) => b.type === 'text' && b.text.startsWith('[channels]')) + .map((b) => (b as { text: string }).text); + assert.equal(noticeTexts().length, 1); + assert.match(noticeTexts()[0], /Selected for opening by subscription policy/); + assert.match(noticeTexts()[0], /admission decision; channel_list shows current transport state/); + assert.doesNotMatch(noticeTexts()[0], /Now open/); + + // Reconstruct the registry from the same Chronicle state before retrying. + // The admission persists even though its first transport attempt failed. + const retry = fixture(onNotice, framework.getStore()); + retry.registry.setSubscriptionPolicy('test', 'auto'); + await retry.registry.handleChanged('test', { added: [descriptor()] }); + assert.equal(retry.isOpen(), true); + assert.equal(retry.notices.length, 0); + assert.equal(noticeTexts().length, 1); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('an unconfirmed target retries explicit close instead of claiming already closed', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + f.calls.length = 0; + f.held.add('x'); + const closing = f.tool('close'); + await tick(); + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B')] }); + f.pending.shift()!.fail(); + assert.equal((await closing).success, false); + assert.equal(f.isOpen(), false); + assert.equal(f.actual.get('x'), true, 'failed close did not change the server state'); + f.held.delete('x'); + assert.equal((await f.tool('close')).success, true); + assert.equal(f.actual.get('x'), false, 'explicit retry must actually close the channel'); + assert.equal(f.calls.length, 2, 'unconfirmed is different from confirmed closed'); + assert.equal((await f.tool('close')).success, true); + assert.equal(f.calls.length, 2, 'the confirmed-closed fast path remains available'); +}); + +for (const kind of ['speech', 'reply'] as const) { + test(kind + ' joins a pending close even after retargeting clears open confirmation', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + f.calls.length = 0; + f.held.add('x'); + const closing = f.tool('close'); + await tick(); + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B')] }); + assert.equal(f.isOpen(), false); + const delivery = kind === 'speech' + ? f.registry.routeSpeech('resident', 'waiting reply', 'x') + : f.registry.openIfClosedForSend('x', 'test'); + await f.drain(closing, delivery); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.equal((await closing).success, true); + assert.equal(f.isOpen(), false); + assert.deepEqual(f.publishedWhileOpen, []); + assert.equal(f.calls.every((call) => call.kind === 'close'), true); + if (kind === 'speech') { + assert.equal(await delivery, null); + assert.equal(f.routeFailures.length, 1); + } else { + assert.equal((await delivery as { status: string }).status, 'open-failed'); + } + }); +} + +for (const change of ['address', 'type'] as const) { + test('failed corrective open does not reuse the old ' + change + ' confirmation', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + f.calls.length = 0; + f.held.add('x'); + const backscroll = f.registry.handleChannelToolCall('channel_open', { + serverId: 'test', channelId: 'x', backscroll: 5, + }); + await tick(); + const speech = f.registry.routeSpeech('resident', 'waiting reply', 'x'); + const updated = change === 'address' + ? descriptor('x', true, 'B') + : { ...descriptor('x', true, 'A'), type: 'new-type' }; + await f.registry.handleChanged('test', { updated: [updated] }); + assert.equal(f.isOpen(), false, 'old target confirmation is invalid after retargeting'); + f.pending.shift()!.finish(); // successful receipt for A + await tick(); + f.pending.shift()!.fail(); // corrective open for the new target fails + await f.drain(backscroll, speech); + assert.equal((await backscroll).success, false); + assert.equal(await speech, null); + assert.equal(f.isOpen(), false); + assert.deepEqual(f.publishedWhileOpen, []); + assert.equal(f.routeFailures.length, 1); + }); +} + +test('a failed RPC from a replaced connection cannot preserve its old open confirmation', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.held.add('x'); + const backscroll = f.registry.handleChannelToolCall('channel_open', { + serverId: 'test', channelId: 'x', backscroll: 5, + }); + await tick(); + const speech = f.registry.routeSpeech('resident', 'waiting reply', 'x'); + f.replaceServer(); + f.pending.shift()!.fail(); + await f.drain(backscroll, speech); + assert.equal((await backscroll).success, false); + assert.equal(await speech, null); + assert.equal(f.isOpen(), false); + assert.deepEqual(f.publishedWhileOpen, []); + assert.equal(f.routeFailures.length, 1); +}); + +test('descriptor churn exhausts convergence and fails the tool plus joined deliveries instead of hanging', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.calls.length = 0; + f.held.add('x'); + let toolSettled = false, speechSettled = false, replySettled = false; + const opening = f.registry.handleChannelToolCall('channel_open', { + serverId: 'test', channelId: 'x', backscroll: 5, + }).then((result) => { toolSettled = true; return result; }); + await tick(); + const speech = f.registry.routeSpeech('resident', 'waiting reply', 'x') + .then((result) => { speechSettled = true; return result; }); + const reply = f.registry.openIfClosedForSend('x', 'test') + .then((result) => { replySettled = true; return result; }); + try { + // Every RPC succeeds promptly, but its target is superseded before the + // receipt arrives. Bound the test itself without a hanging provider. + for (let i = 0; i < 12 && !toolSettled; i++) { + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'churn-' + i)] }); + f.pending.shift()?.finish(); + await tick(); + } + assert.equal(toolSettled, true, 'the whole convergence operation must be bounded'); + assert.equal(speechSettled, true, 'joined speech must receive a terminal result'); + assert.equal(replySettled, true, 'reply preparation must receive a terminal result'); + assert.equal(f.calls.length, 5, 'at most five transport attempts per operation'); + assert.equal((await opening).success, false); + assert.match((await opening).error!, /did not converge/); + assert.equal(await speech, null); + assert.equal((await reply).status, 'open-failed'); + assert.equal(f.isOpen(), false, 'an unconfirmed current target is not reported open'); + assert.deepEqual(f.publishedWhileOpen, []); + assert.equal(f.routeFailures.length, 1); + assert.match(f.routeFailures[0].reason, /did not converge/); + assert.ok(f.traces.some((e) => e.type === 'mcpl:channel-reconcile-failed' && + String(e.error).includes('did not converge'))); + } finally { + // Original code needs one stable receipt to release its unbounded loop. + await f.drain(opening, speech, reply); + } + f.held.delete('x'); + const retry = await f.registry.routeSpeech('resident', 'a later stable retry', 'x'); + assert.equal(retry?.delivered, true); + assert.equal(f.calls.length, 6, 'later delivery reconfirms the latest target before publishing'); + assert.equal((await f.tool('close')).success, true, 'exhaustion releases the queue'); + assert.equal((await f.tool('open')).success, true, 'a later stable decision gets a fresh retry budget'); +}); + +test('background reconciliation reports supersession exhaustion and still permits other channels', async () => { + const f = fixture(); + f.held.add('x'); + let settled = false; + const reconcile = f.registry.handleChanged('test', { added: [descriptor('x', true)] }) + .then(() => { settled = true; }); + await tick(); + try { + await f.registry.handleChanged('test', { added: [descriptor('y', true)] }); + assert.equal(f.actual.get('y'), true); + for (let i = 0; i < 12 && !settled; i++) { + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'churn-' + i)] }); + f.pending.shift()?.finish(); + await tick(); + } + assert.equal(settled, true); + assert.equal(f.calls.filter((call) => call.id === 'x').length, 5); + const failures = f.traces.filter((e) => e.type === 'mcpl:channel-reconcile-failed'); + assert.equal(failures.length, 1); + assert.match(String(failures[0].error), /did not converge/); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open', 'failure does not erase desired intent'); + } finally { await f.drain(reconcile); } +}); + +test('finite descriptor supersession can converge on its fifth transport attempt', async () => { + const f = fixture(); + f.held.add('x'); + const reconcile = f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await tick(); + for (let i = 0; i < 4; i++) { + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'target-' + i)] }); + f.pending.shift()!.finish(); + await tick(); + } + await f.drain(reconcile); + assert.equal(f.calls.length, 5); + assert.equal(f.isOpen(), true); + assert.deepEqual(f.calls.at(-1)?.address, { value: 'target-3' }); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-reconcile-failed').length, 0); +}); + +test('a newer tune-out decision keeps transport open when an older close completes', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.held.add('x'); + const close = f.tool('close'); + await tick(); + f.registry.enterTuneOut('test', 'x', { + epochId: 'new-epoch', cadenceSeconds: 60, backlogCap: 20, maxWakes: 3, startedAtSequence: 1, + }, 'agent-tool'); + await f.drain(close); + assert.equal(f.registry.getDesiredState('test', 'x'), 'tuned-out'); + assert.equal(f.actual.get('x'), true); + assert.equal(f.isOpen(), true); + assert.equal((await close).success, false, 'superseded close must not report current state as closed'); +}); + +test('an explicit open answers only once the current registration is open', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + f.held.add('x'); + let openWhenAnswered: boolean | undefined; + const opening = f.tool('open').then((result) => { openWhenAnswered = f.isOpen(); return result; }); + await tick(); + assert.equal(f.pending.length, 1); + // The server registers the channel again while the open is in flight. + const reRegistered = f.registry.handleChanged('test', { added: [descriptor()] }); + await f.drain(opening, reRegistered); + assert.equal((await opening).success, true); + assert.equal(openWhenAnswered, true, 'a receipt for the replaced registration confirms nothing about the new one'); + assert.equal(f.isOpen(), true); +}); + +test('an explicit open waits for a close in flight rather than answering already open', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.held.add('x'); + const closing = f.tool('close'); + await tick(); + const reopening = f.tool('open'); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open', 'the next open meets an open decision and a pending close'); + let settled = false; + const again = f.tool('open').then((result) => { settled = true; return result; }); + await tick(); + assert.equal(settled, false, 'entry.open is provisional while the close is in flight'); + await f.drain(closing, reopening, again); + assert.equal((await again).success, true); + assert.equal(f.actual.get('x'), true); + assert.equal(f.isOpen(), true); +}); + +for (const delivery of ['reply', 'speech'] as const) { + test('a ' + delivery + ' into a channel whose close failed opens it again, and says so', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.calls.length = 0; + f.held.add('x'); + const closing = f.tool('close'); + await tick(); + f.pending.shift()!.fail(); + assert.equal((await closing).success, false); + await tick(); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.equal(f.isOpen(), true, 'a failed same-target close leaves the transport known open'); + f.held.delete('x'); + if (delivery === 'reply') { + const reply = await f.registry.openIfClosedForSend('x', 'test'); + assert.equal(reply.status, 'opened', 'the reply turned a channel decided closed open again'); + } else { + assert.equal((await f.registry.routeSpeech('resident', 'hello', 'x'))?.delivered, true); + assert.equal(f.notices.length, 1, 'the speech turned a channel decided closed open again'); + assert.deepEqual(f.publishedWhileOpen, [true]); + } + assert.equal(f.registry.getDesiredState('test', 'x'), 'open'); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-opened-by-send').length, 1); + assert.deepEqual(f.calls.map((c) => c.kind), ['close', 'open'], 'the transport is confirmed again after the failed close'); + }); +} + +test('speech into a channel decided closed keeps publishing where no open can succeed, without the lifecycle grant', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + assert.equal(f.isOpen(), true); + f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.lifecycle')); + const closing = await f.tool('close'); + assert.equal(closing.success, false); + assert.match(String(closing.error), /channels\.lifecycle/); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed', 'the refused close is still recorded as a decision'); + f.calls.length = 0; + assert.equal((await f.registry.routeSpeech('resident', 'hello', 'x'))?.delivered, true); + assert.deepEqual(f.publishedWhileOpen, [true]); + assert.deepEqual(f.calls, [], 'no open is attempted where it could not succeed'); + assert.equal(f.notices.length, 0); +}); + +test('a failed explicit open says the channel stays selected for opening, and a later delivery opens it quietly', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + f.calls.length = 0; + f.held.add('x'); + const opening = f.tool('open'); + await tick(); + f.pending.shift()!.fail(); + const result = await opening; + assert.equal(result.success, false); + assert.match(String(result.error), /stays selected for opening/); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open', 'the decision outlives a failed round-trip'); + f.held.delete('x'); + await tick(); + assert.equal((await f.registry.routeSpeech('resident', 'hello', 'x'))?.delivered, true); + assert.equal(f.notices.length, 0, 'the resident was already told the channel stays selected for opening'); + assert.deepEqual(f.calls.map((c) => c.kind), ['open', 'open']); +}); + +for (const delivery of ['speech', 'reply'] as const) { + test(delivery + ' into a channel decided open whose transport lost confirmation is not announced as opening it', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + // A retarget opens the new target; that open fails, so the channel stays + // decided open with its transport unconfirmed. + f.held.add('x'); + const retarget = f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B')] }); + await tick(); + f.pending.shift()!.fail(); + await retarget; + f.held.delete('x'); + assert.equal(f.isOpen(), false, 'a failed open leaves the new target unconfirmed'); + f.calls.length = 0; + if (delivery === 'speech') { + assert.equal((await f.registry.routeSpeech('resident', 'hello', 'x'))?.delivered, true); + assert.deepEqual(f.publishedWhileOpen, [true]); + } else { + assert.equal((await f.registry.openIfClosedForSend('x', 'test')).status, 'already-open'); + } + assert.deepEqual(f.calls.map((c) => [c.kind, c.address]), [['open', { value: 'B' }]], 'the new target is confirmed first'); + assert.equal(f.notices.length, 0, 'the channel was never closed'); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-opened-by-send').length, 0); + assert.equal(f.registry.getDesiredState('test', 'x'), 'open'); + }); + + test(delivery + ' into a tuned-out channel whose transport lost confirmation keeps the tune-out, unannounced', 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; + f.held.delete('x'); + assert.equal(f.isOpen(), false); + if (delivery === 'speech') { + assert.equal((await f.registry.routeSpeech('resident', 'hello', 'x'))?.delivered, true); + assert.deepEqual(f.publishedWhileOpen, [true]); + } else { + assert.equal((await f.registry.openIfClosedForSend('x', 'test')).status, 'already-open'); + } + assert.equal(f.registry.getDesiredState('test', 'x'), 'tuned-out'); + assert.equal(f.registry.getTuneOutState('test', 'x')?.params.epochId, 'kept-epoch'); + assert.equal(f.isOpen(), true); + assert.equal(f.notices.length, 0, 'its traffic is still diverted, so nothing new reaches the resident'); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-opened-by-send').length, 0); + }); +} + +test('a corrective close that cannot converge reports its exhaustion once', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + f.calls.length = 0; + // Asked to open x, the server opens y instead, and y's corrective close is + // retargeted before each receipt. + f.answeredFor.set('x', 'y'); + f.held.add('y'); + assert.equal((await f.tool('open')).success, false); + await tick(); + assert.equal(f.pending.length, 1, 'the corrective close for y is in flight'); + try { + for (let i = 0; i < 12 && f.pending.length > 0; i++) { + await f.registry.handleChanged('test', { updated: [descriptor('y', false, 'churn-' + i)] }); + f.pending.shift()!.finish(); + await tick(); + } + assert.equal(f.pending.length, 0, 'the corrective close is bounded'); + assert.equal(f.calls.filter((c) => c.id === 'y' && c.kind === 'close').length, 5); + const failures = f.traces.filter((e) => e.type === 'mcpl:channel-reconcile-failed' && e.channelId === 'y'); + assert.equal(failures.length, 1); + assert.match(String(failures[0].error), /did not converge/); + } finally { + f.held.delete('y'); + while (f.pending.length > 0) f.pending.shift()!.finish(); + } +}); + +test('speech joined to a failed backscroll still reaches a channel that stayed open', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.held.add('x'); + const backscroll = f.registry.handleChannelToolCall('channel_open', { serverId: 'test', channelId: 'x', backscroll: 5 }); + await tick(); + const speech = f.registry.routeSpeech('resident', 'joined', 'x'); + f.pending.shift()!.fail(); + assert.equal((await backscroll).success, false); + assert.equal((await speech)?.delivered, true, 'a failed redundant open leaves the channel usable for a joined delivery'); + assert.deepEqual(f.publishedWhileOpen, [true]); +}); + +test('a reply into a tuned-out channel whose transport is open needs no lifecycle RPC', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.registry.enterTuneOut('test', 'x', { + epochId: 'quiet', cadenceSeconds: 60, backlogCap: 20, maxWakes: 3, startedAtSequence: 1, + }, 'agent-tool'); + f.calls.length = 0; + assert.equal((await f.registry.openIfClosedForSend('x', 'test')).status, 'already-open'); + assert.deepEqual(f.calls, []); + assert.equal(f.registry.getTuneOutState('test', 'x')?.params.epochId, 'quiet'); +}); + +test('a failed reply open reports the label of the registration that replaced it', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + f.held.add('x'); + const reply = f.registry.openIfClosedForSend('x', 'test'); + await tick(); + const replacement = f.registry.handleChanged('test', { + removed: ['x'], added: [{ ...descriptor('x', false, 'replacement'), label: 'Replacement label' }], + }); + f.pending.shift()!.fail(); + const result = await reply; + assert.equal(result.status, 'open-failed'); + assert.equal(result.label, 'Replacement label'); + await f.drain(replacement); +}); + +test('a channel removed from its batch holds up none of its siblings, and fails nothing', async () => { + const f = fixture(); + f.held.add('x'); + const first = f.registry.handleChanged('test', { added: [descriptor('x')] }); + await tick(); + assert.equal(f.pending.length, 1, "x's first operation is in flight"); + f.held.add('y'); + const batch = f.registry.handleChanged('test', { added: [descriptor('y', true), descriptor('x')] }); + await tick(); + assert.equal(f.pending.length, 2, "y's operation is in flight too"); + await f.registry.handleChanged('test', { removed: ['x'] }); + f.pending.pop()!.finish(); // y's receipt; x's first operation stays in flight + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.isOpen('y'), true, "y opened while x's operation was still in flight"); + await f.drain(first, batch); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-reconcile-failed').length, 0, + 'the removed channel is cancelled, not failed'); +}); + +test("a batch's channels reconcile side by side: one held RPC delays none of its siblings", async () => { + const f = fixture(); + f.held.add('x'); + const batch = f.registry.handleChanged('test', { added: [descriptor('x', true), descriptor('y', true)] }); + for (let i = 0; i < 5; i++) await tick(); + assert.equal(f.isOpen('y'), true, "y's open didn't wait for x's"); + assert.equal(f.isOpen('x'), false); + await f.drain(batch); + assert.equal(f.isOpen('x'), true); +}); + +test('a descriptor the registration rejected is neither admitted nor announced', async () => { + const f = fixture(); + f.registry.setSubscriptionPolicy('test', 'auto'); + // The first registration migrates the legacy policy; the second is admitted by it. + await f.registry.handleRegister('test', { channels: [descriptor('x')] }); + await f.registry.handleRegister('test', { channels: [descriptor('y'), { ...descriptor('rejected'), id: '' }] }); + assert.deepEqual(f.notices.map((n) => n.channels.map((c) => c.channelId)), [['y']]); + assert.equal(f.registry.getDesiredState('test', ''), undefined); +}); + +test('a failed initial close leaves the channel unconfirmed, so channel_close closes it for real', async () => { + const f = fixture(); + f.held.add('x'); + const registration = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + f.pending.shift()!.fail(); + await registration; + f.held.delete('x'); + f.calls.length = 0; + assert.equal(dataOf(await f.tool('close')).status, 'closed'); + assert.deepEqual(f.calls.map((c) => c.kind), ['close'], 'a failed close may have closed nothing'); + assert.equal(dataOf(await f.tool('close')).status, 'already closed', 'its receipt confirms the close'); + assert.equal(f.calls.length, 1); +}); + +test("a re-registration's entry is unconfirmed until a receipt, so channel_close never answers already closed for it", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + assert.equal(dataOf(await f.tool('close')).status, 'already closed', 'its initial close was confirmed'); + // The server registers again, without the lifecycle grant this time: the + // host sends nothing, and nothing confirms the new entry closed. + f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.lifecycle')); + f.calls.length = 0; + await f.registry.handleRegister('test', { channels: [descriptor()] }); + assert.equal(f.calls.length, 0); + const result = await f.tool('close'); + assert.equal(result.success, false); + assert.match(String(result.error), /channels\.lifecycle not in/, 'refused for the grant, not answered already closed'); +}); + +test('traffic after a retarget confirms nothing: only a receipt for the new target does', 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(); + await f.registry.handleIncoming('test', { messages: [message()] }); + assert.equal(f.isOpen(), false, "the old subscription's traffic doesn't confirm the new target"); + f.pending.shift()!.fail(); + await retarget; + assert.equal(f.isOpen(), false, 'and a failed open for the new target leaves it unconfirmed'); +}); + +test('traffic for a channel decided closed leaves its confirmed close standing', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + assert.equal((await f.tool('close')).success, true); + await f.registry.handleIncoming('test', { messages: [message()] }); + assert.equal(f.isOpen(), false); +}); + +test('a close that lands while the coalescer holds a message stands', async () => { + let admit!: () => void; + const f = fixture(undefined, undefined, async (_serverId, held) => { + await new Promise((resolve) => { admit = resolve; }); + return { messageId: held.messageId, accepted: true }; + }); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + const incoming = f.registry.handleIncoming('test', { messages: [{ ...message(), coalesce: { key: 'subject' } }] }); + await tick(); + assert.equal((await f.tool('close')).success, true); + admit(); + await incoming; + assert.equal(f.isOpen(), false, 'the close receipt stands'); +}); + +test('a retarget of a channel decided open opens the new target; one decided closed sends nothing', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A'), descriptor('y', false, 'A')] }); + f.calls.length = 0; + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B'), descriptor('y', false, 'B')] }); + assert.deepEqual(f.calls.map((c) => [c.kind, c.id, c.address]), [['open', 'x', { value: 'B' }]]); + assert.equal(f.isOpen('x'), true); + const closeY = await f.registry.handleChannelToolCall('channel_close', { serverId: 'test', channelId: 'y' }); + assert.equal(dataOf(closeY).status, 'closed', "y's new target is unconfirmed, so its close is sent"); + assert.deepEqual(f.calls.at(-1), { kind: 'close', id: 'y', address: undefined }); +}); + +for (const decision of ['open', 'tuned-out'] as const) { + test(`a retarget while the old target's open is in flight opens the new target when that open fails, for a channel decided ${decision}`, async () => { + // Greptile's P1 on #210: the retarget queued nothing, since an operation + // was running, and that operation threw on its failed RPC without + // reading the new target. + const f = fixture(); + f.held.add('x'); + const opening = f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + await tick(); + if (decision === 'tuned-out') { + f.registry.enterTuneOut('test', 'x', { + epochId: 'e1', cadenceSeconds: 60, backlogCap: 20, maxWakes: 3, startedAtSequence: 1, + }, 'agent-tool'); + } + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B')] }); + f.held.delete('x'); + f.pending.shift()!.fail(); + await f.drain(opening); + assert.deepEqual(f.calls.map((c) => [c.kind, c.address]), [['open', { value: 'A' }], ['open', { value: 'B' }]]); + assert.equal(f.isOpen(), true); + assert.equal(f.registry.getDesiredState('test', 'x'), decision); + assert.deepEqual(f.traces.filter((t) => t.type === 'mcpl:channel-reconcile-failed'), []); + }); +} + +test("a retarget while the old target's close is in flight sends nothing more when that close fails", async () => { + // As for any retarget of a channel decided closed, the new target is left + // unconfirmed for channel_close to close for real. + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); + f.calls.length = 0; + f.held.add('x'); + const close = f.tool('close'); + await tick(); + await f.registry.handleChanged('test', { updated: [descriptor('x', true, 'B')] }); + f.held.delete('x'); + f.pending.shift()!.fail(); + await f.drain(close); + assert.equal((await close).success, false); + assert.deepEqual(f.calls.map((c) => c.kind), ['close']); + assert.equal(f.registry.getDesiredState('test', 'x'), 'closed'); + assert.equal(f.isOpen(), false); +}); + +test("removing a server cancels its queued lifecycle work, even across a re-registration", async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + f.calls.length = 0; + f.held.add('x'); + const closing = f.tool('close'); + await tick(); + const reopening = f.tool('open'); + f.registry.removeServer('test'); + const reregistration = f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + await f.drain(closing, reopening, reregistration); + assert.equal((await reopening).success, false, 'the open queued before the removal never runs'); + assert.match(String((await reopening).error), /removed while this operation was queued/); + assert.deepEqual(f.calls.map((c) => c.kind), ['close', 'open'], "the close in flight, then the re-registration's own open"); + assert.equal(f.isOpen(), true); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-reconcile-failed').length, 0); +}); + +test('a substitute removed while its corrective close is queued is cancelled, not traced as failed', async () => { + const f = fixture(); + f.held.add('y'); + const registration = f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + await tick(); + assert.equal(f.pending.length, 1, "y's initial close is in flight"); + f.answeredFor.set('x', 'y'); + const opening = f.tool('open'); + for (let i = 0; i < 3; i++) await tick(); + await f.registry.handleChanged('test', { removed: ['y'] }); + await f.drain(registration, opening); + assert.equal((await opening).success, false); + assert.equal(f.traces.filter((e) => e.type === 'mcpl:channel-reconcile-failed').length, 0); +}); + +test('declining an invitation closes an unconfirmed transport for real, and waits for pending work', async () => { + const f = fixture(); + const decline = () => f.registry.handleChannelToolCall('channel_decline', { serverId: 'test', channelId: 'x', messageId: 'm1' }); + f.held.add('x'); + const registration = f.registry.handleChanged('test', { added: [descriptor()] }); + await tick(); + f.pending.shift()!.fail(); + await registration; + f.held.delete('x'); + f.calls.length = 0; + const unconfirmed = await decline(); + assert.equal(unconfirmed.success, true); + assert.equal(dataOf(unconfirmed).status, 'remained closed'); + assert.equal(dataOf(unconfirmed).closeError, undefined); + assert.deepEqual(f.calls.map((c) => c.kind), ['close'], 'the unconfirmed transport is closed for real'); + f.calls.length = 0; + assert.equal((await decline()).success, true); + assert.equal(f.calls.length, 0, 'a confirmed close with nothing pending needs no RPC'); + + // A re-registration's close in flight: the decline queues behind it. + f.held.add('x'); + const reregistration = f.registry.handleChanged('test', { removed: ['x'], added: [descriptor()] }); + await tick(); + assert.equal(f.pending.length, 1); + const pending = decline(); + await f.drain(reregistration, pending); + assert.equal((await pending).success, true); + assert.deepEqual(f.calls.map((c) => c.kind), ['close', 'close'], 'the decline went through the queue behind the close'); +}); + +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 }; + for (let i = 0; i < 3; i++) await tick(); + assert.equal(internals.channelLifecycleTails.size, 0); + assert.equal(internals.lifecycleOutstanding.size, 0); +}); + +test('the notices say what happened, name the opt-out, and state no rule about later sends', async () => { + const dir = mkdtempSync(join(tmpdir(), 'channel-notice-text-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'resident', model: 'test-model', systemPrompt: 'Test.' }], + modules: [], syncIntervalMs: 0, + }); + try { + const recordNotice = (framework as unknown as { + recordChannelAutoOpenNotice( + agent: undefined, + channels: Array<{ channelId: string; label?: string }>, + cause: 'subscription-policy' | 'opened-by-delivery' | 'opened-by-reply', + ): void; + }).recordChannelAutoOpenNotice.bind(framework); + recordNotice(undefined, [{ channelId: 'discord:g1:c1', label: '#general' }], 'opened-by-reply'); + recordNotice(undefined, [{ channelId: 'c2' }], 'opened-by-delivery'); + recordNotice(undefined, [{ channelId: 'c3' }, { channelId: 'c4', label: 'ops' }], 'subscription-policy'); + const texts = framework.getAgent('resident')!.getContextManager().getAllMessages() + .flatMap((m) => m.content) + .filter((b) => b.type === 'text' && b.text.startsWith('[channels]')) + .map((b) => (b as { text: string }).text); + assert.deepEqual(texts, [ + "[channels] Now open: #general (discord:g1:c1): your own message into the closed channel opened it. " + + "What an open channel delivers to you depends on its server. If you don't want it, use channel_close; your choice outranks policy.", + "[channels] Now open: c2: your own message into the closed channel opened it. " + + "What an open channel delivers to you depends on its server. If you don't want it, use channel_close; your choice outranks policy.", + '[channels] Selected for opening by subscription policy: c3, ops (c4). ' + + 'This records the admission decision; channel_list shows current transport state. ' + + "What an open channel delivers to you depends on its server. If you don't want one of them, use channel_close; your choice outranks policy.", + ]); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test("a server the host doesn't direct opens its own channels: its traffic marks them open, and a decline sends nothing", async () => { + const f = fixture(); + f.setGrant(ALL_CAPABILITY_PATHS.filter((path) => path !== 'channels.lifecycle')); + await f.registry.handleChanged('test', { added: [descriptor()] }); + await f.registry.handleIncoming('test', { messages: [message()] }); + assert.equal(f.isOpen(), true, 'traffic is the only evidence such a server gives'); + await f.registry.handleChanged('test', { updated: [descriptor('x', false, 'B')] }); + assert.equal(f.isOpen(), false); + await f.registry.handleIncoming('test', { messages: [message('x', 'm2')] }); + assert.equal(f.isOpen(), true); + await f.registry.handleChanged('test', { added: [descriptor('y')] }); + const declined = await f.registry.handleChannelToolCall('channel_decline', { serverId: 'test', channelId: 'y', messageId: 'm3' }); + assert.equal(declined.success, true); + assert.equal(dataOf(declined).closeError, undefined); + assert.equal(f.calls.length, 0, 'the host sends no lifecycle RPC to a server it does not direct'); +}); + +test('a failed open leaves the channel unconfirmed: traffic afterwards marks nothing open', async () => { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor()] }); + f.held.add('x'); + const opening = f.tool('open'); + await tick(); + f.pending.shift()!.fail(); + assert.equal((await opening).success, false); + f.held.delete('x'); + await f.registry.handleIncoming('test', { messages: [message()] }); + assert.equal(f.isOpen(), false, 'an RPC can fail after the server acted, and traffic does not settle which'); +}); + +test('disconnecting a server removes its channel entries before its connection closes', async () => { + const dir = mkdtempSync(join(tmpdir(), 'disconnect-order-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'resident', model: 'test-model', systemPrompt: 'Test.' }], + modules: [], syncIntervalMs: 0, + }); + try { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + let channelsAtClose: number | undefined; + const internals = framework as unknown as { channelRegistry: ChannelRegistry | null; mcplServerRegistry: unknown }; + internals.channelRegistry = f.registry; + internals.mcplServerRegistry = { + removeServer: async () => { channelsAtClose = f.registry.listChannelsRaw().length; }, + getAllServers: () => [], + closeAll: async () => {}, + }; + await framework.disconnectMcplServer('test'); + assert.equal(channelsAtClose, 0, 'nothing queued for its channels can run against the closing connection'); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); + +test('a channel its server registers while a disconnect closes it is removed with the rest', async () => { + const dir = mkdtempSync(join(tmpdir(), 'disconnect-late-register-')); + const framework = await AgentFramework.create({ + storePath: join(dir, 'store'), membrane: new MockMembrane().asMembrane(), + agents: [{ name: 'resident', model: 'test-model', systemPrompt: 'Test.' }], + modules: [], syncIntervalMs: 0, + }); + try { + const f = fixture(); + await f.registry.handleChanged('test', { added: [descriptor('x', true)] }); + const internals = framework as unknown as { channelRegistry: ChannelRegistry | null; mcplServerRegistry: unknown }; + internals.channelRegistry = f.registry; + let lateRegistered = false; + internals.mcplServerRegistry = { + // A connect still setting up when the disconnect began: a channel + // change it had received is handled while its connection closes, after + // the disconnect removed the entries it found. + removeServer: async () => { + if (lateRegistered) return; + await f.registry.handleChanged('test', { added: [descriptor('y')] }); + lateRegistered = f.registry.listChannelsRaw().some((c) => c.descriptor.id === 'y'); + }, + getAllServers: () => [], + closeAll: async () => {}, + }; + await framework.disconnectMcplServer('test'); + assert.equal(lateRegistered, true, 'the channel was registered while the connection closed'); + assert.deepEqual(f.registry.listChannelsRaw(), [], 'no channel of the disconnected server outlives its disconnect'); + } finally { + await framework.stop(); + rmSync(dir, { recursive: true, force: true }); + } +}); diff --git a/test/channel-open-substitution.test.ts b/test/channel-open-substitution.test.ts new file mode 100644 index 000000000..2e92b1054 --- /dev/null +++ b/test/channel-open-substitution.test.ts @@ -0,0 +1,245 @@ +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { ChannelRegistry } from '../src/mcpl/channel-registry.js'; +import { logValue } from '../src/mcpl/log-value.js'; +import { CapabilityGrant, ALL_CAPABILITY_PATHS } from '../src/mcpl/capability-grant.js'; +import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; +import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; +import type { ChannelDescriptor, ChannelIncomingMessage, ChannelsOpenResult } from '../src/mcpl/types.js'; + +// A server that answers channels/open for another channel than the one named, +// as discord-mcpl's base did for a stale id: it fell back to its first channel +// of the type, opened that, and returned that channel's history. + +const tick = () => new Promise((resolve) => setImmediate(resolve)); + +function descriptor(id: string, initiallyOpen = false, address = id): ChannelDescriptor { + return { id, type: 'test', label: id, direction: 'bidirectional', initiallyOpen, address: { value: address } }; +} + +function historyOf(channelId: string): ChannelIncomingMessage[] { + return [{ + channelId, messageId: `${channelId}-1`, author: { id: 'u1', name: 'someone' }, + timestamp: '2026-10-09T00:00:00Z', content: [{ type: 'text', text: `from ${channelId}` }], + }]; +} + +function fixture(answer: (requested: string) => ChannelsOpenResult | Promise) { + const calls: Array<{ kind: 'open' | 'close'; id: string }> = []; + const published: string[] = []; + const traces: Array<{ type: string; [key: string]: unknown }> = []; + const routeFailures: Array<{ reason: string }> = []; + const server = { + grant: new CapabilityGrant(new Set(ALL_CAPABILITY_PATHS), []), + sendChannelsOpen: async (params: { channelId?: string }) => { + calls.push({ kind: 'open', id: params.channelId! }); + return answer(params.channelId!); + }, + sendChannelsClose: async (params: { channelId: string }) => { + calls.push({ kind: 'close', id: params.channelId }); + return { closed: true }; + }, + sendChannelsPublish: async (params: { channelId: string }) => { + published.push(params.channelId); + return { delivered: true }; + }, + }; + const registry = new ChannelRegistry( + { getServer: () => server } as unknown as McplServerRegistry, + {} as FeatureSetManager, () => {}, (event) => traces.push(event), + { onRouteFailure: (info) => routeFailures.push(info) }, + ); + const isOpen = (id: string) => registry.getOpenChannels().some((e) => e.descriptor.id === id); + const substituted = () => traces.filter((t) => t.type === 'mcpl:channel-open-substituted'); + const openX = (backscroll?: number) => + registry.handleChannelToolCall('channel_open', { serverId: 'test', channelId: 'x', ...(backscroll ? { backscroll } : {}) }); + return { registry, calls, published, traces, routeFailures, isOpen, substituted, openX }; +} + +/** Run `work` with console.error captured, returning what it printed. */ +async function capturingErrors(work: () => Promise): Promise<{ value: T; lines: string[] }> { + const lines: string[] = []; + const original = console.error; + console.error = (...args: unknown[]) => { lines.push(args.map(String).join(' ')); }; + try { + return { value: await work(), lines }; + } finally { + console.error = original; + } +} + +test('an open answered for another channel fails, and that channel\'s history goes nowhere', async () => { + const f = fixture(() => ({ channel: descriptor('y'), history: historyOf('y') })); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + const { value: result, lines } = await capturingErrors(() => f.openX(5)); + + assert.equal(result.success, false); + assert.match(String(result.error), /answered channels\/open for x with a different channel/); + assert.equal(result.data, undefined, 'no history, or anything else from the answer, reaches the resident'); + assert.equal(f.isOpen('x'), false, 'the requested channel is not marked open'); + assert.equal(f.isOpen('y'), false, 'nor is the channel the server answered for'); + + const [trace, ...more] = f.substituted(); + assert.equal(more.length, 0); + assert.equal(trace?.serverId, 'test'); + assert.equal(trace?.channelId, 'x'); + assert.equal(trace?.answeredFor, 'y'); + assert.deepEqual(lines.filter((l) => l.startsWith('[channel-open-substituted]')), + ['[channel-open-substituted] server=test requested=x answered=y']); +}); + +test('an answer naming no channel but carrying another channel\'s history fails the same way', async () => { + const f = fixture(() => ({ history: historyOf('y') }) as unknown as ChannelsOpenResult); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + const { value: result } = await capturingErrors(() => f.openX(5)); + assert.equal(result.success, false); + assert.equal(result.data, undefined); + assert.equal(f.isOpen('x'), false); + assert.equal(f.substituted()[0]?.answeredFor, 'y'); +}); + +test('the channel opened instead is closed again when the host wants it closed', async () => { + const f = fixture(() => ({ channel: descriptor('y') })); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + f.calls.length = 0; + await capturingErrors(() => f.openX()); + await tick(); + assert.deepEqual(f.calls, [{ kind: 'open', id: 'x' }, { kind: 'close', id: 'y' }], + 'the server opened y, so the host closes it through y\'s own lifecycle'); + assert.equal(f.registry.getDesiredState('test', 'y'), 'closed'); + assert.equal(f.isOpen('y'), false); +}); + +test('the channel opened instead is left alone when the host wants it open', async () => { + const f = fixture((requested) => ({ channel: descriptor(requested === 'x' ? 'y' : requested) })); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y', true)] }); + assert.equal(f.isOpen('y'), true); + f.calls.length = 0; + await capturingErrors(() => f.openX()); + await tick(); + assert.deepEqual(f.calls, [{ kind: 'open', id: 'x' }], 'no corrective close, and no reopening of y'); + assert.equal(f.isOpen('y'), true); +}); + +test('the channel opened instead needs nothing once it is no longer registered', async () => { + const f = fixture(() => ({ channel: descriptor('y') })); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + await f.registry.handleChanged('test', { removed: ['y'] }); + f.calls.length = 0; + await capturingErrors(() => f.openX()); + await tick(); + assert.equal(f.substituted().length, 1, 'the substitution is still reported'); + assert.deepEqual(f.calls, [{ kind: 'open', id: 'x' }], 'nothing is sent for y'); + assert.equal(f.traces.filter((t) => t.type === 'mcpl:channel-reconcile-failed').length, 0, + 'an unregistered channel has no lifecycle to fail'); +}); + +test('a channel admitted open stays closed on a substituted answer, reports it once, and speech into it fails loudly', async () => { + const f = fixture(() => ({ channel: descriptor('y') })); + await capturingErrors(() => f.registry.handleChanged('test', { added: [descriptor('x', true), descriptor('y')] })); + assert.equal(f.isOpen('x'), false); + assert.equal(f.substituted().length, 1); + assert.equal(f.traces.filter((t) => t.type === 'mcpl:channel-reconcile-failed').length, 0, + 'the substitution is its own diagnostic, not also a generic reconcile failure'); + + const { value: delivery } = await capturingErrors(() => f.registry.routeSpeech('resident', 'hello', 'x')); + assert.equal(delivery, null); + assert.deepEqual(f.published, [], 'nothing is sent into a channel the server would not confirm'); + assert.equal(f.routeFailures.length, 1); + assert.match(f.routeFailures[0]!.reason, /answered channels\/open for x with a different channel/); +}); + +test('a known-open channel loses its confirmation on a substituted answer, and reopens once the server answers for it', async () => { + let answerFor: string | undefined; + const f = fixture((requested) => ({ channel: descriptor(answerFor ?? requested) })); + await f.registry.handleChanged('test', { added: [descriptor('x', true), descriptor('y')] }); + assert.equal(f.isOpen('x'), true); + + answerFor = 'y'; + const { value: backscroll } = await capturingErrors(() => f.openX(5)); + assert.equal(backscroll.success, false); + assert.equal(f.isOpen('x'), false, 'an answer for another channel confirms nothing about this one'); + + answerFor = undefined; + await tick(); + f.calls.length = 0; + const delivery = await f.registry.routeSpeech('resident', 'hello', 'x'); + assert.equal(delivery?.delivered, true); + assert.deepEqual(f.calls, [{ kind: 'open', id: 'x' }], 'delivery reopens the channel before sending'); + assert.deepEqual(f.published, ['x']); +}); + +test('a substituted answer on a receipt made stale by retargeting is retried, not failed', async () => { + let release!: (answer: ChannelsOpenResult) => void; + let first = true; + const f = fixture((requested) => { + if (!first) return { channel: descriptor(requested) }; + first = false; + return new Promise((resolve) => { release = resolve; }); + }); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + f.calls.length = 0; + const { value: opened } = await capturingErrors(async () => { + const opening = f.openX(); + await tick(); + await f.registry.handleChanged('test', { updated: [descriptor('x', false, 'new-target')] }); + release({ channel: descriptor('y') }); + return opening; + }); + assert.equal(opened.success, true, 'the stale receipt is not trusted either way; the retry answers for x'); + assert.equal(f.isOpen('x'), true); + assert.equal(f.substituted().length, 1, 'the substitution the server made is still reported'); + await tick(); + assert.deepEqual(f.calls.filter((c) => c.id === 'y'), [{ kind: 'close', id: 'y' }], + 'and the channel it opened instead is still closed again'); +}); + +test("the substitution line and its trace print each id as the module's other log lines do, an odd one escaped", async () => { + const forged = 'y\n[channel-open-substituted] server=other requested=x answered=x'; + const f = fixture(() => ({ channel: { ...descriptor('y'), id: forged } })); + await f.registry.handleChanged('test', { added: [descriptor('x')] }); + const { lines } = await capturingErrors(() => f.openX()); + const logged = lines.filter((l) => l.includes('channel-open-substituted')); + assert.deepEqual(logged, [`[channel-open-substituted] server=test requested=x answered=${logValue(forged)}`]); + const answered = logged[0]!.slice(logged[0]!.indexOf(' answered=') + ' answered='.length); + assert.doesNotMatch(answered, /[\n =]/, 'the forged id adds no line, field or value of its own'); + assert.equal(f.substituted()[0]?.answeredFor, logValue(forged), 'the trace records the same escaped form'); +}); + +test('an answer whose channel id is not a string fails the open, and closes nothing', async () => { + for (const answer of [ + () => ({ channel: { ...descriptor('y'), id: 5 } }) as unknown as ChannelsOpenResult, + () => ({ history: [{ ...historyOf('y')[0]!, channelId: false }] }) as unknown as ChannelsOpenResult, + ]) { + const f = fixture(answer); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + f.calls.length = 0; + const { value: result } = await capturingErrors(() => f.openX()); + await tick(); + assert.equal(result.success, false); + assert.match(String(result.error), /with a channel id that is not a string/); + assert.equal(f.isOpen('x'), false); + assert.match(String(f.substituted()[0]?.answeredFor), /^(5|false)$/, 'logValue prints a non-string id as its JSON text'); + assert.deepEqual(f.calls, [{ kind: 'open', id: 'x' }], 'nothing registered by that id to close'); + } +}); + +test('an answer for the requested channel, with or without its history, or one naming no channel, opens it as before', async () => { + for (const answer of [ + (requested: string) => ({ channel: descriptor(requested), history: historyOf(requested) }), + (requested: string) => ({ channel: descriptor(requested) }), + () => ({}) as ChannelsOpenResult, + // A null id names no channel, in the answer or in a history item. + (requested: string) => ({ channel: { ...descriptor(requested), id: null } }) as unknown as ChannelsOpenResult, + (requested: string) => ({ history: [{ ...historyOf(requested)[0]!, channelId: null }] }) as unknown as ChannelsOpenResult, + ]) { + const f = fixture(answer); + await f.registry.handleChanged('test', { added: [descriptor('x'), descriptor('y')] }); + f.calls.length = 0; + const result = await f.openX(5); + assert.equal(result.success, true); + assert.equal(f.isOpen('x'), true); + assert.equal(f.substituted().length, 0); + assert.deepEqual(f.calls, [{ kind: 'open', id: 'x' }]); + } +}); diff --git a/test/mcpl-gate-roundtrip.test.ts b/test/mcpl-gate-roundtrip.test.ts index 3ede04df9..1d10485b4 100644 --- a/test/mcpl-gate-roundtrip.test.ts +++ b/test/mcpl-gate-roundtrip.test.ts @@ -35,7 +35,9 @@ function makeGate(config: GateConfig) { function makeRegistry(shouldTriggerInference?: (c: string, m: Record) => boolean) { const pushed: ProcessEvent[] = []; const registry = new ChannelRegistry( - {} as never, + // No server connection: the host directs no channel lifecycle here, so + // traffic is what marks a channel open. + { getServer: () => undefined } as never, {} as never, (ev) => pushed.push(ev), () => {}, diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index 444bdc09a..7809a1dc2 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -623,7 +623,7 @@ describe('notices about an agent\'s own action', () => { state.recordChannelAutoOpenNotice(fork.name, [{ channelId: 'c-1', label: '#one' }], 'opened-by-reply'); assert.equal(forkCm.mock.callCount(), 1); assert.equal(primary.mock.callCount(), 0); - assert.match(JSON.stringify(forkCm.mock.calls[0]!.arguments[1]), /Now open: #one \(c-1\) — your own message into the closed channel engaged it/); + assert.match(JSON.stringify(forkCm.mock.calls[0]!.arguments[1]), /Now open: #one \(c-1\): your own message into the closed channel opened it/); state.recordChannelAutoOpenNotice(undefined, [{ channelId: 'c-2' }], 'subscription-policy'); assert.equal(primary.mock.callCount(), 1, 'no agent caused a policy open'); diff --git a/test/trunk-channel-routing.test.ts b/test/trunk-channel-routing.test.ts index 0852c2ca5..2da99d131 100644 --- a/test/trunk-channel-routing.test.ts +++ b/test/trunk-channel-routing.test.ts @@ -424,7 +424,7 @@ describe('Trunk channel routing (item-3 redux)', () => { const one = lastText(); assert.match( one, - /Now open: #general \(acme\) \(slack:C1\) — your own message into the closed channel engaged it\. What an open channel delivers to you depends on its server\. If you don't want it, use channel_close/, + /Now open: #general \(acme\) \(slack:C1\): your own message into the closed channel opened it\. What an open channel delivers to you depends on its server\. If you don't want it, use channel_close/, ); assert.doesNotMatch(one, /will reach you/); @@ -432,7 +432,7 @@ describe('Trunk channel routing (item-3 redux)', () => { const two = lastText(); assert.match( two, - /Now open: slack:C1, slack:C2 — your subscription policy just admitted them\. What an open channel delivers to you depends on its server\. If you don't want one of them, use channel_close/, + /Selected for opening by subscription policy: slack:C1, slack:C2\. This records the admission decision; channel_list shows current transport state\. What an open channel delivers to you depends on its server\. If you don't want one of them, use channel_close/, ); assert.doesNotMatch(two, /will reach you/); await framework.stop();