feat(workspace): persist runtime worker 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/worker-runtime/src/runtime.rs

@@ -3381,6 +3381,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,40 @@ 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_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)
@@ -16546,46 +16605,21 @@ fn workers_response(
     api: WorkspaceApi,
 ) -> ApiResult<workspace_api::ListResponse<workspace_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,
         );
@@ -16595,8 +16629,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(),
     })
 }
 
@@ -17321,6 +17355,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)?
@@ -17365,6 +17439,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,
@@ -21349,6 +21462,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 =
@@ -29479,6 +29593,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
@@ -29550,12 +29665,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(
@@ -30814,6 +30925,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(
@@ -30827,8 +30939,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/store.rs

@@ -5,6 +5,9 @@ use std::time::Duration;
 
 use async_trait::async_trait;
 use flow::{CompiledFlowDefinition, FlowSourceKind, compile_flow_source};
+use protocol::subscription::{
+    SubscriptionWorker, SubscriptionWorkerAvailability, SubscriptionWorkerId,
+};
 use rusqlite::backup::Backup;
 use rusqlite::{Connection, OpenFlags, OptionalExtension, TransactionBehavior, params};
 use serde::{Deserialize, Serialize};
@@ -18,7 +21,7 @@ use crate::workspace_deletion::WorkspaceDeletionStore;
 use crate::{Error, Result};
 
 const OLDEST_SCHEMA_VERSION: i64 = 50;
-const LATEST_SCHEMA_VERSION: i64 = 61;
+const LATEST_SCHEMA_VERSION: i64 = 62;
 const SCHEMA_BASELINE_NAME: &str = "workspace schema baseline";
 const WORKSPACE_RUNTIME_BINDINGS_MIGRATION_NAME: &str = "workspace runtime bindings";
 const RUNTIME_BINDING_AUDIT_MIGRATION_NAME: &str = "workspace Runtime binding revision and audit";
@@ -36,6 +39,44 @@ const WORKDIR_CREDENTIAL_CANDIDATE_SNAPSHOT_MIGRATION_NAME: &str =
     "Workdir create credential candidate snapshots";
 const RUNTIME_REMOVAL_OPERATION_MIGRATION_NAME: &str = "guarded Runtime removal operations";
 const REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME: &str = "remove obsolete Worker run generation";
+const WORKER_REGISTRY_PROJECTION_MIGRATION_NAME: &str =
+    "durable Runtime-backed Worker registry projection";
+const WORKER_REGISTRY_PROJECTION_SCHEMA: &str = r#"
+CREATE TABLE IF NOT EXISTS 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 IF NOT EXISTS 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 IF NOT EXISTS worker_registry_projection_revisions (
+    workspace_id TEXT PRIMARY KEY,
+    revision INTEGER NOT NULL
+);
+CREATE TABLE IF NOT EXISTS 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
+);
+"#;
 
 const MIGRATIONS: &[Migration] = &[
     Migration {
@@ -93,6 +134,11 @@ const MIGRATIONS: &[Migration] = &[
         name: REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME,
         apply: migrate_worker_run_generation_v60_to_v61,
     },
+    Migration {
+        version: 62,
+        name: WORKER_REGISTRY_PROJECTION_MIGRATION_NAME,
+        apply: migrate_worker_registry_projection_v61_to_v62,
+    },
 ];
 
 #[derive(Clone, Copy)]
@@ -488,6 +534,34 @@ pub struct WorkerRegistryRecord {
     pub updated_at: String,
 }
 
+/// Runtime-backed observation joined to the durable Worker catalog.
+///
+/// The Runtime payload remains intact so protocol projection is lossless. Ordering
+/// metadata is persisted separately to reject stale reconnect snapshots and events.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WorkerRegistryObservationRecord {
+    pub worker: SubscriptionWorker,
+    pub availability: SubscriptionWorkerAvailability,
+    pub connection_generation: u64,
+    pub subject_revision: u64,
+    pub snapshot_revision: u64,
+    pub projection_revision: u64,
+    pub observed_at: String,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WorkerRegistryProjectionRecord {
+    pub registry: WorkerRegistryRecord,
+    pub resource_key: Option<String>,
+    pub observation: Option<WorkerRegistryObservationRecord>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct WorkerRegistryProjectionCommit {
+    pub revision: u64,
+    pub changed_workers: Vec<RuntimeWorkerRef>,
+}
+
 /// Durable authority describing which Runtime Worker another Runtime Worker may
 /// discover and control. Revoked grants remain as audit evidence but are never
 /// returned by active-grant queries.
@@ -1252,6 +1326,52 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         workspace_id: &str,
         limit: usize,
     ) -> Result<Vec<WorkerRegistryRecord>>;
+    fn worker_registry_projection_snapshot(
+        &self,
+        workspace_id: &str,
+        limit: usize,
+    ) -> Result<(u64, Vec<WorkerRegistryProjectionRecord>)>;
+    fn worker_registry_projection(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+    ) -> Result<Option<WorkerRegistryProjectionRecord>>;
+    fn reconcile_worker_registry_snapshot(
+        &self,
+        workspace_id: &str,
+        runtime_id: &str,
+        connection_generation: u64,
+        snapshot_revision: u64,
+        workers: &[SubscriptionWorker],
+        observed_at: &str,
+    ) -> Result<WorkerRegistryProjectionCommit>;
+    fn apply_worker_registry_observation(
+        &self,
+        workspace_id: &str,
+        runtime_id: &str,
+        connection_generation: u64,
+        worker: &SubscriptionWorker,
+        observed_at: &str,
+    ) -> Result<WorkerRegistryProjectionCommit>;
+    fn mark_worker_registry_observation_unavailable(
+        &self,
+        workspace_id: &str,
+        runtime_id: &str,
+        connection_generation: u64,
+        worker_id: Option<&str>,
+        subject_revision: Option<u64>,
+        observed_at: &str,
+    ) -> Result<WorkerRegistryProjectionCommit>;
+    fn publish_worker_registry_catalog_change(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+    ) -> Result<WorkerRegistryProjectionCommit>;
+    fn publish_worker_registry_removal(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+    ) -> Result<WorkerRegistryProjectionCommit>;
     fn update_worker_retention(
         &self,
         workspace_id: &str,
@@ -5534,6 +5654,265 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         })
     }
 
+    fn worker_registry_projection_snapshot(
+        &self,
+        workspace_id: &str,
+        limit: usize,
+    ) -> Result<(u64, Vec<WorkerRegistryProjectionRecord>)> {
+        self.with_conn(|conn| {
+            let revision = current_worker_projection_revision(conn, workspace_id)?;
+            let mut statement = conn.prepare(
+                "SELECT runtime_id, worker_id FROM worker_registry WHERE workspace_id = ?1 ORDER BY updated_at DESC LIMIT ?2",
+            )?;
+            let identities = statement.query_map(params![workspace_id, limit as i64], |row| {
+                Ok(RuntimeWorkerRef { runtime_id: row.get(0)?, worker_id: row.get(1)? })
+            })?.collect::<std::result::Result<Vec<_>, _>>()?;
+            let records = identities.iter()
+                .map(|worker| read_worker_registry_projection(conn, workspace_id, worker))
+                .collect::<Result<Vec<_>>>()?
+                .into_iter().flatten().collect();
+            Ok((revision, records))
+        })
+    }
+
+    fn worker_registry_projection(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+    ) -> Result<Option<WorkerRegistryProjectionRecord>> {
+        self.with_conn(|conn| read_worker_registry_projection(conn, workspace_id, worker))
+    }
+
+    fn reconcile_worker_registry_snapshot(
+        &self,
+        workspace_id: &str,
+        runtime_id: &str,
+        connection_generation: u64,
+        snapshot_revision: u64,
+        workers: &[SubscriptionWorker],
+        observed_at: &str,
+    ) -> Result<WorkerRegistryProjectionCommit> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let (current_generation, current_snapshot) =
+                worker_projection_cursor(&tx, workspace_id, runtime_id)?;
+            if connection_generation < current_generation
+                || (connection_generation == current_generation
+                    && snapshot_revision <= current_snapshot)
+            {
+                let revision = current_worker_projection_revision(&tx, workspace_id)?;
+                tx.commit()?;
+                return Ok(WorkerRegistryProjectionCommit {
+                    revision,
+                    changed_workers: Vec::new(),
+                });
+            }
+            upsert_worker_projection_cursor(
+                &tx,
+                workspace_id,
+                runtime_id,
+                connection_generation,
+                snapshot_revision,
+            )?;
+            let catalog = worker_registry_identities_for_runtime(&tx, workspace_id, runtime_id)?;
+            let runtime_workers = workers
+                .iter()
+                .map(|worker| (worker.worker_id.as_str(), worker))
+                .collect::<HashMap<_, _>>();
+            for worker in workers {
+                if !catalog
+                    .iter()
+                    .any(|item| item.worker_id == worker.worker_id.as_str())
+                {
+                    insert_worker_projection_orphan_diagnostic(
+                        &tx,
+                        workspace_id,
+                        runtime_id,
+                        worker.worker_id.as_str(),
+                        "runtime_worker_without_backend_catalog",
+                        observed_at,
+                    )?;
+                }
+            }
+            let mut changed_workers = Vec::new();
+            for identity in catalog {
+                let changed = if let Some(worker) = runtime_workers.get(identity.worker_id.as_str())
+                {
+                    upsert_worker_observation(
+                        &tx,
+                        workspace_id,
+                        runtime_id,
+                        connection_generation,
+                        snapshot_revision,
+                        worker,
+                        SubscriptionWorkerAvailability::Observed,
+                        observed_at,
+                    )?
+                } else {
+                    set_worker_observation_unavailable(
+                        &tx,
+                        workspace_id,
+                        &identity,
+                        connection_generation,
+                        None,
+                        snapshot_revision,
+                        observed_at,
+                    )?
+                };
+                if changed {
+                    changed_workers.push(identity);
+                }
+            }
+            let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?;
+            tx.commit()?;
+            Ok(WorkerRegistryProjectionCommit {
+                revision,
+                changed_workers,
+            })
+        })
+    }
+
+    fn apply_worker_registry_observation(
+        &self,
+        workspace_id: &str,
+        runtime_id: &str,
+        connection_generation: u64,
+        worker: &SubscriptionWorker,
+        observed_at: &str,
+    ) -> Result<WorkerRegistryProjectionCommit> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let identity = RuntimeWorkerRef {
+                runtime_id: runtime_id.to_string(),
+                worker_id: worker.worker_id.to_string(),
+            };
+            if !worker_registry_identity_exists(&tx, workspace_id, &identity)? {
+                insert_worker_projection_orphan_diagnostic(
+                    &tx,
+                    workspace_id,
+                    runtime_id,
+                    worker.worker_id.as_str(),
+                    "runtime_worker_without_backend_catalog",
+                    observed_at,
+                )?;
+                let revision = current_worker_projection_revision(&tx, workspace_id)?;
+                tx.commit()?;
+                return Ok(WorkerRegistryProjectionCommit {
+                    revision,
+                    changed_workers: Vec::new(),
+                });
+            }
+            let changed = upsert_worker_observation(
+                &tx,
+                workspace_id,
+                runtime_id,
+                connection_generation,
+                0,
+                worker,
+                SubscriptionWorkerAvailability::Observed,
+                observed_at,
+            )?;
+            let changed_workers = if changed { vec![identity] } else { Vec::new() };
+            let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?;
+            tx.commit()?;
+            Ok(WorkerRegistryProjectionCommit {
+                revision,
+                changed_workers,
+            })
+        })
+    }
+
+    fn mark_worker_registry_observation_unavailable(
+        &self,
+        workspace_id: &str,
+        runtime_id: &str,
+        connection_generation: u64,
+        worker_id: Option<&str>,
+        subject_revision: Option<u64>,
+        observed_at: &str,
+    ) -> Result<WorkerRegistryProjectionCommit> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let identities = if let Some(worker_id) = worker_id {
+                vec![RuntimeWorkerRef {
+                    runtime_id: runtime_id.to_string(),
+                    worker_id: worker_id.to_string(),
+                }]
+            } else {
+                worker_registry_identities_for_runtime(&tx, workspace_id, runtime_id)?
+            };
+            let mut changed_workers = Vec::new();
+            for identity in identities {
+                if !worker_registry_identity_exists(&tx, workspace_id, &identity)? {
+                    insert_worker_projection_orphan_diagnostic(
+                        &tx,
+                        workspace_id,
+                        runtime_id,
+                        identity.worker_id.as_str(),
+                        "runtime_worker_without_backend_catalog",
+                        observed_at,
+                    )?;
+                    continue;
+                }
+                if set_worker_observation_unavailable(
+                    &tx,
+                    workspace_id,
+                    &identity,
+                    connection_generation,
+                    subject_revision,
+                    0,
+                    observed_at,
+                )? {
+                    changed_workers.push(identity);
+                }
+            }
+            let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?;
+            tx.commit()?;
+            Ok(WorkerRegistryProjectionCommit {
+                revision,
+                changed_workers,
+            })
+        })
+    }
+
+    fn publish_worker_registry_catalog_change(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+    ) -> Result<WorkerRegistryProjectionCommit> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let changed_workers = if worker_registry_identity_exists(&tx, workspace_id, worker)? {
+                vec![worker.clone()]
+            } else {
+                Vec::new()
+            };
+            let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?;
+            tx.commit()?;
+            Ok(WorkerRegistryProjectionCommit {
+                revision,
+                changed_workers,
+            })
+        })
+    }
+
+    fn publish_worker_registry_removal(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+    ) -> Result<WorkerRegistryProjectionCommit> {
+        self.with_conn_mut(|conn| {
+            let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let changed_workers = vec![worker.clone()];
+            let revision = commit_worker_projection_changes(&tx, workspace_id, &changed_workers)?;
+            tx.commit()?;
+            Ok(WorkerRegistryProjectionCommit {
+                revision,
+                changed_workers,
+            })
+        })
+    }
+
     fn update_worker_retention(
         &self,
         workspace_id: &str,
@@ -8140,6 +8519,297 @@ fn read_worker_registry_record(row: &rusqlite::Row<'_>) -> rusqlite::Result<Work
     })
 }
 
+fn read_worker_registry_projection(
+    conn: &Connection,
+    workspace_id: &str,
+    worker: &RuntimeWorkerRef,
+) -> Result<Option<WorkerRegistryProjectionRecord>> {
+    let registry = conn
+        .query_row(
+            "SELECT workspace_id, runtime_id, worker_id, display_name, profile, retention_state, transcript_ref, session_ref, summary_ref, diagnostics_ref, created_at, updated_at FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3",
+            params![workspace_id, worker.runtime_id, worker.worker_id],
+            read_worker_registry_record,
+        )
+        .optional()?;
+    let Some(registry) = registry else {
+        return Ok(None);
+    };
+    let resource_key = conn
+        .query_row(
+            "SELECT resource_key FROM workspace_resource_keys WHERE workspace_id = ?1 AND resource_kind = 'worker' AND resource_id = ?2",
+            params![workspace_id, worker.worker_id],
+            |row| row.get(0),
+        )
+        .optional()?;
+    let observation = conn
+        .query_row(
+            "SELECT availability, worker_json, connection_generation, subject_revision, snapshot_revision, projection_revision, observed_at FROM worker_registry_observations WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3",
+            params![workspace_id, worker.runtime_id, worker.worker_id],
+            |row| {
+                let availability: String = row.get(0)?;
+                let worker_json: Option<String> = row.get(1)?;
+                Ok((availability, worker_json, row.get::<_, i64>(2)?, row.get::<_, i64>(3)?, row.get::<_, i64>(4)?, row.get::<_, i64>(5)?, row.get::<_, String>(6)?))
+            },
+        )
+        .optional()?;
+    let observation = observation
+        .map(
+            |(
+                availability,
+                worker_json,
+                generation,
+                subject_revision,
+                snapshot_revision,
+                projection_revision,
+                observed_at,
+            )|
+             -> Result<_> {
+                let availability = match availability.as_str() {
+                    "observed" => SubscriptionWorkerAvailability::Observed,
+                    "unavailable" => SubscriptionWorkerAvailability::Unavailable,
+                    other => {
+                        return Err(Error::InvalidInput(format!(
+                            "invalid Worker observation availability `{other}`"
+                        )));
+                    }
+                };
+                let mut observed_worker: SubscriptionWorker = if let Some(worker_json) = worker_json
+                {
+                    serde_json::from_str(&worker_json).map_err(|error| {
+                        Error::InvalidInput(format!("invalid stored Worker observation: {error}"))
+                    })?
+                } else {
+                    SubscriptionWorker {
+                        worker_id: SubscriptionWorkerId::new(worker.worker_id.to_string())
+                            .map_err(|error| Error::InvalidInput(error.to_string()))?,
+                        runtime_id: Some(worker.runtime_id.to_string()),
+                        resource_key: resource_key.clone(),
+                        availability,
+                        subject_revision: subject_revision.max(0) as u64,
+                        state: protocol::subscription::SubscriptionWorkerState::Stopped,
+                        worker_state: None,
+                        has_running_internal_workers: false,
+                        workspace_id: Some(workspace_id.to_string()),
+                        display_name: Some(registry.display_name.clone()),
+                        profile: registry.profile.clone(),
+                        repository_id: None,
+                        repository_key: None,
+                        working_directory_id: None,
+                    }
+                };
+                observed_worker.availability = availability;
+                observed_worker.resource_key = resource_key.clone();
+                Ok(WorkerRegistryObservationRecord {
+                    worker: observed_worker,
+                    availability,
+                    connection_generation: generation.max(0) as u64,
+                    subject_revision: subject_revision.max(0) as u64,
+                    snapshot_revision: snapshot_revision.max(0) as u64,
+                    projection_revision: projection_revision.max(0) as u64,
+                    observed_at,
+                })
+            },
+        )
+        .transpose()?;
+    Ok(Some(WorkerRegistryProjectionRecord {
+        registry,
+        resource_key,
+        observation,
+    }))
+}
+
+fn current_worker_projection_revision(conn: &Connection, workspace_id: &str) -> Result<u64> {
+    Ok(conn
+        .query_row(
+            "SELECT revision FROM worker_registry_projection_revisions WHERE workspace_id = ?1",
+            params![workspace_id],
+            |row| row.get::<_, i64>(0),
+        )
+        .optional()?
+        .unwrap_or(0)
+        .max(0) as u64)
+}
+
+fn worker_projection_cursor(
+    conn: &Connection,
+    workspace_id: &str,
+    runtime_id: &str,
+) -> Result<(u64, u64)> {
+    Ok(conn.query_row(
+        "SELECT connection_generation, snapshot_revision FROM worker_registry_projection_cursors WHERE workspace_id = ?1 AND runtime_id = ?2",
+        params![workspace_id, runtime_id],
+        |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
+    ).optional()?.map(|(generation, revision)| (generation.max(0) as u64, revision.max(0) as u64)).unwrap_or((0, 0)))
+}
+
+fn upsert_worker_projection_cursor(
+    conn: &Connection,
+    workspace_id: &str,
+    runtime_id: &str,
+    generation: u64,
+    snapshot_revision: u64,
+) -> Result<()> {
+    conn.execute(
+        "INSERT INTO worker_registry_projection_cursors (workspace_id, runtime_id, connection_generation, snapshot_revision) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(workspace_id, runtime_id) DO UPDATE SET connection_generation = excluded.connection_generation, snapshot_revision = excluded.snapshot_revision",
+        params![workspace_id, runtime_id, generation as i64, snapshot_revision as i64],
+    )?;
+    Ok(())
+}
+
+fn worker_registry_identities_for_runtime(
+    conn: &Connection,
+    workspace_id: &str,
+    runtime_id: &str,
+) -> Result<Vec<RuntimeWorkerRef>> {
+    let mut statement = conn.prepare("SELECT runtime_id, worker_id FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 ORDER BY worker_id")?;
+    Ok(statement
+        .query_map(params![workspace_id, runtime_id], |row| {
+            Ok(RuntimeWorkerRef {
+                runtime_id: row.get(0)?,
+                worker_id: row.get(1)?,
+            })
+        })?
+        .collect::<std::result::Result<Vec<_>, _>>()?)
+}
+
+fn worker_registry_identity_exists(
+    conn: &Connection,
+    workspace_id: &str,
+    worker: &RuntimeWorkerRef,
+) -> Result<bool> {
+    Ok(conn.query_row(
+        "SELECT 1 FROM worker_registry WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3",
+        params![workspace_id, worker.runtime_id, worker.worker_id],
+        |_| Ok(()),
+    ).optional()?.is_some())
+}
+
+fn insert_worker_projection_orphan_diagnostic(
+    conn: &Connection,
+    workspace_id: &str,
+    runtime_id: &str,
+    worker_id: &str,
+    reason: &str,
+    observed_at: &str,
+) -> Result<()> {
+    conn.execute(
+        "INSERT INTO worker_registry_projection_diagnostics (workspace_id, runtime_id, worker_id, reason, observed_at) VALUES (?1, ?2, ?3, ?4, ?5)",
+        params![workspace_id, runtime_id, worker_id, reason, observed_at],
+    )?;
+    Ok(())
+}
+
+fn upsert_worker_observation(
+    conn: &Connection,
+    workspace_id: &str,
+    runtime_id: &str,
+    connection_generation: u64,
+    snapshot_revision: u64,
+    worker: &SubscriptionWorker,
+    availability: SubscriptionWorkerAvailability,
+    observed_at: &str,
+) -> Result<bool> {
+    let existing = conn.query_row(
+        "SELECT availability, worker_json, connection_generation, subject_revision FROM worker_registry_observations WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3",
+        params![workspace_id, runtime_id, worker.worker_id.as_str()],
+        |row| Ok((row.get::<_, String>(0)?, row.get::<_, Option<String>>(1)?, row.get::<_, i64>(2)?, row.get::<_, i64>(3)?)),
+    ).optional()?;
+    if let Some((_, _, generation, revision)) = existing.as_ref() {
+        let generation = (*generation).max(0) as u64;
+        let revision = (*revision).max(0) as u64;
+        if connection_generation < generation
+            || (connection_generation == generation && worker.subject_revision <= revision)
+        {
+            return Ok(false);
+        }
+    }
+    let mut persisted_worker = worker.clone();
+    persisted_worker.runtime_id = Some(runtime_id.to_string());
+    persisted_worker.availability = availability;
+    let worker_json = serde_json::to_string(&persisted_worker).map_err(|error| {
+        Error::InvalidInput(format!("failed to encode Worker observation: {error}"))
+    })?;
+    let availability_text = match availability {
+        SubscriptionWorkerAvailability::Observed => "observed",
+        SubscriptionWorkerAvailability::Unavailable => "unavailable",
+    };
+    let changed = existing
+        .as_ref()
+        .is_none_or(|(old_availability, old_json, _, _)| {
+            old_availability != availability_text
+                || old_json.as_deref() != Some(worker_json.as_str())
+        });
+    conn.execute(
+        "INSERT INTO worker_registry_observations (workspace_id, runtime_id, worker_id, availability, worker_json, connection_generation, subject_revision, snapshot_revision, projection_revision, observed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 0, ?9) ON CONFLICT(workspace_id, runtime_id, worker_id) DO UPDATE SET availability = excluded.availability, worker_json = excluded.worker_json, connection_generation = excluded.connection_generation, subject_revision = excluded.subject_revision, snapshot_revision = MAX(worker_registry_observations.snapshot_revision, excluded.snapshot_revision), observed_at = excluded.observed_at",
+        params![workspace_id, runtime_id, worker.worker_id.as_str(), availability_text, worker_json, connection_generation as i64, worker.subject_revision as i64, snapshot_revision as i64, observed_at],
+    )?;
+    Ok(changed)
+}
+
+fn set_worker_observation_unavailable(
+    conn: &Connection,
+    workspace_id: &str,
+    worker: &RuntimeWorkerRef,
+    connection_generation: u64,
+    subject_revision: Option<u64>,
+    snapshot_revision: u64,
+    observed_at: &str,
+) -> Result<bool> {
+    let existing = conn.query_row(
+        "SELECT availability, connection_generation, subject_revision, worker_json FROM worker_registry_observations WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3",
+        params![workspace_id, worker.runtime_id, worker.worker_id],
+        |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?, row.get::<_, i64>(2)?, row.get::<_, Option<String>>(3)?)),
+    ).optional()?;
+    if let Some((availability, generation, revision, _)) = existing.as_ref() {
+        let generation = (*generation).max(0) as u64;
+        let revision = (*revision).max(0) as u64;
+        if connection_generation < generation
+            || (connection_generation == generation
+                && subject_revision.is_some_and(|incoming| incoming <= revision))
+        {
+            return Ok(false);
+        }
+        if availability == "unavailable"
+            && subject_revision.is_none()
+            && connection_generation == generation
+        {
+            return Ok(false);
+        }
+    }
+    let revision = subject_revision
+        .or_else(|| existing.as_ref().map(|item| item.2.max(0) as u64))
+        .unwrap_or(0);
+    let worker_json = existing.and_then(|item| item.3);
+    conn.execute(
+        "INSERT INTO worker_registry_observations (workspace_id, runtime_id, worker_id, availability, worker_json, connection_generation, subject_revision, snapshot_revision, projection_revision, observed_at) VALUES (?1, ?2, ?3, 'unavailable', ?4, ?5, ?6, ?7, 0, ?8) ON CONFLICT(workspace_id, runtime_id, worker_id) DO UPDATE SET availability = 'unavailable', connection_generation = excluded.connection_generation, subject_revision = excluded.subject_revision, snapshot_revision = MAX(worker_registry_observations.snapshot_revision, excluded.snapshot_revision), observed_at = excluded.observed_at",
+        params![workspace_id, worker.runtime_id, worker.worker_id, worker_json, connection_generation as i64, revision as i64, snapshot_revision as i64, observed_at],
+    )?;
+    Ok(true)
+}
+
+fn commit_worker_projection_changes(
+    conn: &Connection,
+    workspace_id: &str,
+    changed_workers: &[RuntimeWorkerRef],
+) -> Result<u64> {
+    let current = current_worker_projection_revision(conn, workspace_id)?;
+    if changed_workers.is_empty() {
+        return Ok(current);
+    }
+    let revision = current.saturating_add(1);
+    conn.execute(
+        "INSERT INTO worker_registry_projection_revisions (workspace_id, revision) VALUES (?1, ?2) ON CONFLICT(workspace_id) DO UPDATE SET revision = excluded.revision",
+        params![workspace_id, revision as i64],
+    )?;
+    for worker in changed_workers {
+        conn.execute(
+            "UPDATE worker_registry_observations SET projection_revision = ?1 WHERE workspace_id = ?2 AND runtime_id = ?3 AND worker_id = ?4",
+            params![revision as i64, workspace_id, worker.runtime_id, worker.worker_id],
+        )?;
+    }
+    Ok(revision)
+}
+
 fn read_worker_control_grant_record(
     row: &rusqlite::Row<'_>,
 ) -> rusqlite::Result<WorkerControlGrantRecord> {
@@ -9784,11 +10454,22 @@ fn migrate_runtime_removal_operations_v59_to_v60(conn: &Connection) -> Result<()
 fn migrate_worker_run_generation_v60_to_v61(conn: &Connection) -> Result<()> {
     let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
     tx.execute_batch("ALTER TABLE worker_removal_operations DROP COLUMN run_generation;")?;
+    tx.execute(
+        "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
+        params![61, REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME],
+    )?;
+    tx.commit()?;
+    Ok(())
+}
+
+fn migrate_worker_registry_projection_v61_to_v62(conn: &Connection) -> Result<()> {
+    let tx = rusqlite::Transaction::new_unchecked(conn, TransactionBehavior::Exclusive)?;
+    tx.execute_batch(WORKER_REGISTRY_PROJECTION_SCHEMA)?;
     tx.execute(
         "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
         params![
             LATEST_SCHEMA_VERSION,
-            REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME
+            WORKER_REGISTRY_PROJECTION_MIGRATION_NAME
         ],
     )?;
     tx.commit()?;
@@ -11166,6 +11847,10 @@ mod tests {
                     version: 61,
                     name: REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME.to_string(),
                 },
+                WorkspaceSchemaMigrationStep {
+                    version: 62,
+                    name: WORKER_REGISTRY_PROJECTION_MIGRATION_NAME.to_string(),
+                },
             ]
         );
 
@@ -11216,6 +11901,7 @@ mod tests {
                             61,
                             REMOVE_WORKER_RUN_GENERATION_MIGRATION_NAME.to_string(),
                         ),
+                        (62, WORKER_REGISTRY_PROJECTION_MIGRATION_NAME.to_string()),
                     ]
                 );
                 assert!(!table_exists(conn, "trusted_runtime_records")?);
@@ -11286,7 +11972,7 @@ mod tests {
                 .iter()
                 .map(|migration| migration.version)
                 .collect::<Vec<_>>(),
-            vec![52, 53, 54, 55, 56, 57, 58, 59, 60, 61]
+            vec![52, 53, 54, 55, 56, 57, 58, 59, 60, 61, 62]
         );
         SqliteWorkspaceStore::migrate_database(&path).unwrap();
         let conn = Connection::open(&path).unwrap();
@@ -11294,7 +11980,7 @@ mod tests {
             current_schema_version(&conn).unwrap(),
             LATEST_SCHEMA_VERSION
         );
-        assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 12);
+        assert_eq!(workspace_schema_migration_history(&conn).unwrap().len(), 13);
     }
 
     #[test]
@@ -14143,6 +14829,189 @@ INSERT INTO worker_registry (
         );
     }
 
+    #[tokio::test]
+    async fn worker_projection_reconciles_snapshots_and_preserves_unavailable_lifecycle() {
+        let temp = tempfile::tempdir().unwrap();
+        let store = SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: "local-dev".to_string(),
+                owner_account_id: "owner-account".to_string(),
+                display_name: "Local Dev".to_string(),
+                state: "active".to_string(),
+                created_at: "1".to_string(),
+                updated_at: "1".to_string(),
+            })
+            .await
+            .unwrap();
+        let catalog = WorkerRegistryRecord {
+            workspace_id: "local-dev".to_string(),
+            worker: RuntimeWorkerRef::new("embedded", "known"),
+            display_name: "Known Worker".to_string(),
+            profile: None,
+            retention_state: "normal".to_string(),
+            transcript_ref: None,
+            session_ref: None,
+            summary_ref: None,
+            diagnostics_ref: None,
+            created_at: "1".to_string(),
+            updated_at: "1".to_string(),
+        };
+        store.upsert_worker_registry(&catalog).unwrap();
+        assert_eq!(
+            store
+                .resource_key("local-dev", WorkspaceResourceKind::Worker, "known")
+                .unwrap()
+                .as_deref(),
+            Some("W-1")
+        );
+        assert_eq!(
+            store
+                .worker_registry_projection("local-dev", &catalog.worker)
+                .unwrap()
+                .unwrap()
+                .resource_key
+                .as_deref(),
+            Some("W-1")
+        );
+        let observed = SubscriptionWorker {
+            worker_id: SubscriptionWorkerId::new("known").unwrap(),
+            runtime_id: Some("embedded".to_string()),
+            resource_key: None,
+            availability: SubscriptionWorkerAvailability::Observed,
+            subject_revision: 4,
+            worker_state: None,
+            state: protocol::subscription::SubscriptionWorkerState::Running,
+            has_running_internal_workers: true,
+            workspace_id: Some("local-dev".to_string()),
+            display_name: Some("Known Worker".to_string()),
+            profile: None,
+            repository_id: None,
+            repository_key: None,
+            working_directory_id: None,
+        };
+        let orphan = SubscriptionWorker {
+            worker_id: SubscriptionWorkerId::new("orphan").unwrap(),
+            ..observed.clone()
+        };
+        let first = store
+            .reconcile_worker_registry_snapshot(
+                "local-dev",
+                "embedded",
+                1,
+                10,
+                &[observed.clone(), orphan],
+                "2",
+            )
+            .unwrap();
+        assert_eq!(first.changed_workers, vec![catalog.worker.clone()]);
+        let (_, records) = store
+            .worker_registry_projection_snapshot("local-dev", 10)
+            .unwrap();
+        assert_eq!(
+            records.len(),
+            1,
+            "orphan Runtime rows must not create catalog rows"
+        );
+        let observation = records[0].observation.as_ref().unwrap();
+        assert_eq!(
+            observation.availability,
+            SubscriptionWorkerAvailability::Observed
+        );
+        assert_eq!(
+            observation.worker.state,
+            protocol::subscription::SubscriptionWorkerState::Running
+        );
+
+        let unavailable = store
+            .reconcile_worker_registry_snapshot("local-dev", "embedded", 1, 11, &[], "3")
+            .unwrap();
+        assert_eq!(unavailable.changed_workers, vec![catalog.worker.clone()]);
+        let projection = store
+            .worker_registry_projection("local-dev", &catalog.worker)
+            .unwrap()
+            .unwrap();
+        let observation = projection.observation.unwrap();
+        assert_eq!(
+            observation.availability,
+            SubscriptionWorkerAvailability::Unavailable
+        );
+        assert_eq!(
+            observation.worker.state,
+            protocol::subscription::SubscriptionWorkerState::Running,
+            "unavailable must retain the last lifecycle observation rather than becoming stopped"
+        );
+
+        let duplicate = store
+            .reconcile_worker_registry_snapshot("local-dev", "embedded", 1, 11, &[], "4")
+            .unwrap();
+        assert!(duplicate.changed_workers.is_empty());
+    }
+
+    #[tokio::test]
+    async fn late_runtime_event_does_not_recreate_retention_deleted_catalog_worker() {
+        let temp = tempfile::tempdir().unwrap();
+        let store = SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: "local-dev".to_string(),
+                owner_account_id: "owner-account".to_string(),
+                display_name: "Local Dev".to_string(),
+                state: "active".to_string(),
+                created_at: "1".to_string(),
+                updated_at: "1".to_string(),
+            })
+            .await
+            .unwrap();
+        let worker_ref = RuntimeWorkerRef::new("embedded", "removed");
+        store
+            .upsert_worker_registry(&WorkerRegistryRecord {
+                workspace_id: "local-dev".to_string(),
+                worker: worker_ref.clone(),
+                display_name: "Removed Worker".to_string(),
+                profile: None,
+                retention_state: "normal".to_string(),
+                transcript_ref: None,
+                session_ref: None,
+                summary_ref: None,
+                diagnostics_ref: None,
+                created_at: "1".to_string(),
+                updated_at: "1".to_string(),
+            })
+            .unwrap();
+        assert!(
+            store
+                .delete_worker_registry("local-dev", &worker_ref)
+                .unwrap()
+        );
+        let event = SubscriptionWorker {
+            worker_id: SubscriptionWorkerId::new("removed").unwrap(),
+            runtime_id: Some("embedded".to_string()),
+            resource_key: None,
+            availability: SubscriptionWorkerAvailability::Observed,
+            subject_revision: 99,
+            worker_state: None,
+            state: protocol::subscription::SubscriptionWorkerState::Running,
+            has_running_internal_workers: false,
+            workspace_id: Some("local-dev".to_string()),
+            display_name: None,
+            profile: None,
+            repository_id: None,
+            repository_key: None,
+            working_directory_id: None,
+        };
+        let commit = store
+            .apply_worker_registry_observation("local-dev", "embedded", 2, &event, "2")
+            .unwrap();
+        assert!(commit.changed_workers.is_empty());
+        assert!(
+            store
+                .worker_registry_projection("local-dev", &worker_ref)
+                .unwrap()
+                .is_none()
+        );
+    }
+
     #[tokio::test]
     async fn worker_control_grants_are_idempotent_scoped_and_support_internal_invalidation() {
         let dir = tempfile::tempdir().unwrap();

crates/workspace-server/src/worker_projection.rs

@@ -0,0 +1,276 @@
+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)?;
+        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,10 @@ 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_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(project_registry_worker)
-        });
+        .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,184 +335,102 @@ 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(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,
+    record: crate::store::WorkerRegistryProjectionRecord,
-    workers: &mut HashMap<String, BTreeMap<String, SubscriptionWorker>>,
+) -> Option<SubscriptionWorker> {
-    runtime_id: &str,
+    let runtime_id = record.registry.worker.runtime_id.clone();
-    snapshot: SubscriptionSnapshot,
+    let mut worker = if let Some(observation) = record.observation {
-) {
+        let mut worker = observation.worker;
-    let SubscriptionSnapshot::Workers {
+        worker.availability = observation.availability;
-        workers: snapshot_workers,
+        worker.subject_revision = observation.projection_revision;
-    } = snapshot
+        worker
-    else {
+    } else {
-        return;
+        SubscriptionWorker {
-    };
+            worker_id: protocol::subscription::SubscriptionWorkerId::new(
-    let mut projected = BTreeMap::new();
+                record.registry.worker.worker_id.clone(),
-    for mut worker in snapshot_workers {
+            )
-        worker.runtime_id = Some(runtime_id.to_string());
+            .ok()?,
-        let Ok(Some(resource_key)) = api.store.resource_key(
+            runtime_id: Some(runtime_id.clone()),
-            &api.config.workspace_id,
+            resource_key: record.resource_key.clone(),
-            WorkspaceResourceKind::Worker,
+            availability: protocol::subscription::SubscriptionWorkerAvailability::Unavailable,
-            worker.worker_id.as_str(),
+            subject_revision: 0,
-        ) else {
+            worker_state: None,
-            continue;
+            state: protocol::subscription::SubscriptionWorkerState::Stopped,
-        };
+            has_running_internal_workers: false,
-        worker.resource_key = Some(resource_key);
+            workspace_id: Some(record.registry.workspace_id.clone()),
-        if !project_repository_key(api, &mut worker) {
+            display_name: Some(record.registry.display_name.clone()),
-            continue;
+            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);
-    workers.insert(runtime_id.to_string(), projected);
+    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;
+    Some(worker)
 }
 
-fn project_repository_key(api: &WorkspaceApi, worker: &mut SubscriptionWorker) -> bool {
+fn sort_workers(workers: &mut [SubscriptionWorker]) {
-    let Some(repository_id) = worker.repository_id.take() else {
+    workers.sort_by(|left, right| {
-        return true;
+        left.runtime_id
-    };
+            .cmp(&right.runtime_id)
-    let Ok(Some(repository)) = api
+            .then_with(|| left.worker_id.cmp(&right.worker_id))
-        .store
+    });
-        .get_repository(&api.config.workspace_id, &repository_id)
-    else {
-        return false;
-    };
-    worker.repository_key = Some(repository.repository_key);
-    true
 }
 
 async fn send_event(
@@ -586,23 +470,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,