chore: update staging with latest dev

app/db/admin.py

@@ -10,6 +10,7 @@
 import hmac
 import io
 import json
+import logging
 import os
 import secrets
 from datetime import UTC, datetime
@@ -36,8 +37,10 @@
 from ..auth.validator import fetch_userinfo_claims, verify_access_token_claims
 from ..config import Settings, get_settings
 from ..routes.dependencies import get_db
+from ..schemas.workflows.shared import PipelineStatus
 from ..services.credits import launch_credit_cost
 from ..services.globus_transfer import reset_failed_output_transfers
+from ..services.job_sync import force_resync_run_outputs
 from . import engine
 from .models import job_queue
 from .models.core import (
@@ -51,10 +54,18 @@
     WorkflowRun,
 )
 
+logger = logging.getLogger(__name__)
+
 _ADMIN_TEMPLATES_DIR = os.path.join(
     os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "templates"
 )
 
+# The built-in displays/relation.html builds its href from the foreign model's
+# raw pk, which 500s ("May not contain path separators") for a pk containing
+# "/" (e.g. an S3 object key). Use for any field pointing at
+# S3ObjectAdmin/RunInputAdmin/RunOutputAdmin.
+_SAFE_RELATION_TEMPLATE = "displays/safe_relation.html"
+
 DEFAULT_DB_ADMIN_REQUIRED_ROLE = "biocommons/role/sbp/admin"
 DEFAULT_DB_ADMIN_ROLES_CLAIM = "https://biocommons.org.au/roles"
 DEFAULT_DB_ADMIN_SESSION_COOKIE = "sbp_admin_session"
@@ -289,6 +300,91 @@ def get_search_query(self, request: Request, term: str) -> Any:
         )
         return or_(base_clause, owner_clause)
 
+    @action(
+        name="force_resync_outputs",
+        text="Force resync outputs",
+        confirmation=(
+            "Re-scan S3 and resync result outputs for the selected completed "
+            "runs, even though they're already marked as synced? Submits a "
+            "new output transfer first for any prefix the current spec "
+            "requires but hasn't been transferred yet. Use this after a "
+            "results-utils change adds a new output category or classifier, "
+            "so existing runs pick up newly recognised files."
+        ),
+        submit_btn_text="Force resync",
+        submit_btn_class="btn-warning",
+        icon_class="fa-solid fa-arrows-rotate",
+    )
+    async def force_resync_outputs_action(self, request: Request, pks: list[Any]) -> str:
+        runs = await self.find_by_pks(request, pks)
+        return await _force_resync_workflow_runs(request.state.session, runs)
+
+    @row_action(
+        name="force_resync_outputs",
+        text="Force resync outputs",
+        confirmation=(
+            "Re-scan S3 and resync result outputs for this run, even though "
+            "it's already marked as synced? Submits a new output transfer "
+            "first if the current spec requires a prefix that hasn't been "
+            "transferred yet."
+        ),
+        submit_btn_text="Force resync",
+        submit_btn_class="btn-warning",
+        action_btn_class="btn-warning",
+        icon_class="fa-solid fa-arrows-rotate",
+    )
+    async def row_action_force_resync_outputs(self, request: Request, pk: Any) -> str:
+        run = await self.find_by_pk(request, pk)
+        if run is None:
+            raise ActionFailed("Workflow run not found.")
+        return await _force_resync_workflow_runs(request.state.session, [run])
+
+
+async def _force_resync_workflow_runs(db: Session, runs: list[WorkflowRun]) -> str:
+    """Force-resync result outputs for the given completed workflow runs.
+
+    See force_resync_run_outputs: bypasses the usual "already synced"
+    short-circuit and submits any output transfer still missing, so runs
+    pick up categories/files results-utils didn't recognise before.
+    """
+    processed = 0
+    submitted = 0
+    skipped = 0
+    errored = 0
+    outputs_synced = 0
+
+    for run in runs:
+        if run.seqera_final_status != PipelineStatus.SUCCEEDED.value:
+            skipped += 1
+            continue
+        try:
+            outcome = await force_resync_run_outputs(db, run)
+        except Exception:
+            db.rollback()
+            logger.exception("Force resync of outputs failed for workflow run %s", run.id)
+            errored += 1
+            continue
+
+        if outcome.ready:
+            processed += 1
+            outputs_synced += outcome.outputs_synced
+        else:
+            submitted += 1
+
+    if processed == 0 and submitted == 0 and errored == 0:
+        raise ActionFailed(
+            "No completed runs were selected - only SUCCEEDED runs can be force-resynced."
+        )
+
+    summary = [f"{processed} run(s) resynced ({outputs_synced} output(s) found)"]
+    if submitted:
+        summary.append(f"{submitted} submitted new output transfer(s), not yet ready to resync")
+    if skipped:
+        summary.append(f"{skipped} skipped (not completed)")
+    if errored:
+        summary.append(f"{errored} errored")
+    return ", ".join(summary) + "."
+
 
 def _build_workflow_runs_csv(session: Session) -> str:
     """Build a CSV dump of every workflow run row.
@@ -472,8 +568,8 @@ class S3ObjectAdmin(UrlSafePrimaryKeyModelView):
     fields = [
         "object_key",
         "uri",
-        HasMany("run_inputs", identity="run-input"),
-        HasMany("run_outputs", identity="run-output"),
+        HasMany("run_inputs", identity="run-input", display_template=_SAFE_RELATION_TEMPLATE),
+        HasMany("run_outputs", identity="run-output", display_template=_SAFE_RELATION_TEMPLATE),
         "version_id",
         "size_bytes",
     ]
@@ -485,16 +581,26 @@ async def repr(self, obj: Any, request: Request) -> str:
 class RunInputAdmin(UrlSafePrimaryKeyModelView):
     fields = [
         HasOne("run", identity="workflow-run"),
-        HasOne("s3_object", identity="s3-object"),
+        HasOne("s3_object", identity="s3-object", display_template=_SAFE_RELATION_TEMPLATE),
     ]
 
 
 class RunOutputAdmin(UrlSafePrimaryKeyModelView):
     fields = [
         HasOne("run", identity="workflow-run"),
-        HasOne("s3_object", identity="s3-object"),
+        HasOne("s3_object", identity="s3-object", display_template=_SAFE_RELATION_TEMPLATE),
     ]
 
+    def get_search_query(self, request: Request, term: str) -> Any:
+        # `run` is a relationship, not a column, so the default search only
+        # matches the raw run_id UUID. Add job name via a correlated EXISTS.
+        base_clause = super().get_search_query(request, term)
+        run_name_clause = exists().where(
+            WorkflowRun.id == RunOutput.run_id,
+            WorkflowRun.run_name.ilike(f"%{term}%"),
+        )
+        return or_(base_clause, run_name_clause)
+
 
 class DataTransferAdmin(ModelView):
     fields = [

app/routes/workflow/jobs.py

@@ -26,6 +26,8 @@
     JobDetailsResponse,
     JobListItem,
     JobListResponse,
+    PipelineStatus,
+    UIStatus,
     map_pipeline_status_to_ui,
 )
 from ...services.job_utils import (
@@ -80,12 +82,25 @@ def _resolve_stored_score(owned_run: WorkflowRun | None) -> float | None:
 
 
 def _get_stored_terminal_ui_status(run: WorkflowRun) -> str | None:
+    """Resolve UI status from stored columns alone, no live Seqera call needed.
+
+    A run isn't "Completed" just because Seqera succeeded - it's only Completed
+    once output transfers have synced too (see `WorkflowRun.results_sync_status`).
+    Still syncing -> In progress; a permanently failed output transfer -> Failed.
+    """
     if run.seqera_final_status is None:
         return None
     status = run.seqera_final_status.strip().upper()
     if status not in TERMINAL_SEQERA_STATUSES:
         return None
-    return map_pipeline_status_to_ui(status)
+    if status != PipelineStatus.SUCCEEDED.value:
+        return map_pipeline_status_to_ui(status)
+    sync_status = run.results_sync_status
+    if sync_status == "syncing":
+        return UIStatus.IN_PROGRESS.value
+    if sync_status == "partial":
+        return UIStatus.FAILED.value
+    return UIStatus.COMPLETED.value
 
 
 @router.post("/{run_id}/cancel", response_model=CancelWorkflowResponse)