feat(gateway/v1): real SSE streaming on the openrouter lane; honest 400 elsewhere
lakehouse/auditor 13 blocking issues: cloud: claim not backed — "pretending to have streamed. Verified live: chat.completion.chunk frames from"
lakehouse/auditor 13 blocking issues: cloud: claim not backed — "pretending to have streamed. Verified live: chat.completion.chunk frames from"
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) <noreply@anthropic.com>
This commit is contained in:
@@ -208,9 +208,8 @@ pub struct ChatRequest {
|
|||||||
pub temperature: Option<f64>,
|
pub temperature: Option<f64>,
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub max_tokens: Option<u32>,
|
pub max_tokens: Option<u32>,
|
||||||
/// Accepted for shape-compat but ignored in the thin slice —
|
/// Real SSE passthrough on the openrouter lane (2026-09-20); other lanes reject
|
||||||
/// Phase 38 returns non-streaming even when the client asked for it.
|
/// stream=true with a 400 rather than silently returning a non-streamed body.
|
||||||
/// Phase 39+ wires real streaming.
|
|
||||||
#[serde(default)]
|
#[serde(default)]
|
||||||
pub stream: Option<bool>,
|
pub stream: Option<bool>,
|
||||||
/// Non-OpenAI extension. Passes through to the provider's thinking
|
/// 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]
|
#[test]
|
||||||
fn uses_tools_detects_definitions_and_history() {
|
fn uses_tools_detects_definitions_and_history() {
|
||||||
let mut r = mk_req(None, "openrouter/z-ai/glm-5.3-flash");
|
let mut r = mk_req(None, "openrouter/z-ai/glm-5.3-flash");
|
||||||
@@ -425,14 +443,11 @@ async fn chat(
|
|||||||
State(state): State<V1State>,
|
State(state): State<V1State>,
|
||||||
headers: axum::http::HeaderMap,
|
headers: axum::http::HeaderMap,
|
||||||
Json(req): Json<ChatRequest>,
|
Json(req): Json<ChatRequest>,
|
||||||
) -> Result<Json<ChatResponse>, (StatusCode, String)> {
|
) -> Result<axum::response::Response, (StatusCode, String)> {
|
||||||
|
use axum::response::IntoResponse as _;
|
||||||
if req.messages.is_empty() {
|
if req.messages.is_empty() {
|
||||||
return Err((StatusCode::BAD_REQUEST, "messages must be non-empty".into()));
|
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
|
// Provider resolution: explicit `req.provider` wins; otherwise
|
||||||
// infer from a model-name prefix. Phase 39 PRD gate example:
|
// infer from a model-name prefix. Phase 39 PRD gate example:
|
||||||
// `model: "openrouter/openai/gpt-4o-mini"` → provider "openrouter",
|
// `model: "openrouter/openai/gpt-4o-mini"` → provider "openrouter",
|
||||||
@@ -464,6 +479,45 @@ async fn chat(
|
|||||||
std::borrow::Cow::Owned(cloned)
|
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() {
|
let (resp, used_provider) = match provider.as_str() {
|
||||||
"ollama" | "local" | "" => {
|
"ollama" | "local" | "" => {
|
||||||
let r = ollama::chat(&state.ai_client, &*req_for_adapter)
|
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;
|
provider_usage.total_tokens += resp.usage.total_tokens as u64;
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(Json(resp))
|
Ok(Json(resp).into_response())
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn usage(State(state): State<V1State>) -> impl IntoResponse {
|
async fn usage(State(state): State<V1State>) -> impl IntoResponse {
|
||||||
|
|||||||
@@ -61,6 +61,52 @@ pub fn resolve_openrouter_key() -> Option<String> {
|
|||||||
None
|
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<reqwest::Response, String> {
|
||||||
|
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::<String>()));
|
||||||
|
}
|
||||||
|
Ok(resp)
|
||||||
|
}
|
||||||
|
|
||||||
pub async fn chat(
|
pub async fn chat(
|
||||||
key: &str,
|
key: &str,
|
||||||
req: &ChatRequest,
|
req: &ChatRequest,
|
||||||
|
|||||||
Reference in New Issue
Block a user