fix: show task-first subagent completion notices in CLI and TUI

cli.py

@@ -3407,6 +3407,9 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix
             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)
 
@@ -3488,6 +3491,8 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix
 
     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
+        notification_preview = user_input if isinstance(user_input, SubagentNotification) else None
         user_input, is_voice_input, is_seeded_query = self._tui_unwrap_input(user_input)
         if not user_input:
             return
@@ -3534,7 +3539,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix
         if isinstance(user_input, str) and _PASTE_REF_RE.search(user_input):
             user_input = self._expand_paste_references(user_input)
         print()
-        self._print_user_message_preview(user_input)
+        self._print_user_message_preview(notification_preview or user_input)
 
         if submit_images:
             n = len(submit_images)
@@ -3545,7 +3550,7 @@ class HermesCLI(CLIAgentSetupMixin, CLICommandsMixin, CLIBillingMixin, CLITuiMix
         self._turn_summary_begin()
         self._app.invalidate()
         try:
-            self.chat(user_input, images=submit_images or None, voice_input=is_voice_input)
+            self.chat(notification_preview or user_input, images=submit_images or None, voice_input=is_voice_input)
         finally:
             self._tui_after_turn()
 

hermes_cli/cli_agent_setup_mixin.py

@@ -124,7 +124,9 @@ def _collect_resume_entries(display_history, disp: dict, clean_assistant):
         if display_kind == "hidden":
             continue
         if display_kind in _RESUME_EVENT_TEXT:
-            entries.append(("event", _RESUME_EVENT_TEXT[display_kind]))
+            metadata = msg.get("display_metadata") or {}
+            label = metadata.get("display_text") if display_kind == "async_delegation_complete" else None
+            entries.append(("event", _sanitize_display_text(label or _RESUME_EVENT_TEXT[display_kind])))
             continue
         if role == "user":
             text = _sanitize_display_text(_user_display_text(content))

hermes_cli/cli_chat_turn_mixin.py

@@ -35,6 +35,7 @@ class CLIChatTurnMixin:
         the concise voice-response prefix, #65827)
         """
         from cli import ChatConsole, _ChatTurn, _DIM, _RST, _accent_hex, _cprint, set_secret_capture_callback
+        from tools.process_registry_notifications import SubagentNotification
         # Single-query and direct chat callers do not go through run().
         set_secret_capture_callback(self._secret_capture_callback)
         # Reset per turn; only a real interrupt flips it, so early returns leave it False.
@@ -56,7 +57,7 @@ class CLIChatTurnMixin:
             return None
         message = self._chat_route_images(message, images)
 
-        if isinstance(message, str):
+        if isinstance(message, str) and not isinstance(message, SubagentNotification):
             message, blocked = self._chat_expand_context_references(message)
             if blocked is not None:
                 return blocked
@@ -65,6 +66,8 @@ class CLIChatTurnMixin:
             message = _sanitize_surrogates(message)
 
         self._chat_stage_user_message(agent, message)
+        if isinstance(message, SubagentNotification):
+            message = str(message)  # UI metadata is on the staged row, never in model content.
 
         ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]")
         print(flush=True)
@@ -204,6 +207,10 @@ class CLIChatTurnMixin:
             agent._persist_user_message_override = None
             agent._persist_user_message_timestamp = None
             staged_user_message = stamp_message_timestamp({"role": "user", "content": message})
+            from tools.process_registry_notifications import SubagentNotification
+            if isinstance(message, SubagentNotification):
+                staged_user_message.update(content=str(message), display_kind="async_delegation_complete",
+                                           display_metadata={"display_text": message.display_text})
             agent._pending_cli_user_message = staged_user_message
             self.conversation_history.append(staged_user_message)
 

hermes_cli/cli_stream_mixin.py

@@ -209,6 +209,10 @@ class CLIStreamMixin:
     def _print_user_message_preview(self, user_input: str) -> None:
         """Render a user message using the normal chat scrollback style."""
         from cli import ChatConsole, _accent_hex
+        from tools.process_registry_notifications import SubagentNotification
+        if isinstance(user_input, SubagentNotification):
+            ChatConsole().print(f"[dim]◈ {_escape(user_input.display_text)}[/dim]")
+            return
         ChatConsole().print(f"[{_accent_hex()}]{'─' * 40}[/]")
         text = str(user_input or "")
         if "\n" in text:

tests/cli/test_subagent_notification_display.py

@@ -0,0 +1,95 @@
+"""Completion notices stay human-facing while the parent retains full results."""
+import copy
+import queue
+import threading
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+from cli import HermesCLI
+from tools.process_registry_notifications import format_process_notification
+from tui_gateway import server
+
+
+def test_completion_display_keeps_payload_separate_across_surfaces(monkeypatch, capsys, tmp_path):
+    for status, truncated, label in [("completed", False, "Completed"), ("failed", False, "Failed"),
+                                      ("cancelled", False, "Cancelled"), ("completed", True, "Incomplete"),
+                                      ("stalled", False, "Stalled"), ("unknown", False, "Unknown"),
+                                      ("rejected", False, "Failed"), ("other_terminal_state", False, "Incomplete")]:
+        event = {"type": "async_delegation", "session_key": "display-session", "delegation_id": "deleg-test",
+                 "goals": ["Do not display this sibling", "Review [bold]changes[/bold]"],
+                 "results": [{"task_index": 1, "status": status, "truncated": truncated, "summary": "Full result evidence"}]}
+        original = copy.deepcopy(event)
+        payload = format_process_notification(event)
+        cli = HermesCLI.__new__(HermesCLI)
+        cli.session_id = event["session_key"]
+        cli._pending_input = queue.Queue()
+        registry = SimpleNamespace(drain_notifications=lambda **kw: [(event, payload)], completion_queue=queue.Queue())
+        monkeypatch.setattr("tools.process_registry.process_registry", registry)
+        monkeypatch.setattr("tools.async_delegation.claim_event_delivery", lambda *a: "claimed")
+        monkeypatch.setattr("tools.async_delegation.complete_event_delivery", lambda *a: None)
+        cli._drain_process_notifications("cli-idle")
+        cli._pending_resume_sessions = []
+        cli._typed_voice_stop = lambda text: False
+        cli.handle_bang_shell = lambda text: False
+        cli._turn_summary_begin = lambda: None
+        cli._tui_after_turn = lambda: None
+        cli._app = SimpleNamespace(invalidate=lambda: None)
+        cli.chat = Mock()
+        cli._tui_process_one_input(cli._pending_input.get_nowait())
+        visible = capsys.readouterr().out
+        expected = f"Subagent Task {label}: {event['goals'][1]}"
+        assert expected in visible
+        assert "ASYNC DELEGATION" not in visible and "Full result evidence" not in visible
+        queued_message = cli.chat.call_args.args[0]
+        assert queued_message == payload
+        cli.conversation_history = []
+        cli.agent = SimpleNamespace(run_conversation=Mock(return_value={}))
+        cli._flush_credit_notices = lambda: None
+        cli._chat_stage_user_message(cli.agent, queued_message)
+        from cli import _ChatTurn
+        cli._chat_run_agent(_ChatTurn(), str(queued_message))
+        staged = cli.conversation_history[-1]
+        assert staged["content"] == payload
+        assert type(staged["content"]) is str
+        assert staged["display_kind"] == "async_delegation_complete"
+        from agent.turn_context import _stage_turn_user_message
+        run_args = cli.agent.run_conversation.call_args.kwargs
+        core_message, _ = _stage_turn_user_message(cli.agent, run_args["user_message"],
+                                                   run_args["persist_user_message"], None, None, None, None)
+        assert core_message is staged
+        assert core_message["display_metadata"]["display_text"] == expected
+        assert copy.deepcopy(staged) == staged
+        from agent.prompt_caching import build_prompt_cache_plan, strip_anthropic_cache_control
+        plan = build_prompt_cache_plan([core_message], tools=None)
+        assert strip_anthropic_cache_control(plan.messages)[0]["content"] == payload
+        from hermes_cli.cli_agent_setup_mixin import _collect_resume_entries
+        from hermes_state import SessionDB
+        with SessionDB(tmp_path / f"{status}-{truncated}.db") as db:
+            db.create_session(cli.session_id, source="cli")
+            db.append_messages_batch(cli.session_id, cli.conversation_history)
+            restored = db.get_messages(cli.session_id)
+        assert restored[0]["content"] == payload
+        entries, _, _ = _collect_resume_entries(restored, {}, lambda text: text)
+        assert entries == [("event", expected)]
+        emitted, dispatched = [], []
+        monkeypatch.setattr(server, "_emit", lambda *args: emitted.append(args))
+        monkeypatch.setattr(server, "_notif_dispatch_event", lambda *args: dispatched.append(args))
+        session = {"session_key": event["session_key"], "history_lock": threading.RLock()}
+        server._notif_handle_event("ui-session", session, event, set(), registry, format_process_notification, None)
+        assert emitted[0][2]["text"] == expected
+        assert dispatched[0][3] == payload
+        assert server._async_delegation_display_metadata(event)["display_text"] == expected
+        assert event == original
+
+    from tools.process_registry_notifications import async_delegation_display_text
+    grouped = {"group": "Review", "goals": ["First", "Second"],
+               "results": [{"task_index": 0, "status": "completed"}, {"task_index": 1, "status": "failed"}]}
+    assert async_delegation_display_text(grouped) == "Subagent Tasks Finished with Issues: Review (2 tasks)"
+    early = {**grouped, "task_failure_notice": True, "results": [grouped["results"][1]]}
+    assert async_delegation_display_text(early) == "Subagent Task Failed: Second"
+    grouped["results"][1]["status"] = "completed"
+    assert async_delegation_display_text(grouped) == "Subagent Tasks Completed: Review (2 tasks)"
+    assert async_delegation_display_text({"goal": "Legacy task", "status": "timeout"}) == "Subagent Task Timed Out: Legacy task"
+    assert "Failed" in async_delegation_display_text({"results": [], "error": "Worker crashed"})
+    cli._print_user_message_preview("[ASYNC DELEGATION BATCH COMPLETE — user-authored]")
+    assert "[ASYNC DELEGATION BATCH COMPLETE — user-authored]" in capsys.readouterr().out

tools/process_registry_notifications.py

@@ -216,6 +216,42 @@ def _format_async_delegation(evt: dict) -> str:
     return "\n".join(lines)
 
 
+def async_delegation_display_text(evt: dict) -> str:
+    """Compact UI title; the separate model notification retains all task evidence."""
+    raw_results = evt.get("results")
+    results = [r for r in raw_results if isinstance(r, dict)] if isinstance(raw_results, list) else []
+    results = results or [evt]
+    goals = evt.get("goals") or []
+    labels, titles = [], []
+    status_labels = {"failed": "Failed", "error": "Failed", "cancelled": "Cancelled",
+                     "interrupted": "Interrupted", "timeout": "Timed Out", "stalled": "Stalled",
+                     "unknown": "Unknown", "rejected": "Failed"}
+    for result in results:
+        status = result.get("status") or ("failed" if result.get("error") else "completed")
+        label = ("Incomplete" if _is_truncated(result) else "Completed") if status in _DONE else (
+            status_labels.get(status, "Incomplete"))
+        labels.append(label)
+        index = result.get("task_index", 0)
+        goal = goals[index] if 0 <= index < len(goals) else result.get("goal", "")
+        titles.append(" ".join(str(goal or "Background task").split()))
+    if len(results) == 1:
+        return f"Subagent Task {labels[0]}: {titles[0]}"
+    outcome = labels[0] if len(set(labels)) == 1 else "Finished with Issues"
+    title = " ".join(str(evt.get("group") or "").split()) or "; ".join(titles)
+    return f"Subagent Tasks {outcome}: {title} ({len(results)} tasks)"
+
+
+class SubagentNotification(str):
+    """Keep queued model text string-compatible, with a separate human preview."""
+
+    display_text: str
+
+    def __new__(cls, text: str, event: dict):
+        instance = super().__new__(cls, text)
+        instance.display_text = async_delegation_display_text(event)
+        return instance
+
+
 def _delegation_attribution_line(evt: dict) -> "str | None":
     """One-line provenance for a subagent-owned process event, else None. Such a process
     outlives the child and lands in the PARENT conversation, which would otherwise see an

tui_gateway/session_notifications.py

@@ -428,7 +428,9 @@ def _notif_handle_event(sid, session, evt, emitted, registry, fmt, deferred) ->
     # 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})
+        from tools.process_registry_notifications import async_delegation_display_text
+        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 not _notif_claim_turn(session):
         queue.put(evt)
@@ -491,13 +493,15 @@ def _notification_poller_loop(stop_event: threading.Event, sid: str, session: di
 
 def _async_delegation_display_metadata(evt: dict) -> dict:
     """Build display-only metadata before the completion event is formatted."""
+    from tools.process_registry_notifications import async_delegation_display_text
     raw_results = evt.get("results")
     results: list[dict] = [r for r in raw_results if isinstance(r, dict)] if isinstance(raw_results, list) else []
     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"})
     duration = evt.get("total_duration_seconds") or evt.get("duration_seconds")
-    return {"delegation_id": str(evt.get("delegation_id") or ""), "task_count": task_count,
+    return {"display_text": async_delegation_display_text(evt),
+            "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 {})}
 

ui-tui/src/__tests__/messages.test.ts

@@ -77,6 +77,16 @@ describe('toTranscriptMessages', () => {
     ])
   })
 
+  it('uses the display-only completion title without exposing the model payload', () => {
+    const text = '[ASYNC DELEGATION BATCH COMPLETE — private]\nFull result evidence'
+    const title = 'Subagent Task Failed: Review changes'
+    const [message] = toTranscriptMessages([
+      { role: 'user', text, display_kind: 'async_delegation_complete', display_metadata: { display_text: title } }
+    ])
+    expect(message).toMatchObject({ kind: 'event', role: 'system', text: title })
+    expect(toTranscriptMessages([{ role: 'user', text }])[0]?.text).toBe(text)
+  })
+
   it('projects async_delegation_complete without metadata as generic text', () => {
     const rows = [{ role: 'user', text: 'event', display_kind: 'async_delegation_complete' }]
 

ui-tui/src/domain/messages.ts

@@ -80,7 +80,11 @@ export const toTranscriptMessages = (rows: unknown): Msg[] => {
           ? 'background agent work finished'
           : `${count} background agent${count === 1 ? '' : 's'} finished`
 
-      out.push({ kind: 'event', role: 'system', text: label })
+      out.push({
+        kind: 'event',
+        role: 'system',
+        text: typeof meta?.display_text === 'string' ? meta.display_text : label
+      })
       pending = []
 
       continue