From e0dca77d02cca8a990e3c1ebffd3dab1e9191b80 Mon Sep 17 00:00:00 2001 From: Praveen Perera Date: Thu, 22 Oct 2026 15:32:11 -0500 Subject: [PATCH 1/3] Cap concurrent calls per upstream server A script that ran about 45 Linear searches under asyncio.gather sent them all at once, and Linear rate limited 14 of them. The gateway had no limit per server and never retries a call, so the agent had to notice and rerun searches by hand. Each server now takes max_concurrent (default 8, 1 to 4096). Calls past it wait their turn, up to the tool timeout, and one that gets no turn fails with a TimeoutError without being sent. Builtin servers are not capped, so a script's 1Password finds still share one prompt. A spoke's wait for a hub call covers the hub's queue too. When a tool's error reads as a rate limit, the error names the cap and the file to lower it in, the included one for a server kept there. --- docs/operations.md | 1 + src/config.rs | 76 ++++++++++++++++++++++++------- src/gateway.rs | 55 +++++++++++++++++++++-- src/hub.rs | 7 +-- src/hub/client.rs | 6 ++- src/upstream.rs | 50 ++++++++++++++++++++- tests/fixtures/fixture_server.rs | 44 +++++++++++++++++- tests/gateway/concurrency.rs | 77 ++++++++++++++++++++++++++++++++ tests/gateway/main.rs | 1 + 9 files changed, 291 insertions(+), 26 deletions(-) create mode 100644 tests/gateway/concurrency.rs diff --git a/docs/operations.md b/docs/operations.md index 8e45565..324e21d 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -794,6 +794,7 @@ policy = "deny" - `oauth = true` on a `url` server enables `oauth login` (optional `scopes`, `client_name`, or `client_id` and `client_secret` for a client registered ahead of time, and `authorization_params`). - Rarely used stdio servers: `lazy = true` starts the process on the first call and stops it after `idle_secs` (default 600) without calls; `/healthz` shows it as `idle` meanwhile and its tools stay searchable. A failed start fails that call only, and the next call tries again. +- Parallel calls: the gateway sends a server at most `max_concurrent` calls at once (default 8), and the rest of a script's `asyncio.gather` waits its turn, so a burst of calls does not trip the server's rate limit. A call that gets no turn within `tool_timeout_secs` fails with a `TimeoutError` without being sent, and nothing is retried. The setting takes 1 to 4096. When a tool's error reads as a rate limit (`rate limit`, `ratelimit` or `too many requests`), the script's error names this setting and the file to change it in, which is the included file for a server kept in one, so the agent can ask for it to be lowered, e.g. `max_concurrent = 4` under `[servers.linear]`. Builtin servers, such as `secrets`, answer inside the daemon and are not capped, so a script's finds still reach the owner as one prompt. - Keys stay in 1Password: a `headers`, `env` or `url` value can hold `{{ op://vault/item/field }}` references, as in `op inject`, e.g. `headers = { authorization = "Bearer {{ op://Dev/Context7/credential }}" }`. The daemon runs `op read` the first time the server connects and keeps the value in memory until it exits. A clean stop (`service restart`, `update`, a reboot) hands the values to the next daemon sealed, so it asks nothing again; to pick up a rotated key, run `codemode-mcp service restart --fresh`. With the 1Password app integration the Mac asks once when the daemon starts (a locked or unanswered prompt shows `secret_unavailable` in `/healthz` and retries every 10 minutes). `[secrets] op` names the `op` binary (default `op` on `PATH`). A `url` that carries the key, as Zoho's MCP URLs do, should embed the reference where the key sits: `url = "https://-.zohomcp.com/mcp/{{ op://Personal/Zoho MCP - InfraOps/key }}/message"`. References in a `url` may sit only in path segments and query names or values, never in the scheme, credentials, host, port or fragment, and the `url` needs a literal host; anything else is refused when the config loads. A key embedded in a `url` may hold only ASCII letters, digits and `- . _ ~ + / = : @ ! * , ;` (at least 16 bytes), which covers hex, base64, base64url, JWT-like and colon-separated keys; any other character, such as `\`, quotes, `<`, `>`, `^`, `{`, `}`, `|`, `[`, `]`, `#`, `?`, `%`, `&`, `$`, `(`, `)` or a space, refuses the server with a configuration error. It is encoded as data where it sits, so `/` or `+` in a key stays part of that one segment or query value. The secrets are exactly the values 1Password returns, in the encodings libraries print them in (percent-encoded in either case, JSON-escaped, HTML character references such as `+` or `+`, `+` read as a space), matched ignoring ASCII case, and any run of 8 or more of its bytes is redacted too, since rmcp cuts invalid responses to 256 bytes and can leave only part of a key; nothing else in the URL is guessed to be a key. A value read for an embedded reference that is itself an `http(s)` URL is refused with a configuration error naming the reference, and dropped here and from the phones' copies, so the next attempt asks again instead of replaying it. A `url` that is one whole reference (`url = "{{ op://Personal/Zoho MCP infraops/credential }}"`) is refused when the config loads, with or without OAuth, because codemode cannot tell which part of such a URL is the key; embed the reference instead, e.g. `url = "https://host/mcp/{{ op://Vault/Item/key }}"`. Log text is matched through nested escape layers (a client's `Debug` output escapes the JSON escapes of a response body), up to 4 passes. Only the configured template is ever shown (companion prompts name the reference, never the value), and the resolved URL is redacted from errors, `/healthz` and the daemon's logs. Whatever OAuth would store (credentials, pending state, the client registration) or print (the authorization link) is refused when it holds a key, and the link's `resource` parameter is left out only when it holds the key. - Per tool: `policy` (`allow`, `ask`, `deny`) and, for `ask`, `grant` (`session` or `per_call`). A `session` grant is given to the calling client, not to one MCP session: it covers every running session of that client name for 15 or 60 minutes. - The server is session-less: it issues no `mcp-session-id`, for any protocol version. Clients are told apart by the client name in the request `_meta`, the `initialize` handshake or the `user-agent` header. While `execute` or `wait` is waiting, it sends progress notifications every `limits.progress_interval_ms` (15000) to clients that sent a `progressToken`. diff --git a/src/config.rs b/src/config.rs index f6251b4..4e2217a 100644 --- a/src/config.rs +++ b/src/config.rs @@ -147,6 +147,10 @@ pub struct Config { /// nothing can be written back #[serde(skip)] path: Option, + /// For servers set up across included files, the file that sets each + /// one's `max_concurrent`, or else the file holding its table + #[serde(skip)] + server_files: BTreeMap, } impl Default for Config { @@ -169,6 +173,7 @@ impl Default for Config { hub: None, servers: BTreeMap::new(), path: None, + server_files: BTreeMap::new(), } } } @@ -261,6 +266,15 @@ impl Config { self.path.as_deref() } + /// The file to change `servers..max_concurrent` in: the one + /// that sets it, or else the one holding the server's table + pub fn max_concurrent_file(&self, server: &str) -> Option<&Path> { + self.server_files + .get(server) + .map(PathBuf::as_path) + .or(self.path()) + } + /// Parses and validates configuration text; `path` is used in errors and /// to resolve relative `include` paths /// @@ -309,21 +323,35 @@ impl Config { } merged.add(table, path)?; - merged - .table - .try_into() - .map_err(|source| ConfigError::ParseIncluded { - files: format!( - "{} together with {}", - path.display(), - included_paths - .iter() - .map(|included| included.display().to_string()) - .collect::>() - .join(", ") - ), - source, - })? + let Merged { table, origins } = merged; + let mut config: Self = + table + .try_into() + .map_err(|source| ConfigError::ParseIncluded { + files: format!( + "{} together with {}", + path.display(), + included_paths + .iter() + .map(|included| included.display().to_string()) + .collect::>() + .join(", ") + ), + source, + })?; + // a value already set must be changed where it is, or the + // files would clash + for name in config.servers.keys() { + let key = ["servers", name, "max_concurrent"].map(str::to_owned); + if let Some(file) = (1..=key.len()) + .rev() + .find_map(|end| origins.get(&key[..end])) + { + config.server_files.insert(name.clone(), file.clone()); + } + } + + config } }; config.validate()?; @@ -693,6 +721,14 @@ fn default_idle_secs() -> u64 { 10 * 60 } +/// Highest `max_concurrent`; more is no limit in practice, and the +/// semaphore that enforces it panics past its own maximum +const MAX_CONCURRENT: usize = 4096; + +fn default_max_concurrent() -> usize { + 8 +} + fn default_op() -> String { "op".to_owned() } @@ -1254,6 +1290,11 @@ pub struct ServerConfig { /// Seconds a `lazy` server may go without calls before it is stopped #[serde(default = "default_idle_secs")] pub idle_secs: u64, + /// Calls the gateway sends this server at once; the rest wait their + /// turn, so a script's `asyncio.gather` cannot trip the server's rate + /// limit. Builtin servers answer inside the daemon and are not limited + #[serde(default = "default_max_concurrent")] + pub max_concurrent: usize, /// Endpoint of a Streamable HTTP server; may hold `{{ op://... }}` /// references for a URL that carries a key, and is then never shown /// resolved @@ -1485,6 +1526,11 @@ impl ServerConfig { } fn validate(&self, name: &str) -> Result<(), ConfigError> { + if !(1..=MAX_CONCURRENT).contains(&self.max_concurrent) { + return Err(ConfigError::Invalid(format!( + "servers.{name}.max_concurrent must be from 1 to {MAX_CONCURRENT}" + ))); + } for (tool, tool_override) in &self.tools { tool_override.validate(name, tool)?; } diff --git a/src/gateway.rs b/src/gateway.rs index 07da4e5..bc1cbab 100644 --- a/src/gateway.rs +++ b/src/gateway.rs @@ -119,7 +119,9 @@ impl ToolCallError { match self { Self::Unknown(_) => ExcType::NameError, Self::Denied(_) | Self::NotApproved { .. } => ExcType::PermissionError, - Self::Upstream(UpstreamError::TimedOut { .. }) => ExcType::TimeoutError, + Self::Upstream(UpstreamError::TimedOut { .. } | UpstreamError::Busy { .. }) => { + ExcType::TimeoutError + } Self::ResultTooLarge { .. } => ExcType::ValueError, Self::Hub(error) => match error.kind { HubErrorKind::Unknown => ExcType::NameError, @@ -140,7 +142,9 @@ impl ToolCallError { match self { Self::Unknown(_) => Outcome::NameError, Self::Denied(_) | Self::NotApproved { .. } => Outcome::Denied, - Self::Upstream(UpstreamError::TimedOut { .. }) => Outcome::Timeout, + Self::Upstream(UpstreamError::TimedOut { .. } | UpstreamError::Busy { .. }) => { + Outcome::Timeout + } Self::Hub(error) => match error.kind { HubErrorKind::Unknown => Outcome::NameError, HubErrorKind::Denied => Outcome::Denied, @@ -1042,7 +1046,7 @@ impl Gateway { .map_err(|error| match error { // the upstream's own error message can echo configured secrets UpstreamError::Service(rmcp::service::ServiceError::McpError(_)) => { - ToolCallError::ToolFailed(self.redactor.apply(&error.public_message())) + self.tool_failed(&call.server, self.redactor.apply(&error.public_message())) } error => ToolCallError::Upstream(error), })?; @@ -1298,6 +1302,33 @@ fn omit_none_arguments(entry: &ToolEntry, mut arguments: Map) -> } impl Gateway { + /// A tool's error, with where to lower `max_concurrent` when it reads + /// as the server's rate limit, so the agent can say what to change + fn tool_failed(&self, server: &str, message: String) -> ToolCallError { + let limit = self + .config + .servers + .get(server) + .filter(|config| config.builtin.is_none()) + .map(|config| config.max_concurrent); + let Some(limit) = limit.filter(|_| looks_rate_limited(&message)) else { + return ToolCallError::ToolFailed(message); + }; + let config = self.config.max_concurrent_file(server).map_or_else( + || "the codemode-mcp config".to_owned(), + |path| format!("`{}`", path.display()), + ); + + // the path is the owner's own, but anything shown passes the + // redactor, which a value read from 1Password may share text with + ToolCallError::ToolFailed(self.redactor.apply(&format!( + "{message}\n\n`{server}` looks rate limited. codemode sends it at most {limit} calls \ + at once; make fewer calls at once, or ask the user to set a lower `max_concurrent` under \ + `[servers.{server}]` in {config} on the machine that runs `{server}`, then run \ + `codemode-mcp service restart`" + ))) + } + /// Converts an MCP tool result to the value a script receives /// /// Structured content wins. Without it, text that is a JSON object or @@ -1335,7 +1366,7 @@ impl Gateway { self.redactor.apply(&text.join("\n")) }; - return Err(ToolCallError::ToolFailed(message)); + return Err(self.tool_failed(&call.server, message)); } if matches!(call.output, OutputSpec::Text) { // a text tool that sends only structured content still gives the @@ -1361,6 +1392,22 @@ impl Gateway { } } +/// Whether an upstream's error reads as a rate limit, such as HTTP 429's +/// `Too Many Requests` +fn looks_rate_limited(message: &str) -> bool { + let message = message.to_ascii_lowercase(); + + [ + "rate limit", + "rate-limit", + "rate_limit", + "ratelimit", + "too many requests", + ] + .iter() + .any(|phrase| message.contains(phrase)) +} + /// Parses text blocks as one JSON object or array; scalars stay text because /// a tool answering `123` or `true` as prose is more likely than as data fn json_from_text(blocks: &[String]) -> Option { diff --git a/src/hub.rs b/src/hub.rs index c55b8ef..a4cd325 100644 --- a/src/hub.rs +++ b/src/hub.rs @@ -195,9 +195,10 @@ impl From<&ToolCallError> for HubError { let kind = match error { ToolCallError::Unknown(_) => HubErrorKind::Unknown, ToolCallError::Denied(_) | ToolCallError::NotApproved { .. } => HubErrorKind::Denied, - ToolCallError::Upstream(crate::upstream::UpstreamError::TimedOut { .. }) => { - HubErrorKind::TimedOut - } + ToolCallError::Upstream( + crate::upstream::UpstreamError::TimedOut { .. } + | crate::upstream::UpstreamError::Busy { .. }, + ) => HubErrorKind::TimedOut, ToolCallError::ResultTooLarge { .. } => HubErrorKind::TooLarge, ToolCallError::Hub(error) => error.kind, _ => HubErrorKind::Failed, diff --git a/src/hub/client.rs b/src/hub/client.rs index 6ecf970..5522fd8 100644 --- a/src/hub/client.rs +++ b/src/hub/client.rs @@ -279,12 +279,16 @@ impl HubClient { /// Runs a call of the hub's tool there; `tool_timeout` is the spoke's /// own limit, which the hub's may exceed + /// + /// The hub may first queue the call behind the server's + /// `max_concurrent` for up to its timeout, so the wait covers that too pub async fn call( &self, call: &ForwardedCall, tool_timeout: Duration, ) -> Result { - self.post("/call", call, tool_timeout + QUICK).await + self.post("/call", call, tool_timeout.saturating_mul(2) + QUICK) + .await } /// The hub's verdict on request `id`, after a long poll; `None` while it diff --git a/src/upstream.rs b/src/upstream.rs index 4f764ca..f8136a0 100644 --- a/src/upstream.rs +++ b/src/upstream.rs @@ -32,7 +32,7 @@ use serde_json::{Map, Value}; use tokio::{ io::{AsyncBufReadExt, BufReader, DuplexStream}, process::Command, - sync::{Mutex, Notify, watch}, + sync::{Mutex, Notify, Semaphore, SemaphorePermit, watch}, task::JoinHandle, }; @@ -140,6 +140,20 @@ pub enum UpstreamError { /// Timeout that elapsed timeout: Duration, }, + /// Every `max_concurrent` slot stayed taken for the call's timeout; the + /// call was never sent, so failing it is safe + #[error( + "`{server}` had {limit} calls running and this one got no turn within {waited:?}; \ + it was not sent. Make fewer calls at once" + )] + Busy { + /// Server name + server: String, + /// Its `max_concurrent` + limit: usize, + /// How long the call waited + waited: Duration, + }, /// The server asked for something other than a tool result #[error("server asked for interactive input, which is not supported")] InputRequired, @@ -170,7 +184,9 @@ impl UpstreamError { Self::Header(_) | Self::Url | Self::UrlValue(_) => ErrorCategory::Configuration, Self::Secret(_) => ErrorCategory::Secret, Self::Initialize(_) => ErrorCategory::Initialize, - Self::ConnectTimeout(_) | Self::TimedOut { .. } => ErrorCategory::Timeout, + Self::ConnectTimeout(_) | Self::TimedOut { .. } | Self::Busy { .. } => { + ErrorCategory::Timeout + } Self::Service(_) | Self::InputRequired | Self::SetupFailed { .. } => { ErrorCategory::Protocol } @@ -188,6 +204,7 @@ impl UpstreamError { match self { Self::Unavailable { .. } | Self::TimedOut { .. } + | Self::Busy { .. } | Self::InputRequired | Self::SetupFailed { .. } | Self::NeedsLogin(_) => self.to_string(), @@ -346,6 +363,9 @@ pub struct Upstream { setup_epoch: AtomicU64, /// Serializes setup, so concurrent first calls run it once setup_turn: Mutex<()>, + /// Caps the calls in flight at `max_concurrent`; none for a builtin + /// server, whose finds must all reach the owner as one prompt + call_slots: Option, /// Logins for `oauth = true` servers oauth: Option>, /// Where the last tool list is kept for the next start @@ -554,6 +574,7 @@ impl Upstream { arguments: Map, ) -> Result { let _in_use = self.begin_use(); + let _slot = self.call_slot().await?; let (peer, epoch) = self.awake_peer().await?; self.ensure_setup(&peer, epoch, tool).await?; @@ -561,6 +582,27 @@ impl Upstream { .await } + /// Waits for one of the `max_concurrent` slots, so the server is never + /// sent more calls at once and nothing has to be retried + /// + /// The wait is bounded by the call's own timeout, so a call forwarded + /// from a spoke answers before the spoke stops waiting for it + async fn call_slot(&self) -> Result>, UpstreamError> { + let Some(slots) = &self.call_slots else { + return Ok(None); + }; + + match tokio::time::timeout(self.call_timeout, slots.acquire()).await { + // the semaphore is never closed + Ok(permit) => Ok(permit.ok()), + Err(_) => Err(UpstreamError::Busy { + server: self.name.clone(), + limit: self.config.max_concurrent, + waited: self.call_timeout, + }), + } + } + /// Adds the server's configured `arguments` that `tool` declares and the /// caller left out; the stubs hide them, so scripts never pass them fn with_fixed_arguments( @@ -1386,6 +1428,10 @@ impl Upstreams { refresh_turn: Mutex::new(()), setup_epoch: AtomicU64::new(0), setup_turn: Mutex::new(()), + call_slots: config + .builtin + .is_none() + .then(|| Semaphore::new(config.max_concurrent)), oauth: oauth.clone(), store: store.clone(), reconnect: watch::channel(0).0, diff --git a/tests/fixtures/fixture_server.rs b/tests/fixtures/fixture_server.rs index 0d140bc..58d7892 100644 --- a/tests/fixtures/fixture_server.rs +++ b/tests/fixtures/fixture_server.rs @@ -3,7 +3,7 @@ use std::{ sync::{ Arc, - atomic::{AtomicBool, Ordering}, + atomic::{AtomicBool, AtomicUsize, Ordering}, }, time::Duration, }; @@ -64,6 +64,28 @@ struct EchoParams { delay_ms: u64, } +#[derive(Deserialize, schemars::JsonSchema)] +struct OverlapParams { + /// How long to stay in flight + delay_ms: u64, +} + +#[derive(Serialize, schemars::JsonSchema)] +struct Overlap { + /// Most calls of this tool in flight at once so far + peak: usize, +} + +#[derive(Deserialize, schemars::JsonSchema)] +struct FailParams { + /// Text of the error to answer with + message: String, +} + +/// Calls of `overlap` running now, and the most seen at once +static IN_FLIGHT: AtomicUsize = AtomicUsize::new(0); +static PEAK: AtomicUsize = AtomicUsize::new(0); + #[derive(Deserialize, schemars::JsonSchema)] struct UseTokenParams { /// The token to use @@ -316,6 +338,26 @@ impl Fixture { }) } + #[tool(description = "Answer with an error carrying the message")] + fn fail( + &self, + Parameters(FailParams { message }): Parameters, + ) -> Result { + Err(message) + } + + #[tool(description = "Stay in flight for delay_ms, then report the most calls seen at once")] + async fn overlap(&self, Parameters(params): Parameters) -> Json { + let running = IN_FLIGHT.fetch_add(1, Ordering::SeqCst) + 1; + PEAK.fetch_max(running, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(params.delay_ms)).await; + IN_FLIGHT.fetch_sub(1, Ordering::SeqCst); + + Json(Overlap { + peak: PEAK.load(Ordering::SeqCst), + }) + } + #[tool(description = "The largest u64, which no i64 can hold")] fn wide(&self) -> Json { Json(Wide { n: u64::MAX }) diff --git a/tests/gateway/concurrency.rs b/tests/gateway/concurrency.rs new file mode 100644 index 0000000..8b25ffa --- /dev/null +++ b/tests/gateway/concurrency.rs @@ -0,0 +1,77 @@ +//! `max_concurrent`: calls past a server's cap wait their turn, and a rate +//! limit error says where to lower the cap + +use super::harness::{Harness, Setup, TestResult}; + +#[tokio::test(flavor = "multi_thread")] +async fn calls_past_the_cap_wait_and_rate_limits_name_the_setting() -> TestResult { + let fixture = env!("CARGO_BIN_EXE_codemode-fixture"); + let extra = format!("[servers.roomy]\ncommand = \"{fixture}\"\n"); + // a server kept in an included file is changed there, not in config.toml + let shared = std::env::temp_dir().join(format!( + "codemode-concurrency-{}-servers.toml", + std::process::id() + )); + std::fs::write( + &shared, + format!("[servers.shared]\ncommand = \"{fixture}\"\nmax_concurrent = 3\n"), + )?; + let top = format!("include = [\"{}\"]", shared.display()); + let Some(harness) = Harness::with(Setup { + server: "max_concurrent = 2", + extra: &extra, + top: &top, + ..Setup::default() + }) + .await? + else { + return Ok(()); + }; + + // every call still runs; none is refused for waiting + let body = harness + .execute_json( + "import asyncio\n\ + capped = await asyncio.gather(*[fixture_overlap(delay_ms=200) for _ in range(6)])\n\ + roomy = await asyncio.gather(*[roomy_overlap(delay_ms=200) for _ in range(12)])\n\ + [len(capped), max(r[\"peak\"] for r in capped), len(roomy), max(r[\"peak\"] for r in roomy)]", + ) + .await?; + assert_eq!(body["result"], serde_json::json!([6, 2, 12, 8]), "{body}"); + + let body = harness + .execute_json( + "messages: list[str] = []\n\ + for message in [\"429 Too Many Requests\", \"no such issue\"]:\n \ + try:\n await fixture_fail(message=message)\n \ + except RuntimeError as error:\n messages.append(str(error))\n\ + messages", + ) + .await?; + let limited = body["result"][0].as_str().unwrap_or_default(); + assert!( + limited.contains("at most 2 calls") + && limited.contains("`max_concurrent` under `[servers.fixture]`") + && limited.contains("config.toml"), + "{body}" + ); + let other = body["result"][1].as_str().unwrap_or_default(); + assert_eq!(other, "no such issue", "{body}"); + + let body = harness + .execute_json( + "message = \"\"\n\ + try:\n await shared_fail(message=\"Rate limit exceeded\")\n\ + except RuntimeError as error:\n message = str(error)\n\ + message", + ) + .await?; + let _ = std::fs::remove_file(&shared); + let limited = body["result"].as_str().unwrap_or_default(); + assert!( + limited.contains("at most 3 calls") && limited.contains(&format!("`{}`", shared.display())), + "{body}" + ); + + Ok(()) +} diff --git a/tests/gateway/main.rs b/tests/gateway/main.rs index 365b789..47efbf1 100644 --- a/tests/gateway/main.rs +++ b/tests/gateway/main.rs @@ -5,6 +5,7 @@ mod admin_gate; mod apple; mod approvals; mod companion; +mod concurrency; mod durable; mod harness; mod lazy; From 7b0a39390d3d1301c4fc15e3e006fc324a237f86 Mon Sep 17 00:00:00 2001 From: Praveen Perera Date: Thu, 22 Oct 2026 15:42:56 -0500 Subject: [PATCH 2/3] Wait out the hub's own timeout for hub calls A spoke waited twice its own tool timeout for a call forwarded to the hub. A hub with a longer timeout could queue a call and then run it past that, so the spoke reported a timeout for a call the hub sent. The hub's catalog now lists each server's call timeout, and the spoke waits twice the longer of the two. An older hub lists none, so the spoke's own timeout stands in. --- docs/operations.md | 2 +- src/gateway.rs | 17 ++++++++++++++++- src/hub.rs | 4 ++++ src/hub/client.rs | 6 +++--- src/upstream.rs | 5 +++++ 5 files changed, 29 insertions(+), 5 deletions(-) diff --git a/docs/operations.md b/docs/operations.md index 324e21d..06e3e79 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -794,7 +794,7 @@ policy = "deny" - `oauth = true` on a `url` server enables `oauth login` (optional `scopes`, `client_name`, or `client_id` and `client_secret` for a client registered ahead of time, and `authorization_params`). - Rarely used stdio servers: `lazy = true` starts the process on the first call and stops it after `idle_secs` (default 600) without calls; `/healthz` shows it as `idle` meanwhile and its tools stay searchable. A failed start fails that call only, and the next call tries again. -- Parallel calls: the gateway sends a server at most `max_concurrent` calls at once (default 8), and the rest of a script's `asyncio.gather` waits its turn, so a burst of calls does not trip the server's rate limit. A call that gets no turn within `tool_timeout_secs` fails with a `TimeoutError` without being sent, and nothing is retried. The setting takes 1 to 4096. When a tool's error reads as a rate limit (`rate limit`, `ratelimit` or `too many requests`), the script's error names this setting and the file to change it in, which is the included file for a server kept in one, so the agent can ask for it to be lowered, e.g. `max_concurrent = 4` under `[servers.linear]`. Builtin servers, such as `secrets`, answer inside the daemon and are not capped, so a script's finds still reach the owner as one prompt. +- Parallel calls: the gateway sends a server at most `max_concurrent` calls at once (default 8), and the rest of a script's `asyncio.gather` waits its turn, so a burst of calls does not trip the server's rate limit. A call that gets no turn within `tool_timeout_secs` fails with a `TimeoutError` without being sent, and nothing is retried. A spoke calling one of the hub's servers waits out the hub's queue too, using the longer of its own `tool_timeout_secs` and the hub's. The setting takes 1 to 4096. When a tool's error reads as a rate limit (`rate limit`, `ratelimit` or `too many requests`), the script's error names this setting and the file to change it in, which is the included file for a server kept in one, so the agent can ask for it to be lowered, e.g. `max_concurrent = 4` under `[servers.linear]`. Builtin servers, such as `secrets`, answer inside the daemon and are not capped, so a script's finds still reach the owner as one prompt. - Keys stay in 1Password: a `headers`, `env` or `url` value can hold `{{ op://vault/item/field }}` references, as in `op inject`, e.g. `headers = { authorization = "Bearer {{ op://Dev/Context7/credential }}" }`. The daemon runs `op read` the first time the server connects and keeps the value in memory until it exits. A clean stop (`service restart`, `update`, a reboot) hands the values to the next daemon sealed, so it asks nothing again; to pick up a rotated key, run `codemode-mcp service restart --fresh`. With the 1Password app integration the Mac asks once when the daemon starts (a locked or unanswered prompt shows `secret_unavailable` in `/healthz` and retries every 10 minutes). `[secrets] op` names the `op` binary (default `op` on `PATH`). A `url` that carries the key, as Zoho's MCP URLs do, should embed the reference where the key sits: `url = "https://-.zohomcp.com/mcp/{{ op://Personal/Zoho MCP - InfraOps/key }}/message"`. References in a `url` may sit only in path segments and query names or values, never in the scheme, credentials, host, port or fragment, and the `url` needs a literal host; anything else is refused when the config loads. A key embedded in a `url` may hold only ASCII letters, digits and `- . _ ~ + / = : @ ! * , ;` (at least 16 bytes), which covers hex, base64, base64url, JWT-like and colon-separated keys; any other character, such as `\`, quotes, `<`, `>`, `^`, `{`, `}`, `|`, `[`, `]`, `#`, `?`, `%`, `&`, `$`, `(`, `)` or a space, refuses the server with a configuration error. It is encoded as data where it sits, so `/` or `+` in a key stays part of that one segment or query value. The secrets are exactly the values 1Password returns, in the encodings libraries print them in (percent-encoded in either case, JSON-escaped, HTML character references such as `+` or `+`, `+` read as a space), matched ignoring ASCII case, and any run of 8 or more of its bytes is redacted too, since rmcp cuts invalid responses to 256 bytes and can leave only part of a key; nothing else in the URL is guessed to be a key. A value read for an embedded reference that is itself an `http(s)` URL is refused with a configuration error naming the reference, and dropped here and from the phones' copies, so the next attempt asks again instead of replaying it. A `url` that is one whole reference (`url = "{{ op://Personal/Zoho MCP infraops/credential }}"`) is refused when the config loads, with or without OAuth, because codemode cannot tell which part of such a URL is the key; embed the reference instead, e.g. `url = "https://host/mcp/{{ op://Vault/Item/key }}"`. Log text is matched through nested escape layers (a client's `Debug` output escapes the JSON escapes of a response body), up to 4 passes. Only the configured template is ever shown (companion prompts name the reference, never the value), and the resolved URL is redacted from errors, `/healthz` and the daemon's logs. Whatever OAuth would store (credentials, pending state, the client registration) or print (the authorization link) is refused when it holds a key, and the link's `resource` parameter is left out only when it holds the key. - Per tool: `policy` (`allow`, `ask`, `deny`) and, for `ask`, `grant` (`session` or `per_call`). A `session` grant is given to the calling client, not to one MCP session: it covers every running session of that client name for 15 or 60 minutes. - The server is session-less: it issues no `mcp-session-id`, for any protocol version. Clients are told apart by the client name in the request `_meta`, the `initialize` handshake or the `user-agent` header. While `execute` or `wait` is waiting, it sends progress notifications every `limits.progress_interval_ms` (15000) to clients that sent a `progressToken`. diff --git a/src/gateway.rs b/src/gateway.rs index bc1cbab..8aaf57b 100644 --- a/src/gateway.rs +++ b/src/gateway.rs @@ -1021,8 +1021,19 @@ impl Gateway { approval: call.hub_request.clone(), sensitive: call.sensitive, }; + // the hub's own limit governs how long the call may queue and + // run there; an older hub does not say, so ours stands in + let timeout = self + .hub_view() + .servers + .iter() + .find(|server| server.name == call.server) + .and_then(|server| server.call_timeout_secs) + .map_or(self.config.tool_timeout(), |secs| { + Duration::from_secs(secs).max(self.config.tool_timeout()) + }); let answer = hub - .call(&forwarded, self.config.tool_timeout()) + .call(&forwarded, timeout) .await .map_err(|error| hub_unreachable(&error))?; let value = match answer { @@ -1187,6 +1198,10 @@ impl Gateway { .map(|server| server.aliases.clone()) .unwrap_or_default(), tools: export_tools(&self.config, snapshot), + call_timeout_secs: self + .upstreams + .get(&snapshot.name) + .map(|upstream| upstream.call_timeout().as_secs()), }) }) .collect(); diff --git a/src/hub.rs b/src/hub.rs index a4cd325..3e3d1af 100644 --- a/src/hub.rs +++ b/src/hub.rs @@ -93,6 +93,10 @@ pub struct HubServer { pub aliases: Vec, /// Its tools pub tools: Vec, + /// Seconds the hub lets a call of it wait for a turn under + /// `max_concurrent`, and then run; absent from older hubs + #[serde(default, skip_serializing_if = "Option::is_none")] + pub call_timeout_secs: Option, } /// Body of `POST /hub/v1/gate`: asks the hub whether a call may run now diff --git a/src/hub/client.rs b/src/hub/client.rs index 5522fd8..0d00af6 100644 --- a/src/hub/client.rs +++ b/src/hub/client.rs @@ -277,11 +277,11 @@ impl HubClient { self.post("/gate", call, QUICK).await } - /// Runs a call of the hub's tool there; `tool_timeout` is the spoke's - /// own limit, which the hub's may exceed + /// Runs a call of the hub's tool there; `tool_timeout` is the longer of + /// the spoke's limit and the one the hub lists for the server /// /// The hub may first queue the call behind the server's - /// `max_concurrent` for up to its timeout, so the wait covers that too + /// `max_concurrent` for up to that timeout, so the wait covers that too pub async fn call( &self, call: &ForwardedCall, diff --git a/src/upstream.rs b/src/upstream.rs index f8136a0..fd6e406 100644 --- a/src/upstream.rs +++ b/src/upstream.rs @@ -459,6 +459,11 @@ impl Upstream { self.connected_peer().map(|_| ()) } + /// How long one call may wait for a turn, and then how long it may run + pub fn call_timeout(&self) -> Duration { + self.call_timeout + } + /// Whether the catalog offers this server's tools as callable pub fn callable(&self) -> bool { self.config.lazy || self.status() == UpstreamStatus::Connected From 6b6630e336ce443cfeb9b6a91d6cbe86c76a58b4 Mon Sep 17 00:00:00 2001 From: Praveen Perera Date: Thu, 22 Oct 2026 15:44:58 -0500 Subject: [PATCH 3/3] Hold a hub's listed call timeout to a day A hub names its own call timeout now, and doubling a huge one then adding the allowance would overflow the spoke's deadline and panic. --- src/hub/client.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/hub/client.rs b/src/hub/client.rs index 0d00af6..490a48a 100644 --- a/src/hub/client.rs +++ b/src/hub/client.rs @@ -40,6 +40,8 @@ use crate::{ pub const ID_FILE: &str = "spoke.id"; /// How long a request that needs no person may take const QUICK: Duration = Duration::from_secs(30); +/// The longest call timeout a forwarded call is waited out for +const LONGEST_CALL: Duration = Duration::from_secs(24 * 60 * 60); /// How long a request that may wait for the owner may take: a long poll, /// or a read the owner approves on a device const PATIENT: Duration = Duration::from_secs(15 * 60); @@ -287,8 +289,11 @@ impl HubClient { call: &ForwardedCall, tool_timeout: Duration, ) -> Result { - self.post("/call", call, tool_timeout.saturating_mul(2) + QUICK) - .await + // the hub names its own timeout, so a huge one is held to a day + // rather than overflowing the deadline + let wait = tool_timeout.min(LONGEST_CALL).saturating_mul(2) + QUICK; + + self.post("/call", call, wait).await } /// The hub's verdict on request `id`, after a long poll; `None` while it