diff --git a/cli.py b/cli.py index 3ab970a7b9..feadecd07d 100644 --- a/cli.py +++ b/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 diff --git a/evals/completion_backlog_probe.py b/evals/completion_backlog_probe.py new file mode 100644 index 0000000000..3e6bf61979 --- /dev/null +++ b/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() diff --git a/hermes_cli/cli_process_notifications.py b/hermes_cli/cli_process_notifications.py new file mode 100644 index 0000000000..f5dd6697b6 --- /dev/null +++ b/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 + diff --git a/tests/cli/test_completion_backlog.py b/tests/cli/test_completion_backlog.py new file mode 100644 index 0000000000..20f5e3e757 --- /dev/null +++ b/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 diff --git a/tools/process_registry_notifications.py b/tools/process_registry_notifications.py index 867155080b..5a82b2cb7e 100644 --- a/tools/process_registry_notifications.py +++ b/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: diff --git a/tui_gateway/prompt_turn.py b/tui_gateway/prompt_turn.py index f89bbe8855..30a95ddcdb 100644 --- a/tui_gateway/prompt_turn.py +++ b/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) diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 6524f3e3f1..3e196f7233 100644 --- a/tui_gateway/session_notifications.py +++ b/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)