Title-generation thread is never unsubscribed, so its MCP server processes outlive session/close #628

// Repro: codex-acp's title-generation thread is never unsubscribed, so the MCP server processes codex starts
// for it outlive `session/close` and run until codex-acp exits.
//
// Usage:  node title-thread-repro.mts /path/to/node_modules/@agentclientprotocol/codex-acp/dist/index.js
// Needs:  Node >= 22.18 (runs TypeScript directly), macOS or Linux, a logged-in codex.
// Cost:   two one-word model prompts, plus the title turn the adapter itself starts for session A.
//
// It is self-contained and edits no config. It points CODEX_PATH at a generated wrapper that starts the
// bundled codex with `-c mcp_servers.title_repro_stub...`, so every codex thread in this one codex-acp process
// starts one stub MCP server (this same file, run with `--stub`). Counting those stubs counts loaded threads.
//
//   Session A (control): first prompt is an ordinary prompt.
//   Session B:           first prompt is `/rename ...`, then the same ordinary prompt.
//
// Expected if nothing leaked: 0 stub processes once A and B are closed (codex unloads a thread about 60 s
// after its last subscriber leaves). Observed: 1, started during A's first turn, belonging to no ACP session.
import { spawn, execSync } from "node:child_process";
import { chmodSync, mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs";
import { createRequire } from "node:module";
import { tmpdir } from "node:os";
import path from "node:path";
import readline from "node:readline";
import { fileURLToPath } from "node:url";

const self = fileURLToPath(import.meta.url);

if (process.argv[2] === "--stub") {
  runStub();
} else {
  await main();
}

/** Minimal stdio MCP server: answers initialize / tools/list / ping, then idles until stdin closes. */
function runStub(): void {
  const rl = readline.createInterface({ input: process.stdin });
  const send = (m: unknown) => process.stdout.write(JSON.stringify(m) + "\n");
  rl.on("line", (line) => {
    let m: any;
    try { m = JSON.parse(line); } catch { return; }
    if (m.id === undefined) return; // notification
    if (m.method === "initialize") {
      send({ jsonrpc: "2.0", id: m.id, result: {
        protocolVersion: m.params?.protocolVersion ?? "2025-06-18",
        capabilities: { tools: {} },
        serverInfo: { name: "title-repro-stub", version: "0.0.1" },
      } });
    } else if (m.method === "tools/list") {
      send({ jsonrpc: "2.0", id: m.id, result: { tools: [] } });
    } else if (m.method === "ping") {
      send({ jsonrpc: "2.0", id: m.id, result: {} });
    } else {
      send({ jsonrpc: "2.0", id: m.id, error: { code: -32601, message: "not implemented" } });
    }
  });
  rl.on("close", () => process.exit(0));
}

async function main(): Promise<void> {
  const acp = process.argv[2];
  if (!acp) {
    console.error("usage: node title-thread-repro.mts /path/to/@agentclientprotocol/codex-acp/dist/index.js");
    process.exit(2);
  }
  const sleep = (ms: number) => new Promise<void>((r) => setTimeout(r, ms));
  const log = (s: string) => console.log(`${new Date().toISOString().slice(11, 19)}Z ${s}`);

  // A wrapper codex that adds one config-level stdio MCP server for the whole app-server process.
  const marker = `title-repro-stub-${process.pid}`;
  const work = mkdtempSync(path.join(tmpdir(), "codex-acp-title-repro-"));
  const cwd = path.join(work, "project");
  mkdirSync(cwd);
  const codexJs = createRequire(path.resolve(acp)).resolve("@openai/codex/bin/codex.js");
  const sh = (s: string) => `'${s.replace(/'/g, `'\\''`)}'`;
  const wrapper = path.join(work, "codex-with-stub-mcp.sh");
  writeFileSync(wrapper, [
    "#!/bin/sh",
    `exec ${sh(process.execPath)} ${sh(codexJs)} "$@" \\`,
    `  -c ${sh(`mcp_servers.title_repro_stub.command=${JSON.stringify(process.execPath)}`)} \\`,
    `  -c ${sh(`mcp_servers.title_repro_stub.args=${JSON.stringify([self, "--stub", marker])}`)}`,
    "",
  ].join("\n"));
  chmodSync(wrapper, 0o755);

  // Stub processes only: the app-server's own command line carries the marker inside a quoted `-c` value.
  const stubs = (): string[] => {
    try {
      return execSync(`pgrep -f -- "--stub ${marker}"`, { encoding: "utf8" }).split("\n").filter(Boolean);
    } catch { return []; }
  };
  const count = (label: string): string[] => {
    const pids = stubs();
    log(`${label}: ${pids.length} stub MCP server process(es) [${pids.join(", ")}]`);
    return pids;
  };
  const waitForCount = async (n: number, timeoutMs: number): Promise<boolean> => {
    for (let waited = 0; waited < timeoutMs; waited += 500) {
      if (stubs().length >= n) return true;
      await sleep(500);
    }
    return false;
  };

  const child = spawn(process.execPath, [acp], {
    stdio: ["pipe", "pipe", "ignore"],
    detached: true,
    env: { ...process.env, CODEX_PATH: wrapper },
  });
  let buf = "";
  let nextId = 1;
  const pending = new Map<number, (m: any) => void>();
  child.stdout.on("data", (d: Buffer) => {
    buf += d;
    let i: number;
    while ((i = buf.indexOf("\n")) >= 0) {
      const line = buf.slice(0, i);
      buf = buf.slice(i + 1);
      let m: any;
      try { m = JSON.parse(line); } catch { continue; }
      if (m.id !== undefined && !m.method && pending.has(m.id)) {
        pending.get(m.id)!(m);
        pending.delete(m.id);
      } else if (m.method && m.id !== undefined) {
        // Any request from the agent (for example a permission prompt) is refused.
        child.stdin.write(JSON.stringify({ jsonrpc: "2.0", id: m.id, error: { code: -32601, message: "unsupported" } }) + "\n");
      } else if (m.method === "session/update" && m.params?.update?.sessionUpdate === "session_info_update") {
        const title = m.params.update.title;
        if (title !== undefined) log(`  session_info_update ${String(m.params.sessionId).slice(0, 13)} title=${JSON.stringify(title)}`);
      }
    }
  });
  const req = (method: string, params: unknown, timeoutMs = 60_000): Promise<any> =>
    new Promise((resolve, reject) => {
      const id = nextId++;
      pending.set(id, resolve);
      child.stdin.write(JSON.stringify({ jsonrpc: "2.0", id, method, params }) + "\n");
      setTimeout(() => { if (pending.delete(id)) reject(new Error(`timeout: ${method}`)); }, timeoutMs);
    });
  const prompt = async (name: string, sessionId: string, text: string): Promise<any> => {
    const r = await req("session/prompt", { sessionId, prompt: [{ type: "text", text }] }, 120_000);
    if (r.error) throw new Error(`prompt ${name}: ${JSON.stringify(r.error)}`);
    log(`prompt ${name} ${JSON.stringify(text)} -> stopReason=${r.result?.stopReason} usage=${r.result?.usage ? "reported" : "null"}`);
    return r.result;
  };
  const PROMPT = "Reply with the single word OK. Do not use tools.";

  let verdict = "INCONCLUSIVE";
  try {
    const init = await req("initialize", { protocolVersion: 1, clientCapabilities: {} });
    log(`agent ${JSON.stringify(init.result?.agentInfo ?? {})}`);

    const a = await req("session/new", { cwd, mcpServers: [] });
    if (a.error) throw new Error(`session/new A: ${JSON.stringify(a.error)}`);
    const A: string = a.result.sessionId;
    log(`session A (control) = ${A}`);
    if (!(await waitForCount(1, 30_000))) {
      throw new Error("setup: the stub MCP server did not start for session A; nothing was sent to the model");
    }
    const afterAOpen = count("A open, no prompt yet");

    await prompt("A", A, PROMPT);
    await sleep(15_000);
    const afterATurn = count("A, 15 s after its first turn");
    const extra = afterATurn.filter((p) => !afterAOpen.includes(p));

    const b = await req("session/new", { cwd, mcpServers: [] });
    if (b.error) throw new Error(`session/new B: ${JSON.stringify(b.error)}`);
    const B: string = b.result.sessionId;
    log(`session B (/rename first) = ${B}`);
    await waitForCount(afterATurn.length + 1, 30_000);
    const afterBOpen = count("B open, no prompt yet");

    const renamed = await prompt("B", B, "/rename title-repro-b");
    if (renamed?.usage) log("  note: /rename reported model usage, so this adapter did not handle it locally");
    await prompt("B", B, PROMPT);
    await sleep(15_000);
    const afterBTurn = count("B, 15 s after /rename and its first turn");

    const ca = await req("session/close", { sessionId: A });
    const cb = await req("session/close", { sessionId: B });
    log(`session/close A -> ${JSON.stringify(ca.error ?? ca.result)}; B -> ${JSON.stringify(cb.error ?? cb.result)}`);
    let left: string[] = [];
    for (let t = 10; t <= 90; t += 10) {
      await sleep(10_000);
      left = count(`${t} s after closing A and B`);
    }

    log("");
    log(`A's first turn added ${extra.length} thread(s) beyond the session's own [${extra.join(", ")}]`);
    log(`B's /rename + first turn added ${afterBTurn.length - afterBOpen.length}`);
    if (left.length > 0) {
      const fromTitle = left.every((p) => extra.includes(p));
      verdict = `REPRODUCED: ${left.length} stub MCP server process(es) still running 90 s after every session was closed`
        + ` (expected 0)${fromTitle ? "; each was started during A's first turn" : ""}`;
    } else {
      verdict = "NOT REPRODUCED: no stub MCP server process is left after closing every session";
    }
  } catch (e) {
    verdict = `INCONCLUSIVE: ${(e as Error).message}`;
  } finally {
    log(verdict);
    try { process.kill(-child.pid!, "SIGTERM"); } catch { /* already gone */ }
    await sleep(4000);
    log(`after terminating codex-acp: ${stubs().length} stub MCP server process(es)`);
    rmSync(work, { recursive: true, force: true });
    process.exit(verdict.startsWith("REPRODUCED") ? 1 : 0);
  }
}