diff --git a/apps/maple-agent/Cargo.lock b/apps/maple-agent/Cargo.lock index c54e046eb..d139f141f 100644 --- a/apps/maple-agent/Cargo.lock +++ b/apps/maple-agent/Cargo.lock @@ -5254,7 +5254,7 @@ dependencies = [ [[package]] name = "maple-proxy" -version = "0.4.3" +version = "0.4.4" dependencies = [ "anyhow", "async-stream", diff --git a/apps/maple-research/frontend/src-tauri/Cargo.lock b/apps/maple-research/frontend/src-tauri/Cargo.lock index 496fda66d..926ab2049 100644 --- a/apps/maple-research/frontend/src-tauri/Cargo.lock +++ b/apps/maple-research/frontend/src-tauri/Cargo.lock @@ -4754,7 +4754,7 @@ dependencies = [ [[package]] name = "maple-proxy" -version = "0.4.3" +version = "0.4.4" dependencies = [ "anyhow", "async-stream", diff --git a/proxy/.env.example b/proxy/.env.example index 298251b00..496ee6bb3 100644 --- a/proxy/.env.example +++ b/proxy/.env.example @@ -33,7 +33,7 @@ MAPLE_ENABLE_CORS=false # Backend request setup and non-streaming response timeout, in seconds MAPLE_REQUEST_TIMEOUT_SECS=300 # Maximum idle time between streaming chunks, in seconds -MAPLE_STREAM_IDLE_TIMEOUT_SECS=300 +MAPLE_STREAM_IDLE_TIMEOUT_SECS=360 # Rust Logging (optional) # RUST_LOG=info,maple_proxy=debug diff --git a/proxy/Cargo.lock b/proxy/Cargo.lock index f8abcc58e..ad6286a8b 100644 --- a/proxy/Cargo.lock +++ b/proxy/Cargo.lock @@ -1419,7 +1419,7 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "maple-proxy" -version = "0.4.3" +version = "0.4.4" dependencies = [ "anyhow", "async-stream", diff --git a/proxy/Cargo.toml b/proxy/Cargo.toml index d977133d6..47921e635 100644 --- a/proxy/Cargo.toml +++ b/proxy/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "maple-proxy" -version = "0.4.3" +version = "0.4.4" edition = "2021" authors = ["OpenSecret"] description = "Lightweight OpenAI-compatible proxy server for Maple/OpenSecret TEE infrastructure" diff --git a/proxy/Dockerfile b/proxy/Dockerfile index f2a5e9954..2666e5af9 100644 --- a/proxy/Dockerfile +++ b/proxy/Dockerfile @@ -51,7 +51,7 @@ ENV MAPLE_HOST=0.0.0.0 \ MAPLE_DEBUG=false \ MAPLE_ENABLE_CORS=true \ MAPLE_REQUEST_TIMEOUT_SECS=300 \ - MAPLE_STREAM_IDLE_TIMEOUT_SECS=300 \ + MAPLE_STREAM_IDLE_TIMEOUT_SECS=360 \ RUST_LOG=info # Expose the port diff --git a/proxy/README.md b/proxy/README.md index 48c9a74d8..33be79540 100644 --- a/proxy/README.md +++ b/proxy/README.md @@ -54,7 +54,7 @@ maple-proxy = "0.3.2" Crates.io publishing remains separate from Maple application releases; the example above uses the latest published crate version. -This checkout prepares proxy `0.4.3`, including `POST /v1/systemone` introduced +This checkout prepares proxy `0.4.4`, including `POST /v1/systemone` introduced in `0.4.2`, with a compatible registry requirement for `maple-sdk` starting at `4.0.1`. The published proxy crate in the example above predates this Transport V2 update. The standalone binary's lockfile selects its @@ -79,10 +79,18 @@ export MAPLE_CACHE_NAMESPACE_ROOT="$(openssl rand -base64 32)" # Generate once; export MAPLE_DEBUG=true # Enable debug logging export MAPLE_ENABLE_CORS=false # Default; see browser warning below export MAPLE_REQUEST_TIMEOUT_SECS=300 # Backend request timeout -export MAPLE_STREAM_IDLE_TIMEOUT_SECS=300 # Streaming idle timeout between chunks +export MAPLE_STREAM_IDLE_TIMEOUT_SECS=360 # Streaming idle timeout between chunks ``` -Both timeout defaults are five minutes. `MAPLE_REQUEST_TIMEOUT_SECS` covers +The request timeout defaults to five minutes; the streaming idle timeout defaults +to six minutes. OpenSecret allows five minutes between provider chunks for all +chat models, including pauses while a provider buffers tool arguments. The extra +minute lets the backend return its structured timeout error before the proxy +closes the stream. Existing installations must upgrade the proxy or increase +`MAPLE_STREAM_IDLE_TIMEOUT_SECS` to adopt this margin. Explicit overrides remain +supported; choose a proxy idle timeout longer than the backend provider idle limit. + +`MAPLE_REQUEST_TIMEOUT_SECS` covers sending a request through completion of a non-streaming response body. For SSE, it covers response setup; once the stream starts, `MAPLE_STREAM_IDLE_TIMEOUT_SECS` limits the wait for each next chunk without imposing a total stream duration. @@ -467,7 +475,7 @@ docker pull ghcr.io/mapleprivacylabs/maple-proxy:latest docker run -p 8080:8080 \ -e MAPLE_BACKEND_URL=https://enclave.trymaple.ai \ -e MAPLE_REQUEST_TIMEOUT_SECS=300 \ - -e MAPLE_STREAM_IDLE_TIMEOUT_SECS=300 \ + -e MAPLE_STREAM_IDLE_TIMEOUT_SECS=360 \ ghcr.io/mapleprivacylabs/maple-proxy:latest ``` @@ -565,7 +573,7 @@ environment: - MAPLE_BACKEND_URL=https://enclave.trymaple.ai # Production backend - MAPLE_ENABLE_CORS=true # Enable for web apps - MAPLE_REQUEST_TIMEOUT_SECS=300 # Backend request timeout - - MAPLE_STREAM_IDLE_TIMEOUT_SECS=300 # Streaming idle timeout + - MAPLE_STREAM_IDLE_TIMEOUT_SECS=360 # Streaming idle timeout - RUST_LOG=info # Logging level # - MAPLE_API_KEY=xxx # Only for private deployments! # - MAPLE_CACHE_NAMESPACE_ROOT=xxx # Stable secret for cache continuity diff --git a/proxy/docker-compose.yml b/proxy/docker-compose.yml index 8e49a0d1a..fe24bc95e 100644 --- a/proxy/docker-compose.yml +++ b/proxy/docker-compose.yml @@ -33,7 +33,7 @@ services: - MAPLE_DEBUG=${MAPLE_DEBUG:-false} - MAPLE_ENABLE_CORS=${MAPLE_ENABLE_CORS:-true} - MAPLE_REQUEST_TIMEOUT_SECS=${MAPLE_REQUEST_TIMEOUT_SECS:-300} - - MAPLE_STREAM_IDLE_TIMEOUT_SECS=${MAPLE_STREAM_IDLE_TIMEOUT_SECS:-300} + - MAPLE_STREAM_IDLE_TIMEOUT_SECS=${MAPLE_STREAM_IDLE_TIMEOUT_SECS:-360} - RUST_LOG=${RUST_LOG:-info} # Production-grade restart policy diff --git a/proxy/src/config.rs b/proxy/src/config.rs index d086c4220..94141de06 100644 --- a/proxy/src/config.rs +++ b/proxy/src/config.rs @@ -4,7 +4,8 @@ use serde::Serialize; use std::{net::SocketAddr, time::Duration}; pub const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 300; -pub const DEFAULT_STREAM_IDLE_TIMEOUT_SECS: u64 = 300; +// Leave room for the backend's 300s provider-idle error to reach the client. +pub const DEFAULT_STREAM_IDLE_TIMEOUT_SECS: u64 = 360; #[derive(Parser, Debug, Clone)] #[command(name = "maple-proxy")] diff --git a/proxy/src/proxy.rs b/proxy/src/proxy.rs index 9ec10de7c..531f8d912 100644 --- a/proxy/src/proxy.rs +++ b/proxy/src/proxy.rs @@ -854,7 +854,7 @@ mod tests { let mut transport = MockTransport::new(vec![Ok(upstream)]); transport.start_delay = Duration::from_millis(600); let transport = Arc::new(transport); - // Short test-only durations on a paused clock; production defaults stay 300s. + // Short test-only durations on a paused clock; production defaults are unchanged. let config = test_config() .with_api_key("fixture-key".into()) .with_request_timeout_secs(1) @@ -953,6 +953,59 @@ mod tests { assert!(stream.next().await.is_none()); } + #[tokio::test(start_paused = true)] + async fn default_stream_idle_budget_preserves_backend_timeout_error() { + let config = Config::new("127.0.0.1".into(), 8080, "http://localhost:3000".into()); + let body: OpenSecretResponseBody = Box::pin(async_stream::stream! { + yield Ok(Bytes::from_static(b"data: {\"choices\":[]}\n\n")); + // Backend provider reads may now wait five minutes before reporting failure. + tokio::time::sleep(Duration::from_secs(300)).await; + yield Ok(Bytes::from_static(b"data: {\"error\":{\"code\":\"upstream_timeout\"}}\n\n")); + }); + let mut upstream = http::Response::new(body); + upstream + .headers_mut() + .insert(header::CONTENT_TYPE, "text/event-stream".parse().unwrap()); + let started = Instant::now(); + let response = build_downstream_response( + upstream, + config.stream_idle_timeout(), + started + config.request_timeout(), + ); + let mut stream = response.into_body().into_data_stream(); + assert_eq!( + stream.next().await.unwrap().unwrap(), + "data: {\"choices\":[]}\n\n" + ); + assert_eq!( + stream.next().await.unwrap().unwrap(), + "data: {\"error\":{\"code\":\"upstream_timeout\"}}\n\n" + ); + assert_eq!(started.elapsed(), Duration::from_secs(300)); + assert!(stream.next().await.is_none()); + } + + #[tokio::test(start_paused = true)] + async fn default_stream_idle_budget_still_drops_silent_upstream() { + let config = Config::new("127.0.0.1".into(), 8080, "http://localhost:3000".into()); + let dropped = Arc::new(AtomicBool::new(false)); + let guard = DropMarker(Arc::clone(&dropped)); + let body: OpenSecretResponseBody = Box::pin(async_stream::stream! { + let _guard = guard; + std::future::pending::<()>().await; + yield Ok(Bytes::new()); + }); + let started = Instant::now(); + let mut stream = stream_with_timeouts(body, config.stream_idle_timeout(), None); + assert_eq!( + stream.next().await.unwrap().unwrap_err().kind(), + io::ErrorKind::TimedOut + ); + assert_eq!(started.elapsed(), Duration::from_secs(360)); + assert!(dropped.load(Ordering::SeqCst)); + assert!(stream.next().await.is_none()); + } + #[tokio::test(start_paused = true)] async fn dropping_downstream_body_releases_pending_upstream() { let dropped = Arc::new(AtomicBool::new(false)); diff --git a/services/opensecret/src/web/openai.rs b/services/opensecret/src/web/openai.rs index a804baef0..dc0c2fbab 100644 --- a/services/opensecret/src/web/openai.rs +++ b/services/opensecret/src/web/openai.rs @@ -69,8 +69,11 @@ use tracing::{debug, error, info, warn}; use uuid::Uuid; // Timeout constants for provider requests -const REQUEST_TIMEOUT_SECS: u64 = 120; // Request timeout (generous for large non-streaming responses) -const STREAM_CHUNK_TIMEOUT_SECS: u64 = 120; // Per-chunk timeout for streaming reads +// Request timeout (generous for large non-streaming responses). +const REQUEST_TIMEOUT_SECS: u64 = 120; +// Providers may buffer a complete tool call before emitting its arguments. +// This is an idle read limit for every provider/model, not a total generation deadline. +const STREAM_CHUNK_TIMEOUT_SECS: u64 = 300; const TTS_BILLING_CHECK_TIMEOUT: Duration = Duration::from_secs(5); const TTS_PROVIDER_TIMEOUT: Duration = Duration::from_secs(120); const MAX_TTS_INPUT_CHARS: usize = 100_000; @@ -5975,6 +5978,79 @@ mod tests { assert_eq!(decrypted, plaintext.as_bytes()); } + #[tokio::test] + async fn mock_provider_buffered_tool_calls_reset_stream_idle_timeout() { + let app = Router::new().route( + "/v1/chat/completions", + post(|| async { + let body = async_stream::stream! { + yield Ok::<_, std::io::Error>(bytes::Bytes::from_static( + b"data: {\"choices\":[{\"delta\":{\"reasoning\":\"planning\"}}]}\n\n", + )); + for index in 0..2 { + // Scale one production second to one millisecond: each buffered + // call exceeds the old 120s limit, but fits the new 300s limit. + tokio::time::sleep(Duration::from_millis(150)).await; + let frame = json!({"choices": [{"delta": {"tool_calls": [{ + "index": index, "type": "function", + "function": {"name": "write_file", "arguments": "{\"content\":\"complete\"}"} + }]}}]}); + yield Ok(bytes::Bytes::from(format!("data: {frame}\n\n"))); + } + tokio::time::sleep(Duration::from_millis(150)).await; + yield Ok(bytes::Bytes::from_static(b"data: [DONE]\n\n")); + }; + axum::http::Response::builder() + .header(header::CONTENT_TYPE, "text/event-stream") + .body(axum::body::Body::from_stream(body)) + .expect("mock buffered tool stream") + }), + ); + let (trace, server) = call_mock_provider(app).await; + let response = trace.result.expect("provider response start"); + let pinned = pinned_test_completion(); + let attempt = pinned + .begin_execution() + .begin_attempt(pinned.route.identity()); + let (tx, mut rx) = mpsc::channel(8); + let started = tokio::time::Instant::now(); + let result = process_completion_stream( + response, + "fixture-model", + attempt, + &tx, + Duration::from_millis(STREAM_CHUNK_TIMEOUT_SECS), + None, + false, + ) + .await; + drop(tx); + server.abort(); + + assert!(matches!( + result.terminal, + AttemptTerminal::Completed { + evidence: CompletionEvidence::ProviderDone, + .. + } + )); + // Timely chunks renew the idle budget even after total runtime exceeds it. + assert!(started.elapsed() > Duration::from_millis(STREAM_CHUNK_TIMEOUT_SECS)); + let mut chunks = Vec::new(); + while let Some(chunk) = rx.recv().await { + if let CompletionChunk::StreamChunk(json) = chunk { + chunks.push(json); + } + } + assert_eq!(chunks.len(), 3); + assert_eq!(chunks[0]["choices"][0]["delta"]["reasoning"], "planning"); + for (index, chunk) in chunks[1..].iter().enumerate() { + let call = &chunk["choices"][0]["delta"]["tool_calls"][0]; + assert_eq!(call["index"], index); + assert_eq!(call["function"]["arguments"], "{\"content\":\"complete\"}"); + } + } + #[tokio::test] async fn mock_provider_pending_stream_has_one_timeout_terminal_result() { let app = Router::new().route(