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
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
// Shared streaming tool-call accumulator (EVE-672)
//
// Native streaming drivers assemble tool calls from fragments spread across many
// SSE chunks: OpenAI Chat Completions keys fragments by a numeric `index`,
// OpenAI Open Responses keys them by an `item_id` string, and Gemini pushes
// whole calls. Each driver previously open-coded this into an
// `Arc<Mutex<Vec<...>>>` with subtly different growth, argument-append, and
// finalize rules. This module centralizes the accumulation so the rules live in
// one tested place and the drivers share a single `StreamToolCallAccumulator`.
//
// The accumulator keeps arguments as an un-parsed `String` fragment buffer and
// parses the JSON exactly once at finalize (EVE-636: `push_str` is amortized
// O(total), versus re-parsing a `Value` per delta which was O(n^2)). It exposes
// two finalize modes matching the historical driver behavior:
//
// - `take_finalized`: normal finish path — malformed argument JSON degrades to
// an empty object (`{}`), because the provider signalled a real tool-call
// completion.
// - `take_pending_strict`: fallback flush at `[DONE]`/end-of-stream without a
// tool-call finish — malformed argument JSON causes the call to be dropped,
// since there was no explicit completion to trust.
use crate::tool_types::ToolCall;
use serde_json::{Value, json};
/// One in-progress tool call being assembled from streamed fragments.
#[derive(Debug, Clone)]
struct PartialToolCall {
/// Fragment key: the numeric `index` (Chat Completions) or the stream
/// `item_id` (Open Responses). Whole-call providers (Gemini) leave it empty.
key: String,
/// Provider call id, applied to the finalized [`ToolCall::id`].
id: String,
/// Function name.
name: String,
/// Accumulated JSON argument fragments (parsed once at finalize).
arguments: String,
}
/// Accumulates streamed tool-call fragments into finalized [`ToolCall`]s.
///
/// Fragments are addressed by a string `key` (the provider's per-call index or
/// item id). Calls are emitted in first-seen order.
#[derive(Debug, Default)]
pub struct StreamToolCallAccumulator {
calls: Vec<PartialToolCall>,
}
impl StreamToolCallAccumulator {
/// Create an empty accumulator.
pub fn new() -> Self {
Self::default()
}
/// Whether any fragments have been accumulated.
pub fn is_empty(&self) -> bool {
self.calls.is_empty()
}
fn slot(&mut self, key: &str) -> &mut PartialToolCall {
if let Some(pos) = self.calls.iter().position(|c| c.key == key) {
return &mut self.calls[pos];
}
self.calls.push(PartialToolCall {
key: key.to_string(),
id: String::new(),
name: String::new(),
arguments: String::new(),
});
self.calls.last_mut().expect("just pushed")
}
/// Apply an OpenAI Chat Completions streamed tool-call delta, keyed by the
/// chunk's numeric `index`. Any of id/name/arguments may be absent in a
/// given delta; argument fragments are appended in place.
pub fn apply_indexed_delta(
&mut self,
index: u32,
id: Option<&str>,
name: Option<&str>,
arguments: Option<&str>,
) {
let slot = self.slot(&index.to_string());
if let Some(id) = id {
slot.id = id.to_string();
}
if let Some(name) = name {
slot.name = name.to_string();
}
if let Some(args) = arguments {
slot.arguments.push_str(args);
}
}
/// Append an Open Responses `function_call_arguments.delta` fragment, keyed
/// by `item_id`. Creates the slot if the item was not yet announced.
pub fn append_arguments(&mut self, item_id: &str, delta: &str) {
self.slot(item_id).arguments.push_str(delta);
}
/// Record an Open Responses `output_item.added` function call, keyed by
/// `item_id`, setting its name and provider `call_id`.
pub fn set_item(&mut self, item_id: &str, call_id: &str, name: &str) {
let slot = self.slot(item_id);
slot.id = call_id.to_string();
slot.name = name.to_string();
}
/// Push a fully-formed tool call (Gemini emits whole `functionCall` parts
/// with already-parsed arguments, so there is nothing to reassemble).
pub fn push_complete(&mut self, id: String, name: String, arguments: Value) {
self.calls.push(PartialToolCall {
key: String::new(),
id,
name,
// Store as a compact JSON string so the single finalize path parses
// it back identically to the fragment-assembled calls.
arguments: arguments.to_string(),
});
}
/// Drain all accumulated calls, parsing each argument string once. Malformed
/// argument JSON degrades to an empty object — the normal finish path, where
/// the provider explicitly signalled tool-call completion.
pub fn take_finalized(&mut self) -> Vec<ToolCall> {
std::mem::take(&mut self.calls)
.into_iter()
.map(|c| ToolCall {
id: c.id,
name: c.name,
arguments: parse_arguments(&c.arguments).unwrap_or_else(|| json!({})),
})
.collect()
}
/// Drain all accumulated calls for a *fallback* flush (end-of-stream without
/// an explicit tool-call finish). Malformed argument JSON drops the call
/// rather than fabricating an empty object, since no completion was signalled.
pub fn take_pending_strict(&mut self) -> Vec<ToolCall> {
std::mem::take(&mut self.calls)
.into_iter()
.filter_map(|c| {
Some(ToolCall {
id: c.id,
name: c.name,
arguments: parse_arguments(&c.arguments)?,
})
})
.collect()
}
/// Drain accumulated calls, keeping only those with a non-empty name and
/// parsing arguments once (empty/malformed → `{}`). Matches the Open
/// Responses finalize path, which skips announced-but-unnamed items.
pub fn take_named(&mut self) -> Vec<ToolCall> {
std::mem::take(&mut self.calls)
.into_iter()
.filter(|c| !c.name.is_empty())
.map(|c| ToolCall {
id: c.id,
name: c.name,
arguments: parse_arguments(&c.arguments).unwrap_or_else(|| json!({})),
})
.collect()
}
}
/// Parse an accumulated argument fragment buffer into JSON. An empty buffer is
/// treated as an empty object; a non-empty buffer that fails to parse returns
/// `None` so callers can choose their fallback.
fn parse_arguments(buffer: &str) -> Option<Value> {
if buffer.is_empty() {
return Some(json!({}));
}
serde_json::from_str(buffer).ok()
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn indexed_delta_reassembles_fragmented_arguments() {
let mut acc = StreamToolCallAccumulator::new();
acc.apply_indexed_delta(0, Some("call_a"), Some("get_weather"), Some(""));
acc.apply_indexed_delta(0, None, None, Some("{\"city\":"));
acc.apply_indexed_delta(0, None, None, Some("\"Paris\"}"));
let calls = acc.take_finalized();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].id, "call_a");
assert_eq!(calls[0].name, "get_weather");
assert_eq!(calls[0].arguments, json!({"city": "Paris"}));
assert!(acc.is_empty());
}
#[test]
fn indexed_delta_grows_to_index_and_preserves_order() {
let mut acc = StreamToolCallAccumulator::new();
// Deltas can arrive for index 1 before index 0 has all fields, but the
// first-seen order is preserved.
acc.apply_indexed_delta(0, Some("c0"), Some("first"), Some("{}"));
acc.apply_indexed_delta(1, Some("c1"), Some("second"), Some("{}"));
let calls = acc.take_finalized();
assert_eq!(calls[0].name, "first");
assert_eq!(calls[1].name, "second");
}
#[test]
fn empty_arguments_finalize_to_empty_object() {
let mut acc = StreamToolCallAccumulator::new();
acc.apply_indexed_delta(0, Some("c"), Some("noop"), None);
let calls = acc.take_finalized();
assert_eq!(calls[0].arguments, json!({}));
}
#[test]
fn take_finalized_degrades_malformed_json_to_empty_object() {
let mut acc = StreamToolCallAccumulator::new();
acc.apply_indexed_delta(0, Some("c"), Some("bad"), Some("{not json"));
let calls = acc.take_finalized();
assert_eq!(calls[0].arguments, json!({}));
}
#[test]
fn take_pending_strict_drops_malformed_json() {
let mut acc = StreamToolCallAccumulator::new();
acc.apply_indexed_delta(0, Some("c"), Some("bad"), Some("{not json"));
acc.apply_indexed_delta(1, Some("d"), Some("good"), Some("{\"ok\":true}"));
let calls = acc.take_pending_strict();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "good");
}
#[test]
fn item_keyed_flow_matches_open_responses() {
let mut acc = StreamToolCallAccumulator::new();
acc.set_item("fc_1", "call_1", "get_weather");
acc.append_arguments("fc_1", "{\"city\":");
acc.append_arguments("fc_1", "\"Paris\"}");
let calls = acc.take_named();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].id, "call_1");
assert_eq!(calls[0].name, "get_weather");
assert_eq!(calls[0].arguments, json!({"city": "Paris"}));
}
#[test]
fn take_named_skips_unnamed_items() {
let mut acc = StreamToolCallAccumulator::new();
// An arguments delta arrived before the item was announced with a name.
acc.append_arguments("fc_orphan", "{}");
acc.set_item("fc_1", "call_1", "named");
acc.append_arguments("fc_1", "{}");
let calls = acc.take_named();
assert_eq!(calls.len(), 1);
assert_eq!(calls[0].name, "named");
}
#[test]
fn push_complete_serializes_and_reparses_identically() {
let mut acc = StreamToolCallAccumulator::new();
acc.push_complete("call_0".into(), "f".into(), json!({"a": 1, "b": "x"}));
let calls = acc.take_finalized();
assert_eq!(calls[0].arguments, json!({"a": 1, "b": "x"}));
}
}