From e25360e30ac07dcedf4d7c5d65a67e96dd825e4a Mon Sep 17 00:00:00 2001
From: Dustin Washington <dwash96@gmail.com>
Date: Mon, 14 Sep 2026 07:58:20 -0400
Subject: [PATCH 1/6] Automatically background auth-like commands to allow
 users to enter passwords, fix auto-PTY on linux background commands

---
 cecli/tools/command.py | 68 ++++++++++++++++++++++++++++++++++++++----
 1 file changed, 62 insertions(+), 6 deletions(-)

diff --git a/cecli/tools/command.py b/cecli/tools/command.py
index 6fd9c3be2aa..1177850bd4b 100644
--- a/cecli/tools/command.py
+++ b/cecli/tools/command.py
@@ -2,6 +2,7 @@
 import fnmatch
 import os
 import platform
+import re
 
 # PTY support for interactive commands (avoids pipe buffering issues)
 try:
@@ -22,6 +23,40 @@
 from cecli.tools.utils.responses import ToolResponse
 from cecli.tools.validations import ToolValidations
 
+# Commands an LLM is likely to run during development where the user must
+# type input for the task to proceed (passwords, passphrases, host-key
+# confirmations, credential logins, editor handoffs). Matching commands run
+# with user_input_required=True so the user can respond and the command can
+# complete. Patterns are matched case-insensitively against the full command.
+#
+# Long-running commands are already moved to the background by the timeout
+# mechanism, and session-style tools (ssh, su, database REPLs) read from
+# stdin when backgrounded, so they need no special handling here.
+#
+# Read-only viewers and live-monitoring tools (less, more, man, top, htop,
+# watch, tail -f, docker logs -f) are intentionally excluded: an LLM would
+# run those in the background to watch output rather than wait for typed
+# input, so forcing interactivity would block the background use case.
+INTERACTIVE_COMMAND_PATTERNS = [
+    # Privilege escalation, user switching, and password entry
+    r"^\s*(sudo|doas|runas|passwd)\b",
+    # Remote access: passwords, key passphrases, host-key confirmations
+    r"^\s*(scp|rsync|ssh-keygen|ssh-add|ssh-copy-id)\b",
+    # Passphrase prompts (gpg is also used for commit signing)
+    r"^\s*(gpg|gpg2)\b",
+    r"^\s*openssl\s+(enc|pkcs12|pkey|genpkey|rsa|genrsa|req)\b",
+    # Interactive credential / login flows
+    r"^\s*(gh|docker|npm|yarn|pnpm|az|aws|gcloud|heroku|firebase|vercel|netlify)\s+(auth|login|logout|configure|sso)\b",
+    # Editors: the user must edit content for the task to proceed
+    r"^\s*(vi|vim|nvim|nano|emacs|pico)\b",
+    # Git flows that hand off to an editor or need hunk-by-hunk input
+    r"^\s*git\s+(add\s+(-p|--patch)|commit\s+(-e|--edit)|rebase\s+(-i|--interactive)|mergetool|config\s+(-e|--edit))\b",
+    # Windows credential / remote-execution tools
+    r"^\s*(net\s+use|get-credential|psexec|cmdkey)\b",
+    # Config editors that open an interactive editor
+    r"^\s*(crontab\s+-e|visudo)\b",
+]
+
 
 class Tool(BaseTool):
     NORM_NAME = "command"
@@ -66,17 +101,19 @@ class Tool(BaseTool):
                         "type": "string",
                         "description": (
                             "Input to send. Use with background=True to send at "
-                            "start time, or with background_key + action='stdin'."
+                            "start time, or with background_key + action='stdin'. "
+                            "End the input with a newline to submit a line to an "
+                            "interactive prompt."
                         ),
                     },
                     "pty": {
                         "type": "boolean",
                         "description": (
-                            "Use a pseudo-terminal (PTY). Auto-enabled on Unix for "
-                            "background commands. Useful for interactive programs "
-                            "like 'vi' or 'top'."
+                            "Use a pseudo-terminal (PTY). Auto-enabled on Unix "
+                            "when omitted; set false to force pipe mode. A PTY lets "
+                            "you send stdin to long-running background commands."
                         ),
-                        "default": False,
+                        "default": None,
                     },
                     "user_input_required": {
                         "type": "boolean",
@@ -120,7 +157,7 @@ async def execute(
         background_key=None,
         action=None,
         stdin=None,
-        pty=False,
+        pty=None,
         user_input_required=False,
         timeout=0,
         **kwargs,
@@ -129,6 +166,8 @@ async def execute(
         Execute a shell command or interact with background processes.
 
         For new commands: provide 'command' (and optionally 'background', 'stdin', 'pty').
+        PTY is auto-enabled on Unix when 'pty' is omitted, so long-running
+        backgrounded commands can receive input via background_key + action='stdin'.
         When 'user_input_required' is True, runs the command interactively using a
         pseudo-terminal (PTY), allowing the user to provide inputs like passwords
         or navigate terminal interfaces.
@@ -182,6 +221,12 @@ async def execute(
             background = True
             command = command.strip()[:-1].strip()
 
+        # Force interactive handling for commands known to prompt for input
+        # (e.g. sudo, passphrase, credential, and editor prompts) so the user
+        # can respond and the command can complete.
+        if cls._requires_user_input(command):
+            user_input_required = True
+
         # Get user confirmation
         confirmed = await cls._get_confirmation(coder, command, background)
         if not confirmed:
@@ -662,6 +707,17 @@ async def _handle_errors(cls, coder, command_string, e):
         response.append_error(f"Error executing command: {str(e)}")
         return response
 
+    @classmethod
+    def _requires_user_input(cls, command_string):
+        """Return True if command matches a known interactive-input pattern."""
+        if not command_string:
+            return False
+
+        return any(
+            re.search(pattern, command_string, re.IGNORECASE)
+            for pattern in INTERACTIVE_COMMAND_PATTERNS
+        )
+
     @classmethod
     def format_output(cls, coder, mcp_server, tool_response):
         """Format output for Command tool."""

From e80051cce1ffbd15803ea1f7c7c6d1729be67e4c Mon Sep 17 00:00:00 2001
From: Dustin Washington <dwash96@gmail.com>
Date: Mon, 14 Sep 2026 20:11:08 -0400
Subject: [PATCH 2/6] - Update background command injection to be page based
 with a faster debounce cycle than other post messages - Allow models to
 intentionally pull the latest command arguments - Automatically require user
 input for auth-like commands like `sudo`

---
 cecli/coders/agent_coder.py                   |  67 +++++--
 cecli/helpers/background_commands.py          | 127 ++++++++++++-
 cecli/helpers/conversation/integration.py     |  80 ++++++++-
 cecli/tools/command.py                        | 126 +++++++++----
 tests/basic/test_background_commands.py       | 167 +++++++++++++++++
 .../test_background_command_injection.py      | 170 ++++++++++++++++++
 tests/tools/test_command_timeout_paging.py    | 104 +++++------
 tests/tools/test_resource_manager_paging.py   |  24 ++-
 8 files changed, 742 insertions(+), 123 deletions(-)
 create mode 100644 tests/conversations/test_background_command_injection.py

diff --git a/cecli/coders/agent_coder.py b/cecli/coders/agent_coder.py
index 4384283dea3..9a5a9d04cb2 100644
--- a/cecli/coders/agent_coder.py
+++ b/cecli/coders/agent_coder.py
@@ -652,6 +652,10 @@ def format_chat_chunks(self):
         # Add post-message context blocks (priority 250 - between CUR and REMINDER)
         ConversationService.get_chunks(self).add_post_message_context_blocks()
 
+        # Background command output is debounced independently so it is not
+        # re-dumped every turn as its contents mutate slightly
+        ConversationService.get_chunks(self).add_background_command_output()
+
         # Add sub-agent states context block (same priority as post-message blocks)
         ConversationService.get_chunks(self).add_sub_agent_states()
 
@@ -1845,25 +1849,44 @@ def get_background_command_output(self):
         """
         Get background command output to append after the main message.
 
-        Returns:
-            String containing formatted background command output, or empty string if none
+        Emits a roster of active commands (keeping command keys in context)
+        plus any new incremental output. Output that has been paged to disk is
+        advertised by command key and read on demand through ``ResourceManager``
+        paging.
         """
-        # Get output from all running background commands
-        bg_outputs = BackgroundCommandManager.get_all_command_outputs(clear=True)
+        command_info = BackgroundCommandManager.list_background_commands()
 
-        if not bg_outputs:
+        if not command_info:
             return ""
 
-        # Get command info to show actual command strings
-        command_info = BackgroundCommandManager.list_background_commands()
+        new_outputs = {}
+        for command_key in command_info:
+            output = BackgroundCommandManager.get_new_command_output(command_key)
+            if output.strip():
+                new_outputs[command_key] = output
 
-        # Create formatted output for background commands
         output = "--- Background Commands Output ---\n"
-        for command_key, cmd_output in bg_outputs.items():
-            if cmd_output.strip():  # Only add if there's output
-                # Get the actual command string if available
-                command_str = command_info.get(command_key, {}).get("command", command_key)
-                output += f"\n[bg: {command_str}]\n{cmd_output}\n"
+        output += "Commands:\n"
+
+        paged_keys = []
+        for command_key, info in sorted(command_info.items()):
+            status = "running" if info.get("running", False) else "finished"
+            pages = info.get("pages", 0)
+            page_note = f"pages 1-{pages}" if pages else "no pages yet"
+            output += (
+                f"- {command_key} [{status}] `{info.get('command', command_key)}`"
+                f" — {info.get('total_chars', 0):,} chars, {page_note}\n"
+            )
+            if pages:
+                paged_keys.append(command_key)
+
+        for command_key, cmd_output in new_outputs.items():
+            output += f"\nNew output ({command_key}):\n{cmd_output}\n"
+
+        if paged_keys:
+            output += "\nPaged output is available via `ResourceManager` (up to 3 pages):\n"
+            for command_key in paged_keys:
+                output += f'{{"paging": [{{"target": "{command_key}", "page": 1}}]}}\n'
 
         # Clean up stale (finished) background commands after reading their output
         for command_key, info in command_info.items():
@@ -1872,6 +1895,24 @@ def get_background_command_output(self):
 
         return output
 
+    def get_background_command_state(self):
+        """
+        Return a lightweight status snapshot of tracked background commands.
+
+        Maps each command key to its running state and page count so the
+        injection layer can detect finish and page-flush transitions cheaply
+        without consuming any output.
+        """
+        command_info = BackgroundCommandManager.list_background_commands()
+
+        return {
+            key: {
+                "running": bool(info.get("running", False)),
+                "pages": info.get("pages", 0),
+            }
+            for key, info in command_info.items()
+        }
+
     def get_git_status(self):
         """
         Generate a git status context block for repository information.
diff --git a/cecli/helpers/background_commands.py b/cecli/helpers/background_commands.py
index a698ac1313c..1a7f846558c 100644
--- a/cecli/helpers/background_commands.py
+++ b/cecli/helpers/background_commands.py
@@ -106,6 +106,110 @@ def size(self) -> int:
             return len(self.buffer)
 
 
+class PagedOutputBuffer:
+    """
+    Thread-safe output window that spills full pages to disk.
+
+    Output accumulates in memory until it reaches ``page_size`` characters.
+    Full pages are written to ``pages_dir`` as ``{n}.txt`` and dropped from
+    memory, so the in-memory window never grows beyond ``page_size``. Unlike
+    ``CircularBuffer``, ``total_added`` is a monotonic stream offset that is
+    never reset, so incremental readers can resume after content has been
+    flushed; anything older than the in-memory window lives on disk.
+    """
+
+    def __init__(self, page_size: int = 4096, pages_dir: Optional[str] = None):
+        self.page_size = max(1, int(page_size))
+        self.pages_dir = pages_dir
+        self.buffer = deque()
+        self.lock = threading.Lock()
+        self.total_added = 0
+        self.window_start = 0
+        self.page_count = 0
+
+    def append(self, text: str) -> None:
+        """Append text, spilling full pages to disk once the window fills."""
+        if not text:
+            return
+
+        with self.lock:
+            self.buffer.extend(text)
+            self.total_added += len(text)
+
+            if self.total_added - self.window_start >= self.page_size:
+                self._spill_locked()
+
+    def get_all(self, clear: bool = False) -> str:
+        """Return the in-memory window (content not yet flushed to disk)."""
+        with self.lock:
+            result = "".join(self.buffer)
+
+            if clear:
+                self.buffer.clear()
+                self.window_start = self.total_added
+
+            return result
+
+    def get_new_output(self, last_read_position: int) -> Tuple[str, int]:
+        """Return window content past ``last_read_position`` and the new offset.
+
+        Content older than the window has already been paged to disk, so the
+        read position is clamped forward to the window start rather than
+        replaying bytes that are only available as pages.
+        """
+        with self.lock:
+            if last_read_position >= self.total_added:
+                return "", self.total_added
+
+            start = max(last_read_position, self.window_start)
+            new_output = "".join(self.buffer)[start - self.window_start :]
+
+            return new_output, self.total_added
+
+    def clear(self) -> None:
+        """Drop the in-memory window; flushed pages are unaffected."""
+        with self.lock:
+            self.buffer.clear()
+            self.window_start = self.total_added
+
+    def size(self) -> int:
+        """Get current buffer size in characters."""
+        with self.lock:
+            return len(self.buffer)
+
+    def _spill_locked(self) -> None:
+        content = "".join(self.buffer)
+
+        if not self.pages_dir:
+            # Without a page directory, retain only the newest page in memory.
+            if len(content) > self.page_size:
+                dropped = len(content) - self.page_size
+                content = content[dropped:]
+                self.window_start += dropped
+                self.buffer = deque(content)
+
+            return
+
+        while len(content) >= self.page_size:
+            page = content[: self.page_size]
+            content = content[self.page_size :]
+            self.page_count += 1
+            self._write_page_locked(self.page_count, page)
+            self.window_start += self.page_size
+
+        self.buffer = deque(content)
+
+    def _write_page_locked(self, page_number: int, content: str) -> None:
+        os.makedirs(self.pages_dir, exist_ok=True)
+        abs_path = os.path.join(self.pages_dir, f"{page_number}.txt")
+        tmp_path = f"{abs_path}.tmp"
+
+        with safe_open(tmp_path, "w") as page_file:
+            page_file.write(content)
+
+        os.replace(tmp_path, abs_path)
+
+
 class InputBuffer:
     """
     Thread-safe buffer for queuing input to be sent to a process.
@@ -442,6 +546,9 @@ def start_background_command(
         existing_input_buffer: Optional[InputBuffer] = None,
         use_pty: bool = False,
         master_fd: Optional[int] = None,
+        command_key: Optional[str] = None,
+        page_size: Optional[int] = None,
+        pages_dir: Optional[str] = None,
     ) -> str:
         """
         Start a command in background.
@@ -452,15 +559,23 @@ def start_background_command(
             cwd: Working directory for command
             max_buffer_size: Maximum buffer size for output
             existing_process: Optional existing subprocess.Popen to register
-            existing_buffer: Optional existing CircularBuffer to use
+            existing_buffer: Optional existing buffer to use (CircularBuffer or PagedOutputBuffer)
             persist: If True, output buffer won't be cleared when read
+            command_key: Optional pre-generated command key; generated when omitted
+            page_size: Characters per page when paging output to disk
+            pages_dir: Directory where full output pages are written
 
         Returns:
             Command key for future reference
         """
         try:
-            # Use existing buffer or create new one
-            buffer = existing_buffer or CircularBuffer(max_size=max_buffer_size)
+            # Use existing buffer or create a paged/circular one
+            if existing_buffer is not None:
+                buffer = existing_buffer
+            elif page_size and pages_dir:
+                buffer = PagedOutputBuffer(page_size=page_size, pages_dir=pages_dir)
+            else:
+                buffer = CircularBuffer(max_size=max_buffer_size)
 
             # Use existing process or start new one
             # Use provided master_fd (e.g., from _execute_with_timeout) or default to None
@@ -523,7 +638,7 @@ def start_background_command(
             )
 
             # Generate unique key and store
-            command_key = cls._generate_command_key(command)
+            command_key = command_key or cls._generate_command_key(command)
 
             with cls._lock:
                 cls._background_commands[command_key] = bg_process
@@ -691,6 +806,10 @@ def list_background_commands(cls) -> Dict[str, Dict[str, any]]:
                     "command": bg_process.command,
                     "running": bg_process.is_alive(),
                     "buffer_size": bg_process.buffer.size(),
+                    "pages": getattr(bg_process.buffer, "page_count", 0),
+                    "total_chars": getattr(
+                        bg_process.buffer, "total_added", bg_process.buffer.size()
+                    ),
                     "start_time": bg_process.start_time,
                     "end_time": bg_process.end_time,
                     "duration": (
diff --git a/cecli/helpers/conversation/integration.py b/cecli/helpers/conversation/integration.py
index 202b00b7665..0cf9073b6a0 100644
--- a/cecli/helpers/conversation/integration.py
+++ b/cecli/helpers/conversation/integration.py
@@ -930,7 +930,9 @@ def add_post_message_context_blocks(self) -> None:
         """
         Add post-message context blocks to conversation (priority 250).
 
-        Post-message blocks include: tool_context/write_context, background_command_output
+        Post-message blocks include: todo_list, context_summary, tool_context,
+        and write_context. Background command output is injected separately via
+        ``add_background_command_output`` with its own debounce.
         """
         coder = self.get_coder()
         if not coder:
@@ -972,12 +974,6 @@ def add_post_message_context_blocks(self) -> None:
                         if write_context:
                             message_blocks["write_context"] = write_context
 
-        # Add background command output if any
-        if hasattr(coder, "get_background_command_output"):
-            bg_output = coder.get_background_command_output()
-            if bg_output:
-                message_blocks["background_command_output"] = bg_output
-
         # Add post-message blocks to conversation manager with stable hash keys
         for block_type, block_content in message_blocks.items():
             ConversationService.get_manager(coder).add_message(
@@ -989,6 +985,56 @@ def add_post_message_context_blocks(self) -> None:
                 force=True,
             )
 
+    def add_background_command_output(self, frequency=5):
+        """
+        Inject background command output at most once every ``frequency`` turns,
+        except when a command finishes or flushes a new page. Those transitions
+        bypass the debounce so important changes surface immediately.
+
+        Debounced independently from the other post-message blocks: the injected
+        content mutates slightly as commands produce output, so re-adding it
+        every turn would churn the conversation tail without the usual hash-key
+        deduplication catching it.
+        """
+        coder = self.get_coder()
+        if not coder:
+            return
+
+        if not hasattr(coder, "use_enhanced_context") or not coder.use_enhanced_context:
+            return
+
+        if not hasattr(coder, "get_background_command_output"):
+            return
+
+        last_turn = self.message_tracker.get("background_command_output")
+        due = last_turn is None or coder.turn_count - last_turn >= frequency
+
+        previous_state = self.message_tracker.get("background_command_state")
+        state = None
+        if hasattr(coder, "get_background_command_state"):
+            state = coder.get_background_command_state()
+
+        significant = self._has_background_signal(previous_state, state)
+        if state is not None:
+            self.message_tracker["background_command_state"] = state
+
+        if not significant and not due:
+            return
+
+        bg_output = coder.get_background_command_output()
+        if not bg_output:
+            return
+
+        self.message_tracker["background_command_output"] = coder.turn_count
+        ConversationService.get_manager(coder).add_message(
+            message_dict={"role": "user", "content": bg_output},
+            tag=MessageTag.STATIC,
+            priority=DEFAULT_TAG_PRIORITY[MessageTag.REMINDER] + 25,
+            mark_for_delete=0,
+            hash_key=("post_message", "background_command_output"),
+            force=True,
+        )
+
     def add_sub_agent_states(self) -> None:
         """
         Add sub-agent states context block to conversation (priority 250).
@@ -1056,6 +1102,26 @@ def debounce_message_injection(self, coder, message_type="default", frequency=10
 
         return not should_send
 
+    @staticmethod
+    def _has_background_signal(previous, current):
+        """Return True when a command finished or flushed a new page since ``previous``."""
+        if not current or not previous:
+            return False
+
+        for key, info in current.items():
+            prior = previous.get(key)
+
+            if prior is None:
+                return True
+
+            if prior.get("running") and not info.get("running"):
+                return True
+
+            if info.get("pages", 0) > prior.get("pages", 0):
+                return True
+
+        return False
+
     def _cancel_post_message_injections(self, modulus=10):
         coder = self.get_coder()
         if not coder:
diff --git a/cecli/tools/command.py b/cecli/tools/command.py
index 1177850bd4b..42a4e82ac78 100644
--- a/cecli/tools/command.py
+++ b/cecli/tools/command.py
@@ -86,15 +86,16 @@ class Tool(BaseTool):
                         "type": "string",
                         "description": (
                             "Key of an existing background command to interact with. "
-                            "Use with 'action' (stdin/stop)."
+                            "Use with 'action' (stdin/stop/tail)."
                         ),
                     },
                     "action": {
                         "type": "string",
-                        "enum": ["stdin", "stop"],
+                        "enum": ["stdin", "stop", "tail"],
                         "description": (
                             "Action on a background command. Requires background_key: "
-                            "'stdin' to send input, 'stop' to terminate."
+                            "'stdin' to send input, 'stop' to terminate, 'tail' to read "
+                            "the latest output."
                         ),
                     },
                     "stdin": {
@@ -171,7 +172,7 @@ async def execute(
         When 'user_input_required' is True, runs the command interactively using a
         pseudo-terminal (PTY), allowing the user to provide inputs like passwords
         or navigate terminal interfaces.
-        For background interactions: provide 'background_key' + 'action' (stdin/stop).
+        For background interactions: provide 'background_key' + 'action' (stdin/stop/tail).
 
         Commands run with timeout from agent_config['command_timeout'] (default: 30 seconds),
         """
@@ -201,8 +202,11 @@ async def execute(
             elif action == "stop":
                 return await cls._stop_background_command(coder, background_key)
 
+            elif action == "tail":
+                return await cls._tail_background_command(coder, background_key)
+
             else:
-                response.append_error(f"Unknown action '{action}'. Use one of: stdin, stop.")
+                response.append_error(f"Unknown action '{action}'. Use one of: stdin, stop, tail.")
                 return response
 
         if not command:
@@ -320,12 +324,17 @@ async def _execute_background(cls, coder, command_string, use_pty=None, stdin=No
             use_pty = platform.system() != "Windows"
 
         # Use static manager to start background command
+        command_key, page_size, pages_dir = cls._paging_config(coder, command_string)
+
         command_key = BackgroundCommandManager.start_background_command(
             command_string,
             verbose=coder.verbose,
             cwd=coder.root,
-            max_buffer_size=4096,
+            max_buffer_size=page_size or 4096,
             use_pty=use_pty,
+            command_key=command_key,
+            page_size=page_size,
+            pages_dir=pages_dir,
         )
 
         # Send stdin to the background command if provided
@@ -352,7 +361,7 @@ async def _execute_with_timeout(cls, coder, command_string, timeout, use_pty=Non
         import asyncio
         import subprocess
 
-        from cecli.helpers.background_commands import CircularBuffer
+        from cecli.helpers.background_commands import PagedOutputBuffer
 
         response = ToolResponse(cls.NORM_NAME)
 
@@ -364,8 +373,9 @@ async def _execute_with_timeout(cls, coder, command_string, timeout, use_pty=Non
         if use_pty is None:
             use_pty = platform.system() != "Windows"
 
-        # Create output buffer
-        buffer = CircularBuffer(max_size=4096)
+        # Create output buffer (paged when context management is enabled)
+        command_key, page_size, pages_dir = cls._paging_config(coder, command_string)
+        buffer = PagedOutputBuffer(page_size=page_size or 4096, pages_dir=pages_dir)
 
         # Decide whether to use PTY
         master_fd = None
@@ -424,6 +434,7 @@ async def _execute_with_timeout(cls, coder, command_string, timeout, use_pty=Non
             existing_buffer=buffer,
             persist=True,
             master_fd=master_fd,
+            command_key=command_key,
         )
 
         # Now monitor the process with an event-driven race instead of
@@ -475,30 +486,10 @@ async def _execute_with_timeout(cls, coder, command_string, timeout, use_pty=Non
 
             command_completed = wait_task in done
             output_content = buffer.get_all(clear=command_completed) or ""
-            # Tokens are roughly 3-4 characters
-            output_limit = int(coder.large_file_token_threshold * 3.5)
-
-            if coder.context_management_enabled and len(output_content) > output_limit * 1.25:
-                folder_path, file_list, alias_paths = (
-                    BackgroundCommandManager.save_paginated_output(
-                        output=output_content,
-                        command_key=command_key,
-                        page_size=output_limit,
-                        abs_root_path_func=coder.abs_root_path,
-                        local_agent_folder_func=coder.local_agent_folder,
-                    )
-                )
-                total_size = len(output_content)
+            pages_notice = cls._pages_notice(command_key, getattr(buffer, "page_count", 0))
+            if pages_notice:
                 output_content = (
-                    f"[Large Response ({total_size} characters). "
-                    f"Output saved in {len(file_list)} pages.]\n"
-                    f"Command key: {command_key}\n"
-                    f"Pages: 1-{len(file_list)}\n"
-                    "Use `ResourceManager` to view up to 3 pages at a time:\n"
-                    f'{{"paging": [{{"target": "{command_key}", "page": 1}}]}}\n'
-                    "Change page or add entries to read other pages (maximum 3 entries). "
-                    "Do not use add, read_only, or standard CLI tools to view command output "
-                    "files. Pages are returned directly, not added to file context."
+                    f"{output_content}\n\n{pages_notice}" if output_content else pages_notice
                 )
 
             if command_completed:
@@ -570,8 +561,8 @@ async def _execute_foreground(cls, coder, command_string):
 
         # Format the output for the result message
         output_content = combined_output or ""
-        output_limit = coder.large_file_token_threshold
-        if coder.context_management_enabled and len(output_content) > output_limit * 1.25:
+        output_limit = cls._page_size(coder)
+        if coder.context_management_enabled and len(output_content) > output_limit:
             # Generate a unique key for file naming
             fg_key = BackgroundCommandManager._generate_command_key(command_string)
             # Save full output to paginated files instead of truncating
@@ -718,6 +709,73 @@ def _requires_user_input(cls, command_string):
             for pattern in INTERACTIVE_COMMAND_PATTERNS
         )
 
+    @classmethod
+    def _page_size(cls, coder):
+        """Characters per output page (~3.5 characters per LLM token)."""
+        return max(1, int(getattr(coder, "large_file_token_threshold", 8192) * 3.5))
+
+    @classmethod
+    def _paging_config(cls, coder, command_string):
+        """Return (command_key, page_size, pages_dir) when output paging is enabled."""
+        if not getattr(coder, "context_management_enabled", False):
+            return None, None, None
+
+        page_size = cls._page_size(coder)
+        command_key = BackgroundCommandManager._generate_command_key(command_string)
+        pages_dir = coder.abs_root_path(coder.local_agent_folder(command_key))
+
+        return command_key, page_size, pages_dir
+
+    @classmethod
+    async def _tail_background_command(cls, coder, command_key):
+        """Return the latest in-memory output and page roster for a background command."""
+        command_info = BackgroundCommandManager.list_background_commands()
+        info = command_info.get(command_key)
+
+        response = ToolResponse(cls.NORM_NAME)
+        if not info:
+            response.append_error(f"Background command {command_key} not found.")
+            return response
+
+        status = "running" if info.get("running", False) else "finished"
+        output = BackgroundCommandManager.get_new_command_output(command_key)
+        pages = info.get("pages", 0)
+
+        lines = [
+            f"Background command {command_key} [{status}]: {info.get('command', command_key)}",
+            f"Output so far: {info.get('total_chars', 0):,} chars",
+        ]
+        if pages:
+            lines.append(f"Paged output: pages 1-{pages}. Read with ResourceManager paging.")
+            lines.append(f'{{"paging": [{{"target": "{command_key}", "page": 1}}]}}')
+
+        if output.strip():
+            lines.append("New output since last read:")
+            lines.append(output)
+        else:
+            lines.append("No new output since last read.")
+
+        response.append_result("\n".join(lines))
+
+        return response
+
+    @staticmethod
+    def _pages_notice(command_key, page_count):
+        """Guidance for reading command output that has been paged to disk."""
+        if not page_count:
+            return ""
+
+        return (
+            f"[Output paged to disk: {page_count} page(s).]\n"
+            f"Command key: {command_key}\n"
+            f"Pages: 1-{page_count}\n"
+            "Use `ResourceManager` to view up to 3 pages at a time:\n"
+            f'{{"paging": [{{"target": "{command_key}", "page": 1}}]}}\n'
+            "Change the page number to read other pages (maximum 3 entries). "
+            "Do not use add, read_only, or standard CLI tools to view command output "
+            "files. Pages are returned directly, not added to file context."
+        )
+
     @classmethod
     def format_output(cls, coder, mcp_server, tool_response):
         """Format output for Command tool."""
diff --git a/tests/basic/test_background_commands.py b/tests/basic/test_background_commands.py
index 5ca98746186..1f7f079a274 100644
--- a/tests/basic/test_background_commands.py
+++ b/tests/basic/test_background_commands.py
@@ -49,8 +49,10 @@ def readline(self):
 _install_stubs()
 
 from cecli.helpers.background_commands import (  # noqa: E402
+    BackgroundCommandManager,
     BackgroundProcess,
     CircularBuffer,
+    PagedOutputBuffer,
 )
 
 
@@ -235,3 +237,168 @@ def readline(self):
     success, output, exit_code = bg_process.stop()
     assert success is True
     assert exit_code == -1  # terminate() sets returncode to -1 in MockProcess
+
+
+def test_paged_output_buffer_spills_pages_to_disk(tmp_path):
+    """Full pages are flushed to disk and dropped from the in-memory window."""
+    buffer = PagedOutputBuffer(page_size=5, pages_dir=str(tmp_path / "pages"))
+
+    buffer.append("abc")
+    assert buffer.get_all() == "abc"
+    assert buffer.page_count == 0
+
+    buffer.append("de")
+    assert buffer.page_count == 1
+    assert buffer.get_all() == ""
+    assert buffer.total_added == 5
+    assert (tmp_path / "pages" / "1.txt").read_text(encoding="utf-8") == "abcde"
+
+    buffer.append("fgh")
+    assert buffer.page_count == 1
+    assert buffer.get_all() == "fgh"
+
+    buffer.append("ij")
+    assert buffer.page_count == 2
+    assert buffer.get_all() == ""
+    assert (tmp_path / "pages" / "2.txt").read_text(encoding="utf-8") == "fghij"
+
+    # Atomic writes leave no temporary files behind
+    assert not list((tmp_path / "pages").glob("*.tmp"))
+
+
+def test_paged_output_buffer_incremental_reads_clamp_to_window(tmp_path):
+    """Readers resume monotonically; already-paged content is not replayed."""
+    buffer = PagedOutputBuffer(page_size=4, pages_dir=str(tmp_path / "p"))
+
+    assert buffer.get_new_output(0) == ("", 0)
+
+    buffer.append("abcd")
+    assert buffer.get_new_output(0) == ("", 4)
+
+    buffer.append("ef")
+    assert buffer.get_new_output(4) == ("ef", 6)
+    assert buffer.get_new_output(6) == ("", 6)
+
+
+def test_paged_output_buffer_without_pages_dir_is_bounded():
+    """With no page directory the window simply keeps the newest page."""
+    buffer = PagedOutputBuffer(page_size=5, pages_dir=None)
+
+    buffer.append("abcdefgh")
+
+    assert buffer.get_all() == "defgh"
+    assert buffer.page_count == 0
+
+
+def test_tail_background_command_reports_output_and_pages(monkeypatch):
+    """The tail action reports status, page roster, and new output."""
+    import asyncio
+
+    from cecli.tools.command import Tool as CommandTool
+
+    monkeypatch.setattr(
+        BackgroundCommandManager,
+        "list_background_commands",
+        lambda: {
+            "bg_1_1234": {
+                "command": "pytest -q",
+                "running": True,
+                "pages": 3,
+                "total_chars": 42,
+            }
+        },
+    )
+    monkeypatch.setattr(
+        BackgroundCommandManager, "get_new_command_output", lambda key: "new line\n"
+    )
+
+    response = asyncio.run(CommandTool._tail_background_command(object(), "bg_1_1234"))
+    content = response.to_dict()["result"][0]["content"]
+
+    assert "bg_1_1234" in content
+    assert "running" in content
+    assert "pages 1-3" in content
+    assert '{"paging": [{"target": "bg_1_1234", "page": 1}]}' in content
+    assert "new line" in content
+
+
+def test_tail_background_command_missing_key(monkeypatch):
+    import asyncio
+
+    from cecli.tools.command import Tool as CommandTool
+
+    monkeypatch.setattr(BackgroundCommandManager, "list_background_commands", lambda: {})
+
+    response = asyncio.run(CommandTool._tail_background_command(object(), "bg_9_9999"))
+
+    assert response.to_dict()["errors"]
+
+
+def test_get_background_command_output_roster_incremental_and_pages(monkeypatch):
+    """Injection lists a stable roster, new output, and page guidance."""
+    from cecli.coders.agent_coder import AgentCoder
+
+    monkeypatch.setattr(
+        BackgroundCommandManager,
+        "list_background_commands",
+        lambda: {
+            "bg_1_1234": {
+                "command": "pytest -q",
+                "running": True,
+                "pages": 2,
+                "total_chars": 100,
+            },
+            "bg_2_5678": {
+                "command": "npm run build",
+                "running": True,
+                "pages": 0,
+                "total_chars": 12,
+            },
+        },
+    )
+    monkeypatch.setattr(
+        BackgroundCommandManager,
+        "get_new_command_output",
+        lambda key: f"out-{key}\n",
+    )
+    stopped = []
+    monkeypatch.setattr(
+        BackgroundCommandManager, "stop_background_command", lambda key: stopped.append(key)
+    )
+
+    output = AgentCoder.get_background_command_output(object())
+
+    assert "bg_1_1234" in output
+    assert "pages 1-2" in output
+    assert "no pages yet" in output
+    assert "out-bg_1_1234" in output
+    assert '{"paging": [{"target": "bg_1_1234", "page": 1}]}' in output
+    assert stopped == []
+
+
+def test_get_background_command_output_stops_finished_commands(monkeypatch):
+    """Finished commands are reported once and then removed from tracking."""
+    from cecli.coders.agent_coder import AgentCoder
+
+    monkeypatch.setattr(
+        BackgroundCommandManager,
+        "list_background_commands",
+        lambda: {
+            "bg_3_0001": {
+                "command": "true",
+                "running": False,
+                "pages": 0,
+                "total_chars": 0,
+            }
+        },
+    )
+    monkeypatch.setattr(BackgroundCommandManager, "get_new_command_output", lambda key: "")
+    stopped = []
+    monkeypatch.setattr(
+        BackgroundCommandManager, "stop_background_command", lambda key: stopped.append(key)
+    )
+
+    output = AgentCoder.get_background_command_output(object())
+
+    assert "finished" in output
+    assert stopped == ["bg_3_0001"]
diff --git a/tests/conversations/test_background_command_injection.py b/tests/conversations/test_background_command_injection.py
new file mode 100644
index 00000000000..f154d57adaa
--- /dev/null
+++ b/tests/conversations/test_background_command_injection.py
@@ -0,0 +1,170 @@
+"""Tests for the independently debounced background command output injection."""
+
+import uuid
+
+from cecli.helpers.conversation import ConversationService
+
+
+class MockCoder:
+    def __init__(self):
+        self.uuid = str(uuid.uuid4())
+        self.use_enhanced_context = True
+        self.turn_count = 0
+        self.output_calls = 0
+        self.output = "roster"
+
+    def get_background_command_output(self):
+        self.output_calls += 1
+        return self.output
+
+
+def _make_chunks(coder):
+    manager = ConversationService.get_manager(coder)
+    manager.reset()
+    chunks = ConversationService.get_chunks(coder)
+    chunks.message_tracker = {}
+    return chunks, manager
+
+
+def test_background_command_output_debounced_to_every_five_turns():
+    coder = MockCoder()
+    chunks, manager = _make_chunks(coder)
+
+    for turn in range(7):
+        coder.turn_count = turn
+        chunks.add_background_command_output(frequency=5)
+
+    # Injected once up front, then not again until turn 5
+    assert coder.output_calls == 2
+    assert [message["content"] for message in manager.get_messages_dict()] == ["roster"]
+
+
+def test_background_command_output_replaces_same_hash_key():
+    coder = MockCoder()
+    chunks, manager = _make_chunks(coder)
+
+    coder.turn_count = 0
+    chunks.add_background_command_output(frequency=5)
+
+    coder.output = "roster v2"
+    coder.turn_count = 5
+    chunks.add_background_command_output(frequency=5)
+
+    assert [message["content"] for message in manager.get_messages_dict()] == ["roster v2"]
+
+
+def test_background_command_output_skipped_without_enhanced_context():
+    coder = MockCoder()
+    coder.use_enhanced_context = False
+    chunks, manager = _make_chunks(coder)
+
+    coder.turn_count = 0
+    chunks.add_background_command_output(frequency=5)
+
+    assert coder.output_calls == 0
+    assert manager.get_messages_dict() == []
+
+
+def test_empty_background_command_output_does_not_consume_window():
+    coder = MockCoder()
+    coder.output = ""
+    chunks, manager = _make_chunks(coder)
+
+    coder.turn_count = 0
+    chunks.add_background_command_output(frequency=5)
+    coder.turn_count = 1
+    chunks.add_background_command_output(frequency=5)
+
+    # Empty output never injects and never marks the tracker, so polling continues
+    assert coder.output_calls == 2
+    assert manager.get_messages_dict() == []
+
+    coder.output = "roster"
+    coder.turn_count = 2
+    chunks.add_background_command_output(frequency=5)
+
+    assert coder.output_calls == 3
+    assert [message["content"] for message in manager.get_messages_dict()] == ["roster"]
+
+
+class SignalCoder(MockCoder):
+    def __init__(self):
+        super().__init__()
+        self.state = {}
+
+    def get_background_command_state(self):
+        return self.state
+
+
+def test_finish_transition_bypasses_debounce():
+    coder = SignalCoder()
+    chunks, manager = _make_chunks(coder)
+
+    coder.state = {"bg_1_0001": {"running": True, "pages": 0}}
+    coder.turn_count = 0
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 1
+
+    # Still running, nothing changed, and not due -> skip
+    coder.turn_count = 1
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 1
+
+    # Finished -> inject immediately despite the frequency window
+    coder.state = {"bg_1_0001": {"running": False, "pages": 0}}
+    coder.turn_count = 2
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 2
+
+
+def test_new_page_flush_bypasses_debounce():
+    coder = SignalCoder()
+    chunks, _ = _make_chunks(coder)
+
+    coder.state = {"bg_1_0001": {"running": True, "pages": 0}}
+    coder.turn_count = 0
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 1
+
+    coder.state = {"bg_1_0001": {"running": True, "pages": 1}}
+    coder.turn_count = 1
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 2
+
+
+def test_new_command_appearance_bypasses_debounce():
+    coder = SignalCoder()
+    chunks, _ = _make_chunks(coder)
+
+    coder.state = {"bg_1_0001": {"running": True, "pages": 0}}
+    coder.turn_count = 0
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 1
+
+    coder.state = {
+        "bg_1_0001": {"running": True, "pages": 0},
+        "bg_2_0002": {"running": True, "pages": 0},
+    }
+    coder.turn_count = 1
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 2
+
+
+def test_running_command_without_change_respects_frequency():
+    coder = SignalCoder()
+    chunks, _ = _make_chunks(coder)
+
+    coder.state = {"bg_1_0001": {"running": True, "pages": 0}}
+    coder.turn_count = 0
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 1
+
+    for turn in (1, 2, 3, 4):
+        coder.turn_count = turn
+        chunks.add_background_command_output(frequency=5)
+
+    assert coder.output_calls == 1
+
+    coder.turn_count = 5
+    chunks.add_background_command_output(frequency=5)
+    assert coder.output_calls == 2
diff --git a/tests/tools/test_command_timeout_paging.py b/tests/tools/test_command_timeout_paging.py
index d2cc2f8d37c..c4fb654a12f 100644
--- a/tests/tools/test_command_timeout_paging.py
+++ b/tests/tools/test_command_timeout_paging.py
@@ -33,35 +33,43 @@ async def elapsed_command(monkeypatch, tmp_path):
         context_blocks_cache={},
         edit_allowed=False,
         interrupt_event=asyncio.Event(),
+        large_file_token_threshold=8,
+        context_management_enabled=True,
     )
     manager = background_commands.BackgroundCommandManager
     target = "bg_1_1234"
     process = Mock()
     popen = Mock(return_value=process)
-    buffer = background_commands.CircularBuffer()
-    get_all = Mock(wraps=buffer.get_all)
-    monkeypatch.setattr(buffer, "get_all", get_all)
-    monkeypatch.setattr(background_commands, "CircularBuffer", Mock(return_value=buffer))
+    state = {"output": "", "buffer": None}
+
+    real_buffer_cls = background_commands.PagedOutputBuffer
+
+    def make_buffer(page_size=4096, pages_dir=None):
+        buffer = real_buffer_cls(page_size=page_size, pages_dir=pages_dir)
+        state["buffer"] = buffer
+        if state["output"]:
+            buffer.append(state["output"])
+        return buffer
+
+    monkeypatch.setattr(background_commands, "PagedOutputBuffer", make_buffer)
     monkeypatch.setattr("subprocess.Popen", popen)
+    monkeypatch.setattr(manager, "_generate_command_key", Mock(return_value=target))
     start = Mock(return_value=target)
     stop = Mock()
-    save = Mock(wraps=manager.save_paginated_output)
     monkeypatch.setattr(manager, "start_background_command", start)
     monkeypatch.setattr(manager, "stop_background_command", stop)
-    monkeypatch.setattr(manager, "save_paginated_output", save)
     pending_tasks = []
 
     async def pending_wait(*args, **kwargs):
         pending_tasks.append(asyncio.current_task())
         await asyncio.get_running_loop().create_future()
 
-    to_thread = Mock(side_effect=pending_wait)
-    monkeypatch.setattr(asyncio, "to_thread", to_thread)
+    monkeypatch.setattr(asyncio, "to_thread", Mock(side_effect=pending_wait))
 
     async def execute(output, threshold=8, enabled=True):
         coder.large_file_token_threshold = threshold
         coder.context_management_enabled = enabled
-        buffer.append(output)
+        state["output"] = output
         response = await CommandTool._execute_with_timeout(
             coder, "pending command", 0.001, use_pty=False
         )
@@ -75,14 +83,12 @@ async def execute(output, threshold=8, enabled=True):
         assert len(pending_tasks) == 1
         assert not pending_tasks[0].done()
         assert not coder.interrupt_event.is_set()
-        to_thread.assert_called_once_with(process.wait)
         popen.assert_called_once()
         start.assert_called_once()
         assert start.call_args.kwargs["existing_process"] is process
-        assert start.call_args.kwargs["existing_buffer"] is buffer
+        assert start.call_args.kwargs["existing_buffer"] is state["buffer"]
         assert start.call_args.kwargs["persist"] is True
-        get_all.assert_called_once_with(clear=False)
-        assert buffer.get_all() == output
+        assert start.call_args.kwargs["command_key"] == (target if enabled else None)
         stop.assert_not_called()
         process.wait.assert_not_called()
         process.terminate.assert_not_called()
@@ -90,7 +96,7 @@ async def execute(output, threshold=8, enabled=True):
         return content
 
     try:
-        yield SimpleNamespace(execute=execute, coder=coder, target=target, save=save)
+        yield SimpleNamespace(execute=execute, coder=coder, target=target, state=state)
     finally:
         for task in pending_tasks:
             task.cancel()
@@ -99,59 +105,38 @@ async def execute(output, threshold=8, enabled=True):
 
 
 @pytest.mark.asyncio
-async def test_elapsed_timeout_saves_pages_readable_without_adding_context(elapsed_command):
-    output = "first line: café\nsecond line\n" * 3
-    content = await elapsed_command.execute(output)
+async def test_elapsed_timeout_pages_output_and_exposes_command_keys(elapsed_command):
     coder = elapsed_command.coder
     target = elapsed_command.target
     page_size = int(coder.large_file_token_threshold * 3.5)
-    expected_pages = [
-        output[index : index + page_size] for index in range(0, len(output), page_size)
-    ]
+    output = "first line: café\nsecond line\n" * 3
+    content = await elapsed_command.execute(output)
+    num_pages = len(output) // page_size
 
+    assert num_pages >= 1
     assert output not in content
-    assert f"Large Response ({len(output)} characters)" in content
-    assert f"Output saved in {len(expected_pages)} pages." in content
+    assert f"Output paged to disk: {num_pages} page(s)." in content
     assert "ResourceManager" in content
     assert "not added to file context" in content
     assert "command_key::" not in content
     example = next(line for line in content.splitlines() if line.startswith('{"paging"'))
     assert json.loads(example) == {"paging": [{"target": target, "page": 1}]}
-    elapsed_command.save.assert_called_once_with(
-        output=output,
-        command_key=target,
-        page_size=page_size,
-        abs_root_path_func=coder.abs_root_path,
-        local_agent_folder_func=coder.local_agent_folder,
-    )
+
     folder = Path(coder.abs_root_path(coder.local_agent_folder(target)))
     assert {path.name for path in folder.iterdir()} == {
-        f"{page}.txt" for page in range(1, len(expected_pages) + 1)
+        f"{page}.txt" for page in range(1, num_pages + 1)
     }
     saved_pages = [
-        (folder / f"{page}.txt").read_text(encoding="utf-8")
-        for page in range(1, len(expected_pages) + 1)
+        (folder / f"{page}.txt").read_text(encoding="utf-8") for page in range(1, num_pages + 1)
     ]
-    assert saved_pages == expected_pages
-    assert "".join(saved_pages) == output
+    assert "".join(saved_pages) == output[: num_pages * page_size]
 
     editable_before = coder.abs_fnames.copy()
     read_only_before = coder.abs_read_only_fnames.copy()
-    for index in range(0, len(expected_pages), 3):
-        batch = expected_pages[index : index + 3]
-        response = await ResourceManagerTool.execute(
-            coder,
-            paging=[
-                {"target": target, "page": page}
-                for page in range(index + 1, index + len(batch) + 1)
-            ],
-        )
-        result = response.to_dict()
-        assert result["errors"] == []
-        assert len(result["result"]) == len(batch)
-        for item, expected in zip(result["result"], batch):
-            assert item["content"].endswith(expected)
-
+    response = await ResourceManagerTool.execute(coder, paging=[{"target": target, "page": 1}])
+    result = response.to_dict()
+    assert result["errors"] == []
+    assert result["result"][0]["content"].endswith(output[:page_size])
     assert coder.abs_fnames == editable_before
     assert coder.abs_read_only_fnames == read_only_before
     coder._add_file_to_context.assert_not_called()
@@ -159,23 +144,27 @@ async def test_elapsed_timeout_saves_pages_readable_without_adding_context(elaps
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    "threshold,length,paged", [(8, 35, False), (8, 36, True), (9, 38, False), (9, 39, True)]
+    "threshold,length,paged", [(8, 27, False), (8, 28, True), (9, 30, False), (9, 31, True)]
 )
-async def test_elapsed_timeout_paging_uses_strict_rounded_threshold(
+async def test_elapsed_timeout_paging_triggers_at_page_size(
     elapsed_command, threshold, length, paged
 ):
     output = "x" * length
     content = await elapsed_command.execute(output, threshold=threshold)
 
     if paged:
-        elapsed_command.save.assert_called_once()
-        assert elapsed_command.save.call_args.kwargs["page_size"] == int(threshold * 3.5)
-        assert "Large Response" in content
+        page_size = int(threshold * 3.5)
+        assert "Output paged to disk: 1 page(s)." in content
         assert output not in content
+        folder = Path(
+            elapsed_command.coder.abs_root_path(
+                elapsed_command.coder.local_agent_folder(elapsed_command.target)
+            )
+        )
+        assert (folder / "1.txt").read_text(encoding="utf-8") == output[:page_size]
     else:
-        elapsed_command.save.assert_not_called()
         assert f"Output captured so far:\n{output}\n" in content
-        assert "Large Response" not in content
+        assert "Output paged to disk" not in content
 
 
 @pytest.mark.asyncio
@@ -188,5 +177,4 @@ async def test_elapsed_timeout_keeps_small_empty_or_unmanaged_output_inline(
     content = await elapsed_command.execute(output, enabled=enabled)
 
     assert f"Output captured so far:\n{output}\n" in content
-    assert "Large Response" not in content
-    elapsed_command.save.assert_not_called()
+    assert "Output paged to disk" not in content
diff --git a/tests/tools/test_resource_manager_paging.py b/tests/tools/test_resource_manager_paging.py
index 04ecc1576b9..1237b29dc6e 100644
--- a/tests/tools/test_resource_manager_paging.py
+++ b/tests/tools/test_resource_manager_paging.py
@@ -284,9 +284,9 @@ async def test_command_large_output_guidance_uses_paging_array(
     manager = command.BackgroundCommandManager
     target = "bg_1_1234"
     output = "large command output\n" * 50
+    monkeypatch.setattr(manager, "_generate_command_key", Mock(return_value=target))
     save = Mock(return_value=("pages", ["1.txt", "2.txt"], ["command_key::old/1.txt"]))
     monkeypatch.setattr(manager, "save_paginated_output", save)
-    monkeypatch.setattr(manager, "_generate_command_key", Mock(return_value=target))
 
     if execution_path == "foreground":
         monkeypatch.setattr(command, "run_cmd_subprocess", Mock(return_value=(0, output)))
@@ -297,9 +297,15 @@ async def test_command_large_output_guidance_uses_paging_array(
         monkeypatch.setattr("subprocess.Popen", Mock(return_value=process))
         monkeypatch.setattr(manager, "start_background_command", Mock(return_value=target))
         monkeypatch.setattr(manager, "stop_background_command", Mock())
-        buffer = Mock()
-        buffer.get_all.return_value = output
-        monkeypatch.setattr(background_commands, "CircularBuffer", Mock(return_value=buffer))
+
+        real_buffer_cls = background_commands.PagedOutputBuffer
+
+        def make_buffer(page_size=4096, pages_dir=None):
+            buffer = real_buffer_cls(page_size=page_size, pages_dir=pages_dir)
+            buffer.append(output)
+            return buffer
+
+        monkeypatch.setattr(background_commands, "PagedOutputBuffer", make_buffer)
         response = await command.Tool._execute_with_timeout(coder, "echo test", 30, use_pty=False)
 
     result = response.to_dict()
@@ -310,6 +316,10 @@ async def test_command_large_output_guidance_uses_paging_array(
     assert "ResourceManager" in content
     assert "command_key::" not in content
     assert "not added to file context" in content
-    save.assert_called_once()
-    assert save.call_args.kwargs["output"] == output
-    assert save.call_args.kwargs["command_key"] == target
+
+    if execution_path == "foreground":
+        save.assert_called_once()
+        assert save.call_args.kwargs["output"] == output
+        assert save.call_args.kwargs["command_key"] == target
+    else:
+        save.assert_not_called()

From 42b3f06cfec2f81b7ba56e18226f0f3e1cb346ab Mon Sep 17 00:00:00 2001
From: Dustin Washington <dwash96@gmail.com>
Date: Tue, 15 Sep 2026 02:18:28 -0400
Subject: [PATCH 3/6] Add MCP server default request timeout, make sure
 connect_all() errors can't block

---
 cecli/helpers/coroutines.py      | 19 +++++++
 cecli/mcp/manager.py             | 69 +++++++++++++++---------
 cecli/mcp/oauth.py               | 91 +++++++++++++++++++++++++-------
 cecli/mcp/server.py              | 51 ++++++++++++++++--
 cecli/tui/worker.py              | 38 +++++++++++--
 cecli/website/docs/config/mcp.md | 31 +++++++++++
 tests/mcp/test_manager_retry.py  | 41 ++++++++++++--
 7 files changed, 284 insertions(+), 56 deletions(-)

diff --git a/cecli/helpers/coroutines.py b/cecli/helpers/coroutines.py
index 968410ecf48..0d2ae4c8294 100644
--- a/cecli/helpers/coroutines.py
+++ b/cecli/helpers/coroutines.py
@@ -109,3 +109,22 @@ async def interruptible(coroutine, interrupt_event):
         return main_task.result(), False
     except asyncio.CancelledError:
         return None, True
+
+
+def task_is_cancelling() -> bool:
+    """Return True when the running asyncio task has a pending cancellation.
+
+    Used to tell a genuine cancellation of the caller apart from cancellation
+    errors that transports (e.g. MCP's anyio TaskGroups) surface for ordinary
+    connection failures.
+    """
+    task = asyncio.current_task()
+    if task is None:
+        return False
+
+    cancelling = getattr(task, "cancelling", None)
+    if cancelling is not None:
+        return cancelling() > 0
+
+    # Python 3.10 has no Task.cancelling(); fall back to the private flag.
+    return bool(getattr(task, "_must_cancel", False))
diff --git a/cecli/mcp/manager.py b/cecli/mcp/manager.py
index 9f3cb429208..a2216e0267b 100644
--- a/cecli/mcp/manager.py
+++ b/cecli/mcp/manager.py
@@ -1,6 +1,7 @@
 import asyncio
 
-from cecli.mcp.server import LocalServer, McpServer
+from cecli.helpers.coroutines import task_is_cancelling
+from cecli.mcp.server import DEFAULT_MCP_REQUEST_TIMEOUT, LocalServer, McpServer
 from cecli.tools.utils.registry import ToolRegistry
 
 
@@ -196,9 +197,10 @@ async def connect_server(self, name: str) -> bool:
             return True
 
         # Retry with exponential backoff for transient connection failures.
-        # Note: This also fixes a latent bug where asyncio.CancelledError was
-        # silently caught and treated as a connection failure. CancelledError is
-        # now re-raised to properly propagate cancellation.
+        # A genuine cancellation must still propagate, but some transports
+        # (MCP's streamable-HTTP runs its writer inside an anyio TaskGroup)
+        # surface ordinary connection failures as CancelledError, so only a
+        # real cancellation of this task is re-raised.
         # When io is None (e.g., during from_servers before IO is assigned),
         # _log_warning and _log_error silently return — retries still happen
         # but with no user-visible feedback. This is intentional.
@@ -206,38 +208,55 @@ async def connect_server(self, name: str) -> bool:
         delay = 1.0
         backoff = 2.0
         max_delay = 30.0
+        # Bound each attempt so a server that accepts a connection but never
+        # speaks MCP cannot wedge startup. The SDK read timeout usually fires
+        # first; this wait_for is a backstop for hangs inside the transport.
+        try:
+            base_timeout = float(server._request_timeout_seconds())
+        except (TypeError, ValueError):
+            base_timeout = DEFAULT_MCP_REQUEST_TIMEOUT
+
+        if base_timeout <= 0:
+            base_timeout = DEFAULT_MCP_REQUEST_TIMEOUT
+
+        attempt_timeout = base_timeout + 5
 
         for attempt in range(1, max_retries + 1):
+            error = None
+
             try:
-                session = await server.connect()
-                tools_result = await session.list_tools()
+                session = await asyncio.wait_for(server.connect(), timeout=attempt_timeout)
+                tools_result = await asyncio.wait_for(session.list_tools(), timeout=attempt_timeout)
                 tools = _mcp_tools_to_openai_tools(tools_result.tools)
                 self._server_tools[server.name] = tools
                 self._connected_servers.add(server)
                 self._log_verbose(f"Connected to MCP server: {name}")
                 return True
             except asyncio.CancelledError:
-                raise
+                if task_is_cancelling():
+                    raise
+                error = "connection cancelled by transport"
             except Exception as e:
-                if attempt < max_retries and server.name != "unnamed-server":
-                    self._log_warning(
-                        f"Connection attempt {attempt} failed for {name}, "
-                        f"retrying in {delay}s... ({e})"
-                    )
+                error = e
 
-                    await asyncio.sleep(delay)
-                    delay = min(delay * backoff, max_delay)
-                else:
-                    if server.name != "unnamed-server":
-                        self._log_error(
-                            f"Failed to connect to MCP server {name} "
-                            f"after {max_retries} attempts: {e}"
-                        )
-                    if server.is_connected:
-                        # Session was established but tool listing failed; tear
-                        # it down so the transport/subprocess doesn't leak.
-                        await server.disconnect()
-                    return False
+            if attempt < max_retries and server.name != "unnamed-server":
+                self._log_warning(
+                    f"Connection attempt {attempt} failed for {name}, "
+                    f"retrying in {delay}s... ({error})"
+                )
+                await asyncio.sleep(delay)
+                delay = min(delay * backoff, max_delay)
+            else:
+                if server.name != "unnamed-server":
+                    self._log_error(
+                        f"Failed to connect to MCP server {name} "
+                        f"after {max_retries} attempts: {error}"
+                    )
+                if server.is_connected:
+                    # Session was established but tool listing failed; tear
+                    # it down so the transport/subprocess doesn't leak.
+                    await server.disconnect()
+                return False
 
     async def disconnect_server(self, name: str) -> bool:
         """
diff --git a/cecli/mcp/oauth.py b/cecli/mcp/oauth.py
index ec5390f77b5..31ff00f0475 100644
--- a/cecli/mcp/oauth.py
+++ b/cecli/mcp/oauth.py
@@ -15,20 +15,29 @@
 from cecli.decoding import safe_open
 
 
-def create_oauth_callback_server(
-    port, path="/callback"
-) -> Tuple[Callable[[], Awaitable[Tuple[str, str]]], Callable[[], None]]:
+def create_oauth_callback_server(port, path="/callback") -> Tuple[
+    Callable[[], Awaitable[Tuple[str, str]]],
+    Callable[[], None],
+    Callable[[], None],
+]:
     """
     Create a local HTTP server to handle OAuth callback.
 
+    The listener is started lazily via the returned ``ensure_started`` callable
+    so servers that never actually trigger OAuth don't leave a daemon HTTP
+    server (and its bound port) running for the life of the process.
+
     Returns:
-        Tuple of (async callback handler function, shutdown function)
+        Tuple of (async callback handler, shutdown function, start function)
     """
     auth_code = None
     state = None
     server_error = None
     callback_received = threading.Event()
     server = None
+    server_thread = None
+    server_started = threading.Event()
+    start_lock = threading.Lock()
 
     class OAuthCallbackHandler(http.server.SimpleHTTPRequestHandler):
         def do_GET(self):
@@ -78,27 +87,71 @@ def do_GET(self):
         def log_message(self, format, *args):
             pass
 
-    # Start server in a separate thread
     def start_server():
-        nonlocal server
+        nonlocal server, server_error
+        srv = None
         try:
-            server = socketserver.TCPServer(("localhost", port), OAuthCallbackHandler)
-            server.serve_forever()
+            srv = socketserver.TCPServer(("localhost", port), OAuthCallbackHandler)
+            server = srv
+            server_started.set()
+            srv.serve_forever()
         except Exception as e:
-            server_error = f"Server error: {e}"  # noqa
+            server_error = f"Server error: {e}"
+            server_started.set()
             callback_received.set()
+        finally:
+            if srv is not None:
+                try:
+                    srv.server_close()
+                except Exception:
+                    pass
+
+    def ensure_started():
+        """Start the callback listener once, blocking briefly for the bind."""
+        nonlocal server_thread
+        with start_lock:
+            if server_started.is_set():
+                return
+
+            server_thread = threading.Thread(
+                target=start_server, daemon=True, name="oauth-callback-server"
+            )
+            server_thread.start()
+
+        # Wait for the bind to complete so the browser redirect cannot race the
+        # listener coming up.
+        server_started.wait(timeout=5)
 
-    server_thread = threading.Thread(target=start_server, daemon=True)
-    server_thread.start()
-
-    # Shutdown function
     def shutdown():
+        """Stop the callback listener. Idempotent and safe to call repeatedly.
+
+        ``socketserver.shutdown()`` blocks until ``serve_forever`` exits; async
+        callers must run this in a thread (see ``asyncio.to_thread``) so a stuck
+        listener can never wedge the event loop.
+        """
         nonlocal server
-        if server:
-            server.shutdown()
+        with start_lock:
+            srv = server
             server = None
 
+        if srv is None:
+            return
+
+        try:
+            srv.shutdown()
+        except Exception:
+            pass
+        finally:
+            try:
+                srv.server_close()
+            except Exception:
+                pass
+
     async def get_auth_code() -> Tuple[str, str]:
+        # Backstop for callers that didn't start the listener via the redirect
+        # handler first (e.g. a resumed flow).
+        ensure_started()
+
         # Wait for callback to be received
         MINUTES = 5
         timeout = MINUTES * 60
@@ -106,23 +159,23 @@ async def get_auth_code() -> Tuple[str, str]:
         start_time = time.time()
         while not callback_received.is_set():
             if time.time() - start_time > timeout:
-                shutdown()
+                await asyncio.to_thread(shutdown)
                 raise Exception(f"OAuth callback timed out after {MINUTES} minutes")
 
             # Small sleep to avoid busy waiting
             await asyncio.sleep(0.1)
 
         if server_error:
-            shutdown()
+            await asyncio.to_thread(shutdown)
             raise Exception(server_error)
 
         if not auth_code:
-            shutdown()
+            await asyncio.to_thread(shutdown)
             raise Exception("No authorization code received")
 
         return auth_code, state
 
-    return get_auth_code, shutdown
+    return get_auth_code, shutdown, ensure_started
 
 
 def get_token_file_path():
diff --git a/cecli/mcp/server.py b/cecli/mcp/server.py
index a38ab2ebd12..a7bf9c172b5 100644
--- a/cecli/mcp/server.py
+++ b/cecli/mcp/server.py
@@ -5,6 +5,7 @@
 import threading
 import webbrowser
 from contextlib import AsyncExitStack
+from datetime import timedelta
 from enum import Enum, auto
 from urllib.parse import urlparse
 
@@ -28,6 +29,10 @@
 MIN_KEEPALIVE_INTERVAL = 5
 MAX_KEEPALIVE_INTERVAL = 300
 FAILED_PING_THRESHOLD = 3
+# Default per-request timeout (seconds) for MCP handshake/tool calls. Two
+# minutes is generous enough for slow first-run bootstraps (uvx/npx/Docker) while
+# still bounding a server that accepts a connection but never speaks MCP.
+DEFAULT_MCP_REQUEST_TIMEOUT = 120
 
 logger = logging.getLogger(__name__)
 
@@ -246,7 +251,13 @@ async def _open_session(self):
                 stdio_client(server_params, errlog=err_file)
             )
             read, write = stdio_transport
-            session = await self.exit_stack.enter_async_context(ClientSession(read, write))
+            session = await self.exit_stack.enter_async_context(
+                ClientSession(
+                    read,
+                    write,
+                    read_timeout_seconds=timedelta(seconds=self._request_timeout_seconds()),
+                )
+            )
             await session.initialize()
             self.session = session
 
@@ -323,6 +334,25 @@ async def _run_session(self):
                 self.session = None
                 self._connection_loop = None
 
+    def _request_timeout_seconds(self) -> float:
+        """Per-request timeout (seconds) for the MCP handshake and tool calls.
+
+        A wedged transport would otherwise block startup forever; the timeout is
+        applied to the SDK ``ClientSession`` so a silent server raises (and can be
+        retried or reported as failed) instead of hanging. Overridable per server
+        via the ``timeout`` config key.
+        """
+        raw = self.config.get("timeout")
+        if raw is not None:
+            try:
+                value = float(raw)
+                if value > 0:
+                    return value
+            except (TypeError, ValueError):
+                pass
+
+        return DEFAULT_MCP_REQUEST_TIMEOUT
+
 
 class HttpBasedMcpServer(McpServer):
     """Base class for HTTP-based MCP servers (HTTP streaming and SSE)."""
@@ -376,12 +406,16 @@ async def _create_oauth_provider(self):
 
         redirect_uri = f"http://localhost:{port}/callback"
 
-        get_auth_code, shutdown = create_oauth_callback_server(port)
+        get_auth_code, shutdown, ensure_callback_server = create_oauth_callback_server(port)
 
         # Store shutdown function for cleanup
         self._oauth_shutdown = shutdown
 
         async def handle_redirect(auth_url: str) -> None:
+            # Start the local listener before opening the browser so the OAuth
+            # redirect can never race the callback server binding.
+            ensure_callback_server()
+
             if self.io:
                 self.io.tool_output(f"\nAuthentication required for MCP server: {self.name}")
                 self.io.tool_output("\nPlease open this URL in your browser to authenticate:")
@@ -440,7 +474,13 @@ async def _open_session(self):
 
         read, write = _unpack_transport(transport)
 
-        session = await self.exit_stack.enter_async_context(ClientSession(read, write))
+        session = await self.exit_stack.enter_async_context(
+            ClientSession(
+                read,
+                write,
+                read_timeout_seconds=timedelta(seconds=self._request_timeout_seconds()),
+            )
+        )
         await session.initialize()
         self.session = session
 
@@ -580,7 +620,10 @@ async def _close_session(self, cancel_keepalive: bool = True):
             logger.info(f"Keepalive task stopped for {self.name}")
 
         if hasattr(self, "_oauth_shutdown"):
-            self._oauth_shutdown()
+            # Run the blocking socketserver shutdown off-loop so a stuck callback
+            # server can't wedge the MCP event loop (and so the surrounding
+            # wait_for timeout can still fire).
+            await asyncio.to_thread(self._oauth_shutdown)
 
         self._http_client = None
 
diff --git a/cecli/tui/worker.py b/cecli/tui/worker.py
index 9553bbe13d0..e5fa92fdd7e 100644
--- a/cecli/tui/worker.py
+++ b/cecli/tui/worker.py
@@ -10,6 +10,7 @@
 from cecli.coders import Coder
 from cecli.commands import ReloadProgramSignal, SwitchCoderSignal
 from cecli.helpers.conversation import ConversationService, MessageTag
+from cecli.helpers.coroutines import task_is_cancelling
 
 logger = logging.getLogger(__name__)
 # Suppress asyncio task destroyed warnings during shutdown
@@ -60,10 +61,14 @@ def _run_thread(self):
 
         try:
             self.loop.run_until_complete(self._async_run())
-        except BaseException:
-            # Catch anything that could bring down the thread, and just let it exit.
-            # This includes KeyboardInterrupt, SystemExit, etc.
-            pass
+        except BaseException as e:
+            # A normal stop() stops the loop, which makes run_until_complete
+            # raise RuntimeError; that and a cancellation after running=False
+            # are expected shutdown paths, not crashes.
+            graceful = not self.running and isinstance(e, (asyncio.CancelledError, RuntimeError))
+            if not graceful:
+                logger.error("Coder worker thread stopped unexpectedly", exc_info=e)
+                self._notify_crash(e)
         finally:
             self._cleanup_loop()
 
@@ -117,6 +122,14 @@ async def _async_run(self):
         if mcp_manager is not None:
             try:
                 await mcp_manager.connect_all()
+            except asyncio.CancelledError:
+                # connect_all uses gather; a single transport (e.g. MCP's
+                # streamable-HTTP) can surface an ordinary connection failure
+                # as CancelledError and abort the whole gather. Only propagate
+                # a genuine cancellation of this worker.
+                if task_is_cancelling():
+                    raise
+                logger.warning("MCP connect_all was cancelled by a server transport; continuing")
             except Exception as e:
                 logger.error("Failed to connect MCP servers in worker: %s", e, exc_info=True)
 
@@ -285,6 +298,23 @@ def stop(self):
         if self.thread and self.thread.is_alive():
             self.thread.join(timeout=2.0)
 
+    def _notify_crash(self, exc):
+        """Tell the TUI the worker died so it can surface the error and exit.
+
+        Without this the TUI keeps running with a dead worker and appears hung.
+        """
+        try:
+            self.output_queue.put(
+                {
+                    "type": "error",
+                    "message": f"Worker stopped unexpectedly: {exc!r}",
+                    "coder_uuid": getattr(self.coder, "uuid", None),
+                }
+            )
+            self.output_queue.put({"type": "exit"})
+        except Exception:
+            pass
+
     def _create_event_loop(self):
         """Create the event loop used by the coder worker thread.
 
diff --git a/cecli/website/docs/config/mcp.md b/cecli/website/docs/config/mcp.md
index 9bd5e516673..8af15affee2 100644
--- a/cecli/website/docs/config/mcp.md
+++ b/cecli/website/docs/config/mcp.md
@@ -30,6 +30,37 @@ mcp-servers:
       keepalive_interval: 60 # Send a heartbeat every 60 seconds
 ```
 
+### Request Timeout
+
+`timeout`: (Optional) The per-request timeout in **seconds** for a server's connection handshake and subsequent requests. This keeps a server that accepts a connection but never responds from blocking cecli indefinitely at startup.
+
+- `timeout`: (Optional) A positive number of seconds.
+  - If not provided, it defaults to **120** seconds (2 minutes) — generous enough for slow first-run bootstraps while still bounding a server that never responds.
+  - It bounds the `initialize` handshake, the initial tool listing (`list_tools`), and later requests made on that server's session.
+
+A server that stays silent past its timeout is marked as failed to connect (it is retried up to **3** times) instead of hanging cecli, so increase `timeout` for servers that legitimately take a while to start — for example a first-run `uvx`/`npx` download or a Docker image pull.
+
+Example with a longer timeout:
+
+```yaml
+mcp-servers:
+  mcpServers:
+    serena:
+      transport: stdio
+      command: uvx
+      args: [
+        "--from",
+        "git+https://github.com/oraios/serena",
+        "serena",
+        "start-mcp-server",
+        "--context",
+        "ide",
+        "--project",
+        "/path/to/project"
+      ]
+      timeout: 300 # Allow extra time for a first-run build
+```
+
 You have two ways of sharing your MCP server configuration with cecli.
 
 
diff --git a/tests/mcp/test_manager_retry.py b/tests/mcp/test_manager_retry.py
index 850938702cd..734d4807245 100644
--- a/tests/mcp/test_manager_retry.py
+++ b/tests/mcp/test_manager_retry.py
@@ -323,15 +323,48 @@ async def test_connect_server_propagates_cancelled_error_during_retry(mock_serve
 
 
 @pytest.mark.asyncio
-async def test_connect_server_propagates_cancelled_error_during_connect(mock_server, mock_io):
-    """TC-008: connect_server re-raises CancelledError when server.connect() raises it."""
+async def test_connect_server_treats_transport_cancellation_as_failure(mock_server, mock_io):
+    """TC-008: a transport-level CancelledError is a failed attempt, not a propagated cancel.
+
+    MCP's streamable-HTTP transport surfaces an unreachable server as
+    CancelledError from its anyio TaskGroup; it must not abort startup or the
+    caller, so it is retried and reported like any other connection failure.
+    """
     manager = McpServerManager(servers=[mock_server], io=mock_io)
     mock_server.connect.side_effect = asyncio.CancelledError()
 
+    with patch("asyncio.sleep"):
+        result = await manager.connect_server("test-server")
+
+    assert result is False
+    assert mock_server.connect.call_count == 3
+    assert mock_io.tool_error.call_count == 1
+
+
+# ---------------------------------------------------------------------------
+# TC-008b: connect_server propagates cancellation of the calling task
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_connect_server_propagates_genuine_cancellation(mock_server, mock_io):
+    """TC-008b: cancelling the calling task still propagates out of connect_server."""
+    manager = McpServerManager(servers=[mock_server], io=mock_io)
+
+    started = asyncio.Event()
+
+    async def _slow_connect():
+        started.set()
+        await asyncio.sleep(3600)
+
+    mock_server.connect.side_effect = _slow_connect
+    task = asyncio.create_task(manager.connect_server("test-server"))
+    await started.wait()
+    task.cancel()
+
     with pytest.raises(asyncio.CancelledError):
-        await manager.connect_server("test-server")
+        await task
 
-    assert mock_server.connect.call_count == 1
     mock_io.tool_error.assert_not_called()
 
 

From aead8d1dd234cce963a35561547230fc9bc8accc Mon Sep 17 00:00:00 2001
From: Dustin Washington <dwash96@gmail.com>
Date: Tue, 15 Sep 2026 02:33:40 -0400
Subject: [PATCH 4/6] Update cancellation expectations for python 3.10

---
 cecli/helpers/coroutines.py     | 7 ++++++-
 tests/mcp/test_manager_retry.py | 9 +++++++++
 2 files changed, 15 insertions(+), 1 deletion(-)

diff --git a/cecli/helpers/coroutines.py b/cecli/helpers/coroutines.py
index 0d2ae4c8294..3676d676b55 100644
--- a/cecli/helpers/coroutines.py
+++ b/cecli/helpers/coroutines.py
@@ -117,6 +117,11 @@ def task_is_cancelling() -> bool:
     Used to tell a genuine cancellation of the caller apart from cancellation
     errors that transports (e.g. MCP's anyio TaskGroups) surface for ordinary
     connection failures.
+
+    Reliable on Python 3.11+ (``Task.cancelling()``). On 3.10 there is no public
+    signal: ``_must_cancel`` is already cleared by the time the CancelledError is
+    delivered, so this is best-effort and usually reports False. Callers must
+    therefore treat False as "not known to be cancelling" rather than proof.
     """
     task = asyncio.current_task()
     if task is None:
@@ -126,5 +131,5 @@ def task_is_cancelling() -> bool:
     if cancelling is not None:
         return cancelling() > 0
 
-    # Python 3.10 has no Task.cancelling(); fall back to the private flag.
+    # Python 3.10 fallback: best-effort only (see docstring).
     return bool(getattr(task, "_must_cancel", False))
diff --git a/tests/mcp/test_manager_retry.py b/tests/mcp/test_manager_retry.py
index 734d4807245..976a7eac2d2 100644
--- a/tests/mcp/test_manager_retry.py
+++ b/tests/mcp/test_manager_retry.py
@@ -15,6 +15,7 @@
 """
 
 import asyncio
+import sys
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -346,6 +347,14 @@ async def test_connect_server_treats_transport_cancellation_as_failure(mock_serv
 # ---------------------------------------------------------------------------
 
 
+# Python 3.10 cannot tell a real cancellation from a transport-level one: it has no
+# Task.cancelling(), _must_cancel is already cleared when the handler runs, and the
+# traceback loses the origin frame. task_is_cancelling() therefore reports False
+# there, so this propagation guarantee only holds on 3.11+.
+@pytest.mark.skipif(
+    sys.version_info < (3, 11),
+    reason="Task.cancelling() is required to distinguish a real cancel from a transport cancel",
+)
 @pytest.mark.asyncio
 async def test_connect_server_propagates_genuine_cancellation(mock_server, mock_io):
     """TC-008b: cancelling the calling task still propagates out of connect_server."""

From 24cf3948ae71e006992888d9d547edd9588c417f Mon Sep 17 00:00:00 2001
From: Dustin Washington <dwash96@gmail.com>
Date: Tue, 15 Sep 2026 02:59:18 -0400
Subject: [PATCH 5/6] Make MCP connection timeout changes more
 explicit/idiomatic

---
 cecli/helpers/coroutines.py |  6 +++---
 cecli/mcp/manager.py        | 14 ++++++++-----
 cecli/mcp/oauth.py          | 35 ++++++++++++++++++---------------
 cecli/mcp/server.py         | 39 ++++++++++++++++++++-----------------
 cecli/tui/worker.py         | 15 +++++++++-----
 5 files changed, 62 insertions(+), 47 deletions(-)

diff --git a/cecli/helpers/coroutines.py b/cecli/helpers/coroutines.py
index 3676d676b55..3d4b4b46089 100644
--- a/cecli/helpers/coroutines.py
+++ b/cecli/helpers/coroutines.py
@@ -127,9 +127,9 @@ def task_is_cancelling() -> bool:
     if task is None:
         return False
 
-    cancelling = getattr(task, "cancelling", None)
-    if cancelling is not None:
-        return cancelling() > 0
+    cancelling_fn = getattr(task, "cancelling", None)
+    if cancelling_fn is not None:
+        return cancelling_fn() > 0
 
     # Python 3.10 fallback: best-effort only (see docstring).
     return bool(getattr(task, "_must_cancel", False))
diff --git a/cecli/mcp/manager.py b/cecli/mcp/manager.py
index a2216e0267b..2063c09fc77 100644
--- a/cecli/mcp/manager.py
+++ b/cecli/mcp/manager.py
@@ -4,6 +4,10 @@
 from cecli.mcp.server import DEFAULT_MCP_REQUEST_TIMEOUT, LocalServer, McpServer
 from cecli.tools.utils.registry import ToolRegistry
 
+# Slack added on top of a server's request timeout for the connect/list_tools
+# backstop, covering time the transport spends outside the SDK's read timeout.
+CONNECT_BACKSTOP_GRACE_SECONDS = 5
+
 
 class McpServerManager:
     """
@@ -204,7 +208,8 @@ async def connect_server(self, name: str) -> bool:
         # When io is None (e.g., during from_servers before IO is assigned),
         # _log_warning and _log_error silently return — retries still happen
         # but with no user-visible feedback. This is intentional.
-        max_retries = 3 if server.name != "unnamed-server" else 1
+        is_unnamed = server.name == "unnamed-server"
+        max_retries = 1 if is_unnamed else 3
         delay = 1.0
         backoff = 2.0
         max_delay = 30.0
@@ -219,10 +224,9 @@ async def connect_server(self, name: str) -> bool:
         if base_timeout <= 0:
             base_timeout = DEFAULT_MCP_REQUEST_TIMEOUT
 
-        attempt_timeout = base_timeout + 5
+        attempt_timeout = base_timeout + CONNECT_BACKSTOP_GRACE_SECONDS
 
         for attempt in range(1, max_retries + 1):
-            error = None
 
             try:
                 session = await asyncio.wait_for(server.connect(), timeout=attempt_timeout)
@@ -239,7 +243,7 @@ async def connect_server(self, name: str) -> bool:
             except Exception as e:
                 error = e
 
-            if attempt < max_retries and server.name != "unnamed-server":
+            if attempt < max_retries:
                 self._log_warning(
                     f"Connection attempt {attempt} failed for {name}, "
                     f"retrying in {delay}s... ({error})"
@@ -247,7 +251,7 @@ async def connect_server(self, name: str) -> bool:
                 await asyncio.sleep(delay)
                 delay = min(delay * backoff, max_delay)
             else:
-                if server.name != "unnamed-server":
+                if not is_unnamed:
                     self._log_error(
                         f"Failed to connect to MCP server {name} "
                         f"after {max_retries} attempts: {error}"
diff --git a/cecli/mcp/oauth.py b/cecli/mcp/oauth.py
index 31ff00f0475..5804e9bdd95 100644
--- a/cecli/mcp/oauth.py
+++ b/cecli/mcp/oauth.py
@@ -14,6 +14,10 @@
 
 from cecli.decoding import safe_open
 
+# How long ``ensure_started`` waits for the callback listener to bind before
+# giving up so the browser redirect can't race the listener coming up.
+CALLBACK_BIND_TIMEOUT_SECONDS = 5
+
 
 def create_oauth_callback_server(port, path="/callback") -> Tuple[
     Callable[[], Awaitable[Tuple[str, str]]],
@@ -35,7 +39,6 @@ def create_oauth_callback_server(port, path="/callback") -> Tuple[
     server_error = None
     callback_received = threading.Event()
     server = None
-    server_thread = None
     server_started = threading.Event()
     start_lock = threading.Lock()
 
@@ -100,27 +103,19 @@ def start_server():
             server_started.set()
             callback_received.set()
         finally:
-            if srv is not None:
-                try:
-                    srv.server_close()
-                except Exception:
-                    pass
+            _close_quietly(srv)
 
     def ensure_started():
         """Start the callback listener once, blocking briefly for the bind."""
-        nonlocal server_thread
         with start_lock:
             if server_started.is_set():
                 return
 
-            server_thread = threading.Thread(
-                target=start_server, daemon=True, name="oauth-callback-server"
-            )
-            server_thread.start()
+            threading.Thread(target=start_server, daemon=True, name="oauth-callback-server").start()
 
         # Wait for the bind to complete so the browser redirect cannot race the
         # listener coming up.
-        server_started.wait(timeout=5)
+        server_started.wait(timeout=CALLBACK_BIND_TIMEOUT_SECONDS)
 
     def shutdown():
         """Stop the callback listener. Idempotent and safe to call repeatedly.
@@ -142,10 +137,7 @@ def shutdown():
         except Exception:
             pass
         finally:
-            try:
-                srv.server_close()
-            except Exception:
-                pass
+            _close_quietly(srv)
 
     async def get_auth_code() -> Tuple[str, str]:
         # Backstop for callers that didn't start the listener via the redirect
@@ -278,3 +270,14 @@ async def set_client_info(self, client_info: OAuthClientInformationFull) -> None
 
         all_tokens[self.server_name]["client_info"] = json.loads(client_info.model_dump_json())
         save_mcp_oauth_tokens(all_tokens)
+
+
+def _close_quietly(server) -> None:
+    """Close an HTTP server socket, ignoring a missing server or close errors."""
+    if server is None:
+        return
+
+    try:
+        server.server_close()
+    except Exception:
+        pass
diff --git a/cecli/mcp/server.py b/cecli/mcp/server.py
index a7bf9c172b5..c3758e76a55 100644
--- a/cecli/mcp/server.py
+++ b/cecli/mcp/server.py
@@ -251,15 +251,7 @@ async def _open_session(self):
                 stdio_client(server_params, errlog=err_file)
             )
             read, write = stdio_transport
-            session = await self.exit_stack.enter_async_context(
-                ClientSession(
-                    read,
-                    write,
-                    read_timeout_seconds=timedelta(seconds=self._request_timeout_seconds()),
-                )
-            )
-            await session.initialize()
-            self.session = session
+            session = await self._enter_client_session(read, write)
 
         return session
 
@@ -334,6 +326,25 @@ async def _run_session(self):
                 self.session = None
                 self._connection_loop = None
 
+    async def _enter_client_session(self, read, write):
+        """Enter a client session on the shared exit stack and initialize it.
+
+        Builds the SDK ``ClientSession`` with the server's configured request
+        timeout so the stdio and HTTP transports share identical timeout
+        behavior, then stores the initialized session on ``self.session``.
+        """
+        session = await self.exit_stack.enter_async_context(
+            ClientSession(
+                read,
+                write,
+                read_timeout_seconds=timedelta(seconds=self._request_timeout_seconds()),
+            )
+        )
+        await session.initialize()
+        self.session = session
+
+        return session
+
     def _request_timeout_seconds(self) -> float:
         """Per-request timeout (seconds) for the MCP handshake and tool calls.
 
@@ -474,15 +485,7 @@ async def _open_session(self):
 
         read, write = _unpack_transport(transport)
 
-        session = await self.exit_stack.enter_async_context(
-            ClientSession(
-                read,
-                write,
-                read_timeout_seconds=timedelta(seconds=self._request_timeout_seconds()),
-            )
-        )
-        await session.initialize()
-        self.session = session
+        session = await self._enter_client_session(read, write)
 
         await self.start_keepalive()
 
diff --git a/cecli/tui/worker.py b/cecli/tui/worker.py
index e5fa92fdd7e..b1602657e7f 100644
--- a/cecli/tui/worker.py
+++ b/cecli/tui/worker.py
@@ -62,11 +62,7 @@ def _run_thread(self):
         try:
             self.loop.run_until_complete(self._async_run())
         except BaseException as e:
-            # A normal stop() stops the loop, which makes run_until_complete
-            # raise RuntimeError; that and a cancellation after running=False
-            # are expected shutdown paths, not crashes.
-            graceful = not self.running and isinstance(e, (asyncio.CancelledError, RuntimeError))
-            if not graceful:
+            if not self._is_graceful_shutdown(e):
                 logger.error("Coder worker thread stopped unexpectedly", exc_info=e)
                 self._notify_crash(e)
         finally:
@@ -298,6 +294,15 @@ def stop(self):
         if self.thread and self.thread.is_alive():
             self.thread.join(timeout=2.0)
 
+    def _is_graceful_shutdown(self, exc) -> bool:
+        """Return True when a worker-loop exception is an expected shutdown artifact.
+
+        A normal stop() stops the loop, which surfaces from run_until_complete
+        as RuntimeError; a cancellation after running goes False surfaces as
+        CancelledError. Neither is a crash.
+        """
+        return not self.running and isinstance(exc, (asyncio.CancelledError, RuntimeError))
+
     def _notify_crash(self, exc):
         """Tell the TUI the worker died so it can surface the error and exit.
 

From 51e92ff6518bd1f937c205973813822080109cc8 Mon Sep 17 00:00:00 2001
From: Dustin Washington <dwash96@gmail.com>
Date: Thu, 17 Sep 2026 09:16:27 -0400
Subject: [PATCH 6/6] Omit temperature entirely for github copilot provider

---
 cecli/helpers/model_config/agent.py | 11 +++++++++--
 cecli/models.py                     |  6 ++++--
 tests/basic/test_models.py          | 12 ++++++++++++
 tests/helpers/test_model_config.py  | 17 +++++++++++++++++
 4 files changed, 42 insertions(+), 4 deletions(-)

diff --git a/cecli/helpers/model_config/agent.py b/cecli/helpers/model_config/agent.py
index 863bf13c251..f5e9615f89b 100644
--- a/cecli/helpers/model_config/agent.py
+++ b/cecli/helpers/model_config/agent.py
@@ -9,7 +9,7 @@
 
 from typing import Dict, Optional
 
-from .identifiers import is_anthropic
+from .identifiers import is_anthropic, is_github_copilot
 from .utils import supports_reasoning
 
 
@@ -67,7 +67,14 @@ def derive_agent_config(provider: Optional[str], route: str, record: Optional[Di
         "uses_messages_api": uses_messages_api,
     }
 
-    if reasoning or record.get("supports_adaptive_thinking"):
+    if (
+        reasoning
+        or record.get("supports_adaptive_thinking")
+        or is_github_copilot(provider, route, record)
+    ):
+        # Reasoning, adaptive and GitHub Copilot models do not take an explicit
+        # sampling temperature; omit the parameter entirely rather than sending
+        # the generic default of 0.
         agent["use_temperature"] = False
 
     return agent
diff --git a/cecli/models.py b/cecli/models.py
index 351d689f574..04d5e199474 100644
--- a/cecli/models.py
+++ b/cecli/models.py
@@ -1348,8 +1348,10 @@ async def send_completion(
                     temperature = float(self.use_temperature)
             kwargs["temperature"] = temperature
         else:
-            if override_kwargs and override_kwargs.get("temperature", None):
-                override_kwargs.pop("temperature", None)
+            # Omit temperature entirely when the model does not use it; the
+            # key must be dropped even when its override value is falsy (0).
+            if override_kwargs and "temperature" in override_kwargs:
+                override_kwargs.pop("temperature")
 
         effective_tools = tools
 
diff --git a/tests/basic/test_models.py b/tests/basic/test_models.py
index 748f8b8072b..a7357abe027 100644
--- a/tests/basic/test_models.py
+++ b/tests/basic/test_models.py
@@ -511,6 +511,18 @@ def test_use_temperature_settings(self):
         model.use_temperature = 0.7
         assert model.use_temperature == 0.7
 
+    @patch("cecli.models.litellm.acompletion")
+    async def test_use_temperature_false_omits_temperature(self, mock_completion):
+        model = Model("github/o1-mini")
+        model.extra_params = {}
+        messages = [{"role": "user", "content": "Hello"}]
+
+        await model.send_completion(
+            messages, functions=None, stream=False, override_kwargs={"temperature": 0}
+        )
+
+        assert "temperature" not in mock_completion.call_args.kwargs
+
     @patch("cecli.models.litellm.acompletion")
     async def test_request_timeout_default(self, mock_completion):
         model = Model("gpt-4")
diff --git a/tests/helpers/test_model_config.py b/tests/helpers/test_model_config.py
index b228cae5a2a..80f2a90e37a 100644
--- a/tests/helpers/test_model_config.py
+++ b/tests/helpers/test_model_config.py
@@ -634,6 +634,23 @@ def test_non_gpt_github_copilot_stays_chat():
     assert config["llm"]["mode"] == "chat"
 
 
+def test_github_copilot_models_omit_temperature():
+    """Copilot models never send a sampling temperature, even in chat mode.
+
+    Copilot models are absent from the individual model configs, so the
+    metadata-derived agent block supplies the rule.
+    """
+    record = _record(
+        litellm_provider="github_copilot",
+        supports_reasoning=False,
+        supported_endpoints=["/v1/chat/completions"],
+    )
+    config = get_default_config("github_copilot/gpt-4o", [{"github_copilot/gpt-4o": record}])
+
+    assert config["llm"]["mode"] == "chat"
+    assert config["agent"]["use_temperature"] is False
+
+
 def test_adaptive_thinking_sets_use_temperature_false():
     record = _record(supports_reasoning=False, supports_adaptive_thinking=True)
     config = get_default_config("adaptive", [{"adaptive": record}])