From d6875994b86a9e113851b4e28f3f9c55eac4831e Mon Sep 17 00:00:00 2001 From: Mason Date: Wed, 30 Sep 2026 17:56:45 -0700 Subject: [PATCH] feat(mcp): add bounded live resource subscriptions --- Cargo.lock | 1 + crates/adapters/gfa-mcp/Cargo.toml | 1 + crates/adapters/gfa-mcp/README.md | 3 + crates/adapters/gfa-mcp/src/lib.rs | 17 ++- crates/adapters/gfa-mcp/src/live.rs | 109 +++++++++++++++++++ crates/apps/gfa-server/src/lib.rs | 3 + crates/apps/gfa-server/src/mcp_live_tests.rs | 100 +++++++++++++++++ 7 files changed, 233 insertions(+), 1 deletion(-) create mode 100644 crates/adapters/gfa-mcp/src/live.rs create mode 100644 crates/apps/gfa-server/src/mcp_live_tests.rs diff --git a/Cargo.lock b/Cargo.lock index a9a5321..56ea55c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -814,6 +814,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 1940bb5..43985f3 100644 --- a/crates/adapters/gfa-mcp/README.md +++ b/crates/adapters/gfa-mcp/README.md @@ -37,3 +37,6 @@ The resource catalog lists each installed game's compact rules. URI templates ex `simulate_moves` returns compact final states for independent lines and the first illegal action in each line. `analyze_position` returns ranked recommendations with labeled provider advice and expected game returns kept separate from engine scores. Both use the host-selected seat, honor the match's stored assist permissions, and support historical turns without changing live state. Simulation usage is durably accounted by the service. Optional analysis budgets go through the same validated worker limits as REST. + + +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 3c8d02e..6b47f49 100644 --- a/crates/adapters/gfa-mcp/src/lib.rs +++ b/crates/adapters/gfa-mcp/src/lib.rs @@ -5,6 +5,7 @@ mod play; mod play_types; mod resources; mod types; +mod live; use gfa_api_types::ApiError; use gfa_core::Viewer; use gfa_service::GameService; @@ -32,6 +33,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. @@ -43,7 +45,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\"}).")?, @@ -204,6 +206,7 @@ impl ServerHandler for McpServer { ServerCapabilities::builder() .enable_tools() .enable_resources() + .enable_resources_subscribe() .enable_prompts() .build(), ) @@ -252,6 +255,18 @@ 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..a164797 --- /dev/null +++ b/crates/adapters/gfa-mcp/src/live.rs @@ -0,0 +1,109 @@ +use crate::{play_types::StateOutput,McpServer}; +use gfa_service::GameService; +use gfa_core::Viewer; +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 56d0682..827b63f 100644 --- a/crates/apps/gfa-server/src/lib.rs +++ b/crates/apps/gfa-server/src/lib.rs @@ -269,3 +269,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..1b2cd51 --- /dev/null +++ b/crates/apps/gfa-server/src/mcp_live_tests.rs @@ -0,0 +1,100 @@ +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=().serve(client_io).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(()) +}