perf(workflow): batch the workflow-list first page into one journal RPC (kill N+1) (#325)

The DB-page workflow-list path did query_ids + a per-id `load` loop — one journal
RPC per run (N+1). Bounded to 200 (via the graphql default limit) that's still
201 serialized RPCs on the journal plugin's stdio host, which queued behind other
traffic and made /workflows intermittently multi-second.

Add journal_client::list_page + WorkflowStateManager::list_page: fetch the `limit`
newest runs in ONE bounded journal/list RPC (the plugin already supports the
limit). The query() first-page path (offset 0) now does one list RPC + a cheap
ids-only count for `total`, instead of the load loop. Offset>0 keeps the id-walk
(the list RPC has no offset). Additive; list_all/query_ids unchanged.

cargo check/fmt/clippy clean; orchestrator-core tests 160 passed.

Author: Shooksie

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

@@ -369,6 +369,19 @@ impl WorkflowServiceApi for FileServiceHub {
     async fn query(&self, query: WorkflowQuery) -> Result<ListPage<OrchestratorWorkflow>> {
         if workflow_query_can_use_db_page(&query) {
             let manager = self.workflow_manager();
+            // Fast path for the first page (the workflow-list UI + graphql poll):
+            // fetch the bounded runs in ONE journal RPC, plus a cheap ids-only
+            // 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).
+            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)?;
+                    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 mut items = Vec::with_capacity(ids.len());
             for id in ids {

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

@@ -469,6 +469,27 @@ pub(crate) fn list(
     Ok(resp.runs.into_iter().filter_map(|run| from_journal_run(run).ok()).collect())
 }
 
+/// Like [`list`] but bounded to the `limit` newest rows in a SINGLE RPC — the
+/// fast path for the workflow-list UI. Replaces `query_ids` + a per-id `load`
+/// loop (N+1 — one RPC per run), which serialized behind other traffic on the
+/// journal plugin's stdio host and made the list intermittently slow.
+pub(crate) fn list_page(
+    plugin: &DiscoveredPlugin,
+    project_root: &Path,
+    status: Option<WorkflowStatus>,
+    limit: usize,
+) -> Result<Vec<OrchestratorWorkflow>> {
+    let query = JournalQuery {
+        status: status.map(|s| vec![status_wire(s).to_string()]).unwrap_or_default(),
+        workflow_ref: None,
+        updated_since: None,
+        limit: Some(limit as u32),
+    };
+    let value = run_blocking(call(plugin, project_root, METHOD_JOURNAL_LIST, query))??;
+    let resp: ListResult = serde_json::from_value(value).context("decoding journal ListResult")?;
+    Ok(resp.runs.into_iter().filter_map(|run| from_journal_run(run).ok()).collect())
+}
+
 /// All run ids matching `status` (None = all). The caller paginates client-side.
 pub(crate) fn query_ids(
     plugin: &DiscoveredPlugin,

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

@@ -362,6 +362,23 @@ impl WorkflowStateManager {
         Ok(workflows)
     }
 
+    /// The `limit` newest runs (optionally status-filtered) in a SINGLE journal
+    /// RPC — the fast path for the workflow-list first page (avoids the per-id
+    /// load loop / N+1). Backend order (newest-first) mirrors `query_ids`.
+    pub fn list_page(
+        &self,
+        status: Option<crate::types::WorkflowStatus>,
+        limit: usize,
+    ) -> Result<Vec<OrchestratorWorkflow>> {
+        if let Some(plugin) = self.journal_plugin() {
+            return super::journal_client::list_page(plugin, &self.project_root, status, limit);
+        }
+        // SQLite dev fallback: fetch all and truncate (no plugin round-trips).
+        let mut all = self.list_all()?;
+        all.truncate(limit);
+        Ok(all)
+    }
+
     pub fn list_all(&self) -> Result<Vec<OrchestratorWorkflow>> {
         if let Some(plugin) = self.journal_plugin() {
             return super::journal_client::list(plugin, &self.project_root, &[]);