feat(delegation): structured stall metadata + live per-child status in /agents

Completes #51690 on top of the salvaged #60378 timeout metadata:

- async_delegation: terminal 'stalled' events now carry structured
  stall context (stalled_after_quiet_seconds, stall_threshold_seconds,
  stall_phase idle|in_tool, stall_grace_seconds) on both single and
  batch paths, persisted in the durable row so restart-restored events
  keep it. Mirrors the sync path's timeout_seconds/timed_out_after_
  seconds/timeout_phase from #60378.
- list_async_delegations(): exposes seconds_since_progress and live
  children_activity (per-child api_calls, current_tool,
  seconds_since_activity) sampled from the dispatch's progress_fn
  outside the records lock; private monitor bookkeeping and callables
  never leak.
- /agents (CLI + gateway): background delegations render per-child
  activity rows, quiet-time hints, and the stalling state; gateway
  section is new (previously async delegations were invisible there).
  New locale key gateway.agents.background_delegations in all 17
  catalogs.

Tests: stall-metadata event shape, live-listing projection, gateway
/agents rendering (real registry dispatch, sabotage-verified), sync
timeout metadata fields, non-timeout None contract.

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"))
 

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'}")

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"

locales/ar.yaml

@@ -58,6 +58,7 @@ gateway:
     more:                  "... و{count} آخر"
     running_processes:     "**العمليات الخلفية الجارية:** {count}"
     async_jobs:            "**مهام البوابة غير المتزامنة:** {count}"
+    background_delegations: "**التفويضات في الخلفية:** {count}"
     none:                  "لا يوجد وكلاء نشطون أو مهام جارية."
     state_starting:        "قيد البدء"
     state_running:         "قيد التشغيل"

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"

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"

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"

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"

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"

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"

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"

locales/ja.yaml

@@ -53,6 +53,7 @@ gateway:
     more:                  "... 他に {count} 件"
     running_processes:     "**実行中のバックグラウンドプロセス:** {count}"
     async_jobs:            "**ゲートウェイ非同期ジョブ:** {count}"
+    background_delegations: "**バックグラウンド委任:** {count}"
     none:                  "アクティブなエージェントや実行中のタスクはありません。"
     state_starting:        "起動中"
     state_running:         "実行中"

locales/ko.yaml

@@ -53,6 +53,7 @@ gateway:
     more:                  "... 외 {count}개 더"
     running_processes:     "**실행 중인 백그라운드 프로세스:** {count}"
     async_jobs:            "**게이트웨이 비동기 작업:** {count}"
+    background_delegations: "**백그라운드 위임:** {count}"
     none:                  "활성 에이전트나 실행 중인 작업이 없습니다."
     state_starting:        "시작 중"
     state_running:         "실행 중"

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"

locales/ru.yaml

@@ -53,6 +53,7 @@ gateway:
     more:                  "... и ещё {count}"
     running_processes:     "**Выполняющиеся фоновые процессы:** {count}"
     async_jobs:            "**Асинхронные задачи шлюза:** {count}"
+    background_delegations: "**Фоновые делегирования:** {count}"
     none:                  "Нет активных агентов или выполняющихся задач."
     state_starting:        "запускается"
     state_running:         "выполняется"

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"

locales/uk.yaml

@@ -53,6 +53,7 @@ gateway:
     more:                  "... і ще {count}"
     running_processes:     "**Фонові процеси, що виконуються:** {count}"
     async_jobs:            "**Асинхронні задачі гейтвея:** {count}"
+    background_delegations: "**Фонові делегування:** {count}"
     none:                  "Немає активних агентів або задач."
     state_starting:        "запускається"
     state_running:         "виконується"

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:         "執行中"

locales/zh.yaml

@@ -53,6 +53,7 @@ gateway:
     more:                  "... 还有 {count} 个"
     running_processes:     "**运行中的后台进程：** {count}"
     async_jobs:            "**网关异步任务：** {count}"
+    background_delegations: "**后台委派：** {count}"
     none:                  "没有活跃的代理或运行中的任务。"
     state_starting:        "启动中"
     state_running:         "运行中"

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

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()

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

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: