feat(delegation): live-viewable subagent transcripts — tail your subagents while they work (#67479)

* feat(delegation): live-viewable subagent transcripts for delegate_task

Each child now streams an append-only, human-readable log to
<hermes_home>/cache/delegation/live/<delegation_id>/task-<n>.log while it
runs, and the dispatch return includes the paths so the caller can tail
them immediately instead of waiting blind for the consolidated summary.

- New tools/delegation_live_log.py: LiveTranscriptWriter (per-event append
  + flush, one-line rendering with truncation, never raises into the agent
  loop), wrap_progress_callback (tees the child's existing
  tool_progress_callback events into the log, preserves the _flush
  contract), dispatch-time creation with pre-headered files so tail -f
  attaches immediately, manifest.json (goals/task count/per-task status),
  and 7-day retention pruning on new dispatches.
- delegate_task: wraps each child's progress callback with the writer;
  sync results and background dispatch responses gain live_transcripts
  (+ hint field on dispatch); per-task result entries carry
  live_transcript; transcripts finalized with exit-reason markers.
- async_delegation: dispatch_async_delegation_batch accepts an optional
  delegation_id so the live/ dir name matches the returned handle; the
  completion event carries live_transcripts.
- process_registry: consolidated batch-completion block references each
  task's live transcript path.
- Tool schema description documents the live_transcripts return surface;
  docs gain a 'Live Transcripts' section with a tail -f example.

Placement under cache/delegation means the logs are mounted read-only
into remote terminal backends for free. Side-channel only: zero changes
to message content, so prompt caching is unaffected. Transcript-OUT only
— no overlap with the subagent control surfaces of PR #66046.

* fix(delegation): label the kickoff transcript line as user — it is the child's one user message
This commit is contained in:
Teknium 2026-07-19 10:29:14 -07:00 committed by GitHub
parent 0385e15544
commit 299e409f15
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 963 additions and 2 deletions

View file

@ -0,0 +1,483 @@
"""Tests for tools/delegation_live_log.py — live subagent transcripts.
Covers:
- writer event rendering + truncation + append/flush semantics
- failure-swallowing when the target dir is unwritable
- the tool_progress_callback observe() demux (assistant/tool events in order)
- dispatch-time creation: paths pre-created with a header, manifest written
- retention pruning of stale live dirs
- delegate_task return-shape: live_transcripts in sync + background dispatch
"""
import json
import os
import threading
import time
from pathlib import Path
from unittest.mock import MagicMock
import pytest
from tools import delegation_live_log as dll
from tools.delegation_live_log import (
LiveTranscriptWriter,
create_live_transcripts,
live_transcript_root,
prune_stale_live_dirs,
update_manifest_statuses,
wrap_progress_callback,
)
# ---------------------------------------------------------------------------
# Writer unit tests
# ---------------------------------------------------------------------------
def test_writer_precreates_file_with_header():
w = LiveTranscriptWriter("deleg_test1", 0, "do the thing", context="some ctx")
assert w.path is not None and w.path.exists()
text = w.path.read_text(encoding="utf-8")
assert "Hermes subagent live transcript" in text
assert "delegation: deleg_test1" in text
assert "goal: do the thing" in text
assert "kickoff" in text
assert "some ctx" in text
# Lives under the hermes cache/delegation/live root, named task-<n>.log
assert w.path.name == "task-0.log"
assert w.path.parent.name == "deleg_test1"
assert w.path.parent.parent == live_transcript_root()
def test_writer_event_lines_append_in_order_and_flush_immediately():
w = LiveTranscriptWriter("deleg_order", 1, "goal")
w.assistant_text("I'll inspect the repo first.")
w.tool_start("terminal", "ls -la /tmp")
w.tool_result("terminal", result="file1\nfile2", duration=1.234, is_error=False)
w.thinking("hmm, next step")
# No close() needed: every event is flushed on write.
lines = w.path.read_text(encoding="utf-8").splitlines()
body = [ln for ln in lines if "|" in ln and not ln.startswith("=")]
joined = "\n".join(body)
assert "assistant" in joined and "I'll inspect the repo first." in joined
assert "-> terminal(ls -la /tmp)" in joined
assert "terminal ok 1.2s: file1 file2" in joined
assert "hmm, next step" in joined
# Ordering: assistant before tool before result before think
idx = {k: joined.index(k) for k in ("I'll inspect", "-> terminal", "terminal ok", "hmm,")}
assert idx["I'll inspect"] < idx["-> terminal"] < idx["terminal ok"] < idx["hmm,"]
def test_writer_truncates_long_text_with_elision_note():
w = LiveTranscriptWriter("deleg_trunc", 0, "g")
w.assistant_text("x" * 5000)
w.tool_result("web_search", result="y" * 5000)
text = w.path.read_text(encoding="utf-8")
assert "…(+" in text # elision marker present
# No line carries the full 5000 chars
assert all(len(ln) < 1200 for ln in text.splitlines())
def test_writer_collapses_newlines_to_single_line_events():
w = LiveTranscriptWriter("deleg_nl", 0, "g")
before = len(w.path.read_text(encoding="utf-8").splitlines())
w.assistant_text("line1\nline2\n\nline3")
after = w.path.read_text(encoding="utf-8").splitlines()
assert len(after) == before + 1
assert "line1 line2 line3" in after[-1]
def test_writer_swallows_failures_when_dir_unwritable(tmp_path):
# Point the writer at a root that is actually a FILE — mkdir will fail.
bogus_root = tmp_path / "not-a-dir"
bogus_root.write_text("occupied")
w = LiveTranscriptWriter("deleg_fail", 0, "g", root=bogus_root)
assert w.path is None
# All writes must be silent no-ops.
w.assistant_text("hello")
w.tool_start("terminal", "ls")
w.marker("done")
w.observe("tool.completed", "terminal", result="x")
w.finalize({"status": "completed"})
def test_writer_disables_itself_after_write_failure():
w = LiveTranscriptWriter("deleg_disable", 0, "g")
# Delete the parent dir out from under it and make writing impossible by
# replacing the path with a directory.
p = w.path
p.unlink()
p.mkdir()
w.assistant_text("should not raise")
assert w._ok is False
w.assistant_text("still silent") # no raise on subsequent calls
def test_stream_deltas_buffer_and_flush_as_one_line():
w = LiveTranscriptWriter("deleg_stream", 0, "g")
w.add_stream_delta("Hello ")
w.add_stream_delta("world, ")
w.add_stream_delta("streaming.")
# Not yet flushed
assert "Hello world" not in w.path.read_text(encoding="utf-8")
w.flush_stream()
text = w.path.read_text(encoding="utf-8")
assert "Hello world, streaming." in text
# tool_start also flushes pending stream text first
w.add_stream_delta("more text")
w.tool_start("read_file", "foo.py")
text = w.path.read_text(encoding="utf-8")
assert text.index("more text") < text.index("-> read_file")
# ---------------------------------------------------------------------------
# observe() demux — the tool_progress_callback seam
# ---------------------------------------------------------------------------
def test_observe_maps_child_callback_events_to_lines():
w = LiveTranscriptWriter("deleg_observe", 0, "g")
w.observe("subagent.start", preview="kick off the goal")
w.observe("_thinking", "first line of thinking")
w.observe("reasoning.available", "_thinking", "deep reasoning text", None)
w.observe("tool.started", "terminal", "ls /tmp", {"command": "ls /tmp"})
w.observe("tool.completed", "terminal", None, None,
duration=0.5, is_error=False, result="ok output")
w.observe("subagent.text", preview="final reply ")
w.observe("subagent.text", preview="streamed in parts")
w.observe("subagent.complete", preview="short", status="completed",
duration_seconds=3.2, summary="did the thing")
text = w.path.read_text(encoding="utf-8")
assert "kick off the goal" in text
assert "first line of thinking" in text
assert "deep reasoning text" in text
assert "-> terminal(ls /tmp)" in text
assert "terminal ok 0.5s: ok output" in text
assert "final reply streamed in parts" in text
assert "status=completed" in text
assert "did the thing" in text
def test_observe_marks_tool_errors():
w = LiveTranscriptWriter("deleg_err", 0, "g")
w.observe("tool.completed", "web_search", None, None,
is_error=True, result="Error: boom")
assert "web_search ERROR" in w.path.read_text(encoding="utf-8")
def test_finalize_records_budget_exhaustion_and_errors():
w = LiveTranscriptWriter("deleg_final", 0, "g")
w.finalize({"status": "failed", "exit_reason": "max_iterations",
"error": "Subagent did not produce a response."})
text = w.path.read_text(encoding="utf-8")
assert "end status=failed" in text
assert "exit_reason=max_iterations" in text
assert "iteration budget exhausted" in text
assert "did not produce a response" in text
def test_wrap_progress_callback_tees_and_preserves_inner():
w = LiveTranscriptWriter("deleg_wrap", 0, "g")
seen = []
def inner(event_type, tool_name=None, preview=None, args=None, **kw):
seen.append((event_type, tool_name))
inner_flushed = []
inner._flush = lambda: inner_flushed.append(True)
cb = wrap_progress_callback(inner, w)
cb("tool.started", "terminal", "echo hi", None)
cb("_thinking", "pondering")
assert seen == [("tool.started", "terminal"), ("_thinking", "pondering")]
text = w.path.read_text(encoding="utf-8")
assert "-> terminal(echo hi)" in text and "pondering" in text
# _flush contract preserved
cb._flush()
assert inner_flushed == [True]
def test_wrap_progress_callback_with_no_inner_still_records():
w = LiveTranscriptWriter("deleg_noinner", 0, "g")
cb = wrap_progress_callback(None, w)
cb("tool.started", "read_file", "a.py", None)
cb._flush() # must not raise
assert "-> read_file(a.py)" in w.path.read_text(encoding="utf-8")
def test_wrap_progress_callback_writer_failure_does_not_block_inner():
w = LiveTranscriptWriter("deleg_wfail", 0, "g")
w.observe = MagicMock(side_effect=RuntimeError("disk on fire"))
seen = []
cb = wrap_progress_callback(lambda *a, **k: seen.append(a), w)
cb("tool.started", "terminal", "x", None) # must not raise
assert len(seen) == 1
# ---------------------------------------------------------------------------
# Dispatch-time creation + manifest + retention
# ---------------------------------------------------------------------------
def test_create_live_transcripts_precreates_paths_and_manifest():
tasks = [{"goal": "task A"}, {"goal": "task B", "context": "ctx B"}]
deleg_id, writers, paths = create_live_transcripts(tasks, context="shared ctx")
assert deleg_id and deleg_id.startswith("deleg_")
assert len(writers) == 2 and all(w is not None for w in writers)
assert len(paths) == 2
for i, p in enumerate(paths):
assert os.path.isabs(p)
assert p.endswith(f"task-{i}.log")
assert Path(p).exists() # tail -f works immediately
manifest = json.loads(
(live_transcript_root() / deleg_id / "manifest.json").read_text()
)
assert manifest["task_count"] == 2
assert manifest["tasks"][0]["goal"] == "task A"
assert manifest["tasks"][0]["status"] == "running"
assert manifest["tasks"][1]["log"] == paths[1]
# Per-task context beats shared context in the kickoff line.
assert "ctx B" in Path(paths[1]).read_text(encoding="utf-8")
def test_update_manifest_statuses():
tasks = [{"goal": "a"}, {"goal": "b"}]
deleg_id, _writers, _paths = create_live_transcripts(tasks)
update_manifest_statuses(deleg_id, [
{"task_index": 0, "status": "completed", "exit_reason": "completed"},
{"task_index": 1, "status": "error"},
])
manifest = json.loads(
(live_transcript_root() / deleg_id / "manifest.json").read_text()
)
assert manifest["tasks"][0]["status"] == "completed"
assert manifest["tasks"][1]["status"] == "error"
assert "completed" in manifest
def test_update_manifest_statuses_none_id_is_noop():
update_manifest_statuses(None, [{"task_index": 0, "status": "completed"}])
def test_prune_stale_live_dirs():
root = live_transcript_root()
old_dir = root / "deleg_old00001"
new_dir = root / "deleg_new00001"
old_dir.mkdir(parents=True)
new_dir.mkdir(parents=True)
(old_dir / "task-0.log").write_text("old")
(new_dir / "task-0.log").write_text("new")
stale = time.time() - 8 * 86400
os.utime(old_dir, (stale, stale))
removed = prune_stale_live_dirs(max_age_days=7)
assert removed == 1
assert not old_dir.exists()
assert new_dir.exists()
def test_create_live_transcripts_survives_root_failure(monkeypatch):
monkeypatch.setattr(
dll, "live_transcript_root",
lambda: (_ for _ in ()).throw(RuntimeError("no home")),
)
deleg_id, writers, paths = create_live_transcripts([{"goal": "g"}])
assert deleg_id is None
assert writers == [None]
assert paths == []
# ---------------------------------------------------------------------------
# delegate_task return-shape integration
# ---------------------------------------------------------------------------
def _make_parent():
parent = MagicMock()
parent._delegate_depth = 0
parent.session_id = "sess-live"
parent._interrupt_requested = False
parent._active_children = []
parent._active_children_lock = None
return parent
_CREDS = {
"model": "m", "provider": None, "base_url": None, "api_key": None,
"api_mode": None, "command": None, "args": None,
}
def _fake_run(task_index, goal, child=None, parent_agent=None, **kw):
return {
"task_index": task_index, "status": "completed",
"summary": f"done: {goal}", "api_calls": 1,
"duration_seconds": 0.1, "model": "m", "exit_reason": "completed",
}
def test_delegate_task_sync_result_includes_live_transcripts(monkeypatch):
import tools.delegate_tool as dt
parent = _make_parent()
fake_child = MagicMock()
fake_child._delegate_role = "leaf"
fake_child.tool_progress_callback = None
monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
monkeypatch.setattr(dt, "_run_single_child", _fake_run)
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: _CREDS)
out = json.loads(dt.delegate_task(goal="sync goal", parent_agent=parent))
assert "live_transcripts" in out
assert len(out["live_transcripts"]) == 1
p = Path(out["live_transcripts"][0])
assert p.exists()
assert "sync goal" in p.read_text(encoding="utf-8")
# Per-task entries carry their own path + a terminal marker was written.
assert out["results"][0]["live_transcript"] == str(p)
assert "end status=completed" in p.read_text(encoding="utf-8")
def test_delegate_task_background_dispatch_includes_live_transcripts(monkeypatch):
import tools.delegate_tool as dt
from tools import async_delegation as ad
from tools.process_registry import process_registry
parent = _make_parent()
fake_child = MagicMock()
fake_child._delegate_role = "leaf"
fake_child._subagent_id = "s1"
fake_child.tool_progress_callback = None
gate = threading.Event()
def slow_child(task_index, goal, child=None, parent_agent=None, **kw):
gate.wait(timeout=60)
return _fake_run(task_index, goal)
monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
monkeypatch.setattr(dt, "_run_single_child", slow_child)
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: _CREDS)
out = json.loads(dt.delegate_task(
goal="bg goal", background=True, parent_agent=parent,
))
try:
assert out["status"] == "dispatched"
assert "live_transcripts" in out
assert len(out["live_transcripts"]) == 1
live = Path(out["live_transcripts"][0])
# Pre-created at dispatch time — tail -f attaches immediately,
# while the child is still running behind the gate.
assert live.exists()
assert "bg goal" in live.read_text(encoding="utf-8")
assert "live_transcripts_hint" in out
# The dir name matches the returned delegation handle.
assert live.parent.name == out["delegation_id"]
finally:
gate.set()
# Drain the completion so it can't leak into other tests.
deadline = time.time() + 30
evt = None
while time.time() < deadline:
try:
evt = process_registry.completion_queue.get(timeout=0.5)
break
except Exception:
continue
ad._reset_for_tests()
assert evt is not None
# The completion event carries the same paths for the consolidated block.
assert evt.get("live_transcripts") == out["live_transcripts"]
assert evt["results"][0]["live_transcript"] == out["live_transcripts"][0]
def test_batch_dispatch_creates_one_log_per_task(monkeypatch):
import tools.delegate_tool as dt
parent = _make_parent()
def make_child(**kw):
c = MagicMock()
c._delegate_role = "leaf"
c.tool_progress_callback = None
return c
monkeypatch.setattr(dt, "_build_child_agent", make_child)
monkeypatch.setattr(dt, "_run_single_child", _fake_run)
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: _CREDS)
out = json.loads(dt.delegate_task(
tasks=[{"goal": "alpha"}, {"goal": "beta"}], parent_agent=parent,
))
assert len(out["live_transcripts"]) == 2
names = [Path(p).name for p in out["live_transcripts"]]
assert names == ["task-0.log", "task-1.log"]
# Both under the same delegation dir
parents = {Path(p).parent for p in out["live_transcripts"]}
assert len(parents) == 1
for p, goal in zip(out["live_transcripts"], ("alpha", "beta")):
assert goal in Path(p).read_text(encoding="utf-8")
def test_child_progress_events_land_in_live_log(monkeypatch):
"""Events fired through the child's (wrapped) tool_progress_callback land
in the transcript file in order the seam the real agent loop drives."""
import tools.delegate_tool as dt
parent = _make_parent()
built = []
def make_child(**kw):
c = MagicMock()
c._delegate_role = "leaf"
c.tool_progress_callback = None
built.append(c)
return c
def run_child(task_index, goal, child=None, parent_agent=None, **kw):
# Simulate what agent/tool_executor.py + conversation_loop.py emit.
cb = child.tool_progress_callback
cb("_thinking", "planning the work")
cb("tool.started", "terminal", "echo hi", {"command": "echo hi"})
cb("tool.completed", "terminal", None, None,
duration=0.2, is_error=False, result="hi")
return _fake_run(task_index, goal)
monkeypatch.setattr(dt, "_build_child_agent", make_child)
monkeypatch.setattr(dt, "_run_single_child", run_child)
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: _CREDS)
out = json.loads(dt.delegate_task(goal="observable goal", parent_agent=parent))
text = Path(out["live_transcripts"][0]).read_text(encoding="utf-8")
assert "planning the work" in text
assert "-> terminal(echo hi)" in text
assert "terminal ok 0.2s: hi" in text
assert text.index("planning") < text.index("-> terminal") < text.index("terminal ok")
def test_delegate_task_proceeds_when_transcripts_unavailable(monkeypatch):
"""Live-log failure must never break delegation itself."""
import tools.delegate_tool as dt
from tools import delegation_live_log as _dll
parent = _make_parent()
fake_child = MagicMock()
fake_child._delegate_role = "leaf"
fake_child.tool_progress_callback = None
monkeypatch.setattr(dt, "_build_child_agent", lambda **kw: fake_child)
monkeypatch.setattr(dt, "_run_single_child", _fake_run)
monkeypatch.setattr(dt, "_resolve_delegation_credentials", lambda *a, **k: _CREDS)
monkeypatch.setattr(
_dll, "live_transcript_root",
lambda: (_ for _ in ()).throw(RuntimeError("nope")),
)
out = json.loads(dt.delegate_task(goal="resilient", parent_agent=parent))
assert out["results"][0]["status"] == "completed"
assert "live_transcripts" not in out
if __name__ == "__main__":
import sys
sys.exit(pytest.main([__file__, "-v"]))

View file

@ -655,6 +655,7 @@ def dispatch_async_delegation_batch(
origin_ui_session_id: str = "",
interrupt_fn: Optional[Callable[[], None]] = None,
max_async_children: int = _DEFAULT_MAX_ASYNC_CHILDREN,
delegation_id: Optional[str] = None,
) -> Dict[str, Any]:
"""Dispatch a WHOLE fan-out batch as ONE background unit.
@ -676,7 +677,7 @@ def dispatch_async_delegation_batch(
``{"status": "rejected", "error": ...}`` when the async pool is at
capacity.
"""
delegation_id = _new_delegation_id()
delegation_id = delegation_id or _new_delegation_id()
dispatched_at = time.time()
n = len(goals)
# A combined goal label for status listings / the completion header.
@ -805,6 +806,10 @@ def _finalize_batch(
# The full per-task results list — the formatter renders a
# consolidated multi-task block from this.
"results": combined.get("results") or [],
# Per-task live transcript log paths (cache/delegation/live/...).
# They persist after completion and double as the full-fidelity
# operational record of each child's run.
"live_transcripts": combined.get("live_transcripts"),
"error": combined.get("error"),
"total_duration_seconds": combined.get("total_duration_seconds"),
"dispatched_at": dispatched_at,

View file

@ -2555,6 +2555,21 @@ def delegate_task(
# Track goal labels for progress display (truncated for readability)
task_labels = [t["goal"][:40] for t in task_list]
# Live transcripts: one pre-headered append-only log per task under
# cache/delegation/live/<delegation_id>/task-<n>.log so the caller can
# tail each child's operations while it runs (side-channel only — zero
# effect on message content or prompt caching). Best-effort: on failure
# live_paths is empty and delegation proceeds exactly as before.
from tools.delegation_live_log import (
create_live_transcripts,
update_manifest_statuses,
wrap_progress_callback,
)
live_deleg_id, live_writers, live_paths = create_live_transcripts(
task_list, context
)
# Save parent tool names BEFORE any child construction mutates the global.
# _build_child_agent() calls AIAgent() which calls get_tool_definitions(),
# which overwrites model_tools._last_resolved_tool_names with child's toolset.
@ -2594,6 +2609,17 @@ def delegate_task(
)
# Override with correct parent tool names (before child construction mutated global)
child._delegate_saved_tool_names = _parent_tool_names
# Tee the child's progress events into its live transcript log.
# wrap_progress_callback preserves the inner callback contract
# (including the _flush attribute) and never lets writer failures
# reach the agent loop. When no parent display exists the inner
# callback is None and the wrapper still records events.
_writer = live_writers[i] if i < len(live_writers) else None
if _writer is not None:
child.tool_progress_callback = wrap_progress_callback(
getattr(child, "tool_progress_callback", None), _writer
)
child._live_transcript_path = str(_writer.path)
children.append((i, t, child))
finally:
# Authoritative restore: reset global to parent's tool names after all children built
@ -2839,10 +2865,33 @@ def delegate_task(
total_duration = round(time.monotonic() - overall_start, 2)
return {
# Close out the live transcripts: terminal marker per task + manifest
# status update. The files are retained (retention pruning happens on
# future dispatches) — they double as the full-fidelity operational
# record alongside the summary spill files.
for entry in results:
_idx = entry.get("task_index", -1)
_w = (
live_writers[_idx]
if isinstance(_idx, int) and 0 <= _idx < len(live_writers)
else None
)
if _w is not None:
try:
_w.finalize(entry)
except Exception:
logger.debug("Live transcript finalize failed", exc_info=True)
if _idx < len(live_paths):
entry["live_transcript"] = live_paths[_idx]
update_manifest_statuses(live_deleg_id, results)
combined: Dict[str, Any] = {
"results": results,
"total_duration_seconds": total_duration,
}
if live_paths:
combined["live_transcripts"] = list(live_paths)
return combined
# ----- Background dispatch: run the WHOLE batch as one async unit -----
# When background is true, the entire fan-out runs on the daemon executor
@ -2964,6 +3013,9 @@ def delegate_task(
runner=_batch_runner,
interrupt_fn=_batch_interrupt,
max_async_children=_get_max_async_children(),
# Reuse the live-transcript directory's id (when created) so the
# returned delegation_id matches cache/delegation/live/<id>/.
delegation_id=live_deleg_id,
)
if dispatch.get("status") == "dispatched":
@ -2988,6 +3040,14 @@ def delegate_task(
"goals": _goals,
"note": note,
}
if live_paths:
payload["live_transcripts"] = list(live_paths)
payload["live_transcripts_hint"] = (
"Each subagent streams a human-readable transcript of its "
"operations to the file listed above (append-only, one per "
"task). Read or `tail -f` these paths at any time to watch "
"a child work while it runs."
)
return json.dumps(payload, ensure_ascii=False)
# Pool at capacity / schedule failure — children are still attached
@ -3330,6 +3390,13 @@ def _build_top_level_description() -> str:
"batch returns one handle, runs N subagents concurrently, and delivers "
"one consolidated result after ALL of them finish. Do NOT wait or poll; "
"just continue with other work after dispatching.\n\n"
"LIVE TRANSCRIPTS: the dispatch response includes 'live_transcripts'"
"one append-only human-readable log file per task (under "
"cache/delegation/live/<delegation_id>/). Each child streams its "
"assistant text, tool calls, and tool results there while it runs. "
"Read (or `tail -f` in a terminal) those paths any time you or the "
"user want to see what a subagent is actually doing instead of "
"waiting for the final summary.\n\n"
"WHEN TO USE delegate_task:\n"
"- Reasoning-heavy subtasks (debugging, code review, research synthesis)\n"
"- Tasks that would flood your context with intermediate data\n"

View file

@ -0,0 +1,385 @@
"""Live, tail-able transcripts for delegated subagents.
Every ``delegate_task`` dispatch creates one append-only, human-readable log
per child under::
<hermes_home>/cache/delegation/live/<delegation_id>/task-<n>.log
The files are pre-created with a header at dispatch time (so ``tail -f``
attaches immediately) and then stream one line per child event: assistant
text, thinking, tool calls, tool results, and lifecycle markers. The paths
are returned from ``delegate_task`` so the parent agent (or the user) can
watch a child work instead of waiting blind for the consolidated summary.
Placement under ``cache/delegation`` is deliberate: that directory is
mounted read-only into remote terminal backends (Docker/Modal/SSH) via
``credential_files._CACHE_DIRS``, so the logs are readable from any backend.
Design constraints:
* **Never raise into the agent loop.** Every write is wrapped; the first
failure disables the writer and degrades to a debug log.
* **Survive child crashes.** Files are opened in append mode per write
no long-lived handle to lose, every event is flushed when written.
* **Side-channel only.** Nothing here touches message content, so prompt
caching is unaffected.
* **No config knobs.** Retention is a module constant (7 days), pruned
opportunistically on each new dispatch.
"""
from __future__ import annotations
import json
import logging
import shutil
import threading
import time
import uuid
from pathlib import Path
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
# Live transcript directories older than this are pruned on new dispatches.
LIVE_RETENTION_DAYS = 7
# Per-line truncation budgets (chars). The .log is a compact operational
# view, not the full-fidelity record — the child's SessionDB transcript and
# the summary spill files carry complete text.
_ASSISTANT_MAX = 600
_THINKING_MAX = 300
_ARGS_MAX = 220
_RESULT_MAX = 400
_KICKOFF_MAX = 500
# Stream deltas are buffered and flushed as one assistant line when another
# event type arrives (or on completion). Cap the buffer so a huge streamed
# reply can't hold memory hostage.
_STREAM_BUFFER_FLUSH_CHARS = 4000
def live_transcript_root() -> Path:
"""Root directory for live transcripts (profile-safe, never ~/.hermes)."""
from hermes_constants import get_hermes_dir
return get_hermes_dir("cache/delegation", "delegation_cache") / "live"
def new_live_delegation_id() -> str:
"""Same shape as async_delegation's ids so the dir name matches the handle."""
return f"deleg_{uuid.uuid4().hex[:8]}"
def _one_line(text: Any, limit: int) -> str:
"""Collapse to a single line and truncate with an elided-chars note."""
s = str(text or "")
s = " ".join(s.split()) # collapse newlines/runs of whitespace
if len(s) > limit:
omitted = len(s) - limit
s = s[:limit] + f" …(+{omitted} chars)"
return s
class LiveTranscriptWriter:
"""Append-only human-readable event log for ONE subagent task.
All methods are best-effort: the first write failure flips ``_ok`` off
and subsequent calls become no-ops (debug-logged). Never raises.
"""
def __init__(self, delegation_id: str, task_index: int, goal: str,
context: Optional[str] = None, root: Optional[Path] = None):
self.delegation_id = delegation_id
self.task_index = task_index
self._ok = True
self._lock = threading.Lock()
self._stream_buf: List[str] = []
self._stream_len = 0
try:
base = (root if root is not None else live_transcript_root())
d = base / delegation_id
d.mkdir(parents=True, exist_ok=True)
self.path: Optional[Path] = d / f"task-{task_index}.log"
header = [
"=== Hermes subagent live transcript ===",
f"delegation: {delegation_id} task: {task_index}",
f"goal: {_one_line(goal, _KICKOFF_MAX)}",
f"started: {time.strftime('%Y-%m-%d %H:%M:%S')}",
"(append-only; streams while the subagent runs — tail -f me)",
"=" * 40,
]
self.path.write_text("\n".join(header) + "\n", encoding="utf-8")
self.event("user", "kickoff: " + _one_line(goal, _KICKOFF_MAX)
+ (f" | context: {_one_line(context, _KICKOFF_MAX)}" if context else ""))
except Exception as exc:
logger.debug("Live transcript init failed (%s task %s): %s",
delegation_id, task_index, exc)
self._ok = False
self.path = None
# ── low-level ────────────────────────────────────────────────────────
def event(self, role: str, text: str) -> None:
"""Append one ``HH:MM:SS role ⟩ text`` line. Flushed per event."""
if not self._ok or self.path is None:
return
line = f"{time.strftime('%H:%M:%S')} {role:<9}| {text}\n"
try:
with self._lock:
# Append mode per write: no held handle, survives child crash,
# and the close() acts as the flush.
with open(self.path, "a", encoding="utf-8") as fh:
fh.write(line)
except Exception as exc:
self._ok = False
logger.debug("Live transcript write failed (%s): %s", self.path, exc)
# ── typed helpers ────────────────────────────────────────────────────
def assistant_text(self, text: str) -> None:
t = _one_line(text, _ASSISTANT_MAX)
if t:
self.event("assistant", t)
def thinking(self, text: str) -> None:
t = _one_line(text, _THINKING_MAX)
if t:
self.event("think", t)
def tool_start(self, name: str, args_preview: Any = None) -> None:
self.flush_stream()
args = _one_line(args_preview, _ARGS_MAX)
self.event("tool", f"-> {name or '?'}({args})")
def tool_result(self, name: str, result: Any = None,
duration: Any = None, is_error: bool = False) -> None:
status = "ERROR" if is_error else "ok"
dur = ""
try:
if duration is not None:
dur = f" {float(duration):.1f}s"
except (TypeError, ValueError):
pass
self.event("result", f"{name or '?'} {status}{dur}: "
f"{_one_line(result, _RESULT_MAX)}")
def marker(self, text: str) -> None:
"""Lifecycle marker: start / final / error / interrupt / budget."""
self.flush_stream()
self.event("final", _one_line(text, _ASSISTANT_MAX))
# ── streamed reply buffering ─────────────────────────────────────────
def add_stream_delta(self, delta: str) -> None:
"""Buffer streamed assistant reply text; flushed as one line."""
if not delta or not self._ok:
return
self._stream_buf.append(delta)
self._stream_len += len(delta)
if self._stream_len >= _STREAM_BUFFER_FLUSH_CHARS:
self.flush_stream()
def flush_stream(self) -> None:
if not self._stream_buf:
return
text = "".join(self._stream_buf)
self._stream_buf = []
self._stream_len = 0
self.assistant_text(text)
# ── event demux (the tool_progress_callback surface) ─────────────────
def observe(self, event_type: Any, tool_name: Any = None,
preview: Any = None, args: Any = None, **kwargs: Any) -> None:
"""Map a child tool_progress_callback event onto transcript lines.
Mirrors the shapes emitted by agent/tool_executor.py,
agent/conversation_loop.py, and tools/delegate_tool._run_single_child.
Unknown events are ignored. Never raises (event() swallows I/O).
"""
et = str(event_type or "")
if et == "tool.started":
self.tool_start(str(tool_name or ""), preview if preview else args)
elif et == "tool.completed":
self.tool_result(
str(tool_name or ""),
result=kwargs.get("result"),
duration=kwargs.get("duration"),
is_error=bool(kwargs.get("is_error")),
)
elif et == "_thinking":
# Fired as cb("_thinking", <text>) — the text rides in the
# tool_name positional slot (see conversation_loop.py).
self.thinking(str(tool_name or preview or ""))
elif et == "reasoning.available":
# cb("reasoning.available", "_thinking", <text>, None)
self.thinking(str(preview or ""))
elif et == "subagent.text":
self.add_stream_delta(str(preview or ""))
elif et == "subagent.start":
self.event("start", _one_line(preview, _KICKOFF_MAX))
elif et == "subagent.complete":
self.flush_stream()
status = kwargs.get("status", "?")
dur = kwargs.get("duration_seconds")
parts = [f"status={status}"]
if dur is not None:
parts.append(f"duration={dur}s")
summary = kwargs.get("summary") or preview
if summary:
parts.append(f"summary: {_one_line(summary, _RESULT_MAX)}")
self.marker(" ".join(parts))
def finalize(self, entry: Dict[str, Any]) -> None:
"""Terminal marker from the aggregated result entry.
Adds exit-reason detail the subagent.complete event doesn't carry
(budget exhaustion via exit_reason=max_iterations, errors, etc.).
"""
parts = [f"end status={entry.get('status', '?')}"]
exit_reason = entry.get("exit_reason")
if exit_reason:
parts.append(f"exit_reason={exit_reason}")
if exit_reason == "max_iterations":
parts.append("(iteration budget exhausted)")
if entry.get("error"):
parts.append(f"error: {_one_line(entry['error'], _RESULT_MAX)}")
self.marker(" ".join(parts))
def wrap_progress_callback(inner_cb, writer: LiveTranscriptWriter):
"""Wrap a child's tool_progress_callback so events also land in the log.
``inner_cb`` may be None (no parent display) the wrapper still records.
Writer failures never propagate; inner callback behavior is unchanged
(its own exceptions are handled by callers exactly as before).
Preserves the ``_flush`` attribute contract used by _run_single_child.
"""
def _cb(event_type, tool_name=None, preview=None, args=None, **kwargs):
try:
writer.observe(event_type, tool_name, preview, args, **kwargs)
except Exception as exc: # noqa: BLE001 — must never hit the agent loop
logger.debug("Live transcript observe failed: %s", exc)
if inner_cb is not None:
inner_cb(event_type, tool_name, preview, args, **kwargs)
def _flush():
try:
writer.flush_stream()
except Exception:
pass
inner_flush = getattr(inner_cb, "_flush", None)
if callable(inner_flush):
inner_flush()
_cb._flush = _flush
return _cb
# ── dispatch-time helpers ────────────────────────────────────────────────
def create_live_transcripts(
task_list: List[Dict[str, Any]],
context: Optional[str] = None,
delegation_id: Optional[str] = None,
) -> tuple[Optional[str], List[Optional[LiveTranscriptWriter]], List[str]]:
"""Create one pre-headered writer per task + a manifest.json.
Returns ``(delegation_id, writers, paths)``. On any top-level failure
returns ``(None, [None]*n, [])`` so delegation proceeds untouched.
Also opportunistically prunes stale live dirs (retention).
"""
n = len(task_list)
try:
prune_stale_live_dirs()
except Exception:
pass
try:
deleg_id = delegation_id or new_live_delegation_id()
writers: List[Optional[LiveTranscriptWriter]] = []
paths: List[str] = []
for i, t in enumerate(task_list):
w = LiveTranscriptWriter(
deleg_id, i, str(t.get("goal", "")),
context=t.get("context") or context,
)
writers.append(w if w.path is not None else None)
if w.path is not None:
paths.append(str(w.path))
if not paths:
return None, [None] * n, []
_write_manifest(deleg_id, task_list, paths)
return deleg_id, writers, paths
except Exception as exc:
logger.debug("Live transcript creation failed: %s", exc)
return None, [None] * n, []
def _manifest_path(delegation_id: str) -> Path:
return live_transcript_root() / delegation_id / "manifest.json"
def _write_manifest(delegation_id: str, task_list: List[Dict[str, Any]],
paths: List[str]) -> None:
try:
manifest = {
"delegation_id": delegation_id,
"started": time.strftime("%Y-%m-%d %H:%M:%S"),
"task_count": len(task_list),
"tasks": [
{
"index": i,
"goal": str(t.get("goal", ""))[:500],
"log": paths[i] if i < len(paths) else None,
"status": "running",
}
for i, t in enumerate(task_list)
],
}
_manifest_path(delegation_id).write_text(
json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8"
)
except Exception as exc:
logger.debug("Live transcript manifest write failed: %s", exc)
def update_manifest_statuses(delegation_id: Optional[str],
results: List[Dict[str, Any]]) -> None:
"""Best-effort per-task status update once the batch has aggregated."""
if not delegation_id:
return
try:
mp = _manifest_path(delegation_id)
manifest = json.loads(mp.read_text(encoding="utf-8"))
by_index = {r.get("task_index"): r for r in results if isinstance(r, dict)}
for task in manifest.get("tasks", []):
r = by_index.get(task.get("index"))
if r is not None:
task["status"] = r.get("status", task.get("status"))
if r.get("exit_reason"):
task["exit_reason"] = r["exit_reason"]
manifest["completed"] = time.strftime("%Y-%m-%d %H:%M:%S")
mp.write_text(json.dumps(manifest, indent=2, ensure_ascii=False),
encoding="utf-8")
except Exception as exc:
logger.debug("Live transcript manifest update failed: %s", exc)
def prune_stale_live_dirs(max_age_days: int = LIVE_RETENTION_DAYS) -> int:
"""Remove live/<delegation_id> dirs older than the retention window.
Returns how many were removed. Fully best-effort.
"""
removed = 0
try:
root = live_transcript_root()
if not root.is_dir():
return 0
cutoff = time.time() - max_age_days * 86400
for child in root.iterdir():
try:
if child.is_dir() and child.stat().st_mtime < cutoff:
shutil.rmtree(child, ignore_errors=True)
removed += 1
except OSError:
continue
except Exception as exc:
logger.debug("Live transcript pruning failed: %s", exc)
return removed

View file

@ -2119,6 +2119,11 @@ def _format_async_delegation(evt: dict) -> str:
+ (f": {r_error}" if r_error else "")
+ ")"
)
r_live = r.get("live_transcript")
if r_live:
lines.append(
f"Full live transcript (complete tool/assistant trace): {r_live}"
)
return "\n".join(lines)
age = ""

View file

@ -206,6 +206,22 @@ The TUI ships a `/agents` overlay (alias `/tasks`) that turns recursive `delegat
The classic CLI just prints `/agents` as a text summary; the TUI is where the overlay shines. See [TUI — Slash commands](/user-guide/tui#slash-commands).
## Live Transcripts
Every `delegate_task` dispatch also creates one **append-only, human-readable log per task** so you (or the parent agent) can watch a subagent work in real time instead of waiting for the consolidated summary:
```
<hermes_home>/cache/delegation/live/<delegation_id>/task-<n>.log
```
The dispatch response includes the paths as `live_transcripts`, and the files are pre-created at dispatch time, so this works immediately:
```bash
tail -f ~/.hermes/cache/delegation/live/deleg_ab12cd34/task-0.log
```
Each line is timestamped and shows the child's assistant text, thinking snippets, tool calls (`-> tool_name({args})`), tool results, and a final status marker. A `manifest.json` in the same directory describes the batch (goals, task count, per-task status). The logs persist after completion — they double as the full-fidelity operational record alongside the summary — and directories older than 7 days are pruned automatically on new dispatches. Because they live under `cache/delegation`, they are also readable from remote terminal backends (Docker/Modal/SSH).
## Depth Limit and Nested Orchestration
By default, delegation is **flat**: a parent (depth 0) spawns children (depth 1), and those children cannot delegate further. This prevents runaway recursive delegation.