feat(cli): add --process-type option to override process execution mode (#183)

# Summary

Adds `--process-type` CLI option to `plugboard process run` for
switching between `LocalProcess` and `RayProcess` execution modes,
overriding YAML configuration. Uses `Literal` type for type-safe
validation.

# Changes

- **CLI Enhancement**
- Added `--process-type {local|ray}` option to `run` command with
`Literal["local", "ray"]` type
- Typer automatically validates input and shows helpful error messages
for invalid values

- **Schema Enhancement**
- Added `RAY_CONNECTOR_CLS_PATH` constant to connector schema for
consistency
- Implemented `ProcessSpec.override_process_type()` method with
mapping-based logic
- Method handles process type, connector builder, and state backend
updates atomically
- When switching to `ray`: `LocalProcess` → `RayProcess`,
`AsyncioConnector` → `RayConnector`, `DictStateBackend` →
`RayStateBackend`
  - When switching to `local`: reverses the above transformations
  - Preserves connector builder args during override
- Only modifies incompatible components (e.g., won't change
`RayConnector` when already on Ray)

- **Tests**
  - Unit tests for local override and ray override
  - Validation testing handled by typer's built-in Literal type support

**Example usage:**
```bash
# Override any YAML config to run locally
plugboard process run ray-config.yaml --process-type local

# Force Ray execution regardless of YAML
plugboard process run local-config.yaml --process-type ray
```

**Implementation Note:** The override logic is implemented in the schema
layer (`ProcessSpec.override_process_type()`) using a mapping-based
approach rather than hard-coded if/else branches in the CLI, keeping
business logic close to the data model and improving maintainability.
Type safety is enforced using `Literal` types, allowing typer to handle
validation automatically.

<!-- START COPILOT ORIGINAL PROMPT -->



<details>

<summary>Original prompt</summary>

> 
> ----
> 
> *This section details on the original issue you should resolve*
> 
> <issue_title>CLI option to switch between local and Ray
process</issue_title>
> <issue_description>See
#49 (comment)
> 
> We would like to be able to switch between Local and Ray processes
from the CLI.</issue_description>
> 
> <agent_instructions>We need an option on the CLI to switch between
`RayProcess` and `LocalProcess` by overriding whatever is specified in
the input YAML. Note that when changing from local to ray, it is not
enough just to change the `Process` type: you will also need to make
sure the `connector_builder` and `State` type are changed if they are
specified in the YAML and not compatible with Ray.</agent_instructions>
> 
> ## Comments on the Issue (you are @copilot in this section)
> 
> <comments>
> </comments>
> 


</details>



<!-- START COPILOT CODING AGENT SUFFIX -->

- Fixes #60

<!-- START COPILOT CODING AGENT TIPS -->
---

✨ Let Copilot coding agent [set things up for
you](https://github.com/plugboard-dev/plugboard/issues/new?title=✨+Set+up+Copilot+instructions&body=Configure%20instructions%20for%20this%20repository%20as%20documented%20in%20%5BBest%20practices%20for%20Copilot%20coding%20agent%20in%20your%20repository%5D%28https://gh.io/copilot-coding-agent-tips%29%2E%0A%0A%3COnboard%20this%20repo%3E&assignees=copilot)
— coding agent works faster and does higher quality work when set up for
your repo.

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: toby-coleman <13170610+toby-coleman@users.noreply.github.com>
Co-authored-by: Toby Coleman <toby@tobycoleman.com>
Co-authored-by: Toby Coleman <toby-coleman@users.noreply.github.com>

Author: Copilot

plugboard-schemas/plugboard_schemas/__init__.py

@@ -20,6 +20,7 @@
 from .config import ConfigSpec, ProcessConfigSpec
 from .connector import (
     DEFAULT_CONNECTOR_CLS_PATH,
+    RAY_CONNECTOR_CLS_PATH,
     ConnectorBuilderArgsDict,
     ConnectorBuilderArgsSpec,
     ConnectorBuilderSpec,
@@ -83,6 +84,7 @@
     "ProcessSpec",
     "ProcessArgsDict",
     "ProcessArgsSpec",
+    "RAY_CONNECTOR_CLS_PATH",
     "RAY_STATE_BACKEND_CLS_PATH",
     "Resource",
     "StateBackendSpec",

plugboard-schemas/plugboard_schemas/connector.py

@@ -11,6 +11,7 @@
 
 
 DEFAULT_CONNECTOR_CLS_PATH: str = "plugboard.connector.AsyncioConnector"
+RAY_CONNECTOR_CLS_PATH: str = "plugboard.connector.RayConnector"
 
 
 class ConnectorMode(StrEnum):

plugboard-schemas/plugboard_schemas/process.py

@@ -8,7 +8,12 @@
 
 from ._common import PlugboardBaseModel
 from .component import ComponentSpec
-from .connector import DEFAULT_CONNECTOR_CLS_PATH, ConnectorBuilderSpec, ConnectorSpec
+from .connector import (
+    DEFAULT_CONNECTOR_CLS_PATH,
+    RAY_CONNECTOR_CLS_PATH,
+    ConnectorBuilderSpec,
+    ConnectorSpec,
+)
 from .state import DEFAULT_STATE_BACKEND_CLS_PATH, RAY_STATE_BACKEND_CLS_PATH, StateBackendSpec
 
 
@@ -96,3 +101,52 @@ def _validate_type(cls, value: _t.Any) -> str:
                 "plugboard.process.ray_process.RayProcess": "plugboard.process.RayProcess",
             }.get(value, value)
         return value
+
+    def override_process_type(self, process_type: _t.Literal["local", "ray"]) -> Self:
+        """Override the process type and update connector/state to be compatible.
+
+        Args:
+            process_type: The process type to use ("local" or "ray")
+
+        Returns:
+            A new ProcessSpec with the overridden process type and compatible settings
+        """
+        # Map process type to full class path
+        type_map = {
+            "local": "plugboard.process.LocalProcess",
+            "ray": "plugboard.process.RayProcess",
+        }
+        new_process_type = type_map[process_type]
+
+        # Map of connector types that should be replaced
+        connector_type_map = {
+            "ray": {DEFAULT_CONNECTOR_CLS_PATH: RAY_CONNECTOR_CLS_PATH},
+            "local": {RAY_CONNECTOR_CLS_PATH: DEFAULT_CONNECTOR_CLS_PATH},
+        }
+
+        # Map of state types that should be replaced
+        state_type_map = {
+            "ray": {DEFAULT_STATE_BACKEND_CLS_PATH: RAY_STATE_BACKEND_CLS_PATH},
+            "local": {RAY_STATE_BACKEND_CLS_PATH: DEFAULT_STATE_BACKEND_CLS_PATH},
+        }
+
+        # Prepare updates
+        updates: dict[str, _t.Any] = {"type": new_process_type}
+
+        # Update connector if needed
+        current_connector_type = self.connector_builder.type
+        if current_connector_type in connector_type_map[process_type]:
+            new_connector_type = connector_type_map[process_type][current_connector_type]
+            updates["connector_builder"] = self.connector_builder.model_copy(
+                update={"type": new_connector_type}
+            )
+
+        # Update state if needed
+        current_state_type = self.args.state.type
+        if current_state_type in state_type_map[process_type]:
+            new_state_type = state_type_map[process_type][current_state_type]
+            new_state = self.args.state.model_copy(update={"type": new_state_type})
+            updates["args"] = self.args.model_copy(update={"state": new_state})
+
+        # Apply updates and return new instance
+        return self.model_copy(update=updates)

plugboard/cli/process/__init__.py

@@ -92,6 +92,16 @@ def run(
             help="Job ID for the process. If not provided, a random job ID will be generated.",
         ),
     ] = None,
+    process_type: Annotated[
+        _t.Optional[_t.Literal["local", "ray"]],
+        typer.Option(
+            "--process-type",
+            help=(
+                "Override the process type. "
+                "Options: 'local' for LocalProcess, 'ray' for RayProcess."
+            ),
+        ),
+    ] = None,
 ) -> None:
     """Run a Plugboard process."""
     config_spec = _read_yaml(config)
@@ -100,6 +110,12 @@ def run(
         # Override job ID in config file if set
         config_spec.plugboard.process.args.state.args.job_id = job_id
 
+    if process_type is not None:
+        # Use the ProcessSpec method to override the process type
+        config_spec.plugboard.process = config_spec.plugboard.process.override_process_type(
+            process_type  # type: ignore[arg-type]
+        )
+
     with Progress(
         SpinnerColumn("arrow3"),
         TextColumn("[progress.description]{task.description}"),

tests/unit/test_cli.py

@@ -88,6 +88,49 @@ def test_cli_process_diagram() -> None:
     assert "flowchart" in result.stdout
 
 
+@pytest.mark.asyncio
+async def test_cli_process_run_with_local_override() -> None:
+    """Tests the process run command with --process-type local."""
+    with patch("plugboard.cli.process.ProcessBuilder") as mock_process_builder:
+        mock_process = AsyncMock()
+        mock_process_builder.build.return_value = mock_process
+        result = runner.invoke(
+            app,
+            ["process", "run", "tests/data/minimal-process.yaml", "--process-type", "local"],
+        )
+        # CLI must run without error
+        assert result.exit_code == 0
+        assert "Process complete" in result.stdout
+        # Process must be built with LocalProcess type
+        mock_process_builder.build.assert_called_once()
+        call_args = mock_process_builder.build.call_args
+        process_spec = call_args[0][0]
+        assert process_spec.type == "plugboard.process.LocalProcess"
+        assert process_spec.connector_builder.type == "plugboard.connector.AsyncioConnector"
+
+
+@pytest.mark.asyncio
+async def test_cli_process_run_with_ray_override() -> None:
+    """Tests the process run command with --process-type ray."""
+    with patch("plugboard.cli.process.ProcessBuilder") as mock_process_builder:
+        mock_process = AsyncMock()
+        mock_process_builder.build.return_value = mock_process
+        result = runner.invoke(
+            app,
+            ["process", "run", "tests/data/minimal-process.yaml", "--process-type", "ray"],
+        )
+        # CLI must run without error
+        assert result.exit_code == 0
+        assert "Process complete" in result.stdout
+        # Process must be built with RayProcess type
+        mock_process_builder.build.assert_called_once()
+        call_args = mock_process_builder.build.call_args
+        process_spec = call_args[0][0]
+        assert process_spec.type == "plugboard.process.RayProcess"
+        assert process_spec.connector_builder.type == "plugboard.connector.RayConnector"
+        assert process_spec.args.state.type == "plugboard.state.RayStateBackend"
+
+
 def test_cli_process_validate() -> None:
     """Tests the process validate command."""
     result = runner.invoke(app, ["process", "validate", "tests/data/minimal-process.yaml"])