fix: complete Runtime verification cutover

crates/worker-runtime/src/http_server.rs

@@ -2126,15 +2126,28 @@ async fn require_runtime_auth(
             unix_now_seconds(),
         ) {
             Ok(context) => {
-                if state.workspace_auth.as_ref().is_some_and(|workspace_auth| {
+                let workspace_verification_exists = match state.workspace_auth.as_deref() {
-                    workspace_auth
+                    Some(workspace_auth) => match workspace_auth
-                        .verifier
+                        .verifications
-                        .has_active_workspace_issuer(&context.workspace_id)
+                        .get(&context.workspace_id, workspace_auth.signer.runtime_id())
-                }) {
+                    {
+                        Ok(record) => record.is_some(),
+                        Err(error) => {
+                            return RuntimeHttpRestError::new(
+                                StatusCode::SERVICE_UNAVAILABLE,
+                                "workspace_runtime_verification_unavailable",
+                                error.to_string(),
+                            )
+                            .into_response();
+                        }
+                    },
+                    None => false,
+                };
+                if workspace_verification_exists {
                     return RuntimeHttpRestError::new(
                         StatusCode::FORBIDDEN,
                         "workspace_identity_required",
-                        "Legacy Server-issued capability is disabled for this Workspace",
+                        "Legacy Server-issued capability is disabled after signed Workspace Runtime verification",
                     )
                     .into_response();
                 }

crates/workspace-server/src/hosts.rs

@@ -3433,11 +3433,10 @@ fn workspace_runtime_operation(method: &str, path_and_query: &str) -> &'static s
     {
         return "workers:create";
     }
-    if path.ends_with("/input")
+    if path.ends_with("/workspace-api") {
-        || path.ends_with("/restore")
+        return "workers:create";
-        || path.ends_with("/workspace-api")
+    }
-        || path.contains("/attachments")
+    if path.ends_with("/input") || path.ends_with("/restore") || path.contains("/attachments") {
-    {
         return "workers:input";
     }
     if path.ends_with("/stop") || path.ends_with("/cancel") {
@@ -4556,13 +4555,20 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
         &self,
         worker_id: &str,
     ) -> Option<crate::observation::RuntimeObservationSource> {
+        let path = format!("/v1/workers/{worker_id}/protocol/ws");
+        let bearer_token = match &self.workspace_authorization {
+            Some(authorization) => authorization
+                .issue("GET", &path, "workers:protocol", Some(worker_id), &[])
+                .ok(),
+            None => self
+                .runtime_capability_token(&path)
+                .or_else(|| self.bearer_token.clone()),
+        };
         Some(crate::observation::RuntimeObservationSource::remote_ws(
             crate::observation::RuntimeObservationSourceConfig {
                 worker: RuntimeWorkerRef::new(&self.runtime_id, worker_id),
                 endpoint: self.ws_endpoint(worker_id),
-                bearer_token: self
+                bearer_token,
-                    .runtime_capability_token(&format!("/v1/workers/{worker_id}/protocol"))
-                    .or_else(|| self.bearer_token.clone()),
             },
         ))
     }
@@ -6631,6 +6637,19 @@ mod tests {
         assert!(!format!("{failure:?}").contains("secret-token"));
     }
 
+    #[test]
+    fn workspace_runtime_operation_matches_runtime_permission_classifier() {
+        let worker_id = EmbeddedWorkerId::from_legacy_u64(1).to_string();
+        assert_eq!(
+            workspace_runtime_operation("POST", &format!("/v1/workers/{worker_id}/workspace-api")),
+            "workers:create"
+        );
+        assert_eq!(
+            workspace_runtime_operation("GET", &format!("/v1/workers/{worker_id}/protocol/ws")),
+            "workers:protocol"
+        );
+    }
+
     #[test]
     fn remote_runtime_registry_routes_commands_without_browser_secret_leaks() {
         let worker_id = EmbeddedWorkerId::from_legacy_u64(1).to_string();

crates/workspace-server/src/server.rs

@@ -2123,23 +2123,12 @@ impl WorkspaceApi {
             ))
         })?;
         let runtime_binding_store = store.clone();
-        let configured_runtime_endpoints = config
-            .remote_runtime_sources
-            .iter()
-            .filter_map(|source| {
-                (source.workspace_id.as_deref() == Some(config.workspace_id.as_str()))
-                    .then(|| (source.runtime_id.clone(), source.base_url.clone()))
-            })
-            .collect::<HashMap<_, _>>();
         let expected_runtime_bindings = store
             .list_workspace_runtime_bindings(&config.workspace_id, false)
             .await?
             .into_iter()
             .filter(|binding| binding.runtime_id != EMBEDDED_RUNTIME_ID)
-            .filter(|binding| binding.state == StoredRuntimeBindingState::Verified)
+            .filter(|binding| binding.state != StoredRuntimeBindingState::Revoked)
-            .filter(|binding| {
-                configured_runtime_endpoints.get(&binding.runtime_id) == Some(&binding.base_url)
-            })
             .map(|binding| {
                 (
                     (binding.workspace_id.clone(), binding.runtime_id.clone()),
@@ -2171,6 +2160,7 @@ impl WorkspaceApi {
                         .unwrap_or(false)
                 })
         });
+        {
             let active_expectations = api
                 .runtime_binding_expectations
                 .read()
@@ -2183,7 +2173,21 @@ impl WorkspaceApi {
                         .unregister_runtime(&source.runtime_id);
                 }
             }
-        drop(active_expectations);
+        }
+        for binding in api
+            .store
+            .list_workspace_runtime_bindings(&api.config.workspace_id, false)
+            .await?
+            .into_iter()
+            .filter(|binding| {
+                binding.authentication_mode
+                    == crate::store::WorkspaceRuntimeAuthenticationMode::WorkspaceIdentity
+            })
+            .filter(|binding| binding.state != StoredRuntimeBindingState::Revoked)
+        {
+            let activate = binding.state == StoredRuntimeBindingState::Verified;
+            api.register_workspace_runtime_binding(binding, activate)?;
+        }
         Ok(api)
     }
 
@@ -2342,6 +2346,37 @@ impl WorkspaceApi {
         self.config.workspace_id.as_str()
     }
 
+    fn register_workspace_runtime_binding(
+        &self,
+        binding: WorkspaceRuntimeBinding,
+        activate: bool,
+    ) -> Result<()> {
+        let backend_url = self
+            .config
+            .backend_base_url
+            .clone()
+            .unwrap_or_else(|| "http://127.0.0.1:8787".to_string());
+        let mut remote_config = remote_runtime_config_from_binding(&binding)
+            .map_err(|diagnostic| Error::Store(diagnostic.message))?;
+        remote_config.workspace_authorization = Some(WorkspaceRuntimeAuthorization::new(
+            self.store.clone(),
+            self.signing_identities.clone(),
+            backend_url.clone(),
+            activate.then_some(binding),
+        ));
+        let remote_runtime = RemoteWorkerRuntime::new(
+            remote_config.clone(),
+            self.config.workspace_id.clone(),
+            backend_url,
+        )
+        .map(|runtime| runtime.with_resource_broker(self.resource_broker.clone()))
+        .map_err(|error| error.into_error())?;
+        self.runtime.register_or_replace(remote_runtime);
+        self.runtime_subscription_broker
+            .register_remote_runtime(remote_config);
+        Ok(())
+    }
+
     pub fn runtime_subscription_broker(&self) -> &RuntimeSubscriptionBroker {
         &self.runtime_subscription_broker
     }
@@ -13629,16 +13664,20 @@ async fn create_remote_runtime(
         updated_at: now,
         revoked_at: None,
     };
-    let (mutation, _) = api
+    let (mutation, binding) = api
         .store
         .put_workspace_runtime_binding_key(record, request.expected_revision, &actor.account_id)
         .await?;
     api.runtime_binding_expectations
         .write()
         .unwrap_or_else(std::sync::PoisonError::into_inner)
-        .remove(&(api.config.workspace_id.clone(), runtime_id.clone()));
+        .insert(
+            (api.config.workspace_id.clone(), runtime_id.clone()),
+            binding.clone(),
+        );
     api.runtime_subscription_broker
         .unregister_runtime(&runtime_id);
+    api.register_workspace_runtime_binding(binding, false)?;
     let resource = workspace_runtime_resources_response(&api, &api.config.workspace_id)
         .await?
         .items
@@ -13778,6 +13817,7 @@ async fn perform_workspace_runtime_verification(
         .await
         .map_err(|error| error.to_string())?;
 
+    let result = async {
         let challenge_body = serde_json::to_vec(&challenge).map_err(|error| error.to_string())?;
         let challenge_claims = WorkspaceCapabilityClaims {
             issuer: backend_url.to_string(),
@@ -13888,12 +13928,27 @@ async fn perform_workspace_runtime_verification(
             last_outcome: "verified".to_string(),
             verified_at: Some(verified_at.clone()),
             checked_at: verified_at,
-        ..pending
+            ..pending.clone()
         };
         api.store
             .complete_workspace_runtime_verification(&verified)
             .await
             .map_err(|error| error.to_string())
+    }
+    .await;
+    if result.is_err() {
+        let checked_at = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);
+        let _ = api
+            .store
+            .record_workspace_runtime_verification_outcome_if_current(
+                &pending,
+                "failed",
+                "verification_failed",
+                &checked_at,
+            )
+            .await;
+    }
+    result
 }
 
 async fn test_runtime_connection(
@@ -13918,6 +13973,7 @@ async fn test_runtime_connection(
     if binding.authentication_mode == StoredRuntimeAuthenticationMode::WorkspaceIdentity {
         match perform_workspace_runtime_verification(&api, api.runtime.clone(), &binding).await {
             Ok(verified_binding) => {
+                api.register_workspace_runtime_binding(verified_binding.clone(), true)?;
                 api.runtime_binding_expectations
                     .write()
                     .unwrap_or_else(std::sync::PoisonError::into_inner)
@@ -13933,22 +13989,6 @@ async fn test_runtime_connection(
                     .map_err(|error| Error::Store(format!("{error:?}")))?;
             }
             Err(message) => {
-                if let Ok(Some(mut evidence)) = api
-                    .store
-                    .get_workspace_runtime_verification(api.workspace_id(), &runtime_id)
-                    .await
-                {
-                    if evidence.state != "verified" {
-                        evidence.state = "failed".to_string();
-                        evidence.verified_at = None;
-                    }
-                    evidence.last_outcome = "verification_failed".to_string();
-                    evidence.checked_at = Utc::now().to_rfc3339();
-                    let _ = api
-                        .store
-                        .record_workspace_runtime_verification_attempt(&evidence)
-                        .await;
-                }
                 let mut result = runtime_connection_test_failure(
                     api.workspace_id(),
                     &runtime_id,
@@ -13981,15 +14021,19 @@ async fn test_runtime_connection(
 
     if ping.is_err()
         && binding.authentication_mode == StoredRuntimeAuthenticationMode::WorkspaceIdentity
-        && let Some(mut evidence) = api
+        && let Some(evidence) = api
             .store
             .get_workspace_runtime_verification(api.workspace_id(), &runtime_id)
             .await?
     {
-        evidence.last_outcome = "connectivity_failed".to_string();
+        let checked_at = Utc::now().to_rfc3339();
-        evidence.checked_at = Utc::now().to_rfc3339();
         api.store
-            .record_workspace_runtime_verification_attempt(&evidence)
+            .record_workspace_runtime_verification_outcome_if_current(
+                &evidence,
+                "failed",
+                "connectivity_failed",
+                &checked_at,
+            )
             .await?;
     }
     let current_binding = api
@@ -16505,7 +16549,6 @@ fn validate_public_runtime_id(runtime_id: &str) -> ApiResult<()> {
     Ok(())
 }
 
-#[cfg(test)]
 fn remote_runtime_config_from_binding(
     binding: &crate::store::WorkspaceRuntimeBinding,
 ) -> std::result::Result<RemoteRuntimeConfig, RuntimeDiagnostic> {
@@ -20349,6 +20392,98 @@ mod tests {
         assert!(!serialized.contains("materialized_path"));
     }
 
+    #[tokio::test]
+    async fn verified_workspace_runtime_binding_is_restored_into_live_registry() {
+        let dir = tempfile::tempdir().unwrap();
+        let root = dir.path().to_path_buf();
+        let api = test_api(&root).await;
+        let actor = test_owner_actor();
+        let identity = api
+            .signing_identities
+            .provision_existing(&api.config.workspace_id, &actor.account_id)
+            .unwrap();
+        let runtime_identity = RuntimeIdentityMaterial::generate("restored-runtime").unwrap();
+        api.store
+            .upsert_workspace_runtime_binding_record(
+                WorkspaceRuntimeBinding {
+                    workspace_id: api.config.workspace_id.clone(),
+                    runtime_id: "restored-runtime".to_string(),
+                    display_name: "Restored Runtime".to_string(),
+                    base_url: "https://8.8.8.8".to_string(),
+                    public_key: runtime_identity.public_key,
+                    public_key_fingerprint: String::new(),
+                    binding_revision: 1,
+                    state: StoredRuntimeBindingState::Configured,
+                    authentication_mode: StoredRuntimeAuthenticationMode::WorkspaceIdentity,
+                    workspace_key_id: Some(identity.key_id.clone()),
+                    workspace_key_generation: Some(identity.revision),
+                    created_at: "1".to_string(),
+                    updated_at: "1".to_string(),
+                    revoked_at: None,
+                },
+                false,
+            )
+            .await
+            .unwrap();
+        let configured = api
+            .store
+            .get_workspace_runtime_binding(&api.config.workspace_id, "restored-runtime")
+            .await
+            .unwrap()
+            .unwrap();
+        let evidence = crate::store::WorkspaceRuntimeVerificationEvidence {
+            workspace_id: configured.workspace_id.clone(),
+            runtime_id: configured.runtime_id.clone(),
+            binding_revision: configured.binding_revision,
+            workspace_key_id: identity.key_id,
+            workspace_identity_revision: identity.revision,
+            workspace_trust_generation: identity.revision,
+            runtime_public_key_fingerprint: configured.public_key_fingerprint.clone(),
+            runtime_identity_revision: 1,
+            challenge_id: "restart-challenge".to_string(),
+            state: "verified".to_string(),
+            last_outcome: "verified".to_string(),
+            verified_at: Some("2".to_string()),
+            checked_at: "2".to_string(),
+        };
+        api.store
+            .record_workspace_runtime_verification_attempt(&evidence)
+            .await
+            .unwrap();
+        api.store
+            .complete_workspace_runtime_verification(&evidence)
+            .await
+            .unwrap();
+        drop(api);
+
+        let restored_config = test_server_config(&root);
+        let restored_store =
+            SqliteWorkspaceStore::open(restored_config.database_path.clone()).unwrap();
+        let restored = WorkspaceApi::new(restored_config, Arc::new(restored_store))
+            .await
+            .unwrap();
+        assert!(
+            restored
+                .runtime
+                .list_runtimes(100)
+                .items
+                .iter()
+                .any(|runtime| runtime.runtime_id == "restored-runtime"),
+            "verified persisted Runtime binding must return to the live registry"
+        );
+        assert!(
+            restored
+                .runtime_binding_expectations
+                .read()
+                .unwrap()
+                .contains_key(&(
+                    restored.config.workspace_id.clone(),
+                    "restored-runtime".to_string(),
+                )),
+            "restored Runtime must retain its revision-fenced binding expectation"
+        );
+    }
+
     #[tokio::test]
     async fn remote_runtime_registration_is_workspace_scoped_revisioned_and_configured() {
         let dir = tempfile::tempdir().unwrap();
@@ -20416,18 +20551,18 @@ mod tests {
                 .list_runtimes(100)
                 .items
                 .iter()
-                .all(|runtime| runtime.runtime_id != "configured-runtime"),
+                .any(|runtime| runtime.runtime_id == "configured-runtime"),
-            "configured binding must remove a stale active Runtime projection"
+            "configured binding must remain registered so verification can reach it"
         );
         assert!(
-            !api.runtime_binding_expectations
+            api.runtime_binding_expectations
                 .read()
                 .unwrap()
                 .contains_key(&(
                     api.config.workspace_id.clone(),
                     "configured-runtime".to_string(),
                 )),
-            "configured binding must not remain a control expectation"
+            "configured binding must remain fenced by its current binding revision"
         );
         let replacement_identity = RuntimeIdentityMaterial::generate("configured-runtime").unwrap();
         let generic_put = scoped_put_runtime_trust_key(

crates/workspace-server/src/store.rs

@@ -771,6 +771,13 @@ pub trait ControlPlaneStore: Send + Sync + WorkspaceDeletionStore {
         &self,
         evidence: &WorkspaceRuntimeVerificationEvidence,
     ) -> Result<()>;
+    async fn record_workspace_runtime_verification_outcome_if_current(
+        &self,
+        expected: &WorkspaceRuntimeVerificationEvidence,
+        state: &str,
+        outcome: &str,
+        checked_at: &str,
+    ) -> Result<bool>;
     async fn complete_workspace_runtime_verification(
         &self,
         evidence: &WorkspaceRuntimeVerificationEvidence,
@@ -2336,6 +2343,65 @@ impl SqliteWorkspaceStore {
         })
     }
 
+    pub fn record_workspace_runtime_verification_outcome_if_current(
+        &self,
+        expected: &WorkspaceRuntimeVerificationEvidence,
+        state: &str,
+        outcome: &str,
+        checked_at: &str,
+    ) -> Result<bool> {
+        validate_workspace_runtime_verification(expected)?;
+        if !matches!(state, "pending" | "verified" | "failed") {
+            return Err(Error::Store(
+                "invalid Workspace Runtime verification state".to_string(),
+            ));
+        }
+        if !matches!(
+            outcome,
+            "challenge_issued" | "verified" | "verification_failed" | "connectivity_failed"
+        ) {
+            return Err(Error::Store(
+                "invalid Workspace Runtime verification outcome".to_string(),
+            ));
+        }
+        if checked_at.is_empty() || checked_at.len() > 128 {
+            return Err(Error::Store(
+                "invalid Workspace Runtime verification checked_at".to_string(),
+            ));
+        }
+        self.with_conn(|conn| {
+            let changed = conn.execute(
+                r#"UPDATE workspace_runtime_verifications
+                   SET state = CASE WHEN state = 'verified' THEN state ELSE ?10 END,
+                       last_outcome = ?11,
+                       checked_at = ?12
+                   WHERE workspace_id = ?1 AND runtime_id = ?2
+                     AND binding_revision = ?3
+                     AND workspace_key_id = ?4
+                     AND workspace_identity_revision = ?5
+                     AND workspace_trust_generation = ?6
+                     AND runtime_public_key_fingerprint = ?7
+                     AND runtime_identity_revision = ?8
+                     AND challenge_id = ?9"#,
+                params![
+                    expected.workspace_id,
+                    expected.runtime_id,
+                    expected.binding_revision,
+                    expected.workspace_key_id,
+                    expected.workspace_identity_revision,
+                    expected.workspace_trust_generation,
+                    expected.runtime_public_key_fingerprint,
+                    expected.runtime_identity_revision,
+                    expected.challenge_id,
+                    state,
+                    outcome,
+                    checked_at,
+                ],
+            )?;
+            Ok(changed == 1)
+        })
+    }
+
     pub fn complete_workspace_runtime_verification(
         &self,
         evidence: &WorkspaceRuntimeVerificationEvidence,
@@ -2348,6 +2414,36 @@ impl SqliteWorkspaceStore {
         }
         self.with_conn_mut(|conn| {
             let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
+            let current_attempt = tx.query_row(
+                r#"SELECT EXISTS(
+                       SELECT 1 FROM workspace_runtime_verifications
+                       WHERE workspace_id = ?1 AND runtime_id = ?2
+                         AND binding_revision = ?3
+                         AND workspace_key_id = ?4
+                         AND workspace_identity_revision = ?5
+                         AND workspace_trust_generation = ?6
+                         AND runtime_public_key_fingerprint = ?7
+                         AND runtime_identity_revision = ?8
+                         AND challenge_id = ?9
+                   )"#,
+                params![
+                    evidence.workspace_id,
+                    evidence.runtime_id,
+                    evidence.binding_revision,
+                    evidence.workspace_key_id,
+                    evidence.workspace_identity_revision,
+                    evidence.workspace_trust_generation,
+                    evidence.runtime_public_key_fingerprint,
+                    evidence.runtime_identity_revision,
+                    evidence.challenge_id,
+                ],
+                |row| row.get::<_, bool>(0),
+            )?;
+            if !current_attempt {
+                return Err(Error::RuntimeBindingConflict(
+                    "Runtime verification attempt was superseded".to_string(),
+                ));
+            }
             let changed = tx.execute(
                 r#"UPDATE workspace_runtime_bindings
                    SET state = 'verified', updated_at = ?7
@@ -3193,6 +3289,18 @@ impl ControlPlaneStore for SqliteWorkspaceStore {
         SqliteWorkspaceStore::record_workspace_runtime_verification_attempt(self, evidence)
     }
 
+    async fn record_workspace_runtime_verification_outcome_if_current(
+        &self,
+        expected: &WorkspaceRuntimeVerificationEvidence,
+        state: &str,
+        outcome: &str,
+        checked_at: &str,
+    ) -> Result<bool> {
+        SqliteWorkspaceStore::record_workspace_runtime_verification_outcome_if_current(
+            self, expected, state, outcome, checked_at,
+        )
+    }
+
     async fn complete_workspace_runtime_verification(
         &self,
         evidence: &WorkspaceRuntimeVerificationEvidence,
@@ -9857,6 +9965,9 @@ mod tests {
             verified_at: Some("2".to_string()),
             checked_at: "2".to_string(),
         };
+        store
+            .record_workspace_runtime_verification_attempt(&evidence)
+            .unwrap();
         let verified = store
             .complete_workspace_runtime_verification(&evidence)
             .unwrap();
@@ -9886,6 +9997,43 @@ mod tests {
         store
             .complete_workspace_runtime_verification(&evidence)
             .unwrap();
+        let newer_pending = WorkspaceRuntimeVerificationEvidence {
+            challenge_id: "challenge-b".to_string(),
+            state: "pending".to_string(),
+            last_outcome: "challenge_issued".to_string(),
+            verified_at: None,
+            checked_at: "4".to_string(),
+            ..evidence.clone()
+        };
+        store
+            .record_workspace_runtime_verification_attempt(&newer_pending)
+            .unwrap();
+        let newer_verified = WorkspaceRuntimeVerificationEvidence {
+            state: "verified".to_string(),
+            last_outcome: "verified".to_string(),
+            verified_at: Some("5".to_string()),
+            checked_at: "5".to_string(),
+            ..newer_pending
+        };
+        store
+            .complete_workspace_runtime_verification(&newer_verified)
+            .unwrap();
+        assert!(
+            !store
+                .record_workspace_runtime_verification_outcome_if_current(
+                    &pending_retry,
+                    "failed",
+                    "verification_failed",
+                    "6",
+                )
+                .unwrap()
+        );
+        assert_eq!(
+            store
+                .get_workspace_runtime_verification("workspace-a", "runtime-a")
+                .unwrap(),
+            Some(newer_verified.clone())
+        );
         drop(store);
 
         let reopened = SqliteWorkspaceStore::open(&path).unwrap();
@@ -9893,7 +10041,7 @@ mod tests {
             reopened
                 .get_workspace_runtime_verification("workspace-a", "runtime-a")
                 .unwrap(),
-            Some(evidence)
+            Some(newer_verified)
         );
         assert_eq!(
             reopened