github-featured/github--spec-kit
2
0
Fork 0
Code Issues Pull Requests Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(workflows): validate dispatch defaults (#4181)

Browse Source 
* fix(workflows): validate dispatch defaults

* fix(workflows): validate dispatch defaults on resume

---------

Co-authored-by: root <kinsonnee@gmail.com>
This commit is contained in:
WOLIKIMCHENG
2026-08-20 05:23:50 +08:00
committed by GitHub
parent 14bbfd52e5
commit 7e48738e26
2 changed files with 252 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/specify_cli/workflows/engine.py
+56 -5
View File
@@ -61,11 +61,15 @@ class WorkflowDefinition:
self.schema_version: str = data.get("schema_version", "1.0")
# Defaults
self.default_integration: str | None = workflow.get("integration")
self.default_model: str | None = workflow.get("model")
self.default_options: dict[str, Any] = workflow.get("options") or {}
if not isinstance(self.default_options, dict):
self.default_options = {}
# Keep malformed values intact until ``validate_workflow`` can report
# them. ``None`` remains the supported "no defaults" form for options
# and retains its existing runtime representation as an empty mapping.
self.default_integration: Any = workflow.get("integration")
self.default_model: Any = workflow.get("model")
raw_default_options = workflow.get("options")
self.default_options: Any = (
{} if raw_default_options is None else raw_default_options
)
# Advisory pre-conditions (spec-kit version / integrations a workflow
# expects). Validated by ``validate_workflow`` (recognized keys only;
@@ -140,6 +144,40 @@ def _get_valid_step_types() -> set[str]:
}
def _dispatch_default_errors(definition: WorkflowDefinition) -> list[str]:
"""Return validation errors for workflow defaults inherited by dispatch steps."""
errors: list[str] = []
if (
definition.default_integration is not None
and not isinstance(definition.default_integration, str)
):
errors.append(
"'workflow.integration' must be a string or null, got "
f"{type(definition.default_integration).__name__} "
f"({definition.default_integration!r})."
)
if (
definition.default_model is not None
and not isinstance(definition.default_model, str)
):
errors.append(
"'workflow.model' must be a string or null, got "
f"{type(definition.default_model).__name__} "
f"({definition.default_model!r})."
)
if not isinstance(definition.default_options, dict):
errors.append(
"'workflow.options' must be a mapping or null, got "
f"{type(definition.default_options).__name__} "
f"({definition.default_options!r})."
)
return errors
def validate_workflow(definition: WorkflowDefinition) -> list[str]:
"""Validate a workflow definition and return a list of error messages.
@@ -197,6 +235,11 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
f"semantic versioning (expected X.Y.Z)."
)
# Workflow-level dispatch defaults are inherited by command and prompt
# steps. Validate their shapes before an invalid value reaches dispatch, or
# (for options) is silently normalized away during construction.
errors.extend(_dispatch_default_errors(definition))
# -- Inputs -----------------------------------------------------------
if not isinstance(definition.inputs, dict):
errors.append("'inputs' must be a mapping (or omitted).")
@@ -947,6 +990,10 @@ class WorkflowEngine:
-------
The final ``RunState`` after execution completes (or pauses).
"""
dispatch_default_errors = _dispatch_default_errors(definition)
if dispatch_default_errors:
raise ValueError(" ".join(dispatch_default_errors))
from . import STEP_REGISTRY
effective_run_id = run_id
@@ -1048,6 +1095,10 @@ class WorkflowEngine:
else:
definition = self.load_workflow(state.workflow_id)
dispatch_default_errors = _dispatch_default_errors(definition)
if dispatch_default_errors:
raise ValueError(" ".join(dispatch_default_errors))
# Merge any newly-supplied inputs over the persisted ones and
# re-validate through the same typing path as the initial run.
if inputs:
tests/test_workflows.py
+196
View File
@@ -4520,6 +4520,94 @@ steps:
errors = validate_workflow(definition)
assert errors == []
@pytest.mark.parametrize(
"field, bad_value",
[
("integration", ["claude"]),
("integration", {"name": "claude"}),
("integration", False),
("model", ["gpt-5"]),
("model", {"name": "gpt-5"}),
("model", 0),
("options", ["max_tokens"]),
("options", "max_tokens"),
("options", False),
],
)
def test_rejects_invalid_workflow_dispatch_defaults(self, field, bad_value):
"""Top-level dispatch defaults must retain their invalid shape for
validation instead of being passed to a step or normalized to ``{}``.
"""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow
definition = WorkflowDefinition(
{
"workflow": {
"id": "test",
"name": "Test",
"version": "1.0.0",
field: bad_value,
},
"steps": [{"id": "step-one", "command": "speckit.specify"}],
}
)
errors = validate_workflow(definition)
assert any(f"workflow.{field}" in error for error in errors), errors
assert any(type(bad_value).__name__ in error for error in errors), errors
if field == "options":
assert definition.default_options == bad_value
def test_preserves_valid_workflow_dispatch_defaults(self):
"""String and mapping defaults stay available unchanged to steps."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow
defaults = {
"integration": "claude",
"model": "gpt-5",
"options": {"max_tokens": 8000},
}
definition = WorkflowDefinition(
{
"workflow": {
"id": "test",
"name": "Test",
"version": "1.0.0",
**defaults,
},
"steps": [{"id": "step-one", "command": "speckit.specify"}],
}
)
assert definition.default_integration == defaults["integration"]
assert definition.default_model == defaults["model"]
assert definition.default_options == defaults["options"]
assert validate_workflow(definition) == []
def test_accepts_null_workflow_dispatch_defaults(self):
"""Null integration/model inherit at runtime and null options stays {}."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow
definition = WorkflowDefinition(
{
"workflow": {
"id": "test",
"name": "Test",
"version": "1.0.0",
"integration": None,
"model": None,
"options": None,
},
"steps": [{"id": "step-one", "command": "speckit.specify"}],
}
)
assert definition.default_integration is None
assert definition.default_model is None
assert definition.default_options == {}
assert validate_workflow(definition) == []
def test_no_steps(self):
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow
@@ -5165,6 +5253,36 @@ steps:
class TestWorkflowEngine:
"""Test WorkflowEngine execution."""
@pytest.mark.parametrize(
("field", "value"),
[
("integration", ["claude"]),
("model", {"name": "gpt-5"}),
("options", ["max_tokens"]),
],
)
def test_execute_rejects_invalid_workflow_dispatch_defaults(
self, project_dir, field, value
):
from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
definition = WorkflowDefinition(
{
"workflow": {
"id": "invalid-dispatch-defaults",
"name": "Invalid dispatch defaults",
"version": "1.0.0",
field: value,
},
"steps": [],
}
)
with pytest.raises(ValueError, match=f"workflow.{field}"):
WorkflowEngine(project_dir).execute(definition)
assert not (project_dir / ".specify" / "workflows" / "runs").exists()
def test_load_from_file(self, sample_workflow_file, project_dir):
from specify_cli.workflows.engine import WorkflowEngine
@@ -6684,6 +6802,45 @@ steps:
# and abort the run.
class TestWorkflowDispatchDefaultExecution:
"""Execution safeguards for defaults inherited by dispatch steps."""
@pytest.mark.parametrize(
"defaults",
[
{
"integration": "claude",
"model": "gpt-5",
"options": {"max_tokens": 8000},
},
{"integration": None, "model": None, "options": None},
],
)
def test_execute_accepts_valid_and_null_dispatch_defaults(
self, project_dir, defaults
):
"""Defaults with supported shapes remain executable without validation."""
from specify_cli.workflows.base import RunStatus
from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
definition = WorkflowDefinition(
{
"workflow": {
"id": "valid-defaults",
"name": "Valid Defaults",
"version": "1.0.0",
**defaults,
},
"steps": [],
}
)
state = WorkflowEngine(project_dir).execute(definition)
assert state.status == RunStatus.COMPLETED
assert state.step_results == {}
class TestContinueOnError:
"""Test the `continue_on_error` step-level field."""
@@ -10962,6 +11119,45 @@ steps:
with pytest.raises(ValueError):
engine.resume(state.run_id, {"count": "not-a-number"})
def test_resume_rejects_legacy_invalid_options_before_state_mutation(
self, project_dir, monkeypatch
):
from specify_cli.workflows.base import RunStatus
from specify_cli.workflows.engine import RunState, WorkflowDefinition
definition = WorkflowDefinition.from_string(self._WF_NUM)
engine = self._engine(project_dir)
state = engine.execute(definition)
assert state.status == RunStatus.PAUSED
workflow_copy = (
project_dir
/ ".specify"
/ "workflows"
/ "runs"
/ state.run_id
/ "workflow.yml"
)
workflow_copy.write_text(
self._WF_NUM.replace(
'version: "1.0.0"', 'version: "1.0.0"\n options: [max_tokens]'
),
encoding="utf-8",
)
def fail_step_context(*args, **kwargs):
raise AssertionError("StepContext must not be created")
monkeypatch.setattr("specify_cli.workflows.engine.StepContext", fail_step_context)
with pytest.raises(ValueError, match="'workflow.options' must be a mapping or null"):
engine.resume(state.run_id, {"count": "5"})
reloaded = RunState.load(state.run_id, project_dir)
assert reloaded.status == RunStatus.PAUSED
assert reloaded.error is None
assert reloaded.inputs["count"] == 1
def test_retry_verdict_input_is_consumed_and_can_be_replaced(self, project_dir):
import json as _json
from specify_cli.workflows.engine import WorkflowDefinition