From fc2b9f6772c63bc9ca60f49e1e83fdef5414f44c Mon Sep 17 00:00:00 2001 From: Tamir Date: Fri, 5 Jun 2026 18:52:03 +0000 Subject: [PATCH 1/2] =?UTF-8?q?feat(telegram):=20multi-session=20threads?= =?UTF-8?q?=20=E2=80=94=20topics=20run=20in=20parallel?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each Telegram topic already gets its own session and the runner keeps independent per-thread serial queues (threadQueues), but the handler gated every incoming message on the global isMainBusy() — any in-flight run anywhere blocked all topics. Add per-thread busy tracking in the runner (isThreadBusy/isGlobalBusy) and check only the target topic's queue (or the global session for non-topic chats), so each topic is a fully independent session running in parallel. A second message to the same topic still gets the polite busy reply. --- src/commands/telegram.ts | 8 +++++--- src/runner.ts | 18 ++++++++++++++++++ 2 files changed, 23 insertions(+), 3 deletions(-) diff --git a/src/commands/telegram.ts b/src/commands/telegram.ts index d709432a..05d1360b 100644 --- a/src/commands/telegram.ts +++ b/src/commands/telegram.ts @@ -1,4 +1,4 @@ -import { ensureProjectClaudeMd, run, runUserMessage, runFork, killActive, isMainBusy, compactCurrentSession, compactCurrentThreadSession, isRateLimited, getRateLimitResetAt, getPermissionMode, setPermissionMode, type PermissionMode } from "../runner"; +import { ensureProjectClaudeMd, run, runUserMessage, runFork, killActive, isThreadBusy, isGlobalBusy, compactCurrentSession, compactCurrentThreadSession, isRateLimited, getRateLimitResetAt, getPermissionMode, setPermissionMode, type PermissionMode } from "../runner"; import { wrapUntrusted } from "../prompt-safety"; import { isAllowed } from "../allowlist"; import { extractErrorDetail } from "../messaging"; @@ -1444,14 +1444,16 @@ async function handleMessage(message: TelegramMessage): Promise { ); } const prefixedPrompt = promptParts.join("\n"); - const busy = isMainBusy(); + // Per-thread busy check: only reject if THIS topic's queue (or the global + // session for non-topic chats) is mid-run. Different topics run in parallel. + const busy = sessionKey ? isThreadBusy(sessionKey) : isGlobalBusy(); const verbose = verboseChats.has(chatId); const modelOverride = chatModels.get(chatId); let result; let streamMsgId: number | null = null; let hadToolLines = false; if (busy) { - await sendMessage(config.token, chatId, "Claude is busy — try again in a moment, or use /fork for a quick parallel task.", threadId); + await sendMessage(config.token, chatId, "Claude is still working on the previous message in this topic — try again in a moment, or use /fork for a quick parallel task.", threadId); return; } else { const stream = makeStreamCallback(config.token, chatId, threadId, { verbose }); diff --git a/src/runner.ts b/src/runner.ts index 7d0ee6a1..00cacd2c 100644 --- a/src/runner.ts +++ b/src/runner.ts @@ -298,6 +298,22 @@ export function isMainBusy(): boolean { return mainRunCount > 0; } +// Busy state per thread queue + global queue, so handlers can allow +// parallel runs across different topics/threads while still rejecting +// a second message to the SAME thread (or the global session). +const busyThreads = new Set(); +let busyGlobalCount = 0; + +/** True while THIS thread's queue is processing a task. */ +export function isThreadBusy(threadId: string): boolean { + return busyThreads.has(threadId); +} + +/** True while a global-session (non-thread) run is in flight. */ +export function isGlobalBusy(): boolean { + return busyGlobalCount > 0; +} + function extractRateLimitMessage(stdout: string, stderr: string): string | null { const candidates = [stdout, stderr]; for (const text of candidates) { @@ -1031,6 +1047,7 @@ async function execClaude( onToolEvent?: (line: string) => void ): Promise { mainRunCount++; + if (threadId) busyThreads.add(threadId); else busyGlobalCount++; persistRunCount(); try { await mkdir(LOGS_DIR, { recursive: true }); @@ -1426,6 +1443,7 @@ async function execClaude( return result; } finally { mainRunCount--; + if (threadId) busyThreads.delete(threadId); else busyGlobalCount = Math.max(0, busyGlobalCount - 1); persistRunCount(); } } From 37c95e070a78a4713a27703f36bb4080b918efa7 Mon Sep 17 00:00:00 2001 From: Tamir Date: Fri, 5 Jun 2026 18:52:41 +0000 Subject: [PATCH 2/2] feat(telegram): queue messages sent mid-run instead of rejecting them MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A message arriving while the same topic (or the global session) is mid-run was rejected with a busy notice, forcing the user to resend. The runner already serializes runs per thread via enqueue(), so the message can simply be submitted and execute next, in order, with full session context via --resume — like Claude Code's message queue. While queued, the bot reacts 👀 to the incoming message as a lightweight received-and-waiting ack. The typing indicator already runs while queued, and the stream preview message is only created on first output chunk, so nothing else changes visibly until the queued run starts. --- src/commands/telegram.ts | 28 +++++++++++++--------------- 1 file changed, 13 insertions(+), 15 deletions(-) diff --git a/src/commands/telegram.ts b/src/commands/telegram.ts index 05d1360b..352caf91 100644 --- a/src/commands/telegram.ts +++ b/src/commands/telegram.ts @@ -1444,24 +1444,22 @@ async function handleMessage(message: TelegramMessage): Promise { ); } const prefixedPrompt = promptParts.join("\n"); - // Per-thread busy check: only reject if THIS topic's queue (or the global - // session for non-topic chats) is mid-run. Different topics run in parallel. + // Per-thread queue: if THIS topic (or the global session for non-topic + // chats) is mid-run, the message queues behind the current run — like + // Claude Code — instead of being rejected. run() serializes per thread, + // so queued messages execute in order with full session context. React + // 👀 so the user knows the message was received and is waiting its turn. const busy = sessionKey ? isThreadBusy(sessionKey) : isGlobalBusy(); - const verbose = verboseChats.has(chatId); - const modelOverride = chatModels.get(chatId); - let result; - let streamMsgId: number | null = null; - let hadToolLines = false; if (busy) { - await sendMessage(config.token, chatId, "Claude is still working on the previous message in this topic — try again in a moment, or use /fork for a quick parallel task.", threadId); - return; - } else { - const stream = makeStreamCallback(config.token, chatId, threadId, { verbose }); - result = await runUserMessage("telegram", prefixedPrompt, sessionKey, undefined, stream.onChunk, stream.onToolEvent, modelOverride); - const streamResult = await stream.waitForStreamMsg(); - streamMsgId = streamResult.msgId; - hadToolLines = streamResult.hadToolLines; + await sendReaction(config.token, chatId, message.message_id, "👀").catch(() => {}); } + const verbose = verboseChats.has(chatId); + const modelOverride = chatModels.get(chatId); + const stream = makeStreamCallback(config.token, chatId, threadId, { verbose }); + const result = await runUserMessage("telegram", prefixedPrompt, sessionKey, undefined, stream.onChunk, stream.onToolEvent, modelOverride); + const streamResult = await stream.waitForStreamMsg(); + const streamMsgId: number | null = streamResult.msgId; + const hadToolLines = streamResult.hadToolLines; if (result.exitCode !== 0) { const isTimedOut = result.exitCode === 124;