fix(channels): channel_open and channel_close promise delivery only to an agent whose window holds the channel's traffic

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.

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;

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<Record<'channel_open' | 'channel_close', string>> = {
+  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',

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<string, unknown>;
+  conversationRouter?: { touch(channelId: string, now?: number): void; getBinding(channelId: string): unknown };
+  conversationAgentHomes: Map<string, string>;
+  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<string, unknown> = { 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();
+    }
+  });
+});