more pios tasks, Assign pioreactors UX

Author: CamDavidsonPilon

CHANGELOG.md

@@ -11,12 +11,14 @@
 
 #### Enhancements
 
- - Added `/api/units/<pioreactor_unit>/jobs/stop/experiments/<experiment>` to mirror worker stop-all behavior.
  - Normalized calibration route parameter naming to `calibration_name` across API and unit API calibration endpoints for more consistent endpoint templates.
+ - Added `pios jobs list` and `pios jobs list running` to inspect worker job history and running jobs from the leader CLI.
+ - Added `GET /unit_api/jobs` for per-unit job history (ordered by newest first), complementing the existing `GET /unit_api/jobs/running` endpoint.
 
 #### Bug fixes
 
  - Fixed `/api/config/units/$broadcast` to correctly merge each unit's own `config_<unit>.ini` instead of using a shared `config_$broadcast.ini` path.
+ - Improved the UI's "Assign Pioreactors" dialog: units already assigned to another experiment can now be selected for reassignment, their status text updates to "Will be unassigned from <experiment>" when selected, and "Select all" now applies consistently to all listed units.
 
 ### 26.2.3
 

core/pioreactor/cli/pios.py

@@ -4,6 +4,7 @@
 """
 import os
 from concurrent.futures import ThreadPoolExecutor
+from typing import Any
 
 import click
 from msgspec import DecodeError
@@ -18,6 +19,7 @@
 from pioreactor.exc import SSHError
 from pioreactor.logging import create_logger
 from pioreactor.mureq import HTTPException
+from pioreactor.pubsub import get_from
 from pioreactor.pubsub import post_into
 from pioreactor.utils.job_manager import ClusterJobManager
 from pioreactor.utils.networking import cp_file_across_cluster
@@ -209,6 +211,107 @@ def parse_click_arguments(input_list: list[str]) -> dict:  # TODO: typed dict
 
         return {"args": args, "options": opts}
 
+    def _format_timestamp_to_seconds(timestamp: str | None) -> str:
+        if timestamp is None:
+            return ""
+
+        from pioreactor.utils.timing import to_datetime
+
+        try:
+            dt = to_datetime(timestamp)
+        except ValueError:
+            return timestamp
+
+        return dt.replace(microsecond=0).strftime("%Y-%m-%dT%H:%M:%S")
+
+    def _format_job_history_line(
+        job_id: int,
+        job_name: str,
+        experiment: str,
+        job_source: str | None,
+        unit: str,
+        started_at: str,
+        ended_at: str | None,
+    ) -> str:
+        job_source_display = job_source or "unknown"
+        ended_at_display = _format_timestamp_to_seconds(ended_at) or "still running"
+        job_id_label = click.style(f"[job_id={job_id}]", fg="cyan")
+        job_name_label = click.style(job_name, fg="green", bold=True)
+        ended_at_label = (
+            click.style(ended_at_display, fg="yellow", bold=True) if ended_at is None else ended_at_display
+        )
+
+        return (
+            f"{job_id_label} {job_name_label} "
+            f"experiment={experiment}, source={job_source_display}, "
+            f"started_at={_format_timestamp_to_seconds(started_at)}, ended_at={ended_at_label}"
+        )
+
+    def _show_cluster_job_history(
+        units: tuple[str, ...],
+        experiments: tuple[str, ...],
+        *,
+        running_only: bool = False,
+    ) -> None:
+        units = resolve_target_units(units, experiments, active_only=True, include_leader=None)
+        if len(units) == 0:
+            click.echo("No jobs recorded.")
+            return
+
+        endpoint = "/unit_api/jobs/running" if running_only else "/unit_api/jobs"
+        all_rows: list[tuple[int, str, str, str | None, str, str, str | None]] = []
+
+        def _thread_function(unit: str) -> tuple[bool, str, list[dict[str, Any]]]:
+            try:
+                response = get_from(resolve_to_address(unit), endpoint)
+                response.raise_for_status()
+                payload = response.json()
+                if not isinstance(payload, list):
+                    raise ValueError("Expected list payload")
+                return True, unit, payload
+            except (HTTPException, ValueError) as e:
+                click.echo(f"Unable to list jobs on {unit}: {e}", err=True)
+                return False, unit, []
+
+        with ThreadPoolExecutor(max_workers=len(units)) as executor:
+            results = executor.map(_thread_function, units)
+
+        for _, unit, rows in results:
+            for raw_row in rows:
+                if not isinstance(raw_row, dict):
+                    continue
+
+                try:
+                    job_id = int(raw_row["job_id"])
+                    job_name = str(raw_row["job_name"])
+                    experiment = str(raw_row["experiment"])
+                    job_source_raw = raw_row.get("job_source")
+                    row_unit = str(raw_row.get("unit", unit))
+                    started_at = str(raw_row["started_at"])
+                    ended_at_raw = raw_row.get("ended_at")
+                except (KeyError, TypeError, ValueError):
+                    continue
+
+                all_rows.append(
+                    (
+                        job_id,
+                        job_name,
+                        experiment,
+                        None if job_source_raw is None else str(job_source_raw),
+                        row_unit,
+                        started_at,
+                        None if ended_at_raw is None else str(ended_at_raw),
+                    )
+                )
+
+        if not all_rows:
+            click.echo("No jobs recorded.")
+            return
+
+        all_rows.sort(key=lambda job_row: job_row[5], reverse=True)
+        for job_row in all_rows:
+            click.echo(_format_job_history_line(*job_row))
+
     def universal_identifier_to_all_active_workers(workers: tuple[str, ...]) -> tuple[str, ...]:
         try:
             active_workers = get_active_workers_in_inventory()
@@ -782,6 +885,23 @@ def _thread_function(unit: str) -> bool:
         if not all(results):
             raise click.Abort()
 
+    @pios.group(name="jobs", short_help="job-related commands")
+    def jobs():
+        """Interact with worker jobs."""
+        pass
+
+    @jobs.group(name="list", short_help="list jobs current and previous", invoke_without_command=True)
+    @which_units
+    @click.pass_context
+    def list_jobs(ctx, units: tuple[str, ...], experiments: tuple[str, ...]) -> None:
+        if ctx.invoked_subcommand is None:
+            _show_cluster_job_history(units, experiments, running_only=False)
+
+    @list_jobs.command(name="running", short_help="show status of running job(s)")
+    @which_units
+    def list_running_jobs(units: tuple[str, ...], experiments: tuple[str, ...]) -> None:
+        _show_cluster_job_history(units, experiments, running_only=True)
+
     @pios.command("kill", short_help="kill a job(s) on workers")
     @click.option("--all-jobs", is_flag=True, help="kill all worker jobs")
     @click.option("--experiment", type=click.STRING)

core/pioreactor/web/tasks.py

@@ -38,6 +38,7 @@
 from pioreactor import whoami
 from pioreactor.config import config as pioreactor_config
 from pioreactor.logging import create_logger
+from pioreactor.models import CORE_MODELS
 from pioreactor.mureq import HTTPErrorStatus
 from pioreactor.mureq import HTTPException
 from pioreactor.mureq import Response
@@ -286,6 +287,11 @@ def check_model_hardware(model_name: str, model_version: str) -> None:
     if model_version != "1.5":
         return
 
+    if (model_name, model_version) in CORE_MODELS:
+        display_name = CORE_MODELS[(model_name, model_version)].display_name
+    else:
+        display_name = f"{model_name} {model_version}"
+
     try:
         addresses = _get_adc_addresses_for_model(model_name, model_version)
     except exc.HardwareNotFoundError as err:
@@ -295,21 +301,19 @@ def check_model_hardware(model_name: str, model_version: str) -> None:
         return
 
     if not addresses:
-        logger.debug(
-            f"Hardware check found no ADC addresses for {model_name} {model_version} on {get_unit_name()}."
-        )
+        logger.debug(f"Hardware check found no ADC addresses for {display_name} on {get_unit_name()}.")
         return
 
     missing = sorted(addr for addr in addresses if not hardware.is_i2c_device_present(addr))
     if missing:
         missing_hex = ", ".join(hex(addr) for addr in missing)
         logger.warning(
-            f"Hardware check failed for {model_name} {model_version} on {get_unit_name()}: "
+            f"Hardware check failed for {display_name} on {get_unit_name()}: "
             f"missing I2C devices at {missing_hex}."
         )
         return
 
-    logger.notice(f"Correct hardware found for {model_name} {model_version} on {get_unit_name()}.")
+    logger.notice(f"Correct hardware found for {display_name} on {get_unit_name()}.")
     return
 
 

core/pioreactor/web/unit_api.py

@@ -594,6 +594,25 @@ def stop_jobs() -> DelayedResponseReturnValue:
     return create_task_response(task)
 
 
+@unit_api_bp.route("/jobs", methods=["GET"])
+def get_jobs() -> ResponseReturnValue:
+    jobs = query_temp_local_metadata_db(
+        """
+        SELECT
+            job_id,
+            job_name,
+            experiment,
+            job_source,
+            unit,
+            started_at,
+            ended_at
+        FROM pio_job_metadata
+        ORDER BY started_at DESC
+        """
+    )
+    return jsonify(jobs)
+
+
 @unit_api_bp.route("/jobs/running/experiments/<experiment>", methods=["GET"])
 def get_running_jobs_for_experiment(experiment: str) -> ResponseReturnValue:
     jobs = query_temp_local_metadata_db(

core/tests/test_cli.py

@@ -243,6 +243,74 @@ def test_pios_kill_requests() -> None:
     assert bucket[1].json == {"experiment": "demo"}
 
 
+def test_pios_jobs_list_requests_history_endpoint(monkeypatch) -> None:
+    class DummyResponse:
+        def raise_for_status(self) -> None:
+            return
+
+        def json(self) -> list[dict[str, str | int | None]]:
+            return [
+                {
+                    "job_id": 42,
+                    "job_name": "stirring",
+                    "experiment": "_testing_experiment",
+                    "job_source": "cli",
+                    "unit": "unit1",
+                    "started_at": "2026-01-01T00:00:00.000Z",
+                    "ended_at": "2026-01-01T00:10:00.000Z",
+                }
+            ]
+
+    captured: list[tuple[str, str]] = []
+
+    def fake_get_from(address: str, endpoint: str, **_kwargs):
+        captured.append((address, endpoint))
+        return DummyResponse()
+
+    monkeypatch.setattr("pioreactor.cli.pios.get_from", fake_get_from)
+
+    runner = CliRunner()
+    result = runner.invoke(pios, ["jobs", "list", "--units", "unit1"])
+    assert result.exit_code == 0
+    assert captured == [("unit1.local", "/unit_api/jobs")]
+    assert "[job_id=42]" in result.output
+    assert "stirring" in result.output
+
+
+def test_pios_jobs_list_running_requests_running_endpoint(monkeypatch) -> None:
+    class DummyResponse:
+        def raise_for_status(self) -> None:
+            return
+
+        def json(self) -> list[dict[str, str | int | None]]:
+            return [
+                {
+                    "job_id": 43,
+                    "job_name": "od_reading",
+                    "experiment": "_testing_experiment",
+                    "job_source": "cli",
+                    "unit": "unit1",
+                    "started_at": "2026-01-01T00:00:00.000Z",
+                    "ended_at": None,
+                }
+            ]
+
+    captured: list[tuple[str, str]] = []
+
+    def fake_get_from(address: str, endpoint: str, **_kwargs):
+        captured.append((address, endpoint))
+        return DummyResponse()
+
+    monkeypatch.setattr("pioreactor.cli.pios.get_from", fake_get_from)
+
+    runner = CliRunner()
+    result = runner.invoke(pios, ["jobs", "list", "running", "--units", "unit1"])
+    assert result.exit_code == 0
+    assert captured == [("unit1.local", "/unit_api/jobs/running")]
+    assert "[job_id=43]" in result.output
+    assert "still running" in result.output
+
+
 def test_pio_job_info_lists_job() -> None:
     runner = CliRunner()
     job_name = "test_job"

core/tests/web/test_unit_api.py

@@ -167,3 +167,80 @@ class DummyTask:
     data = resp.get_json()
     assert data["task_id"] == "task-123"
     assert captured["args"] == ("install", "pioreactor-air-bubbler")
+
+
+def test_get_jobs_returns_history(client) -> None:
+    from time import sleep
+
+    from pioreactor.utils.job_manager import JobManager
+
+    with JobManager() as jm:
+        old_job_id = jm.register_and_set_running(
+            unit="unit1",
+            experiment="exp_old",
+            job_name="old_job",
+            job_source="test",
+            pid=1001,
+            leader="leader",
+            is_long_running_job=False,
+        )
+        jm.set_not_running(old_job_id)
+
+        sleep(0.02)
+
+        newest_job_id = jm.register_and_set_running(
+            unit="unit1",
+            experiment="exp_new",
+            job_name="new_job",
+            job_source="test",
+            pid=1002,
+            leader="leader",
+            is_long_running_job=False,
+        )
+
+    response = client.get("/unit_api/jobs")
+    assert response.status_code == 200
+    rows = response.get_json()
+    assert isinstance(rows, list)
+    assert [row["job_id"] for row in rows[:2]] == [newest_job_id, old_job_id]
+    assert set(rows[0]) == {
+        "job_id",
+        "job_name",
+        "experiment",
+        "job_source",
+        "unit",
+        "started_at",
+        "ended_at",
+    }
+
+
+def test_get_running_jobs_endpoint_filters_results(client) -> None:
+    from pioreactor.utils.job_manager import JobManager
+
+    with JobManager() as jm:
+        stopped_job_id = jm.register_and_set_running(
+            unit="unit1",
+            experiment="exp_old",
+            job_name="old_job",
+            job_source="test",
+            pid=1003,
+            leader="leader",
+            is_long_running_job=False,
+        )
+        jm.set_not_running(stopped_job_id)
+
+        running_job_id = jm.register_and_set_running(
+            unit="unit1",
+            experiment="exp_new",
+            job_name="new_job",
+            job_source="test",
+            pid=1004,
+            leader="leader",
+            is_long_running_job=False,
+        )
+
+    response = client.get("/unit_api/jobs/running")
+    assert response.status_code == 200
+    rows = response.get_json()
+    assert isinstance(rows, list)
+    assert [row["job_id"] for row in rows] == [running_job_id]