The channel invitation, channel_open, channel_close and the "Now open" notice describe discord-mcpl's delivery, which slack-mcpl doesn't share #406

/**
 * What an agent on agent-framework is told about opening and closing a Slack
 * channel, against what slack-mcpl delivers. End to end, in one process:
 * agent-framework's host (its built dist, MCPL over WebSocket) connected to
 * slack-mcpl's server running on slack-mcpl's own test fake of Slack.
 *
 * Run from a slack-mcpl checkout with its dependencies installed, with
 * agent-framework built:
 *   AF_ROOT=/path/to/agent-framework SLACK_ROOT=$PWD [SCENARIO=policy] \
 *     node --import tsx /path/to/probe.mts
 * SCENARIO: lifecycle (default) or policy. It prints its report as JSON, and
 * also writes it to $OUT_DIR when that is set.
 */

import { PassThrough } from 'node:stream';
import { createRequire } from 'node:module';
import { mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from 'node:fs';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';

const AF = process.env.AF_ROOT!;
/** lifecycle (default): open, close and slack-mcpl's subscription; policy: a
 *  channel the host's subscription policy opens. */
const SCENARIO = process.env.SCENARIO ?? 'lifecycle';
const SLACK = process.env.SLACK_ROOT!;

const { AgentFramework } = await import(pathToFileURL(`${AF}/dist/src/index.js`).href);
const { MockMembrane, MockYieldingStream, createMockResponse } = await import(pathToFileURL(`${AF}/dist/test/helpers/mock-membrane.js`).href);
const { WebSocketServer } = createRequire(`${AF}/package.json`)('ws');
const harness = await import(pathToFileURL(`${SLACK}/test/harness.ts`).href);
const { SlackMcplServer } = await import(pathToFileURL(`${SLACK}/src/server.ts`).href);
const corePkgDir = `${SLACK}/node_modules/@animalabs/mcpl-core`;
const corePkg = JSON.parse(readFileSync(`${corePkgDir}/package.json`, 'utf8'));
const coreEntry = (corePkg.exports?.['.']?.import ?? corePkg.exports?.['.']?.default ?? corePkg.module ?? corePkg.main) as string;
const { McplConnection } = await import(pathToFileURL(join(corePkgDir, coreEntry)).href);

const tempDir = mkdtempSync(join(tmpdir(), 'slack-model-probe-'));
// slack-mcpl keeps its subscriptions in a file; give it a private one.
process.env.SLACK_SUBSCRIPTIONS_FILE = join(tempDir, 'subscriptions.json');

// ---- slack-mcpl over a fake Slack, served to each WebSocket client ----
const { slack, socket, posts } = harness.fakeSlack({});
const server = new SlackMcplServer(slack, {});
const wire: Array<{ method: string; channel?: string; tags?: string[] }> = [];
const wss = new WebSocketServer({ host: '127.0.0.1', port: 0 });
await new Promise<void>((resolve) => wss.once('listening', () => resolve()));
const port = (wss.address() as { port: number }).port;
wss.on('connection', (ws: any) => {
  const toServer = new PassThrough();
  const toClient = new PassThrough();
  ws.on('message', (data: Buffer) => {
    const text = data.toString('utf8');
    toServer.write(text.endsWith('\n') ? text : text + '\n');
  });
  ws.on('close', () => toServer.end());
  let pending = '';
  toClient.on('data', (chunk: Buffer) => {
    pending += chunk.toString('utf8');
    let nl: number;
    while ((nl = pending.indexOf('\n')) >= 0) {
      const line = pending.slice(0, nl);
      pending = pending.slice(nl + 1);
      if (!line.trim()) continue;
      try {
        const msg = JSON.parse(line);
        if (msg.method === 'push/event' || msg.method === 'channels/incoming') {
          const p = msg.params ?? {};
          wire.push({
            method: msg.method,
            channel: p.origin?.mcplChannelId ?? p.messages?.[0]?.channelId,
            tags: p.tags ?? p.messages?.[0]?.tags,
          });
        }
      } catch { /* not JSON: forwarded as is */ }
      ws.send(line);
    }
  });
  void server.serve(McplConnection.fromStreams(toServer, toClient));
});

// ---- agent-framework, connected to it ----
// Any turn the host runs says nothing, so no speech moves the channel's state.
const membrane = new MockMembrane();
let turns = 0;
membrane.streamYielding = (request: unknown) => {
  turns++;
  membrane.calls.push(request);
  const stream = new MockYieldingStream([createMockResponse([], 'end_turn')]);
  membrane.lastStream = stream;
  return stream;
};
membrane.complete = async (request: unknown) => {
  membrane.calls.push(request);
  return createMockResponse([], 'end_turn');
};
const framework = await AgentFramework.create({
  storePath: join(tempDir, 'store'),
  membrane: membrane.asMembrane(),
  agents: [{ name: 'resident', model: 'test-model', systemPrompt: 'You are a resident.' }],
  modules: [],
  mcplServers: [{
    id: 'slack',
    url: `ws://127.0.0.1:${port}`,
    ...(SCENARIO === 'policy' ? { channelSubscription: 'auto' as const } : {}),
  }],
});
framework.start();

const cm = () => framework.getAgent('resident')!.getContextManager();
const textOf = (m: { content: Array<{ type: string; text?: string }> }) =>
  m.content.filter((b) => b.type === 'text').map((b) => b.text).join('\n');
const userTexts = () => (cm().queryMessages({}).messages as Array<{ participant: string; content: Array<{ type: string; text?: string }> }>)
  .filter((m) => m.participant !== 'resident' && m.participant !== 'assistant').map(textOf);
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
async function until(pred: () => boolean, what: string, ms = 10_000) {
  const end = Date.now() + ms;
  while (!pred()) {
    if (Date.now() > end) throw new Error(`timed out waiting for ${what}`);
    await sleep(25);
  }
}
const channelState = async () => {
  const r = await framework.executeToolCall({ id: `list-${Date.now()}`, name: 'channel_list', input: {}, callerAgentName: 'resident' });
  return (r.data as Array<{ id: string; open: boolean; desired: unknown }>).filter((c) => c.id.startsWith('slack:C'));
};

await until(() => {
  const reg = (framework as unknown as { channelRegistry?: { getDescriptor(id: string): unknown } }).channelRegistry;
  return !!reg?.getDescriptor('slack:C1');
}, 'slack:C1 registered');

const steps: Array<Record<string, unknown>> = [];
async function step(name: string, action: () => Promise<unknown>, fromStart = false) {
  const wireBefore = fromStart ? 0 : wire.length;
  const ctxBefore = fromStart ? 0 : userTexts().length;
  const result = await action();
  await sleep(600);
  steps.push({
    step: name,
    result: result === undefined ? undefined : result,
    slackSent: wire.slice(wireBefore),
    agentReceived: userTexts().slice(ctxBefore),
    channel: await channelState(),
  });
}
const tool = (name: string, input: Record<string, unknown>) =>
  framework.executeToolCall({ id: `${name}-${Date.now()}`, name, input, callerAgentName: 'resident' });
const toolNames: string[] = framework.getAllTools().map((t: { name: string }) => t.name);
const unsubscribe = toolNames.find((n) => n.endsWith('unsubscribe_channel'))!;
const subscribe = toolNames.find((n) => n.endsWith('--subscribe_channel'))!;

let ts = 100;
const say = (text: string) => socket.emitMessage({ type: 'message', channel: 'C1', user: 'U2', text, ts: `${++ts}.1` });

if (SCENARIO === 'policy') {
  // The policy's first pass migrates silently; a conversation Slack lists
  // later is admitted by the policy with the "[channels] Now open" notice.
  await step('at start', async () => undefined, true);
  const web = (slack as unknown as {
    web: { conversations: { list: () => Promise<unknown>; info: (a: { channel: string }) => Promise<unknown> } };
  }).web;
  const names: Record<string, string> = { C1: 'general', C2: 'random' };
  web.conversations.list = async () => ({
    channels: Object.entries(names).map(([id, name]) => ({ id, name, is_member: true, is_im: false, is_mpim: false })),
  });
  // The harness's fake names every conversation "general"; name #random as Slack would.
  web.conversations.info = async ({ channel }) => ({
    channel: { id: channel, name: names[channel] ?? channel, is_member: true, is_im: false, is_mpim: false },
  });
  const refresh = toolNames.find((n) => n.endsWith('--refresh_channels'))!;
  await step(`1. #random appears in Slack, and the agent calls slack-mcpl's ${refresh}`, async () => {
    const r = await tool(refresh, {});
    await until(() => userTexts().some((t) => t.includes('Now open')), 'the policy notice');
    return r;
  });
  const sayIn = (channel: string, text: string) =>
    socket.emitMessage({ type: 'message', channel, user: 'U2', text, ts: `${++ts}.1` });
  await step('2. an ordinary message in #random, which the policy opened', () => sayIn('C2', 'ordinary chatter in random'));
  await step('3. a mention in #random', () => sayIn('C2', '<@UBOT> hello from random'));
  await step('4. an ordinary message in #random after the mention', () => sayIn('C2', 'more chatter in random'));
} else {
await step('at start', async () => undefined);
await step('1. a mention in #general, which the host has closed', () => say('<@UBOT> can you look at this?'));
await step('2. an ordinary message there, the channel still closed in the host', () => say('ordinary chatter, not addressed'));
await step('3. the agent calls channel_close', () => tool('channel_close', { channelId: 'slack:C1' }));
await step('4. another ordinary message after channel_close', () => say('more chatter after the close'));
await step(`5. the agent calls slack-mcpl's ${unsubscribe}`, () => tool(unsubscribe, { channelId: 'C1' }));
await step('6. the agent calls channel_open', () => tool('channel_open', { channelId: 'slack:C1' }));
await step('7. an ordinary message while the host has it open, unsubscribed in slack-mcpl', () => say('chatter while open but unsubscribed'));
await step('8. a mention while the host has it open, unsubscribed', () => say('<@UBOT> are you there?'));
await step(`9. the agent calls slack-mcpl's ${subscribe}, then channel_close`, async () => {
  const s = await tool(subscribe, { channelId: 'C1' });
  const c = await tool('channel_close', { channelId: 'slack:C1' });
  return { subscribe: s, close: c };
});
await step('10. an ordinary message, closed in the host and subscribed in slack-mcpl', () => say('chatter while closed but subscribed'));
await step(`11. the agent calls slack-mcpl's ${unsubscribe} again (closed in the host)`, () => tool(unsubscribe, { channelId: 'C1' }));
await step('12. the agent posts in #general with channel_publish', () => tool('channel_publish', { channelId: 'slack:C1', content: 'Here is what I found.' }));
await step("13. a reply in the thread of the agent's post (Slack's ts 100.1), without a mention", () =>
  socket.emitMessage({ type: 'message', channel: 'C1', user: 'U2', text: 'thanks, one more question', ts: '130.1', thread_ts: '100.1' }));
}

const toolText = (name: string) => framework.getAllTools().find((t: { name: string }) => t.name === name)?.description;
const report = {
  scenario: SCENARIO,
  versions: { node: process.version },
  turnsRun: turns,
  tools: { channel_open: toolText('channel_open'), channel_close: toolText('channel_close'), unsubscribe, subscribe },
  slackPosts: posts,
  steps,
};
const out = JSON.stringify(report, null, 2);
console.log(out);
const dir = process.env.OUT_DIR;
if (dir) {
  mkdirSync(dir, { recursive: true });
  writeFileSync(join(dir, `slack-model-probe-${SCENARIO}.json`), out + '\n');
}
await framework.stop();
wss.close();
rmSync(tempDir, { recursive: true, force: true });
process.exit(0);