diff --git a/RUN.md b/RUN.md index b523302..a846341 100644 --- a/RUN.md +++ b/RUN.md @@ -934,6 +934,16 @@ cargo test -p ardur-fused-runtime --test memory_recall spend. The cost-gate enforces this server-side and returns a structured error to the channel before the next provider call when the ceiling is hit. +## Turn timeout + +`ARDUR_HTTP_TURN_TIMEOUT_SECS=30` (default `30`) bounds how long the +synchronous `POST /chat` and ACP HTTP handlers wait on a single turn before +returning `504 Gateway Timeout`. Raise it for workloads with long tool loops +or slow providers. The value must be a positive integer number of seconds; a +`0` or unparseable value is rejected at boot. When the wait elapses the turn is +cancelled before it commits a receipt or bills cost, so a `504` never pairs +with a silently-billed turn the client could not observe. + ## Troubleshooting **Events not arriving.** diff --git a/crates/e2e-tests/tests/scenario_mcp_server.rs b/crates/e2e-tests/tests/scenario_mcp_server.rs index e5777c3..92b3cd3 100644 --- a/crates/e2e-tests/tests/scenario_mcp_server.rs +++ b/crates/e2e-tests/tests/scenario_mcp_server.rs @@ -42,6 +42,7 @@ async fn spawn_server(bearer: &str) -> String { enable_http_tool: false, http_allowlist: Vec::new(), file_tool_root: None, + http_turn_timeout: std::time::Duration::from_secs(30), slack_enabled: true, slack_bot_token: Some("xoxb-e2e".to_string()), slack_signing_secret: Some("e2e-signing-secret-0000000000".to_string()), diff --git a/crates/e2e-tests/tests/scenario_server_slack_round_trip.rs b/crates/e2e-tests/tests/scenario_server_slack_round_trip.rs index e9182ca..07bdaff 100644 --- a/crates/e2e-tests/tests/scenario_server_slack_round_trip.rs +++ b/crates/e2e-tests/tests/scenario_server_slack_round_trip.rs @@ -89,6 +89,7 @@ async fn server_routes_signed_slack_message_through_runtime_to_chat_post_message enable_http_tool: false, http_allowlist: Vec::new(), file_tool_root: None, + http_turn_timeout: std::time::Duration::from_secs(30), slack_enabled: true, slack_bot_token: Some(BOT_TOKEN.to_string()), slack_signing_secret: Some(SIGNING_SECRET.to_string()), diff --git a/crates/e2e-tests/tests/scenario_streaming_chat_sse.rs b/crates/e2e-tests/tests/scenario_streaming_chat_sse.rs index 98a7751..0421302 100644 --- a/crates/e2e-tests/tests/scenario_streaming_chat_sse.rs +++ b/crates/e2e-tests/tests/scenario_streaming_chat_sse.rs @@ -38,6 +38,7 @@ fn test_config(data_dir: &tempfile::TempDir) -> Config { enable_http_tool: false, http_allowlist: Vec::new(), file_tool_root: None, + http_turn_timeout: std::time::Duration::from_secs(30), slack_enabled: true, slack_bot_token: Some("redacted-test-token".to_string()), slack_signing_secret: Some("streaming-e2e-signing-secret".to_string()), diff --git a/crates/server/src/config.rs b/crates/server/src/config.rs index 6428883..c6ecaf9 100644 --- a/crates/server/src/config.rs +++ b/crates/server/src/config.rs @@ -15,6 +15,7 @@ use std::fmt; use std::path::PathBuf; +use std::time::Duration; use ardur_provider_selector::{ProviderKind, SELECTOR_ENV}; use ardur_tool_registry::{BuiltinOpts, HttpFetchOpts}; @@ -176,6 +177,13 @@ pub struct Config { /// `file.list` built-in tools (`ARDUR_FILE_TOOL_ROOT`). `Some(root)` /// registers all three confined to it; `None` registers no file tool. pub file_tool_root: Option, + /// How long a synchronous `POST /chat` (and ACP) turn may run before the + /// HTTP surface stops waiting on it (`ARDUR_HTTP_TURN_TIMEOUT_SECS`, default + /// `30`). When the wait elapses the client receives `504`; the worker + /// observes the dropped reply channel and cancels the in-flight turn before + /// it commits a receipt or bills cost, so the `504` is never paired with a + /// silently-billed turn (issue #359). Must be a positive number of seconds. + pub http_turn_timeout: Duration, } /// A required environment variable was unset or empty. @@ -254,6 +262,7 @@ impl fmt::Debug for Config { .field("enable_http_tool", &self.enable_http_tool) .field("http_allowlist", &self.http_allowlist) .field("file_tool_root", &self.file_tool_root) + .field("http_turn_timeout", &self.http_turn_timeout) .finish() } } @@ -414,6 +423,28 @@ impl Config { let http_allowlist = parse_csv(optional("ARDUR_HTTP_ALLOWLIST").as_deref()); let file_tool_root = optional("ARDUR_FILE_TOOL_ROOT").map(PathBuf::from); + // The synchronous-turn wait ceiling. A slow-but-legitimate turn (a long + // tool loop, a slow provider) should be able to outlast the default 30s + // without the operator having to fork the code — so it is a knob, not a + // constant. A zero or unparseable value is rejected rather than silently + // treated as "no timeout". + let http_turn_timeout = match optional("ARDUR_HTTP_TURN_TIMEOUT_SECS") { + None => Duration::from_secs(30), + Some(raw) => { + let secs = raw.parse::().map_err(|e| ConfigError::Invalid { + var: "ARDUR_HTTP_TURN_TIMEOUT_SECS", + reason: format!("`{raw}` is not a valid u64: {e}"), + })?; + if secs == 0 { + return Err(ConfigError::Invalid { + var: "ARDUR_HTTP_TURN_TIMEOUT_SECS", + reason: "must be a positive number of seconds".to_string(), + }); + } + Duration::from_secs(secs) + } + }; + Ok(Self { anthropic_api_key, slack_enabled, @@ -473,6 +504,7 @@ impl Config { enable_http_tool, http_allowlist, file_tool_root, + http_turn_timeout, }) } diff --git a/crates/server/src/routes.rs b/crates/server/src/routes.rs index 8f5100f..87cb309 100644 --- a/crates/server/src/routes.rs +++ b/crates/server/src/routes.rs @@ -16,7 +16,6 @@ use std::convert::Infallible; use std::fmt::Write as _; use std::sync::Arc; -use std::time::Duration; use axum::Router; use axum::body::{Body, Bytes}; @@ -43,7 +42,6 @@ use crate::state::{ }; const HTTP_BODY_LIMIT_BYTES: usize = 64 * 1024; -const HTTP_TURN_TIMEOUT: Duration = Duration::from_secs(30); /// Build the application router over the shared [`AppState`]. /// @@ -767,7 +765,7 @@ async fn chat(State(state): State>, headers: HeaderMap, body: Byte let session_id = request.session_id.unwrap_or_default(); match tokio::time::timeout( - HTTP_TURN_TIMEOUT, + state.http_turn_timeout(), state.submit_chat(request.message, session_id), ) .await @@ -979,7 +977,7 @@ async fn acp(State(state): State>, headers: HeaderMap, body: Bytes } }; match tokio::time::timeout( - HTTP_TURN_TIMEOUT, + state.http_turn_timeout(), state.submit_chat(prompt, SessionId::new()), ) .await diff --git a/crates/server/src/state.rs b/crates/server/src/state.rs index 6ec0a41..54ac705 100644 --- a/crates/server/src/state.rs +++ b/crates/server/src/state.rs @@ -41,7 +41,7 @@ use std::panic::AssertUnwindSafe; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex, OnceLock}; -use std::time::{SystemTime, UNIX_EPOCH}; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use ardur_cap_token::{ BiscuitCapTokenIssuer, CapScope, CapTokenIssuer, HolderId as CapHolderId, KeyPair, PublicKey, @@ -346,6 +346,11 @@ pub struct AppState { receipt_jwks: ardur_receipt::Jwks, /// Turn-outcome and security-denial counters, shared with the worker. security_metrics: Arc, + /// How long the synchronous `/chat` + ACP handlers wait on a turn before + /// returning `504` (`ARDUR_HTTP_TURN_TIMEOUT_SECS`, default `30s`). The + /// worker cancels the turn when the wait elapses, so the timeout bounds the + /// client's wait without billing for an unobservable turn (issue #359). + http_turn_timeout: Duration, } /// The data [`build_router`](crate::build_router) needs to mount the §6.0 MCP @@ -611,6 +616,7 @@ impl AppState { telegram, receipt_jwks, security_metrics, + http_turn_timeout: config.http_turn_timeout, })) } @@ -632,6 +638,13 @@ impl AppState { self.cost_budget_cents } + /// How long the synchronous `/chat` + ACP handlers wait on a turn before + /// returning `504` (`ARDUR_HTTP_TURN_TIMEOUT_SECS`, default `30s`). + #[must_use] + pub fn http_turn_timeout(&self) -> Duration { + self.http_turn_timeout + } + /// The tool ids minted into session cap-tokens for runtime turns. #[must_use] pub fn tool_allowlist(&self) -> &[String] { @@ -1173,7 +1186,7 @@ impl Processor { let HttpTurn { message, session_id, - reply, + mut reply, } = turn; let token = match self.mint_session_token(now_unix()) { @@ -1200,7 +1213,33 @@ impl Processor { requested_provider: None, }; - let outcome = match self.runtime.submit(request).await { + // Cancel the turn if the HTTP caller goes away before it settles. The + // client-facing turn timeout firing (or the client simply hanging up) + // drops the `submit_chat` future, which drops the oneshot receiver and + // resolves `reply.closed()`. Racing that against the turn — the sync + // mirror of the streaming path's `events.closed()` guard — lets us drop + // the in-flight `submit` future before it commits the receipt, journal, + // and cost side effects (all of which happen at `.await` points inside + // `submit`). Without this, the worker ran the turn to completion and + // minted+billed a receipt the caller was told `504` for and never saw + // (issue #359). `biased` prefers the completion arm so a turn that + // finished right at the deadline still reports its already-committed + // outcome rather than being needlessly discarded. + let submit = self.runtime.submit(request); + tokio::pin!(submit); + let submit_result = tokio::select! { + biased; + result = &mut submit => result, + () = reply.closed() => { + tracing::info!( + session_id = %session_id.0, + "HTTP turn abandoned by caller before completion; cancelling turn (no receipt minted)" + ); + return; + } + }; + + let outcome = match submit_result { Ok(result) => { self.security_metrics.record_ok(); let tools_called = self.tools_called_since(receipts_before); @@ -1944,6 +1983,7 @@ mod tests { telegram: Arc::new(OnceLock::new()), receipt_jwks: ardur_receipt::Jwks::new(), security_metrics: Arc::new(SecurityMetrics::default()), + http_turn_timeout: Duration::from_secs(30), }; assert!(state.worker_alive()); diff --git a/crates/server/tests/chat_turn_timeout.rs b/crates/server/tests/chat_turn_timeout.rs new file mode 100644 index 0000000..d3342c8 --- /dev/null +++ b/crates/server/tests/chat_turn_timeout.rs @@ -0,0 +1,167 @@ +//! Regression coverage for issue #359 — the synchronous `POST /chat` turn +//! timeout must not bill for a turn the client was told timed out. +//! +//! Before the fix, the HTTP handler wrapped `submit_chat` in a fixed 30s +//! `tokio::time::timeout`, but the turn itself ran on a detached worker thread +//! that owns the `!Send` fused runtime. When the wait elapsed, the handler +//! returned `504` and dropped its side of the reply channel — yet the worker +//! kept running the turn to completion, minting a receipt and billing cost for +//! a reply the client never received. These tests pin down both halves of the +//! fix: the timeout is configurable, and a timed-out turn is cancelled before +//! it commits a receipt. + +mod support; + +use std::sync::Arc; +use std::time::Duration; + +use ardur_provider_runtime::{ + CompletionRequest, CompletionResponse, FinishReason, Provider, ProviderError, ProviderId, + RateCard, Usage, +}; +use ardur_runtime::CostTuple; +use ardur_server::{AppState, Config, build_router, example_registry}; +use async_trait::async_trait; +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use serde_json::Value; + +/// A provider whose `complete` sleeps for `delay` before returning a *billed* +/// reply — long enough to outrun a short HTTP turn timeout. +struct SlowProvider { + rate_card: RateCard, + delay: Duration, +} + +impl SlowProvider { + fn new(delay: Duration) -> Arc { + Arc::new(Self { + rate_card: RateCard::anthropic_2026_q2_v1(), + delay, + }) + } +} + +#[async_trait] +impl Provider for SlowProvider { + async fn complete(&self, _req: CompletionRequest) -> Result { + tokio::time::sleep(self.delay).await; + Ok(CompletionResponse { + content: "slow but complete".to_string(), + finish_reason: FinishReason::Stop, + usage: Usage { + tokens_in: 10, + tokens_out: 20, + cost_cents: None, + }, + cost: CostTuple { + tokens_in: 10, + tokens_out: 20, + cents: 5, + ..CostTuple::default() + }, + raw_provider_response: None, + }) + } + + fn id(&self) -> ProviderId { + ProviderId("slow".to_string()) + } + + fn supports_streaming(&self) -> bool { + false + } + + fn rate_card(&self) -> &RateCard { + &self.rate_card + } +} + +/// Boot state + router over `provider` with a `timeout`-bounded HTTP turn wait. +async fn boot_with_timeout( + provider: Arc, + timeout: Duration, +) -> (Arc, axum::Router) { + let dir = Box::leak(Box::new(tempfile::tempdir().expect("tempdir"))); + let mut config: Config = support::test_config(dir, None); + config.http_turn_timeout = timeout; + let tools = Arc::new(example_registry("slow", "in-memory")); + let state = AppState::boot(&config, provider, tools) + .await + .expect("AppState boots"); + let router = build_router(Arc::clone(&state)); + (state, router) +} + +fn chat_request(message: &str) -> Request { + Request::builder() + .method("POST") + .uri("/chat") + .header("content-type", "application/json") + .header("authorization", format!("Bearer {}", support::CHAT_TOKEN)) + .body(Body::from( + serde_json::json!({ "message": message }).to_string(), + )) + .expect("request builds") +} + +/// The turn outruns the (tiny) configured timeout: the client must get `504`, +/// and — the crux of #359 — no receipt may be minted for it, because the +/// timed-out turn is cancelled before it commits any receipt/billing side +/// effect. A wait past the provider delay proves the turn was cancelled rather +/// than merely still in flight. +#[tokio::test] +async fn timed_out_turn_returns_504_and_mints_no_receipt() { + // Provider round takes ~600ms; the HTTP surface only waits 100ms. + let provider = SlowProvider::new(Duration::from_millis(600)); + let (state, router) = boot_with_timeout( + Arc::clone(&provider) as Arc, + Duration::from_millis(100), + ) + .await; + + assert_eq!(state.receipt_count(), 0, "no receipts before the turn"); + + let (status, bytes) = support::oneshot(router, chat_request("hello")).await; + assert_eq!( + status, + StatusCode::GATEWAY_TIMEOUT, + "a turn that outruns the timeout returns 504" + ); + let json: Value = serde_json::from_slice(&bytes).expect("response body is JSON"); + assert!( + json["error"] + .as_str() + .unwrap_or_default() + .contains("timed out"), + "504 body names the timeout: {json}" + ); + + // Wait well past the provider delay: had the turn *not* been cancelled, the + // detached worker would have finished the provider round and minted+billed a + // receipt by now. Cancellation means the count stays at zero. + tokio::time::sleep(Duration::from_millis(900)).await; + assert_eq!( + state.receipt_count(), + 0, + "a timed-out turn must not mint (or bill) a receipt the client never saw" + ); +} + +/// The happy path with a generous timeout: a turn that finishes within the +/// budget returns `200` and *does* mint a receipt. Guards against the fix +/// over-cancelling legitimate turns. +#[tokio::test] +async fn turn_within_timeout_succeeds_and_mints_receipt() { + let provider = SlowProvider::new(Duration::from_millis(50)); + let (state, router) = + boot_with_timeout(provider as Arc, Duration::from_secs(30)).await; + + let (status, _bytes) = support::oneshot(router, chat_request("hello")).await; + assert_eq!(status, StatusCode::OK, "a prompt turn returns 200"); + assert_eq!( + state.receipt_count(), + 1, + "a completed turn mints exactly one receipt" + ); +} diff --git a/crates/server/tests/support/mod.rs b/crates/server/tests/support/mod.rs index 72f2867..52e4cc4 100644 --- a/crates/server/tests/support/mod.rs +++ b/crates/server/tests/support/mod.rs @@ -91,6 +91,7 @@ pub fn test_config(data_dir: &TempDir, slack_base: Option) -> Config { enable_http_tool: false, http_allowlist: Vec::new(), file_tool_root: None, + http_turn_timeout: std::time::Duration::from_secs(30), } }