mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-20 09:45:47 +00:00
feat(subagents): show effective model and token usage on task cards (#4049)
* feat(subagents): show runtime metadata on task cards * fix(subagents): stop task-card render loop and dedupe model fetches Address code review on the runtime-metadata cards: - P1 render loop: the terminal ToolMessage is re-parsed on every MessageList render and always carries modelName/usage, so the presence-based setTasks condition fired a fresh state object each render -> "Maximum update depth exceeded". computeNextSubtask now returns a value-compared `changed` flag and a pure subtaskNotification() routes terminal transitions through the deferred after-render path while skipping no-op re-parses. - Per-card useModels refetch: add staleTime: Infinity to the ["models"] query so every subtask card shares one /api/models fetch instead of refetching on each mount. * make format * refactor(subagents): dedupe token-usage validators + tidy event narrowing Address PR review follow-ups: - DRY: extract one shared token-usage validator per side. Backend status_contract.normalize_token_usage() now backs both the terminal ToolMessage metadata and the subagent.step/.end run events (step_events.py), and frontend messages/usage.normalizeTokenUsage() backs both the live task_running event (lifecycle.ts) and the terminal ToolMessage metadata (subtask-result.ts). Prevents the input/output/ total_tokens validation from drifting across the four former copies. - Nit: onCustomEvent narrows event.type once instead of re-checking the object shape per branch; the redundant task_started early-return (already validated by taskEventToSubtaskUpdate) is dropped.
This commit is contained in:
Nan Gao
@@ -707,7 +707,7 @@ Use `/compact` in the Web UI composer to summarize older context for the current
|
||||
|
||||
Complex tasks rarely fit in a single pass. DeerFlow decomposes them.
|
||||
|
||||
The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. When token usage tracking is enabled, completed sub-agent usage is attributed back to the dispatching step.
|
||||
The lead agent can spawn sub-agents on the fly — each with its own scoped context, tools, and termination conditions. Sub-agents run in parallel when possible, report back structured results, and the lead agent synthesizes everything into a coherent output. Long-running sub-agents compact older history when summarization is enabled and re-inject the summary as guarded, hidden durable context before continuing, so recent assistant/tool activity remains grounded in the task. Provider/model request failures are reported as failed sub-agent tasks rather than successful results, so the lead agent and Web UI can react to them correctly. Collapsed sub-agent cards show the effective model and, when the provider returns usage metadata, a cumulative token total that updates after each completed sub-agent LLM call and persists after a reload. When token usage tracking is enabled, completed sub-agent usage is also attributed back to the dispatching step.
|
||||
|
||||
This is how DeerFlow handles tasks that take minutes to hours: a research task might fan out into a dozen sub-agents, each exploring a different angle, then converge into a single report — or a website — or a slide deck with generated visuals. One harness, many hands.
|
||||
|
||||
|
||||
+1
-1
@@ -377,7 +377,7 @@ Proxied through nginx: `/api/langgraph/*` → Gateway LangGraph-compatible runti
|
||||
**Built-in Agents**: `general-purpose` (all tools except `task`) and `bash` (command specialist)
|
||||
**Execution**: Dual thread pool - `_scheduler_pool` (3 workers) + `_execution_pool` (3 workers)
|
||||
**Concurrency**: `MAX_CONCURRENT_SUBAGENTS = 3` enforced by `SubagentLimitMiddleware` (truncates excess tool calls in `after_model`); default subagent timeout `subagents.timeout_seconds=1800` (30 min) and built-in `general-purpose` `max_turns=150` (raised from 100/15-min so deep-research subtasks stop hitting `GraphRecursionError` out of the box)
|
||||
**Flow**: `task()` tool → `SubagentExecutor` → background thread → poll 5s → SSE events → result
|
||||
**Flow**: `task()` tool → `SubagentExecutor` → background thread → poll 5s → SSE events → result. `task_started` carries the resolved effective model name. The per-subagent `SubagentTokenCollector` publishes a cumulative usage snapshot to the shared `SubagentResult` after every completed LLM response; the next `task_running` event carries that snapshot, so collapsed workspace cards can update without re-accounting parent-run totals. Terminal ToolMessage metadata (`subagent_model_name`, `subagent_token_usage`) and the persisted `subagent.end` event retain the model/usage after reload; absent provider usage stays absent rather than being estimated as zero.
|
||||
**Events**: `task_started`, `task_running`, `task_completed`/`task_failed`/`task_timed_out`
|
||||
**Handled LLM failures**: `LLMErrorHandlingMiddleware` deliberately converts provider/model exceptions into an `AIMessage` so the graph can end cleanly, stamping `additional_kwargs.deerflow_error_fallback=true` plus error metadata. Clean graph termination does not imply subagent success: `SubagentExecutor` inspects the last assistant message at terminalization and maps a marked fallback to `SubagentStatus.FAILED`, which then emits `task_failed` and the existing structured `subagent_error`. Only the marker is authoritative — error-looking assistant prose without it remains a normal completed result, so neither the executor nor frontend parses display text as a status protocol.
|
||||
**Guardrail caps & `stop_reason` (#3875 Phase 2)**: three independent axes can end a subagent run early, and all now surface *why* through one additive field rather than a new status enum. **Turn axis**: `recursion_limit` on the subagent `run_config` equals `max_turns`, so exhausting the turn budget raises `GraphRecursionError` from `agent.astream`; `executor.py::_aexecute` catches it specifically (before the generic `except Exception`). **Token axis**: `TokenBudgetMiddleware` is attached per-agent via `build_subagent_runtime_middlewares` from `subagents.token_budget` (default `max_tokens` **coupled to `summarization.enabled`** — 1,000,000 when subagent summarization is on, 2,000,000 when off, warn at 0.7, hard-stop at 1.0; a user-set budget always wins regardless of the switch — #3875 Phase 3; a backstop against a subagent that burns tokens on trivial work). It does *not* raise: at the hard-stop threshold it strips the in-flight turn's tool calls, forces `finish_reason="stop"`, and lets the run complete naturally with a final answer. **Loop axis**: `LoopDetectionMiddleware` (attached at the same point) catches repeated identical tool-call sets — or one tool *type* called many times with varying args — and its hard-stop likewise strips `tool_calls` and forces a final answer without raising, recording `loop_capped`. Each guard exposes its cap on a per-`run_id` `consume_stop_reason(run_id)` accessor; `_aexecute` collects **every** middleware with that method (duck-typed via `hasattr`, so the executor has no import coupling to the guard classes) and surfaces the first non-`None` reason — adding a future guard needs no executor change. **Surfacing**: whichever axis fired, `_aexecute` stamps a normal status plus an additive reason — `completed` + `stop_reason=token_capped|turn_capped|loop_capped` when a usable final answer (or partial recovered from the last streamed chunk via `_extract_final_result` → `utils/messages.py::message_content_to_text`, returning a `"No response Generated"` sentinel when no text survived) was produced; `failed` + `stop_reason=turn_capped` when nothing usable survived. `SubagentResult.stop_reason` flows through `task_tool.py::_task_result_command` → `format_subagent_result_message` (renders `Task Succeeded (capped: ...)` / `Task failed (capped: ...)`) and `make_subagent_additional_kwargs`, which stamps the additive `subagent_stop_reason` key alongside the normal `subagent_status`. **Why additive, not an enum**: a new status value would break v1 consumers; an optional field is ignored by older frontends and ledger readers, so the cross-language contract (`contracts/subagent_status_contract.json` v2 + `subagents/status_contract.py` + `frontend/.../subtask-result.ts`, pinned by `test_status_values_match_contract` / `test_stop_reason_values_match_contract`) stays backward-compatible. The durable delegation ledger captures `stop_reason` onto the entry and renders model-facing guidance ("hit a guardrail cap with a partial result; reuse it, retry tighter, or raise the per-agent budget (`max_turns` / `token_budget`)") so the lead reuses a capped completion knowingly instead of mistaking it for a clean one. (Phase 1 shipped this surfacing as a `MAX_TURNS_REACHED` status enum in #3949; Phase 2 replaced that enum with the additive `stop_reason` field per the agreed design — the `max_turns_reached` status value and `SubagentStatus.MAX_TURNS_REACHED` are gone.)
|
||||
|
||||
@@ -110,6 +110,12 @@ class SubagentResult:
|
||||
if self.ai_messages is None:
|
||||
self.ai_messages = []
|
||||
|
||||
def update_token_usage_records(self, records: list[dict[str, int | str | None]]) -> None:
|
||||
"""Publish the latest cumulative collector snapshot while still running."""
|
||||
with self._state_lock:
|
||||
if not self.status.is_terminal:
|
||||
self.token_usage_records = list(records)
|
||||
|
||||
def try_set_terminal(
|
||||
self,
|
||||
status: SubagentStatus,
|
||||
@@ -807,6 +813,7 @@ class SubagentExecutor:
|
||||
return result
|
||||
|
||||
final_state = chunk
|
||||
result.update_token_usage_records(collector.snapshot_records())
|
||||
|
||||
# Capture every step message (assistant turns AND tool outputs)
|
||||
# appended since the last chunk. A single super-step can append
|
||||
|
||||
@@ -15,6 +15,10 @@ consumers read the structured facts carried inside
|
||||
backend recorded.
|
||||
- ``subagent_result_brief`` / ``subagent_result_sha256`` (optional):
|
||||
bounded completed-result metadata plus a digest of the full result.
|
||||
- ``subagent_model_name`` (optional): effective DeerFlow model identifier used
|
||||
by this delegated run.
|
||||
- ``subagent_token_usage`` (optional): final cumulative ``input_tokens`` /
|
||||
``output_tokens`` / ``total_tokens`` snapshot when the provider reported it.
|
||||
|
||||
The shared fixture at ``contracts/subagent_status_contract.json`` pins
|
||||
the enum values across Python and TypeScript.
|
||||
@@ -25,13 +29,15 @@ from __future__ import annotations
|
||||
import hashlib
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from typing import Literal, NotRequired, TypedDict
|
||||
from typing import Any, Literal, NotRequired, TypedDict
|
||||
|
||||
SUBAGENT_STATUS_KEY = "subagent_status"
|
||||
SUBAGENT_STOP_REASON_KEY = "subagent_stop_reason"
|
||||
SUBAGENT_ERROR_KEY = "subagent_error"
|
||||
SUBAGENT_RESULT_BRIEF_KEY = "subagent_result_brief"
|
||||
SUBAGENT_RESULT_SHA256_KEY = "subagent_result_sha256"
|
||||
SUBAGENT_MODEL_NAME_KEY = "subagent_model_name"
|
||||
SUBAGENT_TOKEN_USAGE_KEY = "subagent_token_usage"
|
||||
SUBAGENT_METADATA_TEXT_MAX_CHARS = 2000
|
||||
|
||||
#: The producer always emits ``hashlib.sha256(...).hexdigest()`` — 64
|
||||
@@ -127,7 +133,9 @@ def make_subagent_additional_kwargs(
|
||||
result: str | None = None,
|
||||
error: str | None = None,
|
||||
stop_reason: SubagentStopReasonValue | None = None,
|
||||
) -> dict[str, str]:
|
||||
model_name: str | None = None,
|
||||
token_usage: Mapping[str, object] | None = None,
|
||||
) -> dict[str, object]:
|
||||
"""Build the ``additional_kwargs`` payload the middleware stamps.
|
||||
|
||||
Drops the error field when blank so the JSON wire format never carries
|
||||
@@ -145,7 +153,7 @@ def make_subagent_additional_kwargs(
|
||||
raise ValueError(f"invalid subagent status {status!r}; expected one of {SUBAGENT_STATUS_VALUES}")
|
||||
if stop_reason is not None and stop_reason not in SUBAGENT_STOP_REASON_VALUES:
|
||||
raise ValueError(f"invalid subagent stop_reason {stop_reason!r}; expected one of {SUBAGENT_STOP_REASON_VALUES}")
|
||||
payload: dict[str, str] = {SUBAGENT_STATUS_KEY: status}
|
||||
payload: dict[str, object] = {SUBAGENT_STATUS_KEY: status}
|
||||
if status in _RESULT_BEARING_STATUSES and isinstance(result, str) and result.strip():
|
||||
payload[SUBAGENT_RESULT_BRIEF_KEY] = _bound_metadata_text(result)
|
||||
payload[SUBAGENT_RESULT_SHA256_KEY] = hashlib.sha256(result.encode("utf-8")).hexdigest()
|
||||
@@ -155,9 +163,36 @@ def make_subagent_additional_kwargs(
|
||||
payload[SUBAGENT_ERROR_KEY] = _bound_metadata_text(error)
|
||||
if stop_reason is not None:
|
||||
payload[SUBAGENT_STOP_REASON_KEY] = stop_reason
|
||||
if isinstance(model_name, str) and model_name.strip():
|
||||
payload[SUBAGENT_MODEL_NAME_KEY] = model_name.strip()
|
||||
normalized_usage = normalize_token_usage(token_usage)
|
||||
if normalized_usage is not None:
|
||||
payload[SUBAGENT_TOKEN_USAGE_KEY] = normalized_usage
|
||||
return payload
|
||||
|
||||
|
||||
def normalize_token_usage(value: Any) -> dict[str, int] | None:
|
||||
"""Validate a cumulative token-usage mapping into the contract shape.
|
||||
|
||||
The single shared validator for both metadata surfaces — the terminal
|
||||
``ToolMessage`` metadata (here) and the persisted ``subagent.step`` /
|
||||
``subagent.end`` run events (``step_events.py``). Keeping one function
|
||||
prevents the two from drifting (e.g. one later accepting an extra token
|
||||
field the other rejects, silently dropping usage on one path). Requires
|
||||
non-negative ``int`` values for all three keys — ``bool`` is rejected — and
|
||||
returns ``None`` for any non-mapping or malformed input.
|
||||
"""
|
||||
if not isinstance(value, Mapping):
|
||||
return None
|
||||
normalized: dict[str, int] = {}
|
||||
for key in ("input_tokens", "output_tokens", "total_tokens"):
|
||||
amount = value.get(key)
|
||||
if isinstance(amount, bool) or not isinstance(amount, int) or amount < 0:
|
||||
return None
|
||||
normalized[key] = amount
|
||||
return normalized
|
||||
|
||||
|
||||
def format_subagent_result_message(
|
||||
status: SubagentStatusValue,
|
||||
*,
|
||||
|
||||
@@ -25,6 +25,8 @@ from langchain_core.messages import AIMessage, BaseMessage, ToolMessage
|
||||
|
||||
from deerflow.utils.messages import message_content_to_text
|
||||
|
||||
from .status_contract import normalize_token_usage
|
||||
|
||||
#: Default per-step character cap for the ``text`` field. Tool outputs (web
|
||||
#: search results, file contents) can be large; this cap bounds the persisted
|
||||
#: run-event row and the streamed frame. It only affects display/storage — the
|
||||
@@ -223,6 +225,12 @@ def subagent_run_event(chunk: Any) -> dict[str, Any] | None:
|
||||
status = _TERMINAL_EVENT_STATUS.get(event)
|
||||
if status is not None:
|
||||
content: dict[str, Any] = {"task_id": task_id, "status": status}
|
||||
model_name = chunk.get("model_name")
|
||||
if isinstance(model_name, str) and model_name.strip():
|
||||
content["model_name"] = model_name.strip()
|
||||
usage = normalize_token_usage(chunk.get("usage"))
|
||||
if usage is not None:
|
||||
content["usage"] = usage
|
||||
# The final result/error can be a multi-page report; cap it so the
|
||||
# persisted run-event row stays bounded (it is also kept verbatim on the
|
||||
# terminal ToolMessage, which the card reads separately).
|
||||
|
||||
@@ -200,6 +200,8 @@ def _task_result_command(
|
||||
result: str | None = None,
|
||||
error: str | None = None,
|
||||
stop_reason: SubagentStopReasonValue | None = None,
|
||||
model_name: str | None = None,
|
||||
usage: dict[str, int] | None = None,
|
||||
) -> Command:
|
||||
content, metadata_error = format_subagent_result_message(status, result=result, error=error, stop_reason=stop_reason)
|
||||
return Command(
|
||||
@@ -209,7 +211,14 @@ def _task_result_command(
|
||||
content=content,
|
||||
tool_call_id=tool_call_id,
|
||||
name="task",
|
||||
additional_kwargs=make_subagent_additional_kwargs(status, result=result, error=metadata_error, stop_reason=stop_reason),
|
||||
additional_kwargs=make_subagent_additional_kwargs(
|
||||
status,
|
||||
result=result,
|
||||
error=metadata_error,
|
||||
stop_reason=stop_reason,
|
||||
model_name=model_name,
|
||||
token_usage=usage,
|
||||
),
|
||||
)
|
||||
]
|
||||
}
|
||||
@@ -398,7 +407,14 @@ async def task_tool(
|
||||
|
||||
writer = get_stream_writer()
|
||||
# Send Task Started message'
|
||||
writer({"type": "task_started", "task_id": task_id, "description": description})
|
||||
writer(
|
||||
{
|
||||
"type": "task_started",
|
||||
"task_id": task_id,
|
||||
"description": description,
|
||||
"model_name": effective_model,
|
||||
}
|
||||
)
|
||||
|
||||
try:
|
||||
while True:
|
||||
@@ -420,6 +436,11 @@ async def task_tool(
|
||||
logger.info(f"[trace={trace_id}] Task {task_id} status: {result.status.value}")
|
||||
last_status = result.status
|
||||
|
||||
# The collector publishes cumulative records. Reuse one snapshot for
|
||||
# both live progress and the terminal event so the frontend can
|
||||
# replace, rather than add, its per-task total.
|
||||
usage = _summarize_usage(getattr(result, "token_usage_records", None))
|
||||
|
||||
# Check for new AI messages and send task_running events
|
||||
ai_messages = result.ai_messages or []
|
||||
current_message_count = len(ai_messages)
|
||||
@@ -434,17 +455,26 @@ async def task_tool(
|
||||
"message": message,
|
||||
"message_index": i + 1, # 1-based index for display
|
||||
"total_messages": current_message_count,
|
||||
"usage": usage,
|
||||
"model_name": effective_model,
|
||||
}
|
||||
)
|
||||
logger.info(f"[trace={trace_id}] Task {task_id} sent message #{i + 1}/{current_message_count}")
|
||||
last_message_count = current_message_count
|
||||
|
||||
# Check if task completed, failed, or timed out
|
||||
usage = _summarize_usage(getattr(result, "token_usage_records", None))
|
||||
if result.status == SubagentStatus.COMPLETED:
|
||||
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
|
||||
_report_subagent_usage(runtime, result)
|
||||
writer({"type": "task_completed", "task_id": task_id, "result": result.result, "usage": usage})
|
||||
writer(
|
||||
{
|
||||
"type": "task_completed",
|
||||
"task_id": task_id,
|
||||
"result": result.result,
|
||||
"usage": usage,
|
||||
"model_name": effective_model,
|
||||
}
|
||||
)
|
||||
logger.info(f"[trace={trace_id}] Task {task_id} completed after {poll_count} polls")
|
||||
cleanup_background_task(task_id)
|
||||
# stop_reason carries a guardrail cap (token_capped / turn_capped)
|
||||
@@ -455,11 +485,21 @@ async def task_tool(
|
||||
status="completed",
|
||||
result=result.result,
|
||||
stop_reason=result.stop_reason,
|
||||
model_name=effective_model,
|
||||
usage=usage,
|
||||
)
|
||||
elif result.status == SubagentStatus.FAILED:
|
||||
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
|
||||
_report_subagent_usage(runtime, result)
|
||||
writer({"type": "task_failed", "task_id": task_id, "error": result.error, "usage": usage})
|
||||
writer(
|
||||
{
|
||||
"type": "task_failed",
|
||||
"task_id": task_id,
|
||||
"error": result.error,
|
||||
"usage": usage,
|
||||
"model_name": effective_model,
|
||||
}
|
||||
)
|
||||
logger.error(f"[trace={trace_id}] Task {task_id} failed: {result.error}")
|
||||
cleanup_background_task(task_id)
|
||||
# A turn-capped run with no usable output surfaces as failed +
|
||||
@@ -470,28 +510,50 @@ async def task_tool(
|
||||
status="failed",
|
||||
error=result.error,
|
||||
stop_reason=result.stop_reason,
|
||||
model_name=effective_model,
|
||||
usage=usage,
|
||||
)
|
||||
elif result.status == SubagentStatus.CANCELLED:
|
||||
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
|
||||
_report_subagent_usage(runtime, result)
|
||||
writer({"type": "task_cancelled", "task_id": task_id, "error": result.error, "usage": usage})
|
||||
writer(
|
||||
{
|
||||
"type": "task_cancelled",
|
||||
"task_id": task_id,
|
||||
"error": result.error,
|
||||
"usage": usage,
|
||||
"model_name": effective_model,
|
||||
}
|
||||
)
|
||||
logger.info(f"[trace={trace_id}] Task {task_id} cancelled: {result.error}")
|
||||
cleanup_background_task(task_id)
|
||||
return _task_result_command(
|
||||
tool_call_id=tool_call_id,
|
||||
status="cancelled",
|
||||
error=result.error,
|
||||
model_name=effective_model,
|
||||
usage=usage,
|
||||
)
|
||||
elif result.status == SubagentStatus.TIMED_OUT:
|
||||
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
|
||||
_report_subagent_usage(runtime, result)
|
||||
writer({"type": "task_timed_out", "task_id": task_id, "error": result.error, "usage": usage})
|
||||
writer(
|
||||
{
|
||||
"type": "task_timed_out",
|
||||
"task_id": task_id,
|
||||
"error": result.error,
|
||||
"usage": usage,
|
||||
"model_name": effective_model,
|
||||
}
|
||||
)
|
||||
logger.warning(f"[trace={trace_id}] Task {task_id} timed out: {result.error}")
|
||||
cleanup_background_task(task_id)
|
||||
return _task_result_command(
|
||||
tool_call_id=tool_call_id,
|
||||
status="timed_out",
|
||||
error=result.error,
|
||||
model_name=effective_model,
|
||||
usage=usage,
|
||||
)
|
||||
|
||||
# Still running, wait before next poll
|
||||
@@ -507,7 +569,14 @@ async def task_tool(
|
||||
_report_subagent_usage(runtime, result)
|
||||
usage = _summarize_usage(getattr(result, "token_usage_records", None))
|
||||
_cache_subagent_usage(tool_call_id, usage, enabled=cache_token_usage)
|
||||
writer({"type": "task_timed_out", "task_id": task_id, "usage": usage})
|
||||
writer(
|
||||
{
|
||||
"type": "task_timed_out",
|
||||
"task_id": task_id,
|
||||
"usage": usage,
|
||||
"model_name": effective_model,
|
||||
}
|
||||
)
|
||||
# The task may still be running in the background. Signal cooperative
|
||||
# cancellation and schedule deferred cleanup to remove the entry from
|
||||
# _background_tasks once the background thread reaches a terminal state.
|
||||
@@ -518,6 +587,8 @@ async def task_tool(
|
||||
tool_call_id=tool_call_id,
|
||||
status="polling_timed_out",
|
||||
error=message,
|
||||
model_name=effective_model,
|
||||
usage=usage,
|
||||
)
|
||||
except asyncio.CancelledError:
|
||||
# Signal the background subagent thread to stop cooperatively.
|
||||
|
||||
@@ -802,6 +802,67 @@ class TestAsyncExecutionPath:
|
||||
assert result.result == "real result"
|
||||
assert result.error is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aexecute_exposes_collected_usage_before_subagent_finishes(self, classes, base_config, mock_agent, msg, monkeypatch):
|
||||
"""Polling callers can read a cumulative token snapshot while running."""
|
||||
from deerflow.subagents import executor as executor_module
|
||||
|
||||
SubagentExecutor = classes["SubagentExecutor"]
|
||||
SubagentResult = classes["SubagentResult"]
|
||||
SubagentStatus = classes["SubagentStatus"]
|
||||
collectors = []
|
||||
yielded = asyncio.Event()
|
||||
release = asyncio.Event()
|
||||
|
||||
class Collector:
|
||||
def __init__(self, caller):
|
||||
self.records = []
|
||||
collectors.append(self)
|
||||
|
||||
def snapshot_records(self):
|
||||
return list(self.records)
|
||||
|
||||
async def streaming_agent(*args, **kwargs):
|
||||
collectors[0].records = [
|
||||
{
|
||||
"source_run_id": "subagent-llm-1",
|
||||
"caller": "subagent:test-agent",
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 20,
|
||||
"total_tokens": 120,
|
||||
}
|
||||
]
|
||||
yielded.set()
|
||||
yield {"messages": [msg.human("Task"), msg.ai("Working", "m1")]}
|
||||
await release.wait()
|
||||
|
||||
monkeypatch.setattr(executor_module, "SubagentTokenCollector", Collector)
|
||||
mock_agent.astream = streaming_agent
|
||||
executor = SubagentExecutor(config=base_config, tools=[], thread_id="test-thread")
|
||||
result_holder = SubagentResult(
|
||||
task_id="task-1",
|
||||
trace_id="trace-1",
|
||||
status=SubagentStatus.RUNNING,
|
||||
)
|
||||
|
||||
with patch.object(executor, "_create_agent", return_value=mock_agent):
|
||||
running = asyncio.create_task(executor._aexecute("Task", result_holder=result_holder))
|
||||
await yielded.wait()
|
||||
await asyncio.sleep(0)
|
||||
await asyncio.sleep(0)
|
||||
assert result_holder.status == SubagentStatus.RUNNING
|
||||
assert result_holder.token_usage_records == [
|
||||
{
|
||||
"source_run_id": "subagent-llm-1",
|
||||
"caller": "subagent:test-agent",
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 20,
|
||||
"total_tokens": 120,
|
||||
}
|
||||
]
|
||||
release.set()
|
||||
await running
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_aexecute_collects_ai_messages(self, classes, base_config, mock_agent, msg):
|
||||
"""Test that AI messages are collected during streaming."""
|
||||
|
||||
@@ -8,12 +8,14 @@ from pathlib import Path
|
||||
from deerflow.subagents.status_contract import (
|
||||
SUBAGENT_ERROR_KEY,
|
||||
SUBAGENT_METADATA_TEXT_MAX_CHARS,
|
||||
SUBAGENT_MODEL_NAME_KEY,
|
||||
SUBAGENT_RESULT_BRIEF_KEY,
|
||||
SUBAGENT_RESULT_SHA256_KEY,
|
||||
SUBAGENT_STATUS_KEY,
|
||||
SUBAGENT_STATUS_VALUES,
|
||||
SUBAGENT_STOP_REASON_KEY,
|
||||
SUBAGENT_STOP_REASON_VALUES,
|
||||
SUBAGENT_TOKEN_USAGE_KEY,
|
||||
_bound_metadata_text,
|
||||
format_subagent_result_message,
|
||||
make_subagent_additional_kwargs,
|
||||
@@ -49,6 +51,22 @@ def test_make_subagent_additional_kwargs_includes_status():
|
||||
assert kwargs == {SUBAGENT_STATUS_KEY: "completed"}
|
||||
|
||||
|
||||
def test_make_subagent_additional_kwargs_carries_terminal_runtime_metadata():
|
||||
kwargs = make_subagent_additional_kwargs(
|
||||
"completed",
|
||||
result="done",
|
||||
model_name="claude-3-7-sonnet",
|
||||
token_usage={"input_tokens": 100, "output_tokens": 20, "total_tokens": 120},
|
||||
)
|
||||
|
||||
assert kwargs[SUBAGENT_MODEL_NAME_KEY] == "claude-3-7-sonnet"
|
||||
assert kwargs[SUBAGENT_TOKEN_USAGE_KEY] == {
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 20,
|
||||
"total_tokens": 120,
|
||||
}
|
||||
|
||||
|
||||
def test_make_subagent_additional_kwargs_includes_error_when_present():
|
||||
kwargs = make_subagent_additional_kwargs("failed", error="boom")
|
||||
assert kwargs == {SUBAGENT_STATUS_KEY: "failed", SUBAGENT_ERROR_KEY: "boom"}
|
||||
|
||||
@@ -315,11 +315,21 @@ def test_run_event_for_task_running_carries_step_payload():
|
||||
|
||||
|
||||
def test_run_event_for_terminal_status():
|
||||
record = subagent_run_event({"type": "task_completed", "task_id": "call_1", "result": "done"})
|
||||
record = subagent_run_event(
|
||||
{
|
||||
"type": "task_completed",
|
||||
"task_id": "call_1",
|
||||
"result": "done",
|
||||
"model_name": "claude-3-7-sonnet",
|
||||
"usage": {"input_tokens": 100, "output_tokens": 20, "total_tokens": 120},
|
||||
}
|
||||
)
|
||||
|
||||
assert record["event_type"] == "subagent.end"
|
||||
assert record["content"]["status"] == "completed"
|
||||
assert record["content"]["result"] == "done"
|
||||
assert record["content"]["model_name"] == "claude-3-7-sonnet"
|
||||
assert record["content"]["usage"]["total_tokens"] == 120
|
||||
|
||||
failed = subagent_run_event({"type": "task_failed", "task_id": "call_1", "error": "boom"})
|
||||
assert failed["content"]["status"] == "failed"
|
||||
|
||||
@@ -15,10 +15,12 @@ from deerflow.sandbox.security import LOCAL_BASH_SUBAGENT_DISABLED_MESSAGE
|
||||
from deerflow.subagents.config import SubagentConfig
|
||||
from deerflow.subagents.status_contract import (
|
||||
SUBAGENT_ERROR_KEY,
|
||||
SUBAGENT_MODEL_NAME_KEY,
|
||||
SUBAGENT_RESULT_BRIEF_KEY,
|
||||
SUBAGENT_RESULT_SHA256_KEY,
|
||||
SUBAGENT_STATUS_KEY,
|
||||
SUBAGENT_STOP_REASON_KEY,
|
||||
SUBAGENT_TOKEN_USAGE_KEY,
|
||||
)
|
||||
|
||||
# Use module import so tests can patch the exact symbols referenced inside task_tool().
|
||||
@@ -117,6 +119,18 @@ def test_task_result_command_derives_content_from_status_payload():
|
||||
assert completed.additional_kwargs[SUBAGENT_STATUS_KEY] == "completed"
|
||||
assert completed.additional_kwargs[SUBAGENT_RESULT_BRIEF_KEY] == "done"
|
||||
|
||||
completed_with_runtime_metadata = _task_tool_message(
|
||||
task_tool_module._task_result_command(
|
||||
tool_call_id="tc-completed-metadata",
|
||||
status="completed",
|
||||
result="done",
|
||||
model_name="claude-3-7-sonnet",
|
||||
usage={"input_tokens": 100, "output_tokens": 20, "total_tokens": 120},
|
||||
)
|
||||
)
|
||||
assert completed_with_runtime_metadata.additional_kwargs[SUBAGENT_MODEL_NAME_KEY] == "claude-3-7-sonnet"
|
||||
assert completed_with_runtime_metadata.additional_kwargs[SUBAGENT_TOKEN_USAGE_KEY]["total_tokens"] == 120
|
||||
|
||||
failed = _task_tool_message(
|
||||
task_tool_module._task_result_command(
|
||||
tool_call_id="tc-failed",
|
||||
@@ -431,9 +445,68 @@ def test_task_tool_emits_running_and_completed_events(monkeypatch):
|
||||
|
||||
event_types = [e["type"] for e in events]
|
||||
assert event_types == ["task_started", "task_running", "task_running", "task_completed"]
|
||||
assert events[0]["model_name"] == "ark-model"
|
||||
assert events[-1]["result"] == "all done"
|
||||
|
||||
|
||||
def test_task_tool_emits_cumulative_usage_on_running_event(monkeypatch):
|
||||
config = _make_subagent_config()
|
||||
runtime = _make_runtime()
|
||||
events = []
|
||||
usage_records = [
|
||||
{
|
||||
"source_run_id": "subagent-call-1",
|
||||
"caller": "subagent:general-purpose",
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 20,
|
||||
"total_tokens": 120,
|
||||
}
|
||||
]
|
||||
responses = iter(
|
||||
[
|
||||
_make_result(
|
||||
FakeSubagentStatus.RUNNING,
|
||||
ai_messages=[{"id": "m1", "content": "researching"}],
|
||||
token_usage_records=usage_records,
|
||||
),
|
||||
_make_result(
|
||||
FakeSubagentStatus.COMPLETED,
|
||||
result="done",
|
||||
token_usage_records=usage_records,
|
||||
),
|
||||
]
|
||||
)
|
||||
|
||||
monkeypatch.setattr(task_tool_module, "SubagentStatus", FakeSubagentStatus)
|
||||
monkeypatch.setattr(
|
||||
task_tool_module,
|
||||
"SubagentExecutor",
|
||||
type("DummyExecutor", (), {"__init__": lambda self, **kwargs: None, "execute_async": lambda self, prompt, task_id=None: task_id}),
|
||||
)
|
||||
monkeypatch.setattr(task_tool_module, "get_subagent_config", lambda _: config)
|
||||
monkeypatch.setattr(task_tool_module, "get_background_task_result", lambda _: next(responses))
|
||||
monkeypatch.setattr(task_tool_module, "get_stream_writer", lambda: events.append)
|
||||
monkeypatch.setattr(task_tool_module.asyncio, "sleep", _no_sleep)
|
||||
monkeypatch.setattr(task_tool_module, "_report_subagent_usage", lambda *_: None)
|
||||
monkeypatch.setattr("deerflow.tools.get_available_tools", lambda **kwargs: [])
|
||||
|
||||
_run_task_tool(
|
||||
runtime=runtime,
|
||||
description="research",
|
||||
prompt="find facts",
|
||||
subagent_type="general-purpose",
|
||||
tool_call_id="tc-live-usage",
|
||||
)
|
||||
|
||||
running = next(event for event in events if event["type"] == "task_running")
|
||||
assert running["usage"] == {
|
||||
"input_tokens": 100,
|
||||
"output_tokens": 20,
|
||||
"total_tokens": 120,
|
||||
}
|
||||
assert running["model_name"] == "ark-model"
|
||||
|
||||
|
||||
def test_task_tool_propagates_tool_groups_to_subagent(monkeypatch):
|
||||
"""Verify tool_groups from parent metadata are passed to get_available_tools(groups=...)."""
|
||||
config = _make_subagent_config()
|
||||
|
||||
+1
-1
@@ -79,7 +79,7 @@ Human input requests are a structured message protocol layered on normal chat hi
|
||||
- **Thread hooks** (`useThreadStream`, `useSubmitThread`, `useThreads`) are the primary API interface
|
||||
- **LangGraph client** is a singleton obtained via `getAPIClient()` in `core/api/`
|
||||
- **Environment validation** uses `@t3-oss/env-nextjs` with Zod schemas (`src/env.js`). Skip with `SKIP_ENV_VALIDATION=1`
|
||||
- **Subtask step history** (`core/tasks/`) — the subtask card shows a subagent's full step timeline (#3779): its assistant reasoning turns interleaved with the tools it ran. `Subtask.steps[]` is accumulated live from `task_running` events (appended via `mergeSteps`, not overwritten) and backfilled on expand for historical runs by `fetchSubtaskSteps`, which pages the events endpoint scoped to one task (GET `/runs/{runId}/events?event_types=subagent.step&task_id=…&after_seq=…`) until a short page, so the run-wide limit can't truncate the timeline. `core/tasks/steps.ts` is the pure model: `messageToStep` (live), `eventsToSteps` (reload), `mergeSteps` (dedup by `message_index`), and `stepsForDisplay` (what the card renders — keeps tool steps + AI steps with text, drops the trailing final-answer AI step when completed since it's shown as `result`). `core/tasks/subtask-update.ts::computeNextSubtask` is the pure per-subtask state transition (merge step deltas, keep terminal status stable); `core/tasks/context.tsx`'s `useUpdateSubtask` applies it against a `tasksRef` mirroring the latest state (not a closure snapshot), so a late-resolving `fetchSubtaskSteps` backfill merges into current state instead of clobbering SSE steps or sibling subtasks that arrived meanwhile. The owning `run_id` is carried onto history content messages in `buildVisibleHistoryMessages` so the card can resolve the events endpoint.
|
||||
- **Subtask step history and runtime metadata** (`core/tasks/`) — the subtask card shows a subagent's full step timeline (#3779): its assistant reasoning turns interleaved with the tools it ran. `Subtask.steps[]` is accumulated live from `task_running` events (appended via `mergeSteps`, not overwritten) and backfilled on expand for historical runs by `fetchSubtaskSteps`, which pages the events endpoint scoped to one task (GET `/runs/{runId}/events?event_types=subagent.step&task_id=…&after_seq=…`) until a short page, so the run-wide limit can't truncate the timeline. `task_started` carries the effective `model_name`; `task_running` carries a cumulative usage snapshot after each completed LLM call. `core/tasks/lifecycle.ts` normalizes these additive events, and `computeNextSubtask` keeps the largest cumulative total so replayed or late SSE frames cannot double-count or roll the folded card backward. Terminal ToolMessage metadata (`subagent_model_name` / `subagent_token_usage`) restores the same values from normal history after reload; no per-card event fetch is needed. `core/tasks/steps.ts` is the pure step model: `messageToStep` (live), `eventsToSteps` (reload), `mergeSteps` (dedup by `message_index`), and `stepsForDisplay` (what the card renders — keeps tool steps + AI steps with text, drops the trailing final-answer AI step when completed since it's shown as `result`). `core/tasks/context.tsx`'s `useUpdateSubtask` applies updates against a `tasksRef` mirroring the latest state (not a closure snapshot), so a late-resolving `fetchSubtaskSteps` backfill merges into current state instead of clobbering SSE steps or sibling subtasks that arrived meanwhile. The owning `run_id` is carried onto history content messages in `buildVisibleHistoryMessages` so the card can resolve the events endpoint.
|
||||
|
||||
### Interaction Ownership
|
||||
|
||||
|
||||
@@ -19,11 +19,16 @@ import { Button } from "@/components/ui/button";
|
||||
import { ShineBorder } from "@/components/ui/shine-border";
|
||||
import { useI18n } from "@/core/i18n/hooks";
|
||||
import { hasToolCalls } from "@/core/messages/utils";
|
||||
import { useModels } from "@/core/models/hooks";
|
||||
import { useRehypeSplitWordsIntoSpans } from "@/core/rehype";
|
||||
import { streamdownPluginsWithWordAnimation } from "@/core/streamdown";
|
||||
import { SafeStreamdown } from "@/core/streamdown/components";
|
||||
import { fetchSubtaskSteps } from "@/core/tasks/api";
|
||||
import { useSubtask, useUpdateSubtask } from "@/core/tasks/context";
|
||||
import {
|
||||
formatSubtaskTokenUsage,
|
||||
resolveSubtaskModelLabel,
|
||||
} from "@/core/tasks/presentation";
|
||||
import { stepsForDisplay } from "@/core/tasks/steps";
|
||||
import { explainLastToolCall } from "@/core/tools/utils";
|
||||
import { cn } from "@/lib/utils";
|
||||
@@ -50,7 +55,19 @@ export function SubtaskCard({
|
||||
const [collapsed, setCollapsed] = useState(true);
|
||||
const rehypePlugins = useRehypeSplitWordsIntoSpans(isLoading);
|
||||
const task = useSubtask(taskId)!;
|
||||
const { models, tokenUsageEnabled } = useModels();
|
||||
const updateSubtask = useUpdateSubtask();
|
||||
const modelLabel = resolveSubtaskModelLabel(task.modelName, models);
|
||||
const tokenLabel = tokenUsageEnabled
|
||||
? formatSubtaskTokenUsage(task.usage)
|
||||
: undefined;
|
||||
const runtimeUsageLabel = tokenUsageEnabled
|
||||
? tokenLabel
|
||||
? `${tokenLabel} ${t.tokenUsage.label}`
|
||||
: task.status === "in_progress"
|
||||
? t.tokenUsage.collecting
|
||||
: t.tokenUsage.unavailableShort
|
||||
: undefined;
|
||||
|
||||
// The card shows the subagent's step timeline (#3779): its reasoning turns
|
||||
// (AI text) interleaved with the tools it ran (by name). See stepsForDisplay
|
||||
@@ -137,6 +154,19 @@ export function SubtaskCard({
|
||||
task.status === "failed" ? "text-red-500 opacity-67" : "",
|
||||
)}
|
||||
>
|
||||
{modelLabel && (
|
||||
<span className="max-w-32 truncate" title={modelLabel}>
|
||||
{modelLabel}
|
||||
</span>
|
||||
)}
|
||||
{runtimeUsageLabel && (
|
||||
<span
|
||||
className="max-w-28 truncate"
|
||||
title={runtimeUsageLabel}
|
||||
>
|
||||
{runtimeUsageLabel}
|
||||
</span>
|
||||
)}
|
||||
{icon}
|
||||
<FlipDisplay
|
||||
className="max-w-[420px] truncate pb-1"
|
||||
|
||||
@@ -613,6 +613,7 @@ export const enUS: Translations = {
|
||||
unavailable:
|
||||
"No token usage yet. Usage appears only after a successful model response when the provider returns usage_metadata.",
|
||||
unavailableShort: "No usage returned",
|
||||
collecting: "Collecting tokens",
|
||||
note: "Header totals use persisted thread usage, plus visible in-flight usage while a run is still streaming. Per-turn and debug usage come from currently visible messages only. Totals may differ from provider billing pages.",
|
||||
presets: {
|
||||
off: "Off",
|
||||
|
||||
@@ -502,6 +502,7 @@ export interface Translations {
|
||||
view: string;
|
||||
unavailable: string;
|
||||
unavailableShort: string;
|
||||
collecting: string;
|
||||
note: string;
|
||||
presets: {
|
||||
off: string;
|
||||
|
||||
@@ -590,6 +590,7 @@ export const zhCN: Translations = {
|
||||
unavailable:
|
||||
"暂无 Token 用量。只有模型成功返回且供应商提供 usage_metadata 时才会显示。",
|
||||
unavailableShort: "未返回用量",
|
||||
collecting: "统计中",
|
||||
note: "顶部总量优先使用后端持久化的线程用量；当当前回复仍在流式返回时，还会叠加可见的进行中用量。每轮和调试用量只来自当前可见消息，可能与平台账单页不完全一致。",
|
||||
presets: {
|
||||
off: "关闭",
|
||||
|
||||
@@ -77,6 +77,40 @@ export function accumulateUsage(messages: Message[]): TokenUsage | null {
|
||||
return hasUsage ? cumulative : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Validate a raw `{input,output,total}_tokens` object into {@link TokenUsage}.
|
||||
*
|
||||
* The single shared validator for both sub-agent usage surfaces — the live
|
||||
* `task_running` event (`core/tasks/lifecycle.ts`) and the terminal ToolMessage
|
||||
* metadata (`core/tasks/subtask-result.ts`). Keeping one function stops the two
|
||||
* from drifting (e.g. one accepting an extra token field the other rejects).
|
||||
* Every key must be a finite, non-negative number or the whole snapshot is
|
||||
* rejected as `undefined`.
|
||||
*/
|
||||
export function normalizeTokenUsage(value: unknown): TokenUsage | undefined {
|
||||
if (typeof value !== "object" || value === null) {
|
||||
return undefined;
|
||||
}
|
||||
const record = value as Record<string, unknown>;
|
||||
const inputTokens = nonNegativeNumber(record.input_tokens);
|
||||
const outputTokens = nonNegativeNumber(record.output_tokens);
|
||||
const totalTokens = nonNegativeNumber(record.total_tokens);
|
||||
if (
|
||||
inputTokens === undefined ||
|
||||
outputTokens === undefined ||
|
||||
totalTokens === undefined
|
||||
) {
|
||||
return undefined;
|
||||
}
|
||||
return { inputTokens, outputTokens, totalTokens };
|
||||
}
|
||||
|
||||
function nonNegativeNumber(value: unknown): number | undefined {
|
||||
return typeof value === "number" && Number.isFinite(value) && value >= 0
|
||||
? value
|
||||
: undefined;
|
||||
}
|
||||
|
||||
export function hasNonZeroUsage(
|
||||
usage: TokenUsage | null | undefined,
|
||||
): usage is TokenUsage {
|
||||
|
||||
@@ -8,6 +8,12 @@ export function useModels({ enabled = true }: { enabled?: boolean } = {}) {
|
||||
queryFn: () => loadModels(),
|
||||
enabled,
|
||||
refetchOnWindowFocus: false,
|
||||
// Model config changes rarely and every subtask card mounts its own
|
||||
// observer of this query; without a staleTime each newly-mounted card would
|
||||
// refetch /api/models on mount (default staleTime: 0). Treat the list as
|
||||
// fresh for the session so a long conversation with many cards issues one
|
||||
// request, not one per card.
|
||||
staleTime: Infinity,
|
||||
});
|
||||
return {
|
||||
models: data?.models ?? [],
|
||||
|
||||
@@ -7,7 +7,7 @@ import {
|
||||
useState,
|
||||
} from "react";
|
||||
|
||||
import { computeNextSubtask } from "./subtask-update";
|
||||
import { computeNextSubtask, subtaskNotification } from "./subtask-update";
|
||||
import type { Subtask } from "./types";
|
||||
|
||||
export interface SubtaskContextValue {
|
||||
@@ -74,18 +74,23 @@ export function useUpdateSubtask() {
|
||||
// fetchSubtaskSteps().then(updateSubtask) resolving late would write a stale
|
||||
// map, clobbering SSE steps/status and sibling subtasks added meanwhile (#3779).
|
||||
const current = tasksRef.current;
|
||||
const { next, becameTerminal } = computeNextSubtask(
|
||||
const { next, becameTerminal, changed } = computeNextSubtask(
|
||||
current[task.id],
|
||||
task,
|
||||
);
|
||||
|
||||
current[task.id] = next;
|
||||
|
||||
if (task.latestMessage || task.steps) {
|
||||
// Gate on an actual state change, not mere field presence. The terminal
|
||||
// ToolMessage is re-parsed on every MessageList render and always carries
|
||||
// modelName/usage, so a presence check would setTasks({...}) with a fresh
|
||||
// reference each render — an infinite loop. `subtaskNotification` routes a
|
||||
// terminal transition through the deferred (after-render) path and skips
|
||||
// no-op re-parses entirely.
|
||||
const notify = subtaskNotification(task, { becameTerminal, changed });
|
||||
if (notify === "eager") {
|
||||
setTasks({ ...current });
|
||||
} else if (becameTerminal) {
|
||||
// Defer the render to the after-render effect so a terminal-only update
|
||||
// does not loop with MessageList's same-render pending write.
|
||||
} else if (notify === "deferred") {
|
||||
shouldNotifyAfterRenderRef.current = true;
|
||||
}
|
||||
},
|
||||
|
||||
@@ -0,0 +1,65 @@
|
||||
import { normalizeTokenUsage } from "../messages/usage";
|
||||
|
||||
import type { Subtask } from "./types";
|
||||
|
||||
type TaskStartedEvent = {
|
||||
type: "task_started";
|
||||
task_id: string;
|
||||
model_name?: unknown;
|
||||
};
|
||||
|
||||
type TaskRunningEvent = {
|
||||
type: "task_running";
|
||||
task_id: string;
|
||||
model_name?: unknown;
|
||||
usage?: unknown;
|
||||
};
|
||||
|
||||
/** Convert an additive task lifecycle event into a task-state update. */
|
||||
export function taskEventToSubtaskUpdate(
|
||||
event: unknown,
|
||||
): (Partial<Subtask> & { id: string }) | null {
|
||||
if (!isRecord(event)) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const taskId = event.task_id;
|
||||
if (typeof taskId !== "string" || !taskId.trim()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (event.type === "task_started") {
|
||||
const started = event as TaskStartedEvent;
|
||||
const modelName =
|
||||
typeof started.model_name === "string" && started.model_name.trim()
|
||||
? started.model_name.trim()
|
||||
: undefined;
|
||||
return {
|
||||
id: taskId,
|
||||
...(modelName ? { modelName } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
if (event.type === "task_running") {
|
||||
const running = event as TaskRunningEvent;
|
||||
const usage = normalizeTokenUsage(running.usage);
|
||||
const modelName = normalizeModelName(running.model_name);
|
||||
return usage || modelName
|
||||
? {
|
||||
id: taskId,
|
||||
...(modelName ? { modelName } : {}),
|
||||
...(usage ? { usage } : {}),
|
||||
}
|
||||
: null;
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null;
|
||||
}
|
||||
|
||||
function normalizeModelName(value: unknown): string | undefined {
|
||||
return typeof value === "string" && value.trim() ? value.trim() : undefined;
|
||||
}
|
||||
@@ -0,0 +1,21 @@
|
||||
import { formatTokenCount, type TokenUsage } from "@/core/messages/usage";
|
||||
import type { Model } from "@/core/models/types";
|
||||
|
||||
/** Return the user-facing label for a configured subagent model. */
|
||||
export function resolveSubtaskModelLabel(
|
||||
modelName: string | undefined,
|
||||
models: Model[],
|
||||
): string | undefined {
|
||||
if (!modelName) {
|
||||
return undefined;
|
||||
}
|
||||
return (
|
||||
models.find((model) => model.name === modelName)?.display_name ?? modelName
|
||||
);
|
||||
}
|
||||
|
||||
export function formatSubtaskTokenUsage(
|
||||
usage: TokenUsage | undefined,
|
||||
): string | undefined {
|
||||
return usage ? formatTokenCount(usage.totalTokens) : undefined;
|
||||
}
|
||||
@@ -1,5 +1,7 @@
|
||||
import type { Message } from "@langchain/langgraph-sdk";
|
||||
|
||||
import { normalizeTokenUsage } from "../messages/usage";
|
||||
|
||||
import type { Subtask } from "./types";
|
||||
|
||||
export type SubtaskStatus = Subtask["status"];
|
||||
@@ -8,6 +10,8 @@ export interface SubtaskResultUpdate {
|
||||
status: SubtaskStatus;
|
||||
result?: string;
|
||||
error?: string;
|
||||
modelName?: string;
|
||||
usage?: Subtask["usage"];
|
||||
/**
|
||||
* Why a guardrail cap ended the run early (``token_capped`` / ``turn_capped``
|
||||
* / ``loop_capped``), when the backend stamps ``subagent_stop_reason``. A
|
||||
@@ -25,7 +29,8 @@ export interface SubtaskResultUpdate {
|
||||
* The values mirror the Python contract in
|
||||
* ``backend/packages/harness/deerflow/subagents/status_contract.py``
|
||||
* (``SUBAGENT_STATUS_KEY`` / ``SUBAGENT_ERROR_KEY`` /
|
||||
* ``SUBAGENT_RESULT_BRIEF_KEY`` / ``SUBAGENT_RESULT_SHA256_KEY``). The
|
||||
* ``SUBAGENT_RESULT_BRIEF_KEY`` / ``SUBAGENT_RESULT_SHA256_KEY`` /
|
||||
* ``SUBAGENT_MODEL_NAME_KEY`` / ``SUBAGENT_TOKEN_USAGE_KEY``). The
|
||||
* result metadata fields are optional and bounded: ``subagent_result_brief``
|
||||
* carries a trimmed summary for completed tasks and
|
||||
* ``subagent_result_sha256`` carries the full-result digest. The
|
||||
@@ -37,6 +42,8 @@ export const SUBAGENT_STOP_REASON_KEY = "subagent_stop_reason";
|
||||
export const SUBAGENT_ERROR_KEY = "subagent_error";
|
||||
export const SUBAGENT_RESULT_BRIEF_KEY = "subagent_result_brief";
|
||||
export const SUBAGENT_RESULT_SHA256_KEY = "subagent_result_sha256";
|
||||
export const SUBAGENT_MODEL_NAME_KEY = "subagent_model_name";
|
||||
export const SUBAGENT_TOKEN_USAGE_KEY = "subagent_token_usage";
|
||||
/**
|
||||
* Why a guardrail cap ended a subagent run early (#3875 Phase 2). Mirrors the
|
||||
* Python ``SUBAGENT_STOP_REASON_VALUES`` and the shared fixture's
|
||||
@@ -54,6 +61,8 @@ const STRUCTURED_SUBAGENT_KEYS = [
|
||||
SUBAGENT_ERROR_KEY,
|
||||
SUBAGENT_RESULT_BRIEF_KEY,
|
||||
SUBAGENT_RESULT_SHA256_KEY,
|
||||
SUBAGENT_MODEL_NAME_KEY,
|
||||
SUBAGENT_TOKEN_USAGE_KEY,
|
||||
];
|
||||
|
||||
const SUCCESS_PREFIX = "Task Succeeded. Result:";
|
||||
@@ -129,6 +138,14 @@ export function parseSubtaskResult(
|
||||
if (stopReason) {
|
||||
update.stopReason = stopReason;
|
||||
}
|
||||
const modelName = readStructuredModelName(additionalKwargs);
|
||||
if (modelName) {
|
||||
update.modelName = modelName;
|
||||
}
|
||||
const usage = readStructuredTokenUsage(additionalKwargs);
|
||||
if (usage) {
|
||||
update.usage = usage;
|
||||
}
|
||||
return update;
|
||||
}
|
||||
|
||||
@@ -239,3 +256,16 @@ function readStructuredStopReason(
|
||||
? value
|
||||
: undefined;
|
||||
}
|
||||
|
||||
function readStructuredModelName(
|
||||
additionalKwargs: Record<string, unknown> | null | undefined,
|
||||
): string | undefined {
|
||||
const value = additionalKwargs?.[SUBAGENT_MODEL_NAME_KEY];
|
||||
return typeof value === "string" && value.trim() ? value.trim() : undefined;
|
||||
}
|
||||
|
||||
function readStructuredTokenUsage(
|
||||
additionalKwargs: Record<string, unknown> | null | undefined,
|
||||
): Subtask["usage"] | undefined {
|
||||
return normalizeTokenUsage(additionalKwargs?.[SUBAGENT_TOKEN_USAGE_KEY]);
|
||||
}
|
||||
|
||||
@@ -22,7 +22,7 @@ export function isTerminalSubtaskStatus(status: Subtask["status"] | undefined) {
|
||||
export function computeNextSubtask(
|
||||
previous: Subtask | undefined,
|
||||
task: Partial<Subtask> & { id: string },
|
||||
): { next: Subtask; becameTerminal: boolean } {
|
||||
): { next: Subtask; becameTerminal: boolean; changed: boolean } {
|
||||
const previousStatus = previous?.status;
|
||||
|
||||
// MessageList writes the pending task tool-call state before parsing the
|
||||
@@ -40,8 +40,90 @@ export function computeNextSubtask(
|
||||
next.steps = mergeSteps(previous?.steps ?? [], task.steps);
|
||||
}
|
||||
|
||||
// Usage events are cumulative snapshots. A delayed older frame must never
|
||||
// make the folded card appear to spend fewer tokens than it already did.
|
||||
if (
|
||||
task.usage &&
|
||||
previous?.usage &&
|
||||
task.usage.totalTokens < previous.usage.totalTokens
|
||||
) {
|
||||
next.usage = previous.usage;
|
||||
}
|
||||
|
||||
const becameTerminal =
|
||||
isTerminalSubtaskStatus(next.status) && previousStatus !== next.status;
|
||||
|
||||
return { next, becameTerminal };
|
||||
return { next, becameTerminal, changed: subtaskChanged(previous, next) };
|
||||
}
|
||||
|
||||
/**
|
||||
* Did `next` materially differ from `previous`?
|
||||
*
|
||||
* The terminal ToolMessage is re-parsed on *every* MessageList render, and
|
||||
* `parseSubtaskResult` rebuilds `modelName`/`usage` into a fresh object each
|
||||
* time. Comparing by value (not reference) is what lets the hook skip a
|
||||
* redundant `setTasks` for an idempotent re-parse — without it the new object
|
||||
* identity drives an infinite render loop once a subagent finishes.
|
||||
*/
|
||||
function subtaskChanged(prev: Subtask | undefined, next: Subtask): boolean {
|
||||
if (!prev) {
|
||||
return true;
|
||||
}
|
||||
return (
|
||||
prev.status !== next.status ||
|
||||
prev.modelName !== next.modelName ||
|
||||
prev.result !== next.result ||
|
||||
prev.error !== next.error ||
|
||||
prev.stopReason !== next.stopReason ||
|
||||
prev.subagent_type !== next.subagent_type ||
|
||||
prev.description !== next.description ||
|
||||
prev.prompt !== next.prompt ||
|
||||
prev.latestMessage !== next.latestMessage ||
|
||||
prev.steps !== next.steps ||
|
||||
!usageEquals(prev.usage, next.usage)
|
||||
);
|
||||
}
|
||||
|
||||
function usageEquals(a: Subtask["usage"], b: Subtask["usage"]): boolean {
|
||||
if (a === b) {
|
||||
return true;
|
||||
}
|
||||
if (!a || !b) {
|
||||
return false;
|
||||
}
|
||||
return (
|
||||
a.inputTokens === b.inputTokens &&
|
||||
a.outputTokens === b.outputTokens &&
|
||||
a.totalTokens === b.totalTokens
|
||||
);
|
||||
}
|
||||
|
||||
export type SubtaskNotification = "eager" | "deferred" | "none";
|
||||
|
||||
/**
|
||||
* Decide how `useUpdateSubtask` should publish a computed transition.
|
||||
*
|
||||
* - `deferred`: a terminal transition. These arrive while MessageList renders
|
||||
* (it parses the ToolMessage inline), so we must not `setTasks` mid-render;
|
||||
* the hook flips a ref and publishes in an after-render effect instead.
|
||||
* - `eager`: a live SSE update (steps / latestMessage / model / usage) that
|
||||
* actually changed state — publish immediately from the async callback.
|
||||
* - `none`: nothing changed. Critically, a re-parsed terminal result still
|
||||
* carries `modelName`/`usage`, so gating on `changed` (not mere presence) is
|
||||
* what stops the render loop.
|
||||
*/
|
||||
export function subtaskNotification(
|
||||
task: Partial<Subtask> & { id: string },
|
||||
transition: { becameTerminal: boolean; changed: boolean },
|
||||
): SubtaskNotification {
|
||||
if (transition.becameTerminal) {
|
||||
return "deferred";
|
||||
}
|
||||
if (
|
||||
transition.changed &&
|
||||
(task.latestMessage || task.steps || task.modelName || task.usage)
|
||||
) {
|
||||
return "eager";
|
||||
}
|
||||
return "none";
|
||||
}
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
import type { AIMessage } from "@langchain/langgraph-sdk";
|
||||
|
||||
import type { TokenUsage } from "../messages/usage";
|
||||
|
||||
import type { SubtaskStep } from "./steps";
|
||||
|
||||
export interface Subtask {
|
||||
@@ -7,6 +9,10 @@ export interface Subtask {
|
||||
status: "in_progress" | "completed" | "failed";
|
||||
subagent_type: string;
|
||||
description: string;
|
||||
/** Effective DeerFlow model selected for this subagent run. */
|
||||
modelName?: string;
|
||||
/** Latest cumulative token snapshot reported while the subagent runs. */
|
||||
usage?: TokenUsage;
|
||||
latestMessage?: AIMessage;
|
||||
/**
|
||||
* Full ordered step history (assistant turns + tool outputs) of the subagent.
|
||||
|
||||
@@ -23,6 +23,7 @@ import type { FileInMessage } from "../messages/utils";
|
||||
import type { LocalSettings } from "../settings";
|
||||
import { isSidecarThread, SIDECAR_METADATA_KEY } from "../sidecar/thread";
|
||||
import { useUpdateSubtask } from "../tasks/context";
|
||||
import { taskEventToSubtaskUpdate } from "../tasks/lifecycle";
|
||||
import { messageToStep } from "../tasks/steps";
|
||||
import type { UploadedFileInfo } from "../uploads";
|
||||
import { promptInputFilePartToFile, uploadFiles } from "../uploads";
|
||||
@@ -1016,12 +1017,20 @@ export function useThreadStream({
|
||||
}
|
||||
},
|
||||
onCustomEvent(event: unknown) {
|
||||
if (
|
||||
typeof event === "object" &&
|
||||
event !== null &&
|
||||
"type" in event &&
|
||||
event.type === "task_running"
|
||||
) {
|
||||
// Narrow `event.type` once; taskEventToSubtaskUpdate already validated the
|
||||
// task_* events, so the per-branch re-narrowing below reads this single
|
||||
// source of truth instead of re-checking the object shape each time.
|
||||
const eventType =
|
||||
typeof event === "object" && event !== null && "type" in event
|
||||
? (event as { type: unknown }).type
|
||||
: undefined;
|
||||
|
||||
const taskUpdate = taskEventToSubtaskUpdate(event);
|
||||
if (taskUpdate) {
|
||||
updateSubtask(taskUpdate);
|
||||
}
|
||||
|
||||
if (eventType === "task_running") {
|
||||
const e = event as {
|
||||
type: "task_running";
|
||||
task_id: string;
|
||||
@@ -1039,17 +1048,11 @@ export function useThreadStream({
|
||||
return;
|
||||
}
|
||||
|
||||
if (
|
||||
typeof event === "object" &&
|
||||
event !== null &&
|
||||
"type" in event &&
|
||||
event.type === "llm_retry" &&
|
||||
"message" in event &&
|
||||
typeof event.message === "string" &&
|
||||
event.message.trim()
|
||||
) {
|
||||
const e = event as { type: "llm_retry"; message: string };
|
||||
toast(e.message);
|
||||
if (eventType === "llm_retry") {
|
||||
const e = event as { type: "llm_retry"; message?: unknown };
|
||||
if (typeof e.message === "string" && e.message.trim()) {
|
||||
toast(e.message);
|
||||
}
|
||||
}
|
||||
},
|
||||
onError(error) {
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import { describe, expect, it } from "@rstest/core";
|
||||
|
||||
import { taskEventToSubtaskUpdate } from "@/core/tasks/lifecycle";
|
||||
|
||||
describe("taskEventToSubtaskUpdate", () => {
|
||||
it("maps a task-start event to the effective model for that task", () => {
|
||||
expect(
|
||||
taskEventToSubtaskUpdate({
|
||||
type: "task_started",
|
||||
task_id: "call-1",
|
||||
description: "Research auth",
|
||||
model_name: "claude-3-7-sonnet",
|
||||
}),
|
||||
).toEqual({
|
||||
id: "call-1",
|
||||
modelName: "claude-3-7-sonnet",
|
||||
});
|
||||
});
|
||||
|
||||
it("maps a running event to its cumulative token snapshot", () => {
|
||||
expect(
|
||||
taskEventToSubtaskUpdate({
|
||||
type: "task_running",
|
||||
task_id: "call-1",
|
||||
model_name: "claude-3-7-sonnet",
|
||||
usage: {
|
||||
input_tokens: 100,
|
||||
output_tokens: 20,
|
||||
total_tokens: 120,
|
||||
},
|
||||
}),
|
||||
).toEqual({
|
||||
id: "call-1",
|
||||
modelName: "claude-3-7-sonnet",
|
||||
usage: {
|
||||
inputTokens: 100,
|
||||
outputTokens: 20,
|
||||
totalTokens: 120,
|
||||
},
|
||||
});
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,36 @@
|
||||
import { describe, expect, it } from "@rstest/core";
|
||||
|
||||
import {
|
||||
formatSubtaskTokenUsage,
|
||||
resolveSubtaskModelLabel,
|
||||
} from "@/core/tasks/presentation";
|
||||
|
||||
describe("resolveSubtaskModelLabel", () => {
|
||||
it("prefers the configured display name and falls back to the model identifier", () => {
|
||||
expect(
|
||||
resolveSubtaskModelLabel("claude-3-7-sonnet", [
|
||||
{
|
||||
id: "model-1",
|
||||
name: "claude-3-7-sonnet",
|
||||
model: "claude-3-7-sonnet@20250219",
|
||||
display_name: "Claude 3.7 Sonnet",
|
||||
},
|
||||
]),
|
||||
).toBe("Claude 3.7 Sonnet");
|
||||
|
||||
expect(resolveSubtaskModelLabel("unlisted-model", [])).toBe(
|
||||
"unlisted-model",
|
||||
);
|
||||
});
|
||||
|
||||
it("formats only reported cumulative token usage", () => {
|
||||
expect(formatSubtaskTokenUsage(undefined)).toBeUndefined();
|
||||
expect(
|
||||
formatSubtaskTokenUsage({
|
||||
inputTokens: 10_000,
|
||||
outputTokens: 2_345,
|
||||
totalTokens: 12_345,
|
||||
}),
|
||||
).toBe("12.3K");
|
||||
});
|
||||
});
|
||||
@@ -6,9 +6,11 @@ import { describe, expect, it } from "@rstest/core";
|
||||
|
||||
import {
|
||||
SUBAGENT_ERROR_KEY,
|
||||
SUBAGENT_MODEL_NAME_KEY,
|
||||
SUBAGENT_RESULT_BRIEF_KEY,
|
||||
SUBAGENT_STATUS_KEY,
|
||||
SUBAGENT_STOP_REASON_KEY,
|
||||
SUBAGENT_TOKEN_USAGE_KEY,
|
||||
derivePendingSubtaskStatus,
|
||||
hasSubtaskToolResult,
|
||||
parseSubtaskResult,
|
||||
@@ -147,6 +149,28 @@ describe("parseSubtaskResult — structured additional_kwargs (preferred path)",
|
||||
expect(parsed.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("restores terminal model and token usage metadata", () => {
|
||||
expect(
|
||||
parseSubtaskResult("Task Succeeded. Result: done", {
|
||||
[SUBAGENT_STATUS_KEY]: "completed",
|
||||
[SUBAGENT_MODEL_NAME_KEY]: "claude-3-7-sonnet",
|
||||
[SUBAGENT_TOKEN_USAGE_KEY]: {
|
||||
input_tokens: 100,
|
||||
output_tokens: 20,
|
||||
total_tokens: 120,
|
||||
},
|
||||
}),
|
||||
).toMatchObject({
|
||||
status: "completed",
|
||||
modelName: "claude-3-7-sonnet",
|
||||
usage: {
|
||||
inputTokens: 100,
|
||||
outputTokens: 20,
|
||||
totalTokens: 120,
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("collapses cancelled / timed_out / polling_timed_out to failed for the card UI", () => {
|
||||
for (const backendStatus of [
|
||||
"cancelled",
|
||||
|
||||
@@ -4,6 +4,7 @@ import type { SubtaskStep } from "@/core/tasks/steps";
|
||||
import {
|
||||
computeNextSubtask,
|
||||
isTerminalSubtaskStatus,
|
||||
subtaskNotification,
|
||||
} from "@/core/tasks/subtask-update";
|
||||
import type { Subtask } from "@/core/tasks/types";
|
||||
|
||||
@@ -56,6 +57,23 @@ describe("computeNextSubtask", () => {
|
||||
expect(next.steps?.map((s) => s.message_index)).toEqual([1, 2, 3]);
|
||||
});
|
||||
|
||||
it("keeps the latest cumulative token snapshot when an older event arrives late", () => {
|
||||
const previous = baseTask({
|
||||
usage: { inputTokens: 200, outputTokens: 40, totalTokens: 240 },
|
||||
});
|
||||
|
||||
const { next } = computeNextSubtask(previous, {
|
||||
id: "t1",
|
||||
usage: { inputTokens: 100, outputTokens: 20, totalTokens: 120 },
|
||||
});
|
||||
|
||||
expect(next.usage).toEqual({
|
||||
inputTokens: 200,
|
||||
outputTokens: 40,
|
||||
totalTokens: 240,
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps a terminal status stable against a late in_progress write", () => {
|
||||
const previous = baseTask({ status: "completed" });
|
||||
|
||||
@@ -96,6 +114,96 @@ describe("computeNextSubtask", () => {
|
||||
expect(next.steps?.map((s) => s.message_index)).toEqual([1]);
|
||||
expect(becameTerminal).toBe(false);
|
||||
});
|
||||
|
||||
it("reports changed=false when a terminal update carries identical runtime metadata", () => {
|
||||
// The terminal ToolMessage is re-parsed on every MessageList render, so the
|
||||
// same modelName/usage arrive again as a *new* object each time. Value-equal
|
||||
// re-application must not be flagged as a change or the card loops forever.
|
||||
const previous = baseTask({
|
||||
status: "completed",
|
||||
result: "done",
|
||||
modelName: "opus",
|
||||
usage: { inputTokens: 100, outputTokens: 20, totalTokens: 120 },
|
||||
});
|
||||
|
||||
const { changed } = computeNextSubtask(previous, {
|
||||
id: "t1",
|
||||
status: "completed",
|
||||
result: "done",
|
||||
modelName: "opus",
|
||||
usage: { inputTokens: 100, outputTokens: 20, totalTokens: 120 },
|
||||
});
|
||||
|
||||
expect(changed).toBe(false);
|
||||
});
|
||||
|
||||
it("reports changed=true when runtime metadata actually differs", () => {
|
||||
const previous = baseTask({
|
||||
status: "completed",
|
||||
modelName: "opus",
|
||||
usage: { inputTokens: 100, outputTokens: 20, totalTokens: 120 },
|
||||
});
|
||||
|
||||
const { changed } = computeNextSubtask(previous, {
|
||||
id: "t1",
|
||||
status: "completed",
|
||||
modelName: "opus",
|
||||
usage: { inputTokens: 300, outputTokens: 60, totalTokens: 360 },
|
||||
});
|
||||
|
||||
expect(changed).toBe(true);
|
||||
});
|
||||
});
|
||||
|
||||
describe("subtaskNotification", () => {
|
||||
it("defers a terminal transition (arrives during render, must not setState mid-render)", () => {
|
||||
expect(
|
||||
subtaskNotification(
|
||||
{ id: "t1", status: "completed", modelName: "opus" },
|
||||
{ becameTerminal: true, changed: true },
|
||||
),
|
||||
).toBe("deferred");
|
||||
});
|
||||
|
||||
it("eagerly reflects a live SSE update that actually changed", () => {
|
||||
expect(
|
||||
subtaskNotification(
|
||||
{
|
||||
id: "t1",
|
||||
usage: { inputTokens: 300, outputTokens: 60, totalTokens: 360 },
|
||||
},
|
||||
{ becameTerminal: false, changed: true },
|
||||
),
|
||||
).toBe("eager");
|
||||
});
|
||||
|
||||
it("does nothing when a re-parsed terminal result carries unchanged metadata", () => {
|
||||
// Regression: modelName/usage are present on every terminal re-parse, but the
|
||||
// state did not change. Firing setTasks here is the render loop (P1).
|
||||
expect(
|
||||
subtaskNotification(
|
||||
{
|
||||
id: "t1",
|
||||
status: "completed",
|
||||
modelName: "opus",
|
||||
usage: { inputTokens: 100, outputTokens: 20, totalTokens: 120 },
|
||||
},
|
||||
{ becameTerminal: false, changed: false },
|
||||
),
|
||||
).toBe("none");
|
||||
});
|
||||
|
||||
it("does nothing when a replayed SSE usage snapshot did not change state", () => {
|
||||
expect(
|
||||
subtaskNotification(
|
||||
{
|
||||
id: "t1",
|
||||
usage: { inputTokens: 100, outputTokens: 20, totalTokens: 120 },
|
||||
},
|
||||
{ becameTerminal: false, changed: false },
|
||||
),
|
||||
).toBe("none");
|
||||
});
|
||||
});
|
||||
|
||||
describe("isTerminalSubtaskStatus", () => {
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
# Plan: Subagent Card Runtime Metadata
|
||||
|
||||
> Source PRD: Conversation request approved on 2026-07-10 — show live token usage and the effective LLM name on collapsed subagent cards.
|
||||
|
||||
## Architectural decisions
|
||||
|
||||
- **Runtime identity**: Every metadata update is keyed by the existing subagent `task_id`, so parallel delegations in one lead-agent turn remain isolated.
|
||||
- **Usage schema**: Runtime payloads carry cumulative `input_tokens`, `output_tokens`, and `total_tokens`. They are snapshots, not deltas, so replayed or out-of-order stream frames cannot double-count usage.
|
||||
- **Update cadence**: “Live” means after each completed subagent LLM response. Providers generally do not expose authoritative usage before a response completes.
|
||||
- **Model identity**: The wire contract carries the effective DeerFlow model name resolved for the subagent. The UI prefers the configured display name and falls back to the raw model name. Provider deployment identifiers remain observability data, not the primary card label.
|
||||
- **Live and durable sources**: Custom task lifecycle events drive in-flight updates. Terminal ToolMessage metadata restores the same values from checkpointed chat history. `subagent.end` events retain the terminal snapshot for audit/debug consumers.
|
||||
- **Compatibility**: All protocol additions are optional. Older runs render without runtime metadata, missing provider usage renders as unavailable rather than zero, and no database migration is required because existing JSON payloads are extended additively.
|
||||
- **Existing totals**: Parent-run and thread token accounting remains unchanged; card metadata is a presentation projection and must not report usage to `RunJournal` a second time.
|
||||
- **Feature gate**: Token rendering follows the existing `token_usage.enabled` setting. Model identity may still be shown when token rendering is disabled.
|
||||
|
||||
---
|
||||
|
||||
## Phase 1: Live Model Identity
|
||||
|
||||
**User stories**: A user can collapse a running subagent card and immediately see which configured LLM is executing it.
|
||||
|
||||
### What to build
|
||||
|
||||
Carry the effective subagent model name through the task-start lifecycle event and merge it into the task state by `task_id`. Resolve the friendly model display name in the workspace and render it in the collapsed-card header without displacing the existing status indicator.
|
||||
|
||||
### Acceptance criteria
|
||||
|
||||
- [ ] A running collapsed card shows its effective model as soon as the task-start event arrives.
|
||||
- [ ] Parallel subagents using different models display the correct model on their own cards.
|
||||
- [ ] The configured display name is preferred; an unknown model falls back to the raw identifier.
|
||||
- [ ] Older task events without a model continue to render normally.
|
||||
|
||||
---
|
||||
|
||||
## Phase 2: Live Cumulative Token Usage
|
||||
|
||||
**User stories**: A user watching a collapsed running subagent card sees its token total increase after each completed subagent LLM call.
|
||||
|
||||
### What to build
|
||||
|
||||
Publish the collector’s latest cumulative usage snapshot while the subagent is running and attach it to task progress events. Merge snapshots into frontend task state as authoritative cumulative values, then render the formatted total beside the model label. Preserve the existing parent-run accounting path without adding another accounting write.
|
||||
|
||||
### Acceptance criteria
|
||||
|
||||
- [ ] The first completed subagent LLM call updates the collapsed card from “collecting” to a non-zero token total when usage is available.
|
||||
- [ ] Later calls replace the card snapshot with the new cumulative total rather than adding the total again.
|
||||
- [ ] Replayed, duplicate, or older progress events never double-count or decrease the displayed total.
|
||||
- [ ] Concurrent subagents maintain independent totals keyed by `task_id`.
|
||||
- [ ] Providers that omit usage metadata show an unavailable/collecting state, never a fabricated zero.
|
||||
|
||||
---
|
||||
|
||||
## Phase 3: Terminal Durability and Edge Paths
|
||||
|
||||
**User stories**: A user sees the same final model and token usage after completion, failure, cancellation, timeout, or page reload.
|
||||
|
||||
### What to build
|
||||
|
||||
Stamp the final model and cumulative usage into the existing structured task ToolMessage metadata and the persisted `subagent.end` event. Teach history reconstruction to read the optional metadata, with live snapshots and terminal history converging on the same task model. Cover every terminal status and tolerate legacy/malformed metadata safely.
|
||||
|
||||
### Acceptance criteria
|
||||
|
||||
- [ ] Completed, failed, cancelled, and timed-out cards retain their final model and usage.
|
||||
- [ ] Reloading a thread restores metadata from normal message history without one request per card.
|
||||
- [ ] Persisted `subagent.end` events contain the same terminal snapshot for audit/debug use.
|
||||
- [ ] Legacy cards without metadata and providers without usage remain readable and show an explicit unavailable state.
|
||||
- [ ] The right-side thread Token Usage total is unchanged and still counts subagent usage exactly once.
|
||||
- [ ] Backend tests, frontend unit tests, type checks, formatting, and relevant regression suites pass.
|
||||
Reference in New Issue
Block a user