From c29b787567ed4d594b75b6db6f07403fd6f68520 Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Sun, 16 Aug 2026 20:38:02 +0100 Subject: [PATCH 01/10] Added: framework-owned RunEvent streaming type in core hooks Introduce the run_event module defining RunEvent, the #[non_exhaustive] item type a run stream yields, together with the distilled transcript records RunComplete carries. - RunEvent covers one variant per observable streaming milestone: run start, text and thinking deltas, tool call start and completion, tool execution, output-ready, run complete, error, and cancellation. New variants may be appended later; consumers match with a wildcard arm. - RunComplete carries a distilled transcript of RunMessage records, each with its RunMessageRole, text, RunToolCallSummary calls, and an optional RunToolResultSummary. The record serves display and audit; model-replay detail stays with the underlying agent. - Every new type derives serde Serialize/Deserialize so events survive serialization boundaries unchanged. - Wire the module into hooks via mod run_event; and pub use self::run_event::*;, with a Public API doc list entry for the new types. - Tests pin serde roundtrips for the full-shape RunComplete transcript (externally tagged wire shape) and for every remaining variant. Purely additive: existing core hook tests are untouched and passing; clippy clean. --- src/reloaded-code-core/src/hooks/mod.rs | 9 + .../src/hooks/run_event/mod.rs | 248 ++++++++++++++++++ 2 files changed, 257 insertions(+) create mode 100644 src/reloaded-code-core/src/hooks/run_event/mod.rs diff --git a/src/reloaded-code-core/src/hooks/mod.rs b/src/reloaded-code-core/src/hooks/mod.rs index bbbd286..05ec5e1 100644 --- a/src/reloaded-code-core/src/hooks/mod.rs +++ b/src/reloaded-code-core/src/hooks/mod.rs @@ -20,6 +20,13 @@ //! - [`HookRunContext`] - Context given to hook run lifecycle events //! - [`EndReason`] - Why a run ended //! +//! Run event types: +//! - [`RunEvent`] - Framework-owned event yielded by run streams +//! - [`RunMessage`] - Distilled transcript message in a completed run +//! - [`RunMessageRole`] - Author role of a transcript message +//! - [`RunToolCallSummary`] - Distilled tool call summary +//! - [`RunToolResultSummary`] - Distilled tool result summary +//! //! Observers are plain hooks: code before `original` is "start", code //! after is "end". They participate in the same hook chain. //! @@ -37,11 +44,13 @@ pub use self::builder::HookSetBuilder; pub use self::hook_set::HookSet; +pub use self::run_event::*; pub use self::run_hook::*; pub use self::tool_hook::*; mod builder; mod hook_set; +mod run_event; mod run_hook; mod tool_hook; diff --git a/src/reloaded-code-core/src/hooks/run_event/mod.rs b/src/reloaded-code-core/src/hooks/run_event/mod.rs new file mode 100644 index 0000000..4bc296a --- /dev/null +++ b/src/reloaded-code-core/src/hooks/run_event/mod.rs @@ -0,0 +1,248 @@ +//! Run event types: the framework-owned streaming item type. +//! +//! [`RunEvent`] is the item type a run stream yields. Adapters +//! translate their vendor-specific stream events into it, so consumers +//! match one stable framework-owned enum instead of vendor types. +//! +//! # Transcript distillation +//! +//! [`RunEvent::RunComplete`] carries a distilled transcript +//! ([`RunMessage`]): each message records its role, text, and tool +//! call/result summaries. The record serves display and audit; +//! consumers needing model-replay detail must use the underlying +//! agent directly. +//! +//! # Extensibility +//! +//! [`RunEvent`] is `#[non_exhaustive]`: variants may be appended +//! without a breaking release. Consumers match it with a wildcard arm. + +use serde::{Deserialize, Serialize}; + +/// Framework-owned event yielded by a run stream. +/// +/// One variant per observable streaming milestone: run start, text +/// and thinking deltas, tool activity (call start, call complete, +/// executed), output-ready, run complete, error, and cancellation. +/// +/// The enum is `#[non_exhaustive]`: variants may be appended in a +/// future release without a breaking change, so matches outside this +/// crate need a wildcard arm. +#[non_exhaustive] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub enum RunEvent { + /// The run started. + RunStart { + /// Identifier of the started run. + run_id: String, + }, + /// Incremental assistant text arrived. + TextDelta { + /// Text fragment appended since the previous delta. + text: String, + }, + /// Incremental reasoning text arrived (reasoning models). + ThinkingDelta { + /// Thinking fragment appended since the previous delta. + text: String, + }, + /// A tool call started; its arguments may still be streaming. + ToolCallStart { + /// Name of the tool being called. + tool_name: String, + /// Call id correlating this call with its completion and result. + tool_call_id: Option, + }, + /// A tool call's arguments finished streaming. + ToolCallComplete { + /// Name of the tool being called. + tool_name: String, + /// Call id correlating this call with its start and result. + tool_call_id: Option, + }, + /// A tool finished executing. + ToolExecuted { + /// Name of the tool that ran. + tool_name: String, + /// Call id of the executed call. + tool_call_id: Option, + /// Whether the tool reported success. + success: bool, + /// Error text when the tool failed. + error: Option, + }, + /// The run's final output is ready to consume. + OutputReady, + /// The run completed. + RunComplete { + /// Identifier of the completed run. + run_id: String, + /// Distilled transcript of the run. + messages: Vec, + }, + /// The run failed. + Error { + /// Human-readable description of the failure. + message: String, + }, + /// The run was cancelled. + Cancelled { + /// Partial text accumulated before cancellation. + partial_text: Option, + /// Partial thinking content accumulated before cancellation. + partial_thinking: Option, + /// Tool names whose calls were still in progress when + /// cancelled. + pending_tools: Vec, + }, +} + +/// Distilled transcript message carried by [`RunEvent::RunComplete`]. +/// +/// One participant turn: what it said ([`Self::text`]), which tools it +/// requested ([`Self::tool_calls`]), or which tool result it returns +/// ([`Self::tool_result`]). Missing fields mean the message carries no +/// content of that kind. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RunMessage { + /// Author of the message. + pub role: RunMessageRole, + /// Text content of the message, if any. + pub text: Option, + /// Tool calls the message requests, in order. + pub tool_calls: Vec, + /// Tool result the message returns, if any. + pub tool_result: Option, +} + +/// Author role of a [`RunMessage`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum RunMessageRole { + /// Framework-level instruction. + System, + /// Human input. + User, + /// Model output. + Assistant, + /// Tool output answering an assistant tool call. + Tool, +} + +/// Distilled summary of one tool call requested during a run. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RunToolCallSummary { + /// Name of the requested tool. + pub tool_name: String, + /// Call id correlating the call with its result. + pub tool_call_id: Option, + /// JSON-serialized arguments, when the call carries any. + pub arguments_json: Option, +} + +/// Distilled summary of one tool result returned during a run. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct RunToolResultSummary { + /// Call id of the tool call this result answers. + pub tool_call_id: Option, + /// Result payload rendered as text for display and audit. + pub output: String, +} + +#[cfg(test)] +mod tests { + use super::*; + + /// Full-shape transcript: every role, a tool call, and a tool result. + fn complete_transcript() -> Vec { + vec![ + RunMessage { + role: RunMessageRole::System, + text: Some("sys".into()), + tool_calls: Vec::new(), + tool_result: None, + }, + RunMessage { + role: RunMessageRole::User, + text: Some("read a.txt".into()), + tool_calls: Vec::new(), + tool_result: None, + }, + RunMessage { + role: RunMessageRole::Assistant, + text: Some("checking".into()), + tool_calls: vec![RunToolCallSummary { + tool_name: "read_file".into(), + tool_call_id: Some("call_1".into()), + arguments_json: Some(r#"{"path":"a.txt"}"#.into()), + }], + tool_result: None, + }, + RunMessage { + role: RunMessageRole::Tool, + text: None, + tool_calls: Vec::new(), + tool_result: Some(RunToolResultSummary { + tool_call_id: Some("call_1".into()), + output: "contents".into(), + }), + }, + ] + } + + #[test] + fn run_complete_serde_roundtrip_preserves_transcript() { + let event = RunEvent::RunComplete { + run_id: "run-42".into(), + messages: complete_transcript(), + }; + let json = serde_json::to_string(&event).unwrap(); + // Pin the wire shape: variants are externally tagged, so the + // variant name is the JSON object key consumers see. + assert!(json.starts_with("{\"RunComplete\":")); + let restored: RunEvent = serde_json::from_str(&json).unwrap(); + assert_eq!(restored, event); + } + + #[test] + fn run_event_variants_serde_roundtrip() { + let events = vec![ + RunEvent::RunStart { + run_id: "run-42".into(), + }, + RunEvent::TextDelta { + text: "chunk".into(), + }, + RunEvent::ThinkingDelta { + text: "thought".into(), + }, + RunEvent::ToolCallStart { + tool_name: "read_file".into(), + tool_call_id: Some("call_1".into()), + }, + RunEvent::ToolCallComplete { + tool_name: "read_file".into(), + tool_call_id: Some("call_1".into()), + }, + RunEvent::ToolExecuted { + tool_name: "read_file".into(), + tool_call_id: Some("call_1".into()), + success: false, + error: Some("missing".into()), + }, + RunEvent::OutputReady, + RunEvent::Error { + message: "boom".into(), + }, + RunEvent::Cancelled { + partial_text: Some("partial".into()), + partial_thinking: None, + pending_tools: vec!["read_file".into()], + }, + ]; + for event in events { + let json = serde_json::to_string(&event).unwrap(); + let restored: RunEvent = serde_json::from_str(&json).unwrap(); + assert_eq!(restored, event); + } + } +} From f10d937a1e6d53be8a8a9cdbb4269a30396743a1 Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Sun, 16 Aug 2026 21:50:27 +0100 Subject: [PATCH 02/10] Changed: stream RunEvent items from HookedAgent::run_stream Replace the vendor event stream and the synthetic hooked branch with a lazy, caller-driven mapping over the inner SerdesAI agent stream. - New agent_runtime/stream_events module maps each vendor stream event to the framework-owned RunEvent, distills RunComplete transcripts into the core record types, and drops vendor-only events (ContextInfo, ContextCompressed, RequestStart, ToolCallDelta, ResponseComplete); the module docs record the dropped set and its observable information loss. - run_stream accepts full UserContent prompts (image and multi-part pass through), no longer consults registered run hooks, and propagates inner failures: vendor error events map to RunEvent::Error and the inner AgentRunError still terminates the stream unchanged. - The synthetic buffered hooked branch, its text-only prompt restriction, and the write-only run-extras plumbing it fed are removed; run() dispatch, output, usage, and error restoration are unchanged and its existing tests pass unmodified. - The streaming mock emits incremental multi-delta text (16-char chunks on char boundaries) and stamps scripted tool calls with a correlation id. - RunEvent is re-exported from the adapter crate, and the serdesai-task example matches RunEvent arms only. - reloaded-code-core bumps to 0.2.1 (workspace requirement raised to match) so packaged builds resolve the new core API. --- src/Cargo.lock | 2 +- src/Cargo.toml | 2 +- src/reloaded-code-core/Cargo.toml | 2 +- .../examples/serdesai-task.rs | 150 +-- .../src/agent_runtime/mod.rs | 4 +- .../src/agent_runtime/stream_events.rs | 951 ++++++++++++++++++ .../src/agent_runtime/task.rs | 140 +-- src/reloaded-code-serdesai/src/lib.rs | 3 + src/reloaded-code-serdesai/src/mock.rs | 67 +- 9 files changed, 1081 insertions(+), 240 deletions(-) create mode 100644 src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs diff --git a/src/Cargo.lock b/src/Cargo.lock index 1d65247..d25b1da 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -2886,7 +2886,7 @@ dependencies = [ [[package]] name = "reloaded-code-core" -version = "0.2.0" +version = "0.2.1" dependencies = [ "ahash", "bitcode", diff --git a/src/Cargo.toml b/src/Cargo.toml index b213db4..2c0f84d 100644 --- a/src/Cargo.toml +++ b/src/Cargo.toml @@ -76,7 +76,7 @@ serdes-ai-models = { version = "0.2.6", default-features = false } serdes-ai-streaming = "0.2" # Internal crates -reloaded-code-core = { version = "0.2.0", path = "reloaded-code-core", default-features = false } +reloaded-code-core = { version = "0.2.1", path = "reloaded-code-core", default-features = false } reloaded-code-bubblewrap = { version = "0.1.0", path = "reloaded-code-bubblewrap" } reloaded-code-agents = { version = "0.1.0", path = "reloaded-code-agents" } reloaded-code-models-dev = { version = "0.1.0", path = "reloaded-code-models-dev" } diff --git a/src/reloaded-code-core/Cargo.toml b/src/reloaded-code-core/Cargo.toml index 711a772..49697c8 100644 --- a/src/reloaded-code-core/Cargo.toml +++ b/src/reloaded-code-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "reloaded-code-core" -version = "0.2.0" +version = "0.2.1" edition = "2021" description = "Lightweight, high-performance core types and utilities for coding tools - framework agnostic" repository = "https://github.com/Reloaded-Project/ReloadedCode" diff --git a/src/reloaded-code-serdesai/examples/serdesai-task.rs b/src/reloaded-code-serdesai/examples/serdesai-task.rs index e8fd254..46ddc72 100644 --- a/src/reloaded-code-serdesai/examples/serdesai-task.rs +++ b/src/reloaded-code-serdesai/examples/serdesai-task.rs @@ -9,10 +9,10 @@ use futures::StreamExt; use reloaded_code_agents::{AgentCatalog, AgentLoader, AgentRuntimeBuilder}; -use reloaded_code_core::{CredentialResolver, TaskInput, resolve_workspace_root}; +use reloaded_code_core::{CredentialResolver, resolve_workspace_root}; use reloaded_code_models_dev::ModelsDevCatalog; -use reloaded_code_serdesai::{AgentBuildContext, AgentDefaults}; -use serdes_ai::{AgentStreamEvent, UserContent}; +use reloaded_code_serdesai::{AgentBuildContext, AgentDefaults, RunEvent}; +use serdes_ai::UserContent; use std::{ fmt::Write, io::{self, Write as IoWrite}, @@ -26,17 +26,9 @@ const API_KEY_VALUE: &str = ""; // <-- Set your API key here const MODEL_ID: &str = "synthetic/hf:zai-org/GLM-4.7-Flash"; struct OpenStreamTag { - message_id: u32, tag: &'static str, } -struct PendingToolCall { - message_id: u32, - tool_name: String, - tool_call_id: Option, - args: String, -} - #[tokio::main] async fn main() -> Result<(), Box> { let agents_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -88,72 +80,30 @@ async fn main() -> Result<(), Box> { let prompt = UserContent::text(prompt); let prompt_text = render_user_content(&prompt); - println!("\n=== Transcript (message ids, streamed where possible) ==="); - log_xml(0, "user", &prompt_text); + println!("\n=== Transcript (streamed where possible) ==="); + log_xml("user", &prompt_text); let mut stream = agent.run_stream(prompt, ()).await?; - let mut current_message_id = 0u32; - let mut request_count = 0u32; let mut tool_call_count = 0u32; // Tracks the currently-open streaming XML tag so we can append deltas without reopening. let mut open_tag: Option = None; - let mut pending_tool_calls = Vec::with_capacity(4); while let Some(event) = stream.next().await { match event? { - AgentStreamEvent::RequestStart { step } => { - close_stream_xml(&mut open_tag); - current_message_id = step; - request_count = request_count.saturating_add(1); + RunEvent::ThinkingDelta { text } => { + write_stream_delta(&mut open_tag, "thinking", &text); } - AgentStreamEvent::ThinkingDelta { text } => { - write_stream_delta(&mut open_tag, current_message_id, "thinking", &text); + RunEvent::TextDelta { text } => { + write_stream_delta(&mut open_tag, "assistant", &text); } - AgentStreamEvent::TextDelta { text } => { - write_stream_delta(&mut open_tag, current_message_id, "assistant", &text); - } - AgentStreamEvent::ToolCallStart { - tool_name, - tool_call_id, - } => { + RunEvent::ToolCallStart { tool_name, .. } => { close_stream_xml(&mut open_tag); - log_xml(current_message_id, "tool", &tool_name); - pending_tool_calls.push(PendingToolCall { - message_id: current_message_id, - tool_name, - tool_call_id, - args: String::new(), - }); - } - AgentStreamEvent::ToolCallDelta { - delta, - tool_call_id, - } => { - // Accumulate streamed JSON args into the matching pending call. - if let Some(call) = - find_pending_tool_call_mut(&mut pending_tool_calls, tool_call_id.as_deref()) - { - call.args.push_str(&delta); - } + log_xml("tool", &tool_name); } - AgentStreamEvent::ToolCallComplete { tool_call_id, .. } => { + RunEvent::ToolCallComplete { .. } => { tool_call_count = tool_call_count.saturating_add(1); - if let Some(call) = - take_pending_tool_call(&mut pending_tool_calls, tool_call_id.as_deref()) - { - let tag = if call.tool_name == "task" { - "task-input" - } else { - "tool-input" - }; - let content = render_tool_input(&call.tool_name, &call.args); - log_xml(call.message_id, tag, &content); - } } - AgentStreamEvent::ResponseComplete { .. } => { - close_stream_xml(&mut open_tag); - } - AgentStreamEvent::RunComplete { .. } => { + RunEvent::RunComplete { .. } => { close_stream_xml(&mut open_tag); } _ => {} @@ -162,52 +112,25 @@ async fn main() -> Result<(), Box> { close_stream_xml(&mut open_tag); - println!( - "Root agent activity: {} model requests, {} tool calls", - request_count, tool_call_count - ); + println!("Root agent activity: {tool_call_count} tool calls"); Ok(()) } -fn find_pending_tool_call_mut<'a>( - pending: &'a mut [PendingToolCall], - tool_call_id: Option<&str>, -) -> Option<&'a mut PendingToolCall> { - // Most providers include a tool_call_id; fall back to the last pending call otherwise. - match tool_call_id { - Some(tool_call_id) => pending - .iter_mut() - .rev() - .find(|call| call.tool_call_id.as_deref() == Some(tool_call_id)), - None => pending.last_mut(), - } -} - -fn log_xml(message_id: u32, tag: &str, content: &str) { +fn log_xml(tag: &str, content: &str) { // Long or multiline content gets block-style tags; short content fits on one line. if content.contains('\n') || content.len() > 120 { - println!(""); + println!("<{tag}>"); println!("{content}"); println!(""); return; } let mut line = String::with_capacity(content.len() + tag.len() * 2 + 18); - let _ = write!(line, "{content}"); + let _ = write!(line, "<{tag}>{content}"); println!("{line}"); } -fn render_tool_input(tool_name: &str, args_text: &str) -> String { - match serde_json::from_str::(args_text) { - Ok(args) if tool_name == "task" => render_task_input(&args), - Ok(args) => { - serde_json::to_string_pretty(&args).expect("tool args serialization should succeed") - } - Err(_) => args_text.to_string(), - } -} - fn render_user_content(content: &UserContent) -> String { match content { UserContent::Text(text) => text.clone(), @@ -216,37 +139,17 @@ fn render_user_content(content: &UserContent) -> String { } } -fn take_pending_tool_call( - pending: &mut Vec, - tool_call_id: Option<&str>, -) -> Option { - let index = match tool_call_id { - Some(tool_call_id) => pending - .iter() - .rposition(|call| call.tool_call_id.as_deref() == Some(tool_call_id)), - None => pending.len().checked_sub(1), - }?; - Some(pending.remove(index)) -} - -fn write_stream_delta( - open_tag: &mut Option, - message_id: u32, - tag: &'static str, - text: &str, -) { +fn write_stream_delta(open_tag: &mut Option, tag: &'static str, text: &str) { if text.is_empty() { return; } - // If the message or tag changed, close the previous open tag and start a new one. - let is_same = open_tag - .as_ref() - .is_some_and(|t| t.message_id == message_id && t.tag == tag); + // If the tag changed, close the previous open tag and start a new one. + let is_same = open_tag.as_ref().is_some_and(|t| t.tag == tag); if !is_same { close_stream_xml(open_tag); - println!(""); - *open_tag = Some(OpenStreamTag { message_id, tag }); + println!("<{tag}>"); + *open_tag = Some(OpenStreamTag { tag }); } print!("{text}"); @@ -259,12 +162,3 @@ fn close_stream_xml(open_tag: &mut Option) { println!("", tag.tag); } } - -fn render_task_input(args: &serde_json::Value) -> String { - // Try to decode into the typed TaskInput shape; fall back to raw JSON. - serde_json::from_value::(args.clone()) - .and_then(|input| serde_json::to_string_pretty(&input)) - .unwrap_or_else(|_| { - serde_json::to_string_pretty(args).expect("task args serialization should succeed") - }) -} diff --git a/src/reloaded-code-serdesai/src/agent_runtime/mod.rs b/src/reloaded-code-serdesai/src/agent_runtime/mod.rs index b128cf5..f1932ac 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/mod.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/mod.rs @@ -6,7 +6,8 @@ //! //! # Public API //! - [`AgentBuildContext`] - Shared context that builds runnable agents by name. -//! - [`HookedAgent`] - Built agent wrapper that dispatches through run hooks. +//! - [`HookedAgent`] - Built agent wrapper that dispatches `run()` through run +//! hooks and streams framework-owned events from `run_stream()`. //! - [`AgentBuildError`] - Build-time failures. pub use build::AgentBuildError; @@ -20,6 +21,7 @@ pub(crate) use task::{TaskBuildContext, build_agent}; mod build; mod model; mod provider_bridge; +mod stream_events; mod task; #[cfg(test)] pub(crate) mod test_stubs; diff --git a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs new file mode 100644 index 0000000..e1b9f32 --- /dev/null +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -0,0 +1,951 @@ +//! Vendor stream events mapped to framework-owned run events. +//! +//! [`RunEventStream`] wraps the vendor [`AgentStream`] and lazily maps each +//! [`AgentStreamEvent`] to a [`RunEvent`] as the consumer polls it. SerdesAI +//! event types stay inside this module; consumers of +//! [`HookedAgent::run_stream`][task] only ever see [`RunEvent`] items. +//! +//! # Dropped vendor-only events +//! +//! The vendor emits five events with no [`RunEvent`] counterpart; the +//! mapping drops them: +//! +//! - `ContextInfo` and `ContextCompressed`: context-size telemetry emitted +//! before each model request, and compression notices. Context metrics +//! are not surfaced anywhere else. +//! - `RequestStart` and `ResponseComplete`: model-request step boundaries. +//! Dropping them removes the step index consumers could use to group +//! events by model request. +//! - `ToolCallDelta`: incremental tool-call argument fragments. Dropping it +//! removes streamed argument assembly; complete arguments remain +//! available in the [`RunEvent::RunComplete`] transcript as +//! [`RunToolCallSummary::arguments_json`]. +//! +//! Observable information loss: step boundaries and incremental tool-call +//! arguments no longer stream, and context telemetry is not surfaced. +//! +//! [task]: super::task::HookedAgent::run_stream + +use futures::{Stream, StreamExt}; +use reloaded_code_core::hooks::{ + RunEvent, RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary, +}; +use serdes_ai::core::{ + ModelRequest, ModelRequestPart, ModelResponse, ModelResponsePart, UserContent, +}; +use serdes_ai::{AgentStream, AgentStreamEvent}; +use std::pin::Pin; +use std::task::{Context, Poll}; + +/// Lazy caller-driven [`Stream`] mapping a vendor [`AgentStream`] to +/// framework-owned [`RunEvent`] items. +/// +/// Polling this stream drives the vendor stream, which the vendor already +/// runs on its own background task; no channel, spawn, or shared agent +/// handle is added on this side. Vendor-only events (see the module docs) +/// are dropped rather than yielded. +pub(super) struct RunEventStream { + /// Owned vendor stream, driven by the vendor's own background task. + inner: AgentStream, +} + +impl RunEventStream { + /// Wraps an already-started vendor stream. + pub(super) fn new(inner: AgentStream) -> Self { + Self { inner } + } +} + +impl Stream for RunEventStream { + type Item = Result; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + // Dropped events yield nothing, so keep polling until a mappable + // event, an error, or the stream's end arrives. + let inner = &mut self.get_mut().inner; + loop { + match inner.poll_next_unpin(cx) { + Poll::Ready(Some(Ok(event))) => { + if let Some(event) = map_vendor_event(event) { + return Poll::Ready(Some(Ok(event))); + } + } + Poll::Ready(Some(Err(error))) => return Poll::Ready(Some(Err(error))), + Poll::Ready(None) => return Poll::Ready(None), + Poll::Pending => return Poll::Pending, + } + } + } +} + +/// Maps one vendor event to its framework-owned counterpart. +/// +/// Returns `None` for vendor-only events; see the module docs for the +/// dropped set and its observable information loss. +fn map_vendor_event(event: AgentStreamEvent) -> Option { + Some(match event { + AgentStreamEvent::RunStart { run_id } => RunEvent::RunStart { run_id }, + AgentStreamEvent::TextDelta { text } => RunEvent::TextDelta { text }, + AgentStreamEvent::ThinkingDelta { text } => RunEvent::ThinkingDelta { text }, + AgentStreamEvent::ToolCallStart { + tool_name, + tool_call_id, + } => RunEvent::ToolCallStart { + tool_name, + tool_call_id, + }, + AgentStreamEvent::ToolCallComplete { + tool_name, + tool_call_id, + } => RunEvent::ToolCallComplete { + tool_name, + tool_call_id, + }, + AgentStreamEvent::ToolExecuted { + tool_name, + tool_call_id, + success, + error, + } => RunEvent::ToolExecuted { + tool_name, + tool_call_id, + success, + error, + }, + AgentStreamEvent::OutputReady => RunEvent::OutputReady, + AgentStreamEvent::RunComplete { run_id, messages } => RunEvent::RunComplete { + run_id, + messages: distill_messages(messages), + }, + AgentStreamEvent::Error { message } => RunEvent::Error { message }, + AgentStreamEvent::Cancelled { + partial_text, + partial_thinking, + pending_tools, + } => RunEvent::Cancelled { + partial_text, + partial_thinking, + pending_tools, + }, + AgentStreamEvent::ContextInfo { .. } + | AgentStreamEvent::ContextCompressed { .. } + | AgentStreamEvent::RequestStart { .. } + | AgentStreamEvent::ToolCallDelta { .. } + | AgentStreamEvent::ResponseComplete { .. } => return None, + }) +} + +/// Distills the vendor run transcript into framework-owned records. +/// +/// One [`RunMessage`] per vendor part: system prompts, user prompts, and +/// retry feedback map to their authoring roles; tool returns map to +/// [`RunMessageRole::Tool`] with a [`RunToolResultSummary`]; model +/// responses map to [`RunMessageRole::Assistant`] with joined text plus +/// [`RunToolCallSummary`] entries. Thinking and file parts carry no +/// distilled representation and are skipped. +fn distill_messages(messages: Vec) -> Vec { + // Upper bound: one distilled message per part across all requests. + let part_count = messages.iter().map(|message| message.parts.len()).sum(); + let mut distilled = Vec::with_capacity(part_count); + for message in messages { + for part in message.parts { + match part { + ModelRequestPart::SystemPrompt(part) => { + distilled.push(authored_message(RunMessageRole::System, part.content)); + } + ModelRequestPart::UserPrompt(part) => { + distilled.push(authored_message( + RunMessageRole::User, + user_content_text(part.content), + )); + } + ModelRequestPart::RetryPrompt(part) => { + distilled.push(authored_message( + RunMessageRole::User, + part.content.message().to_owned(), + )); + } + ModelRequestPart::ToolReturn(part) => { + distilled.push(tool_result_message( + part.tool_call_id, + part.content.to_string_content(), + )); + } + ModelRequestPart::BuiltinToolReturn(part) => { + // Structured content (search results, code output) has + // no text projection; serialize it so the audit trail + // keeps it. + let output = serde_json::to_string(&part.content) + .unwrap_or_else(|_| format!("{:?}", part.content)); + distilled.push(tool_result_message(Some(part.tool_call_id), output)); + } + ModelRequestPart::ModelResponse(response) => { + if let Some(message) = distill_model_response(*response) { + distilled.push(message); + } + } + } + } + } + distilled +} + +/// Builds a text-only [`RunMessage`] for an authoring-side part. +fn authored_message(role: RunMessageRole, text: String) -> RunMessage { + RunMessage { + role, + text: Some(text), + tool_calls: Vec::new(), + tool_result: None, + } +} + +/// Distills one assistant model response. +/// +/// Returns `None` when the response carries no distilled content +/// (thinking-only, file-only, or empty). +fn distill_model_response(response: ModelResponse) -> Option { + let text = response.text_content(); + let text = (!text.is_empty()).then_some(text); + let mut tool_calls = Vec::new(); + for part in &response.parts { + if let ModelResponsePart::ToolCall(call) = part { + tool_calls.push(RunToolCallSummary { + tool_name: call.tool_name.clone(), + tool_call_id: call.tool_call_id.clone(), + arguments_json: call.args.to_json_string().ok(), + }); + } + } + if text.is_none() && tool_calls.is_empty() { + return None; + } + Some(RunMessage { + role: RunMessageRole::Assistant, + text, + tool_calls, + tool_result: None, + }) +} + +/// Builds a [`RunMessageRole::Tool`] record answering a tool call. +fn tool_result_message(tool_call_id: Option, output: String) -> RunMessage { + RunMessage { + role: RunMessageRole::Tool, + text: None, + tool_calls: Vec::new(), + tool_result: Some(RunToolResultSummary { + tool_call_id, + output, + }), + } +} + +/// Renders user prompt content as audit text. +/// +/// Plain text passes through; multi-part prompts are serialized so image +/// and mixed content stay observable in the transcript. +fn user_content_text(content: UserContent) -> String { + match content { + UserContent::Text(text) => text, + UserContent::Parts(parts) => { + serde_json::to_string(&parts).unwrap_or_else(|_| format!("{parts:?}")) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::agent_runtime::AgentBuildContext; + use crate::agent_runtime::task::HookedAgent; + use crate::agent_runtime::test_stubs::{ + SerdesTestFactory, agent, allow_tools, catalog, credentials, workspace_root, + }; + use crate::mock::{FunctionModel, Streamed, tool_then_text}; + use futures::StreamExt; + use reloaded_code_agents::{AgentCatalog, AgentDefaults, AgentMode, AgentRuntimeBuilder}; + use reloaded_code_core::hooks::{ + HookRunContext, HookSet, RunConfig, RunHook, RunHookFuture, RunOriginal, + }; + use reloaded_code_core::{ToolCatalogEntry, ToolCatalogKind}; + use serde_json::json; + use serdes_ai::core::messages::request::RetryPromptPart; + use serdes_ai::core::{ + BuiltinToolReturnContent, BuiltinToolReturnPart, SystemPromptPart, ToolReturnPart, + UserPromptPart, + }; + use serdes_ai_models::{ModelError, ModelProfile}; + use std::sync::{Arc, Mutex}; + + // ======================================================================== + // Vendor event mapping + // ======================================================================== + + #[test] + fn map_vendor_event_drops_vendor_only_variants() { + let dropped = vec![ + AgentStreamEvent::ContextInfo { + estimated_tokens: 1, + request_bytes: 4, + context_limit: None, + }, + AgentStreamEvent::ContextCompressed { + original_tokens: 10, + compressed_tokens: 5, + strategy: "truncate".into(), + messages_before: 2, + messages_after: 1, + }, + AgentStreamEvent::RequestStart { step: 1 }, + AgentStreamEvent::ToolCallDelta { + delta: "{\"a\":".into(), + tool_call_id: Some("call_1".into()), + }, + AgentStreamEvent::ResponseComplete { step: 1 }, + ]; + for event in dropped { + assert!( + map_vendor_event(event).is_none(), + "vendor-only event should be dropped" + ); + } + } + + #[test] + fn map_vendor_event_preserves_mapped_variant_payloads() { + let Some(RunEvent::Error { message }) = map_vendor_event(AgentStreamEvent::Error { + message: "boom".into(), + }) else { + panic!("error event should map"); + }; + assert_eq!(message, "boom"); + + let Some(RunEvent::Cancelled { + partial_text, + partial_thinking, + pending_tools, + }) = map_vendor_event(AgentStreamEvent::Cancelled { + partial_text: Some("partial".into()), + partial_thinking: None, + pending_tools: vec!["read".into()], + }) + else { + panic!("cancelled event should map"); + }; + assert_eq!(partial_text.as_deref(), Some("partial")); + assert!(partial_thinking.is_none()); + assert_eq!(pending_tools, vec!["read".to_string()]); + + // A mislabel of the thinking arm (both sides are bare text records) + // would compile, so pin the variant identity here. + let Some(RunEvent::ThinkingDelta { text }) = + map_vendor_event(AgentStreamEvent::ThinkingDelta { text: "hmm".into() }) + else { + panic!("thinking delta should map to the thinking variant"); + }; + assert_eq!(text, "hmm"); + + // The call-start and call-complete arms share one field shape, so + // pin each variant identity and its id separately. + let Some(RunEvent::ToolCallStart { + tool_name, + tool_call_id, + }) = map_vendor_event(AgentStreamEvent::ToolCallStart { + tool_name: "read".into(), + tool_call_id: Some("call_1".into()), + }) + else { + panic!("tool call start should map to the start variant"); + }; + assert_eq!( + (tool_name.as_str(), tool_call_id.as_deref()), + ("read", Some("call_1")) + ); + + let Some(RunEvent::ToolCallComplete { + tool_name, + tool_call_id, + }) = map_vendor_event(AgentStreamEvent::ToolCallComplete { + tool_name: "read".into(), + tool_call_id: Some("call_1".into()), + }) + else { + panic!("tool call complete should map to the complete variant"); + }; + assert_eq!( + (tool_name.as_str(), tool_call_id.as_deref()), + ("read", Some("call_1")) + ); + + let Some(RunEvent::ToolExecuted { + tool_name, + tool_call_id, + success, + error, + }) = map_vendor_event(AgentStreamEvent::ToolExecuted { + tool_name: "read".into(), + tool_call_id: Some("call_1".into()), + success: false, + error: Some("missing".into()), + }) + else { + panic!("tool executed event should map"); + }; + assert_eq!(tool_name, "read"); + assert_eq!(tool_call_id.as_deref(), Some("call_1")); + assert!(!success); + assert_eq!(error.as_deref(), Some("missing")); + } + + // ======================================================================== + // Transcript distillation + // ======================================================================== + + #[test] + fn distill_messages_renders_tool_flow_transcript() { + let response = ModelResponse::with_parts(vec![ + ModelResponsePart::text("checking"), + ModelResponsePart::tool_call("read_file", json!({"path": "a.txt"})), + ]); + let request = ModelRequest::with_parts(vec![ + ModelRequestPart::SystemPrompt(SystemPromptPart::new("sys")), + ModelRequestPart::UserPrompt(UserPromptPart::new("read a.txt")), + ModelRequestPart::ModelResponse(Box::new(response)), + ModelRequestPart::ToolReturn( + ToolReturnPart::success("read_file", "contents").with_tool_call_id("call_1"), + ), + ModelRequestPart::RetryPrompt(RetryPromptPart::new("retry feedback")), + ModelRequestPart::BuiltinToolReturn(BuiltinToolReturnPart::new( + "web_search", + BuiltinToolReturnContent::Other { + kind: "custom".into(), + data: json!({"hits": 1}), + }, + "call_9", + )), + ]); + + let messages = distill_messages(vec![request]); + + let expected = vec![ + RunMessage { + role: RunMessageRole::System, + text: Some("sys".into()), + tool_calls: Vec::new(), + tool_result: None, + }, + RunMessage { + role: RunMessageRole::User, + text: Some("read a.txt".into()), + tool_calls: Vec::new(), + tool_result: None, + }, + RunMessage { + role: RunMessageRole::Assistant, + text: Some("checking".into()), + tool_calls: vec![RunToolCallSummary { + tool_name: "read_file".into(), + tool_call_id: None, + arguments_json: Some(r#"{"path":"a.txt"}"#.into()), + }], + tool_result: None, + }, + RunMessage { + role: RunMessageRole::Tool, + text: None, + tool_calls: Vec::new(), + tool_result: Some(RunToolResultSummary { + tool_call_id: Some("call_1".into()), + output: "contents".into(), + }), + }, + RunMessage { + role: RunMessageRole::User, + text: Some("retry feedback".into()), + tool_calls: Vec::new(), + tool_result: None, + }, + ]; + assert_eq!(messages.len(), 6); + assert_eq!(messages[..5], expected[..]); + + // The builtin tool return distills as a Tool turn whose output is + // the serialized structured content; the exact vendor tag shape + // stays unpinned while the payload stays observable. + let builtin = messages[5] + .tool_result + .as_ref() + .expect("builtin tool result should distill"); + assert_eq!(messages[5].role, RunMessageRole::Tool); + assert_eq!(builtin.tool_call_id, Some("call_9".to_string())); + assert!( + builtin.output.contains("custom") && builtin.output.contains("hits"), + "serialized builtin content should stay observable: {}", + builtin.output + ); + } + + #[test] + fn distill_messages_serializes_multipart_user_prompts() { + let request = ModelRequest::with_parts(vec![ModelRequestPart::UserPrompt( + UserPromptPart::new(UserContent::Parts(vec![ + serdes_ai::core::UserContentPart::text("part one"), + serdes_ai::core::UserContentPart::text("part two"), + ])), + )]); + + let messages = distill_messages(vec![request]); + + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].role, RunMessageRole::User); + let text = messages[0] + .text + .as_deref() + .expect("multipart prompt should render as text"); + assert!( + text.contains("part one") && text.contains("part two"), + "serialized parts should stay observable: {text}" + ); + } + + #[test] + fn distill_messages_skips_responses_without_distillable_content() { + let response = ModelResponse::with_parts(vec![ModelResponsePart::thinking("internal")]); + let request = + ModelRequest::with_parts(vec![ModelRequestPart::ModelResponse(Box::new(response))]); + + assert!(distill_messages(vec![request]).is_empty()); + } + + // ======================================================================== + // HookedAgent::run_stream integration + // ======================================================================== + + /// Run hook that records every dispatch it observes. Clones share the + /// record so the test can register one copy and inspect another. + #[derive(Clone)] + struct DispatchRecorder { + dispatches: Arc>>, + } + + impl RunHook for DispatchRecorder { + fn hook<'a>( + &'a self, + ctx: &'a HookRunContext<'a>, + config: RunConfig, + original: RunOriginal<'a>, + ) -> RunHookFuture<'a> { + self.dispatches + .lock() + .expect("dispatches should not be poisoned") + .push(ctx.run_id.to_string()); + original.call(ctx, config) + } + } + + /// Model whose streaming requests fail before yielding any events. + struct FailingStreamModel { + profile: ModelProfile, + } + + impl FailingStreamModel { + fn new() -> Self { + Self { + profile: ModelProfile::default(), + } + } + } + + #[async_trait::async_trait] + impl serdes_ai_models::Model for FailingStreamModel { + fn name(&self) -> &str { + "failing-stream-model" + } + + fn system(&self) -> &str { + "test" + } + + fn profile(&self) -> &ModelProfile { + &self.profile + } + + async fn request( + &self, + _messages: &[ModelRequest], + _settings: &serdes_ai::core::ModelSettings, + _params: &serdes_ai_models::ModelRequestParameters, + ) -> Result { + Err(ModelError::api("upstream exploded")) + } + + async fn request_stream( + &self, + _messages: &[ModelRequest], + _settings: &serdes_ai::core::ModelSettings, + _params: &serdes_ai_models::ModelRequestParameters, + ) -> Result { + Err(ModelError::api("upstream exploded")) + } + } + + /// Builds a hooked `caller` agent with no tools, running `model`. + fn streamed_agent( + model: impl serdes_ai_models::Model + 'static, + hooks: HookSet, + ) -> HookedAgent { + let runtime = AgentRuntimeBuilder::new() + .catalog(AgentCatalog::from_entries([agent( + "caller", + AgentMode::Primary, + allow_tools(&[]), + "prompt", + )])) + .defaults(AgentDefaults::with_model("openrouter/openai/gpt-4.1-mini")) + .hooks(hooks) + .build() + .expect("runtime should build"); + + let context = AgentBuildContext::new( + Arc::new(runtime), + Arc::new(catalog()), + Arc::new(credentials()), + workspace_root(), + ) + .with_model_override(model); + context.build("caller").expect("build should succeed") + } + + /// Builds a hooked `caller` agent whose only tool is a custom `ping` + /// tool returning "pong", running `model`. + fn streamed_agent_with_ping_tool( + model: impl serdes_ai_models::Model + 'static, + hooks: HookSet, + ) -> HookedAgent { + let runtime = AgentRuntimeBuilder::new() + .catalog(AgentCatalog::from_entries([agent( + "caller", + AgentMode::Primary, + allow_tools(&["ping"]), + "prompt", + )])) + .tools(vec![ToolCatalogEntry::new("ping", ToolCatalogKind::Custom)]) + .custom_tool(SerdesTestFactory::new( + "ping", + "Use ping to check connectivity.", + "pong", + )) + .defaults(AgentDefaults::with_model("openrouter/openai/gpt-4.1-mini")) + .hooks(hooks) + .build() + .expect("runtime should build"); + + let context = AgentBuildContext::new( + Arc::new(runtime), + Arc::new(catalog()), + Arc::new(credentials()), + workspace_root(), + ) + .with_model_override(model); + context.build("caller").expect("build should succeed") + } + + /// Collects every event of one `run_stream` call into owned events. + async fn collect_events(agent: &HookedAgent, prompt: impl Into) -> Vec { + let mut stream = agent + .run_stream(prompt, ()) + .await + .expect("stream should start"); + let mut events = Vec::new(); + while let Some(event) = stream.next().await { + events.push(event.expect("stream item should be ok")); + } + events + } + + #[tokio::test] + async fn run_stream_yields_incremental_text_deltas_with_run_hooks_registered() { + const RESPONSE: &str = "ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"; + let model = Streamed::new(FunctionModel::new(move |_, _| { + ModelResponse::text(RESPONSE) + })); + let recorder = DispatchRecorder { + dispatches: Arc::new(Mutex::new(Vec::new())), + }; + let hooks = HookSet::builder().run_hook(recorder.clone()).build(); + let hooked = streamed_agent(model, hooks); + + let events = collect_events(&hooked, "hello").await; + + let delta_count = events + .iter() + .filter(|event| matches!(event, RunEvent::TextDelta { .. })) + .count(); + let first_delta = events + .iter() + .position(|event| matches!(event, RunEvent::TextDelta { .. })); + let complete_index = events + .iter() + .position(|event| matches!(event, RunEvent::RunComplete { .. })) + .expect("run should complete"); + assert!( + delta_count > 1, + "expected multiple incremental text deltas, got {delta_count}" + ); + assert!( + first_delta < Some(complete_index), + "a text delta must arrive before run completion" + ); + + // Registered run hooks stay inert on the streaming path. + assert!( + recorder + .dispatches + .lock() + .expect("recorder not poisoned") + .is_empty(), + "run hooks must not fire on run_stream" + ); + + // Concatenated deltas equal the model's response text. + let streamed_text: String = events + .iter() + .filter_map(|event| match event { + RunEvent::TextDelta { text } => Some(text.as_str()), + _ => None, + }) + .collect(); + assert_eq!(streamed_text, RESPONSE); + } + + #[tokio::test] + async fn run_stream_run_complete_carries_real_run_id_and_faithful_transcript() { + // The model echoes the last user prompt so the transcript ties to + // the prompt text. + let model = Streamed::new(FunctionModel::new(|messages, _| { + let prompt = messages + .iter() + .rev() + .flat_map(|message| message.user_prompts()) + .next() + .and_then(|prompt| prompt.content.as_text()) + .unwrap_or_default() + .to_string(); + ModelResponse::text(format!("echo: {prompt}")) + })); + let hooked = streamed_agent(model, HookSet::builder().build()); + + let events = collect_events(&hooked, "transcript probe").await; + + let mut start_run_id = None; + let mut streamed_text = String::new(); + let mut complete = None; + for event in events { + match event { + RunEvent::RunStart { run_id } => start_run_id = Some(run_id), + RunEvent::TextDelta { text } => streamed_text.push_str(&text), + RunEvent::RunComplete { run_id, messages } => { + complete = Some((run_id, messages)); + } + _ => {} + } + } + let (run_id, messages) = complete.expect("run should complete"); + assert!( + !run_id.is_empty(), + "RunComplete must carry the inner run id" + ); + assert_eq!( + start_run_id.as_deref(), + Some(run_id.as_str()), + "RunComplete id must match the started run" + ); + + // Transcript consistency: the user turn carries the prompt, the + // assistant turn carries exactly what was streamed. + let user_text = messages + .iter() + .find(|message| message.role == RunMessageRole::User) + .and_then(|message| message.text.as_deref()); + assert_eq!(user_text, Some("transcript probe")); + let assistant_text = messages + .iter() + .find(|message| message.role == RunMessageRole::Assistant) + .and_then(|message| message.text.as_deref()); + assert_eq!(assistant_text, Some(streamed_text.as_str())); + } + + #[tokio::test] + async fn run_stream_accepts_multipart_prompt() { + let model = Streamed::new(FunctionModel::new(|_, _| ModelResponse::text("handled"))); + let hooked = streamed_agent(model, HookSet::builder().build()); + + let prompt = UserContent::Parts(vec![ + serdes_ai::core::UserContentPart::text("describe this"), + serdes_ai::core::UserContentPart::image_url("https://example.invalid/image.png"), + ]); + + let events = collect_events(&hooked, prompt).await; + + assert!( + events + .iter() + .any(|event| matches!(event, RunEvent::RunComplete { .. })), + "multipart prompt should stream to completion" + ); + } + + #[tokio::test] + async fn run_stream_reports_tool_activity_consistent_with_transcript() { + // Scripted flow: the first turn calls `ping`, then the final turn + // answers with text once the tool return is in the history. + let model = tool_then_text("ping", json!({"target": "example.com"}), "after the tool"); + let hooked = streamed_agent_with_ping_tool(model, HookSet::builder().build()); + + let events = collect_events(&hooked, "use the tool").await; + + let position = |predicate: &dyn Fn(&RunEvent) -> bool| { + events.iter().position(|event| predicate(event)) + }; + let call_start = position(&|event| { + matches!(event, RunEvent::ToolCallStart { tool_name, .. } if tool_name == "ping") + }) + .expect("tool call start should stream"); + // The scripted mock stamps the call id; every later event and + // transcript record must correlate on it. + let streamed_call_id = match &events[call_start] { + RunEvent::ToolCallStart { tool_call_id, .. } => tool_call_id.clone(), + other => panic!("expected a tool call start, got {other:?}"), + }; + assert_eq!( + streamed_call_id.as_deref(), + Some("call_mock"), + "the scripted call id must stream through the start event" + ); + let call_complete = position(&|event| { + matches!(event, RunEvent::ToolCallComplete { tool_call_id, .. } + if tool_call_id == &streamed_call_id) + }) + .expect("tool call complete should stream"); + let executed = position(&|event| { + matches!( + event, + RunEvent::ToolExecuted { tool_name, tool_call_id, success, error: None, .. } + if tool_name == "ping" && *success && tool_call_id == &streamed_call_id + ) + }) + .expect("successful tool execution should stream"); + assert!( + call_start < call_complete && call_complete < executed, + "tool events must stream in call-start, call-complete, executed order" + ); + + // Output-ready arrives after the final text but before completion. + let output_ready = position(&|event| matches!(event, RunEvent::OutputReady)) + .expect("output-ready should stream"); + let complete = position(&|event| matches!(event, RunEvent::RunComplete { .. })) + .expect("run should complete"); + assert_eq!( + complete, + events.len() - 1, + "RunComplete should be the last event" + ); + assert!( + output_ready < complete, + "output-ready must precede run completion" + ); + + // The final answer streams as deltas after the tool ran. + let streamed_answer: String = events[executed + 1..] + .iter() + .filter_map(|event| match event { + RunEvent::TextDelta { text } => Some(text.as_str()), + _ => None, + }) + .collect(); + + // Transcript consistency with the streamed tool activity: the + // assistant turn records the call and arguments, the tool turn + // records the real result, and the closing assistant turn matches + // the streamed answer. + let (run_id, messages) = match events.last() { + Some(RunEvent::RunComplete { run_id, messages }) => (run_id, messages), + other => panic!("RunComplete should be the last event, got {other:?}"), + }; + assert!( + events.iter().any(|event| matches!( + event, + RunEvent::RunStart { run_id: started } if started == run_id + )), + "RunComplete id must match the started run" + ); + let call_summary = messages + .iter() + .flat_map(|message| &message.tool_calls) + .find(|call| call.tool_name == "ping") + .expect("transcript should record the ping call"); + assert_eq!( + call_summary.arguments_json.as_deref(), + Some(r#"{"target":"example.com"}"#) + ); + assert_eq!( + call_summary.tool_call_id, streamed_call_id, + "transcript call summary must carry the streamed call id" + ); + let tool_output = messages + .iter() + .find(|message| message.role == RunMessageRole::Tool) + .and_then(|message| message.tool_result.as_ref()) + .expect("transcript should record the tool result"); + assert_eq!(tool_output.output, "pong"); + assert_eq!( + tool_output.tool_call_id, streamed_call_id, + "transcript tool result must answer the streamed call id" + ); + let final_answer = messages + .iter() + .filter(|message| message.role == RunMessageRole::Assistant) + .filter_map(|message| message.text.as_deref()) + .last() + .expect("closing assistant turn should carry text"); + assert_eq!(final_answer, streamed_answer); + } + + #[tokio::test] + async fn run_stream_maps_vendor_error_event_and_ends_with_inner_error() { + let hooked = streamed_agent(FailingStreamModel::new(), HookSet::builder().build()); + + let mut stream = hooked + .run_stream("trigger the failure", ()) + .await + .expect("stream should start"); + let mut mapped_error_message = None; + let mut terminal = None; + while let Some(item) = stream.next().await { + match item { + Ok(RunEvent::Error { message }) => mapped_error_message = Some(message), + other => terminal = Some(other), + } + } + + // The vendor failure surfaces as the mapped error event first... + let message = + mapped_error_message.expect("vendor failure should surface as RunEvent::Error"); + assert!( + message.contains("upstream exploded"), + "mapped error should carry the vendor message: {message}" + ); + // ...then the inner error terminates the stream unchanged. + let error = terminal + .expect("stream should end with the inner error") + .expect_err("terminal item should be the inner error"); + assert!( + matches!( + error, + serdes_ai::agent::AgentRunError::Model(ModelError::Api { .. }) + ), + "inner model failure should keep its variant, got: {error:?}" + ); + } +} diff --git a/src/reloaded-code-serdesai/src/agent_runtime/task.rs b/src/reloaded-code-serdesai/src/agent_runtime/task.rs index 39e04f9..62b43fa 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/task.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/task.rs @@ -2,22 +2,23 @@ //! //! # Public API //! - [`AgentBuildContext`] - Reusable shared inputs for building runnable agents. -//! - [`HookedAgent`] - Built agent wrapper that dispatches through run hooks. +//! - [`HookedAgent`] - Built agent wrapper that dispatches `run()` through run +//! hooks and streams framework-owned events from `run_stream()`. #[cfg(not(all(feature = "linux-bubblewrap", target_os = "linux")))] use super::build::Profile; use super::build::{AgentBuildError, attach_standard_tools, prepare_build}; +use super::stream_events::RunEventStream; use crate::task::TaskHandle; use futures::Stream; use reloaded_code_agents::AgentRuntime; #[cfg(all(feature = "linux-bubblewrap", target_os = "linux"))] use reloaded_code_bubblewrap::{CreateSandboxError, Preset, Profile, TempSandboxDirs}; use reloaded_code_core::hooks::{ - EndReason, HookRunContext, HookSet, ModelSettingsOverrides, PreambleRole, RunConfig, + EndReason, HookRunContext, HookSet, ModelSettingsOverrides, PreambleRole, RunConfig, RunEvent, RunExecutor, RunHookFuture, RunOutput, RunUsage, }; use reloaded_code_core::{CredentialLookup, CredentialResolver, models::ModelCatalog}; -use serdes_ai::core::ModelRequest; use serdes_ai::{Agent, AgentBuilder, RunOptions}; #[cfg(any(test, feature = "mock"))] use serdes_ai_models::BoxedModel; @@ -36,10 +37,13 @@ pub struct AgentBuildContext>, } -/// Lightweight newtype around a built SerdesAI `Agent` that dispatches -/// `run()` and `run_stream()` through the core `HookSet::dispatch_run` hook -/// chain when run hooks are registered. Passes through directly when no hooks -/// are present for zero overhead. +/// Lightweight newtype around a built SerdesAI `Agent`. +/// +/// `run()` dispatches through the core `HookSet::dispatch_run` hook chain +/// when run hooks are registered and passes through directly otherwise. +/// `run_stream()` streams framework-owned [`RunEvent`]s lazily mapped from +/// the vendor stream; registered run hooks are not consulted on the +/// streaming path, so run-hook config injection applies to `run()` only. pub struct HookedAgent { inner: Agent<(), String>, hooks: HookSet, @@ -63,7 +67,7 @@ pub struct HookedAgentRunResult { /// /// On inner failure the hook chain sees a [`ToolError::Execution`] projection /// while the original [`AgentRunError`] is parked in `error`; the dispatch -/// site in `run_with_extras` restores the original when the failure reaches +/// site in `run_hooked` restores the original when the failure reaches /// the caller untouched. /// /// [`ToolError::Execution`]: reloaded_code_core::ToolError::Execution @@ -72,24 +76,10 @@ struct SerdesRunExecutor<'a> { agent: &'a Agent<(), String>, prompt: String, deps: (), - /// Slot the executor fills with the inner response's run metadata. - extras: Arc>>, /// Slot the executor fills with the inner run's original failure. error: Arc>>, } -/// Inner-agent run metadata captured alongside the output so streaming -/// callers can emit a faithful `RunComplete` event. -#[derive(Default)] -struct AgentRunExtras { - /// Run identifier assigned by the inner agent, or the wrapper's - /// identifier when a hook replaced the run without calling `original`. - run_id: String, - /// Complete message history from the inner run; empty when a hook - /// replaced the run without calling `original`. - messages: Vec, -} - /// Shared owned state for builds that may happen later during Task delegation. #[derive(Clone)] pub(crate) struct TaskBuildContext @@ -351,15 +341,10 @@ impl HookedAgent { prompt: impl Into, deps: (), ) -> Result { - let (result, _extras) = self.run_with_extras(prompt.into(), deps).await?; - Ok(result) + self.run_hooked(prompt.into(), deps).await } - /// Shared implementation behind `run` and `run_stream`. - /// - /// Returns the run result plus the inner agent's run id and message - /// history, which `run_stream` needs for its synthetic `RunComplete` - /// event. + /// Hooked-run implementation behind `run`. /// /// # Errors /// @@ -368,21 +353,15 @@ impl HookedAgent { /// and the failure reaches the caller untouched. /// - Returns [`serdes_ai::agent::AgentRunError::Other`] when a run hook /// returns or substitutes its own error during dispatch. - async fn run_with_extras( + async fn run_hooked( &self, prompt: String, deps: (), - ) -> Result<(HookedAgentRunResult, AgentRunExtras), serdes_ai::agent::AgentRunError> { + ) -> Result { if self.hooks.run_hooks_is_empty() { let response = self.inner.run(prompt, deps).await?; - let serdes_ai::agent::AgentRunResult { - output, - run_id, - messages, - .. - } = response; - let extras = AgentRunExtras { run_id, messages }; - return Ok((HookedAgentRunResult { content: output }, extras)); + let serdes_ai::agent::AgentRunResult { output, .. } = response; + return Ok(HookedAgentRunResult { content: output }); } // Wrapper-assigned run id for the hook context. The inner agent @@ -396,13 +375,11 @@ impl HookedAgent { }; let config = RunConfig::default(); - let extras_slot = Arc::new(Mutex::new(None)); let error_slot = Arc::new(Mutex::new(None)); let executor = SerdesRunExecutor { agent: &self.inner, prompt, deps, - extras: Arc::clone(&extras_slot), error: Arc::clone(&error_slot), }; @@ -414,71 +391,41 @@ impl HookedAgent { Err(dispatched) => return Err(restore_run_error(dispatched, &error_slot)), }; - // A hook that skipped `original` leaves the slot empty; fall back to - // the wrapper id so downstream events still carry a stable identifier. - let extras = extras_slot - .lock() - .expect("extras slot should not be poisoned") - .take() - .unwrap_or(AgentRunExtras { - run_id, - messages: Vec::new(), - }); - - Ok((HookedAgentRunResult::from_run_output(output), extras)) + Ok(HookedAgentRunResult::from_run_output(output)) } - /// Runs the agent in streaming mode. + /// Runs the agent in streaming mode, yielding framework-owned + /// [`RunEvent`]s. /// - /// When no run hooks are registered this delegates directly to the inner - /// agent's `run_stream`. When hooks are present it reuses the hooked - /// non-stream path and emits a synthetic stream containing the final - /// text output plus a `RunComplete` event carrying the real run id and - /// message history. + /// Starts the inner agent's stream and lazily maps each vendor event as + /// the stream is polled, so real incremental text and thinking deltas + /// reach the caller as they arrive. The mapped + /// [`RunEvent::RunComplete`] carries the inner run's id and a distilled + /// transcript. The prompt accepts full + /// [`UserContent`][serdes_ai::core::UserContent]; image and multi-part + /// prompts pass through to the vendor unchanged. + /// + /// Registered run hooks are not consulted on the streaming path; + /// preamble, system-prompt, and model-settings injection apply to + /// [`HookedAgent::run`] only. /// /// # Errors /// /// - Returns the inner agent's [`serdes_ai::agent::AgentRunError`] - /// unchanged when the inner agent fails (direct stream or hooked run) - /// and the failure reaches the caller untouched. - /// - Returns [`serdes_ai::agent::AgentRunError::Other`] when the prompt is not - /// representable as text, or when a run hook returns or substitutes its - /// own error during dispatch. + /// unchanged when starting the stream fails. + /// - The stream itself yields the inner error as an `Err` item when the + /// run fails mid-stream; a vendor error event surfaces as the mapped + /// [`RunEvent::Error`] variant instead. pub async fn run_stream( &self, prompt: impl Into, deps: (), ) -> Result< - Pin< - Box< - dyn Stream< - Item = Result, - > + Send, - >, - >, + Pin> + Send>>, serdes_ai::agent::AgentRunError, > { - let prompt = prompt.into(); - if self.hooks.run_hooks_is_empty() { - let stream = self.inner.run_stream(prompt, deps).await?; - return Ok(Box::pin(stream)); - } - let text = prompt.as_text().ok_or_else(|| { - serdes_ai::agent::AgentRunError::Other(anyhow::anyhow!( - "run hooks require a text prompt; image or multi-part prompts are unsupported" - )) - })?; - let (result, extras) = self.run_with_extras(text.to_string(), deps).await?; - let text = result.output().to_string(); - let events = vec![ - Ok(serdes_ai::AgentStreamEvent::TextDelta { text }), - Ok(serdes_ai::AgentStreamEvent::OutputReady), - Ok(serdes_ai::AgentStreamEvent::RunComplete { - run_id: extras.run_id, - messages: extras.messages, - }), - ]; - Ok(Box::pin(futures::stream::iter(events))) + let inner = self.inner.run_stream(prompt, deps).await?; + Ok(Box::pin(RunEventStream::new(inner))) } } @@ -595,7 +542,6 @@ impl<'a> RunExecutor for SerdesRunExecutor<'a> { prompt = format!("{}\n\n{prompt}", sections.join("\n\n")); } - let extras = Arc::clone(&self.extras); let error = Arc::clone(&self.error); let run_options = run_options_with_overrides(agent, config.model_settings_overrides); #[allow(clippy::let_unit_value)] @@ -617,17 +563,11 @@ impl<'a> RunExecutor for SerdesRunExecutor<'a> { return Err(projection); } }; - // Read borrowed fields before moving run_id and messages out. let content = response.output().to_string(); let usage = RunUsage { prompt_tokens: response.usage.request_tokens, completion_tokens: response.usage.response_tokens, }; - let inner_extras = AgentRunExtras { - run_id: response.run_id, - messages: response.messages, - }; - *extras.lock().expect("extras slot should not be poisoned") = Some(inner_extras); Ok(RunOutput { content, reason: EndReason::Completed, @@ -801,7 +741,7 @@ mod tests { read as read_meta, task as task_meta, write as write_meta, }; use serde_json::json; - use serdes_ai::core::{ModelResponse, ModelSettings}; + use serdes_ai::core::{ModelRequest, ModelResponse, ModelSettings}; use std::collections::HashSet; use std::path::PathBuf; use std::sync::Mutex; diff --git a/src/reloaded-code-serdesai/src/lib.rs b/src/reloaded-code-serdesai/src/lib.rs index 103091f..a04b9bc 100644 --- a/src/reloaded-code-serdesai/src/lib.rs +++ b/src/reloaded-code-serdesai/src/lib.rs @@ -33,6 +33,9 @@ pub use reloaded_code_agents::{ AgentDefaults, AgentRuntime, AgentRuntimeBuilder, ModelResolutionError, ResolvedModel, resolve_model_with_catalog, }; +/// Re-export [`RunEvent`], the framework-owned item type yielded by +/// [`HookedAgent::run_stream`]. +pub use reloaded_code_core::hooks::RunEvent; pub mod agent_ext; pub mod agent_runtime; diff --git a/src/reloaded-code-serdesai/src/mock.rs b/src/reloaded-code-serdesai/src/mock.rs index c6edb4d..9f6afdd 100644 --- a/src/reloaded-code-serdesai/src/mock.rs +++ b/src/reloaded-code-serdesai/src/mock.rs @@ -25,6 +25,7 @@ use async_trait::async_trait; use futures::stream; use serdes_ai::core::{ FinishReason, ModelRequest, ModelResponse, ModelResponsePart, ModelResponseStreamEvent, + ToolCallPart, }; use serdes_ai_models::Model as ModelTrait; pub use serdes_ai_models::{FunctionModel, MockModel, TestModel}; @@ -34,6 +35,13 @@ use serdes_ai_models::{ ModelCapability, ModelError, ModelProfile, ModelRequestParameters, StreamedResponse, }; +// ============================================================================ +// Private helpers +// ============================================================================ + +/// Characters per streamed text chunk; see [`response_to_stream_events`]. +const STREAM_CHUNK_CHARS: usize = 16; + // ============================================================================ // Streamed - wrapper that adds streaming support to any Model // ============================================================================ @@ -292,16 +300,37 @@ fn extract_tool_return_text(tr: &serdes_ai::core::ToolReturnPart) -> String { serde_json::to_string_pretty(&val).unwrap_or_else(|_| format!("{:?}", tr.content)) } -// ============================================================================ -// Private helpers -// ============================================================================ - fn response_to_stream_events(response: ModelResponse) -> Vec { - let mut events = Vec::with_capacity(response.parts.len() * 2 + 1); + // Estimate: two boundary events per part plus one delta per chunk + // (byte length bounds the char-chunk count from above). + let estimated: usize = response + .parts + .iter() + .map(|part| match part { + ModelResponsePart::Text(text) => text.content.len() / STREAM_CHUNK_CHARS + 3, + _ => 3, + }) + .sum(); + let mut events = Vec::with_capacity(estimated); for (index, part) in response.parts.into_iter().enumerate() { - events.push(ModelResponseStreamEvent::part_start(index, part)); - events.push(ModelResponseStreamEvent::part_end(index)); + match part { + // Text streams incrementally: an empty start, then bounded + // chunks, so the agent layer surfaces multiple text deltas + // instead of one whole-part event. + ModelResponsePart::Text(text) => { + events.push(ModelResponseStreamEvent::part_start( + index, + ModelResponsePart::text(""), + )); + push_text_delta_events(&mut events, index, &text.content); + events.push(ModelResponseStreamEvent::part_end(index)); + } + part => { + events.push(ModelResponseStreamEvent::part_start(index, part)); + events.push(ModelResponseStreamEvent::part_end(index)); + } + } } events @@ -309,10 +338,32 @@ fn response_to_stream_events(response: ModelResponse) -> Vec ModelResponse { ModelResponse::with_parts(vec![ ModelResponsePart::text(format!("Calling {tool_name}...")), - ModelResponsePart::tool_call(tool_name, args.clone()), + ModelResponsePart::ToolCall( + ToolCallPart::new(tool_name, args.clone()).with_tool_call_id("call_mock"), + ), ]) .with_finish_reason(FinishReason::ToolCall) } + +/// Pushes one `text_delta` event per [`STREAM_CHUNK_CHARS`] character chunk +/// of `text`, chunking on char boundaries so multi-byte text stays intact. +fn push_text_delta_events(events: &mut Vec, index: usize, text: &str) { + let mut remaining = text; + while !remaining.is_empty() { + let end = remaining + .char_indices() + .nth(STREAM_CHUNK_CHARS) + .map_or(remaining.len(), |(offset, _)| offset); + events.push(ModelResponseStreamEvent::text_delta( + index, + &remaining[..end], + )); + remaining = &remaining[end..]; + } +} From 53443da04fd32b6e318c1a34be31930a8fdb2dae Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Sun, 16 Aug 2026 22:50:46 +0100 Subject: [PATCH 03/10] Added: optional step, context, and tool-call-delta RunEvents - Add five optional RunEvent variants (StepStart, StepEnd, ContextInfo, ContextCompressed, ToolCallDelta), surfaced from the vendor stream through an exhaustive adapter mapping. Each optional variant documents that backends may omit it and absence is normal; StepEnd documents the vendor's response-before-tool-execution ordering caveat. - Restore the serdesai-task example's pre-regression printed output: message-id tag prefixes, streamed tool-argument blocks with typed TaskInput rendering, and the model-request/tool-call summary line, now driven by the new variants. - Make distill_model_response size its outputs exactly in one scan and move (not clone) tool-call fields. - Bump reloaded-code-core to 0.2.2 (workspace requirement matches) per the additive convention. --- src/Cargo.lock | 2 +- src/Cargo.toml | 2 +- src/reloaded-code-core/Cargo.toml | 2 +- .../src/hooks/run_event/mod.rs | 91 +++- .../examples/serdesai-task.rs | 140 +++++- .../src/agent_runtime/stream_events.rs | 414 +++++++++++++----- 6 files changed, 521 insertions(+), 130 deletions(-) diff --git a/src/Cargo.lock b/src/Cargo.lock index d25b1da..c42ee01 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -2886,7 +2886,7 @@ dependencies = [ [[package]] name = "reloaded-code-core" -version = "0.2.1" +version = "0.2.2" dependencies = [ "ahash", "bitcode", diff --git a/src/Cargo.toml b/src/Cargo.toml index 2c0f84d..aa46b5c 100644 --- a/src/Cargo.toml +++ b/src/Cargo.toml @@ -76,7 +76,7 @@ serdes-ai-models = { version = "0.2.6", default-features = false } serdes-ai-streaming = "0.2" # Internal crates -reloaded-code-core = { version = "0.2.1", path = "reloaded-code-core", default-features = false } +reloaded-code-core = { version = "0.2.2", path = "reloaded-code-core", default-features = false } reloaded-code-bubblewrap = { version = "0.1.0", path = "reloaded-code-bubblewrap" } reloaded-code-agents = { version = "0.1.0", path = "reloaded-code-agents" } reloaded-code-models-dev = { version = "0.1.0", path = "reloaded-code-models-dev" } diff --git a/src/reloaded-code-core/Cargo.toml b/src/reloaded-code-core/Cargo.toml index 49697c8..e48d7b4 100644 --- a/src/reloaded-code-core/Cargo.toml +++ b/src/reloaded-code-core/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "reloaded-code-core" -version = "0.2.1" +version = "0.2.2" edition = "2021" description = "Lightweight, high-performance core types and utilities for coding tools - framework agnostic" repository = "https://github.com/Reloaded-Project/ReloadedCode" diff --git a/src/reloaded-code-core/src/hooks/run_event/mod.rs b/src/reloaded-code-core/src/hooks/run_event/mod.rs index 4bc296a..5a23f8d 100644 --- a/src/reloaded-code-core/src/hooks/run_event/mod.rs +++ b/src/reloaded-code-core/src/hooks/run_event/mod.rs @@ -21,9 +21,10 @@ use serde::{Deserialize, Serialize}; /// Framework-owned event yielded by a run stream. /// -/// One variant per observable streaming milestone: run start, text -/// and thinking deltas, tool activity (call start, call complete, -/// executed), output-ready, run complete, error, and cancellation. +/// One variant per observable streaming milestone: run start, step +/// boundaries, context telemetry, text and thinking deltas, tool +/// activity (call start, argument deltas, call complete, executed), +/// output-ready, run complete, error, and cancellation. /// /// The enum is `#[non_exhaustive]`: variants may be appended in a /// future release without a breaking change, so matches outside this @@ -36,6 +37,43 @@ pub enum RunEvent { /// Identifier of the started run. run_id: String, }, + /// A model-request step started; the step's tool events may follow + /// the matching [`RunEvent::StepEnd`] (see that variant). + /// + /// Optional: emitted only by backends that report step boundaries; + /// absence is normal. + StepStart { + /// Index of the started step; numbering is backend-defined. + step: u32, + }, + /// Context-size telemetry measured while building a model request. + /// + /// Optional: emitted only by backends that report context metrics; + /// absence is normal. + ContextInfo { + /// Estimated token count of the request. + estimated_tokens: usize, + /// Serialized request size in bytes (messages plus tools). + request_bytes: usize, + /// Model's context window limit, when known. + context_limit: Option, + }, + /// Context was compressed to fit within limits. + /// + /// Optional: emitted only by backends that compress context and + /// report it; absence is normal. + ContextCompressed { + /// Token count before compression. + original_tokens: usize, + /// Token count after compression. + compressed_tokens: usize, + /// Strategy used, e.g. "truncate" or "summarize". + strategy: String, + /// Number of messages before compression. + messages_before: usize, + /// Number of messages after compression. + messages_after: usize, + }, /// Incremental assistant text arrived. TextDelta { /// Text fragment appended since the previous delta. @@ -53,6 +91,17 @@ pub enum RunEvent { /// Call id correlating this call with its completion and result. tool_call_id: Option, }, + /// Incremental tool-call argument fragment arrived. + /// + /// Optional: a backend may omit it; absence is normal. Complete + /// arguments remain available in the [`RunEvent::RunComplete`] + /// transcript as [`RunToolCallSummary::arguments_json`]. + ToolCallDelta { + /// Call id correlating this fragment with its call. + tool_call_id: Option, + /// Argument fragment appended since the previous delta. + delta: String, + }, /// A tool call's arguments finished streaming. ToolCallComplete { /// Name of the tool being called. @@ -71,6 +120,19 @@ pub enum RunEvent { /// Error text when the tool failed. error: Option, }, + /// A model-request step's response finished. + /// + /// Backends that execute tools inside a step emit this before the + /// step's tool calls run, so tool events may arrive after the + /// matching [`RunEvent::StepEnd`]. + /// + /// Optional: emitted only by backends that report step boundaries; + /// absence is normal. + StepEnd { + /// Index of the finished step, matching its + /// [`RunEvent::StepStart`]. + step: u32, + }, /// The run's final output is ready to consume. OutputReady, /// The run completed. @@ -209,6 +271,24 @@ mod tests { RunEvent::RunStart { run_id: "run-42".into(), }, + RunEvent::StepStart { step: 0 }, + RunEvent::ContextInfo { + estimated_tokens: 128, + request_bytes: 512, + context_limit: Some(8192), + }, + RunEvent::ContextInfo { + estimated_tokens: 128, + request_bytes: 512, + context_limit: None, + }, + RunEvent::ContextCompressed { + original_tokens: 9000, + compressed_tokens: 4000, + strategy: "truncate".into(), + messages_before: 20, + messages_after: 6, + }, RunEvent::TextDelta { text: "chunk".into(), }, @@ -219,6 +299,10 @@ mod tests { tool_name: "read_file".into(), tool_call_id: Some("call_1".into()), }, + RunEvent::ToolCallDelta { + tool_call_id: Some("call_1".into()), + delta: "{\"path\":".into(), + }, RunEvent::ToolCallComplete { tool_name: "read_file".into(), tool_call_id: Some("call_1".into()), @@ -229,6 +313,7 @@ mod tests { success: false, error: Some("missing".into()), }, + RunEvent::StepEnd { step: 0 }, RunEvent::OutputReady, RunEvent::Error { message: "boom".into(), diff --git a/src/reloaded-code-serdesai/examples/serdesai-task.rs b/src/reloaded-code-serdesai/examples/serdesai-task.rs index 46ddc72..4490d23 100644 --- a/src/reloaded-code-serdesai/examples/serdesai-task.rs +++ b/src/reloaded-code-serdesai/examples/serdesai-task.rs @@ -4,12 +4,16 @@ //! orchestrator through [`AgentBuildContext::build`], and runs one //! prompt that should delegate exactly once to `reader`. //! +//! Transcript tags carry the model-request step index (``); +//! step events are optional, so a backend that omits them prints every +//! tag under step 0. +//! //! Run: Edit the API_KEY_NAME and API_KEY_VALUE constants below, then: //! cargo run --example serdesai-task -p reloaded-code-serdesai use futures::StreamExt; use reloaded_code_agents::{AgentCatalog, AgentLoader, AgentRuntimeBuilder}; -use reloaded_code_core::{CredentialResolver, resolve_workspace_root}; +use reloaded_code_core::{CredentialResolver, TaskInput, resolve_workspace_root}; use reloaded_code_models_dev::ModelsDevCatalog; use reloaded_code_serdesai::{AgentBuildContext, AgentDefaults, RunEvent}; use serdes_ai::UserContent; @@ -26,9 +30,17 @@ const API_KEY_VALUE: &str = ""; // <-- Set your API key here const MODEL_ID: &str = "synthetic/hf:zai-org/GLM-4.7-Flash"; struct OpenStreamTag { + message_id: u32, tag: &'static str, } +struct PendingToolCall { + message_id: u32, + tool_name: String, + tool_call_id: Option, + args: String, +} + #[tokio::main] async fn main() -> Result<(), Box> { let agents_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR")) @@ -80,28 +92,70 @@ async fn main() -> Result<(), Box> { let prompt = UserContent::text(prompt); let prompt_text = render_user_content(&prompt); - println!("\n=== Transcript (streamed where possible) ==="); - log_xml("user", &prompt_text); + println!("\n=== Transcript (message ids, streamed where possible) ==="); + log_xml(0, "user", &prompt_text); let mut stream = agent.run_stream(prompt, ()).await?; + let mut current_message_id = 0u32; + let mut request_count = 0u32; let mut tool_call_count = 0u32; // Tracks the currently-open streaming XML tag so we can append deltas without reopening. let mut open_tag: Option = None; + let mut pending_tool_calls = Vec::with_capacity(4); while let Some(event) = stream.next().await { match event? { + RunEvent::StepStart { step } => { + close_stream_xml(&mut open_tag); + current_message_id = step; + request_count = request_count.saturating_add(1); + } RunEvent::ThinkingDelta { text } => { - write_stream_delta(&mut open_tag, "thinking", &text); + write_stream_delta(&mut open_tag, current_message_id, "thinking", &text); } RunEvent::TextDelta { text } => { - write_stream_delta(&mut open_tag, "assistant", &text); + write_stream_delta(&mut open_tag, current_message_id, "assistant", &text); } - RunEvent::ToolCallStart { tool_name, .. } => { + RunEvent::ToolCallStart { + tool_name, + tool_call_id, + } => { close_stream_xml(&mut open_tag); - log_xml("tool", &tool_name); + log_xml(current_message_id, "tool", &tool_name); + pending_tool_calls.push(PendingToolCall { + message_id: current_message_id, + tool_name, + tool_call_id, + args: String::new(), + }); + } + RunEvent::ToolCallDelta { + delta, + tool_call_id, + } => { + // Accumulate streamed JSON args into the matching pending call. + if let Some(index) = + pending_tool_call_index(&pending_tool_calls, tool_call_id.as_deref()) + { + pending_tool_calls[index].args.push_str(&delta); + } } - RunEvent::ToolCallComplete { .. } => { + RunEvent::ToolCallComplete { tool_call_id, .. } => { tool_call_count = tool_call_count.saturating_add(1); + if let Some(call) = + take_pending_tool_call(&mut pending_tool_calls, tool_call_id.as_deref()) + { + let tag = if call.tool_name == "task" { + "task-input" + } else { + "tool-input" + }; + let content = render_tool_input(&call.tool_name, &call.args); + log_xml(call.message_id, tag, &content); + } + } + RunEvent::StepEnd { .. } => { + close_stream_xml(&mut open_tag); } RunEvent::RunComplete { .. } => { close_stream_xml(&mut open_tag); @@ -112,25 +166,38 @@ async fn main() -> Result<(), Box> { close_stream_xml(&mut open_tag); - println!("Root agent activity: {tool_call_count} tool calls"); + println!( + "Root agent activity: {} model requests, {} tool calls", + request_count, tool_call_count + ); Ok(()) } -fn log_xml(tag: &str, content: &str) { +fn log_xml(message_id: u32, tag: &str, content: &str) { // Long or multiline content gets block-style tags; short content fits on one line. if content.contains('\n') || content.len() > 120 { - println!("<{tag}>"); + println!(""); println!("{content}"); println!(""); return; } let mut line = String::with_capacity(content.len() + tag.len() * 2 + 18); - let _ = write!(line, "<{tag}>{content}"); + let _ = write!(line, "{content}"); println!("{line}"); } +fn render_tool_input(tool_name: &str, args_text: &str) -> String { + match serde_json::from_str::(args_text) { + Ok(args) if tool_name == "task" => render_task_input(&args), + Ok(args) => { + serde_json::to_string_pretty(&args).expect("tool args serialization should succeed") + } + Err(_) => args_text.to_string(), + } +} + fn render_user_content(content: &UserContent) -> String { match content { UserContent::Text(text) => text.clone(), @@ -139,17 +206,31 @@ fn render_user_content(content: &UserContent) -> String { } } -fn write_stream_delta(open_tag: &mut Option, tag: &'static str, text: &str) { +fn take_pending_tool_call( + pending: &mut Vec, + tool_call_id: Option<&str>, +) -> Option { + pending_tool_call_index(pending, tool_call_id).map(|index| pending.remove(index)) +} + +fn write_stream_delta( + open_tag: &mut Option, + message_id: u32, + tag: &'static str, + text: &str, +) { if text.is_empty() { return; } - // If the tag changed, close the previous open tag and start a new one. - let is_same = open_tag.as_ref().is_some_and(|t| t.tag == tag); + // If the message or tag changed, close the previous open tag and start a new one. + let is_same = open_tag + .as_ref() + .is_some_and(|t| t.message_id == message_id && t.tag == tag); if !is_same { close_stream_xml(open_tag); - println!("<{tag}>"); - *open_tag = Some(OpenStreamTag { tag }); + println!(""); + *open_tag = Some(OpenStreamTag { message_id, tag }); } print!("{text}"); @@ -162,3 +243,28 @@ fn close_stream_xml(open_tag: &mut Option) { println!("", tag.tag); } } + +/// Index of the pending call an event addresses: match the call id +/// from the newest call, or the last pending call when ids are absent. +fn pending_tool_call_index( + pending: &[PendingToolCall], + tool_call_id: Option<&str>, +) -> Option { + match tool_call_id { + // Most backends include a tool_call_id; fall back to the last + // pending call otherwise. + Some(tool_call_id) => pending + .iter() + .rposition(|call| call.tool_call_id.as_deref() == Some(tool_call_id)), + None => pending.len().checked_sub(1), + } +} + +fn render_task_input(args: &serde_json::Value) -> String { + // Try to decode into the typed TaskInput shape; fall back to raw JSON. + serde_json::from_value::(args.clone()) + .and_then(|input| serde_json::to_string_pretty(&input)) + .unwrap_or_else(|_| { + serde_json::to_string_pretty(args).expect("task args serialization should succeed") + }) +} diff --git a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs index e1b9f32..dd0d4ff 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -5,24 +5,21 @@ //! event types stay inside this module; consumers of //! [`HookedAgent::run_stream`][task] only ever see [`RunEvent`] items. //! -//! # Dropped vendor-only events +//! # Optional events //! -//! The vendor emits five events with no [`RunEvent`] counterpart; the -//! mapping drops them: +//! Step boundaries, context telemetry, and streamed tool-call +//! arguments map through when the vendor reports them. The vendor +//! delivers whole tool-call arguments as a single delta when the +//! model does not stream fragments, so [`RunEvent::ToolCallDelta`] +//! items arrive for every tool call with non-empty arguments. //! -//! - `ContextInfo` and `ContextCompressed`: context-size telemetry emitted -//! before each model request, and compression notices. Context metrics -//! are not surfaced anywhere else. -//! - `RequestStart` and `ResponseComplete`: model-request step boundaries. -//! Dropping them removes the step index consumers could use to group -//! events by model request. -//! - `ToolCallDelta`: incremental tool-call argument fragments. Dropping it -//! removes streamed argument assembly; complete arguments remain -//! available in the [`RunEvent::RunComplete`] transcript as -//! [`RunToolCallSummary::arguments_json`]. +//! Ordering caveat: the vendor emits `ResponseComplete` before +//! executing a step's tool calls, so [`RunEvent::StepEnd`] closes the +//! step before its [`RunEvent::ToolExecuted`] items arrive. //! -//! Observable information loss: step boundaries and incremental tool-call -//! arguments no longer stream, and context telemetry is not surfaced. +//! The match over [`AgentStreamEvent`] is exhaustive: a new vendor +//! variant fails compilation here, keeping vendor coupling inside +//! this module. //! //! [task]: super::task::HookedAgent::run_stream @@ -42,8 +39,7 @@ use std::task::{Context, Poll}; /// /// Polling this stream drives the vendor stream, which the vendor already /// runs on its own background task; no channel, spawn, or shared agent -/// handle is added on this side. Vendor-only events (see the module docs) -/// are dropped rather than yielded. +/// handle is added on this side. pub(super) struct RunEventStream { /// Owned vendor stream, driven by the vendor's own background task. inner: AgentStream, @@ -60,31 +56,46 @@ impl Stream for RunEventStream { type Item = Result; fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { - // Dropped events yield nothing, so keep polling until a mappable - // event, an error, or the stream's end arrives. let inner = &mut self.get_mut().inner; - loop { - match inner.poll_next_unpin(cx) { - Poll::Ready(Some(Ok(event))) => { - if let Some(event) = map_vendor_event(event) { - return Poll::Ready(Some(Ok(event))); - } - } - Poll::Ready(Some(Err(error))) => return Poll::Ready(Some(Err(error))), - Poll::Ready(None) => return Poll::Ready(None), - Poll::Pending => return Poll::Pending, - } + match inner.poll_next_unpin(cx) { + Poll::Ready(Some(Ok(event))) => Poll::Ready(Some(Ok(map_vendor_event(event)))), + Poll::Ready(Some(Err(error))) => Poll::Ready(Some(Err(error))), + Poll::Ready(None) => Poll::Ready(None), + Poll::Pending => Poll::Pending, } } } /// Maps one vendor event to its framework-owned counterpart. /// -/// Returns `None` for vendor-only events; see the module docs for the -/// dropped set and its observable information loss. -fn map_vendor_event(event: AgentStreamEvent) -> Option { - Some(match event { +/// The match is exhaustive over the vendor enum, so vendor drift +/// fails compilation here instead of leaking vendor types. +fn map_vendor_event(event: AgentStreamEvent) -> RunEvent { + match event { AgentStreamEvent::RunStart { run_id } => RunEvent::RunStart { run_id }, + AgentStreamEvent::RequestStart { step } => RunEvent::StepStart { step }, + AgentStreamEvent::ContextInfo { + estimated_tokens, + request_bytes, + context_limit, + } => RunEvent::ContextInfo { + estimated_tokens, + request_bytes, + context_limit, + }, + AgentStreamEvent::ContextCompressed { + original_tokens, + compressed_tokens, + strategy, + messages_before, + messages_after, + } => RunEvent::ContextCompressed { + original_tokens, + compressed_tokens, + strategy, + messages_before, + messages_after, + }, AgentStreamEvent::TextDelta { text } => RunEvent::TextDelta { text }, AgentStreamEvent::ThinkingDelta { text } => RunEvent::ThinkingDelta { text }, AgentStreamEvent::ToolCallStart { @@ -94,6 +105,13 @@ fn map_vendor_event(event: AgentStreamEvent) -> Option { tool_name, tool_call_id, }, + AgentStreamEvent::ToolCallDelta { + delta, + tool_call_id, + } => RunEvent::ToolCallDelta { + tool_call_id, + delta, + }, AgentStreamEvent::ToolCallComplete { tool_name, tool_call_id, @@ -112,6 +130,7 @@ fn map_vendor_event(event: AgentStreamEvent) -> Option { success, error, }, + AgentStreamEvent::ResponseComplete { step } => RunEvent::StepEnd { step }, AgentStreamEvent::OutputReady => RunEvent::OutputReady, AgentStreamEvent::RunComplete { run_id, messages } => RunEvent::RunComplete { run_id, @@ -127,12 +146,7 @@ fn map_vendor_event(event: AgentStreamEvent) -> Option { partial_thinking, pending_tools, }, - AgentStreamEvent::ContextInfo { .. } - | AgentStreamEvent::ContextCompressed { .. } - | AgentStreamEvent::RequestStart { .. } - | AgentStreamEvent::ToolCallDelta { .. } - | AgentStreamEvent::ResponseComplete { .. } => return None, - }) + } } /// Distills the vendor run transcript into framework-owned records. @@ -205,18 +219,32 @@ fn authored_message(role: RunMessageRole, text: String) -> RunMessage { /// Returns `None` when the response carries no distilled content /// (thinking-only, file-only, or empty). fn distill_model_response(response: ModelResponse) -> Option { - let text = response.text_content(); - let text = (!text.is_empty()).then_some(text); - let mut tool_calls = Vec::new(); + // One sizing scan gives both outputs exact capacity, so neither + // grows mid-build. + let mut text_len = 0usize; + let mut tool_call_count = 0usize; for part in &response.parts { - if let ModelResponsePart::ToolCall(call) = part { - tool_calls.push(RunToolCallSummary { - tool_name: call.tool_name.clone(), - tool_call_id: call.tool_call_id.clone(), + match part { + ModelResponsePart::Text(text) => text_len += text.content.len(), + ModelResponsePart::ToolCall(_) => tool_call_count += 1, + _ => {} + } + } + let mut text = String::with_capacity(text_len); + let mut tool_calls = Vec::with_capacity(tool_call_count); + // Consuming the parts moves the call fields instead of cloning. + for part in response.parts { + match part { + ModelResponsePart::Text(part_text) => text.push_str(&part_text.content), + ModelResponsePart::ToolCall(call) => tool_calls.push(RunToolCallSummary { + tool_name: call.tool_name, + tool_call_id: call.tool_call_id, arguments_json: call.args.to_json_string().ok(), - }); + }), + _ => {} } } + let text = (!text.is_empty()).then_some(text); if text.is_none() && tool_calls.is_empty() { return None; } @@ -283,49 +311,96 @@ mod tests { // ======================================================================== #[test] - fn map_vendor_event_drops_vendor_only_variants() { - let dropped = vec![ - AgentStreamEvent::ContextInfo { - estimated_tokens: 1, - request_bytes: 4, - context_limit: None, - }, - AgentStreamEvent::ContextCompressed { - original_tokens: 10, - compressed_tokens: 5, - strategy: "truncate".into(), - messages_before: 2, - messages_after: 1, - }, - AgentStreamEvent::RequestStart { step: 1 }, - AgentStreamEvent::ToolCallDelta { - delta: "{\"a\":".into(), - tool_call_id: Some("call_1".into()), - }, - AgentStreamEvent::ResponseComplete { step: 1 }, - ]; - for event in dropped { - assert!( - map_vendor_event(event).is_none(), - "vendor-only event should be dropped" - ); - } + fn map_vendor_event_preserves_optional_variant_payloads() { + // Shared field shapes across variants compile even when + // mislabelled, so each mapping pins its variant identity. + let RunEvent::StepStart { step } = + map_vendor_event(AgentStreamEvent::RequestStart { step: 2 }) + else { + panic!("request start should map to the step-start variant"); + }; + assert_eq!(step, 2); + + let RunEvent::StepEnd { step } = + map_vendor_event(AgentStreamEvent::ResponseComplete { step: 2 }) + else { + panic!("response complete should map to the step-end variant"); + }; + assert_eq!(step, 2); + + let RunEvent::ContextInfo { + estimated_tokens, + request_bytes, + context_limit, + } = map_vendor_event(AgentStreamEvent::ContextInfo { + estimated_tokens: 128, + request_bytes: 512, + context_limit: Some(8192), + }) + else { + panic!("context info should map"); + }; + assert_eq!((estimated_tokens, request_bytes), (128, 512)); + assert_eq!(context_limit, Some(8192)); + + let RunEvent::ContextCompressed { + original_tokens, + compressed_tokens, + strategy, + messages_before, + messages_after, + } = map_vendor_event(AgentStreamEvent::ContextCompressed { + original_tokens: 10, + compressed_tokens: 5, + strategy: "truncate".into(), + messages_before: 2, + messages_after: 1, + }) + else { + panic!("context compressed should map"); + }; + assert_eq!( + ( + original_tokens, + compressed_tokens, + messages_before, + messages_after + ), + (10, 5, 2, 1) + ); + assert_eq!(strategy, "truncate"); + + // The vendor also emits whole arguments as one delta, so this + // mapping runs on the mock path; pinned here with a fragment + // shape to prove pass-through. + let RunEvent::ToolCallDelta { + tool_call_id, + delta, + } = map_vendor_event(AgentStreamEvent::ToolCallDelta { + delta: "{\"a\":".into(), + tool_call_id: Some("call_1".into()), + }) + else { + panic!("tool call delta should map"); + }; + assert_eq!(tool_call_id.as_deref(), Some("call_1")); + assert_eq!(delta, "{\"a\":"); } #[test] fn map_vendor_event_preserves_mapped_variant_payloads() { - let Some(RunEvent::Error { message }) = map_vendor_event(AgentStreamEvent::Error { + let RunEvent::Error { message } = map_vendor_event(AgentStreamEvent::Error { message: "boom".into(), }) else { panic!("error event should map"); }; assert_eq!(message, "boom"); - let Some(RunEvent::Cancelled { + let RunEvent::Cancelled { partial_text, partial_thinking, pending_tools, - }) = map_vendor_event(AgentStreamEvent::Cancelled { + } = map_vendor_event(AgentStreamEvent::Cancelled { partial_text: Some("partial".into()), partial_thinking: None, pending_tools: vec!["read".into()], @@ -339,7 +414,7 @@ mod tests { // A mislabel of the thinking arm (both sides are bare text records) // would compile, so pin the variant identity here. - let Some(RunEvent::ThinkingDelta { text }) = + let RunEvent::ThinkingDelta { text } = map_vendor_event(AgentStreamEvent::ThinkingDelta { text: "hmm".into() }) else { panic!("thinking delta should map to the thinking variant"); @@ -348,10 +423,10 @@ mod tests { // The call-start and call-complete arms share one field shape, so // pin each variant identity and its id separately. - let Some(RunEvent::ToolCallStart { + let RunEvent::ToolCallStart { tool_name, tool_call_id, - }) = map_vendor_event(AgentStreamEvent::ToolCallStart { + } = map_vendor_event(AgentStreamEvent::ToolCallStart { tool_name: "read".into(), tool_call_id: Some("call_1".into()), }) @@ -363,10 +438,10 @@ mod tests { ("read", Some("call_1")) ); - let Some(RunEvent::ToolCallComplete { + let RunEvent::ToolCallComplete { tool_name, tool_call_id, - }) = map_vendor_event(AgentStreamEvent::ToolCallComplete { + } = map_vendor_event(AgentStreamEvent::ToolCallComplete { tool_name: "read".into(), tool_call_id: Some("call_1".into()), }) @@ -378,12 +453,12 @@ mod tests { ("read", Some("call_1")) ); - let Some(RunEvent::ToolExecuted { + let RunEvent::ToolExecuted { tool_name, tool_call_id, success, error, - }) = map_vendor_event(AgentStreamEvent::ToolExecuted { + } = map_vendor_event(AgentStreamEvent::ToolExecuted { tool_name: "read".into(), tool_call_id: Some("call_1".into()), success: false, @@ -407,6 +482,9 @@ mod tests { let response = ModelResponse::with_parts(vec![ ModelResponsePart::text("checking"), ModelResponsePart::tool_call("read_file", json!({"path": "a.txt"})), + // Text after the call proves the consuming pass keeps every + // text part in order, not just the leading one. + ModelResponsePart::text("done"), ]); let request = ModelRequest::with_parts(vec![ ModelRequestPart::SystemPrompt(SystemPromptPart::new("sys")), @@ -443,7 +521,7 @@ mod tests { }, RunMessage { role: RunMessageRole::Assistant, - text: Some("checking".into()), + text: Some("checkingdone".into()), tool_calls: vec![RunToolCallSummary { tool_name: "read_file".into(), tool_call_id: None, @@ -796,39 +874,67 @@ mod tests { ); } - #[tokio::test] - async fn run_stream_reports_tool_activity_consistent_with_transcript() { - // Scripted flow: the first turn calls `ping`, then the final turn - // answers with text once the tool return is in the history. + /// Stream window of the scripted ping call: the run's events plus + /// the call's start/complete positions and stamped call id, shared + /// by the tool-activity and optional-events tests. + struct PingCallWindow { + events: Vec, + call_start: usize, + call_id: Option, + call_complete: usize, + } + + /// Runs the scripted ping flow (the first turn calls `ping`, then + /// the final turn answers with text) and locates the ping call's + /// stream window. + async fn run_scripted_ping_flow() -> PingCallWindow { let model = tool_then_text("ping", json!({"target": "example.com"}), "after the tool"); let hooked = streamed_agent_with_ping_tool(model, HookSet::builder().build()); - let events = collect_events(&hooked, "use the tool").await; - let position = |predicate: &dyn Fn(&RunEvent) -> bool| { - events.iter().position(|event| predicate(event)) - }; - let call_start = position(&|event| { + let call_start = position(&events, &|event| { matches!(event, RunEvent::ToolCallStart { tool_name, .. } if tool_name == "ping") }) .expect("tool call start should stream"); - // The scripted mock stamps the call id; every later event and - // transcript record must correlate on it. - let streamed_call_id = match &events[call_start] { + let call_id = match &events[call_start] { RunEvent::ToolCallStart { tool_call_id, .. } => tool_call_id.clone(), other => panic!("expected a tool call start, got {other:?}"), }; + let call_complete = position(&events, &|event| { + matches!(event, RunEvent::ToolCallComplete { tool_call_id, .. } + if tool_call_id == &call_id) + }) + .expect("tool call complete should stream"); + PingCallWindow { + events, + call_start, + call_id, + call_complete, + } + } + + /// Index of the first event matching `predicate`. + fn position(events: &[RunEvent], predicate: &dyn Fn(&RunEvent) -> bool) -> Option { + events.iter().position(|event| predicate(event)) + } + + #[tokio::test] + async fn run_stream_reports_tool_activity_consistent_with_transcript() { + let PingCallWindow { + events, + call_start, + call_id: streamed_call_id, + call_complete, + } = run_scripted_ping_flow().await; + + // The scripted mock stamps the call id; every later event and + // transcript record must correlate on it. assert_eq!( streamed_call_id.as_deref(), Some("call_mock"), "the scripted call id must stream through the start event" ); - let call_complete = position(&|event| { - matches!(event, RunEvent::ToolCallComplete { tool_call_id, .. } - if tool_call_id == &streamed_call_id) - }) - .expect("tool call complete should stream"); - let executed = position(&|event| { + let executed = position(&events, &|event| { matches!( event, RunEvent::ToolExecuted { tool_name, tool_call_id, success, error: None, .. } @@ -842,10 +948,12 @@ mod tests { ); // Output-ready arrives after the final text but before completion. - let output_ready = position(&|event| matches!(event, RunEvent::OutputReady)) + let output_ready = position(&events, &|event| matches!(event, RunEvent::OutputReady)) .expect("output-ready should stream"); - let complete = position(&|event| matches!(event, RunEvent::RunComplete { .. })) - .expect("run should complete"); + let complete = position(&events, &|event| { + matches!(event, RunEvent::RunComplete { .. }) + }) + .expect("run should complete"); assert_eq!( complete, events.len() - 1, @@ -912,6 +1020,98 @@ mod tests { assert_eq!(final_answer, streamed_answer); } + #[tokio::test] + async fn run_stream_surfaces_optional_events_when_backend_emits_them() { + let PingCallWindow { + events, + call_start, + call_id: streamed_call_id, + call_complete, + } = run_scripted_ping_flow().await; + + // Both model-request steps report start and end boundaries, so + // consumers can group the stream by step index; this backend + // numbers steps from one. + let started_steps: Vec = events + .iter() + .filter_map(|event| match event { + RunEvent::StepStart { step } => Some(*step), + _ => None, + }) + .collect(); + assert_eq!(started_steps, vec![1, 2], "both steps should report starts"); + let ended_steps: Vec = events + .iter() + .filter_map(|event| match event { + RunEvent::StepEnd { step } => Some(*step), + _ => None, + }) + .collect(); + assert_eq!(ended_steps, vec![1, 2], "both steps should report ends"); + let first_step_start = position(&events, &|event| { + matches!(event, RunEvent::StepStart { step: 1 }) + }) + .expect("first step should start"); + let first_content = position(&events, &|event| { + matches!( + event, + RunEvent::TextDelta { .. } | RunEvent::ToolCallStart { .. } + ) + }) + .expect("step content should stream"); + assert!( + first_step_start < first_content, + "step start must precede the step's content events" + ); + let first_step_end = position(&events, &|event| { + matches!(event, RunEvent::StepEnd { step: 1 }) + }) + .expect("first step should end"); + let second_step_start = position(&events, &|event| { + matches!(event, RunEvent::StepStart { step: 2 }) + }) + .expect("second step should start"); + assert!( + first_step_end < second_step_start, + "steps must not interleave" + ); + + // Context telemetry arrives per model request. + let context_infos: Vec<&RunEvent> = events + .iter() + .filter(|event| matches!(event, RunEvent::ContextInfo { .. })) + .collect(); + assert_eq!(context_infos.len(), 2, "one context info per step"); + for info in context_infos { + let RunEvent::ContextInfo { request_bytes, .. } = info else { + unreachable!("filtered to context info"); + }; + assert!( + *request_bytes > 0, + "serialized request size should be positive" + ); + } + + // Streamed argument assembly: the ping call's deltas arrive + // between its start and completion and concatenate to the + // call's arguments. The mock delivers arguments whole, so this + // exercises the vendor's single-delta path. + let streamed_args: String = events[call_start + 1..call_complete] + .iter() + .filter_map(|event| match event { + RunEvent::ToolCallDelta { + tool_call_id, + delta, + } if tool_call_id == &streamed_call_id => Some(delta.as_str()), + _ => None, + }) + .collect(); + assert_eq!( + streamed_args, r#"{"target":"example.com"}"#, + "argument deltas must concatenate to the call's arguments" + ); + } + #[tokio::test] async fn run_stream_maps_vendor_error_event_and_ends_with_inner_error() { let hooked = streamed_agent(FailingStreamModel::new(), HookSet::builder().build()); From 2e10d44cdb133ba91dbb99edeb9a5e12542c762c Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Sun, 16 Aug 2026 23:59:57 +0100 Subject: [PATCH 04/10] Changed: table-drive map_vendor_event tests with rstest - Converted the two hand-written map_vendor_event unit tests into rstest parameterized tables: optional events (StepStart, StepEnd, ContextInfo, ContextCompressed, ToolCallDelta) and always-emitted events (RunStart, TextDelta, ThinkingDelta, ToolCallStart, ToolCallComplete, ToolExecuted, Error, Cancelled). - Each case now asserts whole-event equality, so a mislabelled mapping arm fails even when the shared field shapes compile. - Added RunStart and TextDelta coverage the previous tests lacked. Tests-only change; production code untouched. --- .../src/agent_runtime/stream_events.rs | 254 ++++++++---------- 1 file changed, 113 insertions(+), 141 deletions(-) diff --git a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs index dd0d4ff..5d84237 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -297,6 +297,7 @@ mod tests { HookRunContext, HookSet, RunConfig, RunHook, RunHookFuture, RunOriginal, }; use reloaded_code_core::{ToolCatalogEntry, ToolCatalogKind}; + use rstest::rstest; use serde_json::json; use serdes_ai::core::messages::request::RetryPromptPart; use serdes_ai::core::{ @@ -310,167 +311,138 @@ mod tests { // Vendor event mapping // ======================================================================== - #[test] - fn map_vendor_event_preserves_optional_variant_payloads() { - // Shared field shapes across variants compile even when - // mislabelled, so each mapping pins its variant identity. - let RunEvent::StepStart { step } = - map_vendor_event(AgentStreamEvent::RequestStart { step: 2 }) - else { - panic!("request start should map to the step-start variant"); - }; - assert_eq!(step, 2); - - let RunEvent::StepEnd { step } = - map_vendor_event(AgentStreamEvent::ResponseComplete { step: 2 }) - else { - panic!("response complete should map to the step-end variant"); - }; - assert_eq!(step, 2); - - let RunEvent::ContextInfo { - estimated_tokens, - request_bytes, - context_limit, - } = map_vendor_event(AgentStreamEvent::ContextInfo { + /// Optional events map field-for-field when the backend emits them; + /// absence is normal on other backends. Whole-event equality pins + /// variant identity too, so a mislabelled arm fails even when the + /// field shapes compile. + #[rstest] + #[case::step_start( + AgentStreamEvent::RequestStart { step: 2 }, + RunEvent::StepStart { step: 2 } + )] + #[case::step_end( + AgentStreamEvent::ResponseComplete { step: 2 }, + RunEvent::StepEnd { step: 2 } + )] + #[case::context_info( + AgentStreamEvent::ContextInfo { estimated_tokens: 128, request_bytes: 512, context_limit: Some(8192), - }) - else { - panic!("context info should map"); - }; - assert_eq!((estimated_tokens, request_bytes), (128, 512)); - assert_eq!(context_limit, Some(8192)); - - let RunEvent::ContextCompressed { - original_tokens, - compressed_tokens, - strategy, - messages_before, - messages_after, - } = map_vendor_event(AgentStreamEvent::ContextCompressed { + }, + RunEvent::ContextInfo { + estimated_tokens: 128, + request_bytes: 512, + context_limit: Some(8192), + } + )] + #[case::context_compressed( + AgentStreamEvent::ContextCompressed { original_tokens: 10, compressed_tokens: 5, strategy: "truncate".into(), messages_before: 2, messages_after: 1, - }) - else { - panic!("context compressed should map"); - }; - assert_eq!( - ( - original_tokens, - compressed_tokens, - messages_before, - messages_after - ), - (10, 5, 2, 1) - ); - assert_eq!(strategy, "truncate"); - - // The vendor also emits whole arguments as one delta, so this - // mapping runs on the mock path; pinned here with a fragment - // shape to prove pass-through. - let RunEvent::ToolCallDelta { - tool_call_id, - delta, - } = map_vendor_event(AgentStreamEvent::ToolCallDelta { + }, + RunEvent::ContextCompressed { + original_tokens: 10, + compressed_tokens: 5, + strategy: "truncate".into(), + messages_before: 2, + messages_after: 1, + } + )] + // The vendor delivers whole arguments as one delta when the model + // does not stream fragments; fragment shape pinned for pass-through. + #[case::tool_call_delta( + AgentStreamEvent::ToolCallDelta { delta: "{\"a\":".into(), tool_call_id: Some("call_1".into()), - }) - else { - panic!("tool call delta should map"); - }; - assert_eq!(tool_call_id.as_deref(), Some("call_1")); - assert_eq!(delta, "{\"a\":"); + }, + RunEvent::ToolCallDelta { + tool_call_id: Some("call_1".into()), + delta: "{\"a\":".into(), + } + )] + fn map_vendor_event_preserves_optional_variant_payloads( + #[case] event: AgentStreamEvent, + #[case] expected: RunEvent, + ) { + assert_eq!(map_vendor_event(event), expected); } - #[test] - fn map_vendor_event_preserves_mapped_variant_payloads() { - let RunEvent::Error { message } = map_vendor_event(AgentStreamEvent::Error { - message: "boom".into(), - }) else { - panic!("error event should map"); - }; - assert_eq!(message, "boom"); - - let RunEvent::Cancelled { - partial_text, - partial_thinking, - pending_tools, - } = map_vendor_event(AgentStreamEvent::Cancelled { - partial_text: Some("partial".into()), - partial_thinking: None, - pending_tools: vec!["read".into()], - }) - else { - panic!("cancelled event should map"); - }; - assert_eq!(partial_text.as_deref(), Some("partial")); - assert!(partial_thinking.is_none()); - assert_eq!(pending_tools, vec!["read".to_string()]); - - // A mislabel of the thinking arm (both sides are bare text records) - // would compile, so pin the variant identity here. - let RunEvent::ThinkingDelta { text } = - map_vendor_event(AgentStreamEvent::ThinkingDelta { text: "hmm".into() }) - else { - panic!("thinking delta should map to the thinking variant"); - }; - assert_eq!(text, "hmm"); - - // The call-start and call-complete arms share one field shape, so - // pin each variant identity and its id separately. - let RunEvent::ToolCallStart { - tool_name, - tool_call_id, - } = map_vendor_event(AgentStreamEvent::ToolCallStart { + /// Always-emitted events map field-for-field. + #[rstest] + #[case::run_start( + AgentStreamEvent::RunStart { run_id: "run_1".into() }, + RunEvent::RunStart { run_id: "run_1".into() } + )] + #[case::text_delta( + AgentStreamEvent::TextDelta { text: "hello".into() }, + RunEvent::TextDelta { text: "hello".into() } + )] + // Both text arms share one field shape; whole-event equality pins + // the thinking variant separately. + #[case::thinking_delta( + AgentStreamEvent::ThinkingDelta { text: "hmm".into() }, + RunEvent::ThinkingDelta { text: "hmm".into() } + )] + // Call-start and call-complete share one field shape; each pinned. + #[case::tool_call_start( + AgentStreamEvent::ToolCallStart { tool_name: "read".into(), tool_call_id: Some("call_1".into()), - }) - else { - panic!("tool call start should map to the start variant"); - }; - assert_eq!( - (tool_name.as_str(), tool_call_id.as_deref()), - ("read", Some("call_1")) - ); - - let RunEvent::ToolCallComplete { - tool_name, - tool_call_id, - } = map_vendor_event(AgentStreamEvent::ToolCallComplete { + }, + RunEvent::ToolCallStart { tool_name: "read".into(), tool_call_id: Some("call_1".into()), - }) - else { - panic!("tool call complete should map to the complete variant"); - }; - assert_eq!( - (tool_name.as_str(), tool_call_id.as_deref()), - ("read", Some("call_1")) - ); - - let RunEvent::ToolExecuted { - tool_name, - tool_call_id, - success, - error, - } = map_vendor_event(AgentStreamEvent::ToolExecuted { + } + )] + #[case::tool_call_complete( + AgentStreamEvent::ToolCallComplete { + tool_name: "read".into(), + tool_call_id: Some("call_1".into()), + }, + RunEvent::ToolCallComplete { + tool_name: "read".into(), + tool_call_id: Some("call_1".into()), + } + )] + #[case::tool_executed( + AgentStreamEvent::ToolExecuted { tool_name: "read".into(), tool_call_id: Some("call_1".into()), success: false, error: Some("missing".into()), - }) - else { - panic!("tool executed event should map"); - }; - assert_eq!(tool_name, "read"); - assert_eq!(tool_call_id.as_deref(), Some("call_1")); - assert!(!success); - assert_eq!(error.as_deref(), Some("missing")); + }, + RunEvent::ToolExecuted { + tool_name: "read".into(), + tool_call_id: Some("call_1".into()), + success: false, + error: Some("missing".into()), + } + )] + #[case::error( + AgentStreamEvent::Error { message: "boom".into() }, + RunEvent::Error { message: "boom".into() } + )] + #[case::cancelled( + AgentStreamEvent::Cancelled { + partial_text: Some("partial".into()), + partial_thinking: None, + pending_tools: vec!["read".into()], + }, + RunEvent::Cancelled { + partial_text: Some("partial".into()), + partial_thinking: None, + pending_tools: vec!["read".into()], + } + )] + fn map_vendor_event_preserves_mapped_variant_payloads( + #[case] event: AgentStreamEvent, + #[case] expected: RunEvent, + ) { + assert_eq!(map_vendor_event(event), expected); } // ======================================================================== From edfa528e43e2c1e716fc5253b66473a9528aba87 Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Mon, 17 Aug 2026 00:20:56 +0100 Subject: [PATCH 05/10] Removed: redundant stream_events test cases Trimmed the stream_events test module from 11 test functions to 7 (-106/+42 lines) with zero coverage loss by folding redundant cases into broader tests: - run-id/user-turn assertions merged into the tool-activity test - multipart-prompt acceptance merged into the incremental-deltas test (its prompt is now UserContent::Parts) - thinking-skip and multipart-serialization cases merged into the tool-flow transcript test Production code is untouched. cargo test -p reloaded-code-serdesai --lib stream_events: 18 passed, 0 failed. --- .../src/agent_runtime/stream_events.rs | 148 +++++------------- 1 file changed, 42 insertions(+), 106 deletions(-) diff --git a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs index 5d84237..bb839e8 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -462,6 +462,11 @@ mod tests { ModelRequestPart::SystemPrompt(SystemPromptPart::new("sys")), ModelRequestPart::UserPrompt(UserPromptPart::new("read a.txt")), ModelRequestPart::ModelResponse(Box::new(response)), + // Thinking carries no distilled representation; the length + // assertion below proves it contributes no message. + ModelRequestPart::ModelResponse(Box::new(ModelResponse::with_parts(vec![ + ModelResponsePart::thinking("internal"), + ]))), ModelRequestPart::ToolReturn( ToolReturnPart::success("read_file", "contents").with_tool_call_id("call_1"), ), @@ -475,8 +480,16 @@ mod tests { "call_9", )), ]); + // A follow-up request with a multipart user prompt covers + // multi-request distillation and the parts-serialization branch. + let follow_up = ModelRequest::with_parts(vec![ModelRequestPart::UserPrompt( + UserPromptPart::new(UserContent::Parts(vec![ + serdes_ai::core::UserContentPart::text("part one"), + serdes_ai::core::UserContentPart::text("part two"), + ])), + )]); - let messages = distill_messages(vec![request]); + let messages = distill_messages(vec![request, follow_up]); let expected = vec![ RunMessage { @@ -517,7 +530,7 @@ mod tests { tool_result: None, }, ]; - assert_eq!(messages.len(), 6); + assert_eq!(messages.len(), 7); assert_eq!(messages[..5], expected[..]); // The builtin tool return distills as a Tool turn whose output is @@ -534,40 +547,20 @@ mod tests { "serialized builtin content should stay observable: {}", builtin.output ); - } - #[test] - fn distill_messages_serializes_multipart_user_prompts() { - let request = ModelRequest::with_parts(vec![ModelRequestPart::UserPrompt( - UserPromptPart::new(UserContent::Parts(vec![ - serdes_ai::core::UserContentPart::text("part one"), - serdes_ai::core::UserContentPart::text("part two"), - ])), - )]); - - let messages = distill_messages(vec![request]); - - assert_eq!(messages.len(), 1); - assert_eq!(messages[0].role, RunMessageRole::User); - let text = messages[0] + // The multipart prompt from the follow-up request renders as + // serialized text so mixed content stays observable. + assert_eq!(messages[6].role, RunMessageRole::User); + let multipart = messages[6] .text .as_deref() .expect("multipart prompt should render as text"); assert!( - text.contains("part one") && text.contains("part two"), - "serialized parts should stay observable: {text}" + multipart.contains("part one") && multipart.contains("part two"), + "serialized parts should stay observable: {multipart}" ); } - #[test] - fn distill_messages_skips_responses_without_distillable_content() { - let response = ModelResponse::with_parts(vec![ModelResponsePart::thinking("internal")]); - let request = - ModelRequest::with_parts(vec![ModelRequestPart::ModelResponse(Box::new(response))]); - - assert!(distill_messages(vec![request]).is_empty()); - } - // ======================================================================== // HookedAgent::run_stream integration // ======================================================================== @@ -726,7 +719,14 @@ mod tests { let hooks = HookSet::builder().run_hook(recorder.clone()).build(); let hooked = streamed_agent(model, hooks); - let events = collect_events(&hooked, "hello").await; + // The multipart prompt doubles as stream-path acceptance + // coverage: structured prompts must stream to completion, and + // the model closure ignores prompt content. + let prompt = UserContent::Parts(vec![ + serdes_ai::core::UserContentPart::text("hello"), + serdes_ai::core::UserContentPart::image_url("https://example.invalid/image.png"), + ]); + let events = collect_events(&hooked, prompt).await; let delta_count = events .iter() @@ -769,83 +769,6 @@ mod tests { assert_eq!(streamed_text, RESPONSE); } - #[tokio::test] - async fn run_stream_run_complete_carries_real_run_id_and_faithful_transcript() { - // The model echoes the last user prompt so the transcript ties to - // the prompt text. - let model = Streamed::new(FunctionModel::new(|messages, _| { - let prompt = messages - .iter() - .rev() - .flat_map(|message| message.user_prompts()) - .next() - .and_then(|prompt| prompt.content.as_text()) - .unwrap_or_default() - .to_string(); - ModelResponse::text(format!("echo: {prompt}")) - })); - let hooked = streamed_agent(model, HookSet::builder().build()); - - let events = collect_events(&hooked, "transcript probe").await; - - let mut start_run_id = None; - let mut streamed_text = String::new(); - let mut complete = None; - for event in events { - match event { - RunEvent::RunStart { run_id } => start_run_id = Some(run_id), - RunEvent::TextDelta { text } => streamed_text.push_str(&text), - RunEvent::RunComplete { run_id, messages } => { - complete = Some((run_id, messages)); - } - _ => {} - } - } - let (run_id, messages) = complete.expect("run should complete"); - assert!( - !run_id.is_empty(), - "RunComplete must carry the inner run id" - ); - assert_eq!( - start_run_id.as_deref(), - Some(run_id.as_str()), - "RunComplete id must match the started run" - ); - - // Transcript consistency: the user turn carries the prompt, the - // assistant turn carries exactly what was streamed. - let user_text = messages - .iter() - .find(|message| message.role == RunMessageRole::User) - .and_then(|message| message.text.as_deref()); - assert_eq!(user_text, Some("transcript probe")); - let assistant_text = messages - .iter() - .find(|message| message.role == RunMessageRole::Assistant) - .and_then(|message| message.text.as_deref()); - assert_eq!(assistant_text, Some(streamed_text.as_str())); - } - - #[tokio::test] - async fn run_stream_accepts_multipart_prompt() { - let model = Streamed::new(FunctionModel::new(|_, _| ModelResponse::text("handled"))); - let hooked = streamed_agent(model, HookSet::builder().build()); - - let prompt = UserContent::Parts(vec![ - serdes_ai::core::UserContentPart::text("describe this"), - serdes_ai::core::UserContentPart::image_url("https://example.invalid/image.png"), - ]); - - let events = collect_events(&hooked, prompt).await; - - assert!( - events - .iter() - .any(|event| matches!(event, RunEvent::RunComplete { .. })), - "multipart prompt should stream to completion" - ); - } - /// Stream window of the scripted ping call: the run's events plus /// the call's start/complete positions and stamped call id, shared /// by the tool-activity and optional-events tests. @@ -960,6 +883,19 @@ mod tests { )), "RunComplete id must match the started run" ); + assert!( + !run_id.is_empty(), + "RunComplete must carry the inner run id" + ); + let user_text = messages + .iter() + .find(|message| message.role == RunMessageRole::User) + .and_then(|message| message.text.as_deref()); + assert_eq!( + user_text, + Some("use the tool"), + "the user turn should carry the prompt verbatim" + ); let call_summary = messages .iter() .flat_map(|message| &message.tool_calls) From b2483381bbb9871f0fd3526a66abc9472c8c7962 Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Mon, 17 Aug 2026 00:36:07 +0100 Subject: [PATCH 06/10] Changed: eliminate hidden clones in RunEvent transcript distillation Vendor accessors (`to_string_content`, `RetryContent::message`, `ToolCallArgs::to_json_string`) take `&self` and clone, but this module owns the values being read. Match on the owned variants first so plain text tool returns (largest strings in the transcript), retry text, and raw-string tool arguments move their Strings; non-text variants keep the vendor accessor fallback. `distill_model_response` also gains a single-text fast path: one text part (the common assistant shape) moves into `RunMessage.text` instead of copying through a joined buffer. Multi-text responses still join via the exact-capacity buffer. - Behavior identical; full test suite, clippy `-D warnings`, and docs pass via `.cargo/verify.sh` --- .../src/agent_runtime/stream_events.rs | 60 ++++++++++++++----- 1 file changed, 45 insertions(+), 15 deletions(-) diff --git a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs index bb839e8..f1068c6 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -27,6 +27,7 @@ use futures::{Stream, StreamExt}; use reloaded_code_core::hooks::{ RunEvent, RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary, }; +use serdes_ai::core::messages::{RetryContent, ToolCallArgs, ToolReturnContent}; use serdes_ai::core::{ ModelRequest, ModelRequestPart, ModelResponse, ModelResponsePart, UserContent, }; @@ -174,16 +175,24 @@ fn distill_messages(messages: Vec) -> Vec { )); } ModelRequestPart::RetryPrompt(part) => { - distilled.push(authored_message( - RunMessageRole::User, - part.content.message().to_owned(), - )); + // Plain-text retry content moves its owned String; + // structured content still renders through the vendor + // accessor, which only borrows. + let text = match part.content { + RetryContent::Text(text) => text, + other => other.message().to_owned(), + }; + distilled.push(authored_message(RunMessageRole::User, text)); } ModelRequestPart::ToolReturn(part) => { - distilled.push(tool_result_message( - part.tool_call_id, - part.content.to_string_content(), - )); + // Text returns (the common large case) move their + // owned String; other variants render through the + // vendor accessor, which clones. + let output = match part.content { + ToolReturnContent::Text { content } => content, + other => other.to_string_content(), + }; + distilled.push(tool_result_message(part.tool_call_id, output)); } ModelRequestPart::BuiltinToolReturn(part) => { // Structured content (search results, code output) has @@ -219,32 +228,53 @@ fn authored_message(role: RunMessageRole, text: String) -> RunMessage { /// Returns `None` when the response carries no distilled content /// (thinking-only, file-only, or empty). fn distill_model_response(response: ModelResponse) -> Option { - // One sizing scan gives both outputs exact capacity, so neither - // grows mid-build. + // One sizing scan gives both outputs exact capacity, plus the text + // part count: a single text part (the common shape) moves its + // String instead of copying through a joined buffer. let mut text_len = 0usize; + let mut text_part_count = 0usize; let mut tool_call_count = 0usize; for part in &response.parts { match part { - ModelResponsePart::Text(text) => text_len += text.content.len(), + ModelResponsePart::Text(text) => { + text_len += text.content.len(); + text_part_count += 1; + } ModelResponsePart::ToolCall(_) => tool_call_count += 1, _ => {} } } - let mut text = String::with_capacity(text_len); let mut tool_calls = Vec::with_capacity(tool_call_count); + let mut joined = (text_part_count != 1).then(|| String::with_capacity(text_len)); + let mut single: Option = None; // Consuming the parts moves the call fields instead of cloning. for part in response.parts { match part { - ModelResponsePart::Text(part_text) => text.push_str(&part_text.content), + ModelResponsePart::Text(part_text) => { + if let Some(buffer) = joined.as_mut() { + buffer.push_str(&part_text.content); + } else { + single = Some(part_text.content); + } + } ModelResponsePart::ToolCall(call) => tool_calls.push(RunToolCallSummary { tool_name: call.tool_name, tool_call_id: call.tool_call_id, - arguments_json: call.args.to_json_string().ok(), + // Raw-string args move their owned String; parsed JSON + // serializes through the vendor accessor. + arguments_json: match call.args { + ToolCallArgs::String(raw) => Some(raw), + other => other.to_json_string().ok(), + }, }), _ => {} } } - let text = (!text.is_empty()).then_some(text); + // Multi-text responses keep their joined buffer; both paths drop + // to `None` when no text survived. + let text = single + .or_else(|| joined.filter(|joined| !joined.is_empty())) + .filter(|text| !text.is_empty()); if text.is_none() && tool_calls.is_empty() { return None; } From 577b02d800a91d4543379a4e34a1cf573dd712ad Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Mon, 17 Aug 2026 01:12:22 +0100 Subject: [PATCH 07/10] Changed: document run-hook skip rationale in run_stream docs Add a `# Remarks` section to `HookedAgent::run_stream` in the SerdesAI adapter explaining why registered run hooks are deliberately skipped on the streaming path: the core run-hook chain resolves to one completed `RunOutput`, so dispatching it there would buffer the whole run before the first event and defeat streaming. State explicitly that run-hook config injection (preamble, system prompt, model settings overrides) applies to `HookedAgent::run` only. Doc-comment-only change; no behavior change. Validation: rust-llm-tidy clean; cargo test -p reloaded-code-serdesai passed 140 lib and 17 doc tests, 0 failed. --- src/reloaded-code-serdesai/src/agent_runtime/task.rs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/reloaded-code-serdesai/src/agent_runtime/task.rs b/src/reloaded-code-serdesai/src/agent_runtime/task.rs index 62b43fa..1fbefc6 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/task.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/task.rs @@ -405,9 +405,13 @@ impl HookedAgent { /// [`UserContent`][serdes_ai::core::UserContent]; image and multi-part /// prompts pass through to the vendor unchanged. /// - /// Registered run hooks are not consulted on the streaming path; - /// preamble, system-prompt, and model-settings injection apply to - /// [`HookedAgent::run`] only. + /// # Remarks + /// + /// Registered run hooks are skipped on this path. The core run-hook + /// chain resolves to one completed `RunOutput`, so dispatching it here + /// would buffer the whole run before the first event and defeat + /// streaming. Preamble, system-prompt, and model-settings injection + /// therefore apply to [`HookedAgent::run`] only. /// /// # Errors /// From ccbee201ddea10a1314aca92ccaa6d990b86dc30 Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Mon, 17 Aug 2026 01:16:37 +0100 Subject: [PATCH 08/10] Changed: re-export RunEvent transcript payload types from serdesai Consumers matching on RunEvent::RunComplete needed the message shape types (RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary) but had no path to them without depending on reloaded-code-core directly. Re-export them alongside RunEvent so the crate's public surface names every transcript payload type it exposes. --- src/reloaded-code-serdesai/src/lib.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/reloaded-code-serdesai/src/lib.rs b/src/reloaded-code-serdesai/src/lib.rs index a04b9bc..7eeb293 100644 --- a/src/reloaded-code-serdesai/src/lib.rs +++ b/src/reloaded-code-serdesai/src/lib.rs @@ -34,8 +34,12 @@ pub use reloaded_code_agents::{ resolve_model_with_catalog, }; /// Re-export [`RunEvent`], the framework-owned item type yielded by -/// [`HookedAgent::run_stream`]. -pub use reloaded_code_core::hooks::RunEvent; +/// [`HookedAgent::run_stream`], together with its transcript payload +/// types ([`RunMessage`], [`RunMessageRole`], [`RunToolCallSummary`], +/// [`RunToolResultSummary`]). +pub use reloaded_code_core::hooks::{ + RunEvent, RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary, +}; pub mod agent_ext; pub mod agent_runtime; From d1f8de62b22b8bc49bba549e4e1f767b7d873d81 Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Mon, 17 Aug 2026 01:28:20 +0100 Subject: [PATCH 09/10] Changed: give scripted mock tool calls per-turn call ids - tool_call_response stamped every scripted tool call with the fixed id "call_mock", so two_tools_then_text correlated both calls to one id in streamed events and transcripts. The id now derives per scripted turn as call_mock_{turn + 1}, and the helper doc states the derivation and its 0-based turn. - Pin the first- and second-turn ids plus their distinctness with a unit test driving two_tools_then_text directly, and expect call_mock_1 in the stream-events call-id correlation test. --- .../src/agent_runtime/stream_events.rs | 2 +- src/reloaded-code-serdesai/src/mock.rs | 70 +++++++++++++++++-- 2 files changed, 64 insertions(+), 8 deletions(-) diff --git a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs index f1068c6..ec1ac5c 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -856,7 +856,7 @@ mod tests { // transcript record must correlate on it. assert_eq!( streamed_call_id.as_deref(), - Some("call_mock"), + Some("call_mock_1"), "the scripted call id must stream through the start event" ); let executed = position(&events, &|event| { diff --git a/src/reloaded-code-serdesai/src/mock.rs b/src/reloaded-code-serdesai/src/mock.rs index 9f6afdd..9d0b4a5 100644 --- a/src/reloaded-code-serdesai/src/mock.rs +++ b/src/reloaded-code-serdesai/src/mock.rs @@ -192,7 +192,7 @@ pub fn tool_then_text( ModelResponse::text(text) } else { // First call: emit a tool call so the agent executes the real tool. - tool_call_response(&tool_name, &args) + tool_call_response(&tool_name, &args, 0) } }); @@ -241,8 +241,8 @@ pub fn two_tools_then_text( let answered_calls = messages.iter().flat_map(|m| m.tool_returns()).count(); match answered_calls { - 0 => tool_call_response(&first_name, &first_args), - 1 => tool_call_response(&second_name, &second_args), + 0 => tool_call_response(&first_name, &first_args, 0), + 1 => tool_call_response(&second_name, &second_args, 1), _ => { let tool_results: String = messages .iter() @@ -339,13 +339,15 @@ fn response_to_stream_events(response: ModelResponse) -> Vec ModelResponse { +/// The call id is `call_mock_{turn + 1}`, where `turn` is the 0-based +/// scripted turn number, so each scripted call correlates separately in +/// streamed events and transcripts. +fn tool_call_response(tool_name: &str, args: &serde_json::Value, turn: usize) -> ModelResponse { ModelResponse::with_parts(vec![ ModelResponsePart::text(format!("Calling {tool_name}...")), ModelResponsePart::ToolCall( - ToolCallPart::new(tool_name, args.clone()).with_tool_call_id("call_mock"), + ToolCallPart::new(tool_name, args.clone()) + .with_tool_call_id(format!("call_mock_{}", turn + 1)), ), ]) .with_finish_reason(FinishReason::ToolCall) @@ -367,3 +369,57 @@ fn push_text_delta_events(events: &mut Vec, index: usi remaining = &remaining[end..]; } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use serdes_ai::core::{ModelRequestPart, ToolReturnPart}; + + /// Id stamped on the first tool call in `response`, if it has one. + fn scripted_call_id(response: &ModelResponse) -> Option<&str> { + response.parts.iter().find_map(|part| match part { + ModelResponsePart::ToolCall(call) => call.tool_call_id.as_deref(), + _ => None, + }) + } + + /// Each scripted turn stamps its own call id, so two-call flows keep + /// distinct correlation keys in streamed events and transcripts. + #[tokio::test] + async fn two_tools_then_text_assigns_distinct_call_ids_per_turn() { + let model = two_tools_then_text( + ("read", json!({"file_path": "a.txt"})), + ("write", json!({"file_path": "b.txt", "content": "notes"})), + "Run finished.", + ); + + // No answered calls in an empty history: the first turn runs. + let first = model + .request( + &[], + &ModelSettings::default(), + &ModelRequestParameters::default(), + ) + .await + .expect("first scripted turn should respond"); + + // One answered call in the history advances the script. + let history = vec![ModelRequest::with_parts(vec![ + ModelRequestPart::ToolReturn( + ToolReturnPart::success("read", "fixture").with_tool_call_id("call_mock_1"), + ), + ])]; + let second = model + .request( + &history, + &ModelSettings::default(), + &ModelRequestParameters::default(), + ) + .await + .expect("second scripted turn should respond"); + + assert_eq!(scripted_call_id(&first), Some("call_mock_1")); + assert_eq!(scripted_call_id(&second), Some("call_mock_2")); + } +} From fdb8e3b44c4a44db2f194d22e1ee48c236012a97 Mon Sep 17 00:00:00 2001 From: Sewer56 Date: Mon, 17 Aug 2026 01:50:11 +0100 Subject: [PATCH 10/10] Changed: Redact user files in history --- .../src/agent_runtime/stream_events.rs | 79 +++++++++++++++++-- 1 file changed, 74 insertions(+), 5 deletions(-) diff --git a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs index ec1ac5c..81371e9 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -27,9 +27,12 @@ use futures::{Stream, StreamExt}; use reloaded_code_core::hooks::{ RunEvent, RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary, }; -use serdes_ai::core::messages::{RetryContent, ToolCallArgs, ToolReturnContent}; +use serdes_ai::core::messages::{ + AudioContent, DocumentContent, FileContent, ImageContent, RetryContent, ToolCallArgs, + ToolReturnContent, VideoContent, +}; use serdes_ai::core::{ - ModelRequest, ModelRequestPart, ModelResponse, ModelResponsePart, UserContent, + ModelRequest, ModelRequestPart, ModelResponse, ModelResponsePart, UserContent, UserContentPart, }; use serdes_ai::{AgentStream, AgentStreamEvent}; use std::pin::Pin; @@ -301,17 +304,48 @@ fn tool_result_message(tool_call_id: Option, output: String) -> RunMessa /// Renders user prompt content as audit text. /// -/// Plain text passes through; multi-part prompts are serialized so image -/// and mixed content stay observable in the transcript. +/// Plain text passes through; multi-part prompts are serialized so +/// image and mixed content stay observable in the transcript. Inline +/// binary parts collapse to a media type plus byte length placeholder +/// so payloads never bloat the transcript. fn user_content_text(content: UserContent) -> String { match content { UserContent::Text(text) => text, UserContent::Parts(parts) => { - serde_json::to_string(&parts).unwrap_or_else(|_| format!("{parts:?}")) + // After the map: binary parts are placeholders; text and + // URL parts serialize exactly as before. + let redacted: Vec<_> = parts.iter().map(redact_binary_part).collect(); + serde_json::to_string(&redacted).unwrap_or_else(|_| format!("{redacted:?}")) } } } +/// Serializes one prompt part for the audit trail. +/// +/// Text and URL parts keep their JSON shape; binary parts become a +/// placeholder carrying their media type and byte length. +fn redact_binary_part(part: &UserContentPart) -> serde_json::Value { + let placeholder = |kind: &str, media_type: &str, bytes: usize| serde_json::json!({ "type": kind, "media_type": media_type, "bytes": bytes }); + match part { + UserContentPart::Image { + image: ImageContent::Binary(binary), + } => placeholder("image", binary.media_type.mime_type(), binary.data.len()), + UserContentPart::Audio { + audio: AudioContent::Binary(binary), + } => placeholder("audio", binary.media_type.mime_type(), binary.data.len()), + UserContentPart::Video { + video: VideoContent::Binary(binary), + } => placeholder("video", binary.media_type.mime_type(), binary.data.len()), + UserContentPart::Document { + document: DocumentContent::Binary(binary), + } => placeholder("document", binary.media_type.mime_type(), binary.data.len()), + UserContentPart::File { + file: FileContent::Binary(binary), + } => placeholder("file", binary.mime_type.as_str(), binary.data.len()), + other => serde_json::to_value(other).unwrap_or(serde_json::Value::Null), + } +} + #[cfg(test)] mod tests { use super::*; @@ -591,6 +625,41 @@ mod tests { ); } + #[test] + fn user_content_text_redacts_binary_but_keeps_text_and_urls() { + use serdes_ai::core::messages::ImageMediaType; + + // Base64 of the binary bytes must never reach the audit text. + let content = UserContent::parts(vec![ + UserContentPart::text("look at this"), + UserContentPart::image_url("https://example.invalid/image.png"), + UserContentPart::image_binary(vec![1, 2, 3, 4], ImageMediaType::Png), + UserContentPart::File { + file: FileContent::binary(vec![9, 9], "application/pdf"), + }, + ]); + + let rendered = user_content_text(content); + + assert!( + rendered.contains("look at this") + && rendered.contains("https://example.invalid/image.png"), + "text and URL parts keep their serialization: {rendered}" + ); + assert!( + rendered.contains(r#""type":"image""#) + && rendered.contains(r#""media_type":"image/png""#) + && rendered.contains(r#""bytes":4"#) + && rendered.contains(r#""media_type":"application/pdf""#) + && rendered.contains(r#""bytes":2"#), + "binary parts render as media type plus length: {rendered}" + ); + assert!( + !rendered.contains("AQIDBA==") && !rendered.contains("CQk="), + "base64 payloads must not leak: {rendered}" + ); + } + // ======================================================================== // HookedAgent::run_stream integration // ========================================================================