From e3b815ed757c5686295aff7c65243068cb4025c6 Mon Sep 17 00:00:00 2001 From: Nosko Artem Date: Thu, 6 Aug 2026 12:03:29 +0300 Subject: [PATCH 1/2] fix: guarantee startup message delivery --- src/ccbot/bot/_session_create.py | 43 +--- src/ccbot/bot/app.py | 10 + src/ccbot/bot/callbacks/dir_browser.py | 16 +- src/ccbot/bot/callbacks/more_menu.py | 3 + src/ccbot/bot/callbacks/switcher.py | 3 + src/ccbot/bot/callbacks/window_picker.py | 31 ++- src/ccbot/bot/commands/lifecycle.py | 21 +- src/ccbot/bot/messages.py | 220 ++++++++++++------- src/ccbot/session.py | 52 ++++- src/ccbot/startup_queue.py | 255 +++++++++++++++++++++++ src/ccbot/tmux_manager.py | 94 ++++++--- tests/ccbot/test_codex_backend.py | 37 +++- tests/ccbot/test_startup_queue.py | 242 +++++++++++++++++++++ tests/e2e/test_inbound_routing.py | 9 +- 14 files changed, 859 insertions(+), 177 deletions(-) create mode 100644 src/ccbot/startup_queue.py create mode 100644 tests/ccbot/test_startup_queue.py diff --git a/src/ccbot/bot/_session_create.py b/src/ccbot/bot/_session_create.py index 32963178..ff06f0c2 100644 --- a/src/ccbot/bot/_session_create.py +++ b/src/ccbot/bot/_session_create.py @@ -16,15 +16,13 @@ from telegram.ext import ContextTypes -from ..handlers.directory_browser import take_pending_text -from ..handlers.message_sender import safe_edit, safe_send +from ..handlers.message_sender import safe_edit from ..handlers.notifications import ( detach_paused_cards_at_message, paint_card_on_carrier, ) from ..i18n import t from ..local_terminal import open_terminal_for_window -from ..naming import maybe_auto_name from ..session import session_manager from ..tmux_manager import tmux_manager @@ -79,8 +77,6 @@ async def create_and_activate_session( ) if not success: await safe_edit(query, f"❌ {message}") - if context.user_data is not None: - context.user_data.pop("_pending_text", None) return logger.info( @@ -128,6 +124,13 @@ async def create_and_activate_session( session_manager.set_session_claude_id(sess.id, ws.session_id) session_manager.set_active_session(user.id, sess.id) + # Every inbound captured since the user pressed Start is now owned by + # this window. The drain waits for proven TUI readiness and replays the + # original Telegram updates in order. + from ..startup_queue import bind_startup_queue + + bind_startup_queue(user.id, created_wid) + # Transition the carrier from dir-browser to the new session's # empty live card in place. No separate "Created. Send messages # here." notice — that was a dead-end stub; the live card itself @@ -178,33 +181,3 @@ async def _bind_lifecycle_in_background() -> None: open_terminal_for_window(created_wid, user_id=user.id), name=f"local-terminal:{created_wid}", ) - - # Forward any pending text held while the picker was up. ``take_pending_text`` - # drops a stale stash (older than PENDING_TEXT_TTL_S) so a message typed - # hours ago against a since-gone session can never be injected here — the - # 2026-06-28 "medical insurance" misroute. - pending_text = take_pending_text(context.user_data) - if pending_text: - logger.debug( - "Forwarding pending text to window %s (len=%d)", - created_wname, - len(pending_text), - ) - send_ok, send_msg = await session_manager.send_to_window( - created_wid, pending_text - ) - if not send_ok: - logger.warning("Failed to forward pending text: %s", send_msg) - await safe_send( - context.bot, - user.id, - f"❌ Failed to send pending message: {send_msg}", - ) - elif len(pending_text) >= 20: - # Name the session from its actual first human message (this - # pending text), not from whatever inbound happens to land - # first. ``maybe_auto_name`` is one-shot (re-entrancy guard), - # so a later message can't override it. - asyncio.create_task( - maybe_auto_name(sess.id, pending_text, getattr(user, "id", None)) - ) diff --git a/src/ccbot/bot/app.py b/src/ccbot/bot/app.py index 2a0d60fa..7616e25d 100644 --- a/src/ccbot/bot/app.py +++ b/src/ccbot/bot/app.py @@ -25,6 +25,8 @@ filters, ) +from ..startup_queue import capture_startup_message + from ..config import config from ..handlers.quota_alerts import quota_alerts_loop from ..handlers.notifications import card_timer_loop @@ -516,6 +518,14 @@ def create_bot() -> "Application[Any, Any, Any, Any, Any, Any]": logger.info("TG proxy enabled: %s", config.tg_proxy_url) application = builder.build() + # Group -1 runs before commands and content handlers. It is a no-op unless + # a new-session flow is open; while open it captures the update and stops + # it from leaking to the previously-active session. + application.add_handler( + MessageHandler(filters.ALL & ~filters.StatusUpdate.ALL, capture_startup_message), + group=-1, + ) + # Visible menu commands. application.add_handler(CommandHandler("history", history_command)) application.add_handler(CommandHandler("screenshot", screenshot_command)) diff --git a/src/ccbot/bot/callbacks/dir_browser.py b/src/ccbot/bot/callbacks/dir_browser.py index cc3238b3..169f89ef 100644 --- a/src/ccbot/bot/callbacks/dir_browser.py +++ b/src/ccbot/bot/callbacks/dir_browser.py @@ -214,9 +214,15 @@ async def handle( return True if data == CB_DIR_CANCEL: + from ...startup_queue import cancel_startup_queue + + unsent = cancel_startup_queue(user.id) clear_browse_state(context.user_data) await _close_modal(query, user.id, context) - await query.answer() + await query.answer( + f"Cancelled; {unsent} queued item(s) were not sent" if unsent else None, + show_alert=bool(unsent), + ) return True if data.startswith(CB_SESSION_SELECT): @@ -272,13 +278,19 @@ async def handle( return True if data == CB_SESSION_CANCEL: + from ...startup_queue import cancel_startup_queue + + unsent = cancel_startup_queue(user.id) clear_session_picker_state(context.user_data) if context.user_data is not None: context.user_data.pop("_selected_path", None) context.user_data.pop(SESSIONS_PAGE_KEY, None) clear_browse_state(context.user_data) await _close_modal(query, user.id, context) - await query.answer() + await query.answer( + f"Cancelled; {unsent} queued item(s) were not sent" if unsent else None, + show_alert=bool(unsent), + ) return True if data == CB_SESSION_BACK: diff --git a/src/ccbot/bot/callbacks/more_menu.py b/src/ccbot/bot/callbacks/more_menu.py index 85d038bc..1f37a05f 100644 --- a/src/ccbot/bot/callbacks/more_menu.py +++ b/src/ccbot/bot/callbacks/more_menu.py @@ -45,6 +45,9 @@ async def _emit_new_flow( query: CallbackQuery, context: ContextTypes.DEFAULT_TYPE, user: Any ) -> None: """Open the directory browser from the Menu screen.""" + from ...startup_queue import begin_startup_queue + + begin_startup_queue(user.id) from ...handlers.directory_browser import ( BROWSE_DIRS_KEY, BROWSE_PAGE_KEY, diff --git a/src/ccbot/bot/callbacks/switcher.py b/src/ccbot/bot/callbacks/switcher.py index b1cf13b7..69be4771 100644 --- a/src/ccbot/bot/callbacks/switcher.py +++ b/src/ccbot/bot/callbacks/switcher.py @@ -239,6 +239,9 @@ async def _seed_bg_status(old_sess: _Session) -> None: # Pause the active session first so its events buffer silently # while the user picks a directory; events catch up when the # user switches back via the switcher. + from ...startup_queue import begin_startup_queue + + begin_startup_queue(user.id) active = session_manager.get_active_session(user.id) if active is not None: pause_card_view(user.id, active.id) diff --git a/src/ccbot/bot/callbacks/window_picker.py b/src/ccbot/bot/callbacks/window_picker.py index 07b8a574..b106aaa0 100644 --- a/src/ccbot/bot/callbacks/window_picker.py +++ b/src/ccbot/bot/callbacks/window_picker.py @@ -24,9 +24,8 @@ UNBOUND_WINDOWS_KEY, build_directory_browser, clear_window_picker_state, - take_pending_text, ) -from ...handlers.message_sender import safe_edit, safe_send +from ...handlers.message_sender import safe_edit from ...session import session_manager from ...tmux_manager import tmux_manager from .._common import open_more_in_place @@ -78,22 +77,16 @@ async def handle( await safe_edit(query, f"✅ Bound to window `{display}`") - pending_text = take_pending_text(context.user_data) - if pending_text: - send_ok, send_msg = await session_manager.send_to_window( - selected_wid, pending_text - ) - if not send_ok: - logger.warning("Failed to forward pending text: %s", send_msg) - await safe_send( - context.bot, - user.id, - f"❌ Failed to send pending message: {send_msg}", - ) + from ...startup_queue import bind_startup_queue + + bind_startup_queue(user.id, selected_wid) await query.answer("Bound") return True if data == CB_WIN_NEW: + from ...startup_queue import begin_startup_queue + + begin_startup_queue(user.id) clear_window_picker_state(context.user_data) start_path = str(Path.home()) msg_text, keyboard, subdirs = await build_directory_browser( @@ -109,11 +102,15 @@ async def handle( return True if data == CB_WIN_CANCEL: + from ...startup_queue import cancel_startup_queue + + unsent = cancel_startup_queue(user.id) clear_window_picker_state(context.user_data) - if context.user_data is not None: - context.user_data.pop("_pending_text", None) await open_more_in_place(query, user.id) - await query.answer() + await query.answer( + f"Cancelled; {unsent} queued item(s) were not sent" if unsent else None, + show_alert=bool(unsent), + ) return True return False diff --git a/src/ccbot/bot/commands/lifecycle.py b/src/ccbot/bot/commands/lifecycle.py index cf4b2a67..ad60a0df 100644 --- a/src/ccbot/bot/commands/lifecycle.py +++ b/src/ccbot/bot/commands/lifecycle.py @@ -35,6 +35,7 @@ from ...i18n import t from ...session import Session, session_manager from ...tmux_manager import tmux_manager +from ...startup_queue import begin_startup_queue, bind_startup_queue from .._common import ( active_window, is_user_allowed, @@ -60,6 +61,10 @@ async def new_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non if not update.message: return + # Open capture before authentication, filesystem and Telegram awaits. + # Non-blocking voice handlers can otherwise race into the old session. + begin_startup_queue(user.id) + args = (update.message.text or "").split(maxsplit=2) name_arg = args[1] if len(args) > 1 else "" path_arg = args[2] if len(args) > 2 else "" @@ -84,10 +89,13 @@ async def new_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non if not success: await safe_reply(update.message, f"❌ {message}") return - hook_ok = await session_manager.wait_for_session_map_entry( - created_wid, timeout=5.0 + session_manager.mark_window_starting( + created_wid, + backend=session_manager.agent_backend, + resume=False, + bot=context.bot, + user_id=user.id, ) - del hook_ok sess = session_manager.create_session( name=name_arg or created_wname or "", window_id=created_wid, @@ -97,10 +105,15 @@ async def new_command(update: Update, context: ContextTypes.DEFAULT_TYPE) -> Non if ws.session_id: session_manager.set_session_claude_id(sess.id, ws.session_id) session_manager.set_active_session(user.id, sess.id) + bind_startup_queue(user.id, created_wid) if session_manager.get_user_settings(user.id).get("local_terminal") == "auto": from ...local_terminal import open_terminal_for_window + import asyncio - await open_terminal_for_window(created_wid, user_id=user.id) + asyncio.create_task( + open_terminal_for_window(created_wid, user_id=user.id), + name=f"local-terminal:{created_wid}", + ) await safe_reply( update.message, f"✅ Session `{sess.name}` ({sess.id}) created at {target_path}", diff --git a/src/ccbot/bot/messages.py b/src/ccbot/bot/messages.py index 65a8f820..b8b0ec67 100644 --- a/src/ccbot/bot/messages.py +++ b/src/ccbot/bot/messages.py @@ -35,7 +35,6 @@ STATE_SELECTING_SESSION, STATE_SELECTING_WINDOW, build_directory_browser, - stash_pending_text, ) from ..handlers.interactive_ui import ( get_interactive_window, @@ -173,14 +172,50 @@ def _transcript_contains_voice_text( async def _wait_for_voice_transcript( checkpoint: _VoiceTranscriptCheckpoint | None, text: str, + *, + wid: str | None = None, ) -> bool | None: - """Wait for delivery proof; ``None`` means no transcript was available.""" - if checkpoint is None: + """Wait for exact delivery proof in the target session transcript. + + A fresh Codex session has no rollout/session_map binding before its first + accepted prompt. In that case keep polling the binding and scan the new + transcript from byte zero instead of treating "no checkpoint" as success. + """ + if checkpoint is None and wid is None: return None + if checkpoint is None and wid is not None: + provisional = session_manager.window_states.get(wid) + if not isinstance(provisional, WindowState): + # A real fresh-session flow always publishes provisional state + # before exposing the card. Missing state means this is a legacy + # caller (or a focused unit-test double), so transcript proof is + # not available on this path. + return None loop = asyncio.get_running_loop() deadline = loop.time() + _VOICE_TRANSCRIPT_CONFIRM_TIMEOUT while True: - if await asyncio.to_thread(_transcript_contains_voice_text, checkpoint, text): + if checkpoint is None and wid is not None: + await session_manager.load_session_map() + state = session_manager.window_states.get(wid) + if isinstance(state, WindowState) and state.session_id: + path = ( + Path(state.transcript_path) + if state.transcript_path + else None + ) + if path is None or not path.is_file(): + if state.backend == "codex": + from ..codex_session_io import build_session_file_path + else: + from ..session_claude_io import build_session_file_path + path = build_session_file_path(state.session_id, state.cwd) + if path is not None and path.is_file(): + checkpoint = _VoiceTranscriptCheckpoint( + path=path, offset=0, backend=state.backend + ) + if checkpoint is not None and await asyncio.to_thread( + _transcript_contains_voice_text, checkpoint, text + ): return True remaining = deadline - loop.time() if remaining <= 0: @@ -188,6 +223,42 @@ async def _wait_for_voice_transcript( await asyncio.sleep(min(_VOICE_TRANSCRIPT_CONFIRM_POLL, remaining)) +async def _send_with_delivery_proof( + wid: str, text: str, sess: Session | None +) -> tuple[bool, str]: + """Send one prompt and require an exact Codex transcript acknowledgement.""" + transcript_checkpoint = _voice_transcript_checkpoint(wid) + message = "" + for attempt in range(1, 3): + success, message = await session_manager.send_to_window(wid, text) + if not success: + continue + if message.startswith("Queued for "): + return True, message + if sess is None or sess.backend != "codex": + return True, message + if not await tmux_manager.ensure_codex_prompt_submitted(wid, text): + message = "Codex kept the text in its input field" + continue + # TUI slash commands do not become ordinary user_message rows. + if text.lstrip().startswith("/"): + return True, message + confirmed = await _wait_for_voice_transcript( + transcript_checkpoint, text, wid=wid + ) + if confirmed is True or confirmed is None: + return True, message + logger.warning( + "Codex delivery absent from transcript; retrying exact prompt " + "window=%s attempt=%d/2 text_len=%d", + wid, + attempt, + len(text), + ) + message = "Prompt did not appear in the Codex transcript" + return False, message or "Delivery was not acknowledged" + + def _enqueue_voice( user_id: int, wid: str ) -> tuple[asyncio.Future[bool] | None, asyncio.Future[bool]]: @@ -417,13 +488,13 @@ async def _intercept_if_pending_ui( async def forward_command_handler( update: Update, context: ContextTypes.DEFAULT_TYPE -) -> None: +) -> bool: """Forward an unhandled /command as a slash to the active Claude session.""" user = update.effective_user if not user or not is_user_allowed(user.id): - return + return False if not update.message: - return + return False cmd_text = update.message.text or "" cc_slash = cmd_text.split("@")[0] # strip bot mention @@ -432,15 +503,15 @@ async def forward_command_handler( await safe_reply( update.message, "❌ No active session. Use /new to create one." ) - return + return False if not await _await_prior_voice(user.id, wid): - return + return False w = await tmux_manager.find_window_by_id(wid) if not w: display = session_manager.get_display_name(wid) await safe_reply(update.message, f"❌ Window '{display}' no longer exists.") - return + return False display = session_manager.get_display_name(wid) logger.info( @@ -448,10 +519,10 @@ async def forward_command_handler( ) await fire_typing(context.bot, user.id, "forward_command", window_id=wid) if await _intercept_if_pending_ui(context.bot, user.id, wid, update.message): - return + return False sess = session_manager.find_session_by_window(wid) async with _card_repost_bracket(context.bot, user.id, sess) as repost: - success, message = await session_manager.send_to_window(wid, cc_slash) + success, message = await _send_with_delivery_proof(wid, cc_slash, sess) if success: # /clear: drop the session association so we re-detect once a # new session id is written by the next user message. @@ -469,6 +540,8 @@ async def forward_command_handler( repost.commit() else: await safe_reply(update.message, f"❌ {message}") + return False + return True # --- non-text catch-all --- @@ -541,7 +614,7 @@ def _hidden_link_urls(msg: Any) -> list[str]: async def unsupported_content_handler( update: Update, context: ContextTypes.DEFAULT_TYPE -) -> None: +) -> bool: """Catch-all for messages without a dedicated handler. When the message carries a caption (typical for forwarded channel @@ -554,15 +627,15 @@ async def unsupported_content_handler( caption to salvage. """ if not update.message: - return + return False user = update.effective_user if not user or not is_user_allowed(user.id): - return + return False msg = update.message wid_for_queue = active_window(user.id) if wid_for_queue is not None: if not await _await_prior_voice(user.id, wid_for_queue): - return + return False caption = (msg.caption or "").strip() if caption: @@ -572,7 +645,7 @@ async def unsupported_content_handler( msg, "❌ No active session. Send a text message first or use /new.", ) - return + return False w = await tmux_manager.find_window_by_id(wid) if not w: display = session_manager.get_display_name(wid) @@ -581,7 +654,7 @@ async def unsupported_content_handler( f"❌ Window '{display}' no longer exists.\n" "Send a message to start a new session.", ) - return + return False prefix = _forward_attribution(msg) hidden_urls = _hidden_link_urls(msg) @@ -593,19 +666,21 @@ async def unsupported_content_handler( await fire_typing(context.bot, user.id, "caption_forward", window_id=wid) if await _intercept_if_pending_ui(context.bot, user.id, wid, msg): - return + return False sess = session_manager.find_session_by_window(wid) async with _card_repost_bracket(context.bot, user.id, sess) as repost: - success, message = await session_manager.send_to_window(wid, text_to_send) + success, message = await _send_with_delivery_proof( + wid, text_to_send, sess + ) if not success: await safe_reply(msg, f"❌ {message}") - return + return False if sess is not None: session_manager.touch_session(sess.id) repost.commit() # No success reply — the user just sent the message; they know # they sent it. Errors above still surface. - return + return True logger.debug("Unsupported content from user %d", user.id) await safe_reply( @@ -613,6 +688,7 @@ async def unsupported_content_handler( "⚠ Only text, photo, and voice messages are supported. " "Stickers, video, and other media cannot be forwarded to Claude Code.", ) + return True # --- inbox file plumbing (photo + document share this) --- @@ -646,10 +722,10 @@ async def _forward_inbox_file( rel_path = str(file_path) text_to_send = f"{caption}\n\n{rel_path}" if caption.strip() else rel_path await fire_typing(bot, user_id, "inbox_file_forward", window_id=wid, label=label) - return await session_manager.send_to_window(wid, text_to_send) + return await _send_with_delivery_proof(wid, text_to_send, sess) -async def photo_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: +async def photo_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: """Drop the user's photo into the active session's inbox + notify Claude.""" user = update.effective_user if not user or not is_user_allowed(user.id): @@ -657,10 +733,10 @@ async def photo_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N # allowlist is private; unauthorized senders should see the bot # as inert (no "not authorized" copy that signals "you found the # right bot, just not the right user"). - return + return False if not update.message or not update.message.photo: - return + return False wid = active_window(user.id) if wid is None: @@ -668,9 +744,9 @@ async def photo_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N update.message, "❌ No active session. Send a text message first or use /new.", ) - return + return False if not await _await_prior_voice(user.id, wid): - return + return False w = await tmux_manager.find_window_by_id(wid) if not w: @@ -680,7 +756,7 @@ async def photo_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N f"❌ Window '{display}' no longer exists.\n" "Send a message to start a new session.", ) - return + return False sess = session_manager.find_session_by_window(wid) workdir = sess.workdir if sess and sess.workdir else str(ccbot_dir() / "images") @@ -691,7 +767,7 @@ async def photo_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N except BadRequest as e: if _is_file_too_big(e): await safe_reply(update.message, _FILE_TOO_BIG_MSG) - return + return False raise filename = f"{photo.file_unique_id}.jpg" @@ -702,18 +778,19 @@ async def _fetch(target: Path) -> None: caption = update.message.caption or "" if await _intercept_if_pending_ui(context.bot, user.id, wid, update.message): - return + return False async with _card_repost_bracket(context.bot, user.id, sess) as repost: success, message = await _forward_inbox_file( user.id, wid, user.id, file_path, caption, "image", context.bot ) if not success: await safe_reply(update.message, f"❌ {message}") - return + return False repost.commit() + return True -async def document_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: +async def document_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: """Drop the user's document into the active session's inbox + notify Claude.""" user = update.effective_user if not user or not is_user_allowed(user.id): @@ -721,10 +798,10 @@ async def document_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) - # allowlist is private; unauthorized senders should see the bot # as inert (no "not authorized" copy that signals "you found the # right bot, just not the right user"). - return + return False if not update.message or not update.message.document: - return + return False wid = active_window(user.id) if wid is None: @@ -732,9 +809,9 @@ async def document_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) - update.message, "❌ No active session. Send a text message first or use /new.", ) - return + return False if not await _await_prior_voice(user.id, wid): - return + return False w = await tmux_manager.find_window_by_id(wid) if not w: @@ -744,7 +821,7 @@ async def document_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) - f"❌ Window '{display}' no longer exists.\n" "Send a message to start a new session.", ) - return + return False doc = update.message.document sess = session_manager.find_session_by_window(wid) @@ -755,7 +832,7 @@ async def document_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) - except BadRequest as e: if _is_file_too_big(e): await safe_reply(update.message, _FILE_TOO_BIG_MSG) - return + return False raise async def _fetch(target: Path) -> None: @@ -765,15 +842,16 @@ async def _fetch(target: Path) -> None: caption = update.message.caption or "" if await _intercept_if_pending_ui(context.bot, user.id, wid, update.message): - return + return False async with _card_repost_bracket(context.bot, user.id, sess) as repost: success, message = await _forward_inbox_file( user.id, wid, user.id, file_path, caption, "document", context.bot ) if not success: await safe_reply(update.message, f"❌ {message}") - return + return False repost.commit() + return True # --- voice --- @@ -792,7 +870,7 @@ async def _clear_voice_pending_marker(bot: Bot, user_id: int, sess: Session) -> logger.debug("voice-pending marker clear failed: %s", e) -async def voice_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: +async def voice_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: """Queue a voice turn, then transcribe it without letting later messages pass.""" user = update.effective_user if ( @@ -802,24 +880,23 @@ async def voice_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> N or not update.message.voice or resolve_voice_backend(user.id) == "off" ): - await _process_voice(update, context) - return + return await _process_voice(update, context) wid = active_window(user.id) if wid is None: - await _process_voice(update, context) - return + return await _process_voice(update, context) previous, barrier = _enqueue_voice(user.id, wid) delivered = False try: if previous is not None and not await _wait_for_voice(previous): - return + return False delivered = await _process_voice( update, context, pinned_wid=wid, queue_barrier=barrier ) finally: _release_voice(user.id, wid, barrier, delivered=delivered) + return delivered async def _process_voice( @@ -996,7 +1073,9 @@ async def _process_voice( # can be an approval raised by the successfully delivered turn, especially # for the second voice in a queue. Prefer the authoritative transcript and # only use the pane heuristic when no matching user row appears. - transcript_confirmed = await _wait_for_voice_transcript(transcript_checkpoint, text) + transcript_confirmed = await _wait_for_voice_transcript( + transcript_checkpoint, text, wid=wid + ) if transcript_confirmed is True: logger.info( "Voice delivery confirmed by transcript user=%d window=%s", @@ -1164,14 +1243,17 @@ async def _resolve_active_window( Returns the window id when there is a live active session window. Returns None when ``text_handler`` must ``return`` instead — either because there is no active session (a directory browser is opened - with the pending text stashed) or because the active session's + with the message queued) or because the active session's window is gone (it's marked lost, state cleared, and the user told). """ assert update.message is not None wid = active_window(user_id) if wid is None: # No active session — start a directory browser to create one. - # The pending text is held in user_data and forwarded after creation. + from ..startup_queue import begin_startup_queue, enqueue_startup_message + + begin_startup_queue(user_id) + enqueue_startup_message(update, context) logger.info("No active session: showing directory browser (user=%d)", user_id) start_path = str(Path.home()) msg_text, keyboard, subdirs = await build_directory_browser( @@ -1182,7 +1264,6 @@ async def _resolve_active_window( context.user_data[BROWSE_PATH_KEY] = start_path context.user_data[BROWSE_PAGE_KEY] = 0 context.user_data[BROWSE_DIRS_KEY] = subdirs - stash_pending_text(context.user_data, text) await safe_reply(update.message, msg_text, reply_markup=keyboard) return None @@ -1273,25 +1354,12 @@ async def _dispatch_text_to_active( intent_sess_id = sess.id if (owns_card and sess is not None) else None try: _t0 = _time.time() - success, message = await session_manager.send_to_window(wid, text) + success, message = await _send_with_delivery_proof(wid, text, sess) metrics.observe("tg_to_claude_latency_ms", (_time.time() - _t0) * 1000.0) metrics.inc("tg_messages_in") if not success: metrics.inc("tg_send_failures") - await safe_reply(update.message, f"❌ {message}") - return False - queued_for_startup = message.startswith("Queued for ") - if ( - sess is not None - and sess.backend == "codex" - and not queued_for_startup - and not await tmux_manager.ensure_codex_prompt_submitted(wid, text) - ): - metrics.inc("tg_send_failures") - await safe_reply( - update.message, - "❌ Codex kept the text in its input field; the prompt was not sent.", - ) + await safe_reply(update.message, f"❌ Delivery not confirmed: {message}") return False # Immediate typing-indicator so the user sees feedback within @@ -1374,29 +1442,29 @@ async def _dispatch_text_to_active( end_repost_intent(user_id, intent_sess_id) -async def text_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: +async def text_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> bool: user = update.effective_user if not user or not is_user_allowed(user.id): # Drop the message silently — no reply, no callback ack. The # allowlist is private; unauthorized senders should see the bot # as inert (no "not authorized" copy that signals "you found the # right bot, just not the right user"). - return + return False if not update.message or not update.message.text: - return + return False text = update.message.text queued_wid = active_window(user.id) if queued_wid is not None: if not await _await_prior_voice(user.id, queued_wid): - return + return False # A pending /login flow owns the next message: it's the OAuth code, not a # prompt. Must run before session routing — the code would otherwise be # typed into a pane (and echoed into that session's transcript). if await maybe_consume_code(update, context): - return + return True # Ignore text while a picker UI is mid-flight. state = context.user_data.get(STATE_KEY) if context.user_data else None @@ -1406,14 +1474,14 @@ async def text_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No STATE_SELECTING_SESSION, ): await safe_reply(update.message, "Please use the picker above, or tap Cancel.") - return + return False if await _route_reply_quote(update, user.id, text): - return + return True wid = await _resolve_active_window(update, context, user.id, text) if wid is None: - return + return False await fire_typing(context.bot, user.id, "text_handler", window_id=wid) @@ -1425,9 +1493,9 @@ async def text_handler(update: Update, context: ContextTypes.DEFAULT_TYPE) -> No # Enter submits). Surface the prompt to the user and bail before # send_to_window — the user must answer via the keyboard. if await _intercept_if_pending_ui(context.bot, user.id, wid, update.message): - return + return False - await _dispatch_text_to_active(update, context, user.id, wid, text) + return await _dispatch_text_to_active(update, context, user.id, wid, text) # Re-export so existing callers (callbacks/dir_browser.py) keep working. diff --git a/src/ccbot/session.py b/src/ccbot/session.py index b3265f56..97914589 100644 --- a/src/ccbot/session.py +++ b/src/ccbot/session.py @@ -425,8 +425,16 @@ async def _load_session_map_unlocked(self) -> None: self.window_display_names[window_id] = new_wname changed = True - # Clean up window_states entries not in current session_map. - stale_wids = [w for w in self.window_states if w and w not in valid_wids] + # A fresh Codex window has no session_map entry until its first prompt + # is accepted. Keep provisional state for every bot Session still + # bound to a window; deleting it here removed the transcript binding + # and made first-turn delivery impossible to prove. + bound_wids = {sess.window_id for sess in self.sessions.values() if sess.window_id} + stale_wids = [ + w + for w in self.window_states + if w and w not in valid_wids and w not in bound_wids + ] for wid in stale_wids: logger.info("Removing stale window_state: %s", wid) del self.window_states[wid] @@ -1072,7 +1080,7 @@ def mark_window_resuming( """ self.mark_window_starting( window_id, - backend=self.agent_backend, + backend="claude", resume=True, bot=bot, user_id=user_id, @@ -1157,9 +1165,17 @@ async def _typing_keepalive() -> None: _typing_keepalive(), name=f"resume-settle-typing:{window_id}" ) try: - settled = await self._wait_for_resume_settle( - window_id, backend=backend, resume=resume - ) + settled = False + while not settled: + settled = await self._wait_for_resume_settle( + window_id, backend=backend, resume=resume + ) + if not settled: + logger.error( + "startup gate remains closed for window %s; " + "TUI readiness is still unproven", + window_id, + ) logger.info( "startup gate cleared for window %s " "(settled=%s backend=%s resume=%s, background)", @@ -1230,9 +1246,16 @@ def _pane_has_ready_input(pane: str, backend: str) -> bool: or "sign in with chatgpt" in lower or "sign in with device code" in lower or "provide your own api key" in lower + or "update available!" in lower ): return False marker = "›" if backend == "codex" else "❯" + if backend == "codex" and "openai codex" not in lower: + # Artem's shell prompt also starts with `›`. Marker-only + # detection cleared the startup queue while the pane was still a + # shell, causing the first Telegram turn to be typed into startup + # chrome and discarded. + return False # The live input row is pinned near the bottom. Restricting detection # to the tail avoids mistaking a historical user row for readiness # while a resumed transcript is still being restored. @@ -1240,6 +1263,23 @@ def _pane_has_ready_input(pane: str, backend: str) -> bool: line.lstrip().startswith(marker) for line in pane.strip().splitlines()[-6:] ) + async def wait_for_window_ready(self, window_id: str) -> bool: + """Wait until the startup gate has observed the real agent input UI.""" + while window_id in self._resuming_windows: + task = self._resume_settle_tasks.get(window_id) + if task is None: + await asyncio.sleep(_RESUME_SETTLE_POLL) + continue + try: + await asyncio.shield(task) + except asyncio.CancelledError: + return False + except Exception: + logger.exception("startup readiness task failed: %s", window_id) + return False + window = await tmux_manager.find_window_by_id(window_id) + return window is not None + async def _wait_for_resume_settle( self, window_id: str, diff --git a/src/ccbot/startup_queue.py b/src/ccbot/startup_queue.py new file mode 100644 index 00000000..1c17edaa --- /dev/null +++ b/src/ccbot/startup_queue.py @@ -0,0 +1,255 @@ +"""Ordered inbound queue for a session-creation flow. + +The queue starts when the user opens the new-session picker, before the first +filesystem/UI await. A high-priority Telegram handler then captures every +message while the picker is open or the agent TUI is booting. Once the new +tmux window is genuinely ready, messages are replayed through the normal +handlers in Telegram order. + +Entries are removed only after their handler reports success. A failed entry +and everything behind it stay queued, so a transient delivery failure cannot +silently turn into message loss. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections import deque +from dataclasses import dataclass, field +from telegram import Update +from telegram.ext import ApplicationHandlerStop, ContextTypes + +logger = logging.getLogger(__name__) + + +@dataclass +class QueuedInbound: + update: Update + context: ContextTypes.DEFAULT_TYPE + sequence: int + + +@dataclass +class StartupFlow: + user_id: int + entries: deque[QueuedInbound] = field(default_factory=deque) + next_sequence: int = 1 + window_id: str | None = None + drain_task: asyncio.Task[None] | None = None + + +_flows: dict[int, StartupFlow] = {} + + +def begin_startup_queue(user_id: int) -> StartupFlow: + """Open (or retain) the queue for the user's in-progress new session.""" + flow = _flows.get(user_id) + if flow is None: + flow = StartupFlow(user_id=user_id) + _flows[user_id] = flow + logger.info("startup queue opened user=%d", user_id) + return flow + + +def has_startup_queue(user_id: int) -> bool: + return user_id in _flows + + +def pending_startup_count(user_id: int) -> int: + flow = _flows.get(user_id) + return len(flow.entries) if flow is not None else 0 + + +def cancel_startup_queue(user_id: int) -> int: + """Explicitly abandon a cancelled flow and return its unsent count.""" + flow = _flows.pop(user_id, None) + if flow is None: + return 0 + if flow.drain_task is not None and not flow.drain_task.done(): + flow.drain_task.cancel() + count = len(flow.entries) + logger.info("startup queue cancelled user=%d pending=%d", user_id, count) + return count + + +async def capture_startup_message( + update: Update, context: ContextTypes.DEFAULT_TYPE +) -> None: + """High-priority PTB handler that captures messages during Start. + + Raising :class:`ApplicationHandlerStop` prevents the normal handler from + routing the same update to the previously-active session. + """ + user = update.effective_user + if user is None or update.message is None: + return + flow = _flows.get(user.id) + if flow is None: + return + text = (update.message.text or "").strip() + if text.startswith(("/login", "/new")): + # Control-plane commands must be able to repair/restart a failed + # creation flow. begin_startup_queue() retains the existing entries. + return + if text and not text.startswith("/"): + # Authentication codes are control-plane input, never agent prompts. + # Let the normal text handler consume them while retaining the queued + # user turns for the next successful Start attempt. + from .codex_auth import get_flow + + if get_flow(user.id) is not None: + return + enqueue_startup_message(update, context) + raise ApplicationHandlerStop + + +def enqueue_startup_message( + update: Update, context: ContextTypes.DEFAULT_TYPE +) -> QueuedInbound | None: + """Append an update to an already-open flow without stopping dispatch.""" + user = update.effective_user + if user is None or update.message is None: + return None + flow = _flows.get(user.id) + if flow is None: + return None + entry = QueuedInbound(update=update, context=context, sequence=flow.next_sequence) + flow.next_sequence += 1 + flow.entries.append(entry) + logger.info( + "startup queue captured user=%d seq=%d message_id=%s pending=%d", + user.id, + entry.sequence, + getattr(update.message, "message_id", None), + len(flow.entries), + ) + return entry + + +async def _replay(entry: QueuedInbound) -> bool: + """Replay one captured update through its regular inbound handler.""" + # Lazy import avoids a cycle: bot.messages imports the capture handler for + # application registration. + from .bot.messages import ( + document_handler, + forward_command_handler, + photo_handler, + text_handler, + unsupported_content_handler, + voice_handler, + ) + + message = entry.update.message + if message is None: + return True + if message.voice: + result = await voice_handler(entry.update, entry.context) + elif message.photo: + result = await photo_handler(entry.update, entry.context) + elif message.document: + result = await document_handler(entry.update, entry.context) + elif message.text: + if message.text.startswith("/"): + result = await forward_command_handler(entry.update, entry.context) + else: + result = await text_handler(entry.update, entry.context) + else: + result = await unsupported_content_handler(entry.update, entry.context) + # Legacy handlers returned None on success. New delivery-aware paths + # return a bool; preserve compatibility while they are migrated. + return result is not False + + +async def _drain(user_id: int, window_id: str) -> None: + from .session import session_manager + + flow = _flows.get(user_id) + if flow is None: + return + try: + ready = await session_manager.wait_for_window_ready(window_id) + if not ready: + logger.error( + "startup queue kept closed: window never became ready " + "user=%d window=%s pending=%d", + user_id, + window_id, + len(flow.entries), + ) + return + while flow.entries: + entry = flow.entries[0] + try: + delivered = await _replay(entry) + except asyncio.CancelledError: + raise + except Exception as exc: + logger.exception( + "startup queue replay failed user=%d window=%s seq=%d: %s", + user_id, + window_id, + entry.sequence, + exc, + ) + return + if not delivered: + logger.error( + "startup queue delivery unconfirmed user=%d window=%s " + "seq=%d pending=%d", + user_id, + window_id, + entry.sequence, + len(flow.entries), + ) + return + flow.entries.popleft() + logger.info( + "startup queue delivered user=%d window=%s seq=%d remaining=%d", + user_id, + window_id, + entry.sequence, + len(flow.entries), + ) + # No await between the empty check and removal. A capture either + # appended before this point and was drained, or observes no flow and + # follows the now-active session's normal delivery path. + if _flows.get(user_id) is flow and not flow.entries: + _flows.pop(user_id, None) + logger.info("startup queue drained user=%d window=%s", user_id, window_id) + finally: + if _flows.get(user_id) is flow: + flow.drain_task = None + + +def bind_startup_queue(user_id: int, window_id: str) -> asyncio.Task[None] | None: + """Bind the current flow to its new window and start ordered draining.""" + flow = _flows.get(user_id) + if flow is None: + return None + flow.window_id = window_id + if flow.drain_task is not None and not flow.drain_task.done(): + return flow.drain_task + flow.drain_task = asyncio.create_task( + _drain(user_id, window_id), name=f"startup-queue:{user_id}:{window_id}" + ) + return flow.drain_task + + +def reset_startup_queues_for_test() -> None: + """Test-only cleanup for the module-global registry.""" + for flow in _flows.values(): + if flow.drain_task is not None and not flow.drain_task.done(): + flow.drain_task.cancel() + _flows.clear() + + +__all__ = [ + "begin_startup_queue", + "bind_startup_queue", + "cancel_startup_queue", + "capture_startup_message", + "enqueue_startup_message", + "has_startup_queue", + "pending_startup_count", +] diff --git a/src/ccbot/tmux_manager.py b/src/ccbot/tmux_manager.py index 591c8b64..5fe619f3 100644 --- a/src/ccbot/tmux_manager.py +++ b/src/ccbot/tmux_manager.py @@ -89,8 +89,51 @@ def _send_lock_for(self, window_id: str) -> asyncio.Lock: return lock @staticmethod - def _accept_codex_directory_trust(pane: object) -> bool: - """Handle only Codex's known directory startup prompts. + def _handle_codex_startup_screen(pane: object) -> tuple[bool, bool]: + """Handle one captured startup screen. + + Returns ``(acted, terminal)``. Terminal means the normal Codex input + is ready or the pane can no longer be inspected. + """ + capture = getattr(pane, "capture_pane", None) + send_keys = getattr(pane, "send_keys", None) + if not callable(capture) or not callable(send_keys): + return False, True + try: + lines = capture() + text = "\n".join(lines) if isinstance(lines, list) else str(lines) + except Exception: + return False, True + if _CODEX_TRUST_PROMPT in text and _CODEX_TRUST_YES in text: + send_keys("", enter=True) + logger.info("Accepted Codex directory trust prompt") + return True, False + if ( + "Choose working directory to resume this session" in text + and "1. Use session directory" in text + and "2. Use current directory" in text + and "Press enter to continue" in text + ): + send_keys("Down", enter=False) + send_keys("", enter=True) + logger.info("Selected current directory for Codex resume") + return True, False + if ( + "Update available!" in text + and "1. Update now" in text + and "2. Skip" in text + and "Press enter to continue" in text + ): + # Never mutate the host toolchain from a Telegram session. + send_keys("Down", enter=False) + send_keys("", enter=True) + logger.info("Skipped Codex CLI update prompt") + return True, False + return False, "OpenAI Codex" in text and "›" in text + + @classmethod + def _accept_codex_directory_trust(cls, pane: object) -> bool: + """Synchronously handle known prompts (small helper/test surface). Codex can show this before its normal input box even in full-access mode. The bot already owns the selected working directory, so leaving @@ -99,36 +142,27 @@ def _accept_codex_directory_trust(pane: object) -> bool: the current directory that the user selected for this bot session. Poll briefly because the Node wrapper needs a moment to draw prompts. """ - capture = getattr(pane, "capture_pane", None) - send_keys = getattr(pane, "send_keys", None) - if not callable(capture) or not callable(send_keys): - return False accepted = False for _ in range(30): time.sleep(0.15) - try: - lines = capture() - text = "\n".join(lines) if isinstance(lines, list) else str(lines) - except Exception: + acted, terminal = cls._handle_codex_startup_screen(pane) + accepted = accepted or acted + if terminal: return accepted - if _CODEX_TRUST_PROMPT in text and _CODEX_TRUST_YES in text: - send_keys("", enter=True) - logger.info("Accepted Codex directory trust prompt") - accepted = True - continue - if ( - "Choose working directory to resume this session" in text - and "1. Use session directory" in text - and "2. Use current directory" in text - and "Press enter to continue" in text - ): - send_keys("Down", enter=False) - send_keys("", enter=True) - logger.info("Selected current directory for Codex resume") - accepted = True - continue - # The regular input box is ready, so there is no trust prompt. - if "OpenAI Codex" in text and "›" in text: + return accepted + + @classmethod + async def _watch_codex_startup_screens(cls, pane: object) -> bool: + """Cancellation-safe long watcher for cold Codex launches.""" + accepted = False + attempts = max(30, int(max(config.resume_settle_timeout, 4.5) / 0.15)) + for _ in range(attempts): + await asyncio.sleep(0.15) + acted, terminal = await asyncio.to_thread( + cls._handle_codex_startup_screen, pane + ) + accepted = accepted or acted + if terminal: return accepted return accepted @@ -789,7 +823,7 @@ def _create_and_start() -> tuple[bool, str, str, str]: # draws its startup UI. The background task accepts only the two # known directory prompts; normal input is never confirmed. task = asyncio.create_task( - asyncio.to_thread(self._accept_codex_directory_trust, created_pane), + self._watch_codex_startup_screens(created_pane), name=f"codex-startup-trust:{result[3]}", ) self._startup_tasks.add(task) @@ -798,6 +832,8 @@ def _finish_startup_task(done: asyncio.Task[bool]) -> None: self._startup_tasks.discard(done) try: done.result() + except asyncio.CancelledError: + return except Exception as e: logger.warning("Codex startup prompt handler failed: %s", e) diff --git a/tests/ccbot/test_codex_backend.py b/tests/ccbot/test_codex_backend.py index 3ffdb62c..3872baa5 100644 --- a/tests/ccbot/test_codex_backend.py +++ b/tests/ccbot/test_codex_backend.py @@ -104,6 +104,37 @@ def send_keys(self, _value: str, enter: bool = True) -> None: assert TmuxManager._accept_codex_directory_trust(Pane()) is False +def test_codex_update_prompt_is_skipped(monkeypatch: pytest.MonkeyPatch) -> None: + screens = iter( + [ + [ + "Update available! 0.146.0 -> 0.146.1", + "› 1. Update now (runs npm install)", + " 2. Skip", + " 3. Skip until next version", + "Press enter to continue", + ], + ["OpenAI Codex", "›"], + ] + ) + + class Pane: + def __init__(self) -> None: + self.sent: list[tuple[str, bool]] = [] + + def capture_pane(self) -> list[str]: + return next(screens) + + def send_keys(self, value: str, enter: bool = True) -> None: + self.sent.append((value, enter)) + + pane = Pane() + monkeypatch.setattr("ccbot.tmux_manager.time.sleep", lambda _delay: None) + + assert TmuxManager._accept_codex_directory_trust(pane) is True + assert pane.sent == [("Down", False), ("", True)] + + def test_codex_resume_uses_selected_current_directory( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -610,12 +641,12 @@ def new_window(self, **_kwargs): mgr = TmuxManager() monkeypatch.setattr(mgr, "get_or_create_session", lambda: Session()) - def wait_for_trust(_pane: object) -> bool: + async def wait_for_trust(_pane: object) -> bool: trust_started.set() - release_trust.wait(timeout=2.0) + await asyncio.to_thread(release_trust.wait, 2.0) return True - monkeypatch.setattr(mgr, "_accept_codex_directory_trust", wait_for_trust) + monkeypatch.setattr(mgr, "_watch_codex_startup_screens", wait_for_trust) async def no_existing(_name: str): return None diff --git a/tests/ccbot/test_startup_queue.py b/tests/ccbot/test_startup_queue.py new file mode 100644 index 00000000..8d819c17 --- /dev/null +++ b/tests/ccbot/test_startup_queue.py @@ -0,0 +1,242 @@ +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from telegram.ext import ApplicationHandlerStop + +from ccbot.session import SessionManager +from ccbot.session_models import Session, WindowState +from ccbot.startup_queue import ( + _replay, + begin_startup_queue, + bind_startup_queue, + capture_startup_message, + enqueue_startup_message, + has_startup_queue, + pending_startup_count, + reset_startup_queues_for_test, +) + + +def _update( + seq: int, + *, + text: str | None = None, + voice: bool = False, + photo: bool = False, + document: bool = False, +) -> MagicMock: + message = SimpleNamespace( + message_id=seq, + text=text, + voice=object() if voice else None, + photo=[object()] if photo else [], + document=object() if document else None, + ) + update = MagicMock() + update.effective_user = SimpleNamespace(id=42) + update.message = message + return update + + +@pytest.fixture(autouse=True) +def _clean_queue() -> None: + reset_startup_queues_for_test() + yield + reset_startup_queues_for_test() + + +@pytest.mark.asyncio +async def test_capture_stops_old_session_routing_and_preserves_order() -> None: + context = MagicMock() + begin_startup_queue(42) + + with pytest.raises(ApplicationHandlerStop): + await capture_startup_message(_update(10, text="first"), context) + with pytest.raises(ApplicationHandlerStop): + await capture_startup_message(_update(11, voice=True), context) + + assert pending_startup_count(42) == 2 + + +@pytest.mark.asyncio +async def test_auth_code_bypasses_agent_startup_queue() -> None: + context = MagicMock() + begin_startup_queue(42) + with patch("ccbot.codex_auth.get_flow", return_value=object()): + await capture_startup_message(_update(10, text="one-time-code"), context) + assert pending_startup_count(42) == 0 + + +@pytest.mark.asyncio +async def test_new_command_can_retry_failed_creation_flow() -> None: + context = MagicMock() + begin_startup_queue(42) + await capture_startup_message(_update(10, text="/new retry /tmp"), context) + assert pending_startup_count(42) == 0 + + +@pytest.mark.asyncio +async def test_drain_includes_messages_arriving_while_window_becomes_ready() -> None: + context = MagicMock() + begin_startup_queue(42) + enqueue_startup_message(_update(1, text="first"), context) + + first_started = asyncio.Event() + release_first = asyncio.Event() + seen: list[int] = [] + + async def replay(entry) -> bool: + seen.append(entry.sequence) + if entry.sequence == 1: + first_started.set() + await release_first.wait() + return True + + with ( + patch( + "ccbot.session.session_manager.wait_for_window_ready", + new=AsyncMock(return_value=True), + ), + patch("ccbot.startup_queue._replay", side_effect=replay), + ): + task = bind_startup_queue(42, "@9") + assert task is not None + await first_started.wait() + enqueue_startup_message(_update(2, text="second"), context) + release_first.set() + await task + + assert seen == [1, 2] + assert not has_startup_queue(42) + + +@pytest.mark.asyncio +async def test_unconfirmed_head_is_retained_with_everything_behind_it() -> None: + context = MagicMock() + begin_startup_queue(42) + enqueue_startup_message(_update(1, text="first"), context) + enqueue_startup_message(_update(2, text="second"), context) + + with ( + patch( + "ccbot.session.session_manager.wait_for_window_ready", + new=AsyncMock(return_value=True), + ), + patch("ccbot.startup_queue._replay", new=AsyncMock(return_value=False)), + ): + task = bind_startup_queue(42, "@9") + assert task is not None + await task + + assert has_startup_queue(42) + assert pending_startup_count(42) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("update", "handler"), + [ + (_update(1, text="hello"), "text_handler"), + (_update(2, text="/compact"), "forward_command_handler"), + (_update(3, voice=True), "voice_handler"), + (_update(4, photo=True), "photo_handler"), + (_update(5, document=True), "document_handler"), + (_update(6), "unsupported_content_handler"), + ], +) +async def test_replay_covers_every_inbound_kind(update, handler: str) -> None: + context = MagicMock() + entry = SimpleNamespace(update=update, context=context, sequence=1) + target = AsyncMock(return_value=True) + with patch(f"ccbot.bot.messages.{handler}", new=target): + assert await _replay(entry) + target.assert_awaited_once_with(update, context) + + +def test_shell_prompt_is_not_codex_readiness() -> None: + assert not SessionManager._pane_has_ready_input("zsh\n› codex", "codex") + assert SessionManager._pane_has_ready_input( + "│ >_ OpenAI Codex (v0.146.0) │\n\n› Ask anything", "codex" + ) + + +@pytest.mark.asyncio +async def test_session_map_poll_keeps_fresh_bound_window_state( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + from ccbot.config import config + + monkeypatch.setattr(SessionManager, "_load_state", lambda self: None) + monkeypatch.setattr(SessionManager, "save_state", lambda self: None) + session_map = tmp_path / "session_map.json" + session_map.write_text("{}") + monkeypatch.setattr(config, "session_map_file", session_map) + manager = SessionManager() + manager.window_states["@9"] = WindowState(backend="codex", cwd=str(tmp_path)) + manager.sessions["fresh"] = Session( + id="fresh", name="fresh", window_id="@9", backend="codex" + ) + + await manager.load_session_map() + + assert "@9" in manager.window_states + + +@pytest.mark.asyncio +async def test_first_turn_can_be_confirmed_after_binding_appears(tmp_path: Path) -> None: + from ccbot.bot.messages import _wait_for_voice_transcript + + transcript = tmp_path / "rollout.jsonl" + transcript.write_text( + json.dumps( + { + "type": "event_msg", + "payload": {"type": "user_message", "message": "first prompt"}, + } + ) + + "\n" + ) + state = WindowState( + session_id="sid", + cwd=str(tmp_path), + window_name="new", + backend="codex", + transcript_path=str(transcript), + ) + fake_manager = MagicMock() + fake_manager.window_states = {"@9": state} + fake_manager.load_session_map = AsyncMock() + + with patch("ccbot.bot.messages.session_manager", fake_manager): + assert await _wait_for_voice_transcript(None, "first prompt", wid="@9") + + +@pytest.mark.asyncio +async def test_delivery_retries_until_exact_transcript_ack() -> None: + from ccbot.bot.messages import _send_with_delivery_proof + + fake_manager = MagicMock() + fake_manager.send_to_window = AsyncMock(return_value=(True, "Sent")) + fake_session = SimpleNamespace(backend="codex") + with ( + patch("ccbot.bot.messages.session_manager", fake_manager), + patch( + "ccbot.bot.messages.tmux_manager.ensure_codex_prompt_submitted", + new=AsyncMock(return_value=True), + ), + patch( + "ccbot.bot.messages._wait_for_voice_transcript", + new=AsyncMock(side_effect=[False, True]), + ), + patch("ccbot.bot.messages._voice_transcript_checkpoint", return_value=None), + ): + ok, _ = await _send_with_delivery_proof("@9", "do it", fake_session) + + assert ok + assert fake_manager.send_to_window.await_count == 2 diff --git a/tests/e2e/test_inbound_routing.py b/tests/e2e/test_inbound_routing.py index 3ca555ec..4facd51d 100644 --- a/tests/e2e/test_inbound_routing.py +++ b/tests/e2e/test_inbound_routing.py @@ -74,7 +74,7 @@ async def test_text_routes_to_active_session_window(fake_tmux, fake_bot): @pytest.mark.asyncio async def test_text_with_no_active_session_opens_dir_browser(fake_tmux, fake_bot): # No active session at all → handler must NOT send_keys; it opens the - # directory browser instead (pending text stashed in user_data). + # directory browser instead (pending update retained in startup queue). user = FakeUser(USER_ID) msg = FakeReplyMessage( message_id=5002, chat_id=USER_ID, bot=fake_bot, text="hello there" @@ -85,11 +85,10 @@ async def test_text_with_no_active_session_opens_dir_browser(fake_tmux, fake_bot await text_handler(update, ctx) assert fake_tmux.sent == [] - # The pending text is held (timestamped) for forwarding after session - # creation; ``take_pending_text`` reads it back with a freshness guard. - from ccbot.handlers.directory_browser import take_pending_text + from ccbot.startup_queue import cancel_startup_queue, pending_startup_count - assert take_pending_text(ctx.user_data) == "hello there" + assert pending_startup_count(USER_ID) == 1 + assert cancel_startup_queue(USER_ID) == 1 @pytest.mark.asyncio From c80269577026d33784ac760882821aaf96473c36 Mon Sep 17 00:00:00 2001 From: Nosko Artem Date: Thu, 6 Aug 2026 12:05:20 +0300 Subject: [PATCH 2/2] style: format startup delivery changes --- src/ccbot/bot/app.py | 4 +++- src/ccbot/bot/messages.py | 10 ++-------- src/ccbot/session.py | 4 +++- tests/ccbot/test_startup_queue.py | 4 +++- 4 files changed, 11 insertions(+), 11 deletions(-) diff --git a/src/ccbot/bot/app.py b/src/ccbot/bot/app.py index 7616e25d..4a6e4a17 100644 --- a/src/ccbot/bot/app.py +++ b/src/ccbot/bot/app.py @@ -522,7 +522,9 @@ def create_bot() -> "Application[Any, Any, Any, Any, Any, Any]": # a new-session flow is open; while open it captures the update and stops # it from leaking to the previously-active session. application.add_handler( - MessageHandler(filters.ALL & ~filters.StatusUpdate.ALL, capture_startup_message), + MessageHandler( + filters.ALL & ~filters.StatusUpdate.ALL, capture_startup_message + ), group=-1, ) diff --git a/src/ccbot/bot/messages.py b/src/ccbot/bot/messages.py index b8b0ec67..cdd30916 100644 --- a/src/ccbot/bot/messages.py +++ b/src/ccbot/bot/messages.py @@ -198,11 +198,7 @@ async def _wait_for_voice_transcript( await session_manager.load_session_map() state = session_manager.window_states.get(wid) if isinstance(state, WindowState) and state.session_id: - path = ( - Path(state.transcript_path) - if state.transcript_path - else None - ) + path = Path(state.transcript_path) if state.transcript_path else None if path is None or not path.is_file(): if state.backend == "codex": from ..codex_session_io import build_session_file_path @@ -669,9 +665,7 @@ async def unsupported_content_handler( return False sess = session_manager.find_session_by_window(wid) async with _card_repost_bracket(context.bot, user.id, sess) as repost: - success, message = await _send_with_delivery_proof( - wid, text_to_send, sess - ) + success, message = await _send_with_delivery_proof(wid, text_to_send, sess) if not success: await safe_reply(msg, f"❌ {message}") return False diff --git a/src/ccbot/session.py b/src/ccbot/session.py index 97914589..335032b9 100644 --- a/src/ccbot/session.py +++ b/src/ccbot/session.py @@ -429,7 +429,9 @@ async def _load_session_map_unlocked(self) -> None: # is accepted. Keep provisional state for every bot Session still # bound to a window; deleting it here removed the transcript binding # and made first-turn delivery impossible to prove. - bound_wids = {sess.window_id for sess in self.sessions.values() if sess.window_id} + bound_wids = { + sess.window_id for sess in self.sessions.values() if sess.window_id + } stale_wids = [ w for w in self.window_states diff --git a/tests/ccbot/test_startup_queue.py b/tests/ccbot/test_startup_queue.py index 8d819c17..8aef6415 100644 --- a/tests/ccbot/test_startup_queue.py +++ b/tests/ccbot/test_startup_queue.py @@ -189,7 +189,9 @@ async def test_session_map_poll_keeps_fresh_bound_window_state( @pytest.mark.asyncio -async def test_first_turn_can_be_confirmed_after_binding_appears(tmp_path: Path) -> None: +async def test_first_turn_can_be_confirmed_after_binding_appears( + tmp_path: Path, +) -> None: from ccbot.bot.messages import _wait_for_voice_transcript transcript = tmp_path / "rollout.jsonl"