PyModel · elkaix · Jun 30, 2026 · Jun 30, 2026 · Jun 30, 2026 · Jun 30, 2026
@@ -15,6 +15,10 @@ GitHub Releases page; `0.8.0` is the new starting line.
 
 ## Unreleased
 
+- Add a built-in `Workflow` tool that runs a deterministic Python script orchestrating
+  multiple subagents via `agent()`, `parallel()`, and `pipeline()`, with live progress,
+  structured-schema output, and one-shot approval.
+
 ## 0.53.0 (2026-06-23)
 
 - **GPT/ChatGPT session re-authentication crash fixed.** A fatal "Could not

@@ -24,6 +24,7 @@ agent:
     - "pythinker_code.tools.worktree:ExitWorktree"
     - "pythinker_code.tools.goal:UpdateGoal"
     - "pythinker_code.tools.progress:Progress"
+    - "pythinker_code.tools.workflow:Workflow"
     - "pythinker_code.tools.suggest:Suggest"
     - "pythinker_code.tools.memory:Memory"
     - "pythinker_code.tools.recall:Recall"

@@ -0,0 +1,244 @@
+import hashlib
+import json
+import re
+from pathlib import Path
+from typing import Any, override
+
+import jsonschema
+from pydantic import BaseModel, Field
+from pythinker_core.message import ContentPart, TextPart
+from pythinker_core.tooling import CallableTool2, ToolError, ToolOk, ToolReturnValue
+
+from pythinker_code.soul import wire_send
+from pythinker_code.soul.agent import Runtime
+from pythinker_code.tools.agent import AgentTool
+from pythinker_code.tools.agent import Params as AgentParams
+from pythinker_code.tools.utils import load_desc
+from pythinker_code.tools.workflow.display import WorkflowSnapshot, render_progress
+from pythinker_code.tools.workflow.engine import (
+    AgentEndEvent,
+    AgentOptions,
+    AgentStartEvent,
+    RunWorkflowHooks,
+    WorkflowScriptError,
+    parse_workflow_script,
+    run_workflow,
+)
+from pythinker_code.utils.logging import logger
+from pythinker_code.wire.types import ProgressNote
+
+_FENCE_RE = re.compile(r"^```(?:py|python)?\s*\n([\s\S]*?)\n```$", re.IGNORECASE)
+_JSON_BLOCK_RE = re.compile(r"```json\s*\n([\s\S]*?)\n```", re.IGNORECASE)
+_SUMMARY_SEP = "\n[summary]\n"
+
+
+class Params(BaseModel):
+    script: str = Field(
+        description=(
+            "Raw Python workflow script (no Markdown fences). First statement must be "
+            '`meta = {"name": ..., "description": ...}`. Must call agent() at least once '
+            "and return a JSON-serializable value."
+        )
+    )
+    args: Any | None = Field(
+        default=None,
+        description="Optional JSON value exposed to the script as the global `args`.",
+    )
+    token_budget: int | None = Field(
+        default=None,
+        description=(
+            "Optional cap on estimated total tokens spent by spawned subagents. When set, "
+            "the script's `budget.remaining()` reaches 0 once the cap is hit and further "
+            "agent() calls raise inside the engine: within parallel()/pipeline(), that one "
+            "call is caught and returns None (logged), but a bare `await agent(...)` propagates "
+            "and fails the whole workflow run. Leave unset for no cap (budget.remaining() "
+            "stays unbounded)."
+        ),
+        ge=1,
+    )
+
+
+class Workflow(CallableTool2[Params]):
+    name: str = "Workflow"
+    params: type[Params] = Params
+    emits_tool_execution_started_after_approval = True
+
+    def __init__(self, runtime: Runtime) -> None:
+        super().__init__(description=load_desc(Path(__file__).parent / "description.md"))
+        self._runtime = runtime
+        self._agent_tool = AgentTool(runtime)
+
+    @override
+    async def __call__(self, params: Params) -> ToolReturnValue:
+        if self._runtime.role != "root":
+            return ToolError(
+                message="Workflow can only run from the root agent; subagents may not "
+                "launch nested workflows.",
+                brief="Not allowed in subagent",
+            )
+
+        script = _strip_fences(params.script)
+        try:
+            meta, _ = parse_workflow_script(script)
+        except WorkflowScriptError as exc:
+            return ToolError(message=str(exc), brief="Invalid workflow script")
+
+        fingerprint = _fingerprint(script, params.args)
+        if self._runtime.approval.is_orchestration_approved(fingerprint):
+            from pythinker_code.soul.toolset import emit_current_tool_execution_started
+
+            emit_current_tool_execution_started()
+        else:
+            approval = await self._runtime.approval.request(
+                self.name,
+                "run workflow orchestration",
+                f"Run workflow `{meta.name}`: {meta.description}",
+            )
+            if not approval:
+                return approval.rejection_error()
+            self._runtime.approval.approve_orchestration(fingerprint)
+
+        snapshot = WorkflowSnapshot(name=meta.name, description=meta.description)
+
+        def emit() -> None:
+            wire_send(ProgressNote(title=f"Workflow {meta.name}", body=render_progress(snapshot)))
+
+        def on_log(message: str) -> None:
+            snapshot.logs.append(message)
+            emit()
+
+        def on_phase(title: str) -> None:
+            snapshot.add_phase(title)
+            emit()
+
+        def on_agent_start(event: AgentStartEvent) -> None:
+            snapshot.start_agent(event.agent_id, event.label, event.phase)
+            emit()
+
+        def on_agent_end(event: AgentEndEvent) -> None:
+            snapshot.end_agent(event.agent_id, error=event.error)
+            emit()
+
+        hooks = RunWorkflowHooks(
+            on_log=on_log,
+            on_phase=on_phase,
+            on_agent_start=on_agent_start,
+            on_agent_end=on_agent_end,
+        )
+
+        try:
+            result = await run_workflow(
+                script,
+                agent_runner=self._agent_runner,
+                args=params.args,
+                cwd=str(self._runtime.work_dir),
+                concurrency=max(1, self._runtime.config.background.max_running_tasks),
+                token_budget=params.token_budget,
+                hooks=hooks,
+            )
+        except WorkflowScriptError as exc:
+            return ToolError(message=str(exc), brief="Invalid workflow script")
+        except Exception as exc:  # noqa: BLE001 - converted to a typed ToolError below
+            # asyncio.CancelledError is a BaseException in 3.12+ and is NOT caught here;
+            # it propagates so the host can abort the tool. The engine already fires
+            # on_agent_end(..., error="cancelled") for in-flight agents before
+            # re-raising, so the incrementally-emitted snapshot already shows them.
+            logger.exception("Workflow run failed")
+            return ToolError(message=f"workflow failed: {exc}", brief="Workflow failed")
+
+        if result.agent_count == 0:
+            return ToolError(
+                message="workflow scripts must call agent() at least once; this workflow "
+                "ran no subagents.",
+                brief="No agents run",
+            )
+
+        output = json.dumps(result.result, indent=2, default=str)
+        return ToolOk(
+            output=output,
+            message=f"Workflow {meta.name} completed with {result.agent_count} agent(s).",
+            brief=f"{result.agent_count} agent(s), {len(result.phases)} phase(s)",
+        )
+
+    async def _agent_runner(self, prompt: str, opts: AgentOptions) -> Any:
+        if opts.schema is None:
+            text, is_error, message = await self._run_child(prompt, opts)
+            if is_error:
+                raise RuntimeError(message or "subagent failed")
+            return text
+
+        # Schema mode: instruct, parse, validate; one retry with the error fed back.
+        instructed = prompt + "\n\n" + _schema_instructions(opts.schema)
+        last_error = ""
+        for _attempt in range(2):
+            text, is_error, message = await self._run_child(instructed, opts)
+            if is_error:
+                raise RuntimeError(message or "subagent failed")
+            value, err = _parse_and_validate(text, opts.schema)
+            if err is None:
+                return value
+            last_error = err
+            instructed = (
+                prompt
+                + "\n\n"
+                + _schema_instructions(opts.schema)
+                + f"\n\nYour previous output was invalid: {err}\nReturn a valid ```json block."
+            )
+        raise RuntimeError(f"subagent never produced schema-valid output: {last_error}")
+
+    async def _run_child(self, prompt: str, opts: AgentOptions) -> tuple[str, bool, str]:
+        child_params = AgentParams(
+            description=opts.label or "workflow agent",
+            prompt=prompt,
+            subagent_type=opts.agent_type or "coder",
+            model=opts.model,
+            run_in_background=False,
+        )
+        ret = await self._agent_tool(child_params)
+        output = ret.output if isinstance(ret.output, str) else _content_text(ret.output)
+        return _extract_summary(output), ret.is_error, ret.message
+
+
+def _strip_fences(script: str) -> str:
+    text = script.strip()
+    match = _FENCE_RE.match(text)
+    return match.group(1).strip() if match else text
+
+
+def _fingerprint(script: str, args: Any) -> str:
+    payload = json.dumps({"script": script, "args": args}, sort_keys=True, default=str)
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _extract_summary(output: str) -> str:
+    if _SUMMARY_SEP in output:
+        return output.split(_SUMMARY_SEP, 1)[1].strip()
+    return output.strip()
+
+
+def _content_text(parts: list[ContentPart]) -> str:
+    # AgentTool returns a string; this is a defensive fallback for ContentPart lists.
+    return "".join(part.text for part in parts if isinstance(part, TextPart))
+
+
+def _schema_instructions(schema: dict[str, Any]) -> str:
+    return (
+        "Final output contract: your final message MUST end with a single fenced "
+        "```json block whose contents validate against this JSON Schema:\n"
+        + json.dumps(schema, indent=2)
+        + "\nDo not write prose after the json block."
+    )
+
+
+def _parse_and_validate(text: str, schema: dict[str, Any]) -> tuple[Any, str | None]:
+    blocks = _JSON_BLOCK_RE.findall(text)
+    raw = blocks[-1] if blocks else text
+    try:
+        value = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        return None, f"output was not valid JSON ({exc})"
+    try:
+        jsonschema.validate(value, schema)
+    except jsonschema.ValidationError as exc:
+        return None, f"output did not match schema ({exc.message})"
+    return value, None
@@ -0,0 +1,22 @@
+Execute a deterministic Python workflow that orchestrates multiple subagents with `agent()`, `parallel()`, and `pipeline()`. Use this only when the user explicitly asks for a workflow, fan-out, or multi-agent orchestration, or when a task decomposes into many independent subtasks worth running concurrently and synthesizing.
+
+`script` is required raw Python (no Markdown fences). `args` (optional) is a JSON value exposed to
+the script as the global `args`. `token_budget` (optional) caps estimated total token spend across
+all spawned subagents — set it when you want the workflow to self-throttle via
+`budget.remaining()`; leave it unset for no cap.
+
+Rules:
+
+- The first statement MUST be `meta = {"name": "short_snake_case", "description": "non-empty description"}`. `meta` must be a literal dict (no function calls or interpolation). `meta["phases"]` is optional documentation; live progress is driven by `phase(title)` at runtime.
+- After `meta`, write plain Python. Do NOT use `import`, `time`, `random`, `datetime`, `os`, `sys`, `open`, `eval`, or `exec` — scripts must be deterministic.
+- The script must call `agent()` at least once. End by `return`-ing a compact JSON-serializable value.
+
+Available globals:
+
+- `agent(prompt, opts=None)` — spawn one subagent; returns its final text, or a validated dict when `opts={"schema": <json-schema-dict>}`. Other opts: `label` (short, 2-5 words), `phase`, `model`, `agent_type` (a built-in subagent type, e.g. `"explore"`, `"coder"`, `"review"`). Always `await` it. A failed agent returns `None` and logs — check for `None` before synthesizing.
+- `parallel([awaitables])` — run awaitables concurrently, results in input order: `await parallel([agent("a"), agent("b")])`. Pass awaitables, NOT functions.
+- `pipeline(items, *stages)` — run each item through sequential stages while items fan out. Each stage is called `(prev, original, index)` and may be sync or async.
+- `phase(title)` — start a progress group. Names may be conditional or built in a loop; do not predeclare speculative phases.
+- `log(message)`, `args` (the optional JSON `args` input), `cwd`, `budget` (`.total`, `.spent()`, `.remaining()` — reflects the `token_budget` tool parameter; unbounded if it was not set).
+
+Include enough context and file paths in each `agent()` prompt — subagents do NOT inherit the parent conversation. Add a final synthesis `agent()` (or a plain return) that combines results into `{ "ok": ..., ... }`.
@@ -0,0 +1,105 @@
+"""Compact progress snapshot + renderer for the Workflow tool.
+
+A port of the reference display module, trimmed to a single text block suitable
+for a `ProgressNote` wire event.
+"""
+
+from __future__ import annotations
+
+_STATUS_ICON = {"running": "●", "done": "✓", "error": "✗", "skipped": "-"}
+
+
+class AgentSnapshot:
+    __slots__ = ("id", "label", "phase", "status")
+
+    def __init__(self, id: int, label: str, phase: str | None) -> None:
+        self.id = id
+        self.label = label
+        self.phase = phase
+        self.status = "running"
+
+
+class WorkflowSnapshot:
+    __slots__ = ("name", "description", "phases", "current_phase", "agents", "logs")
+
+    def __init__(self, name: str, description: str = "") -> None:
+        self.name = name
+        self.description = description
+        self.phases: list[str] = []
+        self.current_phase: str | None = None
+        self.agents: list[AgentSnapshot] = []
+        self.logs: list[str] = []
+
+    def add_phase(self, title: str | None) -> None:
+        if not title:
+            return
+        self.current_phase = title
+        if title not in self.phases:
+            self.phases.append(title)
+
+    def start_agent(self, agent_id: int, label: str, phase: str | None) -> AgentSnapshot:
+        self.add_phase(phase)
+        agent = AgentSnapshot(agent_id, label, phase)
+        self.agents.append(agent)
+        return agent
+
+    def end_agent(self, agent_id: int, *, error: str | None = None) -> None:
+        # Matched by the engine's stable dispatch id, not by label: two
+        # concurrent agents may share the same caller-supplied label, and a
+        # label-based reverse search can mark the wrong entry done/error if
+        # they finish out of dispatch order.
+        for agent in self.agents:
+            if agent.id == agent_id and agent.status == "running":
+                agent.status = "error" if error else "done"
+                return
+
+    def mark_running_skipped(self) -> None:
+        for agent in self.agents:
+            if agent.status == "running":
+                agent.status = "skipped"
+
+    @property
+    def running_count(self) -> int:
+        return sum(1 for a in self.agents if a.status == "running")
+
+    @property
+    def done_count(self) -> int:
+        return sum(1 for a in self.agents if a.status == "done")
+
+    @property
+    def error_count(self) -> int:
+        return sum(1 for a in self.agents if a.status == "error")
+
+    @property
+    def skipped_count(self) -> int:
+        return sum(1 for a in self.agents if a.status == "skipped")
+
+
+def render_progress(snapshot: WorkflowSnapshot, max_agents: int = 6, max_logs: int = 3) -> str:
+    state = ""
+    if snapshot.error_count:
+        state = f", {snapshot.error_count} errors"
+    elif snapshot.running_count:
+        state = f", {snapshot.running_count} running"
+    lines = [
+        f"◆ Workflow: {snapshot.name} ({snapshot.done_count}/{len(snapshot.agents)} done{state})"
+    ]
+    rendered: set[int] = set()
+    phase_order = list(snapshot.phases)
+    if snapshot.current_phase and snapshot.current_phase not in phase_order:
+        phase_order.append(snapshot.current_phase)
+    for phase in phase_order:
+        agents = [a for a in snapshot.agents if a.phase == phase]
+        if not agents and snapshot.current_phase != phase:
+            continue
+        done = sum(1 for a in agents if a.status == "done")
+        lines.append(f"  {phase} {done}/{len(agents)}")
+        for agent in agents[-max_agents:]:
+            rendered.add(agent.id)
+            lines.append(f"    #{agent.id} {_STATUS_ICON.get(agent.status, '?')} {agent.label}")
+    unphased = [a for a in snapshot.agents if a.id not in rendered]
+    for agent in unphased[-max_agents:]:
+        lines.append(f"    #{agent.id} {_STATUS_ICON.get(agent.status, '?')} {agent.label}")
+    for message in snapshot.logs[-max_logs:]:
+        lines.append(f"  log: {message}")
+    return "\n".join(lines)