From cbc24a4b1960981c1ba8e14227af430674159590 Mon Sep 17 00:00:00 2001 From: Claw Date: Sun, 20 Sep 2026 07:33:47 -0500 Subject: [PATCH] feat(gateway/v1): real SSE streaming on the openrouter lane; honest 400 elsewhere MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The gateway accepted `stream: true` and returned a single JSON body — the code said 'Phase 39+ wires real streaming' and it never happened. A streaming client waits for SSE frames that never arrive and reports 'no response': that is what broke openclaw's gateway lane today and cost six orchestrator rounds before the cause was found. Now: openrouter/openrouter_free forward the upstream SSE byte-for-byte; every other lane rejects stream=true with a 400 that names the lane, instead of pretending to have streamed. Verified live: chat.completion.chunk frames from :3100, and the non-streaming path unchanged. Streamed calls skip usage metering and Langfuse (stated in the code and logged per call) — metering needs the final usage chunk and belongs in its own change. Co-Authored-By: Claude Opus 5 (1M context) --- crates/gateway/src/v1/mod.rs | 72 +++++++++++++++++++++++++---- crates/gateway/src/v1/openrouter.rs | 46 ++++++++++++++++++ 2 files changed, 109 insertions(+), 9 deletions(-) diff --git a/crates/gateway/src/v1/mod.rs b/crates/gateway/src/v1/mod.rs index cdf264c..b604587 100644 --- a/crates/gateway/src/v1/mod.rs +++ b/crates/gateway/src/v1/mod.rs @@ -208,9 +208,8 @@ pub struct ChatRequest { pub temperature: Option, #[serde(default)] pub max_tokens: Option, - /// Accepted for shape-compat but ignored in the thin slice — - /// Phase 38 returns non-streaming even when the client asked for it. - /// Phase 39+ wires real streaming. + /// Real SSE passthrough on the openrouter lane (2026-09-20); other lanes reject + /// stream=true with a 400 rather than silently returning a non-streamed body. #[serde(default)] pub stream: Option, /// Non-OpenAI extension. Passes through to the provider's thinking @@ -352,6 +351,25 @@ mod resolve_provider_tests { } } + /// Streaming is a per-lane capability, and the routing decision must be visible in a test: + /// before 2026-09-20 the gateway accepted stream=true on every lane and returned a single + /// JSON body, so a client waiting for SSE frames hung until its own timeout. + #[test] + fn streaming_is_openrouter_only() { + let mut r = mk_req(None, "openrouter/z-ai/glm-5.3-flash"); + r.stream = Some(true); + assert_eq!(resolve_provider(&r).0, "openrouter", "openrouter lane streams"); + + let mut local = mk_req(None, "qwen3.5:latest"); + local.stream = Some(true); + assert_eq!( + resolve_provider(&local).0, + "ollama", + "a local model resolves to ollama, where the handler rejects stream=true with a 400 \ + instead of silently returning a non-streamed body" + ); + } + #[test] fn uses_tools_detects_definitions_and_history() { let mut r = mk_req(None, "openrouter/z-ai/glm-5.3-flash"); @@ -425,14 +443,11 @@ async fn chat( State(state): State, headers: axum::http::HeaderMap, Json(req): Json, -) -> Result, (StatusCode, String)> { +) -> Result { + use axum::response::IntoResponse as _; if req.messages.is_empty() { return Err((StatusCode::BAD_REQUEST, "messages must be non-empty".into())); } - if req.stream.unwrap_or(false) { - tracing::warn!("/v1/chat: stream=true requested but Phase 38 returns non-streaming"); - } - // Provider resolution: explicit `req.provider` wins; otherwise // infer from a model-name prefix. Phase 39 PRD gate example: // `model: "openrouter/openai/gpt-4o-mini"` → provider "openrouter", @@ -464,6 +479,45 @@ async fn chat( std::borrow::Cow::Owned(cloned) }; + // STREAMING (2026-09-20). Phase 38 accepted `stream: true` and returned a single JSON body; + // a client waiting for SSE frames hangs and then reports "no response". Forward the upstream + // SSE verbatim on the lane that supports it, and say so plainly on the lanes that do not + // instead of pretending. Streamed calls skip usage metering + Langfuse (see chat_stream). + if req.stream.unwrap_or(false) { + if !matches!(provider.as_str(), "openrouter" | "openrouter_free") { + return Err(( + StatusCode::BAD_REQUEST, + format!( + "stream=true is only supported on the openrouter lane via this gateway \ + (resolved provider: {provider}); resend with stream=false or use an \ + OpenRouter model id" + ), + )); + } + let key = state.openrouter_key.as_deref().ok_or(( + StatusCode::SERVICE_UNAVAILABLE, + "OPENROUTER_API_KEY not configured".to_string(), + ))?; + let upstream = openrouter::chat_stream(key, &*req_for_adapter) + .await + .map_err(|e| (StatusCode::BAD_GATEWAY, format!("openrouter stream: {e}")))?; + tracing::info!( + target: "v1.chat", + provider = %provider, + model = %effective_model, + "streaming passthrough — usage metering and Langfuse are skipped for this call", + ); + let body = axum::body::Body::from_stream(upstream.bytes_stream()); + return Ok(( + [ + (axum::http::header::CONTENT_TYPE, "text/event-stream"), + (axum::http::header::CACHE_CONTROL, "no-cache"), + ], + body, + ) + .into_response()); + } + let (resp, used_provider) = match provider.as_str() { "ollama" | "local" | "" => { let r = ollama::chat(&state.ai_client, &*req_for_adapter) @@ -647,7 +701,7 @@ async fn chat( provider_usage.total_tokens += resp.usage.total_tokens as u64; } - Ok(Json(resp)) + Ok(Json(resp).into_response()) } async fn usage(State(state): State) -> impl IntoResponse { diff --git a/crates/gateway/src/v1/openrouter.rs b/crates/gateway/src/v1/openrouter.rs index 8eca184..dd490d7 100644 --- a/crates/gateway/src/v1/openrouter.rs +++ b/crates/gateway/src/v1/openrouter.rs @@ -61,6 +61,52 @@ pub fn resolve_openrouter_key() -> Option { None } +/// Streaming passthrough (2026-09-20). The gateway used to accept `stream: true` and return a +/// single non-streamed JSON body — a client waiting for SSE frames hangs until its own timeout +/// and reports "no response" (this is what broke openclaw's gateway lane). We now forward the +/// upstream SSE byte-for-byte. +/// +/// The caller streams the body straight to its client, so nothing here parses the chunks: +/// usage metering and Langfuse tracing are SKIPPED for streamed calls, and the handler logs +/// that. Metering streamed responses needs the final usage chunk and belongs in its own change. +pub async fn chat_stream(key: &str, req: &ChatRequest) -> Result { + let model = req.model.strip_prefix("openrouter/").unwrap_or(&req.model).to_string(); + let body = ORChatBody { + model, + messages: req.messages.iter().map(|m| ORMessage { + role: m.role.clone(), + content: m.content.clone(), + tool_calls: m.tool_calls.clone(), + tool_call_id: m.tool_call_id.clone(), + name: m.name.clone(), + }).collect(), + tools: req.tools.clone(), + tool_choice: req.tool_choice.clone(), + max_tokens: Some(req.max_tokens.unwrap_or(800)), + temperature: req.temperature.unwrap_or(0.3), + stream: true, + }; + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(or_timeout_secs())) + .build() + .map_err(|e| format!("build client: {e}"))?; + let resp = client + .post(format!("{}/chat/completions", OR_BASE_URL)) + .bearer_auth(key) + .header("HTTP-Referer", "https://vcp.devop.live") + .header("X-Title", "Lakehouse Scrum") + .json(&serde_json::to_value(&body).map_err(|e| format!("serialize body: {e}"))?) + .send() + .await + .map_err(|e| format!("openrouter.ai unreachable: {e}"))?; + if !resp.status().is_success() { + let code = resp.status(); + let text = resp.text().await.unwrap_or_else(|_| "?".into()); + return Err(format!("openrouter {code}: {}", text.chars().take(300).collect::())); + } + Ok(resp) +} + pub async fn chat( key: &str, req: &ChatRequest,