From d4029684022990cd1800ca321843c454eb559332 Mon Sep 17 00:00:00 2001 From: Greta-2425 Date: Wed, 21 Oct 2026 17:05:13 -0400 Subject: [PATCH 1/2] fix(channels): channel_open and channel_close promise delivery only to an agent whose window holds the channel's traffic channel_close told every agent "Messages that address you still reach you", and channel_open offered history "preceding the message that invited you". Both hold for a holder: a resident of an unrouted host, whose shared slot an incoming message is stored in, and a conversation fork for its home. Neither holds for an ephemeral run, whose store is its own, or on a host that routes conversations for any agent but a fork, since a channel's messages go to its fork or nowhere. A run that opened a channel to read it would wait for messages that go elsewhere. getToolsForAgent now gives a non-holder its own words for the two tools (NON_HOLDER_CHANNEL_DESCRIPTIONS). Opening or closing a channel does so for every agent on this host, since the decision is recorded per server and channel; without that clause a run could take its close as costing nothing, when it closes the channel for the residents too. Opening it brings the caller none of its messages, and closing it stops none reaching it: a push that names a channel is stored in the residents' shared slot whether or not the channel is open, so on a routed host the residents get the channel's pushes, and only those two clauses hold on both lanes. channel_open's history comes back in the call, as handleToolOpen returns it. A holder keeps the registry's words, so its cached tool list doesn't change. A fork is a holder by its home rather than its binding (holdsChannelTrafficForTools), so one whose binding expired keeps its words through its closure turn instead of sending its whole window uncached for a sentence false for that turn; holdsChannelTraffic, which #598's notices read, still goes by the binding. The resident part of both is one helper, residentHoldsTraffic. On a host with no residents the delivery agent is the first registered agent, so that run is a holder, and when it ends the next run's words change to the holder's. Spec: test/channel-tool-words.test.ts. Residents of an unrouted host keep the registry's words; an ephemeral run's request carries the non-holder's; a routed host's trunk gets the non-holder's, and its fork the registry's, in its closure turn's request too, after the idle sweep unbinds its home. Found by Petra-1012; designed by Ada-1017; read by Esther-2584. Co-Authored-By: Claude Opus 5.5 --- ...hannel-tool-words-for-non-holders.fixed.md | 14 ++ src/framework.ts | 25 ++- src/mcpl/channel-registry.ts | 27 ++++ test/channel-tool-words.test.ts | 147 ++++++++++++++++++ 4 files changed, 212 insertions(+), 1 deletion(-) create mode 100644 changelog.d/channel-tool-words-for-non-holders.fixed.md create mode 100644 test/channel-tool-words.test.ts diff --git a/changelog.d/channel-tool-words-for-non-holders.fixed.md b/changelog.d/channel-tool-words-for-non-holders.fixed.md new file mode 100644 index 000000000..27fa2ee90 --- /dev/null +++ b/changelog.d/channel-tool-words-for-non-holders.fixed.md @@ -0,0 +1,14 @@ +- `channel_open` and `channel_close` promise delivery only to an agent whose + window holds the channel's traffic. An ephemeral run, and on a host that + routes conversations every agent but a conversation fork, are now told that + opening or closing a channel does so for every agent on the host, that + opening it brings them none of its messages and closing it stops none + reaching them, and that `channel_open`'s history comes back in its result. + Before, they were told "Messages that address you still reach you" and + offered history "preceding the message that invited you", though a channel's + messages go to the residents, or to its fork or nowhere. Residents of a host + that doesn't route conversations, and conversation forks, keep the words they + had, so their cached tool lists don't change; a fork keeps them through its + closure turn after its binding expires. On a host with no residents, the run channel messages are stored + with is a holder, so when it ends, the next run's words change to the + holder's. diff --git a/src/framework.ts b/src/framework.ts index 9bea1fd49..05a0ded7a 100644 --- a/src/framework.ts +++ b/src/framework.ts @@ -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, desiredStateInLog, destinationKey, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome, type SelectedForOpening } from './mcpl/channel-registry.js'; +import { ChannelRegistry, connectorText, desiredStateInLog, destinationKey, journalToolDescription, NON_HOLDER_CHANNEL_DESCRIPTIONS, 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'; @@ -6084,6 +6084,10 @@ export class AgentFramework { ); } if (tool.name === 'journal' && !canSkipReply) return { ...tool, description: journalToolDescription(false) }; + // A holder's words are the registry's, so its cached tool list doesn't change. + if ((tool.name === 'channel_open' || tool.name === 'channel_close') && !this.holdsChannelTrafficForTools(agentName)) { + return { ...tool, description: NON_HOLDER_CHANNEL_DESCRIPTIONS[tool.name] }; + } return tool; }); } @@ -15568,6 +15572,25 @@ export class AgentFramework { return channelIds.every((id) => id === home && this.conversationRouter?.getBinding(id)?.agentName === who); } if (this.conversationRouter) return false; + return this.residentHoldsTraffic(who); + } + + /** + * Whether `who`'s window holds the traffic of the channels it may open or close, for the words its channel_open and + * channel_close carry. A conversation fork does for its home, the one + * channel it may close, decided by the home rather than the binding, so a fork whose binding expired keeps its words + * through its closure turn: changing them then would send its whole window uncached, for a sentence false for that + * turn only. Otherwise as holdsChannelTraffic has it: on a host that routes conversations, the trunk holds none, and + * otherwise the delivery agent and the readers of the residents' shared slot do. An ephemeral run holds none. + */ + private holdsChannelTrafficForTools(who: string): boolean { + if (this.conversationAgentHomes.has(who)) return true; + if (this.conversationRouter) return false; + return this.residentHoldsTraffic(who); + } + + /** Whether `who` is the delivery agent, or reads the residents' shared slot it stores into. */ + private residentHoldsTraffic(who: string): boolean { const delivery = this.deliveryAgent(); const agent = this.agents.get(who); if (delivery === undefined || agent === undefined) return false; diff --git a/src/mcpl/channel-registry.ts b/src/mcpl/channel-registry.ts index d667fff4e..6307678a1 100644 --- a/src/mcpl/channel-registry.ts +++ b/src/mcpl/channel-registry.ts @@ -452,6 +452,33 @@ export function journalToolDescription(canSkipReply: boolean): string { 'text is routed' + (canSkipReply ? '; to end the turn without replying, call skip_reply afterwards.' : '.'); } +/** + * channel_open's and channel_close's descriptions for an agent whose window doesn't hold a channel's traffic: an + * ephemeral run, whose store is its own unless it's where channel messages are stored (on a host with no residents, + * the first registered agent), and on a host that routes conversations every agent but a fork, since a channel's + * messages go to its fork or, unrouted, nowhere. The descriptions in CHANNEL_TOOL_DEFINITIONS, written for the agents + * that read a channel, promised such an agent what it never gets: "Messages that address you still reach you", and + * history "preceding the message that invited you" when nothing invites it. Given per agent (AgentFramework's + * getToolsForAgent), so a holder's cached tool list doesn't change. + * + * Both say the call is for every agent: the decision is recorded per server and channel (setDesiredState), so a run's + * close closes the channel for the agents whose windows hold it, which "its messages don't come to you" alone could + * read as costing nothing. They claim only what holds on both lanes: a push that names a channel is stored in the + * residents' shared slot whether or not the channel is open (handleMcplPushEvent), so on a routed host the residents + * get the channel's pushes, and a server may push while a channel is closed and send channels/incoming once it's + * open. So opening brings the caller none of the channel's messages, and closing stops none reaching it. The history + * channel_open asks for comes back in its result (handleToolOpen), so that clause is true for anyone. + */ +export const NON_HOLDER_CHANNEL_DESCRIPTIONS: Readonly> = { + channel_open: + 'Open a channel for every agent on this host. Opening it brings you none of its messages; for the agents they do ' + + 'come to, its server decides what it delivers. Optionally request some of its history, which comes back to you in ' + + 'this call.', + channel_close: + 'Close a channel for every agent on this host. Closing it stops none of its messages reaching you; for the agents ' + + 'they do come to, its server decides what closing it stops.', +}; + const CHANNEL_TOOL_DEFINITIONS: ToolDefinition[] = [ { name: 'channel_list', diff --git a/test/channel-tool-words.test.ts b/test/channel-tool-words.test.ts new file mode 100644 index 000000000..81bf50c7a --- /dev/null +++ b/test/channel-tool-words.test.ts @@ -0,0 +1,147 @@ +/** + * channel_open's and channel_close's words follow whether the agent's window holds a channel's traffic. A holder (a + * resident of an unrouted host, a conversation fork for its home, and on a host with no residents the run channel + * messages are stored with) keeps the registry's words, so its cached tool list doesn't change. A non-holder (any other ephemeral run, and every agent but a fork on a host that routes + * conversations) is promised nothing about delivery: the registry's "Messages that address you still reach you" was + * false for it. + */ +import { describe, it, beforeEach, afterEach } from 'node:test'; +import assert from 'node:assert/strict'; +import { tmpdir } from 'node:os'; +import { mkdtempSync, rmSync } from 'node:fs'; +import { join } from 'node:path'; +import { AgentFramework } from '../src/index.js'; +import type { ProcessEvent } from '../src/index.js'; +import { MockMembrane, createMockResponse } from './helpers/mock-membrane.js'; +import { ChannelRegistry, NON_HOLDER_CHANNEL_DESCRIPTIONS } 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'; + +interface Internals { + channelRegistry: unknown; + agents: Map; + conversationRouter?: { touch(channelId: string, now?: number): void; getBinding(channelId: string): unknown }; + conversationAgentHomes: Map; + lastConversationSweep: number; + sweepExpiredConversations(): void; + getToolsForAgent(agentName: string): Array<{ name: string; description?: string }>; +} + +const internals = (framework: AgentFramework) => framework as unknown as Internals; + +/** The registry's own words: what a holder is given. */ +const registryTools = new ChannelRegistry({} as McplServerRegistry, {} as FeatureSetManager, () => {}, () => {}).getChannelTools(); +const HOLDER = { + open: registryTools.find((t) => t.name === 'channel_open')!.description, + close: registryTools.find((t) => t.name === 'channel_close')!.description, +}; +const NON_HOLDER = { open: NON_HOLDER_CHANNEL_DESCRIPTIONS.channel_open, close: NON_HOLDER_CHANNEL_DESCRIPTIONS.channel_close }; + +/** Give the framework the registry's channel tools, and nothing else of a registry. */ +function withChannelTools(framework: AgentFramework): void { + const registry: Record = { getChannelTools: () => registryTools, getDescriptor: () => undefined }; + internals(framework).channelRegistry = new Proxy(registry, { + get: (target, prop: string) => (prop in target ? target[prop] : () => undefined), + }); +} + +const words = (tools: Array<{ name: string; description?: string }>) => ({ + open: tools.find((t) => t.name === 'channel_open')?.description, + close: tools.find((t) => t.name === 'channel_close')?.description, +}); + +function mention(channelId: string, text: string): ProcessEvent { + return { + type: 'mcpl:channel-incoming', + serverId: 'srv', + channelId, + messageId: `m-${Math.random().toString(36).slice(2)}`, + author: { id: 'U1', name: 'alice' }, + content: [{ type: 'text', text }], + timestamp: new Date().toISOString(), + metadata: { mentioned: true }, + tags: ['chat:addressed'], + triggerInference: true, + } as unknown as ProcessEvent; +} + +describe("channel_open and channel_close's words follow whether the agent's window holds the channel's traffic", () => { + let dir: string; + let membrane: MockMembrane; + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'channel-tool-words-')); + membrane = new MockMembrane(); + }); + afterEach(() => rmSync(dir, { recursive: true, force: true })); + + it("a non-holder's words promise no delivery, and say the call changes nothing that reaches it", () => { + // The holder's words are trunk-channel-routing's to pin (#406). + assert.doesNotMatch(NON_HOLDER.close, /still reach you/); + assert.match(NON_HOLDER.close, /^Close a channel for every agent on this host\. Closing it stops none of its messages reaching you;/); + assert.match(NON_HOLDER.open, /^Open a channel for every agent on this host\. Opening it brings you none of its messages;/); + assert.doesNotMatch(NON_HOLDER.open, /invited you|delivers to you/); + }); + + it("an unrouted host's residents keep the registry's words, and an ephemeral run is given the non-holder's", async () => { + const framework = await AgentFramework.create({ + storePath: join(dir, 's.chronicle'), + membrane: membrane.asMembrane(), + agents: [ + { name: 'main', model: 'test-model', systemPrompt: 'You are main.' }, + { name: 'helper', model: 'test-model', systemPrompt: 'You are helper.' }, + ], + modules: [], + }); + try { + withChannelTools(framework); + // Residents read the shared slot an incoming message is stored in. + assert.deepEqual(words(internals(framework).getToolsForAgent('main')), HOLDER); + assert.deepEqual(words(internals(framework).getToolsForAgent('helper')), HOLDER); + // A run's store is its own, and an incoming message goes to the residents': what its model is sent says so. + const created = await framework.createEphemeralAgent({ name: 'run-1', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all' }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Go.' }]); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'done' }])); + const run = framework.runEphemeralToCompletion(created.agent, created.contextManager); + framework.start(); + await run; + const sent = membrane.calls.at(-1) as unknown as { tools?: Array<{ name: string; description?: string }> }; + assert.deepEqual(words(sent.tools ?? []), NON_HOLDER, JSON.stringify(words(sent.tools ?? []))); + } finally { + await framework.stop(); + } + }); + + it("a routed host's trunk is given the non-holder's words, and a fork keeps the registry's for its home, through its closure turn", async () => { + const framework = await AgentFramework.create({ + storePath: join(dir, 's.chronicle'), + membrane: membrane.asMembrane(), + agents: [{ name: 'trunk', model: 'test-model', systemPrompt: 'You are the trunk.' }], + modules: [], + conversations: { templateAgent: 'trunk' }, + }); + try { + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'ok' }])); + framework.pushEvent(mention('slack:C1', 'are you there?')); + await framework.runUntilIdle(); + const fork = [...internals(framework).agents.keys()].find((name) => name !== 'trunk' && !name.startsWith('subconscious')); + assert.ok(fork, 'the mention made a fork'); + withChannelTools(framework); + assert.deepEqual(words(internals(framework).getToolsForAgent('trunk')), NON_HOLDER); + assert.deepEqual(words(internals(framework).getToolsForAgent(fork)), HOLDER); + // The idle sweep unbinds a fork before its closure turn. Its words don't change for that one turn, which would + // send its whole window uncached: the closure turn's request carries the registry's. + internals(framework).conversationRouter!.touch('slack:C1', 0); + internals(framework).lastConversationSweep = 0; + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'bye' }])); + const before = membrane.calls.length; + internals(framework).sweepExpiredConversations(); + await framework.runUntilIdle(); + assert.equal(internals(framework).conversationRouter!.getBinding('slack:C1'), undefined, 'the sweep unbound the home'); + const sent = membrane.calls.slice(before) as unknown as Array<{ tools?: Array<{ name: string; description?: string }> }>; + assert.equal(sent.length, 1, 'the closure turn was the one request'); + assert.deepEqual(words(sent[0]!.tools ?? []), HOLDER); + } finally { + await framework.stop(); + } + }); +}); From 0743c9380c06f1335c4e14cc2b458c3ebbf9e618 Mon Sep 17 00:00:00 2001 From: Greta-2425 Date: Wed, 21 Oct 2026 18:27:12 -0400 Subject: [PATCH 2/2] test(channels): on a routed host, a resident beside the trunk and an ephemeral run get the non-holder's words The routed-host spec had only the trunk, so a rule that named the trunk alone, or exempted residents other than it, would have passed. A routed host stores a channel's incoming messages only in the fork its channel is bound to, so every agent but a fork is a non-holder: the new spec checks a second resident's words and an ephemeral run's request there. Both run specs now pick the run's request by its system prompt rather than taking the last request sent. Raised by the automated review of #678. Co-Authored-By: Claude Opus 5.5 --- test/channel-tool-words.test.ts | 40 ++++++++++++++++++++++++++++++++- 1 file changed, 39 insertions(+), 1 deletion(-) diff --git a/test/channel-tool-words.test.ts b/test/channel-tool-words.test.ts index 81bf50c7a..3d9d9efbe 100644 --- a/test/channel-tool-words.test.ts +++ b/test/channel-tool-words.test.ts @@ -50,6 +50,14 @@ const words = (tools: Array<{ name: string; description?: string }>) => ({ close: tools.find((t) => t.name === 'channel_close')?.description, }); +/** The one request the run sent, told apart from any resident's by the run's system prompt. */ +function runRequest(membrane: MockMembrane): { tools?: Array<{ name: string; description?: string }> } { + const sent = (membrane.calls as unknown as Array<{ system?: unknown; tools?: Array<{ name: string; description?: string }> }>) + .filter((c) => JSON.stringify(c.system ?? '').includes('Do the task.')); + assert.equal(sent.length, 1, 'the run sent one request'); + return sent[0]!; +} + function mention(channelId: string, text: string): ProcessEvent { return { type: 'mcpl:channel-incoming', @@ -104,7 +112,7 @@ describe("channel_open and channel_close's words follow whether the agent's wind const run = framework.runEphemeralToCompletion(created.agent, created.contextManager); framework.start(); await run; - const sent = membrane.calls.at(-1) as unknown as { tools?: Array<{ name: string; description?: string }> }; + const sent = runRequest(membrane); assert.deepEqual(words(sent.tools ?? []), NON_HOLDER, JSON.stringify(words(sent.tools ?? []))); } finally { await framework.stop(); @@ -144,4 +152,34 @@ describe("channel_open and channel_close's words follow whether the agent's wind await framework.stop(); } }); + + it("on a routed host, every agent but a fork is given the non-holder's words: a resident beside the trunk, and an ephemeral run", async () => { + // A routed host stores a channel's incoming messages only in the fork its channel is bound to, never in a + // resident's or a run's window, so no rule naming the trunk alone would do. + const framework = await AgentFramework.create({ + storePath: join(dir, 's.chronicle'), + membrane: membrane.asMembrane(), + agents: [ + { name: 'trunk', model: 'test-model', systemPrompt: 'You are the trunk.' }, + { name: 'helper', model: 'test-model', systemPrompt: 'You are helper.' }, + ], + modules: [], + conversations: { templateAgent: 'trunk' }, + }); + try { + withChannelTools(framework); + assert.deepEqual(words(internals(framework).getToolsForAgent('trunk')), NON_HOLDER); + assert.deepEqual(words(internals(framework).getToolsForAgent('helper')), NON_HOLDER); + const created = await framework.createEphemeralAgent({ name: 'run-1', model: 'test-model', systemPrompt: 'Do the task.', allowedTools: 'all' }); + created.contextManager.addMessage('user', [{ type: 'text', text: 'Go.' }]); + membrane.pushResponse(createMockResponse([{ type: 'text', text: 'done' }])); + const run = framework.runEphemeralToCompletion(created.agent, created.contextManager); + framework.start(); + await run; + const sent = runRequest(membrane); + assert.deepEqual(words(sent.tools ?? []), NON_HOLDER, JSON.stringify(words(sent.tools ?? []))); + } finally { + await framework.stop(); + } + }); });