Fix long-gap confirmation and duplicated pending questions

src/coding_agent_telegram/router/message_commands.py

@@ -251,6 +251,25 @@ def _long_gap_threshold_seconds(self, provider: str) -> int:
             return 0
         return getattr(self.deps.cfg, provider_config.threshold_field, 0)
 
+    def _mark_active_session_recently_resumed(self, chat_id: int) -> None:
+        """Suppress another idle prompt while an accepted resume warms the session.
+
+        A successful provider turn normally updates its native transcript immediately.
+        This local marker also covers the short interval before that write is visible,
+        and prevents queued messages behind an explicit "Proceed anyway" from asking
+        the same long-gap question again.
+        """
+        active_id, session, _project_path = self._active_session_context(chat_id)
+        if active_id is None or session is None:
+            return
+        provider = str(session.get("provider") or "codex").strip().lower() or "codex"
+        threshold_seconds = self._long_gap_threshold_seconds(provider)
+        if threshold_seconds <= 0:
+            return
+        now_monotonic = time.monotonic()
+        self._prune_session_gap_cache(now_monotonic)
+        self._session_gap_safe_until[f"{provider}:{active_id}"] = now_monotonic + threshold_seconds
+
     async def _maybe_warn_long_gap(
         self,
         update: Update,
@@ -432,6 +451,10 @@ async def replay() -> None:
 
         if action == "longgap:proceed":
             await query.edit_message_text(self._t(update, "runtime.long_gap_proceeding"))
+            # The user has approved resuming this session.  Queue entries waiting
+            # behind the held message share that newly-warmed session, rather than
+            # needing a duplicate confirmation based on the same old activity time.
+            self._mark_active_session_recently_resumed(chat_id)
             await replay()
             return
 

src/coding_agent_telegram/router/queue_processing.py

@@ -294,6 +294,23 @@ async def _dispatch_queued_questions(
         else:
             user_message = queued_messages[0].text
             reply_to_message_id = queued_messages[0].reply_to_message_id
+
+        # Queue dispatch resumes the very same provider session as an ordinary
+        # message.  It must therefore pass through the same idle/cache warning before
+        # it starts the run.  In particular, choosing "Group questions" is only a
+        # batching decision; it must not bypass the user's compact/new-session choice.
+        #
+        # The queue file has already been claimed above.  If the message is held for a
+        # long-gap decision, it is now represented by that pending action instead, so
+        # retire the claimed file and let the callback replay the held message exactly
+        # once.  Returning True lets the drain loop observe the pending action and
+        # stop without putting this batch back on the queue.
+        if await self._maybe_warn_long_gap(queued_update, context, user_message, suppress_working_notice=False):
+            queue_file.unlink(missing_ok=True)
+            self._queue_lock_path(queue_file).unlink(missing_ok=True)
+            self._chat_processing_queue_files.pop(chat_id, None)
+            return True
+
         logger.debug(
             "Dispatching queued question(s) for chat %s grouped=%s count=%s reply_to_message_id=%s.",
             chat_id,
@@ -311,6 +328,7 @@ async def _dispatch_queued_questions(
             {
                 "kind": "message",
                 "user_message": user_message,
+                "reply_to_message_id": reply_to_message_id,
             },
         )
         continued = await self._continue_pending_action(
@@ -319,6 +337,16 @@ async def _dispatch_queued_questions(
             drain_queue_after_completion=False,
         )
         if not continued:
+            # A prerequisite such as selecting a replacement project keeps this
+            # message in persistent pending_action state.  That state is now the sole
+            # owner of the question: re-adding its queue file would run it once when
+            # the prerequisite is resolved and again when the queue later drains.
+            pending_action = self._pending_action(chat_id)
+            if isinstance(pending_action, dict) and pending_action.get("awaiting_prerequisite"):
+                queue_file.unlink(missing_ok=True)
+                self._queue_lock_path(queue_file).unlink(missing_ok=True)
+                self._chat_processing_queue_files.pop(chat_id, None)
+                return True
             self._queue_lock_path(queue_file).unlink(missing_ok=True)
             self._chat_processing_queue_files.pop(chat_id, None)
             queue = self._chat_message_queue_files.setdefault(chat_id, deque())

src/coding_agent_telegram/router/session_common.py

@@ -51,6 +51,11 @@ def _should_queue_incoming_message(self, chat_id: int) -> bool:
         return (
             self._is_project_busy(chat_id)
             or self._has_pending_queue_files(chat_id)
+            # A queue file has already been claimed for dispatch but its message may
+            # still be going through an async preflight check (such as long-gap
+            # confirmation).  Treat that as queued work too, so a concurrently
+            # received message cannot jump ahead of it.
+            or chat_id in self._chat_processing_queue_files
             or self._has_pending_queue_decision(chat_id)
             or isinstance(pending_action, dict)
             or has_pending_photo_album(chat_id)

src/coding_agent_telegram/router/session_lifecycle_commands.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+from types import SimpleNamespace
 
 from telegram import Update
 from telegram.ext import ContextTypes
@@ -24,6 +25,14 @@
 class SessionLifecycleCommandMixin:
     _CREATE_SESSION_TEXT_RE = re.compile(r"^\s*create\s+session\s*:\s*(.*?)\s*$", re.IGNORECASE)
 
+    def _hold_pending_action_for_prerequisite(self, chat_id: int, pending_action: dict[str, object] | None) -> None:
+        """Persist that a pending action is waiting for user-resolvable setup."""
+        if pending_action is None:
+            return
+        held_action = dict(pending_action)
+        held_action["awaiting_prerequisite"] = True
+        self._store_pending_action(chat_id, held_action)
+
     def _parse_create_session_text(self, text: str) -> tuple[bool, str | None]:
         match = self._CREATE_SESSION_TEXT_RE.match(text)
         if not match:
@@ -63,6 +72,7 @@ async def _resolve_session_prerequisites(
         chat_state = self.deps.store.get_chat_state(self.deps.bot_id, chat_id)
         provider = self._selected_provider(chat_state)
         if not provider:
+            self._hold_pending_action_for_prerequisite(chat_id, pending_action)
             await self._prompt_for_provider_selection(
                 update,
                 context,
@@ -75,7 +85,7 @@ async def _resolve_session_prerequisites(
 
         project_folder = str(chat_state.get("current_project_folder") or "").strip()
         if not project_folder:
-            self._store_pending_action(chat_id, pending_action)
+            self._hold_pending_action_for_prerequisite(chat_id, pending_action)
             await send_text(
                 update,
                 context,
@@ -85,13 +95,13 @@ async def _resolve_session_prerequisites(
 
         project_path = resolve_project_path(self.deps.cfg.workspace_root, project_folder)
         if not project_path.exists() or not project_path.is_dir():
-            self._store_pending_action(chat_id, pending_action)
+            self._hold_pending_action_for_prerequisite(chat_id, pending_action)
             await send_text(update, context, self._t(update, "project.project_folder_missing_retry", project_folder=project_folder))
             return None
 
         branch_name = str(chat_state.get("current_branch") or "").strip()
         if self.git.is_git_repo(project_path) and not branch_name:
-            self._store_pending_action(chat_id, pending_action)
+            self._hold_pending_action_for_prerequisite(chat_id, pending_action)
             await self._send_branch_selection_prompt(
                 update,
                 context,
@@ -218,6 +228,10 @@ async def _continue_pending_action(
             if resolved is None:
                 return False
             provider, project_folder, branch_name, project_path = resolved
+            if pending_action.get("awaiting_prerequisite"):
+                pending_action = dict(pending_action)
+                pending_action.pop("awaiting_prerequisite", None)
+                self._store_pending_action(chat_id, pending_action)
             kind = str(pending_action.get("kind") or "")
 
             if kind == "new_session":
@@ -257,9 +271,19 @@ async def _continue_pending_action(
                         return False
                 if not await self._ensure_active_session_ready_for_run(update, context):
                     return False
+                # A deferred queued message may be resumed by a control command such
+                # as /project.  Keep its response associated with the original user
+                # question instead of making Telegram quote that control command.
+                reply_to_message_id = pending_action.get("reply_to_message_id")
+                run_update = update
+                if isinstance(reply_to_message_id, int):
+                    run_update = SimpleNamespace(
+                        effective_chat=update.effective_chat,
+                        message=SimpleNamespace(message_id=reply_to_message_id),
+                    )
                 try:
                     self._last_run_results[chat_id] = await self.runtime.run_active_session(
-                        update,
+                        run_update,
                         context,
                         user_message=user_message,
                         suppress_working_notice=bool(pending_action.get("suppress_working_notice")),
@@ -307,6 +331,7 @@ async def _ensure_active_session_ready_for_run(self, update: Update, context: Co
             return await self._resolve_branch_discrepancy_if_needed(update, context)
 
         pending_action = dict(pending_action)
+        pending_action["awaiting_prerequisite"] = True
         pending_action["branch_resolution"] = {
             "kind": "discrepancy",
             "session_id": active_session_id,

tests/test_command_router.py

@@ -14,6 +14,7 @@
 import pytest
 from coding_agent_telegram.agent_runner import AgentProgressInfo, AgentRunResult, AgentStallInfo
 from coding_agent_telegram.command_router import CommandRouter, RouterDeps
+from coding_agent_telegram.router.queue_processing import QueuedQuestion
 from coding_agent_telegram.router.session_lifecycle_commands import SESSION_PRIMING_PROMPT
 from coding_agent_telegram.config import AppConfig
 from coding_agent_telegram.session_store import SessionStore
@@ -4482,6 +4483,98 @@ def test_long_gap_warning_sent_and_holds_message_when_native_session_idle_past_t
     }
 
 
+def test_grouped_queued_questions_warn_before_resuming_long_idle_session(tmp_path: Path, monkeypatch):
+    """Grouping a queue batch must not bypass the same long-gap guard as a new message."""
+    home = tmp_path / "home"
+    monkeypatch.setenv("HOME", str(home))
+    backend = tmp_path / "backend"
+    backend.mkdir()
+    runner = DummyRunner()
+    cfg = make_config(tmp_path)
+    cfg = AppConfig(**{**cfg.__dict__, "long_gap_warning_enabled": True, "codex_long_gap_seconds": 600})
+    store = SessionStore(cfg.state_file, cfg.state_backup_file)
+    store.create_session("bot-a", 123, "sess_idle", "idle-session", "backend", "codex")
+    seed_codex_native_session(
+        home,
+        session_id="sess_idle",
+        cwd=backend,
+        title="idle-session",
+        branch="",
+        created_at=int(time.time()) - 7200,
+        updated_at=int(time.time()) - 7200,
+        tokens_used=100_000,
+    )
+    router = CommandRouter(RouterDeps(cfg=cfg, store=store, agent_runner=runner, bot_id="bot-a"))
+    router.git = FakeGitManager(is_git_repo=False)
+    queue_file = router._next_queue_file_path(123)
+    queued_questions = [QueuedQuestion("first queued question"), QueuedQuestion("second queued question")]
+    router._write_queue_questions(queue_file, queued_questions)
+    bot = FakeBot()
+    context = SimpleNamespace(args=[], bot=bot)
+
+    continued = asyncio.run(
+        router._dispatch_queued_questions(
+            123,
+            context,
+            queue_file=queue_file,
+            queued_messages=queued_questions,
+            grouped=True,
+        )
+    )
+
+    assert continued is True
+    assert runner.resume_calls == []
+    assert not queue_file.exists()
+    assert 123 not in router._chat_processing_queue_files
+    pending = store.get_chat_state("bot-a", 123)["pending_action"]
+    assert pending["kind"] == "long_gap_confirm"
+    assert "Answer the following queued user questions in order." in pending["user_message"]
+    assert "first queued question" in pending["user_message"]
+    assert "second queued question" in pending["user_message"]
+    buttons = [button.callback_data for row in bot.messages[-1][3].inline_keyboard for button in row]
+    assert buttons == ["longgap:switch", "longgap:compact", "longgap:proceed"]
+
+
+def test_queued_message_waiting_for_replaced_project_runs_once_and_replies_to_original_question(tmp_path: Path):
+    """Resolving a missing project must not duplicate a held queue entry or quote /project."""
+    runner = DummyRunner()
+    cfg = make_config(tmp_path)
+    store = SessionStore(cfg.state_file, cfg.state_backup_file)
+    store.set_current_project_folder("bot-a", 123, "renamed-project")
+    store.create_session("bot-a", 123, "sess_missing", "old-session", "renamed-project", "codex")
+    router = CommandRouter(RouterDeps(cfg=cfg, store=store, agent_runner=runner, bot_id="bot-a"))
+    router.git = FakeGitManager(is_git_repo=False)
+    queue_file = router._next_queue_file_path(123)
+    queued_question = QueuedQuestion("restore the chat history", reply_to_message_id=321)
+    router._write_queue_questions(queue_file, [queued_question])
+    bot = FakeBot()
+    context = SimpleNamespace(args=[], bot=bot)
+
+    continued = asyncio.run(
+        router._dispatch_queued_questions(
+            123,
+            context,
+            queue_file=queue_file,
+            queued_messages=[queued_question],
+            grouped=False,
+        )
+    )
+
+    assert continued is True
+    assert runner.resume_calls == []
+    assert not queue_file.exists()
+    pending = store.get_chat_state("bot-a", 123)["pending_action"]
+    assert pending["user_message"] == "restore the chat history"
+    assert pending["reply_to_message_id"] == 321
+
+    project_update = make_update(text="/project replacement-project", message_id=999)
+    asyncio.run(router.handle_project(project_update, SimpleNamespace(args=["replacement-project"], bot=bot)))
+
+    assert [call["user_message"] for call in runner.resume_calls] == ["restore the chat history"]
+    working_entries = [entry for entry in bot.sent_messages if entry["text"] == "Working on it..."]
+    assert working_entries[-1]["reply_to_message_id"] == 321
+
+
 def test_long_gap_warning_skipped_for_small_session_despite_long_idle(tmp_path: Path, monkeypatch):
     """A session with little accumulated context shouldn't nag just because it sat idle
     -- reprocessing it from scratch is cheap regardless, so the size gate should skip