feat(delegate): tag every subagent progress line with its batch id

Concurrent or nested delegation batches (a parent's 9-way fan-out plus a
child's own 3-way fan-out) printed interleaved `✓ [3/3]` / `✓ [3/9]` lines
with nothing identifying which batch each belongs to.

- CLI: batch header `🔀 [6a66] delegating 9 tasks`; completion lines and
  child tree-view lines become `[6a66 3/9]`; spinner remaining-count tagged.
- Relay: `delegation_id` rides on every `subagent.*` event (TUI gateway
  payload, api_server SSE subagent.start/complete).
- TUI: `[6a66 3/9]` prefix on /agents rows; Desktop Agents pane groups
  workers by exact delegation_id (heuristic shape/time grouping kept for
  older backends) and shows the tag on the group header.
- Tag = last 4 hex of the deleg_xxxxxxxx id (format_batch_tag), same id
  returned by the dispatch and used for cache/delegation/live/<id>/.
This commit is contained in:
Teknium
2026-09-01 22:34:29 -07:00
parent 2adb1a4ea6
commit a2600740e8
12 changed files with 251 additions and 20 deletions
+40 -2
View File
@@ -143,19 +143,56 @@ const flatten = (nodes: readonly SubagentNode[]): SubagentNode[] =>
interface RootGroup {
id: string
delegationIndex: number
/** Short batch tag (`deleg_6a664903` → `6a66`) when the backend sent one. */
batchTag?: string
nodes: SubagentNode[]
taskCount: number
}
/** `deleg_6a664903` → `6a66`; mirrors tools.delegate_tool.format_batch_tag. */
export const batchTagOf = (delegationId: string | undefined): string | undefined => {
if (!delegationId) {
return undefined
}
const short = delegationId.split('_').at(-1)?.slice(0, 4)
return short || undefined
}
function groupDelegations(roots: readonly SubagentNode[]): RootGroup[] {
const groups: RootGroup[] = []
let n = 0
for (const node of roots) {
// Exact grouping when the backend tags workers with their batch id —
// concurrent or nested fan-outs of the same shape must not merge.
if (node.delegationId) {
const byId = groups.find(g => g.id === `delegation:${node.delegationId}`)
if (byId) {
byId.nodes.push(node)
continue
}
n += 1
groups.push({
id: `delegation:${node.delegationId}`,
delegationIndex: n,
batchTag: batchTagOf(node.delegationId),
nodes: [node],
taskCount: node.taskCount
})
continue
}
// Older backends (no delegation_id): heuristic grouping by shape + time.
const prev = groups.at(-1)
const prevTail = prev?.nodes.at(-1)
const closeInTime = prevTail ? Math.abs(node.startedAt - prevTail.startedAt) <= 5_000 : false
const sameShape = prev && node.taskCount > 1 && prev.taskCount === node.taskCount
const sameShape = prev && !prev.batchTag && node.taskCount > 1 && prev.taskCount === node.taskCount
const uniqueStep = prev ? !prev.nodes.some(item => item.taskIndex === node.taskIndex) : false
if (prev && sameShape && closeInTime && uniqueStep) {
@@ -248,7 +285,8 @@ function DelegationGroup({ group, nowMs }: { group: RootGroup; nowMs: number })
return (
<section className="grid min-w-0 gap-3">
<p className="text-[0.66rem] font-medium uppercase tracking-wider text-muted-foreground/70">
{group.delegationIndex > 0 ? t.agents.delegation(group.delegationIndex) : ''}{' '}
{group.delegationIndex > 0 ? t.agents.delegation(group.delegationIndex) : ''}
{group.batchTag ? <span className="ml-1 font-mono text-muted-foreground/60">[{group.batchTag}]</span> : null}{' '}
<span className="text-muted-foreground/50">·</span> {t.agents.workers(group.nodes.length)}
{activeWorkers > 0 ? <span className="text-primary/85"> · {t.agents.workersActive(activeWorkers)}</span> : null}
</p>
+4
View File
@@ -18,6 +18,9 @@ export interface SubagentProgress {
goal: string
/** The child's own stored session id — lets UIs open its session window. */
sessionId?: string
/** Batch (delegation) id — exact grouping key for one fan-out's workers,
* so concurrent/nested batches never merge into one group. */
delegationId?: string
model?: string
status: SubagentStatus
taskCount: number
@@ -189,6 +192,7 @@ function toProgress(payload: SubagentPayload, prev: SubagentProgress | undefined
parentId: str(payload.parent_id) || prev?.parentId || null,
goal: str(payload.goal) || prev?.goal || 'Subagent',
sessionId: str(payload.child_session_id) || prev?.sessionId,
delegationId: str(payload.delegation_id) || prev?.delegationId,
model: str(payload.model) || prev?.model,
status,
taskCount: num(payload.task_count) ?? prev?.taskCount ?? 1,
+1
View File
@@ -225,6 +225,7 @@ def _make_run_event_callback(
"task_index",
"subagent_id",
"child_session_id",
"delegation_id",
"parent_id",
"depth",
"model",
+4 -3
View File
@@ -132,11 +132,12 @@ class TestBuildChildProgressCallback:
parent._delegate_spinner = spinner
parent.tool_progress_callback = None
# task_index=0 in a batch of 3 → prefix "[1]"
# task_index=0 in a batch of 3 → prefix "[1/3]" (batch slot; a
# delegation batch tag is prepended when the id is known)
cb0 = _build_child_progress_callback(0, "test goal", parent, task_count=3)
cb0("tool.started", "web_search", "test", {})
output = buf.getvalue()
assert "[1]" in output
assert "[1/3]" in output
# task_index=2 in a batch of 3 → prefix "[3]"
buf.truncate(0)
@@ -144,7 +145,7 @@ class TestBuildChildProgressCallback:
cb2 = _build_child_progress_callback(2, "test goal", parent, task_count=3)
cb2("tool.started", "web_search", "test", {})
output = buf.getvalue()
assert "[3]" in output
assert "[3/3]" in output
+116
View File
@@ -0,0 +1,116 @@
"""Batch tag on delegation progress lines (#p1-campaign feedback, Sep 2026).
When a parent fans out N subagents and a child fans out its own M, both
batches print ``[n/N]`` completion lines to the same console. Without a
batch tag ``✓ [3/3]`` and ``✓ [3/9]`` are indistinguishable. Every progress
surface must carry the short delegation id.
"""
import types
import pytest
import tools.delegate_tool as dt
from tools.delegate_tool import _batch_prefix, _build_child_progress_callback, format_batch_tag
def test_format_batch_tag_shortens_delegation_handle():
assert format_batch_tag("deleg_6a664903") == "6a66"
assert format_batch_tag("deleg_") == ""
assert format_batch_tag(None) == ""
assert format_batch_tag("") == ""
@pytest.mark.parametrize(
"deleg, idx, count, expected",
[
("deleg_6a664903", 2, 9, "[6a66 3/9] "),
(None, 2, 9, "[3/9] "),
("deleg_6a664903", 0, 1, "[6a66] "),
(None, 0, 1, ""),
],
)
def test_batch_prefix_shapes(deleg, idx, count, expected):
assert _batch_prefix(deleg, idx, count) == expected
class _Spinner:
def __init__(self):
self.lines = []
def print_above(self, line):
self.lines.append(line)
def update_text(self, text):
self.lines.append(f"<spin>{text}")
def test_child_tree_lines_and_relayed_events_carry_batch_tag():
relayed = []
parent = types.SimpleNamespace(
_delegate_spinner=_Spinner(),
tool_progress_callback=lambda et, name=None, preview=None, args=None, **kw: relayed.append((et, kw)),
)
ref = {}
cb = _build_child_progress_callback(2, "triage cluster", parent, 9, subagent_id="sa-2", session_ref=ref)
# Stamped by delegate_task AFTER the callback is built — must be picked up lazily.
ref["delegation_id"] = "deleg_6a664903"
ref["session_id"] = "child-sess"
cb("subagent.start")
cb("tool.started", "terminal", "ls")
tree = parent._delegate_spinner.lines
assert tree[0].startswith(" [6a66 3/9] ├─ 🔀 triage cluster")
assert tree[1].startswith(" [6a66 3/9] ├─ ")
assert all(kw.get("delegation_id") == "deleg_6a664903" for _, kw in relayed)
assert all(kw.get("child_session_id") == "child-sess" for _, kw in relayed)
def test_child_tree_prefix_without_batch_id_is_unchanged():
parent = types.SimpleNamespace(_delegate_spinner=_Spinner(), tool_progress_callback=None)
cb = _build_child_progress_callback(0, "solo goal", parent, 3, session_ref={})
cb("subagent.start")
assert parent._delegate_spinner.lines[0].startswith(" [1/3] ├─ 🔀 solo goal")
def test_batch_completion_lines_are_attributable_across_two_batches(monkeypatch, tmp_path):
"""Two interleaved batches: every ✓ line names its own batch tag, and the
tag equals the delegation_id the dispatch returns."""
monkeypatch.setenv("HERMES_HOME", str(tmp_path / ".hermes"))
(tmp_path / ".hermes").mkdir()
lines = []
parent = types.SimpleNamespace(
session_id="root", model="m", tool_progress_callback=None, _delegate_spinner=None,
_safe_print=lambda line: lines.append(line),
)
monkeypatch.setattr(
dt, "_run_single_child",
lambda task_index, goal, child=None, parent_agent=None, **kw: {
"task_index": task_index, "status": "completed", "summary": "ok",
"error": None, "api_calls": 1, "duration_seconds": 1,
},
)
monkeypatch.setattr(dt, "_build_child_preserving_parent_tools",
lambda **kw: types.SimpleNamespace(tool_progress_callback=None))
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: {
"model": "m", "provider": "openrouter", "base_url": "https://x/v1",
"api_key": "k", "api_mode": "chat_completions"})
import re
for n in (3, 9):
res = dt.delegate_task(
tasks=[{"goal": f"batch of {n}: worker task number {i}"} for i in range(n)],
parent_agent=parent,
)
assert "error" not in str(res)[:20], res
headers = [re.match(r"\s*🔀 \[([0-9a-f]{4})\] delegating (\d+) tasks", l) for l in lines]
headers = [m for m in headers if m]
assert [int(m.group(2)) for m in headers] == [3, 9]
tags = [m.group(1) for m in headers]
assert len(set(tags)) == 2
done = [l for l in lines if "✓ [" in l]
assert len(done) == 12
assert sum(1 for l in done if f"✓ [{tags[0]} " in l and "/3]" in l) == 3
assert sum(1 for l in done if f"✓ [{tags[1]} " in l and "/9]" in l) == 9
+65 -12
View File
@@ -1402,6 +1402,31 @@ def _blocked_toolsets_for_role(role: str) -> List[str]:
)
def format_batch_tag(delegation_id: Optional[str]) -> str:
"""Short human tag identifying which delegation batch a line belongs to.
``deleg_6a664903`` → ``6a66``. Several batches (a parent's fan-out plus
a child's nested fan-out, or two concurrent tools) print interleaved
``[n/N]`` progress lines to the same console; without a batch tag a
``✓ [3/3]`` and a ``✓ [3/9]`` are indistinguishable. Empty string when
no id is known so callers can concatenate unconditionally.
"""
if not isinstance(delegation_id, str) or not delegation_id:
return ""
short = delegation_id.split("_", 1)[-1][:4]
return f"{short}" if short else ""
def _batch_prefix(delegation_id: Optional[str], task_index: int, task_count: int) -> str:
"""``[6a66 3/9] `` for batch children, ``[6a66] `` for a lone child,
``[3/9] `` / ``""`` when the batch id is unknown."""
tag = format_batch_tag(delegation_id)
if task_count > 1:
inner = f"{tag} {task_index + 1}/{task_count}" if tag else f"{task_index + 1}/{task_count}"
return f"[{inner}] "
return f"[{tag}] " if tag else ""
def _emit_parent_console(parent_agent, line: str) -> None:
"""Emit a human-readable progress line to the parent's console.
@@ -1454,8 +1479,14 @@ def _build_child_progress_callback(
if not spinner and not parent_cb:
return None # No display → no callback → zero behavior change
# Show 1-indexed prefix only in batch mode (multiple tasks)
prefix = f"[{task_index + 1}] " if task_count > 1 else ""
# Show 1-indexed prefix only in batch mode (multiple tasks). The batch tag
# (short delegation id) is resolved lazily from session_ref because the
# callback is built before delegate_task stamps ``_delegation_id`` on the
# child; delegate_task drops the id into the same shared ref.
def _prefix() -> str:
deleg = session_ref.get("delegation_id") if session_ref else None
return _batch_prefix(deleg, task_index, task_count)
goal_label = (goal or "").strip()
# Gateway: batch tool names, flush periodically
@@ -1484,6 +1515,8 @@ def _build_child_progress_callback(
# event lets UIs open/inspect the subagent's session directly.
if session_ref and session_ref.get("session_id"):
kw["child_session_id"] = str(session_ref["session_id"])
if session_ref and session_ref.get("delegation_id"):
kw["delegation_id"] = str(session_ref["delegation_id"])
kw["tool_count"] = _tool_count[0]
return kw
@@ -1510,7 +1543,7 @@ def _build_child_progress_callback(
(goal_label[:55] + "...") if len(goal_label) > 55 else goal_label
)
try:
spinner.print_above(f" {prefix}├─ 🔀 {short}")
spinner.print_above(f" {_prefix()}├─ 🔀 {short}")
except Exception as e:
logger.debug("Spinner print_above failed: %s", e)
_relay("subagent.start", preview=preview or goal_label or "", **kwargs)
@@ -1529,7 +1562,7 @@ def _build_child_progress_callback(
duration_seconds=kwargs.get("duration_seconds"),
)
try:
spinner.print_above(f" {prefix}├─ {_fail_line}")
spinner.print_above(f" {_prefix()}├─ {_fail_line}")
except Exception as e:
logger.debug("Spinner print_above failed: %s", e)
_relay("subagent.complete", preview=preview, **kwargs)
@@ -1563,7 +1596,7 @@ def _build_child_progress_callback(
if spinner:
short = (text[:55] + "...") if len(text) > 55 else text
try:
spinner.print_above(f' {prefix}├─ 💭 "{short}"')
spinner.print_above(f' {_prefix()}├─ 💭 "{short}"')
except Exception as e:
logger.debug("Spinner print_above failed: %s", e)
_relay("subagent.thinking", preview=text)
@@ -1583,12 +1616,12 @@ def _build_child_progress_callback(
summary_text = tool_name or preview or ""
if spinner and summary_text:
try:
spinner.print_above(f" {prefix}├─ 🔀 {summary_text}")
spinner.print_above(f" {_prefix()}├─ 🔀 {summary_text}")
except Exception as e:
logger.debug("Spinner print_above failed: %s", e)
if parent_cb:
try:
parent_cb("subagent_progress", f"{prefix}{summary_text}")
parent_cb("subagent_progress", f"{_prefix()}{summary_text}")
except Exception as e:
logger.debug("Parent callback relay failed: %s", e)
return
@@ -1610,7 +1643,7 @@ def _build_child_progress_callback(
from agent.display import get_tool_emoji
emoji = get_tool_emoji(tool_name or "")
line = f" {prefix}├─ {emoji} {tool_name}"
line = f" {_prefix()}├─ {emoji} {tool_name}"
if short:
line += f' "{short}"'
try:
@@ -1623,14 +1656,14 @@ def _build_child_progress_callback(
_batch.append(tool_name or "")
if len(_batch) >= _BATCH_SIZE:
summary = ", ".join(_batch)
_relay("subagent.progress", preview=f"🔀 {prefix}{summary}")
_relay("subagent.progress", preview=f"🔀 {_prefix()}{summary}")
_batch.clear()
def _flush():
"""Flush remaining batched tool names to gateway on completion."""
if parent_cb and _batch:
summary = ", ".join(_batch)
_relay("subagent.progress", preview=f"🔀 {prefix}{summary}")
_relay("subagent.progress", preview=f"🔀 {_prefix()}{summary}")
_batch.clear()
_callback._flush = _flush
@@ -2138,6 +2171,9 @@ def _build_child_agent(
# Now the child exists, its session id can ride on every relayed event
# (including the spawn_requested below — first emit happens after this).
child_session_ref["session_id"] = getattr(child, "session_id", "") or ""
# Same shared ref receives the batch id once delegate_task stamps it, so
# the display prefix and relayed events can tag which batch this is.
child._progress_identity_ref = child_session_ref
# Set delegation depth so children can't spawn grandchildren
child._delegate_depth = child_depth
# Stash the post-degrade role for introspection (leaf if the
@@ -4091,6 +4127,18 @@ def delegate_task(
live_deleg_id, live_writers, live_paths = create_live_transcripts(
task_list, context, model=creds.get("model"), provider=creds.get("provider")
)
# Announce the batch tag once so the later ``[tag n/N]`` completion lines
# (and any nested batch's lines interleaving with them) are attributable.
if n_tasks > 1 and live_deleg_id:
_hdr = f"🔀 [{format_batch_tag(live_deleg_id)}] delegating {n_tasks} tasks"
_hdr_spinner = getattr(parent_agent, "_delegate_spinner", None)
if _hdr_spinner:
try:
_hdr_spinner.print_above(f" {_hdr}")
except Exception:
_emit_parent_console(parent_agent, f" {_hdr}")
else:
_emit_parent_console(parent_agent, f" {_hdr}")
# Capture the ORIGINATING session's wake target BEFORE any child agent is
# constructed: _build_child_agent() -> AIAgent() -> agent_init calls
@@ -4179,6 +4227,9 @@ def delegate_task(
# attribution (child-started background processes report under it).
if live_deleg_id:
setattr(child, "_delegation_id", live_deleg_id)
_ident_ref = getattr(child, "_progress_identity_ref", None)
if isinstance(_ident_ref, dict):
_ident_ref["delegation_id"] = live_deleg_id
children.append((i, t, child))
def _execute_and_aggregate(*, honor_parent_interrupt: bool = True) -> dict:
@@ -4314,7 +4365,9 @@ def delegate_task(
status = entry.get("status", "?")
icon = "✓" if status == "completed" else "✗"
remaining = n_tasks - completed_count
completion_line = f"{icon} [{idx+1}/{n_tasks}] {label} ({dur}s)"
_tag = format_batch_tag(live_deleg_id)
_slot = f"{_tag} {idx+1}/{n_tasks}" if _tag else f"{idx+1}/{n_tasks}"
completion_line = f"{icon} [{_slot}] {label} ({dur}s)"
# Failed/errored/timed-out children: say WHY on the
# same line, cleaned to one short human-readable
# fragment — a bare ✗ reads as "silently dropped".
@@ -4336,7 +4389,7 @@ def delegate_task(
if spinner_ref and remaining > 0:
try:
spinner_ref.update_text(
f"🔀 {remaining} task{'s' if remaining != 1 else ''} remaining"
f"🔀 {'[' + _tag + '] ' if _tag else ''}{remaining} task{'s' if remaining != 1 else ''} remaining"
)
except Exception as e:
logger.debug("Spinner update_text failed: %s", e)
+2
View File
@@ -8194,6 +8194,8 @@ def _on_tool_progress(
payload["parent_id"] = str(_kwargs["parent_id"])
if _kwargs.get("child_session_id"):
payload["child_session_id"] = str(_kwargs["child_session_id"])
if _kwargs.get("delegation_id"):
payload["delegation_id"] = str(_kwargs["delegation_id"])
if _kwargs.get("depth") is not None:
payload["depth"] = int(_kwargs["depth"])
if _kwargs.get("model"):
+2
View File
@@ -1040,6 +1040,7 @@ class TurnController {
}
const base: SubagentProgress = existing ?? {
delegationId: p.delegation_id,
depth: p.depth ?? 0,
goal: p.goal,
id,
@@ -1071,6 +1072,7 @@ class TurnController {
...base,
apiCalls: p.api_calls ?? base.apiCalls,
costUsd: p.cost_usd ?? base.costUsd,
delegationId: p.delegation_id ?? base.delegationId,
depth: p.depth ?? base.depth,
filesRead: p.files_read ?? base.filesRead,
filesWritten: p.files_written ?? base.filesWritten,
+8 -1
View File
@@ -332,7 +332,14 @@ function SubagentAccordion({
? 'warn'
: 'dim'
const prefix = item.taskCount > 1 ? `[${item.index + 1}/${item.taskCount}] ` : ''
// `[6a66 3/9]` when the gateway tags the batch; `[3/9]` on older gateways.
const batchTag = item.delegationId?.split('_').at(-1)?.slice(0, 4)
const prefix =
item.taskCount > 1
? `[${batchTag ? `${batchTag} ` : ''}${item.index + 1}/${item.taskCount}] `
: batchTag
? `[${batchTag}] `
: ''
const goalLabel = item.goal || `Subagent ${item.index + 1}`
const title = `${prefix}${open ? goalLabel : compactPreview(goalLabel, 60)}`
const summary = compactPreview((item.summary || '').replace(/\s+/g, ' ').trim(), 72)
+3
View File
@@ -543,6 +543,9 @@ export interface RollbackRestoreResponse {
export interface SubagentEventPayload {
api_calls?: number
cost_usd?: number
/** Batch (delegation) id this subagent belongs to — distinguishes
* interleaved `[n/N]` progress from concurrent or nested fan-outs. */
delegation_id?: string
depth?: number
duration_seconds?: number
files_read?: string[]
+3
View File
@@ -25,6 +25,9 @@ export type SubagentStatus = 'completed' | 'error' | 'failed' | 'interrupted' |
export interface SubagentProgress {
apiCalls?: number
costUsd?: number
/** Batch (delegation) id — tags `[n/N]` rows so concurrent/nested fan-outs
* are distinguishable. Absent on older gateways. */
delegationId?: string
depth: number
durationSeconds?: number
filesRead?: string[]
@@ -477,8 +477,9 @@ When the agent delegates work to background subagents, the stream also carries
`subagent.start` and `subagent.complete` lifecycle events, so clients can
observe delegation outcomes — including timeouts and failures — instead of the
run going silent while a child works. The `subagent.complete` payload carries
the child's status, summary, duration, token/cost figures, and a
`child_session_id` for correlation; free-text fields pass forced secret
the child's status, summary, duration, token/cost figures, a
`child_session_id` for correlation, and the `delegation_id` of the batch it
belongs to (so concurrent or nested fan-outs stay distinguishable); free-text fields pass forced secret
redaction before leaving the process. Per-tool child events
(`subagent.tool`, progress ticks) are intentionally **not** forwarded — they
are high-volume UI noise; use the per-child live transcript files for