diff --git a/external_plugins/telegram/.claude-plugin/plugin.json b/external_plugins/telegram/.claude-plugin/plugin.json index 41b42894..8dc878c6 100644 --- a/external_plugins/telegram/.claude-plugin/plugin.json +++ b/external_plugins/telegram/.claude-plugin/plugin.json @@ -1,7 +1,7 @@ { "name": "telegram", "description": "Telegram channel for Claude Code \u2014 messaging bridge with built-in access control. Manage pairing, allowlists, and policy via /telegram:access.", - "version": "0.0.7", + "version": "0.0.8", "keywords": [ "telegram", "messaging", diff --git a/external_plugins/telegram/server.ts b/external_plugins/telegram/server.ts index 6bc0ebc6..82a7a97a 100644 --- a/external_plugins/telegram/server.ts +++ b/external_plugins/telegram/server.ts @@ -55,27 +55,7 @@ if (!TOKEN) { const INBOX_DIR = join(STATE_DIR, 'inbox') const PID_FILE = join(STATE_DIR, 'bot.pid') -// Telegram allows exactly one getUpdates consumer per token. If a previous -// session crashed (SIGKILL, terminal closed) its server.ts grandchild can -// survive as an orphan and hold the slot forever, so every new session sees -// 409 Conflict. Kill any stale holder before we start polling. mkdirSync(STATE_DIR, { recursive: true, mode: 0o700 }) -try { - const stale = parseInt(readFileSync(PID_FILE, 'utf8'), 10) - if (stale > 1 && stale !== process.pid) { - process.kill(stale, 0) - // PID files race with OS PID recycling — verify the holder is actually a - // server.ts process before SIGTERM. Otherwise a recycled PID can point at - // our own bun-run wrapper (kills our stdin → immediate self-shutdown) or - // an unrelated user process. - const cmd = execFileSync('ps', ['-p', String(stale), '-o', 'args='], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'ignore'] }) - if (cmd.includes('server.ts')) { - process.stderr.write(`telegram channel: replacing stale poller pid=${stale}\n`) - process.kill(stale, 'SIGTERM') - } - } -} catch {} -writeFileSync(PID_FILE, String(process.pid)) // Last-resort safety net — without these the process dies silently on any // unhandled promise rejection. With them it logs and keeps serving tools. @@ -998,12 +978,86 @@ bot.catch(err => { process.stderr.write(`telegram channel: handler error (polling continues): ${err.error}\n`) }) +// Telegram allows exactly one getUpdates consumer per token, so exactly one +// server.ts polls at a time; bot.pid records the current holder. A live +// healthy holder is an incumbent serving another Claude Code session — never +// kill it (gh-81571: an earlier startup guard SIGTERMed any live holder, so +// starting a second session stole the channel from the first and, when the +// second session exited, no poller remained at all and the bot went silent). +// Instead: +// - slot free (no pid file, holder dead, or pid recycled to some other +// program) → claim it and poll +// - live holder → standby: outbound tools stay fully usable, and a watcher +// claims the slot the moment the holder goes away (session exit, crash — +// the orphan watchdog above reaps pollers whose CLI died) +// so the last session standing always ends up holding the channel, and the +// pid file is removed by its owner in shutdown(). +function livePollerPid(): number | null { + let holder: number + try { + holder = parseInt(readFileSync(PID_FILE, 'utf8'), 10) + } catch { return null } // no pid file — slot is free + if (!(holder > 1) || holder === process.pid) return null + try { + process.kill(holder, 0) // throws ESRCH once the process is gone + } catch (err) { + // EPERM = alive but owned by another user — never fight over the slot. + return (err as NodeJS.ErrnoException).code === 'EPERM' ? holder : null + } + // PID liveness alone can't tell an incumbent poller from an unrelated + // process that recycled its pid — check the process identity too. + // /proc//cmdline (Linux) needs no subprocess; ps covers macOS. + try { + const cmdline = readFileSync(`/proc/${holder}/cmdline`, 'utf8') + return cmdline.includes('server.ts') ? holder : null + } catch {} + try { + const args = execFileSync('ps', ['-p', String(holder), '-o', 'args='], { encoding: 'utf8', stdio: ['ignore', 'pipe', 'ignore'] }) + return args.includes('server.ts') ? holder : null + } catch { + // Identity unverifiable (Windows has no ps). Treat the slot as free + // rather than deferring forever to an unknown pid; if it IS a live + // poller, the 409 retry loop in startPolling reports the conflict + // instead of us killing anything. + return null + } +} + +let polling = false +function tryBecomePoller(): void { + if (polling || shuttingDown) return + const holder = livePollerPid() + if (holder !== null) return // healthy incumbent — leave it alone + writeFileSync(PID_FILE, String(process.pid)) + // Two standbys can race to claim; last writer owns the file, everyone else + // re-reads, sees a different pid, and stays in standby. + try { + if (parseInt(readFileSync(PID_FILE, 'utf8'), 10) !== process.pid) return + } catch { return } + polling = true + startPolling() +} + +tryBecomePoller() +if (!polling) { + process.stderr.write( + `telegram channel: another session's poller holds this channel — ` + + `outbound tools active, standing by to take over inbound when it exits\n`, + ) + const standbyWatcher = setInterval(() => { + tryBecomePoller() + if (polling || shuttingDown) clearInterval(standbyWatcher) + }, 2000) + standbyWatcher.unref() +} + // Retry polling with backoff on any error. Previously only 409 was retried — // a single ETIMEDOUT/ECONNRESET/DNS failure rejected bot.start(), the catch // returned, and polling stopped permanently while the process stayed alive // (MCP stdin keeps it running). Outbound tools kept working but the bot was // deaf to inbound messages until a full restart. -void (async () => { +function startPolling(): void { + void (async () => { for (let attempt = 1; ; attempt++) { try { await bot.start({ @@ -1042,4 +1096,5 @@ void (async () => { await new Promise(r => setTimeout(r, delay)) } } -})() + })() +} diff --git a/external_plugins/telegram/test/poller-lifecycle.test.ts b/external_plugins/telegram/test/poller-lifecycle.test.ts new file mode 100644 index 00000000..f496469b --- /dev/null +++ b/external_plugins/telegram/test/poller-lifecycle.test.ts @@ -0,0 +1,122 @@ +// Regression test for anthropics/claude-code#81571: starting a second +// Claude Code session must not kill the Telegram poller a first session is +// using, and the polling slot must hand off to a surviving server when the +// holder exits. +// +// Run from this directory (deps installed via `bun install`): +// bun test test/poller-lifecycle.test.ts +// +// The pid-slot logic runs before any network call, so a dummy token is +// enough — polling errors from the fake token are expected and irrelevant. + +import { afterEach, expect, test } from 'bun:test' +import { mkdtempSync, readFileSync, rmSync, writeFileSync } from 'fs' +import { tmpdir } from 'os' +import { join } from 'path' + +const SERVER = join(import.meta.dir, '..', 'server.ts') + +type Server = ReturnType +const spawned: Server[] = [] + +function startServer(stateDir: string): Server { + const proc = Bun.spawn(['bun', SERVER], { + env: { ...process.env, TELEGRAM_STATE_DIR: stateDir }, + stdin: 'pipe', // held open — the server treats stdin EOF as session exit + stdout: 'ignore', + stderr: 'ignore', + }) + spawned.push(proc) + return proc +} + +function pidFileContents(stateDir: string): number | null { + try { + return parseInt(readFileSync(join(stateDir, 'bot.pid'), 'utf8'), 10) + } catch { + return null + } +} + +async function waitFor(cond: () => boolean, ms: number): Promise { + const deadline = Date.now() + ms + while (Date.now() < deadline) { + if (cond()) return true + await new Promise(r => setTimeout(r, 100)) + } + return cond() +} + +function isAlive(pid: number): boolean { + try { + process.kill(pid, 0) + return true + } catch { + return false + } +} + +function makeStateDir(): string { + const dir = mkdtempSync(join(tmpdir(), 'tg-poller-test-')) + writeFileSync(join(dir, '.env'), 'TELEGRAM_BOT_TOKEN=123456789:AAHdummy_token_for_pid_logic_only\n') + return dir +} + +afterEach(() => { + for (const proc of spawned) { + try { + proc.kill() + } catch {} + } + spawned.length = 0 +}) + +test('second server does not kill the incumbent poller; slot hands off on exit', async () => { + const stateDir = makeStateDir() + try { + // A starts and claims the polling slot. + const a = startServer(stateDir) + expect(await waitFor(() => pidFileContents(stateDir) === a.pid, 10_000)).toBe(true) + + // B starts while A is healthy: B must defer, A must survive as holder. + const b = startServer(stateDir) + expect(await waitFor(() => pidFileContents(stateDir) === b.pid, 4_000)).toBe(false) + expect(isAlive(a.pid)).toBe(true) + expect(isAlive(b.pid)).toBe(true) + expect(pidFileContents(stateDir)).toBe(a.pid) + + // A's session ends (stdin EOF): B's standby watcher takes over the slot. + a.stdin.end() + await a.exited + expect(await waitFor(() => pidFileContents(stateDir) === b.pid, 10_000)).toBe(true) + expect(isAlive(b.pid)).toBe(true) + + // Last session exits: the owner removes its pid file. + b.stdin.end() + await b.exited + expect(await waitFor(() => pidFileContents(stateDir) === null, 5_000)).toBe(true) + } finally { + rmSync(stateDir, { recursive: true, force: true }) + } +}, 60_000) + +test('a standby exiting leaves the incumbent and its pid file untouched', async () => { + const stateDir = makeStateDir() + try { + const a = startServer(stateDir) + expect(await waitFor(() => pidFileContents(stateDir) === a.pid, 10_000)).toBe(true) + + // Short-lived second session: starts, defers, exits. + const b = startServer(stateDir) + await new Promise(r => setTimeout(r, 1_500)) + b.stdin.end() + await b.exited + + // The incumbent still holds the slot — the exact end state #81571 broke + // (no poller left and no pid file at all). + expect(isAlive(a.pid)).toBe(true) + expect(pidFileContents(stateDir)).toBe(a.pid) + } finally { + rmSync(stateDir, { recursive: true, force: true }) + } +}, 60_000)