feat: add dynamic workflows

frontend/terminal/src/App.tsx

@@ -2,6 +2,7 @@ import React, {useDeferredValue, useEffect, useMemo, useRef, useState} from 'rea
 import {Box, Text, useApp, useInput} from 'ink';
 
 import {readClipboardImage, type ImageAttachment} from './clipboardImage.js';
+import {filterCommandItems} from './commandPalette.js';
 import {CommandPicker} from './components/CommandPicker.js';
 import {ConversationView} from './components/ConversationView.js';
 import {ModalHost} from './components/ModalHost.js';
@@ -145,13 +146,7 @@ function AppInner({config}: {config: FrontendConfig}): React.JSX.Element {
 	}, [deferredTranscript]);
 
 	// Command hints
-	const commandHints = useMemo(() => {
-		const value = input.trim();
-		if (!value.startsWith('/')) {
-			return [] as string[];
-		}
-		return session.commands.filter((cmd) => cmd.startsWith(value)).slice(0, 10);
-	}, [session.commands, input]);
+	const commandHints = useMemo(() => filterCommandItems(session.commandItems, input), [session.commandItems, input]);
 
 	const showPicker = commandHints.length > 0 && !session.busy && !session.modal && !selectModal;
 	const outputStyle = String(session.status.output_style ?? 'default');
@@ -392,8 +387,8 @@ function AppInner({config}: {config: FrontendConfig}): React.JSX.Element {
 				const selected = commandHints[pickerIndex];
 				if (selected) {
 					setInput('');
-					if (!handleCommand(selected)) {
-						onSubmit(selected);
+					if (!handleCommand(selected.name)) {
+						onSubmit(selected.name);
 					}
 				}
 				return;
@@ -405,7 +400,7 @@ function AppInner({config}: {config: FrontendConfig}): React.JSX.Element {
 					// the user can hit Enter immediately to run it, or keep
 					// typing to add args. The trailing space made it look like
 					// Tab was "committing" with a token, which broke the flow.
-					setInput(selected);
+					setInput(selected.name);
 				}
 				return;
 			}

frontend/terminal/src/commandPalette.test.ts

@@ -0,0 +1,58 @@
+import assert from 'node:assert/strict';
+import test from 'node:test';
+
+import {filterCommandItems, normalizeCommandItems} from './commandPalette.js';
+import type {CommandItemPayload} from './types.js';
+
+const commands: CommandItemPayload[] = [
+	{
+		name: '/help',
+		description: 'Show available commands',
+		aliases: [],
+	},
+	{
+		name: '/workflows',
+		description: 'List, run, pause, resume, and save dynamic workflows',
+		aliases: ['/wf'],
+	},
+	{
+		name: '/permissions',
+		description: 'Show or update permission mode',
+		aliases: [],
+	},
+];
+
+test('normalizes structured command metadata and legacy command names', () => {
+	const items = normalizeCommandItems(
+		['help', '/status'],
+		[{name: 'help', description: 'Show help', aliases: ['h']}],
+	);
+
+	assert.deepEqual(items, [
+		{name: '/help', description: 'Show help', aliases: ['/h']},
+		{name: '/status', description: '', aliases: []},
+	]);
+});
+
+test('shows the palette for bare slash input', () => {
+	const items = filterCommandItems(commands, '/');
+
+	assert.deepEqual(items.map((item) => item.name), ['/help', '/workflows', '/permissions']);
+});
+
+test('matches aliases but returns the canonical command name', () => {
+	const items = filterCommandItems(commands, '/wf');
+
+	assert.equal(items[0].name, '/workflows');
+	assert.equal(items[0].matchedAlias, '/wf');
+});
+
+test('supports fuzzy command search', () => {
+	const items = filterCommandItems(commands, '/wrkflw');
+
+	assert.equal(items[0].name, '/workflows');
+});
+
+test('hides command palette after the command token has arguments', () => {
+	assert.deepEqual(filterCommandItems(commands, '/workflows run'), []);
+});

frontend/terminal/src/commandPalette.ts

@@ -0,0 +1,154 @@
+import type {CommandItemPayload} from './types.js';
+
+export type CommandPaletteItem = {
+	name: string;
+	description: string;
+	aliases: string[];
+	matchedAlias?: string;
+	score: number;
+};
+
+function withSlash(value: string): string {
+	const trimmed = value.trim();
+	if (!trimmed) {
+		return '';
+	}
+	return trimmed.startsWith('/') ? trimmed : `/${trimmed}`;
+}
+
+function commandToken(value: string): string {
+	return withSlash(value).slice(1).toLowerCase();
+}
+
+export function normalizeCommandItems(
+	commands: string[],
+	commandItems: CommandItemPayload[] | null | undefined,
+): CommandItemPayload[] {
+	const byName = new Map<string, CommandItemPayload>();
+
+	for (const item of commandItems ?? []) {
+		const name = withSlash(String(item.name ?? ''));
+		if (!name) {
+			continue;
+		}
+		byName.set(name, {
+			name,
+			description: item.description ?? '',
+			aliases: (item.aliases ?? []).map((alias) => withSlash(String(alias))).filter(Boolean),
+		});
+	}
+
+	for (const command of commands) {
+		const name = withSlash(command);
+		if (!name || byName.has(name)) {
+			continue;
+		}
+		byName.set(name, {name, description: '', aliases: []});
+	}
+
+	return [...byName.values()];
+}
+
+export function filterCommandItems(
+	commandItems: CommandItemPayload[],
+	input: string,
+	limit = 12,
+): CommandPaletteItem[] {
+	const trimmed = input.trimStart();
+	if (!trimmed.startsWith('/')) {
+		return [];
+	}
+
+	const rawQuery = trimmed.slice(1);
+	if (/\s/.test(rawQuery)) {
+		return [];
+	}
+
+	const query = rawQuery.toLowerCase();
+	const scored = commandItems
+		.map((item, index) => scoreItem(item, query, index))
+		.filter((item): item is CommandPaletteItem => item !== null)
+		.sort((a, b) => a.score - b.score || a.name.localeCompare(b.name));
+
+	return scored.slice(0, limit);
+}
+
+function scoreItem(item: CommandItemPayload, query: string, index: number): CommandPaletteItem | null {
+	const name = withSlash(item.name);
+	if (!name) {
+		return null;
+	}
+	const aliases = (item.aliases ?? []).map((alias) => withSlash(alias)).filter(Boolean);
+	const description = item.description ?? '';
+	if (!query) {
+		return {name, description, aliases, score: index};
+	}
+
+	const candidates = [name, ...aliases];
+	let bestScore = Number.POSITIVE_INFINITY;
+	let matchedAlias: string | undefined;
+
+	for (const candidate of candidates) {
+		const token = commandToken(candidate);
+		const score = scoreToken(token, query);
+		if (score < bestScore) {
+			bestScore = score;
+			matchedAlias = candidate === name ? undefined : candidate;
+		}
+	}
+
+	const descriptionIndex = description.toLowerCase().indexOf(query);
+	if (descriptionIndex >= 0) {
+		bestScore = Math.min(bestScore, 80 + descriptionIndex);
+	}
+
+	if (!Number.isFinite(bestScore)) {
+		return null;
+	}
+
+	return {
+		name,
+		description,
+		aliases,
+		matchedAlias,
+		score: bestScore + index / 1000,
+	};
+}
+
+function scoreToken(token: string, query: string): number {
+	if (token === query) {
+		return 0;
+	}
+	if (token.startsWith(query)) {
+		return 10 + token.length - query.length;
+	}
+	const segmentIndex = token
+		.split(/[-_]/)
+		.findIndex((segment) => segment.startsWith(query));
+	if (segmentIndex >= 0) {
+		return 25 + segmentIndex;
+	}
+	const includesIndex = token.indexOf(query);
+	if (includesIndex >= 0) {
+		return 35 + includesIndex;
+	}
+	const fuzzyScore = fuzzyMatchScore(token, query);
+	if (fuzzyScore !== null) {
+		return 55 + fuzzyScore;
+	}
+	return Number.POSITIVE_INFINITY;
+}
+
+function fuzzyMatchScore(token: string, query: string): number | null {
+	let tokenIndex = 0;
+	let score = 0;
+	for (const char of query) {
+		const foundAt = token.indexOf(char, tokenIndex);
+		if (foundAt < 0) {
+			return null;
+		}
+		score += foundAt - tokenIndex;
+		tokenIndex = foundAt + 1;
+	}
+	return score;
+}

frontend/terminal/src/components/CommandPicker.tsx

@@ -1,11 +1,13 @@
 import React from 'react';
 import {Box, Text} from 'ink';
 
+import type {CommandPaletteItem} from '../commandPalette.js';
+
 function CommandPickerInner({
 	hints,
 	selectedIndex,
 }: {
-	hints: string[];
+	hints: CommandPaletteItem[];
 	selectedIndex: number;
 }): React.JSX.Element | null {
 	if (hints.length === 0) {
@@ -18,18 +20,33 @@ function CommandPickerInner({
 			{hints.map((hint, i) => {
 				const isSelected = i === selectedIndex;
 				return (
-					<Box key={hint}>
-						<Text color={isSelected ? 'cyan' : undefined} bold={isSelected}>
-							{isSelected ? '\u276F ' : '  '}
-							{hint}
-						</Text>
-						{isSelected ? <Text dimColor> [enter]</Text> : null}
+					<Box key={hint.name} flexDirection="column">
+						<Box>
+							<Text color={isSelected ? 'cyan' : undefined} bold={isSelected}>
+								{isSelected ? '\u276F ' : '  '}
+								{hint.name}
+							</Text>
+							{hint.matchedAlias ? <Text dimColor>  alias {hint.matchedAlias}</Text> : null}
+							{isSelected ? <Text dimColor>  [enter]</Text> : null}
+						</Box>
+						{isSelected && hint.description ? (
+							<Box paddingLeft={2}>
+								<Text dimColor>{truncate(hint.description, 86)}</Text>
+							</Box>
+						) : null}
 					</Box>
 				);
 			})}
-			<Text dimColor> {'\u2191\u2193'} navigate{'  '}{'\u23CE'} select{'  '}esc dismiss</Text>
+			<Text dimColor> {'\u2191\u2193'} navigate{'  '}tab complete{'  '}{'\u23CE'} run{'  '}esc dismiss</Text>
 		</Box>
 	);
 }
 
 export const CommandPicker = React.memo(CommandPickerInner);
+
+function truncate(value: string, maxLength: number): string {
+	if (value.length <= maxLength) {
+		return value;
+	}
+	return `${value.slice(0, Math.max(0, maxLength - 1))}…`;
+}

frontend/terminal/src/hooks/useBackendSession.ts

@@ -5,6 +5,7 @@ import readline from 'node:readline';
 import type {
 	BackendEvent,
 	BridgeSessionSnapshot,
+	CommandItemPayload,
 	FrontendConfig,
 	McpServerSnapshot,
 	SelectOptionPayload,
@@ -13,6 +14,7 @@ import type {
 	TaskSnapshot,
 	TranscriptItem,
 } from '../types.js';
+import {normalizeCommandItems} from '../commandPalette.js';
 
 const PROTOCOL_PREFIX = 'OHJSON:';
 const ASSISTANT_DELTA_FLUSH_MS = 50;
@@ -27,6 +29,7 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 	const [status, setStatus] = useState<Record<string, unknown>>({});
 	const [tasks, setTasks] = useState<TaskSnapshot[]>([]);
 	const [commands, setCommands] = useState<string[]>([]);
+	const [commandItems, setCommandItems] = useState<CommandItemPayload[]>([]);
 	const [mcpServers, setMcpServers] = useState<McpServerSnapshot[]>([]);
 	const [bridgeSessions, setBridgeSessions] = useState<BridgeSessionSnapshot[]>([]);
 	const [modal, setModal] = useState<Record<string, unknown> | null>(null);
@@ -191,7 +194,9 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			startTransition(() => {
 				setTasks(event.tasks ?? []);
 			});
-			setCommands(event.commands ?? []);
+			const nextCommands = event.commands ?? [];
+			setCommands(nextCommands);
+			setCommandItems(normalizeCommandItems(nextCommands, event.command_items));
 			const mcpSnapshot = stableStringify(event.mcp_servers ?? []);
 			lastMcpSnapshotRef.current = mcpSnapshot;
 			startTransition(() => {
@@ -441,6 +446,7 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			status,
 			tasks,
 			commands,
+			commandItems,
 			mcpServers,
 			bridgeSessions,
 			modal,
@@ -457,6 +463,6 @@ export function useBackendSession(config: FrontendConfig, onExit: (code?: number
 			setBusyLabel,
 			sendRequest,
 		}),
-		[assistantBuffer, bridgeSessions, busy, busyLabel, commands, mcpServers, modal, ready, selectRequest, status, swarmNotifications, swarmTeammates, tasks, todoMarkdown, transcript]
+		[assistantBuffer, bridgeSessions, busy, busyLabel, commandItems, commands, mcpServers, modal, ready, selectRequest, status, swarmNotifications, swarmTeammates, tasks, todoMarkdown, transcript]
 	);
 }

frontend/terminal/src/types.ts

@@ -79,6 +79,7 @@ export type BackendEvent = {
 	mcp_servers?: McpServerSnapshot[] | null;
 	bridge_sessions?: BridgeSessionSnapshot[] | null;
 	commands?: string[] | null;
+	command_items?: CommandItemPayload[] | null;
 	modal?: Record<string, unknown> | null;
 	select_options?: SelectOptionPayload[] | null;
 	tool_name?: string | null;
@@ -96,3 +97,9 @@ export type BackendEvent = {
 	swarm_teammates?: SwarmTeammateSnapshot[] | null;
 	swarm_notifications?: SwarmNotificationSnapshot[] | null;
 };
+
+export type CommandItemPayload = {
+	name: string;
+	description?: string | null;
+	aliases?: string[] | null;
+};

src/openharness/cli.py

@@ -2153,7 +2153,7 @@ def main(
     effort: str | None = typer.Option(
         None,
         "--effort",
-        help="Effort level for the session (low, medium, high, xhigh/max)",
+        help="Effort level for the session (low, medium, high, xhigh/max, ultracode)",
         rich_help_panel="Model & Effort",
     ),
     verbose: bool = typer.Option(

src/openharness/commands/registry.py

@@ -92,6 +92,8 @@ from openharness.skills import load_skill_registry
 from openharness.skills.types import SkillDefinition
 from openharness.tasks import get_task_manager
 from openharness.plugins.types import PluginCommandDefinition
+from openharness.tools.workflow_tool import format_workflow_run, list_saved_workflows_text
+from openharness.workflows.manager import get_workflow_manager
 
 if TYPE_CHECKING:
     from openharness.config.settings import ProviderProfile
@@ -1237,8 +1239,8 @@ def create_default_command_registry(
             return CommandResult(message=f"Reasoning effort: {current}")
         if value == "max":
             value = "xhigh"
-        if value not in {"low", "medium", "high", "xhigh"}:
-            return CommandResult(message="Usage: /effort [show|low|medium|high|xhigh]")
+        if value not in {"low", "medium", "high", "xhigh", "ultracode"}:
+            return CommandResult(message="Usage: /effort [show|low|medium|high|xhigh|ultracode]")
         settings.effort = value
         save_settings(settings)
         context.engine.set_effort(value)
@@ -2048,6 +2050,70 @@ def create_default_command_registry(
             )
         )
 
+    async def _workflows_handler(args: str, context: CommandContext) -> CommandResult:
+        manager = get_workflow_manager()
+        tokens = args.split(maxsplit=2)
+        if not tokens or tokens[0] in {"list", "runs"}:
+            runs = manager.list_runs(limit=20)
+            if not runs:
+                return CommandResult(message="No workflow runs.")
+            return CommandResult(
+                message="\n".join(
+                    f"{run.id} {run.status} {run.name} agents={run.agent_count} cached={run.cached_count}"
+                    for run in runs
+                )
+            )
+        if tokens[0] in {"saved", "list-saved"}:
+            return CommandResult(message=list_saved_workflows_text(context.cwd))
+        if tokens[0] == "run" and len(tokens) >= 2:
+            name_or_path = args[len("run ") :].strip()
+            try:
+                run = await manager.start_saved(
+                    cwd=context.cwd,
+                    name_or_path=name_or_path,
+                    model=context.engine.model,
+                )
+            except ValueError as exc:
+                return CommandResult(message=str(exc))
+            return CommandResult(message=f"Started workflow {run.id}: {run.name}")
+        if tokens[0] in {"show", "output"} and len(tokens) >= 2:
+            run = manager.get(tokens[1])
+            if run is None:
+                return CommandResult(message=f"Workflow run not found: {tokens[1]}")
+            return CommandResult(message=format_workflow_run(run, include_output=tokens[0] == "output"))
+        if tokens[0] == "pause" and len(tokens) >= 2:
+            try:
+                run = await manager.pause(tokens[1])
+            except ValueError as exc:
+                return CommandResult(message=str(exc))
+            return CommandResult(message=f"Paused workflow {run.id}: {run.status}")
+        if tokens[0] == "stop" and len(tokens) >= 2:
+            try:
+                run = await manager.stop(tokens[1])
+            except ValueError as exc:
+                return CommandResult(message=str(exc))
+            return CommandResult(message=f"Stopped workflow {run.id}: {run.status}")
+        if tokens[0] == "resume" and len(tokens) >= 2:
+            try:
+                run = await manager.resume(run_id=tokens[1], model=context.engine.model)
+            except ValueError as exc:
+                return CommandResult(message=str(exc))
+            return CommandResult(message=f"Resumed workflow {run.id}: {run.status}")
+        if tokens[0] == "save" and len(tokens) >= 3:
+            run_id = tokens[1]
+            name = tokens[2]
+            try:
+                path = manager.save_run_script(cwd=context.cwd, run_id=run_id, name=name)
+            except ValueError as exc:
+                return CommandResult(message=str(exc))
+            return CommandResult(message=f"Saved workflow {run_id} as {path}")
+        return CommandResult(
+            message=(
+                "Usage: /workflows "
+                "[list|saved|run NAME_OR_PATH|show ID|output ID|pause ID|resume ID|stop ID|save ID NAME]"
+            )
+        )
+
     async def _autopilot_handler(args: str, context: CommandContext) -> CommandResult:
         store = RepoAutopilotStore(context.cwd)
         tokens = args.split()
@@ -2450,6 +2516,7 @@ def create_default_command_registry(
     registry.register(SlashCommand("turns", "Show or update maximum agentic turn count", _turns_handler))
     registry.register(SlashCommand("continue", "Continue the previous tool loop if it was interrupted", _continue_handler))
     registry.register(SlashCommand("stop", "Interrupt the running turn from TUI/ohmo channels", _stop_handler))
+    registry.register(SlashCommand("workflows", "List, run, pause, resume, and save dynamic workflows", _workflows_handler))
     registry.register(
         SlashCommand(
             "provider",

src/openharness/config/paths.py

@@ -82,6 +82,27 @@ def get_tasks_dir() -> Path:
     return tasks_dir
 
 
+def get_workflows_dir() -> Path:
+    """Return the workflow runtime storage directory."""
+    workflows_dir = get_data_dir() / "workflows"
+    workflows_dir.mkdir(parents=True, exist_ok=True)
+    return workflows_dir
+
+
+def get_user_workflows_dir() -> Path:
+    """Return the user-level saved workflow directory."""
+    workflows_dir = get_config_dir() / "workflows"
+    workflows_dir.mkdir(parents=True, exist_ok=True)
+    return workflows_dir
+
+
+def get_project_workflows_dir(cwd: str | Path) -> Path:
+    """Return the project-level saved workflow directory."""
+    workflows_dir = get_project_config_dir(cwd) / "workflows"
+    workflows_dir.mkdir(parents=True, exist_ok=True)
+    return workflows_dir
+
+
 def get_feedback_dir() -> Path:
     """Return the feedback storage directory."""
     feedback_dir = get_data_dir() / "feedback"

src/openharness/engine/query.py

@@ -974,6 +974,7 @@ async def _execute_tool_call(
             metadata={
                 "tool_registry": context.tool_registry,
                 "ask_user_prompt": context.ask_user_prompt,
+                "model": context.model,
                 **(context.tool_metadata or {}),
             },
             hook_executor=context.hook_executor,

src/openharness/tools/__init__.py

@@ -43,6 +43,7 @@ from openharness.tools.todo_write_tool import TodoWriteTool
 from openharness.tools.tool_search_tool import ToolSearchTool
 from openharness.tools.web_fetch_tool import WebFetchTool
 from openharness.tools.web_search_tool import WebSearchTool
+from openharness.tools.workflow_tool import WorkflowCreateTool, WorkflowRunTool, WorkflowStatusTool
 
 
 def create_default_tool_registry(mcp_manager=None) -> ToolRegistry:
@@ -88,6 +89,9 @@ def create_default_tool_registry(mcp_manager=None) -> ToolRegistry:
         SendMessageTool(),
         TeamCreateTool(),
         TeamDeleteTool(),
+        WorkflowCreateTool(),
+        WorkflowRunTool(),
+        WorkflowStatusTool(),
     ):
         registry.register(tool)
     if mcp_manager is not None:

src/openharness/tools/workflow_tool.py

@@ -0,0 +1,174 @@
+"""Tools for creating and running dynamic workflows."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from pydantic import BaseModel, Field
+
+from openharness.tools.base import BaseTool, ToolExecutionContext, ToolResult
+from openharness.workflows.manager import get_workflow_manager, list_saved_workflows
+
+
+WORKFLOW_DSL_HELP = """\
+Dynamic workflow scripts are deterministic JavaScript. The runtime injects:
+- workflow(name, async () => result): name and run the workflow.
+- agent(prompt, opts): run one OpenHarness subagent. opts may include phase, model, name, team, system_prompt, permissions.
+- parallel(items, async (item, index) => result): barrier parallel map.
+- pipeline(items, [stage1, stage2]): each item independently passes through all stages without a stage barrier.
+- phase(name, async () => result): group progress and token accounting.
+- log(message): append progress.
+
+The script cannot directly use filesystem, shell, network, Date, Math.random, require, import, process, or fetch.
+Agents perform all file/shell work through normal OpenHarness tools and permissions.
+Return the final workflow result from workflow(...).
+"""
+
+
+class WorkflowCreateInput(BaseModel):
+    """Arguments for creating a reusable workflow."""
+
+    name: str = Field(description="Short workflow name, used as the saved command name")
+    script: str = Field(description="Deterministic JavaScript workflow script using workflow/agent/parallel/pipeline/phase/log")
+    description: str = Field(default="", description="Short description of what the workflow does")
+    scope: str = Field(default="project", description="Save scope: project or user")
+    run_immediately: bool = Field(default=True, description="Start a run after saving the workflow")
+    wait_for_completion: bool = Field(default=False, description="Wait for the workflow run to finish before returning")
+    max_concurrency: int | None = Field(default=None, description="Optional concurrent agent cap, default min(16, cpu-2)")
+    max_agents: int = Field(default=1000, description="Hard cap for agent() calls in one run")
+
+
+class WorkflowRunInput(BaseModel):
+    """Arguments for starting a saved workflow."""
+
+    name_or_path: str = Field(description="Saved workflow name or path to a .js workflow script")
+    wait_for_completion: bool = Field(default=False, description="Wait for the workflow run to finish before returning")
+    max_concurrency: int | None = Field(default=None, description="Optional concurrent agent cap")
+    max_agents: int = Field(default=1000, description="Hard cap for agent() calls in one run")
+
+
+class WorkflowStatusInput(BaseModel):
+    """Arguments for inspecting a workflow run."""
+
+    run_id: str = Field(description="Workflow run ID returned by workflow_create or workflow_run")
+
+
+class WorkflowCreateTool(BaseTool):
+    """Create and optionally run a dynamic workflow."""
+
+    name = "workflow_create"
+    description = "Create a saved dynamic workflow and optionally run it.\n\n" + WORKFLOW_DSL_HELP
+    input_model = WorkflowCreateInput
+
+    async def execute(self, arguments: WorkflowCreateInput, context: ToolExecutionContext) -> ToolResult:
+        manager = get_workflow_manager()
+        scope = arguments.scope if arguments.scope in {"project", "user"} else "project"
+        saved_path = manager.save_script(
+            cwd=context.cwd,
+            name=arguments.name,
+            script=arguments.script,
+            scope=scope,
+        )
+        lines = [f"Saved workflow {arguments.name!r} to {saved_path}"]
+        metadata = {"workflow_path": str(saved_path)}
+        if arguments.run_immediately:
+            run = await manager.start_inline(
+                cwd=context.cwd,
+                name=arguments.name,
+                script=arguments.script,
+                description=arguments.description,
+                model=_active_model(context),
+                wait=arguments.wait_for_completion,
+                max_concurrency=arguments.max_concurrency,
+                max_agents=arguments.max_agents,
+            )
+            lines.append(_format_run_summary(run))
+            metadata["run_id"] = run.id
+            metadata["status"] = run.status
+        return ToolResult(output="\n".join(lines), metadata=metadata)
+
+
+class WorkflowRunTool(BaseTool):
+    """Run a saved dynamic workflow."""
+
+    name = "workflow_run"
+    description = "Run a saved dynamic workflow by name or .js path.\n\n" + WORKFLOW_DSL_HELP
+    input_model = WorkflowRunInput
+
+    async def execute(self, arguments: WorkflowRunInput, context: ToolExecutionContext) -> ToolResult:
+        try:
+            run = await get_workflow_manager().start_saved(
+                cwd=context.cwd,
+                name_or_path=arguments.name_or_path,
+                model=_active_model(context),
+                wait=arguments.wait_for_completion,
+                max_concurrency=arguments.max_concurrency,
+                max_agents=arguments.max_agents,
+            )
+        except ValueError as exc:
+            return ToolResult(output=str(exc), is_error=True)
+        return ToolResult(output=_format_run_summary(run), metadata={"run_id": run.id, "status": run.status})
+
+
+class WorkflowStatusTool(BaseTool):
+    """Inspect one dynamic workflow run."""
+
+    name = "workflow_status"
+    description = "Inspect the status, phases, agents, and final output of a dynamic workflow run."
+    input_model = WorkflowStatusInput
+
+    async def execute(self, arguments: WorkflowStatusInput, context: ToolExecutionContext) -> ToolResult:
+        del context
+        run = get_workflow_manager().get(arguments.run_id)
+        if run is None:
+            return ToolResult(output=f"Workflow run not found: {arguments.run_id}", is_error=True)
+        return ToolResult(output=format_workflow_run(run, include_output=True), metadata={"run_id": run.id, "status": run.status})
+
+
+def format_workflow_run(run, *, include_output: bool = False) -> str:
+    """Render a workflow run for tools and slash commands."""
+    lines = [
+        _format_run_summary(run),
+        f"script: {run.script_path}",
+        f"agents: {run.agent_count} live, {run.cached_count} cached",
+    ]
+    if run.error:
+        lines.append(f"error: {run.error}")
+    if run.phases:
+        lines.append("phases:")
+        for phase in run.phases.values():
+            lines.append(
+                f"- {phase.name}: {phase.status}, agents={phase.agent_count}, cached={phase.cached_count}"
+            )
+    if run.agents:
+        lines.append("agent calls:")
+        for agent in run.agents.values():
+            suffix = f", task={agent.task_id}" if agent.task_id else ""
+            lines.append(f"- {agent.id}: {agent.status}, phase={agent.phase or '-'}{suffix}")
+    if run.logs:
+        lines.append("logs:")
+        for message in run.logs[-10:]:
+            lines.append(f"- {message}")
+    if include_output and run.result is not None:
+        lines.append("result:")
+        lines.append(str(run.result))
+    return "\n".join(lines)
+
+
+def list_saved_workflows_text(cwd: str | Path) -> str:
+    """Render saved workflows visible from a cwd."""
+    paths = list_saved_workflows(cwd)
+    if not paths:
+        return "No saved workflows."
+    return "\n".join(str(path) for path in paths)
+
+
+def _format_run_summary(run) -> str:
+    return f"Workflow {run.id} {run.status}: {run.name}"
+
+
+def _active_model(context: ToolExecutionContext) -> str | None:
+    model = context.metadata.get("model")
+    if isinstance(model, str) and model.strip():
+        return model.strip()
+    return None

src/openharness/ui/backend_host.py

@@ -34,7 +34,13 @@ from openharness.engine.stream_events import (
 from openharness.output_styles import load_output_styles
 from openharness.tasks import get_task_manager
 from openharness.ui.coordinator_drain import drain_coordinator_async_agents
-from openharness.ui.protocol import BackendEvent, FrontendImageAttachment, FrontendRequest, TranscriptItem
+from openharness.ui.protocol import (
+    BackendEvent,
+    CommandSnapshot,
+    FrontendImageAttachment,
+    FrontendRequest,
+    TranscriptItem,
+)
 from openharness.ui.runtime import build_runtime, close_runtime, handle_line, start_runtime
 from openharness.services.session_backend import SessionBackend
 
@@ -115,11 +121,20 @@ class ReactBackendHost:
             include_project_memory=self._config.include_project_memory,
         )
         await start_runtime(self._bundle)
+        command_items = [
+            CommandSnapshot(
+                name=f"/{command.name}",
+                description=command.description,
+                aliases=[f"/{alias}" for alias in command.aliases],
+            )
+            for command in self._bundle.commands.list_commands()
+        ]
         await self._emit(
             BackendEvent.ready(
                 self._bundle.app_state.get(),
                 get_task_manager().list_tasks(),
-                [f"/{command.name}" for command in self._bundle.commands.list_commands()],
+                [item.name for item in command_items],
+                command_items,
             )
         )
         await self._emit(self._status_snapshot())
@@ -577,6 +592,7 @@ class ReactBackendHost:
                 {"value": "medium", "label": "Medium", "description": "Balanced reasoning", "active": settings.effort == "medium"},
                 {"value": "high", "label": "High", "description": "Deepest reasoning", "active": settings.effort == "high"},
                 {"value": "xhigh", "label": "XHigh", "description": "Extra high reasoning", "active": settings.effort == "xhigh"},
+                {"value": "ultracode", "label": "Ultracode", "description": "Prefer dynamic workflows for substantive tasks", "active": settings.effort == "ultracode"},
             ]
             await self._emit(
                 BackendEvent(

src/openharness/ui/protocol.py

@@ -87,6 +87,14 @@ class TaskSnapshot(BaseModel):
         )
 
 
+class CommandSnapshot(BaseModel):
+    """UI-safe slash command metadata."""
+
+    name: str
+    description: str = ""
+    aliases: list[str] = Field(default_factory=list)
+
+
 class BackendEvent(BaseModel):
     """One event sent from the Python backend to the React frontend."""
 
@@ -118,6 +126,7 @@ class BackendEvent(BaseModel):
     mcp_servers: list[dict[str, Any]] | None = None
     bridge_sessions: list[dict[str, Any]] | None = None
     commands: list[str] | None = None
+    command_items: list[CommandSnapshot] | None = None
     modal: dict[str, Any] | None = None
     tool_name: str | None = None
     tool_input: dict[str, Any] | None = None
@@ -140,6 +149,7 @@ class BackendEvent(BaseModel):
         state: AppState,
         tasks: list[TaskRecord],
         commands: list[str],
+        command_items: list[CommandSnapshot] | None = None,
     ) -> "BackendEvent":
         return cls(
             type="ready",
@@ -148,6 +158,7 @@ class BackendEvent(BaseModel):
             mcp_servers=[],
             bridge_sessions=[],
             commands=commands,
+            command_items=command_items,
         )
 
     @classmethod

src/openharness/ui/runtime.py

@@ -514,6 +514,21 @@ def _truncate(text: str, limit: int) -> str:
     return text[:limit] + "…"
 
 
+def _maybe_ultracode_prompt(prompt: str, effort: str | None) -> str:
+    """Ask the model to use dynamic workflows when ultracode is active."""
+    if (effort or "").strip().lower() != "ultracode":
+        return prompt
+    if "workflow_create" in prompt or "workflow_run" in prompt:
+        return prompt
+    return (
+        "Use dynamic workflows for this substantive task when it benefits from "
+        "parallel subagents or independent verification. Prefer the workflow_create "
+        "tool: write a deterministic JavaScript workflow using workflow(), agent(), "
+        "parallel(), pipeline(), phase(), and log(), run it, and summarize the run.\n\n"
+        f"Task:\n{prompt}"
+    )
+
+
 def _format_pending_tool_results(messages: list[ConversationMessage]) -> str | None:
     """Render a compact summary when we stop after tool execution but before the follow-up model turn."""
     if not messages:
@@ -732,6 +747,8 @@ async def handle_line(
     settings = bundle.current_settings()
     if bundle.enforce_max_turns:
         bundle.engine.set_max_turns(settings.max_turns)
+    if user_message is None:
+        line = _maybe_ultracode_prompt(line, settings.effort)
     latest_user_prompt = line or (user_message.text if user_message is not None else "")
     system_prompt = build_runtime_system_prompt(
         settings,

src/openharness/workflows/__init__.py

@@ -0,0 +1,20 @@
+"""Dynamic workflow runtime."""
+
+from openharness.workflows.manager import WorkflowManager, get_workflow_manager
+from openharness.workflows.runtime import WorkflowRuntime
+from openharness.workflows.store import WorkflowStore
+from openharness.workflows.types import (
+    WorkflowAgentRecord,
+    WorkflowPhaseRecord,
+    WorkflowRunRecord,
+)
+
+__all__ = [
+    "WorkflowAgentRecord",
+    "WorkflowManager",
+    "WorkflowPhaseRecord",
+    "WorkflowRunRecord",
+    "WorkflowRuntime",
+    "WorkflowStore",
+    "get_workflow_manager",
+]

src/openharness/workflows/manager.py

@@ -0,0 +1,251 @@
+"""In-process manager for dynamic workflow runs."""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any
+
+from openharness.config.paths import get_project_workflows_dir, get_user_workflows_dir
+from openharness.utils.fs import atomic_write_text
+from openharness.workflows.runtime import WorkflowAgentRunner, WorkflowRuntime, default_max_concurrency
+from openharness.workflows.store import WorkflowStore
+from openharness.workflows.types import WorkflowRunRecord
+
+
+class WorkflowManager:
+    """Start, inspect, pause, and resume workflow runs."""
+
+    def __init__(self, store: WorkflowStore | None = None) -> None:
+        self.store = store or WorkflowStore()
+        self._tasks: dict[str, asyncio.Task[Any]] = {}
+        self._agent_runner: WorkflowAgentRunner | None = None
+
+    def set_agent_runner(self, runner: WorkflowAgentRunner | None) -> None:
+        """Override the agent runner, primarily for tests."""
+        self._agent_runner = runner
+
+    async def start_inline(
+        self,
+        *,
+        cwd: str | Path,
+        name: str,
+        script: str,
+        description: str = "",
+        model: str | None = None,
+        wait: bool = False,
+        max_concurrency: int | None = None,
+        max_agents: int = 1000,
+    ) -> WorkflowRunRecord:
+        """Create and start an inline workflow script."""
+        run = self.store.create_run(
+            cwd=cwd,
+            name=name,
+            script=script,
+            description=description,
+            max_agents=max_agents,
+            max_concurrency=max_concurrency or default_max_concurrency(),
+        )
+        self._start_runtime(run, model=model, max_concurrency=max_concurrency, max_agents=max_agents)
+        if wait:
+            await self.wait(run.id)
+        latest = self.store.load_run(run.id)
+        return latest or run
+
+    async def start_saved(
+        self,
+        *,
+        cwd: str | Path,
+        name_or_path: str,
+        model: str | None = None,
+        wait: bool = False,
+        max_concurrency: int | None = None,
+        max_agents: int = 1000,
+    ) -> WorkflowRunRecord:
+        """Start a saved workflow by name or path."""
+        resolved = resolve_workflow_script(cwd, name_or_path)
+        if resolved is None:
+            raise ValueError(f"Workflow not found: {name_or_path}")
+        script = resolved.read_text(encoding="utf-8")
+        return await self.start_inline(
+            cwd=cwd,
+            name=resolved.stem,
+            script=script,
+            description=f"Saved workflow: {resolved}",
+            model=model,
+            wait=wait,
+            max_concurrency=max_concurrency,
+            max_agents=max_agents,
+        )
+
+    async def resume(
+        self,
+        *,
+        run_id: str,
+        model: str | None = None,
+        wait: bool = False,
+        max_concurrency: int | None = None,
+    ) -> WorkflowRunRecord:
+        """Resume a paused or failed workflow using its run cache."""
+        run = self.store.clone_run_for_resume(run_id)
+        if run is None:
+            raise ValueError(f"Workflow run not found: {run_id}")
+        self._start_runtime(run, model=model, max_concurrency=max_concurrency, max_agents=run.max_agents)
+        if wait:
+            await self.wait(run.id)
+        latest = self.store.load_run(run.id)
+        return latest or run
+
+    async def pause(self, run_id: str) -> WorkflowRunRecord:
+        """Pause a running workflow by cancelling the runtime task."""
+        task = self._tasks.get(run_id)
+        if task is not None and not task.done():
+            task.cancel()
+            try:
+                await task
+            except asyncio.CancelledError:
+                pass
+        run = self.store.load_run(run_id)
+        if run is None:
+            raise ValueError(f"Workflow run not found: {run_id}")
+        if run.status == "running":
+            run.status = "paused"
+            self.store.write_snapshot(run)
+            self.store.append_event(run_id, "run_paused", {})
+        return run
+
+    async def stop(self, run_id: str) -> WorkflowRunRecord:
+        """Stop a running workflow and mark it killed."""
+        await self.pause(run_id)
+        run = self.store.load_run(run_id)
+        if run is None:
+            raise ValueError(f"Workflow run not found: {run_id}")
+        run.status = "killed"
+        self.store.write_snapshot(run)
+        self.store.append_event(run_id, "run_killed", {})
+        return run
+
+    async def wait(self, run_id: str) -> WorkflowRunRecord:
+        """Wait for a workflow task to finish."""
+        task = self._tasks.get(run_id)
+        if task is not None:
+            try:
+                await task
+            except Exception:
+                pass
+        run = self.store.load_run(run_id)
+        if run is None:
+            raise ValueError(f"Workflow run not found: {run_id}")
+        return run
+
+    def get(self, run_id: str) -> WorkflowRunRecord | None:
+        """Return one workflow run."""
+        return self.store.load_run(run_id)
+
+    def list_runs(self, *, limit: int = 50) -> list[WorkflowRunRecord]:
+        """List workflow runs."""
+        return self.store.list_runs(limit=limit)
+
+    def save_script(
+        self,
+        *,
+        cwd: str | Path,
+        name: str,
+        script: str,
+        scope: str = "project",
+    ) -> Path:
+        """Save a workflow script as a reusable command."""
+        safe_name = _safe_workflow_name(name)
+        target_dir = get_user_workflows_dir() if scope == "user" else get_project_workflows_dir(cwd)
+        path = target_dir / f"{safe_name}.js"
+        atomic_write_text(path, script.rstrip() + "\n")
+        return path
+
+    def save_run_script(
+        self,
+        *,
+        cwd: str | Path,
+        run_id: str,
+        name: str,
+        scope: str = "project",
+    ) -> Path:
+        """Save an existing run's script."""
+        run = self.store.load_run(run_id)
+        if run is None:
+            raise ValueError(f"Workflow run not found: {run_id}")
+        script = Path(run.script_path).read_text(encoding="utf-8")
+        return self.save_script(cwd=cwd, name=name, script=script, scope=scope)
+
+    def _start_runtime(
+        self,
+        run: WorkflowRunRecord,
+        *,
+        model: str | None,
+        max_concurrency: int | None,
+        max_agents: int,
+    ) -> None:
+        runtime = WorkflowRuntime(
+            store=self.store,
+            run=run,
+            model=model,
+            agent_runner=self._agent_runner,
+            max_concurrency=max_concurrency,
+            max_agents=max_agents,
+        )
+        task = asyncio.create_task(runtime.run_script())
+        self._tasks[run.id] = task
+
+        def _cleanup(done: asyncio.Task[Any]) -> None:
+            if self._tasks.get(run.id) is done:
+                self._tasks.pop(run.id, None)
+
+        task.add_done_callback(_cleanup)
+
+
+def list_saved_workflows(cwd: str | Path) -> list[Path]:
+    """List saved workflow scripts visible to a project."""
+    project_dir = get_project_workflows_dir(cwd)
+    user_dir = get_user_workflows_dir()
+    paths = list(project_dir.glob("*.js")) + list(user_dir.glob("*.js"))
+    return sorted(paths, key=lambda path: (path.name, str(path)))
+
+
+def resolve_workflow_script(cwd: str | Path, name_or_path: str) -> Path | None:
+    """Resolve a saved workflow by explicit path or search path name."""
+    raw = name_or_path.strip()
+    if not raw:
+        return None
+    path = Path(raw).expanduser()
+    if path.exists() and path.is_file():
+        return path.resolve()
+    name = raw[:-3] if raw.endswith(".js") else raw
+    safe_name = _safe_workflow_name(name)
+    for base in (get_project_workflows_dir(cwd), get_user_workflows_dir()):
+        candidate = base / f"{safe_name}.js"
+        if candidate.exists():
+            return candidate
+    return None
+
+
+def _safe_workflow_name(name: str) -> str:
+    cleaned = "".join(ch.lower() if ch.isalnum() else "-" for ch in name.strip())
+    while "--" in cleaned:
+        cleaned = cleaned.replace("--", "-")
+    return cleaned.strip("-") or "workflow"
+
+
+_DEFAULT_MANAGER: WorkflowManager | None = None
+
+
+def get_workflow_manager() -> WorkflowManager:
+    """Return the process-local workflow manager."""
+    global _DEFAULT_MANAGER
+    if _DEFAULT_MANAGER is None:
+        _DEFAULT_MANAGER = WorkflowManager()
+    return _DEFAULT_MANAGER
+
+
+def reset_workflow_manager() -> None:
+    """Reset the process-local workflow manager."""
+    global _DEFAULT_MANAGER
+    _DEFAULT_MANAGER = None

src/openharness/workflows/runtime.py

@@ -0,0 +1,573 @@
+"""JavaScript-backed dynamic workflow runtime."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import textwrap
+import time
+from pathlib import Path
+from typing import Any, Protocol
+
+from openharness.swarm.registry import get_backend_registry
+from openharness.swarm.types import TeammateSpawnConfig
+from openharness.tasks import get_task_manager
+from openharness.workflows.store import WorkflowStore, workflow_cache_key
+from openharness.workflows.types import WorkflowAgentRecord, WorkflowPhaseRecord, WorkflowRunRecord
+
+
+class WorkflowAgentRunner(Protocol):
+    """Protocol for executing one workflow subagent call."""
+
+    async def run_agent(
+        self,
+        *,
+        prompt: str,
+        cwd: Path,
+        model: str | None,
+        phase: str | None,
+        metadata: dict[str, Any],
+    ) -> tuple[str, str | None, str | None]:
+        """Return ``(output, task_id, agent_id)``."""
+
+
+class SubprocessWorkflowAgentRunner:
+    """Run workflow subagents through OpenHarness' subprocess swarm backend."""
+
+    async def run_agent(
+        self,
+        *,
+        prompt: str,
+        cwd: Path,
+        model: str | None,
+        phase: str | None,
+        metadata: dict[str, Any],
+    ) -> tuple[str, str | None, str | None]:
+        registry = get_backend_registry()
+        executor = registry.get_executor("subprocess")
+        name = str(metadata.get("name") or metadata.get("subagent_type") or "workflow-agent")
+        team = str(metadata.get("team") or "workflow")
+        result = await executor.spawn(
+            TeammateSpawnConfig(
+                name=name,
+                team=team,
+                prompt=prompt,
+                cwd=str(cwd),
+                parent_session_id=str(metadata.get("parent_session_id") or "workflow"),
+                model=model,
+                system_prompt=metadata.get("system_prompt") if isinstance(metadata.get("system_prompt"), str) else None,
+                system_prompt_mode=metadata.get("system_prompt_mode") if metadata.get("system_prompt_mode") in {"append", "replace", "default"} else None,
+                permissions=[str(item) for item in metadata.get("permissions", [])] if isinstance(metadata.get("permissions"), list) else [],
+                task_type="local_agent",
+            )
+        )
+        if not result.success:
+            raise RuntimeError(result.error or "failed to spawn workflow agent")
+
+        manager = get_task_manager()
+        while True:
+            task = manager.get_task(result.task_id)
+            if task is None:
+                raise RuntimeError(f"workflow agent task disappeared: {result.task_id}")
+            if task.status in {"completed", "failed", "killed"}:
+                output = manager.read_task_output(result.task_id, max_bytes=200_000)
+                if task.status != "completed":
+                    raise RuntimeError(output.strip() or f"workflow agent {result.agent_id} {task.status}")
+                return output, result.task_id, result.agent_id
+            await asyncio.sleep(0.25)
+
+
+class WorkflowRuntime:
+    """Execute a dynamic workflow script and coordinate subagents."""
+
+    def __init__(
+        self,
+        *,
+        store: WorkflowStore,
+        run: WorkflowRunRecord,
+        model: str | None = None,
+        agent_runner: WorkflowAgentRunner | None = None,
+        max_concurrency: int | None = None,
+        max_agents: int | None = None,
+    ) -> None:
+        self.store = store
+        self.run = run
+        self.model = model
+        self.agent_runner = agent_runner or SubprocessWorkflowAgentRunner()
+        self.max_concurrency = max_concurrency or run.max_concurrency or default_max_concurrency()
+        self.max_agents = max_agents or run.max_agents or 1000
+        self._semaphore = asyncio.Semaphore(max(1, self.max_concurrency))
+        self._write_lock = asyncio.Lock()
+        self._node_process: asyncio.subprocess.Process | None = None
+        self._pending_rpc: set[asyncio.Task[None]] = set()
+
+    async def run_script(self) -> Any:
+        """Run the workflow script to completion."""
+        script_path = Path(self.run.script_path)
+        if not script_path.exists():
+            raise FileNotFoundError(f"workflow script not found: {script_path}")
+
+        self.run.status = "running"
+        self.run.started_at = time.time()
+        self.run.ended_at = None
+        self.run.error = None
+        self.run.max_concurrency = self.max_concurrency
+        self.run.max_agents = self.max_agents
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "run_started", {"max_concurrency": self.max_concurrency})
+
+        driver_path = self.store.run_dir(self.run.id) / "driver.js"
+        driver_path.write_text(_NODE_DRIVER, encoding="utf-8")
+
+        try:
+            result = await self._run_node(driver_path, script_path)
+        except asyncio.CancelledError:
+            await self._kill_node()
+            self.run.status = "paused"
+            self.run.ended_at = time.time()
+            self.store.write_snapshot(self.run)
+            self.store.append_event(self.run.id, "run_paused", {})
+            raise
+        except Exception as exc:
+            await self._kill_node()
+            self._finalize_open_phases("failed")
+            self.run.status = "failed"
+            self.run.error = str(exc)
+            self.run.ended_at = time.time()
+            self.store.write_snapshot(self.run)
+            self.store.append_event(self.run.id, "run_failed", {"error": str(exc)})
+            raise
+
+        self._finalize_open_phases("completed")
+        self.run.status = "completed"
+        self.run.result = result
+        self.run.ended_at = time.time()
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "run_completed", {"result": result})
+        return result
+
+    async def _run_node(self, driver_path: Path, script_path: Path) -> Any:
+        self._node_process = await asyncio.create_subprocess_exec(
+            "node",
+            str(driver_path),
+            str(script_path),
+            cwd=str(Path(self.run.cwd)),
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        assert self._node_process.stdout is not None
+        assert self._node_process.stderr is not None
+
+        stderr_task = asyncio.create_task(self._drain_stderr(self._node_process.stderr))
+        final_seen = False
+        final_result: Any = None
+
+        try:
+            while True:
+                raw = await self._node_process.stdout.readline()
+                if not raw:
+                    break
+                try:
+                    message = json.loads(raw.decode("utf-8"))
+                except json.JSONDecodeError as exc:
+                    raise RuntimeError(f"workflow runtime emitted invalid JSON: {raw!r}") from exc
+
+                op = message.get("op")
+                if op == "__done":
+                    final_seen = True
+                    if message.get("ok"):
+                        final_result = message.get("result")
+                    else:
+                        raise RuntimeError(str(message.get("error") or "workflow failed"))
+                    continue
+
+                task = asyncio.create_task(self._handle_rpc(message))
+                self._pending_rpc.add(task)
+                task.add_done_callback(self._pending_rpc.discard)
+
+            if self._pending_rpc:
+                await asyncio.gather(*list(self._pending_rpc))
+            return_code = await self._node_process.wait()
+            await stderr_task
+            if return_code != 0:
+                raise RuntimeError(f"workflow runtime exited with code {return_code}")
+            if not final_seen:
+                raise RuntimeError("workflow runtime exited without a result")
+            return final_result
+        finally:
+            stderr_task.cancel()
+
+    async def _handle_rpc(self, message: dict[str, Any]) -> None:
+        rpc_id = message.get("id")
+        op = message.get("op")
+        try:
+            if op == "agent":
+                result = await self._agent(message)
+            elif op == "phase_start":
+                result = self._phase_start(str(message.get("name") or "phase"))
+            elif op == "phase_end":
+                result = self._phase_end(str(message.get("name") or "phase"))
+            elif op == "log":
+                result = self._log(str(message.get("message") or ""))
+            elif op == "workflow_start":
+                result = self._workflow_start(str(message.get("name") or self.run.name))
+            elif op == "workflow_end":
+                result = self._workflow_end(str(message.get("name") or self.run.name), message.get("result"))
+            elif op == "workflow_error":
+                result = self._workflow_error(str(message.get("name") or self.run.name), str(message.get("error") or ""))
+            else:
+                raise RuntimeError(f"unknown workflow rpc op: {op}")
+            await self._respond(rpc_id, ok=True, result=result)
+        except Exception as exc:
+            await self._respond(rpc_id, ok=False, error=str(exc))
+
+    async def _respond(self, rpc_id: Any, *, ok: bool, result: Any = None, error: str | None = None) -> None:
+        process = self._node_process
+        if process is None or process.stdin is None or process.stdin.is_closing():
+            return
+        payload = {"id": rpc_id, "ok": ok, "result": result, "error": error}
+        data = json.dumps(payload, ensure_ascii=True, default=str) + "\n"
+        async with self._write_lock:
+            process.stdin.write(data.encode("utf-8"))
+            await process.stdin.drain()
+
+    async def _agent(self, message: dict[str, Any]) -> Any:
+        prompt = str(message.get("prompt") or "")
+        opts = message.get("opts")
+        if not isinstance(opts, dict):
+            opts = {}
+        phase = str(opts.get("phase") or self.run.current_phase or "") or None
+        cache_key = workflow_cache_key(script_hash=self.run.script_hash, prompt=prompt, opts=opts)
+        cached = self.store.read_cache(self.run.id, cache_key)
+        agent_id = f"agent_{cache_key[:12]}"
+
+        if cached is not None:
+            self.run.cached_count += 1
+            self._increment_phase(phase, cached=True)
+            record = WorkflowAgentRecord(
+                id=agent_id,
+                cache_key=cache_key,
+                prompt=prompt,
+                phase=phase,
+                status="cached",
+                output=str(cached),
+                model=str(opts.get("model") or self.model or ""),
+                started_at=time.time(),
+                ended_at=time.time(),
+                cached=True,
+            )
+            self.run.agents[agent_id] = record
+            self.store.write_snapshot(self.run)
+            self.store.append_event(self.run.id, "agent_cached", {"agent_id": agent_id, "phase": phase, "cache_key": cache_key})
+            return cached
+
+        if self.run.agent_count >= self.max_agents:
+            raise RuntimeError(f"workflow exceeded max agent calls ({self.max_agents})")
+
+        self.run.agent_count += 1
+        self._increment_phase(phase, cached=False)
+        record = WorkflowAgentRecord(
+            id=agent_id,
+            cache_key=cache_key,
+            prompt=prompt,
+            phase=phase,
+            status="pending",
+            model=str(opts.get("model") or self.model or ""),
+            metadata=dict(opts),
+        )
+        self.run.agents[agent_id] = record
+        self.store.write_snapshot(self.run)
+
+        async with self._semaphore:
+            record.status = "running"
+            record.started_at = time.time()
+            self.store.write_snapshot(self.run)
+            self.store.append_event(self.run.id, "agent_started", {"agent_id": agent_id, "phase": phase, "cache_key": cache_key})
+            try:
+                output, task_id, spawned_agent_id = await self.agent_runner.run_agent(
+                    prompt=prompt,
+                    cwd=Path(self.run.cwd),
+                    model=str(opts.get("model") or self.model or "") or None,
+                    phase=phase,
+                    metadata=opts,
+                )
+            except Exception as exc:
+                record.status = "failed"
+                record.error = str(exc)
+                record.ended_at = time.time()
+                self.store.write_snapshot(self.run)
+                self.store.append_event(self.run.id, "agent_failed", {"agent_id": agent_id, "error": str(exc)})
+                raise
+
+            record.status = "completed"
+            record.task_id = task_id
+            if spawned_agent_id:
+                record.metadata["spawned_agent_id"] = spawned_agent_id
+            record.output = output
+            record.ended_at = time.time()
+            self.store.write_cache(self.run.id, cache_key, output)
+            self.store.write_snapshot(self.run)
+            self.store.append_event(self.run.id, "agent_completed", {"agent_id": agent_id, "phase": phase, "task_id": task_id})
+            return output
+
+    def _workflow_start(self, name: str) -> dict[str, str]:
+        self.run.name = name or self.run.name
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "workflow_started", {"name": self.run.name})
+        return {"name": self.run.name}
+
+    def _workflow_end(self, name: str, result: Any) -> dict[str, str]:
+        self.run.name = name or self.run.name
+        self.run.result = result
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "workflow_completed", {"name": self.run.name, "result": result})
+        return {"name": self.run.name}
+
+    def _workflow_error(self, name: str, error: str) -> dict[str, str]:
+        self.run.name = name or self.run.name
+        self.run.error = error
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "workflow_error", {"name": self.run.name, "error": error})
+        return {"name": self.run.name}
+
+    def _phase_start(self, name: str) -> dict[str, str]:
+        phase = self.run.phases.get(name)
+        if phase is None:
+            phase = WorkflowPhaseRecord(name=name)
+            self.run.phases[name] = phase
+        phase.status = "running"
+        phase.started_at = phase.started_at or time.time()
+        self.run.current_phase = name
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "phase_started", {"phase": name})
+        return {"phase": name}
+
+    def _phase_end(self, name: str) -> dict[str, str]:
+        phase = self.run.phases.get(name)
+        if phase is None:
+            phase = WorkflowPhaseRecord(name=name)
+            self.run.phases[name] = phase
+        phase.status = "completed"
+        phase.ended_at = time.time()
+        if self.run.current_phase == name:
+            self.run.current_phase = None
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "phase_completed", {"phase": name})
+        return {"phase": name}
+
+    def _log(self, message: str) -> dict[str, int]:
+        if message:
+            self.run.logs.append(message)
+            self.run.logs = self.run.logs[-200:]
+        self.store.write_snapshot(self.run)
+        self.store.append_event(self.run.id, "log", {"message": message})
+        return {"log_count": len(self.run.logs)}
+
+    def _increment_phase(self, phase_name: str | None, *, cached: bool) -> None:
+        if not phase_name:
+            return
+        phase = self.run.phases.get(phase_name)
+        if phase is None:
+            phase = WorkflowPhaseRecord(name=phase_name, status="running", started_at=time.time())
+            self.run.phases[phase_name] = phase
+        phase.agent_count += 1
+        if cached:
+            phase.cached_count += 1
+
+    def _finalize_open_phases(self, status: str) -> None:
+        now = time.time()
+        for phase in self.run.phases.values():
+            if phase.status == "running":
+                phase.status = status
+                phase.ended_at = phase.ended_at or now
+        self.run.current_phase = None
+
+    async def _drain_stderr(self, stream: asyncio.StreamReader) -> None:
+        while True:
+            chunk = await stream.readline()
+            if not chunk:
+                return
+            text = chunk.decode("utf-8", errors="replace").strip()
+            if text:
+                self.store.append_event(self.run.id, "runtime_stderr", {"message": text})
+
+    async def _kill_node(self) -> None:
+        process = self._node_process
+        if process is None or process.returncode is not None:
+            return
+        process.kill()
+        await process.wait()
+
+
+def default_max_concurrency() -> int:
+    cpu_count = os.cpu_count() or 2
+    return max(1, min(16, cpu_count - 2))
+
+
+_NODE_DRIVER = textwrap.dedent(
+    r"""
+    const fs = require("fs");
+    const vm = require("vm");
+    const readline = require("readline");
+
+    const scriptPath = process.argv[2];
+    const source = fs.readFileSync(scriptPath, "utf8");
+    const forbiddenPatterns = [
+      /\bDate\s*\.\s*now\s*\(/,
+      /\bnew\s+Date\s*\(/,
+      /\bMath\s*\.\s*random\s*\(/,
+      /\brequire\s*\(/,
+      /\bimport\s*\(/,
+      /\bfetch\s*\(/,
+    ];
+    for (const pattern of forbiddenPatterns) {
+      if (pattern.test(source)) {
+        throw new Error(`Forbidden workflow API matched ${pattern}`);
+      }
+    }
+
+    let nextId = 1;
+    const pending = new Map();
+    const rl = readline.createInterface({ input: process.stdin });
+    rl.on("line", line => {
+      let msg;
+      try {
+        msg = JSON.parse(line);
+      } catch (error) {
+        return;
+      }
+      const waiter = pending.get(msg.id);
+      if (!waiter) return;
+      pending.delete(msg.id);
+      if (msg.ok) {
+        waiter.resolve(msg.result);
+      } else {
+        waiter.reject(new Error(msg.error || "workflow rpc failed"));
+      }
+    });
+
+    const rootWorkflows = [];
+
+    function rpc(op, payload) {
+      const id = nextId++;
+      const msg = Object.assign({ id, op }, payload || {});
+      process.stdout.write(JSON.stringify(msg) + "\n");
+      return new Promise((resolve, reject) => pending.set(id, { resolve, reject }));
+    }
+
+    function workflow(name, fn) {
+      const run = (async () => {
+        await rpc("workflow_start", { name });
+        try {
+          const result = await fn();
+          await rpc("workflow_end", { name, result });
+          return result;
+        } catch (error) {
+          await rpc("workflow_error", { name, error: String(error && error.stack || error) });
+          throw error;
+        }
+      })();
+      rootWorkflows.push(run);
+      return run;
+    }
+
+    async function agent(prompt, opts) {
+      return await rpc("agent", { prompt: String(prompt), opts: opts || {} });
+    }
+
+    async function parallel(items, fn) {
+      if (!Array.isArray(items)) throw new Error("parallel(items, fn) requires an array");
+      if (typeof fn !== "function") throw new Error("parallel(items, fn) requires a function");
+      return await Promise.all(items.map((item, index) => fn(item, index)));
+    }
+
+    async function pipeline(items, stages) {
+      if (!Array.isArray(items)) throw new Error("pipeline(items, stages) requires an array of items");
+      if (!Array.isArray(stages)) throw new Error("pipeline(items, stages) requires an array of stage functions");
+      return await Promise.all(items.map(async (item, index) => {
+        let value = item;
+        for (const stage of stages) {
+          if (typeof stage !== "function") throw new Error("pipeline stages must be functions");
+          value = await stage(value, index);
+        }
+        return value;
+      }));
+    }
+
+    async function phase(name, fn) {
+      await rpc("phase_start", { name: String(name) });
+      try {
+        return await fn();
+      } finally {
+        await rpc("phase_end", { name: String(name) });
+      }
+    }
+
+    async function log(message) {
+      await rpc("log", { message: String(message) });
+    }
+
+    const safeMath = Object.create(Math);
+    safeMath.random = function () {
+      throw new Error("Math.random is disabled in deterministic workflows");
+    };
+
+    function DisabledDate() {
+      throw new Error("Date is disabled in deterministic workflows");
+    }
+    DisabledDate.now = function () {
+      throw new Error("Date.now is disabled in deterministic workflows");
+    };
+
+    const context = {
+      workflow,
+      agent,
+      parallel,
+      pipeline,
+      phase,
+      log,
+      console: { log: (...args) => log(args.join(" ")) },
+      JSON,
+      Array,
+      Object,
+      String,
+      Number,
+      Boolean,
+      Promise,
+      Error,
+      RegExp,
+      Map,
+      Set,
+      Math: safeMath,
+      Date: DisabledDate,
+    };
+
+    async function main() {
+      const wrapped = `(async () => {\n${source}\n})()`;
+      const vmContext = vm.createContext(context, {
+        codeGeneration: { strings: false, wasm: false },
+      });
+      const script = new vm.Script(wrapped, { filename: scriptPath });
+      const result = await script.runInContext(vmContext, { timeout: 1000 });
+      if (typeof result === "undefined" && rootWorkflows.length === 1) {
+        return await rootWorkflows[0];
+      }
+      if (typeof result === "undefined" && rootWorkflows.length > 1) {
+        return await Promise.all(rootWorkflows);
+      }
+      return result;
+    }
+
+    main()
+      .then(result => {
+        rl.close();
+        process.stdout.write(JSON.stringify({ op: "__done", ok: true, result }) + "\n", () => process.exit(0));
+      })
+      .catch(error => {
+        rl.close();
+        process.stdout.write(JSON.stringify({ op: "__done", ok: false, error: String(error && error.stack || error) }) + "\n", () => process.exit(1));
+      });
+    """
+).strip() + "\n"

src/openharness/workflows/store.py

@@ -0,0 +1,191 @@
+"""Persistent storage for dynamic workflow runs."""
+
+from __future__ import annotations
+
+import json
+import time
+from dataclasses import asdict
+from hashlib import sha1, sha256
+from pathlib import Path
+from typing import Any
+from uuid import uuid4
+
+from openharness.config.paths import get_workflows_dir
+from openharness.utils.fs import atomic_write_text
+from openharness.workflows.types import (
+    WorkflowAgentRecord,
+    WorkflowPhaseRecord,
+    WorkflowRunRecord,
+)
+
+
+def stable_json(value: Any) -> str:
+    """Return deterministic JSON for cache keys."""
+    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"), default=str)
+
+
+def workflow_script_hash(script: str) -> str:
+    """Return the stable hash for a workflow script."""
+    return sha256(script.encode("utf-8")).hexdigest()
+
+
+def workflow_cache_key(*, script_hash: str, prompt: str, opts: dict[str, Any] | None) -> str:
+    """Return the deterministic cache key for one agent call."""
+    payload = stable_json(
+        {
+            "script_hash": script_hash,
+            "prompt": prompt,
+            "opts": opts or {},
+        }
+    )
+    return sha256(payload.encode("utf-8")).hexdigest()
+
+
+class WorkflowStore:
+    """JSONL journal and snapshot storage for workflow runs."""
+
+    def __init__(self, root: str | Path | None = None) -> None:
+        self.root = Path(root).expanduser().resolve() if root is not None else get_workflows_dir()
+        self.runs_dir = self.root / "runs"
+        self.runs_dir.mkdir(parents=True, exist_ok=True)
+
+    def create_run(
+        self,
+        *,
+        cwd: str | Path,
+        name: str,
+        script: str,
+        description: str = "",
+        max_agents: int = 1000,
+        max_concurrency: int = 1,
+    ) -> WorkflowRunRecord:
+        """Create a new run directory and initial snapshot."""
+        run_id = f"w{uuid4().hex[:10]}"
+        run_dir = self.run_dir(run_id)
+        run_dir.mkdir(parents=True, exist_ok=True)
+        (run_dir / "cache").mkdir(exist_ok=True)
+        script_path = run_dir / "script.js"
+        atomic_write_text(script_path, script.rstrip() + "\n")
+        now = time.time()
+        record = WorkflowRunRecord(
+            id=run_id,
+            name=name,
+            cwd=str(Path(cwd).resolve()),
+            script_path=str(script_path),
+            script_hash=workflow_script_hash(script),
+            status="pending",
+            description=description,
+            created_at=now,
+            max_agents=max_agents,
+            max_concurrency=max_concurrency,
+        )
+        self.write_snapshot(record)
+        self.append_event(run_id, "run_created", asdict(record))
+        return record
+
+    def run_dir(self, run_id: str) -> Path:
+        """Return a run directory path."""
+        return self.runs_dir / run_id
+
+    def snapshot_path(self, run_id: str) -> Path:
+        return self.run_dir(run_id) / "snapshot.json"
+
+    def journal_path(self, run_id: str) -> Path:
+        return self.run_dir(run_id) / "journal.jsonl"
+
+    def cache_path(self, run_id: str, key: str) -> Path:
+        return self.run_dir(run_id) / "cache" / f"{key}.json"
+
+    def load_run(self, run_id: str) -> WorkflowRunRecord | None:
+        """Load one run snapshot."""
+        path = self.snapshot_path(run_id)
+        if not path.exists():
+            return None
+        return self._record_from_payload(json.loads(path.read_text(encoding="utf-8")))
+
+    def list_runs(self, *, limit: int = 50) -> list[WorkflowRunRecord]:
+        """List known workflow runs, newest first."""
+        records: list[WorkflowRunRecord] = []
+        for path in sorted(self.runs_dir.glob("*/snapshot.json"), key=lambda item: item.stat().st_mtime, reverse=True):
+            try:
+                records.append(self._record_from_payload(json.loads(path.read_text(encoding="utf-8"))))
+            except (OSError, json.JSONDecodeError, TypeError, ValueError):
+                continue
+            if len(records) >= limit:
+                break
+        return records
+
+    def write_snapshot(self, record: WorkflowRunRecord) -> None:
+        """Persist a workflow run snapshot."""
+        run_dir = self.run_dir(record.id)
+        run_dir.mkdir(parents=True, exist_ok=True)
+        (run_dir / "cache").mkdir(exist_ok=True)
+        atomic_write_text(
+            self.snapshot_path(record.id),
+            json.dumps(asdict(record), indent=2, ensure_ascii=True, default=str) + "\n",
+        )
+
+    def append_event(self, run_id: str, event: str, payload: dict[str, Any] | None = None) -> None:
+        """Append a journal event."""
+        entry = {
+            "ts": time.time(),
+            "event": event,
+            "payload": payload or {},
+        }
+        with self.journal_path(run_id).open("a", encoding="utf-8") as handle:
+            handle.write(json.dumps(entry, ensure_ascii=True, default=str) + "\n")
+
+    def read_cache(self, run_id: str, key: str) -> Any | None:
+        """Read a cached agent result for a run."""
+        path = self.cache_path(run_id, key)
+        if not path.exists():
+            return None
+        try:
+            payload = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, json.JSONDecodeError):
+            return None
+        return payload.get("result")
+
+    def write_cache(self, run_id: str, key: str, result: Any) -> None:
+        """Write a cached agent result for a run."""
+        path = self.cache_path(run_id, key)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        atomic_write_text(path, json.dumps({"result": result}, ensure_ascii=True, default=str) + "\n")
+
+    def clone_run_for_resume(self, run_id: str) -> WorkflowRunRecord | None:
+        """Load a run for resume while preserving its cache directory."""
+        record = self.load_run(run_id)
+        if record is None:
+            return None
+        record.status = "pending"
+        record.error = None
+        record.result = None
+        record.started_at = None
+        record.ended_at = None
+        for agent in record.agents.values():
+            if agent.status not in {"completed", "cached"}:
+                agent.status = "pending"
+                agent.error = None
+                agent.ended_at = None
+        self.write_snapshot(record)
+        self.append_event(run_id, "run_resumed", {})
+        return record
+
+    def _record_from_payload(self, payload: dict[str, Any]) -> WorkflowRunRecord:
+        phases = {
+            name: WorkflowPhaseRecord(**value)
+            for name, value in dict(payload.get("phases") or {}).items()
+        }
+        agents = {
+            name: WorkflowAgentRecord(**value)
+            for name, value in dict(payload.get("agents") or {}).items()
+        }
+        payload = dict(payload)
+        payload["phases"] = phases
+        payload["agents"] = agents
+        return WorkflowRunRecord(**payload)
+
+
+def project_digest(cwd: str | Path) -> str:
+    """Return a short stable project digest for future storage partitioning."""
+    return sha1(str(Path(cwd).resolve()).encode("utf-8")).hexdigest()[:12]

src/openharness/workflows/types.py

@@ -0,0 +1,71 @@
+"""Data models for dynamic workflow runs."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Literal
+
+
+WorkflowRunStatus = Literal["pending", "running", "paused", "completed", "failed", "killed"]
+WorkflowAgentStatus = Literal["pending", "running", "cached", "completed", "failed", "killed"]
+WorkflowPhaseStatus = Literal["pending", "running", "completed", "failed"]
+
+
+@dataclass
+class WorkflowAgentRecord:
+    """One subagent invocation inside a workflow run."""
+
+    id: str
+    cache_key: str
+    prompt: str
+    phase: str | None = None
+    status: WorkflowAgentStatus = "pending"
+    task_id: str | None = None
+    output: str | None = None
+    error: str | None = None
+    model: str | None = None
+    started_at: float | None = None
+    ended_at: float | None = None
+    cached: bool = False
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class WorkflowPhaseRecord:
+    """A named workflow phase shown in progress views."""
+
+    name: str
+    status: WorkflowPhaseStatus = "pending"
+    started_at: float | None = None
+    ended_at: float | None = None
+    agent_count: int = 0
+    cached_count: int = 0
+    token_count: int = 0
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class WorkflowRunRecord:
+    """Runtime snapshot for a workflow run."""
+
+    id: str
+    name: str
+    cwd: str
+    script_path: str
+    script_hash: str
+    status: WorkflowRunStatus = "pending"
+    description: str = ""
+    result: Any = None
+    error: str | None = None
+    created_at: float = 0.0
+    started_at: float | None = None
+    ended_at: float | None = None
+    agent_count: int = 0
+    cached_count: int = 0
+    max_agents: int = 1000
+    max_concurrency: int = 1
+    current_phase: str | None = None
+    phases: dict[str, WorkflowPhaseRecord] = field(default_factory=dict)
+    agents: dict[str, WorkflowAgentRecord] = field(default_factory=dict)
+    logs: list[str] = field(default_factory=list)
+    metadata: dict[str, Any] = field(default_factory=dict)

tests/test_ui/test_react_backend.py

@@ -21,7 +21,8 @@ from openharness.ui.backend_host import (
     _format_transcript_line,
     run_backend_host,
 )
-from openharness.ui.protocol import BackendEvent, FrontendImageAttachment, FrontendRequest
+from openharness.state.app_state import AppState
+from openharness.ui.protocol import BackendEvent, CommandSnapshot, FrontendImageAttachment, FrontendRequest
 from openharness.ui.runtime import build_runtime, close_runtime, start_runtime
 
 
@@ -141,6 +142,21 @@ def test_format_transcript_line_mentions_attached_images():
     ) == "inspect\n[2 images attached]"
 
 
+def test_ready_event_includes_structured_command_metadata():
+    event = BackendEvent.ready(
+        AppState(model="test-model", permission_mode="default", theme="default"),
+        [],
+        ["/help"],
+        [CommandSnapshot(name="/help", description="Show available commands", aliases=["/h"])],
+    )
+
+    payload = event.model_dump()
+    assert payload["commands"] == ["/help"]
+    assert payload["command_items"] == [
+        {"name": "/help", "description": "Show available commands", "aliases": ["/h"]}
+    ]
+
+
 @pytest.mark.asyncio
 async def test_run_backend_host_accepts_permission_mode(monkeypatch):
     captured: dict[str, str | None] = {}

tests/test_workflows.py

@@ -0,0 +1,216 @@
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+from openharness.commands.registry import CommandContext, create_default_command_registry
+from openharness.tools.base import ToolExecutionContext
+from openharness.tools.workflow_tool import WorkflowCreateInput, WorkflowCreateTool
+from openharness.workflows.manager import get_workflow_manager, reset_workflow_manager
+from openharness.workflows.runtime import WorkflowRuntime
+from openharness.workflows.store import WorkflowStore
+
+
+@pytest.fixture(autouse=True)
+def _reset_workflow_manager():
+    reset_workflow_manager()
+    yield
+    reset_workflow_manager()
+
+
+class EchoWorkflowAgentRunner:
+    def __init__(self, *, fail_if_called: bool = False) -> None:
+        self.calls: list[str] = []
+        self.running = 0
+        self.max_running = 0
+        self.fail_if_called = fail_if_called
+
+    async def run_agent(self, *, prompt: str, cwd: Path, model: str | None, phase: str | None, metadata: dict):
+        del cwd, model, metadata
+        if self.fail_if_called:
+            raise AssertionError("cache hit should avoid live agent execution")
+        self.calls.append(f"{phase or '-'}:{prompt}")
+        self.running += 1
+        self.max_running = max(self.max_running, self.running)
+        try:
+            await asyncio.sleep(0.03)
+            return f"OUT[{phase or '-'}]:{prompt}", f"task-{len(self.calls)}", None
+        finally:
+            self.running -= 1
+
+
+WORKFLOW_SCRIPT = """
+return await workflow("demo-workflow", async () => {
+  const items = ["a", "b", "c"];
+  const first = await phase("fanout", async () => {
+    return await parallel(items, item => agent(`inspect ${item}`, { phase: "fanout", name: `worker-${item}` }));
+  });
+  const reviewed = await phase("review", async () => {
+    return await pipeline(first, [
+      item => agent(`review ${item}`, { phase: "review" }),
+      item => agent(`finalize ${item}`, { phase: "review" }),
+    ]);
+  });
+  await log(`reviewed ${reviewed.length} items`);
+  return reviewed.join("\\n");
+});
+"""
+
+BARE_WORKFLOW_SCRIPT = """
+workflow("bare-workflow", async () => {
+  const result = await phase("single", async () => {
+    return await agent("inspect bare workflow", { phase: "single", name: "worker" });
+  });
+  return `bare:${result}`;
+});
+"""
+
+LABELED_AGENT_PHASE_SCRIPT = """
+return workflow("labeled-agent-phase", async () => {
+  const results = await parallel(["a", "b"], item => {
+    return agent(`inspect ${item}`, { phase: "labeled-review" });
+  });
+  return results.join(",");
+});
+"""
+
+
+@pytest.mark.asyncio
+async def test_workflow_runtime_runs_parallel_pipeline_and_records_snapshot(tmp_path: Path):
+    store = WorkflowStore(tmp_path / "wf")
+    run = store.create_run(cwd=tmp_path, name="demo", script=WORKFLOW_SCRIPT, max_concurrency=4)
+    runner = EchoWorkflowAgentRunner()
+
+    result = await WorkflowRuntime(store=store, run=run, agent_runner=runner, model="test-model").run_script()
+
+    assert "finalize OUT[review]" in result
+    assert len(runner.calls) == 9
+    assert runner.max_running > 1
+
+    saved = store.load_run(run.id)
+    assert saved is not None
+    assert saved.status == "completed"
+    assert saved.agent_count == 9
+    assert saved.phases["fanout"].agent_count == 3
+    assert saved.phases["review"].agent_count == 6
+    assert "reviewed 3 items" in saved.logs
+
+
+@pytest.mark.asyncio
+async def test_workflow_runtime_resume_uses_cached_agent_results(tmp_path: Path):
+    store = WorkflowStore(tmp_path / "wf")
+    run = store.create_run(cwd=tmp_path, name="demo", script=WORKFLOW_SCRIPT, max_concurrency=4)
+    await WorkflowRuntime(
+        store=store,
+        run=run,
+        agent_runner=EchoWorkflowAgentRunner(),
+        model="test-model",
+    ).run_script()
+
+    resumed = store.clone_run_for_resume(run.id)
+    assert resumed is not None
+    result = await WorkflowRuntime(
+        store=store,
+        run=resumed,
+        agent_runner=EchoWorkflowAgentRunner(fail_if_called=True),
+        model="test-model",
+    ).run_script()
+
+    assert "finalize OUT[review]" in result
+    saved = store.load_run(run.id)
+    assert saved is not None
+    assert saved.status == "completed"
+    assert saved.cached_count >= 9
+
+
+@pytest.mark.asyncio
+async def test_workflow_runtime_waits_for_bare_workflow_call(tmp_path: Path):
+    store = WorkflowStore(tmp_path / "wf")
+    run = store.create_run(cwd=tmp_path, name="bare", script=BARE_WORKFLOW_SCRIPT, max_concurrency=2)
+    runner = EchoWorkflowAgentRunner()
+
+    result = await WorkflowRuntime(store=store, run=run, agent_runner=runner, model="test-model").run_script()
+
+    assert result == "bare:OUT[single]:inspect bare workflow"
+    assert runner.calls == ["single:inspect bare workflow"]
+
+    saved = store.load_run(run.id)
+    assert saved is not None
+    assert saved.status == "completed"
+    assert saved.agent_count == 1
+    assert saved.phases["single"].agent_count == 1
+
+
+@pytest.mark.asyncio
+async def test_workflow_runtime_completes_agent_labeled_phases(tmp_path: Path):
+    store = WorkflowStore(tmp_path / "wf")
+    run = store.create_run(cwd=tmp_path, name="labeled", script=LABELED_AGENT_PHASE_SCRIPT, max_concurrency=2)
+
+    await WorkflowRuntime(
+        store=store,
+        run=run,
+        agent_runner=EchoWorkflowAgentRunner(),
+        model="test-model",
+    ).run_script()
+
+    saved = store.load_run(run.id)
+    assert saved is not None
+    assert saved.status == "completed"
+    assert saved.phases["labeled-review"].status == "completed"
+    assert saved.phases["labeled-review"].agent_count == 2
+
+
+@pytest.mark.asyncio
+async def test_workflow_create_tool_saves_and_runs_workflow(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    reset_workflow_manager()
+    manager = get_workflow_manager()
+    manager.set_agent_runner(EchoWorkflowAgentRunner())
+
+    result = await WorkflowCreateTool().execute(
+        WorkflowCreateInput(
+            name="Demo Tool Workflow",
+            script=WORKFLOW_SCRIPT,
+            run_immediately=True,
+            wait_for_completion=True,
+            max_concurrency=4,
+        ),
+        ToolExecutionContext(cwd=tmp_path, metadata={"model": "test-model"}),
+    )
+
+    assert result.is_error is False
+    assert "Saved workflow" in result.output
+    assert "completed" in result.output
+    assert result.metadata["run_id"]
+    assert (tmp_path / ".openharness" / "workflows" / "demo-tool-workflow.js").exists()
+
+
+@pytest.mark.asyncio
+async def test_workflows_slash_command_runs_and_shows_saved_workflow(tmp_path: Path, monkeypatch: pytest.MonkeyPatch):
+    monkeypatch.setenv("OPENHARNESS_CONFIG_DIR", str(tmp_path / "config"))
+    monkeypatch.setenv("OPENHARNESS_DATA_DIR", str(tmp_path / "data"))
+    reset_workflow_manager()
+    manager = get_workflow_manager()
+    manager.set_agent_runner(EchoWorkflowAgentRunner())
+    manager.save_script(cwd=tmp_path, name="demo", script=WORKFLOW_SCRIPT)
+
+    registry = create_default_command_registry()
+    context = CommandContext(
+        engine=SimpleNamespace(model="test-model"),  # type: ignore[arg-type]
+        cwd=str(tmp_path),
+    )
+
+    command, args = registry.lookup("/workflows run demo")
+    result = await command.handler(args, context)
+    assert "Started workflow" in result.message
+    run_id = result.message.split()[2].rstrip(":")
+
+    await manager.wait(run_id)
+    command, args = registry.lookup(f"/workflows output {run_id}")
+    output_result = await command.handler(args, context)
+    assert "completed" in output_result.message
+    assert "result:" in output_result.message