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
5 changes: 5 additions & 0 deletions src/specify_cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -1936,6 +1936,11 @@ def _read_integration_json(project_root: Path) -> dict[str, Any]:
console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
console.print(f"[dim]Details:[/dim] {exc}")
raise typer.Exit(1)
except UnicodeDecodeError as exc:
console.print(f"[red]Error:[/red] {path} is not valid UTF-8.")
console.print(f"Please fix or delete {INTEGRATION_JSON} and retry.")
console.print(f"[dim]Details:[/dim] {exc}")
raise typer.Exit(1)
except OSError as exc:
console.print(f"[red]Error:[/red] Could not read {path}.")
console.print(f"Please fix file permissions or delete {INTEGRATION_JSON} and retry.")
Expand Down
85 changes: 84 additions & 1 deletion src/specify_cli/workflows/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

import yaml

from ..integration_state import INTEGRATION_JSON
from .base import RunStatus, StepContext, StepResult, StepStatus


Expand Down Expand Up @@ -143,6 +144,35 @@ def validate_workflow(definition: WorkflowDefinition) -> list[str]:
f"Must be 'string', 'number', or 'boolean'."
)

# Validate the default eagerly so authoring mistakes (e.g. a
# default not in the declared enum, or a non-numeric default for
# a number input) surface at install/validation time instead of
# at workflow-execution time. ``"auto"`` for the integration
# input is a runtime-resolved sentinel, so only the
# enum-membership check is exempted for that exact case — the
# declared type is still enforced (e.g. ``type: number`` paired
# with ``default: "auto"`` is still rejected).
if "default" in input_def:
default_value = input_def["default"]
is_auto_integration = (
input_name == "integration" and default_value == "auto"
)
validation_input_def: dict[str, Any] = input_def
if is_auto_integration and "enum" in input_def:
validation_input_def = {
key: value
for key, value in input_def.items()
if key != "enum"
}
try:
WorkflowEngine._coerce_input(
input_name, default_value, validation_input_def
)
except ValueError as exc:
errors.append(
f"Input {input_name!r} has invalid default: {exc}"
)

# -- Steps ------------------------------------------------------------
if not isinstance(definition.steps, list):
errors.append("'steps' must be a list.")
Expand Down Expand Up @@ -715,12 +745,65 @@ def _resolve_inputs(
name, provided[name], input_def
)
elif "default" in input_def:
resolved[name] = input_def["default"]
default_value = self._resolve_default(name, input_def["default"])
# If the integration default could not be resolved against
# project state and falls back to the literal ``"auto"``
# sentinel, exempt it from enum-membership coercion so a
# workflow that lists specific integrations in ``enum`` does
# not crash at runtime — declared type is still enforced.
coerce_input_def = input_def
if (
name == "integration"
and default_value == "auto"
and "enum" in input_def
):
coerce_input_def = {
key: value
for key, value in input_def.items()
if key != "enum"
}
resolved[name] = self._coerce_input(
name, default_value, coerce_input_def
)
elif input_def.get("required", False):
Comment on lines 747 to 768
Copy link

Copilot AI May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_resolve_inputs now coerces/validates default values via _coerce_input, which is a user-visible behavior change (invalid defaults can now raise at runtime). Since validate_workflow() doesn’t currently validate defaults against declared types/enums, consider adding that validation there so workflows fail fast at install/validation time rather than during execution.

Copilot uses AI. Check for mistakes.
msg = f"Required input {name!r} not provided."
raise ValueError(msg)
return resolved

def _resolve_default(self, name: str, default: Any) -> Any:
"""Resolve special default sentinels against project state.

For the ``integration`` input, ``"auto"`` resolves to the integration
recorded in ``.specify/integration.json`` so workflows dispatch to the
AI the project was actually initialized with, instead of a hardcoded
value baked into the workflow YAML.
"""
if name == "integration" and default == "auto":
resolved = self._load_project_integration()
if resolved is not None:
return resolved
return default

def _load_project_integration(self) -> str | None:
"""Read the active integration key from ``.specify/integration.json``.

Returns None when the file is missing or malformed; callers are
expected to fall back to a literal default.
"""
path = self.project_root / INTEGRATION_JSON
if not path.is_file():
return None
try:
data = json.loads(path.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError, UnicodeDecodeError):
return None
if not isinstance(data, dict):
return None
value = data.get("integration")
if isinstance(value, str) and value:
return value
return None
Comment on lines +787 to +805
Copy link

Copilot AI May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

_load_project_integration() duplicates the integration.json parsing logic that already exists in specify_cli.__init__._read_integration_json, but with different error handling semantics (CLI exits loudly vs. workflow engine silently falls back). To avoid drift and keep behavior consistent, consider extracting a shared low-level helper (e.g., “try-read integration.json → dict|None”) into a common module and layering the CLI/engine-specific error handling on top.

Copilot uses AI. Check for mistakes.

@staticmethod
def _coerce_input(
name: str, value: Any, input_def: dict[str, Any]
Expand Down
235 changes: 235 additions & 0 deletions tests/test_workflows.py
Original file line number Diff line number Diff line change
Expand Up @@ -1495,6 +1495,241 @@ def test_execute_missing_required_input(self, project_dir):
with pytest.raises(ValueError, match="Required input"):
engine.execute(definition, {})

def test_integration_auto_default_uses_project_integration(self, project_dir):
"""`integration: auto` should resolve to .specify/integration.json's integration."""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

Comment on lines +1498 to +1501
Copy link

Copilot AI May 1, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PR description mentions “4 new regression tests”, but this diff adds more than four new test cases in this block. Consider updating the PR description/test plan to match the actual test coverage so reviewers know the full scope.

Copilot uses AI. Check for mistakes.
specify_dir = project_dir / ".specify"
specify_dir.mkdir(parents=True, exist_ok=True)
(specify_dir / "integration.json").write_text(
json.dumps({"integration": "opencode", "version": "0.7.4"}),
encoding="utf-8",
)

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "auto-default"
name: "Auto Default"
version: "1.0.0"
inputs:
integration:
type: string
default: "auto"
""")
engine = WorkflowEngine(project_dir)
resolved = engine._resolve_inputs(definition, {})
assert resolved["integration"] == "opencode"

def test_integration_auto_default_falls_back_when_no_integration_json(self, project_dir):
"""`integration: auto` should keep the literal "auto" when project state is missing.

The engine itself must not invent an integration when
``.specify/integration.json`` is absent; any later validation or
command resolution will handle an unresolved ``"auto"`` value.
"""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "auto-fallback"
name: "Auto Fallback"
version: "1.0.0"
inputs:
integration:
type: string
default: "auto"
""")
engine = WorkflowEngine(project_dir)
resolved = engine._resolve_inputs(definition, {})
assert resolved["integration"] == "auto"

def test_integration_explicit_input_overrides_auto(self, project_dir):
"""An explicit --input integration=X must win over `auto` even when integration.json exists."""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

specify_dir = project_dir / ".specify"
specify_dir.mkdir(parents=True, exist_ok=True)
(specify_dir / "integration.json").write_text(
json.dumps({"integration": "opencode"}),
encoding="utf-8",
)

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "explicit-wins"
name: "Explicit Wins"
version: "1.0.0"
inputs:
integration:
type: string
default: "auto"
""")
engine = WorkflowEngine(project_dir)
resolved = engine._resolve_inputs(definition, {"integration": "claude"})
assert resolved["integration"] == "claude"

def test_integration_auto_ignores_malformed_integration_json(self, project_dir):
"""A malformed integration.json must not crash — fall back to the literal default."""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

specify_dir = project_dir / ".specify"
specify_dir.mkdir(parents=True, exist_ok=True)
(specify_dir / "integration.json").write_text("{not json", encoding="utf-8")

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "auto-malformed"
name: "Auto Malformed"
version: "1.0.0"
inputs:
integration:
type: string
default: "auto"
""")
engine = WorkflowEngine(project_dir)
resolved = engine._resolve_inputs(definition, {})
assert resolved["integration"] == "auto"

def test_integration_auto_ignores_non_utf8_integration_json(self, project_dir):
"""A non-UTF8 integration.json must not crash — fall back to the literal default."""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

specify_dir = project_dir / ".specify"
specify_dir.mkdir(parents=True, exist_ok=True)
# 0xFF is invalid as the leading byte of a UTF-8 sequence, so
# ``Path.read_text(encoding="utf-8")`` raises UnicodeDecodeError.
(specify_dir / "integration.json").write_bytes(b"\xff\xfe\x00\x00")

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "auto-non-utf8"
name: "Auto Non UTF-8"
version: "1.0.0"
inputs:
integration:
type: string
default: "auto"
""")
engine = WorkflowEngine(project_dir)
resolved = engine._resolve_inputs(definition, {})
assert resolved["integration"] == "auto"

def test_default_value_is_validated_against_enum(self, project_dir):
"""Defaults must run through the same coercion/enum check as provided inputs."""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "default-enum"
name: "Default Enum"
version: "1.0.0"
inputs:
scope:
type: string
default: "not-in-enum"
enum: ["full", "backend-only", "frontend-only"]
""")
engine = WorkflowEngine(project_dir)
with pytest.raises(ValueError, match="not in allowed values"):
engine._resolve_inputs(definition, {})

def test_default_value_is_coerced_to_declared_type(self, project_dir):
"""A numeric default declared as a string should still be coerced like a provided input."""
from specify_cli.workflows.engine import WorkflowEngine, WorkflowDefinition

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "default-coerce"
name: "Default Coerce"
version: "1.0.0"
inputs:
retries:
type: number
default: "3"
""")
engine = WorkflowEngine(project_dir)
resolved = engine._resolve_inputs(definition, {})
assert resolved["retries"] == 3
assert isinstance(resolved["retries"], int)

def test_validate_workflow_rejects_invalid_default(self):
"""Authoring-time validation should reject defaults that violate enum."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "bad-default"
name: "Bad Default"
version: "1.0.0"
inputs:
scope:
type: string
default: "not-in-enum"
enum: ["full", "backend-only", "frontend-only"]
steps:
- id: noop
type: gate
message: "noop"
options: [approve]
""")
errors = validate_workflow(definition)
assert any("invalid default" in e for e in errors), errors

def test_validate_workflow_exempts_integration_auto_sentinel(self):
"""``integration: auto`` is a runtime-resolved sentinel and must not fail validation."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "auto-ok"
name: "Auto OK"
version: "1.0.0"
inputs:
integration:
type: string
default: "auto"
enum: ["copilot", "claude", "gemini"]
steps:
- id: noop
type: gate
message: "noop"
options: [approve]
""")
errors = validate_workflow(definition)
assert not any("invalid default" in e for e in errors), errors

def test_validate_workflow_still_checks_type_for_auto_sentinel(self):
"""The ``auto`` exemption only skips enum-membership; declared type is still enforced."""
from specify_cli.workflows.engine import WorkflowDefinition, validate_workflow

definition = WorkflowDefinition.from_string("""
schema_version: "1.0"
workflow:
id: "auto-bad-type"
name: "Auto Bad Type"
version: "1.0.0"
inputs:
integration:
type: number
default: "auto"
steps:
- id: noop
type: gate
message: "noop"
options: [approve]
""")
errors = validate_workflow(definition)
assert any("invalid default" in e for e in errors), errors


# ===== State Persistence Tests =====

Expand Down
Loading