feat(telegram): queue messages sent mid-run instead of rejecting them

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,22 +1444,22 @@ async function handleMessage(message: TelegramMessage): Promise<void> {
       );
     }
     const prefixedPrompt = promptParts.join("\n");
-    const busy = isMainBusy();
-    const verbose = verboseChats.has(chatId);
-    const modelOverride = chatModels.get(chatId);
-    let result;
-    let streamMsgId: number | null = null;
-    let hadToolLines = false;
+    // 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();
     if (busy) {
-      await sendMessage(config.token, chatId, "Claude is busy — 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;

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<string>();
+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<RunResult> {
   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();
   }
 }