Workflow polish: result delivery, retry, worktree, /workflows TUI, integration test

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`

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.
 

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",
 ]

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,

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()
 

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")