diff --git a/src/ccbot/bot/_session_create.py b/src/ccbot/bot/_session_create.py index c7eb536b..32963178 100644 --- a/src/ccbot/bot/_session_create.py +++ b/src/ccbot/bot/_session_create.py @@ -91,64 +91,28 @@ async def create_and_activate_session( user.id, resume_session_id, ) - # `claude --resume` records a new session_id in the hook, but messages - # still write to the resumed JSONL. The card seeds from that existing - # transcript, so for a resume we must resolve the canonical session_id - # BEFORE painting — wait for the hook (or fall back to the known resume - # id on timeout), then override window_state to track it. - # - # A fresh session has nothing to seed, so this pre-paint wait is - # skipped entirely (see below): the empty card goes up the instant the - # window exists instead of blocking on claude's 2-5s boot + the - # SessionStart hook. The hook is confirmed (and the fresh session_id - # bound) after the paint, before any pending text is forwarded. + # Publish the session immediately, while the agent process boots in the + # pane. Every send is queued until the real TUI input prompt appears. + # This covers fresh starts, normal resumes, and long resume compaction + # with one ordering-preserving gate. + session_manager.mark_window_starting( + created_wid, + backend=session_manager.agent_backend, + resume=resume_session_id is not None, + bot=context.bot, + user_id=user.id, + ) + + # A resumed transcript id is already authoritative. Bind it before paint + # instead of waiting up to 15 seconds for a lifecycle hook; the hook is + # reconciled in the background below. if resume_session_id: - # A near-limit transcript auto-compacts on resume (60-110s); flag - # the window so any prompt that arrives while we're still - # compacting buffers into _pending_sends instead of being typed - # mid-compaction. The background watcher drains the buffer after - # the pane settles AND refreshes Telegram TYPING in the meantime - # so the chat doesn't look frozen. - if session_manager.agent_backend == "claude": - session_manager.mark_window_resuming( - created_wid, bot=context.bot, user_id=user.id - ) ws = session_manager.get_window_state(created_wid) - if session_manager.agent_backend == "codex": - # ``codex resume`` preserves the rollout id, so binding it is - # deterministic and must not wait for a SessionStart hook that can - # arrive only after the CLI finishes booting. - ws.session_id = resume_session_id - ws.cwd = str(selected_path) - ws.window_name = created_wname - ws.backend = "codex" - session_manager.save_state() - hook_ok = True - else: - hook_ok = await session_manager.wait_for_session_map_entry( - created_wid, timeout=15.0 - ) - if not hook_ok: - logger.warning( - "Hook timed out for resume window %s, " - "manually setting session_id=%s cwd=%s", - created_wid, - resume_session_id, - selected_path, - ) - ws.session_id = resume_session_id - ws.cwd = str(selected_path) - ws.window_name = created_wname - session_manager.save_state() - elif ws.session_id != resume_session_id: - logger.info( - "Resume override: window %s session_id %s -> %s", - created_wid, - ws.session_id, - resume_session_id, - ) - ws.session_id = resume_session_id - session_manager.save_state() + ws.session_id = resume_session_id + ws.cwd = str(selected_path) + ws.window_name = created_wname + ws.backend = session_manager.agent_backend + session_manager.save_state() # Register Session record and make it active. Honor /new if any. pending_name = ( @@ -164,9 +128,6 @@ 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) - if session_manager.get_user_settings(user.id).get("local_terminal") == "auto": - await open_terminal_for_window(created_wid, user_id=user.id) - # 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 @@ -183,16 +144,40 @@ async def create_and_activate_session( # the stale dir-browser body when paint fails. await safe_edit(query, f"✅ {message}") - # Fresh session: claude is still booting, so the hook hasn't written - # the session_id yet. Confirm it now (card already on screen) and bind - # it onto the Session record so the monitor + history follow the right - # transcript and notifications reverse-map to this user — all before - # any pending text is forwarded below. - if not resume_session_id: - await session_manager.wait_for_session_map_entry(created_wid, timeout=5.0) - ws = session_manager.get_window_state(created_wid) - if ws.session_id and not sess.claude_session_id: - session_manager.set_session_claude_id(sess.id, ws.session_id) + async def _bind_lifecycle_in_background() -> None: + """Attach the hook-written session id without delaying Telegram UI.""" + try: + await session_manager.wait_for_session_map_entry(created_wid, timeout=15.0) + live_ws = session_manager.get_window_state(created_wid) + if resume_session_id: + # Claude may expose a transient new id for ``--resume``; + # messages still belong to the requested transcript. + if live_ws.session_id != resume_session_id: + live_ws.session_id = resume_session_id + live_ws.cwd = str(selected_path) + live_ws.window_name = created_wname + live_ws.backend = session_manager.agent_backend + session_manager.save_state() + elif live_ws.session_id and not sess.claude_session_id: + session_manager.set_session_claude_id(sess.id, live_ws.session_id) + except Exception as e: + logger.warning( + "Background lifecycle bind failed for window %s: %s", + created_wid, + e, + ) + + asyncio.create_task( + _bind_lifecycle_in_background(), name=f"session-bind:{created_wid}" + ) + + # Desktop Terminal is a convenience side-effect, never part of the + # session-start critical path. + if session_manager.get_user_settings(user.id).get("local_terminal") == "auto": + asyncio.create_task( + 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 diff --git a/src/ccbot/bot/messages.py b/src/ccbot/bot/messages.py index ea246193..65a8f820 100644 --- a/src/ccbot/bot/messages.py +++ b/src/ccbot/bot/messages.py @@ -1280,9 +1280,11 @@ async def _dispatch_text_to_active( 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") diff --git a/src/ccbot/handlers/archive.py b/src/ccbot/handlers/archive.py index 8edb37b5..402128f4 100644 --- a/src/ccbot/handlers/archive.py +++ b/src/ccbot/handlers/archive.py @@ -12,6 +12,7 @@ from __future__ import annotations +import asyncio import json import logging import re @@ -459,8 +460,6 @@ async def restore_session(bot: Bot, user_id: int, sess: Session) -> tuple[bool, resume_session_id = sess.claude_session_id or None initial_prompt: str | None = None if cross_backend: - import asyncio - from ..session_import import build_import_context, import_prompt try: @@ -482,29 +481,30 @@ async def restore_session(bot: Bot, user_id: int, sess: Session) -> tuple[bool, if not success: return False, message - # A near-limit transcript auto-compacts on resume (60-110s) before it - # accepts input. Flag the window so any prompts that arrive while - # we're still compacting buffer into _pending_sends instead of being - # typed mid-compaction. The background watcher drains the buffer - # once the pane settles AND keeps Telegram TYPING refreshed so the - # chat doesn't look frozen during the wait. - if resume_session_id and target_backend == "claude": - session_manager.mark_window_resuming(created_wid, bot=bot, user_id=user_id) + # Publish the restored window immediately. Prompts sent from Telegram now + # queue until the real TUI input box appears (including long compaction). + session_manager.mark_window_starting( + created_wid, + backend=target_backend, + resume=resume_session_id is not None or initial_prompt is not None, + bot=bot, + user_id=user_id, + ) # Codex ``resume `` keeps the same authoritative rollout id. We # already know everything needed to bind the window, while its SessionStart # hook may not run until the CLI has finished booting. Waiting 15 seconds # here made a normal archive restore look frozen for exactly that long. # Bind Codex immediately; the hook will later add transcript_path and - # self-heal the persisted map. Claude resume remains on the old wait path - # because Claude can report a transient new session id before we override - # it back to the resumed transcript id. + # self-heal the persisted map. Claude's original id is also known, so it + # can be published immediately and reconciled after the hook in background. codex_restore_published = False if resume_session_id and target_backend == "codex": from ..codex_session_io import build_session_file_path transcript_path = build_session_file_path(resume_session_id, workdir) if transcript_path is None or not transcript_path.is_file(): + session_manager.cancel_window_startup(created_wid) await tmux_manager.kill_window(created_wid) return False, "Codex rollout not found; restore was cancelled" try: @@ -516,11 +516,12 @@ async def restore_session(bot: Bot, user_id: int, sess: Session) -> tuple[bool, transcript_path=transcript_path, ) except (OSError, RuntimeError) as e: + session_manager.cancel_window_startup(created_wid) await tmux_manager.kill_window(created_wid) logger.warning("Codex restore binding failed for %s: %s", created_wid, e) return False, "Could not publish Codex restore binding" codex_restore_published = True - else: + elif cross_backend: await session_manager.wait_for_session_map_entry(created_wid, timeout=15.0) # If we did a --resume, override window_state to original sid (Claude allocates a new sid for the resume). @@ -530,10 +531,12 @@ async def restore_session(bot: Bot, user_id: int, sess: Session) -> tuple[bool, ws.session_id = resume_session_id ws.cwd = workdir ws.window_name = created_wname + ws.backend = target_backend session_manager.save_state() elif cross_backend: ws = session_manager.get_window_state(created_wid) if not ws.session_id: + session_manager.cancel_window_startup(created_wid) await tmux_manager.kill_window(created_wid) return ( False, @@ -549,10 +552,39 @@ async def restore_session(bot: Bot, user_id: int, sess: Session) -> tuple[bool, if not codex_restore_published: session_manager.set_session_window(sess.id, created_wid) session_manager.set_active_session(user_id, sess.id) + + if resume_session_id and not codex_restore_published: + + async def _reconcile_resume_binding() -> None: + try: + await session_manager.wait_for_session_map_entry( + created_wid, timeout=15.0 + ) + live_ws = session_manager.get_window_state(created_wid) + if live_ws.session_id != resume_session_id: + live_ws.session_id = resume_session_id + live_ws.cwd = workdir + live_ws.window_name = created_wname + live_ws.backend = target_backend + session_manager.save_state() + except Exception as e: + logger.warning( + "Background archive binding failed for %s: %s", + created_wid, + e, + ) + + asyncio.create_task( + _reconcile_resume_binding(), name=f"archive-bind:{created_wid}" + ) + if session_manager.get_user_settings(user_id).get("local_terminal") == "auto": from ..local_terminal import open_terminal_for_window - 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}", + ) note = "" if resume_session_id: note = " — if it was a large session it may compact for a minute; your first message is held until it's ready." diff --git a/src/ccbot/session.py b/src/ccbot/session.py index 79a1349d..b3265f56 100644 --- a/src/ccbot/session.py +++ b/src/ccbot/session.py @@ -37,7 +37,7 @@ from .config import config from .session_models import ClaudeSession, Session, SessionState, WindowState -from .terminal_parser import parse_status_line +from .terminal_parser import is_interactive_ui, parse_status_line from .tmux_manager import tmux_manager from .transcript_parser import TranscriptParser from .utils import atomic_write_json @@ -1070,6 +1070,30 @@ def mark_window_resuming( supplied, the watcher also keeps Telegram's TYPING indicator alive so the chat doesn't look frozen during a long compaction. """ + self.mark_window_starting( + window_id, + backend=self.agent_backend, + resume=True, + bot=bot, + user_id=user_id, + ) + + def mark_window_starting( + self, + window_id: str, + *, + backend: str, + resume: bool, + bot: "Bot | None" = None, + user_id: int | None = None, + ) -> None: + """Gate sends until a newly-created agent pane can accept input. + + The Session record and Telegram card may be published immediately; + ``send_to_window`` queues prompts in arrival order while this watcher + waits for the real TUI input prompt. ``resume=True`` preserves the + extra Claude compaction settle window. + """ if config.resume_settle_timeout <= 0: return if window_id in self._resuming_windows: @@ -1083,8 +1107,10 @@ def mark_window_resuming( return self._resuming_windows.add(window_id) self._resume_settle_tasks[window_id] = loop.create_task( - self._watch_resume_settle(window_id, bot, user_id), - name=f"resume-settle:{window_id}", + self._watch_resume_settle( + window_id, bot, user_id, backend=backend, resume=resume + ), + name=f"startup-ready:{window_id}", ) async def _watch_resume_settle( @@ -1092,6 +1118,9 @@ async def _watch_resume_settle( window_id: str, bot: "Bot | None", user_id: int | None, + *, + backend: str, + resume: bool, ) -> None: """Background watcher for a resuming window. @@ -1128,29 +1157,49 @@ async def _typing_keepalive() -> None: _typing_keepalive(), name=f"resume-settle-typing:{window_id}" ) try: - settled = await self._wait_for_resume_settle(window_id) + settled = await self._wait_for_resume_settle( + window_id, backend=backend, resume=resume + ) logger.info( - "resume-settle gate cleared for window %s (settled=%s, background)", + "startup gate cleared for window %s " + "(settled=%s backend=%s resume=%s, background)", window_id, settled, + backend, + resume, ) - pending = self._pending_sends.pop(window_id, []) - for i, text in enumerate(pending): - ok = await tmux_manager.send_keys(window_id, text) - if not ok: - logger.warning( - "resume-settle: failed to drain pending send #%d " - "for window %s (text_len=%d)", - i, - window_id, - len(text), - ) - if i < len(pending) - 1: + drained = 0 + while True: + pending = self._pending_sends.pop(window_id, []) + if not pending: + # No await between the final empty check and clearing the + # gate: a concurrent send either joined the batch above or + # observes the cleared gate and sends normally. + self._resuming_windows.discard(window_id) + break + for text in pending: + ok = await tmux_manager.send_keys(window_id, text) + if ok and backend == "codex": + ok = await tmux_manager.ensure_codex_prompt_submitted( + window_id, text + ) + if not ok: + logger.warning( + "startup gate: failed to drain pending send #%d " + "for window %s (text_len=%d)", + drained, + window_id, + len(text), + ) + drained += 1 + # Give the TUI one render tick before submitting another + # queued prompt. A new arrival during this sleep is picked + # up by the next outer-loop batch. await asyncio.sleep(_RESUME_SETTLE_DRAIN_GAP) - if pending: + if drained: logger.info( - "resume-settle: drained %d pending send(s) for window %s", - len(pending), + "startup gate: drained %d pending send(s) for window %s", + drained, window_id, ) except Exception as e: @@ -1169,7 +1218,35 @@ async def _typing_keepalive() -> None: # that didn't get drained so they can't leak forever. self._pending_sends.pop(window_id, None) - async def _wait_for_resume_settle(self, window_id: str) -> bool: + @staticmethod + def _pane_has_ready_input(pane: str, backend: str) -> bool: + """Whether the visible pane ends in the agent's real input box.""" + if not pane or is_interactive_ui(pane) or parse_status_line(pane) is not None: + return False + lower = pane.lower() + if backend == "codex" and ( + "do you trust the contents of this directory?" in lower + or "choose working directory to resume this session" in lower + or "sign in with chatgpt" in lower + or "sign in with device code" in lower + or "provide your own api key" in lower + ): + return False + marker = "›" if backend == "codex" else "❯" + # 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. + return any( + line.lstrip().startswith(marker) for line in pane.strip().splitlines()[-6:] + ) + + async def _wait_for_resume_settle( + self, + window_id: str, + *, + backend: str = "claude", + resume: bool = True, + ) -> bool: """Block until a just-resumed window is safe to type into. A ``claude --resume`` of a near-limit transcript auto-compacts before @@ -1194,10 +1271,21 @@ async def _wait_for_resume_settle(self, window_id: str) -> bool: pane = await tmux_manager.capture_pane(window_id) now = loop.time() busy = bool(pane) and parse_status_line(pane) is not None + ready = bool(pane) and self._pane_has_ready_input(pane or "", backend) if busy: saw_busy = True idle_since = None else: + # Fresh sessions and Codex resumes are ready the moment the + # actual input box appears. Claude resume keeps the historical + # grace/stability rule because compaction may start shortly + # after an initially-idle frame. + if ready and (not resume or backend == "codex"): + return True + if not ready: + idle_since = None + await asyncio.sleep(_RESUME_SETTLE_POLL) + continue if idle_since is None: idle_since = now if saw_busy and (now - idle_since) >= _RESUME_SETTLE_IDLE_STABLE: @@ -1214,14 +1302,22 @@ async def _wait_for_resume_settle(self, window_id: str) -> bool: ) return False + def cancel_window_startup(self, window_id: str) -> None: + """Cancel a readiness gate after its tmux window was rolled back.""" + task = self._resume_settle_tasks.pop(window_id, None) + if task is not None and not task.done(): + task.cancel() + self._resuming_windows.discard(window_id) + self._pending_sends.pop(window_id, None) + async def send_to_window(self, window_id: str, text: str) -> tuple[bool, str]: """Send text to a tmux window by ID. - For windows mid-resume (``mark_window_resuming`` was called and - the background watcher hasn't settled yet), the text is buffered - into ``_pending_sends`` and we return success immediately — the - watcher drains the buffer when the pane is ready. This keeps the - message handler off the hot path of a 60-200s compaction wait. + For newly-created windows whose TUI is still booting, the text is + buffered into ``_pending_sends`` and success is returned immediately. + The background watcher drains the buffer in arrival order once the + real input prompt appears. This covers both ordinary startup and a + long resume compaction without holding the Telegram handler open. """ display = self.get_display_name(window_id) logger.debug( @@ -1238,12 +1334,12 @@ async def send_to_window(self, window_id: str, text: str) -> tuple[bool, str]: queue.append(text) logger.info( "send_to_window buffered: window=%s pending=%d text_len=%d " - "(resume in progress)", + "(startup in progress)", window_id, len(queue), len(text), ) - return True, f"Queued for {display} (session restoring)" + return True, f"Queued for {display} (session starting)" success = await tmux_manager.send_keys(window.window_id, text) if success: return True, f"Sent to {display}" diff --git a/src/ccbot/tmux_manager.py b/src/ccbot/tmux_manager.py index d2d73447..591c8b64 100644 --- a/src/ccbot/tmux_manager.py +++ b/src/ccbot/tmux_manager.py @@ -69,6 +69,11 @@ def __init__(self, session_name: str | None = None): # both messages into one and firing a spurious Enter. See # send_keys. self._send_locks: dict[str, asyncio.Lock] = {} + # Codex directory-trust handling used to block ``create_window`` for + # up to 4.5 seconds. Keep the pollers alive in the background instead; + # session readiness/queued input is handled independently by + # SessionManager's startup gate. + self._startup_tasks: set[asyncio.Task[bool]] = set() def _send_lock_for(self, window_id: str) -> asyncio.Lock: """Return the per-window send lock, creating it on first use. @@ -698,7 +703,10 @@ async def create_window( counter += 1 # Create window in thread + created_pane: object | None = None + def _create_and_start() -> tuple[bool, str, str, str]: + nonlocal created_pane session = self.get_or_create_session() try: # Create new window @@ -716,6 +724,7 @@ def _create_and_start() -> tuple[bool, str, str, str]: if start_claude: pane = window.active_pane if pane: + created_pane = pane if selected_backend == "codex": cmd = config.codex_command if config.codex_flags: @@ -756,8 +765,6 @@ def _create_and_start() -> tuple[bool, str, str, str]: else: cmd = f"{env_prefix} {cmd}" pane.send_keys(cmd, enter=True) - if selected_backend == "codex": - self._accept_codex_directory_trust(pane) logger.info( "Created window '%s' (id=%s) at %s", @@ -776,7 +783,26 @@ def _create_and_start() -> tuple[bool, str, str, str]: logger.error(f"Failed to create window: {e}") return False, f"Failed to create window: {e}", "", "" - return await asyncio.to_thread(_create_and_start) + result = await asyncio.to_thread(_create_and_start) + if result[0] and selected_backend == "codex" and created_pane is not None: + # Do not hold the Telegram callback open while the Node wrapper + # 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), + name=f"codex-startup-trust:{result[3]}", + ) + self._startup_tasks.add(task) + + def _finish_startup_task(done: asyncio.Task[bool]) -> None: + self._startup_tasks.discard(done) + try: + done.result() + except Exception as e: + logger.warning("Codex startup prompt handler failed: %s", e) + + task.add_done_callback(_finish_startup_task) + return result # Global instance with default session name diff --git a/tests/ccbot/test_card_in_front.py b/tests/ccbot/test_card_in_front.py index ec598276..54a86ff8 100644 --- a/tests/ccbot/test_card_in_front.py +++ b/tests/ccbot/test_card_in_front.py @@ -180,6 +180,40 @@ async def _send_and_switch(*args, **kwargs): class TestActiveDispatchPutsCardInFront: + @pytest.mark.asyncio + async def test_codex_startup_queue_skips_early_submit_check(self): + update = _make_update(message_id=500) + context = _make_context() + active = MagicMock(id="sessA", backend="codex") + mock_sm = MagicMock() + mock_sm.find_session_by_window.return_value = active + mock_sm.send_to_window = AsyncMock( + return_value=(True, "Queued for project (session starting)") + ) + ensure = AsyncMock(return_value=False) + + with ( + patch("ccbot.bot.messages.session_manager", mock_sm), + patch("ccbot.bot.messages.is_active_for_user", return_value=True), + patch("ccbot.bot.messages.card_is_below", return_value=True), + patch("ccbot.bot.messages.repost_card", new=AsyncMock()), + patch("ccbot.bot.messages.resume_card_view", new=AsyncMock()), + patch("ccbot.bot.messages.fire_typing", new=AsyncMock()), + patch("ccbot.bot.messages.get_interactive_window", return_value=None), + patch( + "ccbot.bot.messages.tmux_manager.ensure_codex_prompt_submitted", + new=ensure, + ), + ): + from ccbot.bot.messages import _dispatch_text_to_active + + ok = await _dispatch_text_to_active( + update, context, 1, "@5", "sent during startup" + ) + + assert ok is True + ensure.assert_not_awaited() + @pytest.mark.asyncio async def test_card_above_user_message_is_reposted(self): update = _make_update(message_id=500) diff --git a/tests/ccbot/test_codex_backend.py b/tests/ccbot/test_codex_backend.py index 314bb7fa..3ffdb62c 100644 --- a/tests/ccbot/test_codex_backend.py +++ b/tests/ccbot/test_codex_backend.py @@ -4,9 +4,10 @@ import asyncio import json +import threading from pathlib import Path from types import SimpleNamespace -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import ANY, AsyncMock, MagicMock import pytest @@ -200,6 +201,7 @@ async def test_restore_claude_archive_imports_into_codex( monkeypatch.setattr(mgr, "save_state", lambda: None) mgr.agent_backend = "codex" mgr.wait_for_session_map_entry = AsyncMock(return_value=True) # type: ignore[method-assign] + mgr.mark_window_starting = MagicMock() # type: ignore[method-assign] monkeypatch.setattr(archive, "session_manager", mgr) monkeypatch.setattr(config, "config_dir", tmp_path / "ccbot") monkeypatch.setattr(config, "claude_projects_path", tmp_path / "claude-projects") @@ -265,6 +267,7 @@ async def test_restore_codex_archive_does_not_wait_for_hook( monkeypatch.setattr(mgr, "save_state", lambda: None) mgr.agent_backend = "codex" mgr.wait_for_session_map_entry = AsyncMock(return_value=False) # type: ignore[method-assign] + mgr.mark_window_starting = MagicMock() # type: ignore[method-assign] monkeypatch.setattr(archive, "session_manager", mgr) sid = "550e8400-e29b-41d4-a716-446655440000" @@ -298,6 +301,9 @@ async def test_restore_codex_archive_does_not_wait_for_hook( ok, _message = await archive.restore_session(MagicMock(), 42, sess) assert ok is True + mgr.mark_window_starting.assert_called_once_with( # type: ignore[attr-defined] + "@9", backend="codex", resume=True, bot=ANY, user_id=42 + ) mgr.wait_for_session_map_entry.assert_not_awaited() # type: ignore[attr-defined] ws = mgr.get_window_state("@9") assert ws.session_id == sid @@ -582,6 +588,8 @@ async def test_tmux_builds_codex_resume_command( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: sent: list[str] = [] + trust_started = threading.Event() + release_trust = threading.Event() class Pane: def send_keys(self, value: str, enter: bool = True) -> None: @@ -602,6 +610,13 @@ def new_window(self, **_kwargs): mgr = TmuxManager() monkeypatch.setattr(mgr, "get_or_create_session", lambda: Session()) + def wait_for_trust(_pane: object) -> bool: + trust_started.set() + release_trust.wait(timeout=2.0) + return True + + monkeypatch.setattr(mgr, "_accept_codex_directory_trust", wait_for_trust) + async def no_existing(_name: str): return None @@ -629,3 +644,11 @@ async def no_existing(_name: str): assert "/data/data/com.termux/files/usr/bin/codex" in sent[0] assert " resume 550e8400-e29b-41d4-a716-446655440000" in sent[0] assert "--resume" not in sent[0] + # create_window returned even though its startup-prompt worker is still + # blocked. Trust handling must never hold the Telegram callback open. + assert await asyncio.to_thread(trust_started.wait, 1.0) + startup_tasks = tuple(mgr._startup_tasks) + assert len(startup_tasks) == 1 + assert not startup_tasks[0].done() + release_trust.set() + await asyncio.gather(*startup_tasks) diff --git a/tests/ccbot/test_session.py b/tests/ccbot/test_session.py index f61a31cc..70bf74ef 100644 --- a/tests/ccbot/test_session.py +++ b/tests/ccbot/test_session.py @@ -184,6 +184,8 @@ def test_no_active_session_initially(self, mgr: SessionManager) -> None: _BUSY_PANE = "✻ Compacting conversation…\n" + "─" * 26 + "\n❯\n" + "─" * 26 # A settled pane: input chrome only, no spinner line. _IDLE_PANE = "─" * 26 + "\n❯\n" + "─" * 26 +_CODEX_STARTING_PANE = "Starting Codex…" +_CODEX_READY_PANE = "OpenAI Codex\n\n› Write tests for @filename" class TestResumeSettleGate: @@ -205,9 +207,70 @@ def _mock_tmux(self, monkeypatch, capture_side_effect) -> MagicMock: mock_tmux.find_window_by_id = AsyncMock(return_value=MagicMock(window_id="@1")) mock_tmux.capture_pane = AsyncMock(side_effect=capture_side_effect) mock_tmux.send_keys = AsyncMock(return_value=True) + mock_tmux.ensure_codex_prompt_submitted = AsyncMock(return_value=True) monkeypatch.setattr("ccbot.session.tmux_manager", mock_tmux) return mock_tmux + @pytest.mark.asyncio + async def test_fresh_codex_start_queues_until_real_prompt( + self, mgr: SessionManager, monkeypatch, fast_gate + ) -> None: + """A message sent immediately after window creation is held until + Codex has drawn its input box, then submitted exactly once.""" + panes = iter([_CODEX_STARTING_PANE, _CODEX_READY_PANE]) + mock_tmux = self._mock_tmux(monkeypatch, lambda _w: next(panes)) + mgr.mark_window_starting("@1", backend="codex", resume=False) + + ok, message = await mgr.send_to_window("@1", "fix startup") + + assert ok is True + assert message.startswith("Queued for ") + mock_tmux.send_keys.assert_not_awaited() + task = mgr._resume_settle_tasks["@1"] + await task + mock_tmux.send_keys.assert_awaited_once_with("@1", "fix startup") + mock_tmux.ensure_codex_prompt_submitted.assert_awaited_once_with( + "@1", "fix startup" + ) + + @pytest.mark.asyncio + async def test_message_arriving_during_drain_is_not_lost( + self, mgr: SessionManager, monkeypatch, fast_gate + ) -> None: + """The gate keeps draining batches that arrive while a prior queued + prompt is being submitted.""" + mock_tmux = self._mock_tmux(monkeypatch, lambda _w: _CODEX_READY_PANE) + second_queued = False + + async def submit(_wid: str, text: str) -> bool: + nonlocal second_queued + if text == "first" and not second_queued: + second_queued = True + ok, _ = await mgr.send_to_window("@1", "second") + assert ok is True + return True + + mock_tmux.ensure_codex_prompt_submitted.side_effect = submit + mgr.mark_window_starting("@1", backend="codex", resume=False) + ok, _ = await mgr.send_to_window("@1", "first") + assert ok is True + + await mgr._resume_settle_tasks["@1"] + + sent = [call.args[1] for call in mock_tmux.send_keys.await_args_list] + assert sent == ["first", "second"] + + @pytest.mark.parametrize( + "pane", + [ + "Do you trust the contents of this directory?\n› 1. Yes, continue", + "Choose working directory to resume this session\n› 1. Use session directory", + "Sign in with ChatGPT\n› 1. Continue", + ], + ) + def test_codex_startup_screens_are_not_ready(self, pane: str) -> None: + assert SessionManager._pane_has_ready_input(pane, "codex") is False + @pytest.mark.asyncio async def test_holds_until_compaction_ends( self, mgr: SessionManager, monkeypatch, fast_gate