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(); } }