Refactor introduce runners v

SPEC.md

@@ -182,7 +182,7 @@ def get_runner(name: str) -> Runner:
 
 - `Settings.runner = StringField(required=True, max_length=64)`, no default. `job_engine` is
   deleted. `create_settings(runner=...)`.
-- `QuantingEnv.job_engine` -> `runner: str = Field(alias="_RUNNER")`. Underscore prefix, so it
+- `QuantingEnv.job_engine` -> `runner_name: str = Field(alias="_RUNNER_NAME")`. Underscore prefix, so it
   is never exported to a job.
 
 ### 2.4 prepare_job
@@ -195,7 +195,7 @@ resolved absolute strings change flavour.
 ### 2.5 Handler factory and SSH
 
 - `_get_job_handler(runner: Runner)`. `start_job/get_job_status/get_job_result(..., runner_name)`
-  look the runner up. `ssh_sensor.py:58` reads `.runner`.
+  look the runner up. `ssh_sensor.py:58` reads `.runner_name`.
 - Slurm: `SlurmSSHJobHandler(runner.view.resolve(Locations.SLURM), runner.ssh_connection_id_prefix)`.
   The factory raises `AirflowFailException` naming the runner if the prefix is `None`.
 - Docker: unchanged, `DockerJobHandler(DOCKER_HOST_VIEW)`. The host view is a property of the
@@ -392,7 +392,7 @@ pytest, tests next to the existing ones (`shared/tests`, `airflow_src/tests`, `w
   substituted `_CONFIG_PARAMS`, and `_check_content` returns no errors. `_check_content` still
   rejects `..`, `;`, `$` in relative paths, file names, `software`, `config_params`.
 - 7.3 Regression: for a `slurm` runner whose `view` equals the former `absolute_path`
-  values, `QuantingEnv.to_dict()` is byte-identical to before, except `_JOB_ENGINE` -> `_RUNNER`.
+  values, `QuantingEnv.to_dict()` is byte-identical to before, except `_JOB_ENGINE` -> `_RUNNER_NAME`.
   `get_backup_base_path` yields the same string as before for the same yaml values.
 - 7.4 `test_utils.py`: two prefixes select disjoint connection sets.
 - 7.5 `test_job_handler.py`: factory per engine with a `Runner`; unknown engine raises; a

airflow_src/dags/impl/processor_impl.py

@@ -225,7 +225,8 @@ def _create_quanting_env(
         settings_version=settings.version,
         relative_raw_file_path=str(relative_raw_file_path),
         config_params=substituted_params,
-        job_engine=settings.job_engine,
+        # transitional until Settings.runner exists: every in-repo runner is named after its engine
+        runner_name=settings.job_engine,
         year_month_folder=get_created_at_year_month(raw_file),
     )
 
@@ -284,6 +285,7 @@ def _prepare_custom_command(settings: Settings, substituted_params: str) -> str:
     "settings_name",
     "year_month_folder",
     "slurm_mem",
+    "runner_name",
 )
 
 # composed of a yaml base path (admin configuration) and fields checked above, e.g.
@@ -295,7 +297,6 @@ def _prepare_custom_command(settings: Settings, substituted_params: str) -> str:
     "custom_command",
     "config_params",
     "slurm_time",  # contains ":", validated in the webapp
-    "job_engine",  # selectbox value, dispatch only, not exported to the job
 )
 
 
@@ -391,10 +392,7 @@ def submit_job(
 
     output_path.mkdir(parents=True, exist_ok=True)
 
-    job_id = start_job(
-        quanting_env,
-        engine=quanting_env.job_engine,
-    )
+    job_id = start_job(quanting_env, runner_name=quanting_env.runner_name)
 
     # TODO: race condition here, e.g. for file in ERROR status
     update_raw_file(quanting_env.raw_file_id, new_status=RawFileStatus.QUANTING)
@@ -468,7 +466,9 @@ def check_job_result(*, quanting_env_dict: dict, job_id: str, ti: TaskInstance)
     """
     quanting_env = QuantingEnv.from_dict(quanting_env_dict)
 
-    job_status, time_elapsed = get_job_result(job_id, engine=quanting_env.job_engine)
+    job_status, time_elapsed = get_job_result(
+        job_id, runner_name=quanting_env.runner_name
+    )
 
     logging.info(f"Job {job_id} exited with status {job_status}.")
 

airflow_src/plugins/common/quanting_env.py

@@ -42,7 +42,7 @@ class QuantingEnv(BaseModel):
 
     relative_raw_file_path: str = Field(alias="_RELATIVE_RAW_FILE_PATH")
     config_params: str = Field(alias="_CONFIG_PARAMS")
-    job_engine: str = Field(alias="_JOB_ENGINE")
+    runner_name: str = Field(alias="_RUNNER_NAME")
     year_month_folder: str = Field(alias="_YEAR_MONTH_FOLDER")
 
     def to_dict(self) -> dict:

airflow_src/plugins/jobs/job_handler.py

@@ -12,10 +12,12 @@
 
 from shared.keys import EnvVars, JobEngines
 from shared.path_views import CLUSTER_VIEW, DOCKER_HOST_VIEW, Locations
+from shared.runners import Runner, get_runner
 
 
-def _get_job_handler(engine: str) -> "JobHandler":
-    """Factory function to get the appropriate job handler for the given engine."""
+def _get_job_handler(runner: Runner) -> "JobHandler":
+    """Factory function to get the job handler for the engine of the given runner."""
+    engine = runner.engine
     if engine == JobEngines.SLURM:
         from jobs.slurm_ssh_job_handler import SlurmSSHJobHandler
 
@@ -81,28 +83,25 @@ def get_job_result(self, job_id: str) -> tuple[str, int]:
         """
 
 
-def start_job(
-    quanting_env: QuantingEnv,
-    engine: str,
-) -> str:
-    """Start a job using the given job engine.
+def start_job(quanting_env: QuantingEnv, runner_name: str) -> str:
+    """Start a job on the given runner.
 
     Delegates to JobHandler.start_job(), see docs there.
     """
-    handler = _get_job_handler(engine)
+    handler = _get_job_handler(get_runner(runner_name))
     return handler.start_job(quanting_env)
 
 
-def get_job_status(job_id: str, engine: str) -> str:
-    """Get the job status using the given job engine.
+def get_job_status(job_id: str, runner_name: str) -> str:
+    """Get the job status from the given runner.
 
     Delegates to JobHandler.get_job_status(), see docs there.
     """
-    handler = _get_job_handler(engine)
+    handler = _get_job_handler(get_runner(runner_name))
     return handler.get_job_status(job_id)
 
 
-def get_job_result(job_id: str, engine: str) -> tuple[str, int]:
-    """Get the job status and time elapsed using the given job engine."""
-    handler = _get_job_handler(engine)
+def get_job_result(job_id: str, runner_name: str) -> tuple[str, int]:
+    """Get the job status and time elapsed from the given runner."""
+    handler = _get_job_handler(get_runner(runner_name))
     return handler.get_job_result(job_id)

airflow_src/plugins/sensors/ssh_sensor.py

@@ -38,7 +38,7 @@ def __init__(
         self.xcom_source_task_id: str = xcom_source_task_id
         self.quanting_env_source_task_id: str = quanting_env_source_task_id
         self._job_id: str | None = None
-        self._engine: str | None = None
+        self._runner_name: str | None = None
 
     def pre_execute(self, context: dict[str, Any]) -> None:
         """Persist the job id and job engine from XCom."""
@@ -55,13 +55,13 @@ def pre_execute(self, context: dict[str, Any]) -> None:
             task_ids=self.quanting_env_source_task_id,
             map_indexes=ti.map_index,
         )
-        self._engine = QuantingEnv.from_dict(quanting_env_dict).job_engine
+        self._runner_name = QuantingEnv.from_dict(quanting_env_dict).runner_name
 
     def poke(self, context: dict[str, Any]) -> bool:
         """Check the output of the ssh command."""
         del context  # unused
 
-        job_status = get_job_status(self._job_id, self._engine)
+        job_status = get_job_status(self._job_id, self._runner_name)
         logging.info(f"job_status: '{job_status}'")
 
         return job_status not in self.states

airflow_src/tests/conftest.py

@@ -32,7 +32,7 @@
     "settings_version": 1,
     "relative_raw_file_path": "instrument1/1970_01/test_file.raw",
     "config_params": "",
-    "job_engine": "slurm",
+    "runner_name": "slurm",
     "year_month_folder": "1970_01",
 }
 

airflow_src/tests/dags/impl/test_processor_impl.py

@@ -105,7 +105,7 @@ def test_create_quanting_env(
         "PROJECT_ID": "some_project_id",
         "SETTINGS_NAME": "test_settings",
         "SETTINGS_VERSION": 1,
-        "_JOB_ENGINE": "slurm",
+        "_RUNNER_NAME": "slurm",
         "_YEAR_MONTH_FOLDER": "1970_01",
         "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw",
         "_CONFIG_PARAMS": "",
@@ -188,7 +188,7 @@ def test_create_quanting_env_custom_software(
         "PROJECT_ID": "some_project_id",
         "SETTINGS_NAME": "test_custom_settings",
         "SETTINGS_VERSION": 1,
-        "_JOB_ENGINE": "slurm",
+        "_RUNNER_NAME": "slurm",
         "_YEAR_MONTH_FOLDER": "1970_01",
         "_RELATIVE_RAW_FILE_PATH": "instrument1/1970_01/test_file.raw",
         "_CONFIG_PARAMS": expected_config_params,
@@ -387,7 +387,7 @@ def test_check_content_allows_image_name_in_software_field(
 ) -> None:
     """Test that a docker image name in the software field is accepted."""
     quanting_env = make_quanting_env(
-        software="alphakraken-msqc", job_engine=JobEngines.DOCKER
+        software="alphakraken-msqc", runner_name=JobEngines.DOCKER
     )
 
     errors = _check_content(quanting_env, MagicMock(config_params=None))
@@ -562,7 +562,7 @@ def test_submit_job_executes_ssh_command_and_stores_job_id(
     assert output_dir.exists()
     mock_start_job.assert_called_once_with(
         quanting_env,
-        engine="slurm",
+        runner_name="slurm",
     )
     mock_get_raw_file_by_id.assert_called_once_with("test_file.raw")
     mock_update.assert_called_once_with(

airflow_src/tests/plugins/jobs/test_job_handler.py

@@ -2,18 +2,21 @@
 
 import importlib.util
 import sys
+from collections.abc import Callable
 from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
 from airflow.exceptions import AirflowFailException
+from common.quanting_env import QuantingEnv
 from jobs._experimental.file_based_job_handler import FileBasedJobHandler
-from jobs.job_handler import _get_job_handler
+from jobs.job_handler import _get_job_handler, start_job
 from jobs.slurm_ssh_job_handler import SlurmSSHJobHandler
 
 from airflow_src.tests.helpers import yaml_locations
 from shared.keys import JobEngines
 from shared.path_views import DOCKER_HOST_VIEW
+from shared.runners import Runner, get_runner
 
 # `docker` is an optional dependency, cf. requirements_docker_job_engine.txt
 HAS_DOCKER = importlib.util.find_spec("docker") is not None
@@ -24,14 +27,18 @@
 @yaml_locations(slurm=str(SLURM_BASE_DIR))
 def test_get_job_handler_routes_engine_to_handler() -> None:
     """Test that the factory returns the handler matching the requested engine."""
-    assert isinstance(_get_job_handler(JobEngines.SLURM), SlurmSSHJobHandler)
-    assert isinstance(_get_job_handler(JobEngines.FILE_BASED), FileBasedJobHandler)
+    assert isinstance(
+        _get_job_handler(get_runner(JobEngines.SLURM)), SlurmSSHJobHandler
+    )
+    assert isinstance(
+        _get_job_handler(get_runner(JobEngines.FILE_BASED)), FileBasedJobHandler
+    )
 
 
 @yaml_locations(slurm=str(SLURM_BASE_DIR))
 def test_get_job_handler_injects_slurm_base_dir() -> None:
     """Test that the factory reads the base dir and hands it to the Slurm handler."""
-    handler = _get_job_handler(JobEngines.SLURM)
+    handler = _get_job_handler(get_runner(JobEngines.SLURM))
 
     assert handler._cluster_base_dir == SLURM_BASE_DIR
 
@@ -42,7 +49,7 @@ def test_get_job_handler_injects_docker_host_view(
     mock_from_env: MagicMock,  # noqa: ARG001
 ) -> None:
     """Test that the factory hands the docker host view to the docker handler."""
-    handler = _get_job_handler(JobEngines.DOCKER)
+    handler = _get_job_handler(get_runner(JobEngines.DOCKER))
 
     assert handler._docker_host_view is DOCKER_HOST_VIEW
 
@@ -57,7 +64,7 @@ def test_get_job_handler_docker_without_mounts_env(
         patch.object(DOCKER_HOST_VIEW, "_locations", {}),
         pytest.raises(AirflowFailException, match="MOUNTS_PATH"),
     ):
-        _get_job_handler(JobEngines.DOCKER)
+        _get_job_handler(get_runner(JobEngines.DOCKER))
 
 
 def test_get_job_handler_docker_without_optional_dependency() -> None:
@@ -67,4 +74,26 @@ def test_get_job_handler_docker_without_optional_dependency() -> None:
         patch.dict(sys.modules, {"docker": None, "jobs.docker_job_handler": None}),
         pytest.raises(AirflowFailException, match="requirements_docker_job_engine"),
     ):
-        _get_job_handler(JobEngines.DOCKER)
+        _get_job_handler(get_runner(JobEngines.DOCKER))
+
+
+def test_get_job_handler_rejects_unknown_engine() -> None:
+    """Test that a runner with an engine the factory has no branch for is rejected."""
+    runner = Runner(
+        name="k8s",
+        engine="kubernetes",
+        os="linux",
+        view=MagicMock(),
+        ssh_connection_id_prefix=None,
+    )
+
+    with pytest.raises(ValueError, match="kubernetes"):
+        _get_job_handler(runner)
+
+
+def test_start_job_rejects_undeclared_runner(
+    make_quanting_env: Callable[..., QuantingEnv],
+) -> None:
+    """Test that a runner name not in the yaml fails listing the declared runners."""
+    with pytest.raises(KeyError, match=r"'nope'.*\['slurm', 'docker', 'file_based'\]"):
+        start_job(make_quanting_env(), runner_name="nope")

airflow_src/tests/plugins/sensors/test_ssh_sensor.py

@@ -11,7 +11,7 @@
 
 JOB_ID_SOURCE_TASK_ID = "processing.submit_job"
 QUANTING_ENV_SOURCE_TASK_ID = "processing.prepare_job"
-ENGINE = "file_based"
+RUNNER_NAME = "file_based"
 
 
 @patch("plugins.sensors.ssh_sensor.get_job_status")
@@ -25,7 +25,7 @@ def test_poke_executes_ssh_command_and_checks_returned_state(
     mock_ti.map_index = 0
     mock_ti.xcom_pull.side_effect = [
         "12345",
-        make_quanting_env(job_engine=ENGINE).to_dict(),
+        make_quanting_env(runner_name=RUNNER_NAME).to_dict(),
     ]
     mock_get_job_status.return_value = JobStates.RUNNING
     context = {"ti": mock_ti}
@@ -49,7 +49,7 @@ def test_poke_executes_ssh_command_and_checks_returned_state(
         ]
     )
     assert not operator.poke(context)
-    mock_get_job_status.assert_called_once_with("12345", ENGINE)
+    mock_get_job_status.assert_called_once_with("12345", RUNNER_NAME)
 
 
 @pytest.mark.parametrize(
@@ -79,7 +79,7 @@ def test_poke_returns_true_when_state_not_in_running_states(
     mock_ti.map_index = 2
     mock_ti.xcom_pull.side_effect = [
         "12345",
-        make_quanting_env(job_engine=ENGINE).to_dict(),
+        make_quanting_env(runner_name=RUNNER_NAME).to_dict(),
     ]
     mock_get_job_status.return_value = job_status
     context = {"ti": mock_ti}
@@ -103,4 +103,4 @@ def test_poke_returns_true_when_state_not_in_running_states(
         ]
     )
     assert operator.poke(context) is expected_poke
-    mock_get_job_status.assert_called_once_with("12345", ENGINE)
+    mock_get_job_status.assert_called_once_with("12345", RUNNER_NAME)

tasks/todo.md

@@ -104,25 +104,25 @@ yet. Consistency test: every in-repo yaml declares `runners`, engine and os know
 
 ## Phase 2: Jobs are dispatched by runner
 
-### Task 5: `QuantingEnv.runner`, factory takes a `Runner`
+### Task 5: `QuantingEnv.runner_name`, factory takes a `Runner`
 
-**Description:** `QuantingEnv.job_engine` -> `runner: str = Field(alias="_RUNNER")`.
+**Description:** `QuantingEnv.job_engine` -> `runner_name: str = Field(alias="_RUNNER_NAME")`.
 `_get_job_handler(runner: Runner)` dispatches on `runner.engine`; `start_job/get_job_status/
-get_job_result(..., runner_name)` call `get_runner`. `ssh_sensor.py` reads `.runner`.
-`processor_impl` passes `quanting_env.runner` to the job functions and, for this commit only,
-sets `runner=settings.job_engine` (transitional line, removed in T6, named in the commit
+get_job_result(..., runner_name)` call `get_runner`. `ssh_sensor.py` reads `.runner_name`.
+`processor_impl` passes `quanting_env.runner_name` to the job functions and, for this commit only,
+sets `runner_name=settings.job_engine` (transitional line, removed in T6, named in the commit
 message). `runner` joins the strict list of `_check_content`. Slurm branch still resolves
 `CLUSTER_VIEW` for its base dir (T8). Spec 2.3, 2.5 factory signature, 7.5 partial, 9.5.
 
 **Acceptance criteria:**
-- [ ] `QuantingEnv.to_dict()` differs from before only in `_JOB_ENGINE` -> `_RUNNER` (existing full-dict assertion in `test_create_quanting_env`, spec 7.3).
-- [ ] Factory per engine with a `Runner`; unknown engine raises `ValueError`.
-- [ ] `start_job(env, runner_name="nope")` raises `KeyError` listing declared runners (9.5).
-- [ ] `grep -rn job_engine airflow_src --include='*.py'` hits only the transitional line.
+- [x] `QuantingEnv.to_dict()` differs from before only in `_JOB_ENGINE` -> `_RUNNER_NAME` (existing full-dict assertion in `test_create_quanting_env`, spec 7.3).
+- [x] Factory per engine with a `Runner`; unknown engine raises `ValueError`.
+- [x] `start_job(env, runner_name="nope")` raises `KeyError` listing declared runners (9.5).
+- [x] `grep -rn job_engine airflow_src --include='*.py'` hits only the transitional line.
 
 **Verification:**
-- [ ] `pytest airflow_src/tests/plugins/jobs/test_job_handler.py airflow_src/tests/plugins/sensors airflow_src/tests/dags/impl/test_processor_impl.py`
-- [ ] `pytest airflow_src`
+- [x] `pytest airflow_src/tests/plugins/jobs/test_job_handler.py airflow_src/tests/plugins/sensors airflow_src/tests/dags/impl/test_processor_impl.py`
+- [x] `pytest airflow_src`
 
 **Dependencies:** T3, T4
 **Files:** `airflow_src/plugins/common/quanting_env.py`, `airflow_src/plugins/jobs/job_handler.py`, `airflow_src/plugins/sensors/ssh_sensor.py`, `airflow_src/dags/impl/processor_impl.py`, tests: `conftest.py`, `test_job_handler.py`, `test_ssh_sensor.py`, `test_processor_impl.py`
@@ -131,7 +131,7 @@ message). `runner` joins the strict list of `_check_content`. Slurm branch still
 ### Task 6: `Settings.runner` replaces `job_engine`; webapp selectbox
 
 **Description:** `Settings.runner = StringField(required=True, max_length=64)`, `job_engine`
-deleted; `create_settings(runner=...)`. The T5 transitional line becomes `runner=settings.runner`.
+deleted; `create_settings(runner=...)`. The T5 transitional line becomes `runner_name=settings.runner`.
 Webapp (spec 2.7): options `list(RUNNERS)`, default first declared, `SHOW_RUNNER_SELECT`, prefill
 key `runner`, docker-only-custom check via `RUNNERS[runner].engine`, help texts at 314 and 415
 say "runner". Empty `RUNNERS` (spec 1.2.4): no settings form, a notice instead; covers the
@@ -172,7 +172,7 @@ precondition. Spec 2.8, 9.7.
 
 ### Checkpoint 2: Runner flows
 - [ ] Three suites green, pre-commit clean; `grep -n "job_engine" airflow_src/dags/impl/processor_impl.py` empty.
-- [ ] Local stack: create a settings entry in the webapp with runner `slurm`, trigger a quanting DAG with `debug_no_cluster_ssh=true`, the `prepare_job` XCom shows `_RUNNER: slurm` and the same paths as before.
+- [ ] Local stack: create a settings entry in the webapp with runner `slurm`, trigger a quanting DAG with `debug_no_cluster_ssh=true`, the `prepare_job` XCom shows `_RUNNER_NAME: slurm` and the same paths as before.
 - [ ] Migration `--dry-run` runs against a sandbox DB copy; note the distinct names (open question 1).
 - [ ] Human review before Phase 3.