feat: workflow staging

.env.example

@@ -63,3 +63,10 @@ GLOBUS_S3_COLLECTION_ID=dev-globus-s3-collection-id
 GLOBUS_GADI_COLLECTION_ROOT=on-gadi-collection-root
 GLOBUS_INPUT_DIR=on-gadi-input-dir
 GLOBUS_OUTPUT_DIR=on-gadi-output-dir
+
+# GitHub App used to look up workflow repo commits (see workflow_repo_staging.py).
+# Preferred over a personal access token since this is an org-owned service.
+GITHUB_WORKFLOW_STAGING_AUTOMATION_APP_ID=github-app-id
+GITHUB_WORKFLOW_STAGING_AUTOMATION_APP_PRIVATE_KEY=
+# Local-dev-only fallback if you don't want to set up a GitHub App - never use in production.
+GITHUB_WORKFLOW_STAGING_AUTOMATION_TOKEN=

Dockerfile

@@ -6,6 +6,14 @@ ENV PYTHONDONTWRITEBYTECODE=1 \
 
 WORKDIR /app
 
+# git is required by app/services/workflow_repo_staging.py, which clones
+# workflow pipeline repos (with real .git metadata, not just extracted source)
+# to stage them onto Gadi via S3 + Globus - Gadi compute nodes have no network
+# access, so Nextflow can't fetch/clone them itself at run time.
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends git \
+    && rm -rf /var/lib/apt/lists/*
+
 # Install UV
 COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
 

alembic/versions/20260818_033139_add_workflow_repo_staging_cache_b1f4c9a7e021.py

@@ -0,0 +1,29 @@
+"""add workflow repo staging cache"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = 'b1f4c9a7e021'
+down_revision = '70ac6b86efb4'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column('workflows', sa.Column('repo_staged_commit_sha', sa.Text(), nullable=True))
+    op.add_column('workflows', sa.Column('repo_staging_status', sa.String(length=20), nullable=True))
+    op.add_column('workflows', sa.Column('repo_gadi_path', sa.Text(), nullable=True))
+    op.add_column('workflows', sa.Column('repo_staging_transfer_id', sa.Text(), nullable=True))
+    op.add_column('workflows', sa.Column('repo_staging_error_message', sa.Text(), nullable=True))
+    op.add_column('workflows', sa.Column('repo_staging_updated_at', sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column('workflows', 'repo_staging_updated_at')
+    op.drop_column('workflows', 'repo_staging_error_message')
+    op.drop_column('workflows', 'repo_staging_transfer_id')
+    op.drop_column('workflows', 'repo_gadi_path')
+    op.drop_column('workflows', 'repo_staging_status')
+    op.drop_column('workflows', 'repo_staged_commit_sha')

app/config.py

@@ -75,6 +75,7 @@ class AdminSettings(NestedSettings):
     cookie_secure: bool
     session_secret: str
     roles_claim: str
+    title: str = "SBP Backend Admin"
 
     model_config = SettingsConfigDict(env_prefix="DB_ADMIN_")
 
@@ -125,6 +126,30 @@ def reject_empty_values(cls, value: str) -> str:
     model_config = SettingsConfigDict(env_prefix="GLOBUS_")
 
 
+class GithubSettings(NestedSettings):
+    """Credentials for GitHub API calls in workflow_repo_staging.py.
+
+    Prefer the workflow-staging-automation GitHub App
+    (workflow_staging_automation_app_id/_private_key) - org-owned, not tied
+    to a person. workflow_staging_automation_token (a PAT) is a local-dev-only
+    fallback.
+    """
+
+    workflow_staging_automation_app_id: str | None = None
+    workflow_staging_automation_app_private_key: str | None = None
+    workflow_staging_automation_token: str | None = None
+
+    @field_validator("workflow_staging_automation_app_private_key")
+    @classmethod
+    def _normalize_private_key_newlines(cls, value: str | None) -> str | None:
+        # Secrets managers often flatten PEM newlines to literal "\n" - restore them.
+        if value is None:
+            return None
+        return value.replace("\\n", "\n")
+
+    model_config = SettingsConfigDict(env_prefix="GITHUB_")
+
+
 class Settings(BaseSettings):
     """
     Core settings for the app.
@@ -150,6 +175,7 @@ class Settings(BaseSettings):
     admin: AdminSettings = Field(default_factory=AdminSettings)
     auth: AuthSettings = Field(default_factory=AuthSettings)
     globus: GlobusSettings = Field(default_factory=GlobusSettings)
+    github: GithubSettings = Field(default_factory=GithubSettings)
 
     model_config = SettingsConfigDict(env_file=".env", dotenv_filtering="only_existing")
 

app/db/admin.py

@@ -141,12 +141,27 @@ class WorkflowAdmin(ModelView):
         "default_revision",
         "config_path",
         "prerun_script_path",
+        # Cache of the repo checkout currently staged on Gadi via Globus for
+        # this workflow (see app/services/workflow_repo_staging.py) - a single
+        # slot shared by every run, not per-run history.
+        "repo_staged_commit_sha",
+        "repo_staging_status",
+        "repo_gadi_path",
+        "repo_staging_transfer_id",
+        "repo_staging_updated_at",
+        "repo_staging_error_message",
     ]
+    exclude_fields_from_list = ["repo_staging_transfer_id", "repo_staging_error_message"]
 
     _NULLABLE_FIELDS = (
         "description",
         "tool",
         "prerun_script_path",
+        "repo_staged_commit_sha",
+        "repo_staging_status",
+        "repo_gadi_path",
+        "repo_staging_transfer_id",
+        "repo_staging_error_message",
     )
 
     def _nullify_empty_fields(self, obj: Any) -> None:
@@ -681,7 +696,7 @@ def mount_db_admin(app: FastAPI, settings: Settings) -> None:
     # before this mount) so it stays available independently of the dashboard.
     _mount_db_debug_api(app)
     _mount_admin_ui_assets(app)
-    _mount_starlette_admin(app)
+    _mount_starlette_admin(app, settings)
 
 
 def _mount_admin_ui_assets(app: FastAPI) -> None:
@@ -700,7 +715,7 @@ def workflow_run_export_js() -> Response:
     app.include_router(router)
 
 
-def _mount_starlette_admin(app: FastAPI) -> None:
+def _mount_starlette_admin(app: FastAPI, settings: Settings) -> None:
     session_cookie_name = _get_admin_session_cookie_name()
     oauth_state_cookie_name = "sbp_admin_oauth_state"
     oauth_verifier_cookie_name = "sbp_admin_oauth_verifier"
@@ -937,7 +952,7 @@ async def serialize_value(
 
     admin = Admin(
         engine=engine,
-        title=os.getenv("DB_ADMIN_TITLE", "SBP Backend Admin"),
+        title=settings.admin.title,
         templates_dir=_ADMIN_TEMPLATES_DIR,
         auth_provider=Auth0AdminAuthProvider(),
         # Timestamps are stored as UTC; always display them in Sydney/Melbourne

app/db/models/core.py

@@ -53,6 +53,11 @@ class AppUser(Base):
     workflow_runs: Mapped[list[WorkflowRun]] = relationship(back_populates="owner")
 
 
+# Mirrors DataTransferStatus (defined further below) - kept as its own alias
+# since a workflow's repo staging is a distinct cache concept, not a DataTransfer.
+RepoStagingStatus = Literal["pending", "in_progress", "completed", "failed"]
+
+
 class Workflow(Base):
     __tablename__ = "workflows"
     __table_args__ = (
@@ -81,6 +86,20 @@ class Workflow(Base):
     config_path: Mapped[str] = mapped_column(Text, nullable=False)
     prerun_script_path: Mapped[str | None] = mapped_column(Text, nullable=True)
     tool: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # Cache of the most recently staged (repo_url, default_revision) commit on
+    # Gadi, shared across every run of this workflow
+    repo_staged_commit_sha: Mapped[str | None] = mapped_column(Text, nullable=True)
+    repo_staging_status: Mapped[RepoStagingStatus | None] = mapped_column(
+        String(length=20), nullable=True
+    )
+    repo_gadi_path: Mapped[str | None] = mapped_column(Text, nullable=True)
+    # Holds Globus's submission id until submission succeeds, then the real
+    # Globus task id thereafter - same reuse trick as DataTransfer.transfer_id.
+    repo_staging_transfer_id: Mapped[str | None] = mapped_column(Text, nullable=True)
+    repo_staging_error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    repo_staging_updated_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
 
     runs: Mapped[list[WorkflowRun]] = relationship(back_populates="workflow")
 

app/routes/workflows.py

@@ -44,7 +44,7 @@
     SinglePredictionEntity,
     validate_single_prediction_entities,
 )
-from ..services.bindflow_executor import prepare_bindflow_workflow
+from ..services.bindflow_executor import prepare_bindflow_workflow, resolve_bindflow_asset_path
 from ..services.credits import (
     WorkflowCreditsResponse,
     is_credits_enabled,
@@ -69,6 +69,7 @@
 )
 from ..services.seqera_errors import WorkflowLaunchError
 from ..services.wisps_executor import prepare_wisps_workflow
+from ..services.workflow_repo_staging import RepoStagingError, ensure_repo_staging_requested
 from .dependencies import (
     get_client_ip,
     get_current_user_id,
@@ -281,6 +282,64 @@ async def _stage_referenced_samplesheet_file(
     return csv_upload.file_key
 
 
+async def _rewrite_bindflow_settings_asset_columns(
+    *, s3_input_key: str, repo_assets_path: str
+) -> str:
+    """Fill in settings_filters/settings_advanced samplesheet columns with the
+    local Gadi path to bindflow's bundled default JSON files - the frontend
+    leaves these fields unset (see sbp-portal's de-novo-design.ts), so
+    resolve_bindflow_asset_path fills in the known default; any other,
+    genuinely custom value is left untouched.
+
+    Unlike starting_pdb (_stage_referenced_samplesheet_file, above) these
+    columns don't need their own Globus transfer or RunInput/DataTransfer
+    bookkeeping - they reference files that are already part of the workflow
+    repo, staged as a whole. This is a plain string rewrite, re-uploaded only
+    if something actually changed.
+    """
+    try:
+        samplesheet_rows = await read_csv_from_s3(s3_input_key)
+    except S3ConfigurationError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"S3 configuration error: {exc}",
+        ) from exc
+    except S3ServiceError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"Failed to read samplesheet at s3InputKey: {exc}",
+        ) from exc
+    if not samplesheet_rows:
+        raise HTTPException(
+            status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
+            detail="Samplesheet at s3InputKey is empty.",
+        )
+    samplesheet_row = samplesheet_rows[0]
+    changed = False
+    for field_name in ("settings_filters", "settings_advanced"):
+        resolved = resolve_bindflow_asset_path(
+            field_name, samplesheet_row.get(field_name), repo_assets_path=repo_assets_path
+        )
+        if resolved is not None and resolved != samplesheet_row.get(field_name):
+            samplesheet_row[field_name] = resolved
+            changed = True
+    if not changed:
+        return s3_input_key
+    try:
+        csv_upload = await upload_csv_to_s3(samplesheet_row)
+    except S3ConfigurationError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"S3 configuration error: {exc}",
+        ) from exc
+    except S3ServiceError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"Failed to re-upload corrected samplesheet: {exc}",
+        ) from exc
+    return csv_upload.file_key
+
+
 def _stage_wisps_fasta(
     *,
     db_session: Session,
@@ -417,6 +476,28 @@ async def launch_workflow(
             detail=f"Workflow '{workflow.name}' is missing default_revision in workflows table.",
         )
 
+    # Gadi compute nodes have no network access, so Nextflow can't fetch the
+    # pipeline from GitHub itself - it must already be staged there
+    try:
+        repo_staging_locations = ensure_repo_staging_requested(
+            db_session, workflow, settings=settings
+        )
+    except RepoStagingError as exc:
+        raise HTTPException(
+            status_code=status.HTTP_502_BAD_GATEWAY,
+            detail=f"Failed to resolve workflow repo: {exc}",
+        ) from exc
+    # Seqera's launch API validates `pipeline` as a URL, Nextflow's local-path scheme is "file:" with a
+    # single slash before the (already-absolute) path
+    repo_gadi_path = repo_staging_locations.gadi_path
+    pipeline_url = f"file:{repo_gadi_path}"
+    # bindcraft's settings_filters/settings_advanced reference files bundled
+    # inside the workflow repo itself - repo_gadi_path is a bare git repo
+    # (see build_repo_gadi_path) with no working-tree files on disk, so asset
+    # resolution uses the separate plain checkout staged alongside it instead
+    # (see build_repo_assets_gadi_path, _rewrite_bindflow_settings_asset_columns).
+    repo_assets_path = repo_staging_locations.assets_gadi_path
+
     user = db_session.execute(
         select(AppUser.email).where(AppUser.id == current_user_id)
     ).one_or_none()
@@ -518,6 +599,9 @@ async def launch_workflow(
             workflow_name=workflow_name,
             globus_settings=settings.globus,
         )
+        s3_input_key = await _rewrite_bindflow_settings_asset_columns(
+            s3_input_key=s3_input_key, repo_assets_path=repo_assets_path
+        )
     elif is_proteinfold_launch:
         s3_input_key = await _stage_referenced_samplesheet_file(
             db_session=db_session,
@@ -583,7 +667,7 @@ async def launch_workflow(
                 settings=settings,
                 db_session=db_session,
                 workflow_run=workflow_run,
-                pipeline=workflow.repo_url,
+                pipeline=pipeline_url,
                 config_path=workflow.config_path,
                 revision=workflow.default_revision,
                 output_id=str(run_id),
@@ -603,7 +687,7 @@ async def launch_workflow(
                     settings=settings,
                     db_session=db_session,
                     workflow_run=workflow_run,
-                    pipeline=workflow.repo_url,
+                    pipeline=pipeline_url,
                     config_path=workflow.config_path,
                     revision=workflow.default_revision,
                     output_id=str(run_id),
@@ -617,12 +701,14 @@ async def launch_workflow(
                     settings=settings,
                     db_session=db_session,
                     workflow_run=workflow_run,
-                    pipeline=workflow.repo_url,
+                    pipeline=pipeline_url,
                     config_path=workflow.config_path,
                     revision=workflow.default_revision,
                     output_id=str(run_id),
+                    form_data=payload.formData,
                     user_details=user_details,
                     staged_input_location=staged_input_location,
+                    repo_assets_path=repo_assets_path,
                 )
         elif workflow_name in ("interaction-screening", "bulk-prediction"):
             assert wisps_form_data is not None
@@ -633,7 +719,7 @@ async def launch_workflow(
                 settings=settings,
                 db_session=db_session,
                 workflow_run=workflow_run,
-                pipeline=workflow.repo_url,
+                pipeline=pipeline_url,
                 revision=workflow.default_revision,
                 config_path=workflow.config_path,
                 form_data=wisps_form_data,

app/run_scheduler.py

@@ -18,11 +18,13 @@
     submit_pending_jobs,
     sync_completed_workflow_runs,
     sync_data_transfers,
+    sync_workflow_repo_staging,
 )
 
 SUBMIT_INTERVAL = IntervalTrigger(minutes=5)
 SYNC_INTERVAL = IntervalTrigger(minutes=10)
 DATA_TRANSFER_SYNC_INTERVAL = IntervalTrigger(minutes=2)
+REPO_STAGING_SYNC_INTERVAL = IntervalTrigger(minutes=2)
 # Fixed AEST (UTC+10), no DST. Not Australia/Sydney: APScheduler 3.11.3's CronTrigger
 # miscalculates day=1 across Sydney's October DST switch and skips November entirely.
 MONTHLY_TRIGGER = CronTrigger(day=1, hour=0, minute=0, timezone="Australia/Brisbane")
@@ -68,6 +70,20 @@ def main(dry_run: bool = False):
             max_instances=1,
             replace_existing=True,
         )
+        logger.info(
+            f"Adding sync_workflow_repo_staging to scheduler: trigger = {REPO_STAGING_SYNC_INTERVAL}"
+        )
+        SCHEDULER.add_job(
+            sync_workflow_repo_staging,
+            kwargs={"dry_run": dry_run},
+            jobstore="memory",
+            trigger=REPO_STAGING_SYNC_INTERVAL,
+            next_run_time=datetime.now(tz=UTC) + timedelta(minutes=1),
+            id="sync_workflow_repo_staging",
+            misfire_grace_time=60,
+            max_instances=1,
+            replace_existing=True,
+        )
         logger.info(
             f"Adding monthly refresh_user_credits to scheduler: trigger = {MONTHLY_TRIGGER}"
         )