diff --git a/changelog.d/channel-notices-reach-who-can-close.fixed.md b/changelog.d/channel-notices-reach-who-can-close.fixed.md new file mode 100644 index 000000000..70a1f2bb4 --- /dev/null +++ b/changelog.d/channel-notices-reach-who-can-close.fixed.md @@ -0,0 +1,28 @@ +- The notice that a channel opened without the agent asking (an open its + own send or reply caused, or a subscription-policy admission) names + `channel_close` only to an agent that can call it there for every channel + it names, except where several residents share one slot (#291), and in + forks spawned from the primary afterwards (#235). A conversation fork can + close only its home channel, and an agent's `allowedTools` can leave + `channel_close` out. Before, such an agent + was told to use `channel_close`, and couldn't call it, or its call was + refused. Where an agent's own send opened a channel it can't close, the + primary is now told by name too, if it can close it, and the agent is told + that it was. +- The opened notice says "What an open channel delivers to you depends on + its server" only to an agent whose window holds the channel's traffic. + Before, a conversation fork was told this about a channel other than its + home, whose messages never reach it. +- The primary's notices name a conversation fork by the channel it's bound + to, then by its generated name. The notice of an open by an agent that has + ended says "a message from" it, where it said "opened by X's reply" whether + a reply or speech had opened the channel. The reply notice and speech's + failure notice say why a channel is selected for opening again with + "because", where they said "as". +- When a conversation fork or an ephemeral run is released without reading a + channel notice it could act on (a channel its own send opened, or a + channel its reply couldn't open that is selected for opening), the primary + is told by name in its place, as for a reply whose delivery failed, of what + still holds then: a channel closed since is left out, and so is one no + longer selected for opening, or open since. Before, the notice stayed + unread in the released agent's window. diff --git a/src/framework.ts b/src/framework.ts index 2df4807f1..14d2bd10a 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -10,7 +10,7 @@ import { import { JsStore } from '@animalabs/chronicle'; import type { Membrane, ContentBlock, NormalizedRequest, NormalizedResponse, YieldingStream, DetailedUsage, ToolResult as MembraneToolResult, ToolResultContentBlock } from '@animalabs/membrane'; import { MembraneError } from '@animalabs/membrane'; -import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError, stripUnpairedToolBlocks } from '@animalabs/context-manager'; +import { ContextManager, PassthroughStrategy, WindowedPassthroughStrategy, OverBudgetError, UncoveredDropError, stripUnpairedToolBlocks, continuesBody } from '@animalabs/context-manager'; import type { CacheWireReceipt } from './kv-unified-wire.js'; import { SILENT_HEARTBEAT_SEPARATOR, type RequestOnlyPrompt, type RequestOnlyStamps } from './silent-heartbeat.js'; import { SUBCONSCIOUS_TOOLS, SUBCONSCIOUS_TOOL_NAMES, type SubconsciousConfig } from './tune-out/tools.js'; @@ -103,7 +103,7 @@ import { isSilentHeartbeatMarker, isVisiblyEmptyContent, NON_STRING_TEXT_PLACEHO import { parseProsePrefix, parseHybridProsePrefix } from './mcpl/prose-grammar.js'; import { detectKnownToolWrapperProse } from './tool-wrapper-prose-guard.js'; import { InferenceRouter } from './mcpl/inference-router.js'; -import { ChannelRegistry, connectorText, destinationKey, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome, type SelectedForOpening } from './mcpl/channel-registry.js'; +import { ChannelRegistry, connectorText, desiredStateInLog, destinationKey, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome, type SelectedForOpening } from './mcpl/channel-registry.js'; import { KEEP_IDLE_MS, KEEP_NEWEST, ProseDraftStore, draftState, uncertainAttempt, type Draft, type DraftAttempt, type DraftReason, type DraftState, type InheritedRisk } from './prose-drafts.js'; import { ConversationRouter, CONVERSATION_ROUTING_DEPRECATION_NOTICE } from './mcpl/conversation-router.js'; import { INBOUND_SOURCE_KEY, sourceBodyDigest, type InboundAcceptanceObserver, type InboundChannelSource, type InboundLane, type InboundSource, type InboundUnscopedSource } from './mcpl/inbound-source.js'; @@ -2277,6 +2277,89 @@ type McpRestartClaim = { readonly restarting: string; endedBy?: string }; * agent and stream, before the cancel, for the stream's terminal to read. */ type FrameworkCancelKind = 'turn_ended' | 'budget_restart' | 'quiesce_abandoned' | 'shutdown'; +/** What opened a channel the agent didn't ask to open + * (recordChannelAutoOpenNotice). */ +type ChannelOpenedCause = 'subscription-policy' | 'opened-by-delivery' | 'opened-by-reply'; + +/** A channel notice as its builder returns it: its text, its metadata, whom + * it's addressed to (none: where addMessage delivers an unrouted message), + * and what its log line says. A reader's notice that says the primary was + * told carries `untold`, its words without saying so, which + * writeChannelNotices writes instead where the write of the primary's + * notice, before it, failed. */ +interface ChannelNotice { + text: string; + untold?: string; + metadata: MessageMetadata; + target?: { forAgent: string }; + told: string; + shown: string; +} + +/** + * A channel notice's inputs, kept in its metadata (`passOn`) where its reader + * was told it could act: the server its channels are on, and what its + * builder was given. A release that finds the notice unread tells the + * primary in its place, by calling the notice's builder again for the reader + * then gone, and checks each channel against that server's decision + * (channelNoticeForPrimary). A notice whose reader couldn't act carries none: + * the primary was told then, where it could act. + */ +type ChannelNoticePassOn = + | { notice: 'channel-opened'; serverId: string; cause: ChannelOpenedCause; channels: Array<{ channelId: string; label?: string }> } + | { notice: 'reply-open-failed'; serverId: string; channel: { channelId: string; label?: string }; selectedForOpening: SelectedForOpening }; + +/** A channel as the channel notices show it (the opened and reply notices, + * and the lines that log them): its label beside its id, or its id alone. */ +function shownChannel(channel: { channelId: string; label?: string }): string { + return channel.label && channel.label !== channel.channelId ? `${channel.label} (${channel.channelId})` : channel.channelId; +} + +/** A channel as a notice's inputs carry it, read back from stored metadata. */ +function noticeChannel(value: unknown): { channelId: string; label?: string } | null { + const channel = value as { channelId?: unknown; label?: unknown } | null | undefined; + if (typeof channel?.channelId !== 'string' || channel.channelId === '') return null; + return typeof channel.label === 'string' ? { channelId: channel.channelId, label: channel.label } : { channelId: channel.channelId }; +} + +/** A channel notice's inputs (`passOn`), read back from its metadata, which + * a store or a queue may have kept from another build: null when it + * carries none, or none this build reads. */ +function channelNoticePassOn(metadata: unknown): ChannelNoticePassOn | null { + const meta = metadata as { kind?: unknown; passOn?: unknown } | null | undefined; + if (meta?.kind !== 'channel-notice') return null; + const passOn = meta.passOn as { + notice?: unknown; serverId?: unknown; cause?: unknown; channels?: unknown; channel?: unknown; selectedForOpening?: unknown; + } | null | undefined; + if (typeof passOn?.serverId !== 'string' || passOn.serverId === '') return null; + const serverId = passOn.serverId; + if (passOn.notice === 'channel-opened') { + // A policy admission's notice is addressed to no agent of its own and + // carries no inputs (channelOpenedNotices), so only an agent's send reads + // back. + const causes: readonly unknown[] = ['opened-by-delivery', 'opened-by-reply']; + if (!causes.includes(passOn.cause) || !Array.isArray(passOn.channels) || passOn.channels.length === 0) return null; + const channels = passOn.channels.map(noticeChannel); + if (channels.some((c) => c === null)) return null; + return { notice: 'channel-opened', serverId, cause: passOn.cause as ChannelOpenedCause, channels: channels as Array<{ channelId: string; label?: string }> }; + } + if (passOn.notice === 'reply-open-failed') { + const selected: readonly unknown[] = ['unrecorded', 'server-bootstrap', 'subscription-policy']; + const channel = noticeChannel(passOn.channel); + if (!channel || !selected.includes(passOn.selectedForOpening)) return null; + return { notice: 'reply-open-failed', serverId, channel, selectedForOpening: passOn.selectedForOpening as SelectedForOpening }; + } + return null; +} + +/** Whether a notice an agent never read goes to the primary in its place + * when the agent ends: a failed reply's, or a channel notice that carries + * its inputs, since its reader was told it could act (ChannelNoticePassOn). */ +function passesToPrimary(metadata: unknown): boolean { + return (metadata as { kind?: unknown } | null | undefined)?.kind === 'discord-send-failed' || + channelNoticePassOn(metadata) !== null; +} + export class AgentFramework { private toolPresentations = new Map(); private presentationPreviews = new WeakMap(); @@ -3983,23 +4066,24 @@ export class AgentFramework { }); } // A window release the quiesce postponed (releasesAfterResume) never runs - // once the host stops, so its unread pass runs here: each failed reply - // stored unread in that window goes to the primary now, deferred while - // quiesced like any write, so it waits in the queue with the rest. What - // was queued for that agent goes as boot would send it (readdDrained): a - // failed reply to the primary, deferred again under its own id, and - // anything else dropped, logged, since nothing lands while quiesced. So - // no flush that comes later meets a failed reply of an agent that has - // gone, not even one that writes into that agent's own window, as a turn - // of the fork's still assembling or at a tool boundary does - // (flushForStart). Each release's pass is a step of its own, and so is - // the drain: one that throws fails the stop() and leaves the others, and - // the final flush, to run. It runs again after the waits below, for a - // release an end they let settle postpones. + // once the host stops, so its unread pass runs here: each notice stored + // unread in that window that the primary takes in its place (a failed + // reply, or a channel notice its reader could act on) goes to the + // primary now, deferred while quiesced like any write, so it waits in the + // queue with the rest. What was queued for that agent goes as boot would + // send it (readdDrained): such a notice to the primary, deferred again + // under its own id, and anything else dropped, logged, since nothing + // lands while quiesced. So no flush that comes later meets such a notice + // of an agent that has gone, not even one that writes into that agent's + // own window, as a turn of the fork's still assembling or at a tool + // boundary does (flushForStart). Each release's pass is a step of its + // own, and so is the drain: one that throws fails the stop() and leaves + // the others, and the final flush, to run. It runs again after the waits + // below, for a release an end they let settle postpones. const passPostponedReleases = () => { for (const agent of this.releasesAfterResume.values()) { step(() => { - for (const metadata of this.unreadFailedReplies(agent)) this.passFailedReplyToPrimary(agent.name, metadata); + for (const metadata of this.unreadNoticesForPrimary(agent)) this.passNoticeToPrimary(agent.name, metadata); }); } const released = [...this.releasesAfterResume.keys()]; @@ -4123,9 +4207,10 @@ export class AgentFramework { // still loses it at exit. An entry for a target no longer registered, // such as one a release postponed after the pass above leaves, or one a // quiesced boot restored for an agent gone before a crash, is left for - // the flush that meets it, which passes a failed reply on to the primary - // and drops anything else, logged (readdDrained): with a persisted queue, - // the boot or resume flush. A memory-only queue loses it at exit. + // the flush that meets it, which passes a failed reply, or a channel + // notice its reader could act on, on to the primary and drops anything + // else, logged (readdDrained): with a persisted queue, the boot or resume + // flush. A memory-only queue loses it at exit. // // A registered agent's stream that outlasted its wait, or whose cancel // threw, may still write, so that agent's entries stay queued, said once @@ -6579,7 +6664,8 @@ export class AgentFramework { try { // The windows of forks and runs deregistered during the window are // released first, each landing its queued writes and passing on its - // unread failed replies, as at any release. Then flush context writes + // unread failed replies, and the channel notices it could act on, as at + // any release. Then flush context writes // deferred by the quiesce window (module events, api message.send) — // with no turn alive there is no other flush point. The batch is // stored in queue order and acknowledged once, after the chronicle @@ -6651,7 +6737,8 @@ export class AgentFramework { this.handOffDeferredWrites(flush); // A write for a target no longer registered, such as a fork or a run gone // before a crash that left it to boot recovery, goes as readdDrained - // sends it: a failed reply to the primary, anything else dropped, logged. + // sends it: a failed reply, or a channel notice its reader could act on, + // to the primary, anything else dropped, logged. const stored = this.readdDrained(flush, `host-mode ${label}`).length; // Ack = sync the chronicle FIRST, then rewrite the durable queue without // the batch. Never the other way round. @@ -8438,9 +8525,10 @@ export class AgentFramework { // turn's end, is left to its window's release below // (releaseWindowNowOrAfterResume), as for a disposed fork: it lands // there in the run's store, which keeps the run's namespace for - // investigation, now or, while quiesced, at resume. A failed reply - // among it goes on to the primary as the same queue entry, so no ack - // comes between the entry and the primary's notice. + // investigation, now or, while quiesced, at resume. A failed reply, or + // a channel notice the run could act on, among it goes on to the + // primary as the same queue entry, so no ack comes between the entry + // and the primary's notice. // Exact-generation disposal owns these name-keyed entries. Remove them // before deregistration; the late driveStream finally will see the // generation mismatch and leave any future owner untouched. A fork's @@ -15288,60 +15376,228 @@ export class AgentFramework { * 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. + * confirmed. It names `channel_close` as the opt-out (their decision + * outranks policy) only where the agent it's addressed to can call it for + * every channel it names (canCloseChannels): a conversation fork may close + * only its home channel, and an agent's allowedTools can leave + * channel_close out. Others can read it too: the other residents, where + * several share one slot (#291), and forks spawned from the primary + * afterwards, from its window (#235). Where an + * agent's own send opened a channel it can't close, the primary is told by + * name too, if it's registered and can close it, and the agent is told + * that it was. What an open channel delivers is its server's to say (#406), + * so the notice doesn't promise its traffic, and it says even that only to + * an agent whose window holds the channel's traffic (holdsChannelTraffic). + * System message: never triggers inference, never conversational, never + * locus-eligible. */ private recordChannelAutoOpenNotice( agentName: string | undefined, + serverId: string, channels: Array<{ channelId: string; label?: string }>, - cause: 'subscription-policy' | 'opened-by-delivery' | 'opened-by-reply', + cause: ChannelOpenedCause, ): void { if (channels.length === 0) return; - const shown = channels - .map((c) => (c.label && c.label !== c.channelId ? `${c.label} (${c.channelId})` : c.channelId)) - .join(', '); + this.writeChannelNotices(this.channelOpenedNotices(agentName, serverId, channels, cause), `auto-open notice (${cause})`, 'recordChannelAutoOpenNotice'); + } + + /** + * Write a builder's channel notices in order, each on its own: one that + * can't be written is logged, and the rest are still written. A reader's + * notice that says the primary was told comes after the primary's, and is + * written in its `untold` words where the primary's write failed, so it + * never says the primary was told of a notice that wasn't written. A write + * addMessage defers to the primary's next boundary, as it does any write + * to an agent mid-turn, counts as written: a memory-only queue loses it at + * exit, as it loses any deferred write. Through + * framework.addMessage for mid-turn safety (deferral + hear-while-acting + * injection): an open a delivery forces happens inside a live turn, and a + * direct context-manager append there would store the notice out of order. + */ + private writeChannelNotices(notices: readonly ChannelNotice[], what: string, writer: string): void { + let allWritten = true; + for (const notice of notices) { + const text = allWritten || notice.untold === undefined ? notice.text : notice.untold; + try { + this.addMessage('user', [{ type: 'text', text }], notice.metadata, notice.target); + console.error(`[channel] ${what} -> ${notice.told}: ${notice.shown}`); + } catch (err) { + allWritten = false; + console.error(`${writer} failed:`, err); + } + } + } + + /** + * The notices recordChannelAutoOpenNotice writes, in order: the primary's + * by name where the reader can't close the channels, then the reader's, + * which then says the primary was told (`untold`: its words where that + * didn't land). None, logged, for an agent that has ended when there's no + * primary to tell. + * The reader's carries its inputs (`passOn`), the channels' server among + * them, where it's addressed to the agent whose send opened the channels + * and that agent was told it can close them, so a release that finds it + * unread can tell the primary in its place (channelNoticeForPrimary), which + * calls this again for the agent then gone. A policy admission's goes where + * addMessage delivers it, to no agent of its own, and carries none. + */ + private channelOpenedNotices( + agentName: string | undefined, + serverId: string, + channels: Array<{ channelId: string; label?: string }>, + cause: ChannelOpenedCause, + ): ChannelNotice[] { + const shown = channels.map(shownChannel).join(', '); + const ids = channels.map((c) => c.channelId); 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 outranks policy.`; + // What the channels deliver is said only to an agent that holds their + // traffic (holdsChannelTraffic), and channel_close is named only to one + // that can call it for every one of them. + const optOut = (holds: boolean, canClose: boolean) => + (holds ? ' What an open channel delivers to you depends on its server.' : '') + + (canClose ? ` 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 - // after its release) can close nothing, and no channel opens without - // someone being told who can close it: the primary, the one left, is told - // by name, as for a failed reply. The default path would fall back to the - // first registered agent when there is no primary. - let text: string; - let metadata = { system: true, kind: 'channel-notice' } as MessageMetadata; - let target: { forAgent: string } | undefined; - let told: string; - if (agentName && !this.agents.has(agentName)) { - if (!this.primaryAgentName || !this.agents.has(this.primaryAgentName)) { + // after its release) can close nothing, so the primary, the one left, is + // told by name, as for a failed reply, whether or not it can close the + // channels: its advice is judged for it. The default path would fall + // back to the first registered agent when there is no primary. + const primary = this.primaryAgentName ?? undefined; + const ended = !!agentName && !this.agents.has(agentName); + if (ended) { + if (!primary || !this.agents.has(primary)) { console.error(`[channel] ${agentName} has ended, and there is no primary to tell that ${shown} opened`); - return; + return []; } - 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 { - 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'; + const sender = this.agentForPrimary(agentName); + return [{ + text: `[channels] Now open: ${shown}: a message from ${sender.first} into the closed channel opened it; ${sender.again} has ended.` + + optOut(this.holdsChannelTraffic(primary, ids), this.canCloseChannels(primary, ids)), + metadata: { system: true, kind: 'channel-notice', speaker: agentName } as MessageMetadata, + told: `the primary, by name (${agentName} has ended)`, + shown, + }]; + } + // The reader: the agent whose send opened the channels, or for a policy + // admission, the agent addMessage delivers to (deliveryAgent). + const who = agentName || this.deliveryAgent()?.name; + const canClose = this.canCloseChannels(who, ids); + // Where the agent whose send opened them can't close them, as a fork away + // from its home can't, or an agent whose allowedTools leave channel_close + // out, the primary is told by name too, where it's registered and can + // close them, so an agent that can close the channels hears they opened, + // and the agent hears who was told. (The primary can't be that agent: it + // can close them, and that agent can't.) + const tellPrimary = !!agentName && !canClose && primary !== undefined && this.agents.has(primary) && + this.canCloseChannels(primary, ids); + const readerText = (told: string) => (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(this.holdsChannelTraffic(who, ids), canClose) + told; + const reader: ChannelNotice = { + text: readerText(tellPrimary ? ` ${this.primaryForReader(agentName, primary)} has been told and can close ${plural ? 'them' : 'it'}.` : ''), + ...(tellPrimary ? { untold: readerText('') } : {}), + metadata: { + system: true, kind: 'channel-notice', + ...(agentName && canClose ? { passOn: { notice: 'channel-opened', serverId, cause, channels } satisfies ChannelNoticePassOn } : {}), + } as MessageMetadata, + ...(agentName ? { target: { forAgent: agentName } } : {}), + told: agentName ?? 'primary', + shown, + }; + const notices: ChannelNotice[] = []; + if (tellPrimary) { + const sender = this.agentForPrimary(agentName); + notices.push({ + text: `[channels] Now open: ${shown}: a message from ${sender.first} into the closed channel opened it, and ${sender.again} can't close ${plural ? 'them' : 'it'}.` + + optOut(this.holdsChannelTraffic(primary, ids), true), + metadata: { system: true, kind: 'channel-notice', speaker: agentName } as MessageMetadata, + told: `the primary, by name (${agentName} can't close ${plural ? 'them' : 'it'})`, + shown, + }); } - try { - // Through framework.addMessage for mid-turn safety (deferral + - // hear-while-acting injection) — a delivery-forced open happens inside - // a live turn, and a direct context-manager append there would store - // the notice out of order. - this.addMessage('user', [{ type: 'text', text }], metadata, target); - console.error(`[channel] auto-open notice (${cause}) -> ${told}: ${shown}`); - } catch (err) { - console.error('recordChannelAutoOpenNotice failed:', err); + notices.push(reader); + return notices; + } + + /** + * Whether `who` can call channel_close for every one of `channelIds` here: + * its allowedTools hold it (canAgentUseTool), and a conversation fork, + * whose channel_close dispatchChannelToolCall refuses for any channel but + * its home, only for its home. + */ + private canCloseChannels(who: string | undefined, channelIds: readonly string[]): boolean { + if (who === undefined) return false; + const home = this.conversationAgentHomes.get(who); + return channelIds.every((id) => !home || home === id) && this.canAgentUseTool(who, 'channel_close'); + } + + /** + * Whether `who` holds the traffic of every one of `channelIds`, so that + * what an open channel delivers is in its window. Where conversations are + * routed, an incoming message goes to the fork its channel is bound to, or + * nowhere, and never to the primary (handleMcplChannelIncoming), so a fork + * holds only its home's, and only while the router binds its home to it. + * An idle expiry's sweep unbinds the home before the fork's closure turn + * (ConversationRouter.expired), and from then the closing fork holds none; + * a newer fork the home is bound to holds it instead. Elsewhere an + * incoming message is stored with no forAgent, in deliveryAgent()'s store, + * and where that's the residents' shared slot, every agent that reads the + * slot holds it (sharedSlotAgents: the ordinary residents, and the + * subconscious through its merged view); a run's store is its own. (A push + * names its channel and lands in the primary's window either way, but it + * comes whether or not the channel is open, so it isn't what opening one + * delivers.) + */ + private holdsChannelTraffic(who: string | undefined, channelIds: readonly string[]): boolean { + if (who === undefined) return false; + const home = this.conversationAgentHomes.get(who); + if (home !== undefined) { + return channelIds.every((id) => id === home && this.conversationRouter?.getBinding(id)?.agentName === who); } + if (this.conversationRouter) return false; + const delivery = this.deliveryAgent(); + const agent = this.agents.get(who); + if (delivery === undefined || agent === undefined) return false; + return agent === delivery || (this.sharedSlotAgents.has(agent) && this.isOrdinaryResident(delivery)); + } + + /** + * How a notice to the primary names `agentName`: a conversation fork by the + * channel it's bound to, as notices name channels, with its own name once, + * after "named", and as "that fork" after that ("your fork" where the + * primary is the template forks are made from); any other agent by its + * name. `first` is said mid-sentence, so a fork's ends with a comma. A + * fork's home is known while it's registered and through its release; a + * notice passed on later, at resume or at boot, names it by its name. + */ + private agentForPrimary(agentName: string): { first: string; again: string } { + const home = this.conversationAgentHomes.get(agentName); + if (home === undefined) return { first: agentName, again: agentName }; + const label = this.channelRegistry?.getDescriptor(home)?.label; + const bound = label && label !== home ? `${label} (${home})` : home; + const fork = this.forkOfPrimary(agentName) ? 'your fork' : 'the fork'; + return { first: `${fork} in ${bound}, named ${agentName},`, again: 'that fork' }; + } + + /** + * How a notice to `reader` names the primary it says was told: "Your + * primary" to a conversation fork made from the primary, whose + * configuration is the primary's with only its name changed, so its prompt + * may give the primary's name as its own; the primary's name to any other + * agent, a fork made from another template included. + */ + private primaryForReader(reader: string | undefined, primary: string): string { + return reader !== undefined && this.forkOfPrimary(reader) ? 'Your primary' : primary; + } + + /** Whether `agentName` is a conversation fork made from the primary: the + * router's template is the primary (connectome-host's is). */ + private forkOfPrimary(agentName: string): boolean { + return this.conversationAgentHomes.has(agentName) && this.primaryAgentName !== null && + this.conversationRouter?.templateAgent === this.primaryAgentName; } /** @@ -15357,9 +15613,12 @@ export class AgentFramework { * opening, the primary is told by name in its place, if one is registered, * as the opened notice tells it of an open by an agent that has ended, so * an agent that could close the channel hears it may open. It names - * channel_open as the way to try again only to an agent that can call it: - * a conversation fork's channel_open is refused outright + * channel_open as the way to try again only where the agent it's addressed + * to can call it: a conversation fork's channel_open is refused outright * (dispatchChannelToolCall), and an agent's allowedTools can leave it out. + * Others can read it too: the other residents, where several share one + * slot (#291), and forks spawned from the primary afterwards, from its + * window (#235). * Where the channel is selected for opening, it says so and why, as * speech's failure does, and what that means: where the host couldn't * record putting the channel's decision back, a later delivery or @@ -15368,88 +15627,103 @@ export class AgentFramework { * default put back, the host tries again itself, so channel_open isn't * named. Either way channel_close withdraws that, named only to an agent * that can call it here (a fork may close only its home channel). Where - * the reader can't, the primary is told by name too, if it's registered, - * isn't the reader and can close it. System message, as the opened notice - * is. + * the reader can't, the primary is told by name too, if it's registered + * and can close it, and the reader is told that it was. System message, as + * the opened notice is. */ private recordReplyOpenFailedNotice( agentName: string | undefined, + serverId: string, channel: { channelId: string; label?: string }, selectedForOpening?: SelectedForOpening, ): void { - const shown = channel.label && channel.label !== channel.channelId - ? `${channel.label} (${channel.channelId})` - : channel.channelId; + this.writeChannelNotices(this.replyOpenFailedNotices(agentName, serverId, channel, selectedForOpening), 'reply open-failed notice', 'recordReplyOpenFailedNotice'); + } + + /** + * The notices recordReplyOpenFailedNotice writes, in order: the primary's + * by name where the reader can't close a channel selected for opening, + * then the reader's, which then says the primary was told (`untold`: its + * words where that didn't land). None, logged, where the agent has ended and the channel + * stays closed, or there's no primary to tell. The reader's carries its + * inputs (`passOn`), the channel's server among them, where it's addressed + * to the agent whose reply it was, the channel is selected for opening and + * that agent was told it can close it, so a release that finds it unread + * can tell the primary in its place (channelNoticeForPrimary). + */ + private replyOpenFailedNotices( + agentName: string | undefined, + serverId: string, + channel: { channelId: string; label?: string }, + selectedForOpening?: SelectedForOpening, + ): ChannelNotice[] { + const shown = shownChannel(channel); const ended = !!agentName && !this.agents.has(agentName); if (ended && !selectedForOpening) { console.error(`[channel] ${agentName} has ended; not telling it that ${shown} couldn't be opened`); - return; + return []; } const primary = this.primaryAgentName ?? undefined; if (ended && (!primary || !this.agents.has(primary))) { console.error(`[channel] ${agentName} has ended, and there is no primary to tell that ${shown} couldn't be opened and is selected for opening`); - return; + return []; } const who = ended ? primary : agentName || this.deliveryAgent()?.name; const canOpen = who !== undefined && !this.conversationAgentHomes.has(who) && this.canAgentUseTool(who, 'channel_open'); - // A fork's channel_close is refused for any channel but its home, as - // dispatchChannelToolCall reads it. - const home = who !== undefined ? this.conversationAgentHomes.get(who) : undefined; - const canClose = who !== undefined && (!home || home === channel.channelId) && - this.canAgentUseTool(who, 'channel_close'); + const canClose = this.canCloseChannels(who, [channel.channelId]); const selected = selectedForOpening === 'unrecorded' - ? " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again" + ? " It stays selected for opening, because the host couldn't record undoing that: a later delivery or reconnect tries again" : selectedForOpening !== undefined - ? ` It's selected for opening again, as ${selectedForOpening === 'server-bootstrap' ? 'its server announces it open' : 'the subscription policy admits it'}: the host tries again` + ? ` It's selected for opening again, because ${selectedForOpening === 'server-bootstrap' ? 'its server lists it as open by default' : 'the subscription policy admits it'}: the host tries again` : ''; const hostRetries = selectedForOpening === 'server-bootstrap' || selectedForOpening === 'subscription-policy'; - const text = `[channels] Not open: ${shown}. ` + - (ended - ? `${agentName}'s reply there went out, but opening the channel failed; ${agentName} has ended.` - : 'Your reply there went out, but opening the channel failed.') + - (selected ? selected + (canClose ? ', and channel_close withdraws that.' : '.') : '') + - (canOpen && !hostRetries ? ' channel_open tries again.' : ''); - try { - this.addMessage( - 'user', - [{ type: 'text', text }], - (ended - ? { system: true, kind: 'channel-notice', speaker: agentName } - : { system: true, kind: 'channel-notice' }) as MessageMetadata, - agentName && !ended ? { forAgent: agentName } : undefined, - ); - console.error( - `[channel] reply open-failed notice -> ${ended ? `the primary, by name (${agentName} has ended)` : who ?? 'no agent'}: ${shown}`, - ); - } catch (err) { - console.error('recordReplyOpenFailedNotice failed:', err); - } // Where the reader can't close a channel selected for opening, as a fork // away from its home can't, or an agent whose allowedTools leave // channel_close out, the primary is told by name too, where it's // registered and can close it, so an agent that could close the channel - // hears it may open. - const primaryHome = primary !== undefined ? this.conversationAgentHomes.get(primary) : undefined; - if ( - !ended && selectedForOpening !== undefined && who !== undefined && !canClose && - primary !== undefined && primary !== who && this.agents.has(primary) && - (!primaryHome || primaryHome === channel.channelId) && this.canAgentUseTool(primary, 'channel_close') - ) { - const primaryCanOpen = !this.conversationAgentHomes.has(primary) && this.canAgentUseTool(primary, 'channel_open'); - const told = `[channels] Not open: ${shown}. ${who}'s reply there went out, but opening the channel failed, and ${who} can't close it.` + - selected + ', and channel_close withdraws that.' + - (primaryCanOpen && !hostRetries ? ' channel_open tries again.' : ''); - try { - this.addMessage( - 'user', - [{ type: 'text', text: told }], - { system: true, kind: 'channel-notice', speaker: who } as MessageMetadata, - ); - console.error(`[channel] reply open-failed notice -> the primary, by name (${who} can't close it): ${shown}`); - } catch (err) { - console.error('recordReplyOpenFailedNotice failed:', err); - } + // hears it may open, and the reader hears who was told. (The primary + // can't be the reader: it can close it, and the reader can't.) + const tellPrimary = !ended && selectedForOpening !== undefined && who !== undefined && !canClose && + primary !== undefined && this.agents.has(primary) && + this.canCloseChannels(primary, [channel.channelId]); + const sender = ended ? this.agentForPrimary(agentName!) : undefined; + const readerText = (told: string) => `[channels] Not open: ${shown}. ` + + (ended + ? `A reply from ${sender!.first} went out there, but opening the channel failed; ${sender!.again} has ended.` + : 'Your reply there went out, but opening the channel failed.') + + (selected ? selected + (canClose ? ', and channel_close withdraws that.' : '.') : '') + + told + + (canOpen && !hostRetries ? ' channel_open tries again.' : ''); + const notice: ChannelNotice = { + text: readerText(tellPrimary ? ` ${this.primaryForReader(who, primary)} has been told, and can withdraw that.` : ''), + ...(tellPrimary ? { untold: readerText('') } : {}), + metadata: (ended + ? { system: true, kind: 'channel-notice', speaker: agentName } + : { + system: true, kind: 'channel-notice', + ...(agentName && selectedForOpening !== undefined && canClose + ? { passOn: { notice: 'reply-open-failed', serverId, channel, selectedForOpening } satisfies ChannelNoticePassOn } + : {}), + }) as MessageMetadata, + ...(agentName && !ended ? { target: { forAgent: agentName } } : {}), + told: ended ? `the primary, by name (${agentName} has ended)` : who ?? 'no agent', + shown, + }; + const notices: ChannelNotice[] = []; + if (tellPrimary) { + const primaryCanOpen = !this.conversationAgentHomes.has(primary!) && this.canAgentUseTool(primary!, 'channel_open'); + const reader = this.agentForPrimary(who!); + notices.push({ + text: `[channels] Not open: ${shown}. A reply from ${reader.first} went out there, but opening the channel failed, and ${reader.again} can't close it.` + + selected + ', and channel_close withdraws that.' + + (primaryCanOpen && !hostRetries ? ' channel_open tries again.' : ''), + metadata: { system: true, kind: 'channel-notice', speaker: who } as MessageMetadata, + told: `the primary, by name (${who} can't close it)`, + shown, + }); } + notices.push(notice); + return notices; } /** @@ -21580,22 +21854,23 @@ export class AgentFramework { * landed in, and its newest stays the newest (remainingArrival). * * A target no longer registered (a fork or a run that has ended, here or - * before a crash) has no window to land in: a failed reply it never read - * becomes the primary's by-name notice, under the entry's own id and - * place in the queue (queuedFailedReplyForPrimary), and is written with - * the rest; any other write is dropped, logged, by addMessage. */ + * before a crash) has no window to land in: a failed reply it never read, + * or a channel notice it could have acted on, becomes the primary's + * by-name notice, under the entry's own id and place in the queue + * (queuedNoticeForPrimary), and is written with the rest; any other write + * is dropped, logged, by addMessage. */ private readdDrained(drained: DeferredWrite[], where: string, opts?: { bypassDeferral?: boolean }): DeferredWrite[] { let passed = 0; const entries = drained.flatMap((msg) => { if (msg.forAgent === undefined || this.agents.has(msg.forAgent)) return [msg]; - if ((msg.metadata as { kind?: unknown } | undefined)?.kind !== 'discord-send-failed') return [msg]; - const forPrimary = this.queuedFailedReplyForPrimary(msg.forAgent, msg); + if (!passesToPrimary(msg.metadata)) return [msg]; + const forPrimary = this.queuedNoticeForPrimary(msg.forAgent, msg); if (!forPrimary) return []; passed++; return [forPrimary]; }); if (passed > 0) { - console.error(`[deferred-flush] ${where}: ${passed} failed repl${passed === 1 ? 'y' : 'ies'} of an agent that has ended passed to the primary`); + console.error(`[deferred-flush] ${where}: ${passed} notice${passed === 1 ? '' : 's'} of an agent that has ended passed to the primary`); } const bypassFor = (msg: DeferredWrite): string | undefined => opts?.bypassDeferral ? this.deliveryAgent(msg.forAgent)?.name : undefined; @@ -21625,9 +21900,10 @@ export class AgentFramework { * null when the agent is no longer registered (a fork or a run already * released): there is no window to write, so the caller writes nothing, * traces nothing and requests no wake. The drop is logged here, once. A - * failed reply is passed on by name instead (passFailedReplyToPrimary). The - * notice of a channel an agent's own send opened follows the same rule in - * recordChannelAutoOpenNotice, which passes it on by name too. + * failed reply is passed on by name instead (passNoticeToPrimary). The + * notices of a channel an agent's own send opened, or a reply couldn't + * open, follow the same rule in their builders (channelOpenedNotices, + * replyOpenFailedNotices), which pass them on by name too. */ private noticeTargetFor(agentName: string, notice: string): { forAgent: string } | null { if (this.agents.has(agentName)) return { forAgent: agentName }; @@ -21641,30 +21917,31 @@ export class AgentFramework { * (releaseWindowNowOrAfterResume), since a write could land mid-surgery. * * The agent never reads its window again, so every failed-reply notice it - * never read is passed to the primary, by name, once: each one stored in - * its window above its consumed watermark (flushed there at its turn's end, - * or stored while it was idle) and not heard at a tool boundary, and each - * one still queued for it. Writes still - * queued would wait for a boundary it never reaches, and a flush that - * finds it gone would drop all but its failed replies, so they land in its - * own window now, which Chronicle keeps for investigation. One whose - * earlier write may have landed (`landing` unknown) is looked up there by - * its id first: found, it's neither stored again nor passed to the - * primary; where the store can't be searched, it's stored anyway, with a - * line saying it may now be stored twice, or, where storing it fails too, - * that it may be stored once, twice or not at all. Its stream is over or - * cancelled, or, when stop() ends it, may still run past its cancel, and - * then drops each event that comes after the agent has left (driveStream); - * a turn of its still assembling or compiling, as when stop() ends it, - * drops before it sends anything. One path still stores after these - * writes: a `complete` that arrived while the agent still waited for tools - * awaits the spill of its last tool results, and stores them once that - * resolves, even if a watchdog has ended the run meanwhile. A tool boundary - * stores nothing for a round whose turn was stopped while its results were - * prepared. + * never read, and every channel notice it could act on and never read, is + * passed to the primary, by name, once (passesToPrimary): each one stored + * in its window above its consumed watermark (flushed there at its turn's + * end, or stored while it was idle) and not heard at a tool boundary, as + * unreadNoticesForPrimary counts them, and each one still queued for it. + * Writes still queued would wait for a boundary it never reaches, and a + * flush that finds it gone would drop all but the notices it passes on, so + * they land in its own window now, which Chronicle keeps for + * investigation. One whose earlier write may have landed (`landing` + * unknown) is looked up there by its id first: found, it's neither stored + * again nor passed to the primary; where the store can't be searched, it's + * stored anyway, with a line saying it may now be stored twice, or, where + * storing it fails too, that it may be stored once, twice or not at all. + * Its stream is over or cancelled, or, when stop() ends it, may still run + * past its cancel, and then drops each event that comes after the agent + * has left (driveStream); a turn of its still assembling or compiling, as + * when stop() ends it, drops before it sends anything. One path still + * stores after these writes: a `complete` that arrived while the agent + * still waited for tools awaits the spill of its last tool results, and + * stores them once that resolves, even if a watchdog has ended the run + * meanwhile. A tool boundary stores nothing for a round whose turn was + * stopped while its results were prepared. */ private releaseWindow(agent: Agent): void { - for (const metadata of this.unreadFailedReplies(agent)) this.passFailedReplyToPrimary(agent.name, metadata); + for (const metadata of this.unreadNoticesForPrimary(agent)) this.passNoticeToPrimary(agent.name, metadata); const queued = this.drainDeferredFor(agent.name); if (queued.length === 0) return; const cm = agent.getContextManager(); @@ -21674,11 +21951,12 @@ export class AgentFramework { // looked up by its id, as settleUnknownLandings looks one up // elsewhere, but in the released agent's own store. Found, it landed: // it's neither stored again nor passed on below, since the loop above - // has passed its stored copy if it's a failed reply the agent hadn't - // read, by its window as that loop reads it. When the store can't be - // searched, it's stored anyway where the store takes it, since no - // later flush will try again, and a copy its deferredWriteId tells - // apart costs less than a write lost without a trace. + // has passed its stored copy if it's a notice the primary takes in its + // place that the agent hadn't read, by its window as that loop reads + // it. When the store can't be searched, it's stored anyway where the + // store takes it, since no later flush will try again, and a copy its + // deferredWriteId tells apart costs less than a write lost without a + // trace. const found = msg.landing === 'unknown' ? this.landedCopiesById(agent, msg.id) : []; if (found && found.length > 0) { this.logDeferredWriteError(msg, `release of ${agent.name}`, undefined, @@ -21704,11 +21982,12 @@ export class AgentFramework { err, ); } - // A failed reply, now in a window no one reads: the entry itself goes - // on to the primary as its by-name notice, under its own id and seq, so - // a crash before the ack below replays it once, not twice. - if ((msg.metadata as { kind?: unknown } | undefined)?.kind !== 'discord-send-failed') continue; - const passed = this.queuedFailedReplyForPrimary(agent.name, msg); + // A failed reply, or a channel notice its reader could act on, now in a + // window no one reads: the entry itself goes on to the primary as its + // by-name notice, under its own id and seq, so a crash before the ack + // below replays it once, not twice. + if (!passesToPrimary(msg.metadata)) continue; + const passed = this.queuedNoticeForPrimary(agent.name, msg); if (passed) forPrimary.push(passed); } if (forPrimary.length > 0) this.readdDrained(forPrimary, `release of ${agent.name}`); @@ -21716,32 +21995,41 @@ export class AgentFramework { } /** - * The failed-reply notices stored in `agent`'s window that it never read: - * those above the lower of its consumed watermark and the point where the - * watermark's branch and the current one part (branchesPartAt), or every - * one, when it never compiled. The watermark moves at a compile, and at a - * tool boundary whose - * stream declares it carries the messages injected there - * (markContextConsumed), so a notice the live stream heard is under it, and - * one a boundary stored without sending (a turn a tool ended, a budget - * restart, a stream that may drop injections) is above it. + * The notices stored in `agent`'s window that it never read and the primary + * takes in its place (passesToPrimary): its failed replies, and the channel + * notices it was told it could act on. Those are the ones above the lower + * of its consumed watermark and the point where the watermark's branch and + * the current one part (branchesPartAt), or every one, when it has no + * watermark. The watermark is set when the agent is created, and moves at + * a compile, and at a tool boundary whose stream declares it carries the + * messages injected there (markContextConsumed). So a notice such a stream + * carried is under it. + * One a boundary stored without sending (a turn a tool ended, a budget + * restart) is above it, and so is one injected into a stream that doesn't + * declare it carries injections, which may have dropped it: counted as + * unread, it goes to the primary even if the agent heard it, rather than to + * no one where the stream dropped it. */ - private unreadFailedReplies(agent: Agent): MessageMetadata[] { + private unreadNoticesForPrimary(agent: Agent): MessageMetadata[] { try { const cm = agent.getContextManager(); const watermark = agent.getConsumedWatermark(); const floor = watermark ? Math.min(watermark.sequence, this.branchesPartAt(watermark.branch, cm.currentBranch().name)) : Number.NEGATIVE_INFINITY; - return cm.getAllMessages() - .filter((m) => { - const metadata = m.metadata as { kind?: unknown } | undefined; - return m.sequence > floor && metadata?.kind === 'discord-send-failed'; - }) + // A body the store took as more than one message (a sharded body) + // carries the same metadata on each of its shards, and is one notice, + // so a shard that continues the body before it isn't taken again. Which + // messages make up a body is context-manager's rule (continuesBody, + // pairwise in store order): a group id names the text, so the same text + // stored twice is two bodies under one id. + const messages = cm.getAllMessages(); + return messages + .filter((m, i) => m.sequence > floor && passesToPrimary(m.metadata) && !continuesBody(messages[i - 1], m)) .map((m) => m.metadata as MessageMetadata); } catch (err) { // A release must complete whatever its window's read meets. - console.error(`[route-failure] ${agent.name}: could not read its window for unread failed replies:`, err); + console.error(`[route-failure] ${agent.name}: could not read its window for unread notices the primary would take:`, err); return []; } } @@ -21786,7 +22074,8 @@ export class AgentFramework { * deregistered, so its stream can't dispatch anything more and resume * doesn't weigh its context, and a write addressed to it meanwhile is * treated as for any agent no longer registered (dropped, or passed to the - * primary if it's a failed reply). + * primary if it's a failed reply, or a channel notice its reader could act + * on). */ private releaseWindowNowOrAfterResume(agent: Agent): void { if (this.quiesced) this.releasesAfterResume.set(agent.name, agent); @@ -21807,46 +22096,41 @@ export class AgentFramework { } /** - * A fork or an ephemeral run ended before reading that its reply's delivery - * to a channel failed: it was released with the notice stored in its window - * unread, or its delivery failed after release. (A notice still queued for - * it, at release or restored at boot, goes on as the queue entry itself: - * queuedFailedReplyForPrimary.) Tell the primary instead, naming the - * speaker, so an agent who can still act learns of it and can tell the - * human. It gives the reply's length, not its words, which stay in the - * speaker's window. Its wording follows the speaker's own notice - * (onRouteFailure): a delivery that wasn't confirmed may have been posted, - * and neither text says no one saw the reply, since only this channel's - * routing failed. Only for a delivery to a channel; a reply that had no - * channel to go to (no locus) had no human waiting for it. Never throws: - * it runs inside a release, which must complete whatever this write meets. + * A fork or an ephemeral run ended before reading a notice the primary + * takes in its place (passesToPrimary): it was released with the notice + * stored in its window unread, or, for a failed reply, its delivery failed + * after release. (A notice still queued for it, at release or restored at + * boot, goes on as the queue entry itself: queuedNoticeForPrimary.) Tell + * the primary instead, naming the speaker, so an agent who can still act + * learns of it (noticeForPrimary). Never throws: it runs inside a release, + * which must complete whatever this write meets. */ - private passFailedReplyToPrimary(speaker: string, metadata: MessageMetadata | undefined): void { - const notice = this.failedReplyNoticeForPrimary(speaker, metadata); + private passNoticeToPrimary(speaker: string, metadata: MessageMetadata | undefined): void { + const notice = this.noticeForPrimary(speaker, metadata); if (!notice) return; try { this.addMessage('user', notice.content, notice.metadata); } catch (err) { - console.error(`[route-failure] failed to tell the primary that ${speaker}'s reply to ${String(notice.metadata.channelId)} failed:`, err); + console.error(`[route-failure] failed to tell the primary ${notice.about}:`, err); } } /** - * A queued failed-reply entry for `speaker`, who is gone, as the primary's - * by-name notice (passFailedReplyToPrimary's words): the same entry, under - * its own id and seq, now addressed to the primary. Written through the - * drain, it lands stamped with that id, or waits in the queue under it, so - * boot's exactly-once holds for it: a crash before the original's ack - * replays one entry, never the original and a separate notice. An entry - * marked as one whose earlier write may have landed (`landing`) keeps the - * mark, so before it's written it's looked up by its id in the primary's - * window, where it would land now (writeDrainedInOrder). A release has - * looked in the speaker's own window first (releaseWindow). null when no - * one is told (logged): a reply that had no channel to reach, or no - * primary. + * A queued entry for `speaker`, who is gone, that the primary takes in its + * place, as the primary's by-name notice (noticeForPrimary): the same + * entry, under its own id and seq, now addressed to the primary. Written + * through the drain, it lands stamped with that id, or waits in the queue + * under it, so boot's exactly-once holds for it: a crash before the + * original's ack replays one entry, never the original and a separate + * notice. An entry marked as one whose earlier write may have landed + * (`landing`) keeps the mark, so before it's written it's looked up by its + * id in the primary's window, where it would land now + * (writeDrainedInOrder). A release has looked in the speaker's own window + * first (releaseWindow). null when no one is told (logged where it's + * decided). */ - private queuedFailedReplyForPrimary(speaker: string, msg: DeferredWrite): DeferredWrite | null { - const notice = this.failedReplyNoticeForPrimary(speaker, msg.metadata); + private queuedNoticeForPrimary(speaker: string, msg: DeferredWrite): DeferredWrite | null { + const notice = this.noticeForPrimary(speaker, msg.metadata); if (!notice) return null; return { id: msg.id, seq: msg.seq, participant: 'user', content: notice.content, metadata: notice.metadata, heldAt: msg.heldAt, @@ -21854,8 +22138,83 @@ export class AgentFramework { }; } + /** + * The primary's by-name notice in place of one `speaker` never read, and + * what it's about, as a log line says it: a failed reply's + * (failedReplyNoticeForPrimary) or a channel notice's + * (channelNoticeForPrimary). null when no one is told, which is logged + * where it's decided. + */ + private noticeForPrimary( + speaker: string, + metadata: MessageMetadata | undefined, + ): { content: ContentBlock[]; metadata: MessageMetadata; about: string } | null { + const kind = (metadata as { kind?: unknown } | undefined)?.kind; + if (kind === 'discord-send-failed') { + const notice = this.failedReplyNoticeForPrimary(speaker, metadata); + return notice && { ...notice, about: `that ${speaker}'s reply to ${String(notice.metadata.channelId)} failed` }; + } + const passOn = channelNoticePassOn(metadata); + if (!passOn) return null; + const notice = this.channelNoticeForPrimary(speaker, passOn); + return notice && { ...notice, about: `of ${speaker}'s channel notice` }; + } + + /** + * The primary's by-name notice in place of a channel notice `speaker` was + * told it could act on and never read: its builder, called again now that + * `speaker` has ended, gives the words it gives an agent that has ended, + * judged for the primary, or nothing, logged, with no primary to tell. + * They speak of the open or the failure as it was, as the notice did, a + * failed one's reason for its selection included, which an agent's + * channel_open since replaces in the record and not in the words, and + * only of what still holds by each channel's durable decision on its own + * server: the registry's (getDesiredState), or before there is one, as at + * a serving boot's flush, the lifecycle log's (desiredStateInLog). An + * opened channel is kept while it's decided open or tuned out, whatever its + * transport is doing at that moment: the decision is what channel_close + * changes and what the host carries out once the transport is back, and at + * a serving boot no transport is up yet. A failed one is kept only while + * it's still decided open and its transport isn't open, since "Not open" + * would misstate an open one. One tuned out since is left out too: the + * tune-out is a newer decision, by an agent that knows the channel, and + * the notice's reason and its channel_open advice, which a tuned-out + * channel refuses, would misstate it. With nothing left, no one is told, + * logged here. + */ + private channelNoticeForPrimary( + speaker: string, + passOn: ChannelNoticePassOn, + ): { content: ContentBlock[]; metadata: MessageMetadata } | null { + const registry = this.channelRegistry; + const { serverId } = passOn; + const desiredState = (channelId: string) => + (registry ? registry.getDesiredState(serverId, channelId) : desiredStateInLog(this.store, serverId, channelId)); + let notice: ChannelNotice | undefined; + if (passOn.notice === 'channel-opened') { + // A tune-out leaves a channel open, its traffic held. + const channels = passOn.channels.filter((c) => { + const desired = desiredState(c.channelId); + return desired === 'open' || desired === 'tuned-out'; + }); + if (channels.length === 0) { + console.error(`[channel] ${speaker} has ended, and no channel its notice said opened is decided open or tuned out now (${passOn.channels.map(shownChannel).join(', ')}): no one is told`); + return null; + } + [notice] = this.channelOpenedNotices(speaker, serverId, channels, passOn.cause); + } else { + const { channelId } = passOn.channel; + if (desiredState(channelId) !== 'open' || registry?.isChannelOpen(channelId, serverId)) { + console.error(`[channel] ${speaker} has ended, and ${shownChannel(passOn.channel)}, which its notice said was not open and selected for opening, is open now or no longer decided open: no one is told`); + return null; + } + [notice] = this.replyOpenFailedNotices(speaker, serverId, passOn.channel, passOn.selectedForOpening); + } + return notice ? { content: [{ type: 'text', text: notice.text }], metadata: notice.metadata } : null; + } + /** The primary's by-name notice of `speaker`'s failed reply, or null when no - * one is told, which is logged here (see passFailedReplyToPrimary). */ + * one is told, which is logged here (see passNoticeToPrimary). */ private failedReplyNoticeForPrimary( speaker: string, metadata: MessageMetadata | undefined, @@ -21933,8 +22292,8 @@ export class AgentFramework { * slot (sharedSlotAgents). A write whose target is gone isn't among them: * no module reads that agent's store, and the write waits for its window's * release (drainDeferredFor) or, at its flush, is dropped (addMessage), a - * failed reply passing to the primary as its notice - * (queuedFailedReplyForPrimary). + * failed reply, or a channel notice its reader could act on, passing to the + * primary as its notice (queuedNoticeForPrimary). */ private heldWritesFor(reader: Agent): DeferredWrite[] { const heldFor = (m: DeferredWrite): boolean => { @@ -22958,7 +23317,7 @@ export class AgentFramework { }; if (!this.agents.has(conversationId)) { // Released before its delivery failed: no window left to read it. - this.passFailedReplyToPrimary(conversationId, metadata); + this.passNoticeToPrimary(conversationId, metadata); return; } // No-locus failures (headless / WebUI turns with no home or @@ -22990,10 +23349,11 @@ export class AgentFramework { }, // A channel opened without the agent asking (policy admission or an // open forced by delivery) must be announced in the window, with the - // opt-out named. Routed here so registry-driven opens and framework- - // driven opens produce the same durable notice. - onChannelAutoOpened: ({ conversationId, source, channels }) => { - this.recordChannelAutoOpenNotice(conversationId, channels, source); + // opt-out named to a reader that can use it. Routed here so + // registry-driven opens and framework-driven opens produce the same + // durable notice. + onChannelAutoOpened: ({ conversationId, serverId, source, channels }) => { + this.recordChannelAutoOpenNotice(conversationId, serverId, channels, source); }, }, ); @@ -26244,6 +26604,7 @@ export class AgentFramework { ); this.recordChannelAutoOpenNotice( agentName, + serverId, [{ channelId: channelId ?? target, label }], 'opened-by-reply', ); @@ -26261,7 +26622,7 @@ export class AgentFramework { `[channel] ${agentName}: ${toolName} delivered to ${target} but the open-on-send failed` + (tell ? '; the agent is told' : ''), ); - if (tell) this.recordReplyOpenFailedNotice(agentName, { channelId: channelId ?? target, label }, selectedForOpening); + if (tell) this.recordReplyOpenFailedNotice(agentName, serverId, { channelId: channelId ?? target, label }, selectedForOpening); } }) .catch((err) => console.error('[channel] open-on-send error:', err)); diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index c37144d30..f11468ac3 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -238,6 +238,43 @@ interface ChannelLifecycleEvent { acknowledgment?: string; } +/** A lifecycle-log item as a desired-state decision, or null when it isn't + * one this build reads: the one rule the registry's replay + * (initializeLifecycleStore) and desiredStateInLog read the log by. */ +function desiredStateRecord(item: unknown): { + serverId: string; channelId: string; state: DesiredChannelState; source: string; tuneOut?: TuneOutParams; +} | null { + const event = item as Partial | null | undefined; + if (event?.kind !== 'desired-state' || typeof event.serverId !== 'string' || typeof event.channelId !== 'string') return null; + if (!(event.desired === 'open' || event.desired === 'closed' || (event.desired === 'tuned-out' && event.tuneOut))) return null; + return { + serverId: event.serverId, + channelId: event.channelId, + state: event.desired, + source: typeof event.source === 'string' ? event.source : 'unknown', + ...(event.desired === 'tuned-out' ? { tuneOut: event.tuneOut } : {}), + }; +} + +/** + * The desired state `store`'s lifecycle log records for `channelId` on + * `serverId`, for a reader with no registry yet, as a serving boot's + * deferred-write flush is, before initializeMcpl: the last decision wins, as + * the registry's replay reads it (getDesiredState answers once there is a + * registry). undefined when no decision names it. + */ +export function desiredStateInLog(store: JsStore, serverId: string, channelId: string): DesiredChannelState | undefined { + ensureStateRegistered(store, { id: CHANNEL_LIFECYCLE_LOG_ID, strategy: 'append_log' }); + const raw = store.getStateJson(CHANNEL_LIFECYCLE_LOG_ID); + if (!Array.isArray(raw)) return undefined; + let state: DesiredChannelState | undefined; + for (const item of raw) { + const decision = desiredStateRecord(item); + if (decision?.serverId === serverId && decision.channelId === channelId) state = decision.state; + } + return state; +} + /** One append-log record in `CHANNEL_LABEL_HISTORY_LOG_ID`: "at `ts`, this * channelId's label was observed to be `label`". See `labelHistory`. */ interface ChannelLabelSightingEvent { @@ -2056,8 +2093,15 @@ export class ChannelRegistry { return typeof label === 'string' && label.length > 0 ? label : undefined; } - isChannelOpen(channelId: string): boolean { - return this.findChannelEntry(channelId)?.open === true; + /** Whether `channelId`'s transport is open: on `serverId` where the caller + * knows the channel's server, and otherwise on the first server found to + * register that id (findChannelEntry). The map's key joins the two ids + * with a colon, so another server's entry can sit under a key: it has to + * be this server's (as in streamingServerAt). */ + isChannelOpen(channelId: string, serverId?: string): boolean { + if (serverId === undefined) return this.findChannelEntry(channelId)?.open === true; + const entry = this.channels.get(`${serverId}:${channelId}`); + return entry?.serverId === serverId && entry.open === true; } getDesiredState(serverId: string, channelId: string): DesiredChannelState | undefined { @@ -2292,20 +2336,11 @@ export class ChannelRegistry { if (!item || typeof item !== 'object') continue; const event = item as Partial; if (typeof event.serverId !== 'string') continue; - if ( - event.kind === 'desired-state' && - typeof event.channelId === 'string' && - (event.desired === 'open' || event.desired === 'closed' || - (event.desired === 'tuned-out' && event.tuneOut)) - ) { + const decision = desiredStateRecord(item); + if (decision) { this.desiredStates.set( - this.lifecycleKey(event.serverId, event.channelId), - { - state: event.desired, - source: typeof event.source === "string" ? event.source : "unknown", - tuneOut: event.desired === 'tuned-out' ? event.tuneOut : undefined, - wakeCount: 0, - }, + this.lifecycleKey(decision.serverId, decision.channelId), + { state: decision.state, source: decision.source, tuneOut: decision.tuneOut, wakeCount: 0 }, ); } else if ( event.kind === 'tune-out-wake' && @@ -3529,7 +3564,7 @@ export class ChannelRegistry { // The caller reports both: the channel isn't open, but it stays // selected for opening, so reconciliation may open it later. throw new ChannelLifecycleError( - afterClause(lifecycleClause(err), 'the channel stays selected for opening, as this host couldn\'t record ' + + afterClause(lifecycleClause(err), 'the channel stays selected for opening, because this host couldn\'t record ' + `undoing that: ${requestErrorText(putBackErr)}`), 'put-back-unrecorded', undefined, @@ -3576,8 +3611,8 @@ export class ChannelRegistry { } if (selectedAgain) { throw new ChannelLifecycleError( - afterClause(lifecycleClause(err), 'the channel is selected for opening again, as ' + - (selectedAgain === 'server-bootstrap' ? 'its server announces it open' : 'the subscription policy admits it')), + afterClause(lifecycleClause(err), 'the channel is selected for opening again, because ' + + (selectedAgain === 'server-bootstrap' ? 'its server lists it as open by default' : 'the subscription policy admits it')), 'selected-again', undefined, { by: selectedAgain, cause: err }, @@ -4657,7 +4692,8 @@ export class ChannelRegistry { * after an open for a declaration leave the channel open, and the agent * gets the failure for the words. An open by delivery (`openSource` * 'opened-by-delivery', the default) that turned a closed channel open - * also gives the notice that it's open, which says how to close it. + * also gives the notice that it's open, which says how to close it to a + * reader that can. * One that only reconfirmed the transport of a channel already decided * open or tuned out changes no decision, and gives none. When the * registration is gone, the failure says the channel was removed. diff --git a/test/channel-lifecycle-order.test.ts b/test/channel-lifecycle-order.test.ts index 9ee395322..7281bbda7 100644 --- a/test/channel-lifecycle-order.test.ts +++ b/test/channel-lifecycle-order.test.ts @@ -769,9 +769,9 @@ test('failed policy opening records intent once, with truthful resident text thr }); try { const recordNotice = (framework as unknown as { - recordChannelAutoOpenNotice(agent: undefined, channels: Array<{ channelId: string; label?: string }>, cause: 'subscription-policy' | 'opened-by-delivery'): void; + recordChannelAutoOpenNotice(agent: undefined, serverId: string, 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 onNotice: NoticeHandler = (info) => recordNotice(undefined, info.serverId, info.channels, info.source); const f = fixture(onNotice, framework.getStore()); f.registry.setSubscriptionPolicy('test', 'auto'); f.held.add('x'); @@ -1556,13 +1556,14 @@ test('the notices say what happened, name the opt-out, and state no rule about l const recordNotice = (framework as unknown as { recordChannelAutoOpenNotice( agent: undefined, + serverId: string, 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'); + recordNotice(undefined, 'srv', [{ channelId: 'discord:g1:c1', label: '#general' }], 'opened-by-reply'); + recordNotice(undefined, 'srv', [{ channelId: 'c2' }], 'opened-by-delivery'); + recordNotice(undefined, 'srv', [{ 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]')) @@ -1927,14 +1928,14 @@ test("when an open by speech or a reply fails while the server announces the cha const { outcome } = await speechWhoseOpenFails(f, reannounce); assert.equal(outcome.status, 'failed'); assert.equal(outcome.reason, - "opening the channel for these words didn't complete: fixture lifecycle failure; the channel is selected for opening again, as its server announces it open"); + "opening the channel for these words didn't complete: fixture lifecycle failure; the channel is selected for opening again, because its server lists it as open by default"); // A server's own words keep their stop, so what follows is a sentence. const g = fixture(); await g.registry.handleChanged('test', { added: [descriptor('x', false)] }); const answered = await speechWhoseOpenFails(g, () => [g.registry.handleChanged('test', { added: [descriptor('x', true)] })], new McplRequestError('Opening x is not allowed here.', 'error-response')); assert.equal(answered.outcome.reason, "opening the channel for these words didn't complete: Opening x is not allowed here. " + - 'The channel is selected for opening again, as its server announces it open'); + 'The channel is selected for opening again, because its server lists it as open by default'); } else { const { answer } = await replyWhoseOpenFails(f, reannounce); assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true, selectedForOpening: 'server-bootstrap' }); @@ -1955,7 +1956,7 @@ test("when a reply's open fails after the subscription policy came to admit the const { answer } = await replyWhoseOpenFails(f, () => { f.registry.setSubscriptionPolicy('test', 'auto'); return []; }); assert.deepEqual(answer, { status: 'open-failed', channelId: 'x', label: 'x', opening: true, selectedForOpening: 'subscription-policy' }); assert.deepEqual(f.traces.filter((t) => t.type === 'mcpl:channel-open-failed').map((t) => t.error), - ['fixture lifecycle failure; the channel is selected for opening again, as the subscription policy admits it']); + ['fixture lifecycle failure; the channel is selected for opening again, because the subscription policy admits it']); await tick(); await tick(); assert.deepEqual(decision(f.registry), { state: 'open', source: 'subscription-policy' }); assert.equal(f.isOpen(), true); @@ -2230,7 +2231,7 @@ test("when putting back the decision a failed open replaced can't be recorded, s const closes = f.calls.filter((c) => c.kind === 'close').length; const { outcome } = await speechWhoseOpenFails(f); assert.equal(outcome.status, 'failed'); - assert.match(outcome.reason ?? '', /^opening the channel for these words didn't complete: fixture lifecycle failure; the channel stays selected for opening, as this host couldn't record undoing that: ENOSPC/); + assert.match(outcome.reason ?? '', /^opening the channel for these words didn't complete: fixture lifecycle failure; the channel stays selected for opening, because this host couldn't record undoing that: ENOSPC/); assert.deepEqual(decision(f.registry), { state: 'open', source: 'opened-by-delivery' }); await tick(); await tick(); assert.equal(f.calls.filter((c) => c.kind === 'close').length, closes, 'no close is sent: the decision stays open'); @@ -2263,12 +2264,15 @@ test("channel_open and channel_close on a channel already so say what failed whe async function replyOpenFailedNotices(opts: { opening: boolean; directs: boolean; agentName?: string; live?: string[]; homes?: Array<[string, string]>; denied?: string[]; selectedForOpening?: 'unrecorded' | 'server-bootstrap' | 'subscription-policy'; primary?: string; dropsDuring?: true; + opened?: true; }) { const agentName = opts.agentName ?? 'agent'; const told: Array<{ text: string; metadata: unknown; forAgent?: string }> = []; const logged: string[] = []; + // The server's id isn't its tool prefix, so the notice's inputs can't take + // one for the other. const server = { - id: 'srv', + id: 'chat-server', isConnected: true, async sendToolsList() { return { tools: [{ name: 'send_message', inputSchema: { type: 'object' } }] }; }, async sendToolsCall() { return { content: [{ type: 'text', text: 'sent' }] }; }, @@ -2276,8 +2280,8 @@ async function replyOpenFailedNotices(opts: { const framework = Object.create(AgentFramework.prototype) as any; framework.mcplTools = []; framework.mcplToolFeatureSets = new Map(); - framework.mcplServerConfigs = new Map([['srv', { id: 'srv', toolPrefix: 'srv' }]]); - framework.mcplServerRegistry = { getAllServers: () => [server], getServer: (id: string) => (id === 'srv' ? server : null) }; + framework.mcplServerConfigs = new Map([['chat-server', { id: 'chat-server', toolPrefix: 'srv' }]]); + framework.mcplServerRegistry = { getAllServers: () => [server], getServer: (id: string) => (id === 'chat-server' ? server : null) }; framework.checkpointManager = null; framework.turnRouting = new Map(); framework.agents = new Map((opts.live ?? [agentName]).map((name) => [name, { canUseTool: (tool: string) => !(opts.denied ?? []).includes(tool) }])); @@ -2295,11 +2299,13 @@ async function replyOpenFailedNotices(opts: { framework.channelRegistry = { openIfClosedForSend: async () => { opened = true; + if (opts.opened) return { status: 'opened', channelId: 'x', label: '#x' }; return { status: 'open-failed', channelId: 'x', label: '#x', opening: opts.opening, ...(opts.selectedForOpening ? { selectedForOpening: opts.selectedForOpening } : {}) }; }, directsLifecycle: () => opts.directs && !(opts.dropsDuring && opened), }; + const handled = (l: string) => l.includes('open-on-send failed') || l.includes('opened (send implies engagement)'); const error = console.error; console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; try { @@ -2308,15 +2314,15 @@ async function replyOpenFailedNotices(opts: { framework.pushEvent = (event: { type: string; result: { success?: boolean; error?: string } }) => { if (event.type === 'tool-result') resolve(event.result); }; - framework.dispatchMcplToolCall(agentName, { id: 'c1', name: 'srv--send_message', input: { channelId: 'x' } }, 'srv', 'srv'); + framework.dispatchMcplToolCall(agentName, { id: 'c1', name: 'srv--send_message', input: { channelId: 'x' } }, 'chat-server', 'srv'); }); assert.equal(sent.success, true, `the send went out: ${sent.error ?? ''}`); // The open's answer is handled once its console line is written. - for (let i = 0; i < 50 && !logged.some((l) => l.includes('open-on-send failed')); i++) await tick(); + for (let i = 0; i < 50 && !logged.some(handled); i++) await tick(); } finally { console.error = error; } - assert.ok(logged.some((l) => l.includes('open-on-send failed')), 'the failed open was handled'); + assert.ok(logged.some(handled), 'the open was handled'); return { told, logged }; } @@ -2334,19 +2340,19 @@ test("a reply whose open would have opened the channel, and failed, tells the ag '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. channel_open tries again.', ], "a transport that drops while the open is out clears the grant before the failure arrives: whether the host directs the lifecycle is read before the open"); assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true, selectedForOpening: 'unrecorded' })).told.map((t) => t.text), [ - "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again, and channel_close withdraws that. channel_open tries again.", + "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It stays selected for opening, because the host couldn't record undoing that: a later delivery or reconnect tries again, and channel_close withdraws that. channel_open tries again.", ], "where the put-back couldn't be recorded, as speech's failure says, in channel_open's own failure's words"); assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true, selectedForOpening: 'server-bootstrap' })).told.map((t) => t.text), [ - "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It's selected for opening again, as its server announces it open: the host tries again, and channel_close withdraws that.", + "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It's selected for opening again, because its server lists it as open by default: the host tries again, and channel_close withdraws that.", ], "where the server's announcement lifted the default put back, the host tries again itself, so channel_open isn't named"); assert.deepEqual((await replyOpenFailedNotices({ opening: true, directs: true, selectedForOpening: 'subscription-policy' })).told.map((t) => t.text), [ - "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It's selected for opening again, as the subscription policy admits it: the host tries again, and channel_close withdraws that.", + "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It's selected for opening again, because the subscription policy admits it: the host tries again, and channel_close withdraws that.", ], 'where the subscription policy lifted it'); }); -test("the reply notice names channel_open and channel_close only to an agent that can call them; where the replying agent has ended, no one is told unless the channel is selected for opening, and then the primary is, by name", async () => { +test("the reply notice names channel_open and channel_close only where the agent it's addressed to can call them; where the replying agent has ended, no one is told unless the channel is selected for opening, and then the primary is, by name", async () => { const facts = '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.'; - const stays = " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again"; + const stays = " It stays selected for opening, because the host couldn't record undoing that: a later delivery or reconnect tries again"; const fork = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'x']] }); assert.deepEqual(fork.told.map((t) => t.text), [facts], "a conversation fork's channel_open is refused outright"); @@ -2356,6 +2362,13 @@ test("the reply notice names channel_open and channel_close only to an agent tha const forkHome = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'x']], selectedForOpening: 'unrecorded' }); assert.deepEqual(forkHome.told.map((t) => t.text), [facts + stays + ', and channel_close withdraws that.'], 'a fork may close its home channel, though not open it'); + assert.equal((forkHome.told[0]!.metadata as { passOn?: { serverId?: unknown } }).passOn?.serverId, 'chat-server', + "its inputs name the send's server, not its tool prefix"); + // A send that opens its channel tells the agent, and its inputs name the + // send's server too. + const sendOpened = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'x']], opened: true }); + assert.match(sendOpened.told[0]?.text ?? '', /^\[channels\] Now open: #x \(x\): your own message into the closed channel opened it\./); + assert.equal((sendOpened.told[0]!.metadata as { passOn?: { serverId?: unknown } }).passOn?.serverId, 'chat-server'); const forkElsewhere = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'y']], selectedForOpening: 'unrecorded' }); assert.deepEqual(forkElsewhere.told.map((t) => t.text), [facts + stays + '.'], "a fork's channel_close is refused for a channel that isn't its home"); @@ -2366,7 +2379,7 @@ test("the reply notice names channel_open and channel_close only to an agent tha assert.deepEqual(noClose.told.map((t) => t.text), [facts + stays + '. channel_open tries again.'], 'an agent whose allowedTools leave channel_close out'); - const again = " It's selected for opening again, as its server announces it open: the host tries again"; + const again = " It's selected for opening again, because its server lists it as open by default: the host tries again"; const forkAgain = await replyOpenFailedNotices({ opening: true, directs: true, homes: [['agent', 'x']], selectedForOpening: 'server-bootstrap' }); assert.deepEqual(forkAgain.told.map((t) => t.text), [facts + again + ', and channel_close withdraws that.'], 'a fork told the server lifted its home channel may close it'); @@ -2375,7 +2388,7 @@ test("the reply notice names channel_open and channel_close only to an agent tha assert.deepEqual(ended.told, [], 'the channel stays closed, so nothing follows from the failure'); assert.ok(ended.logged.some((l) => l.includes("agent has ended; not telling it that #x (x) couldn't be opened"))); - const endedFacts = "[channels] Not open: #x (x). agent's reply there went out, but opening the channel failed; agent has ended."; + const endedFacts = "[channels] Not open: #x (x). A reply from agent went out there, but opening the channel failed; agent has ended."; const endedAgain = await replyOpenFailedNotices({ opening: true, directs: true, live: ['primary'], primary: 'primary', selectedForOpening: 'server-bootstrap' }); assert.deepEqual(endedAgain.told, [{ text: endedFacts + again + ', and channel_close withdraws that.', @@ -2399,7 +2412,7 @@ test("the reply notice names channel_open and channel_close only to an agent tha test("a reply notice with no agent named, or an empty name, goes out unrouted, with its advice judged for, and its log naming, the agent addMessage delivers to", () => { const facts = '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.'; - const stays = " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again"; + const stays = " It stays selected for opening, because the host couldn't record undoing that: a later delivery or reconnect tries again"; const cases: Array<{ why: string; name?: string; agents: Array<[string, string[]]>; primary: string | null; selectedForOpening?: 'unrecorded' | 'server-bootstrap' | 'subscription-policy'; @@ -2414,7 +2427,7 @@ test("a reply notice with no agent named, or an empty name, goes out unrouted, w reader: 'primary', text: facts + stays + '. channel_open tries again.' }, { why: 'selected again by the server, a primary that can open and close', agents: [['primary', []]], primary: 'primary', selectedForOpening: 'server-bootstrap', reader: 'primary', - text: facts + " It's selected for opening again, as its server announces it open: the host tries again, and channel_close withdraws that." }, + text: facts + " It's selected for opening again, because its server lists it as open by default: the host tries again, and channel_close withdraws that." }, { why: 'no primary: the first registered agent', agents: [['first', ['channel_open']], ['second', []]], primary: null, reader: 'first', text: facts }, { why: 'a primary that is not the first registered agent', agents: [['first', ['channel_open']], ['second', []]], primary: 'second', @@ -2435,7 +2448,7 @@ test("a reply notice with no agent named, or an empty name, goes out unrouted, w const error = console.error; console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; try { - framework.recordReplyOpenFailedNotice(c.name, { channelId: 'x', label: '#x' }, c.selectedForOpening); + framework.recordReplyOpenFailedNotice(c.name, 'srv', { channelId: 'x', label: '#x' }, c.selectedForOpening); } finally { console.error = error; } @@ -2446,9 +2459,11 @@ test("a reply notice with no agent named, or an empty name, goes out unrouted, w test("where a reply's reader can't close a channel selected for opening, the primary is told by name too, if it can close it", () => { const facts = '[channels] Not open: #x (x). Your reply there went out, but opening the channel failed.'; - const stays = " It stays selected for opening, as the host couldn't record undoing that: a later delivery or reconnect tries again"; - const again = " It's selected for opening again, as its server announces it open: the host tries again"; - const byName = (who: string) => `[channels] Not open: #x (x). ${who}'s reply there went out, but opening the channel failed, and ${who} can't close it.`; + const stays = " It stays selected for opening, because the host couldn't record undoing that: a later delivery or reconnect tries again"; + const again = " It's selected for opening again, because its server lists it as open by default: the host tries again"; + const byName = (first: string, again: string) => `[channels] Not open: #x (x). A reply from ${first} went out there, but opening the channel failed, and ${again} can't close it.`; + const forkByName = byName('your fork in y, named fork,', 'that fork'); + const told = (primary: string) => ` ${primary} has been told, and can withdraw that.`; const cases: Array<{ why: string; name: string; agents: Array<[string, string[]]>; homes?: Array<[string, string]>; primary: string | null; selectedForOpening?: 'unrecorded' | 'server-bootstrap'; @@ -2457,24 +2472,34 @@ test("where a reply's reader can't close a channel selected for opening, the pri { why: 'a fork away from its home, selected again by its server', name: 'fork', agents: [['fork', []], ['primary', []]], homes: [['fork', 'y']], primary: 'primary', selectedForOpening: 'server-bootstrap', told: [ - { text: facts + again + '.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, - { text: byName('fork') + again + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: forkByName + again + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: facts + again + '.' + told('Your primary'), metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, ] }, { why: "an agent whose allowedTools leave channel_close out, the put-back unrecorded", name: 'helper', agents: [['helper', ['channel_close']], ['primary', []]], primary: 'primary', selectedForOpening: 'unrecorded', told: [ - { text: facts + stays + '. channel_open tries again.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'helper' }, - { text: byName('helper') + stays + ', and channel_close withdraws that. channel_open tries again.', metadata: { system: true, kind: 'channel-notice', speaker: 'helper' } }, + { text: byName('helper', 'helper') + stays + ', and channel_close withdraws that. channel_open tries again.', metadata: { system: true, kind: 'channel-notice', speaker: 'helper' } }, + { text: facts + stays + '.' + told('primary') + ' channel_open tries again.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'helper' }, ] }, { why: 'a reader who can close it, beside a primary that can too', name: 'helper', agents: [['helper', []], ['primary', []]], primary: 'primary', selectedForOpening: 'server-bootstrap', - told: [{ text: facts + again + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'helper' }] }, + // Its inputs, so a release that finds it unread tells the primary. + told: [{ text: facts + again + ', and channel_close withdraws that.', forAgent: 'helper', metadata: { system: true, kind: 'channel-notice', + passOn: { notice: 'reply-open-failed', serverId: 'srv', channel: { channelId: 'x', label: '#x' }, selectedForOpening: 'server-bootstrap' } } }] }, + { why: 'a reader who can close it, the put-back unrecorded', name: 'helper', agents: [['helper', []], ['primary', []]], + primary: 'primary', selectedForOpening: 'unrecorded', + told: [{ text: facts + stays + ', and channel_close withdraws that. channel_open tries again.', forAgent: 'helper', metadata: { system: true, kind: 'channel-notice', + passOn: { notice: 'reply-open-failed', serverId: 'srv', channel: { channelId: 'x', label: '#x' }, selectedForOpening: 'unrecorded' } } }] }, { why: "a primary that can close it but not open it, the put-back unrecorded", name: 'fork', agents: [['fork', []], ['primary', ['channel_open']]], homes: [['fork', 'y']], primary: 'primary', selectedForOpening: 'unrecorded', told: [ - { text: facts + stays + '.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, - { text: byName('fork') + stays + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: forkByName + stays + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: facts + stays + '.' + told('Your primary'), metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, ] }, + { why: 'a fork that has ended, its home still known while it is released, the channel selected again', name: 'fork', agents: [['primary', []]], + homes: [['fork', 'y']], primary: 'primary', selectedForOpening: 'server-bootstrap', + told: [{ text: '[channels] Not open: #x (x). A reply from your fork in y, named fork, went out there, but opening the channel failed; that fork has ended.' + + again + ', and channel_close withdraws that.', metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }] }, { why: 'a fork away from its home, the channel left closed', name: 'fork', agents: [['fork', []], ['primary', []]], homes: [['fork', 'y']], primary: 'primary', told: [{ text: facts, metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }] }, @@ -2494,6 +2519,8 @@ test("where a reply's reader can't close a channel selected for opening, the pri framework.primaryAgentName = c.primary; framework.usedEphemeralAgentNames = new Set(); framework.conversationAgentHomes = new Map(c.homes ?? []); + // Forks are made where conversations are routed, here from the primary. + framework.conversationRouter = c.homes ? { templateAgent: 'primary' } : undefined; framework.addMessage = (_participant: string, content: Array<{ text: string }>, metadata: unknown, routed?: { forAgent?: string }) => { told.push({ text: content[0]!.text, metadata, ...(routed ? { forAgent: routed.forAgent } : {}) }); return 'm'; @@ -2501,7 +2528,7 @@ test("where a reply's reader can't close a channel selected for opening, the pri const error = console.error; console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; try { - framework.recordReplyOpenFailedNotice(c.name, { channelId: 'x', label: '#x' }, c.selectedForOpening); + framework.recordReplyOpenFailedNotice(c.name, 'srv', { channelId: 'x', label: '#x' }, c.selectedForOpening); } finally { console.error = error; } @@ -2512,6 +2539,244 @@ test("where a reply's reader can't close a channel selected for opening, the pri } }); +// The opened notice follows the reply notice's rule: channel_close is named +// only where the agent it's addressed to can call it for every channel named +// (other residents of one slot, and forks spawned from the primary later, may +// read it too: #291, #235), and where an +// agent's send opened a channel it can't close, the primary is told by name +// too, if it can close it. The reader's notice carries its inputs only where +// it was told it can close the channels, so a release passes on only that. +test('the opened notice names channel_close only where the agent it\'s addressed to can call it, says what the channel delivers only to the agent that holds its traffic, and tells the primary by name where the agent whose send opened it can\'t close it', () => { + // What the channel delivers: said to the agent whose window holds its + // traffic. Where conversations are routed (router), that's a fork for its + // home alone; elsewhere, the agent addMessage delivers to. + const optOut = ' What an open channel delivers to you depends on its server.'; + const close = " If you don't want it, use channel_close; your choice outranks policy."; + const closeMany = " If you don't want one of them, use channel_close; your choice outranks policy."; + const own = '[channels] Now open: #x (x): your own message into the closed channel opened it.'; + const opened = (passOn: unknown) => ({ system: true, kind: 'channel-notice', passOn }); + const fromReply = { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: [{ channelId: 'x', label: '#x' }] }; + const cases: Array<{ + why: string; name?: string; cause?: 'opened-by-reply' | 'subscription-policy'; channels?: Array<{ channelId: string; label?: string }>; + agents: Array<[string, string[]]>; homes?: Array<[string, string]>; labels?: Array<[string, string]>; isolated?: string[]; + router?: boolean; template?: string; expired?: string[]; subconscious?: string; primary: string | null; + told: Array<{ text: string; metadata: unknown; forAgent?: string }>; + }> = [ + { why: "a second resident that can close it, which reads the residents' shared slot", name: 'helper', agents: [['helper', []], ['primary', []]], primary: 'primary', + told: [{ text: own + optOut + close, metadata: opened(fromReply), forAgent: 'helper' }] }, + { why: 'a run that can close it, whose store is its own and holds no channel traffic', name: 'run', agents: [['run', []], ['primary', []]], isolated: ['run'], primary: 'primary', + told: [{ text: own + close, metadata: opened(fromReply), forAgent: 'run' }] }, + { why: 'a fork away from its home', name: 'fork', agents: [['fork', []], ['primary', []]], homes: [['fork', 'y']], router: true, primary: 'primary', + told: [ + { text: "[channels] Now open: #x (x): a message from your fork in y, named fork, into the closed channel opened it, and that fork can't close it." + close, + metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: own + ' Your primary has been told and can close it.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, + ] }, + { why: 'a fork away from a home the registry labels', name: 'fork', agents: [['fork', []], ['primary', []]], homes: [['fork', 'y']], labels: [['y', '#dm-y']], router: true, primary: 'primary', + told: [ + { text: "[channels] Now open: #x (x): a message from your fork in #dm-y (y), named fork, into the closed channel opened it, and that fork can't close it." + close, + metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: own + ' Your primary has been told and can close it.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, + ] }, + { why: "a fork made from a template that isn't the primary", name: 'fork', agents: [['fork', []], ['primary', []], ['other', []]], homes: [['fork', 'y']], + router: true, template: 'other', primary: 'primary', + told: [ + { text: "[channels] Now open: #x (x): a message from the fork in y, named fork, into the closed channel opened it, and that fork can't close it." + close, + metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: own + ' primary has been told and can close it.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, + ] }, + { why: 'a fork at its home', name: 'fork', agents: [['fork', []], ['primary', []]], homes: [['fork', 'x']], router: true, primary: 'primary', + told: [{ text: own + optOut + close, metadata: opened(fromReply), forAgent: 'fork' }] }, + { why: 'a fork at its home whose binding the sweep has ended, closing: its home routes to it no more', name: 'fork', agents: [['fork', []], ['primary', []]], + homes: [['fork', 'x']], router: true, expired: ['fork'], primary: 'primary', + told: [{ text: own + close, metadata: opened(fromReply), forAgent: 'fork' }] }, + { why: 'a fork at its home, which the router has bound to a newer fork since: its home routes to that one', name: 'fork', + agents: [['fork', []], ['newer', []], ['primary', []]], homes: [['fork', 'x'], ['newer', 'x']], router: true, primary: 'primary', + told: [{ text: own + close, metadata: opened(fromReply), forAgent: 'fork' }] }, + { why: 'the subconscious, which reads the shared slot through its merged view', name: 'sub', agents: [['sub', []], ['primary', []]], subconscious: 'sub', primary: 'primary', + told: [{ text: own + optOut + close, metadata: opened(fromReply), forAgent: 'sub' }] }, + { why: 'a second resident whose allowedTools leave channel_close out', name: 'helper', agents: [['helper', ['channel_close']], ['primary', []]], primary: 'primary', + told: [ + { text: "[channels] Now open: #x (x): a message from helper into the closed channel opened it, and helper can't close it." + optOut + close, + metadata: { system: true, kind: 'channel-notice', speaker: 'helper' } }, + { text: own + optOut + ' primary has been told and can close it.', metadata: { system: true, kind: 'channel-notice' }, forAgent: 'helper' }, + ] }, + { why: "a primary that can't close it either", name: 'fork', agents: [['fork', []], ['primary', ['channel_close']]], homes: [['fork', 'y']], router: true, primary: 'primary', + told: [{ text: own, metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }] }, + { why: 'a primary named but not registered', name: 'fork', agents: [['fork', []], ['other', []]], homes: [['fork', 'y']], router: true, primary: 'gone', + told: [{ text: own, metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }] }, + { why: 'the primary itself, without channel_close', name: 'primary', agents: [['primary', ['channel_close']]], primary: 'primary', + told: [{ text: own + optOut, metadata: { system: true, kind: 'channel-notice' }, forAgent: 'primary' }] }, + { why: "several channels, one away from a fork's home", name: 'fork', channels: [{ channelId: 'y', label: '#y' }, { channelId: 'x', label: '#x' }], + agents: [['fork', []], ['primary', []]], homes: [['fork', 'y']], router: true, primary: 'primary', + told: [ + { text: "[channels] Now open: #y (y), #x (x): a message from your fork in y, named fork, into the closed channel opened it, and that fork can't close them." + closeMany, + metadata: { system: true, kind: 'channel-notice', speaker: 'fork' } }, + { text: '[channels] Now open: #y (y), #x (x): your own message into the closed channel opened it. Your primary has been told and can close them.', + metadata: { system: true, kind: 'channel-notice' }, forAgent: 'fork' }, + ] }, + { why: 'a policy admission, to the primary that can close it', cause: 'subscription-policy', agents: [['primary', []]], primary: 'primary', + // Addressed to no agent of its own, so it carries no inputs to pass on. + told: [{ text: '[channels] Selected for opening by subscription policy: #x (x). This records the admission decision; channel_list shows current transport state.' + optOut + close, + metadata: { system: true, kind: 'channel-notice' } }] }, + { why: 'a policy admission where conversations are routed, to the primary, whose window holds no channel traffic', cause: 'subscription-policy', agents: [['primary', []]], router: true, primary: 'primary', + told: [{ text: '[channels] Selected for opening by subscription policy: #x (x). This records the admission decision; channel_list shows current transport state.' + close, + metadata: { system: true, kind: 'channel-notice' } }] }, + { why: "a policy admission, to a primary that can't close it", cause: 'subscription-policy', agents: [['primary', ['channel_close']]], primary: 'primary', + told: [{ text: '[channels] Selected for opening by subscription policy: #x (x). This records the admission decision; channel_list shows current transport state.' + optOut, + metadata: { system: true, kind: 'channel-notice' } }] }, + { why: "a policy admission with no primary, to the first agent, a run whose store is its own, which can't close it", cause: 'subscription-policy', + agents: [['worker', ['channel_close']]], isolated: ['worker'], primary: null, + told: [{ text: '[channels] Selected for opening by subscription policy: #x (x). This records the admission decision; channel_list shows current transport state.' + optOut, + metadata: { system: true, kind: 'channel-notice' } }] }, + { why: "an agent that has ended, with a primary that can't close it", name: 'gone', agents: [['primary', ['channel_close']]], primary: 'primary', + told: [{ text: '[channels] Now open: #x (x): a message from gone into the closed channel opened it; gone has ended.' + optOut, + metadata: { system: true, kind: 'channel-notice', speaker: 'gone' } }] }, + { why: 'an agent that has ended, where conversations are routed', name: 'gone', agents: [['primary', []]], router: true, primary: 'primary', + told: [{ text: '[channels] Now open: #x (x): a message from gone into the closed channel opened it; gone has ended.' + close, + metadata: { system: true, kind: 'channel-notice', speaker: 'gone' } }] }, + ]; + for (const c of cases) { + const told: Array<{ text: string; metadata: unknown; forAgent?: string }> = []; + const logged: string[] = []; + const framework = Object.create(AgentFramework.prototype) as any; + framework.agents = new Map(c.agents.map(([name, denied]) => [name, { name, canUseTool: (tool: string) => !denied.includes(tool) }])); + framework.primaryAgentName = c.primary; + framework.usedEphemeralAgentNames = new Set(); + framework.conversationAgentHomes = new Map(c.homes ?? []); + // Each fork is bound to its home, unless the sweep has ended its binding; + // of two forks with one home, the router binds the later. + const bound = new Map((c.homes ?? []).filter(([fork]) => !(c.expired ?? []).includes(fork)).map(([fork, home]) => [home, fork])); + framework.conversationRouter = c.router + ? { templateAgent: c.template ?? 'primary', getBinding: (id: string) => (bound.has(id) ? { channelId: id, agentName: bound.get(id) } : undefined) } + : undefined; + framework.subconsciousAgentName = c.subconscious; + // Residents, and the subconscious through its merged view, read one shared + // slot; a fork's store, or a run's, is its own. + framework.sharedSlotAgents = new Set([...framework.agents.values()] + .filter((a: { name: string }) => !framework.conversationAgentHomes.has(a.name) && !(c.isolated ?? []).includes(a.name))); + const labels = new Map(c.labels ?? []); + framework.channelRegistry = { getDescriptor: (id: string) => (labels.has(id) ? { id, label: labels.get(id) } : undefined) }; + framework.addMessage = (_participant: string, content: Array<{ text: string }>, metadata: unknown, routed?: { forAgent?: string }) => { + told.push({ text: content[0]!.text, metadata, ...(routed ? { forAgent: routed.forAgent } : {}) }); + return 'm'; + }; + const error = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + framework.recordChannelAutoOpenNotice(c.name, 'srv', c.channels ?? [{ channelId: 'x', label: '#x' }], c.cause ?? 'opened-by-reply'); + } finally { + console.error = error; + } + assert.deepEqual(told, c.told, c.why); + if (c.told.length === 2) { + assert.ok(logged.some((l) => l.startsWith(`[channel] auto-open notice (opened-by-reply) -> the primary, by name (${c.name} can't close`)), `${c.why}: the log says why`); + } + } +}); + +// Each notice is written on its own: one that can't be written leaves the +// other to be written still. The primary's is written first, and the +// reader's says the primary was told only where the primary's landed. +test("each channel notice is written on its own, and the reader's says the primary was told only where the primary's landed, for both notices", () => { + for (const write of ['opened', 'reply'] as const) { + for (const refused of ['reader', 'primary'] as const) { + const told: Array<{ text: string; forAgent?: string }> = []; + const logged: string[] = []; + const framework = Object.create(AgentFramework.prototype) as any; + framework.agents = new Map([['fork', { name: 'fork', canUseTool: () => true }], ['primary', { name: 'primary', canUseTool: () => true }]]); + framework.primaryAgentName = 'primary'; + framework.usedEphemeralAgentNames = new Set(); + framework.conversationAgentHomes = new Map([['fork', 'y']]); + framework.conversationRouter = { templateAgent: 'primary' }; + framework.addMessage = (_participant: string, content: Array<{ text: string }>, _metadata: unknown, routed?: { forAgent?: string }) => { + // The reader's notice is addressed to the fork; the primary's to no one. + if ((refused === 'reader') === !!routed?.forAgent) throw new Error('store refused the write'); + told.push({ text: content[0]!.text, ...(routed?.forAgent ? { forAgent: routed.forAgent } : {}) }); + return 'm'; + }; + const error = console.error; + console.error = (...args: unknown[]) => { logged.push(args.map(String).join(' ')); }; + try { + if (write === 'opened') framework.recordChannelAutoOpenNotice('fork', 'srv', [{ channelId: 'x', label: '#x' }], 'opened-by-reply'); + else framework.recordReplyOpenFailedNotice('fork', 'srv', { channelId: 'x', label: '#x' }, 'server-bootstrap'); + } finally { + console.error = error; + } + const why = `${write}, the ${refused}'s write refused`; + assert.equal(told.length, 1, `${why}: the other notice is written`); + if (refused === 'reader') { + assert.equal(told[0]!.forAgent, undefined, why); + assert.match(told[0]!.text, /that fork can't close it/, why); + } else { + assert.equal(told[0]!.forAgent, 'fork', why); + assert.equal(told[0]!.text, write === 'opened' + ? '[channels] Now open: #x (x): your own message into the closed channel opened it.' + : "[channels] Not open: #x (x). Your reply there went out, but opening the channel failed. It's selected for opening again, because its server lists it as open by default: the host tries again.", + `${why}: the reader isn't told the primary was`); + } + assert.ok(logged.some((l) => l.startsWith(write === 'opened' ? 'recordChannelAutoOpenNotice failed:' : 'recordReplyOpenFailedNotice failed:')), + `${why}: the failure is logged`); + } + } +}); + +// A release finds a channel notice its reader could act on unread: the +// primary is told in its place, in the words the notice's builder gives an +// agent that has ended, judged for the primary. A notice whose inputs don't +// read back as this build writes them, or that carries none, isn't passed on. +test("a channel notice's inputs give the primary's by-name notice once its reader has ended, and malformed ones give none", () => { + const framework = Object.create(AgentFramework.prototype) as any; + framework.agents = new Map([['primary', { name: 'primary', canUseTool: (tool: string) => tool !== 'channel_open' }]]); + framework.primaryAgentName = 'primary'; + framework.usedEphemeralAgentNames = new Set(); + framework.conversationAgentHomes = new Map(); + // Every channel still as the notice says: open, or still waiting to open. + framework.channelRegistry = { getDesiredState: () => 'open', isChannelOpen: () => false }; + const error = console.error; + console.error = () => {}; + try { + const reply = framework.noticeForPrimary('gone', { system: true, kind: 'channel-notice', + passOn: { notice: 'reply-open-failed', serverId: 'srv', channel: { channelId: 'x', label: '#x' }, selectedForOpening: 'unrecorded' } }); + assert.deepEqual(reply?.content, [{ type: 'text', text: + "[channels] Not open: #x (x). A reply from gone went out there, but opening the channel failed; gone has ended. " + + "It stays selected for opening, because the host couldn't record undoing that: a later delivery or reconnect tries again, and channel_close withdraws that." }], + "judged for the primary, which can close it and can't call channel_open"); + assert.deepEqual(reply?.metadata, { system: true, kind: 'channel-notice', speaker: 'gone' }); + for (const passOn of [ + undefined, + { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: [] }, + { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-whim', channels: [{ channelId: 'x' }] }, + { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: [{ channelId: '' }] }, + { notice: 'reply-open-failed', serverId: 'srv', channel: { channelId: 'x' } }, + { notice: 'reply-open-failed', serverId: 'srv', channel: { channelId: 'x' }, selectedForOpening: 'maybe' }, + { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: [{ channelId: 7 }] }, + { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: 'x' }, + { notice: 'channel-opened', serverId: 'srv', cause: 'subscription-policy', channels: [{ channelId: 'x' }] }, + { notice: 'reply-open-failed', serverId: 'srv', channel: null, selectedForOpening: 'unrecorded' }, + { notice: 'something-else', serverId: 'srv' }, + // Each channel is checked on its own server, so a notice must name one. + { notice: 'channel-opened', cause: 'opened-by-reply', channels: [{ channelId: 'x' }] }, + { notice: 'channel-opened', serverId: '', cause: 'opened-by-reply', channels: [{ channelId: 'x' }] }, + { notice: 'reply-open-failed', serverId: 7, channel: { channelId: 'x' }, selectedForOpening: 'unrecorded' }, + ]) { + assert.equal(framework.noticeForPrimary('gone', { system: true, kind: 'channel-notice', ...(passOn ? { passOn } : {}) }), null, JSON.stringify(passOn)); + } + assert.equal(framework.noticeForPrimary('gone', { system: true, kind: 'routing-notice', passOn: { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: [{ channelId: 'x' }] } }), null, + 'only a channel notice'); + // A channel with no label, or one whose label isn't text, is named by its + // id, and so is one whose label is its id. + const unlabeled = framework.noticeForPrimary('gone', { system: true, kind: 'channel-notice', + passOn: { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-delivery', channels: [{ channelId: 'x', label: 9 }] } }); + assert.match(unlabeled?.content[0]?.text ?? '', /^\[channels\] Now open: x: a message from gone into the closed channel opened it; gone has ended\./); + const labelIsId = framework.noticeForPrimary('gone', { system: true, kind: 'channel-notice', + passOn: { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-delivery', channels: [{ channelId: 'x', label: 'x' }] } }); + assert.match(labelIsId?.content[0]?.text ?? '', /^\[channels\] Now open: x: a message from gone into the closed channel opened it; gone has ended\./); + } finally { + console.error = error; + } +}); + test("a reply's open that only reconfirms a tuned-out channel's transport, and fails, isn't one that would have opened it", async () => { const f = fixture(); await f.registry.handleChanged('test', { added: [descriptor('x', true, 'A')] }); diff --git a/test/channel-lifecycle-provenance.test.ts b/test/channel-lifecycle-provenance.test.ts index d27a776f2..ecb46dd52 100644 --- a/test/channel-lifecycle-provenance.test.ts +++ b/test/channel-lifecycle-provenance.test.ts @@ -1,6 +1,10 @@ import { test } from 'node:test'; import assert from 'node:assert/strict'; -import { ChannelRegistry } from '../src/mcpl/channel-registry.js'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { JsStore } from '@animalabs/chronicle'; +import { ChannelRegistry, desiredStateInLog } from '../src/mcpl/channel-registry.js'; import type { McplServerRegistry } from '../src/mcpl/server-registry.js'; import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; @@ -13,7 +17,7 @@ import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js'; * predated the server declaring its bootstrap preference. */ -function makeRegistry() { +function makeRegistry(store?: JsStore) { const serverRegistry = { getServer: (_id: string) => ({ sendChannelsPublish: async () => ({ delivered: true }) }), } as unknown as McplServerRegistry; @@ -22,6 +26,7 @@ function makeRegistry() { {} as FeatureSetManager, () => {}, () => {}, + store ? { store } : undefined, ); const r = registry as unknown as { ensureInitialDesiredState(serverId: string, ch: { id: string; type: string; label: string; direction: string; initiallyOpen?: boolean }): void; @@ -61,3 +66,58 @@ test('initiallyOpen at first sight registers open with server-bootstrap provenan r.ensureInitialDesiredState('eidoverse', ch(true)); assert.equal(r.getDesiredState('eidoverse', 'world:commons'), 'open'); }); + +test("isChannelOpen, given a server, reads only that server's channel, though the registry's key joins the two ids with a colon", () => { + const { registry } = makeRegistry(); + // Server "a" registers channel "b:c": its key is "a:b:c", which is also + // what server "a:b" and channel "c" would make. + (registry as unknown as { channels: Map }).channels + .set('a:b:c', { serverId: 'a', descriptor: { id: 'b:c', type: 'a', label: '#c' }, open: true }); + assert.equal(registry.isChannelOpen('b:c', 'a'), true, "a's own channel"); + assert.equal(registry.isChannelOpen('c', 'a:b'), false, "another server's entry under the same key"); +}); + +test("desiredStateInLog reads a server's channel's last decision from a store's lifecycle log, by the rule a registry replayed from it reads", () => { + const dir = mkdtempSync(join(tmpdir(), 'desired-state-in-log-')); + const store = JsStore.openOrCreate({ path: join(dir, 'store') }); + try { + assert.equal(desiredStateInLog(store, 'a', 'c-1'), undefined, 'a store no registry has written to'); + const { registry, r } = makeRegistry(store); + r.setDesiredState('a', 'c-1', 'open', 'opened-by-reply'); + r.setDesiredState('a', 'c-1', 'closed', 'agent-tool'); + r.setDesiredState('a', 'c-2', 'open', 'opened-by-reply'); + // One channel id on two servers, each with its own decision. + r.setDesiredState('a', 'c-3', 'open', 'opened-by-reply'); + r.setDesiredState('b', 'c-3', 'closed', 'agent-tool'); + registry.enterTuneOut('a', 'c-9', { epochId: 'e-9', cadenceSeconds: 60, backlogCap: 10, maxWakes: 3, startedAtSequence: 0 }, 'agent-tool'); + // Records the rule doesn't read as a decision: another kind, a channel id + // that isn't text, a tune-out with no params, a state no build writes. + // A decision with no source is still a decision. + for (const item of [ + { kind: 'channel-renamed', serverId: 'a', channelId: 'c-5', desired: 'open', source: 'agent-tool' }, + { kind: 'desired-state', serverId: 'a', channelId: 6, desired: 'open', source: 'agent-tool' }, + { kind: 'desired-state', serverId: 'a', channelId: 'c-7', desired: 'tuned-out', source: 'agent-tool' }, + { kind: 'desired-state', serverId: 'a', channelId: 'c-8', desired: 'maybe', source: 'agent-tool' }, + { kind: 'desired-state', serverId: 'a', channelId: 'c-10', desired: 'open' }, + ]) store.appendToStateJson('mcpl/channel-lifecycle', { ...item, timestamp: new Date().toISOString() } as never); + assert.equal(desiredStateInLog(store, 'a', 'c-1'), 'closed', 'the last decision wins'); + assert.equal(desiredStateInLog(store, 'a', 'c-2'), 'open'); + assert.equal(desiredStateInLog(store, 'a', 'c-3'), 'open', "a's own decision"); + assert.equal(desiredStateInLog(store, 'b', 'c-3'), 'closed', "b's own decision"); + assert.equal(desiredStateInLog(store, 'b', 'c-2'), undefined, "another server's channel"); + assert.equal(desiredStateInLog(store, 'a', 'c-9'), 'tuned-out'); + for (const [server, id] of [['a', 'c-5'], ['a', '6'], ['a', 'c-7'], ['a', 'c-8']]) { + assert.equal(desiredStateInLog(store, server!, id!), undefined, `${server}:${id} isn't a decision`); + } + assert.equal(desiredStateInLog(store, 'a', 'c-10'), 'open', 'a decision with no source'); + assert.equal(desiredStateInLog(store, 'a', 'c-4'), undefined, 'none on record'); + const replayed = makeRegistry(store).registry; + for (const [server, id] of [['a', 'c-1'], ['a', 'c-2'], ['a', 'c-3'], ['b', 'c-3'], ['b', 'c-2'], ['a', 'c-9'], + ['a', 'c-5'], ['a', '6'], ['a', 'c-7'], ['a', 'c-8'], ['a', 'c-10'], ['a', 'c-4']]) { + assert.equal(replayed.getDesiredState(server!, id!), desiredStateInLog(store, server!, id!), `${server}:${id}`); + } + } finally { + store.close(); + rmSync(dir, { recursive: true, force: true }); + } +}); diff --git a/test/conversation-router.test.ts b/test/conversation-router.test.ts index 407ff16ec..1a055d4f8 100644 --- a/test/conversation-router.test.ts +++ b/test/conversation-router.test.ts @@ -125,7 +125,7 @@ test('rebind after expiry gets a fresh generation and agent name', () => { if (d1.kind !== 'spawn') assert.fail('expected spawn'); router.bind('slack:C1', d1.agentName, d1.generation, T0); - router.unbind('slack:C1'); // the framework does this as the closure turn begins + router.unbind('slack:C1'); // the framework does this as the closure turn begins, just before starting it (sweepExpiredConversations) const d2 = router.route({ channelId: 'slack:C1', mentioned: true, kind: 'channel' }); assert.equal(d2.kind, 'spawn'); diff --git a/test/route-failure-targeting.test.ts b/test/route-failure-targeting.test.ts index 13b170826..569cc13a4 100644 --- a/test/route-failure-targeting.test.ts +++ b/test/route-failure-targeting.test.ts @@ -1,6 +1,6 @@ import { describe, it, beforeEach, afterEach } from 'node:test'; import assert from 'node:assert/strict'; -import { mkdtempSync, rmSync } from 'node:fs'; +import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { AgentFramework, ShutdownTimeoutError } from '../src/index.js'; @@ -33,7 +33,7 @@ function internals(framework: AgentFramework) { scriptWake(agentName: string, scriptId: string, text: string, label: string): Promise; closingConversationAgents: Set; activeStreams: Map>; - recordChannelAutoOpenNotice(agentName: string | undefined, channels: Array<{ channelId: string; label?: string }>, cause: 'subscription-policy' | 'opened-by-delivery' | 'opened-by-reply'): void; + recordChannelAutoOpenNotice(agentName: string | undefined, serverId: string, channels: Array<{ channelId: string; label?: string }>, cause: 'subscription-policy' | 'opened-by-delivery' | 'opened-by-reply'): void; addMessage(participant: string, content: ContentBlock[], metadata?: Record, opts?: { forAgent?: string }): string; }; } @@ -152,6 +152,13 @@ describe('notices about an agent\'s own action', () => { return framework.getAgent(binding.agentName)!; } + /** A durable decision for `channel` on `server`, as an open or a close + * records it in the registry's lifecycle log. */ + function decide(channel: string, desired: 'open' | 'closed', source = 'opened-by-reply', on = framework, server = 'srv') { + (internals(on).channelRegistry as unknown as { setDesiredState(serverId: string, channelId: string, desired: string, source: string): void }) + .setDesiredState(server, channel, desired, source); + } + async function runEphemeral(proseRouting: 'locus' | 'explicit', speech: string) { const created = await framework.createEphemeralAgent({ name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting, @@ -1167,20 +1174,120 @@ describe('notices about an agent\'s own action', () => { const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); const forkCm = t.mock.method(fork.getContextManager(), 'addMessage'); - state.recordChannelAutoOpenNotice(fork.name, [{ channelId: 'c-1', label: '#one' }], 'opened-by-reply'); + // The fork's own home channel, which it can close. + state.recordChannelAutoOpenNotice(fork.name, 'srv', [{ channelId: 'missing-channel', label: '#home' }], '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 opened it/); + assert.match(JSON.stringify(forkCm.mock.calls[0]!.arguments[1]), + /Now open: #home \(missing-channel\): your own message into the closed channel opened it\. .*use channel_close/); - state.recordChannelAutoOpenNotice(undefined, [{ channelId: 'c-2' }], 'subscription-policy'); + state.recordChannelAutoOpenNotice(undefined, 'srv', [{ channelId: 'c-2' }], 'subscription-policy'); assert.equal(primary.mock.callCount(), 1, 'no agent caused a policy open'); assert.equal(forkCm.mock.callCount(), 1); }); + it("keeps the server the registry names for an open a delivery forced, as the framework's callback writes the notice's inputs", async () => { + const fork = await spawnFork(); + const registry = internals(framework).channelRegistry as unknown as { + onChannelAutoOpened(info: { conversationId?: string; serverId: string; source: string; channels: Array<{ channelId: string; label?: string }> }): void; + }; + registry.onChannelAutoOpened({ conversationId: fork.name, serverId: 'other-server', source: 'opened-by-delivery', channels: [{ channelId: 'missing-channel', label: '#home' }] }); + const notice = fork.getContextManager().getAllMessages() + .find((m) => (m.metadata as { kind?: unknown } | undefined)?.kind === 'channel-notice'); + assert.equal((notice?.metadata as { passOn?: { serverId?: unknown } } | undefined)?.passOn?.serverId, 'other-server'); + }); + + it("tells a fork whose send opened a channel away from its home that it opened, without channel_close, which it can't call there, and tells the primary by name, which can", async (t) => { + const fork = await spawnFork(); + const state = internals(framework); + const primary = t.mock.method(framework.getAgent('primary')!.getContextManager(), 'addMessage'); + const forkCm = t.mock.method(fork.getContextManager(), 'addMessage'); + + state.recordChannelAutoOpenNotice(fork.name, 'srv', [{ channelId: 'c-1', label: '#one' }], 'opened-by-reply'); + + assert.equal(forkCm.mock.callCount(), 1); + assert.equal(JSON.stringify(forkCm.mock.calls[0]!.arguments[1]), JSON.stringify([{ type: 'text', text: + '[channels] Now open: #one (c-1): your own message into the closed channel opened it. Your primary has been told and can close it.' }])); + assert.equal(primary.mock.callCount(), 1, 'the primary is told too'); + assert.equal(JSON.stringify(primary.mock.calls[0]!.arguments[1]), JSON.stringify([{ type: 'text', text: + `[channels] Now open: #one (c-1): a message from your fork in missing-channel, named ${fork.name}, into the closed channel opened it, and that fork can't close it. ` + + "If you don't want it, use channel_close; your choice outranks policy." }])); + assert.equal((primary.mock.calls[0]!.arguments[2] as { speaker?: unknown } | undefined)?.speaker, fork.name); + + // The advice left out is advice the fork can't take: its channel_close + // there is refused. + const results: Array<{ success?: boolean; error?: string }> = []; + const push = framework.pushEvent.bind(framework); + (framework as unknown as { pushEvent: (e: { type: string; result?: { success?: boolean; error?: string } }) => void }).pushEvent = (e) => { + if (e.type === 'tool-result') results.push(e.result!); else push(e as never); + }; + try { + (framework as unknown as { dispatchChannelToolCall(agent: string, call: { id: string; name: string; input: Record }): void }) + .dispatchChannelToolCall(fork.name, { id: 'close-away', name: 'channel_close', input: { channelId: 'c-1' } }); + for (let i = 0; i < 50 && results.length === 0; i++) await new Promise((resolve) => setImmediate(resolve)); + } finally { + (framework as unknown as { pushEvent: unknown }).pushEvent = push; + } + assert.equal(results[0]?.success, false); + assert.match(results[0]?.error ?? '', /closing c-1 is not allowed/); + }); + + it('tells the primary, by name, of a channel notice a fork could act on and never read, when the fork is released', async () => { + const fork = await spawnFork(); + const state = internals(framework); + decide('missing-channel', 'open'); + state.recordChannelAutoOpenNotice(fork.name, 'srv', [{ channelId: 'missing-channel', label: '#home' }], 'opened-by-reply'); + + reap(fork.name); + + assert.equal(framework.getAgent(fork.name), null, 'released'); + const told = framework.getAgent('primary')!.getContextManager().getAllMessages() + .filter((m) => JSON.stringify(m.content).includes('#home')); + assert.equal(told.length, 1, 'the primary is told once'); + assert.equal((told[0]!.content[0] as { text: string }).text, + `[channels] Now open: #home (missing-channel): a message from your fork in missing-channel, named ${fork.name}, into the closed channel opened it; that fork has ended. ` + + "If you don't want it, use channel_close; your choice outranks policy."); + assert.equal((told[0]!.metadata as { speaker?: unknown } | undefined)?.speaker, fork.name); + }); + + it('tells the primary nothing, logged, of a channel a fork\'s unread notice said opened, once the channel has closed', async (t) => { + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + const fork = await spawnFork(); + decide('missing-channel', 'open'); + internals(framework).recordChannelAutoOpenNotice(fork.name, 'srv', [{ channelId: 'missing-channel', label: '#home' }], 'opened-by-reply'); + decide('missing-channel', 'closed', 'agent-tool'); + + reap(fork.name); + + assert.equal(framework.getAgent(fork.name), null, 'released'); + const told = framework.getAgent('primary')!.getContextManager().getAllMessages() + .filter((m) => JSON.stringify(m.content).includes('#home')); + assert.equal(told.length, 0, 'the primary is told nothing'); + assert.ok(errors.includes( + `[channel] ${fork.name} has ended, and no channel its notice said opened is decided open or tuned out now (#home (missing-channel)): no one is told`)); + }); + + it("passes on no channel notice its reader couldn't act on: the primary was told then, once", async () => { + const fork = await spawnFork(); + const state = internals(framework); + // Decided open, so only the notice's missing inputs keep it back. + decide('c-1', 'open'); + state.recordChannelAutoOpenNotice(fork.name, 'srv', [{ channelId: 'c-1', label: '#one' }], 'opened-by-reply'); + + reap(fork.name); + + assert.equal(framework.getAgent(fork.name), null, 'released'); + const told = framework.getAgent('primary')!.getContextManager().getAllMessages() + .filter((m) => JSON.stringify(m.content).includes('#one')); + assert.equal(told.length, 1, 'only the notice of the time'); + assert.match(JSON.stringify(told[0]!.content), new RegExp(`your fork in missing-channel, named ${fork.name}, into the closed channel opened it, and that fork can't close it`)); + }); + it('tells the primary, by name, of a channel an ended agent\'s send opened, and the primary\'s channel_close closes it', async (t) => { const state = internals(framework); const channels = (state.channelRegistry as unknown as { channels: Map }).channels; - // The channel the ended run's reply opened, on a server that lets it close. + // The channel the ended run's send opened, on a server that lets it close. channels.set('srv:opened-1', { serverId: 'srv', descriptor: { id: 'opened-1', type: 'srv', label: '#opened' }, open: true } as never); const closed: unknown[] = []; const server = { @@ -1190,12 +1297,12 @@ describe('notices about an agent\'s own action', () => { }; t.mock.method(state.mcplServerRegistry, 'getServer', () => server as never); - state.recordChannelAutoOpenNotice('released-run', [{ channelId: 'opened-1', label: '#opened' }], 'opened-by-delivery'); + state.recordChannelAutoOpenNotice('released-run', 'srv', [{ channelId: 'opened-1', label: '#opened' }], 'opened-by-delivery'); const told = framework.getAgent('primary')!.getContextManager().getAllMessages() .filter((m) => JSON.stringify(m.content).includes('Now open: #opened (opened-1)')); assert.equal(told.length, 1, 'the primary is told once'); - assert.match(JSON.stringify(told[0]!.content), /opened by released-run's reply; released-run has ended\. .*use channel_close/); + assert.match(JSON.stringify(told[0]!.content), /a message from released-run into the closed channel opened it; released-run has ended\. .*use channel_close/); assert.equal((told[0]!.metadata as { speaker?: unknown } | undefined)?.speaker, 'released-run'); assert.equal(state.deferredMessages.length, 0); @@ -1204,7 +1311,7 @@ describe('notices about an agent\'s own action', () => { .dispatchChannelToolCall('primary', { id: 'close-1', name: 'channel_close', input: { channelId: 'opened-1' } }); for (let i = 0; i < 50 && channels.get('srv:opened-1')!.open; i++) await new Promise((resolve) => setImmediate(resolve)); assert.deepEqual(closed, [{ channelId: 'opened-1' }]); - assert.equal(channels.get('srv:opened-1')!.open, false, 'the channel the reply opened is closed'); + assert.equal(channels.get('srv:opened-1')!.open, false, 'the channel the send opened is closed'); }); it('tells no one, logged, of a channel an ended agent\'s send opened when there is no primary', async (t) => { @@ -1215,7 +1322,7 @@ describe('notices about an agent\'s own action', () => { const name = state.primaryAgentName; state.primaryAgentName = null; try { - state.recordChannelAutoOpenNotice('released-run', [{ channelId: 'opened-2' }], 'opened-by-delivery'); + state.recordChannelAutoOpenNotice('released-run', 'srv', [{ channelId: 'opened-2' }], 'opened-by-delivery'); } finally { state.primaryAgentName = name; } @@ -1266,6 +1373,198 @@ describe('notices about an agent\'s own action', () => { assert.equal(state.deferredMessages.length, 0); }); + it('passes a restored channel notice its gone reader could act on to the primary, by name, under its own id; one it couldn\'t is dropped, logged', async (t) => { + const state = internals(framework); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + decide('c-4', 'open'); + decide('c-5', 'open', 'server-bootstrap'); + (state.deferredMessages as unknown as Array>).push( + { + id: 'w-4', seq: 4, participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: '[channels] Now open: #four (c-4): your own message into the closed channel opened it.' }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: [{ channelId: 'c-4', label: '#four' }] } }, + }, + { + id: 'w-5', seq: 5, participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: '[channels] Not open: #five (c-5). Your reply there went out, but opening the channel failed.' }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'reply-open-failed', serverId: 'srv', channel: { channelId: 'c-5', label: '#five' }, selectedForOpening: 'server-bootstrap' } }, + }, + { + // Its reader couldn't act on it, or an earlier build stored it: no inputs. + id: 'w-6', seq: 6, participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: '[channels] Now open: #six (c-6): your own message into the closed channel opened it.' }], + metadata: { system: true, kind: 'channel-notice' }, + }, + ); + + await state.flushDeferredWrites('boot-recovery'); + + const told = framework.getAgent('primary')!.getContextManager().getAllMessages() + .filter((m) => JSON.stringify(m.content).includes('conversation-gone')); + assert.deepEqual(told.map((m) => (m.content[0] as { text: string }).text), [ + "[channels] Now open: #four (c-4): a message from conversation-gone into the closed channel opened it; conversation-gone has ended. " + + "If you don't want it, use channel_close; your choice outranks policy.", + "[channels] Not open: #five (c-5). A reply from conversation-gone went out there, but opening the channel failed; conversation-gone has ended. " + + "It's selected for opening again, because its server lists it as open by default: the host tries again, and channel_close withdraws that.", + ]); + assert.deepEqual(told.map((m) => (m.metadata as { deferredWriteId?: unknown; speaker?: unknown }).deferredWriteId), ['w-4', 'w-5']); + assert.deepEqual(told.map((m) => (m.metadata as { speaker?: unknown }).speaker), ['conversation-gone', 'conversation-gone']); + const all = framework.getAgent('primary')!.getContextManager().getAllMessages().map((m) => JSON.stringify(m.content)); + assert.equal(all.filter((c) => c.includes('#six')).length, 0); + assert.ok(errors.some((e) => /"conversation-gone" is not a registered agent — message dropped \(participant=user\)/.test(e))); + assert.equal(state.deferredMessages.length, 0); + }); + + it('passes on, of a restored channel notice, only what still holds: the channels still decided open or tuned out, and a failed one still decided open and not open', async (t) => { + const state = internals(framework); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + decide('c-7', 'open'); + decide('c-8', 'open'); + decide('c-8', 'closed', 'agent-tool'); + // Selected for opening, and open since. + decide('c-9', 'open', 'server-bootstrap'); + (state.channelRegistry as unknown as { channels: Map }).channels + .set('srv:c-9', { serverId: 'srv', descriptor: { id: 'c-9', type: 'srv', label: '#nine' }, open: true }); + // Selected for opening on its own server, where it isn't open, while + // another server's channel with the same id is. + decide('c-22', 'open', 'server-bootstrap'); + (state.channelRegistry as unknown as { channels: Map }).channels + .set('other:c-22', { serverId: 'other', descriptor: { id: 'c-22', type: 'other', label: '#twenty-two' }, open: true }); + // Selected for opening, and closed since. + decide('c-10', 'open', 'server-bootstrap'); + decide('c-10', 'closed', 'agent-tool'); + // Opened, and tuned out since: still open, its traffic held. + decide('c-13', 'open'); + state.channelRegistry.enterTuneOut('srv', 'c-13', + { epochId: 'e-13', cadenceSeconds: 60, backlogCap: 10, maxWakes: 3, startedAtSequence: 0 }, 'agent-tool'); + // One channel id on two servers: each notice is checked by its own + // server's decision. + decide('c-14', 'open'); + decide('c-14', 'closed', 'agent-tool', framework, 'other'); + // Selected for opening, and tuned out since: a newer decision, by the + // agent that tuned it out, so its failed notice is left out. + decide('c-15', 'open', 'server-bootstrap'); + state.channelRegistry.enterTuneOut('srv', 'c-15', + { epochId: 'e-15', cadenceSeconds: 60, backlogCap: 10, maxWakes: 3, startedAtSequence: 0 }, 'agent-tool'); + const opened = (id: string, channels: Array<{ channelId: string; label: string }>, serverId = 'srv') => ({ + id, seq: Number(id.slice(2)), participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: '[channels] Now open: your own message into the closed channel opened it.' }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'channel-opened', serverId, cause: 'opened-by-delivery', channels } }, + }); + const failed = (id: string, channel: { channelId: string; label: string }) => ({ + id, seq: Number(id.slice(2)), participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: '[channels] Not open. Your reply there went out, but opening the channel failed.' }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'reply-open-failed', serverId: 'srv', channel, selectedForOpening: 'server-bootstrap' } }, + }); + (state.deferredMessages as unknown as Array>).push( + opened('w-7', [{ channelId: 'c-7', label: '#seven' }, { channelId: 'c-8', label: '#eight' }]), + opened('w-8', [{ channelId: 'c-8', label: '#eight' }]), + failed('w-9', { channelId: 'c-9', label: '#nine' }), + failed('w-10', { channelId: 'c-10', label: '#ten' }), + opened('w-13', [{ channelId: 'c-13', label: '#thirteen' }]), + opened('w-14', [{ channelId: 'c-14', label: '#fourteen' }]), + failed('w-15', { channelId: 'c-15', label: '#fifteen' }), + opened('w-16', [{ channelId: 'c-14', label: '#fourteen' }], 'other'), + failed('w-22', { channelId: 'c-22', label: '#twenty-two' }), + ); + + await state.flushDeferredWrites('boot-recovery'); + + const told = framework.getAgent('primary')!.getContextManager().getAllMessages() + .filter((m) => JSON.stringify(m.content).includes('conversation-gone')); + assert.deepEqual(told.map((m) => (m.content[0] as { text: string }).text), [ + '[channels] Now open: #seven (c-7): a message from conversation-gone into the closed channel opened it; conversation-gone has ended. ' + + "If you don't want it, use channel_close; your choice outranks policy.", + '[channels] Now open: #thirteen (c-13): a message from conversation-gone into the closed channel opened it; conversation-gone has ended. ' + + "If you don't want it, use channel_close; your choice outranks policy.", + '[channels] Now open: #fourteen (c-14): a message from conversation-gone into the closed channel opened it; conversation-gone has ended. ' + + "If you don't want it, use channel_close; your choice outranks policy.", + "[channels] Not open: #twenty-two (c-22). A reply from conversation-gone went out there, but opening the channel failed; conversation-gone has ended. " + + "It's selected for opening again, because its server lists it as open by default: the host tries again, and channel_close withdraws that.", + ]); + assert.deepEqual(told.map((m) => (m.metadata as { deferredWriteId?: unknown }).deferredWriteId), ['w-7', 'w-13', 'w-14', 'w-22']); + for (const line of [ + '[channel] conversation-gone has ended, and no channel its notice said opened is decided open or tuned out now (#eight (c-8)): no one is told', + '[channel] conversation-gone has ended, and #nine (c-9), which its notice said was not open and selected for opening, is open now or no longer decided open: no one is told', + '[channel] conversation-gone has ended, and #ten (c-10), which its notice said was not open and selected for opening, is open now or no longer decided open: no one is told', + '[channel] conversation-gone has ended, and #fifteen (c-15), which its notice said was not open and selected for opening, is open now or no longer decided open: no one is told', + '[channel] conversation-gone has ended, and no channel its notice said opened is decided open or tuned out now (#fourteen (c-14)): no one is told', + ]) assert.ok(errors.includes(line), line); + assert.equal(state.deferredMessages.length, 0); + }); + + it('reads a stranded channel notice\'s channels from the lifecycle log at a serving boot, before there is a registry', async (t) => { + decide('c-11', 'open'); + decide('c-12', 'open'); + decide('c-12', 'closed', 'agent-tool'); + // A failed reply's channel, selected for opening still, and one closed + // since. + decide('c-17', 'open', 'server-bootstrap'); + decide('c-18', 'open', 'server-bootstrap'); + decide('c-18', 'closed', 'agent-tool'); + // Opened, and tuned out since, which the log records with its params. + decide('c-19', 'open'); + internals(framework).channelRegistry.enterTuneOut('srv', 'c-19', + { epochId: 'e-19', cadenceSeconds: 60, backlogCap: 10, maxWakes: 3, startedAtSequence: 0 }, 'agent-tool'); + // One channel id on two servers, each with its own decision. + decide('c-20', 'open'); + decide('c-20', 'closed', 'agent-tool', framework, 'other'); + await framework.stop(); + const storePath = join(tempDir, 'test.chronicle'); + const notice = (id: string, channelId: string, label: string, serverId = 'srv') => ({ + id, seq: Number(id.slice(2)), participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: `[channels] Now open: ${label} (${channelId}): your own message into the closed channel opened it.` }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'channel-opened', serverId, cause: 'opened-by-reply', channels: [{ channelId, label }] } }, + }); + const failed = (id: string, channelId: string, label: string) => ({ + id, seq: Number(id.slice(2)), participant: 'user', forAgent: 'conversation-gone', + content: [{ type: 'text', text: `[channels] Not open: ${label} (${channelId}). Your reply there went out, but opening the channel failed.` }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'reply-open-failed', serverId: 'srv', channel: { channelId, label }, selectedForOpening: 'server-bootstrap' } }, + }); + // A resume that didn't finish its flush leaves the flag cleared and the + // queue behind, which a serving boot lands before MCPL is brought up. + mkdirSync(join(storePath, 'recovery'), { recursive: true }); + writeFileSync(join(storePath, 'recovery', 'host-mode.json'), JSON.stringify({ version: 1, quiesced: false })); + writeFileSync(join(storePath, 'recovery', 'deferred-writes.json'), JSON.stringify({ + version: 1, pending: [ + notice('w-11', 'c-11', '#eleven'), notice('w-12', 'c-12', '#twelve'), + failed('w-17', 'c-17', '#seventeen'), failed('w-18', 'c-18', '#eighteen'), + notice('w-19', 'c-19', '#nineteen'), + notice('w-20', 'c-20', '#twenty'), notice('w-21', 'c-20', '#twenty', 'other'), + ], + })); + const errors: string[] = []; + t.mock.method(console, 'error', (...args: unknown[]) => { errors.push(args.map(String).join(' ')); }); + + framework = await AgentFramework.create({ + storePath, + membrane: membrane.asMembrane(), + agents: [ + { name: 'primary', model: 'test-model', systemPrompt: 'Primary.', proseRouting: 'locus' }, + { name: 'secondary', model: 'test-model', systemPrompt: 'Secondary.' }, + ], + modules: [], + conversations: { templateAgent: 'primary' }, + }); + + assert.equal(internals(framework).channelRegistry, null, 'the boot landed them with no registry'); + const told = framework.getAgent('primary')!.getContextManager().getAllMessages() + .filter((m) => JSON.stringify(m.content).includes('conversation-gone')); + assert.deepEqual(told.map((m) => (m.metadata as { deferredWriteId?: unknown }).deferredWriteId), ['w-11', 'w-17', 'w-19', 'w-20']); + const texts = told.map((m) => (m.content[0] as { text: string }).text); + assert.match(texts[0]!, /^\[channels\] Now open: #eleven \(c-11\): a message from conversation-gone into the closed channel opened it; conversation-gone has ended\./); + assert.match(texts[1]!, /^\[channels\] Not open: #seventeen \(c-17\)\. A reply from conversation-gone went out there, but opening the channel failed; conversation-gone has ended\. It's selected for opening again/); + assert.match(texts[2]!, /^\[channels\] Now open: #nineteen \(c-19\): a message from conversation-gone/); + assert.match(texts[3]!, /^\[channels\] Now open: #twenty \(c-20\): a message from conversation-gone/); + for (const line of [ + '[channel] conversation-gone has ended, and no channel its notice said opened is decided open or tuned out now (#twelve (c-12)): no one is told', + '[channel] conversation-gone has ended, and #eighteen (c-18), which its notice said was not open and selected for opening, is open now or no longer decided open: no one is told', + '[channel] conversation-gone has ended, and no channel its notice said opened is decided open or tuned out now (#twenty (c-20)): no one is told', + ]) assert.ok(errors.includes(line), line); + }); + it('passes a restored failure notice on as the same queue entry while the primary is mid-turn: it waits under its own id', async () => { const state = internals(framework); (state.deferredMessages as unknown as Array>).push({ @@ -1285,14 +1584,43 @@ describe('notices about an agent\'s own action', () => { assert.match(JSON.stringify(entry.content), /A reply by conversation-gone \(12 chars\) could not be delivered/); }); + it('counts a body the store shards as one notice, and the same text stored again as a notice of its own', async () => { + const state = internals(framework) as unknown as { unreadNoticesForPrimary(agent: unknown): Array<{ tag?: unknown }> }; + const agent = framework.getAgent('secondary')!; + const cm = agent.getContextManager(); + // Shard every notice into one message per block, as a strategy's ingress + // chunking does. The group id names the text, as context-manager's + // chunker's does, so the same text shards under the same id each time + // it's stored. + (cm as unknown as { strategy: Record }).strategy.chunkIngressMessage = (_participant: string, content: ContentBlock[]) => + content.length > 1 + ? { + bodyGroupId: `bg-${content.map((block) => (block as { text?: string }).text).join('')}`, + shards: content.map((block, shardIndex) => ({ content: [block], shardIndex })), + } + : null; + const notice = (tag: string, words: string) => cm.addMessage('user', + [{ type: 'text', text: `[discord-send-failed] ${words}` }, { type: 'text', text: ' and the rest of it' }], + { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never); + + notice('first', 'the same words'); + notice('second', 'the same words'); + notice('third', 'other words'); + + const stored = cm.getAllMessages().filter((m) => (m.metadata as { kind?: unknown } | undefined)?.kind === 'discord-send-failed'); + assert.deepEqual(stored.map((m) => m.shardIndex), [0, 1, 0, 1, 0, 1], 'each body is stored as two shards'); + assert.equal(stored[2]!.bodyGroupId, stored[0]!.bodyGroupId, 'the same text is stored again under the same group id'); + assert.deepEqual(state.unreadNoticesForPrimary(agent).map((m) => m.tag), ['first', 'second', 'third']); + }); + it('counts as read, across branches, only what an agent\'s window shares with the branch it read up to the point they part', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { unreadNoticesForPrimary(agent: unknown): Array<{ tag?: unknown }> }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.unreadNoticesForPrimary(agent).map((m) => m.tag); const read = notice('read'); agent.markContextConsumed(); @@ -1312,13 +1640,13 @@ describe('notices about an agent\'s own action', () => { }); it('counts as unread a notice stored on a branch after the agent read past the point where it parts from another', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { unreadNoticesForPrimary(agent: unknown): Array<{ tag?: unknown }> }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.unreadNoticesForPrimary(agent).map((m) => m.tag); notice('before the branch'); const home = store.currentBranch().name; @@ -1336,13 +1664,13 @@ describe('notices about an agent\'s own action', () => { }); it('counts as unread what the current branch has from a branch between it and the one the agent read on', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { unreadNoticesForPrimary(agent: unknown): Array<{ tag?: unknown }> }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.unreadNoticesForPrimary(agent).map((m) => m.tag); notice('read at home'); const home = store.currentBranch().name; @@ -1363,13 +1691,13 @@ describe('notices about an agent\'s own action', () => { }); it('counts as unread, across a branch made after the agent read, what was stored after its watermark', async () => { - const state = internals(framework) as unknown as { unreadFailedReplies(agent: unknown): Array<{ tag?: unknown }> }; + const state = internals(framework) as unknown as { unreadNoticesForPrimary(agent: unknown): Array<{ tag?: unknown }> }; const agent = framework.getAgent('secondary')!; const cm = agent.getContextManager(); const store = framework.getStore(); const notice = (tag: string) => cm.getMessage(cm.addMessage('user', [{ type: 'text', text: `[discord-send-failed] ${tag}` }], { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', tag } as never))!; - const tags = () => state.unreadFailedReplies(agent).map((m) => m.tag); + const tags = () => state.unreadNoticesForPrimary(agent).map((m) => m.tag); notice('read'); agent.markContextConsumed(); @@ -1629,7 +1957,26 @@ describe('notices about an agent\'s own action', () => { assert.equal(count(messages, 'A reply by worker'), 1, 'and, still queued at the release, it goes on to the primary by name'); }); - it('passes a failed reply still queued when a run ends unquiesced on to the primary as the same queue entry', async () => { + const queuedForTheRun = [ + { + what: 'a failed reply', + content: [{ type: 'text', text: `${SEND_FAILED} Your previous reply (5 chars) could not be delivered to #x (missing-channel).` }], + metadata: { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 5 }, + inRun: SEND_FAILED, + toPrimary: 'A reply by worker', + opened: null, + }, + { + what: 'a channel notice the run could act on', + content: [{ type: 'text', text: '[channels] Now open: #seven (c-7): your own message into the closed channel opened it.' }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-delivery', channels: [{ channelId: 'c-7', label: '#seven' }] } }, + inRun: 'Now open: #seven', + toPrimary: "a message from worker into the closed channel opened it; worker has ended", + opened: 'c-7', + }, + ]; + for (const queued of queuedForTheRun) it(`passes ${queued.what} still queued when a run ends unquiesced on to the primary as the same queue entry`, async () => { + if (queued.opened) decide(queued.opened, 'open', 'opened-by-delivery', runFramework); let release!: () => void; const released = new Promise((resolve) => { release = resolve; }); runMembrane.streamYielding = (request) => { @@ -1646,12 +1993,10 @@ describe('notices about an agent\'s own action', () => { const ended = done.then(() => null, (error: unknown) => error); for (let i = 0; i < 400 && !state.activeStreams.has('worker'); i++) await new Promise((resolve) => setTimeout(resolve, 5)); assert.ok(state.activeStreams.has('worker'), 'the run\'s stream started'); - // A failed reply queued for the run's turn end, which the run's stall - // then ends first. + // A notice queued for the run's turn end, which the run's stall then + // ends first. (state.deferredMessages as unknown as Array>).push({ - id: 'w-late', seq: 1_000_000, participant: 'user', forAgent: 'worker', - content: [{ type: 'text', text: `${SEND_FAILED} Your previous reply (5 chars) could not be delivered to #x (missing-channel).` }], - metadata: { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 5 }, + id: 'w-late', seq: 1_000_000, participant: 'user', forAgent: 'worker', content: queued.content, metadata: queued.metadata, }); silenceRun(runFramework, 'worker'); assert.match(String(await ended), /stalled/, 'the idle watchdog ended the run'); @@ -1663,11 +2008,11 @@ describe('notices about an agent\'s own action', () => { const archive = await ContextManager.open({ store: runFramework.getStore(), namespace: 'subagent/worker', isolate: true, strategy: new PassthroughStrategy(), }); - const record = archive.getAllMessages().filter((m) => JSON.stringify(m.content).includes(SEND_FAILED)); + const record = archive.getAllMessages().filter((m) => JSON.stringify(m.content).includes(queued.inRun)); assert.equal(record.length, 1, 'it landed in the run\'s window'); assert.equal(idOf(record[0]), 'w-late'); const told = runFramework.getAgent('primary')!.getContextManager().getAllMessages() - .filter((m) => JSON.stringify(m.content).includes('A reply by worker')); + .filter((m) => JSON.stringify(m.content).includes(queued.toPrimary)); assert.equal(told.length, 1, 'the primary is told once'); assert.equal(idOf(told[0]), 'w-late', 'as the same entry, under its id'); }); @@ -1699,15 +2044,30 @@ describe('notices about an agent\'s own action', () => { assert.equal(count(await archived(), 'Written for the run while the host was quiesced.'), 1, 'and lands in its window at resume'); }); - it('passes a failed reply a run stored unread on to the primary\'s queue when the host stops before the release it waited for', async () => { + const storedUnread = [ + { + what: 'a failed reply', + content: [{ type: 'text' as const, text: `${SEND_FAILED} Your previous reply (5 chars) could not be delivered to #x (missing-channel).` }], + metadata: { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 5 }, + toPrimary: 'A reply by worker', + opened: null, + }, + { + what: 'a channel notice the run could act on', + content: [{ type: 'text' as const, text: '[channels] Now open: #eight (c-8): your own message into the closed channel opened it.' }], + metadata: { system: true, kind: 'channel-notice', passOn: { notice: 'channel-opened', serverId: 'srv', cause: 'opened-by-reply', channels: [{ channelId: 'c-8', label: '#eight' }] } }, + toPrimary: "a message from worker into the closed channel opened it; worker has ended", + opened: 'c-8', + }, + ]; + for (const unread of storedUnread) it(`passes ${unread.what} a run stored unread on to the primary\'s queue when the host stops before the release it waited for`, async () => { + if (unread.opened) decide(unread.opened, 'open', 'opened-by-reply', runFramework); const created = await runFramework.createEphemeralAgent({ name: 'worker', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all', proseRouting: 'explicit', }); created.contextManager.addMessage('user', [{ type: 'text', text: 'Run once.' }]); // Stored in its window, and never read: the run never compiled. - created.contextManager.addMessage('user', - [{ type: 'text', text: `${SEND_FAILED} Your previous reply (5 chars) could not be delivered to #x (missing-channel).` }], - { system: true, kind: 'discord-send-failed', channelId: 'missing-channel', reason: 'no registered channel', textLen: 5 } as never); + created.contextManager.addMessage('user', unread.content, unread.metadata as never); const state = internals(runFramework) as unknown as ReturnType & { releasesAfterResume: Map }; // Not started, so its startup watchdog ends it, while the host is quiesced. const done = runFramework.runEphemeralToCompletion(created.agent, created.contextManager, { startupTimeoutMs: 150 }); @@ -1718,7 +2078,7 @@ describe('notices about an agent\'s own action', () => { await runFramework.stop(); const passed = (state.deferredMessages as unknown as Array<{ forAgent?: string; content: ContentBlock[]; metadata?: Record }>) - .filter((m) => JSON.stringify(m.content).includes('A reply by worker')); + .filter((m) => JSON.stringify(m.content).includes(unread.toPrimary)); assert.equal(passed.length, 1, 'queued for the primary, for resume or boot'); assert.equal(passed[0]!.forAgent, undefined); assert.equal(passed[0]!.metadata?.speaker, 'worker'); @@ -1771,7 +2131,7 @@ describe('notices about an agent\'s own action', () => { assert.equal(state.deferredMessages.length, 0); assert.equal(count(await archived(), 'Queued for the run.'), 1, 'what was queued landed in its window'); - assert.ok(errors.some((e) => e.includes('[route-failure] worker: could not read its window for unread failed replies'))); + assert.ok(errors.some((e) => e.includes('[route-failure] worker: could not read its window for unread notices the primary would take'))); }); it('resumes past a run that ended while the host was quiesced: resume doesn\'t weigh its context', async (t) => { diff --git a/test/stop-step-throws.test.ts b/test/stop-step-throws.test.ts index 7d6db96e4..27d0f8644 100644 --- a/test/stop-step-throws.test.ts +++ b/test/stop-step-throws.test.ts @@ -115,7 +115,7 @@ const sentCompletes: string[] = []; * stop() for its first pass, or while it waits (from the liveness watchdog's * stop(), which runs between its two passes) for its pass after the waits. * Other agents, such as the run whose end releases its window, read their - * unread replies as usual. + * unread notices as usual. */ function postponedReleases(framework: AgentFramework, error: Error, when: 'before stop()' | 'while stop() waits'): () => void { unreadPassed.length = 0; @@ -129,14 +129,14 @@ function postponedReleases(framework: AgentFramework, error: Error, when: 'befor if (when === 'before stop()') postpone(); else watchdog.stop = function (this: unknown) { postpone(); return stop.call(this); }; const owner = framework as unknown as Record; - const usual = (owner.unreadFailedReplies as (agent: unknown) => unknown[]).bind(framework); - owner.unreadFailedReplies = (agent: { name: string }) => { + const usual = (owner.unreadNoticesForPrimary as (agent: unknown) => unknown[]).bind(framework); + owner.unreadNoticesForPrimary = (agent: { name: string }) => { if (agent.name === 'first') throw error; if (agent.name !== 'second') return usual(agent); unreadPassed.push(agent.name); return []; }; - return () => { delete owner.unreadFailedReplies; watchdog.stop = stop; }; + return () => { delete owner.unreadNoticesForPrimary; watchdog.stop = stop; }; } function checkPostponedReleases(framework: AgentFramework): void { diff --git a/test/trunk-channel-routing.test.ts b/test/trunk-channel-routing.test.ts index 26fc8fa3d..05e6f86e2 100644 --- a/test/trunk-channel-routing.test.ts +++ b/test/trunk-channel-routing.test.ts @@ -456,6 +456,7 @@ describe('Trunk channel routing (item-3 redux)', () => { const notice = internals(framework) as unknown as { recordChannelAutoOpenNotice( agentName: string | undefined, + serverId: string, channels: Array<{ channelId: string; label?: string }>, cause: 'subscription-policy' | 'opened-by-delivery' | 'opened-by-reply', ): void; @@ -468,7 +469,7 @@ describe('Trunk channel routing (item-3 redux)', () => { // Each form an agent can read: a delivery or a send opens one channel, and // only the subscription policy opens several at once. - notice.recordChannelAutoOpenNotice('scout', [{ channelId: 'slack:C1', label: '#general (acme)' }], 'opened-by-delivery'); + notice.recordChannelAutoOpenNotice('scout', 'slack', [{ channelId: 'slack:C1', label: '#general (acme)' }], 'opened-by-delivery'); const one = lastText(); assert.match( one, @@ -476,7 +477,7 @@ describe('Trunk channel routing (item-3 redux)', () => { ); assert.doesNotMatch(one, /will reach you/); - notice.recordChannelAutoOpenNotice('scout', [{ channelId: 'slack:C1' }, { channelId: 'slack:C2' }], 'subscription-policy'); + notice.recordChannelAutoOpenNotice('scout', 'slack', [{ channelId: 'slack:C1' }, { channelId: 'slack:C2' }], 'subscription-policy'); const two = lastText(); assert.match( two,