diff --git a/src/Cargo.lock b/src/Cargo.lock index 1d65247..c42ee01 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.2" dependencies = [ "ahash", "bitcode", diff --git a/src/Cargo.toml b/src/Cargo.toml index b213db4..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.0", 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 711a772..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.0" +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/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..5a23f8d --- /dev/null +++ b/src/reloaded-code-core/src/hooks/run_event/mod.rs @@ -0,0 +1,333 @@ +//! 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, 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 +/// 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, + }, + /// 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. + 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, + }, + /// 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. + 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, + }, + /// 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. + 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::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(), + }, + RunEvent::ThinkingDelta { + text: "thought".into(), + }, + RunEvent::ToolCallStart { + 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()), + }, + RunEvent::ToolExecuted { + tool_name: "read_file".into(), + tool_call_id: Some("call_1".into()), + success: false, + error: Some("missing".into()), + }, + RunEvent::StepEnd { step: 0 }, + 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); + } + } +} diff --git a/src/reloaded-code-serdesai/examples/serdesai-task.rs b/src/reloaded-code-serdesai/examples/serdesai-task.rs index e8fd254..4490d23 100644 --- a/src/reloaded-code-serdesai/examples/serdesai-task.rs +++ b/src/reloaded-code-serdesai/examples/serdesai-task.rs @@ -4,6 +4,10 @@ //! 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 @@ -11,8 +15,8 @@ use futures::StreamExt; use reloaded_code_agents::{AgentCatalog, AgentLoader, AgentRuntimeBuilder}; use reloaded_code_core::{CredentialResolver, TaskInput, 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}, @@ -101,18 +105,18 @@ async fn main() -> Result<(), Box> { while let Some(event) = stream.next().await { match event? { - AgentStreamEvent::RequestStart { step } => { + RunEvent::StepStart { step } => { close_stream_xml(&mut open_tag); current_message_id = step; request_count = request_count.saturating_add(1); } - AgentStreamEvent::ThinkingDelta { text } => { + RunEvent::ThinkingDelta { text } => { write_stream_delta(&mut open_tag, current_message_id, "thinking", &text); } - AgentStreamEvent::TextDelta { text } => { + RunEvent::TextDelta { text } => { write_stream_delta(&mut open_tag, current_message_id, "assistant", &text); } - AgentStreamEvent::ToolCallStart { + RunEvent::ToolCallStart { tool_name, tool_call_id, } => { @@ -125,18 +129,18 @@ async fn main() -> Result<(), Box> { args: String::new(), }); } - AgentStreamEvent::ToolCallDelta { + RunEvent::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()) + if let Some(index) = + pending_tool_call_index(&pending_tool_calls, tool_call_id.as_deref()) { - call.args.push_str(&delta); + pending_tool_calls[index].args.push_str(&delta); } } - AgentStreamEvent::ToolCallComplete { tool_call_id, .. } => { + 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()) @@ -150,10 +154,10 @@ async fn main() -> Result<(), Box> { log_xml(call.message_id, tag, &content); } } - AgentStreamEvent::ResponseComplete { .. } => { + RunEvent::StepEnd { .. } => { close_stream_xml(&mut open_tag); } - AgentStreamEvent::RunComplete { .. } => { + RunEvent::RunComplete { .. } => { close_stream_xml(&mut open_tag); } _ => {} @@ -170,20 +174,6 @@ async fn main() -> Result<(), Box> { 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) { // Long or multiline content gets block-style tags; short content fits on one line. if content.contains('\n') || content.len() > 120 { @@ -220,13 +210,7 @@ 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)) + pending_tool_call_index(pending, tool_call_id).map(|index| pending.remove(index)) } fn write_stream_delta( @@ -260,6 +244,22 @@ fn close_stream_xml(open_tag: &mut Option) { } } +/// 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()) 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..81371e9 --- /dev/null +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -0,0 +1,1158 @@ +//! 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. +//! +//! # Optional events +//! +//! 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. +//! +//! 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. +//! +//! 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 + +use futures::{Stream, StreamExt}; +use reloaded_code_core::hooks::{ + RunEvent, RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary, +}; +use serdes_ai::core::messages::{ + AudioContent, DocumentContent, FileContent, ImageContent, RetryContent, ToolCallArgs, + ToolReturnContent, VideoContent, +}; +use serdes_ai::core::{ + ModelRequest, ModelRequestPart, ModelResponse, ModelResponsePart, UserContent, UserContentPart, +}; +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. +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> { + let inner = &mut self.get_mut().inner; + 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. +/// +/// 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 { + tool_name, + tool_call_id, + } => RunEvent::ToolCallStart { + tool_name, + tool_call_id, + }, + AgentStreamEvent::ToolCallDelta { + delta, + tool_call_id, + } => RunEvent::ToolCallDelta { + tool_call_id, + delta, + }, + 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::ResponseComplete { step } => RunEvent::StepEnd { step }, + 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, + }, + } +} + +/// 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) => { + // 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) => { + // 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 + // 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 { + // 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(); + text_part_count += 1; + } + ModelResponsePart::ToolCall(_) => tool_call_count += 1, + _ => {} + } + } + 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) => { + 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, + // 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(), + }, + }), + _ => {} + } + } + // 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; + } + 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. 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) => { + // 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::*; + 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 rstest::rstest; + 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 + // ======================================================================== + + /// 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), + }, + 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, + }, + 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()), + }, + 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); + } + + /// 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()), + }, + RunEvent::ToolCallStart { + tool_name: "read".into(), + tool_call_id: Some("call_1".into()), + } + )] + #[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()), + }, + 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); + } + + // ======================================================================== + // 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"})), + // 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")), + 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"), + ), + ModelRequestPart::RetryPrompt(RetryPromptPart::new("retry feedback")), + ModelRequestPart::BuiltinToolReturn(BuiltinToolReturnPart::new( + "web_search", + BuiltinToolReturnContent::Other { + kind: "custom".into(), + data: json!({"hits": 1}), + }, + "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, follow_up]); + + 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("checkingdone".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(), 7); + 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 + ); + + // 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!( + multipart.contains("part one") && multipart.contains("part two"), + "serialized parts should stay observable: {multipart}" + ); + } + + #[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 + // ======================================================================== + + /// 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); + + // 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() + .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); + } + + /// 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 call_start = position(&events, &|event| { + matches!(event, RunEvent::ToolCallStart { tool_name, .. } if tool_name == "ping") + }) + .expect("tool call start should stream"); + 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_1"), + "the scripted call id must stream through the start event" + ); + let executed = position(&events, &|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(&events, &|event| matches!(event, RunEvent::OutputReady)) + .expect("output-ready should stream"); + let complete = position(&events, &|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" + ); + 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) + .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_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()); + + 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..1fbefc6 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,45 @@ 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. + /// + /// # 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 /// /// - 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 +546,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 +567,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 +745,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..7eeb293 100644 --- a/src/reloaded-code-serdesai/src/lib.rs +++ b/src/reloaded-code-serdesai/src/lib.rs @@ -33,6 +33,13 @@ 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`], 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; diff --git a/src/reloaded-code-serdesai/src/mock.rs b/src/reloaded-code-serdesai/src/mock.rs index c6edb4d..9d0b4a5 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 // ============================================================================ @@ -184,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) } }); @@ -233,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() @@ -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,88 @@ 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::tool_call(tool_name, args.clone()), + ModelResponsePart::ToolCall( + ToolCallPart::new(tool_name, args.clone()) + .with_tool_call_id(format!("call_mock_{}", turn + 1)), + ), ]) .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..]; + } +} + +#[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")); + } +}