fix: interrupt revoked Plugin transports and enforce full control-plane exclusions

crates/plugin-runtime/src/owner.rs

@@ -44,6 +44,12 @@ pub trait PluginHostOwner: Send + Sync {
     }
 }
 
+/// Finite generated-client response budget for the supported one-MiB binary
+/// body, its worst-case JSON numeric-array expansion, headers and envelope.
+/// Set `response_body_limit` on the injected client before constructing a Remote
+/// owner; a smaller operator-selected limit may reject otherwise valid replies.
+pub const PLUGIN_HOST_RESPONSE_BODY_LIMIT: usize = 4 * 1024 * 1024 + 512 * 1024;
+
 /// Uses an injected, already configured generated client (including its existing
 /// Runtime request authorizer). This adapter never creates or owns credentials.
 pub struct RemotePluginHostOwner<A> {

crates/workspace-server/src/plugin_https/test_support.rs

@@ -108,6 +108,7 @@ pub(crate) struct Peer {
     pub(super) root: reqwest::Certificate,
     pub(super) observations: mpsc::Receiver<Vec<u8>>,
     pub(super) connections: Arc<AtomicUsize>,
+    closed: mpsc::Receiver<()>,
     task: JoinHandle<()>,
     _directory: tempfile::TempDir,
 }
@@ -164,6 +165,7 @@ impl Peer {
         let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
         let address = listener.local_addr().unwrap();
         let (observed, observations) = mpsc::channel(16);
+        let (closed_sender, closed) = mpsc::channel(16);
         let connections = Arc::new(AtomicUsize::new(0));
         let count = connections.clone();
         let mut replies = VecDeque::from(replies);
@@ -176,6 +178,7 @@ impl Peer {
                         count.fetch_add(1, Ordering::SeqCst);
                         let acceptor = acceptor.clone();
                         let observed = observed.clone();
+                        let closed_sender = closed_sender.clone();
                         let reply = replies.pop_front().unwrap_or(Reply::Close);
                         handlers.spawn(async move {
                             let Ok(mut socket) = acceptor.accept(socket).await else { return; };
@@ -185,7 +188,7 @@ impl Peer {
                                 let Ok(length) = socket.read(&mut buffer).await else { return; };
                                 if length == 0 { return; }
                                 bytes.extend_from_slice(&buffer[..length]);
-                                assert!(bytes.len() <= 64 * 1024, "test request exceeded fixture bound");
+                                assert!(bytes.len() <= 2 * 1024 * 1024, "test request exceeded fixture bound");
                                 if let Some(end) = bytes.windows(4).position(|part| part == b"\r\n\r\n") {
                                     let text = String::from_utf8_lossy(&bytes[..end]).to_lowercase();
                                     let length = text.lines().find_map(|line| line.strip_prefix("content-length: "))
@@ -197,7 +200,13 @@ impl Peer {
                             match reply {
                                 Reply::Bytes(bytes) => { let _ = socket.write_all(&bytes).await; let _ = socket.shutdown().await; }
                                 Reply::StallAfter(bytes) => { let _ = socket.write_all(&bytes).await; pending::<()>().await; }
-                                Reply::Stall => pending::<()>().await,
+                                Reply::Stall => {
+                                    let mut byte = [0];
+                                    match socket.read(&mut byte).await {
+                                        Ok(0) | Err(_) => { let _ = closed_sender.send(()).await; }
+                                        Ok(_) => panic!("unexpected bytes after complete request"),
+                                    }
+                                },
                                 Reply::Close => {},
                                 Reply::Gated(release) => {
                                     if let Ok(bytes) = release.await {
@@ -217,6 +226,7 @@ impl Peer {
             root: reqwest::Certificate::from_pem(&certificate).unwrap(),
             observations,
             connections,
+            closed,
             task,
             _directory: directory,
         }
@@ -232,6 +242,19 @@ impl Peer {
         self.connections.load(Ordering::SeqCst)
     }
 
+    pub(crate) async fn closed(&mut self) {
+        tokio::time::timeout(Duration::from_secs(2), self.closed.recv())
+            .await
+            .unwrap()
+            .unwrap();
+    }
+
+    pub(crate) fn transport_with_policy(&self, policy: TransportPolicy) -> HttpsTransport {
+        let mut transport = self.transport_for_requests(4);
+        transport.policy = policy;
+        transport
+    }
+
     pub(crate) fn transport(&self) -> HttpsTransport {
         self.transport_for_requests(1)
     }

crates/workspace-server/src/server.rs

@@ -307,6 +307,9 @@ pub struct ServerConfig {
     pub runtime_event_sources: Vec<RuntimeObservationSourceConfig>,
     pub remote_runtime_sources: Vec<RemoteRuntimeConfig>,
     pub backend_base_url: Option<String>,
+    // Operational deny inventory survives Workspace capability filtering. It is
+    // never deserialized from Workspace configuration or returned to a guest.
+    pub(crate) control_plane_endpoints: std::collections::BTreeSet<String>,
 }
 
 impl ServerConfig {
@@ -344,6 +347,7 @@ impl ServerConfig {
             runtime_event_sources: Vec::new(),
             remote_runtime_sources: Vec::new(),
             backend_base_url: None,
+            control_plane_endpoints: Default::default(),
         }
     }
 
@@ -414,6 +418,28 @@ impl ServerConfig {
         self
     }
 
+    fn plugin_control_plane_endpoints(&self) -> std::collections::BTreeSet<String> {
+        let AuthConfig::Passkey {
+            public_base_url, ..
+        } = &self.auth;
+        self.control_plane_endpoints
+            .iter()
+            .chain(std::iter::once(public_base_url))
+            .chain(self.backend_base_url.iter())
+            .chain(
+                self.remote_runtime_sources
+                    .iter()
+                    .map(|source| &source.base_url),
+            )
+            .chain(
+                self.runtime_event_sources
+                    .iter()
+                    .map(|source| &source.endpoint),
+            )
+            .cloned()
+            .collect()
+    }
+
     fn for_catalog_workspace(
         &self,
         workspace: &WorkspaceRecord,
@@ -456,6 +482,7 @@ impl ServerConfig {
         scoped.repositories = repositories;
         // Runtime trust is server-global. Only explicitly assigned sources enter
         // this Workspace's registry and receive Workspace-scoped capabilities.
+        scoped.control_plane_endpoints = self.plugin_control_plane_endpoints();
         scoped.remote_runtime_sources.retain(|runtime| {
             runtime.workspace_id.as_deref() == Some(workspace.workspace_id.as_str())
         });

crates/workspace-server/src/server_db_execution_tests.rs

@@ -77,6 +77,7 @@ impl Fixture {
             runtime_event_sources: Vec::new(),
             remote_runtime_sources: Vec::new(),
             backend_base_url: Some("http://localhost:8787".into()),
+            control_plane_endpoints: Default::default(),
         };
         let store = Arc::new(
             SqliteWorkspaceStore::open_for_test(&config.database_path)

crates/workspace-server/src/server_plugin_host.rs

@@ -77,7 +77,14 @@ use tokio_util::sync::CancellationToken;
 const RECEIPT_RETENTION_MS: i64 = 30 * 24 * 60 * 60 * 1000;
 
 pub(super) fn router(service: Arc<ServerApiContractService>) -> Router {
-    server_api_axum::plugin_host(service).layer(DefaultBodyLimit::max(2 * 1024 * 1024))
+    // Vec<u8> uses JSON's numeric array representation (up to four wire bytes
+    // per decoded byte). Reserve bounded room for IDs, URL/query and escaped
+    // headers; the shared owner still enforces every decoded field/body limit.
+    let body_ceiling = crate::plugin_binding::PluginConnectionPolicyCeiling::default()
+        .limits
+        .max_request_bytes as usize;
+    server_api_axum::plugin_host(service)
+        .layer(DefaultBodyLimit::max(4 * body_ceiling + 512 * 1024))
 }
 
 fn denied() -> PluginHostError {
@@ -623,8 +630,7 @@ async fn execute_with_transport(
                         None => Arc::new(transport(&api).await?),
                     };
                     let cancellation = CancellationToken::new();
-                    let result = transport
-                        .execute_with_guard(
+                    let sending = transport.execute_with_guard(
                         binding,
                         prepared,
                         secret.as_deref().map(|v| v.as_str()),
@@ -641,14 +647,24 @@ async fn execute_with_transport(
                             .map(|_| ())
                             .map_err(|_| https::TransportError::Denied)
                         },
-                        )
-                        .await
-                        .map(|response| PluginHttpResponse {
+                    );
+                    // Monitor the existing durable owners rather than maintaining
+                    // another revocation registry. Dropping the transport future
+                    // shuts down its single-use client/socket. A stalled authority
+                    // owner also closes the attempt; it is never permission to run
+                    // until the HTTP deadline. Admission remains consumed even if
+                    // the shutdown acknowledgement or completion write is lost.
+                    let result = tokio::select! {
+                        result = sending => result.map(|response| PluginHttpResponse {
                             status: response.status,
                             headers: response.headers,
                             body: response.body,
-                        })
-                        .map_err(sent_transport_error);
+                        }).map_err(sent_transport_error),
+                        error = monitor_invocation(api.clone(), &context, &workspace, &invocation_id, operation.clone()) => {
+                            cancellation.cancel();
+                            Err(error)
+                        }
+                    };
                     let store = api.config_store.clone();
                     let r = record;
                     let fp = fingerprint;
@@ -699,25 +715,92 @@ async fn execute_with_transport(
     }
 }
 
+async fn monitor_invocation(
+    api: Arc<WorkspaceApi>,
+    context: &ServerRequestContext,
+    workspace: &str,
+    invocation_id: &str,
+    operation: PluginHostOperation,
+) -> PluginHostError {
+    loop {
+        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
+        let checked = tokio::time::timeout(
+            std::time::Duration::from_secs(1),
+            invocation(
+                api.clone(),
+                context,
+                workspace,
+                invocation_id,
+                operation.clone(),
+            ),
+        )
+        .await;
+        let error = match checked {
+            Ok(Ok(_)) => continue,
+            Ok(Err(error)) => error,
+            Err(_) => policy::map_error(policy::failure(PluginHostErrorCode::Unavailable)),
+        };
+        // After admission the request may have reached the peer. Neither a
+        // failed check nor a local connection shutdown proves remote rollback.
+        return PluginHostError {
+            effect: PluginEffectObservation::Indeterminate,
+            ..error
+        };
+    }
+}
+
 async fn transport(
     api: &WorkspaceApi,
 ) -> std::result::Result<https::HttpsTransport, PluginHostError> {
+    let policy = transport_policy(api, |host, port| async move {
+        tokio::net::lookup_host((host, port))
+            .await
+            .map(|addresses| addresses.take(65).collect::<Vec<_>>())
+            .map_err(|_| https::TransportError::Dns)
+    })
+    .await?;
+    Ok(https::HttpsTransport::new(policy))
+}
+
+pub(super) async fn transport_policy<F, Fut>(
+    api: &WorkspaceApi,
+    resolve: F,
+) -> std::result::Result<https::TransportPolicy, PluginHostError>
+where
+    F: Fn(String, u16) -> Fut,
+    Fut: std::future::Future<
+            Output = std::result::Result<Vec<std::net::SocketAddr>, https::TransportError>,
+        >,
+{
     let mut hosts = std::collections::BTreeSet::new();
     let mut ips = std::collections::BTreeSet::new();
-    let AuthConfig::Passkey {
-        public_base_url, ..
-    } = &api.config.auth;
-    for address in std::iter::once(public_base_url)
-        .chain(api.config.backend_base_url.iter())
-        .chain(
-            api.config
-                .remote_runtime_sources
+    // Current server-owned bindings include dynamically registered endpoints in
+    // other Workspaces, even when they were absent from the startup config.
+    let store = api.config_store.clone();
+    let registered = store
+        .db_executor()
+        .run(move || {
+            store.with_conn(|conn| {
+                let mut statement =
+                    conn.prepare("SELECT DISTINCT base_url FROM workspace_runtime_bindings")?;
+                let rows = statement.query_map([], |row| row.get::<_, String>(0))?;
+                Ok(rows.collect::<std::result::Result<Vec<_>, _>>()?)
+            })
+        })
+        .await
+        .map_err(policy::map_error)?;
+    for address in api
+        .config
+        .plugin_control_plane_endpoints()
         .iter()
-                .map(|source| &source.base_url),
-        )
+        .chain(registered.iter())
     {
         if let Ok(url) = url::Url::parse(address) {
+            if !matches!(url.scheme(), "http" | "https" | "ws" | "wss") {
+                continue;
+            }
             if let Some(host) = url.host_str() {
+                let host = host.trim_start_matches('[').trim_end_matches(']');
                 hosts.insert(host.to_ascii_lowercase());
                 if let Ok(ip) = host.parse() {
                     ips.insert(ip);
@@ -725,12 +808,15 @@ async fn transport(
                     let port = url.port_or_known_default().ok_or_else(denied)?;
                     let resolved = tokio::time::timeout(
                         std::time::Duration::from_secs(5),
-                        tokio::net::lookup_host((host, port)),
+                        resolve(host.to_owned(), port),
                     )
                     .await
                     .map_err(|_| transport_error(https::TransportError::Dns))?
                     .map_err(|_| transport_error(https::TransportError::Dns))?;
-                    for (index, address) in resolved.enumerate() {
+                    if resolved.is_empty() {
+                        return Err(transport_error(https::TransportError::Dns));
+                    }
+                    for (index, address) in resolved.into_iter().enumerate() {
                         if index >= 64 {
                             return Err(transport_error(https::TransportError::Dns));
                         }
@@ -740,10 +826,10 @@ async fn transport(
             }
         }
     }
-    Ok(https::HttpsTransport::new(https::TransportPolicy {
+    Ok(https::TransportPolicy {
         denied_addresses: ips,
         denied_hosts: hosts,
-    }))
+    })
 }
 fn sent_transport_error(error: https::TransportError) -> PluginHostError {
     match error {

crates/workspace-server/src/server_plugin_host_tests.rs

@@ -1258,6 +1258,7 @@ async fn remote_owner_for(
     }));
     let client = server_api::ServerApiClient::builder(format!("http://{address}"))
         .unwrap()
+        .response_body_limit(yoi_plugin_runtime::owner::PLUGIN_HOST_RESPONSE_BODY_LIMIT)
         .client(
             reqwest::Client::builder()
                 .no_proxy()
@@ -1557,3 +1558,323 @@ async fn reserved_job_without_canonical_job_binding_cannot_fall_back_to_ordinary
     assert_eq!(f.count("plugin_effect_receipts"), 0);
     f.shutdown().await;
 }
+
+#[tokio::test]
+async fn active_remote_and_embedded_sends_close_on_cancel_revoke_permission_loss_and_owner_failure()
+{
+    use crate::plugin_https::test_support::Reply;
+    use server_api::PluginEffectObservation;
+    for remote in [false, true] {
+        for action in ["cancel", "revoke", "permission", "unavailable"] {
+            let mut peer = Peer::new(HOST, vec![Reply::Stall]).await;
+            let mut f = Fixture::new(Some(peer_binding(&peer))).await;
+            f.api.plugin_https_transport = Some(Arc::new(peer.transport_for_requests(4)));
+            let (remote_owner, server) =
+                remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await;
+            let owner = if remote {
+                remote_owner
+            } else {
+                f.owner(Some(peer.transport_for_requests(4))).await
+            };
+            let host = f.host(owner.clone(), "active-interruption").await;
+            let id = host.context().invocation_id.clone();
+            let request = send(&id, PluginHttpMethod::Post, "once");
+            let mut recovered_api = f.api.clone();
+            recovered_api.config_store =
+                Arc::new(SqliteWorkspaceStore::open_for_test(&f.api.config.database_path).unwrap());
+            let attempt = owner.call(request.clone());
+            tokio::pin!(attempt);
+            tokio::select! {
+                result = &mut attempt => panic!("request must reach the stalled peer: {result:?}"),
+                _ = peer.observed() => {},
+            }
+            match action {
+                "cancel" => host.cancel().await.unwrap(),
+                "revoke" => f.revoke(),
+                "permission" => f.deny(),
+                "unavailable" => f
+                    .api
+                    .config_store
+                    .with_conn(|conn| {
+                        conn.execute_batch(
+                            "ALTER TABLE plugin_execution_plans RENAME TO unavailable_plugin_plans",
+                        )?;
+                        Ok(())
+                    })
+                    .unwrap(),
+                _ => unreachable!(),
+            }
+            // No response is released. HTTP has a five-second timeout; authority
+            // loss must close the owner transport substantially before that.
+            let error = tokio::time::timeout(std::time::Duration::from_secs(2), &mut attempt)
+                .await
+                .expect("active transport did not stop")
+                .unwrap_err();
+            assert_eq!(
+                error.effect,
+                PluginEffectObservation::Indeterminate,
+                "{remote}/{action}"
+            );
+            peer.closed().await;
+            if action == "unavailable" {
+                f.api
+                    .config_store
+                    .with_conn(|conn| {
+                        conn.execute_batch(
+                            "ALTER TABLE unavailable_plugin_plans RENAME TO plugin_execution_plans",
+                        )?;
+                        Ok(())
+                    })
+                    .unwrap();
+            }
+            if action == "permission" {
+                set_policy(
+                    &f.api,
+                    &f.subject,
+                    &f.installed.installation.installation_id,
+                    true,
+                );
+            }
+            // Another actual owner with an independent SQLite connection has no
+            // local attempt state. Reopen after ordinary cancellation/revocation;
+            // the unavailable-owner fault deliberately changes SQLite DDL quotes.
+            if action != "unavailable" {
+                recovered_api.config_store = Arc::new(
+                    SqliteWorkspaceStore::open_for_test(&f.api.config.database_path).unwrap(),
+                );
+            }
+            let recovered = plugin_host::EmbeddedPluginHostOwner::new(
+                Arc::new(recovered_api),
+                f.verified_context().await,
+                TEST_WORKSPACE_ID.into(),
+            )
+            .with_transport(peer.transport_for_requests(4));
+            assert!(recovered.call(request).await.is_err());
+            assert_eq!(f.count("plugin_effect_receipts"), 1);
+            assert_eq!(
+                peer.connection_count(),
+                1,
+                "recovery resent {remote}/{action}"
+            );
+            drop(server);
+            f.shutdown().await;
+        }
+    }
+}
+
+#[tokio::test]
+async fn binary_ceiling_matches_generated_remote_router_and_embedded_owner_limits() {
+    let ceiling = crate::plugin_binding::PluginConnectionPolicyCeiling::default().limits;
+    let body = vec![255; ceiling.max_request_bytes as usize];
+    let mut peer = Peer::new(
+        HOST,
+        vec![response(200, &[], &body), response(200, &[], &body)],
+    )
+    .await;
+    let mut binding = peer_binding(&peer);
+    binding.limits.max_request_bytes = ceiling.max_request_bytes;
+    binding.limits.max_response_bytes = ceiling.max_response_bytes;
+    let mut f = Fixture::new(Some(binding)).await;
+    f.api.plugin_https_transport = Some(Arc::new(peer.transport_for_requests(4)));
+    let embedded = f.owner(Some(peer.transport_for_requests(4))).await;
+    let (remote, server) = remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await;
+    let host = f.host(remote.clone(), "binary-boundary").await;
+    let id = &host.context().invocation_id;
+    for (index, owner) in [&embedded, &remote].into_iter().enumerate() {
+        let mut input = send(id, PluginHttpMethod::Post, &format!("binary-{index}"));
+        if let PluginHostRequest::Send { request, .. } = &mut input {
+            request.body = body.clone();
+        }
+        assert!(serde_json::to_vec(&input).unwrap().len() > 4 * 1024 * 1024);
+        let PluginHostResponse::Response { response } = owner.call(input.clone()).await.unwrap()
+        else {
+            panic!("expected response")
+        };
+        assert_eq!(response.status, 200);
+        assert_eq!(response.body.len(), body.len());
+        assert!(response.body.iter().all(|byte| *byte == 255));
+        let observed = peer.observed().await;
+        assert!(observed.ends_with(&body));
+        assert_eq!(peer.connection_count(), index + 1);
+        // Replays cross adapters without a new connection.
+        embedded.call(input).await.unwrap();
+        let mut oversized = send(id, PluginHttpMethod::Post, &format!("oversized-{index}"));
+        if let PluginHostRequest::Send { request, .. } = &mut oversized {
+            request.body = vec![255; body.len() + 1];
+        }
+        let error = owner.call(oversized).await.unwrap_err();
+        assert_eq!(error.code, PluginHostErrorCode::QuotaExceeded);
+        assert_eq!(peer.connection_count(), index + 1);
+    }
+    assert_eq!(f.count("plugin_effect_receipts"), 2);
+    let mut enormous = send(id, PluginHttpMethod::Post, "wire-overflow");
+    if let PluginHostRequest::Send { request, .. } = &mut enormous {
+        request.body = vec![255; 2 * body.len()];
+    }
+    let path = format!("/api/w/{TEST_WORKSPACE_ID}/plugin-host");
+    let response = build_router(f.api.clone())
+        .oneshot(runtime_source_request(
+            &f.identity,
+            Some(&f.source.worker_id),
+            "POST",
+            &path,
+            serde_json::to_vec(&enormous).unwrap(),
+        ))
+        .await
+        .unwrap();
+    assert_eq!(response.status(), StatusCode::PAYLOAD_TOO_LARGE);
+    assert_eq!(peer.connection_count(), 2);
+    drop(server);
+    f.shutdown().await;
+}
+
+#[tokio::test]
+async fn other_workspace_control_plane_hosts_ips_and_dns_aliases_never_connect() {
+    for dynamic in [false, true] {
+        for host in [HOST, "8.8.8.8", "runtime-alias.example.com"] {
+            let peer = Peer::new(HOST, vec![]).await;
+            let mut binding = peer_binding(&peer);
+            binding.origin = format!("https://{host}:{}", peer.public_address().port());
+            let mut f = Fixture::new(Some(binding)).await;
+            let endpoint = format!("https://{HOST}:{}", peer.public_address().port());
+            if dynamic {
+                let mut workspace = test_workspace();
+                workspace.workspace_id = "workspace-b".into();
+                f.api.store.upsert_workspace(&workspace).await.unwrap();
+                f.api
+                    .store
+                    .upsert_workspace_runtime_binding_record(
+                        WorkspaceRuntimeBinding {
+                            workspace_id: "workspace-b".into(),
+                            runtime_id: "only-b".into(),
+                            display_name: "Only B".into(),
+                            base_url: endpoint,
+                            public_key: f.identity.public_key.clone(),
+                            public_key_fingerprint: String::new(),
+                            binding_id: String::new(),
+                            authentication_mode:
+                                StoredRuntimeAuthenticationMode::LegacyServerIssuer,
+                            created_at: "1".into(),
+                            updated_at: "1".into(),
+                            revoked_at: None,
+                        },
+                        true,
+                    )
+                    .await
+                    .unwrap();
+            } else {
+                let mut operational = f.api.config.clone();
+                operational.remote_runtime_sources.push(
+                    RemoteRuntimeConfig::new("only-b", "Only B", endpoint, None)
+                        .with_workspace_id("workspace-b"),
+                );
+                f.api.config = operational
+                    .for_catalog_workspace(&test_workspace(), vec![])
+                    .unwrap();
+                assert!(
+                    !f.api
+                        .config
+                        .remote_runtime_sources
+                        .iter()
+                        .any(|source| source.runtime_id == "only-b")
+                );
+            }
+            // Only external DNS/socket/trust are substituted. The operational
+            // inventory and concrete shared owner/adapters remain production code.
+            let policy = plugin_host::transport_policy(&f.api, |name, port| async move {
+                let ip = if name == HOST { "8.8.8.8" } else { "9.9.9.9" };
+                Ok(vec![format!("{ip}:{port}").parse().unwrap()])
+            })
+            .await
+            .unwrap();
+            assert!(policy.denied_hosts.contains(HOST));
+            assert!(
+                policy
+                    .denied_addresses
+                    .contains(&"8.8.8.8".parse().unwrap())
+            );
+            f.api.plugin_https_transport =
+                Some(Arc::new(peer.transport_with_policy(policy.clone())));
+            let embedded = f.owner(Some(peer.transport_with_policy(policy))).await;
+            let (remote, server) =
+                remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await;
+            let invocation = f.host(embedded.clone(), "forbidden-control-plane").await;
+            for owner in [&embedded, &remote] {
+                assert_denied(
+                    owner
+                        .call(send(
+                            &invocation.context().invocation_id,
+                            PluginHttpMethod::Get,
+                            "blocked",
+                        ))
+                        .await,
+                );
+            }
+            assert_eq!(peer.connection_count(), 0, "{dynamic}/{host}");
+            drop(server);
+            f.shutdown().await;
+        }
+    }
+}
+
+#[tokio::test]
+async fn stalled_authority_monitor_closes_active_transport_and_consumes_effect() {
+    use crate::plugin_https::test_support::Reply;
+    for remote in [false, true] {
+        let mut peer = Peer::new(HOST, vec![Reply::Stall]).await;
+        let mut f = Fixture::new(Some(peer_binding(&peer))).await;
+        f.api.plugin_https_transport = Some(Arc::new(peer.transport_for_requests(4)));
+        let (remote_owner, server) =
+            remote_owner_for(&f, &f.source.worker_id, TEST_WORKSPACE_ID).await;
+        let owner = if remote {
+            remote_owner
+        } else {
+            f.owner(Some(peer.transport_for_requests(4))).await
+        };
+        let host = f.host(owner.clone(), "monitor-unavailable").await;
+        let request = send(
+            &host.context().invocation_id,
+            PluginHttpMethod::Post,
+            "once",
+        );
+        let attempt = owner.call(request.clone());
+        tokio::pin!(attempt);
+        tokio::select! {
+            result = &mut attempt => panic!("request must reach peer: {result:?}"),
+            _ = peer.observed() => {},
+        }
+        // Hold the actual authority DB connection. The monitor cannot complete
+        // its read, and may not treat a past permission as continued authority.
+        let store = f.api.config_store.clone();
+        let (started, waiting) = tokio::sync::oneshot::channel();
+        let (release, gate) = tokio::sync::oneshot::channel();
+        let held = tokio::task::spawn_blocking(move || {
+            store.with_conn(|_| {
+                let _ = started.send(());
+                let _ = gate.blocking_recv();
+                Ok(())
+            })
+        });
+        waiting.await.unwrap();
+        tokio::select! {
+            result = &mut attempt => panic!("DB is held but owner finished: {result:?}"),
+            _ = peer.closed() => {},
+        }
+        release.send(()).unwrap();
+        held.await.unwrap().unwrap();
+        let error = tokio::time::timeout(std::time::Duration::from_secs(3), &mut attempt)
+            .await
+            .unwrap()
+            .unwrap_err();
+        assert_eq!(
+            error.effect,
+            server_api::PluginEffectObservation::Indeterminate
+        );
+        assert_eq!(error.code, PluginHostErrorCode::Unavailable);
+        assert!(owner.call(request).await.is_err());
+        assert_eq!(f.count("plugin_effect_receipts"), 1);
+        assert_eq!(peer.connection_count(), 1);
+        drop(server);
+        f.shutdown().await;
+    }
+}

crates/workspace-server/src/worker_source/runtime_request.rs

@@ -454,6 +454,7 @@ mod tests {
             runtime_event_sources: vec![],
             remote_runtime_sources: vec![],
             backend_base_url: Some("server-auth".into()),
+            control_plane_endpoints: Default::default(),
         };
         let identity = RuntimeIdentityMaterial::generate("runtime-auth").unwrap();
         store