V1.6.0

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.

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": (

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:

cecli/helpers/coroutines.py

@@ -109,3 +109,27 @@ 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.
+
+    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:
+        return False
+
+    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))