feat(gateway): expose workspace service route snapshots and watches

crates/openshell-cli/tests/sandbox_exec_streaming_integration.rs

@@ -163,6 +163,7 @@ macro_rules! mock_gateway {
     ) => {
         #[tonic::async_trait]
         impl OpenShell for MockGateway {
+
             $(async fn $method(&self, _: Request<$request>) -> Result<Response<$response>, Status> {
                 Err(Status::unimplemented("unused test RPC"))
             })*

crates/openshell-sdk/src/client.rs

@@ -12,7 +12,7 @@ use crate::auth::{BearerSlot, EdgeAuthInterceptor, bearer_metadata};
 use crate::config::{AuthConfig, ClientConfig};
 use crate::error::{Result, SdkError};
 use crate::pagination::{Page, Pager};
-use crate::raw::{AuthedGrpcClient, AuthedSandboxUpdatesClient};
+use crate::raw::{AuthedGrpcClient, AuthedSandboxUpdatesClient, AuthedServiceDiscoveryClient};
 use crate::refresh::{RefreshedToken, TokenSource};
 use crate::routing::OwnerCache;
 use crate::types::{
@@ -149,6 +149,23 @@ impl OpenShellClient {
         Ok(self.raw_sandbox_updates())
     }
 
+    /// Authenticated client for the separate ingress discovery service.
+    /// Shares the channel and live auth slot; like [`Self::raw_grpc`], does not
+    /// refresh or retry credentials by itself.
+    pub fn raw_service_discovery(&self) -> AuthedServiceDiscoveryClient {
+        proto::service_discovery_client::ServiceDiscoveryClient::with_interceptor(
+            self.channel.clone(),
+            self.interceptor.clone(),
+        )
+    }
+
+    /// Like [`Self::raw_service_discovery`], refreshing bearer credentials first.
+    /// Reactive retry remains the raw caller's responsibility.
+    pub async fn raw_service_discovery_fresh(&self) -> Result<AuthedServiceDiscoveryClient> {
+        self.ensure_fresh().await?;
+        Ok(self.raw_service_discovery())
+    }
+
     /// Force an OIDC refresh and write the new token into the live bearer
     /// slot, regardless of expiry. Returns `true` when a refresher is wired
     /// and a fresh token was minted, `false` for static auth. Use after a

crates/openshell-sdk/src/raw.rs

@@ -21,6 +21,7 @@
 pub use openshell_core::proto;
 pub use openshell_core::proto::open_shell_client::OpenShellClient as GrpcClient;
 pub use openshell_core::proto::sandbox_updates_client::SandboxUpdatesClient;
+pub use openshell_core::proto::service_discovery_client::ServiceDiscoveryClient;
 pub use openshell_core::proto::{
     CreateSandboxRequest, CreateSandboxTemplateRequest, CreateWorkspaceRequest,
     DeleteSandboxRequest, DeleteSandboxTemplateRequest, DeleteWorkspaceRequest, ExecSandboxRequest,
@@ -53,3 +54,11 @@ pub type AuthedSandboxUpdatesClient = SandboxUpdatesClient<
         crate::EdgeAuthInterceptor,
     >,
 >;
+
+/// Application ingress discovery using the SDK transport and live auth slot.
+pub type AuthedServiceDiscoveryClient = ServiceDiscoveryClient<
+    tonic::service::interceptor::InterceptedService<
+        tonic::transport::Channel,
+        crate::EdgeAuthInterceptor,
+    >,
+>;

crates/openshell-sdk/tests/client_mock.rs

@@ -1195,6 +1195,35 @@ impl SandboxUpdates for TestOpenShell {
     }
 }
 
+#[tonic::async_trait]
+impl proto::service_discovery_server::ServiceDiscovery for TestOpenShell {
+    async fn get_service_route_snapshot(
+        &self,
+        request: tonic::Request<proto::GetServiceRouteSnapshotRequest>,
+    ) -> Result<Response<proto::GetServiceRouteSnapshotResponse>, Status> {
+        if let Some(expected) = &self.state.require_bearer
+            && request
+                .metadata()
+                .get("authorization")
+                .and_then(|v| v.to_str().ok())
+                != Some(expected.as_str())
+        {
+            return Err(Status::unauthenticated("invalid or missing bearer token"));
+        }
+        Ok(Response::new(proto::GetServiceRouteSnapshotResponse {
+            snapshot: Some(proto::ServiceRouteSnapshot::default()),
+        }))
+    }
+    type WatchServiceRoutesStream =
+        tokio_stream::wrappers::ReceiverStream<Result<proto::WatchServiceRoutesResponse, Status>>;
+    async fn watch_service_routes(
+        &self,
+        _: tonic::Request<proto::WatchServiceRoutesRequest>,
+    ) -> Result<Response<Self::WatchServiceRoutesStream>, Status> {
+        Err(Status::unimplemented("unused"))
+    }
+}
+
 /// Spin up the mock gateway, return its endpoint URL.
 async fn start_mock(state: Arc<MockState>) -> String {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
@@ -1204,11 +1233,15 @@ async fn start_mock(state: Arc<MockState>) -> String {
     let updates = SandboxUpdatesServer::new(TestOpenShell {
         state: state.clone(),
     });
+    let discovery = proto::service_discovery_server::ServiceDiscoveryServer::new(TestOpenShell {
+        state: state.clone(),
+    });
     let svc = OpenShellServer::new(TestOpenShell { state });
     tokio::spawn(async move {
         let _ = tonic::transport::Server::builder()
             .add_service(svc)
             .add_service(updates)
+            .add_service(discovery)
             .serve_with_incoming(stream)
             .await;
     });
@@ -2457,6 +2490,58 @@ async fn raw_sandbox_updates_shares_transport_and_live_auth_slot() {
     assert_eq!(*state.last_update.lock().await, Some(request));
 }
 
+#[tokio::test]
+async fn raw_discovery_shares_transport_and_refreshes_the_live_auth_slot() {
+    let state = Arc::new(MockState {
+        require_bearer: Some("Bearer fresh-token".into()),
+        ..Default::default()
+    });
+    let endpoint = start_mock(state).await;
+    let calls = Arc::new(AtomicU32::new(0));
+    let mut config = ClientConfig::new(&endpoint);
+    config.auth = Some(AuthConfig::Oidc {
+        token: "stale-token".into(),
+        expires_at: Some(
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap()
+                .as_secs()
+                + 5,
+        ),
+        refresh: Some(Arc::new(OneShotRefresher {
+            calls: calls.clone(),
+        })),
+    });
+    let client = OpenShellClient::connect(config).await.unwrap();
+    let mut discovery = client.raw_service_discovery();
+    let request = proto::GetServiceRouteSnapshotRequest {
+        workspace_scope: Some(proto::workspace_selector("team")),
+    };
+    assert_eq!(
+        discovery
+            .get_service_route_snapshot(request.clone())
+            .await
+            .unwrap_err()
+            .code(),
+        tonic::Code::Unauthenticated
+    );
+    assert_eq!(calls.load(Ordering::SeqCst), 0);
+    client
+        .raw_service_discovery_fresh()
+        .await
+        .unwrap()
+        .get_service_route_snapshot(request.clone())
+        .await
+        .unwrap();
+    assert_eq!(calls.load(Ordering::SeqCst), 1);
+    discovery.get_service_route_snapshot(request).await.unwrap();
+    client
+        .raw_grpc()
+        .health(proto::HealthRequest {})
+        .await
+        .unwrap();
+}
+
 #[tokio::test]
 async fn watch_logs_forwards_logs_and_warnings() {
     let state = Arc::new(MockState {

crates/openshell-server/README.md

@@ -295,3 +295,38 @@ timeouts; bounded record attempts prevent a malformed or stalled record from
 starving later cleanup. Failed attempts are revisited on the next scan.
 Discovery consumers must not treat these
 records as application health or external ingress readiness.
+
+## Workspace service discovery
+
+Both discovery RPCs require `supervisor_app_listener.enabled = true`. When the
+listener is disabled, they return `FAILED_PRECONDITION` without collecting routes
+or starting a watch.
+
+`openshell.v1.ServiceDiscovery.GetServiceRouteSnapshot` and `WatchServiceRoutes` expose accepted installations
+for one explicit active workspace. They require `sandbox:read` and workspace
+user access; omitted/all-workspaces selectors and sandbox principals are rejected.
+Dedicated discovery machine grants are separate work.
+
+Each successful response replaces the complete view. Consumers preserve the
+returned hostname as TLS SNI and HTTP authority when dialing the supervisor
+endpoint, remove absent routes, and expire cached entries at `expires_at` even
+when discovery is unavailable. An error is not an empty replacement. The opaque
+snapshot revision includes lease renewals; `installation_version` remains stable
+across those renewals and changes when installation identity changes. These
+records establish neither application health nor ingress readiness.
+
+Collection validates each installation once; lease renewals during a scan do not
+invalidate the view. This is not a cross-resource transaction. Reads are bounded
+to 1,000 records, 4,096 routes, 4 MiB, and ten seconds. A per-workspace cache shares
+one collection for a one-second refresh interval, retains at most 16 workspaces,
+and never returns expired routes. At most four collections and 16 unary snapshot
+requests run concurrently per gateway. Unary admission permits four requests per
+principal and eight per workspace. Overflow returns `RESOURCE_EXHAUSTED`; storage
+failure or timeout returns `UNAVAILABLE`; malformed records return `INTERNAL`.
+Watches recheck bearer authentication and workspace authorization each second,
+use a one-message buffer, and end after one minute or a five-second blocked send.
+Each gateway allows 32 watches, at most four per principal and 16 per workspace.
+Cache reuse adds at most one refresh interval to collection freshness; consumers
+must independently honor the advertised lease expiry.
+Reconnect and relist on stream termination; revisions are opaque
+equality tokens, not resumable cursors or an ordering across gateway replicas.

crates/openshell-server/src/grpc/mod.rs

@@ -14,6 +14,7 @@ pub mod supervisor_host;
 pub use sandbox::mint_persisted_authentication;
 mod service;
 pub mod service_advertisement;
+pub mod service_discovery;
 mod validation;
 pub mod workspace;
 
@@ -1435,3 +1436,22 @@ mod tests {
         assert_eq!(public.protocol_minor, 2);
     }
 }
+
+#[tonic::async_trait]
+impl openshell_core::proto::service_discovery_server::ServiceDiscovery for OpenShellService {
+    async fn get_service_route_snapshot(
+        &self,
+        request: Request<openshell_core::proto::GetServiceRouteSnapshotRequest>,
+    ) -> Result<Response<openshell_core::proto::GetServiceRouteSnapshotResponse>, Status> {
+        service_discovery::get_snapshot(&self.state, request).await
+    }
+
+    type WatchServiceRoutesStream = service_discovery::RouteWatchStream;
+
+    async fn watch_service_routes(
+        &self,
+        request: Request<openshell_core::proto::WatchServiceRoutesRequest>,
+    ) -> Result<Response<Self::WatchServiceRoutesStream>, Status> {
+        service_discovery::watch(&self.state, request).await
+    }
+}