feat(mcp): notify clients when watched match state changes

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

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.

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<GameService>,
     viewer: Viewer,
     tools: Arc<Vec<Tool>>,
+    watches: Arc<live::Watches>,
 }
 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::<Empty, GamesOutput>("list_games", "Discover installed games before choosing one. Example: list_games({}).")?,
             tool::<InfoArgs, BriefingOutput>("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::<GameArgs, OpponentsOutput>("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<rmcp::model::SubscriptionFilter> {
+        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<RoleServer>,
+    ) -> Result<(), ErrorData> {
+        self.subscribe_state(request.uri, context).await
+    }
+    async fn unsubscribe(
+        &self,
+        request: rmcp::model::UnsubscribeRequestParams,
+        context: RequestContext<RoleServer>,
+    ) -> Result<(), ErrorData> {
+        self.unsubscribe_state(&request.uri, context).await
+    }
     fn get_tool(&self, name: &str) -> Option<Tool> {
         self.tools.iter().find(|tool| tool.name == name).cloned()
     }

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<ClientConfig>,
+    uri: String,
+    task: JoinHandle<()>,
+}
+impl Drop for LegacyWatch {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+pub(crate) struct Watches {
+    capacity: Arc<Semaphore>,
+    legacy: Mutex<Vec<LegacyWatch>>,
+}
+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<String, ErrorData> {
+    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::<Vec<_>>();
+        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(&current) {
+                    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<RoleServer>,
+    ) -> 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<RoleServer>,
+    ) -> 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<GameService>,
+    viewer: Viewer,
+    uri: String,
+    peer: Peer<RoleServer>,
+    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;
+        }
+    }
+}

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;