diff --git a/src/coding_agent_telegram/router/message_commands.py b/src/coding_agent_telegram/router/message_commands.py index 4eca46c..31d9fea 100644 --- a/src/coding_agent_telegram/router/message_commands.py +++ b/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 diff --git a/src/coding_agent_telegram/router/queue_processing.py b/src/coding_agent_telegram/router/queue_processing.py index d09bfa4..609e7a7 100644 --- a/src/coding_agent_telegram/router/queue_processing.py +++ b/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()) diff --git a/src/coding_agent_telegram/router/session_common.py b/src/coding_agent_telegram/router/session_common.py index ec094c0..b26a086 100644 --- a/src/coding_agent_telegram/router/session_common.py +++ b/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) diff --git a/src/coding_agent_telegram/router/session_lifecycle_commands.py b/src/coding_agent_telegram/router/session_lifecycle_commands.py index 59042e0..2b5f63f 100644 --- a/src/coding_agent_telegram/router/session_lifecycle_commands.py +++ b/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, diff --git a/tests/test_command_router.py b/tests/test_command_router.py index 96bfa14..eb4657b 100644 --- a/tests/test_command_router.py +++ b/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