diff --git a/tests/tools/test_delegation_live_log.py b/tests/tools/test_delegation_live_log.py
new file mode 100644
index 000000000000..e73370cb84b5
--- /dev/null
+++ b/tests/tools/test_delegation_live_log.py
@@ -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"]))
diff --git a/tools/async_delegation.py b/tools/async_delegation.py
index c743decf644e..6811d6867fdb 100644
--- a/tools/async_delegation.py
+++ b/tools/async_delegation.py
@@ -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,
diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py
index 12f94a180a8e..6a0e4c0dcd6a 100644
--- a/tools/delegate_tool.py
+++ b/tools/delegate_tool.py
@@ -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"
diff --git a/tools/delegation_live_log.py b/tools/delegation_live_log.py
new file mode 100644
index 000000000000..da157f623c89
--- /dev/null
+++ b/tools/delegation_live_log.py
@@ -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
diff --git a/tools/process_registry.py b/tools/process_registry.py
index 0c7616db0630..97cf89b3bda8 100644
--- a/tools/process_registry.py
+++ b/tools/process_registry.py
@@ -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 = ""
diff --git a/website/docs/user-guide/features/delegation.md b/website/docs/user-guide/features/delegation.md
index e27887461351..36dddd819d4c 100644
--- a/website/docs/user-guide/features/delegation.md
+++ b/website/docs/user-guide/features/delegation.md
@@ -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.