feat(runner): publish the project catalog before Git enrichment

crates/webcodex-runner/src/webcodex_runner/projects.rs

@@ -24,6 +24,7 @@ mod lifecycle;
 mod managed_worktree;
 mod registration;
 
+pub(crate) use catalog::load_runner_project_metadata;
 pub(crate) use catalog::{
     find_project_shell_context, find_project_shell_context_by_id,
     load_runner_project_summaries_from_dir, project_root_fingerprint,

crates/webcodex-runner/src/webcodex_runner/projects/catalog.rs

@@ -674,6 +674,7 @@ fn load_runner_project_with_incarnation(file: &Path) -> Result<RunnerProjectFile
 
 fn load_runner_project_summaries_from_dir_with_shutdown(
     dir: &Path,
+    include_git: bool,
     shutdown: Option<&AtomicBool>,
 ) -> Vec<RunnerProjectSummary> {
     let entries = match std::fs::read_dir(dir) {
@@ -725,20 +726,40 @@ fn load_runner_project_summaries_from_dir_with_shutdown(
         }
         warn_empty_hook_commands(&file, &project);
         projects.push(runner_project_summary_with_shutdown(
-            &project, updated_at, true, shutdown,
+            &project,
+            updated_at,
+            include_git,
+            shutdown,
         ));
     }
     projects.sort_by(|a, b| a.id.cmp(&b.id));
     projects
 }
 
 pub(crate) fn load_runner_project_summaries_from_dir(dir: &Path) -> Vec<RunnerProjectSummary> {
-    load_runner_project_summaries_from_dir_with_shutdown(dir, None)
+    load_runner_project_summaries_from_dir_with_shutdown(dir, true, None)
+}
+
+/// The validated registration catalog without any Git probe. Absent Git fields
+/// mean unknown or pending, never a clean or empty repository.
+pub(crate) fn load_runner_project_metadata(
+    cfg: &RunnerConfig,
+    shutdown: Option<&AtomicBool>,
+) -> Vec<RunnerProjectSummary> {
+    load_runner_project_catalog(cfg, false, shutdown)
 }
 
 fn load_runner_project_summaries(
     cfg: &RunnerConfig,
     shutdown: Option<&AtomicBool>,
+) -> Vec<RunnerProjectSummary> {
+    load_runner_project_catalog(cfg, true, shutdown)
+}
+
+fn load_runner_project_catalog(
+    cfg: &RunnerConfig,
+    include_git: bool,
+    shutdown: Option<&AtomicBool>,
 ) -> Vec<RunnerProjectSummary> {
     // Loaded configs always carry a materialized project_registry_dir; a bare
     // test-built config that cannot derive one reports the error instead of
@@ -750,7 +771,7 @@ fn load_runner_project_summaries(
             return Vec::new();
         }
     };
-    load_runner_project_summaries_from_dir_with_shutdown(&dir, shutdown)
+    load_runner_project_summaries_from_dir_with_shutdown(&dir, include_git, shutdown)
 }
 
 impl RunnerProjectCache {

crates/webcodex-runner/src/webcodex_runner/transport.rs

@@ -54,7 +54,7 @@ use project_inventory::{
 use project_inventory::{
     log_project_inventory_degraded, paged_sync_after_registration, polling_projects_for_poll,
     InitialProjectInventory, InitialProjectScanState, PollingProjectRefresh, ProjectInventorySync,
-    ProjectScanReceiver, StreamingProjectInventoryCoordinator,
+    ProjectScanPublication, ProjectScanReceiver, StreamingProjectInventoryCoordinator,
 };
 use registration::{build_register_request_with_provider_status, register, RegisterRecoveryAction};
 use reqwest::blocking::Client;

crates/webcodex-runner/src/webcodex_runner/transport/project_inventory.rs

@@ -37,14 +37,24 @@ pub(super) enum InitialProjectInventory {
     Disabled,
 }
 
-pub(super) type ProjectScanReceiver =
-    tokio::sync::watch::Receiver<Option<std::sync::Arc<Vec<RunnerProjectSummary>>>>;
+/// One stage of the initial scan. The registration catalog is published first
+/// with unknown Git fields, then replaced once when Git enrichment completes.
+#[derive(Clone)]
+pub(super) struct ProjectScanPublication {
+    pub(super) projects: std::sync::Arc<Vec<RunnerProjectSummary>>,
+    pub(super) complete: bool,
+    /// Inventory generation the catalog was read under. A later Project
+    /// mutation makes it stale even if a stream has not consumed it yet.
+    pub(super) generation: u64,
+}
+
+pub(super) type ProjectScanReceiver = tokio::sync::watch::Receiver<Option<ProjectScanPublication>>;
 
 /// Shared across transport sessions so a mutation fences an in-flight scan even
 /// after its original stream disconnects.
 pub(super) struct InitialProjectScanState {
     pub(super) receiver: Option<ProjectScanReceiver>,
-    pub(super) invalidated: bool,
+    pub(super) generation: u64,
     pub(super) changed: tokio::sync::watch::Sender<()>,
 }
 
@@ -53,7 +63,7 @@ impl Default for InitialProjectScanState {
         let (changed, _) = tokio::sync::watch::channel(());
         Self {
             receiver: None,
-            invalidated: false,
+            generation: 0,
             changed,
         }
     }
@@ -468,6 +478,33 @@ impl StreamingProjectInventoryCoordinator {
         }
         self.project_cache.invalidate();
         let projects = runtime.project_summaries(&mut self.project_cache, cfg);
+        self.replace_projects(projects, transport, out_tx, reason_code);
+    }
+
+    /// A mutation during Git enrichment must not leave the older catalog
+    /// visible until the worker finishes, and needs no Git probe to correct it.
+    pub(super) fn refresh_from_project_metadata(
+        &mut self,
+        transport: StreamTransport,
+        cfg: &RunnerConfig,
+        runtime: &RunnerRuntimeState,
+        out_tx: &tokio::sync::mpsc::Sender<RunnerEnvelope>,
+        reason_code: &str,
+    ) {
+        if !self.supported {
+            return;
+        }
+        let projects = runtime.project_metadata_summaries(cfg);
+        self.replace_projects(projects, transport, out_tx, reason_code);
+    }
+
+    fn replace_projects(
+        &mut self,
+        projects: Vec<RunnerProjectSummary>,
+        transport: StreamTransport,
+        out_tx: &tokio::sync::mpsc::Sender<RunnerEnvelope>,
+        reason_code: &str,
+    ) {
         let projects_count = projects.len();
         self.sync = Some(ProjectInventorySync::new(projects));
         self.retry_backoff.reset();

crates/webcodex-runner/src/webcodex_runner/transport/runtime.rs

@@ -2,6 +2,12 @@
 
 use super::*;
 
+enum ProjectScanPublishOutcome {
+    Published,
+    Stale,
+    Stopped,
+}
+
 impl RunnerRuntimeState {
     pub(crate) fn new(cfg: &RunnerConfig, path: PathBuf) -> Self {
         Self::with_shutdown_budget(cfg, path, DEFAULT_SHUTDOWN_BUDGET)
@@ -87,23 +93,36 @@ impl RunnerRuntimeState {
         cache.get_with_shutdown(cfg, Some(shutdown.as_ref()))
     }
 
-    /// Registration and keepalive must not wait for serial Git probes. Reuse
-    /// an in-flight scan across reconnects, so only one scanner can run per
-    /// process. The existing shutdown flag, Git tree ownership, and bounded
-    /// background-thread drain still own cancellation and cleanup.
+    pub(super) fn project_metadata_summaries(
+        &self,
+        cfg: &RunnerConfig,
+    ) -> Vec<RunnerProjectSummary> {
+        let shutdown = self.shutdown_flag();
+        crate::webcodex_runner::projects::load_runner_project_metadata(cfg, Some(shutdown.as_ref()))
+    }
+
+    /// Registration, keepalive and the first authorized catalog must not wait
+    /// for serial Git probes. Reuse an in-flight scan across reconnects, so
+    /// only one scanner can run per process. The existing shutdown flag, Git
+    /// tree ownership, and bounded background-thread drain still own
+    /// cancellation and cleanup.
     pub(super) fn initial_project_scan(
         &self,
         cfg: &RunnerConfig,
     ) -> Result<ProjectScanReceiver, String> {
+        let metadata_runtime = self.clone();
+        let metadata_cfg = cfg.clone();
         let runtime = self.clone();
         let cfg = cfg.clone();
-        self.initial_project_scan_with(move || {
-            runtime.project_summaries(&mut RunnerProjectCache::default(), &cfg)
-        })
+        self.initial_project_scan_with(
+            move || metadata_runtime.project_metadata_summaries(&metadata_cfg),
+            move || runtime.project_summaries(&mut RunnerProjectCache::default(), &cfg),
+        )
     }
 
     pub(super) fn initial_project_scan_with(
         &self,
+        mut scan_metadata: impl FnMut() -> Vec<RunnerProjectSummary> + Send + 'static,
         mut scan_projects: impl FnMut() -> Vec<RunnerProjectSummary> + Send + 'static,
     ) -> Result<ProjectScanReceiver, String> {
         let mut scan = self
@@ -114,12 +133,15 @@ impl RunnerRuntimeState {
             return Err("project_inventory_scan_stopped".to_string());
         }
         if let Some(receiver) = scan.receiver.as_ref() {
-            if receiver.borrow().is_none() && receiver.has_changed().is_ok() {
+            let complete = receiver
+                .borrow()
+                .as_ref()
+                .is_some_and(|publication| publication.complete);
+            if !complete && receiver.has_changed().is_ok() {
                 return Ok(receiver.clone());
             }
         }
         self.background_threads.reap_finished();
-        scan.invalidated = false;
         let (sender, receiver) = tokio::sync::watch::channel(None);
         let runtime = self.clone();
         let handle = std::thread::Builder::new()
@@ -128,31 +150,87 @@ impl RunnerRuntimeState {
                 if runtime.shutdown_requested() {
                     break;
                 }
-                let projects = scan_projects();
-                let mut scan = runtime
-                    .initial_project_scan
-                    .lock()
-                    .unwrap_or_else(|e| e.into_inner());
-                if runtime.shutdown_requested() {
-                    break;
+                let generation = runtime.project_inventory_generation();
+                let metadata = scan_metadata();
+                match runtime.publish_project_scan(&sender, generation, metadata, false) {
+                    ProjectScanPublishOutcome::Published => {}
+                    ProjectScanPublishOutcome::Stale => continue,
+                    ProjectScanPublishOutcome::Stopped => break,
                 }
-                if scan.invalidated {
-                    // Keep the one owned worker, but discard a catalog observed
-                    // before a Project mutation. Reconnects share this receiver.
-                    scan.invalidated = false;
-                    continue;
+                let projects = scan_projects();
+                match runtime.publish_project_scan(&sender, generation, projects, true) {
+                    ProjectScanPublishOutcome::Stale => continue,
+                    ProjectScanPublishOutcome::Published | ProjectScanPublishOutcome::Stopped => {
+                        break
+                    }
                 }
-                // Serialize publication with invalidation so an obsolete result
-                // cannot become the pending scan of a later connection.
-                let _ = sender.send(Some(Arc::new(projects)));
-                break;
             })
             .map_err(|_| "project_inventory_scan_start_failed".to_string())?;
         self.register_background_thread(handle);
         scan.receiver = Some(receiver.clone());
         Ok(receiver)
     }
 
+    /// Serialize publication with invalidation so an obsolete catalog cannot
+    /// become the pending scan of a later connection. Reconnects share the
+    /// receiver, so a stale result restarts the one owned worker instead.
+    fn publish_project_scan(
+        &self,
+        sender: &tokio::sync::watch::Sender<Option<ProjectScanPublication>>,
+        generation: u64,
+        projects: Vec<RunnerProjectSummary>,
+        complete: bool,
+    ) -> ProjectScanPublishOutcome {
+        let scan = self
+            .initial_project_scan
+            .lock()
+            .unwrap_or_else(|e| e.into_inner());
+        if self.shutdown_requested() {
+            return ProjectScanPublishOutcome::Stopped;
+        }
+        if scan.generation != generation {
+            return ProjectScanPublishOutcome::Stale;
+        }
+        let _ = sender.send(Some(ProjectScanPublication {
+            projects: Arc::new(projects),
+            complete,
+            generation,
+        }));
+        ProjectScanPublishOutcome::Published
+    }
+
+    fn project_inventory_generation(&self) -> u64 {
+        self.initial_project_scan
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+            .generation
+    }
+
+    /// Wait for a catalog this stream has not published. A catalog read before
+    /// the latest mutation stays unpublished even when it is already buffered.
+    pub(super) async fn next_project_scan_publication(
+        &self,
+        receiver: &mut ProjectScanReceiver,
+        last: Option<&ProjectScanPublication>,
+    ) -> Result<ProjectScanPublication, &'static str> {
+        loop {
+            let current = receiver.borrow_and_update().clone();
+            if let Some(publication) = current {
+                let already_published =
+                    last.is_some_and(|last| Arc::ptr_eq(&last.projects, &publication.projects));
+                if !already_published
+                    && publication.generation == self.project_inventory_generation()
+                {
+                    return Ok(publication);
+                }
+            }
+            receiver
+                .changed()
+                .await
+                .map_err(|_| "project_inventory_scan_failed")?;
+        }
+    }
+
     pub(super) fn project_inventory_changes(&self) -> tokio::sync::watch::Receiver<()> {
         self.initial_project_scan
             .lock()
@@ -166,7 +244,7 @@ impl RunnerRuntimeState {
             .initial_project_scan
             .lock()
             .unwrap_or_else(|e| e.into_inner());
-        scan.invalidated = true;
+        scan.generation = scan.generation.wrapping_add(1);
         // Runtime ownership preserves completion notifications from dispatches
         // whose original stream has already disconnected. Watch coalesces bursts.
         scan.changed.send_replace(());