diff --git a/Cargo.lock b/Cargo.lock index bbc959f..93bf23d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -843,6 +843,7 @@ dependencies = [ "rmcp", "serde", "serde_json", + "tokio", ] [[package]] diff --git a/crates/adapters/gfa-mcp/Cargo.toml b/crates/adapters/gfa-mcp/Cargo.toml index f6fe5dc..d7a4400 100644 --- a/crates/adapters/gfa-mcp/Cargo.toml +++ b/crates/adapters/gfa-mcp/Cargo.toml @@ -11,6 +11,7 @@ gfa-service.workspace = true rmcp.workspace = true serde.workspace = true serde_json.workspace = true +tokio.workspace = true [lints] workspace = true diff --git a/crates/adapters/gfa-mcp/README.md b/crates/adapters/gfa-mcp/README.md index 9636c35..1f92c73 100644 --- a/crates/adapters/gfa-mcp/README.md +++ b/crates/adapters/gfa-mcp/README.md @@ -43,3 +43,6 @@ Compact match briefings use the same text-board and canonical-move state as `get `gfa serve` also serves streamable MCP at `/mcp` on the same loopback listener. `--mcp-seat 1` selects a player and `--mcp-spectator` selects read-only access; seat 0 is the default. This remains single-user local mode and shares REST's strict peer, Host, Origin, and response-header policy. SDK request bodies are limited to 64 KiB. Transport responses are stateless JSON when possible, with SDK protocol negotiation for legacy and current clients. No legacy session table is retained. Shutdown cancels SDK streams before HTTP drains. rmcp's streamable HTTP feature supplies protocol handling and SSE framing rather than a separate RPC implementation. + + +Live state watches support current `subscriptions/listen` and legacy stdio `resources/subscribe`/`unsubscribe`. A shared adapter permits at most 16 concurrent watches/streams, with at most 8 URIs in a modern stream and 64 KiB per projected snapshot. Only match state URIs are accepted. Polling every two seconds catches changes committed outside MCP, including same-turn controls. Modern streams emit an initial refresh notification; unchanged snapshots stay quiet. Slow writes, disconnects, explicit cancellation, and dropped handlers release capacity. Legacy watches are keyed by connection identity so identical client names cannot unsubscribe one another. The stateless HTTP host supports modern listen streams; older HTTP clients can poll state tools. Tokio supplies bounded tasks, synchronization, cancellation points, and timers for this transport concern. diff --git a/crates/adapters/gfa-mcp/src/lib.rs b/crates/adapters/gfa-mcp/src/lib.rs index 071f276..7d143bb 100644 --- a/crates/adapters/gfa-mcp/src/lib.rs +++ b/crates/adapters/gfa-mcp/src/lib.rs @@ -1,6 +1,7 @@ //! Official-SDK MCP adapter over the transport-independent lifecycle service. mod briefing; mod history; +mod live; mod planning; mod play; mod play_types; @@ -33,6 +34,7 @@ pub struct McpServer { service: Arc, viewer: Viewer, tools: Arc>, + watches: Arc, } impl McpServer { /// Build the game-discovery tools. Hosts must authorize the supplied viewer first. @@ -44,7 +46,7 @@ impl McpServer { "Select an authorized player or spectator view.", )); } - Ok(Self { service, viewer, tools: Arc::new(vec![ + Ok(Self { service, viewer, watches: Arc::new(live::Watches::default()), tools: Arc::new(vec![ tool::("list_games", "Discover installed games before choosing one. Example: list_games({}).")?, tool::("get_game_info", "Read rules and the API loop before playing; use match_id for an existing match. Example: get_game_info({\"game_id\":\"chess\",\"detail\":\"compact\"}).")?, tool::("list_opponents", "Find installed opponents and their measured levels before creating a match. Example: list_opponents({\"game_id\":\"chess\"}).")?, @@ -210,6 +212,7 @@ impl ServerHandler for McpServer { ServerCapabilities::builder() .enable_tools() .enable_resources() + .enable_resources_subscribe() .enable_prompts() .build(), ) @@ -258,6 +261,29 @@ impl ServerHandler for McpServer { ) .map(Into::into) } + fn accepted_subscription_filter( + &self, + requested: &rmcp::model::SubscriptionFilter, + ) -> Option { + Some(self.accepted_watches(requested)) + } + async fn listen(&self, context: rmcp::service::SubscriptionContext) -> Result<(), ErrorData> { + self.listen_states(context).await + } + async fn subscribe( + &self, + request: rmcp::model::SubscribeRequestParams, + context: RequestContext, + ) -> Result<(), ErrorData> { + self.subscribe_state(request.uri, context).await + } + async fn unsubscribe( + &self, + request: rmcp::model::UnsubscribeRequestParams, + context: RequestContext, + ) -> Result<(), ErrorData> { + self.unsubscribe_state(&request.uri, context).await + } fn get_tool(&self, name: &str) -> Option { self.tools.iter().find(|tool| tool.name == name).cloned() } diff --git a/crates/adapters/gfa-mcp/src/live.rs b/crates/adapters/gfa-mcp/src/live.rs new file mode 100644 index 0000000..49037d8 --- /dev/null +++ b/crates/adapters/gfa-mcp/src/live.rs @@ -0,0 +1,207 @@ +use crate::{play_types::StateOutput, McpServer}; +use gfa_core::Viewer; +use gfa_service::GameService; +use rmcp::{ + model::{ClientConfig, ResourceUpdatedNotificationParam, SubscriptionFilter}, + service::{Peer, RequestContext, SubscriptionContext}, + ErrorData, RoleServer, +}; +use std::{sync::Arc, time::Duration}; +use tokio::{ + sync::{Mutex, Semaphore}, + task::JoinHandle, +}; + +const MAX_URIS: usize = 8; +const POLL: Duration = Duration::from_secs(2); +const IO_TIMEOUT: Duration = Duration::from_secs(3); +struct LegacyWatch { + // The SDK returns clones of this connection's stable Arc. Pointer identity + // distinguishes clients with identical names without trusting client IDs. + identity: Arc, + uri: String, + task: JoinHandle<()>, +} +impl Drop for LegacyWatch { + fn drop(&mut self) { + self.task.abort(); + } +} +pub(crate) struct Watches { + capacity: Arc, + legacy: Mutex>, +} +impl Default for Watches { + fn default() -> Self { + Self { + capacity: Arc::new(Semaphore::new(16)), + legacy: Mutex::new(Vec::new()), + } + } +} +fn state_id(uri: &str) -> Option<&str> { + let id = uri.strip_prefix("gfa://matches/")?.strip_suffix("/state")?; + (!id.is_empty() + && id.len() <= 128 + && id + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || b"_-".contains(&byte))) + .then_some(id) +} +fn unavailable() -> ErrorData { + ErrorData::internal_error("Resource subscription unavailable", None) +} +async fn snapshot(service: &GameService, viewer: Viewer, uri: &str) -> Result { + let id = state_id(uri).ok_or_else(|| { + ErrorData::invalid_params("Only gfa://matches/{match_id}/state can be watched", None) + })?; + let state = tokio::time::timeout(IO_TIMEOUT, service.get_state(id, viewer)) + .await + .map_err(|_| unavailable())? + .map_err(|error| ErrorData::invalid_params(error.message, None))?; + let state = StateOutput::from_state(state, viewer).map_err(|_| unavailable())?; + let encoded = serde_json::to_string(&state).map_err(|_| unavailable())?; + if encoded.len() > 64 * 1024 { + return Err(ErrorData::invalid_params( + "State exceeds the live watch size limit", + None, + )); + } + Ok(encoded) +} +impl McpServer { + pub(crate) fn accepted_watches(&self, requested: &SubscriptionFilter) -> SubscriptionFilter { + let mut uris = requested + .resource_subscriptions + .as_ref() + .into_iter() + .flatten() + .filter(|uri| state_id(uri).is_some()) + .take(MAX_URIS) + .cloned() + .collect::>(); + uris.sort(); + uris.dedup(); + SubscriptionFilter::builder() + .resource_subscriptions(uris) + .build() + } + pub(crate) async fn listen_states( + &self, + context: SubscriptionContext, + ) -> Result<(), ErrorData> { + let _permit = self + .watches + .capacity + .clone() + .try_acquire_owned() + .map_err(|_| unavailable())?; + let uris = context + .accepted() + .resource_subscriptions + .clone() + .unwrap_or_default(); + if uris.is_empty() { + return Ok(()); + } + let mut previous = vec![None; uris.len()]; + let mut tick = tokio::time::interval(POLL); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + tokio::select! { + _=context.cancelled()=>return Ok(()), + _=tick.tick()=>{} + } + for (index, uri) in uris.iter().enumerate() { + let current = tokio::select! { + _=context.cancelled()=>return Ok(()), + state=snapshot(&self.service,self.viewer,uri)=>state?, + }; + if previous[index].as_ref() != Some(¤t) { + tokio::select! { + _=context.cancelled()=>return Ok(()), + sent=tokio::time::timeout(IO_TIMEOUT,context.sink().notify_resource_updated(uri))=> { + sent.map_err(|_|unavailable())?.map_err(|_|unavailable())?; + } + } + previous[index] = Some(current); + } + } + } + } + pub(crate) async fn subscribe_state( + &self, + uri: String, + context: RequestContext, + ) -> Result<(), ErrorData> { + let identity = context.peer.peer_info().ok_or_else(unavailable)?; + let current = snapshot(&self.service, self.viewer, &uri).await?; + let mut watches = self.watches.legacy.lock().await; + watches.retain(|watch| !watch.task.is_finished()); + if watches + .iter() + .any(|watch| Arc::ptr_eq(&watch.identity, &identity) && watch.uri == uri) + { + return Ok(()); + } + let permit = self + .watches + .capacity + .clone() + .try_acquire_owned() + .map_err(|_| unavailable())?; + let service = self.service.clone(); + let viewer = self.viewer; + let watched_uri = uri.clone(); + let task = tokio::spawn(async move { + let _permit = permit; + legacy_loop(service, viewer, watched_uri, context.peer, current).await; + }); + watches.push(LegacyWatch { + identity, + uri, + task, + }); + Ok(()) + } + pub(crate) async fn unsubscribe_state( + &self, + uri: &str, + context: RequestContext, + ) -> Result<(), ErrorData> { + let identity = context.peer.peer_info().ok_or_else(unavailable)?; + self.watches.legacy.lock().await.retain(|watch| { + !watch.task.is_finished() + && !(watch.uri == uri && Arc::ptr_eq(&watch.identity, &identity)) + }); + Ok(()) + } +} +async fn legacy_loop( + service: Arc, + viewer: Viewer, + uri: String, + peer: Peer, + mut previous: String, +) { + loop { + tokio::time::sleep(POLL).await; + if peer.is_transport_closed() { + return; + } + let Ok(current) = snapshot(&service, viewer, &uri).await else { + return; + }; + if current != previous { + let sent = tokio::time::timeout( + IO_TIMEOUT, + peer.notify_resource_updated(ResourceUpdatedNotificationParam::new(&uri)), + ) + .await; + if !matches!(sent, Ok(Ok(()))) { + return; + } + previous = current; + } + } +} diff --git a/crates/apps/gfa-server/src/lib.rs b/crates/apps/gfa-server/src/lib.rs index fd22bef..861aa36 100644 --- a/crates/apps/gfa-server/src/lib.rs +++ b/crates/apps/gfa-server/src/lib.rs @@ -283,3 +283,6 @@ mod mcp_resource_tests; #[cfg(test)] mod mcp_plan_tests; + +#[cfg(test)] +mod mcp_live_tests; diff --git a/crates/apps/gfa-server/src/mcp_live_tests.rs b/crates/apps/gfa-server/src/mcp_live_tests.rs new file mode 100644 index 0000000..3838f7d --- /dev/null +++ b/crates/apps/gfa-server/src/mcp_live_tests.rs @@ -0,0 +1,277 @@ +use super::{ + tests::{config, fixture}, + *, +}; +use gfa_core::Viewer; +use gfa_mcp::McpServer; +use rmcp::{ + model::{ServerNotification, SubscriptionFilter}, + ServiceExt, +}; +use serde_json::json; +use std::time::Duration; + +#[tokio::test] +async fn mcp_live_resources_notify_on_external_commits_and_cancel() -> Result<(), ServerError> { + let directory = tempfile::tempdir()?; + let app = fixture(&config(&directory)).await?; + let initial = app + .service + .create_match( + serde_json::from_value(json!({"game_id":"tictactoe"}))?, + Viewer::Player(0), + ) + .await?; + let uri = format!("gfa://matches/{}/state", initial.match_id); + let adapter = McpServer::new(app.service.clone(), Viewer::Spectator)?; + let (client_io, server_io) = tokio::io::duplex(64 * 1024); + let task = tokio::spawn(async move { adapter.serve(server_io).await }); + let client = rmcp::service::serve_client_with_lifecycle( + (), + client_io, + rmcp::service::ClientLifecycleMode::Discover { + preferred_versions: vec![rmcp::model::ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + let server = task.await??; + let filter = SubscriptionFilter::builder() + .resource_subscriptions([uri.clone(), "file:///secret".into()]) + .tools_list_changed() + .build(); + let mut subscription = client.listen(filter).await?; + assert_eq!( + subscription.acknowledged().resource_subscriptions, + Some(vec![uri.clone()]) + ); + assert_ne!(subscription.acknowledged().tools_list_changed, Some(true)); + let notification = tokio::time::timeout(Duration::from_secs(5), subscription.next()) + .await?? + .ok_or("initial notification")?; + assert!( + matches!(notification,ServerNotification::ResourceUpdatedNotification(update) if update.params.uri==uri) + ); + assert!( + tokio::time::timeout(Duration::from_millis(250), subscription.next()) + .await + .is_err() + ); + // Change the service directly, outside this MCP connection. + app.service + .make_move( + &initial.match_id, + serde_json::from_value( + json!({"seat":0,"turn":0,"action":"r1c1","reasoning":"private live reasoning"}), + )?, + None, + ) + .await?; + let notification = tokio::time::timeout(Duration::from_secs(5), subscription.next()) + .await?? + .ok_or("move notification")?; + assert!(!serde_json::to_string(¬ification)?.contains("private live reasoning")); + assert!( + matches!(notification,ServerNotification::ResourceUpdatedNotification(update) if update.params.uri==uri) + ); + // Draw offers change state without increasing the action turn. + app.service + .offer_draw( + &initial.match_id, + gfa_api_types::ControlRequest { seat: 0, turn: 1 }, + ) + .await?; + let notification = tokio::time::timeout(Duration::from_secs(5), subscription.next()) + .await?? + .ok_or("control notification")?; + assert!( + matches!(notification,ServerNotification::ResourceUpdatedNotification(update) if update.params.uri==uri) + ); + subscription.cancel().await?; + assert!(subscription.next().await?.is_none()); + client.cancel().await?; + server.waiting().await?; + app.store.close().await; + Ok(()) +} + +#[tokio::test] +async fn mcp_live_resources_bound_uri_filters() -> Result<(), ServerError> { + let directory = tempfile::tempdir()?; + let app = fixture(&config(&directory)).await?; + let adapter = McpServer::new(app.service.clone(), Viewer::Player(0))?; + use rmcp::ServerHandler; + let uris = (0..30) + .map(|index| format!("gfa://matches/m_{index}/state")) + .collect::>(); + let accepted = adapter + .accepted_subscription_filter( + &SubscriptionFilter::builder() + .resource_subscriptions(uris) + .build(), + ) + .ok_or("filter")?; + assert_eq!( + accepted.resource_subscriptions.as_ref().map(Vec::len), + Some(8) + ); + for uri in [ + "gfa://matches/../state", + "gfa://matches/x/state?seat=1", + "gfa://games/chess/info", + "gfa://matches//state", + ] { + let accepted = adapter + .accepted_subscription_filter( + &SubscriptionFilter::builder() + .resource_subscriptions([uri]) + .build(), + ) + .ok_or("filter")?; + assert!(accepted + .resource_subscriptions + .unwrap_or_default() + .is_empty()); + } + app.store.close().await; + Ok(()) +} + +struct LegacyClient(tokio::sync::mpsc::Sender); +impl rmcp::ClientHandler for LegacyClient { + fn get_info(&self) -> rmcp::model::ClientConfig { + rmcp::model::ClientConfig::default() + .with_protocol_version(rmcp::model::ProtocolVersion::V_2025_11_25) + } + async fn on_resource_updated( + &self, + params: rmcp::model::ResourceUpdatedNotificationParam, + _: rmcp::service::NotificationContext, + ) { + let _ = self.0.try_send(params.uri); + } +} +#[tokio::test] +#[allow(deprecated)] +async fn legacy_watch_unsubscribe_is_scoped_to_the_connection() -> Result<(), ServerError> { + let directory = tempfile::tempdir()?; + let app = fixture(&config(&directory)).await?; + let initial = app + .service + .create_match( + serde_json::from_value(json!({"game_id":"tictactoe"}))?, + Viewer::Player(0), + ) + .await?; + let uri = format!("gfa://matches/{}/state", initial.match_id); + let shared = McpServer::new(app.service.clone(), Viewer::Spectator)?; + let mut clients = Vec::new(); + let mut servers = Vec::new(); + let mut receivers = Vec::new(); + for _ in 0..2 { + let (send, receive) = tokio::sync::mpsc::channel(4); + let (client_io, server_io) = tokio::io::duplex(64 * 1024); + let handler = shared.clone(); + let task = tokio::spawn(async move { handler.serve(server_io).await }); + let client = LegacyClient(send).serve(client_io).await?; + client + .subscribe(rmcp::model::SubscribeRequestParams::new(&uri)) + .await?; + // Repeated subscriptions do not consume more capacity. + client + .subscribe(rmcp::model::SubscribeRequestParams::new(&uri)) + .await?; + clients.push(client); + servers.push(task.await??); + receivers.push(receive); + } + clients[0] + .unsubscribe(rmcp::model::UnsubscribeRequestParams::new(&uri)) + .await?; + app.service + .make_move( + &initial.match_id, + serde_json::from_value(json!({"seat":0,"turn":0,"action":"r1c1"}))?, + None, + ) + .await?; + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), receivers[1].recv()) + .await? + .as_deref(), + Some(uri.as_str()) + ); + assert!( + tokio::time::timeout(Duration::from_millis(250), receivers[0].recv()) + .await + .is_err() + ); + for client in clients { + client.cancel().await?; + } + for server in servers { + server.waiting().await?; + } + drop(shared); + app.store.close().await; + Ok(()) +} + +#[tokio::test] +async fn modern_watch_capacity_is_released_after_cancellation() -> Result<(), ServerError> { + let directory = tempfile::tempdir()?; + let app = fixture(&config(&directory)).await?; + let initial = app + .service + .create_match( + serde_json::from_value(json!({"game_id":"tictactoe"}))?, + Viewer::Player(0), + ) + .await?; + let uri = format!("gfa://matches/{}/state", initial.match_id); + let adapter = McpServer::new(app.service.clone(), Viewer::Spectator)?; + let (client_io, server_io) = tokio::io::duplex(64 * 1024); + let task = tokio::spawn(async move { adapter.serve(server_io).await }); + let client = rmcp::service::serve_client_with_lifecycle( + (), + client_io, + rmcp::service::ClientLifecycleMode::Discover { + preferred_versions: vec![rmcp::model::ProtocolVersion::V_2026_07_28], + }, + ) + .await?; + let server = task.await??; + let filter = SubscriptionFilter::builder() + .resource_subscriptions([uri]) + .build(); + let mut subscriptions = Vec::new(); + for _ in 0..16 { + let mut subscription = client.listen(filter.clone()).await?; + assert!(subscription.next().await?.is_some()); + subscriptions.push(subscription); + } + // The SDK acknowledges the accepted filter before invoking our handler. + // The exhausted handler then closes the stream with a protocol error. + let mut excess = client.listen(filter.clone()).await?; + assert!(tokio::time::timeout(Duration::from_secs(2), excess.next()) + .await? + .is_err()); + subscriptions[0].cancel().await?; + let mut replacement = tokio::time::timeout(Duration::from_secs(2), async { + loop { + let mut candidate = client.listen(filter.clone()).await?; + if matches!(candidate.next().await, Ok(Some(_))) { + return Ok::<_, rmcp::service::ServiceError>(candidate); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await??; + replacement.cancel().await?; + for subscription in &mut subscriptions { + subscription.cancel().await?; + } + client.cancel().await?; + server.waiting().await?; + app.store.close().await; + Ok(()) +}