perf(workflow): push workflow_ref (type) filter to the fast DB-page path

Filtering the /workflows list by type was slow: workflow_query_can_use_db_page
excluded workflow_ref, so a filtered query fell back to list_all() — fetching
every run with full blobs and filtering in memory (~6s). The journal backend's
list/query_ids already support a workflow_ref filter, so thread it through
list_page + query_ids and let workflow_ref stay on the bounded DB-page path.
Filtered lists are now server-side bounded + accurate-total, same as unfiltered.
Folded into rc.20.

Author: Shooksie

crates/orchestrator-core/src/services/workflow_impl.rs

@@ -149,9 +149,11 @@ fn query_workflows(workflows: Vec<OrchestratorWorkflow>, query: &WorkflowQuery)
 }
 
 fn workflow_query_can_use_db_page(query: &WorkflowQuery) -> bool {
+    // `workflow_ref` (the type filter) is pushed down to the journal backend's
+    // bounded list/query_ids, so it stays on the fast DB-page path; task_id /
+    // phase_id / search still fall back to the in-memory scan.
     query.page.limit.is_some()
         && matches!(query.sort, WorkflowQuerySort::StartedAt)
-        && query.filter.workflow_ref.is_none()
         && query.filter.task_id.is_none()
         && query.filter.phase_id.is_none()
         && query.filter.search_text.as_deref().is_none_or(|value| value.trim().is_empty())
@@ -398,15 +400,16 @@ impl WorkflowServiceApi for FileServiceHub {
             // query for `total`, instead of query_ids + a per-id load loop (N+1 —
             // one RPC per run, which serialized behind other journal traffic and
             // made the list intermittently multi-second).
+            let workflow_ref = query.filter.workflow_ref.as_deref();
             if query.page.offset == 0 {
                 if let Some(limit) = query.page.limit {
-                    let (_ids, total) = manager.query_ids(query.page, query.filter.status)?;
-                    let items = manager.list_page(query.filter.status, limit)?;
+                    let (_ids, total) = manager.query_ids(query.page, query.filter.status, workflow_ref)?;
+                    let items = manager.list_page(query.filter.status, workflow_ref, limit)?;
                     return Ok(ListPage::new(items, total, query.page));
                 }
             }
             // Offset > 0: the journal list RPC has no offset, so keep the id-walk.
-            let (ids, total) = manager.query_ids(query.page, query.filter.status)?;
+            let (ids, total) = manager.query_ids(query.page, query.filter.status, workflow_ref)?;
             let mut items = Vec::with_capacity(ids.len());
             for id in ids {
                 if let Ok(workflow) = manager.load(&id) {

crates/orchestrator-core/src/workflow/journal_client.rs

@@ -477,11 +477,12 @@ pub(crate) fn list_page(
     plugin: &DiscoveredPlugin,
     project_root: &Path,
     status: Option<WorkflowStatus>,
+    workflow_ref: Option<&str>,
     limit: usize,
 ) -> Result<Vec<OrchestratorWorkflow>> {
     let query = JournalQuery {
         status: status.map(|s| vec![status_wire(s).to_string()]).unwrap_or_default(),
-        workflow_ref: None,
+        workflow_ref: workflow_ref.map(str::to_string),
         updated_since: None,
         limit: Some(limit as u32),
     };
@@ -593,10 +594,11 @@ pub(crate) fn query_ids(
     plugin: &DiscoveredPlugin,
     project_root: &Path,
     status: Option<WorkflowStatus>,
+    workflow_ref: Option<&str>,
 ) -> Result<Vec<String>> {
     let query = JournalQuery {
         status: status.map(|s| vec![status_wire(s).to_string()]).unwrap_or_default(),
-        workflow_ref: None,
+        workflow_ref: workflow_ref.map(str::to_string),
         updated_since: None,
         limit: Some(QUERY_IDS_LIMIT),
     };

crates/orchestrator-core/src/workflow/state_manager.rs

@@ -368,13 +368,17 @@ impl WorkflowStateManager {
     pub fn list_page(
         &self,
         status: Option<crate::types::WorkflowStatus>,
+        workflow_ref: Option<&str>,
         limit: usize,
     ) -> Result<Vec<OrchestratorWorkflow>> {
         if let Some(plugin) = self.journal_plugin() {
-            return super::journal_client::list_page(plugin, &self.project_root, status, limit);
+            return super::journal_client::list_page(plugin, &self.project_root, status, workflow_ref, limit);
         }
-        // SQLite dev fallback: fetch all and truncate (no plugin round-trips).
+        // SQLite dev fallback: fetch all, filter by ref, truncate (no plugin round-trips).
         let mut all = self.list_all()?;
+        if let Some(wref) = workflow_ref {
+            all.retain(|w| w.workflow_ref.as_deref() == Some(wref));
+        }
         all.truncate(limit);
         Ok(all)
     }
@@ -476,17 +480,31 @@ impl WorkflowStateManager {
         &self,
         page: ListPageRequest,
         status: Option<crate::types::WorkflowStatus>,
+        workflow_ref: Option<&str>,
     ) -> Result<(Vec<String>, usize)> {
         if let Some(plugin) = self.journal_plugin() {
             // The journal protocol's query_ids has no offset/total, so fetch the
-            // full matching id set from the plugin and paginate client-side to
-            // preserve the (page, total) contract.
-            let all_ids = super::journal_client::query_ids(plugin, &self.project_root, status)?;
+            // full matching id set from the plugin (status + workflow_ref filtered
+            // server-side) and paginate client-side to preserve the (page, total)
+            // contract.
+            let all_ids = super::journal_client::query_ids(plugin, &self.project_root, status, workflow_ref)?;
             let total = all_ids.len();
             let (start, end) = page.bounds(total);
             let ids = all_ids.into_iter().skip(start).take(end.saturating_sub(start)).collect();
             return Ok((ids, total));
         }
+        // SQLite dev fallback: a workflow_ref filter is applied in memory (small
+        // local store) to avoid a combinatorial SQL branch; the plugin path above
+        // is the production one.
+        if let Some(wref) = workflow_ref {
+            let mut runs = self.list_all()?;
+            runs.retain(|w| status.is_none_or(|s| w.status == s) && w.workflow_ref.as_deref() == Some(wref));
+            runs.sort_by(|a, b| b.started_at.cmp(&a.started_at).then_with(|| a.id.cmp(&b.id)));
+            let total = runs.len();
+            let (start, end) = page.bounds(total);
+            let ids = runs[start..end].iter().map(|w| w.id.clone()).collect();
+            return Ok((ids, total));
+        }
         let conn = self.open_db()?;
 
         let total: usize = match status {