diff --git a/agent/tool_executor.py b/agent/tool_executor.py index b454e593b5..da36c6a2d1 100644 --- a/agent/tool_executor.py +++ b/agent/tool_executor.py @@ -749,7 +749,9 @@ def _dispatch_authorized_once( from agent.terminal_approval_batch import prepare_current_terminal prepare_current_terminal(ref) _advance_start_order(lambda: _begin_tool_execution(agent, ref, display_index)) - return _run_with_activity_heartbeat(agent, ref.name, lambda: execute(ref.args)) + from agent.tool_progress_relay import bound_tool_progress + with bound_tool_progress(agent.tool_progress_callback, ref.name, ref.call_id): + return _run_with_activity_heartbeat(agent, ref.name, lambda: execute(ref.args)) def _run_agent_tool_execution_middleware( diff --git a/agent/tool_progress_relay.py b/agent/tool_progress_relay.py new file mode 100644 index 0000000000..a7f24f7dd8 --- /dev/null +++ b/agent/tool_progress_relay.py @@ -0,0 +1,68 @@ +"""Live progress from inside a running tool call (MCP ``notifications/progress``) onto the agent's +``tool_progress_callback`` as ``tool.progress`` events. + +The executor binds a relay for each dispatched call (``bound_tool_progress``); a tool reads it on its +own thread with ``current_tool_progress()`` and may invoke the result from any thread (the MCP client +fires it on the MCP event loop). Emission is throttled per call so a chatty server cannot flood the +UI; the first update and the one reaching ``total`` always pass. +""" + +from __future__ import annotations + +import contextlib +import contextvars +import logging +import threading +import time +from typing import Callable, Iterator, Optional + +logger = logging.getLogger(__name__) + +TOOL_PROGRESS_EVENT = "tool.progress" +_MIN_INTERVAL_S = 0.5 + +ProgressReporter = Callable[[float, Optional[float], Optional[str]], None] +_current: contextvars.ContextVar[Optional[ProgressReporter]] = contextvars.ContextVar("tool_progress_reporter", default=None) + + +def _reporter(callback, tool_name: str, tool_call_id: str) -> ProgressReporter: + lock = threading.Lock() + last = [0.0] + + def report(progress: float, total: Optional[float] = None, message: Optional[str] = None) -> None: + done = total is not None and total > 0 and progress >= total + now = time.monotonic() + with lock: + if last[0] and not done and now - last[0] < _MIN_INTERVAL_S: + return + last[0] = now + percent = round(min(max(progress / total, 0.0), 1.0) * 100) if total else None + try: + callback(TOOL_PROGRESS_EVENT, tool_name, message or None, None, tool_call_id=tool_call_id, + progress=float(progress), total=None if total is None else float(total), percent=percent) + except Exception: # health: allow BLE001 -- a UI sink failing must never fail the tool call + logger.debug("tool.progress callback failed for %s", tool_name, exc_info=True) + return report + + +@contextlib.contextmanager +def bound_tool_progress(callback, tool_name: str, tool_call_id: str) -> Iterator[None]: + """Bind this call's relay for the block (no-op when the agent has no progress callback).""" + token = _current.set(_reporter(callback, tool_name, tool_call_id or "") if callback else None) + try: + yield + finally: + _current.reset(token) + + +def current_tool_progress() -> Optional[ProgressReporter]: + """The running call's reporter, or None outside a dispatched tool call.""" + return _current.get() + + +def tool_progress_payload(name: Optional[str], message: Optional[str], kwargs: dict) -> dict: + """The ``tool.progress`` wire payload every viewer surface publishes (gateway event stream and + the legacy TUI gateway): ``percent`` is 0-100 when the server sent a ``total``, else None.""" + return {"tool_id": str(kwargs.get("tool_call_id") or ""), "name": str(name or ""), + "progress": float(kwargs.get("progress") or 0.0), "total": kwargs.get("total"), + "percent": kwargs.get("percent"), "message": str(message) if message else None} diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts index 88646d431a..6cd610631d 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/index.ts @@ -27,7 +27,7 @@ import { handleOnboardingEvent } from './onboarding' import { handleControlEvent } from './session-control' import { handleSessionInfoEvent } from './session-info' import { handleStatusEvent } from './status' -import { handleToolEvent } from './tools' +import { handleToolEvent, handleToolProgressEvent } from './tools' import type { GatewayEventContext, GatewayEventDeps, GatewayEventHandler } from './types' export type { GatewayEventDeps } from './types' @@ -91,6 +91,7 @@ const HANDLERS: GatewayEventHandler[] = [ handleDeferredPickEvent, handleControlEvent, handleMessageStreamEvent, + handleToolProgressEvent, handleToolEvent, handleInputRequestEvent, handleDesktopBridgeEvent, diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/tools-progress.test.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/tools-progress.test.ts new file mode 100644 index 0000000000..98a6e5b569 --- /dev/null +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/tools-progress.test.ts @@ -0,0 +1,41 @@ +import type { GatewayEvent } from '@hermes/shared' +import { afterEach, expect, it, vi } from 'vitest' + +import { $toolProgress, $toolProgressFor, formatToolProgress } from '@/store/tool-progress' + +import { handleToolProgressEvent } from './tools' +import type { GatewayEventContext } from './types' + +function deliver(event: GatewayEvent) { + return handleToolProgressEvent({ + event, + payload: event.payload, + sessionId: 's1', + occurredAt: 1, + isActiveEvent: true, + deps: { + flushQueuedDeltas: vi.fn(), + updateSessionState: vi.fn(), + sessionInterrupted: () => false, + upsertToolCall: vi.fn(), + nativeSubagentSessionsRef: { current: new Set() } + } + } as unknown as GatewayEventContext) +} + +afterEach(() => $toolProgress.set({})) + +it('shows tool.progress on its running row and drops it when the call completes', () => { + const progress = { + type: 'tool.progress', + session_id: 's1', + payload: { tool_id: 'c1', name: 'mcp__meshy__refine', progress: 42, total: 100, percent: 42, message: 'refining' } + } as unknown as GatewayEvent + + expect(deliver(progress)).toBe(true) + // MCP rows render each batched call as `:`. + expect(formatToolProgress($toolProgressFor('c1:0').get())).toBe('42% · refining') + + deliver({ type: 'tool.complete', session_id: 's1', payload: { tool_id: 'c1', name: 'mcp__meshy__refine' } } as GatewayEvent) + expect($toolProgressFor('c1:0').get()).toBeUndefined() +}) diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/tools.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/tools.ts index 4815d49852..a04ad8bfd2 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/tools.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/tools.ts @@ -15,6 +15,7 @@ import { reportMcpToolResult } from '@/store/suggestion-providers/repair' import { invalidateSkillSuggestionIndex } from '@/store/suggestion-providers/skill' import { restoreSessionTodosFromSnapshot } from '@/store/todos' import { setSessionDraftingTool } from '@/store/tool-drafting' +import { clearToolProgress, setToolProgress } from '@/store/tool-progress' import { notifyWorkspaceChanged, toolChangedPath, toolMayMutateFiles } from '@/store/workspace-events' import { SUBAGENT_EVENT_TYPES, toTodoPayload } from '../utils' @@ -69,6 +70,31 @@ function appliedLayout(name: string | undefined, result: ToolCallMessagePart['re return undefined } +/** tool.progress (live MCP progress for a running call): a row annotation, never a transcript + * mutation. Also retires it on tool.complete, which then falls through to handleToolEvent. */ +export function handleToolProgressEvent({ deps, event, payload, sessionId }: GatewayEventContext): boolean { + const toolId = typeof payload?.tool_id === 'string' ? payload.tool_id : '' + + if (event.type === 'tool.complete') { + clearToolProgress(toolId) + + return false + } + + if (event.type !== 'tool.progress') { + return false + } + + if (sessionId && !deps.sessionInterrupted(sessionId)) { + setToolProgress(toolId, { + message: typeof payload?.message === 'string' && payload.message ? payload.message : null, + percent: typeof payload?.percent === 'number' ? payload.percent : null + }) + } + + return true +} + /** tool.generating / tool.start / tool.complete / subagent.*. */ export function handleToolEvent(ctx: GatewayEventContext): boolean { const { deps, event, payload, sessionId, isActiveEvent, occurredAt } = ctx diff --git a/apps/desktop/src/components/assistant-ui/tool/fallback.tsx b/apps/desktop/src/components/assistant-ui/tool/fallback.tsx index 312506f7d9..6b05f7d800 100644 --- a/apps/desktop/src/components/assistant-ui/tool/fallback.tsx +++ b/apps/desktop/src/components/assistant-ui/tool/fallback.tsx @@ -52,6 +52,7 @@ import { recordPreviewArtifact } from '@/store/preview-status' import { sessionApprovalRequest } from '@/store/prompts' import { $showToolActivity } from '@/store/tool-activity' import { $toolRowDismissed, dismissToolRow } from '@/store/tool-dismiss' +import { $toolProgressFor, formatToolProgress } from '@/store/tool-progress' import { $anyToolDisclosureOpen, $hideCodeDiffs, @@ -344,6 +345,22 @@ function useDisclosureOpen(disclosureId: string, fallbackOpen = false): boolean return persistedOpen ?? fallbackOpen } +/** A running row's trailing meta: live MCP progress (`42% · message`, when the server sends it) + the timer. */ +function ToolRunningMeta({ seconds, toolCallId = '' }: { seconds: number; toolCallId?: string }) { + const progressText = formatToolProgress(useStore($toolProgressFor(toolCallId))) + + return ( + <> + {progressText && ( + + {progressText} + + )} + + + ) +} + function ToolEntry({ part }: ToolEntryProps) { const { t } = useI18n() const copy = t.assistant.tool @@ -541,7 +558,7 @@ function ToolEntry({ part }: ToolEntryProps) { const trailing = !embedded ? ( - {isPending && } + {isPending && } ) : undefined diff --git a/apps/desktop/src/lib/chat-messages/types.ts b/apps/desktop/src/lib/chat-messages/types.ts index 4e10210e45..2473f0ce2a 100644 --- a/apps/desktop/src/lib/chat-messages/types.ts +++ b/apps/desktop/src/lib/chat-messages/types.ts @@ -87,6 +87,8 @@ export type GatewayEventPayload = { name?: string tool_id?: string tool_call_id?: string + /** tool.progress: 0-100 when the MCP server sent a total. */ + percent?: number | null args?: unknown arguments?: unknown context?: string diff --git a/apps/desktop/src/lib/gateway-events.ts b/apps/desktop/src/lib/gateway-events.ts index d775343ec3..c1de3fdd45 100644 --- a/apps/desktop/src/lib/gateway-events.ts +++ b/apps/desktop/src/lib/gateway-events.ts @@ -39,6 +39,7 @@ export const UNSCOPED_STREAM_EVENT_TYPES = new Set([ 'thinking.delta', 'tool.complete', 'tool.generating', + 'tool.progress', 'tool.start', 'vault.code.expire', 'vault.code.request', diff --git a/apps/desktop/src/store/tool-progress.ts b/apps/desktop/src/store/tool-progress.ts new file mode 100644 index 0000000000..d90a6d61e4 --- /dev/null +++ b/apps/desktop/src/store/tool-progress.ts @@ -0,0 +1,50 @@ +import { atom, computed, type ReadableAtom } from 'nanostores' + +/** Latest live progress of a running tool call (`tool.progress`, from MCP `notifications/progress`). */ +export interface ToolProgress { + message: null | string + percent: null | number +} + +// Keyed by tool call id. Ephemeral by design: progress only exists while a call +// runs, so it is never persisted and the row drops it once the call completes. +export const $toolProgress = atom>({}) + +const progressCache = new Map>() + +export function $toolProgressFor(toolCallId: string): ReadableAtom { + let cached = progressCache.get(toolCallId) + + if (!cached) { + // MCP rows render per batched call as `:`; progress is reported for the call. + const callId = toolCallId.replace(/:\d+$/, '') + cached = computed($toolProgress, rows => rows[toolCallId] ?? rows[callId]) + progressCache.set(toolCallId, cached) + } + + return cached +} + +export function setToolProgress(toolCallId: string, progress: ToolProgress): void { + if (toolCallId) { + $toolProgress.set({ ...$toolProgress.get(), [toolCallId]: progress }) + } +} + +export function clearToolProgress(toolCallId: string): void { + const rows = $toolProgress.get() + + if (toolCallId && rows[toolCallId]) { + const { [toolCallId]: _done, ...rest } = rows + $toolProgress.set(rest) + } +} + +/** `42% · refining mesh` — the text a running tool row shows beside its timer. */ +export function formatToolProgress(progress: ToolProgress | undefined): string { + if (!progress) { + return '' + } + + return [progress.percent === null ? '' : `${progress.percent}%`, progress.message ?? ''].filter(Boolean).join(' · ') +} diff --git a/apps/shared/src/gateway-contract.generated.ts b/apps/shared/src/gateway-contract.generated.ts index 639c49f98a..5a3f082e55 100644 --- a/apps/shared/src/gateway-contract.generated.ts +++ b/apps/shared/src/gateway-contract.generated.ts @@ -5748,6 +5748,15 @@ export interface ToolOutputRiskPayload { findings: string[] redacted: boolean } +/** ``agent.tool_progress_relay.tool_progress_payload`` — an MCP server's ``notifications/progress`` for a running call, throttled to one per 500 ms per call. */ +export interface ToolProgressPayload { + tool_id: string + name: string + progress: number + total?: number | null + percent?: number | null + message?: string | null +} /** ``tool_progress._normalize_todo_state`` — full task snapshot. */ export interface TodoUpdatedPayload { todos: unknown[] @@ -7081,6 +7090,8 @@ export interface BackendGatewayEventMap { 'tool.generating': ToolGeneratingPayload /** Tool output was classified as risky (prompt-injection / secret findings). */ 'tool.output_risk': ToolOutputRiskPayload + /** Live progress of a running tool call (MCP notifications/progress). */ + 'tool.progress': ToolProgressPayload /** A tool call began (stable id + full args). */ 'tool.start': ToolStartPayload /** Barge-in: the spoken interjection interrupted the turn; no payload. */ @@ -7169,6 +7180,7 @@ export const GATEWAY_EVENT_TYPES = [ 'tool.complete', 'tool.generating', 'tool.output_risk', + 'tool.progress', 'tool.start', 'voice.interrupted', 'voice.partial', diff --git a/apps/shared/src/gateway-contract.openrpc.json b/apps/shared/src/gateway-contract.openrpc.json index 0d8c05765f..bea828a5c2 100644 --- a/apps/shared/src/gateway-contract.openrpc.json +++ b/apps/shared/src/gateway-contract.openrpc.json @@ -43809,6 +43809,67 @@ "title": "ToolOutputRiskPayload", "type": "object" }, + "ToolProgressPayload": { + "additionalProperties": false, + "description": "``agent.tool_progress_relay.tool_progress_payload`` \u2014 an MCP server's\n``notifications/progress`` for a running call, throttled to one per 500 ms per call.", + "properties": { + "tool_id": { + "title": "Tool Id", + "type": "string" + }, + "name": { + "title": "Name", + "type": "string" + }, + "progress": { + "title": "Progress", + "type": "number" + }, + "total": { + "anyOf": [ + { + "type": "number" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Total" + }, + "percent": { + "anyOf": [ + { + "type": "integer" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Percent" + }, + "message": { + "anyOf": [ + { + "type": "string" + }, + { + "type": "null" + } + ], + "default": null, + "title": "Message" + } + }, + "required": [ + "tool_id", + "name", + "progress" + ], + "title": "ToolProgressPayload", + "type": "object" + }, "ToolShowRow": { "additionalProperties": false, "properties": { @@ -49177,6 +49238,18 @@ } ] }, + { + "name": "tool.progress", + "summary": "Live progress of a running tool call (MCP notifications/progress).", + "params": [ + { + "name": "payload", + "schema": { + "$ref": "#/components/schemas/ToolProgressPayload" + } + } + ] + }, { "name": "tool.start", "summary": "A tool call began (stable id + full args).", diff --git a/gateway/local_viewer_events.py b/gateway/local_viewer_events.py index de6bec9a74..c3353f94f0 100644 --- a/gateway/local_viewer_events.py +++ b/gateway/local_viewer_events.py @@ -135,6 +135,10 @@ def publish_progress(turn, event_type, name, preview, kwargs) -> None: _publish_subagent(turn, event_type, _subagent_payload( event_type, name, preview, kwargs, show_reasoning=show_reasoning(turn))) return + if event_type == "tool.progress": + from agent.tool_progress_relay import tool_progress_payload + turn._publish_execution(event_type, tool_progress_payload(name, preview, kwargs)) + return if event_type != "reasoning.available" and event_type not in _MOA_KEYS: return from gateway.session_cli_turn import classic_cli_turn diff --git a/hermes_cli/cli_stream_mixin.py b/hermes_cli/cli_stream_mixin.py index 00ed37e8f7..a00e1890c5 100644 --- a/hermes_cli/cli_stream_mixin.py +++ b/hermes_cli/cli_stream_mixin.py @@ -763,6 +763,7 @@ class CLIStreamMixin: if event_type == "tool.completed": self._tool_start_time = 0.0 + self.__dict__.pop("_tool_spinner_base", None) self._turn_summary_record( function_name, kwargs.get("result"), kwargs.get("is_error", False)) # Focus view: count the hidden scrollback line for the post-turn recovery report. @@ -810,20 +811,28 @@ class CLIStreamMixin: self._invalidate() return if event_type != "tool.started": - return + return self._show_tool_progress(event_type, preview, kwargs.get("percent")) if function_name and not function_name.startswith("_"): from agent.display import get_tool_preview_max_len, tool_row_emoji label = preview or function_name _pl = get_tool_preview_max_len() if _pl > 0 and len(label) > _pl: label = label[:_pl - 3] + "..." - self._spinner_text = f"{tool_row_emoji(function_name, function_args)} {label}" + self._spinner_text = self._tool_spinner_base = f"{tool_row_emoji(function_name, function_args)} {label}" self._tool_start_time = time.monotonic() # Store args for stacked scrollback line on completion self._pending_tool_info.setdefault(function_name, []).append( function_args if function_args is not None else {}) self._invalidate() + def _show_tool_progress(self, event_type, message, percent) -> None: + """Live MCP progress rides the running tool's spinner line: `` · 42% · message``.""" + base = self.__dict__.get("_tool_spinner_base") + if event_type == "tool.progress" and base: + parts = [base] + ([f"{percent}%"] if percent is not None else []) + ([message] if message else []) + self._spinner_text = " · ".join(parts) + self._invalidate() + def _on_tool_start(self, tool_call_id: str, function_name: str, function_args: dict): """Capture local before-state for write-capable tools.""" from cli import logger diff --git a/hermes_cli/gateway_agent.py b/hermes_cli/gateway_agent.py index 9a1a086702..14214673a7 100644 --- a/hermes_cli/gateway_agent.py +++ b/hermes_cli/gateway_agent.py @@ -952,6 +952,12 @@ class GatewayAgent: if self.tool_gen_callback is not None: self.tool_gen_callback(payload.get("name")) + def _on_tool_progress(self, payload, stream_callback, starts): + if self.tool_progress_callback is not None: + self.tool_progress_callback("tool.progress", payload.get("name"), payload.get("message"), None, + tool_call_id=payload.get("tool_id"), progress=payload.get("progress"), + total=payload.get("total"), percent=payload.get("percent")) + def _on_tool_start(self, payload, stream_callback, starts): self._close_segment(starts) name = payload.get("name") or payload.get("tool_name") or "tool" @@ -1386,6 +1392,7 @@ _HANDLERS = { "message.delta": GatewayAgent._on_delta, "message.interim": GatewayAgent._on_interim, "tool.generating": GatewayAgent._on_tool_generating, + "tool.progress": GatewayAgent._on_tool_progress, "tool.start": GatewayAgent._on_tool_start, "tool.complete": GatewayAgent._on_tool_complete, "todo.updated": GatewayAgent._on_todo, diff --git a/tests/tools/test_mcp_tool_progress.py b/tests/tools/test_mcp_tool_progress.py new file mode 100644 index 0000000000..5e051534cf --- /dev/null +++ b/tests/tools/test_mcp_tool_progress.py @@ -0,0 +1,59 @@ +"""An MCP server's ``notifications/progress`` during ``tools/call`` reaches the agent's +``tool_progress_callback`` as throttled ``tool.progress`` events for that call id. Real stdio child, +real config, real SDK progress token round trip.""" + +import json +import sys +import textwrap + +import pytest + +SERVER = textwrap.dedent('''\ + import asyncio + from mcp.server.mcpserver import MCPServer, Context + app = MCPServer("prog") + + @app.tool() + async def slow(ctx: Context) -> str: + for i in range(1, 21): + await asyncio.sleep(0.1) + await ctx.report_progress(i * 5, 100, f"stage {i}") + return "done" + + app.run() +''') + + +@pytest.mark.no_isolate +@pytest.mark.platforms("posix") +def test_mcp_progress_notifications_reach_tool_progress_callback(monkeypatch, tmp_path): + pytest.importorskip("mcp.server.mcpserver") + home = tmp_path / "home" + home.mkdir() + (home / "server.py").write_text(SERVER, encoding="utf-8") + (home / "config.yaml").write_text( + f"mcp_servers:\n prog:\n command: {json.dumps(sys.executable)}\n args: [{json.dumps(str(home / 'server.py'))}]\n", + encoding="utf-8") + monkeypatch.setenv("HERMES_HOME", str(home)) + + from agent.tool_progress_relay import bound_tool_progress + from tools import mcp_tool_discovery as disc + from tools import mcp_tool_loop as _loop + from tools.registry import registry + + events = [] + try: + assert "mcp__prog__slow" in disc.discover_mcp_tools() + with bound_tool_progress(lambda *a, **kw: events.append((a, kw)), "mcp__prog__slow", "call-7"): + result = registry.dispatch("mcp__prog__slow", {}) + assert "done" in result + finally: + from tools.mcp_tool_lifecycle import shutdown_mcp_servers + shutdown_mcp_servers() + _loop._stop_mcp_loop() + + assert events and all(a[0] == "tool.progress" and a[1] == "mcp__prog__slow" for a, _ in events) + assert {kw["tool_call_id"] for _, kw in events} == {"call-7"} + # 20 notifications over ~2 s, throttled to <= one per 500 ms, the final 100% always delivered. + assert 2 <= len(events) <= 8 + assert events[-1][1]["percent"] == 100 and events[-1][0][2] == "stage 20" diff --git a/tools/mcp_tool_handlers.py b/tools/mcp_tool_handlers.py index e5c4de2de0..3723e11918 100644 --- a/tools/mcp_tool_handlers.py +++ b/tools/mcp_tool_handlers.py @@ -379,7 +379,7 @@ async def _track_inflight_rpc(server: Any, server_name: str, op: str, *, retry_s inflight.discard(task) -async def _call_tool_racing_stdio_death(server, server_name: str, tool_name: str, args: dict): +async def _call_tool_racing_stdio_death(server, server_name: str, tool_name: str, args: dict, progress_callback=None): """``session.call_tool`` that fails fast when the stdio child is/gets dead: pre-call (a dead child must not hold the slot for the full timeout) and mid-call (race against ``_watch_stdio_children``). Both raise :class:`_StdioChildExited` for the respawn path, which @@ -393,7 +393,10 @@ async def _call_tool_racing_stdio_death(server, server_name: str, tool_name: str f"MCP stdio subprocess for '{server_name}' had already exited when the call was dispatched", in_flight=False, ) - _call_coro = server.session.call_tool(tool_name, arguments=args) + # A progress callback makes the SDK send ``_meta.progressToken``, the spec's opt-in for + # ``notifications/progress``; without one the server must stay silent. + extra = {"progress_callback": progress_callback} if progress_callback is not None else {} + _call_coro = server.session.call_tool(tool_name, arguments=args, **extra) _watch_children = getattr(server, "_watch_stdio_children", None) if not (inspect.iscoroutinefunction(_watch_children) and asyncio.iscoroutine(_call_coro)): # Stubbed sessions return a non-awaitable, or there is no child-watcher to race: plain await. @@ -582,12 +585,20 @@ def _make_tool_handler(server_name: str, tool_name: str, tool_timeout: float, *, # pre-dispatch so the auth recoverer keeps its retry for every tool. read_only = _tool_is_read_only(server_name, tool_name) image_paths: list[str] = [] # the LAST attempt's cached images (a recoverer may retry _call) + # Read on the agent's thread: the coroutine below runs on the MCP loop, outside this context. + from agent.tool_progress_relay import current_tool_progress + report = current_tool_progress() + + async def _on_progress(progress, total, message): + if report is not None: + report(progress, total, message) async def _call(): async with server._rpc_lock, _track_inflight_rpc(server, server_name, op, retry_safe=read_only): server._pending_call_context = contextvars.copy_context() # for the elicitation callback try: - result = await _call_tool_racing_stdio_death(server, server_name, tool_name, args) + result = await _call_tool_racing_stdio_death( + server, server_name, tool_name, args, _on_progress if report is not None else None) finally: server._pending_call_context = None if getattr(server, "_mark_session_proven", None) is not None: # round-trip done: transport healthy diff --git a/tui_gateway/contracts/events.py b/tui_gateway/contracts/events.py index b44328f5f0..5ab556f2f7 100644 --- a/tui_gateway/contracts/events.py +++ b/tui_gateway/contracts/events.py @@ -409,6 +409,21 @@ class ToolOutputRiskPayload(Payload): event("tool.output_risk", ToolOutputRiskPayload, doc="Tool output was classified as risky (prompt-injection / secret findings).") +class ToolProgressPayload(Payload): + """``agent.tool_progress_relay.tool_progress_payload`` — an MCP server's + ``notifications/progress`` for a running call, throttled to one per 500 ms per call.""" + + tool_id: str + name: str + progress: float + total: float | None = None + percent: int | None = None + message: str | None = None + + +event("tool.progress", ToolProgressPayload, doc="Live progress of a running tool call (MCP notifications/progress).") + + class TodoUpdatedPayload(Payload): """``tool_progress._normalize_todo_state`` — full task snapshot.""" @@ -846,7 +861,8 @@ __all__ = [ "SetupReadyPayload", "SideAgentCompletePayload", "SkinPayload", "StatusUpdatePayload", "StreamDeltaPayload", "SubagentEventPayload", "SubagentOutputTailEntry", "TerminalClosePayload", "TerminalOutputPayload", "TipShowPayload", "TodoUpdatedPayload", "ToolCompletePayload", - "ToolGeneratingPayload", "ToolLabel", "ToolLabelKind", "ToolOutputRiskPayload", "ToolStartPayload", + "ToolGeneratingPayload", "ToolLabel", "ToolLabelKind", "ToolOutputRiskPayload", "ToolProgressPayload", + "ToolStartPayload", "TurnStatus", "VoicePartialPayload", "VoiceStatusPayload", "VoiceTranscriptPayload", "WakeDetectedPayload", ] diff --git a/tui_gateway/tool_progress.py b/tui_gateway/tool_progress.py index 8e427a9ae4..db024b80a4 100644 --- a/tui_gateway/tool_progress.py +++ b/tui_gateway/tool_progress.py @@ -380,6 +380,13 @@ def _on_tool_complete(sid: str, tool_call_id: str, name: str, args: dict, result # `tool.started` is dropped on purpose: _on_tool_start already emits the authoritative tool.start with # the stable id and args; an id-less duplicate row makes the desktop live view diverge from history. +def _progress_tool_progress(sid, name, preview, kw): + # Live MCP progress on a running tool row: tool chrome, so it follows display.tool_progress. + if _tool_progress_enabled(sid): + from agent.tool_progress_relay import tool_progress_payload + _emit("tool.progress", sid, tool_progress_payload(name, preview, kw)) + + def _progress_output_risk(sid, name, preview, kw): # A risk badge on a tool row: tool chrome, so it follows display.tool_progress. metadata = kw.get("risk_metadata") @@ -494,7 +501,8 @@ def _progress_subagent(sid: str, name: str, preview, kw, event_type): # event_type -> (handler, requires): `requires` names the arg that must be truthy for the row to be # emitted at all ("name" / "preview" / None). _PROGRESS_HANDLERS = { - "tool.output_risk": (_progress_output_risk, "name"), "reasoning.available": (_progress_reasoning, "preview"), + "tool.output_risk": (_progress_output_risk, "name"), "tool.progress": (_progress_tool_progress, "name"), + "reasoning.available": (_progress_reasoning, "preview"), "moa.reference": (_progress_moa_reference, "name"), # Answer-only drops MoA content (references, progress/phase lines that land in the reasoning # disclosure or activity log) but keeps this: a bare state transition both clients use only diff --git a/website/docs/developer-guide/desktop-plugin-sdk.md b/website/docs/developer-guide/desktop-plugin-sdk.md index 24a0e136b8..2b2eb438b7 100644 --- a/website/docs/developer-guide/desktop-plugin-sdk.md +++ b/website/docs/developer-guide/desktop-plugin-sdk.md @@ -1223,6 +1223,35 @@ listener can't affect app dispatch. Every `host` door is async-safe: a sync thro from an internal helper (e.g. no desktop bridge in a plain browser) becomes a rejection your `.catch()` sees, never an error-boundary crash. +#### Live tool progress — `tool.progress` {#tool-progress-event} + +While a tool call runs, an MCP server that sends `notifications/progress` (the +MCP spec's progress channel; Hermes opts in by sending `_meta.progressToken` on +every `tools/call`) reaches the renderer as a `tool.progress` event. The native +tool row shows it as `42% · ` next to its timer; a plugin can render +its own view: + +```javascript +ctx.onEvent('tool.progress', ({ session_id, payload }) => { + // payload: { tool_id, name, progress, total, percent, message } + if (payload.name.startsWith('mcp__meshy__')) showRefineProgress(payload.tool_id, payload.percent, payload.message) +}) +ctx.onEvent('tool.complete', ({ payload }) => clearRefineProgress(payload.tool_id)) +``` + +| Field | Type | Meaning | +|---|---|---| +| `tool_id` | `string` | The call id; the same `tool_id` as the call's `tool.start` / `tool.complete` | +| `name` | `string` | Registered tool name (`mcp____`) | +| `progress` | `number` | The server's raw progress value | +| `total` | `number \| null` | The server's total, when it sent one | +| `percent` | `number \| null` | `0`–`100` (rounded, clamped) when `total` is set, else `null` | +| `message` | `string \| null` | The server's progress message | + +Events are throttled to at most one per 500 ms per call; the first update and +the one that reaches `total` always arrive. They are never stored in the +transcript, so a reopened session shows only the finished row. + `ctx.os` is the curated OS door — every way a plugin reaches outside the app window, in one namespace attributed to your plugin. `ctx.os.notify` posts a **native OS notification** — the same Electron pipeline the app's own