feat(workflows): make shell step timeout configurable

src/specify_cli/workflows/steps/shell/__init__.py

@@ -25,6 +25,14 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         run_cmd = str(run_cmd)
 
         cwd = context.project_root or "."
+        # Defensive: the engine does not auto-validate step config, so an
+        # invalid ``timeout`` (string, None, ...) would otherwise raise a
+        # TypeError from subprocess.run() and crash the whole run.  Mirror
+        # the engine's handling of unvalidated ``continue_on_error`` by
+        # only honoring well-formed values and falling back to the default.
+        timeout = config.get("timeout", 300)
+        if isinstance(timeout, bool) or not isinstance(timeout, int) or timeout <= 0:
+            timeout = 300
 
         # NOTE: shell=True is required to support pipes, redirects, and
         # multi-command expressions in workflow YAML.  Workflow authors
@@ -37,7 +45,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
                 capture_output=True,
                 text=True,
                 cwd=cwd,
-                timeout=300,
+                timeout=timeout,
             )
             output = {
                 "exit_code": proc.returncode,
@@ -74,7 +82,7 @@ def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
         except subprocess.TimeoutExpired:
             return StepResult(
                 status=StepStatus.FAILED,
-                error="Shell command timed out after 300 seconds.",
+                error=f"Shell command timed out after {timeout} seconds.",
                 output={"exit_code": -1, "stdout": "", "stderr": "timeout"},
             )
         except OSError as exc:
@@ -106,4 +114,16 @@ def validate(self, config: dict[str, Any]) -> list[str]:
                 f"Shell step {config.get('id', '?')!r}: 'output_format' must "
                 f"be 'json' when present, got {output_format!r}."
             )
+        if "timeout" in config:
+            timeout = config["timeout"]
+            # bool is an int subclass, so reject it explicitly.
+            if (
+                isinstance(timeout, bool)
+                or not isinstance(timeout, int)
+                or timeout <= 0
+            ):
+                errors.append(
+                    f"Shell step {config.get('id', '?')!r}: 'timeout' must be a "
+                    f"positive integer (seconds) when present, got {timeout!r}."
+                )
         return errors

tests/test_workflows.py

@@ -1286,6 +1286,106 @@ def test_validate_accepts_string_and_expression_run(self):
         assert step.validate({"id": "s", "run": "echo hi"}) == []
         assert step.validate({"id": "s", "run": "{{ steps.x.output }}"}) == []
 
+    def test_timeout_is_configurable(self, monkeypatch):
+        """A 'timeout' field overrides the 300s default (#3327)."""
+        import subprocess as sp
+
+        from specify_cli.workflows.steps.shell import ShellStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        seen = {}
+        real_run = sp.run
+
+        def spy_run(*args, **kwargs):
+            seen["timeout"] = kwargs.get("timeout")
+            return real_run(*args, **kwargs)
+
+        monkeypatch.setattr(
+            "specify_cli.workflows.steps.shell.subprocess.run", spy_run
+        )
+        step = ShellStep()
+        result = step.execute(
+            {"id": "t", "run": "echo hi", "timeout": 1800}, StepContext()
+        )
+        assert result.status == StepStatus.COMPLETED
+        assert seen["timeout"] == 1800
+
+    def test_timeout_defaults_to_300(self, monkeypatch):
+        import subprocess as sp
+
+        from specify_cli.workflows.steps.shell import ShellStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        seen = {}
+        real_run = sp.run
+
+        def spy_run(*args, **kwargs):
+            seen["timeout"] = kwargs.get("timeout")
+            return real_run(*args, **kwargs)
+
+        monkeypatch.setattr(
+            "specify_cli.workflows.steps.shell.subprocess.run", spy_run
+        )
+        result = ShellStep().execute({"id": "t", "run": "echo hi"}, StepContext())
+        assert result.status == StepStatus.COMPLETED
+        assert seen["timeout"] == 300
+
+    def test_timeout_error_reports_configured_value(self, monkeypatch):
+        import subprocess as sp
+
+        from specify_cli.workflows.steps.shell import ShellStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        def raise_timeout(*args, **kwargs):
+            raise sp.TimeoutExpired(cmd="x", timeout=kwargs.get("timeout"))
+
+        monkeypatch.setattr(
+            "specify_cli.workflows.steps.shell.subprocess.run", raise_timeout
+        )
+        result = ShellStep().execute(
+            {"id": "t", "run": "sleep 999", "timeout": 7}, StepContext()
+        )
+        assert result.status == StepStatus.FAILED
+        assert "7 seconds" in result.error
+
+    @pytest.mark.parametrize("bad", [0, -5, "600", 1.5, None, True])
+    def test_execute_ignores_unvalidated_bad_timeout(self, bad, monkeypatch):
+        """execute() falls back to 300 when config skipped validation (#3327)."""
+        import subprocess as sp
+
+        from specify_cli.workflows.steps.shell import ShellStep
+        from specify_cli.workflows.base import StepContext, StepStatus
+
+        seen = {}
+        real_run = sp.run
+
+        def spy_run(*args, **kwargs):
+            seen["timeout"] = kwargs.get("timeout")
+            return real_run(*args, **kwargs)
+
+        monkeypatch.setattr(
+            "specify_cli.workflows.steps.shell.subprocess.run", spy_run
+        )
+        result = ShellStep().execute(
+            {"id": "t", "run": "echo hi", "timeout": bad}, StepContext()
+        )
+        assert result.status == StepStatus.COMPLETED
+        assert seen["timeout"] == 300
+
+    @pytest.mark.parametrize("bad", [0, -5, "600", 1.5, None, True])
+    def test_validate_rejects_bad_timeout(self, bad):
+        from specify_cli.workflows.steps.shell import ShellStep
+
+        errors = ShellStep().validate({"id": "s", "run": "echo hi", "timeout": bad})
+        assert any("'timeout'" in e for e in errors)
+
+    def test_validate_accepts_positive_int_timeout(self):
+        from specify_cli.workflows.steps.shell import ShellStep
+
+        assert (
+            ShellStep().validate({"id": "s", "run": "echo hi", "timeout": 1800}) == []
+        )
+
 
     def test_output_format_json_exposes_data(self, tmp_path):
         from specify_cli.workflows.steps.shell import ShellStep