fix(*): live spawn record, live usage and a leaner node panel subtitle

docs/specs/2026-09-18-desk-tasks-list-design.md

@@ -80,7 +80,14 @@ stores it uncapped; a spawn's is the head of `.error.md`, or of `.out.md` when `
 `tokens_in` / `tokens_out` (null = the lane cannot report; never zero), `tool_call_count`
 (null = zero or unreported, an acknowledged ambiguity of `as_meta`), `tool_failure_count`,
 `has_output`, `prompt_template`, `inputs`, `skills`, `mcps`, `files[]`
-(`{path, op: write|edit, add, del, size}`).
+(`{path, op: write|edit, add, del, size}`). While a node runs, the usage, the tool counts and
+the files are read off the activity being collected for it in this process (the same
+in-memory account `subagent.context` and `dag.node` serve a transcript from), since the record
+on disk carries them only once the run finishes; what the lane has not reported yet stays null,
+and a node that is not running takes nothing from that index (its key is a record id unique per
+conversation only). A dag node that has finished while its run has not keeps the account the
+runner set aside for it at its end (`activity.record_settled`) until the manifest is written,
+so its usage does not vanish between the two.
 
 ### 2. Status derivation
 
@@ -116,13 +123,20 @@ dispatch leaves no record and is withdrawn by the `subagent.status{cancelled}` f
 
 `dag.node(run_id, node, session_key)` and `subagent.context(id, session_id)` return
 `messages[]` in the `session.resume` shape; the tab draws them with the transcript's own
-renderer, including the synthetic `role=console` row an in-flight `cli` lane emits.
+renderer, including the synthetic `role=console` row an in-flight `cli` lane emits. The
+record is re-read on every status transition and, for a dag node, on each `dag.node_updated`
+frame; a spawn gets no per-step frame, so its record is re-read on a one-second beat while it
+runs (a beat is skipped while a read is still out), the cadence the transcript's spawn card
+already reads on. Each such read of a running node also re-reads its row through
+`tasks.list(kind, id)`, which is how the panel's token total moves during the run.
 
 ### 5. Live updates
 
 `dag.run_started` builds a row (`status=running`; `nodes[].subagent` becomes `agent`; no
 timestamps, so `started_at` starts from the run id's prefix); `dag.node_updated` moves a
-node (timestamps only on `running` / `completed` / `failed` / `exception`);
+node (timestamps only on `running` / `completed` / `failed` / `exception`) and, on a node's
+own terminal frame, re-reads the row so the account the runner set aside for the node reaches
+the page before the run's own end;
 `dag.run_completed` ends a row (a hard stop sends `{stopped: true}` and no `files`);
 `dag.run_replanned` marks the old row `cancelled` with `replan`; `subagent.status` builds
 a spawn row at `pending` with `handle = instance ?? task_id`, and `call_id` from `running`
@@ -147,9 +161,10 @@ truth after a reconnect or a terminal event.
 `features/tasks/` derives its types from `generated.ts`, reads through `tasks.list`, and
 draws the mock: the list row, the strip (three chips and an overflow), the running count on
 the tab, the pane (status bar with `interrupted` and `cancelled`, a stop action, the why
-banner naming the failed node, file and diff chips, the board with tool and failure counts
-and a lane per shared `(agent, instance)`), the node panel (header with tokens or "not
-reported", context and order tabs, a chat dock for a stateful agent's instance). The desk's
+banner naming the failed node, file and diff chips, the board with tool counts and a lane
+per shared `(agent, instance)`), the node panel (header with the agent, its status word, the
+duration and the token total when the lane reported one, context and order tabs, a chat dock
+for a stateful agent's instance). The desk's
 deliverables and diff tabs gain a task-derived group beside the session-level rows; a
 task file's diff is built on click from the node's messages with the page's existing hunk
 builders. The offline page answers `tasks.list` from `src/rpc/fixtures/tasks.ts`.

i18n/messages.json

@@ -7099,10 +7099,6 @@
       "en": "{n} tokens",
       "zh": "{n} tokens"
     },
-    "gui.tasks.tokens_none": {
-      "en": "usage not reported",
-      "zh": "不报用量"
-    },
     "gui.tasks.no_output": {
       "en": "no output",
       "zh": "无输出"
@@ -7111,10 +7107,6 @@
       "en": "{n} tools",
       "zh": "{n} 次工具"
     },
-    "gui.tasks.tools_failed_n": {
-      "en": "{n} failed",
-      "zh": "{n} 次失败"
-    },
     "gui.tasks.stop": {
       "en": "Stop",
       "zh": "停止"

raven/agent/subagent/activity.py

@@ -27,7 +27,7 @@
 from __future__ import annotations
 
 import time
-from collections.abc import Awaitable, Callable, Iterator
+from collections.abc import Awaitable, Callable, Iterable, Iterator
 from contextlib import contextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass, field
@@ -205,6 +205,33 @@ def as_meta(self) -> dict[str, Any]:
 their ``collecting`` block."""
 
 
+_settled: dict[str, dict[str, Any]] = {}
+"""The account of a run's node that has finished while its run has not.
+
+A dag node's account reaches disk with the run's manifest, written once the
+whole run is over; between the node's own end (when ``collecting`` drops it from
+``_live``) and that write, nothing else holds it, and a reader that showed the
+node's usage while it ran would show nothing the moment it finished. The runner
+records the account here as the node settles and forgets the run's entries once
+the manifest is written."""
+
+
+def record_settled(key: str, meta: dict[str, Any]) -> None:
+    """Set aside a finished node's account under its live key."""
+    _settled[key] = dict(meta)
+
+
+def settled(key: str) -> dict[str, Any] | None:
+    """The account of a node that finished while its run has not, or None."""
+    return _settled.get(key)
+
+
+def forget_settled(keys: Iterable[str]) -> None:
+    """Drop the accounts a run set aside, once its manifest carries them."""
+    for key in keys:
+        _settled.pop(key, None)
+
+
 _live_instances: dict[tuple[str, str, str], RunActivity] = {}
 """The same activities, addressed the way a *conversation* reader has to ask.
 
@@ -595,9 +622,12 @@ def note_transcript(messages: list[dict[str, Any]] | None) -> None:
     "RunActivity",
     "collecting",
     "current",
+    "forget_settled",
     "set_transcript",
     "live",
     "live_instance",
+    "record_settled",
+    "settled",
     "merge_file_change",
     "note_alive",
     "note_closing",

raven/agent/subagent/dag_runner.py

@@ -683,6 +683,9 @@ async def run_dag(
         await _reap_carried(carried)
         _mark_stopped(status)
         await _record_outcome(store, status, cancelled=True)
+        # No manifest will carry them: a stopped run's set-aside accounts
+        # would otherwise outlive it for the life of the process.
+        activity.forget_settled(node_live_key(store.run_id, nid) for nid in status)
         # This run's own memory-record pollers, scheduled for nodes that had
         # already completed before this cancellation landed, would otherwise
         # keep polling with nothing left to reap them.
@@ -1681,6 +1684,11 @@ async def _run_node(
                     finally:
                         stall_watch.cancel()
                         node_activity[node.id] = did.as_meta()
+                        # Set aside for the reader until the manifest carries
+                        # it: `collecting` drops the live entry as this block
+                        # exits, and the manifest is written once the whole
+                        # run is over.
+                        activity.record_settled(node_live_key(store.run_id, node.id), node_activity[node.id])
             # The whole answer when the reply cap cut one: the in-context copy of
             # a terminal output is capped again on the way out (see
             # `_terminal_outputs`), and this file is what the reader, the next
@@ -1958,6 +1966,9 @@ async def _finalize(
 
     summary = _tally(status)
     await store.write_manifest(manifest)
+    # The manifest now carries every node's account; the copies set aside for
+    # the nodes that finished before it was written have done their job.
+    activity.forget_settled(node_live_key(store.run_id, nid) for nid in by_id)
     await _record_outcome(store, status)
     return DagRunResult(
         run_id=store.run_id,

raven/rpc/methods/tasks.py

@@ -20,8 +20,9 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
+from raven.agent.subagent import activity as run_activity
 from raven.agent.subagent.activity import merge_file_change
-from raven.agent.subagent.dag_store import REGISTRY_FILENAME, RUNNING, UNRECORDED
+from raven.agent.subagent.dag_store import REGISTRY_FILENAME, RUNNING, UNRECORDED, node_live_key
 from raven.agent.subagent.history import dag_root, nodes_root, session_history_root
 from raven.agent.subagent.instances import get_registry
 from raven.rpc.methods.instances import _graph_of
@@ -114,6 +115,31 @@ def _files_of(source: dict[str, Any]) -> list[dict[str, Any]]:
     return folded
 
 
+def _overlay_live(node: dict[str, Any], live: Any) -> None:
+    """Fill a running node's usage, tool counts and files from the activity
+    being collected for it in this process.
+
+    The record on disk carries those only once the run finishes (``as_meta``
+    is written by ``finish``), so without this a node reads as reporting
+    nothing for the whole of its run. Only what the lane has said so far is
+    taken: a lane that has not spoken keeps its null.
+    """
+    # A node that is not running has an account of its own on disk (or none),
+    # and the live index is keyed by a record id that is unique per conversation
+    # only: another conversation's run under the same id must not fill it.
+    if live is None or node["status"] != "running":
+        return
+    for key in ("tokens_in", "tokens_out"):
+        if node[key] is None:
+            node[key] = _int_or_none(getattr(live, key, None))
+    calls = getattr(live, "tool_calls", None)
+    if node["tool_call_count"] is None and calls:
+        node["tool_call_count"] = len(calls)
+        node["tool_failure_count"] = len(getattr(live, "tool_failures", None) or [])
+    if not node["files"]:
+        node["files"] = _files_of({"files": list(getattr(live, "files", None) or [])})
+
+
 def _counts(statuses: list[str]) -> dict[str, int]:
     counts = {key: 0 for key in _COUNT_KEYS}
     for status in statuses:
@@ -262,6 +288,7 @@ def _spawn_row(files: "_NodeFiles", agent_loop_factory: "AgentLoopFactory | None
         "prompt_template": None,
         "files": _files_of(meta),
     }
+    _overlay_live(node, run_activity.live(files.node_id))
     task_summary = meta.get("task_summary") or meta.get("label") or _label_from_prompt(files) or None
     return {
         "id": files.node_id,
@@ -390,14 +417,18 @@ def _dag_row(
             continue
         nid = gnode["id"]
         entry = manifest.get(nid) if isinstance(manifest.get(nid), dict) else None
+        live_key = node_live_key(run_dir.name, nid)
+        # A node that finished while its run has not has no manifest entry yet;
+        # its account is what the runner set aside at the node's end.
+        account = entry if entry is not None else (run_activity.settled(live_key) or {})
         status, started, ended = _dag_node_state(run_dir.name, nid, entry, registry_nodes, by_node)
         if status in _NOT_LIVE_PENDING and not live:
             status = "interrupted"
         if status == "skipped":
             # Never ran: the registry stamps it with the moment the run was
             # finalized, which is not a clock this node ever had.
             started = ended = None
-        call_count, failure_count = _tool_counts(entry or {})
+        call_count, failure_count = _tool_counts(account)
         node: dict[str, Any] = {
             "node_id": nid,
             "node_summary": gnode.get("node_summary") or None,
@@ -408,20 +439,21 @@ def _dag_row(
             "started_at": started,
             "ended_at": ended,
             "error": _clip((entry or {}).get("error")),
-            "tokens_in": _int_or_none((entry or {}).get("tokens_in")),
-            "tokens_out": _int_or_none((entry or {}).get("tokens_out")),
+            "tokens_in": _int_or_none(account.get("tokens_in")),
+            "tokens_out": _int_or_none(account.get("tokens_out")),
             "tool_call_count": call_count,
             "tool_failure_count": failure_count,
             "has_output": _dag_has_output(nid, entry, registry_nodes.get(nid), nodes_dir),
             "prompt_template": gnode.get("prompt_template") or None,
-            "files": _files_of(entry or {}),
+            "files": _files_of(account),
         }
         if gnode.get("inputs") is not None:
             node["inputs"] = gnode["inputs"]
         if gnode.get("skills") is not None:
             node["skills"] = list(gnode["skills"])
         if gnode.get("mcps") is not None:
             node["mcps"] = list(gnode["mcps"])
+        _overlay_live(node, run_activity.live(live_key))
         nodes.append(node)
         statuses.append(status)
         if isinstance(started, int):