diff --git a/server/src/gateway/tests.rs b/server/src/gateway/tests.rs index 6fb5b96d..a1e5c80c 100644 --- a/server/src/gateway/tests.rs +++ b/server/src/gateway/tests.rs @@ -848,6 +848,196 @@ async fn gateway_mcp_sse_proxy_streams_progress_and_final_response() { ); } +// ── SSE path honors explicit `project` and the create-guard (story 1229) ── +// +// Story 1225 added explicit-project routing and the create-without-project +// guard, but only on the buffered `tools/call` path — the SSE branch above +// (Accept: text/event-stream + `_meta.progressToken`) still always proxied +// to `state.active_url()` and never ran the guard. These tests exercise the +// SSE branch specifically so a regression here fails a test, unlike 1225's +// tests at `create_tool_with_explicit_project_lands_in_non_active_project` / +// `read_tool_with_explicit_project_reads_from_non_active_project` in +// `http/gateway/mcp.rs`, which bypass SSE entirely. + +fn sse_body_for(id: i64, text: &str) -> String { + let final_resp = serde_json::json!({ + "jsonrpc": "2.0", + "id": id, + "result": { "content": [{ "type": "text", "text": text }] } + }); + format!("data: {final_resp}\n\n") +} + +async fn make_two_http_project_state() -> ( + Arc, + mockito::ServerGuard, + mockito::ServerGuard, +) { + let mut mock_alpha = mockito::Server::new_async().await; + let mut mock_beta = mockito::Server::new_async().await; + mock_alpha + .mock("POST", "/mcp") + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_body(sse_body_for(1, "handled_by:alpha")) + .create_async() + .await; + mock_beta + .mock("POST", "/mcp") + .with_status(200) + .with_header("content-type", "text/event-stream") + .with_body(sse_body_for(1, "handled_by:beta")) + .create_async() + .await; + + let mut projects = BTreeMap::new(); + projects.insert( + "alpha".to_string(), + ProjectEntry::with_url(mock_alpha.url()), + ); + projects.insert("beta".to_string(), ProjectEntry::with_url(mock_beta.url())); + let config = GatewayConfig { + projects, + sled_tokens: BTreeMap::new(), + release_channels: BTreeMap::new(), + }; + let state = Arc::new(GatewayState::new(config, PathBuf::new(), 3000).unwrap()); + assert_eq!(*state.active_project.read().await, "alpha"); + (state, mock_alpha, mock_beta) +} + +fn sse_data_lines(body: &str) -> Vec<&str> { + body.lines() + .filter_map(|l| l.strip_prefix("data: ")) + .collect() +} + +#[tokio::test] +async fn sse_create_tool_with_explicit_project_lands_in_non_active_project() { + let (state, _mock_alpha, _mock_beta) = make_two_http_project_state().await; + let app = poem::Route::new() + .at("/mcp", poem::post(gateway_mcp_post_handler)) + .data(state); + let cli = poem::test::TestClient::new(app); + + let rpc_body = serde_json::to_vec(&serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "create_bug", + "arguments": { "project": "beta", "title": "x", "description": "y" }, + "_meta": { "progressToken": "tok1" } + } + })) + .unwrap(); + let resp = cli + .post("/mcp") + .header("content-type", "application/json") + .header("accept", "text/event-stream") + .body(rpc_body) + .send() + .await; + let body = resp.0.into_body().into_string().await.unwrap(); + let lines = sse_data_lines(&body); + assert_eq!( + lines.len(), + 1, + "expected exactly one SSE data event: {body}" + ); + let ev: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert_eq!( + ev["result"]["content"][0]["text"], "handled_by:beta", + "explicit project=beta on the SSE path must land in beta even though \ + alpha is active: {ev}" + ); +} + +#[tokio::test] +async fn sse_read_tool_with_explicit_project_reads_from_non_active_project() { + let (state, _mock_alpha, _mock_beta) = make_two_http_project_state().await; + let app = poem::Route::new() + .at("/mcp", poem::post(gateway_mcp_post_handler)) + .data(state); + let cli = poem::test::TestClient::new(app); + + let rpc_body = serde_json::to_vec(&serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "list_upcoming", + "arguments": { "project": "beta" }, + "_meta": { "progressToken": "tok1" } + } + })) + .unwrap(); + let resp = cli + .post("/mcp") + .header("content-type", "application/json") + .header("accept", "text/event-stream") + .body(rpc_body) + .send() + .await; + let body = resp.0.into_body().into_string().await.unwrap(); + let lines = sse_data_lines(&body); + assert_eq!( + lines.len(), + 1, + "expected exactly one SSE data event: {body}" + ); + let ev: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert_eq!( + ev["result"]["content"][0]["text"], "handled_by:beta", + "explicit project=beta on the SSE path must read from beta (audited \ + generically, not via a hand-listed subset — story 1229 AC 4): {ev}" + ); +} + +#[tokio::test] +async fn sse_create_tool_without_project_errors_when_multiple_projects_registered() { + let (state, _mock_alpha, _mock_beta) = make_two_http_project_state().await; + let app = poem::Route::new() + .at("/mcp", poem::post(gateway_mcp_post_handler)) + .data(state); + let cli = poem::test::TestClient::new(app); + + let rpc_body = serde_json::to_vec(&serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "tools/call", + "params": { + "name": "create_bug", + "arguments": { "title": "x", "description": "y" }, + "_meta": { "progressToken": "tok1" } + } + })) + .unwrap(); + let resp = cli + .post("/mcp") + .header("content-type", "application/json") + .header("accept", "text/event-stream") + .body(rpc_body) + .send() + .await; + let body = resp.0.into_body().into_string().await.unwrap(); + let lines = sse_data_lines(&body); + assert_eq!( + lines.len(), + 1, + "expected exactly one SSE error event: {body}" + ); + let ev: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert!( + ev["error"]["message"] + .as_str() + .unwrap() + .contains("requires an explicit `project`"), + "SSE create call omitting `project` with >1 project registered must \ + error, not silently proxy to the active project: {ev}" + ); +} + /// Non-SSE `tools/call` requests must be routed over the live sled-uplink WS /// (story 1180: the gateway no longer falls back to HTTP for MCP proxying) /// and return a plain `application/json` body. diff --git a/server/src/http/gateway/mcp.rs b/server/src/http/gateway/mcp.rs index 840fdf4f..8b4513ba 100644 --- a/server/src/http/gateway/mcp.rs +++ b/server/src/http/gateway/mcp.rs @@ -295,7 +295,10 @@ pub async fn gateway_mcp_post_handler( // SSE proxy: tools/call with Accept: text/event-stream + progressToken for // non-gateway tools is forwarded to the sled's SSE endpoint so progress - // notifications flow through to the gateway client unchanged. + // notifications flow through to the gateway client unchanged. This must + // apply the same explicit-project resolution and create-guard as the + // buffered `tools/call` path below (story 1229: they had drifted apart, + // silently bypassing both on the SSE path). if rpc.method == "tools/call" { let accepts_sse = req .header("accept") @@ -313,7 +316,15 @@ pub async fn gateway_mcp_post_handler( .and_then(|v| v.as_str()) .unwrap_or(""); if !GATEWAY_TOOLS.contains(&tool_name) { - return proxy_and_respond_sse(&state, &bytes, rpc.id).await; + return match extract_explicit_project(&rpc.params) { + Some(project) => { + proxy_and_respond_sse_for_project(&state, project, &bytes, rpc.id).await + } + None => match create_guard_error(&state, tool_name).await { + Some(msg) => sse_error_response(rpc.id, -32602, msg), + None => proxy_and_respond_sse(&state, &bytes, rpc.id).await, + }, + }; } } } @@ -340,35 +351,20 @@ pub async fn gateway_mcp_post_handler( // Story 1208 AC 1: an explicit `project` argument on any // proxied tool call targets that project directly, without // requiring a prior `switch_project`. - let explicit_project = rpc - .params - .get("arguments") - .and_then(|a| a.get("project")) - .and_then(|v| v.as_str()) - .filter(|p| !p.is_empty()); - match explicit_project { + match extract_explicit_project(&rpc.params) { Some(project) => { proxy_and_respond_for_project(&state, project, &bytes, rpc.id).await } - None if is_create_tool(tool_name) => { + None => match create_guard_error(&state, tool_name).await { // Story 1225 AC 3: with >1 project registered, a // create call omitting `project` is ambiguous — fail // loudly instead of silently filing into whichever // project happens to be active. - let project_count = state.projects.read().await.len(); - if project_count > 1 { - to_json_response(JsonRpcResponse::error( - rpc.id, - -32602, - format!( - "'{tool_name}' requires an explicit `project` argument when more than one project is registered (see list_projects) — the active project is not used implicitly for creates." - ), - )) - } else { + Some(msg) => to_json_response(JsonRpcResponse::error(rpc.id, -32602, msg)), + None => { proxy_and_respond_with_resolved_project(&state, &bytes, rpc.id).await } - } - None => proxy_and_respond_with_resolved_project(&state, &bytes, rpc.id).await, + }, } } } @@ -376,6 +372,36 @@ pub async fn gateway_mcp_post_handler( } } +/// Pull the optional per-call `project` argument out of a `tools/call` +/// request's params (story 1208 AC 1), shared by the SSE and buffered +/// `tools/call` paths so both resolve routing identically (story 1229). +fn extract_explicit_project(params: &Value) -> Option<&str> { + params + .get("arguments") + .and_then(|a| a.get("project")) + .and_then(|v| v.as_str()) + .filter(|p| !p.is_empty()) +} + +/// Returns an error message when `tool_name` is a create-tool called without +/// an explicit `project` while more than one project is registered (story +/// 1225 AC 3), or `None` when the call may proceed against the active +/// project. Shared by the SSE and buffered `tools/call` paths so a fix to one +/// can't silently miss the other, as happened in story 1229. +async fn create_guard_error(state: &GatewayState, tool_name: &str) -> Option { + if !is_create_tool(tool_name) { + return None; + } + let project_count = state.projects.read().await.len(); + if project_count > 1 { + Some(format!( + "'{tool_name}' requires an explicit `project` argument when more than one project is registered (see list_projects) — the active project is not used implicitly for creates." + )) + } else { + None + } +} + /// Proxy a request to the active project and format the response. /// /// Prefers the live sled-uplink WebSocket when one is attached (story 899 @@ -473,14 +499,43 @@ async fn proxy_and_respond_for_project( /// /// On sled disconnect mid-stream a JSON-RPC error event is emitted so the /// client does not hang forever. -#[allow(clippy::string_slice)] // pos from buf.find('\n'); '\n' is ASCII so pos and pos+1 are valid boundaries async fn proxy_and_respond_sse(state: &GatewayState, bytes: &[u8], id: Option) -> Response { let url = match state.active_url().await { Ok(u) => u, Err(e) => return sse_error_response(id, -32603, e.to_string()), }; + stream_mcp_call_sse(state, &url, bytes, id).await +} - let resp = match gateway::io::proxy_mcp_call_sse(&state.client, &url, bytes).await { +/// Stream an MCP tool call via SSE to an explicitly named project (story +/// 1229), rather than always targeting the active project — the SSE +/// counterpart of `proxy_and_respond_for_project`. +async fn proxy_and_respond_sse_for_project( + state: &GatewayState, + project: &str, + bytes: &[u8], + id: Option, +) -> Response { + let url = match state.url_for_project(project).await { + Ok(u) => u, + Err(e) => return sse_error_response(id, -32602, e.to_string()), + }; + stream_mcp_call_sse(state, &url, bytes, id).await +} + +/// Shared SSE streaming body for `proxy_and_respond_sse` and +/// `proxy_and_respond_sse_for_project` — proxies to `url` and re-emits each +/// `data:` event from the sled to the originating gateway client without +/// buffering. On sled disconnect mid-stream a JSON-RPC error event is emitted +/// so the client does not hang forever. +#[allow(clippy::string_slice)] // pos from buf.find('\n'); '\n' is ASCII so pos and pos+1 are valid boundaries +async fn stream_mcp_call_sse( + state: &GatewayState, + url: &str, + bytes: &[u8], + id: Option, +) -> Response { + let resp = match gateway::io::proxy_mcp_call_sse(&state.client, url, bytes).await { Ok(r) => r, Err(e) => return sse_error_response(id, -32603, format!("proxy error: {e}")), }; diff --git a/server/src/service/gateway/mod.rs b/server/src/service/gateway/mod.rs index 9dab1da9..9b2c509f 100644 --- a/server/src/service/gateway/mod.rs +++ b/server/src/service/gateway/mod.rs @@ -361,6 +361,28 @@ impl GatewayState { }) } + /// Get the URL of an explicitly named project (story 1229), without + /// reading or mutating `active_project`. + /// + /// Returns `Err` when the project is unknown or has no URL configured + /// (WS-uplink only) — mirrors [`GatewayState::active_url`]'s error shape + /// so callers can format both the same way. + pub async fn url_for_project(&self, project: &str) -> Result { + self.projects + .read() + .await + .get(project) + .ok_or_else(|| Error::ProjectNotFound(format!("unknown project '{project}'")))? + .url + .clone() + .ok_or_else(|| { + Error::ProjectNotFound(format!( + "project '{project}' has no URL configured \ + (use sled-uplink WS or add url to projects.toml)" + )) + }) + } + /// Register a live sled connection for the given project. pub async fn register_sled_connection(&self, project_name: String, conn: SledConnection) { self.sled_connections