channel_publish posts into a closed channel without opening it, and before a pending open is answered #729

// Probe (Barnaby-2576), F261: channel_publish's path (handleToolPublish)
// beside plain speech's (ChannelRegistry.publish), on one registry.
// A: a channel registered closed (channels/changed, initiallyOpen false).
// B: a channel whose channel_open is still waiting for the server's answer,
//    which then fails.
// For each, the probe prints the server calls in order and each path's answer.
// Run with tsx from the agent-framework checkout's root.
import { ChannelRegistry } from '../src/mcpl/channel-registry.js';
import { CapabilityGrant, ALL_CAPABILITY_PATHS } from '../src/mcpl/capability-grant.js';
import type { McplServerRegistry } from '../src/mcpl/server-registry.js';
import type { FeatureSetManager } from '../src/mcpl/feature-set-manager.js';
import type { ChannelDescriptor, ChannelsOpenResult } from '../src/mcpl/types.js';

console.log(`node ${process.version}, ${process.platform}`);
const tick = () => new Promise<void>((resolve) => setImmediate(resolve));
const sleep = (ms: number) => new Promise<void>((resolve) => setTimeout(resolve, ms));

function descriptor(id: string, initiallyOpen = false): ChannelDescriptor {
  return { id, type: 'test', label: id, direction: 'bidirectional', initiallyOpen, address: { value: id } };
}

function fixture(openAnswer: (id: string) => Promise<ChannelsOpenResult>) {
  const calls: string[] = [];
  const server = {
    grant: new CapabilityGrant(new Set(ALL_CAPABILITY_PATHS), []),
    sendChannelsOpen: async (params: { channelId?: string }) => {
      calls.push(`channels/open ${params.channelId}`);
      try {
        const answer = await openAnswer(params.channelId!);
        calls.push(`channels/open ${params.channelId} answered`);
        return answer;
      } catch (err) {
        calls.push(`channels/open ${params.channelId} failed: ${(err as Error).message}`);
        throw err;
      }
    },
    sendChannelsClose: async (params: { channelId: string }) => { calls.push(`channels/close ${params.channelId}`); return { closed: true }; },
    sendChannelsPublish: async (params: { channelId: string }) => { calls.push(`channels/publish ${params.channelId}`); return { delivered: true, messageId: 'm1' }; },
  };
  const registry = new ChannelRegistry(
    { getServer: () => server } as unknown as McplServerRegistry,
    {} as FeatureSetManager, () => {}, () => {},
  );
  return { registry, calls };
}

const quiet = async <T>(work: () => Promise<T>): Promise<T> => {
  const original = console.error;
  console.error = () => {};
  try { return await work(); } finally { console.error = original; }
};

// A: a closed channel.
for (const path of ['channel_publish', 'publish()'] as const) {
  const f = fixture(async (id) => ({ channel: descriptor(id, true) }));
  await f.registry.handleChanged('test', { added: [descriptor('z')] });
  f.calls.length = 0;
  const answer = await quiet(() => path === 'channel_publish'
    ? f.registry.handleChannelToolCall('channel_publish', { channelId: 'z', content: 'hello' }).then((r) => ({ success: r.success, error: r.error }))
    : f.registry.publish('conv', 'hello', { serverId: 'test', channelId: 'z' }).then((o) => ({ status: o.status, reason: (o as { reason?: string }).reason })));
  await tick();
  console.log(`\n== A, closed channel z, via ${path}`);
  console.log(`decision for z: ${JSON.stringify(f.registry.getDesiredState('test', 'z'))}; z open after: ${f.registry.isChannelOpen({ serverId: 'test', channelId: 'z' })}`);
  console.log(`server calls: ${JSON.stringify(f.calls)}`);
  console.log(`answer: ${JSON.stringify(answer)}`);
}

// B: a channel whose open is in flight and then fails.
for (const path of ['channel_publish', 'publish()'] as const) {
  let fail!: (err: Error) => void;
  const f = fixture(() => new Promise<ChannelsOpenResult>((_resolve, reject) => { fail = reject; }));
  await f.registry.handleChanged('test', { added: [descriptor('w')] });
  f.calls.length = 0;
  const opening = quiet(() => f.registry.handleChannelToolCall('channel_open', { serverId: 'test', channelId: 'w' }));
  await sleep(10);
  const sending = quiet(() => path === 'channel_publish'
    ? f.registry.handleChannelToolCall('channel_publish', { channelId: 'w', content: 'hello' }).then((r) => ({ success: r.success, error: r.error }))
    : f.registry.publish('conv', 'hello', { serverId: 'test', channelId: 'w' }).then((o) => ({ status: o.status, reason: (o as { reason?: string }).reason })));
  await sleep(10);
  const callsBeforeAnswer = [...f.calls];
  fail(new Error('server refused the open'));
  const [opened, sent] = await Promise.all([opening, sending]);
  await tick();
  console.log(`\n== B, channel w with channel_open in flight (then failing), via ${path}`);
  console.log(`server calls before the open was answered: ${JSON.stringify(callsBeforeAnswer)}`);
  console.log(`server calls in all: ${JSON.stringify(f.calls)}`);
  console.log(`channel_open answer: success=${opened.success}`);
  console.log(`send answer: ${JSON.stringify(sent)}`);
  console.log(`w open after: ${f.registry.isChannelOpen({ serverId: 'test', channelId: 'w' })}`);
}