diff --git a/src/Cargo.lock b/src/Cargo.lock index c42ee016..c72c14b4 100644 --- a/src/Cargo.lock +++ b/src/Cargo.lock @@ -2886,7 +2886,7 @@ dependencies = [ [[package]] name = "reloaded-code-core" -version = "0.2.2" +version = "0.2.3" dependencies = [ "ahash", "bitcode", diff --git a/src/Cargo.toml b/src/Cargo.toml index aa46b5c7..e8d015bc 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.2", path = "reloaded-code-core", default-features = false } +reloaded-code-core = { version = "0.2.3", 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/docs/src/hooks.md b/src/docs/src/hooks.md index 0143d1d2..ca803631 100644 --- a/src/docs/src/hooks.md +++ b/src/docs/src/hooks.md @@ -2,8 +2,9 @@ Hooks let your code see, change, or stop things the agent does. -Tool and run hooks are wired into the [SerdesAI] agent pipeline: registered -hooks intercept real tool calls and agent runs end to end. +Tool, run, and run-event hooks are wired into the [SerdesAI] agent +pipeline: registered hooks intercept real tool calls, agent runs, and +streamed run events end to end. Tool hooks work like game mods. Each hook gets an `original` function. @@ -247,6 +248,64 @@ error still propagates to the caller. An outer hook that skips `original` never reaches this hook, so do not rely on it for cleanup that must run on every path. +### Intercept streamed events + +Run-event hooks fire only on `run_stream()`. Each event passes the +registered `RunEventHook`s, in registration order, only until one +suppresses it with `Ok(None)` or rejects it with `Err(ToolError)`; +a rejection also terminates the stream. + +Per event, a hook returns one of three decisions: + +- `Ok(Some(event))` publishes it, changed or unchanged. +- `Ok(None)` suppresses it. Later hooks and the consumer never see it. +- `Err(ToolError)` stops dispatch; the stream ends with + `Err(AgentRunError::Other)`. + +This hook forwards the events a TUI renders. The consumer still sees +every event: + +```rust +use reloaded_code_core::{HookSet, RunEventContext, RunEventHookResult}; +use reloaded_code_serdesai::{RunEvent, RunEventHook}; + +struct TuiSender { /* channel to the UI thread */ } + +impl TuiSender { + /// Stub: cheap, non-blocking send; the UI thread draws. + fn send_to_tui(&self, _event: &RunEvent) {} +} + +struct ForwardToTui { + tui: TuiSender, +} + +impl RunEventHook for ForwardToTui { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + // Forward only what the TUI renders + match &event { + RunEvent::TextDelta { .. } | RunEvent::RunComplete { .. } => { + self.tui.send_to_tui(&event); + } + _ => {} + } + Ok(Some(event)) + } +} + +let hooks = HookSet::builder() + .run_event_hook(ForwardToTui { tui: TuiSender {} }) + .build(); +``` + +Hooks run synchronously at token rate; keep per-event work cheap. Each +call sees one event. Text can split across several `TextDelta`s, so +buffer cross-event context inside the hook. + +Full example: [serdesai-run-event-hook] rewrites text deltas to +uppercase and suppresses the output-ready milestone +(`cargo run --example serdesai-run-event-hook -p reloaded-code-serdesai --features mock`). + ## Available types ### Tool hook types @@ -272,12 +331,21 @@ that must run on every path. | [`RunExecutor`] | Final callable used at the end of the run hook chain. | | [`RunUsage`] | Token usage for a completed run. | +### Run event hook types + +| Type | Purpose | +| ---------------------- | ----------------------------------------------------- | +| [`RunEventHook`] | Observes, rewrites, or suppresses one streamed event. | +| [`RunEventContext`] | Agent and model names for the event's stream. | +| [`RunEvent`] | Framework-owned event yielded by a run stream. | +| [`RunEventHookResult`] | Publish, rewrite, suppress, or Err(ToolError). | + ### Container types -| Type | Purpose | -| ------------------ | ------------------------------------------------- | -| [`HookSet`] | Stores tool hooks, run hooks, and compact events. | -| [`HookSetBuilder`] | Builder for [`HookSet`]. | +| Type | Purpose | +| ------------------ | ----------------------------------------------------------- | +| [`HookSet`] | Stores tool, run, and run-event hooks, plus compact events. | +| [`HookSetBuilder`] | Builder for [`HookSet`]. | ## How tool hooks stack @@ -340,6 +408,10 @@ passes `HookSet::default()`. - **Empty fast path.** `dispatch_tool` calls the real tool directly when you set no hooks. +- **Mode-scoped run hooks.** `RunHook` fires only on `run()`; + `RunEventHook` fires only on `run_stream()`. Each hook point stays + inert on the other path. + [`ToolHook`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/trait.ToolHook.html [`ToolOriginal`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/struct.ToolOriginal.html @@ -356,8 +428,13 @@ passes `HookSet::default()`. [`RunOutput`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/struct.RunOutput.html [`RunExecutor`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/trait.RunExecutor.html [`RunUsage`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/struct.RunUsage.html +[`RunEventHook`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/trait.RunEventHook.html +[`RunEventContext`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/struct.RunEventContext.html +[`RunEvent`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/enum.RunEvent.html +[`RunEventHookResult`]: https://docs.rs/reloaded-code-core/latest/reloaded_code_core/type.RunEventHookResult.html [SerdesAI]: https://crates.io/crates/serdes-ai [serdesai-tool-hook]: https://github.com/Reloaded-Project/ReloadedCode/blob/main/src/reloaded-code-serdesai/examples/hooks/tool/serdesai-tool-hook.rs [serdesai-tool-block]: https://github.com/Reloaded-Project/ReloadedCode/blob/main/src/reloaded-code-serdesai/examples/hooks/tool/serdesai-tool-block.rs [serdesai-tool-chain]: https://github.com/Reloaded-Project/ReloadedCode/blob/main/src/reloaded-code-serdesai/examples/hooks/tool/serdesai-tool-chain.rs [serdesai-run-hook]: https://github.com/Reloaded-Project/ReloadedCode/blob/main/src/reloaded-code-serdesai/examples/hooks/run/serdesai-run-hook.rs +[serdesai-run-event-hook]: https://github.com/Reloaded-Project/ReloadedCode/blob/main/src/reloaded-code-serdesai/examples/hooks/run/serdesai-run-event-hook.rs diff --git a/src/reloaded-code-core/Cargo.toml b/src/reloaded-code-core/Cargo.toml index e48d7b43..5c209b85 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.2" +version = "0.2.3" 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/builder.rs b/src/reloaded-code-core/src/hooks/builder.rs index b95ffdd5..56897896 100644 --- a/src/reloaded-code-core/src/hooks/builder.rs +++ b/src/reloaded-code-core/src/hooks/builder.rs @@ -1,6 +1,6 @@ //! HookSetBuilder — builder for constructing a [`HookSet`]. -use crate::hooks::{HookSet, RunHook, SessionCompactFn, ToolHook, INLINE_CAP}; +use crate::hooks::{HookSet, RunEventHook, RunHook, SessionCompactFn, ToolHook, INLINE_CAP}; use std::fmt; use std::sync::Arc; use tinyvec::TinyVec; @@ -10,6 +10,7 @@ use tinyvec::TinyVec; pub struct HookSetBuilder { pub(super) tool_hooks: Vec>, pub(super) run_hooks: Vec>, + pub(super) run_event_hooks: Vec>, pub(super) session_compact: TinyVec<[Option; INLINE_CAP]>, } @@ -67,6 +68,26 @@ impl HookSetBuilder { self } + /// Registers a run-event hook. + /// + /// Hooks run in registration order on every streamed event, + /// before the stream consumer sees it. Streaming path only: + /// run-event hooks never fire during a non-streaming `run()`. + #[inline] + #[must_use] + pub fn run_event_hook(mut self, hook: impl RunEventHook) -> Self { + self.run_event_hooks.push(Arc::new(hook)); + self + } + + /// Registers an already shared run-event hook. + #[inline] + #[must_use] + pub fn shared_run_event_hook(mut self, hook: Arc) -> Self { + self.run_event_hooks.push(hook); + self + } + /// Builds the `HookSet` from the configured hooks. #[inline] #[must_use] @@ -74,6 +95,7 @@ impl HookSetBuilder { HookSet { tool_hooks: self.tool_hooks, run_hooks: self.run_hooks, + run_event_hooks: self.run_event_hooks, session_compact: self.session_compact, } } @@ -84,6 +106,7 @@ impl fmt::Debug for HookSetBuilder { f.debug_struct("HookSetBuilder") .field("tool_hooks", &self.tool_hooks.len()) .field("run_hooks", &self.run_hooks.len()) + .field("run_event_hooks", &self.run_event_hooks.len()) .field("session_compact", &self.session_compact.len()) .finish() } @@ -92,6 +115,7 @@ impl fmt::Debug for HookSetBuilder { #[cfg(test)] mod tests { use super::*; + use crate::hooks::run_event::{RunEvent, RunEventContext, RunEventHook, RunEventHookResult}; use crate::hooks::run_hook::{HookRunContext, RunConfig, RunHookFuture, RunOriginal}; use crate::hooks::tool_hook::{ToolCallContext, ToolHookFuture, ToolOriginal, ToolRequest}; @@ -166,10 +190,52 @@ mod tests { assert_eq!(hooks.run_hooks().len(), 1); } + #[test] + fn run_event_hook_registration_makes_hook_set_non_empty() { + struct NoopEvent; + impl RunEventHook for NoopEvent { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + Ok(Some(event)) + } + } + + let hooks = HookSetBuilder::new().run_event_hook(NoopEvent).build(); + assert!(!hooks.is_empty()); + assert!(!hooks.run_event_hooks_is_empty()); + } + + #[test] + fn shared_run_event_hook_registration() { + struct NoopEvent; + impl RunEventHook for NoopEvent { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + Ok(Some(event)) + } + } + + let shared: Arc = Arc::new(NoopEvent); + let hooks = HookSetBuilder::new().shared_run_event_hook(shared).build(); + assert!(!hooks.run_event_hooks_is_empty()); + } + #[test] fn builder_debug_includes_run_hooks() { let builder = HookSetBuilder::new(); let debug = format!("{:?}", builder); assert!(debug.contains("run_hooks")); } + + #[test] + fn builder_debug_includes_run_event_hooks() { + struct NoopEvent; + impl RunEventHook for NoopEvent { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + Ok(Some(event)) + } + } + + let builder = HookSetBuilder::new().run_event_hook(NoopEvent); + let debug = format!("{builder:?}"); + assert!(debug.contains("run_event_hooks: 1")); + } } diff --git a/src/reloaded-code-core/src/hooks/hook_set.rs b/src/reloaded-code-core/src/hooks/hook_set.rs index 2d674c65..499cfdb2 100644 --- a/src/reloaded-code-core/src/hooks/hook_set.rs +++ b/src/reloaded-code-core/src/hooks/hook_set.rs @@ -1,8 +1,9 @@ //! HookSet — container and dispatch for all registered hooks and lifecycle events. use crate::hooks::{ - HookRunContext, RunConfig, RunExecutor, RunHook, RunHookFuture, RunOriginal, SessionCompactFn, - ToolCallContext, ToolExecutor, ToolHook, ToolHookFuture, ToolOriginal, ToolRequest, INLINE_CAP, + HookRunContext, RunConfig, RunEvent, RunEventContext, RunEventHook, RunEventHookResult, + RunExecutor, RunHook, RunHookFuture, RunOriginal, SessionCompactFn, ToolCallContext, + ToolExecutor, ToolHook, ToolHookFuture, ToolOriginal, ToolRequest, INLINE_CAP, }; use std::fmt; use std::sync::Arc; @@ -13,6 +14,7 @@ use tinyvec::TinyVec; pub struct HookSet { pub(super) tool_hooks: Vec>, pub(super) run_hooks: Vec>, + pub(super) run_event_hooks: Vec>, pub(super) session_compact: TinyVec<[Option; INLINE_CAP]>, } @@ -21,7 +23,10 @@ impl HookSet { #[inline] #[must_use] pub fn is_empty(&self) -> bool { - self.tool_hooks.is_empty() && self.run_hooks.is_empty() && self.session_compact.is_empty() + self.tool_hooks.is_empty() + && self.run_hooks.is_empty() + && self.run_event_hooks.is_empty() + && self.session_compact.is_empty() } /// Returns `true` if no tool hooks are registered. @@ -38,6 +43,13 @@ impl HookSet { self.run_hooks.is_empty() } + /// Returns `true` if no run-event hooks are registered. + #[inline] + #[must_use] + pub fn run_event_hooks_is_empty(&self) -> bool { + self.run_event_hooks.is_empty() + } + /// Returns registered tool hooks in dispatch order. #[inline] #[must_use] @@ -95,6 +107,36 @@ impl HookSet { RunOriginal::new(&self.run_hooks, real_run).call(ctx, config) } + /// Dispatches one streamed run event through the run-event hook chain. + /// + /// Hooks apply in registration order; each hook receives the + /// previous hook's output event. A suppression from any hook ends + /// the chain for that event. If no run-event hooks are registered, + /// the event is returned unchanged without entering the chain. + /// + /// # Errors + /// Returns [`ToolError`] if any hook in the chain returns an error; + /// dispatch stops at the first error. + /// + /// [`ToolError`]: crate::ToolError + #[inline] + pub fn dispatch_run_event( + &self, + ctx: &RunEventContext<'_>, + mut event: RunEvent, + ) -> RunEventHookResult { + if self.run_event_hooks.is_empty() { + return Ok(Some(event)); + } + for hook in &self.run_event_hooks { + match hook.hook(ctx, event)? { + Some(next) => event = next, + None => return Ok(None), + } + } + Ok(Some(event)) + } + /// Dispatches compact events. Name preserved — compact is its own concept, distinct from "run". #[inline] pub fn dispatch_session_compact(&self, ctx: &HookRunContext<'_>) { @@ -109,6 +151,7 @@ impl fmt::Debug for HookSet { f.debug_struct("HookSet") .field("tool_hooks", &self.tool_hooks.len()) .field("run_hooks", &self.run_hooks.len()) + .field("run_event_hooks", &self.run_event_hooks.len()) .field("session_compact", &self.session_compact.len()) .finish() } @@ -117,10 +160,11 @@ impl fmt::Debug for HookSet { #[cfg(test)] mod tests { use super::*; + use crate::hooks::run_event::{RunEvent, RunEventContext, RunEventHook, RunEventHookResult}; use crate::hooks::run_hook::{ EndReason, RunConfig, RunExecutor, RunHook, RunHookFuture, RunOriginal, RunOutput, RunUsage, }; - use crate::ToolOutput; + use crate::{ToolError, ToolOutput}; use serde_json::json; use std::sync::atomic::{AtomicUsize, Ordering}; @@ -525,6 +569,150 @@ mod tests { ); } + // --- Run event dispatch tests --------------------------------------------- + + fn event_ctx() -> RunEventContext<'static> { + RunEventContext { + agent_name: "coder", + model_name: "gpt-5.6-luna", + } + } + + #[test] + fn dispatch_run_event_empty_chain_returns_event_unchanged() { + let hooks = HookSet::default(); + assert!(hooks.run_event_hooks_is_empty()); + let event = RunEvent::TextDelta { text: "hi".into() }; + let decision = hooks + .dispatch_run_event(&event_ctx(), event.clone()) + .unwrap(); + assert_eq!(decision, Some(event)); + } + + #[test] + fn dispatch_run_event_rewrite_publishes_rewritten_event() { + struct UpperCase; + impl RunEventHook for UpperCase { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { text } => Ok(Some(RunEvent::TextDelta { + text: text.to_uppercase(), + })), + other => Ok(Some(other)), + } + } + } + + let hooks = HookSet::builder().run_event_hook(UpperCase).build(); + let decision = hooks + .dispatch_run_event(&event_ctx(), RunEvent::TextDelta { text: "raw".into() }) + .unwrap(); + assert_eq!(decision, Some(RunEvent::TextDelta { text: "RAW".into() })); + } + + #[test] + fn dispatch_run_event_suppression_stops_the_chain() { + struct SuppressText; + impl RunEventHook for SuppressText { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { .. } => Ok(None), + other => Ok(Some(other)), + } + } + } + struct MustNotRun; + impl RunEventHook for MustNotRun { + fn hook(&self, _ctx: &RunEventContext<'_>, _event: RunEvent) -> RunEventHookResult { + panic!("later hooks must not see a suppressed event"); + } + } + + let hooks = HookSet::builder() + .run_event_hook(SuppressText) + .run_event_hook(MustNotRun) + .build(); + let decision = hooks + .dispatch_run_event( + &event_ctx(), + RunEvent::TextDelta { + text: "secret".into(), + }, + ) + .unwrap(); + assert_eq!(decision, None); + } + + #[test] + fn dispatch_run_event_applies_hooks_in_registration_order() { + struct Tag(&'static str); + impl RunEventHook for Tag { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { text } => Ok(Some(RunEvent::TextDelta { + text: format!("{text}-{}", self.0), + })), + other => Ok(Some(other)), + } + } + } + + let hooks = HookSet::builder() + .run_event_hook(Tag("first")) + .run_event_hook(Tag("second")) + .build(); + let decision = hooks + .dispatch_run_event(&event_ctx(), RunEvent::TextDelta { text: "x".into() }) + .unwrap(); + // The second hook must see the first hook's rewrite, proving + // registration order. + assert_eq!( + decision, + Some(RunEvent::TextDelta { + text: "x-first-second".into() + }) + ); + } + + #[test] + fn dispatch_run_event_error_propagates_and_stops_the_chain() { + struct Fail; + impl RunEventHook for Fail { + fn hook(&self, _ctx: &RunEventContext<'_>, _event: RunEvent) -> RunEventHookResult { + Err(ToolError::validation("hook rejected the event")) + } + } + struct MustNotRun; + impl RunEventHook for MustNotRun { + fn hook(&self, _ctx: &RunEventContext<'_>, _event: RunEvent) -> RunEventHookResult { + panic!("later hooks must not run after a hook error"); + } + } + + let hooks = HookSet::builder() + .run_event_hook(Fail) + .run_event_hook(MustNotRun) + .build(); + let error = hooks + .dispatch_run_event(&event_ctx(), RunEvent::OutputReady) + .unwrap_err(); + assert!(matches!(error, ToolError::Validation { .. })); + } + + #[test] + fn hook_set_debug_includes_run_event_hooks_count() { + struct Passthrough; + impl RunEventHook for Passthrough { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + Ok(Some(event)) + } + } + + let hooks = HookSet::builder().run_event_hook(Passthrough).build(); + let debug = format!("{hooks:?}"); + assert!(debug.contains("run_event_hooks: 1")); + } + #[tokio::test] async fn session_compact_dispatch_untouched() { static COMPACTS: AtomicUsize = AtomicUsize::new(0); diff --git a/src/reloaded-code-core/src/hooks/mod.rs b/src/reloaded-code-core/src/hooks/mod.rs index 05ec5e13..a2ddfc67 100644 --- a/src/reloaded-code-core/src/hooks/mod.rs +++ b/src/reloaded-code-core/src/hooks/mod.rs @@ -1,4 +1,5 @@ -//! Hook infrastructure for tool hooks and run lifecycle hooks. +//! Hook infrastructure for tool hooks, run lifecycle hooks, and run +//! event hooks. //! //! # Public API //! @@ -27,6 +28,10 @@ //! - [`RunToolCallSummary`] - Distilled tool call summary //! - [`RunToolResultSummary`] - Distilled tool result summary //! +//! Run event hook types: +//! - [`RunEventHook`] - Observes, rewrites, or suppresses streamed run events +//! - [`RunEventContext`] - Agent and model names for a run-event hook call +//! //! Observers are plain hooks: code before `original` is "start", code //! after is "end". They participate in the same hook chain. //! 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 5a23f8de..8c392cee 100644 --- a/src/reloaded-code-core/src/hooks/run_event/mod.rs +++ b/src/reloaded-code-core/src/hooks/run_event/mod.rs @@ -1,9 +1,16 @@ -//! Run event types: the framework-owned streaming item type. +//! Run event types: the framework-owned streaming item type, plus the +//! hook that intercepts each event before publication. //! //! [`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. //! +//! # Run-event hooks +//! +//! [`RunEventHook`] sees each streamed event before publication: +//! observe, rewrite, or suppress. It fires only on the streaming +//! path; the run boundary hook [`RunHook`] fires only on `run()`. +//! //! # Transcript distillation //! //! [`RunEvent::RunComplete`] carries a distilled transcript @@ -16,9 +23,31 @@ //! //! [`RunEvent`] is `#[non_exhaustive]`: variants may be appended //! without a breaking release. Consumers match it with a wildcard arm. +//! +//! [`RunHook`]: crate::hooks::RunHook +use crate::ToolError; use serde::{Deserialize, Serialize}; +/// Static context for a run-event hook call. +/// +/// Names only. The run id is deliberately absent: it is learnable +/// from the [`RunEvent::RunStart`] and [`RunEvent::RunComplete`] +/// events the stream itself yields. +#[derive(Debug)] +pub struct RunEventContext<'a> { + /// Name of the agent whose stream produced the event. + pub agent_name: &'a str, + /// Name of the model generating the event stream. + pub model_name: &'a str, +} + +/// Publish decision for one event after the run-event hook chain. +/// +/// `Ok(Some(event))` publishes the event, possibly rewritten by a +/// hook; `Ok(None)` suppresses it; `Err` carries the first hook error. +pub type RunEventHookResult = Result, ToolError>; + /// Framework-owned event yielded by a run stream. /// /// One variant per observable streaming milestone: run start, step @@ -210,6 +239,31 @@ pub struct RunToolResultSummary { pub output: String, } +/// Mode-scoped hook for streamed run events. +/// +/// Fires only on the streaming path: each event a run stream yields +/// passes every registered run-event hook, in registration order, +/// before the stream consumer sees it. +/// +/// Per event, a hook may: +/// - observe: return the event unchanged, +/// - rewrite: return a changed event, +/// - suppress: return `Ok(None)`. +/// +/// Hooks run synchronously at token rate, so per-event work stays +/// cheap (plain string transforms). Each call sees exactly one event; +/// a hook needing cross-event context buffers it internally. +/// +/// [`RunHook`]: crate::hooks::RunHook +pub trait RunEventHook: Send + Sync + 'static { + /// Observes, rewrites, or suppresses one streamed event. + /// + /// # Errors + /// Returns `ToolError` when the hook fails; dispatch stops at the + /// first error and returns it to the caller. + fn hook(&self, ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult; +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/reloaded-code-core/src/hooks/run_hook/mod.rs b/src/reloaded-code-core/src/hooks/run_hook/mod.rs index 34f23f27..ebb5816e 100644 --- a/src/reloaded-code-core/src/hooks/run_hook/mod.rs +++ b/src/reloaded-code-core/src/hooks/run_hook/mod.rs @@ -149,6 +149,13 @@ pub trait RunExecutor: Send + Sync { /// takes ownership, mutates, and passes to `original.call()`. The final /// [`RunExecutor`] consumes it: strings move into the framework's run /// options with zero clones. +/// +/// # Remarks +/// +/// Fires only on the `run()` path - never on streaming runs. +/// Use [`RunEventHook`] to intercept streamed events. +/// +/// [`RunEventHook`]: crate::hooks::RunEventHook pub trait RunHook: Send + Sync + 'static { /// Intercepts a run. /// diff --git a/src/reloaded-code-serdesai/Cargo.toml b/src/reloaded-code-serdesai/Cargo.toml index a4e41a33..0b730f6f 100644 --- a/src/reloaded-code-serdesai/Cargo.toml +++ b/src/reloaded-code-serdesai/Cargo.toml @@ -113,6 +113,11 @@ name = "serdesai-run-chain" path = "examples/hooks/run/serdesai-run-chain.rs" required-features = ["mock"] +[[example]] +name = "serdesai-run-event-hook" +path = "examples/hooks/run/serdesai-run-event-hook.rs" +required-features = ["mock"] + [[example]] name = "serdesai-tool-hook" path = "examples/hooks/tool/serdesai-tool-hook.rs" diff --git a/src/reloaded-code-serdesai/examples/hooks/README.MD b/src/reloaded-code-serdesai/examples/hooks/README.MD index 012899a4..f8205b97 100644 --- a/src/reloaded-code-serdesai/examples/hooks/README.MD +++ b/src/reloaded-code-serdesai/examples/hooks/README.MD @@ -11,6 +11,12 @@ Hooks let your code see or change what an agent does. - Observe start and end: code before `original` is "start", code after is "end" (logging, metrics). +- `RunEventHook` trait + - Fires only on `run_stream()`: observe, rewrite, or suppress each + streamed event before the consumer sees it. + - The scoping is symmetric: `RunHook`s stay inert on the streaming + path. + ### Example programs - serdesai-run-hook @@ -22,6 +28,13 @@ Hooks let your code see or change what an agent does. - Registration order A -> B gives execution A-before, B-before, executor, B-after, A-after. - `cargo run --example serdesai-run-chain -p reloaded-code-serdesai --features mock` +- serdesai-run-event-hook + - Two `RunEventHook`s on one streaming run: one rewrites each text + delta to uppercase, the other suppresses the output-ready + milestone; the printed stream shows the rewritten text and the + missing milestone. + - `cargo run --example serdesai-run-event-hook -p reloaded-code-serdesai --features mock` + ## Tool hooks - `ToolHook` trait diff --git a/src/reloaded-code-serdesai/examples/hooks/run/serdesai-run-event-hook.rs b/src/reloaded-code-serdesai/examples/hooks/run/serdesai-run-event-hook.rs new file mode 100644 index 00000000..a47f37f7 --- /dev/null +++ b/src/reloaded-code-serdesai/examples/hooks/run/serdesai-run-event-hook.rs @@ -0,0 +1,98 @@ +//! `RunEventHook` interception on a streaming run with a mock model. +//! +//! This example registers two `RunEventHook`s via +//! `AgentRuntimeBuilder::hooks()`, builds an agent with +//! `AgentBuildContext::with_model_override()` using a mock model, and +//! consumes `HookedAgent::run_stream()`. +//! +//! One hook rewrites each text delta to uppercase; the other suppresses the +//! output-ready milestone. The printed stream shows the rewritten text where +//! the mock's raw response would be, and no output-ready line. +//! +//! Mode scoping: `RunEventHook` fires only on `run_stream()`. A +//! registered `RunHook` stays inert on this path, so this example +//! registers none. +//! +//! Expected output: +//! Built agent with 0 tools. +//! run started +//! [UppercaseDeltas] "Mock response" -> "MOCK RESPONSE" +//! text: "MOCK RESPONSE" +//! [SuppressOutputReady] dropping output-ready +//! run complete +//! +//! Run with: +//! cargo run --example serdesai-run-event-hook -p reloaded-code-serdesai --features mock + +use futures::StreamExt; +use reloaded_code_agents::AgentCatalog; +use reloaded_code_core::{HookSet, RunEventContext, RunEventHookResult}; +use reloaded_code_serdesai::{RunEvent, RunEventHook}; + +#[path = "../shared.rs"] +mod shared; + +/// Suppresses the output-ready milestone; the consumer never sees it. +struct SuppressOutputReady; + +/// Rewrites every streamed text delta to uppercase before publication. +struct UppercaseDeltas; + +impl RunEventHook for SuppressOutputReady { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + if matches!(event, RunEvent::OutputReady) { + println!("[SuppressOutputReady] dropping output-ready"); + return Ok(None); + } + Ok(Some(event)) + } +} + +impl RunEventHook for UppercaseDeltas { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { text } => { + let upper = text.to_uppercase(); + println!("[UppercaseDeltas] {text:?} -> {upper:?}"); + Ok(Some(RunEvent::TextDelta { text: upper })) + } + other => Ok(Some(other)), + } + } +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + let hooks = HookSet::builder() + .run_event_hook(UppercaseDeltas) + .run_event_hook(SuppressOutputReady) + .build(); + + let catalog = AgentCatalog::from_entries([shared::agent_config( + "event-hook-demo", + "event hook demo", + "You are an event hook demo agent.", + )]); + + let build_context = shared::build_agent_context(catalog, hooks); + + let model = shared::mock_model(); + let agent = build_context + .with_model_override(model) + .build("event-hook-demo")?; + println!("Built agent with {} tools.", agent.tools().len()); + + let mut stream = agent.run_stream("Say hello.", ()).await?; + while let Some(item) = stream.next().await { + match item? { + RunEvent::RunStart { .. } => println!("run started"), + RunEvent::TextDelta { text } => println!("text: {text:?}"), + RunEvent::OutputReady => println!("output ready"), + RunEvent::RunComplete { .. } => println!("run complete"), + // Step and context telemetry still flows through the hooks; + // printing it is skipped to keep the output short. + _ => {} + } + } + Ok(()) +} diff --git a/src/reloaded-code-serdesai/src/agent_runtime/mod.rs b/src/reloaded-code-serdesai/src/agent_runtime/mod.rs index f1932ac8..c996a5ae 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/mod.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/mod.rs @@ -7,7 +7,8 @@ //! # Public API //! - [`AgentBuildContext`] - Shared context that builds runnable agents by name. //! - [`HookedAgent`] - Built agent wrapper that dispatches `run()` through run -//! hooks and streams framework-owned events from `run_stream()`. +//! hooks and streams framework-owned events from `run_stream()`, passing +//! each through the registered run-event hooks. //! - [`AgentBuildError`] - Build-time failures. pub use build::AgentBuildError; 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 81371e91..4615543c 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/stream_events.rs @@ -5,6 +5,14 @@ //! event types stay inside this module; consumers of //! [`HookedAgent::run_stream`][task] only ever see [`RunEvent`] items. //! +//! # Hooks +//! +//! With run-event hooks registered, each mapped event passes the +//! [`RunEventHook`] chain before being yielded; a hook may rewrite the +//! event, suppress it by returning `None`, or fail and end the stream +//! with one [`AgentRunError`][error] item. Without hooks, polling maps +//! and yields directly. +//! //! # Optional events //! //! Step boundaries, context telemetry, and streamed tool-call @@ -21,11 +29,14 @@ //! variant fails compilation here, keeping vendor coupling inside //! this module. //! +//! [`RunEventHook`]: reloaded_code_core::hooks::RunEventHook +//! [error]: serdes_ai::agent::AgentRunError //! [task]: super::task::HookedAgent::run_stream use futures::{Stream, StreamExt}; use reloaded_code_core::hooks::{ - RunEvent, RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary, + HookSet, RunEvent, RunEventContext, RunMessage, RunMessageRole, RunToolCallSummary, + RunToolResultSummary, }; use serdes_ai::core::messages::{ AudioContent, DocumentContent, FileContent, ImageContent, RetryContent, ToolCallArgs, @@ -44,15 +55,58 @@ 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. +/// +/// When run-event hooks are registered, each mapped event passes the +/// [`RunEventHook`] chain before it is yielded: a hook may rewrite or +/// suppress it, and a hook failure yields one [`AgentRunError::Other`] +/// item after which the stream ends. +/// +/// [`RunEventHook`]: reloaded_code_core::hooks::RunEventHook +/// [`AgentRunError::Other`]: serdes_ai::agent::AgentRunError::Other pub(super) struct RunEventStream { /// Owned vendor stream, driven by the vendor's own background task. inner: AgentStream, + /// Run-event hook chain plus the names each hook call receives. + /// `None` when no run-event hooks are registered, so polling maps + /// and yields directly without hook work. + dispatch: Option, + /// Set once a run-event hook failure was yielded; every later poll + /// ends the stream. + terminated: bool, +} + +/// Owned run-event hook dispatch state for one stream. +struct RunEventDispatch { + /// Registered hooks; only the run-event chain is consulted here. + hooks: HookSet, + /// Static agent and model names each hook context carries. + agent_name: String, + model_name: String, } impl RunEventStream { - /// Wraps an already-started vendor stream. - pub(super) fn new(inner: AgentStream) -> Self { - Self { inner } + /// Wraps an already-started vendor stream, applying the agent's + /// run-event hook chain to each mapped event. + /// + /// The stream takes ownership of a hook chain only when one is + /// registered; an empty chain keeps polling on the direct mapping + /// path with no per-event hook work. + pub(super) fn new( + inner: AgentStream, + hooks: &HookSet, + agent_name: &str, + model_name: &str, + ) -> Self { + let dispatch = (!hooks.run_event_hooks_is_empty()).then(|| RunEventDispatch { + hooks: hooks.clone(), + agent_name: agent_name.to_owned(), + model_name: model_name.to_owned(), + }); + Self { + inner, + dispatch, + terminated: false, + } } } @@ -60,16 +114,59 @@ 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, + let this = self.get_mut(); + if this.terminated { + return Poll::Ready(None); + } + let Some(dispatch) = this.dispatch.as_ref() else { + // Empty chain: identical to the direct mapping path. + return match this.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, + }; + }; + // A suppressed event must not surface, so keep polling until an + // event publishes; the surrounding events keep their order. + loop { + match this.inner.poll_next_unpin(cx) { + Poll::Pending => return Poll::Pending, + Poll::Ready(None) => return Poll::Ready(None), + Poll::Ready(Some(Err(error))) => return Poll::Ready(Some(Err(error))), + Poll::Ready(Some(Ok(event))) => { + let ctx = RunEventContext { + agent_name: &dispatch.agent_name, + model_name: &dispatch.model_name, + }; + match dispatch + .hooks + .dispatch_run_event(&ctx, map_vendor_event(event)) + { + Ok(Some(event)) => return Poll::Ready(Some(Ok(event))), + Ok(None) => continue, + Err(error) => { + this.terminated = true; + return Poll::Ready(Some(Err(hook_error_item(error)))); + } + } + } + } } } } +/// Converts a run-event hook failure into the stream's error item. +/// +/// Follows the `run()` path's hook-error translation: the failure is +/// labeled hook-origin and carried as [`AgentRunError::Other`]. The +/// stream ends after yielding it. +/// +/// [`AgentRunError::Other`]: serdes_ai::agent::AgentRunError::Other +fn hook_error_item(error: reloaded_code_core::ToolError) -> serdes_ai::agent::AgentRunError { + serdes_ai::agent::AgentRunError::Other(anyhow::anyhow!("run event hook error: {error}")) +} + /// Maps one vendor event to its framework-owned counterpart. /// /// The match is exhaustive over the vendor enum, so vendor drift @@ -358,9 +455,10 @@ mod tests { use futures::StreamExt; use reloaded_code_agents::{AgentCatalog, AgentDefaults, AgentMode, AgentRuntimeBuilder}; use reloaded_code_core::hooks::{ - HookRunContext, HookSet, RunConfig, RunHook, RunHookFuture, RunOriginal, + HookRunContext, HookSet, RunConfig, RunEventContext, RunEventHook, RunEventHookResult, + RunHook, RunHookFuture, RunOriginal, }; - use reloaded_code_core::{ToolCatalogEntry, ToolCatalogKind}; + use reloaded_code_core::{ToolCatalogEntry, ToolCatalogKind, ToolError}; use rstest::rstest; use serde_json::json; use serdes_ai::core::messages::request::RetryPromptPart; @@ -555,7 +653,7 @@ mod tests { let messages = distill_messages(vec![request, follow_up]); - let expected = vec![ + let expected = [ RunMessage { role: RunMessageRole::System, text: Some("sys".into()), @@ -909,7 +1007,7 @@ mod tests { /// Index of the first event matching `predicate`. fn position(events: &[RunEvent], predicate: &dyn Fn(&RunEvent) -> bool) -> Option { - events.iter().position(|event| predicate(event)) + events.iter().position(predicate) } #[tokio::test] @@ -1022,7 +1120,7 @@ mod tests { .iter() .filter(|message| message.role == RunMessageRole::Assistant) .filter_map(|message| message.text.as_deref()) - .last() + .next_back() .expect("closing assistant turn should carry text"); assert_eq!(final_answer, streamed_answer); } @@ -1155,4 +1253,377 @@ mod tests { "inner model failure should keep its variant, got: {error:?}" ); } + + // ======================================================================== + // Run-event hook chain + // ======================================================================== + + /// Uppercases every text delta before publication. + struct UpperCaseDelta; + + impl RunEventHook for UpperCaseDelta { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { text } => Ok(Some(RunEvent::TextDelta { + text: text.to_uppercase(), + })), + other => Ok(Some(other)), + } + } + } + + /// Suppresses every text delta. + struct SuppressTextDelta; + + impl RunEventHook for SuppressTextDelta { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { .. } => Ok(None), + other => Ok(Some(other)), + } + } + } + + /// Fails on the first text delta it sees. + struct FailOnTextDelta; + + impl RunEventHook for FailOnTextDelta { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { .. } => { + Err(ToolError::validation("event hook rejected the delta")) + } + other => Ok(Some(other)), + } + } + } + + /// Appends a fixed tag to every text delta. + struct TagDelta(&'static str); + + impl RunEventHook for TagDelta { + fn hook(&self, _ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + match event { + RunEvent::TextDelta { text } => Ok(Some(RunEvent::TextDelta { + text: format!("{text}-{}", self.0), + })), + other => Ok(Some(other)), + } + } + } + + /// Records the variant name of every event the chain consults plus + /// the static names of the first hook context. Clones share the + /// records so the test can register one copy and inspect another. + #[derive(Clone, Default)] + struct EventObserver { + variants: Arc>>, + context_names: Arc>>, + } + + impl RunEventHook for EventObserver { + fn hook(&self, ctx: &RunEventContext<'_>, event: RunEvent) -> RunEventHookResult { + let variant = match &event { + RunEvent::RunStart { .. } => "RunStart", + RunEvent::ToolCallStart { .. } => "ToolCallStart", + RunEvent::ToolExecuted { .. } => "ToolExecuted", + RunEvent::TextDelta { .. } => "TextDelta", + RunEvent::OutputReady => "OutputReady", + RunEvent::RunComplete { .. } => "RunComplete", + _ => "other", + }; + self.variants + .lock() + .expect("variants should not be poisoned") + .push(variant); + let mut names = self + .context_names + .lock() + .expect("context names should not be poisoned"); + if names.is_none() { + *names = Some((ctx.agent_name.to_string(), ctx.model_name.to_string())); + } + Ok(Some(event)) + } + } + + #[tokio::test] + async fn run_stream_publishes_only_the_rewritten_delta_text() { + const RESPONSE: &str = "secret payload words"; + let hooks = HookSet::builder().run_event_hook(UpperCaseDelta).build(); + let hooked = streamed_agent( + Streamed::new(FunctionModel::new(move |_, _| { + ModelResponse::text(RESPONSE) + })), + hooks, + ); + + let events = collect_events(&hooked, "hello").await; + + // The original text never surfaces, in whole or per chunk. + for event in &events { + if let RunEvent::TextDelta { text } = event { + assert_eq!( + text, + &text.to_uppercase(), + "the consumer must only see the rewritten delta: {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.to_uppercase()); + } + + #[tokio::test] + async fn run_stream_suppression_hides_event_and_keeps_neighbor_order() { + let hooks = HookSet::builder().run_event_hook(SuppressTextDelta).build(); + let hooked = streamed_agent( + Streamed::new(FunctionModel::new(move |_, _| { + ModelResponse::text("hidden words") + })), + hooks, + ); + + let events = collect_events(&hooked, "hello").await; + + assert!( + !events + .iter() + .any(|event| matches!(event, RunEvent::TextDelta { .. })), + "suppressed deltas must never reach the consumer" + ); + // The surrounding milestones keep their order. + assert!(matches!(events.first(), Some(RunEvent::RunStart { .. }))); + let output_ready = position(&events, &|event| matches!(event, RunEvent::OutputReady)) + .expect("output-ready should survive suppression"); + 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 stay before completion" + ); + } + + #[tokio::test] + async fn run_stream_hook_error_yields_other_item_and_ends_the_stream() { + let hooks = HookSet::builder().run_event_hook(FailOnTextDelta).build(); + let hooked = streamed_agent( + Streamed::new(FunctionModel::new(move |_, _| { + ModelResponse::text("first words then more") + })), + hooks, + ); + + let mut stream = hooked + .run_stream("hello", ()) + .await + .expect("stream should start"); + let mut items = Vec::new(); + while let Some(item) = stream.next().await { + items.push(item); + } + + // The error item is the last item the stream ever yields. + let error = match items.last().expect("stream should yield the hook error") { + Err(error) => error, + Ok(event) => panic!("the final item should be the hook error, got: {event:?}"), + }; + match error { + serdes_ai::agent::AgentRunError::Other(source) => { + let message = source.to_string(); + assert!( + message.contains("run event hook error"), + "hook failure should be labeled hook-origin: {message}" + ); + assert!( + message.contains("event hook rejected the delta"), + "the hook's error text should be preserved: {message}" + ); + } + other => panic!("hook failure should surface as Other, got: {other:?}"), + } + assert_eq!( + items.iter().filter(|item| item.is_err()).count(), + 1, + "the hook error must be the only error item" + ); + // Events before the failing delta were published, and the run + // never completes after the failure. + let published: Vec<_> = items.iter().filter_map(|item| item.as_ref().ok()).collect(); + assert!( + matches!(published.first(), Some(RunEvent::RunStart { .. })), + "events before the failing delta should have been published" + ); + assert!( + !published + .iter() + .any(|event| matches!(event, RunEvent::OutputReady | RunEvent::RunComplete { .. })), + "the stream must end before the run completes: {published:?}" + ); + } + + #[tokio::test] + async fn run_stream_applies_run_event_hooks_in_registration_order() { + let hooks = HookSet::builder() + .run_event_hook(TagDelta("first")) + .run_event_hook(TagDelta("second")) + .build(); + let hooked = streamed_agent( + Streamed::new(FunctionModel::new(move |_, _| { + ModelResponse::text("chunked text") + })), + hooks, + ); + + let events = collect_events(&hooked, "hello").await; + + // The second hook sees the first hook's rewrite, so every + // published delta carries the tags in registration order; a + // reversed order would yield "-second-first". + let mut checked_deltas = 0; + for event in &events { + if let RunEvent::TextDelta { text } = event { + checked_deltas += 1; + assert!( + text.ends_with("-first-second"), + "registration order must hold, got: {text}" + ); + } + } + assert!( + checked_deltas > 0, + "the stream should publish at least one TextDelta, got: {events:?}" + ); + } + + #[tokio::test] + async fn run_stream_consults_the_hook_chain_for_every_streamed_event() { + let observer = EventObserver::default(); + // The scripted ping flow streams every milestone kind: run and + // step boundaries, context telemetry, text deltas, and the full + // tool-call lifecycle. + let hooked = streamed_agent_with_ping_tool( + tool_then_text("ping", json!({"target": "example.com"}), "after the tool"), + HookSet::builder().run_event_hook(observer.clone()).build(), + ); + + let events = collect_events(&hooked, "use the tool").await; + + // Non-delta milestones pass the chain too, in published order; + // dispatching only deltas would leave these unobserved. + let variants = observer + .variants + .lock() + .expect("variants should not be poisoned") + .clone(); + for milestone in [ + "RunStart", + "ToolCallStart", + "ToolExecuted", + "OutputReady", + "RunComplete", + ] { + assert!( + variants.contains(&milestone), + "the chain must be consulted for {milestone}: {variants:?}" + ); + } + assert_eq!( + variants.last(), + Some(&"RunComplete"), + "the chain must observe the final event" + ); + assert_eq!( + events.len(), + variants.len(), + "the pass-through hook must observe every published event exactly once" + ); + + // The hook context carries the agent's static names; the model + // name comes from the catalog-resolved model, not the mock + // override that serves the requests. + let (agent_name, model_name) = observer + .context_names + .lock() + .expect("context names should not be poisoned") + .clone() + .expect("the hook should have observed a context"); + assert_eq!(agent_name, "caller"); + assert_eq!(model_name, "openai/gpt-4.1-mini"); + } + + /// Replaces run ids with a placeholder so event sequences from + /// separate streams compare equal; run ids are random per run. + /// + /// `ContextInfo` telemetry is zeroed too: the estimate serializes + /// wall-clock part timestamps, whose RFC 3339 fractional-second + /// width (0/3/6/9 digits) can differ between the two streams, so + /// byte counts drift by a few bytes between otherwise identical + /// requests. + fn normalized_events(events: &[RunEvent]) -> Vec { + events + .iter() + .map(|event| match event { + RunEvent::RunStart { .. } => RunEvent::RunStart { + run_id: "".into(), + }, + RunEvent::RunComplete { messages, .. } => RunEvent::RunComplete { + run_id: "".into(), + messages: messages.clone(), + }, + RunEvent::ContextInfo { context_limit, .. } => RunEvent::ContextInfo { + estimated_tokens: 0, + request_bytes: 0, + context_limit: *context_limit, + }, + other => other.clone(), + }) + .collect() + } + + #[tokio::test] + async fn run_stream_without_run_event_hooks_matches_unhooked_stream() { + const RESPONSE: &str = "equivalence probe text"; + // Same deterministic model twice: once with no hooks at all, once + // with a non-empty hook set that registers no run-event hooks, so + // the empty-chain predicate is the only difference. + let plain = streamed_agent( + Streamed::new(FunctionModel::new(move |_, _| { + ModelResponse::text(RESPONSE) + })), + HookSet::builder().build(), + ); + let inert = streamed_agent( + Streamed::new(FunctionModel::new(move |_, _| { + ModelResponse::text(RESPONSE) + })), + HookSet::builder() + .run_hook(DispatchRecorder { + dispatches: Arc::new(Mutex::new(Vec::new())), + }) + .build(), + ); + + let plain_events = collect_events(&plain, "hello").await; + let inert_events = collect_events(&inert, "hello").await; + + assert_eq!( + normalized_events(&plain_events), + normalized_events(&inert_events), + "an empty run-event chain must stream the unhooked sequence" + ); + } } diff --git a/src/reloaded-code-serdesai/src/agent_runtime/task.rs b/src/reloaded-code-serdesai/src/agent_runtime/task.rs index 1fbefc6d..9b47e158 100644 --- a/src/reloaded-code-serdesai/src/agent_runtime/task.rs +++ b/src/reloaded-code-serdesai/src/agent_runtime/task.rs @@ -3,7 +3,8 @@ //! # Public API //! - [`AgentBuildContext`] - Reusable shared inputs for building runnable agents. //! - [`HookedAgent`] - Built agent wrapper that dispatches `run()` through run -//! hooks and streams framework-owned events from `run_stream()`. +//! hooks and streams framework-owned events from `run_stream()`, passing +//! each through the registered run-event hooks. #[cfg(not(all(feature = "linux-bubblewrap", target_os = "linux")))] use super::build::Profile; @@ -42,7 +43,8 @@ pub struct AgentBuildContext, @@ -325,6 +327,10 @@ impl HookedAgent { /// mutations to the prompt text, applies `model_settings_overrides` to /// the per-run model settings, and returns the result. /// + /// Mode-scoped: run hooks fire only on this path. A registered + /// [`RunEventHook`][event-hook] never fires here; it fires only on + /// [`Self::run_stream`]. + /// /// The run-hook context carries a wrapper-generated `run_id`. The inner /// agent assigns its own id for tool hooks; SerdesAI `RunOptions` has no /// field to override it, so the two identifiers cannot be unified here. @@ -336,6 +342,8 @@ 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. + /// + /// [event-hook]: reloaded_code_core::hooks::RunEventHook pub async fn run( &self, prompt: impl Into, @@ -405,21 +413,30 @@ impl HookedAgent { /// [`UserContent`][serdes_ai::core::UserContent]; image and multi-part /// prompts pass through to the vendor unchanged. /// - /// # Remarks + /// Each mapped event passes the registered + /// [`RunEventHook`][event-hook] chain, in registration order, before the + /// caller sees it, so a hook may rewrite or suppress it. With no + /// run-event hooks registered the stream matches the unhooked mapping. /// - /// 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. + /// Run hooks ([`RunHook`][run-hook]) never fire here; they fire only + /// on [`Self::run`]. The run-hook chain resolves to one completed + /// `RunOutput`, so dispatching it would buffer the whole run before + /// the first event and defeat streaming. Preamble, system-prompt, + /// and model-settings injection therefore apply on that path only. /// /// # Errors /// /// - Returns the inner agent's [`serdes_ai::agent::AgentRunError`] /// 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. + /// - The stream yields the inner error as its final `Err` item when + /// the run fails mid-stream; a vendor error event maps to + /// [`RunEvent::Error`] before that final item. + /// - The stream yields [`serdes_ai::agent::AgentRunError::Other`] as + /// its final item when a run-event hook fails; the stream ends after + /// that item. + /// + /// [event-hook]: reloaded_code_core::hooks::RunEventHook + /// [run-hook]: reloaded_code_core::hooks::RunHook pub async fn run_stream( &self, prompt: impl Into, @@ -429,7 +446,12 @@ impl HookedAgent { serdes_ai::agent::AgentRunError, > { let inner = self.inner.run_stream(prompt, deps).await?; - Ok(Box::pin(RunEventStream::new(inner))) + Ok(Box::pin(RunEventStream::new( + inner, + &self.hooks, + &self.agent_name, + &self.model_name, + ))) } } @@ -1054,7 +1076,7 @@ mod tests { /// Model settings overrides: applied per run, merged over the agent's /// configured settings, with prompt-prepend behavior untouched. - + /// /// Run hook that installs fixed model settings overrides before /// delegating to `original`. struct OverridingRunHook { @@ -1166,17 +1188,18 @@ mod tests { hooked.run("hello", ()).await.expect("run should complete"); - let seen = captured - .lock() - .expect("captured settings should not be poisoned"); - assert_eq!(seen.len(), 1, "one model request should have been made"); - assert_eq!(seen[0].temperature, Some(f64::from(0.9_f32))); - assert_eq!( - seen[0].top_p, - Some(f64::from(0.8_f32)), - "agent-configured top_p should be retained" - ); - drop(seen); + { + let seen = captured + .lock() + .expect("captured settings should not be poisoned"); + assert_eq!(seen.len(), 1, "one model request should have been made"); + assert_eq!(seen[0].temperature, Some(f64::from(0.9_f32))); + assert_eq!( + seen[0].top_p, + Some(f64::from(0.8_f32)), + "agent-configured top_p should be retained" + ); + } // Mirror direction: a top_p-only override replaces top_p and keeps // the agent-configured temperature. @@ -1187,16 +1210,18 @@ mod tests { hooked.run("hello", ()).await.expect("run should complete"); - let seen = captured - .lock() - .expect("captured settings should not be poisoned"); - assert_eq!(seen.len(), 1, "one model request should have been made"); - assert_eq!(seen[0].top_p, Some(f64::from(0.6_f32))); - assert_eq!( - seen[0].temperature, - Some(f64::from(0.3_f32)), - "agent-configured temperature should be retained" - ); + { + let seen = captured + .lock() + .expect("captured settings should not be poisoned"); + assert_eq!(seen.len(), 1, "one model request should have been made"); + assert_eq!(seen[0].top_p, Some(f64::from(0.6_f32))); + assert_eq!( + seen[0].temperature, + Some(f64::from(0.3_f32)), + "agent-configured temperature should be retained" + ); + } } #[tokio::test] diff --git a/src/reloaded-code-serdesai/src/lib.rs b/src/reloaded-code-serdesai/src/lib.rs index 7eeb2931..0b5f1fdf 100644 --- a/src/reloaded-code-serdesai/src/lib.rs +++ b/src/reloaded-code-serdesai/src/lib.rs @@ -36,9 +36,12 @@ pub use reloaded_code_agents::{ /// Re-export [`RunEvent`], the framework-owned item type yielded by /// [`HookedAgent::run_stream`], together with its transcript payload /// types ([`RunMessage`], [`RunMessageRole`], [`RunToolCallSummary`], -/// [`RunToolResultSummary`]). +/// [`RunToolResultSummary`]), and [`RunEventHook`], the hook that +/// intercepts each streamed event before publication, together with its +/// [`RunEventContext`] and [`RunEventHookResult`] call types. pub use reloaded_code_core::hooks::{ - RunEvent, RunMessage, RunMessageRole, RunToolCallSummary, RunToolResultSummary, + RunEvent, RunEventContext, RunEventHook, RunEventHookResult, RunMessage, RunMessageRole, + RunToolCallSummary, RunToolResultSummary, }; pub mod agent_ext;