feat(harness): invoke_stream + sub-agent delta propagation + scripted mock streaming (#17)

* feat(harness): invoke_stream — caller-consumable run event stream

Adds AgentHarness::invoke_stream, the streaming counterpart to
invoke_streaming that returns an async Stream of the run's events instead of
only the final AgentRun.

The run's EventSink already fans out every AgentEvent in a single ordered
sequence (model/reasoning deltas, ToolStarted/ToolCompleted lifecycle,
SubAgentStarted/Completed — the latter reach the stream because sub-agents
share the parent's sink). invoke_stream subscribes a ChannelListener that
forwards each EventRecord into an unbounded channel, drives the shared
streaming loop path, and yields AgentStreamItem::Event(..) for each event
followed by exactly one terminal AgentStreamItem::Completed(run) /
Failed(err). It is a thin projection over the same emit fan-out, not a
parallel event system; the run advances only while the stream is polled.

Lineage rides the events themselves (ModelDelta stamps run_id; sub-agent
events stamp depth). Streaming a child agent's own model deltas into the
parent — which needs the streaming flag routed through the sub-agent runner
(children run the non-streaming path today) — is tracked as follow-up.

Tests: event-then-Completed ordering, tool-lifecycle surfacing
(ToolStarted before ToolCompleted), and the Failed terminal on a limit trip.

Ref: docs/tinyagents-vendor-improvement-plan/03-streaming.md Step 1 (V3)

* feat(subagent): propagate child model deltas to the parent stream

A streaming parent (invoke_stream) previously saw a sub-agent's start/complete
lifecycle but none of the child's own model/reasoning deltas: run_child always
drove the child through the unary invoke_in_context path.

Thread the drive mode down to the child so it matches the parent:
- RunContext gains a streaming flag, set by the loop driver (drive).
- ToolExecutionContext carries it (from_run_context), so a tool sees whether
  its caller run is streaming.
- SubAgentTool::call_with_context passes context.streaming to run_child, and
  invoke_in_parent forwards parent.streaming; run_child then drives the child
  through invoke_streaming_in_context when streaming.

Because sub-agents already share the parent's EventSink, the child's deltas
then land on the parent stream stamped with the child's own run_id and depth.
A non-streaming parent leaves streaming=false, so its event stream is
unchanged (direct SubAgent::invoke/invoke_with_events stay non-streaming).

Tests: nested run asserts child ModelDeltas (run id 'researcher-d1…') arrive
before the parent's RunCompleted, plus SubAgentStarted/Completed at depth 1;
and that a non-streaming parent emits no child deltas.

Ref: docs/tinyagents-vendor-improvement-plan/03-streaming.md Step 2 (V3)

* feat(mock): MockModel::streaming_script for incremental streaming tests

Adds a MockBehavior::StreamScript(Vec<ModelStreamItem>) variant and the
MockModel::streaming_script constructor. stream() emits the scripted items
verbatim — so tests exercise truly incremental streaming (fine-grained
text/reasoning/tool-call deltas in an exact order) rather than the one-or-two
synthetic deltas the other constructors replay — while invoke() folds the same
items through a StreamAccumulator to produce the equivalent unary response.

Tests: invoke_stream over a scripted reasoning+text stream surfaces the
reasoning delta and multiple incremental text deltas; invoke folds a
delta-only script into the reconstructed response.

Ref: docs/tinyagents-vendor-improvement-plan/03-streaming.md Step 4 (V3)

Author: senamakel

src/graph/goals/test.rs

@@ -274,6 +274,7 @@ mod tool_tests {
             depth: 0,
             max_turn_output_tokens: None,
             events: EventSink::new(),
+            streaming: false,
             workspace: None,
         }
     }

src/graph/todos/test.rs

@@ -376,6 +376,7 @@ mod tool_tests {
             depth: 0,
             max_turn_output_tokens: None,
             events: EventSink::new(),
+            streaming: false,
             workspace: None,
         }
     }

src/harness/agent_loop/entry.rs

@@ -177,6 +177,10 @@ impl<State: Send + Sync, Ctx: Send + Sync> AgentHarness<State, Ctx> {
     ) -> Result<AgentLoopResult> {
         let run_id = ctx.config.run_id.clone();
         let thread_id = ctx.config.thread_id.clone();
+        // Record the drive mode so tool execution contexts (and the sub-agents
+        // they spawn) can match it — child deltas then propagate to the parent
+        // stream via the shared sink.
+        ctx.streaming = streaming;
 
         let mut status = HarnessRunStatus::new(run_id.clone(), ComponentId::new("agent_loop"));
         if let Some(thread) = thread_id {

src/harness/agent_loop/mod.rs

@@ -94,7 +94,10 @@ use serde_json::Value;
 mod entry;
 mod model_call;
 mod run_loop;
+mod stream;
 mod tools;
 
+pub use stream::AgentStreamItem;
+
 #[cfg(test)]
 mod test;

src/harness/agent_loop/stream.rs

@@ -0,0 +1,178 @@
+//! Caller-consumable streaming entry point ([`AgentHarness::invoke_stream`]).
+//!
+//! [`AgentHarness::invoke_streaming`] drives each model call through
+//! [`ChatModel::stream`][crate::harness::model::ChatModel::stream] but only
+//! returns the fully-accumulated [`AgentRun`] once the run is over — a caller
+//! that wants to *watch* the run unfold has to attach an
+//! [`EventListener`][crate::harness::events::EventListener] or middleware.
+//!
+//! `invoke_stream` closes that gap: it returns an async
+//! [`Stream`][futures::Stream] of [`AgentStreamItem`]s that yields every
+//! [`AgentEvent`][crate::harness::events::AgentEvent] emitted during the run —
+//! model/reasoning deltas, tool-call `ToolStarted`/`ToolCompleted` lifecycle
+//! events, and sub-agent `SubAgentStarted`/`SubAgentCompleted` events (which
+//! reach the stream because sub-agents share the parent's
+//! [`EventSink`][crate::harness::events::EventSink]) — and finishes with a
+//! terminal item carrying the completed [`AgentRun`] (or the failure).
+//!
+//! It is a thin projection over the same ordered
+//! [`EventSink::emit`][crate::harness::events::EventSink::emit] fan-out the
+//! `EventListener` path already uses, not a parallel event system: a listener
+//! forwards each [`EventRecord`] into an unbounded channel that the returned
+//! stream drains. Lineage is carried by the events themselves (`ModelDelta`
+//! stamps `run_id`; sub-agent events stamp `depth`). Streaming a child agent's
+//! own model deltas into the parent requires routing the streaming flag through
+//! the sub-agent runner and is tracked as follow-up work.
+
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+
+use crate::error::Result;
+use crate::harness::context::{RunConfig, RunContext};
+use crate::harness::events::{EventListener, EventRecord};
+use crate::harness::message::Message;
+use crate::harness::middleware::AgentRun;
+use crate::harness::runtime::AgentHarness;
+
+use super::AgentLoopResult;
+
+/// One item yielded by [`AgentHarness::invoke_stream`].
+///
+/// The stream yields zero or more [`AgentStreamItem::Event`]s in emission
+/// order, followed by exactly one terminal item — either
+/// [`AgentStreamItem::Completed`] carrying the final [`AgentRun`], or
+/// [`AgentStreamItem::Failed`] carrying the error string. No items are produced
+/// after the terminal item.
+#[derive(Clone, Debug)]
+pub enum AgentStreamItem {
+    /// A live event emitted during the run. Carries the full
+    /// [`EventRecord`] (id + offset + [`AgentEvent`][crate::harness::events::AgentEvent])
+    /// so consumers keep ordering and lineage fields (`run_id`, `depth`).
+    Event(EventRecord),
+    /// Terminal: the run completed successfully. Boxed because [`AgentRun`] is
+    /// large relative to the event variant.
+    Completed(Box<AgentRun>),
+    /// Terminal: the run failed; carries the error rendered as a string.
+    Failed(String),
+}
+
+/// An [`EventListener`] that forwards every [`EventRecord`] into an unbounded
+/// channel. `on_event` is synchronous and must not block, so the channel is
+/// unbounded; in practice its depth is bounded by run progress (the loop awaits
+/// the network between emits). Send failures (receiver dropped) are ignored so
+/// a dropped stream never disturbs the run.
+struct ChannelListener {
+    tx: tokio::sync::mpsc::UnboundedSender<EventRecord>,
+}
+
+impl EventListener for ChannelListener {
+    fn on_event(&self, record: &EventRecord) {
+        let _ = self.tx.send(record.clone());
+    }
+}
+
+/// Maps a finished run result onto its terminal [`AgentStreamItem`].
+fn terminal_item(result: Result<AgentLoopResult>) -> AgentStreamItem {
+    match result {
+        Ok(loop_result) => AgentStreamItem::Completed(Box::new(loop_result.run)),
+        Err(error) => AgentStreamItem::Failed(error.to_string()),
+    }
+}
+
+/// Drive-phase for the streaming state machine.
+enum Phase<'a> {
+    /// The run future is still executing.
+    Running(Pin<Box<dyn Future<Output = Result<AgentLoopResult>> + 'a>>),
+    /// The run has finished; drain any buffered events, then emit `terminal`.
+    Draining(AgentStreamItem),
+    /// Terminal item already emitted; the stream is exhausted.
+    Done,
+}
+
+impl<State: Send + Sync, Ctx: Send + Sync + 'static> AgentHarness<State, Ctx> {
+    /// Runs the agent loop while streaming every emitted event to the caller.
+    ///
+    /// Returns a [`Stream`][futures::Stream] of [`AgentStreamItem`]s: live
+    /// [`AgentStreamItem::Event`]s in emission order (model/reasoning deltas,
+    /// tool-call lifecycle, sub-agent lifecycle, usage, …) followed by a single
+    /// terminal [`AgentStreamItem::Completed`] / [`AgentStreamItem::Failed`].
+    /// Driving the loop and consuming the stream are the same task: the run
+    /// only makes progress while the stream is polled, so a caller that stops
+    /// polling pauses the run (and dropping the stream cancels it).
+    ///
+    /// This is the streaming counterpart of
+    /// [`AgentHarness::invoke_streaming`]; the run itself is identical (each
+    /// model call is driven through
+    /// [`ChatModel::stream`][crate::harness::model::ChatModel::stream]).
+    pub fn invoke_stream<'a>(
+        &'a self,
+        state: &'a State,
+        ctx_data: Ctx,
+        config: RunConfig,
+        input: Vec<Message>,
+    ) -> impl futures::Stream<Item = AgentStreamItem> + 'a {
+        let ctx = RunContext::new(config, ctx_data);
+        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
+        // Subscribe before driving so no event (starting with `RunStarted`) is
+        // missed. The listener rides the run's `EventSink`, which sub-agents
+        // clone, so their lifecycle events reach this stream too.
+        ctx.events.subscribe(Arc::new(ChannelListener { tx }));
+
+        // `invoke_streaming_in_context_with_status` is the public wrapper over
+        // the shared `drive(.., streaming = true)` path; it drives *our* `ctx`
+        // (with the listener already attached) and hands back the terminal
+        // `AgentLoopResult`.
+        let run_fut: Pin<Box<dyn Future<Output = Result<AgentLoopResult>> + 'a>> =
+            Box::pin(self.invoke_streaming_in_context_with_status(state, ctx, input));
+
+        futures::stream::unfold(
+            (Phase::Running(run_fut), rx),
+            |(phase, mut rx)| async move {
+                match phase {
+                    Phase::Running(mut run_fut) => {
+                        tokio::select! {
+                            biased;
+                            // Prefer draining ready events so the consumer sees
+                            // fine-grained progress rather than a late burst.
+                            maybe = rx.recv() => match maybe {
+                                Some(record) => {
+                                    Some((AgentStreamItem::Event(record), (Phase::Running(run_fut), rx)))
+                                }
+                                None => {
+                                    // All senders dropped (the run's context —
+                                    // and every sub-agent clone of the sink —
+                                    // is gone): the run is finishing. Await it
+                                    // for the terminal item.
+                                    let terminal = terminal_item(run_fut.await);
+                                    Some((terminal, (Phase::Done, rx)))
+                                }
+                            },
+                            result = &mut run_fut => {
+                                // The run finished. Events emitted during this
+                                // final poll may still be buffered; drain them
+                                // ahead of the terminal item.
+                                let terminal = terminal_item(result);
+                                match rx.try_recv() {
+                                    Ok(record) => Some((
+                                        AgentStreamItem::Event(record),
+                                        (Phase::Draining(terminal), rx),
+                                    )),
+                                    Err(_) => Some((terminal, (Phase::Done, rx))),
+                                }
+                            }
+                        }
+                    }
+                    Phase::Draining(terminal) => match rx.try_recv() {
+                        Ok(record) => Some((
+                            AgentStreamItem::Event(record),
+                            (Phase::Draining(terminal), rx),
+                        )),
+                        Err(_) => Some((terminal, (Phase::Done, rx))),
+                    },
+                    Phase::Done => None,
+                }
+            },
+        )
+    }
+}