diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 3bd184f8c72f..7c7035abf2d1 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -1071,7 +1071,67 @@ class GatewaySlashCommandsMixin: ] ) - if not agent_rows and not running_processes and not background_tasks: + # Background (async) delegations — delegate_task(background=true). + # Live per-child activity comes from the registry's progress sampler + # (#51690): api calls, current tool, seconds since last activity. + delegations: list[dict] = [] + try: + from tools.async_delegation import list_async_delegations + delegations = [ + d for d in list_async_delegations() + if d.get("status") in ("running", "stalling", "finalizing") + ] + except Exception: + delegations = [] + if delegations: + lines.extend( + [ + "", + t( + "gateway.agents.background_delegations", + count=len(delegations), + ), + ] + ) + for d in delegations[:12]: + goal = " ".join(str(d.get("goal") or "").split()) + if len(goal) > 70: + goal = goal[:67] + "..." + status = d.get("status", "?") + row = f"- `{d.get('delegation_id', '?')}` · {status}" + if status == "stalling": + quiet = d.get("stalled_after_quiet_seconds") + if quiet is not None: + row += f" · no progress {quiet:.0f}s" + elif d.get("seconds_since_progress", 0) >= 60: + row += f" · quiet {d['seconds_since_progress']:.0f}s" + if goal: + row += f" · {goal}" + lines.append(row) + for i, child in enumerate(d.get("children_activity") or []): + if not isinstance(child, dict): + continue + tool = child.get("current_tool") + doing = f"`{tool}`" if tool else "between turns" + part = ( + f" - child {i + 1}: " + f"{child.get('api_calls', '?')} api calls · {doing}" + ) + idle = child.get("seconds_since_activity") + if idle is not None: + part += f" · active {idle:.0f}s ago" + lines.append(part) + if len(delegations) > 12: + lines.append( + t("gateway.agents.more", count=len(delegations) - 12) + ) + + if ( + not agent_rows + and not running_processes + and not background_tasks + and not delegations + ): lines.append("") lines.append(t("gateway.agents.none")) diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index e447c2cac6be..d6d6d7e135e6 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/hermes_cli/cli_commands_mixin.py @@ -286,15 +286,44 @@ class CLICommandsMixin: delegations = list_async_delegations() except Exception: delegations = [] - running_d = [d for d in delegations if d.get("status") == "running"] + running_d = [ + d for d in delegations + if d.get("status") in ("running", "stalling") + ] if delegations: _cprint(f" Background delegations: {len(running_d)} running") for d in delegations: goal = (d.get("goal") or "")[:60] - _cprint( + status = d.get("status", "?") + line = ( f" {d.get('delegation_id', '?')} · " - f"{d.get('status', '?')} · {goal}" + f"{status} · {goal}" ) + # Live-status detail for in-flight delegations (#51690). + if status == "stalling": + quiet = d.get("stalled_after_quiet_seconds") + if quiet is not None: + line += ( + f" · no progress {quiet:.0f}s — interrupting" + ) + elif status in ("running",): + quiet = d.get("seconds_since_progress") + if quiet is not None and quiet >= 60: + line += f" · quiet {quiet:.0f}s" + _cprint(line) + for i, child in enumerate(d.get("children_activity") or []): + if not isinstance(child, dict): + continue + tool = child.get("current_tool") + doing = f"in {tool}" if tool else "between turns" + part = ( + f" └ child {i + 1}: " + f"{child.get('api_calls', '?')} api calls · {doing}" + ) + idle = child.get("seconds_since_activity") + if idle is not None: + part += f" · last activity {idle:.0f}s ago" + _cprint(part) agent_running = getattr(self, "_agent_running", False) _cprint(f" Agent: {'running' if agent_running else 'idle'}") diff --git a/locales/af.yaml b/locales/af.yaml index 6ade25b452e7..4069477de2aa 100644 --- a/locales/af.yaml +++ b/locales/af.yaml @@ -53,6 +53,7 @@ gateway: more: "... en nog {count}" running_processes: "**Lopende agtergrondprosesse:** {count}" async_jobs: "**Asinchrone werke van die gateway:** {count}" + background_delegations: "**Agtergrond-delegasies:** {count}" none: "Geen aktiewe agente of lopende take nie." state_starting: "begin" state_running: "loop" diff --git a/locales/ar.yaml b/locales/ar.yaml index f2c78e8c6540..eb53bcb3350f 100644 --- a/locales/ar.yaml +++ b/locales/ar.yaml @@ -58,6 +58,7 @@ gateway: more: "... و{count} آخر" running_processes: "**العمليات الخلفية الجارية:** {count}" async_jobs: "**مهام البوابة غير المتزامنة:** {count}" + background_delegations: "**التفويضات في الخلفية:** {count}" none: "لا يوجد وكلاء نشطون أو مهام جارية." state_starting: "قيد البدء" state_running: "قيد التشغيل" diff --git a/locales/de.yaml b/locales/de.yaml index 2c6b83d3359a..06a6f1931cbb 100644 --- a/locales/de.yaml +++ b/locales/de.yaml @@ -53,6 +53,7 @@ gateway: more: "... und {count} weitere" running_processes: "**Laufende Hintergrundprozesse:** {count}" async_jobs: "**Gateway-Async-Jobs:** {count}" + background_delegations: "**Hintergrund-Delegationen:** {count}" none: "Keine aktiven Agenten oder laufenden Aufgaben." state_starting: "startet" state_running: "läuft" diff --git a/locales/en.yaml b/locales/en.yaml index 2bd229e2b52c..d027257c0bdb 100644 --- a/locales/en.yaml +++ b/locales/en.yaml @@ -68,6 +68,7 @@ gateway: more: "... and {count} more" running_processes: "**Running background processes:** {count}" async_jobs: "**Gateway async jobs:** {count}" + background_delegations: "**Background delegations:** {count}" none: "No active agents or running tasks." state_starting: "starting" state_running: "running" diff --git a/locales/es.yaml b/locales/es.yaml index 61e2b4418c04..074cfa9f278c 100644 --- a/locales/es.yaml +++ b/locales/es.yaml @@ -53,6 +53,7 @@ gateway: more: "... y {count} más" running_processes: "**Procesos en segundo plano en ejecución:** {count}" async_jobs: "**Tareas asíncronas del gateway:** {count}" + background_delegations: "**Delegaciones en segundo plano:** {count}" none: "No hay agentes activos ni tareas en ejecución." state_starting: "iniciando" state_running: "en ejecución" diff --git a/locales/fr.yaml b/locales/fr.yaml index d067b26eeeae..166f1fba8741 100644 --- a/locales/fr.yaml +++ b/locales/fr.yaml @@ -53,6 +53,7 @@ gateway: more: "... et {count} de plus" running_processes: "**Processus d'arrière-plan en cours :** {count}" async_jobs: "**Tâches asynchrones du gateway :** {count}" + background_delegations: "**Délégations en arrière-plan :** {count}" none: "Aucun agent actif ni tâche en cours." state_starting: "démarrage" state_running: "en cours" diff --git a/locales/ga.yaml b/locales/ga.yaml index 32ed8d597add..a1522e4215c0 100644 --- a/locales/ga.yaml +++ b/locales/ga.yaml @@ -57,6 +57,7 @@ gateway: more: "... agus {count} eile" running_processes: "**Próisis chúlra ag rith:** {count}" async_jobs: "**Tascanna asincrónacha gateway:** {count}" + background_delegations: "**Tarmligin sa chúlra:** {count}" none: "Níl aon ghníomhairí gníomhacha ná tascanna ag rith." state_starting: "ag tosú" state_running: "ag rith" diff --git a/locales/hu.yaml b/locales/hu.yaml index 97a2e675155f..2fec63848446 100644 --- a/locales/hu.yaml +++ b/locales/hu.yaml @@ -53,6 +53,7 @@ gateway: more: "... és még {count}" running_processes: "**Futó háttérfolyamatok:** {count}" async_jobs: "**Átjáró aszinkron feladatai:** {count}" + background_delegations: "**Háttérdelegálások:** {count}" none: "Nincsenek aktív ügynökök vagy futó feladatok." state_starting: "indul" state_running: "fut" diff --git a/locales/it.yaml b/locales/it.yaml index f390f2879c69..eafcbb1e29d8 100644 --- a/locales/it.yaml +++ b/locales/it.yaml @@ -53,6 +53,7 @@ gateway: more: "... e {count} altri" running_processes: "**Processi in background in esecuzione:** {count}" async_jobs: "**Job asincroni del gateway:** {count}" + background_delegations: "**Delegazioni in background:** {count}" none: "Nessun agente attivo o attività in esecuzione." state_starting: "in avvio" state_running: "in esecuzione" diff --git a/locales/ja.yaml b/locales/ja.yaml index 2682dc586356..ea78e7aa6f7b 100644 --- a/locales/ja.yaml +++ b/locales/ja.yaml @@ -53,6 +53,7 @@ gateway: more: "... 他に {count} 件" running_processes: "**実行中のバックグラウンドプロセス:** {count}" async_jobs: "**ゲートウェイ非同期ジョブ:** {count}" + background_delegations: "**バックグラウンド委任:** {count}" none: "アクティブなエージェントや実行中のタスクはありません。" state_starting: "起動中" state_running: "実行中" diff --git a/locales/ko.yaml b/locales/ko.yaml index df07aa12a380..2411c9f1d179 100644 --- a/locales/ko.yaml +++ b/locales/ko.yaml @@ -53,6 +53,7 @@ gateway: more: "... 외 {count}개 더" running_processes: "**실행 중인 백그라운드 프로세스:** {count}" async_jobs: "**게이트웨이 비동기 작업:** {count}" + background_delegations: "**백그라운드 위임:** {count}" none: "활성 에이전트나 실행 중인 작업이 없습니다." state_starting: "시작 중" state_running: "실행 중" diff --git a/locales/pt.yaml b/locales/pt.yaml index 3559c044c549..ada4682708fb 100644 --- a/locales/pt.yaml +++ b/locales/pt.yaml @@ -53,6 +53,7 @@ gateway: more: "... e mais {count}" running_processes: "**Processos em segundo plano em execução:** {count}" async_jobs: "**Tarefas assíncronas do gateway:** {count}" + background_delegations: "**Delegações em segundo plano:** {count}" none: "Não há agentes ativos nem tarefas em execução." state_starting: "a iniciar" state_running: "em execução" diff --git a/locales/ru.yaml b/locales/ru.yaml index b55c006244bb..c83483eae799 100644 --- a/locales/ru.yaml +++ b/locales/ru.yaml @@ -53,6 +53,7 @@ gateway: more: "... и ещё {count}" running_processes: "**Выполняющиеся фоновые процессы:** {count}" async_jobs: "**Асинхронные задачи шлюза:** {count}" + background_delegations: "**Фоновые делегирования:** {count}" none: "Нет активных агентов или выполняющихся задач." state_starting: "запускается" state_running: "выполняется" diff --git a/locales/tr.yaml b/locales/tr.yaml index 5e5f0a151755..33d2c9383f1a 100644 --- a/locales/tr.yaml +++ b/locales/tr.yaml @@ -53,6 +53,7 @@ gateway: more: "... ve {count} tane daha" running_processes: "**Çalışan arka plan süreçleri:** {count}" async_jobs: "**Gateway asenkron işleri:** {count}" + background_delegations: "**Arka plan delegasyonları:** {count}" none: "Aktif ajan veya çalışan görev yok." state_starting: "başlatılıyor" state_running: "çalışıyor" diff --git a/locales/uk.yaml b/locales/uk.yaml index 4606ebe91a5f..f9e5cfd7d450 100644 --- a/locales/uk.yaml +++ b/locales/uk.yaml @@ -53,6 +53,7 @@ gateway: more: "... і ще {count}" running_processes: "**Фонові процеси, що виконуються:** {count}" async_jobs: "**Асинхронні задачі гейтвея:** {count}" + background_delegations: "**Фонові делегування:** {count}" none: "Немає активних агентів або задач." state_starting: "запускається" state_running: "виконується" diff --git a/locales/zh-hant.yaml b/locales/zh-hant.yaml index 39b7079a7e3d..8df92cc23b22 100644 --- a/locales/zh-hant.yaml +++ b/locales/zh-hant.yaml @@ -53,6 +53,7 @@ gateway: more: "... 還有 {count} 個" running_processes: "**執行中的背景程序:** {count}" async_jobs: "**閘道非同步任務:** {count}" + background_delegations: "**背景委派:** {count}" none: "沒有作用中的代理或執行中的任務。" state_starting: "啟動中" state_running: "執行中" diff --git a/locales/zh.yaml b/locales/zh.yaml index 7cf13776a678..d13d8cdfa185 100644 --- a/locales/zh.yaml +++ b/locales/zh.yaml @@ -53,6 +53,7 @@ gateway: more: "... 还有 {count} 个" running_processes: "**运行中的后台进程:** {count}" async_jobs: "**网关异步任务:** {count}" + background_delegations: "**后台委派:** {count}" none: "没有活跃的代理或运行中的任务。" state_starting: "启动中" state_running: "运行中" diff --git a/tests/gateway/test_agents_command_delegations.py b/tests/gateway/test_agents_command_delegations.py new file mode 100644 index 000000000000..422ec77f1d32 --- /dev/null +++ b/tests/gateway/test_agents_command_delegations.py @@ -0,0 +1,119 @@ +"""Gateway /agents surfaces background delegations with live activity (#51690). + +Drives the REAL GatewayRunner._handle_agents_command against a REAL +async-delegation registry dispatch (no mocked list function), so the test +covers the whole projection: registry record → list_async_delegations() +live sampling → /agents rendering. +""" + +import threading +import time + +import pytest + +from tools import async_delegation as ad +from tools.process_registry import process_registry + + +@pytest.fixture(autouse=True) +def _clean_state(): + ad._reset_for_tests() + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + yield + deadline = time.monotonic() + 2.0 + while ad.active_count() and time.monotonic() < deadline: + time.sleep(0.02) + ad._reset_for_tests() + while not process_registry.completion_queue.empty(): + process_registry.completion_queue.get_nowait() + + +def _make_runner(): + from gateway.run import GatewayRunner + + runner = object.__new__(GatewayRunner) + runner._running_agents = {} + runner._running_agents_ts = {} + runner._background_tasks = set() + runner._session_key_for_source = lambda source: "agent:main:test:dm:1" + return runner + + +class _Event: + source = None + + +@pytest.mark.asyncio +async def test_agents_command_lists_background_delegation_with_activity(): + gate = threading.Event() + base_ts = time.time() - 8.0 + + res = ad.dispatch_async_delegation( + goal="research the delegation stall monitor", + context=None, toolsets=None, role="leaf", model="m", + session_key="agent:main:test:dm:1", max_async_children=1, + runner=lambda: {} if gate.wait(timeout=10) else {}, + progress_fn=lambda: (((2, "web_search", base_ts),), True), + ) + assert res["status"] == "dispatched" + + try: + runner = _make_runner() + out = await runner._handle_agents_command(_Event()) + finally: + gate.set() + + assert res["delegation_id"] in out + assert "running" in out + assert "research the delegation stall monitor" in out + # Live per-child activity sampled from progress_fn. + assert "2 api calls" in out + assert "web_search" in out + + +@pytest.mark.asyncio +async def test_agents_command_marks_stalling_delegation(monkeypatch): + monkeypatch.setattr(ad, "_STALE_CHECK_INTERVAL", 0.03) + monkeypatch.setattr(ad, "_STALE_IDLE_SECONDS", 0.1) + # Long grace so the record stays in 'stalling' while we render. + monkeypatch.setattr(ad, "_STALL_GRACE_SECONDS", 30.0) + gate = threading.Event() + + res = ad.dispatch_async_delegation( + goal="wedged child", context=None, toolsets=None, role="leaf", + model="m", session_key="agent:main:test:dm:1", max_async_children=1, + runner=lambda: {} if gate.wait(timeout=10) else {}, + progress_fn=lambda: ((0, None), False), + ) + assert res["status"] == "dispatched" + + try: + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + items = ad.list_async_delegations() + if any( + d["delegation_id"] == res["delegation_id"] + and d.get("status") == "stalling" + for d in items + ): + break + time.sleep(0.02) + else: + pytest.fail("delegation never reached stalling state") + + runner = _make_runner() + out = await runner._handle_agents_command(_Event()) + finally: + gate.set() + + assert res["delegation_id"] in out + assert "stalling" in out + assert "no progress" in out + + +@pytest.mark.asyncio +async def test_agents_command_no_delegations_keeps_none_message(): + runner = _make_runner() + out = await runner._handle_agents_command(_Event()) + assert "No active agents" in out diff --git a/tests/tools/test_async_delegation.py b/tests/tools/test_async_delegation.py index 09ab7801b51f..1d7d92b40e8b 100644 --- a/tests/tools/test_async_delegation.py +++ b/tests/tools/test_async_delegation.py @@ -380,6 +380,67 @@ def test_streaming_child_counts_as_alive(monkeypatch): assert evt["status"] == "completed" +def test_stalled_event_carries_structured_stall_metadata(monkeypatch): + """The terminal stalled event must expose machine-readable stall context + (#51690) — quiet duration, tripped threshold, phase, grace — mirroring + the sync path's timeout_seconds/timed_out_after_seconds/timeout_phase.""" + _fast_stale_monitor(monkeypatch) + gate = threading.Event() + + res = ad.dispatch_async_delegation( + goal="stall metadata", context=None, toolsets=None, role="leaf", + model="m", session_key="", max_async_children=1, + runner=lambda: {} if gate.wait(timeout=10) else {}, + progress_fn=lambda: ((0, "terminal"), True), + ) + assert res["status"] == "dispatched" + + evt = _drain_for(res["delegation_id"], timeout=5.0) + try: + assert evt is not None + assert evt["status"] == "stalled" + assert evt["stalled_after_quiet_seconds"] >= 0.3 # in-tool threshold + assert evt["stall_threshold_seconds"] == ad._STALE_IN_TOOL_SECONDS + assert evt["stall_phase"] == "in_tool" + assert evt["stall_grace_seconds"] == ad._STALL_GRACE_SECONDS + finally: + gate.set() + + +def test_list_async_delegations_exposes_live_activity(monkeypatch): + """list_async_delegations must expose per-child live activity sampled + from progress_fn plus seconds_since_progress, for /agents UIs (#51690).""" + monkeypatch.setattr(ad, "_STALE_CHECK_INTERVAL", 0.03) + gate = threading.Event() + base_ts = time.time() - 12.0 + + res = ad.dispatch_async_delegation( + goal="live listing", context=None, toolsets=None, role="leaf", + model="m", session_key="", max_async_children=1, + runner=lambda: {} if gate.wait(timeout=10) else {}, + progress_fn=lambda: (((3, "web_search", base_ts),), True), + ) + try: + time.sleep(0.1) # let the monitor stamp _progress_ts at least once + item = next( + d for d in ad.list_async_delegations() + if d["delegation_id"] == res["delegation_id"] + ) + assert item["status"] == "running" + assert item["in_tool"] is True + assert "seconds_since_progress" in item + (child,) = item["children_activity"] + assert child["api_calls"] == 3 + assert child["current_tool"] == "web_search" + assert 10.0 <= child["seconds_since_activity"] <= 20.0 + # Callables and private bookkeeping must never leak. + assert "progress_fn" not in item + assert "interrupt_fn" not in item + assert not any(k.startswith("_") for k in item) + finally: + gate.set() + + def test_stalled_batch_is_interrupted_then_finalized(monkeypatch): _fast_stale_monitor(monkeypatch) gate = threading.Event() diff --git a/tests/tools/test_delegate_subagent_timeout_diagnostic.py b/tests/tools/test_delegate_subagent_timeout_diagnostic.py index daaaf4301f45..d290d601945a 100644 --- a/tests/tools/test_delegate_subagent_timeout_diagnostic.py +++ b/tests/tools/test_delegate_subagent_timeout_diagnostic.py @@ -282,3 +282,47 @@ class TestRunSingleChildTimeoutDump: if logs_dir.is_dir(): dumps = list(logs_dir.glob("subagent-timeout-*.log")) assert dumps == [] + + # ── explicit timeout metadata (#51690, salvaged from PR #60378) ──── + + def test_timeout_result_carries_structured_metadata(self, hermes_home, monkeypatch): + """Parents must be able to distinguish a child_timeout_seconds kill + from other failures without parsing the error string.""" + child = _StubChild(api_call_count=0, hang_seconds=10.0) + result = self._invoke_with_short_timeout(child, monkeypatch) + + assert result["status"] == "timeout" + assert result["timeout_seconds"] == 0.3 + assert result["timed_out_after_seconds"] == result["duration_seconds"] + assert result["timeout_phase"] == "before_first_llm_call" + + def test_timeout_phase_after_llm_calls(self, hermes_home, monkeypatch): + child = _StubChild(api_call_count=5, hang_seconds=10.0) + result = self._invoke_with_short_timeout(child, monkeypatch) + + assert result["timeout_phase"] == "after_llm_calls" + assert result["timeout_seconds"] == 0.3 + + def test_non_timeout_error_has_null_timeout_metadata(self, hermes_home, monkeypatch): + """The metadata fields are timeout-specific — a child that raises + must report them as None so consumers can key on presence.""" + from tools import delegate_tool + monkeypatch.setattr(delegate_tool, "_get_child_timeout", lambda: 30.0) + + child = _StubChild(api_call_count=1, hang_seconds=0.0) + + def _boom(*a, **kw): + raise RuntimeError("child crashed") + + child.run_conversation = _boom + parent = MagicMock() + parent._touch_activity = MagicMock() + parent._current_task_id = None + result = delegate_tool._run_single_child( + task_index=0, goal="test goal", child=child, parent_agent=parent, + ) + + assert result["status"] == "error" + assert result["timeout_seconds"] is None + assert result["timed_out_after_seconds"] is None + assert result["timeout_phase"] is None diff --git a/tools/async_delegation.py b/tools/async_delegation.py index 5081259679b1..dbe0b2f1074f 100644 --- a/tools/async_delegation.py +++ b/tools/async_delegation.py @@ -825,6 +825,16 @@ def _push_completion_event( "completed_at": completed_at, "exit_reason": result.get("exit_reason"), } + # Structured stall metadata (#51690) — additive, present only on + # stall-monitor finalizations. + for _k in ( + "stalled_after_quiet_seconds", + "stall_threshold_seconds", + "stall_phase", + "stall_grace_seconds", + ): + if _k in result: + evt[_k] = result[_k] _persist_completion(evt, result) try: process_registry.completion_queue.put(evt) @@ -1024,6 +1034,16 @@ def _push_batch_completion_event( "dispatched_at": dispatched_at, "completed_at": completed_at, } + # Structured stall metadata (#51690) — additive, present only on + # stall-monitor finalizations. + for _k in ( + "stalled_after_quiet_seconds", + "stall_threshold_seconds", + "stall_phase", + "stall_grace_seconds", + ): + if _k in combined: + evt[_k] = combined[_k] _persist_completion(evt, combined) try: process_registry.completion_queue.put(evt) @@ -1109,6 +1129,13 @@ def _stale_monitor_loop() -> None: if quiet_for >= limit: record["status"] = "stalling" record["_interrupted_at"] = now + # Structured stall context for the terminal event and + # status listings (#51690): how long progress was frozen, + # which threshold applied, and whether the child was + # inside a tool when it went quiet. + record["_stall_quiet_seconds"] = round(quiet_for, 2) + record["_stall_threshold_seconds"] = limit + record["_stall_in_tool"] = bool(in_tool) stalled.append( ( record["delegation_id"], @@ -1152,18 +1179,36 @@ def _finalize_stalled(delegation_id: str) -> None: completed_at - (event_record.get("dispatched_at") or completed_at), 2, ) + quiet_seconds = event_record.get("_stall_quiet_seconds") + threshold_seconds = event_record.get("_stall_threshold_seconds") + stall_in_tool = event_record.get("_stall_in_tool") error = ( f"Async delegation {delegation_id} stalled: the detached subagent " - "stopped making progress (no new API calls or tool activity), did " - "not respond to interruption, and never produced a completion " - "event. The worker may be wedged inside a model API call — this is " - "a known failure mode of long-lived gateway processes (#60203). " - "Re-dispatch the task if it is still needed." + "stopped making progress (no new API calls, tool activity, or " + "streamed tokens), did not respond to interruption, and never " + "produced a completion event. The worker may be wedged inside a " + "model API call — this is a known failure mode of long-lived " + "gateway processes (#60203). Re-dispatch the task if it is still " + "needed." ) logger.error( "Async delegation %s force-finalized as stalled after %.0fs", delegation_id, duration, ) + # Structured stall metadata (#51690): lets parents and UIs distinguish + # a stall-monitor kill from other failures without parsing the error + # string, mirroring the sync path's timeout_seconds/timed_out_after_ + # seconds/timeout_phase fields. + stall_meta = { + "stalled_after_quiet_seconds": quiet_seconds, + "stall_threshold_seconds": threshold_seconds, + "stall_phase": ( + "in_tool" if stall_in_tool + else "idle" if stall_in_tool is not None + else None + ), + "stall_grace_seconds": _STALL_GRACE_SECONDS, + } if event_record.get("is_batch"): _push_batch_completion_event( event_record, @@ -1171,6 +1216,7 @@ def _finalize_stalled(delegation_id: str) -> None: "results": [], "error": error, "total_duration_seconds": duration, + **stall_meta, }, "stalled", ) @@ -1184,27 +1230,102 @@ def _finalize_stalled(delegation_id: str) -> None: "api_calls": 0, "duration_seconds": duration, "exit_reason": "stalled", + **stall_meta, }, "stalled", ) _finish_finalization(delegation_id, "stalled") +def _children_activity_from_token(token: Any, now: float) -> Optional[List]: + """Parse a progress token into per-child activity dicts (best-effort). + + delegate_tool's ``_batch_progress`` emits one ``(api_call_count, + current_tool, last_activity_ts)`` tuple per child. Foreign token shapes + (custom dispatchers) degrade to ``None`` entries rather than raising — + the token contract is intentionally opaque to the registry. + """ + try: + parts = list(token) + except TypeError: + return None + out: List[Optional[Dict[str, Any]]] = [] + for part in parts: + if isinstance(part, (list, tuple)) and len(part) >= 2: + entry: Dict[str, Any] = { + "api_calls": part[0], + "current_tool": part[1], + } + if len(part) >= 3 and isinstance(part[2], (int, float)): + entry["seconds_since_activity"] = round( + max(0.0, now - float(part[2])), 1 + ) + out.append(entry) + else: + out.append(None) + return out + + def list_async_delegations() -> List[Dict[str, Any]]: """Snapshot of async delegations (running + recently completed). - Safe to call from any thread. Excludes the non-serialisable interrupt_fn. + Safe to call from any thread. Excludes the non-serialisable callables + and private monitor bookkeeping, but exposes computed live-status + fields for UIs (#51690): + + - ``seconds_since_progress``: how long the stale monitor has seen a + frozen progress token (running/stalling records). + - ``children_activity``: per-child ``{api_calls, current_tool, + seconds_since_activity}`` sampled live from the dispatch's + ``progress_fn``. + - ``stalled_after_quiet_seconds`` / ``stall_threshold_seconds`` / + ``stall_in_tool``: stall context once the monitor has tripped. """ + now = time.time() + samplers: Dict[str, Callable] = {} with _records_lock: - return [ - { + items = [] + for r in _records.values(): + item = { k: v for k, v in r.items() if k not in {"interrupt_fn", "progress_fn"} and not k.startswith("_") } - for r in _records.values() - ] + status = r.get("status") + if status in ("running", "stalling"): + ts = r.get("_progress_ts") + if ts: + item["seconds_since_progress"] = round(now - ts, 1) + fn = r.get("progress_fn") + if callable(fn): + samplers[r["delegation_id"]] = fn + if status in ("stalling", "stalled"): + for src, dst in ( + ("_stall_quiet_seconds", "stalled_after_quiet_seconds"), + ("_stall_threshold_seconds", "stall_threshold_seconds"), + ("_stall_in_tool", "stall_in_tool"), + ): + if r.get(src) is not None: + item[dst] = r.get(src) + items.append(item) + + # Sample live activity OUTSIDE the lock — progress_fn reads child-agent + # attributes and must never run under _records_lock (a slow or broken + # sampler would block every dispatch/finalize in the process). + for item in items: + fn = samplers.get(item.get("delegation_id")) + if fn is None: + continue + try: + token, in_tool = fn() + except Exception: + continue + activity = _children_activity_from_token(token, now) + if activity is not None: + item["children_activity"] = activity + item["in_tool"] = bool(in_tool) + return items def interrupt_all(reason: str = "shutdown") -> int: