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 941e7a552..297163104 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, TUNE_OUT_REPLACED, connectorText, desiredStateInLog, destinationKey, journalToolDescription, type ChannelToolOrigin, type PublishDestination, type PublishOutcome, type SelectedForOpening } from './mcpl/channel-registry.js'; +import { ChannelRegistry, TUNE_OUT_REPLACED, 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'; @@ -6127,6 +6127,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; }); } @@ -15630,6 +15634,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 07844cd21..ccb4e6c49 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..3d9d9efbe --- /dev/null +++ b/test/channel-tool-words.test.ts @@ -0,0 +1,185 @@ +/** + * 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, +}); + +/** 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', + 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 = runRequest(membrane); + 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(); + } + }); + + 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(); + } + }); +});