fix: coalesce interactive completion backlogs without losing identity

Based on the earliest batching proposal by BrunoBza (#104686) and JoaoMarcos44 structured correction (#104703). Slim redo into topical siblings and shared TUI poller/post-turn routing. Reported by Xipong (#104671). Local shell-to-loopback probes demonstrate 12 to 1 turn dispatches; suites await the campaign test lock.

Co-authored-by: joaomarcos <joaomarcosdias444@gmail.com>

cli.py

@@ -45,6 +45,7 @@ from hermes_cli.cli_model_switch_mixin import CLIModelSwitchMixin
 from hermes_cli.cli_voice_mixin import CLIVoiceMixin
 from hermes_cli.cli_status_bar_mixin import CLIStatusBarMixin
 from hermes_cli.cli_tui_mixin import CLITuiMixin
+from hermes_cli.cli_process_notifications import CLIProcessNotificationsMixin
 from agent.interrupt_compat import request_hard_interrupt
 from agent.pet import render as pet_render
 
@@ -2526,7 +2527,7 @@ from hermes_cli.cli_chat_turn_mixin import CLIChatTurnMixin
 _PASTE_REF_RE = re.compile(r'\[Pasted text #\d+: \d+ lines \u2192 (.+?)\]')
 
 
-class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMixin, CLIStatusBarMixin, CLIVoiceMixin, CLIModelSwitchMixin, CLISessionMixin, CLIStreamMixin, CLIModalMixin, CLITerminalMixin, CLIInfoMixin, CLILoopsMixin, CLIChatTurnMixin):
+class HermesCLI(CLIProcessNotificationsMixin, CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMixin, CLIStatusBarMixin, CLIVoiceMixin, CLIModelSwitchMixin, CLISessionMixin, CLIStreamMixin, CLIModalMixin, CLITerminalMixin, CLIInfoMixin, CLILoopsMixin, CLIChatTurnMixin):
     """Interactive REPL for the Hermes Agent."""
 
     # Seeded -q first message (see _should_seed_interactive); run() re-creates
@@ -3377,40 +3378,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix
             _cprint(f"{_DIM}{_ACCENT}Type /help for available commands{_RST}")
         return True
 
-    def _owns_process_notification(self, event: dict) -> bool:
-        """Whether this session owns a delegation event (pre-compression keys resolve to their continuation; fail closed)."""
-        event_key = str(event.get("session_key") or "")
-        current_key = str(getattr(self, "session_id", "") or "")
-        if not event_key or not current_key:
-            return False
-        if event_key == current_key:
-            return True
-        try:
-            session_db = getattr(self, "_session_db", None)
-            resolved_key = (
-                session_db.resolve_resume_session_id(event_key) if session_db is not None else event_key
-            ) or event_key
-        except Exception:
-            resolved_key = event_key
-        return str(resolved_key) == current_key
-
-    def _drain_process_notifications(self, consumer: str) -> None:
-        """Queue background notifications owned by this session (drained with our stable identity so another window can't claim them)."""
-        from tools.process_registry import process_registry
-        from tools.async_delegation import claim_event_delivery, complete_event_delivery
-
-        for event, synthetic_message in process_registry.drain_notifications(
-            session_key=getattr(self, "session_id", "") or "", owns_event=self._owns_process_notification,
-        ):
-            claim = claim_event_delivery(event, consumer)
-            if claim is None:
-                continue
-            if event.get("type") == "async_delegation":
-                from tools.process_registry_notifications import SubagentNotification
-                synthetic_message = SubagentNotification(synthetic_message, event)
-            self._pending_input.put(synthetic_message)
-            complete_event_delivery(event, claim)
-
     def _drain_interrupt_queue_to_pending_input(self) -> None:
         """Move stray ``_interrupt_queue`` messages into ``_pending_input`` after every turn.
 
@@ -3475,18 +3442,6 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix
             with suppress(Exception):
                 step()
 
-    def _tui_unwrap_input(self, user_input):
-        """Unwrap ``_VoiceInputMessage`` / ``_SeededQueryMessage`` -> ``(text_or_tuple, is_voice_input, is_seeded_query)``."""
-        # Voice-transcribed messages arrive wrapped in a sentinel so only genuine STT output gets the voice
-        # prefix (#65827).
-        is_voice_input = isinstance(user_input, _VoiceInputMessage)
-        if is_voice_input:
-            user_input = user_input.text
-        is_seeded_query = isinstance(user_input, _SeededQueryMessage)
-        if is_seeded_query:
-            user_input = (user_input.text, user_input.images) if user_input.images else user_input.text
-        return user_input, is_voice_input, is_seeded_query
-
     def _tui_process_one_input(self, user_input):
         """Route one submitted input: file drop, /resume pick, ! shell, slash command, or a chat turn."""
         from tools.process_registry_notifications import SubagentNotification

evals/completion_backlog_probe.py

@@ -0,0 +1,171 @@
+"""Local I/O probe for completion batching, not a hosted-model or native UI test.
+
+Run with the repo's Python: evals/completion_backlog_probe.py REPO OUTPUT.json.
+Real shell children feed ProcessRegistry; production CLI/poller/post-turn routing
+feeds a loopback HTTP turn sink. The sink replaces chat/_run_prompt_submit, NOT
+ownership, consumption, queue draining, batching, or formatting.
+"""
+import argparse
+import contextlib
+import json
+import os
+from pathlib import Path
+import queue
+import shlex
+import sys
+import tempfile
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from types import SimpleNamespace
+from unittest.mock import patch
+from urllib.request import Request, urlopen
+
+
+def probe(surface, scenario, directory):
+    from cli import HermesCLI
+    from tools import process_registry as pr
+    from tools.process_registry_notifications import format_process_notification
+    from tui_gateway import server
+
+    directory = Path(directory)
+    directory.mkdir(parents=True, exist_ok=True)
+    registry = pr.ProcessRegistry()
+    received, statuses = [], []
+
+    class Sink(BaseHTTPRequestHandler):
+        def do_POST(self):
+            received.append(json.loads(self.rfile.read(int(self.headers['Content-Length']))))
+            self.send_response(200)
+            self.end_headers()
+            self.wfile.write(b'{"accepted":true}')
+
+        def log_message(self, *_args):
+            return
+
+    wire = ThreadingHTTPServer(('127.0.0.1', 0), Sink)
+    thread = threading.Thread(target=wire.serve_forever, daemon=True)
+    thread.start()
+    session = {'session_key': 'backlog-owner', 'history_lock': threading.RLock()}
+    cli = HermesCLI.__new__(HermesCLI)
+    cli.session_id = session['session_key']
+    cli._session_db = None
+    cli._pending_input = queue.Queue()
+    cli._pending_resume_sessions = []
+    cli._typed_voice_stop = lambda _text: False
+    cli.handle_bang_shell = lambda _text: False
+    cli._print_user_message_preview = lambda _text: None
+    cli._turn_summary_begin = lambda: None
+    cli._app = SimpleNamespace(invalidate=lambda: None)
+    cli._tui_after_turn = lambda: None
+
+    def submit(text):
+        request = Request(f'http://127.0.0.1:{wire.server_port}/turn',
+                          data=json.dumps({'text': str(text)}).encode(),
+                          headers={'Content-Type': 'application/json'})
+        with urlopen(request, timeout=10) as response:
+            assert response.status == 200
+        session['running'] = False
+        return True
+
+    def tui_submit(_rid, _sid, _session, text, **_kwargs):
+        return submit(text)
+
+    cli.chat = lambda text, **_kwargs: submit(text)
+    processes = []
+    try:
+        with contextlib.ExitStack() as stack:
+            stack.enter_context(patch.object(pr, 'process_registry', registry))
+            stack.enter_context(patch.object(pr, 'CHECKPOINT_PATH', directory / 'processes.json'))
+            stack.enter_context(patch.object(server, '_sessions', {'ui-owner': session}))
+            stack.enter_context(patch.object(server, '_get_db', lambda: None))
+            stack.enter_context(patch.object(server, '_emit', lambda *args: statuses.append(args)))
+            stack.enter_context(patch.object(server, '_run_prompt_submit', tui_submit))
+            stack.enter_context(patch.object(server, '_drain_queued_prompt', lambda *_a: False))
+            count = 1 if scenario == 'single' else 12
+            gate = directory / 'release'
+            for index in range(count):
+                code = ('import pathlib,time,sys; p=pathlib.Path(sys.argv[1]); '
+                        '\nwhile not p.exists(): time.sleep(.01)\n'
+                        f'print("BACKLOG_{index}"); sys.exit({7 if index == count - 1 else 0})')
+                process = registry.spawn_local(
+                    f'{shlex.quote(sys.executable)} -c {shlex.quote(code)} {shlex.quote(str(gate))}',
+                    cwd=str(directory), session_key='backlog-owner')
+                process.notify_on_complete = True
+                processes.append(process)
+            gate.touch()
+            deadline = time.monotonic() + 30
+            while registry.completion_queue.qsize() < count and time.monotonic() < deadline:
+                time.sleep(.01)
+            assert registry.completion_queue.qsize() == count
+            raw = list(registry.completion_queue.queue)
+            expected = [format_process_notification(event) for event in raw]
+            if scenario == 'foreign':
+                session['session_key'] = cli.session_id = 'another-owner'
+                server._sessions['actual-owner'] = {'session_key': 'backlog-owner'}
+            if surface == 'cli':
+                cli._drain_process_notifications('cli-idle')
+                if scenario == 'consumed':
+                    for process in processes:
+                        assert registry.wait(process.id, timeout=1)["status"] == "exited"
+                while not cli._pending_input.empty():
+                    cli._tui_process_one_input(cli._pending_input.get_nowait())
+            elif surface == 'post-turn':
+                if scenario == 'consumed':
+                    for process in processes:
+                        assert registry.wait(process.id, timeout=1)["status"] == "exited"
+                server._run_post_turn_followups('probe', 'ui-owner', session, {}, None)
+            else:
+                if scenario == 'consumed':
+                    for process in processes:
+                        assert registry.wait(process.id, timeout=1)["status"] == "exited"
+                stop = threading.Event()
+                # Skip unrelated scheduled jobs; exercise the production poller loop.
+                stack.enter_context(patch.object(server, '_maybe_fire_tui_loop_tick', lambda *_a: None))
+                stack.enter_context(patch.object(server, '_maybe_fire_tui_heartbeat_tick', lambda *_a: None))
+                stack.enter_context(patch.object(server, '_notif_poll_kanban', lambda *_a: None))
+                poller = threading.Thread(target=server._notification_poller_loop,
+                                          args=(stop, 'ui-owner', session))
+                poller.start()
+                deadline = time.monotonic() + (1 if scenario == 'foreign' else 10)
+                while time.monotonic() < deadline:
+                    if scenario != 'foreign' and registry.completion_queue.empty():
+                        break
+                    time.sleep(.01)
+                stop.set()
+                poller.join(10)
+                assert not poller.is_alive()
+            texts = [item['text'] for item in received]
+            return {'surface': surface, 'scenario': scenario, 'children': count,
+                    'wire_turns': len(texts), 'texts': texts,
+                    'all_payloads_preserved': all(any(text in turn for turn in texts) for text in expected),
+                    'single_exact': texts == expected if count == 1 else None,
+                    'queue_remaining': registry.completion_queue.qsize(),
+                    'status_events': len(statuses)}
+    finally:
+        wire.shutdown()
+        wire.server_close()
+        thread.join()
+        for process in processes:
+            if not process.exited:
+                registry.kill_process(process.id)
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument('repo')
+    parser.add_argument('output')
+    args = parser.parse_args()
+    sys.path.insert(0, args.repo)
+    with tempfile.TemporaryDirectory(prefix='completion-probe-') as home:
+        os.environ['HERMES_HOME'] = home
+        os.environ['HOME'] = home
+        results = [probe(surface, scenario, Path(home) / surface / scenario)
+                   for surface in ('cli', 'poller', 'post-turn')
+                   for scenario in ('backlog', 'single', 'consumed', 'foreign')]
+    Path(args.output).write_text(json.dumps(results, indent=2), encoding="utf-8")
+    print(json.dumps([{k: v for k, v in row.items() if k != 'texts'} for row in results], indent=2))
+
+
+if __name__ == '__main__':
+    main()

hermes_cli/cli_process_notifications.py

@@ -0,0 +1,61 @@
+"""CLI notification ownership, structured queueing and last-moment consumption."""
+
+
+class CLIProcessNotificationsMixin:
+    def _owns_process_notification(self, event: dict) -> bool:
+        """Whether this session owns a delegation event (pre-compression keys resolve to their continuation; fail closed)."""
+        event_key = str(event.get("session_key") or "")
+        current_key = str(getattr(self, "session_id", "") or "")
+        if not event_key or not current_key:
+            return False
+        if event_key == current_key:
+            return True
+        try:
+            session_db = getattr(self, "_session_db", None)
+            resolved_key = (
+                session_db.resolve_resume_session_id(event_key) if session_db is not None else event_key
+            ) or event_key
+        except Exception:
+            resolved_key = event_key
+        return str(resolved_key) == current_key
+
+    def _drain_process_notifications(self, consumer: str) -> None:
+        from tools.process_registry import process_registry
+        from tools.async_delegation import claim_event_delivery, complete_event_delivery
+        from tools.process_registry_notifications import (
+            ProcessNotificationBatch, SubagentNotification, group_process_notifications)
+
+        claimed = []
+        for event, text in process_registry.drain_notifications(
+            session_key=getattr(self, "session_id", "") or "", owns_event=self._owns_process_notification,
+        ):
+            claim = claim_event_delivery(event, consumer)
+            if claim is None:
+                continue
+            claimed.append((event, text))
+            complete_event_delivery(event, claim)
+        for notifications in group_process_notifications(claimed):
+            event, text = notifications[0]
+            if event.get("type", "completion") == "completion":
+                pending = ProcessNotificationBatch(notifications)
+            else:
+                pending = SubagentNotification(text, event) if event.get("type") == "async_delegation" else text
+            self._pending_input.put(pending)
+
+    def _tui_unwrap_input(self, user_input):
+        """Unwrap ``_VoiceInputMessage`` / ``_SeededQueryMessage`` -> ``(text_or_tuple, is_voice_input, is_seeded_query)``."""
+        from cli import _VoiceInputMessage, _SeededQueryMessage
+        from tools.process_registry import process_registry
+        from tools.process_registry_notifications import ProcessNotificationBatch
+        if isinstance(user_input, ProcessNotificationBatch):
+            user_input = user_input.render(process_registry)
+        # Voice-transcribed messages arrive wrapped in a sentinel so only genuine STT output gets the voice
+        # prefix (#65827).
+        is_voice_input = isinstance(user_input, _VoiceInputMessage)
+        if is_voice_input:
+            user_input = user_input.text
+        is_seeded_query = isinstance(user_input, _SeededQueryMessage)
+        if is_seeded_query:
+            user_input = (user_input.text, user_input.images) if user_input.images else user_input.text
+        return user_input, is_voice_input, is_seeded_query
+

tests/cli/test_completion_backlog.py

@@ -0,0 +1,21 @@
+"""Completion backlogs preserve results without multiplying autonomous turns."""
+from evals.completion_backlog_probe import probe
+
+
+def test_ready_completions_share_one_turn_across_interactive_routes(tmp_path):
+    for surface in ("cli", "poller", "post-turn"):
+        for scenario in ("backlog", "single"):
+            result = probe(surface, scenario, tmp_path / surface / scenario)
+            assert result["wire_turns"] == 1, result
+            assert result["all_payloads_preserved"], result
+            if scenario == "single":
+                assert result["single_exact"], result
+
+
+def test_consumed_or_foreign_completions_never_start_a_turn(tmp_path):
+    for surface in ("cli", "poller", "post-turn"):
+        for scenario in ("consumed", "foreign"):
+            result = probe(surface, scenario, tmp_path / surface / scenario)
+            assert result["wire_turns"] == 0, result
+            if scenario == "foreign":
+                assert result["queue_remaining"] == result["children"], result

tools/process_registry_notifications.py

@@ -4,11 +4,47 @@ watch_match, watch_disabled, watch_overflow_*, async_delegation) into the
 TUI inject into the agent conversation."""
 
 import time
+from dataclasses import dataclass
 from contextlib import suppress
 
 _DONE = ("completed", "success")
 
 
+@dataclass(frozen=True, slots=True)
+class ProcessNotificationBatch:
+    """Keep completion identity until the owning surface starts its turn."""
+
+    notifications: tuple[tuple[dict, str], ...]
+
+    def render(self, registry) -> str | None:
+        messages = [text for event, text in self.notifications
+                    if not registry.is_completion_consumed(event.get("session_id", ""))]
+        if not messages:
+            return None
+        if len(messages) == 1:
+            return messages[0]
+        header = (f"[IMPORTANT: {len(messages)} background processes completed. "
+                  "Treat these results as one batch and give one consolidated response; "
+                  "preserve failures and actionable results.]")
+        return "\n\n".join((header, *messages))
+
+
+def group_process_notifications(notifications):
+    """Group consecutive completions only; watches and delegations are barriers."""
+    batch = []
+    for event, text in notifications:
+        if event.get("type", "completion") == "completion":
+            batch.append((event, text))
+        else:
+            if batch:
+                yield tuple(batch)
+                batch = []
+            yield ((event, text),)
+    if batch:
+        yield tuple(batch)
+
+
+
 def _format_age(seconds: float) -> str:
     """Human-friendly elapsed string ('18m', '2h3m', '45s')."""
     try:

tui_gateway/prompt_turn.py

@@ -395,22 +395,14 @@ def _run_post_turn_followups(
             session_key=session.get("session_key", ""),
             owns_event=lambda e: _session_owns_notification_event(sid, session, e),
             skip_poll_observed=False)
-        for index, (_evt, synth) in enumerate(drained):
-            with session["history_lock"]:
-                if session.get("running"):
-                    for pending_evt, _pending_synth in drained[index:]:
-                        process_registry.completion_queue.put(pending_evt)
-                    break
-                session["running"] = True
-            from tools.async_delegation import (
-                claim_event_delivery, complete_event_delivery, release_event_delivery)
-            _claim = claim_event_delivery(_evt, "tui-post-turn")
-            if _claim is None:
-                continue
-            _dispatch_followup_turn(
-                rid, sid, session, synth, "completion notification dispatch",
-                on_done=lambda: complete_event_delivery(_evt, _claim),
-                on_error=lambda: release_event_delivery(_evt, _claim))
+        from tools.process_registry_notifications import format_process_notification
+        deferred = []
+        _notif_handle_ready(
+            sid, session, [event for event, _text in drained],
+            session.setdefault("_notification_emitted", set()), process_registry,
+            format_process_notification, deferred)
+        for event in deferred:
+            process_registry.completion_queue.put(event)
     except Exception as _drain_exc:
         _hook_failure("completion queue drain", _drain_exc)
 

tui_gateway/session_notifications.py

@@ -394,7 +394,7 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None
     complete_event_delivery(evt, claim)
 
 
-def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> bool:
+def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred, completions=None) -> bool:
     """Route one dequeued event: foreign (another live session owns it) → requeued, or onto ``deferred`` during the
     shutdown drain; unowned (addressed but unprovable — never adopt an orphan) → dropped, except delegation payloads
     deferred for a resume; ours (or ownerless legacy, kept process-global) → status.update once, then an agent turn if
@@ -432,6 +432,9 @@ def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) ->
         display_text = async_delegation_display_text(evt) if is_delegation else text
         _emit("status.update", sid, {"kind": "process", "text": display_text})
         emitted.add(dedup_key)
+    if evt_type == "completion" and completions is not None:
+        completions.append((evt, text))
+        return True
     if not _notif_claim_turn(session):
         queue.put(evt)
         if deferred is not None:
@@ -442,6 +445,49 @@ def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) ->
     return True
 
 
+def _notif_dispatch_completions(sid, session, notifications, registry, deferred):
+    from tools.process_registry_notifications import ProcessNotificationBatch
+    from tools.async_delegation import claim_event_delivery, complete_event_delivery, release_event_delivery
+
+    if not notifications:
+        return
+    if not _notif_claim_turn(session):
+        for event, _text in notifications:
+            (deferred.append if deferred is not None else registry.completion_queue.put)(event)
+        if deferred is None:
+            time.sleep(0.25)
+        return
+    claimed = [(event, text, claim) for event, text in notifications
+               if (claim := claim_event_delivery(event, "tui-completion-batch")) is not None]
+    text = ProcessNotificationBatch(tuple((event, text) for event, text, _claim in claimed)).render(registry)
+    if text is None:
+        _notif_release_turn(session)
+    try:
+        if text is not None:
+            _notif_submit(f"__notif__{int(time.time() * 1000)}", sid, session, text,
+                          "completion batch dispatch failed")
+    except Exception:
+        for event, _text, claim in claimed:
+            release_event_delivery(event, claim)
+        return
+    for event, _text, claim in claimed:
+        complete_event_delivery(event, claim)
+
+
+def _notif_handle_ready(sid, session, events, emitted, registry, fmt, deferred):
+    """One ready snapshot: ownership and UI emission per event, one turn per completion run."""
+    completions = []
+    for index, event in enumerate(events):
+        if event.get("type", "completion") != "completion":
+            _notif_dispatch_completions(sid, session, completions, registry, deferred)
+            completions = []
+        if not _notif_handle_event(sid, session, event, emitted, registry, fmt, deferred, completions):
+            for remaining in events[index + 1:]:
+                (deferred.append if deferred is not None else registry.completion_queue.put)(remaining)
+            break
+    _notif_dispatch_completions(sid, session, completions, registry, deferred)
+
+
 def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None:
     """Daemon thread (started by _init_session()) that drains the process-global completion_queue for this session
     (ownership routing: _notif_handle_event) and polls ``kanban_notify_subs`` every ``_KANBAN_POLL_SECONDS`` — the
@@ -454,9 +500,9 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
     from tools.process_registry import process_registry
     from tools.process_registry_notifications import format_process_notification
     queue = process_registry.completion_queue
-    emitted: set = set()  # dedup re-queued events so one completion isn't emitted 50 times while busy
-    handle = lambda evt, deferred: _notif_handle_event(  # noqa: E731
-        sid, session, evt, emitted, process_registry, format_process_notification, deferred)
+    emitted = session.setdefault("_notification_emitted", set())
+    handle = lambda events, deferred: _notif_handle_ready(  # noqa: E731
+        sid, session, events, emitted, process_registry, format_process_notification, deferred)
     last_kanban_poll = last_loop_poll = 0.0
     while not stop_event.is_set() and not session.get("_finalized"):
         now = time.monotonic()
@@ -476,17 +522,23 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
             evt = queue.get(timeout=0.5)
         except Exception:
             continue
-        handle(evt, None)
+        ready = [evt]
+        for _ in range(queue.qsize()):
+            try:
+                ready.append(queue.get_nowait())
+            except Exception:
+                break
+        handle(ready, None)
     # Drain remaining events after the stop signal so nothing is lost on shutdown; foreign and orphaned-delegation
     # events are handed back to the shared queue afterwards.
     deferred: list = []
-    while not queue.empty():
+    ready = []
+    for _ in range(queue.qsize()):
         try:
-            evt = queue.get_nowait()
+            ready.append(queue.get_nowait())
         except Exception:
             break
-        if not handle(evt, deferred):
-            break
+    handle(ready, deferred)
     for evt in deferred:
         queue.put(evt)