External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor(agent): rename llm_stream_chunk_timeout and extract test helper

Browse Source 
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 <pfleger.will@gmail.com>
Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
npub1mn7jgtj4w2pd0g0zeuhxsa6jy6p0rewxz4kujt98my82ahfmp72sxjexk7
2026-06-10 14:01:28 -04:00
co-authored by Will Pfleger
parent 06378f49ef
commit dac14a526a
6 changed files with 92 additions and 200 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/sprout-agent/src/config.rs
+11 -10
View File
@@ -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());
crates/sprout-agent/src/llm.rs
+2 -2
View File
@@ -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),
crates/sprout-agent/tests/common/mod.rs
+70
View File
@@ -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<String> {
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
}
crates/sprout-agent/tests/fake_llm.rs
+3 -65
View File
@@ -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<Value>) -> String {
@@ -105,71 +108,6 @@ async fn spawn_fake_llm(responses: Vec<Value>) -> String {
});
url
}
/// Convert a canned OpenAI Chat Completions response into SSE delta events.
fn openai_to_sse_events(response: &Value) -> Vec<String> {
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 {
crates/sprout-agent/tests/golden_transcripts.rs
+3 -60
View File
@@ -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<Value>) -> String {
});
url
}
/// Convert a canned OpenAI Chat Completions response into SSE delta events.
fn openai_to_sse_events(response: &Value) -> Vec<String> {
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",
crates/sprout-agent/tests/regressions.rs
+3 -63
View File
@@ -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<String> {
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",