From 4f644a4c2c77f9f2306c2f9d8ff1e510cb6d4462 Mon Sep 17 00:00:00 2001 From: "glm-5.3-flash" Date: Sun, 30 Aug 2026 22:51:06 +0000 Subject: [PATCH] feat(mcp): cap MCP batch tool at MAX_BATCH_OPERATIONS (PRJ-22) - enforce the same 100-operation cap the HTTP /batch endpoint enforces; over-cap \x60calls\x60 reject with a structured INVALID_INPUT (retryable: false, matching CallError::invalid_input) before any dispatch - hoist MAX_BATCH_OPERATIONS to gateway/mod.rs and reuse it in routes, to_openapi (removing a pre-existing duplicate literal), and to_mcp - state the limit in the batch tool description and add maxItems to the input schema (doc previously advertised no limit) - GatewayDispatch gains a per-instance invoke_count spy accessor so the over-cap test proves zero dispatches (process-global counters raced under the parallel test runner) - tests: over-cap -> INVALID_INPUT + invoke_count()==0; at-cap -> 100 results + invoke_count()==100 verification: scripts/verify.sh (352 passed) and scripts/verify.sh --all-features (468 passed); cargo clippy --all-targets -D warnings and cargo fmt --check clean --- src/adapters/to_mcp.rs | 83 ++++++++++++++++++++++++++++++++++++-- src/adapters/to_openapi.rs | 3 +- src/gateway/dispatch.rs | 12 ++++++ src/gateway/mod.rs | 7 ++++ src/gateway/routes.rs | 2 +- 5 files changed, 101 insertions(+), 6 deletions(-) diff --git a/src/adapters/to_mcp.rs b/src/adapters/to_mcp.rs index 89d3e7c..3014b49 100644 --- a/src/adapters/to_mcp.rs +++ b/src/adapters/to_mcp.rs @@ -46,7 +46,7 @@ use rmcp::transport::{ }; use serde_json::{Map, Value}; -use crate::gateway::GatewayDispatch; +use crate::gateway::{GatewayDispatch, MAX_BATCH_OPERATIONS}; const TOOL_SEARCH: &str = "search"; const TOOL_SCHEMA: &str = "schema"; @@ -112,7 +112,8 @@ fn batch_input_schema() -> Value { }, "required": ["operation"] }, - "description": "The operations to invoke in this batch." + "maxItems": MAX_BATCH_OPERATIONS, + "description": "The operations to invoke in this batch. At most 100 operations per batch.", } }, "required": ["calls"] @@ -223,6 +224,11 @@ impl ToMcpGateway { )); } }; + if calls.len() > MAX_BATCH_OPERATIONS { + return call_error_to_structured_error(CallError::invalid_input(format!( + "batch exceeds the maximum of {MAX_BATCH_OPERATIONS} operations" + ))); + } let mut results: Vec = Vec::with_capacity(calls.len()); for call in calls { @@ -412,7 +418,7 @@ pub(crate) fn gateway_tools() -> Vec { Tool::new( Cow::Borrowed(TOOL_BATCH), Cow::Borrowed( - "Invoke multiple operations in one tool call. Returns {\"results\": [...]} where each entry is {\"isError\": false, \"output\": ...} on success or {\"isError\": true, \"error\": {code, message, retryable}} on failure.", + "Invoke multiple operations in one tool call, executed serially in order (each bounded by the 30 s deadline). Returns {\"results\": [...]} where each entry is {\"isError\": false, \"output\": ...} on success or {\"isError\": true, \"error\": {code, message, retryable}} on failure. At most 100 operations per batch; larger batches are rejected with INVALID_INPUT without dispatching.", ), value_to_object(batch_input_schema()), ), @@ -1171,6 +1177,77 @@ mod tests { ); } + #[tokio::test] + async fn batch_over_cap_returns_invalid_input_without_dispatching() { + let registry = full_registry_with_ops(vec![( + "echo/run".to_string(), + OperationType::Query, + AccessControl::default(), + )]); + let calls: Vec = (0..MAX_BATCH_OPERATIONS + 1) + .map(|i| serde_json::json!({ "operation": "echo/run", "input": { "n": i } })) + .collect(); + let mut args = Map::new(); + args.insert("calls".to_string(), Value::Array(calls)); + let dispatch_spine = dispatch(registry, provider()); + let gateway = ToMcpGateway::new(Arc::clone(&dispatch_spine)); + + let result = invoke_tool(&gateway, "batch", Some(args), None).await; + assert_eq!(result.is_error, Some(true)); + let structured = result.structured_content.expect("structured error present"); + assert_eq!( + structured.get("code"), + Some(&Value::String("INVALID_INPUT".to_string())) + ); + assert_eq!(structured.get("retryable"), Some(&Value::Bool(false))); + let message = structured + .get("message") + .and_then(Value::as_str) + .expect("message"); + assert!( + message.contains("maximum of 100"), + "the cap message must state the limit: {message}" + ); + assert_eq!( + dispatch_spine.invoke_count(), + 0, + "no operation may be dispatched when the batch is over cap" + ); + } + + #[tokio::test] + async fn batch_at_cap_is_accepted() { + let registry = full_registry_with_ops(vec![( + "echo/run".to_string(), + OperationType::Query, + AccessControl::default(), + )]); + let calls: Vec = (0..MAX_BATCH_OPERATIONS) + .map(|i| serde_json::json!({ "operation": "echo/run", "input": { "n": i } })) + .collect(); + let mut args = Map::new(); + args.insert("calls".to_string(), Value::Array(calls)); + let dispatch_spine = dispatch(registry, provider()); + let gateway = ToMcpGateway::new(Arc::clone(&dispatch_spine)); + + let result = invoke_tool(&gateway, "batch", Some(args), None).await; + assert_eq!(result.is_error, Some(false)); + let structured = result.structured_content.expect("structured present"); + let results = structured + .get("results") + .and_then(Value::as_array) + .expect("results array"); + assert_eq!(results.len(), MAX_BATCH_OPERATIONS); + for entry in results { + assert_eq!(entry.get("isError"), Some(&Value::Bool(false))); + } + assert_eq!( + dispatch_spine.invoke_count(), + MAX_BATCH_OPERATIONS, + "every entry in an at-cap batch is dispatched" + ); + } + #[tokio::test] async fn batch_returns_object_with_result_entries() { let registry = full_registry_with_ops(vec![( diff --git a/src/adapters/to_openapi.rs b/src/adapters/to_openapi.rs index b49397f..abb6ea9 100644 --- a/src/adapters/to_openapi.rs +++ b/src/adapters/to_openapi.rs @@ -60,6 +60,7 @@ use alkcall::registry::registration::OperationRegistry; use alkcall::registry::spec::ErrorDefinition; use super::openapi_spec::OpenAPISpec; +use crate::gateway::MAX_BATCH_OPERATIONS; const GATEWAY_VERSION: &str = "1.2.0"; const GATEWAY_TITLE: &str = "alk gateway"; @@ -93,8 +94,6 @@ const SCHEME_BEARER: &str = "bearerAuth"; const RESPONSE_REF: &str = "#/components/schemas/"; -const MAX_BATCH_OPERATIONS: usize = 100; - /// Project the registry into the fixed 6-endpoint gateway doc (ADR-042). /// /// Returns [`AdapterError::SchemaParse`] if the generated doc does not diff --git a/src/gateway/dispatch.rs b/src/gateway/dispatch.rs index e4ec582..b0e62fc 100644 --- a/src/gateway/dispatch.rs +++ b/src/gateway/dispatch.rs @@ -30,6 +30,7 @@ //! fixed window). use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -50,6 +51,7 @@ const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30); pub struct GatewayDispatch { registry: Arc, identity_provider: Arc, + invoke_count: AtomicUsize, } impl GatewayDispatch { @@ -62,6 +64,7 @@ impl GatewayDispatch { Self { registry, identity_provider, + invoke_count: AtomicUsize::new(0), } } @@ -75,6 +78,14 @@ impl GatewayDispatch { &self.identity_provider } + /// How many [`GatewayDispatch::invoke`] calls this spine has + /// served. A test-spy accessor: the over-cap batch tests assert it + /// stays at zero to prove no dispatch happened before the cap + /// rejection (review-002 PRJ-22). + pub fn invoke_count(&self) -> usize { + self.invoke_count.load(Ordering::Relaxed) + } + /// Resolve a bearer token to an identity (the auth-middleware hook). pub fn resolve_bearer(&self, token: &AuthToken) -> Option { self.identity_provider.resolve_from_token(token) @@ -88,6 +99,7 @@ impl GatewayDispatch { op: &str, input: Value, ) -> ResponseEnvelope { + self.invoke_count.fetch_add(1, Ordering::Relaxed); let operation_name = strip_leading_slash(op).to_string(); let request_id = uuid::Uuid::new_v4().to_string(); let context = self.build_root_context(&request_id, &operation_name, identity); diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index 46ef895..c419bb6 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -5,3 +5,10 @@ pub(crate) mod schema_cache; pub use dispatch::GatewayDispatch; pub use routes::CallRequest; + +/// The maximum number of operations accepted in a single batch, +/// enforced identically by the HTTP `POST /batch` endpoint, the +/// `to_openapi` gateway-spec projection, and the MCP `batch` tool +/// (review-002 PRJ-22). One shared constant so the surfaces cannot +/// drift. +pub const MAX_BATCH_OPERATIONS: usize = 100; diff --git a/src/gateway/routes.rs b/src/gateway/routes.rs index 2ad32fd..0d8c46f 100644 --- a/src/gateway/routes.rs +++ b/src/gateway/routes.rs @@ -45,12 +45,12 @@ use serde_json::{json, Value}; use super::dispatch::GatewayDispatch; use super::error::call_error_to_http_response_with_identity; use super::schema_cache::{CompileFailed, PublishSchemaCache}; +use super::MAX_BATCH_OPERATIONS; use crate::server::auth::ResolvedIdentity; use crate::server::state::RouterState; const SERVICES_LIST: &str = "services/list"; const SERVICES_SCHEMA: &str = "services/schema"; -const MAX_BATCH_OPERATIONS: usize = 100; const MAX_PUBLISH_LINE_BYTES: usize = 2 * 1024 * 1024; /// The explicit request-body limit for the whole gateway router