From 27aae7ddf840fb6e7fa41adbe741e428bb496766 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Mon, 6 Jul 2026 19:10:03 -0700 Subject: [PATCH] harness: add optional tool_name to ToolDelta so streamed tool calls self-label A streaming tool call's name arrives on the call-opening fragment, before its arguments. `ToolDelta` previously carried only `{call_id, content}`, so a consumer projecting the stream had no way to label the call until the final response. Add `ToolDelta::tool_name: Option`: - The OpenAI SSE parser surfaces the captured function name on the tool-call delta. - `StreamAccumulator` records the first non-empty name per call id and stamps it onto the reconstructed `ToolCall`, so a stream-only response (no authoritative `Completed`) still names its tool calls. Backward-compatible: `#[serde(default, skip_serializing_if = "Option::is_none")]`, so existing payloads round-trip unchanged. New focused accumulator test. --- src/harness/model/mod.rs | 38 +++++++++++++++++------- src/harness/model/test.rs | 29 ++++++++++++++++++ src/harness/providers/openai/sse.rs | 4 +++ src/harness/tool/types.rs | 8 +++++ tests/e2e_harness_provider_contracts.rs | 3 ++ tests/e2e_middleware_parser_contracts.rs | 1 + 6 files changed, 72 insertions(+), 11 deletions(-) diff --git a/src/harness/model/mod.rs b/src/harness/model/mod.rs index d38b7d9..aa24cb1 100644 --- a/src/harness/model/mod.rs +++ b/src/harness/model/mod.rs @@ -686,8 +686,10 @@ pub struct StreamAccumulator { /// the final message text). reasoning: String, /// Per-call-id accumulated tool-call argument fragments, in first-seen - /// order. - tool_chunks: Vec<(String, String)>, + /// order: `(call_id, arguments, tool_name)`. The name is the first non-empty + /// `ToolDelta::tool_name` seen for the call (providers surface it on the + /// call-opening delta); `None` until one arrives. + tool_chunks: Vec<(String, String, Option)>, /// Most recent usage value seen. usage: Option, /// Authoritative final response, when a `Completed` item was seen. @@ -710,11 +712,15 @@ impl StreamAccumulator { self.text.push_str(&delta.text); self.reasoning.push_str(&delta.reasoning); if let Some(tool_call) = &delta.tool_call { - self.push_tool_chunk(&tool_call.call_id, &tool_call.content); + self.push_tool_chunk( + &tool_call.call_id, + &tool_call.content, + tool_call.tool_name.as_deref(), + ); } } ModelStreamItem::ToolCallDelta(delta) => { - self.push_tool_chunk(&delta.call_id, &delta.content); + self.push_tool_chunk(&delta.call_id, &delta.content, delta.tool_name.as_deref()); } ModelStreamItem::UsageDelta(usage) => { self.usage = Some(*usage); @@ -741,13 +747,23 @@ impl StreamAccumulator { } /// Appends a tool-call argument fragment for `call_id`, preserving - /// first-seen ordering across calls. - fn push_tool_chunk(&mut self, call_id: &str, content: &str) { - if let Some(entry) = self.tool_chunks.iter_mut().find(|(id, _)| id == call_id) { + /// first-seen ordering across calls. Records the first non-empty `tool_name` + /// seen for the call (call-opening deltas carry it; argument fragments do + /// not) so the reconstructed [`ToolCall`] is named. + fn push_tool_chunk(&mut self, call_id: &str, content: &str, tool_name: Option<&str>) { + if let Some(entry) = self.tool_chunks.iter_mut().find(|(id, ..)| id == call_id) { entry.1.push_str(content); + if entry.2.is_none() { + if let Some(name) = tool_name.filter(|n| !n.is_empty()) { + entry.2 = Some(name.to_string()); + } + } } else { - self.tool_chunks - .push((call_id.to_string(), content.to_string())); + self.tool_chunks.push(( + call_id.to_string(), + content.to_string(), + tool_name.filter(|n| !n.is_empty()).map(str::to_string), + )); } } @@ -805,8 +821,8 @@ impl StreamAccumulator { let tool_calls = self .tool_chunks .into_iter() - .map(|(id, args)| ToolCall { - name: String::new(), + .map(|(id, args, name)| ToolCall { + name: name.unwrap_or_default(), arguments: serde_json::from_str(&args).unwrap_or(Value::Null), id, }) diff --git a/src/harness/model/test.rs b/src/harness/model/test.rs index 9195f95..64a86c5 100644 --- a/src/harness/model/test.rs +++ b/src/harness/model/test.rs @@ -684,6 +684,34 @@ fn finish_backfills_usage_from_stream_delta_when_completed_lacks_it() { assert_eq!(finished.message.usage, Some(Usage::new(4, 6))); } +#[test] +fn finish_names_reconstructed_tool_call_from_the_call_opening_delta_name() { + // A call-opening delta carries the tool name (no args yet); subsequent + // argument fragments carry only content. With no authoritative `Completed` + // response, the accumulator must still name the reconstructed tool call from + // the first non-empty `tool_name` it saw for that call id. + use crate::harness::tool::ToolDelta; + + let mut acc = StreamAccumulator::new(); + acc.push(&ModelStreamItem::ToolCallDelta(ToolDelta { + call_id: "call-1".into(), + content: String::new(), + tool_name: Some("search".into()), + })); + acc.push(&ModelStreamItem::ToolCallDelta(ToolDelta { + call_id: "call-1".into(), + content: r#"{"q":"rust"}"#.into(), + tool_name: None, + })); + + let finished = acc.finish().unwrap(); + assert_eq!(finished.message.tool_calls.len(), 1); + let call = &finished.message.tool_calls[0]; + assert_eq!(call.id, "call-1"); + assert_eq!(call.name, "search", "name carried from the opening delta"); + assert_eq!(call.arguments, serde_json::json!({ "q": "rust" })); +} + /// Round-trips a [`ModelStreamItem`] through JSON and asserts the re-serialized /// form is byte-for-byte stable, proving every variant survives serde. fn roundtrip_stream_item(item: ModelStreamItem) { @@ -704,6 +732,7 @@ fn model_stream_item_roundtrips_every_variant() { crate::harness::tool::ToolDelta { call_id: "call-1".into(), content: "{\"q\":1}".into(), + tool_name: None, }, )); roundtrip_stream_item(ModelStreamItem::UsageDelta(Usage::new(3, 5))); diff --git a/src/harness/providers/openai/sse.rs b/src/harness/providers/openai/sse.rs index 5ae76e4..5421765 100644 --- a/src/harness/providers/openai/sse.rs +++ b/src/harness/providers/openai/sse.rs @@ -86,6 +86,10 @@ impl OpenAiStreamAcc { pending.push_back(ModelStreamItem::ToolCallDelta(ToolDelta { call_id, content: args, + // Surface the tool name (captured into `slot.name` from + // the call-opening fragment) so consumers can label the + // call as it streams; the accumulator keeps the first. + tool_name: Some(slot.name.clone()).filter(|n| !n.is_empty()), })); } } diff --git a/src/harness/tool/types.rs b/src/harness/tool/types.rs index c84219f..3edb7b5 100644 --- a/src/harness/tool/types.rs +++ b/src/harness/tool/types.rs @@ -311,6 +311,14 @@ pub struct ToolDelta { pub call_id: String, /// Incremental content fragment. pub content: String, + /// The tool's name, when known. Providers that surface it on the first + /// (call-opening) delta populate this so consumers can label a tool call as + /// soon as it begins — before its arguments have streamed. Subsequent + /// argument fragments leave it `None`; [`StreamAccumulator`] remembers the + /// first non-empty name per `call_id` and stamps it onto the reconstructed + /// [`ToolCall`]. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_name: Option, } /// A tool the harness can invoke during an agent loop. diff --git a/tests/e2e_harness_provider_contracts.rs b/tests/e2e_harness_provider_contracts.rs index 4f402e3..a4b09b5 100644 --- a/tests/e2e_harness_provider_contracts.rs +++ b/tests/e2e_harness_provider_contracts.rs @@ -239,10 +239,12 @@ async fn model_request_response_registry_and_stream_contracts_are_stable() { accumulator.push(&ModelStreamItem::ToolCallDelta(ToolDelta { call_id: "call-1".into(), content: r#"{"a":"#.into(), + tool_name: None, })); accumulator.push(&ModelStreamItem::ToolCallDelta(ToolDelta { call_id: "call-1".into(), content: r#""b"}"#.into(), + tool_name: None, })); accumulator.push(&ModelStreamItem::UsageDelta(Usage::new(4, 5))); accumulator.push(&ModelStreamItem::MessageDelta(MessageDelta { @@ -316,6 +318,7 @@ fn model_profiles_stream_chunks_usage_and_cost_are_additive() { tool_call: Some(ToolDelta { call_id: "tool-1".into(), content: "partial".into(), + tool_name: None, }), }), StreamChunk::Debug("trace".into()), diff --git a/tests/e2e_middleware_parser_contracts.rs b/tests/e2e_middleware_parser_contracts.rs index d9bb41f..12e11d6 100644 --- a/tests/e2e_middleware_parser_contracts.rs +++ b/tests/e2e_middleware_parser_contracts.rs @@ -186,6 +186,7 @@ async fn middleware_stack_runs_lifecycle_hooks_and_builtin_guards() { let mut tool_delta = ToolDelta { call_id: "tool-1".into(), content: "progress".into(), + tool_name: None, }; stack .run_on_tool_delta(&mut ctx, &(), &mut tool_delta)