fix(workflows): guard non-mapping 'inputs:' block in engine._resolve_inputs (#3696)

execute()/resume() run UNVALIDATED definitions (load_workflow does not
validate). WorkflowDefinition stores `inputs` raw, so a non-mapping
`inputs:` block (bare `inputs:` -> None, or `inputs: []`) crashed
_resolve_inputs at `for name, input_def in definition.inputs.items()` with
AttributeError, aborting the whole run.

Return {} when inputs is not a mapping, mirroring validate_workflow's own
`isinstance(definition.inputs, dict)` check. Protects both call sites
(execute and resume); normal dict resolution is unchanged.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ali jawwad
2026-07-24 20:52:28 +05:00
committed by GitHub
parent b0850c97e6
commit ea6843c1fe
2 changed files with 25 additions and 0 deletions

View File

@@ -1400,6 +1400,14 @@ class WorkflowEngine:
) -> dict[str, Any]:
"""Resolve workflow inputs against definitions and provided values."""
resolved: dict[str, Any] = {}
# execute()/resume() accept UNVALIDATED definitions (load_workflow does
# not validate). A non-mapping ``inputs:`` block (bare ``inputs:`` ->
# None, or ``inputs: []``) is stored raw, so iterating ``.items()`` here
# would crash the run with AttributeError. Treat a non-mapping inputs
# block as "no inputs"; validate_workflow reports the malformed shape
# via its own isinstance check.
if not isinstance(definition.inputs, dict):
return {}
for name, input_def in definition.inputs.items():
if not isinstance(input_def, dict):
continue

View File

@@ -3612,6 +3612,23 @@ class TestWorkflowDefinition:
assert definition.id == "test-workflow"
assert len(definition.inputs) == 2
@pytest.mark.parametrize(
"block",
[
"workflow:\n id: w\n name: W\nsteps: []\ninputs: []\n", # list
"workflow:\n id: w\n name: W\nsteps: []\ninputs:\n", # null
],
)
def test_resolve_inputs_tolerates_non_mapping_inputs(self, block):
# execute()/resume() run UNVALIDATED definitions; a non-mapping `inputs:`
# block (list/null) is stored raw and would crash _resolve_inputs at
# `.items()`. It must be treated as "no inputs" instead.
from specify_cli.workflows.engine import WorkflowDefinition, WorkflowEngine
definition = WorkflowDefinition.from_string(block)
resolved = WorkflowEngine()._resolve_inputs(definition, {}) # must not raise
assert resolved == {}
def test_from_string_invalid(self):
from specify_cli.workflows.engine import WorkflowDefinition