Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
63 changes: 59 additions & 4 deletions src/specify_cli/workflows/steps/command/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -66,16 +66,52 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
for key, value in input_data.items():
resolved_input[key] = evaluate_expression(value, context)

# Resolve integration (step → workflow default → project default)
integration = config.get("integration") or context.default_integration
# Resolve integration (step → workflow default → project default).
# Fall back to the workflow default ONLY for a genuinely-unset value
# (missing / YAML-null / empty string). A ``config.get(...) or ...``
# would also swallow a falsey *non-string* ([], {}, 0, False), coercing
# it to the default before the guard below runs — so on an unvalidated
# execute() such a step would silently dispatch with the configured
# default instead of failing. Fall through instead, so every non-string
# reaches the type guard.
integration = config.get("integration")
if integration is None or integration == "":
integration = context.default_integration
if integration and isinstance(integration, str) and "{{" in integration:
integration = evaluate_expression(integration, context)

# Resolve model
model = config.get("model") or context.default_model
# Resolve model (same fallback rationale as 'integration' above).
model = config.get("model")
if model is None or model == "":
model = context.default_model
if model and isinstance(model, str) and "{{" in model:
model = evaluate_expression(model, context)

# A non-string integration/model — a literal list/dict/number that
# skipped validation, an unvalidated workflow-level default, or an
Comment on lines +90 to +91
# expression that resolved to one — crashes downstream: get_integration()
# uses the value as a dict key (raw TypeError on an unhashable list/dict,
# even on a *validated* run) and build_exec_args() feeds model into the
# CLI argv. Fail the step with the contract error rather than taking down
# the whole run, mirroring the 'input'/'options' guards above. ``None``
# stays valid — it means "unset" and falls back to dispatch-not-possible.
if integration is not None and not isinstance(integration, str):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Command step {config.get('id', '?')!r}: 'integration' must "
f"be a string, got {type(integration).__name__}."
),
)
if model is not None and not isinstance(model, str):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Command step {config.get('id', '?')!r}: 'model' must be a "
f"string, got {type(model).__name__}."
),
)

# Merge options (workflow defaults ← step overrides)
options = dict(context.default_options)
step_options = config.get("options", {})
Expand Down Expand Up @@ -217,4 +253,23 @@ def validate(self, config: dict[str, Any]) -> list[str]:
errors.append(
f"Command step {config.get('id', '?')!r}: 'options' must be a mapping."
)
# execute() passes 'integration' to get_integration(), which uses it as a
# dict key — a non-string (list/dict) raises a raw TypeError (unhashable),
# even on a validated run — and feeds 'model' into the CLI argv. Reject a
# literal non-string here, mirroring the sibling type checks. ``None``
# (an explicit ``integration:``/``model:`` YAML null) means "inherit the
# workflow default" and stays valid; an expression like "{{ ... }}" is
# still a str, so it stays valid too.
integration = config.get("integration")
if integration is not None and not isinstance(integration, str):
errors.append(
f"Command step {config.get('id', '?')!r}: 'integration' must be a "
f"string, got {type(integration).__name__}."
)
model = config.get("model")
if model is not None and not isinstance(model, str):
errors.append(
f"Command step {config.get('id', '?')!r}: 'model' must be a "
f"string, got {type(model).__name__}."
)
return errors
63 changes: 59 additions & 4 deletions src/specify_cli/workflows/steps/prompt/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,16 +42,52 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
if not isinstance(prompt, str):
prompt = str(prompt)

# Resolve integration (step → workflow default)
integration = config.get("integration") or context.default_integration
# Resolve integration (step → workflow default).
# Fall back to the workflow default ONLY for a genuinely-unset value
# (missing / YAML-null / empty string). A ``config.get(...) or ...``
# would also swallow a falsey *non-string* ([], {}, 0, False), coercing
# it to the default before the guard below runs — so on an unvalidated
# execute() such a step would silently dispatch with the configured
# default instead of failing. Fall through instead, so every non-string
# reaches the type guard.
integration = config.get("integration")
if integration is None or integration == "":
integration = context.default_integration
if integration and isinstance(integration, str) and "{{" in integration:
integration = evaluate_expression(integration, context)

# Resolve model
model = config.get("model") or context.default_model
# Resolve model (same fallback rationale as 'integration' above).
model = config.get("model")
if model is None or model == "":
model = context.default_model
if model and isinstance(model, str) and "{{" in model:
model = evaluate_expression(model, context)

# A non-string integration/model — a literal list/dict/number that
# skipped validation, an unvalidated workflow-level default, or an
Comment on lines +66 to +67
# expression that resolved to one — crashes downstream: get_integration()
# uses the value as a dict key (raw TypeError on an unhashable list/dict,
# even on a *validated* run) and build_exec_args() feeds model into the
# CLI argv. Fail the step with the contract error rather than taking down
# the whole run. ``None`` stays valid — it means "unset" and falls back
# to dispatch-not-possible.
if integration is not None and not isinstance(integration, str):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Prompt step {config.get('id', '?')!r}: 'integration' must "
f"be a string, got {type(integration).__name__}."
),
)
if model is not None and not isinstance(model, str):
return StepResult(
status=StepStatus.FAILED,
error=(
f"Prompt step {config.get('id', '?')!r}: 'model' must be a "
f"string, got {type(model).__name__}."
),
)

# Attempt CLI dispatch
dispatch_result = self._try_dispatch(
prompt, integration, model, context
Expand Down Expand Up @@ -172,4 +208,23 @@ def validate(self, config: dict[str, Any]) -> list[str]:
f"Prompt step {config.get('id', '?')!r}: 'prompt' must be a "
f"string, got {type(config['prompt']).__name__}."
)
# execute() passes 'integration' to get_integration(), which uses it as a
# dict key — a non-string (list/dict) raises a raw TypeError (unhashable),
# even on a validated run — and feeds 'model' into the CLI argv. Reject a
# literal non-string here, mirroring the 'prompt' check above. ``None``
# (an explicit ``integration:``/``model:`` YAML null) means "inherit the
# workflow default" and stays valid; an expression like "{{ ... }}" is
# still a str, so it stays valid too.
integration = config.get("integration")
if integration is not None and not isinstance(integration, str):
errors.append(
f"Prompt step {config.get('id', '?')!r}: 'integration' must be a "
f"string, got {type(integration).__name__}."
)
model = config.get("model")
if model is not None and not isinstance(model, str):
errors.append(
f"Prompt step {config.get('id', '?')!r}: 'model' must be a "
f"string, got {type(model).__name__}."
)
return errors
160 changes: 160 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -1026,6 +1026,41 @@ def test_validate_rejects_non_mapping_input_and_options(self):
assert res_opt.status is StepStatus.FAILED
assert "'options' must be a mapping" in (res_opt.error or "")

@pytest.mark.parametrize("bad", [["claude"], {"a": 1}, 5, True])
def test_validate_rejects_non_string_integration_and_model(self, bad):
"""A non-string 'integration'/'model' must be rejected at validation.

execute() passes 'integration' to get_integration(), which uses it as a
dict key — an unhashable list/dict raises a raw TypeError there, even on
a validated run — and feeds 'model' into the CLI argv. Mirrors the
'command'/'input'/'options' type checks.
"""
from specify_cli.workflows.steps.command import CommandStep

step = CommandStep()
errs = step.validate({"id": "c", "command": "/x", "integration": bad})
assert any("'integration' must be a string" in e for e in errs), bad
errs = step.validate({"id": "c", "command": "/x", "model": bad})
assert any("'model' must be a string" in e for e in errs), bad

def test_validate_accepts_none_and_expression_integration_model(self):
"""An explicit YAML-null (inherit default) or a '{{ ... }}' expression
integration/model stays valid — only literal non-strings are rejected."""
from specify_cli.workflows.steps.command import CommandStep

step = CommandStep()
assert step.validate(
{"id": "c", "command": "/x", "integration": None, "model": None}
) == []
assert step.validate(
{
"id": "c",
"command": "/x",
"integration": "{{ inputs.agent }}",
"model": "{{ inputs.model }}",
}
) == []

def test_validate_rejects_non_string_command(self):
from specify_cli.workflows.steps.command import CommandStep

Expand Down Expand Up @@ -1061,6 +1096,55 @@ def test_execute_non_string_command_fails_cleanly(self):
assert result.status is StepStatus.FAILED, bad
assert "'command' must be a string" in (result.error or ""), bad

def test_execute_non_string_integration_fails_loudly(self):
"""On an unvalidated run, an unhashable 'integration' would crash
get_integration() (dict.get on a list) with a raw TypeError. execute()
must fail the step with the contract error instead."""
from specify_cli.workflows.steps.command import CommandStep
from specify_cli.workflows.base import StepContext, StepStatus

step = CommandStep()
res = step.execute(
{"id": "c", "command": "speckit.specify", "integration": ["claude"]},
StepContext(),
)
assert res.status is StepStatus.FAILED
assert "'integration' must be a string" in (res.error or "")
# non-string model likewise fails before build_exec_args
res = step.execute(
{"id": "c", "command": "speckit.specify", "integration": "claude", "model": ["m"]},
StepContext(),
)
assert res.status is StepStatus.FAILED
assert "'model' must be a string" in (res.error or "")

@pytest.mark.parametrize("falsey", [[], {}, 0, False])
def test_execute_falsey_non_string_integration_fails_loudly(self, falsey):
"""A *falsey* non-string ([], {}, 0, False) must fail the step, not be
swallowed by an ``or``-fallback to the workflow default.

A ``config.get('integration') or context.default_integration`` coerces a
falsey non-string to the default *before* the type guard runs, so with a
configured default the step would silently dispatch using the wrong
integration instead of surfacing the contract error. The default is set
here so a regression dispatches rather than fails-not-possible."""
from specify_cli.workflows.steps.command import CommandStep
from specify_cli.workflows.base import StepContext, StepStatus

step = CommandStep()
ctx = StepContext(default_integration="claude", default_model="sonnet")
res = step.execute(
{"id": "c", "command": "speckit.specify", "integration": falsey}, ctx
)
assert res.status is StepStatus.FAILED, falsey
assert "'integration' must be a string" in (res.error or ""), falsey
# a falsey non-string model likewise reaches the guard
res = step.execute(
{"id": "c", "command": "speckit.specify", "model": falsey}, ctx
)
assert res.status is StepStatus.FAILED, falsey
assert "'model' must be a string" in (res.error or ""), falsey

def test_step_override_integration(self):
from unittest.mock import patch
from specify_cli.workflows.steps.command import CommandStep
Expand Down Expand Up @@ -1448,6 +1532,82 @@ def test_validate_accepts_expression_prompt(self):
)
assert errors == []

@pytest.mark.parametrize("bad", [["claude"], {"a": 1}, 5, True])
def test_validate_rejects_non_string_integration_and_model(self, bad):
"""A non-string 'integration'/'model' must be rejected at validation.

execute() passes 'integration' to get_integration(), which uses it as a
dict key — an unhashable list/dict raises a raw TypeError there, even on
a validated run — and feeds 'model' into the CLI argv."""
from specify_cli.workflows.steps.prompt import PromptStep

step = PromptStep()
errs = step.validate({"id": "p", "prompt": "hi", "integration": bad})
assert any("'integration' must be a string" in e for e in errs), bad
errs = step.validate({"id": "p", "prompt": "hi", "model": bad})
assert any("'model' must be a string" in e for e in errs), bad

def test_validate_accepts_none_and_expression_integration_model(self):
"""An explicit YAML-null (inherit default) or a '{{ ... }}' expression
integration/model stays valid — only literal non-strings are rejected."""
from specify_cli.workflows.steps.prompt import PromptStep

step = PromptStep()
assert step.validate(
{"id": "p", "prompt": "hi", "integration": None, "model": None}
) == []
assert step.validate(
{
"id": "p",
"prompt": "hi",
"integration": "{{ inputs.agent }}",
"model": "{{ inputs.model }}",
}
) == []

def test_execute_non_string_integration_fails_loudly(self):
"""On an unvalidated run, an unhashable 'integration' would crash
get_integration() (dict.get on a dict) with a raw TypeError. execute()
must fail the step with the contract error instead."""
from specify_cli.workflows.steps.prompt import PromptStep
from specify_cli.workflows.base import StepContext, StepStatus

step = PromptStep()
res = step.execute(
{"id": "p", "prompt": "hi", "integration": {"a": 1}}, StepContext()
)
assert res.status is StepStatus.FAILED
assert "'integration' must be a string" in (res.error or "")
res = step.execute(
{"id": "p", "prompt": "hi", "integration": "claude", "model": ["m"]},
StepContext(),
)
assert res.status is StepStatus.FAILED
assert "'model' must be a string" in (res.error or "")

@pytest.mark.parametrize("falsey", [[], {}, 0, False])
def test_execute_falsey_non_string_integration_fails_loudly(self, falsey):
"""A *falsey* non-string ([], {}, 0, False) must fail the step, not be
swallowed by an ``or``-fallback to the workflow default.

A ``config.get('integration') or context.default_integration`` coerces a
falsey non-string to the default *before* the type guard runs, so with a
configured default the step would silently dispatch using the wrong
integration instead of surfacing the contract error. The default is set
here so a regression dispatches rather than fails-not-possible."""
from specify_cli.workflows.steps.prompt import PromptStep
from specify_cli.workflows.base import StepContext, StepStatus

step = PromptStep()
ctx = StepContext(default_integration="claude", default_model="sonnet")
res = step.execute({"id": "p", "prompt": "hi", "integration": falsey}, ctx)
assert res.status is StepStatus.FAILED, falsey
assert "'integration' must be a string" in (res.error or ""), falsey
# a falsey non-string model likewise reaches the guard
res = step.execute({"id": "p", "prompt": "hi", "model": falsey}, ctx)
assert res.status is StepStatus.FAILED, falsey
assert "'model' must be a string" in (res.error or ""), falsey


class TestShellStep:
"""Test the shell step type."""
Expand Down