From 7aa424dda134ca40e8953597e88091bc1ff69c36 Mon Sep 17 00:00:00 2001 From: Markus Date: Sat, 26 Sep 2026 16:24:11 +0200 Subject: [PATCH] feat(workflows): compose workflows with a unified execution tree Keep invocation results and workflow bindings on the same execution occurrence. Isolate fan-out contexts and resume persisted expansions through one executor. Assisted-by: OpenCode (model: gpt-6-astra, autonomous) --- design/workflow-step.md | 7 +- docs/reference/workflows.md | 67 ++ src/specify_cli/workflows/__init__.py | 2 + src/specify_cli/workflows/_commands.py | 27 +- src/specify_cli/workflows/_execution.py | 576 +++++++++++++++ src/specify_cli/workflows/command_resume.py | 2 +- src/specify_cli/workflows/command_status.py | 9 + src/specify_cli/workflows/composition.py | 147 ++++ src/specify_cli/workflows/engine.py | 511 +++---------- .../workflows/step/gate/__init__.py | 2 + .../workflows/step/workflow/__init__.py | 17 + .../workflows/test_command_status.py | 32 + tests/workflows/test_composition_execution.py | 670 ++++++++++++++++++ workflows/ARCHITECTURE.md | 13 +- workflows/PUBLISHING.md | 2 +- workflows/README.md | 5 +- 16 files changed, 1655 insertions(+), 434 deletions(-) create mode 100644 src/specify_cli/workflows/_execution.py create mode 100644 src/specify_cli/workflows/composition.py create mode 100644 src/specify_cli/workflows/step/workflow/__init__.py create mode 100644 tests/workflows/test_composition_execution.py diff --git a/design/workflow-step.md b/design/workflow-step.md index e9ff094298..ddb902d840 100644 --- a/design/workflow-step.md +++ b/design/workflow-step.md @@ -41,9 +41,10 @@ unless `continue_on_error: true` is set (an explicit abort always stops). The engine calls `validate()` during workflow validation but does not automatically validate a definition passed to `execute()`. Guard invalid configurations in `execute()` too, returning a failed result rather than a -successful default or an unhandled exception. Resume restarts the current -top-level step; a pause inside nested steps re-runs their parent and nested -body. Design side effects accordingly. +successful default or an unhandled exception. New runs persist an execution +tree: resume retries unfinished occurrences and restores completed results and +selected expansions. Legacy states enter through their old top-level index. +Side effects performed before their completion checkpoint can still repeat. The registry holds one shared instance per type. Concurrent `fan-out` can invoke that instance from multiple threads: keep execution stateless and diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index 811ab4ebf4..c27ece7342 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -540,6 +540,73 @@ specify workflow run speckit -i spec="Build a kanban board with drag-and-drop ta | `do-while` | Execute at least once, then loop on condition | | `fan-out` | Dispatch a step for each item in a list | | `fan-in` | Aggregate results from a fan-out step | +| `workflow` | Call an installed workflow with private inputs and declared outputs | + +### Workflow composition + +A `workflow` step executes an installed, enabled workflow in the current project +as a private scope of the same run. Targets can be literal IDs or expressions; +the resolved string must match the ID exactly, including case and whitespace. + +```yaml +inputs: + target: {type: string, required: true} + report: {type: string, required: true} +steps: + - id: investigate + type: workflow + workflow: "{{ inputs.target }}" + input: + report: "{{ inputs.report }}" +``` + +The included workflow sees only declared inputs passed through `input` and its +own step results. Unknown input names, missing required values, and invalid +types/enums fail the call. Existing defaults and `integration: auto` resolution +apply. Values cross back only through explicit declarations: + +```yaml +outputs: + report: + value: "{{ steps.analyze.output.stdout }}" +``` + +The caller reads `{{ steps.investigate.output.report }}`. Output includes +`workflow` and `status`; failures also include `error`, and an abort includes +`aborted: true`. These names and `integration`, `model`, `options`, and `input` +are reserved. Returned values must be JSON-safe. Private inputs, step records, +and logs are not part of the return value. No separate child run is created. + +Failures may be handled with `continue_on_error: true` at either the included +step or workflow-call boundary. Pauses and explicit aborts always stop execution. +Nested composition is allowed, but repeated workflow IDs on the active call +path are cycles. Diamonds are allowed. The maximum included depth is 16, with +the root at depth zero. + +### Execution identity and resume + +Each step occurrence owns a record in a persisted execution tree. Authored step +names are local expression aliases, not global execution IDs. Fan-out items +have independent alias contexts; results remain ordered by item index. + +New runs persist selected branches, dynamic custom-step expansions, loop +iterations, and fan-out items. Resume retries unfinished operations and retains +completed work without reevaluating already selected branches. Workflow targets +and overlay-resolved definitions remain bound even if installations change. +Ordinary resume retains bound inputs; explicit `--input` updates rebind reached, +incomplete calls through their original mappings. Completed calls retain their +results. Failed output evaluation retries finalization without repeating child +commands. + +Snapshots are stored as YAML strings inside the private JSON execution tree, +preserving YAML scalar types. Inputs and results remain JSON values. Legacy runs +without a tree enter through their saved top-level index, then use tree-backed +resume. A tree-backed run left `running` by a crashed process can also be resumed; +only one process may execute or resume a run at a time. A side effect completed +before its checkpoint may execute again after a crash. + +Run/resume/status JSON includes `workflow_scopes` summaries when calls exist and +reports the active nested gate with its `scope_path`. > **Security note:** a `shell` step runs a local command with **your** privileges. There is no capability sandbox — `requires` is an advisory pre-condition block (spec-kit version, integrations), not a runtime gate, so it does **not** restrict what a step can do. In particular there is no `requires.permissions` capability gate: it is rejected by validation precisely because it would imply a sandbox that does not exist. Review any catalog or downloaded workflow before running it, and use a `gate` step to require explicit approval before sensitive or destructive shell commands. diff --git a/src/specify_cli/workflows/__init__.py b/src/specify_cli/workflows/__init__.py index 2bb3de56a5..dca51280e5 100644 --- a/src/specify_cli/workflows/__init__.py +++ b/src/specify_cli/workflows/__init__.py @@ -49,6 +49,7 @@ def _register_builtin_steps() -> None: from .step.fan_in import FanInStep from .step.fan_out import FanOutStep from .step.gate import GateStep + from .step.workflow import WorkflowStep from .step.if_then import IfThenStep from .step.init import InitStep from .step.prompt import PromptStep @@ -62,6 +63,7 @@ def _register_builtin_steps() -> None: _register_step(FanInStep()) _register_step(FanOutStep()) _register_step(GateStep()) + _register_step(WorkflowStep()) _register_step(IfThenStep()) _register_step(InitStep()) _register_step(PromptStep()) diff --git a/src/specify_cli/workflows/_commands.py b/src/specify_cli/workflows/_commands.py index de01789d0c..7d5757504c 100644 --- a/src/specify_cli/workflows/_commands.py +++ b/src/specify_cli/workflows/_commands.py @@ -940,6 +940,12 @@ def _workflow_run_payload(state: Any) -> dict[str, Any]: error = _failed_step_error(state) if error is not None: payload["error"] = error + if getattr(state, "execution", None): + from ._execution import scope_summaries + + scopes = scope_summaries(state.execution) + if scopes: + payload["workflow_scopes"] = scopes return payload @@ -978,21 +984,36 @@ def _gate_outcome(state: Any) -> dict[str, Any] | None: if getattr(state.status, "value", state.status) not in ("paused", "aborted"): return None step = (getattr(state, "step_results", None) or {}).get(state.current_step_id) + step_id = state.current_step_id + scope_path = None + if getattr(state, "execution", None): + from ._execution import active_step + + active = active_step(state.execution) + if active is None: + return None + path, node = active + step_id = path[-1] + scope_path = path[:-1] + step = node.get("result") if not isinstance(step, dict) or not _is_gate_step(step): return None output = step.get("output") or {} # `message`, `options`, and `choice` may be non-string YAML literals in an - # unvalidated workflow (GateStep coerces none of them for the payload), so + # legacy or synthetic records, so # normalise all three for a stable JSON schema: message → str, options → # list[str] | None, choice → str | None (None means no decision yet). message = output.get("message") choice = output.get("choice") - return { - "step_id": state.current_step_id, + detail = { + "step_id": step_id, "message": None if message is None else str(message), "options": _normalize_gate_options(output.get("options")), "choice": None if choice is None else str(choice), } + if scope_path: + detail["scope_path"] = scope_path + return detail def _normalize_gate_options(options: Any) -> list[str] | None: diff --git a/src/specify_cli/workflows/_execution.py b/src/specify_cli/workflows/_execution.py new file mode 100644 index 0000000000..7f540af0b9 --- /dev/null +++ b/src/specify_cli/workflows/_execution.py @@ -0,0 +1,576 @@ +"""A persisted execution tree, traversed identically on first execution and resume. + +An occurrence owns its result and descendants. Authored names are only aliases +in an expression context; tree positions distinguish repeated executions. +""" + +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy +from dataclasses import replace +import threading +from typing import Any + +import yaml + +from .base import StepContext, StepResult, StepStatus +from .composition import bind_inputs, evaluate_outputs, resolve_target, validate_call +from .expressions import evaluate_condition, evaluate_expression + +HALTING = {"paused", "failed", "aborted"} + + +class CheckpointError(RuntimeError): + """Persistence failed; reload the authoritative disk checkpoint before retrying.""" + + +def sequence(steps: list[dict[str, Any]]) -> dict[str, Any]: + return { + "source": yaml.safe_dump(steps, sort_keys=False), + "nodes": [{"phase": "ready"} for _ in steps], + } + + +def validate_execution(tree: Any) -> None: + """Validate stored structure without importing a project's custom steps.""" + + def check_sequence(seq): + if not isinstance(seq, dict) or not isinstance(seq.get("source"), str): + raise ValueError("Invalid execution sequence") + steps = yaml.safe_load(seq["source"]) + nodes = seq.get("nodes") + if ( + not isinstance(steps, list) + or not all(isinstance(s, dict) for s in steps) + or not isinstance(nodes, list) + or len(nodes) != len(steps) + ): + raise ValueError("Invalid execution sequence length or steps") + for step, node in zip(steps, nodes): + if not isinstance(node, dict) or node.get("phase") not in { + "ready", + "children", + "outputs", + "blocked", + "done", + }: + raise ValueError("Invalid execution phase") + result = node.get("result") + if result is not None and ( + not isinstance(result, dict) + or result.get("status") not in {s.value for s in StepStatus} + or not isinstance(result.get("output"), dict) + ): + raise ValueError("Invalid execution result") + if node["phase"] == "done" and result is None: + raise ValueError("Completed execution lacks a result") + if node.get("outcome", "completed") not in {"completed", *HALTING}: + raise ValueError("Invalid execution outcome") + if ( + node["phase"] == "done" + and node.get("outcome", "completed") != "completed" + ): + raise ValueError("Completed execution has a blocking outcome") + if node["phase"] == "blocked" and ( + result is None + or result["status"] not in {"failed", "paused"} + or node.get("outcome") not in HALTING + ): + raise ValueError("Blocked execution lacks a blocking result") + children = node.get("children", []) + if not isinstance(children, list): + raise ValueError("Invalid execution children") + for child in children: + check_sequence(child) + binding = node.get("binding") + if binding is not None: + if step.get("type") != "workflow" or not isinstance(binding, dict): + raise ValueError("Invalid workflow binding") + definition = yaml.safe_load(binding.get("definition", "")) + if ( + not isinstance(definition, dict) + or not isinstance(definition.get("workflow"), dict) + or definition["workflow"].get("id") != binding.get("workflow") + or not isinstance(binding.get("inputs"), dict) + or len(children) != 1 + or not isinstance(definition.get("steps"), list) + or definition["steps"] != yaml.safe_load(children[0]["source"]) + ): + raise ValueError("Invalid bound workflow definition or inputs") + if node["phase"] == "outputs" and binding is None: + raise ValueError("Output finalization requires a workflow binding") + if node["phase"] == "children" and ( + not children or (binding is None and result is None) + ): + raise ValueError("Expanded execution lacks its children or result") + if node["phase"] == "ready" and (result is not None or children or binding): + raise ValueError("Unstarted execution already has progress") + if step.get("type") == "fan-out" and children: + items = (result or {}).get("output", {}).get("items") + if not isinstance(items, list) or len(items) != len(children): + raise ValueError("Fan-out items do not match execution children") + + try: + if not isinstance(tree, dict) or tree.get("version") != 1: + raise ValueError("Unsupported execution version") + offset = tree.get("offset", 0) + if ( + type(offset) is not int + or offset < 0 + or not isinstance(tree.get("initial", {}), dict) + ): + raise ValueError("Invalid execution offset or initial aliases") + check_sequence(tree["sequence"]) + except (KeyError, TypeError, yaml.YAMLError, RecursionError) as exc: + raise ValueError(f"Invalid execution state: {exc}") from exc + + +def active_step(tree): + """First unfinished leaf in execution order, including nested workflow scopes.""" + + def walk(seq, path): + for index, (config, node) in enumerate( + zip(yaml.safe_load(seq["source"]), seq["nodes"]) + ): + if node["phase"] == "done": + continue + here = [*path, config.get("id", f"step-{index}")] + for item, child in enumerate(node.get("children", [])): + child_path = ( + [*here, str(item)] + if config.get("type") in {"fan-out", "while", "do-while"} + else here + ) + found = walk(child, child_path) + if found: + return found + return here, node + return None + + return walk(tree["sequence"], []) + + +def scope_summaries(tree): + """Report workflow boundaries without exposing private inputs or results.""" + summaries = [] + + def walk(seq, path): + for index, (config, node) in enumerate( + zip(yaml.safe_load(seq["source"]), seq["nodes"]) + ): + here = [*path, config.get("id", f"step-{index}")] + binding = node.get("binding") + if binding: + output = node.get("result", {}).get("output", {}) + summaries.append( + { + "scope_path": here, + "workflow_id": binding["workflow"], + "status": output.get("status", "running"), + } + ) + for item, child in enumerate(node.get("children", [])): + child_path = ( + [*here, str(item)] + if config.get("type") in {"fan-out", "while", "do-while"} + else here + ) + walk(child, child_path) + + walk(tree["sequence"], []) + return summaries + + +class Execution: + def __init__(self, engine, state, registry, *, rebind=False): + self.engine, self.state, self.registry = engine, state, registry + self.rebind = rebind + + def commit(self, node=None, changes=None, *, context=None, name=None, public=None): + """Mutate an occurrence and its compatibility views in one checkpoint.""" + with self.state._lock: + if self.state._checkpoint_failed: + raise CheckpointError("A previous checkpoint failed") + if node is not None: + node.update(changes or {}) + if context is not None and "result" in node: + context.steps[name] = node["result"] + if public is not None: + self.state.step_results[public] = node["result"] + self.state.save() + + def log(self, event, name, path, workflow, **fields): + entry = {"event": event, "step_id": name, **fields} + if path: + entry.update(execution_path=list(path), workflow_id=workflow) + self.state.append_log(entry) + + def run( + self, + seq, + context, + ancestry, + *, + path=(), + public=True, + root=False, + loop_alias=None, + ): + steps = yaml.safe_load(seq["source"]) + for index, (config, node) in enumerate(zip(steps, seq["nodes"])): + config = {"id": f"step-{index}", **config} + name = config["id"] + occurrence = (*path, index) + public_name = name if public else None + if root and node["phase"] != "done": + with self.state._lock: + self.state.current_step_index = index + self.state.execution.get( + "offset", 0 + ) + self.state.current_step_id = name + outcome = self.step( + config, node, context, ancestry, occurrence, public_name + ) + if loop_alias is not None and "result" in node: + with self.state._lock: + key = f"{loop_alias[0]}:{name}:{loop_alias[1]}" + self.state.step_results[key] = node["result"] + context.steps[key] = node["result"] + self.state.save() + if outcome in HALTING: + return outcome, node.get("error") + return "completed", None + + def restore(self, node, context): + """Rebuild aliases for completed expansions without re-executing them.""" + for child in node.get("children", []) if "binding" not in node else []: + for index, (config, nested) in enumerate( + zip(yaml.safe_load(child["source"]), child["nodes"]) + ): + if "result" in nested: + context.steps[config.get("id", f"step-{index}")] = nested["result"] + if config.get("type") != "fan-out": + self.restore(nested, context) + + def step(self, config, node, context, ancestry, path, public_name): + name = config.get("id", "step-0") + kind = config.get("type", "command") + if node["phase"] == "done": + context.steps[name] = node["result"] + if kind == "fan-out": + template = node["result"]["output"].get("step_template", {}) + for index, child in enumerate(node.get("children", [])): + result = child["nodes"][0].get("result") + if result is not None: + context.steps[ + f"{name}:{template.get('id', 'item')}:{index}" + ] = result + else: + self.restore(node, context) + return node.get("outcome", "completed") + if node.get("outcome") == "aborted": + return "aborted" + + if kind == "workflow": + return self.workflow(config, node, context, ancestry, path, public_name) + + if node["phase"] in {"ready", "blocked"}: + with self.state._lock: + self.state.current_step_id = name + self.log("step_started", name, path, ancestry[-1], type=kind) + if self.engine.on_step_start is not None: + with self.engine._callback_lock: + self.engine.on_step_start(name, config.get("command", "") or kind) + impl = self.registry.get(kind) + result = ( + impl.execute(config, context) + if impl + else StepResult(StepStatus.FAILED, error=f"Unknown step type: {kind!r}") + ) + if result.status in {StepStatus.FAILED, StepStatus.PAUSED}: + return self.finish( + config, node, result, context, ancestry, path, public_name + ) + children = [sequence(result.next_steps)] if result.next_steps else [] + if kind == "fan-out": + template = result.output.get("step_template", {}) + children = ( + [sequence([template]) for _ in result.output.get("items", [])] + if template + else [] + ) + data = self.record(config, result, context) + if not children: + if kind == "fan-out": + result.output = {**result.output, "results": []} + return self.finish( + config, node, result, context, ancestry, path, public_name + ) + self.commit( + node, + {"phase": "children", "result": data, "children": children}, + context=context, + name=name, + public=public_name, + ) + else: + context.steps[name] = node["result"] + + if kind == "fan-out": + outcome, error, outputs = self.fan_out( + config, node, context, ancestry, path, public_name + ) + data = { + **node["result"], + "output": {**node["result"]["output"], "results": outputs}, + } + self.commit( + node, {"result": data}, context=context, name=name, public=public_name + ) + else: + outcome, error = "completed", None + for iteration, child in enumerate(node["children"]): + outcome, error = self.run( + child, + context, + ancestry, + path=(*path, iteration), + public=public_name is not None, + loop_alias=(name, iteration) + if kind in {"while", "do-while"} + and iteration + and public_name is not None + else None, + ) + if outcome in HALTING: + break + if outcome == "completed" and kind in {"while", "do-while"}: + limit = config.get("max_iterations", 10) + if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1: + limit = 10 + while len(node["children"]) < limit and evaluate_condition( + config.get("condition", False), context + ): + child = sequence(yaml.safe_load(node["children"][0]["source"])) + self.commit(node, {"children": [*node["children"], child]}) + outcome, error = self.run( + child, + context, + ancestry, + path=(*path, len(node["children"]) - 1), + public=public_name is not None, + loop_alias=(name, len(node["children"]) - 1) + if public_name is not None + else None, + ) + if outcome in HALTING: + break + self.commit( + node, + { + "phase": "done" if outcome == "completed" else "children", + "outcome": outcome, + "error": error, + }, + ) + return outcome + + @staticmethod + def record(config, result, context): + output = result.output + data = { + "type": config.get("type", "command"), + "integration": output.get("integration") + or config.get("integration") + or context.default_integration, + "model": output.get("model") + or config.get("model") + or context.default_model, + "options": output.get("options") or config.get("options", {}), + "input": {} + if config.get("type") == "workflow" + else output.get("input") or config.get("input", {}), + "output": output, + "status": result.status.value, + "error": result.error, + } + if data["type"] == "command" and "integration_args" in output: + data.update( + integration_args=output["integration_args"], + integration_options=output["integration_options"], + ) + return data + + def finish(self, config, node, result, context, ancestry, path, public_name): + name = config.get("id", "step-0") + outcome = "completed" + if result.status == StepStatus.PAUSED: + outcome = "paused" + elif result.status == StepStatus.FAILED: + outcome = "aborted" if result.output.get("aborted") else "failed" + if outcome == "failed" and config.get("continue_on_error") is True: + outcome = "completed" + self.commit( + node, + { + "phase": "done" if outcome == "completed" else "blocked", + "result": self.record(config, result, context), + "outcome": outcome, + "error": result.error, + }, + context=context, + name=name, + public=public_name, + ) + self.log("step_completed", name, path, ancestry[-1], status=result.status.value) + if result.status == StepStatus.FAILED: + event = { + "aborted": "workflow_aborted", + "completed": "step_continue_on_error", + }.get(outcome, "step_failed") + self.log(event, name, path, ancestry[-1], error=result.error) + return outcome + + def workflow(self, config, node, context, ancestry, path, public_name): + from .engine import WorkflowDefinition + + name = config.get("id", "step-0") + self.log("step_started", name, path, ancestry[-1], type="workflow") + if self.engine.on_step_start is not None: + with self.engine._callback_lock: + self.engine.on_step_start(name, "workflow") + binding = node.get("binding") + target = binding["workflow"] if binding else config.get("workflow") + try: + if binding is None: + errors = validate_call(config) + if errors: + raise ValueError("; ".join(errors)) + target = evaluate_expression(target, context) + definition = resolve_target(self.state.project_root, target, ancestry) + binding = { + "workflow": target, + "definition": yaml.safe_dump(definition.data, sort_keys=False), + "inputs": bind_inputs(self.engine, definition, config, context), + "workflow_dir": str(definition.source_path.parent) + if definition.source_path + else None, + } + self.commit( + node, + { + "binding": binding, + "children": [sequence(definition.steps)], + "phase": "children", + }, + ) + else: + definition = WorkflowDefinition(yaml.safe_load(binding["definition"])) + if self.rebind: + binding = { + **binding, + "inputs": bind_inputs(self.engine, definition, config, context), + } + self.commit(node, {"binding": binding}) + child_context = StepContext( + inputs=binding["inputs"], + project_root=context.project_root, + run_id=context.run_id, + is_resume=context.is_resume, + workflow_dir=binding["workflow_dir"], + default_integration=definition.default_integration, + default_model=definition.default_model, + default_options=definition.default_options, + ) + outcome, error = self.run( + node["children"][0], + child_context, + (*ancestry, target), + path=(*path, "workflow"), + public=False, + ) + output = {"workflow": target, "status": outcome} + if outcome == "completed": + self.commit(node, {"phase": "outputs"}) + output.update(evaluate_outputs(definition, child_context)) + elif outcome == "aborted": + output["aborted"] = True + if error is not None: + output["error"] = error + status = ( + StepStatus.COMPLETED + if outcome == "completed" + else StepStatus.PAUSED + if outcome == "paused" + else StepStatus.FAILED + ) + result = StepResult(status, output=output, error=error) + except CheckpointError: + raise + except Exception as exc: + target = target if isinstance(target, str) else repr(target) + result = StepResult( + StepStatus.FAILED, + output={"workflow": target, "status": "failed", "error": str(exc)}, + error=str(exc), + ) + return self.finish(config, node, result, context, ancestry, path, public_name) + + def fan_out(self, config, node, context, ancestry, path, public_name): + output = node["result"]["output"] + items = output.get("items", []) + try: + workers = max(1, int(output.get("max_concurrency", 1))) + except (TypeError, ValueError, OverflowError): + workers = 1 + workers = min(workers, len(items)) + initial = deepcopy(context.steps) + halted = threading.Event() + + def run_item(index): + local = replace( + context, steps=deepcopy(initial), item=items[index], inside_fan_out=True + ) + child = node["children"][index] + outcome, error = self.run( + child, local, ancestry, path=(*path, "item", index), public=False + ) + if outcome in HALTING: + halted.set() + record = child["nodes"][0].get("result", {}) + template_name = output.get("step_template", {}).get("id", "item") + with self.state._lock: + key = f"{config['id']}:{template_name}:{index}" + context.steps[key] = record + if public_name is not None: + self.state.step_results[key] = record + self.state.save() + return outcome, error, record.get("output", {}) + + results = [] + if workers <= 1: + for index in range(len(items)): + outcome, error, value = run_item(index) + results.append(value) + if outcome in HALTING: + return outcome, error, results + return "completed", None, results + with ThreadPoolExecutor(max_workers=workers) as pool: + futures = {i: pool.submit(run_item, i) for i in range(workers)} + for index in range(len(items)): + try: + outcome, error, value = futures.pop(index).result() + except BaseException: + for future in futures.values(): + future.cancel() + raise + results.append(value) + if outcome in HALTING: + for future in futures.values(): + future.cancel() + return outcome, error, results + following = index + workers + if following < len(items) and not halted.is_set(): + futures[following] = pool.submit(run_item, following) + return "completed", None, results diff --git a/src/specify_cli/workflows/command_resume.py b/src/specify_cli/workflows/command_resume.py index 3be24f9582..9a50a6b50c 100644 --- a/src/specify_cli/workflows/command_resume.py +++ b/src/specify_cli/workflows/command_resume.py @@ -18,7 +18,7 @@ def workflow_resume( help="Emit the resume outcome as a single JSON object instead of formatted text.", ), ): - """Resume a paused or failed workflow run.""" + """Resume a paused, failed, or crash-interrupted workflow run.""" from . import load_custom_steps from .engine import RunState, WorkflowEngine diff --git a/src/specify_cli/workflows/command_status.py b/src/specify_cli/workflows/command_status.py index d9af4b48d6..f58c990698 100644 --- a/src/specify_cli/workflows/command_status.py +++ b/src/specify_cli/workflows/command_status.py @@ -90,6 +90,15 @@ def workflow_status( s, "white" ) cli.console.print(f" [{sc}]●[/{sc}] {step_id}: {s}") + if state.execution: + from ._execution import scope_summaries + + for scope in scope_summaries(state.execution): + path = " / ".join(scope["scope_path"]) + cli.console.print( + f" {cli._escape_markup(path)} → " + f"{cli._escape_markup(scope['workflow_id'])}: {scope['status']}" + ) else: runs = engine.list_runs() diff --git a/src/specify_cli/workflows/composition.py b/src/specify_cli/workflows/composition.py new file mode 100644 index 0000000000..02c1156177 --- /dev/null +++ b/src/specify_cli/workflows/composition.py @@ -0,0 +1,147 @@ +"""Typed workflow-call boundary; execution and persistence belong to the engine.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .base import StepContext +from .expressions import evaluate_expression + +RESERVED_OUTPUT_NAMES = frozenset( + { + "workflow", + "status", + "error", + "aborted", + "integration", + "model", + "options", + "input", + } +) +MAX_COMPOSITION_DEPTH = 16 + + +def require_json(value: Any) -> None: + """Reject lossy/non-JSON values, including cycles, before checkpointing.""" + try: + encoded = json.dumps(value, allow_nan=False) + if json.loads(encoded) != value: + raise ValueError("JSON encoding changes the value") + except (TypeError, ValueError, RecursionError) as exc: + raise ValueError(f"Value is not JSON-safe: {exc}") from exc + + +def validate_call(config: dict[str, Any]) -> list[str]: + from .engine import _ID_PATTERN + from .overlay.schema import _RESERVED_WORKFLOW_IDS + + errors = [] + target = config.get("workflow") + if not isinstance(target, str): + errors.append("'workflow' must be a string") + elif "{{" not in target and ( + not _ID_PATTERN.fullmatch(target) or target in _RESERVED_WORKFLOW_IDS + ): + errors.append("'workflow' must be an exact, safe, non-reserved workflow ID") + mapping = config.get("input", {}) + if mapping is not None: + if not isinstance(mapping, dict) or any( + not isinstance(k, str) for k in mapping + ): + errors.append("'input' must be a mapping with string keys") + else: + try: + require_json(mapping) + except ValueError as exc: + errors.append(str(exc)) + return errors + + +def validate_outputs(outputs: Any) -> list[str]: + from .engine import _ID_PATTERN + + if not isinstance(outputs, dict): + return ["'outputs' must be a mapping"] + errors = [] + for name, entry in outputs.items(): + if not isinstance(name, str) or not _ID_PATTERN.fullmatch(name): + errors.append(f"Output {name!r} must be a safe identifier") + elif name in RESERVED_OUTPUT_NAMES: + errors.append(f"Output {name!r} is reserved") + if not isinstance(entry, dict) or set(entry) != {"value"}: + errors.append(f"Output {name!r} must contain exactly 'value'") + else: + try: + require_json(entry["value"]) + except ValueError as exc: + errors.append(f"Output {name!r}: {exc}") + return errors + + +def resolve_target(project_root: Path, target: Any, ancestry: tuple[str, ...]): + from .catalog import WorkflowRegistry + from .engine import _ID_PATTERN, validate_workflow + from .overlay import WorkflowResolver + from .overlay.schema import _RESERVED_WORKFLOW_IDS + + if not isinstance(target, str) or not _ID_PATTERN.fullmatch(target): + raise ValueError("Workflow target must be an exact safe workflow ID") + if target in _RESERVED_WORKFLOW_IDS: + raise ValueError(f"Workflow {target!r} is reserved") + if target in ancestry: + raise ValueError( + f"Workflow composition cycle: {' -> '.join((*ancestry, target))}" + ) + if len(ancestry) > MAX_COMPOSITION_DEPTH: + raise ValueError( + f"Workflow composition exceeds maximum depth {MAX_COMPOSITION_DEPTH}" + ) + metadata = WorkflowRegistry(project_root).get(target) + if not isinstance(metadata, dict): + raise ValueError(f"Workflow {target!r} is not installed") + if not metadata.get("enabled", True): + raise ValueError(f"Workflow {target!r} is disabled") + definition = WorkflowResolver(project_root).resolve(target) + if definition.id != target: + raise ValueError( + f"Workflow {target!r} resolves to mismatched ID {definition.id!r}" + ) + errors = validate_workflow(definition) + if errors: + raise ValueError(f"Invalid workflow {target!r}: {'; '.join(errors)}") + return definition + + +def bind_inputs(engine, definition, config: dict[str, Any], context: StepContext): + mapping = config.get("input") + if mapping is None: + mapping = {} + if not isinstance(mapping, dict) or any(not isinstance(k, str) for k in mapping): + raise ValueError("'input' must be a mapping with string keys") + provided = { + key: evaluate_expression(value, context) for key, value in mapping.items() + } + require_json(provided) + unknown = provided.keys() - definition.inputs.keys() + if unknown: + raise ValueError( + f"Undeclared inputs for workflow {definition.id!r}: {sorted(unknown)}" + ) + resolved = engine._resolve_inputs(definition, provided) + require_json(resolved) + return resolved + + +def evaluate_outputs(definition, context: StepContext) -> dict[str, Any]: + errors = validate_outputs(definition.outputs) + if errors: + raise ValueError("; ".join(errors)) + output = { + name: evaluate_expression(entry["value"], context) + for name, entry in definition.outputs.items() + } + require_json(output) + return output diff --git a/src/specify_cli/workflows/engine.py b/src/specify_cli/workflows/engine.py index d81aae3212..d947f355c5 100644 --- a/src/specify_cli/workflows/engine.py +++ b/src/specify_cli/workflows/engine.py @@ -10,14 +10,12 @@ from __future__ import annotations -import dataclasses import json import os import re import tempfile import threading import uuid -from concurrent.futures import Future, ThreadPoolExecutor from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -28,7 +26,7 @@ default_integration_key, try_read_integration_json, ) -from .base import RunStatus, StepContext, StepResult, StepStatus +from .base import RunStatus, StepContext # -- Workflow Definition -------------------------------------------------- @@ -86,6 +84,7 @@ def __init__(self, data: dict[str, Any], source_path: Path | None = None) -> Non # Inputs self.inputs: dict[str, Any] = data.get("inputs", {}) + self.outputs: Any = data.get("outputs", {}) # Steps self.steps: list[dict[str, Any]] = data.get("steps", []) @@ -140,7 +139,7 @@ def _get_valid_step_types() -> set[str]: return set(STEP_REGISTRY.keys()) return { "command", "shell", "prompt", "gate", "if", "init", "slot", - "switch", "while", "do-while", "fan-out", "fan-in", + "switch", "while", "do-while", "fan-out", "fan-in", "workflow", } @@ -183,7 +182,9 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]: An empty list means the workflow is valid. """ - errors: list[str] = [] + from .composition import validate_outputs + + errors: list[str] = validate_outputs(definition.outputs) # -- Schema version --------------------------------------------------- # str() so an unquoted ``schema_version: 1.0`` (YAML float) is accepted — @@ -695,10 +696,14 @@ def __init__( self.current_step_index = 0 self.current_step_id: str | None = None self.step_results: dict[str, dict[str, Any]] = {} + self.execution: dict[str, Any] | None = None + self._checkpoint_failed = False # Guards step_results mutation and save() so a concurrent fan-out cannot # mutate the dict while save() is serializing it (which would raise # "dictionary changed size during iteration"). - self._lock = threading.Lock() + self._lock = threading.RLock() + # Reentrant so an execution transition can update views and call save() + # under the same lock. Step implementations run outside that lock. # Serializes append_log's list append + log.jsonl write so concurrent # fan-out workers cannot interleave or corrupt log lines. Kept separate # from _lock so frequent logging never contends with state saves; since @@ -723,6 +728,10 @@ def record_step_result(self, step_id: str, data: dict[str, Any]) -> None: fan-out). For a sequential run this is an uncontended lock. """ with self._lock: + from ._execution import CheckpointError + + if self._checkpoint_failed: + raise CheckpointError("A previous checkpoint failed; reload the run") self.step_results[step_id] = data def set_step_output(self, step_id: str, output: Any) -> None: @@ -749,6 +758,10 @@ def save(self) -> None: runs_dir.mkdir(parents=True, exist_ok=True) with self._lock: + if self._checkpoint_failed: + from ._execution import CheckpointError + + raise CheckpointError("A previous checkpoint failed; reload the run") # Stamp updated_at inside the lock so the timestamp matches the # snapshot this thread serializes (concurrent savers don't race it). self.updated_at = datetime.now(timezone.utc).isoformat() @@ -765,9 +778,18 @@ def save(self) -> None: "created_at": self.created_at, "updated_at": self.updated_at, "error": self.error, + "inputs": self.inputs, } - self._atomic_write_json(runs_dir / "state.json", state_data) - self._atomic_write_json(runs_dir / "inputs.json", {"inputs": self.inputs}) + if self.execution is not None: + state_data["execution"] = self.execution + try: + self._atomic_write_json(runs_dir / "state.json", state_data) + self._atomic_write_json(runs_dir / "inputs.json", {"inputs": self.inputs}) + except BaseException as exc: + from ._execution import CheckpointError + + self._checkpoint_failed = True + raise CheckpointError(str(exc)) from exc @staticmethod def _atomic_write_json(path: Path, data: dict[str, Any]) -> None: @@ -892,9 +914,14 @@ def load(cls, run_id: str, project_root: Path) -> RunState: state.created_at = state_data.get("created_at", "") state.updated_at = state_data.get("updated_at", "") state.error = state_data.get("error") + state.execution = state_data.get("execution") + if "execution" in state_data: + from ._execution import validate_execution + + validate_execution(state.execution) inputs_path = runs_dir / "inputs.json" - if inputs_path.exists(): + if "inputs" not in state_data and inputs_path.exists(): with open(inputs_path, encoding="utf-8") as f: inputs_data = json.load(f) if not isinstance(inputs_data, dict): @@ -908,6 +935,11 @@ def load(cls, run_id: str, project_root: Path) -> RunState: ) state.inputs = inputs + if "inputs" in state_data: + if not isinstance(state_data["inputs"], dict): + raise ValueError("Invalid run inputs: 'inputs' must be a JSON object") + state.inputs = state_data["inputs"] + return state def append_log(self, entry: dict[str, Any]) -> None: @@ -1089,6 +1121,8 @@ def execute( state.save() return state except Exception as exc: + if state._checkpoint_failed: + raise state.status = RunStatus.FAILED state.error = str(exc) state.append_log({"event": "workflow_failed", "error": str(exc)}) @@ -1106,7 +1140,7 @@ def resume( run_id: str, inputs: dict[str, Any] | None = None, ) -> RunState: - """Resume a paused or failed workflow run. + """Resume a paused/failed run or a tree-backed run interrupted by a crash. When ``inputs`` is provided, the values are merged over the run's persisted inputs and re-resolved through the same typed validation @@ -1115,7 +1149,9 @@ def resume( empty/``None`` ``inputs`` leaves the run's inputs unchanged. """ state = RunState.load(run_id, self.project_root) - if state.status not in (RunStatus.PAUSED, RunStatus.FAILED): + if state.status not in (RunStatus.PAUSED, RunStatus.FAILED) and not ( + state.status == RunStatus.RUNNING and state.execution is not None + ): msg = f"Cannot resume run {run_id!r} with status {state.status.value!r}." raise ValueError(msg) @@ -1144,6 +1180,12 @@ def resume( ) raise ValueError(msg) + if state.execution is not None: + persisted_steps = yaml.safe_load(state.execution["sequence"]["source"]) + offset = state.execution.get("offset", 0) + if persisted_steps != definition.steps[offset:]: + raise ValueError("Invalid execution state: root sequence differs from workflow snapshot") + dispatch_default_errors = _dispatch_default_errors(definition) if dispatch_default_errors: raise ValueError(" ".join(dispatch_default_errors)) @@ -1182,6 +1224,7 @@ def resume( self._execute_steps( remaining_steps, context, state, STEP_REGISTRY, step_offset=step_offset, + rebind=bool(inputs), ) except KeyboardInterrupt: state.status = RunStatus.PAUSED @@ -1189,6 +1232,8 @@ def resume( state.save() return state except Exception as exc: + if state._checkpoint_failed: + raise state.status = RunStatus.FAILED state.error = str(exc) state.append_log({"event": "resume_failed", "error": str(exc)}) @@ -1225,240 +1270,26 @@ def _execute_steps( registry: dict[str, Any], *, step_offset: int = 0, + rebind: bool = False, ) -> None: - """Execute a list of steps sequentially.""" - for i, step_config in enumerate(steps): - step_id = step_config.get("id", f"step-{i}") - step_type = step_config.get("type", "command") - - state.current_step_id = step_id - if step_offset >= 0: - state.current_step_index = step_offset + i - state.save() - - state.append_log( - {"event": "step_started", "step_id": step_id, "type": step_type} - ) - - # Log progress — use the engine's on_step_start callback if set, - # otherwise stay silent (library-safe default). - label = step_config.get("command", "") or step_type - if self.on_step_start is not None: - with self._callback_lock: - self.on_step_start(step_id, label) - - step_impl = registry.get(step_type) - if not step_impl: - state.status = RunStatus.FAILED - state.error = f"Unknown step type: {step_type!r}" - state.append_log( - { - "event": "step_failed", - "step_id": step_id, - "error": f"Unknown step type: {step_type!r}", - } - ) - state.save() - return - - result: StepResult = step_impl.execute(step_config, context) - - # Record step results — prefer resolved values from step output - step_data = { - "type": step_type, - "integration": result.output.get("integration") - or step_config.get("integration") - or context.default_integration, - "model": result.output.get("model") - or step_config.get("model") - or context.default_model, - "options": result.output.get("options") - or step_config.get("options", {}), - "input": result.output.get("input") - or step_config.get("input", {}), - "output": result.output, - "status": result.status.value, - "error": result.error, - } - if step_type == "command" and "integration_args" in result.output: - step_data["integration_args"] = result.output["integration_args"] - step_data["integration_options"] = result.output[ - "integration_options" - ] - self._record_result(context, state, step_id, step_data) - - state.append_log( - { - "event": "step_completed", - "step_id": step_id, - "status": result.status.value, - } - ) - - # Handle gate pauses - if result.status == StepStatus.PAUSED: - state.status = RunStatus.PAUSED - state.save() - return - - # Handle failures - if result.status == StepStatus.FAILED: - # Gate abort (output.aborted) maps to ABORTED status. - # Aborts are deliberate operator decisions, so - # `continue_on_error` does NOT override them — that flag - # is for transient/expected step failures only. - if result.output.get("aborted"): - state.status = RunStatus.ABORTED - state.error = result.error - state.append_log( - { - "event": "workflow_aborted", - "step_id": step_id, - } - ) - state.save() - return - - # `continue_on_error: true` lets the pipeline route - # around the failure instead of halting. The step - # result (including exit_code, stderr, status) is - # still recorded so a downstream `if` or `switch` - # can branch on it (or a `gate` can surface it to the - # operator via message interpolation). Log a single, - # unambiguous event per failure resolution — either - # the run continued past it, or it halted. - # - # Use identity comparison (`is True`) rather than - # truthiness so that only a literal boolean enables - # the behaviour, even if validation was skipped. - # Validation rejects non-bool values at parse time, - # but `WorkflowEngine.execute()` does not auto-validate - # (see `WorkflowEngine.load_workflow`, whose docstring - # explicitly notes "not yet validated; call - # `validate_workflow()` or `engine.validate()` - # separately"), so a caller passing an unvalidated - # definition could otherwise see truthy non-bool - # values like the string `"true"` silently change - # run semantics. - if step_config.get("continue_on_error") is True: - state.append_log( - { - "event": "step_continue_on_error", - "step_id": step_id, - "error": result.error, - } - ) - state.save() - continue - - state.status = RunStatus.FAILED - state.error = result.error - state.append_log( - { - "event": "step_failed", - "step_id": step_id, - "error": result.error, - } - ) - state.save() - return - - # Execute nested steps (from control flow) - # NOTE: Nested steps run with step_offset=-1 so they don't - # update current_step_index. If a nested step pauses, - # resume will re-run the parent step and its nested body. - # A step-path stack for exact nested resume is a future - # enhancement. - if result.next_steps: - self._execute_steps( - result.next_steps, context, state, registry, - step_offset=-1, - ) - if state.status in ( - RunStatus.PAUSED, - RunStatus.FAILED, - RunStatus.ABORTED, - ): - return - - # Loop iteration: while/do-while re-evaluate after body - if step_type in ("while", "do-while"): - from .expressions import evaluate_condition - - max_iters = step_config.get("max_iterations") - # A bool is an int in Python (isinstance(True, int) is True - # and True == 1), so a bool max_iterations would slip past - # the int check and cap the loop at range(0)==1 iteration - # instead of the default. Exclude bools, mirroring the - # while/do-while validators and the continue_on_error guard. - if ( - isinstance(max_iters, bool) - or not isinstance(max_iters, int) - or max_iters < 1 - ): - max_iters = 10 - condition = step_config.get("condition", False) - for _loop_iter in range(max_iters - 1): - if not evaluate_condition(condition, context): - break - # Namespace nested step IDs per iteration - # so logs and state keys are unique. - # Execute one step at a time and alias each - # result back to the unprefixed key so that - # later steps in the same body and the loop - # condition see the latest values. - for ns_idx, ns in enumerate(result.next_steps): - ns_copy = dict(ns) - orig = ns_copy.get("id") - base_id = orig or f"step-{ns_idx}" - ns_copy["id"] = f"{step_id}:{base_id}:{_loop_iter + 1}" - self._execute_steps( - [ns_copy], context, state, registry, - step_offset=-1, - ) - if state.status in ( - RunStatus.PAUSED, - RunStatus.FAILED, - RunStatus.ABORTED, - ): - return - if orig and ns_copy["id"] in context.steps: - self._record_result( - context, state, orig, - context.steps[ns_copy["id"]], - ) - - # Fan-out: execute the nested step template once per item. Honors - # max_concurrency — <=1 runs sequentially (default, historical - # behavior); >1 runs up to that many items concurrently. Either way - # results are assembled in item order under the - # parentId:templateId:index id grammar. - if step_type == "fan-out": - items = result.output.get("items", []) - template = result.output.get("step_template", {}) - if template and items: - fan_out_results = self._run_fan_out( - items, template, step_id, context, state, registry, - result.output.get("max_concurrency", 1), - ) - context.item = None - # Preserve original output and add collected results - fan_out_output = dict(result.output) - fan_out_output["results"] = fan_out_results - # set_step_output updates the recorded dict under the run lock; - # context.steps[step_id] is that same object, so it reflects the - # change too — no separate (unlocked) context mutation needed. - state.set_step_output(step_id, fan_out_output) - if state.status in ( - RunStatus.PAUSED, - RunStatus.FAILED, - RunStatus.ABORTED, - ): - return - else: - # Empty items or no template — normalize output - result.output["results"] = [] - state.set_step_output(step_id, result.output) + """Execute or resume the persisted tree (legacy indices adapt once).""" + from ._execution import Execution, active_step, sequence + from copy import deepcopy + + if state.execution is None: + state.execution = {"version": 1, "sequence": sequence(steps)} + state.execution["offset"] = max(0, step_offset) + state.execution["initial"] = deepcopy(context.steps) + context.steps = deepcopy(state.execution.get("initial", {})) + state.save() + tree = state.execution["sequence"] + executor = Execution(self, state, registry, rebind=rebind) + outcome, error = executor.run(tree, context, (state.workflow_id,), root=True) + active = active_step(state.execution) + if active is not None: + state.current_step_id = active[0][-1] + state.status = RunStatus.RUNNING if outcome == "completed" else RunStatus(outcome) + state.error = error def _run_fan_out( self, @@ -1470,184 +1301,24 @@ def _run_fan_out( registry: dict[str, Any], max_concurrency: Any, ) -> list[Any]: - """Run a fan-out template once per item; return per-item outputs in item order. - - ``max_concurrency`` <= 1 (the default) runs items sequentially, identical - to the historical fan-out behavior. ``max_concurrency`` > 1 runs items on a - bounded thread pool using a sliding submission window of that size: at most - that many items are ever in flight, and no new item is launched once the run - has reached a halting status, so a halt cannot keep starting queued work. - - Results are always returned in item order (never completion order). On a - halt (PAUSED/FAILED/ABORTED) the returned prefix is the items up to and - including the first item *in item order* whose own execution halted the run - — identical to the sequential path. Later items that have not yet started - are cancelled; any already running are allowed to finish but their outputs - are ignored. Halt is attributed per item from that item's recorded result - (not the shared run status, which a concurrently-running later item may have - already flipped), so the prefix never drops the actual halting item. - - ``max_concurrency`` is coerced with ``int()``; a value that cannot be - coerced (``None``, a non-numeric string, ``.inf``/``.nan``, …) or that - coerces to <= 1 runs sequentially, while a numeric string like ``"4"`` or - a float like ``4.0`` is honored. - """ + """Compatibility adapter to the unified fan-out executor.""" + from ._execution import Execution, sequence + if not items: return [] - - halting = (RunStatus.PAUSED, RunStatus.FAILED, RunStatus.ABORTED) - try: - workers = max(1, int(max_concurrency)) - except (TypeError, ValueError, OverflowError): - # OverflowError: int(float("inf")) — a YAML ``max_concurrency: .inf`` - # would otherwise crash the whole run instead of falling back. - workers = 1 - # Never spin up more workers than there is work — bounds a user-controlled - # max_concurrency from over-allocating threads. - workers = min(workers, len(items)) - - base_id = template.get("id", "item") - - def item_id(idx: int) -> str: - # Per-item ID grammar: parentId:templateId:index. - return f"{step_id}:{base_id}:{idx}" - - def run_item(idx: int, item_ctx: StepContext) -> Any: - item_step = dict(template) - item_step["id"] = item_id(idx) - self._execute_steps( - [item_step], item_ctx, state, registry, step_offset=-1, - ) - # Read back through the context that was actually executed against, - # not the outer closure — clearer and robust if StepContext copying - # ever stops sharing the steps dict by reference. - return item_ctx.steps.get(item_step["id"], {}).get("output", {}) - - # Sequential path — identical to the historical behavior. - if workers <= 1: - results: list[Any] = [] - previous_item = context.item - previous_inside_fan_out = context.inside_fan_out - context.inside_fan_out = True - try: - for item_idx, item_val in enumerate(items): - context.item = item_val - results.append(run_item(item_idx, context)) - if state.status in halting: - break - finally: - context.item = previous_item - context.inside_fan_out = previous_inside_fan_out - return results - - # Concurrent path — bounded sliding window; results assembled in item order. - n = len(items) - slots: list[Any] = [None] * n - - def run_isolated(idx: int) -> Any: - # Each item runs against its own context copy so context.item is not - # clobbered across threads; the shared steps dict is written only on the - # disjoint parentId:templateId:index key (GIL-safe on distinct keys). - return run_item( - idx, - dataclasses.replace( - context, - item=items[idx], - inside_fan_out=True, - ), - ) - - def item_halt_status(idx: int) -> RunStatus | None: - # If THIS item's own execution halted the run, return the resulting run - # status; else None. Decided from the item's own recorded result, not - # the shared run status, so a later item's concurrent halt is never - # misattributed here. Mirrors the sequential mapping: PAUSED -> PAUSED; - # FAILED -> ABORTED when aborted, else FAILED, unless continue_on_error - # routes around it. - rec = context.steps.get(item_id(idx)) - if rec is None: - # Ran but recorded nothing — only when the item failed before - # record_step_result (e.g. an unknown step type returns early). - # Every item runs the same template, so the shared run status is - # this item's own outcome; attribute the halt to it. - return state.status if state.status in halting else None - status = rec.get("status") - if status == StepStatus.PAUSED.value: - return RunStatus.PAUSED - if status == StepStatus.FAILED.value: - out = rec.get("output") or {} - if out.get("aborted"): - return RunStatus.ABORTED - if template.get("continue_on_error") is not True: - return RunStatus.FAILED - return None - - # (halting item index, its run status) once a halt is attributed. - halt: tuple[int, RunStatus] | None = None - collected = 0 - with ThreadPoolExecutor(max_workers=workers) as pool: - futures: dict[int, Future] = {} - next_submit = 0 - for idx in range(n): - # Refill the window: keep <= workers in flight, and stop launching - # new items once the run is halting so a halt cannot keep starting - # queued work. Already-submitted futures are still collected in - # item order below. - while ( - next_submit < n - and len(futures) < workers - and state.status not in halting - ): - futures[next_submit] = pool.submit(run_isolated, next_submit) - next_submit += 1 - - fut = futures.pop(idx, None) - if fut is None: - # Safety net: the window submits indices in order and the loop - # breaks at the first halting item, so every collected index has - # an in-flight future. Stop cleanly rather than raise if a future - # change ever breaks that invariant. - break - try: - slots[idx] = fut.result() - except Exception: - # A genuine exception escaping a step (not a normal step - # FAILED, which sets state.status) must not be masked: cancel - # outstanding work and re-raise — with a bare ``raise`` so the - # original traceback is preserved — so the engine marks the run - # failed instead of reporting a vacuous completion. The pool's - # __exit__ still joins any already-running workers. - for other in futures.values(): - other.cancel() - raise - collected = idx + 1 - halt_status = item_halt_status(idx) - if halt_status is not None: - # First halting item in item order: include it (slots[idx] is - # already set), record its status, and cancel everything pending. - halt = (idx, halt_status) - for other in futures.values(): - other.cancel() - break - - if halt is not None: - halted_at, halted_status = halt - # A later in-flight item may have overwritten state.status before the - # pool joined; restore the halting item's own outcome so the final run - # status matches the sequential semantics. - state.status = halted_status - # Restore the halting item's error so it matches the terminal - # status — a concurrent item may have overwritten state.error - # before the pool joined. Assign unconditionally when a record - # exists (even when the halting item's own error is falsy) so a - # third-party step returning FAILED with no message never inherits - # an unrelated concurrent item's error; this mirrors the sequential - # path, which sets state.error = result.error verbatim. - halt_rec = context.steps.get(item_id(halted_at)) - if isinstance(halt_rec, dict): - state.error = halt_rec.get("error") - return slots[: halted_at + 1] - return slots[:collected] + node = { + "phase": "children", + "result": {"output": {"items": items, "step_template": template, + "max_concurrency": max_concurrency}}, + "children": [sequence([template]) for _ in items], + } + outcome, error, results = Execution(self, state, registry).fan_out( + {"id": step_id, "type": "fan-out"}, node, context, + (state.workflow_id,), (), step_id, + ) + state.status = RunStatus.RUNNING if outcome == "completed" else RunStatus(outcome) + state.error = error + return results def _resolve_inputs( self, diff --git a/src/specify_cli/workflows/step/gate/__init__.py b/src/specify_cli/workflows/step/gate/__init__.py index 5aac060c0f..111ca960ed 100644 --- a/src/specify_cli/workflows/step/gate/__init__.py +++ b/src/specify_cli/workflows/step/gate/__init__.py @@ -40,6 +40,8 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult: message = config.get("message", "Review required.") if isinstance(message, str) and "{{" in message: message = evaluate_expression(message, context) + if message is not None: + message = str(message) options = config.get("options", ["approve", "reject"]) on_reject = config.get("on_reject", "abort") diff --git a/src/specify_cli/workflows/step/workflow/__init__.py b/src/specify_cli/workflows/step/workflow/__init__.py new file mode 100644 index 0000000000..0c8f1df328 --- /dev/null +++ b/src/specify_cli/workflows/step/workflow/__init__.py @@ -0,0 +1,17 @@ +"""Installed workflow calls are scoped execution handled by the engine.""" + +from ...base import StepBase, StepResult, StepStatus +from ...composition import validate_call + + +class WorkflowStep(StepBase): + type_key = "workflow" + + def validate(self, config): + return [*super().validate(config), *validate_call(config)] + + def execute(self, config, context): + return StepResult( + status=StepStatus.FAILED, + error="Workflow calls require the workflow engine", + ) diff --git a/tests/specify_cli/workflows/test_command_status.py b/tests/specify_cli/workflows/test_command_status.py index 0a7b50ac13..16f6a01352 100644 --- a/tests/specify_cli/workflows/test_command_status.py +++ b/tests/specify_cli/workflows/test_command_status.py @@ -95,6 +95,38 @@ def test_status_json_single_and_list(self, project_dir): ) assert any(r["run_id"] == rid for r in listing["runs"]) + def test_composed_gate_status_and_resume(self, project_dir): + from specify_cli.workflows.catalog import WorkflowRegistry + + child_dir = project_dir / ".specify" / "workflows" / "child" + child_dir.mkdir(parents=True) + (child_dir / "workflow.yml").write_text( + "workflow: {id: child, name: Child}\n" + "inputs:\n verdict: {type: string, default: ''}\n" + "steps:\n - {id: review, type: gate, message: Review, verdict_input: verdict}\n", + encoding="utf-8", + ) + WorkflowRegistry(project_dir).add("child", {"enabled": True}) + root = self._write_wf( + project_dir, + "workflow: {id: parent, name: Parent}\n" + "inputs:\n verdict: {type: string, default: ''}\n" + "steps:\n - id: call\n type: workflow\n workflow: child\n" + " input: {verdict: '{{ inputs.verdict }}'}\n", + "parent", + ) + run = json.loads(self._invoke(project_dir, ["workflow", "run", str(root), "--json"]).stdout) + assert run["status"] == "paused", run + status = json.loads(self._invoke(project_dir, ["workflow", "status", run["run_id"], "--json"]).stdout) + assert status["gate"] == run["gate"] + assert status["gate"]["scope_path"] == ["call"] + assert status["workflow_scopes"] == [{"scope_path": ["call"], "workflow_id": "child", "status": "paused"}] + human = self._invoke(project_dir, ["workflow", "status", run["run_id"]]) + assert "child: paused" in human.stdout + resumed = self._invoke(project_dir, ["workflow", "resume", run["run_id"], "--input", "verdict=approve", "--json"]) + assert resumed.exit_code == 0 + assert json.loads(resumed.stdout)["status"] == "completed" + class TestWorkflowCliAlignment: diff --git a/tests/workflows/test_composition_execution.py b/tests/workflows/test_composition_execution.py new file mode 100644 index 0000000000..ae1bc88859 --- /dev/null +++ b/tests/workflows/test_composition_execution.py @@ -0,0 +1,670 @@ +"""Behavioral contracts for scoped calls and durable execution occurrences.""" + +from collections import Counter +from datetime import date +import json +import threading + +import pytest +import yaml + +from specify_cli.workflows import STEP_REGISTRY +from specify_cli.workflows.base import RunStatus, StepBase, StepResult, StepStatus +from specify_cli.workflows.engine import RunState, WorkflowDefinition, WorkflowEngine + + +def definition(name, steps, **fields): + return WorkflowDefinition( + {"workflow": {"id": name, "name": name}, "steps": steps, **fields} + ) + + +def install(root, child, enabled=True): + from specify_cli.workflows.catalog import WorkflowRegistry + + directory = root / ".specify" / "workflows" / child.id + directory.mkdir(parents=True, exist_ok=True) + (directory / "workflow.yml").write_text( + yaml.safe_dump(child.data), encoding="utf-8" + ) + registry = WorkflowRegistry(root) + registry.add(child.id, {"version": "1.0.0", "enabled": enabled}) + return directory + + +def call(target="child", **extra): + return {"id": "call", "type": "workflow", "workflow": target, **extra} + + +@pytest.fixture +def probe(monkeypatch): + counts = Counter() + + class Probe(StepBase): + type_key = "probe" + + def execute(self, config, context): + from specify_cli.workflows.expressions import evaluate_expression + + counts[config["id"]] += 1 + value = evaluate_expression(config.get("value"), context) + status = config.get("status", "completed") + if config.get("await") and not context.inputs.get("approve"): + status = "paused" + return StepResult( + StepStatus(status), output={"value": value, **config.get("output", {})} + ) + + monkeypatch.setitem(STEP_REGISTRY, "probe", Probe()) + return counts + + +def test_declared_output_and_scope_isolation(tmp_path, probe): + child = definition( + "child", + [ + {"id": "inspect", "type": "probe", "value": "{{ inputs.value }}"}, + { + "id": "private", + "type": "probe", + "value": "{{ steps.parent.output.value }}", + }, + ], + inputs={"value": {"type": "string"}}, + outputs={ + "value": {"value": "{{ steps.inspect.output.value }}"}, + "hidden": {"value": "{{ steps.private.output.value }}"}, + }, + ) + install(tmp_path, child) + state = WorkflowEngine(tmp_path).execute( + definition( + "parent", + [ + {"id": "parent", "type": "probe", "value": "secret"}, + call(input={"value": "mapped"}), + { + "id": "consume", + "type": "probe", + "value": "{{ steps.call.output.value }}", + }, + ], + ) + ) + assert state.status == RunStatus.COMPLETED + assert state.step_results["call"]["output"] == { + "workflow": "child", + "status": "completed", + "value": "mapped", + "hidden": None, + } + assert state.step_results["consume"]["output"]["value"] == "mapped" + assert "inspect" not in state.step_results + assert len(list((tmp_path / ".specify/workflows/runs").iterdir())) == 1 + + +def test_concurrent_nested_calls_keep_downstream_aliases_local( + tmp_path, monkeypatch, probe +): + install( + tmp_path, + definition( + "child", + [{"id": "work", "type": "probe"}], + inputs={"value": {"type": "number"}}, + outputs={"value": {"value": "{{ inputs.value }}"}}, + ), + ) + barrier = threading.Barrier(2, timeout=5) + consumed = {} + + class Consume(StepBase): + type_key = "consume" + + def execute(self, config, context): + barrier.wait() + consumed[context.item] = context.steps["call"]["output"]["value"] + return StepResult(StepStatus.COMPLETED) + + monkeypatch.setitem(STEP_REGISTRY, "consume", Consume()) + state = WorkflowEngine(tmp_path).execute( + definition( + "parent", + [ + { + "id": "spread", + "type": "fan-out", + "items": [1, 2], + "max_concurrency": 2, + "step": { + "id": "branch", + "type": "if", + "condition": True, + "then": [ + call(input={"value": "{{ item }}"}), + {"id": "consume", "type": "consume"}, + ], + }, + } + ], + ) + ) + assert state.status == RunStatus.COMPLETED + assert consumed == {1: 1, 2: 2} + items = RunState.load(state.run_id, tmp_path).execution["sequence"]["nodes"][0][ + "children" + ] + results = [ + item["nodes"][0]["children"][0]["nodes"][0]["result"]["output"]["value"] + for item in items + ] + assert results == [1, 2] + assert "call" not in state.step_results + + +def test_nested_resume_freezes_branch_binding_and_completed_prefix(tmp_path, probe): + child = definition( + "child", + [ + {"id": "prepare", "type": "probe"}, + {"id": "wait", "type": "probe", "await": True}, + ], + inputs={"approve": {"type": "boolean", "default": False}}, + ) + directory = install(tmp_path, child) + root = definition( + "parent", + [ + { + "id": "route", + "type": "if", + "condition": "{{ inputs.choose }}", + "then": [call(input={"approve": "{{ inputs.approve }}"})], + "else": [{"id": "wrong", "type": "probe"}], + } + ], + inputs={ + "choose": {"type": "boolean", "default": True}, + "approve": {"type": "boolean", "default": False}, + }, + ) + state = WorkflowEngine(tmp_path).execute(root) + assert state.status == RunStatus.PAUSED + (directory / "workflow.yml").write_text("invalid", encoding="utf-8") + state = WorkflowEngine(tmp_path).resume(state.run_id, {"choose": False}) + assert state.status == RunStatus.PAUSED + state = WorkflowEngine(tmp_path).resume(state.run_id, {"approve": True}) + assert state.status == RunStatus.COMPLETED + assert probe == {"prepare": 1, "wait": 3} + + +@pytest.mark.parametrize( + "status,aborted,handled,expected", + [ + ("failed", False, False, "failed"), + ("failed", False, True, "completed"), + ("failed", True, True, "aborted"), + ("paused", False, True, "paused"), + ], +) +def test_call_outcomes(tmp_path, probe, status, aborted, handled, expected): + install( + tmp_path, + definition( + "child", + [ + { + "id": "work", + "type": "probe", + "status": status, + "output": {"aborted": aborted}, + }, + ], + ), + ) + state = WorkflowEngine(tmp_path).execute( + definition( + "parent", + [ + call(continue_on_error=handled), + {"id": "after", "type": "probe"}, + ], + ) + ) + assert state.status.value == expected + assert probe["after"] == (expected == "completed") + assert state.step_results["call"]["output"]["status"] == ( + "aborted" if aborted else status + ) + + +@pytest.mark.parametrize( + "target", [" child", "child\n", "CHILD", "../child", "runs", 123, date(2026, 1, 1)] +) +def test_invalid_targets_are_handled_failures(tmp_path, probe, target): + state = WorkflowEngine(tmp_path).execute( + definition("parent", [call(target, continue_on_error=True)]) + ) + assert state.status == RunStatus.COMPLETED + assert state.step_results["call"]["status"] == "failed" + RunState.load(state.run_id, tmp_path) + + +@pytest.mark.parametrize( + "mapping", [{"unknown": "x"}, {"value": []}, {"value": date(2026, 1, 1)}, [1]] +) +def test_invalid_inputs_do_not_execute_child(tmp_path, probe, mapping): + install( + tmp_path, + definition( + "child", + [{"id": "work", "type": "probe"}], + inputs={"value": {"type": "string", "required": True}}, + ), + ) + state = WorkflowEngine(tmp_path).execute( + definition("parent", [call(input=mapping)]) + ) + assert state.status == RunStatus.FAILED + assert not probe + RunState.load(state.run_id, tmp_path) + + +def test_output_failure_retries_only_finalization(tmp_path, monkeypatch, probe): + import specify_cli.workflows._execution as execution + + install(tmp_path, definition("child", [{"id": "work", "type": "probe"}])) + original = execution.evaluate_outputs + monkeypatch.setattr( + execution, + "evaluate_outputs", + lambda *_: (_ for _ in ()).throw(ValueError("bad output")), + ) + state = WorkflowEngine(tmp_path).execute(definition("parent", [call()])) + assert state.status == RunStatus.FAILED + state = WorkflowEngine(tmp_path).resume(state.run_id) + assert state.status == RunStatus.FAILED + monkeypatch.setattr(execution, "evaluate_outputs", original) + state = WorkflowEngine(tmp_path).resume(state.run_id) + assert state.status == RunStatus.COMPLETED + assert probe["work"] == 1 + + +def test_legacy_resume_adapts_once(tmp_path, probe): + root = definition( + "parent", + [ + {"id": "before", "type": "probe"}, + {"id": "wait", "type": "probe", "await": True}, + ], + inputs={"approve": {"type": "boolean", "default": False}}, + ) + state = WorkflowEngine(tmp_path).execute(root) + path = state.runs_dir / "state.json" + data = json.loads(path.read_text()) + del data["execution"] + path.write_text(json.dumps(data)) + state = WorkflowEngine(tmp_path).resume(state.run_id) + assert state.status == RunStatus.PAUSED + state = WorkflowEngine(tmp_path).resume(state.run_id, {"approve": True}) + assert state.status == RunStatus.COMPLETED + assert probe == {"before": 1, "wait": 3} + + +@pytest.mark.parametrize( + "mutation", + [ + lambda tree: tree.update(version=99), + lambda tree: tree["sequence"].update(nodes=[]), + lambda tree: tree["sequence"]["nodes"][0].update(phase="nonsense"), + ], +) +def test_bad_checkpoint_rejected_without_writes(tmp_path, probe, mutation): + state = WorkflowEngine(tmp_path).execute( + definition("parent", [{"id": "wait", "type": "probe", "await": True}]) + ) + path = state.runs_dir / "state.json" + data = json.loads(path.read_text()) + mutation(data["execution"]) + path.write_text(json.dumps(data)) + before = path.read_bytes() + with pytest.raises(ValueError): + WorkflowEngine(tmp_path).resume(state.run_id) + assert path.read_bytes() == before + + +@pytest.mark.parametrize("failure_after_replace", [False, True]) +def test_checkpoint_failure_never_overwrites_committed_progress( + tmp_path, monkeypatch, probe, failure_after_replace +): + from specify_cli.workflows._execution import CheckpointError + + original = RunState._atomic_write_json + failed = False + + def write(path, data): + nonlocal failed + record = data.get("step_results", {}).get("work") + if path.name == "state.json" and record and not failed: + failed = True + if failure_after_replace: + original(path, data) + raise OSError("checkpoint failure") + original(path, data) + + monkeypatch.setattr(RunState, "_atomic_write_json", staticmethod(write)) + with pytest.raises(CheckpointError, match="checkpoint failure"): + WorkflowEngine(tmp_path).execute( + definition("parent", [{"id": "work", "type": "probe"}]), run_id="fault" + ) + disk = json.loads( + (tmp_path / ".specify/workflows/runs/fault/state.json").read_text() + ) + node = disk["execution"]["sequence"]["nodes"][0] + assert node["phase"] == ("done" if failure_after_replace else "ready") + assert ("work" in disk["step_results"]) is failure_after_replace + state = WorkflowEngine(tmp_path).resume("fault") + assert state.status == RunStatus.COMPLETED + assert probe["work"] == (1 if failure_after_replace else 2) + + +def test_completion_log_failure_does_not_replay_committed_step( + tmp_path, monkeypatch, probe +): + original = RunState.append_log + failed = False + + def log(self, entry): + nonlocal failed + if entry["event"] == "step_completed" and not failed: + failed = True + raise OSError("log failed") + original(self, entry) + + monkeypatch.setattr(RunState, "append_log", log) + with pytest.raises(OSError, match="log failed"): + WorkflowEngine(tmp_path).execute( + definition("parent", [{"id": "work", "type": "probe"}]), run_id="log" + ) + state = WorkflowEngine(tmp_path).resume("log") + assert state.status == RunStatus.COMPLETED + assert probe["work"] == 1 + + +@pytest.mark.parametrize("kind", ["if", "while", "do-while", "fan-out"]) +def test_expansion_resume_preserves_completed_work(tmp_path, probe, kind): + body = [ + {"id": "prepare", "type": "probe"}, + {"id": "wait", "type": "probe", "await": True}, + ] + config = {"id": "outer", "type": kind, "condition": True, "max_iterations": 2} + if kind == "if": + config["then"] = body + elif kind == "fan-out": + config.update( + items=[1, 2], + max_concurrency=2, + step={"id": "branch", "type": "if", "condition": True, "then": body}, + ) + else: + config["steps"] = body + state = WorkflowEngine(tmp_path).execute( + definition( + "parent", + [config], + inputs={"approve": {"type": "boolean", "default": False}}, + ) + ) + assert state.status == RunStatus.PAUSED + state = WorkflowEngine(tmp_path).resume(state.run_id, {"approve": True}) + assert state.status == RunStatus.COMPLETED + assert probe["prepare"] == (1 if kind == "if" else 2) + + +@pytest.mark.parametrize("mode", ["unknown", "disabled", "mismatch", "cycle"]) +def test_resolution_failures_are_call_failures(tmp_path, probe, mode): + if mode != "unknown": + child = definition( + "child", + [call("parent") if mode == "cycle" else {"id": "work", "type": "probe"}], + ) + directory = install(tmp_path, child, enabled=mode != "disabled") + if mode == "mismatch": + child.data["workflow"]["id"] = "other" + (directory / "workflow.yml").write_text(yaml.safe_dump(child.data)) + state = WorkflowEngine(tmp_path).execute( + definition("parent", [call(continue_on_error=True)]) + ) + assert state.status == RunStatus.COMPLETED + assert state.step_results["call"]["status"] == "failed" + assert not probe + + +def test_diamond_and_depth_limit(tmp_path, probe): + install(tmp_path, definition("leaf", [{"id": "work", "type": "probe"}])) + for name in ("left", "right"): + install(tmp_path, definition(name, [call("leaf")])) + state = WorkflowEngine(tmp_path).execute( + definition( + "parent", + [ + {**call("left"), "id": "left"}, + {**call("right"), "id": "right"}, + ], + ) + ) + assert state.status == RunStatus.COMPLETED + assert probe["work"] == 2 + for index in range(1, 18): + install( + tmp_path, + definition( + f"level-{index}", + [call(f"level-{index + 1}")] + if index < 17 + else [{"id": "too-deep", "type": "probe"}], + ), + ) + state = WorkflowEngine(tmp_path).execute(definition("parent", [call("level-1")])) + assert state.status == RunStatus.FAILED + assert "depth 16" in state.error + assert probe["too-deep"] == 0 + + +def test_output_validation_rejects_reserved_and_yaml_native_values(tmp_path, probe): + from specify_cli.workflows.composition import require_json, validate_outputs + + circular = [] + circular.append(circular) + for value in (circular, date(2026, 1, 1), {1: "key"}, float("nan")): + with pytest.raises(ValueError, match="JSON-safe"): + require_json(value) + for outputs in ( + {"status": {"value": "oops"}}, + {"value": {"wrong": 1}}, + {"value": {"value": date(2026, 1, 1)}}, + ): + assert validate_outputs(outputs) + + +def test_nested_gate_reporting_uses_active_occurrence(tmp_path, probe): + from specify_cli.workflows._commands import _workflow_run_payload + + install( + tmp_path, + definition( + "child", + [{"id": "review", "type": "gate", "message": "Approve", "mode": "manual"}], + ), + ) + state = WorkflowEngine(tmp_path).execute( + definition( + "parent", + [ + { + "id": "route", + "type": "if", + "condition": True, + "then": [call()], + } + ], + ) + ) + assert state.status == RunStatus.PAUSED + payload = _workflow_run_payload(RunState.load(state.run_id, tmp_path)) + assert payload["gate"]["step_id"] == "review" + assert payload["gate"]["scope_path"] == ["route", "call"] + assert payload["workflow_scopes"][0]["status"] == "paused" + + +def test_rebind_failure_has_one_failed_caller_outcome(tmp_path, probe): + install( + tmp_path, + definition( + "child", + [{"id": "wait", "type": "probe", "await": True}], + inputs={"approve": {"type": "boolean"}}, + ), + ) + root = definition( + "parent", + [call(input={"approve": "{{ inputs.approve }}"}, continue_on_error=True)], + inputs={"approve": {"type": "string", "default": "false"}}, + ) + state = WorkflowEngine(tmp_path).execute(root) + assert state.status == RunStatus.PAUSED + state = WorkflowEngine(tmp_path).resume(state.run_id, {"approve": "invalid"}) + assert state.status == RunStatus.COMPLETED + node = RunState.load(state.run_id, tmp_path).execution["sequence"]["nodes"][0] + assert node["phase"] == "done" + assert node["result"]["output"]["status"] == "failed" + from specify_cli.workflows._execution import active_step + + assert active_step(state.execution) is None + + +@pytest.mark.parametrize( + "status,handled", [("failed", False), ("paused", False), ("failed", True)] +) +def test_unsuccessful_expansion_never_executes_children( + tmp_path, monkeypatch, probe, status, handled +): + class Expand(StepBase): + type_key = "expand" + + def execute(self, config, context): + return StepResult( + StepStatus(status), next_steps=[{"id": "wrong", "type": "probe"}] + ) + + monkeypatch.setitem(STEP_REGISTRY, "expand", Expand()) + state = WorkflowEngine(tmp_path).execute( + definition( + "parent", + [ + { + "id": "expand", + "type": "expand", + "continue_on_error": handled, + } + ], + ) + ) + assert state.status.value == ("completed" if handled else status) + assert not probe + RunState.load(state.run_id, tmp_path) + + +def test_custom_expansion_is_frozen_across_resume(tmp_path, monkeypatch, probe): + class Expand(StepBase): + type_key = "expand" + + def execute(self, config, context): + probe["expand"] += 1 + return StepResult( + StepStatus.COMPLETED, + next_steps=[ + {"id": f"prefix-{probe['expand']}", "type": "probe"}, + {"id": "wait", "type": "probe", "await": True}, + ], + ) + + monkeypatch.setitem(STEP_REGISTRY, "expand", Expand()) + root = definition( + "parent", + [{"id": "expand", "type": "expand"}], + inputs={"approve": {"type": "boolean", "default": False}}, + ) + state = WorkflowEngine(tmp_path).execute(root) + state = WorkflowEngine(tmp_path).resume(state.run_id, {"approve": True}) + assert state.status == RunStatus.COMPLETED + assert probe == {"expand": 1, "prefix-1": 1, "wait": 2} + + +def test_native_yaml_definition_and_long_id_roundtrip(tmp_path, probe): + target = "a" * 240 + child = definition( + target, [{"id": "review", "type": "gate", "message": date(2026, 1, 1)}] + ) + install(tmp_path, child) + state = WorkflowEngine(tmp_path).execute(definition("parent", [call(target)])) + assert state.status == RunStatus.PAUSED + saved = RunState.load(state.run_id, tmp_path) + binding = saved.execution["sequence"]["nodes"][0]["binding"] + assert yaml.safe_load(binding["definition"])["steps"][0]["message"] == date( + 2026, 1, 1 + ) + assert WorkflowEngine(tmp_path).resume(state.run_id).status == RunStatus.PAUSED + + +def test_exact_depth_limit_is_allowed(tmp_path, probe): + for index in range(1, 17): + install( + tmp_path, + definition( + f"level-{index}", + [call(f"level-{index + 1}")] + if index < 16 + else [{"id": "work", "type": "probe"}], + ), + ) + state = WorkflowEngine(tmp_path).execute(definition("parent", [call("level-1")])) + assert state.status == RunStatus.COMPLETED + assert probe["work"] == 1 + + +def test_aborted_fanout_sibling_is_never_restarted(tmp_path, monkeypatch): + barrier = threading.Barrier(2, timeout=5) + counts = Counter() + + class Mixed(StepBase): + type_key = "mixed" + + def execute(self, config, context): + counts[context.item] += 1 + if not context.is_resume: + barrier.wait() + if context.item == 0: + return StepResult( + StepStatus.COMPLETED if context.is_resume else StepStatus.PAUSED + ) + return StepResult(StepStatus.FAILED, output={"aborted": True}) + + monkeypatch.setitem(STEP_REGISTRY, "mixed", Mixed()) + root = definition( + "parent", + [ + { + "id": "spread", + "type": "fan-out", + "items": [0, 1], + "max_concurrency": 2, + "step": {"id": "mixed", "type": "mixed"}, + } + ], + ) + state = WorkflowEngine(tmp_path).execute(root) + assert state.status == RunStatus.PAUSED + state = WorkflowEngine(tmp_path).resume(state.run_id) + assert state.status == RunStatus.ABORTED + assert counts == {0: 2, 1: 1} diff --git a/workflows/ARCHITECTURE.md b/workflows/ARCHITECTURE.md index 1b06dc2e66..a9ee91c00d 100644 --- a/workflows/ARCHITECTURE.md +++ b/workflows/ARCHITECTURE.md @@ -72,14 +72,16 @@ flowchart LR When a `gate` step pauses execution, the engine persists `current_step_index` and all accumulated `step_results`. On `specify workflow resume `, the engine restores the context and continues from the paused step. -> **Note:** Resume tracking is at the top-level step index only. If a -> nested step (inside `if`/`switch`/`while`) pauses, resume re-runs -> the parent control-flow step and its nested body. A nested step-path -> stack for exact resume is a planned enhancement. +New runs use a versioned execution tree. Each occurrence owns its result, +selected child sequences, and optional workflow binding. The same executor +traverses that tree on initial execution and resume, restoring local aliases +from completed results. Fan-out items have separate contexts. Legacy runs enter +through their top-level index once. Inputs and tree transitions share one atomic +state checkpoint; the inputs file is a compatibility mirror. ## Step Types -The engine ships with 12 built-in step types, each in its own subpackage under `src/specify_cli/workflows/step/`: +The engine ships with 13 built-in step types, each in its own subpackage under `src/specify_cli/workflows/step/`: | Type Key | Class | Purpose | Returns `next_steps`? | |----------|-------|---------|-----------------------| @@ -95,6 +97,7 @@ The engine ships with 12 built-in step types, each in its own subpackage under ` | `do-while` | `DoWhileStep` | Loop, always runs body at least once | Yes (always) | | `fan-out` | `FanOutStep` | Dispatch per item over a collection | No (engine expands) | | `fan-in` | `FanInStep` | Aggregate results from fan-out | No | +| `workflow` | `WorkflowStep` | Execute an installed workflow in a private scope | No (engine enters scope) | ## Step Registry diff --git a/workflows/PUBLISHING.md b/workflows/PUBLISHING.md index 4e26fdfb25..5b7850cbe0 100644 --- a/workflows/PUBLISHING.md +++ b/workflows/PUBLISHING.md @@ -90,7 +90,7 @@ steps: - ✅ `version` follows semantic versioning (X.Y.Z) - ✅ `description` is concise - ✅ All step IDs are unique -- ✅ Step types are valid: `command`, `prompt`, `shell`, `init`, `slot`, `gate`, `if`, `switch`, `while`, `do-while`, `fan-out`, `fan-in` +- ✅ Step types are valid: `command`, `prompt`, `shell`, `init`, `slot`, `gate`, `if`, `switch`, `while`, `do-while`, `fan-out`, `fan-in`, `workflow` - ✅ Required fields present per step type (e.g., `condition` for `if`, `expression` for `switch`) - ✅ Input types are valid: `string`, `number`, `boolean` - ✅ Step IDs do not contain `:` (reserved for engine-generated nested IDs like `parentId:childId`) diff --git a/workflows/README.md b/workflows/README.md index da045bfdf9..28ac151288 100644 --- a/workflows/README.md +++ b/workflows/README.md @@ -85,7 +85,10 @@ The bundled `speckit` workflow only declares `spec` (and optional ## Step Types -Workflows support 12 built-in step types: +Workflows support 13 built-in step types, including `workflow` for calling an +installed workflow with private inputs and declared outputs. See +[workflow composition and resume](../docs/reference/workflows.md#workflow-composition) +for the scope and execution identity contracts. ### Command Steps (default)