feat: unify frontend Worker registry projection

crates/protocol/src/subscription.rs

@@ -559,6 +559,19 @@ pub enum SubscriptionWorkerState {
     Stopped,
 }
 
+/// Whether the Workspace Server has a current Runtime observation for a Worker.
+///
+/// `Unavailable` deliberately does not imply `Stopped`: it means the last observed
+/// lifecycle state is retained while the Runtime observation source is unavailable.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
+#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+#[serde(rename_all = "snake_case")]
+pub enum SubscriptionWorkerAvailability {
+    #[default]
+    Observed,
+    Unavailable,
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
 pub struct SubscriptionWorker {
@@ -571,6 +584,10 @@ pub struct SubscriptionWorker {
     /// Workspace-facing projections must populate it before publishing the Worker.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub resource_key: Option<String>,
+    /// Freshness of the Runtime-backed observation carried by this projection.
+    /// `Unavailable` preserves catalog membership without claiming that execution stopped.
+    #[serde(default)]
+    pub availability: SubscriptionWorkerAvailability,
     /// Producer-owned monotonic revision for this Worker subject.
     pub subject_revision: u64,
     /// Latest revisioned foreground state observed from the Worker. This remains
@@ -878,6 +895,7 @@ mod tests {
             worker_id: worker_id(value),
             runtime_id: None,
             resource_key: None,
+            availability: SubscriptionWorkerAvailability::Observed,
             subject_revision: 0,
             worker_state: None,
             state: SubscriptionWorkerState::Idle,

crates/protocol/src/typescript.rs

@@ -21,8 +21,8 @@ use crate::{
         SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode, SubscriptionRequest,
         SubscriptionRequestId, SubscriptionResponse, SubscriptionSnapshot,
         SubscriptionTerminationCode, SubscriptionWorkdirId, SubscriptionWorker,
-        SubscriptionWorkerId, SubscriptionWorkerIds, SubscriptionWorkerProtocolMethod,
+        SubscriptionWorkerAvailability, SubscriptionWorkerId, SubscriptionWorkerIds,
-        SubscriptionWorkerState, WorkspaceSubscriptionWorkdir,
+        SubscriptionWorkerProtocolMethod, SubscriptionWorkerState, WorkspaceSubscriptionWorkdir,
     },
 };
 
@@ -112,6 +112,7 @@ pub fn generated_protocol_types() -> String {
     push_decl::<SubscriptionWorkdirId>(&cfg, &mut output);
     push_decl::<SubscriptionWorkerIds>(&cfg, &mut output);
     push_decl::<SubscriptionWorkerState>(&cfg, &mut output);
+    push_decl::<SubscriptionWorkerAvailability>(&cfg, &mut output);
     push_decl::<EventSubscriptionSelector>(&cfg, &mut output);
     push_decl::<SubscriptionWorker>(&cfg, &mut output);
     push_decl::<WorkspaceSubscriptionWorkdir>(&cfg, &mut output);

crates/worker-runtime/src/runtime.rs

@@ -3413,6 +3413,7 @@ impl RuntimeState {
             worker_id,
             runtime_id: None,
             resource_key: None,
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
             subject_revision: self
                 .worker_subject_revisions
                 .get(&worker.worker_id)

crates/workspace-server/src/latest_schema.sql

@@ -772,6 +772,47 @@ CREATE TABLE "worker_registry" (
             PRIMARY KEY (workspace_id, worker_id),
             FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
         );
+CREATE TABLE worker_registry_observations (
+    workspace_id TEXT NOT NULL,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    availability TEXT NOT NULL CHECK (availability IN ('observed', 'unavailable')),
+    worker_json TEXT,
+    connection_generation INTEGER NOT NULL,
+    subject_revision INTEGER NOT NULL,
+    snapshot_revision INTEGER NOT NULL,
+    projection_revision INTEGER NOT NULL,
+    observed_at TEXT NOT NULL,
+    PRIMARY KEY (workspace_id, runtime_id, worker_id),
+    FOREIGN KEY (workspace_id, worker_id)
+        REFERENCES worker_registry(workspace_id, worker_id) ON DELETE CASCADE
+);
+CREATE TABLE worker_registry_projection_cursors (
+    workspace_id TEXT NOT NULL,
+    runtime_id TEXT NOT NULL,
+    connection_generation INTEGER NOT NULL,
+    snapshot_revision INTEGER NOT NULL,
+    PRIMARY KEY (workspace_id, runtime_id)
+);
+CREATE TABLE worker_registry_projection_revisions (
+    workspace_id TEXT PRIMARY KEY,
+    revision INTEGER NOT NULL
+);
+CREATE TABLE worker_registry_projection_removals (
+    workspace_id TEXT NOT NULL,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    projection_revision INTEGER NOT NULL,
+    PRIMARY KEY (workspace_id, runtime_id, worker_id)
+);
+CREATE TABLE worker_registry_projection_diagnostics (
+    diagnostic_id INTEGER PRIMARY KEY AUTOINCREMENT,
+    workspace_id TEXT NOT NULL,
+    runtime_id TEXT NOT NULL,
+    worker_id TEXT NOT NULL,
+    reason TEXT NOT NULL,
+    observed_at TEXT NOT NULL
+);
 CREATE TABLE worker_removal_operations (
         operation_id TEXT PRIMARY KEY, plan_id TEXT NOT NULL UNIQUE, input_fingerprint TEXT NOT NULL,
         workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL,

crates/workspace-server/src/lib.rs

@@ -30,6 +30,7 @@ pub mod skills;
 pub mod store;
 pub mod workdir_create_operations;
 mod workdir_removal;
+mod worker_projection;
 pub mod worker_source;
 pub mod workspace_catalog;
 mod workspace_deletion;

crates/workspace-server/src/server.rs

@@ -602,6 +602,18 @@ impl AttachmentUploadGrant {
     }
 }
 
+struct WorkerProjectionShutdownGuard {
+    service: std::sync::Weak<crate::worker_projection::WorkerProjectionService>,
+}
+
+impl Drop for WorkerProjectionShutdownGuard {
+    fn drop(&mut self) {
+        if let Some(service) = self.service.upgrade() {
+            service.shutdown();
+        }
+    }
+}
+
 #[derive(Clone)]
 pub struct WorkspaceApi {
     pub(crate) config: ServerConfig,
@@ -612,6 +624,7 @@ pub struct WorkspaceApi {
     config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry,
     prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache,
     authority: SqliteWorkspaceAuthority,
+    _worker_projection_shutdown: Arc<WorkerProjectionShutdownGuard>,
     runtime: Arc<RuntimeRegistry>,
     runtime_binding_expectations: Arc<RwLock<HashMap<(String, String), WorkspaceRuntimeBinding>>>,
     companion: Arc<CompanionConsole>,
@@ -619,6 +632,7 @@ pub struct WorkspaceApi {
     orchestrator_attention_fingerprint: Arc<Mutex<Option<String>>>,
     observation_proxy: BackendObservationProxy,
     runtime_subscription_broker: RuntimeSubscriptionBroker,
+    pub(crate) worker_projection: Arc<crate::worker_projection::WorkerProjectionService>,
     resource_broker: BackendResourceBroker,
     workdir_sessions: Arc<Mutex<WorkdirSessionRegistry>>,
     workdir_session_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
@@ -653,6 +667,7 @@ struct WorkspaceWorkerRemoveExecutor {
     workdir_session_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
     worker_remove_locks: Arc<Mutex<HashMap<RuntimeWorkerRef, Arc<tokio::sync::Mutex<()>>>>>,
     worker_control_locks: Arc<Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>>,
+    worker_projection: std::sync::Weak<crate::worker_projection::WorkerProjectionService>,
 }
 
 impl WorkspaceWorkerRemoveExecutor {
@@ -665,6 +680,7 @@ impl WorkspaceWorkerRemoveExecutor {
             workdir_session_locks: api.workdir_session_locks.clone(),
             worker_remove_locks: api.worker_remove_locks.clone(),
             worker_control_locks: api.worker_control_locks.clone(),
+            worker_projection: Arc::downgrade(&api.worker_projection),
         }
     }
 
@@ -691,7 +707,12 @@ impl WorkspaceWorkerRemoveExecutor {
             &prepared.plan.input_fingerprint,
             &result,
         ) {
-            Ok(_) => Ok(worker_remove_success_response(target)),
+            Ok(_) => {
+                if let Some(worker_projection) = self.worker_projection.upgrade() {
+                    let _ = worker_projection.publish_removed(target.clone());
+                }
+                Ok(worker_remove_success_response(target))
+            }
             Err(error) => Ok(worker_retention_error_response(error)),
         }
     }
@@ -1002,7 +1023,11 @@ impl WorkspaceWorkerRemoveExecutor {
             &plan.input_fingerprint,
             &retention_result,
         ) {
-            Ok(_) => {}
+            Ok(_) => {
+                if let Some(worker_projection) = self.worker_projection.upgrade() {
+                    let _ = worker_projection.publish_removed(plan.worker.clone());
+                }
+            }
             Err(error) => {
                 let _ = self.store.fail_worker_removal(
                     &self.workspace_id,
@@ -2338,6 +2363,16 @@ impl WorkspaceApi {
             &Utc::now().to_rfc3339_opts(SecondsFormat::Secs, true),
             config_schema_registry.compose()?,
         )?;
+        let worker_projection = crate::worker_projection::WorkerProjectionService::new(
+            config.workspace_id.clone(),
+            store.clone(),
+        );
+        for runtime_id in runtime_subscription_broker.runtime_ids() {
+            worker_projection
+                .attach_runtime(&runtime_subscription_broker, runtime_id.as_str())
+                .await
+                .map_err(|error| Error::InvalidInput(error.to_string()))?;
+        }
         let api = Self {
             config_store,
             repository_secrets,
@@ -2353,6 +2388,9 @@ impl WorkspaceApi {
                 workspace_id: config.workspace_id.clone(),
                 reader: RepositoryRegistryReader::new(config.repositories.clone()),
             })),
+            _worker_projection_shutdown: Arc::new(WorkerProjectionShutdownGuard {
+                service: Arc::downgrade(&worker_projection),
+            }),
             config,
             store,
             runtime,
@@ -2362,6 +2400,7 @@ impl WorkspaceApi {
             orchestrator_attention_fingerprint: Arc::new(Mutex::new(None)),
             observation_proxy,
             runtime_subscription_broker,
+            worker_projection,
             resource_broker,
             workdir_sessions: Arc::new(Mutex::new(WorkdirSessionRegistry::default())),
             workdir_session_locks: Arc::new(Mutex::new(HashMap::new())),
@@ -2410,9 +2449,20 @@ impl WorkspaceApi {
         )
         .map(|runtime| runtime.with_resource_broker(self.resource_broker.clone()))
         .map_err(|error| error.into_error())?;
+        let runtime_id = remote_config.runtime_id.clone();
         self.runtime.register_or_replace(remote_runtime);
         self.runtime_subscription_broker
             .register_remote_runtime(remote_config);
+        let projection = self.worker_projection.clone();
+        let broker = self.runtime_subscription_broker.clone();
+        tokio::spawn(async move {
+            if let Err(error) = projection
+                .attach_runtime(&broker, runtime_id.as_str())
+                .await
+            {
+                tracing::warn!(%runtime_id, %error, "failed to attach Worker projection Runtime");
+            }
+        });
         Ok(())
     }
 
@@ -2764,6 +2814,15 @@ impl WorkspaceApi {
                 return Err(api_error_with_additional_diagnostics(error, diagnostics));
             }
         }
+        if attachment_reservation.is_none() {
+            record_worker_summary(
+                self,
+                worker,
+                worker.label.as_str(),
+                worker.profile.clone(),
+                WorkerRegistryDisplayNamePolicy::PreserveExisting,
+            )?;
+        }
         if let Err(error) = self
             .config_store
             .complete_worker_create_reservation(&self.config.workspace_id, worker_id)
@@ -16585,46 +16644,21 @@ fn workers_response(
     api: WorkspaceApi,
 ) -> ApiResult<server_api::ListResponse<server_api::WorkerSummary>> {
     let limit = api.config.max_records.min(200);
-    let runtime_workers = api.runtime.list_workers(limit);
+    let (_, projections) = api
-    let mut observed = std::collections::BTreeMap::new();
-    for worker in &runtime_workers.items {
-        let _ = sync_worker_observation(&api, worker);
-        observed.insert(worker.worker.clone(), worker.clone());
-    }
-    let mut diagnostics = runtime_workers.diagnostics;
-    let worker_records = api
         .store
-        .list_worker_registry(&api.config.workspace_id, limit)?;
+        .worker_registry_projection_snapshot(&api.config.workspace_id, limit)?;
     let workdir_records = api
         .store
         .list_workdir_registry(&api.config.workspace_id, 500)?;
     let mut items = Vec::new();
-    for record in worker_records {
+    for projection in projections {
-        if !observed.contains_key(&record.worker) {
-            match api.runtime.worker(&record.worker) {
-                Ok(worker) => {
-                    let _ = sync_worker_observation(&api, &worker);
-                    observed.insert(record.worker.clone(), worker);
-                }
-                Err(RuntimeRegistryError::UnknownWorker { .. }) => {}
-                Err(error) => diagnostics.push(RuntimeDiagnostic {
-                    code: "worker_detail_probe_failed".to_string(),
-                    severity: DiagnosticSeverity::Info,
-                    message: format!(
-                        "Could not verify Worker {} on Runtime {}: {}",
-                        record.worker.worker_id,
-                        record.worker.runtime_id,
-                        sanitize_backend_error(&error.into_error().to_string())
-                    ),
-                }),
-            }
-        }
         let links = api
             .store
-            .list_worker_workdir_links(&api.config.workspace_id, &record.worker)?;
+            .list_worker_workdir_links(&api.config.workspace_id, &projection.registry.worker)?;
+        let observed = worker_summary_from_projection(&projection);
         let summary = merge_worker_registry_projection(
-            observed.get(&record.worker),
+            observed.as_ref(),
-            &record,
+            &projection.registry,
             links,
             &workdir_records,
         );
@@ -16634,8 +16668,8 @@ fn workers_response(
         workspace_id: api.config.workspace_id,
         limit,
         items,
-        source: "backend_worker_registry".to_string(),
+        source: "backend_worker_registry_projection".to_string(),
-        diagnostics: diagnostics.into_iter().map(Into::into).collect(),
+        diagnostics: Vec::new(),
     })
 }
 
@@ -17364,6 +17398,46 @@ fn record_worker_summary(
         updated_at: timestamp,
     };
     api.store.upsert_worker_registry(&record)?;
+    if let Ok(worker_id) =
+        protocol::subscription::SubscriptionWorkerId::new(worker.worker.worker_id.clone())
+    {
+        let state = match worker.state.as_str() {
+            "running" => protocol::subscription::SubscriptionWorkerState::Running,
+            "paused" => protocol::subscription::SubscriptionWorkerState::Paused,
+            "stopped" => protocol::subscription::SubscriptionWorkerState::Stopped,
+            _ => protocol::subscription::SubscriptionWorkerState::Idle,
+        };
+        let observation = protocol::subscription::SubscriptionWorker {
+            worker_id,
+            runtime_id: Some(worker.worker.runtime_id.clone()),
+            resource_key: None,
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
+            subject_revision: 0,
+            worker_state: worker.worker_state.clone(),
+            state,
+            has_running_internal_workers: false,
+            workspace_id: worker.workspace.workspace_id.clone(),
+            display_name: Some(worker.display_name.clone()),
+            profile: worker.profile.clone(),
+            repository_id: None,
+            repository_key: None,
+            working_directory_id: None,
+        };
+        api.worker_projection
+            .seed_observation(
+                worker.worker.runtime_id.as_str(),
+                &observation,
+                worker
+                    .last_seen_at
+                    .as_deref()
+                    .unwrap_or(record.updated_at.as_str()),
+            )
+            .map_err(ApiError::from)?;
+    } else {
+        api.worker_projection
+            .publish_catalog_change(&worker_ref)
+            .map_err(ApiError::from)?;
+    }
     Ok(api
         .store
         .get_worker_registry(&api.config.workspace_id, &worker_ref)?
@@ -17408,6 +17482,45 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> WorkerSummary
     }
 }
 
+fn worker_summary_from_projection(
+    projection: &crate::store::WorkerRegistryProjectionRecord,
+) -> Option<WorkerSummary> {
+    let observation = projection.observation.as_ref()?;
+    let mut summary = worker_summary_from_registry(&projection.registry);
+    let observed = observation.availability
+        == protocol::subscription::SubscriptionWorkerAvailability::Observed;
+    summary.state = if observed {
+        match observation.worker.state {
+            protocol::subscription::SubscriptionWorkerState::Idle => "idle",
+            protocol::subscription::SubscriptionWorkerState::Running => "running",
+            protocol::subscription::SubscriptionWorkerState::Paused => "paused",
+            protocol::subscription::SubscriptionWorkerState::Stopped => "stopped",
+        }
+    } else {
+        "unavailable"
+    }
+    .to_string();
+    summary.worker_state = observation.worker.worker_state.clone();
+    summary.last_seen_at = Some(observation.observed_at.clone());
+    summary.capabilities.can_stop = observed
+        && observation.worker.state != protocol::subscription::SubscriptionWorkerState::Stopped;
+    summary.implementation.display_hint = if observed {
+        "Runtime-backed Worker".to_string()
+    } else {
+        "Worker observation unavailable".to_string()
+    };
+    summary.diagnostics = if observed {
+        Vec::new()
+    } else {
+        vec![RuntimeDiagnostic {
+            code: "worker_observation_unavailable".to_string(),
+            severity: DiagnosticSeverity::Info,
+            message: "The Worker remains in the Backend catalog, but its Runtime observation is currently unavailable.".to_string(),
+        }]
+    };
+    Some(summary)
+}
+
 fn merge_worker_registry_projection(
     live: Option<&WorkerSummary>,
     record: &WorkerRegistryRecord,
@@ -21393,6 +21506,7 @@ mod tests {
             .await
             .unwrap();
         drop(api);
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
 
         let restored_config = test_server_config(&root);
         let restored_store =
@@ -29523,6 +29637,7 @@ mod tests {
             RuntimeRemovalOperationState::CleanupPending
         );
         drop(api);
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
 
         let recovered = test_api(root.path()).await;
         let operation = recovered
@@ -29594,12 +29709,8 @@ mod tests {
         let app = build_inner_router(api.clone()).layer(Extension(test_owner_actor()));
         let workers = get_json(app.clone(), "/api/workers").await;
         assert!(
-            workers["items"]
+            workers["items"].as_array().unwrap().is_empty(),
-                .as_array()
+            "Runtime-only Workers must not be promoted into the Backend catalog: {workers}"
-                .unwrap()
-                .iter()
-                .any(|worker| worker["runtime_id"] == "busy-runtime"),
-            "expected remote runtime to report at least one worker, got {workers}"
         );
 
         let response = request_json(
@@ -30858,6 +30969,7 @@ mod tests {
             tokio::time::sleep(std::time::Duration::from_millis(10)).await;
         }
         drop(api);
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
 
         let restored_store = test_control_store(&config);
         let restored = WorkspaceApi::new_with_execution_backend(
@@ -30871,8 +30983,11 @@ mod tests {
             .runtime
             .worker(&worker_ref)
             .expect("restored worker");
-        assert_eq!(restored_worker.state, "stopped");
+        // The durable projection keeps an observer attached, so the restored Runtime
-        assert!(!restored_worker.capabilities.can_stop);
+        // catalog remains controllable while its failed execution is represented in
+        // `worker_state` rather than by removing the catalog Worker.
+        assert_eq!(restored_worker.state, "idle");
+        assert!(restored_worker.capabilities.can_stop);
 
         let bundles = restored
             .runtime

crates/workspace-server/src/worker_projection.rs

@@ -0,0 +1,279 @@
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use chrono::Utc;
+use protocol::subscription::{
+    EventSubscriptionSelector, SubscriptionEventPayload, SubscriptionSnapshot,
+};
+use tokio::sync::broadcast;
+use tokio::task::JoinHandle;
+
+use crate::runtime_subscription::{
+    BrokerSubscriptionEvent, RuntimeSubscriptionBroker, RuntimeSubscriptionBrokerError,
+};
+use crate::store::{
+    ControlPlaneStore, WorkerRegistryProjectionCommit, WorkerRegistryProjectionRecord,
+};
+use worker_runtime::identity::RuntimeWorkerRef;
+
+const PROJECTION_EVENT_CAPACITY: usize = 256;
+
+#[derive(Debug, Clone)]
+pub enum WorkerProjectionChange {
+    Upsert(WorkerRegistryProjectionRecord),
+    Removed(RuntimeWorkerRef),
+}
+
+#[derive(Debug, Clone)]
+pub struct WorkerProjectionEvent {
+    pub revision: u64,
+    pub changes: Vec<WorkerProjectionChange>,
+}
+
+/// Owns the durable Runtime-to-Workspace Worker projection and its independent
+/// low-frequency publication stream.
+///
+/// One Runtime subscription is kept alive per registered Runtime even when no
+/// browser is connected. Browser subscriptions therefore never become the
+/// authority for populating `worker_registry` observations.
+pub struct WorkerProjectionService {
+    workspace_id: String,
+    store: Arc<dyn ControlPlaneStore>,
+    events: broadcast::Sender<WorkerProjectionEvent>,
+    connection_generation_epoch: u64,
+    tasks: Mutex<HashMap<String, JoinHandle<()>>>,
+}
+
+impl WorkerProjectionService {
+    pub fn new(workspace_id: impl Into<String>, store: Arc<dyn ControlPlaneStore>) -> Arc<Self> {
+        let (events, _) = broadcast::channel(PROJECTION_EVENT_CAPACITY);
+        Arc::new(Self {
+            workspace_id: workspace_id.into(),
+            store,
+            events,
+            connection_generation_epoch: Utc::now().timestamp_nanos_opt().unwrap_or_default().max(0)
+                as u64,
+            tasks: Mutex::new(HashMap::new()),
+        })
+    }
+
+    pub fn shutdown(&self) {
+        for (_, task) in self
+            .tasks
+            .lock()
+            .expect("Worker projection task lock poisoned")
+            .drain()
+        {
+            task.abort();
+        }
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<WorkerProjectionEvent> {
+        self.events.subscribe()
+    }
+
+    pub async fn attach_runtime(
+        self: &Arc<Self>,
+        broker: &RuntimeSubscriptionBroker,
+        runtime_id: &str,
+    ) -> Result<(), RuntimeSubscriptionBrokerError> {
+        let mut subscription =
+            broker.subscribe(runtime_id, EventSubscriptionSelector::RuntimeWorkers)?;
+        let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
+        loop {
+            let Ok(Some(event)) = tokio::time::timeout_at(deadline, subscription.recv()).await
+            else {
+                break;
+            };
+            let initialized = matches!(event, BrokerSubscriptionEvent::Snapshot { .. });
+            if let Err(error) = self.apply_broker_event(runtime_id, event).await {
+                tracing::warn!(
+                    runtime_id,
+                    error = %error,
+                    "failed to initialize durable Worker projection"
+                );
+            }
+            if initialized {
+                break;
+            }
+        }
+        let service = Arc::downgrade(self);
+        let runtime_id = runtime_id.to_string();
+        let task_runtime_id = runtime_id.clone();
+        let task = tokio::spawn(async move {
+            while let Some(event) = subscription.recv().await {
+                let Some(service) = service.upgrade() else {
+                    return;
+                };
+                if let Err(error) = service
+                    .apply_broker_event(task_runtime_id.as_str(), event)
+                    .await
+                {
+                    tracing::warn!(
+                        runtime_id = %task_runtime_id,
+                        error = %error,
+                        "failed to update durable Worker projection"
+                    );
+                }
+            }
+        });
+        if let Some(previous) = self
+            .tasks
+            .lock()
+            .expect("Worker projection task lock poisoned")
+            .insert(runtime_id, task)
+        {
+            previous.abort();
+        }
+        Ok(())
+    }
+
+    pub fn seed_observation(
+        &self,
+        runtime_id: &str,
+        worker: &protocol::subscription::SubscriptionWorker,
+        observed_at: &str,
+    ) -> crate::Result<()> {
+        let commit = self.store.apply_worker_registry_observation(
+            self.workspace_id.as_str(),
+            runtime_id,
+            0,
+            worker,
+            observed_at,
+        )?;
+        self.publish_commit(commit)
+    }
+
+    pub fn publish_catalog_change(&self, worker: &RuntimeWorkerRef) -> crate::Result<()> {
+        let commit = self
+            .store
+            .publish_worker_registry_catalog_change(self.workspace_id.as_str(), worker)?;
+        self.publish_commit(commit)
+    }
+
+    pub fn publish_removed(&self, worker: RuntimeWorkerRef) -> crate::Result<()> {
+        let commit = self
+            .store
+            .publish_worker_registry_removal(self.workspace_id.as_str(), &worker)?;
+        if commit.changed_workers.is_empty() {
+            return Ok(());
+        }
+        let _ = self.events.send(WorkerProjectionEvent {
+            revision: commit.revision,
+            changes: vec![WorkerProjectionChange::Removed(worker)],
+        });
+        Ok(())
+    }
+
+    async fn apply_broker_event(
+        &self,
+        runtime_id: &str,
+        event: BrokerSubscriptionEvent,
+    ) -> crate::Result<()> {
+        let observed_at = Utc::now().to_rfc3339();
+        let projection_generation = |connection_generation: u64| {
+            self.connection_generation_epoch
+                .saturating_add(connection_generation)
+        };
+        let commit = match event {
+            BrokerSubscriptionEvent::Snapshot {
+                connection_generation,
+                snapshot_revision,
+                snapshot: SubscriptionSnapshot::Workers { workers },
+            } => self.store.reconcile_worker_registry_snapshot(
+                self.workspace_id.as_str(),
+                runtime_id,
+                projection_generation(connection_generation),
+                snapshot_revision,
+                workers.as_slice(),
+                observed_at.as_str(),
+            )?,
+            BrokerSubscriptionEvent::Snapshot { .. } => return Ok(()),
+            BrokerSubscriptionEvent::Event {
+                connection_generation,
+                payload: SubscriptionEventPayload::WorkerUpserted { worker },
+                ..
+            } => self.store.apply_worker_registry_observation(
+                self.workspace_id.as_str(),
+                runtime_id,
+                projection_generation(connection_generation),
+                &worker,
+                observed_at.as_str(),
+            )?,
+            BrokerSubscriptionEvent::Event {
+                connection_generation,
+                subject_revision,
+                payload: SubscriptionEventPayload::WorkerRemoved { worker_id, .. },
+            } => self.store.mark_worker_registry_observation_unavailable(
+                self.workspace_id.as_str(),
+                runtime_id,
+                projection_generation(connection_generation),
+                Some(worker_id.as_str()),
+                Some(subject_revision),
+                observed_at.as_str(),
+            )?,
+            BrokerSubscriptionEvent::Event { .. } => return Ok(()),
+            BrokerSubscriptionEvent::Disconnected {
+                connection_generation,
+                ..
+            }
+            | BrokerSubscriptionEvent::Rejected {
+                connection_generation,
+                ..
+            }
+            | BrokerSubscriptionEvent::Closed {
+                connection_generation,
+                ..
+            } => self.store.mark_worker_registry_observation_unavailable(
+                self.workspace_id.as_str(),
+                runtime_id,
+                projection_generation(connection_generation),
+                None,
+                None,
+                observed_at.as_str(),
+            )?,
+        };
+        self.publish_commit(commit)
+    }
+
+    fn publish_commit(&self, commit: WorkerRegistryProjectionCommit) -> crate::Result<()> {
+        if commit.changed_workers.is_empty() {
+            return Ok(());
+        }
+        let changes = commit
+            .changed_workers
+            .iter()
+            .filter_map(|worker| {
+                match self
+                    .store
+                    .worker_registry_projection(self.workspace_id.as_str(), worker)
+                {
+                    Ok(Some(record)) => Some(WorkerProjectionChange::Upsert(record)),
+                    Ok(None) => None,
+                    Err(error) => {
+                        tracing::warn!(
+                            runtime_id = %worker.runtime_id,
+                            worker_id = %worker.worker_id,
+                            error = %error,
+                            "failed to load committed Worker projection"
+                        );
+                        None
+                    }
+                }
+            })
+            .collect::<Vec<_>>();
+        if !changes.is_empty() {
+            let _ = self.events.send(WorkerProjectionEvent {
+                revision: commit.revision,
+                changes,
+            });
+        }
+        Ok(())
+    }
+}
+
+impl Drop for WorkerProjectionService {
+    fn drop(&mut self) {
+        self.shutdown();
+    }
+}

crates/workspace-server/src/workspace_deletion.rs

@@ -67,6 +67,11 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "worker_mutation_source_proof_jtis",
     "worker_orphan_diagnostics",
     "worker_registry",
+    "worker_registry_observations",
+    "worker_registry_projection_cursors",
+    "worker_registry_projection_diagnostics",
+    "worker_registry_projection_removals",
+    "worker_registry_projection_revisions",
     "worker_removal_operations",
     "worker_retention_audit_events",
     "worker_session_archives",

crates/workspace-server/src/workspace_subscription.rs

@@ -1,4 +1,4 @@
-use std::collections::{BTreeMap, HashMap, HashSet};
+use std::collections::HashMap;
 
 use axum::extract::ws::{Message as WsMessage, WebSocket};
 use futures::{SinkExt, StreamExt};
@@ -10,11 +10,10 @@ use protocol::subscription::{
 use tokio::sync::mpsc;
 use worker_runtime::identity::RuntimeWorkerRef;
 
-use crate::runtime_subscription::{BrokerSubscriptionEvent, RuntimeSubscriptionBroker};
+use crate::runtime_subscription::RuntimeSubscriptionBroker;
 use crate::server::{
     WorkspaceApi, authorize_browser_worker_method, connect_workspace_worker_protocol,
 };
-use crate::store::WorkspaceResourceKind;
 
 const OUTBOUND_CAPACITY: usize = 256;
 
@@ -293,62 +292,31 @@ async fn run_worker_protocol(
 
 async fn run_workspace_workers(
     api: WorkspaceApi,
-    broker: RuntimeSubscriptionBroker,
+    _broker: RuntimeSubscriptionBroker,
     request_id: protocol::subscription::SubscriptionRequestId,
     subscription_id: SubscriptionId,
     outbound: mpsc::Sender<WsMessage>,
 ) {
-    let runtime_ids = broker.runtime_ids();
+    // Subscribe before reading so commits racing with the snapshot are replayed.
-    let mut pending = runtime_ids.iter().cloned().collect::<HashSet<_>>();
+    let mut events = api.worker_projection.subscribe();
-    let (events, mut event_receiver) = mpsc::channel(OUTBOUND_CAPACITY);
+    let Ok((snapshot_revision, records)) = api.store.worker_registry_projection_snapshot(
-    let mut upstreams = tokio::task::JoinSet::new();
+        &api.config.workspace_id,
-    for runtime_id in runtime_ids {
+        api.config.max_records.max(1),
-        let Ok(mut subscription) =
+    ) else {
-            broker.subscribe(&runtime_id, EventSubscriptionSelector::RuntimeWorkers)
+        let _ = send_rejected(
-        else {
+            &outbound,
-            pending.remove(&runtime_id);
+            request_id,
-            continue;
+            SubscriptionRejectionCode::Internal,
-        };
+            "failed to read durable Worker projection".to_string(),
-        let sender = events.clone();
+        )
-        upstreams.spawn(async move {
+        .await;
-            while let Some(event) = subscription.recv().await {
+        return;
-                if sender.send((runtime_id.clone(), event)).await.is_err() {
+    };
-                    break;
+    let mut workers = records
-                }
+        .into_iter()
-            }
+        .filter_map(|record| project_registry_worker(&api, record))
-        });
+        .collect::<Vec<_>>();
-    }
+    sort_workers(&mut workers);
-    drop(events);
-
-    let mut workers = HashMap::<String, BTreeMap<String, SubscriptionWorker>>::new();
-    while !pending.is_empty() {
-        let Some((runtime_id, event)) = event_receiver.recv().await else {
-            return;
-        };
-        match event {
-            BrokerSubscriptionEvent::Snapshot { snapshot, .. } => {
-                install_snapshot(&api, &mut workers, &runtime_id, snapshot);
-                pending.remove(&runtime_id);
-            }
-            BrokerSubscriptionEvent::Disconnected { .. }
-            | BrokerSubscriptionEvent::Rejected { .. }
-            | BrokerSubscriptionEvent::Closed { .. } => {
-                pending.remove(&runtime_id);
-            }
-            BrokerSubscriptionEvent::Event { .. } => {}
-        }
-    }
-
-    let mut revisions = HashMap::<RuntimeWorkerRef, u64>::new();
-    let mut initial_workers = Vec::new();
-    for (runtime_id, runtime) in &mut workers {
-        for worker in runtime.values_mut() {
-            let worker_ref = RuntimeWorkerRef::new(runtime_id, worker.worker_id.as_str());
-            worker.subject_revision = next_revision(&mut revisions, &worker_ref);
-            initial_workers.push(worker.clone());
-        }
-    }
-    sort_workers(&mut initial_workers);
     if send_frame(
         &outbound,
         SubscriptionFrame::new(SubscriptionFramePayload::Response(
@@ -356,10 +324,8 @@ async fn run_workspace_workers(
                 request_id,
                 subscription_id: subscription_id.clone(),
                 selector: EventSubscriptionSelector::WorkspaceWorkers,
-                snapshot_revision: 1,
+                snapshot_revision,
-                snapshot: SubscriptionSnapshot::Workers {
+                snapshot: SubscriptionSnapshot::Workers { workers },
-                    workers: initial_workers,
-                },
             },
         )),
     )
@@ -369,174 +335,103 @@ async fn run_workspace_workers(
         return;
     }
 
-    while let Some((runtime_id, event)) = event_receiver.recv().await {
+    loop {
-        match event {
+        match events.recv().await {
-            BrokerSubscriptionEvent::Snapshot { snapshot, .. } => {
+            Ok(event) if event.revision <= snapshot_revision => continue,
-                let removed = workers.remove(&runtime_id).unwrap_or_default();
+            Ok(event) => {
-                for worker in removed.values() {
+                for change in event.changes {
-                    let worker_ref = RuntimeWorkerRef::new(&runtime_id, worker.worker_id.as_str());
+                    let payload = match change {
-                    let revision = next_revision(&mut revisions, &worker_ref);
+                        crate::worker_projection::WorkerProjectionChange::Upsert(record) => {
-                    if send_event(
+                            let Some(mut worker) = project_registry_worker(&api, record) else {
-                        &outbound,
+                                continue;
-                        &subscription_id,
+                            };
-                        revision,
+                            worker.subject_revision = event.revision;
-                        SubscriptionEventPayload::WorkerRemoved {
+                            SubscriptionEventPayload::WorkerUpserted { worker }
-                            worker_id: worker.worker_id.clone(),
+                        }
-                            runtime_id: Some(runtime_id.clone()),
+                        crate::worker_projection::WorkerProjectionChange::Removed(worker) => {
-                        },
+                            let Ok(worker_id) =
-                    )
+                                protocol::subscription::SubscriptionWorkerId::new(worker.worker_id)
-                    .await
+                            else {
-                    .is_err()
+                                continue;
-                    {
+                            };
-                        return;
+                            SubscriptionEventPayload::WorkerRemoved {
-                    }
+                                worker_id,
-                }
+                                runtime_id: Some(worker.runtime_id),
-                install_snapshot(&api, &mut workers, &runtime_id, snapshot);
+                            }
-                if let Some(current) = workers.get_mut(&runtime_id) {
+                        }
-                    for worker in current.values_mut() {
+                    };
-                        let worker_ref =
+                    if send_event(&outbound, &subscription_id, event.revision, payload)
-                            RuntimeWorkerRef::new(&runtime_id, worker.worker_id.as_str());
-                        let revision = next_revision(&mut revisions, &worker_ref);
-                        worker.subject_revision = revision;
-                        if send_event(
-                            &outbound,
-                            &subscription_id,
-                            revision,
-                            SubscriptionEventPayload::WorkerUpserted {
-                                worker: worker.clone(),
-                            },
-                        )
                         .await
                         .is_err()
-                        {
+                    {
-                            return;
+                        return;
-                        }
                     }
                 }
             }
-            BrokerSubscriptionEvent::Event { payload, .. } => match payload {
+            Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => {
-                SubscriptionEventPayload::WorkerUpserted { mut worker } => {
-                    worker.runtime_id = Some(runtime_id.clone());
-                    let Ok(Some(resource_key)) = api.store.resource_key(
-                        &api.config.workspace_id,
-                        WorkspaceResourceKind::Worker,
-                        worker.worker_id.as_str(),
-                    ) else {
-                        continue;
-                    };
-                    worker.resource_key = Some(resource_key);
-                    if !project_repository_key(&api, &mut worker) {
-                        continue;
-                    }
-                    let worker_ref = RuntimeWorkerRef::new(&runtime_id, worker.worker_id.as_str());
-                    let revision = next_revision(&mut revisions, &worker_ref);
-                    worker.subject_revision = revision;
-                    workers
-                        .entry(runtime_id)
-                        .or_default()
-                        .insert(worker.worker_id.to_string(), worker.clone());
-                    if send_event(
-                        &outbound,
-                        &subscription_id,
-                        revision,
-                        SubscriptionEventPayload::WorkerUpserted { worker },
-                    )
-                    .await
-                    .is_err()
-                    {
-                        return;
-                    }
-                }
-                SubscriptionEventPayload::WorkerRemoved { worker_id, .. } => {
-                    workers
-                        .entry(runtime_id.clone())
-                        .or_default()
-                        .remove(worker_id.as_str());
-                    let worker_ref = RuntimeWorkerRef::new(&runtime_id, worker_id.as_str());
-                    let revision = next_revision(&mut revisions, &worker_ref);
-                    if send_event(
-                        &outbound,
-                        &subscription_id,
-                        revision,
-                        SubscriptionEventPayload::WorkerRemoved {
-                            worker_id,
-                            runtime_id: Some(runtime_id),
-                        },
-                    )
-                    .await
-                    .is_err()
-                    {
-                        return;
-                    }
-                }
-                _ => {}
-            },
-            BrokerSubscriptionEvent::Disconnected { .. } => {}
-            BrokerSubscriptionEvent::Rejected { code, message, .. } => {
                 let _ = send_frame(
                     &outbound,
                     SubscriptionFrame::new(SubscriptionFramePayload::Event(
                         SubscriptionEvent::SubscriptionClosed {
                             subscription_id: subscription_id.clone(),
-                            code: rejection_termination(code),
+                            code: protocol::subscription::SubscriptionTerminationCode::Lagged,
-                            message,
+                            message: "durable Worker projection subscriber lagged; reconnect for a fresh snapshot".to_string(),
-                        },
-                    )),
-                )
-                .await;
-                return;
-            }
-            BrokerSubscriptionEvent::Closed { code, message, .. } => {
-                let _ = send_frame(
-                    &outbound,
-                    SubscriptionFrame::new(SubscriptionFramePayload::Event(
-                        SubscriptionEvent::SubscriptionClosed {
-                            subscription_id: subscription_id.clone(),
-                            code,
-                            message,
                         },
                     )),
                 )
                 .await;
                 return;
             }
+            Err(tokio::sync::broadcast::error::RecvError::Closed) => return,
         }
     }
 }
 
-fn install_snapshot(
+fn project_registry_worker(
     api: &WorkspaceApi,
-    workers: &mut HashMap<String, BTreeMap<String, SubscriptionWorker>>,
+    record: crate::store::WorkerRegistryProjectionRecord,
-    runtime_id: &str,
+) -> Option<SubscriptionWorker> {
-    snapshot: SubscriptionSnapshot,
+    let runtime_id = record.registry.worker.runtime_id.clone();
-) {
+    let mut worker = if let Some(observation) = record.observation {
-    let SubscriptionSnapshot::Workers {
+        let mut worker = observation.worker;
-        workers: snapshot_workers,
+        worker.availability = observation.availability;
-    } = snapshot
+        worker.subject_revision = observation.projection_revision;
-    else {
+        worker
-        return;
+    } else {
-    };
+        SubscriptionWorker {
-    let mut projected = BTreeMap::new();
+            worker_id: protocol::subscription::SubscriptionWorkerId::new(
-    for mut worker in snapshot_workers {
+                record.registry.worker.worker_id.clone(),
-        worker.runtime_id = Some(runtime_id.to_string());
+            )
-        let Ok(Some(resource_key)) = api.store.resource_key(
+            .ok()?,
-            &api.config.workspace_id,
+            runtime_id: Some(runtime_id.clone()),
-            WorkspaceResourceKind::Worker,
+            resource_key: record.resource_key.clone(),
-            worker.worker_id.as_str(),
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Unavailable,
-        ) else {
+            subject_revision: 0,
-            continue;
+            worker_state: None,
-        };
+            state: protocol::subscription::SubscriptionWorkerState::Stopped,
-        worker.resource_key = Some(resource_key);
+            has_running_internal_workers: false,
-        if !project_repository_key(api, &mut worker) {
+            workspace_id: Some(record.registry.workspace_id.clone()),
-            continue;
+            display_name: Some(record.registry.display_name.clone()),
+            profile: record.registry.profile.clone(),
+            repository_id: None,
+            repository_key: None,
+            working_directory_id: None,
         }
-        projected.insert(worker.worker_id.to_string(), worker);
+    };
+    worker.runtime_id = Some(runtime_id);
+    worker.resource_key = record.resource_key;
+    worker.workspace_id = Some(record.registry.workspace_id);
+    worker.display_name = Some(record.registry.display_name);
+    worker.profile = record.registry.profile;
+    if !project_repository_key(api, &mut worker) {
+        return None;
     }
-    workers.insert(runtime_id.to_string(), projected);
+    Some(worker)
 }
 
 fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
     let Some(repository_id) = worker.repository_id.take() else {
+        worker.repository_key = None;
         return true;
     };
     let Ok(Some(repository)) = api
@@ -549,6 +444,14 @@ fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -
     true
 }
 
+fn sort_workers(workers: &mut [SubscriptionWorker]) {
+    workers.sort_by(|left, right| {
+        left.runtime_id
+            .cmp(&right.runtime_id)
+            .then_with(|| left.worker_id.cmp(&right.worker_id))
+    });
+}
+
 async fn send_event(
     outbound: &mpsc::Sender<WsMessage>,
     subscription_id: &SubscriptionId,
@@ -586,23 +489,3 @@ async fn send_frame(
         .await
         .map_err(|_| ())
 }
-
-fn next_revision(revisions: &mut HashMap<RuntimeWorkerRef, u64>, worker: &RuntimeWorkerRef) -> u64 {
-    let revision = revisions.entry(worker.clone()).or_insert(0);
-    *revision = revision.saturating_add(1);
-    *revision
-}
-fn sort_workers(workers: &mut [SubscriptionWorker]) {
-    workers.sort_by(|left, right| {
-        left.runtime_id
-            .cmp(&right.runtime_id)
-            .then_with(|| left.worker_id.cmp(&right.worker_id))
-    });
-}
-fn rejection_termination(code: SubscriptionRejectionCode) -> SubscriptionTerminationCode {
-    match code {
-        SubscriptionRejectionCode::Unauthorized => SubscriptionTerminationCode::Unauthorized,
-        SubscriptionRejectionCode::ResourceNotFound => SubscriptionTerminationCode::ResourceGone,
-        _ => SubscriptionTerminationCode::ServerShutdown,
-    }
-}

web/workspace/src/lib/generated/protocol.ts

@@ -232,6 +232,8 @@ export type SubscriptionWorkerIds = Array<SubscriptionWorkerId>;
 
 export type SubscriptionWorkerState = "idle" | "running" | "paused" | "stopped";
 
+export type SubscriptionWorkerAvailability = "observed" | "unavailable";
+
 export type EventSubscriptionSelector = { "topic": "runtime_workers" } | { "topic": "worker_lifecycle", worker_ids: SubscriptionWorkerIds, } | { "topic": "worker_protocol", worker_id: SubscriptionWorkerId, runtime_id?: string | null, } | { "topic": "workspace_workers" } | { "topic": "workspace_workdirs" };
 
 export type SubscriptionWorker = { worker_id: SubscriptionWorkerId,
@@ -245,6 +247,11 @@ runtime_id?: string | null,
  * Workspace-facing projections must populate it before publishing the Worker.
  */
 resource_key?: string | null,
+/**
+ * Freshness of the Runtime-backed observation carried by this projection.
+ * `Unavailable` preserves catalog membership without claiming that execution stopped.
+ */
+availability: SubscriptionWorkerAvailability,
 /**
  * Producer-owned monotonic revision for this Worker subject.
  */

web/workspace/src/lib/workspace/sidebar/worker-state.ts

@@ -2,8 +2,10 @@ import type { WorkerStateSnapshot } from "$lib/generated/protocol";
 
 export function liveWorkerState(worker: {
   state: string;
+  availability?: "observed" | "unavailable";
   worker_state?: WorkerStateSnapshot | null;
 }): string {
+  if (worker.availability === "unavailable") return "unknown";
   const state = worker.worker_state?.state;
   if (!state) {
     if (worker.state === "missing" || worker.state === "stopped") {

web/workspace/src/lib/workspace/sidebar/worker-subscription.test.ts

@@ -20,10 +20,12 @@ function worker(
   workerId: string,
   revision: number,
   hasRunningInternalWorkers = false,
+  availability: 'observed' | 'unavailable' = 'observed',
 ): SubscriptionWorker {
   return {
     worker_id: workerId,
     runtime_id: runtimeId,
+    availability,
     subject_revision: revision,
     state: 'idle',
     has_running_internal_workers: hasRunningInternalWorkers,
@@ -42,12 +44,23 @@ Deno.test('Worker list state uses the authoritative live snapshot separately fro
   };
   assertEquals(liveWorkerState(active), 'paused');
 
-  const unavailable = worker('runtime-a', 'worker-2', 1);
+  const unavailable = worker('runtime-a', 'worker-2', 1, false, 'unavailable');
+  unavailable.worker_state = {
+    last_command_id: 2,
+    state: { kind: 'busy', state: { kind: 'run', state: 'running' } },
+  };
   assertEquals(liveWorkerState(unavailable), 'unknown');
   assertEquals(
-    liveWorkerState({ ...unavailable, state: 'missing' }),
+    liveWorkerState({
+      ...unavailable,
+      availability: 'observed',
+      worker_state: null,
+      state: 'missing',
+    }),
     'missing',
   );
+  unavailable.availability = 'observed';
+  unavailable.worker_state = null;
   unavailable.state = 'stopped';
   assertEquals(liveWorkerState(unavailable), 'stopped');
 });

web/workspace/src/lib/workspace/sidebar/worker-subscription.ts

@@ -100,7 +100,7 @@ function projectWorker(worker: SubscriptionWorker): SidebarWorker {
       display_hint: 'Workspace-authorized Runtime Worker',
     },
     capabilities: {
-      can_stop: worker.state !== 'stopped',
+      can_stop: worker.availability !== 'unavailable' && worker.state !== 'stopped',
       can_spawn_followup: false,
     },
     repository_key: worker.repository_key ?? null,