From dac14a526abf6ef93aa3f670cdf41d4cb0c93c38 Mon Sep 17 00:00:00 2001 From: npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7 Date: Wed, 10 Jun 2026 14:01:28 -0400 Subject: [PATCH] refactor(agent): rename llm_stream_chunk_timeout and extract test helper The buffered-body inter-chunk timeout was named llm_stream_chunk_timeout, confusingly similar to stream_chunk_timeout (the SSE inter-chunk timeout). Operators tuning "stream chunk timeout" would set the wrong env var. Rename to llm_body_chunk_timeout / SPROUT_AGENT_LLM_BODY_CHUNK_TIMEOUT_SECS to clearly distinguish the buffered path from the streaming path. Extract the openai_to_sse_events test helper (duplicated in fake_llm.rs, golden_transcripts.rs, and regressions.rs) into tests/common/mod.rs. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- crates/sprout-agent/src/config.rs | 21 +++--- crates/sprout-agent/src/llm.rs | 4 +- crates/sprout-agent/tests/common/mod.rs | 70 +++++++++++++++++++ crates/sprout-agent/tests/fake_llm.rs | 68 +----------------- .../sprout-agent/tests/golden_transcripts.rs | 63 +---------------- crates/sprout-agent/tests/regressions.rs | 66 +---------------- 6 files changed, 92 insertions(+), 200 deletions(-) create mode 100644 crates/sprout-agent/tests/common/mod.rs diff --git a/crates/sprout-agent/src/config.rs b/crates/sprout-agent/src/config.rs index 1f19b56fa..6eea17732 100644 --- a/crates/sprout-agent/src/config.rs +++ b/crates/sprout-agent/src/config.rs @@ -48,13 +48,14 @@ pub struct Config { pub max_rounds: u32, pub max_output_tokens: u32, pub llm_timeout: Duration, - /// Maximum time to wait between consecutive response body chunks from the - /// LLM provider. If no data arrives within this window the request is - /// considered stalled and terminated. This does NOT cap total response - /// time — a stream that keeps producing chunks can run indefinitely. - pub llm_stream_chunk_timeout: Duration, + /// Maximum time to wait between consecutive response body chunks on the + /// non-streaming (buffered) `post()` path. If no data arrives within this + /// window the request is considered stalled and terminated. This does NOT + /// cap total response time — a body that keeps producing chunks can run + /// indefinitely. SSE streaming uses `stream_chunk_timeout` instead. + pub llm_body_chunk_timeout: Duration, /// Inter-chunk timeout for SSE streaming after the first content delta - /// arrives. Tighter than `llm_stream_chunk_timeout` because SSE events + /// arrives. Tighter than `llm_body_chunk_timeout` because SSE events /// arrive frequently once content generation starts. Default 30s. pub stream_chunk_timeout: Duration, pub tool_timeout: Duration, @@ -161,8 +162,8 @@ impl Config { max_rounds: parse_env("SPROUT_AGENT_MAX_ROUNDS", 0)?, max_output_tokens: parse_env("SPROUT_AGENT_MAX_OUTPUT_TOKENS", 32_768)?, llm_timeout: Duration::from_secs(parse_env("SPROUT_AGENT_LLM_TIMEOUT_SECS", 120)?), - llm_stream_chunk_timeout: Duration::from_secs(parse_env( - "SPROUT_AGENT_LLM_STREAM_CHUNK_TIMEOUT_SECS", + llm_body_chunk_timeout: Duration::from_secs(parse_env( + "SPROUT_AGENT_LLM_BODY_CHUNK_TIMEOUT_SECS", 120, )?), stream_chunk_timeout: Duration::from_secs(parse_env( @@ -228,8 +229,8 @@ impl Config { if self.llm_timeout < MIN_TIMEOUT { return Err("config: SPROUT_AGENT_LLM_TIMEOUT_SECS must be >= 1".into()); } - if self.llm_stream_chunk_timeout < MIN_TIMEOUT { - return Err("config: SPROUT_AGENT_LLM_STREAM_CHUNK_TIMEOUT_SECS must be >= 1".into()); + if self.llm_body_chunk_timeout < MIN_TIMEOUT { + return Err("config: SPROUT_AGENT_LLM_BODY_CHUNK_TIMEOUT_SECS must be >= 1".into()); } if self.stream_chunk_timeout < MIN_TIMEOUT { return Err("config: SPROUT_AGENT_STREAM_CHUNK_TIMEOUT_SECS must be >= 1".into()); diff --git a/crates/sprout-agent/src/llm.rs b/crates/sprout-agent/src/llm.rs index 31635d10b..4fe13250a 100644 --- a/crates/sprout-agent/src/llm.rs +++ b/crates/sprout-agent/src/llm.rs @@ -78,7 +78,7 @@ impl Llm { http_stream, auto_upgraded: AtomicBool::new(false), auth, - chunk_timeout: cfg.llm_stream_chunk_timeout, + chunk_timeout: cfg.llm_body_chunk_timeout, stream_chunk_timeout: cfg.stream_chunk_timeout, first_byte_timeout: cfg.llm_timeout, }) @@ -1696,7 +1696,7 @@ mod tests { max_rounds: 10, max_output_tokens: 1024, llm_timeout: Duration::from_secs(10), - llm_stream_chunk_timeout: Duration::from_secs(120), + llm_body_chunk_timeout: Duration::from_secs(120), stream_chunk_timeout: Duration::from_secs(30), tool_timeout: Duration::from_secs(10), mcp_init_timeout: Duration::from_secs(10), diff --git a/crates/sprout-agent/tests/common/mod.rs b/crates/sprout-agent/tests/common/mod.rs new file mode 100644 index 000000000..a68f87d25 --- /dev/null +++ b/crates/sprout-agent/tests/common/mod.rs @@ -0,0 +1,70 @@ +//! Shared helpers for the `sprout-agent` integration tests. A `common/mod.rs` +//! module (rather than a top-level `common.rs`) keeps Cargo from treating this +//! file as its own test binary. + +use serde_json::{json, Value}; + +/// Convert a canned OpenAI Chat Completions response into the SSE delta events +/// a streaming consumer would receive. Emits a content delta (when present), +/// two deltas per tool call (id+name, then arguments), and a final +/// finish-reason chunk. Usage is carried from the original response when +/// present, otherwise a default is supplied. +pub fn openai_to_sse_events(response: &Value) -> Vec { + let mut events = Vec::new(); + let choice = &response["choices"][0]; + let msg = &choice["message"]; + + if let Some(content) = msg.get("content").and_then(Value::as_str) { + if !content.is_empty() { + events.push( + json!({ + "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": null}] + }) + .to_string(), + ); + } + } + + if let Some(tcs) = msg.get("tool_calls").and_then(Value::as_array) { + for (i, tc) in tcs.iter().enumerate() { + let id = tc.get("id").and_then(Value::as_str).unwrap_or(""); + let name = tc["function"] + .get("name") + .and_then(Value::as_str) + .unwrap_or(""); + let args = tc["function"] + .get("arguments") + .and_then(Value::as_str) + .unwrap_or("{}"); + events.push(json!({ + "choices": [{"index": 0, "delta": { + "tool_calls": [{"index": i, "id": id, "function": {"name": name, "arguments": ""}}] + }, "finish_reason": null}] + }).to_string()); + events.push( + json!({ + "choices": [{"index": 0, "delta": { + "tool_calls": [{"index": i, "function": {"arguments": args}}] + }, "finish_reason": null}] + }) + .to_string(), + ); + } + } + + let finish = choice + .get("finish_reason") + .and_then(Value::as_str) + .unwrap_or("stop"); + let mut final_event = json!({ + "choices": [{"index": 0, "delta": {}, "finish_reason": finish}], + }); + if let Some(usage) = response.get("usage") { + final_event["usage"] = usage.clone(); + } else { + final_event["usage"] = json!({"prompt_tokens": 10, "completion_tokens": 5}); + } + events.push(final_event.to_string()); + + events +} diff --git a/crates/sprout-agent/tests/fake_llm.rs b/crates/sprout-agent/tests/fake_llm.rs index 8cf0671e9..0239c7af9 100644 --- a/crates/sprout-agent/tests/fake_llm.rs +++ b/crates/sprout-agent/tests/fake_llm.rs @@ -16,6 +16,9 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio::sync::Mutex; +mod common; +use common::openai_to_sse_events; + // ─── Fake LLM server ──────────────────────────────────────────────────────── async fn spawn_fake_llm(responses: Vec) -> String { @@ -105,71 +108,6 @@ async fn spawn_fake_llm(responses: Vec) -> String { }); url } - -/// Convert a canned OpenAI Chat Completions response into SSE delta events. -fn openai_to_sse_events(response: &Value) -> Vec { - let mut events = Vec::new(); - let choice = &response["choices"][0]; - let msg = &choice["message"]; - - // Text content - if let Some(content) = msg.get("content").and_then(Value::as_str) { - if !content.is_empty() { - events.push( - json!({ - "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": null}] - }) - .to_string(), - ); - } - } - - // Tool calls - if let Some(tcs) = msg.get("tool_calls").and_then(Value::as_array) { - for (i, tc) in tcs.iter().enumerate() { - let id = tc.get("id").and_then(Value::as_str).unwrap_or(""); - let name = tc["function"] - .get("name") - .and_then(Value::as_str) - .unwrap_or(""); - let args = tc["function"] - .get("arguments") - .and_then(Value::as_str) - .unwrap_or("{}"); - // First chunk: id + name - events.push(json!({ - "choices": [{"index": 0, "delta": { - "tool_calls": [{"index": i, "id": id, "function": {"name": name, "arguments": ""}}] - }, "finish_reason": null}] - }).to_string()); - // Second chunk: arguments - events.push( - json!({ - "choices": [{"index": 0, "delta": { - "tool_calls": [{"index": i, "function": {"arguments": args}}] - }, "finish_reason": null}] - }) - .to_string(), - ); - } - } - - // Final chunk with finish_reason - let finish = choice - .get("finish_reason") - .and_then(Value::as_str) - .unwrap_or("stop"); - events.push( - json!({ - "choices": [{"index": 0, "delta": {}, "finish_reason": finish}], - "usage": {"prompt_tokens": 10, "completion_tokens": 5} - }) - .to_string(), - ); - - events -} - // ─── ACP harness ──────────────────────────────────────────────────────────── struct Harness { diff --git a/crates/sprout-agent/tests/golden_transcripts.rs b/crates/sprout-agent/tests/golden_transcripts.rs index 69cd90286..5051a4f6d 100644 --- a/crates/sprout-agent/tests/golden_transcripts.rs +++ b/crates/sprout-agent/tests/golden_transcripts.rs @@ -8,6 +8,9 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio::sync::Mutex; +mod common; +use common::openai_to_sse_events; + struct Harness { child: tokio::process::Child, stdin: tokio::process::ChildStdin, @@ -193,66 +196,6 @@ async fn spawn_fake_llm(responses: Vec) -> String { }); url } - -/// Convert a canned OpenAI Chat Completions response into SSE delta events. -fn openai_to_sse_events(response: &Value) -> Vec { - let mut events = Vec::new(); - let choice = &response["choices"][0]; - let msg = &choice["message"]; - - if let Some(content) = msg.get("content").and_then(Value::as_str) { - if !content.is_empty() { - events.push( - json!({ - "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": null}] - }) - .to_string(), - ); - } - } - - if let Some(tcs) = msg.get("tool_calls").and_then(Value::as_array) { - for (i, tc) in tcs.iter().enumerate() { - let id = tc.get("id").and_then(Value::as_str).unwrap_or(""); - let name = tc["function"] - .get("name") - .and_then(Value::as_str) - .unwrap_or(""); - let args = tc["function"] - .get("arguments") - .and_then(Value::as_str) - .unwrap_or("{}"); - events.push(json!({ - "choices": [{"index": 0, "delta": { - "tool_calls": [{"index": i, "id": id, "function": {"name": name, "arguments": ""}}] - }, "finish_reason": null}] - }).to_string()); - events.push( - json!({ - "choices": [{"index": 0, "delta": { - "tool_calls": [{"index": i, "function": {"arguments": args}}] - }, "finish_reason": null}] - }) - .to_string(), - ); - } - } - - let finish = choice - .get("finish_reason") - .and_then(Value::as_str) - .unwrap_or("stop"); - events.push( - json!({ - "choices": [{"index": 0, "delta": {}, "finish_reason": finish}], - "usage": {"prompt_tokens": 10, "completion_tokens": 5} - }) - .to_string(), - ); - - events -} - fn openai_text(content: &str) -> Value { json!({ "id": "cc-1", "object": "chat.completion", "model": "fake-model", diff --git a/crates/sprout-agent/tests/regressions.rs b/crates/sprout-agent/tests/regressions.rs index 90c4fe673..f1d0f17c6 100644 --- a/crates/sprout-agent/tests/regressions.rs +++ b/crates/sprout-agent/tests/regressions.rs @@ -15,6 +15,9 @@ use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader}; use tokio::net::TcpListener; use tokio::sync::Mutex; +mod common; +use common::openai_to_sse_events; + // ─── Fake LLM that captures requests so we can inspect history ────────────── struct CapturingLlm { @@ -234,69 +237,6 @@ fn openai_tool_call(id: &str, name: &str, args: Value) -> Value { }], }) } - -/// Convert a canned OpenAI Chat Completions response into SSE delta events. -fn openai_to_sse_events(response: &Value) -> Vec { - let mut events = Vec::new(); - let choice = &response["choices"][0]; - let msg = &choice["message"]; - - if let Some(content) = msg.get("content").and_then(Value::as_str) { - if !content.is_empty() { - events.push( - json!({ - "choices": [{"index": 0, "delta": {"content": content}, "finish_reason": null}] - }) - .to_string(), - ); - } - } - - if let Some(tcs) = msg.get("tool_calls").and_then(Value::as_array) { - for (i, tc) in tcs.iter().enumerate() { - let id = tc.get("id").and_then(Value::as_str).unwrap_or(""); - let name = tc["function"] - .get("name") - .and_then(Value::as_str) - .unwrap_or(""); - let args = tc["function"] - .get("arguments") - .and_then(Value::as_str) - .unwrap_or("{}"); - events.push(json!({ - "choices": [{"index": 0, "delta": { - "tool_calls": [{"index": i, "id": id, "function": {"name": name, "arguments": ""}}] - }, "finish_reason": null}] - }).to_string()); - events.push( - json!({ - "choices": [{"index": 0, "delta": { - "tool_calls": [{"index": i, "function": {"arguments": args}}] - }, "finish_reason": null}] - }) - .to_string(), - ); - } - } - - let finish = choice - .get("finish_reason") - .and_then(Value::as_str) - .unwrap_or("stop"); - // Carry usage from the original response if present - let mut final_event = json!({ - "choices": [{"index": 0, "delta": {}, "finish_reason": finish}], - }); - if let Some(usage) = response.get("usage") { - final_event["usage"] = usage.clone(); - } else { - final_event["usage"] = json!({"prompt_tokens": 10, "completion_tokens": 5}); - } - events.push(final_event.to_string()); - - events -} - async fn init_session(h: &mut Harness, mcp_servers: Value) -> String { h.send( "initialize",