feat: use a job scheduler to launch workflows when the system is healthy (SBP-421)

README.md

@@ -42,6 +42,12 @@ FastAPI backend for handling Seqera Platform workflow launches.
    uv run uvicorn app.main:app --reload --host 0.0.0.0 --port 3000
    ```
 
+5. Run the job scheduler locally (you probably want `--dry-run` so jobs aren't submitted to Seqera)
+
+   ```bash
+   uv run python app/run_scheduler.py --dry-run
+   ```
+
 ## API Endpoints
 
 - `GET /health` — Lightweight health probe

alembic/versions/20260702_104632_system_status_cache_5d4995c0bdca.py

@@ -0,0 +1,30 @@
+"""system_status_cache"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '5d4995c0bdca'
+down_revision = 'df9e54119fe9'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('system_status_cache',
+    sa.Column('key', sa.Text(), nullable=False),
+    sa.Column('payload', sa.JSON(), nullable=False),
+    sa.Column('checked_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('expires_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+    sa.PrimaryKeyConstraint('key', name=op.f('pk_system_status_cache'))
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('system_status_cache')
+    # ### end Alembic commands ###

alembic/versions/20260702_144353_null_seqera_id_74ab4858580d.py

@@ -0,0 +1,27 @@
+"""null_seqera_id"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '74ab4858580d'
+down_revision = '5d4995c0bdca'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('workflow_runs', 'seqera_run_id',
+               existing_type=sa.TEXT(),
+               nullable=True)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.alter_column('workflow_runs', 'seqera_run_id',
+               existing_type=sa.TEXT(),
+               nullable=False)
+    # ### end Alembic commands ###

app/db/models/__init__.py

@@ -10,3 +10,4 @@
     WorkflowRun,
 )
 from .job_queue import QueuedJob
+from .system_status import SystemStatusCache

app/db/models/core.py

@@ -61,7 +61,7 @@ class WorkflowRun(Base):
     id: Mapped[UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid4)
     workflow_id: Mapped[UUID | None] = mapped_column(ForeignKey("workflows.id"))
     owner_user_id: Mapped[UUID] = mapped_column(ForeignKey("app_users.id"), nullable=False)
-    seqera_run_id: Mapped[str] = mapped_column(Text, nullable=False)
+    seqera_run_id: Mapped[str | None] = mapped_column(Text, nullable=True, default=None)
     binder_name: Mapped[str | None] = mapped_column(Text, nullable=True)
     sample_id: Mapped[str | None] = mapped_column(Text, nullable=True)
     run_name: Mapped[str | None] = mapped_column(Text, nullable=True)

app/db/models/system_status.py

@@ -0,0 +1,40 @@
+"""Database-backed cache for runtime system health."""
+
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import JSON, DateTime, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from ...schemas.health import SystemStatus
+from .. import Base
+
+
+class SystemStatusCache(Base):
+    """
+    Stores current system health status in the DB, so it can be shared across
+    processes. There should be only one row in the table,
+    with all systems using a common key.
+    """
+
+    __tablename__ = "system_status_cache"
+
+    key: Mapped[str] = mapped_column(Text, primary_key=True)
+    payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
+    checked_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+    def is_fresh(self, now: datetime) -> bool:
+        expires_at = self.expires_at
+        if expires_at.tzinfo is None:
+            expires_at = expires_at.replace(tzinfo=UTC)
+        if now.tzinfo is None:
+            now = now.replace(tzinfo=UTC)
+        return now < expires_at
+
+    def get_status(self) -> SystemStatus:
+        """
+        Return a SystemStatus object from the cache payload.
+        """
+        return SystemStatus.model_validate(self.payload)

app/routes/health.py

@@ -15,10 +15,11 @@
 from __future__ import annotations
 
 from fastapi import APIRouter, Depends
+from sqlalchemy.orm import Session
 
 from ..schemas.health import ComponentsHealthResponse
 from ..services import health
-from .dependencies import get_current_user_id, require_workflow_execution_role
+from .dependencies import get_current_user_id, get_db, require_workflow_execution_role
 
 router = APIRouter(
     tags=["health"],
@@ -27,16 +28,14 @@
 
 
 @router.get("/components", response_model=ComponentsHealthResponse)
-async def get_components_health() -> ComponentsHealthResponse:
+async def get_components_health(db: Session = Depends(get_db)) -> ComponentsHealthResponse:
     """Return a coarse, user-facing health summary for SBP-bundle users.
 
     ``overallStatus`` is the worst status across all monitored components; when it
     is not ``healthy`` a generic ``message`` is included for display on the job
     details page.
 
-    Always served from the short-lived cache (with background refresh); there is
-    no caller-triggered force-refresh here. Forcing a live probe run is reserved
-    for the admin endpoint, since the Tower Agent probe can mutate Seqera state.
+    Always comes from the shared database cache
     """
-    status_obj = await health.get_system_status()
+    status_obj = await health.get_system_status(db)
     return ComponentsHealthResponse.model_validate(health.to_components_health_dict(status_obj))

app/routes/system_status.py

@@ -13,10 +13,12 @@
 from __future__ import annotations
 
 from fastapi import APIRouter, Depends, Query
+from sqlalchemy.orm import Session
 
 from ..db.admin import require_admin_access
 from ..schemas.health import SystemStatusAdminResponse
 from ..services import health
+from .dependencies import get_db
 
 router = APIRouter(
     tags=["system-status"],
@@ -28,9 +30,10 @@
 async def get_admin_system_status(
     refresh: bool = Query(
         default=False,
-        description="Bypass the short-lived cache and re-run the probes now",
+        description="Bypass the shared database cache and re-run the probes now",
     ),
+    db: Session = Depends(get_db),
 ) -> SystemStatusAdminResponse:
     """Return verbose, admin-only runtime health of the submission components."""
-    status_obj = await health.get_system_status(force_refresh=refresh)
+    status_obj = await health.get_system_status(db, force_refresh=refresh)
     return SystemStatusAdminResponse.model_validate(health.to_admin_dict(status_obj))

app/routes/workflow/jobs.py

@@ -23,16 +23,14 @@
     map_pipeline_status_to_ui,
 )
 from ...services.job_utils import (
+    UserJobListRow,
     coerce_workflow_payload,
     ensure_completed_run_score,
     extract_pipeline_status,
     format_tool_name,
     format_workflow_name,
     get_owned_run,
-    get_owned_runs_by_run_id,
-    get_score_by_seqera_run_id,
-    get_tool_by_seqera_run_id,
-    get_workflow_type_by_seqera_run_id,
+    get_user_job_list_rows,
     parse_submit_datetime,
 )
 from ...services.seqera import describe_workflow
@@ -101,44 +99,68 @@ async def list_jobs(
     db: Session = Depends(get_db),
 ) -> JobListResponse:
     """Retrieve a paginated list of the current user's jobs with search and filtering."""
-    owned_runs = get_owned_runs_by_run_id(db, current_user_id)
-    score_by_run_id = get_score_by_seqera_run_id(db, current_user_id)
-    workflow_type_by_run_id = get_workflow_type_by_seqera_run_id(db, current_user_id)
-    tool_by_run_id = get_tool_by_seqera_run_id(db, current_user_id)
+    user_runs = get_user_job_list_rows(db, current_user_id)
     search_text = (search or "").strip().lower()
     allowed_statuses = set(status_filter or [])
 
-    async def _fetch_seqera(run_id: str) -> tuple[str, dict[str, object] | None]:
+    async def _fetch_seqera(user_run: UserJobListRow) -> tuple[str, dict[str, object] | None]:
         """None signals a 4xx (skip); empty dict signals a 5xx/network error (DB fallback)."""
+        seqera_run_id = user_run.seqera_run_id
+        if not seqera_run_id:
+            return user_run.run_id, {}
         try:
-            return run_id, await describe_workflow(run_id)
+            return user_run.run_id, await describe_workflow(seqera_run_id)
         except SeqeraAPIError as exc:
             if exc.status_code is not None and exc.status_code < 500:
-                return run_id, None
-            logger.warning("Seqera unavailable for run %s, using DB fallback: %s", run_id, exc)
-            return run_id, {}
+                return user_run.run_id, None
+            logger.warning(
+                "Seqera unavailable for run %s, using DB fallback: %s",
+                seqera_run_id,
+                exc,
+            )
+            return user_run.run_id, {}
         except Exception as exc:
-            logger.warning("Seqera unreachable for run %s, using DB fallback: %s", run_id, exc)
-            return run_id, {}
-
-    seqera_results = await asyncio.gather(*(_fetch_seqera(r) for r in owned_runs))
+            logger.warning(
+                "Seqera unreachable for run %s, using DB fallback: %s",
+                seqera_run_id,
+                exc,
+            )
+            return user_run.run_id, {}
+
+    seqera_fetch_rows = [
+        user_run
+        for user_run in user_runs
+        if user_run.seqera_run_id and user_run.queued_status not in {"pending", "failed"}
+    ]
+    seqera_results = dict(
+        await asyncio.gather(*(_fetch_seqera(user_run) for user_run in seqera_fetch_rows))
+    )
 
     jobs: list[JobListItem] = []
     seqera_unavailable = False
 
-    for run_id, seqera_payload in seqera_results:
-        if seqera_payload is None:
-            # 4xx: run is inaccessible (not found, wrong workspace, no permission).
-            continue
-
-        owned_run = owned_runs[run_id]
-
-        if seqera_payload:
-            pipeline_status = extract_pipeline_status(seqera_payload)
-            ui_status = map_pipeline_status_to_ui(pipeline_status)
-        else:
-            ui_status = "N/A"
-            seqera_unavailable = True
+    for user_run in user_runs:
+        run_id = user_run.run_id
+        owned_run = user_run.run
+        seqera_run_id = user_run.seqera_run_id
+
+        seqera_payload: dict[str, object] = {}
+        ui_status = "N/A"
+        if user_run.queued_status == "pending":
+            ui_status = "Pending"
+        elif user_run.queued_status == "failed":
+            ui_status = "Failed"
+        elif seqera_run_id:
+            fetched_payload = seqera_results.get(run_id)
+            if fetched_payload is None:
+                # 4xx: run is inaccessible (not found, wrong workspace, no permission).
+                continue
+            if fetched_payload:
+                seqera_payload = fetched_payload
+                pipeline_status = extract_pipeline_status(seqera_payload)
+                ui_status = map_pipeline_status_to_ui(pipeline_status)
+            else:
+                seqera_unavailable = True
 
         if allowed_statuses and ui_status not in allowed_statuses:
             continue
@@ -150,8 +172,8 @@ async def _fetch_seqera(run_id: str) -> tuple[str, dict[str, object] | None]:
         if submitted_at is None:
             submitted_at = datetime.now(UTC)
 
-        workflow_type = workflow_type_by_run_id.get(run_id) or "Unknown"
-        tool = tool_by_run_id.get(run_id, "Unknown")
+        workflow_type = user_run.workflow_type
+        tool = user_run.tool
         job_name = _resolve_job_name(run_id, wf, owned_run)
 
         if (
@@ -161,7 +183,7 @@ async def _fetch_seqera(run_id: str) -> tuple[str, dict[str, object] | None]:
         ):
             continue
 
-        db_score = score_by_run_id.get(run_id)
+        db_score = user_run.score
 
         # A cached score means the job completed at some point; treat it as Completed
         # when Seqera is unreachable and we cannot get the live status.
@@ -181,7 +203,7 @@ async def _fetch_seqera(run_id: str) -> tuple[str, dict[str, object] | None]:
                 status=ui_status,
                 submittedAt=submitted_at,
                 score=score if ui_status == "Completed" else None,
-                finalDesignCount=_resolve_final_design_count(owned_run),
+                finalDesignCount=user_run.final_design_count,
             )
         )