From 8a82a0b07383ac32660fac570862fbbf59ddb0da Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Wed, 2 Sep 2026 23:20:32 -0700 Subject: [PATCH] =?UTF-8?q?refactor(tui):=20session=5Fnotifications=20?= =?UTF-8?q?=E2=80=94=20fold=20render/classify=20into=20the=20single=20even?= =?UTF-8?q?t=20router?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- tui_gateway/session_auto_continue.py | 8 +-- tui_gateway/session_notifications.py | 104 ++++++++++----------------- tui_gateway/slash_worker.py | 16 +++-- 3 files changed, 48 insertions(+), 80 deletions(-) diff --git a/tui_gateway/session_auto_continue.py b/tui_gateway/session_auto_continue.py index 494fbf5e28..6a0244f530 100644 --- a/tui_gateway/session_auto_continue.py +++ b/tui_gateway/session_auto_continue.py @@ -67,10 +67,6 @@ def _ac_release_turn(session: dict, *, unschedule: bool = False) -> None: session["_auto_continue_scheduled"] = False -def _ac_dispatch_failed(what: str, exc: BaseException) -> None: - print(f"[tui_gateway] {what} dispatch failed: {type(exc).__name__}: {exc}", file=sys.stderr) - - def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) -> dict | None: """Kick off a continuation turn for a crash-interrupted session (session.resume cold paths, once the live record is registered). Returns a descriptor for the resume payload when scheduled, else None. @@ -133,7 +129,7 @@ def _maybe_schedule_auto_continue(sid: str, session: dict, session_key: str) -> _emit("message.start", sid) _run_prompt_submit(rid, sid, session, text, display_kind="auto_continue") except Exception as exc: - _ac_dispatch_failed("auto-continue", exc) + _notif_log_failure("auto-continue dispatch failed", exc) _ac_release_turn(session) threading.Thread(target=kickoff, daemon=True).start() @@ -356,7 +352,7 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool: else: _run_prompt_submit(rid, sid, session, queued["text"], **kwargs) except Exception as exc: - _ac_dispatch_failed("queued prompt", exc) + _notif_log_failure("queued prompt dispatch failed", exc) _ac_release_turn(session) dispatch_failed = True if dispatch_failed: diff --git a/tui_gateway/session_notifications.py b/tui_gateway/session_notifications.py index 5652220ae5..9c4a27f442 100644 --- a/tui_gateway/session_notifications.py +++ b/tui_gateway/session_notifications.py @@ -44,12 +44,9 @@ def _notif_resolve_event_key(evt_key: str) -> str: def _notification_event_belongs_elsewhere(sid: str, session: dict, evt: dict) -> bool: - """True if ``evt`` is owned by a *different* live session. - - Background completions carry the ``session_key`` of the session that started the work; async - delegation completions also carry ``origin_ui_session_id`` (the live TUI tab that commissioned - them). All desktop sessions share one process-wide completion queue, so each poller must skip - events it doesn't own or a detached result surfaces in whichever poller dequeued first.""" + """True if ``evt`` is owned by a *different* live session. Background completions carry the + ``session_key`` of the session that started the work; async delegation completions also carry + ``origin_ui_session_id`` (the live TUI tab that commissioned them).""" evt_ui_sid = str(evt.get("origin_ui_session_id") or "") if evt_ui_sid: if evt_ui_sid == str(sid or "") and not session.get("_finalized"): @@ -307,12 +304,10 @@ def _kb_poll_board(_kb, conn, slug: str, session_key: str) -> list: def _collect_kanban_notifications(session: dict) -> list: - """Claim unseen terminal kanban events for this TUI session's subscriptions. - - ``kanban_create`` auto-subscribes TUI/desktop sessions with ``platform="tui"`` and - ``chat_id=HERMES_SESSION_KEY``; there is no "tui" messaging adapter, so this poller is the delivery - path. Same atomic cursor-claim (``claim_unseen_events_for_sub``) as the gateway notifier, so a sub - is delivered exactly once even if a gateway and a TUI poll the same board DB.""" + """Claim unseen terminal kanban events for this session's ``platform="tui"`` subscriptions + (``kanban_create`` auto-subscribes with ``chat_id=HERMES_SESSION_KEY``; no "tui" messaging adapter + exists, so this poller is the delivery path). Same atomic cursor-claim as the gateway notifier, so + a sub is delivered exactly once even if a gateway and a TUI poll the same board DB.""" session_key = str(session.get("session_key") or "") if not session_key or session.get("_finalized"): return [] @@ -374,22 +369,6 @@ def _notif_poll_kanban(sid: str, session: dict) -> None: ) -def _notif_render_event(sid: str, evt: dict, emitted: set, process_registry, format_process_notification) -> Optional[str]: - """Format ``evt`` and emit its status.update once; None means skip the event (consumed completion - or unformattable). Re-queued completions would otherwise re-emit every 0.5s while the session is - busy, while distinct watch_match events from one process must stay visible.""" - if evt.get("type") == "completion" and process_registry.is_completion_consumed(evt.get("session_id", "")): - return None - text = format_process_notification(evt) - if not text: - return None - dedup_key = _notification_event_dedup_key(evt) - if dedup_key not in emitted: - _emit("status.update", sid, {"kind": "process", "text": text}) - emitted.add(dedup_key) - return text - - def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None: """Run the claimed (running=True) agent turn for one notification event.""" from tools.async_delegation import claim_event_delivery, complete_event_delivery, release_event_delivery @@ -411,32 +390,22 @@ def _notif_dispatch_event(sid: str, session: dict, evt: dict, text: str) -> None complete_event_delivery(evt, claim) -def _notif_classify_event(sid: str, session: dict, evt: dict) -> str: - """'foreign' (another live session owns it), 'unowned' (addressed but unprovable — never adopt an - orphan), or 'mine' (ours, or an ownerless legacy notification kept process-global).""" - if _notification_event_belongs_elsewhere(sid, session, evt): - return "foreign" - if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt): - return "unowned" - return "mine" - - def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> bool: - """Route one dequeued event. ``deferred`` is None in the live loop (foreign events go straight back - on the queue) and the shutdown-drain list otherwise (foreign + orphaned delegation payloads are set - aside for a resume). Returns False when the drain must stop (session busy).""" + """Route one dequeued event: foreign (another live session owns it) → back on the queue, 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 idle. Returns False when the drain + must stop (session busy).""" queue = registry.completion_queue - owner = _notif_classify_event(sid, session, evt) - if owner == "foreign": - if deferred is None: - # Leave foreign events for their owner — otherwise a process started in session A surfaces - # its completion in whichever poller wakes first. + if _notification_event_belongs_elsewhere(sid, session, evt): + if deferred is not None: + deferred.append(evt) + else: + # Otherwise a process started in session A surfaces its completion in whichever poller wakes first. queue.put(evt) time.sleep(0.1) - else: - deferred.append(evt) return True - if owner == "unowned": + if _notification_event_requires_owner(evt) and not _session_owns_notification_event(sid, session, evt): is_delegation = evt.get("type") == "async_delegation" origin, key = str(evt.get("origin_ui_session_id") or ""), str(evt.get("session_key") or "") if deferred is None: @@ -452,9 +421,17 @@ def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> evt.get("type", "completion"), origin, key, ) return True - text = _notif_render_event(sid, evt, emitted, registry, fmt) + if evt.get("type") == "completion" and registry.is_completion_consumed(evt.get("session_id", "")): + return True + text = fmt(evt) if not text: return True + # Emit once per dedup key: a re-queued completion would otherwise re-emit every 0.5s while the + # session is busy, while distinct watch_match events from one process must stay visible. + dedup_key = _notification_event_dedup_key(evt) + if dedup_key not in emitted: + _emit("status.update", sid, {"kind": "process", "text": text}) + emitted.add(dedup_key) if not _notif_claim_turn(session): queue.put(evt) if deferred is not None: @@ -468,12 +445,9 @@ def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) -> def _notification_poller_loop(stop_event: threading.Event, sid: str, session: dict) -> None: - """Poll completion_queue and dispatch notifications autonomously (daemon thread started by - _init_session()): emit a status.update (kind=process), then chain an agent turn via - _run_prompt_submit if the session is idle. The queue is process-global: each poller requeues events - owned by another live session and drops addressed events whose owner is gone; ownerless legacy - notifications remain global. Also polls ``kanban_notify_subs`` every ``_KANBAN_POLL_SECONDS`` — the - delivery path for platform="tui" rows.""" + """Daemon thread (started by _init_session()) that drains the process-global completion_queue for + this session (see _notif_handle_event for ownership routing) and polls ``kanban_notify_subs`` every + ``_KANBAN_POLL_SECONDS`` — the delivery path for platform="tui" rows.""" from tools.process_registry import process_registry, format_process_notification queue = process_registry.completion_queue @@ -518,26 +492,22 @@ def _async_delegation_display_metadata(evt: dict) -> dict: task_count = len(results) or 1 completed_count = sum(1 for r in results if r.get("status") in {"completed", "success"}) failed_count = sum(1 for r in results if r.get("status") in {"failed", "error"}) - metadata = { + duration = evt.get("total_duration_seconds") or evt.get("duration_seconds") + return { "delegation_id": str(evt.get("delegation_id") or ""), "task_count": task_count, "completed_count": completed_count or task_count - failed_count, "failed_count": failed_count, + **({"duration_seconds": duration} if isinstance(duration, (int, float)) else {}), } - duration = evt.get("total_duration_seconds") or evt.get("duration_seconds") - if isinstance(duration, (int, float)): - metadata["duration_seconds"] = duration - return metadata _desktop_ui_wired = False def _wire_desktop_sinks() -> None: - """Idempotently wire process-registry and desktop-tool sinks to renderer events. Background-process - output (`agent.terminal.output` chunks) and `process_registry.request_close_terminal` - (`terminal.close`, drops a tab without killing the process) route to the window owning the process - (its gateway session), keyed by process id; desktop-only tools (open_preview, close_preview, - focus_pane) hand back the turn's ``HERMES_UI_SESSION_ID`` as ``sid`` so the event routes to the - window that asked. `_emit` is `_stdout_lock`-guarded, so registry reader / tool threads may call it.""" + """Idempotently wire process-registry and desktop-tool sinks to renderer events: background-process + output (`agent.terminal.output`) and `terminal.close` (drops a tab without killing the process) route + to the window owning the process by process id; desktop-only tools (open_preview, focus_pane, …) pass + the turn's ``HERMES_UI_SESSION_ID`` as ``sid``. `_emit` is `_stdout_lock`-guarded (thread-safe).""" global _desktop_ui_wired from tools.process_registry import process_registry diff --git a/tui_gateway/slash_worker.py b/tui_gateway/slash_worker.py index d7b4b33162..a97593e440 100644 --- a/tui_gateway/slash_worker.py +++ b/tui_gateway/slash_worker.py @@ -88,6 +88,15 @@ def _run(cli: HermesCLI, command: str) -> str: return strip_ansi(buf.getvalue().rstrip()) +def _sw_log(reason: str) -> None: + print(f"[slash-worker] {reason}", file=sys.stderr, flush=True) + + +def _reply(**fields) -> None: + sys.stdout.write(json.dumps(fields) + "\n") + sys.stdout.flush() + + def main(): p = argparse.ArgumentParser(add_help=False) p.add_argument("--session-key", required=True) @@ -109,13 +118,6 @@ def main(): # entry point — any child inheriting fd 0 can flip the flag). _sw_recovery_times: list[float] = [] - def _sw_log(reason: str) -> None: - print(f"[slash-worker] {reason}", file=sys.stderr, flush=True) - - def _reply(**fields) -> None: - sys.stdout.write(json.dumps(fields) + "\n") - sys.stdout.flush() - while True: raw = sys.stdin.readline() if not raw: