1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
//! Regression guards for #1635.
//!
//! Two defects shipped together and reinforced each other:
//!
//! 1. Every OpenAI-compatible provider built its HTTP client with a 60s total
//! request timeout. reqwest's `.timeout()` covers the response body read, so
//! it is a hard wall-clock ceiling on an SSE stream no matter how many chunks
//! are still arriving. Anthropic and Gemini already used 300s, which is why
//! only the compat providers (every custom endpoint, i.e. every cheap
//! thinking-heavy fallback model) died at exactly 60.0s mid-stream.
//! #1635 raised the number to 300s, which moved the wall. #1687 took it off
//! the stream path altogether: the total now bounds non-streaming calls only,
//! and `the_stream_clients_carry_no_total_timeout` below pins that it never
//! comes back.
//! 2. The stream-retry log lines hardcoded `3` while `MAX_STREAM_RETRIES` is 5,
//! emitting `Stream retry 5/3 failed` and `All 3 stream retries failed`. That
//! made the retry budget look smaller than it is and sent the first reading
//! of the incident down the wrong path entirely.
use std::path::Path;
use crate::brain::provider::anthropic::DEFAULT_TIMEOUT as ANTHROPIC_TOTAL_TIMEOUT;
use crate::brain::provider::custom_openai_compatible::DEFAULT_TIMEOUT as COMPAT_TOTAL_TIMEOUT;
use crate::brain::provider::gemini::DEFAULT_TIMEOUT as GEMINI_TOTAL_TIMEOUT;
#[test]
fn compat_total_timeout_is_not_below_the_native_stream_providers() {
assert!(
COMPAT_TOTAL_TIMEOUT >= ANTHROPIC_TOTAL_TIMEOUT,
"OpenAI-compatible total request timeout ({:?}) is below the Anthropic one ({:?}). \
reqwest's client timeout covers the streamed body, so a lower value guillotines \
healthy SSE streams on every custom provider while Anthropic keeps working (#1635).",
COMPAT_TOTAL_TIMEOUT,
ANTHROPIC_TOTAL_TIMEOUT,
);
assert!(
COMPAT_TOTAL_TIMEOUT >= GEMINI_TOTAL_TIMEOUT,
"OpenAI-compatible total request timeout ({:?}) is below the Gemini one ({:?}) (#1635).",
COMPAT_TOTAL_TIMEOUT,
GEMINI_TOTAL_TIMEOUT,
);
}
/// #1687: the total wall clock must never sit on the stream path again.
///
/// reqwest's `.timeout()` bounds the whole exchange INCLUDING the body read, so
/// a client that carries one puts a ceiling on every stream it serves, no matter
/// how healthily chunks arrive. The inter-chunk idle guard in `helpers.rs` is
/// the correct detector for a dead stream, and it is the only one a stream may
/// meet. This scans the source of all three provider families rather than
/// asserting a relationship between two constants, because the defect was never
/// the VALUE of the number: #1635 raised it from 60s to 300s and the mechanism
/// survived intact.
#[test]
fn the_stream_clients_carry_no_total_timeout() {
let manifest = Path::new(env!("CARGO_MANIFEST_DIR"));
let families = [
(
"src/brain/provider/custom_openai_compatible.rs",
"build_stream_http_client",
),
("src/brain/provider/anthropic.rs", "build_stream_client"),
("src/brain/provider/gemini.rs", "build_stream_client"),
];
for (rel, builder) in families {
let src = std::fs::read_to_string(manifest.join(rel))
.unwrap_or_else(|e| panic!("cannot read {rel}: {e}"));
// 1. The stream-client builder exists...
let head = format!("fn {builder}() -> Client {{");
let start = src.find(&head).unwrap_or_else(|| {
panic!(
"{rel}: `{builder}()` is gone — that is the client that \
carries no total ceiling (#1687)"
)
});
let rest = &src[start..];
let body = &rest[..rest
.find(
"
}",
)
.unwrap_or(rest.len())];
// 2a. ...and sets no total timeout.
assert!(
!body.contains(".timeout("),
"{rel}: `{builder}()` sets a total request timeout again, which is a wall clock on \
every stream it serves (#1687).\n{body}"
);
// 2b. ...and `stream()` posts through it, not through the timed client.
let sstart = src
.find("async fn stream(")
.unwrap_or_else(|| panic!("{rel}: no `stream()` found"));
let srest = &src[sstart..];
let send = &srest[..srest
.find(
"
async fn ",
)
.or_else(|| {
srest.find(
"
fn ",
)
})
.unwrap_or(srest.len())];
assert!(
send.contains(".stream_client"),
"{rel}: `stream()` no longer posts on the total-timeout-free client (#1687)"
);
for (idx, line) in send.lines().enumerate() {
assert!(
line.trim() != ".client",
"{rel}: `stream()` still reaches for the timed `client` at relative line {idx} \
— that is the #1687 ceiling back on the stream path"
);
}
}
}
#[test]
fn stream_retry_log_lines_never_hardcode_the_retry_budget() {
let tool_loop =
Path::new(env!("CARGO_MANIFEST_DIR")).join("src/brain/agent/service/tool_loop.rs");
let content = std::fs::read_to_string(&tool_loop)
.unwrap_or_else(|e| panic!("cannot read {}: {e}", tool_loop.display()));
// Anchors: if the retry log lines are ever reshaped, these fail loudly
// instead of letting the scan below pass vacuously.
for anchor in [
"\"Stream retry {}/{} failed: {}\"",
"\"All {} stream retries failed",
"\"5xx retry {}/{} failed: {}\"",
"\"All {} 5xx retries failed",
] {
assert!(
content.contains(anchor),
"expected retry log format {anchor} in tool_loop.rs; the scan below cannot \
guard what it can no longer find (#1635)"
);
}
let mut violations = Vec::new();
for (idx, line) in content.lines().enumerate() {
let lower = line.to_ascii_lowercase();
if !lower.contains("retry") && !lower.contains("retries") {
continue;
}
// `retry {}/3 failed` — the attempt counter with a literal budget.
if digit_follows(line, "retry {}/") {
violations.push((idx + 1, line.trim().to_string()));
continue;
}
// `All 3 stream retries failed` — the exhausted-budget line.
if digit_follows(line, "All ") {
violations.push((idx + 1, line.trim().to_string()));
}
}
assert!(
violations.is_empty(),
"stream retry log lines must interpolate MAX_STREAM_RETRIES, not a literal count \
(#1635). Offending lines in src/brain/agent/service/tool_loop.rs:\n{}",
violations
.iter()
.map(|(line_no, text)| format!(" {line_no}: {text}"))
.collect::<Vec<_>>()
.join("\n"),
);
}
/// True when `marker` occurs in `line` immediately followed by an ASCII digit.
fn digit_follows(line: &str, marker: &str) -> bool {
line.match_indices(marker).any(|(at, _)| {
line[at + marker.len()..]
.chars()
.next()
.is_some_and(|c| c.is_ascii_digit())
})
}