diff --git a/docs/workflow-engine-port-plan.md b/docs/workflow-engine-port-plan.md
index 7a143f90e..f9250f467 100644
--- a/docs/workflow-engine-port-plan.md
+++ b/docs/workflow-engine-port-plan.md
@@ -305,26 +305,28 @@ designed to accept this port.
 | Resume persistence (Phase 9) | `Journal` persisted to the run's file; `resume_from_run_id` reloads it | ✅ |
 | Bundled workflow | `src/workflow/bundled/deep_research.py` (real fan-out → cross-check → cited report) | ✅ |
 
-**Remaining polish (not blocking; documented honestly):**
-- **Rich `/workflows` TUI dialog** — the headless `/workflows` list + the pill label are built and
-  tested; the Textual phases→agents drill-down with the `p`/`x`/`r`/`s` bindings is not (it builds on
-  the `/tasks` panel, which is itself a broken stub — `REPLScreen.focus_task_panel` is missing). The
-  per-run/agent control API it needs already exists (`kill_workflow_task`/`skip_workflow_agent`).
-- **Live pill wiring** — `workflow_pill_label` is tested; threading `runtime_tasks` onto the live
-  `StatusLine` is a small TUI edit not yet made.
-- **`retry_workflow_agent`** reports "unsupported" — re-spawning one agent mid-run needs engine support.
-- **Budget shared-pool / ProgressTracker** — `LiveAgentRunner` builds `RunAgentParams` correctly but
-  isn't wired to `Budget(base_spent=get_total_cost_usd)` / a `ProgressTracker`; do this when the tool
-  constructs the run with the real provider.
-- **Worktree-per-agent** (`wf_<runId>-<idx>`) — `isolation="worktree"` currently runs in-place.
-- **Live integration test** — `LiveAgentRunner`'s `run_agent` composition needs a real/recorded model
-  (the structured-output tool and the whole engine + launcher are unit-tested with a fake runner).
-- **Result delivery** — the final result is captured on the task (`complete_workflow_task`) and shown
-  by `/workflows`; auto-injecting the report back into the conversation on completion (the
-  async-agent `enqueue_*_notification` path) is a follow-up.
-- **Run-file location** — run journals are written under `<cwd>/.clawcodex/workflows/<run_id>.json`
-  (resume reads the same path; internally consistent). The spec's location is
-  `~/.claude/projects/<session>/`; aligning to the session dir is a deliberate, deferred change.
+**Polish — now built (port-plan §10 follow-ups):**
+
+| Item | Module | Tested |
+|---|---|---|
+| Result delivery to the model | `enqueue_workflow_notification` on terminal transitions | ✅ |
+| Run-file location | `get_workflow_run_path` → `~/.clawcodex/transcripts/workflows/` | ✅ |
+| `retry_workflow_agent` | engine bounded retry loop (`WorkflowRun.retry_agent`) | ✅ |
+| ProgressTracker tokens | `LiveAgentRunner` feeds `finalize_agent_tool(progress=…)` | ✅ |
+| Worktree-per-agent | `isolation="worktree"` → `wf_<runId>-<idx>` (`src/workflow/worktree.py`) | ✅ |
+| Live integration test | `LiveAgentRunner` → `run_agent` → real query loop w/ a fake provider | ✅ |
+| `/workflows` TUI dialog | `src/tui/screens/workflow_dialog.py` (list + detail, `x` stop, `r` retry) + opener | ✅ (pilot) |
+| Live pill | `StatusLine` shows "N background workflows" from `runtime_tasks` | ✅ (pilot) |
+
+The integration test caught two real bugs (fixed): tool **dispatch resolves by name from the
+registry**, so a schema agent needs a per-call registry where `StructuredOutput` is the validating
+tool; and that injected tool was **permission-blocked** in the subagent (now explicitly allowed).
+
+**Still remaining (genuinely small, not blocking):**
+- **`p` (pause) / `s` (save)** in the `/workflows` dialog — pause needs an engine pause gate; save
+  needs the `.claude/workflows` write + save dialog. The `x`/`r` actions and the detail view are built.
+- **Budget shared-pool** — `Budget(base_spent=…)` accepts a shared-pool getter, but no token-target
+  source is wired on this path (`budget_total` is `None` by default), so it's a param awaiting a caller.
 
 **Hardening (post adversarial review).** Fixed before merge: the background launch now runs on a
 dedicated daemon thread (`task_manager.start`) so the run outlives the throwaway `asyncio.run`
diff --git a/src/agent/transcript.py b/src/agent/transcript.py
index 3a2898665..529851577 100644
--- a/src/agent/transcript.py
+++ b/src/agent/transcript.py
@@ -83,6 +83,19 @@ def get_agent_transcript_path(agent_id: str) -> str:
     return str(_transcripts_root() / f"{safe_id}.jsonl")
 
 
+def get_workflow_run_path(run_id: str) -> str:
+    """Absolute path to a workflow run's journal file,
+    ``~/.clawcodex/transcripts/workflows/<run_id>.json``.
+
+    Workflow journals live alongside agent transcripts (the per-user session
+    storage root), not under the project tree, so resume state doesn't become
+    VCS noise. ``run_id`` is sanitized like an agent id."""
+    safe_id = _sanitize_agent_id(run_id)
+    root = _transcripts_root() / "workflows"
+    root.mkdir(parents=True, exist_ok=True)
+    return str(root / f"{safe_id}.json")
+
+
 def _sanitize_agent_id(agent_id: str) -> str:
     """Reject path-traversing agent_ids before we touch the filesystem.
 
diff --git a/src/tasks/local_workflow.py b/src/tasks/local_workflow.py
index 087191fc1..946fdf195 100644
--- a/src/tasks/local_workflow.py
+++ b/src/tasks/local_workflow.py
@@ -16,6 +16,7 @@
 
 from __future__ import annotations
 
+import json
 import logging
 from dataclasses import dataclass, field, replace
 from typing import Any, Literal, TYPE_CHECKING
@@ -114,6 +115,7 @@ def _complete(prev: TaskStateBase) -> TaskStateBase:
         return _terminal_replace(prev, status="completed", result=result, summary=_safe_summary(prev.progress))
 
     registry.update(task_id, _complete)
+    enqueue_workflow_notification(task_id, registry, status="completed")
 
 
 def fail_workflow_task(task_id: str, *, error: str, registry: "RuntimeTaskRegistry") -> None:
@@ -123,17 +125,20 @@ def _fail(prev: TaskStateBase) -> TaskStateBase:
         return _terminal_replace(prev, status="failed", error=error)
 
     registry.update(task_id, _fail)
+    enqueue_workflow_notification(task_id, registry, status="failed", error=error)
 
 
 def kill_workflow_task(task_id: str, registry: "RuntimeTaskRegistry") -> None:
     """Abort the whole run (cascades to every subagent) and mark it killed."""
     captured_run: Any = None
+    fired = False
 
     def _kill(prev: TaskStateBase) -> TaskStateBase:
-        nonlocal captured_run
+        nonlocal captured_run, fired
         if not isinstance(prev, LocalWorkflowTaskState) or is_terminal_task_status(prev.status):
             return prev
         captured_run = prev.run
+        fired = True
         return _terminal_replace(prev, status="killed")
 
     registry.update(task_id, _kill)
@@ -143,6 +148,8 @@ def _kill(prev: TaskStateBase) -> TaskStateBase:
             captured_run.controller.abort("workflow_stopped")
         except Exception:
             logger.exception("failed to abort workflow run %s", task_id)
+    if fired:
+        enqueue_workflow_notification(task_id, registry, status="killed")
 
 
 def skip_workflow_agent(task_id: str, agent_key: str, registry: "RuntimeTaskRegistry") -> bool:
@@ -159,10 +166,16 @@ def skip_workflow_agent(task_id: str, agent_key: str, registry: "RuntimeTaskRegi
 
 
 def retry_workflow_agent(task_id: str, agent_key: str, registry: "RuntimeTaskRegistry") -> bool:
-    """Reserved: re-spawning a single agent mid-run needs engine support that
-    does not exist yet. Reports unsupported rather than silently no-op'ing."""
-    logger.info("retry_workflow_agent is not yet supported (task=%s agent=%s)", task_id, agent_key)
-    return False
+    """Re-spawn one in-flight agent by its call-path key (the `r` action).
+    Returns whether a live agent was found to retry."""
+    state = registry.get(task_id)
+    if not isinstance(state, LocalWorkflowTaskState) or state.run is None:
+        return False
+    try:
+        return bool(state.run.retry_agent(agent_key))
+    except Exception:
+        logger.exception("failed to retry agent %s in workflow %s", agent_key, task_id)
+        return False
 
 
 def _safe_summary(progress: Any) -> str | None:
@@ -172,6 +185,86 @@ def _safe_summary(progress: Any) -> str | None:
         return None
 
 
+def _safe_token_total(progress: Any) -> int:
+    """Token total, tolerant of a concurrent mutation on the engine thread."""
+    try:
+        return int(progress.token_total) if progress is not None else 0
+    except Exception:
+        return 0
+
+
+def _render_result(result: Any) -> str | None:
+    """Render a workflow's return value for the <result> section (capped)."""
+    if result is None:
+        return None
+    if isinstance(result, str):
+        text = result
+    else:
+        try:
+            text = json.dumps(result, default=str, ensure_ascii=False)
+        except Exception:
+            text = str(result)
+    return text[:4000]
+
+
+def enqueue_workflow_notification(
+    task_id: str,
+    registry: "RuntimeTaskRegistry",
+    *,
+    status: str,
+    error: str | None = None,
+) -> bool:
+    """Deliver a workflow's terminal result to the model via the shared
+    ``<task-notification>`` queue (mirrors ``enqueue_agent_notification``).
+
+    The ``notified`` flag is check-and-set atomically, so exactly one envelope
+    is delivered even if complete/fail/kill race."""
+    from src.utils.message_queue_manager import enqueue_pending_notification
+    from src.utils.task_notification import build_task_notification_xml
+
+    # Read the token total OUTSIDE the registry mutator: progress is mutated by
+    # the engine on its own daemon thread, so iterating it under the registry
+    # lock could raise "list changed size during iteration". _safe_token_total
+    # guards it; the mutator itself only flips a scalar flag (the stock-agent
+    # pattern in enqueue_agent_notification).
+    snapshot = registry.get(task_id)
+    tokens = _safe_token_total(getattr(snapshot, "progress", None)) if snapshot is not None else 0
+
+    captured: dict[str, Any] = {}
+    should_enqueue = False
+
+    def _mark(prev: TaskStateBase) -> TaskStateBase:
+        nonlocal should_enqueue
+        if not isinstance(prev, LocalWorkflowTaskState) or prev.notified:
+            return prev
+        should_enqueue = True
+        captured.update(
+            name=prev.workflow_name or "workflow",
+            output_file=prev.output_file,
+            result=prev.result,
+            tool_use_id=prev.tool_use_id,
+        )
+        return replace(prev, notified=True)
+
+    registry.update(task_id, _mark)
+    if not should_enqueue:
+        return False
+
+    final_message = _render_result(captured["result"]) if status == "completed" else None
+    xml = build_task_notification_xml(
+        task_id=task_id,
+        description=captured["name"],
+        status=status,  # type: ignore[arg-type]
+        output_file=captured["output_file"],
+        error=error,
+        final_message=final_message,
+        usage={"total_tokens": tokens, "tool_uses": 0, "duration_ms": 0},
+        tool_use_id=captured["tool_use_id"],
+    )
+    enqueue_pending_notification(value=xml, mode="task-notification")
+    return True
+
+
 # ── Task adapter ──────────────────────────────────────────────────────────────
 
 
@@ -194,4 +287,5 @@ async def kill(self, task_id: str, registry: "RuntimeTaskRegistry") -> None:
     "kill_workflow_task",
     "skip_workflow_agent",
     "retry_workflow_agent",
+    "enqueue_workflow_notification",
 ]
diff --git a/src/tool_system/tools/workflow.py b/src/tool_system/tools/workflow.py
index b57882824..dc87c81b1 100644
--- a/src/tool_system/tools/workflow.py
+++ b/src/tool_system/tools/workflow.py
@@ -134,9 +134,9 @@ async def _call(tool_input: dict, context: ToolContext) -> ToolResult:
 
         run_id = "wf_" + uuid.uuid4().hex[:12]
         task_id = generate_task_id("local_workflow")
-        base = Path(context.cwd) if context.cwd else Path.cwd()
-        wf_dir = base / ".clawcodex" / "workflows"
-        output_file = str(wf_dir / f"{run_id}.json")
+        from src.agent.transcript import get_workflow_run_path
+
+        output_file = get_workflow_run_path(run_id)
         runner = factory(context, run_id)
 
         # Same-session resume: replay the prior run's journal if asked.
@@ -145,7 +145,10 @@ async def _call(tool_input: dict, context: ToolContext) -> ToolResult:
         if isinstance(prior_run_id, str) and prior_run_id.strip():
             from src.workflow.launch import load_journal
 
-            resume = load_journal(str(wf_dir / f"{prior_run_id}.json"))
+            try:
+                resume = load_journal(get_workflow_run_path(prior_run_id))
+            except ValueError:
+                resume = None  # malformed run id
 
         coro = run_workflow_task(
             source=source,
diff --git a/src/tui/app.py b/src/tui/app.py
index f84bbbb85..640ebb587 100644
--- a/src/tui/app.py
+++ b/src/tui/app.py
@@ -414,9 +414,20 @@ def _open_phase2_dialog(self, name: str, transcript: Transcript) -> None:
             self._open_message_selector(transcript)
         elif name == "tasks":
             self._open_tasks_dialog(transcript)
+        elif name == "workflows":
+            self._open_workflows_dialog(transcript)
         else:
             transcript.append_system(f"Dialog '{name}' not available.", style="muted")
 
+    def _open_workflows_dialog(self, transcript: Transcript) -> None:
+        registry = getattr(self.tool_context, "runtime_tasks", None)
+        if registry is None:
+            transcript.append_system("Workflows are unavailable.", style="muted")
+            return
+        from src.tui.screens.workflow_dialog import WorkflowsScreen
+
+        self.push_screen(WorkflowsScreen(registry=registry))
+
     def _open_model_picker(self, transcript: Transcript) -> None:
         models = self._list_available_models()
 
diff --git a/src/tui/commands.py b/src/tui/commands.py
index c67dd4a6d..7839b7ac9 100644
--- a/src/tui/commands.py
+++ b/src/tui/commands.py
@@ -51,6 +51,7 @@
     "/diff",
     "/mcp",
     "/tasks",
+    "/workflows",
     "/rewind",
 )
 
@@ -287,6 +288,8 @@ def dispatch_local_command(
         return CommandDispatchResult(handled=True, open_dialog="mcp")
     if name == "/tasks":
         return CommandDispatchResult(handled=True, open_dialog="tasks")
+    if name == "/workflows":
+        return CommandDispatchResult(handled=True, open_dialog="workflows")
     if name == "/rewind":
         return CommandDispatchResult(handled=True, open_dialog="rewind")
 
diff --git a/src/tui/screens/workflow_dialog.py b/src/tui/screens/workflow_dialog.py
new file mode 100644
index 000000000..bfabe2f66
--- /dev/null
+++ b/src/tui/screens/workflow_dialog.py
@@ -0,0 +1,162 @@
+"""The ``/workflows`` modal: a list of runs and a per-run phases→agents detail
+view, with stop (``x``) and retry (``r``) wired to the task API.
+
+``format_workflow_detail`` is a pure helper (unit-tested); the screens are
+exercised with the Textual pilot harness. Pause (``p``) and save (``s``) are
+noted as further work — the engine doesn't pause yet, and save needs the
+``.claude/workflows`` write flow.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Iterator
+
+from rich.text import Text
+from textual.widget import Widget
+from textual.widgets import Static
+
+from src.tasks.local_workflow import (
+    kill_workflow_task,
+    retry_workflow_agent,
+    skip_workflow_agent,
+)
+
+from ..widgets.select_list import SelectList, SelectOption
+from .dialog_base import DialogScreen
+
+
+def format_workflow_detail(state: Any) -> list[str]:
+    """Render a workflow run's phases → agents tree as display lines."""
+    lines = [f"{getattr(state, 'workflow_name', 'workflow')}  [{getattr(state, 'status', '?')}]"]
+    summary = getattr(state, "summary", None)
+    if summary:
+        lines.append(summary)
+    progress = getattr(state, "progress", None)
+    phases = getattr(progress, "phases", None) or []
+    if not phases:
+        lines.append("(no phases yet)")
+    for phase in phases:
+        agents = getattr(phase, "agents", []) or []
+        lines.append(f"▸ {phase.title}  ({len(agents)} agents · {phase.token_total} tokens)")
+        for agent in agents:
+            lines.append(f"    • [{agent.status}] {agent.label}  ({agent.tokens} tok)")
+    return lines
+
+
+def _agent_options(state: Any) -> list[SelectOption]:
+    progress = getattr(state, "progress", None)
+    phases = getattr(progress, "phases", None) or []
+    options: list[SelectOption] = []
+    for phase in phases:
+        for agent in getattr(phase, "agents", []) or []:
+            options.append(
+                SelectOption(
+                    label=f"[{agent.status}] {agent.label}",
+                    value=agent.key or "",
+                    description=f"{agent.tokens} tok · {phase.title}",
+                    disabled=not agent.key,
+                )
+            )
+    return options
+
+
+class WorkflowDetailScreen(DialogScreen[None]):
+    """Phases → agents for one run; x stops an agent, r retries it."""
+
+    footer_hint = "x stop agent · r retry agent · Esc back"
+    BINDINGS = [("x", "stop_agent", "Stop agent"), ("r", "retry_agent", "Retry agent")]
+
+    def __init__(self, *, registry: Any, task_id: str) -> None:
+        super().__init__()
+        self._registry = registry
+        self._task_id = task_id
+        state = registry.get(task_id)
+        self.title_text = f"Workflow · {getattr(state, 'workflow_name', 'run')}"
+        self._select: SelectList | None = None
+
+    def build_body(self) -> Iterator[Widget]:
+        state = self._registry.get(self._task_id)
+        yield Static(Text("\n".join(format_workflow_detail(state)), style="none"), markup=False)
+        options = _agent_options(state)
+        if options:
+            self._select = SelectList(options, allow_cancel=True)
+            yield self._select
+
+    def _post_mount(self) -> None:
+        if self._select is not None:
+            self._select.focus()
+
+    def _current_key(self) -> str | None:
+        if self._select is None or self._select.current is None:
+            return None
+        return str(self._select.current.value) or None
+
+    def action_stop_agent(self) -> None:
+        key = self._current_key()
+        if key:
+            skip_workflow_agent(self._task_id, key, self._registry)
+
+    def action_retry_agent(self) -> None:
+        key = self._current_key()
+        if key:
+            retry_workflow_agent(self._task_id, key, self._registry)
+
+    def on_select_list_selection_cancelled(self, _: SelectList.SelectionCancelled) -> None:
+        self.dismiss(None)
+
+
+class WorkflowsScreen(DialogScreen[None]):
+    """Modal list of workflow runs. Enter opens detail; x stops the run."""
+
+    title_text = "Workflows"
+    footer_hint = "Enter detail · x stop run · Esc close"
+    BINDINGS = [("x", "stop_run", "Stop run")]
+
+    def __init__(self, *, registry: Any) -> None:
+        super().__init__()
+        self._registry = registry
+        self._select: SelectList | None = None
+
+    def _runs(self) -> list[Any]:
+        try:
+            return [t for t in self._registry.all() if getattr(t, "type", None) == "local_workflow"]
+        except Exception:
+            return []
+
+    def _options(self) -> list[SelectOption]:
+        return [
+            SelectOption(
+                label=f"{getattr(r, 'workflow_name', 'workflow')}  [{r.status}]",
+                value=r.id,
+                description=getattr(r, "summary", None) or "",
+            )
+            for r in self._runs()
+        ]
+
+    def build_body(self) -> Iterator[Widget]:
+        options = self._options()
+        if not options:
+            yield Static(Text("No workflow runs.", style="dim"), markup=False)
+            return
+        self._select = SelectList(options, allow_cancel=True)
+        yield self._select
+
+    def _post_mount(self) -> None:
+        if self._select is not None:
+            self._select.focus()
+
+    def on_select_list_option_selected(self, event: SelectList.OptionSelected) -> None:
+        task_id = str(event.option.value)
+        if self._registry.get(task_id) is not None:
+            self.app.push_screen(WorkflowDetailScreen(registry=self._registry, task_id=task_id))
+
+    def on_select_list_selection_cancelled(self, _: SelectList.SelectionCancelled) -> None:
+        self.dismiss(None)
+
+    def action_stop_run(self) -> None:
+        if self._select is None or self._select.current is None:
+            return
+        task_id = str(self._select.current.value)
+        kill_workflow_task(task_id, self._registry)
+        if self._select is not None:
+            self._select.set_options(self._options(), keep_cursor=True)
diff --git a/src/tui/widgets/status_line.py b/src/tui/widgets/status_line.py
index 89a21098f..0c0b5c46e 100644
--- a/src/tui/widgets/status_line.py
+++ b/src/tui/widgets/status_line.py
@@ -45,6 +45,7 @@ class StatusLine(Static):
     turns: reactive[int] = reactive(0)
     is_thinking: reactive[bool] = reactive(False)
     queued: reactive[int] = reactive(0)
+    workflows: reactive[int] = reactive(0)
 
     def __init__(
         self,
@@ -81,10 +82,26 @@ def _tick(self) -> None:
         if self._app_state is not None:
             self.is_thinking = self._app_state.is_thinking
             self.queued = len(self._app_state.queued_prompts)
+        self.workflows = self._count_workflows()
         if self.is_thinking:
             self._frame = (self._frame + 1) % len(_SPINNER_FRAMES)
         self._redraw()
 
+    def _count_workflows(self) -> int:
+        """Number of running/pending workflow runs (for the footer pill)."""
+        try:
+            tool_context = getattr(self.app, "tool_context", None)
+            if tool_context is None:
+                return 0
+            return sum(
+                1
+                for t in tool_context.runtime_tasks.all()
+                if getattr(t, "type", None) == "local_workflow"
+                and getattr(t, "status", "") in ("running", "pending")
+            )
+        except Exception:
+            return 0
+
     # ---- public API ----
     def bind_state(self, state: AppState) -> None:
         self._app_state = state
@@ -125,6 +142,9 @@ def watch_turns(self, _: int) -> None:
     def watch_queued(self, _: int) -> None:
         self._redraw()
 
+    def watch_workflows(self, _: int) -> None:
+        self._redraw()
+
     def _redraw(self) -> None:
         spinner = _SPINNER_FRAMES[self._frame] if self.is_thinking else " "
         self.update(self._compose_text(spinner=spinner))
@@ -164,6 +184,10 @@ def _compose_text(self, *, spinner: str) -> Text:
         right_bits: list[str] = [f"turn {self.turns}"]
         if self.queued:
             right_bits.append(f"queued {self.queued}")
+        if self.workflows:
+            right_bits.append(
+                "1 background workflow" if self.workflows == 1 else f"{self.workflows} background workflows"
+            )
         if state and state.usage:
             in_t = state.usage.get("input_tokens", 0)
             out_t = state.usage.get("output_tokens", 0)
diff --git a/src/workflow/constants.py b/src/workflow/constants.py
index 71970912b..b70c553cc 100644
--- a/src/workflow/constants.py
+++ b/src/workflow/constants.py
@@ -22,6 +22,9 @@
 #: Retry cap for schema-validated structured output (upstream parity).
 MAX_STRUCTURED_OUTPUT_RETRIES = 5
 
+#: How many times a single agent may be re-spawned via the `r` (retry) action.
+MAX_AGENT_RETRIES = 3
+
 #: Hard ceiling on the concurrency cap regardless of core count.
 _CONCURRENCY_HARD_MAX = 16
 
diff --git a/src/workflow/runner.py b/src/workflow/runner.py
index 92452b855..129010f86 100644
--- a/src/workflow/runner.py
+++ b/src/workflow/runner.py
@@ -57,12 +57,37 @@ def __init__(
         self._run_id = run_id
         self._max_turns = max_turns
 
-    async def run(self, spec: AgentSpec, *, abort: AbortController, index: int) -> AgentOutcome:
+    async def run(self, spec: AgentSpec, *, abort: AbortController, index: str) -> AgentOutcome:
+        # isolation="worktree": run the agent in a throwaway git worktree so
+        # parallel file-mutating agents don't collide. Best-effort — if the
+        # worktree can't be created the agent runs in place.
+        if spec.isolation == "worktree":
+            import dataclasses
+            from pathlib import Path as _Path
+
+            from src.workflow.worktree import agent_worktree
+
+            base_cwd = str(self._parent_context.cwd) if getattr(self._parent_context, "cwd", None) else "."
+            async with agent_worktree(self._run_id, index, base_cwd) as wt:
+                context = (
+                    dataclasses.replace(self._parent_context, cwd=_Path(wt))
+                    if wt
+                    else self._parent_context
+                )
+                return await self._run_in_context(spec, context, abort=abort, index=index)
+        return await self._run_in_context(spec, self._parent_context, abort=abort, index=index)
+
+    async def _run_in_context(
+        self, spec: AgentSpec, parent_context: Any, *, abort: AbortController, index: str
+    ) -> AgentOutcome:
         # Imported lazily: ``src.agent`` pulls in the whole agent stack, which
         # the engine core deliberately never imports.
         from src.agent.agent_tool_utils import finalize_agent_tool, resolve_agent_tools
-        from src.agent.constants import WORKFLOW_TOOL_NAME
+        from src.agent.constants import ALL_AGENT_DISALLOWED_TOOLS, WORKFLOW_TOOL_NAME
         from src.agent.run_agent import RunAgentParams, run_agent
+        from src.tasks.progress import ProgressTracker, update_progress_from_message
+        from src.tool_system.registry import ToolRegistry
+        from src.types.messages import AssistantMessage
 
         agent_type = spec.agent_type or self._default_agent_type
         agent_definition = self._resolve_agent(agent_type)
@@ -77,23 +102,40 @@ async def run(self, spec: AgentSpec, *, abort: AbortController, index: int) -> A
         resolved = resolve_agent_tools(agent_definition, self._base_tools, is_async=False)
         worker_tools = [t for t in resolved.resolved_tools if getattr(t, "name", "") != WORKFLOW_TOOL_NAME]
 
-        # isolation="worktree" is not yet wired (run_agent worktree support is a
-        # later phase); a worktree request currently runs in-place.
         collector: Optional[StructuredOutputCollector] = None
         prompt = spec.prompt
+        structured_tool = None
         if spec.schema is not None:
             collector = StructuredOutputCollector(schema=spec.schema)
+            structured_tool = make_structured_output_tool(collector)
             worker_tools = [t for t in worker_tools if getattr(t, "name", "") != SYNTHETIC_OUTPUT_TOOL_NAME]
-            worker_tools.append(make_structured_output_tool(collector))
+            worker_tools.append(structured_tool)
             prompt = spec.prompt + _SCHEMA_NUDGE
         available_tools = worker_tools
 
+        # Tool DISPATCH resolves by name from the registry, not from
+        # ``available_tools`` — so the schema agent needs a per-call registry in
+        # which ``StructuredOutput`` is *our* validating tool (not the stock
+        # no-op) and ``Workflow`` is absent. Built per call so concurrent schema
+        # agents don't share a collector.
+        agent_registry = ToolRegistry()
+        for t in self._tool_registry.list_tools():
+            # Same firewall as the advertised pool: no Agent/Workflow/TaskStop/...
+            # so a subagent can't recurse or escalate via a by-name dispatch.
+            if getattr(t, "name", "") in ALL_AGENT_DISALLOWED_TOOLS:
+                continue
+            if structured_tool is not None and getattr(t, "name", "") == SYNTHETIC_OUTPUT_TOOL_NAME:
+                continue
+            agent_registry.register(t)
+        if structured_tool is not None:
+            agent_registry.register(structured_tool)
+
         params = RunAgentParams(
-            parent_context=self._parent_context,
+            parent_context=parent_context,
             agent_definition=agent_definition,
             prompt=prompt,
             available_tools=available_tools,
-            tool_registry=self._tool_registry,
+            tool_registry=agent_registry,
             provider=self._provider,
             model=spec.model,
             agent_id=agent_id,
@@ -108,16 +150,25 @@ async def run(self, spec: AgentSpec, *, abort: AbortController, index: int) -> A
             use_exact_tools=True,
         )
 
+        # Feed a ProgressTracker so finalize_agent_tool reports chapter-correct
+        # token totals (latest input + cumulative output) rather than the
+        # message.usage fallback — these tokens drive the workflow budget.
+        tracker = ProgressTracker()
         messages: list = []
         try:
             async for message in run_agent(params):
                 messages.append(message)
+                if isinstance(message, AssistantMessage):
+                    try:
+                        update_progress_from_message(tracker, message)
+                    except Exception:  # noqa: BLE001 — progress is best-effort
+                        pass
         except AbortError:
             raise  # cancellation unwinds; the engine marks the agent aborted
 
         # finalize_agent_tool raises if the run produced no assistant message;
         # the engine catches that and resolves agent() to None (a "death").
-        result = finalize_agent_tool(messages, agent_id, {"agent_type": agent_type})
+        result = finalize_agent_tool(messages, agent_id, {"agent_type": agent_type}, progress=tracker)
         tokens = result.total_tokens
         tool_uses = result.total_tool_use_count
 
diff --git a/src/workflow/runtime.py b/src/workflow/runtime.py
index b0b711a87..9ceaf3562 100644
--- a/src/workflow/runtime.py
+++ b/src/workflow/runtime.py
@@ -28,7 +28,7 @@
 
 from .budget import Budget
 from .callpath import CallKey, current_branch, key_to_str, reset_branch, use_branch
-from .constants import MAX_ITEMS_PER_CALL
+from .constants import MAX_AGENT_RETRIES, MAX_ITEMS_PER_CALL
 from .errors import WorkflowError, WorkflowMetaError
 from .journal import MISS, Journal, JournalRecord
 from .primitives import await_item, run_stage
@@ -87,11 +87,23 @@ def __init__(
         # form the UI/task layer carries on each AgentRecord), reachable so the
         # task layer can stop one agent without aborting the whole run.
         self._agent_controllers: dict[str, AbortController] = {}
+        # Keys whose agent has been asked to retry (the `r` action).
+        self._retry_requested: set[str] = set()
 
     @property
     def controller(self) -> AbortController:
         return self._controller
 
+    def retry_agent(self, key: str) -> bool:
+        """Re-spawn one in-flight agent: flag it for retry and abort the current
+        attempt so ``agent()`` runs it again. Returns whether it was live."""
+        controller = self._agent_controllers.get(key)
+        if controller is None:
+            return False
+        self._retry_requested.add(key)
+        controller.abort("agent_retry")
+        return True
+
     @property
     def meta(self) -> WorkflowMeta:
         return self._meta
@@ -161,19 +173,32 @@ async def agent(
         self._budget.check()
 
         record = self._progress.agent_started(self._next_display(), eff_label, eff_phase, key_str)
-        child = create_child_abort_controller(self._controller)
-        self._agent_controllers[key_str] = child
-        try:
-            async with self._scheduler.slot():
-                try:
-                    outcome = await self._runner.run(spec, abort=child, index=key_to_str(key))
-                except AbortError:
-                    self._progress.agent_finished(record, status="failed", error="aborted")
-                    raise
-                except Exception as exc:  # noqa: BLE001 — a subagent death -> None
-                    outcome = AgentOutcome(error=f"{type(exc).__name__}: {exc}")
-        finally:
-            self._agent_controllers.pop(key_str, None)
+        attempts = 0
+        while True:
+            child = create_child_abort_controller(self._controller)
+            self._agent_controllers[key_str] = child
+            try:
+                async with self._scheduler.slot():
+                    try:
+                        outcome = await self._runner.run(spec, abort=child, index=key_str)
+                    except AbortError:
+                        outcome = AgentOutcome(skipped=True)
+                    except Exception as exc:  # noqa: BLE001 — a subagent death -> None
+                        outcome = AgentOutcome(error=f"{type(exc).__name__}: {exc}")
+            finally:
+                self._agent_controllers.pop(key_str, None)
+            # The `r` (retry) action re-spawns a running agent, bounded.
+            if key_str in self._retry_requested and attempts < MAX_AGENT_RETRIES:
+                self._retry_requested.discard(key_str)
+                attempts += 1
+                continue
+            break
+
+        # A run-level kill (the whole controller aborted, vs. a single-agent
+        # skip) propagates to end the run; a lone skip just resolves to None.
+        if outcome.skipped and self._controller.signal.aborted:
+            self._progress.agent_finished(record, status="failed", error="aborted")
+            raise AbortError(self._controller.signal.reason or "aborted")
 
         self._budget.add(outcome.tokens)
         if outcome.error is not None:
diff --git a/src/workflow/structured.py b/src/workflow/structured.py
index e7630f604..c60e93f9e 100644
--- a/src/workflow/structured.py
+++ b/src/workflow/structured.py
@@ -106,6 +106,8 @@ def _call(tool_input: dict, context: Any) -> ToolResult:
             is_error=True,
         )
 
+    from src.permissions.types import PermissionAllowDecision
+
     return build_tool(
         name=SYNTHETIC_OUTPUT_TOOL_NAME,
         input_schema={"type": "object", "additionalProperties": True},
@@ -118,4 +120,7 @@ def _call(tool_input: dict, context: Any) -> ToolResult:
         max_result_size_chars=100_000,
         is_read_only=lambda _input: True,
         is_concurrency_safe=lambda _input: True,
+        # Always allowed — it only records the model's own final answer; without
+        # this the subagent's permission context can block it before validation.
+        check_permissions=lambda tool_input, _ctx: PermissionAllowDecision(updated_input=tool_input),
     )
diff --git a/src/workflow/worktree.py b/src/workflow/worktree.py
new file mode 100644
index 000000000..0ba12ea4a
--- /dev/null
+++ b/src/workflow/worktree.py
@@ -0,0 +1,57 @@
+"""Per-agent git worktree isolation for ``agent(..., isolation="worktree")``.
+
+Each isolated agent runs in a throwaway worktree named ``wf_<runId>-<idx>``,
+created as a sibling of the main working tree and removed when the agent
+finishes. If creation fails (not a git repo, etc.) the context manager yields
+``None`` and the caller runs in place — isolation is best-effort, never fatal.
+
+Cleanup is **on-exit only**: there is no crash-recovery sweep, so a process kill
+(or a ``remove_worktree`` failure) can orphan a ``wf_*`` worktree. The git calls
+run in a thread (``asyncio.to_thread``) so they don't block the workflow's event
+loop / serialize parallel worktree setup.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import AsyncIterator, Optional
+
+from src.utils.git import create_worktree, remove_worktree
+
+logger = logging.getLogger(__name__)
+
+
+def worktree_slug(run_id: str, index: str) -> str:
+    """The ``wf_<runId>-<idx>`` directory name.
+
+    ``index`` is the deterministic call-path key (unique per agent); dots become
+    dashes for a clean filesystem slug. Distinct keys can't collide after the
+    transform because call-path digits are positionally unique."""
+    safe_index = str(index).replace(".", "-")
+    return f"{run_id}-{safe_index}"
+
+
+@asynccontextmanager
+async def agent_worktree(run_id: str, index: str, base_cwd: str) -> AsyncIterator[Optional[str]]:
+    """Create a worktree for one agent and remove it on exit.
+
+    Yields the worktree path, or ``None`` if it couldn't be created."""
+    base = Path(base_cwd).resolve()
+    wt_path = base.parent / worktree_slug(run_id, index)
+    created = False
+    try:
+        created = await asyncio.to_thread(create_worktree, str(wt_path), cwd=str(base))
+    except Exception:  # noqa: BLE001 — isolation is best-effort
+        logger.debug("worktree create failed for %s", wt_path, exc_info=True)
+        created = False
+    try:
+        yield str(wt_path) if created else None
+    finally:
+        if created:
+            try:
+                await asyncio.to_thread(remove_worktree, str(wt_path), cwd=str(base), force=True)
+            except Exception:  # noqa: BLE001
+                logger.debug("worktree remove failed for %s", wt_path, exc_info=True)
diff --git a/tests/tui/test_workflow_dialog.py b/tests/tui/test_workflow_dialog.py
new file mode 100644
index 000000000..964cad1b7
--- /dev/null
+++ b/tests/tui/test_workflow_dialog.py
@@ -0,0 +1,142 @@
+"""Tests for the /workflows TUI dialog + render helper (#1)."""
+
+from __future__ import annotations
+
+import pytest
+
+pytest.importorskip("textual")
+
+from textual.app import App
+
+from src.task_registry import RuntimeTaskRegistry
+from src.tasks.local_workflow import register_workflow_task
+from src.tui.screens.workflow_dialog import (
+    WorkflowDetailScreen,
+    WorkflowsScreen,
+    format_workflow_detail,
+)
+from src.workflow.progress import WorkflowProgress
+
+
+class _FakeController:
+    def __init__(self):
+        self.aborted = None
+
+    def abort(self, reason=None):
+        self.aborted = reason
+
+
+class _FakeRun:
+    def __init__(self):
+        self.controller = _FakeController()
+
+
+def _registry_with_run(task_id="wdlg1"):
+    reg = RuntimeTaskRegistry()
+    prog = WorkflowProgress([{"title": "Search"}])
+    prog.start_phase("Search")
+    rec = prog.agent_started(0, "finder", "Search", "0")
+    prog.agent_finished(rec, status="completed", tokens=5)
+    register_workflow_task(
+        task_id=task_id,
+        run_id="r1",
+        workflow_name="demo",
+        description="demo run",
+        output_file="/tmp/x.json",
+        progress=prog,
+        run=_FakeRun(),
+        registry=reg,
+    )
+    return reg
+
+
+def test_format_workflow_detail_renders_phases_and_agents():
+    reg = _registry_with_run()
+    lines = format_workflow_detail(reg.get("wdlg1"))
+    blob = "\n".join(lines)
+    assert "demo" in blob
+    assert "Search" in blob
+    assert "finder" in blob
+    assert "completed" in blob
+
+
+def test_format_workflow_detail_handles_no_phases():
+    reg = RuntimeTaskRegistry()
+    register_workflow_task(
+        task_id="empty", run_id="r0", workflow_name="w", description="d",
+        output_file="/tmp/y", progress=WorkflowProgress(), run=_FakeRun(), registry=reg,
+    )
+    lines = format_workflow_detail(reg.get("empty"))
+    assert any("no phases" in line.lower() for line in lines)
+
+
+class _Host(App):
+    def __init__(self, registry):
+        super().__init__()
+        self._wf_registry = registry  # NB: not `_registry` — Textual's App owns that
+
+    def on_mount(self):
+        self.push_screen(WorkflowsScreen(registry=self._wf_registry))
+
+
+async def test_workflows_screen_lists_runs():
+    reg = _registry_with_run()
+    app = _Host(reg)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        assert isinstance(app.screen, WorkflowsScreen)
+        # the run is listed in the select
+        assert app.screen._select is not None
+        labels = [o.label for o in app.screen._select.options]
+        assert any("demo" in label for label in labels)
+
+
+async def test_workflows_screen_stop_kills_run():
+    reg = _registry_with_run()
+    app = _Host(reg)
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        await pilot.press("x")  # stop the highlighted run
+        await pilot.pause()
+    state = reg.get("wdlg1")
+    assert state.status == "killed"
+
+
+async def test_status_line_workflow_pill_counts():
+    from pathlib import Path
+
+    from src.tool_system.context import ToolContext
+    from src.tui.widgets.status_line import StatusLine
+
+    reg = _registry_with_run()  # one running workflow
+    status_line = StatusLine(provider="p", model="m", workspace_root=Path("."))
+
+    class _SLHost(App):
+        def __init__(self):
+            super().__init__()
+            self.tool_context = ToolContext(workspace_root=Path("."), runtime_tasks=reg)
+
+        def compose(self):
+            yield status_line
+
+    app = _SLHost()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        status_line._tick()  # deterministic count refresh
+        assert status_line.workflows == 1
+
+
+async def test_detail_screen_opens_with_agents():
+    reg = _registry_with_run()
+
+    class _DetailHost(App):
+        def on_mount(self):
+            self.push_screen(WorkflowDetailScreen(registry=reg, task_id="wdlg1"))
+
+    app = _DetailHost()
+    async with app.run_test() as pilot:
+        await pilot.pause()
+        assert isinstance(app.screen, WorkflowDetailScreen)
+        # the agent select lists the one finished agent
+        assert app.screen._select is not None
+        assert len(app.screen._select.options) == 1
diff --git a/tests/workflow/test_delivery.py b/tests/workflow/test_delivery.py
new file mode 100644
index 000000000..00746d71b
--- /dev/null
+++ b/tests/workflow/test_delivery.py
@@ -0,0 +1,94 @@
+"""Tests for workflow result delivery (#7) and run-file location (#8)."""
+
+from __future__ import annotations
+
+import pytest
+
+from src.agent.transcript import get_workflow_run_path
+from src.task_registry import RuntimeTaskRegistry
+from src.tasks.local_workflow import (
+    complete_workflow_task,
+    kill_workflow_task,
+    register_workflow_task,
+)
+from src.workflow.progress import WorkflowProgress
+
+
+class _FakeRun:
+    class _C:
+        def abort(self, reason=None):
+            pass
+
+    controller = _C()
+
+
+def _register(reg):
+    return register_workflow_task(
+        task_id="wnotify1",
+        run_id="wf_run1",
+        workflow_name="demo",
+        description="demo",
+        output_file="/tmp/x.json",
+        progress=WorkflowProgress(),
+        run=_FakeRun(),
+        registry=reg,
+    )
+
+
+@pytest.fixture
+def captured(monkeypatch):
+    seen: list[dict] = []
+    monkeypatch.setattr(
+        "src.utils.message_queue_manager.enqueue_pending_notification",
+        lambda **kw: seen.append(kw),
+    )
+    return seen
+
+
+# ── #7 result delivery ────────────────────────────────────────────────────────
+
+
+def test_completion_delivers_result_to_model(captured):
+    reg = RuntimeTaskRegistry()
+    state = _register(reg)
+    complete_workflow_task(state.id, result={"answer": 42}, registry=reg)
+
+    assert len(captured) == 1
+    note = captured[0]
+    assert note["mode"] == "task-notification"
+    xml = note["value"]
+    assert "completed" in xml
+    assert "42" in xml  # the result is rendered into the <result> section
+    assert reg.get(state.id).notified is True
+
+
+def test_exactly_one_notification_even_if_complete_then_kill(captured):
+    reg = RuntimeTaskRegistry()
+    state = _register(reg)
+    complete_workflow_task(state.id, result="done", registry=reg)
+    kill_workflow_task(state.id, reg)  # late kill on a completed task
+    assert len(captured) == 1  # notified flag guards duplicate delivery
+
+
+def test_kill_delivers_a_killed_notification(captured):
+    reg = RuntimeTaskRegistry()
+    state = _register(reg)
+    kill_workflow_task(state.id, reg)
+    assert len(captured) == 1
+    assert "killed" in captured[0]["value"]
+
+
+# ── #8 run-file location ──────────────────────────────────────────────────────
+
+
+def test_run_path_lives_under_session_storage():
+    p = get_workflow_run_path("wf_abc123def")
+    assert ".clawcodex" in p
+    assert "transcripts" in p
+    assert "workflows" in p
+    assert p.endswith("wf_abc123def.json")
+
+
+def test_run_path_rejects_traversal():
+    with pytest.raises(ValueError):
+        get_workflow_run_path("../../etc/passwd")
diff --git a/tests/workflow/test_local_workflow.py b/tests/workflow/test_local_workflow.py
index 5432373b9..2a57cba8a 100644
--- a/tests/workflow/test_local_workflow.py
+++ b/tests/workflow/test_local_workflow.py
@@ -32,11 +32,16 @@ class _FakeRun:
     def __init__(self):
         self.controller = _FakeController()
         self.skipped = []
+        self.retried = []
 
     def abort_agent(self, key):
         self.skipped.append(key)
         return key == "0"
 
+    def retry_agent(self, key):
+        self.retried.append(key)
+        return key == "0"
+
 
 def _register(registry, run=None, progress=None):
     progress = progress or WorkflowProgress([{"title": "P"}])
@@ -112,10 +117,13 @@ def test_skip_agent_targets_one_controller():
     assert run.skipped == ["0", "9"]
 
 
-def test_retry_is_unsupported_for_now():
+def test_retry_agent_targets_one():
     reg = RuntimeTaskRegistry()
-    state = _register(reg)
-    assert retry_workflow_agent(state.id, "0", reg) is False
+    run = _FakeRun()
+    state = _register(reg, run=run)
+    assert retry_workflow_agent(state.id, "0", reg) is True
+    assert retry_workflow_agent(state.id, "9", reg) is False
+    assert run.retried == ["0", "9"]
 
 
 async def test_task_adapter_kill():
diff --git a/tests/workflow/test_retry.py b/tests/workflow/test_retry.py
new file mode 100644
index 000000000..18a4b7939
--- /dev/null
+++ b/tests/workflow/test_retry.py
@@ -0,0 +1,72 @@
+"""Tests for the `r` (retry) action — re-spawning a running agent (#3)."""
+
+from __future__ import annotations
+
+from src.workflow.constants import MAX_AGENT_RETRIES
+from src.workflow.runtime import run_workflow
+from src.workflow.types import AgentOutcome
+
+META = 'meta = {"name": "t", "description": "d"}\n'
+
+
+async def test_agent_retry_reruns_and_recovers(make_runner):
+    state = {"n": 0, "run": None}
+
+    def handler(spec, index):
+        state["n"] += 1
+        if state["n"] == 1:
+            state["run"].retry_agent(index)  # ask to retry this very agent
+            return AgentOutcome(skipped=True)
+        return AgentOutcome(text="recovered")
+
+    runner = make_runner(handler=handler)
+    res = await run_workflow(
+        META + 'return await agent("x")',
+        runner=runner,
+        on_start=lambda run: state.__setitem__("run", run),
+    )
+    assert res.ok
+    assert res.value == "recovered"
+    assert state["n"] == 2  # initial attempt + one retry
+
+
+async def test_agent_retry_is_bounded(make_runner):
+    state = {"n": 0, "run": None}
+
+    def handler(spec, index):
+        state["n"] += 1
+        state["run"].retry_agent(index)  # always ask to retry
+        return AgentOutcome(skipped=True)
+
+    runner = make_runner(handler=handler)
+    res = await run_workflow(
+        META + 'return await agent("x")',
+        runner=runner,
+        on_start=lambda run: state.__setitem__("run", run),
+    )
+    assert res.ok
+    assert res.value is None  # retries exhausted -> skipped -> None
+    assert state["n"] == MAX_AGENT_RETRIES + 1
+
+
+async def test_skip_resolves_to_none_without_aborting_the_run(make_runner):
+    # A single-agent abort (skip) must resolve that agent to None and let the
+    # rest of the script continue — it must NOT propagate and end the run.
+    state = {"run": None}
+
+    def handler(spec, index):
+        if spec.prompt == "skipme":
+            state["run"].abort_agent(index)  # skip this one
+            return AgentOutcome(skipped=True)
+        return AgentOutcome(text=spec.prompt)
+
+    runner = make_runner(handler=handler)
+    script = META + (
+        'a = await agent("first")\n'
+        'b = await agent("skipme")\n'
+        'c = await agent("third")\n'
+        "return [a, b, c]\n"
+    )
+    res = await run_workflow(script, runner=runner, on_start=lambda run: state.__setitem__("run", run))
+    assert res.ok
+    assert res.value == ["first", None, "third"]
diff --git a/tests/workflow/test_runner_integration.py b/tests/workflow/test_runner_integration.py
new file mode 100644
index 000000000..5f629318c
--- /dev/null
+++ b/tests/workflow/test_runner_integration.py
@@ -0,0 +1,98 @@
+"""Live(-ish) integration test for LiveAgentRunner (#6).
+
+Drives the real ``run_agent`` + query loop + tool dispatch with a *fake provider*
+(no network/model), exercising the composition the unit tests can't: the
+injected StructuredOutput tool, schema validation, ``finalize_agent_tool``, and
+the C3 firewall (Workflow stripped, StructuredOutput injected into the pool).
+"""
+
+from __future__ import annotations
+
+from src.agent.agent_definitions import GENERAL_PURPOSE_AGENT
+from src.agent.constants import ALL_AGENT_DISALLOWED_TOOLS
+from src.providers.base import ChatResponse
+from src.tool_system.context import ToolContext
+from src.tool_system.defaults import build_default_registry
+from src.utils.abort_controller import create_abort_controller
+from src.workflow.runner import LiveAgentRunner
+from src.workflow.types import AgentSpec
+
+
+class _ScriptedProvider:
+    model = "fake"
+
+    def __init__(self, script: list[ChatResponse]):
+        self._script = script
+        self._turn = 0
+        self.tools_seen: list[list[str]] = []
+
+    def chat(self, messages, tools=None, **kwargs):
+        self.tools_seen.append([t.get("name") for t in (tools or [])])
+        resp = self._script[min(self._turn, len(self._script) - 1)]
+        self._turn += 1
+        return resp
+
+    def chat_stream_response(self, *a, **kw):  # pragma: no cover - not used
+        raise NotImplementedError
+
+
+def _resp(content="", *, tool_uses=None, finish="stop"):
+    return ChatResponse(
+        content=content,
+        model="fake",
+        usage={"input_tokens": 4, "output_tokens": 3},
+        finish_reason=finish,
+        tool_uses=tool_uses,
+    )
+
+
+def _runner(provider, tmp_path, max_turns=4):
+    registry = build_default_registry(provider=provider)
+    ctx = ToolContext(workspace_root=tmp_path)
+    return LiveAgentRunner(
+        provider=provider,
+        tool_registry=registry,
+        parent_context=ctx,
+        base_tools=list(registry.list_tools()),
+        resolve_agent=lambda _t: GENERAL_PURPOSE_AGENT,
+        run_id="wf_itest",
+        max_turns=max_turns,
+    )
+
+
+async def test_text_agent_returns_final_text(tmp_path):
+    provider = _ScriptedProvider([_resp("hello from the agent")])
+    runner = _runner(provider, tmp_path)
+    out = await runner.run(AgentSpec(prompt="hi"), abort=create_abort_controller(), index="0")
+    assert out.text is not None and "hello from the agent" in out.text
+    # Firewall: no disallowed tool (Agent/Workflow/TaskStop/...) in a subagent pool.
+    assert all(not (set(names) & ALL_AGENT_DISALLOWED_TOOLS) for names in provider.tools_seen)
+
+
+async def test_schema_agent_returns_validated_object(tmp_path):
+    schema = {
+        "type": "object",
+        "properties": {"answer": {"type": "integer"}},
+        "required": ["answer"],
+        "additionalProperties": False,
+    }
+    provider = _ScriptedProvider([
+        _resp(tool_uses=[{"id": "s1", "name": "StructuredOutput", "input": {"answer": 42}}], finish="tool_use"),
+        _resp("done"),
+    ])
+    runner = _runner(provider, tmp_path)
+    out = await runner.run(AgentSpec(prompt="produce", schema=schema), abort=create_abort_controller(), index="0")
+    assert out.structured == {"answer": 42}
+    # The injected StructuredOutput tool reached the model; no disallowed tool did.
+    assert any("StructuredOutput" in names for names in provider.tools_seen)
+    assert all(not (set(names) & ALL_AGENT_DISALLOWED_TOOLS) for names in provider.tools_seen)
+
+
+async def test_schema_not_produced_resolves_to_none(tmp_path):
+    schema = {"type": "object", "properties": {"answer": {"type": "integer"}}, "required": ["answer"]}
+    provider = _ScriptedProvider([_resp("I won't use the tool")])
+    runner = _runner(provider, tmp_path)
+    out = await runner.run(AgentSpec(prompt="produce", schema=schema), abort=create_abort_controller(), index="0")
+    assert out.structured is None
+    assert out.error is not None
+    assert "structured output not produced" in out.error  # the schema-miss path, not an incidental error
diff --git a/tests/workflow/test_worktree.py b/tests/workflow/test_worktree.py
new file mode 100644
index 000000000..53ec307ff
--- /dev/null
+++ b/tests/workflow/test_worktree.py
@@ -0,0 +1,47 @@
+"""Tests for per-agent worktree isolation (#5)."""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+from src.workflow.worktree import agent_worktree, worktree_slug
+
+
+def _git(args, cwd):
+    subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True)
+
+
+def test_worktree_slug():
+    assert worktree_slug("wf_abc123def", "0") == "wf_abc123def-0"
+    # nested call-path keys (dots) become a clean filesystem slug
+    assert worktree_slug("wf_abc123def", "0.1.2") == "wf_abc123def-0-1-2"
+
+
+async def test_agent_worktree_creates_and_removes(tmp_path):
+    repo = tmp_path / "repo"
+    repo.mkdir()
+    _git(["init"], repo)
+    _git(["config", "user.email", "t@example.com"], repo)
+    _git(["config", "user.name", "tester"], repo)
+    (repo / "f.txt").write_text("hi", encoding="utf-8")
+    _git(["add", "."], repo)
+    _git(["commit", "-m", "init"], repo)
+
+    captured = None
+    async with agent_worktree("wf_test1234567", "0", str(repo)) as wt:
+        captured = wt
+        assert wt is not None
+        assert Path(wt).is_dir()
+        assert Path(wt).name == "wf_test1234567-0"  # the wf_<runId>-<idx> slug
+        assert (Path(wt) / "f.txt").exists()  # checked out at HEAD
+
+    assert captured is not None
+    assert not Path(captured).exists()  # removed on context exit
+
+
+async def test_agent_worktree_non_git_yields_none(tmp_path):
+    d = tmp_path / "notgit"
+    d.mkdir()
+    async with agent_worktree("wf_x", "0", str(d)) as wt:
+        assert wt is None  # not a git repo -> best-effort, run in place