feat(ui): wire Export/Import job cards to tenant-scoped counts (#554)

crates/observability/src/dashboard.rs

@@ -130,6 +130,18 @@ pub struct DashboardSeries {
     pub points: Vec<DashboardPoint>,
 }
 
+/// Bulk-export jobs for one tenant, split by lifecycle stage.
+///
+/// Carried by [`DashboardSnapshot::export_jobs`]; both figures come from the
+/// same storage read so they are always consistent with each other.
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+pub struct ExportJobCounts {
+    /// Jobs a worker has claimed and is executing (`in-progress`).
+    pub running: u64,
+    /// Jobs accepted and waiting for a worker slot (`accepted`).
+    pub queued: u64,
+}
+
 /// A snapshot of the figures the dashboard renders. Plain data — no storage or
 /// FHIR types — so this crate stays dependency-light.
 #[derive(Clone, Debug, Default)]
@@ -144,6 +156,15 @@ pub struct DashboardSnapshot {
     pub window: DashboardWindow,
     /// Per-type series for the charted resource types, in display order.
     pub series: Vec<DashboardSeries>,
+    /// Bulk-export jobs for the tenant.
+    ///
+    /// `None` when the running storage backend has no bulk-export job store,
+    /// the subsystem is disabled, or the count could not be read — the UI
+    /// renders an explicit "unavailable" state instead of a fabricated zero.
+    pub export_jobs: Option<ExportJobCounts>,
+    /// Non-terminal bulk-submit (import) jobs for the tenant. `None` under the
+    /// same conditions as [`Self::export_jobs`].
+    pub import_jobs_active: Option<u64>,
 }
 
 /// Supplies [`DashboardSnapshot`]s on demand. Implemented in `helios-rest` over
@@ -470,6 +491,8 @@ mod tests {
                         cumulative: 7,
                     }],
                 }],
+                export_jobs: None,
+                import_jobs_active: None,
             }
         }
     }

crates/persistence/src/backends/postgres/bulk_export.rs

@@ -461,6 +461,25 @@ impl BulkExportStorage for PostgresBackend {
         Ok(count as u64)
     }
 
+    async fn count_exports_by_status(
+        &self,
+        tenant: &TenantContext,
+        status: ExportStatus,
+    ) -> StorageResult<u64> {
+        let client = self.get_client().await?;
+        let tenant_id = tenant.tenant_id().as_str();
+        let status = status.to_string();
+        let row = client
+            .query_one(
+                "SELECT COUNT(*) FROM bulk_export_jobs WHERE tenant_id = $1 AND status = $2",
+                &[&tenant_id, &status],
+            )
+            .await
+            .map_err(|e| internal_error(format!("Failed to count exports by status: {e}")))?;
+        let count: i64 = row.get(0);
+        Ok(count as u64)
+    }
+
     async fn list_expired_exports(
         &self,
         now: DateTime<Utc>,

crates/persistence/src/backends/sqlite/bulk_export.rs

@@ -563,6 +563,23 @@ impl BulkExportStorage for SqliteBackend {
         Ok(count as u64)
     }
 
+    async fn count_exports_by_status(
+        &self,
+        tenant: &TenantContext,
+        status: ExportStatus,
+    ) -> StorageResult<u64> {
+        let conn = self.get_connection()?;
+        let tenant_id = tenant.tenant_id().as_str();
+        let count: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM bulk_export_jobs WHERE tenant_id = ?1 AND status = ?2",
+                params![tenant_id, status.to_string()],
+                |row| row.get(0),
+            )
+            .map_err(|e| internal_error(format!("Failed to count exports by status: {e}")))?;
+        Ok(count as u64)
+    }
+
     async fn list_expired_exports(
         &self,
         now: DateTime<Utc>,
@@ -1643,6 +1660,69 @@ mod tests {
         assert_eq!(backend.count_active_exports(&tenant).await.unwrap(), 3);
     }
 
+    #[tokio::test]
+    async fn count_exports_by_status_splits_accepted_and_in_progress_per_tenant() {
+        let backend = create_test_backend();
+        let tenant_a = create_test_tenant();
+        let tenant_b = TenantContext::new(
+            TenantId::new("other-tenant"),
+            TenantPermissions::full_access(),
+        );
+
+        backend
+            .start_export(&tenant_a, test_input(ExportRequest::system()))
+            .await
+            .unwrap();
+        backend
+            .start_export(&tenant_a, test_input(ExportRequest::system()))
+            .await
+            .unwrap();
+
+        // Move one of tenant A's jobs to in-progress via the real worker path.
+        let worker = WorkerId::new("worker-1");
+        let lease = backend
+            .claim_next(&worker, StdDuration::from_secs(60))
+            .await
+            .unwrap()
+            .expect("a job should be claimable");
+        backend
+            .mark_export_in_progress(&tenant_a, &lease.job_id, &worker, lease.fencing_token)
+            .await
+            .unwrap();
+
+        assert_eq!(
+            backend
+                .count_exports_by_status(&tenant_a, ExportStatus::Accepted)
+                .await
+                .unwrap(),
+            1
+        );
+        assert_eq!(
+            backend
+                .count_exports_by_status(&tenant_a, ExportStatus::InProgress)
+                .await
+                .unwrap(),
+            1
+        );
+        assert_eq!(
+            backend
+                .count_exports_by_status(&tenant_a, ExportStatus::Complete)
+                .await
+                .unwrap(),
+            0
+        );
+        assert_eq!(
+            backend
+                .count_exports_by_status(&tenant_b, ExportStatus::Accepted)
+                .await
+                .unwrap(),
+            0
+        );
+
+        // The concurrency-cap aggregate is unaffected by the new method.
+        assert_eq!(backend.count_active_exports(&tenant_a).await.unwrap(), 2);
+    }
+
     #[tokio::test]
     async fn test_get_export_job_metadata() {
         let backend = create_test_backend();

crates/persistence/src/core/bulk_export.rs

@@ -994,6 +994,17 @@ pub trait BulkExportStorage: Send + Sync {
     /// enforce the per-tenant concurrency cap at kickoff.
     async fn count_active_exports(&self, tenant: &TenantContext) -> StorageResult<u64>;
 
+    /// Counts the tenant's export jobs currently in `status`.
+    ///
+    /// Used by the dashboard to show the running (`in-progress`) / queued
+    /// (`accepted`) split. The per-tenant concurrency cap keeps using
+    /// [`count_active_exports`](Self::count_active_exports).
+    async fn count_exports_by_status(
+        &self,
+        tenant: &TenantContext,
+        status: ExportStatus,
+    ) -> StorageResult<u64>;
+
     /// Lists expired completed jobs across *all* tenants, for the cleanup task.
     ///
     /// This is intentionally cross-tenant — the cleanup task is a server-wide

crates/persistence/tests/postgres_tests.rs

@@ -4887,6 +4887,13 @@ mod postgres_integration {
                 .unwrap();
         }
         assert_eq!(backend.count_active_exports(&tenant).await.unwrap(), 2);
+        assert_eq!(
+            backend
+                .count_exports_by_status(&tenant, ExportStatus::Accepted)
+                .await
+                .unwrap(),
+            2
+        );
 
         // Nothing is expired yet.
         let expired_now = backend