chore: merge develop into hare/develop

crates/workspace-server/Cargo.toml

@@ -51,6 +51,7 @@ webauthn-rs = { workspace = true }
 
 [dev-dependencies]
 agen.workspace = true
+session-store.workspace = true
 tempfile.workspace = true
 tools.workspace = true
 yoi = { path = "../yoi" }

crates/workspace-server/src/authority.rs

@@ -820,11 +820,9 @@ impl SqliteWorkspaceAuthority {
                     .map(|worker| (assignment, worker))
             })
             .map(|(assignment, worker)| {
-                let worker_resource_key = self.store.resource_key(
-                    &self.workspace_id,
-                    WorkspaceResourceKind::Worker,
-                    &worker.worker_id,
-                )?;
+                let worker_resource_key = self
+                    .store
+                    .worker_resource_key(&self.workspace_id, &worker)?;
                 Ok::<_, Error>(TicketAssignmentSummary {
                     assignment_id: assignment.assignment_id.clone(),
                     runtime_id: worker.runtime_id,

crates/workspace-server/src/hosts.rs

@@ -534,6 +534,9 @@ pub struct WorkerSpawnRequest {
     /// Backend-authored immutable Workspace Memory settings snapshot.
     #[serde(skip, default)]
     pub resolved_memory_settings: Option<manifest::WorkspaceMemorySettingsSnapshot>,
+    /// Backend-attested subjektiv singleton attachment. Browser/model input cannot set it.
+    #[serde(skip, default)]
+    pub resolved_subjektiv_attached: bool,
     /// Backend-owned feature enablement; client input cannot set it.
     #[serde(skip, default)]
     pub resolved_worker_observation_enabled: bool,
@@ -1217,6 +1220,26 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
         })
     }
 
+    fn session_public_search(
+        &self,
+        _request: runtime_api::SessionPublicSearchRequest,
+    ) -> Result<runtime_api::SessionPublicSearchAvailability, String> {
+        Ok(runtime_api::SessionPublicSearchAvailability::Unavailable {
+            reason: runtime_api::SessionPublicUnavailableReason::StorageUnavailable,
+            message: "Session public search is not supported by this Runtime".to_string(),
+        })
+    }
+
+    fn session_public_read(
+        &self,
+        _request: runtime_api::SessionPublicReadRequest,
+    ) -> Result<runtime_api::SessionPublicReadAvailability, String> {
+        Ok(runtime_api::SessionPublicReadAvailability::Unavailable {
+            reason: runtime_api::SessionPublicUnavailableReason::StorageUnavailable,
+            message: "Session public read is not supported by this Runtime".to_string(),
+        })
+    }
+
     fn observation_source(
         &self,
         _worker_id: &str,
@@ -2073,6 +2096,36 @@ impl RuntimeRegistry {
             })
     }
 
+    pub fn session_public_search(
+        &self,
+        runtime_id: &str,
+        request: runtime_api::SessionPublicSearchRequest,
+    ) -> Result<runtime_api::SessionPublicSearchAvailability, RuntimeRegistryError> {
+        validate_backend_identifier("runtime_id", runtime_id)?;
+        self.runtime(runtime_id)?
+            .session_public_search(request)
+            .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: runtime_id.to_string(),
+                code: "session_public_search_failed".to_string(),
+                message,
+            })
+    }
+
+    pub fn session_public_read(
+        &self,
+        runtime_id: &str,
+        request: runtime_api::SessionPublicReadRequest,
+    ) -> Result<runtime_api::SessionPublicReadAvailability, RuntimeRegistryError> {
+        validate_backend_identifier("runtime_id", runtime_id)?;
+        self.runtime(runtime_id)?
+            .session_public_read(request)
+            .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: runtime_id.to_string(),
+                code: "session_public_read_failed".to_string(),
+                message,
+            })
+    }
+
     pub fn observation_source(
         &self,
         worker: &RuntimeWorkerRef,
@@ -2964,6 +3017,36 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
             .map_err(|error| error.to_string())
     }
 
+    fn session_public_search(
+        &self,
+        request: runtime_api::SessionPublicSearchRequest,
+    ) -> Result<runtime_api::SessionPublicSearchAvailability, String> {
+        if request.workspace_id != self.workspace_id {
+            return Err(
+                "Session public search Workspace does not match embedded Runtime".to_string(),
+            );
+        }
+        let scope = RuntimeWorkspaceScope::new(&request.workspace_id, "embedded-backend");
+        self.runtime
+            .session_public_search_scoped(&scope, &request)
+            .map_err(|error| error.to_string())
+    }
+
+    fn session_public_read(
+        &self,
+        request: runtime_api::SessionPublicReadRequest,
+    ) -> Result<runtime_api::SessionPublicReadAvailability, String> {
+        if request.workspace_id != self.workspace_id {
+            return Err(
+                "Session public read Workspace does not match embedded Runtime".to_string(),
+            );
+        }
+        let scope = RuntimeWorkspaceScope::new(&request.workspace_id, "embedded-backend");
+        self.runtime
+            .session_public_read_scoped(&scope, &request)
+            .map_err(|error| error.to_string())
+    }
+
     fn observation_source(
         &self,
         worker_id: &str,
@@ -3669,6 +3752,9 @@ fn workspace_runtime_operation(method: &str, path_and_query: &str) -> &'static s
     if path.contains("/retention/") || (path.starts_with("/v1/workers/") && method == "DELETE") {
         return "workers:delete";
     }
+    if path.starts_with("/v1/session-public/") && method == "POST" {
+        return "workers:read";
+    }
     if path.starts_with("/v1/workers/") && method == "GET" {
         return "workers:read";
     }
@@ -5007,6 +5093,30 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
         .map_err(|diagnostic| diagnostic.message)
     }
 
+    fn session_public_search(
+        &self,
+        request: runtime_api::SessionPublicSearchRequest,
+    ) -> Result<runtime_api::SessionPublicSearchAvailability, String> {
+        self.run_runtime_api(
+            self.request_timeout,
+            MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
+            move |client| async move { client.session_public_search(request).await },
+        )
+        .map_err(|diagnostic| diagnostic.message)
+    }
+
+    fn session_public_read(
+        &self,
+        request: runtime_api::SessionPublicReadRequest,
+    ) -> Result<runtime_api::SessionPublicReadAvailability, String> {
+        self.run_runtime_api(
+            self.request_timeout,
+            MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
+            move |client| async move { client.session_public_read(request).await },
+        )
+        .map_err(|diagnostic| diagnostic.message)
+    }
+
     fn observation_source(
         &self,
         worker_id: &str,
@@ -5246,6 +5356,7 @@ fn runtime_create_worker_request(
         worker_observation_grants: request.resolved_worker_observation_grants.clone(),
         workspace_api: Some(workspace_api),
         memory_settings: request.resolved_memory_settings.clone(),
+        subjektiv_attached: request.resolved_subjektiv_attached,
     }
 }
 
@@ -5370,6 +5481,7 @@ fn embedded_profile_label(profile: &ProfileSelector) -> Option<String> {
 
 const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation";
 pub(crate) const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation";
+pub(crate) const SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX: &str = "subjektiv-consolidation:";
 const WORKSPACE_ORCHESTRATOR_PROFILE: &str = "orchestrator";
 pub(crate) const WORKSPACE_ORCHESTRATOR_SINGLETON_KEY: &str = "workspace-orchestrator";
 
@@ -5377,7 +5489,7 @@ pub(crate) fn is_reserved_internal_worker_singleton_key(key: &str) -> bool {
     matches!(
         key,
         MEMORY_CONSOLIDATION_SINGLETON_KEY | WORKSPACE_ORCHESTRATOR_SINGLETON_KEY
-    )
+    ) || key.starts_with(SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX)
 }
 
 struct WorkerDisplayMetadata {
@@ -7030,6 +7142,7 @@ mod tests {
             resolved_control_operation: None,
             resolved_workspace_api: Some(test_workspace_api()),
             resolved_memory_settings: Some(test_memory_settings()),
+            resolved_subjektiv_attached: false,
         }
     }
 
@@ -7318,6 +7431,7 @@ mod tests {
                     resolved_control_operation: None,
                     resolved_workspace_api: Some(test_workspace_api()),
                     resolved_memory_settings: Some(test_memory_settings()),
+                    resolved_subjektiv_attached: false,
                 },
             )
             .unwrap();
@@ -7420,6 +7534,7 @@ mod tests {
                     resolved_control_operation: None,
                     resolved_workspace_api: Some(test_workspace_api()),
                     resolved_memory_settings: Some(test_memory_settings()),
+                    resolved_subjektiv_attached: false,
                 },
             )
             .unwrap();
@@ -7460,6 +7575,7 @@ mod tests {
                     resolved_control_operation: None,
                     resolved_workspace_api: Some(test_workspace_api()),
                     resolved_memory_settings: Some(test_memory_settings()),
+                    resolved_subjektiv_attached: false,
                 },
             )
             .unwrap();

crates/workspace-server/src/latest_schema.sql

@@ -753,11 +753,11 @@ CREATE TABLE "worker_control_grants" (
             created_at TEXT NOT NULL,
             revoked_at TEXT,
             PRIMARY KEY (workspace_id, grant_id),
-            UNIQUE (workspace_id, controller_worker_id, operation_id),
-            FOREIGN KEY (workspace_id, controller_worker_id)
-                REFERENCES "worker_registry"(workspace_id, worker_id) ON DELETE CASCADE,
-            FOREIGN KEY (workspace_id, subject_worker_id)
-                REFERENCES "worker_registry"(workspace_id, worker_id) ON DELETE CASCADE
+            UNIQUE (workspace_id, controller_runtime_id, controller_worker_id, operation_id),
+            FOREIGN KEY (workspace_id, controller_runtime_id, controller_worker_id)
+                REFERENCES "worker_registry"(workspace_id, runtime_id, worker_id) ON DELETE CASCADE,
+            FOREIGN KEY (workspace_id, subject_runtime_id, subject_worker_id)
+                REFERENCES "worker_registry"(workspace_id, runtime_id, worker_id) ON DELETE CASCADE
         );
 CREATE TABLE worker_create_reservations (
             workspace_id TEXT NOT NULL,
@@ -832,7 +832,7 @@ CREATE TABLE "worker_registry" (
             diagnostics_ref TEXT,
             created_at TEXT NOT NULL,
             updated_at TEXT NOT NULL,
-            PRIMARY KEY (workspace_id, worker_id),
+            PRIMARY KEY (workspace_id, runtime_id, worker_id),
             FOREIGN KEY (workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE
         );
 CREATE TABLE backend_jobs (
@@ -914,31 +914,15 @@ CREATE TABLE worker_registry_observations (
     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
+    FOREIGN KEY (workspace_id, runtime_id, worker_id)
+        REFERENCES worker_registry(workspace_id, runtime_id, worker_id) ON DELETE CASCADE
 );
 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 (
@@ -973,6 +957,17 @@ CREATE TABLE worker_session_archives (
         committed_at TEXT NOT NULL, expires_at TEXT,
         FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
         FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id));
+CREATE TABLE worker_session_archive_observe_grants (
+        workspace_id TEXT NOT NULL, archive_id TEXT NOT NULL,
+        controller_runtime_id TEXT NOT NULL, controller_worker_id TEXT NOT NULL,
+        subject_runtime_id TEXT NOT NULL, subject_worker_id TEXT NOT NULL,
+        source_grant_id TEXT NOT NULL, granted_at TEXT NOT NULL, revoked_at TEXT,
+        PRIMARY KEY(workspace_id,archive_id,controller_runtime_id,controller_worker_id,source_grant_id),
+        FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE,
+        FOREIGN KEY(archive_id) REFERENCES worker_session_archives(archive_id) ON DELETE CASCADE);
+CREATE INDEX worker_session_archive_observe_grants_controller
+        ON worker_session_archive_observe_grants(
+            workspace_id,controller_runtime_id,controller_worker_id,archive_id);
 CREATE TABLE worker_tombstones (
         workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL, worker_id TEXT NOT NULL,
         display_name TEXT NOT NULL, profile TEXT, worker_created_at TEXT NOT NULL, removed_at TEXT NOT NULL,
@@ -998,9 +993,9 @@ CREATE TABLE "worker_workdir_links" (
             alias TEXT NOT NULL,
             linked_at TEXT NOT NULL,
             unlinked_at TEXT,
-            PRIMARY KEY (workspace_id, worker_id, workdir_id, alias),
-            FOREIGN KEY (workspace_id, worker_id)
-                REFERENCES "worker_registry"(workspace_id, worker_id) ON DELETE CASCADE,
+            PRIMARY KEY (workspace_id, runtime_id, worker_id, workdir_id, alias),
+            FOREIGN KEY (workspace_id, runtime_id, worker_id)
+                REFERENCES "worker_registry"(workspace_id, runtime_id, worker_id) ON DELETE CASCADE,
             FOREIGN KEY (workspace_id, workdir_id)
                 REFERENCES workdir_registry(workspace_id, workdir_id) ON DELETE CASCADE
         );
@@ -1217,11 +1212,11 @@ CREATE UNIQUE INDEX ux_worker_workdir_attachment_reservation_id
     ON worker_workdir_attachment_reservations(workspace_id, reservation_id);
 CREATE INDEX worker_control_grants_controller
             ON worker_control_grants(
-                workspace_id, controller_worker_id, revoked_at
+                workspace_id, controller_runtime_id, controller_worker_id, revoked_at
             );
 CREATE INDEX worker_control_grants_subject
             ON worker_control_grants(
-                workspace_id, subject_worker_id, revoked_at
+                workspace_id, subject_runtime_id, subject_worker_id, revoked_at
             );
 CREATE INDEX worker_create_reservations_worker
             ON worker_create_reservations(workspace_id, worker_id);
@@ -1232,7 +1227,7 @@ CREATE UNIQUE INDEX worker_workdir_links_active_workdir_unique
             ON worker_workdir_links(workspace_id, workdir_id)
             WHERE unlinked_at IS NULL;
 CREATE UNIQUE INDEX worker_workdir_links_active_alias_unique
-            ON worker_workdir_links(workspace_id, worker_id, alias)
+            ON worker_workdir_links(workspace_id, runtime_id, worker_id, alias)
             WHERE unlinked_at IS NULL;
 CREATE INDEX worker_workdir_links_workdir
             ON worker_workdir_links(workspace_id, workdir_id);

crates/workspace-server/src/lib.rs

@@ -112,6 +112,8 @@ pub enum Error {
     },
     #[error("unknown local repository `{0}`")]
     UnknownRepository(String),
+    #[error("subjektiv subject `{0}` was not found in this Workspace")]
+    SubjektivSubjectNotFound(String),
     #[error(
         "Merge Request reopen requires an authenticated Browser session and explicit confirmation"
     )]

crates/workspace-server/src/memory_backend.rs

@@ -1,8 +1,9 @@
 use memory::backend::{
     MemoryBackendAckOutput, MemoryBackendOperation, MemoryBackendOperationResult,
     MemoryDocumentReadOperation, MemoryDocumentUpdateOperation, MemoryQueryOperation,
-    MemoryStagingCloseAction, MemoryStagingCloseOperation, MemoryStagingListOperation,
-    MemoryStagingReadOperation, MemoryStagingWriteOutput, MemoryToolOutput,
+    MemoryResidentSummaryAvailability, MemoryResidentSummaryOutput, MemoryStagingCloseAction,
+    MemoryStagingCloseOperation, MemoryStagingListOperation, MemoryStagingReadOperation,
+    MemoryStagingWriteOutput, MemoryToolOutput,
 };
 use memory::extract::{ExtractedCandidate, StagingEvidence, StagingRecord};
 use memory::schema::{SourceEvidenceRef, SourceRef};
@@ -32,7 +33,7 @@ pub fn execute_memory_backend_operation_with_authority<A: MemoryAuthority>(
                 .map(MemoryBackendOperationResult::ToolOutput)
         }
         MemoryBackendOperation::ResidentSummary(_operation) => Ok(
-            MemoryBackendOperationResult::ToolOutput(execute_resident_summary(authority)?),
+            MemoryBackendOperationResult::ResidentSummary(execute_resident_summary(authority)?),
         ),
         MemoryBackendOperation::AppendAudit(_operation) => Ok(
             MemoryBackendOperationResult::Acknowledged(MemoryBackendAckOutput {
@@ -88,16 +89,18 @@ pub fn execute_memory_backend_operation_with_authority<A: MemoryAuthority>(
     }
 }
 
-fn execute_resident_summary<A: MemoryAuthority>(authority: &A) -> Result<MemoryToolOutput> {
+fn execute_resident_summary<A: MemoryAuthority>(
+    authority: &A,
+) -> Result<MemoryResidentSummaryOutput> {
     let document = authority.ensure_memory_document()?;
     if document.body_md.trim().is_empty() {
-        Ok(MemoryToolOutput {
-            summary: "resident memory summary unavailable".to_string(),
+        Ok(MemoryResidentSummaryOutput {
+            availability: MemoryResidentSummaryAvailability::Unavailable,
             content: None,
         })
     } else {
-        Ok(MemoryToolOutput {
-            summary: "resident memory document collected".to_string(),
+        Ok(MemoryResidentSummaryOutput {
+            availability: MemoryResidentSummaryAvailability::Ready,
             content: Some(document.body_md),
         })
     }

crates/workspace-server/src/retention.rs

@@ -132,6 +132,21 @@ pub struct WorkerTombstone {
     pub operation_id: String,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct WorkerSessionArchiveRecord {
+    pub archive_id: String,
+    pub workspace_id: String,
+    pub worker: RuntimeWorkerRef,
+    pub session_id: String,
+    pub checksum_sha256: String,
+    pub content_bytes: u64,
+    pub policy_id: String,
+    pub policy_revision: u64,
+    pub operation_id: String,
+    pub committed_at: String,
+    pub expires_at: Option<String>,
+}
+
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
 pub struct WorkerOrphanDiagnostic {
     pub diagnostic_id: String,
@@ -454,6 +469,51 @@ impl SqliteWorkspaceStore {
                 if plan.archive_id.as_deref()!=Some(&a.archive_id)||a.workspace_id!=workspace_id||a.source_runtime_id!=plan.worker.runtime_id||a.source_worker_id.to_string()!=plan.worker.worker_id||a.policy_id!=plan.policy_id||a.policy_revision!=plan.policy_revision{return Err(StoreError::InvalidInput("archive manifest mismatch".into()));}
                 let expires_at=match plan.archive_retention { ArchiveRetention::Forever=>None, ArchiveRetention::ForSeconds{seconds}=>{let seconds=i64::try_from(seconds).map_err(|_|StoreError::InvalidInput("archive retention deadline overflow".into()))?;Some((Utc::now()+chrono::Duration::seconds(seconds)).to_rfc3339())} };
                 tx.execute("INSERT OR IGNORE INTO worker_session_archives(archive_id,workspace_id,runtime_id,worker_id,session_id,checksum_sha256,content_bytes,policy_id,policy_revision,operation_id,committed_at,expires_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?12)",params![a.archive_id,workspace_id,plan.worker.runtime_id,plan.worker.worker_id,a.source_session_id,a.content_checksum_sha256,a.content_bytes,plan.policy_id,plan.policy_revision,operation_id,now,expires_at])?;
+                let observe_grants = {
+                    let mut statement = tx.prepare(
+                        "SELECT grant_id, controller_runtime_id, controller_worker_id,
+                                permissions_json, created_at
+                         FROM worker_control_grants
+                         WHERE workspace_id = ?1 AND subject_runtime_id = ?2
+                           AND subject_worker_id = ?3 AND revoked_at IS NULL",
+                    )?;
+                    statement
+                        .query_map(
+                            params![workspace_id, plan.worker.runtime_id, plan.worker.worker_id],
+                            |row| {
+                                Ok((
+                                    row.get::<_, String>(0)?,
+                                    row.get::<_, String>(1)?,
+                                    row.get::<_, String>(2)?,
+                                    row.get::<_, String>(3)?,
+                                    row.get::<_, String>(4)?,
+                                ))
+                            },
+                        )?
+                        .collect::<std::result::Result<Vec<_>, _>>()?
+                };
+                for (grant_id, controller_runtime_id, controller_worker_id, permissions, created_at) in observe_grants {
+                    let permissions: Vec<String> = serde_json::from_str(&permissions)
+                        .map_err(|error| StoreError::Store(format!("invalid Worker control grant permissions: {error}")))?;
+                    if permissions.iter().any(|permission| permission == "observe") {
+                        tx.execute(
+                            "INSERT OR IGNORE INTO worker_session_archive_observe_grants
+                             (workspace_id,archive_id,controller_runtime_id,controller_worker_id,
+                              subject_runtime_id,subject_worker_id,source_grant_id,granted_at)
+                             VALUES(?1,?2,?3,?4,?5,?6,?7,?8)",
+                            params![
+                                workspace_id,
+                                a.archive_id,
+                                controller_runtime_id,
+                                controller_worker_id,
+                                plan.worker.runtime_id,
+                                plan.worker.worker_id,
+                                grant_id,
+                                created_at
+                            ],
+                        )?;
+                    }
+                }
             }else if plan.session_disposition==SessionDisposition::Archive{return Err(StoreError::InvalidInput("archive manifest missing".into()));}
             match plan.diagnostics_disposition {
                 DiagnosticsDisposition::Purge if result.diagnostics_retained => return Err(StoreError::InvalidInput("Runtime retained diagnostics for purge disposition".into())),
@@ -592,6 +652,78 @@ impl SqliteWorkspaceStore {
     ) -> crate::Result<Option<WorkerTombstone>> {
         self.with_conn(|conn|conn.query_row("SELECT display_name,profile,worker_created_at,removed_at,archive_id,policy_id,policy_revision,operation_id FROM worker_tombstones WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![workspace_id,worker.runtime_id,worker.worker_id],|r|Ok(WorkerTombstone{workspace_id:workspace_id.into(),worker:worker.clone(),display_name:r.get(0)?,profile:r.get(1)?,created_at:r.get(2)?,removed_at:r.get(3)?,archive_id:r.get(4)?,policy_id:r.get(5)?,policy_revision:r.get::<_,i64>(6)? as u64,operation_id:r.get(7)?})).optional().map_err(StoreError::from))
     }
+
+    pub fn worker_session_archive(
+        &self,
+        workspace_id: &str,
+        worker: &RuntimeWorkerRef,
+        session_id: &str,
+    ) -> crate::Result<Option<WorkerSessionArchiveRecord>> {
+        self.with_conn(|connection| {
+            connection
+                .query_row(
+                    "SELECT archive_id, checksum_sha256, content_bytes, policy_id,
+                            policy_revision, operation_id, committed_at, expires_at
+                     FROM worker_session_archives
+                     WHERE workspace_id = ?1 AND runtime_id = ?2 AND worker_id = ?3
+                       AND session_id = ?4",
+                    params![
+                        workspace_id,
+                        worker.runtime_id,
+                        worker.worker_id,
+                        session_id
+                    ],
+                    |row| {
+                        Ok(WorkerSessionArchiveRecord {
+                            archive_id: row.get(0)?,
+                            workspace_id: workspace_id.to_string(),
+                            worker: worker.clone(),
+                            session_id: session_id.to_string(),
+                            checksum_sha256: row.get(1)?,
+                            content_bytes: row.get::<_, i64>(2)? as u64,
+                            policy_id: row.get(3)?,
+                            policy_revision: row.get::<_, i64>(4)? as u64,
+                            operation_id: row.get(5)?,
+                            committed_at: row.get(6)?,
+                            expires_at: row.get(7)?,
+                        })
+                    },
+                )
+                .optional()
+                .map_err(StoreError::from)
+        })
+    }
+
+    pub fn worker_session_archive_observable_by(
+        &self,
+        workspace_id: &str,
+        archive_id: &str,
+        controller: &RuntimeWorkerRef,
+        subject: &RuntimeWorkerRef,
+    ) -> crate::Result<bool> {
+        self.with_conn(|connection| {
+            connection
+                .query_row(
+                    "SELECT EXISTS(
+                         SELECT 1 FROM worker_session_archive_observe_grants
+                         WHERE workspace_id = ?1 AND archive_id = ?2
+                           AND controller_runtime_id = ?3 AND controller_worker_id = ?4
+                           AND subject_runtime_id = ?5 AND subject_worker_id = ?6
+                           AND revoked_at IS NULL
+                     )",
+                    params![
+                        workspace_id,
+                        archive_id,
+                        controller.runtime_id,
+                        controller.worker_id,
+                        subject.runtime_id,
+                        subject.worker_id
+                    ],
+                    |row| row.get(0),
+                )
+                .map_err(StoreError::from)
+        })
+    }
 }
 
 fn orphan_diagnostic(
@@ -1170,6 +1302,32 @@ mod tests {
     #[test]
     fn purge_tombstone_commit_is_idempotent() {
         let s = setup();
+        let archived_grant_id = "archive-observe-grant";
+        let archived_controller =
+            RuntimeWorkerRef::new("r", WorkerId::from_legacy_u64(2).to_string());
+        s.with_conn(|conn| {
+            conn.execute(
+                "INSERT INTO worker_registry(\
+                    workspace_id,worker_id,runtime_id,display_name,profile,retention_state,created_at,updated_at\
+                 ) VALUES('w',?1,'r','controller','builtin:companion','normal','created','controller-rev1')",
+                [&archived_controller.worker_id],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+        s.create_worker_control_grant(&crate::store::WorkerControlGrantRecord {
+            workspace_id: "w".into(),
+            grant_id: archived_grant_id.into(),
+            controller: archived_controller.clone(),
+            subject: req().worker,
+            relation: "subject_session".into(),
+            origin: "test".into(),
+            permissions: vec!["observe".into()],
+            operation_id: archived_grant_id.into(),
+            created_at: "2026-10-02T00:00:00Z".into(),
+            revoked_at: None,
+        })
+        .unwrap();
         s.with_conn(|conn| {
             conn.execute("INSERT INTO typed_tickets(workspace_id,ticket_id,slug,title,status,kind,priority,body,workflow_state,workflow_state_explicit) VALUES('w','ticket-old','ticket-old','Old Ticket','open','task','normal','','planning',1)", [])?;
             conn.execute("INSERT INTO worker_registry(workspace_id,worker_id,runtime_id,display_name,profile,retention_state,created_at,updated_at) VALUES('w','1','r','old worker','builtin:coder','normal','created','rev1')", [])?;
@@ -1220,6 +1378,31 @@ mod tests {
                 .state,
             WorkerRemovalPlanState::Succeeded
         );
+        assert!(
+            s.worker_session_archive_observable_by(
+                "w",
+                p.archive_id.as_deref().unwrap(),
+                &archived_controller,
+                &p.worker,
+            )
+            .unwrap(),
+            "the explicit observe grant survives Worker removal for the committed archive"
+        );
+        assert!(
+            s.revoke_worker_control_grant("w", archived_grant_id, "2026-10-02T00:01:00Z")
+                .unwrap(),
+            "the existing grant revocation authority must also revoke its archive projection"
+        );
+        assert!(
+            !s.worker_session_archive_observable_by(
+                "w",
+                p.archive_id.as_deref().unwrap(),
+                &archived_controller,
+                &p.worker,
+            )
+            .unwrap(),
+            "every later cursor page and direct Read must observe archive grant revocation"
+        );
         assert!(s.worker_tombstone("w", &p.worker).unwrap().is_some());
         let reservation_state: String = s.with_conn(|conn| {
             conn.query_row(

crates/workspace-server/src/runtime_subscription.rs

@@ -33,26 +33,19 @@ pub enum RuntimeSubscriptionBrokerError {
 #[derive(Clone, Debug)]
 pub enum BrokerSubscriptionEvent {
     Snapshot {
-        connection_generation: u64,
-        snapshot_revision: u64,
         snapshot: SubscriptionSnapshot,
     },
     Event {
-        connection_generation: u64,
-        subject_revision: u64,
         payload: SubscriptionEventPayload,
     },
     Disconnected {
-        connection_generation: u64,
         message: String,
     },
     Rejected {
-        connection_generation: u64,
         code: SubscriptionRejectionCode,
         message: String,
     },
     Closed {
-        connection_generation: u64,
         code: SubscriptionTerminationCode,
         message: String,
     },
@@ -60,7 +53,6 @@ pub enum BrokerSubscriptionEvent {
 
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct RuntimeSubscriptionBrokerStatus {
-    pub connection_generation: u64,
     pub connected: bool,
     pub desired_selectors: usize,
     pub upstream_subscriptions: usize,
@@ -82,7 +74,30 @@ impl BrokerSubscription {
         &self.selector
     }
     pub async fn recv(&mut self) -> Option<BrokerSubscriptionEvent> {
-        self.receiver.recv().await
+        loop {
+            if let Some(event) = self.receiver.recv().await {
+                return Some(event);
+            }
+            if self.restart().is_err() {
+                return None;
+            }
+        }
+    }
+
+    pub(crate) fn restart(&mut self) -> Result<(), RuntimeSubscriptionBrokerError> {
+        self.commands
+            .send(Command::Unsubscribe(self.downstream_id))
+            .map_err(|_| RuntimeSubscriptionBrokerError::Closed)?;
+        let (events, receiver) = mpsc::channel(DOWNSTREAM_QUEUE_CAPACITY);
+        self.commands
+            .send(Command::Subscribe {
+                downstream_id: self.downstream_id,
+                selector: self.selector.clone(),
+                events,
+            })
+            .map_err(|_| RuntimeSubscriptionBrokerError::Closed)?;
+        self.receiver = receiver;
+        Ok(())
     }
 }
 
@@ -120,10 +135,7 @@ impl RuntimeSubscriptionBroker {
     pub fn register_remote_runtime(&self, config: RemoteRuntimeConfig) -> u64 {
         let generation = self.next_generation.fetch_add(1, Ordering::Relaxed);
         let (commands, receiver) = mpsc::unbounded_channel();
-        let status = Arc::new(RwLock::new(RuntimeSubscriptionBrokerStatus {
-            connection_generation: generation,
-            ..Default::default()
-        }));
+        let status = Arc::new(RwLock::new(RuntimeSubscriptionBrokerStatus::default()));
         let registration = Registration {
             generation,
             commands: commands.clone(),
@@ -156,7 +168,6 @@ impl RuntimeSubscriptionBroker {
         let generation = self.next_generation.fetch_add(1, Ordering::Relaxed);
         let (commands, receiver) = mpsc::unbounded_channel();
         let status = Arc::new(RwLock::new(RuntimeSubscriptionBrokerStatus {
-            connection_generation: generation,
             connected: true,
             ..Default::default()
         }));
@@ -257,7 +268,6 @@ impl RuntimeSubscriptionBroker {
             .clone();
         if !initial_status.connected {
             let _ = initial_events.try_send(BrokerSubscriptionEvent::Disconnected {
-                connection_generation: initial_status.connection_generation,
                 message: "Runtime subscription connection is not currently available".to_string(),
             });
         }
@@ -286,8 +296,7 @@ struct SelectorState {
     downstreams: HashMap<u64, mpsc::Sender<BrokerSubscriptionEvent>>,
     upstream_id: Option<SubscriptionId>,
     pending: bool,
-    snapshot: Option<(u64, SubscriptionSnapshot)>,
-    revisions: HashMap<String, u64>,
+    snapshot: Option<SubscriptionSnapshot>,
 }
 impl SelectorState {
     fn new() -> Self {
@@ -296,7 +305,6 @@ impl SelectorState {
             upstream_id: None,
             pending: false,
             snapshot: None,
-            revisions: HashMap::new(),
         }
     }
 }
@@ -334,11 +342,9 @@ impl State {
             selector.upstream_id = None;
             selector.pending = false;
             selector.snapshot = None;
-            selector.revisions.clear();
             broadcast(
                 &mut selector.downstreams,
                 BrokerSubscriptionEvent::Disconnected {
-                    connection_generation: self.generation,
                     message: message.clone(),
                 },
             );
@@ -347,12 +353,77 @@ impl State {
 }
 
 struct EmbeddedEntry {
+    lifetime: u64,
     downstreams: HashMap<u64, mpsc::Sender<BrokerSubscriptionEvent>>,
-    snapshot_revision: u64,
     snapshot: SubscriptionSnapshot,
     task: tokio::task::JoinHandle<()>,
 }
 
+enum EmbeddedUpdate {
+    Event {
+        selector: EventSubscriptionSelector,
+        lifetime: u64,
+        payload: SubscriptionEventPayload,
+    },
+    Closed {
+        selector: EventSubscriptionSelector,
+        lifetime: u64,
+        code: SubscriptionTerminationCode,
+        message: String,
+    },
+}
+
+fn start_embedded_subscription(
+    runtime: &worker_runtime::Runtime,
+    runtime_id: &str,
+    selector: EventSubscriptionSelector,
+    lifetime: u64,
+    updates: mpsc::UnboundedSender<EmbeddedUpdate>,
+) -> Result<(SubscriptionSnapshot, tokio::task::JoinHandle<()>), worker_runtime::error::RuntimeError>
+{
+    let mut subscription = runtime.subscribe_event_selector(selector.clone())?;
+    let snapshot = project_snapshot_runtime(subscription.snapshot().clone(), runtime_id);
+    let task_runtime_id = runtime_id.to_string();
+    let task = tokio::spawn(async move {
+        loop {
+            match subscription.recv().await {
+                Ok(update) => {
+                    let payload = project_payload_runtime(update.payload, &task_runtime_id);
+                    if updates
+                        .send(EmbeddedUpdate::Event {
+                            selector: selector.clone(),
+                            lifetime,
+                            payload,
+                        })
+                        .is_err()
+                    {
+                        return;
+                    }
+                }
+                Err(worker_runtime::RuntimeSubscriptionRecvError::Lagged) => {
+                    let _ = updates.send(EmbeddedUpdate::Closed {
+                        selector,
+                        lifetime,
+                        code: SubscriptionTerminationCode::Lagged,
+                        message: "embedded Runtime subscription lagged; resubscribing from a fresh snapshot".to_string(),
+                    });
+                    return;
+                }
+                Err(worker_runtime::RuntimeSubscriptionRecvError::Closed) => {
+                    let _ = updates.send(EmbeddedUpdate::Closed {
+                        selector,
+                        lifetime,
+                        code: SubscriptionTerminationCode::ServerShutdown,
+                        message: "embedded Runtime subscription closed".to_string(),
+                    });
+                    return;
+                }
+            }
+        }
+    });
+    Ok((snapshot, task))
+}
+
 async fn run_embedded_connection(
     runtime_id: String,
     runtime: worker_runtime::Runtime,
@@ -364,6 +435,7 @@ async fn run_embedded_connection(
     let (updates, mut update_receiver) = mpsc::unbounded_channel();
     let mut entries = HashMap::<EventSubscriptionSelector, EmbeddedEntry>::new();
     let mut downstream_index = HashMap::<u64, EventSubscriptionSelector>::new();
+    let mut next_lifetime = 1_u64;
     loop {
         tokio::select! {
             command = commands.recv() => match command {
@@ -371,30 +443,38 @@ async fn run_embedded_connection(
                     downstream_index.insert(downstream_id, selector.clone());
                     if let Some(entry) = entries.get_mut(&selector) {
                         let _ = events.try_send(BrokerSubscriptionEvent::Snapshot {
-                            connection_generation: generation,
-                            snapshot_revision: entry.snapshot_revision,
                             snapshot: entry.snapshot.clone(),
                         });
                         entry.downstreams.insert(downstream_id, events);
                     } else {
-                        match runtime.subscribe_event_selector(selector.clone()) {
-                            Ok(mut subscription) => {
-                                let snapshot_revision = subscription.snapshot_revision();
-                                let snapshot = project_snapshot_runtime(subscription.snapshot().clone(), &runtime_id);
-                                let _ = events.try_send(BrokerSubscriptionEvent::Snapshot { connection_generation: generation, snapshot_revision, snapshot: snapshot.clone() });
-                                let sender = updates.clone();
-                                let task_selector = selector.clone();
-                                let task_runtime_id = runtime_id.clone();
-                                let task = tokio::spawn(async move {
-                                    while let Ok(update) = subscription.recv().await {
-                                        let payload = project_payload_runtime(update.payload, &task_runtime_id);
-                                        if sender.send((task_selector.clone(), update.subject_revision, payload)).is_err() { break; }
-                                    }
+                        let lifetime = next_lifetime;
+                        next_lifetime = next_lifetime.saturating_add(1);
+                        match start_embedded_subscription(
+                            &runtime,
+                            &runtime_id,
+                            selector.clone(),
+                            lifetime,
+                            updates.clone(),
+                        ) {
+                            Ok((snapshot, task)) => {
+                                let _ = events.try_send(BrokerSubscriptionEvent::Snapshot {
+                                    snapshot: snapshot.clone(),
                                 });
-                                entries.insert(selector, EmbeddedEntry { downstreams: HashMap::from([(downstream_id, events)]), snapshot_revision, snapshot, task });
+                                entries.insert(
+                                    selector,
+                                    EmbeddedEntry {
+                                        lifetime,
+                                        downstreams: HashMap::from([(downstream_id, events)]),
+                                        snapshot,
+                                        task,
+                                    },
+                                );
                             }
                             Err(error) => {
-                                let _ = events.try_send(BrokerSubscriptionEvent::Rejected { connection_generation: generation, code: SubscriptionRejectionCode::UnsupportedSelector, message: error.to_string() });
+                                let _ = events.try_send(BrokerSubscriptionEvent::Rejected {
+                                    code: SubscriptionRejectionCode::UnsupportedSelector,
+                                    message: error.to_string(),
+                                });
                             }
                         }
                     }
@@ -407,7 +487,7 @@ async fn run_embedded_connection(
                 }
                 Some(Command::Shutdown(replacement)) => {
                     for entry in entries.values_mut() {
-                        broadcast(&mut entry.downstreams, BrokerSubscriptionEvent::Closed { connection_generation: generation, code: SubscriptionTerminationCode::ServerShutdown, message: format!("embedded Runtime generation {generation} was fenced by {replacement}") });
+                        broadcast(&mut entry.downstreams, BrokerSubscriptionEvent::Closed { code: SubscriptionTerminationCode::ServerShutdown, message: format!("embedded Runtime registration {generation} was replaced by {replacement}") });
                         entry.task.abort();
                     }
                     return;
@@ -415,16 +495,68 @@ async fn run_embedded_connection(
                 None => return,
             },
             update = update_receiver.recv() => {
-                let Some((selector, subject_revision, payload)) = update else { return; };
-                if let Some(entry) = entries.get_mut(&selector) {
-                    entry.snapshot_revision = entry.snapshot_revision.saturating_add(1);
-                    apply_event_to_cached_snapshot(&mut entry.snapshot, &payload);
-                    broadcast(&mut entry.downstreams, BrokerSubscriptionEvent::Event { connection_generation: generation, subject_revision, payload });
+                let Some(update) = update else { return; };
+                match update {
+                    EmbeddedUpdate::Event { selector, lifetime, payload } => {
+                        if let Some(entry) = entries
+                            .get_mut(&selector)
+                            .filter(|entry| entry.lifetime == lifetime)
+                        {
+                            apply_event_to_cached_snapshot(&mut entry.snapshot, &payload);
+                            broadcast(&mut entry.downstreams, BrokerSubscriptionEvent::Event { payload });
+                        }
+                    }
+                    EmbeddedUpdate::Closed { selector, lifetime, code, message } => {
+                        if !entries
+                            .get(&selector)
+                            .is_some_and(|entry| entry.lifetime == lifetime)
+                        {
+                            continue;
+                        }
+                        let Some(mut entry) = entries.remove(&selector) else { continue; };
+                        broadcast(
+                            &mut entry.downstreams,
+                            BrokerSubscriptionEvent::Closed { code, message },
+                        );
+                        if entry.downstreams.is_empty() {
+                            continue;
+                        }
+                        let new_lifetime = next_lifetime;
+                        next_lifetime = next_lifetime.saturating_add(1);
+                        match start_embedded_subscription(
+                            &runtime,
+                            &runtime_id,
+                            selector.clone(),
+                            new_lifetime,
+                            updates.clone(),
+                        ) {
+                            Ok((snapshot, task)) => {
+                                broadcast(
+                                    &mut entry.downstreams,
+                                    BrokerSubscriptionEvent::Snapshot {
+                                        snapshot: snapshot.clone(),
+                                    },
+                                );
+                                entry.snapshot = snapshot;
+                                entry.task = task;
+                                entry.lifetime = new_lifetime;
+                                entries.insert(selector, entry);
+                            }
+                            Err(error) => {
+                                broadcast(
+                                    &mut entry.downstreams,
+                                    BrokerSubscriptionEvent::Rejected {
+                                        code: SubscriptionRejectionCode::Internal,
+                                        message: error.to_string(),
+                                    },
+                                );
+                            }
+                        }
+                    }
                 }
             }
         }
         *status.write().expect("broker status poisoned") = RuntimeSubscriptionBrokerStatus {
-            connection_generation: generation,
             connected: true,
             desired_selectors: entries.len(),
             upstream_subscriptions: entries.len(),
@@ -597,10 +729,8 @@ async fn apply_online(
                 .selectors
                 .entry(selector.clone())
                 .or_insert_with(SelectorState::new);
-            if let Some((revision, snapshot)) = &entry.snapshot {
+            if let Some(snapshot) = &entry.snapshot {
                 let _ = events.try_send(BrokerSubscriptionEvent::Snapshot {
-                    connection_generation: state.generation,
-                    snapshot_revision: *revision,
                     snapshot: snapshot.clone(),
                 });
             }
@@ -696,7 +826,6 @@ async fn handle_frame(
             request_id,
             subscription_id,
             selector,
-            snapshot_revision,
             snapshot,
         }) => {
             if state.pending.remove(&request_id) != Some(selector.clone()) {
@@ -706,18 +835,13 @@ async fn handle_frame(
             let entry = state.selectors.get_mut(&selector).ok_or(())?;
             entry.pending = false;
             entry.upstream_id = Some(subscription_id.clone());
-            entry.snapshot = Some((snapshot_revision, snapshot.clone()));
-            entry.revisions = snapshot_revisions(&snapshot);
+            entry.snapshot = Some(snapshot.clone());
             state
                 .upstream_index
                 .insert(subscription_id, selector.clone());
             broadcast(
                 &mut entry.downstreams,
-                BrokerSubscriptionEvent::Snapshot {
-                    connection_generation: state.generation,
-                    snapshot_revision,
-                    snapshot,
-                },
+                BrokerSubscriptionEvent::Snapshot { snapshot },
             );
             if entry.downstreams.is_empty() {
                 maybe_unsubscribe(socket, state, selector).await?;
@@ -734,18 +858,13 @@ async fn handle_frame(
                 if let Some(mut entry) = state.selectors.remove(&selector) {
                     broadcast(
                         &mut entry.downstreams,
-                        BrokerSubscriptionEvent::Rejected {
-                            connection_generation: state.generation,
-                            code,
-                            message,
-                        },
+                        BrokerSubscriptionEvent::Rejected { code, message },
                     );
                 }
             }
         }
         SubscriptionFramePayload::Event(SubscriptionEvent::Event {
             subscription_id,
-            subject_revision,
             payload,
         }) => {
             let selector = state
@@ -756,24 +875,12 @@ async fn handle_frame(
             payload.validate_for_selector(&selector).map_err(|_| ())?;
             let payload = project_payload_runtime(payload, &state.runtime_id);
             let entry = state.selectors.get_mut(&selector).ok_or(())?;
-            if let Some(subject) = event_subject(&payload) {
-                let revision = entry.revisions.entry(subject).or_insert(0);
-                if subject_revision <= *revision {
-                    return Ok(());
-                }
-                *revision = subject_revision;
-            }
-            if let Some((snapshot_revision, snapshot)) = entry.snapshot.as_mut() {
-                *snapshot_revision = snapshot_revision.saturating_add(1);
+            if let Some(snapshot) = entry.snapshot.as_mut() {
                 apply_event_to_cached_snapshot(snapshot, &payload);
             }
             broadcast(
                 &mut entry.downstreams,
-                BrokerSubscriptionEvent::Event {
-                    connection_generation: state.generation,
-                    subject_revision,
-                    payload,
-                },
+                BrokerSubscriptionEvent::Event { payload },
             );
         }
         SubscriptionFramePayload::Event(SubscriptionEvent::SubscriptionClosed {
@@ -785,14 +892,9 @@ async fn handle_frame(
             let should_resubscribe = if let Some(entry) = state.selectors.get_mut(&selector) {
                 entry.upstream_id = None;
                 entry.snapshot = None;
-                entry.revisions.clear();
                 broadcast(
                     &mut entry.downstreams,
-                    BrokerSubscriptionEvent::Closed {
-                        connection_generation: state.generation,
-                        code,
-                        message,
-                    },
+                    BrokerSubscriptionEvent::Closed { code, message },
                 );
                 !entry.downstreams.is_empty()
             } else {
@@ -822,10 +924,9 @@ fn close_all(state: &mut State, replacement: u64) {
         broadcast(
             &mut entry.downstreams,
             BrokerSubscriptionEvent::Closed {
-                connection_generation: state.generation,
                 code: SubscriptionTerminationCode::ServerShutdown,
                 message: format!(
-                    "Runtime connection generation {} was fenced by generation {replacement}",
+                    "Runtime registration {} was replaced by {replacement}",
                     state.generation
                 ),
             },
@@ -846,51 +947,6 @@ fn broadcast(
     downstreams.retain(|id, _| !closed.contains(id));
 }
 
-fn snapshot_revisions(snapshot: &SubscriptionSnapshot) -> HashMap<String, u64> {
-    match snapshot {
-        SubscriptionSnapshot::Workers { workers } => workers
-            .iter()
-            .map(|worker| {
-                (
-                    format!(
-                        "{}:{}",
-                        worker.runtime_id.as_deref().unwrap_or_default(),
-                        worker.worker_id
-                    ),
-                    worker.subject_revision,
-                )
-            })
-            .collect(),
-        SubscriptionSnapshot::WorkerProtocol { worker_id, .. } => {
-            HashMap::from([(worker_id.to_string(), 0)])
-        }
-        SubscriptionSnapshot::WorkspaceWorkdirs { .. } => HashMap::new(),
-    }
-}
-fn event_subject(payload: &SubscriptionEventPayload) -> Option<String> {
-    Some(match payload {
-        SubscriptionEventPayload::WorkerUpserted { worker } => format!(
-            "{}:{}",
-            worker.runtime_id.as_deref().unwrap_or_default(),
-            worker.worker_id
-        ),
-        SubscriptionEventPayload::WorkerRemoved {
-            worker_id,
-            runtime_id,
-        } => format!(
-            "{}:{}",
-            runtime_id.as_deref().unwrap_or_default(),
-            worker_id
-        ),
-        SubscriptionEventPayload::WorkerProtocol { worker_id, .. } => worker_id.to_string(),
-        SubscriptionEventPayload::WorkdirUpserted { workdir } => {
-            workdir.working_directory_id.to_string()
-        }
-        SubscriptionEventPayload::WorkdirRemoved {
-            working_directory_id,
-        } => working_directory_id.to_string(),
-    })
-}
 async fn send_frame(socket: &mut RuntimeSocket, frame: SubscriptionFrame) -> Result<(), ()> {
     frame.validate().map_err(|_| ())?;
     socket
@@ -979,7 +1035,6 @@ fn runtime_token(
 }
 fn update_status(status: &RwLock<RuntimeSubscriptionBrokerStatus>, state: &State, connected: bool) {
     *status.write().expect("broker status poisoned") = RuntimeSubscriptionBrokerStatus {
-        connection_generation: state.generation,
         connected,
         desired_selectors: state
             .selectors

crates/workspace-server/src/runtime_subscription_tests.rs

@@ -74,6 +74,7 @@ fn create_request(name: &str) -> CreateWorkerRequest {
             settings_revision: 1,
             language: "English".to_string(),
         }),
+        subjektiv_attached: false,
     }
 }
 
@@ -264,16 +265,9 @@ async fn replacing_runtime_registration_fences_the_old_generation() {
     assert!(second_generation > first_generation);
     assert!(matches!(
         next_event(&mut old).await,
-        BrokerSubscriptionEvent::Closed {
-            connection_generation,
-            ..
-        } if connection_generation == first_generation
+        BrokerSubscriptionEvent::Closed { .. }
     ));
-    let status = wait_for_status(&broker, "runtime-test", |status| {
-        status.connection_generation == second_generation && status.connected
-    })
-    .await;
-    assert_eq!(status.connection_generation, second_generation);
+    wait_for_status(&broker, "runtime-test", |status| status.connected).await;
     server.abort();
 }
 
@@ -430,6 +424,75 @@ async fn embedded_runtime_uses_in_process_subscription_source() {
     server.abort();
 }
 
+#[tokio::test]
+async fn embedded_downstream_overflow_recovers_through_a_fresh_snapshot() {
+    let (runtime, _config, server) = fixture().await;
+    let worker = runtime.list_workers().unwrap().remove(0);
+    let broker = RuntimeSubscriptionBroker::new("local");
+    broker.register_embedded_runtime("embedded-worker-runtime", runtime.clone());
+    let mut subscription = broker
+        .subscribe(
+            "embedded-worker-runtime",
+            EventSubscriptionSelector::RuntimeWorkers,
+        )
+        .unwrap();
+    assert!(matches!(
+        next_snapshot(&mut subscription).await,
+        BrokerSubscriptionEvent::Snapshot { .. }
+    ));
+
+    let final_command_id = (DOWNSTREAM_QUEUE_CAPACITY as u64) * 3;
+    for command_id in 1..=final_command_id {
+        let mut snapshot = worker.worker_state.clone().unwrap();
+        snapshot.last_command_id = command_id;
+        runtime
+            .observe_worker_event(
+                &worker.worker_ref,
+                protocol::Event::WorkerState { snapshot },
+            )
+            .unwrap();
+    }
+
+    let mut saw_recovery_snapshot = false;
+    let mut converged = false;
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while !converged {
+            match subscription
+                .recv()
+                .await
+                .expect("broker subscription closed")
+            {
+                BrokerSubscriptionEvent::Snapshot {
+                    snapshot: SubscriptionSnapshot::Workers { workers },
+                } => {
+                    saw_recovery_snapshot = true;
+                    converged = workers.iter().any(|candidate| {
+                        candidate.worker_id.as_str() == worker.worker_id.to_string()
+                            && candidate
+                                .worker_state
+                                .as_ref()
+                                .is_some_and(|state| state.last_command_id == final_command_id)
+                    });
+                }
+                BrokerSubscriptionEvent::Event {
+                    payload: SubscriptionEventPayload::WorkerUpserted { worker: candidate },
+                } if saw_recovery_snapshot => {
+                    converged = candidate.worker_id.as_str() == worker.worker_id.to_string()
+                        && candidate
+                            .worker_state
+                            .as_ref()
+                            .is_some_and(|state| state.last_command_id == final_command_id);
+                }
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("overflow recovery did not converge");
+    assert!(saw_recovery_snapshot);
+    server.abort();
+}
+
 #[tokio::test]
 async fn strict_runtime_subscription_rejects_private_endpoint_before_websocket_connect() {
     let config = RemoteRuntimeConfig::new(

crates/workspace-server/src/worker_projection.rs

@@ -11,7 +11,10 @@ use tokio::task::JoinHandle;
 use crate::runtime_subscription::{
     BrokerSubscriptionEvent, RuntimeSubscriptionBroker, RuntimeSubscriptionBrokerError,
 };
-use crate::store::{ControlPlaneStore, WorkerCatalogChange, WorkerRegistryProjectionCommit};
+use crate::store::{
+    ControlPlaneStore, WorkerCatalogChange, WorkerRegistryProjectionCommit,
+    WorkerRegistryProjectionRecord,
+};
 use worker_runtime::identity::RuntimeWorkerRef;
 
 const PROJECTION_EVENT_CAPACITY: usize = 256;
@@ -20,21 +23,42 @@ pub type WorkerProjectionChange = WorkerCatalogChange;
 
 #[derive(Debug, Clone)]
 pub struct WorkerProjectionEvent {
-    pub revision: u64,
     pub changes: Vec<WorkerProjectionChange>,
 }
 
+/// An authoritative durable snapshot followed by changes committed after that snapshot.
+pub struct WorkerProjectionSubscription {
+    snapshot: Vec<WorkerRegistryProjectionRecord>,
+    receiver: broadcast::Receiver<WorkerProjectionEvent>,
+}
+
+impl WorkerProjectionSubscription {
+    pub fn take_snapshot(&mut self) -> Vec<WorkerRegistryProjectionRecord> {
+        std::mem::take(&mut self.snapshot)
+    }
+
+    pub async fn recv(&mut self) -> Result<WorkerProjectionEvent, broadcast::error::RecvError> {
+        self.receiver.recv().await
+    }
+
+    #[cfg(test)]
+    pub fn try_recv(&mut self) -> Result<WorkerProjectionEvent, broadcast::error::TryRecvError> {
+        self.receiver.try_recv()
+    }
+}
+
 /// 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.
+/// The ordering gate covers both durable commits and subscription snapshot creation.
+/// Every subscription therefore starts with one authoritative snapshot and receives
+/// only commits that completed after it, in the same order in which they were stored.
 pub struct WorkerProjectionService {
     workspace_id: String,
     store: Arc<dyn ControlPlaneStore>,
     events: broadcast::Sender<WorkerProjectionEvent>,
-    connection_generation_epoch: u64,
+    ordering: Mutex<()>,
+    attach_gate: tokio::sync::Mutex<()>,
     tasks: Mutex<HashMap<String, JoinHandle<()>>>,
 }
 
@@ -45,8 +69,8 @@ impl WorkerProjectionService {
             workspace_id: workspace_id.into(),
             store,
             events,
-            connection_generation_epoch: Utc::now().timestamp_nanos_opt().unwrap_or_default().max(0)
-                as u64,
+            ordering: Mutex::new(()),
+            attach_gate: tokio::sync::Mutex::new(()),
             tasks: Mutex::new(HashMap::new()),
         })
     }
@@ -62,8 +86,38 @@ impl WorkerProjectionService {
         }
     }
 
-    pub fn subscribe(&self) -> broadcast::Receiver<WorkerProjectionEvent> {
-        self.events.subscribe()
+    pub fn subscribe_ordered(&self, limit: usize) -> crate::Result<WorkerProjectionSubscription> {
+        let _ordering = self
+            .ordering
+            .lock()
+            .expect("Worker projection ordering lock poisoned");
+        let receiver = self.events.subscribe();
+        let snapshot_limit = limit.saturating_add(1);
+        let mut snapshot = self
+            .store
+            .worker_registry_projection_snapshot(self.workspace_id.as_str(), snapshot_limit)?;
+        if snapshot.len() > limit {
+            return Err(crate::Error::Store(format!(
+                "authoritative Worker projection exceeds the configured snapshot limit of {limit} records"
+            )));
+        }
+        snapshot.shrink_to_fit();
+        Ok(WorkerProjectionSubscription { snapshot, receiver })
+    }
+
+    /// Serialize one durable projection mutation with its publication. The closure
+    /// must return the exact projection payload captured by the same transaction.
+    pub(crate) fn publish_ordered<T, E>(
+        &self,
+        mutation: impl FnOnce() -> std::result::Result<(T, WorkerRegistryProjectionCommit), E>,
+    ) -> std::result::Result<T, E> {
+        let _ordering = self
+            .ordering
+            .lock()
+            .expect("Worker projection ordering lock poisoned");
+        let (result, commit) = mutation()?;
+        self.publish_commit(commit);
+        Ok(result)
     }
 
     pub async fn attach_runtime(
@@ -71,6 +125,20 @@ impl WorkerProjectionService {
         broker: &RuntimeSubscriptionBroker,
         runtime_id: &str,
     ) -> Result<(), RuntimeSubscriptionBrokerError> {
+        // Serialize replacement and wait for the old consumer to stop before the new
+        // lifetime can apply its snapshot. Late notifications can therefore never
+        // cross a Runtime ownership boundary.
+        let _attach = self.attach_gate.lock().await;
+        let previous = self
+            .tasks
+            .lock()
+            .expect("Worker projection task lock poisoned")
+            .remove(runtime_id);
+        if let Some(previous) = previous {
+            previous.abort();
+            let _ = previous.await;
+        }
+
         let mut subscription =
             broker.subscribe(runtime_id, EventSubscriptionSelector::RuntimeWorkers)?;
         let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
@@ -80,12 +148,14 @@ impl WorkerProjectionService {
                 break;
             };
             let initialized = matches!(event, BrokerSubscriptionEvent::Snapshot { .. });
-            if let Err(error) = self.apply_broker_event(runtime_id, event).await {
+            if let Err(error) = self.apply_broker_event(runtime_id, event) {
                 tracing::warn!(
                     runtime_id,
                     error = %error,
-                    "failed to initialize durable Worker projection"
+                    "failed to initialize durable Worker projection; restarting from a fresh snapshot"
                 );
+                subscription.restart()?;
+                continue;
             }
             if initialized {
                 break;
@@ -99,26 +169,22 @@ impl WorkerProjectionService {
                 let Some(service) = service.upgrade() else {
                     return;
                 };
-                if let Err(error) = service
-                    .apply_broker_event(task_runtime_id.as_str(), event)
-                    .await
-                {
+                if let Err(error) = service.apply_broker_event(task_runtime_id.as_str(), event) {
                     tracing::warn!(
                         runtime_id = %task_runtime_id,
                         error = %error,
-                        "failed to update durable Worker projection"
+                        "failed to update durable Worker projection; restarting from a fresh snapshot"
                     );
+                    if subscription.restart().is_err() {
+                        return;
+                    }
                 }
             }
         });
-        if let Some(previous) = self
-            .tasks
+        self.tasks
             .lock()
             .expect("Worker projection task lock poisoned")
-            .insert(runtime_id, task)
-        {
-            previous.abort();
-        }
+            .insert(runtime_id, task);
         Ok(())
     }
 
@@ -128,110 +194,96 @@ impl WorkerProjectionService {
         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)
+        self.publish_ordered(|| {
+            self.store
+                .apply_worker_registry_observation(
+                    self.workspace_id.as_str(),
+                    runtime_id,
+                    worker,
+                    observed_at,
+                )
+                .map(|commit| ((), commit))
+        })
     }
 
     pub fn refresh(&self, worker: &RuntimeWorkerRef) -> crate::Result<()> {
-        let commit = self
-            .store
-            .commit_worker_registry_projection_refresh(self.workspace_id.as_str(), worker)?;
-        self.publish_commit(commit)
+        self.publish_ordered(|| {
+            self.store
+                .commit_worker_registry_projection_refresh(self.workspace_id.as_str(), worker)
+                .map(|commit| ((), commit))
+        })
     }
 
-    async fn apply_broker_event(
+    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)
+        self.publish_ordered(|| {
+            let commit = match event {
+                BrokerSubscriptionEvent::Snapshot {
+                    snapshot: SubscriptionSnapshot::Workers { workers },
+                } => self.store.reconcile_worker_registry_snapshot(
+                    self.workspace_id.as_str(),
+                    runtime_id,
+                    workers.as_slice(),
+                    observed_at.as_str(),
+                )?,
+                BrokerSubscriptionEvent::Snapshot { .. } => {
+                    return Ok((
+                        (),
+                        WorkerRegistryProjectionCommit {
+                            changes: Vec::new(),
+                        },
+                    ));
+                }
+                BrokerSubscriptionEvent::Event {
+                    payload: SubscriptionEventPayload::WorkerUpserted { worker },
+                } => self.store.apply_worker_registry_observation(
+                    self.workspace_id.as_str(),
+                    runtime_id,
+                    &worker,
+                    observed_at.as_str(),
+                )?,
+                BrokerSubscriptionEvent::Event {
+                    payload: SubscriptionEventPayload::WorkerRemoved { worker_id, .. },
+                } => self.store.mark_worker_registry_observation_unavailable(
+                    self.workspace_id.as_str(),
+                    runtime_id,
+                    Some(worker_id.as_str()),
+                    observed_at.as_str(),
+                )?,
+                BrokerSubscriptionEvent::Event { .. } => {
+                    return Ok((
+                        (),
+                        WorkerRegistryProjectionCommit {
+                            changes: Vec::new(),
+                        },
+                    ));
+                }
+                BrokerSubscriptionEvent::Disconnected { .. }
+                | BrokerSubscriptionEvent::Rejected { .. }
+                | BrokerSubscriptionEvent::Closed { .. } => {
+                    self.store.mark_worker_registry_observation_unavailable(
+                        self.workspace_id.as_str(),
+                        runtime_id,
+                        None,
+                        observed_at.as_str(),
+                    )?
+                }
+            };
+            Ok(((), commit))
+        })
     }
 
-    /// Broadcast an already committed catalog payload. Persistence always happens
-    /// before this call; a process crash therefore closes subscribers, whose next
-    /// subscription starts from the durable snapshot. Retrying a removal returns
-    /// the same durable change and can heal an in-process missed broadcast.
-    pub(crate) fn publish_commit(
-        &self,
-        commit: WorkerRegistryProjectionCommit,
-    ) -> crate::Result<()> {
+    fn publish_commit(&self, commit: WorkerRegistryProjectionCommit) {
         if commit.changes.is_empty() {
-            return Ok(());
+            return;
         }
         let _ = self.events.send(WorkerProjectionEvent {
-            revision: commit.revision,
             changes: commit.changes,
         });
-        Ok(())
     }
 }
 
@@ -240,3 +292,159 @@ impl Drop for WorkerProjectionService {
         self.shutdown();
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::{Arc, Barrier, Mutex};
+
+    use crate::store::{
+        AccountRecord, ControlPlaneStore, SqliteWorkspaceStore, WorkerCatalogChange,
+        WorkerRegistryRecord, WorkspaceRecord,
+    };
+    use worker_runtime::identity::RuntimeWorkerRef;
+
+    use super::WorkerProjectionService;
+
+    const WORKSPACE_ID: &str = "projection-ordering-test";
+
+    async fn fixture() -> (
+        Arc<SqliteWorkspaceStore>,
+        Arc<WorkerProjectionService>,
+        tempfile::TempDir,
+    ) {
+        let temp = tempfile::tempdir().unwrap();
+        let store = Arc::new(SqliteWorkspaceStore::open(temp.path().join("workspace.db")).unwrap());
+        store
+            .upsert_account(&AccountRecord {
+                account_id: "owner".to_string(),
+                kind: "user".to_string(),
+                handle: "owner".to_string(),
+                display_name: "Owner".to_string(),
+                created_at: "1".to_string(),
+                updated_at: "1".to_string(),
+            })
+            .unwrap();
+        store
+            .upsert_workspace(&WorkspaceRecord {
+                workspace_id: WORKSPACE_ID.to_string(),
+                owner_account_id: "owner".to_string(),
+                display_name: "Projection ordering".to_string(),
+                state: "active".to_string(),
+                created_at: "1".to_string(),
+                updated_at: "1".to_string(),
+            })
+            .await
+            .unwrap();
+        let service = WorkerProjectionService::new(WORKSPACE_ID, store.clone());
+        (store, service, temp)
+    }
+
+    fn worker(runtime_id: &str, worker_id: &str) -> WorkerRegistryRecord {
+        WorkerRegistryRecord {
+            workspace_id: WORKSPACE_ID.to_string(),
+            worker: RuntimeWorkerRef::new(runtime_id, worker_id),
+            display_name: format!("{runtime_id}-{worker_id}"),
+            profile: Some("builtin:coder".to_string()),
+            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(),
+        }
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn snapshot_registration_is_atomic_with_concurrent_publication() {
+        let (store, service, _temp) = fixture().await;
+        let barrier = Arc::new(Barrier::new(2));
+        let subscriber_service = service.clone();
+        let subscriber_barrier = barrier.clone();
+        let subscriber = std::thread::spawn(move || {
+            subscriber_barrier.wait();
+            subscriber_service.subscribe_ordered(10).unwrap()
+        });
+        let publisher_service = service.clone();
+        let publisher_barrier = barrier.clone();
+        let publisher_store = store.clone();
+        let record = worker("runtime-a", "worker-a");
+        let publisher = std::thread::spawn(move || {
+            publisher_barrier.wait();
+            publisher_service
+                .publish_ordered(|| {
+                    publisher_store
+                        .upsert_worker_registry(&record)
+                        .map(|commit| ((), commit))
+                })
+                .unwrap();
+        });
+
+        let mut subscription = subscriber.join().unwrap();
+        publisher.join().unwrap();
+        let in_snapshot = subscription
+            .take_snapshot()
+            .iter()
+            .any(|record| record.registry.worker.worker_id == "worker-a");
+        let in_event = match subscription.try_recv() {
+            Ok(event) => matches!(
+                event.changes.as_slice(),
+                [WorkerCatalogChange::Upsert(record)]
+                    if record.registry.worker.worker_id == "worker-a"
+            ),
+            Err(tokio::sync::broadcast::error::TryRecvError::Empty) => false,
+            Err(error) => panic!("unexpected projection receive error: {error}"),
+        };
+        assert_ne!(
+            in_snapshot, in_event,
+            "the commit must cross the boundary exactly once"
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn concurrent_commits_are_published_in_durable_execution_order() {
+        let (store, service, _temp) = fixture().await;
+        let mut subscription = service.subscribe_ordered(10).unwrap();
+        assert!(subscription.take_snapshot().is_empty());
+        let barrier = Arc::new(Barrier::new(3));
+        let durable_order = Arc::new(Mutex::new(Vec::new()));
+        let mut publishers = Vec::new();
+        for worker_id in ["worker-a", "worker-b"] {
+            let service = service.clone();
+            let store = store.clone();
+            let barrier = barrier.clone();
+            let durable_order = durable_order.clone();
+            let record = worker("runtime-a", worker_id);
+            publishers.push(std::thread::spawn(move || {
+                barrier.wait();
+                service
+                    .publish_ordered(|| {
+                        durable_order
+                            .lock()
+                            .unwrap()
+                            .push(record.worker.worker_id.clone());
+                        store
+                            .upsert_worker_registry(&record)
+                            .map(|commit| ((), commit))
+                    })
+                    .unwrap();
+            }));
+        }
+        barrier.wait();
+        for publisher in publishers {
+            publisher.join().unwrap();
+        }
+
+        let published_order = [
+            subscription.try_recv().unwrap(),
+            subscription.try_recv().unwrap(),
+        ]
+        .into_iter()
+        .map(|event| match event.changes.as_slice() {
+            [WorkerCatalogChange::Upsert(record)] => record.registry.worker.worker_id.clone(),
+            changes => panic!("unexpected projection changes: {changes:?}"),
+        })
+        .collect::<Vec<_>>();
+        assert_eq!(published_order, *durable_order.lock().unwrap());
+    }
+}

crates/workspace-server/src/workspace_catalog.rs

@@ -296,7 +296,7 @@ mod tests {
             display_name: "Workspace A".to_string(),
             repository: InitialRepositoryIntent {
                 uri: repository.path().display().to_string(),
-                repository_key: "main".to_string(),
+                repository_key: "platform".to_string(),
                 default_ref: None,
             },
         };
@@ -322,13 +322,11 @@ mod tests {
         assert_eq!(signing_identity.algorithm, "ed25519");
         assert!(signing_identity.public_key.is_some());
         assert!(signing_identity.public_key_fingerprint.is_some());
-        assert_eq!(
-            store
-                .list_repositories(&created.workspace.workspace_id)
-                .unwrap()
-                .len(),
-            1
-        );
+        let repositories = store
+            .list_repositories(&created.workspace.workspace_id)
+            .unwrap();
+        assert_eq!(repositories.len(), 1);
+        assert_eq!(repositories[0].repository_key, "platform");
         assert!(
             store
                 .load_workspace_config(&created.workspace.workspace_id)
@@ -337,6 +335,42 @@ mod tests {
         );
     }
 
+    #[test]
+    fn create_rejects_noncanonical_repository_keys_without_persisting_a_workspace() {
+        let store = Arc::new(SqliteWorkspaceStore::in_memory().unwrap());
+        let service = WorkspaceCatalogService::new(
+            store.clone(),
+            Arc::new(
+                crate::workspace_signing_identity::InMemoryWorkspaceSigningMaterialStore::default(),
+            ),
+        );
+        let repository = git_repository();
+        let owner_account_id = owner_account(store.as_ref());
+
+        for (operation_key, repository_key) in [
+            ("blank-repository-key", "   "),
+            ("spaced-repository-key", " platform "),
+        ] {
+            let request = WorkspaceCreateRequest {
+                operation_key: operation_key.to_string(),
+                display_name: "Workspace A".to_string(),
+                repository: InitialRepositoryIntent {
+                    uri: repository.path().display().to_string(),
+                    repository_key: repository_key.to_string(),
+                    default_ref: None,
+                },
+            };
+
+            let error = service
+                .create(request, owner_account_id.clone())
+                .unwrap_err()
+                .to_string();
+
+            assert!(error.contains("invalid Repository key"), "{error}");
+            assert!(store.list_workspaces().unwrap().is_empty());
+        }
+    }
+
     #[tokio::test]
     async fn create_recovers_same_reserved_identity_after_material_write_failure() {
         let temp = tempfile::tempdir().unwrap();

crates/workspace-server/src/workspace_deletion.rs

@@ -73,12 +73,11 @@ const WORKSPACE_DELETION_PURGE_TABLES: &[&str] = &[
     "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_archive_observe_grants",
     "worker_session_archives",
     "worker_singleton_owners",
     "worker_tombstones",

crates/workspace-server/src/workspace_subscription.rs

@@ -271,7 +271,6 @@ async fn run_worker_protocol(
                     worker_id: worker_id.clone(),
                     runtime_id: Some(runtime_id),
                 },
-                snapshot_revision: 0,
                 snapshot: SubscriptionSnapshot::WorkerProtocol {
                     worker_id: worker_id.clone(),
                     events: vec![initial_event],
@@ -284,13 +283,10 @@ async fn run_worker_protocol(
     {
         return;
     }
-    let mut subject_revision = 0_u64;
     while let Some(event) = events.recv().await {
-        subject_revision = subject_revision.saturating_add(1);
         let frame =
             SubscriptionFrame::new(SubscriptionFramePayload::Event(SubscriptionEvent::Event {
                 subscription_id: subscription_id.clone(),
-                subject_revision,
                 payload: SubscriptionEventPayload::WorkerProtocol {
                     worker_id: worker_id.clone(),
                     event,
@@ -333,12 +329,10 @@ async fn run_workspace_workers(
     subscription_id: SubscriptionId,
     outbound: mpsc::Sender<WsMessage>,
 ) {
-    // Subscribe before reading so commits racing with the snapshot are replayed.
-    let mut events = api.worker_projection.subscribe();
-    let Ok((snapshot_revision, records)) = api.store.worker_registry_projection_snapshot(
-        &api.config.workspace_id,
-        api.config.max_records.max(1),
-    ) else {
+    let Ok(mut projection) = api
+        .worker_projection
+        .subscribe_ordered(api.config.max_records.max(1))
+    else {
         let _ = send_rejected(
             &outbound,
             request_id,
@@ -348,7 +342,8 @@ async fn run_workspace_workers(
         .await;
         return;
     };
-    let mut workers = records
+    let mut workers = projection
+        .take_snapshot()
         .into_iter()
         .filter_map(|record| project_registry_worker(&api, record))
         .collect::<Vec<_>>();
@@ -360,7 +355,6 @@ async fn run_workspace_workers(
                 request_id,
                 subscription_id: subscription_id.clone(),
                 selector: EventSubscriptionSelector::WorkspaceWorkers,
-                snapshot_revision,
                 snapshot: SubscriptionSnapshot::Workers { workers },
             },
         )),
@@ -372,16 +366,14 @@ async fn run_workspace_workers(
     }
 
     loop {
-        match events.recv().await {
-            Ok(event) if event.revision <= snapshot_revision => continue,
+        match projection.recv().await {
             Ok(event) => {
                 for change in event.changes {
                     let payload = match change {
                         crate::worker_projection::WorkerProjectionChange::Upsert(record) => {
-                            let Some(mut worker) = project_registry_worker(&api, record) else {
+                            let Some(worker) = project_registry_worker(&api, record) else {
                                 continue;
                             };
-                            worker.subject_revision = event.revision;
                             SubscriptionEventPayload::WorkerUpserted { worker }
                         }
                         crate::worker_projection::WorkerProjectionChange::Removed(worker) => {
@@ -396,7 +388,7 @@ async fn run_workspace_workers(
                             }
                         }
                     };
-                    if send_event(&outbound, &subscription_id, event.revision, payload)
+                    if send_event(&outbound, &subscription_id, payload)
                         .await
                         .is_err()
                     {
@@ -431,7 +423,6 @@ fn project_registry_worker(
     let mut worker = if let Some(observation) = record.observation {
         let mut worker = observation.worker;
         worker.availability = observation.availability;
-        worker.subject_revision = observation.projection_revision;
         worker
     } else {
         SubscriptionWorker {
@@ -442,7 +433,6 @@ fn project_registry_worker(
             runtime_id: Some(runtime_id.clone()),
             resource_key: record.resource_key.clone(),
             availability: protocol::subscription::SubscriptionWorkerAvailability::Unavailable,
-            subject_revision: 0,
             worker_state: None,
             state: protocol::subscription::SubscriptionWorkerState::Stopped,
             has_running_internal_workers: false,
@@ -559,14 +549,12 @@ fn sort_workers(workers: &mut [SubscriptionWorker]) {
 async fn send_event(
     outbound: &mpsc::Sender<WsMessage>,
     subscription_id: &SubscriptionId,
-    subject_revision: u64,
     payload: SubscriptionEventPayload,
 ) -> Result<(), ()> {
     send_frame(
         outbound,
         SubscriptionFrame::new(SubscriptionFramePayload::Event(SubscriptionEvent::Event {
             subscription_id: subscription_id.clone(),
-            subject_revision,
             payload,
         })),
     )
@@ -617,7 +605,6 @@ mod tests {
             runtime_id: Some("runtime-1".to_string()),
             resource_key: Some("W-1".to_string()),
             availability: protocol::subscription::SubscriptionWorkerAvailability::Observed,
-            subject_revision: 1,
             worker_state: None,
             state: protocol::subscription::SubscriptionWorkerState::Idle,
             has_running_internal_workers: false,