feat(harness): invoke_stream + sub-agent delta propagation + scripted mock streaming#17
Conversation
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)
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)
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)
|
Warning Review limit reachedYou’ve reached a temporary PR review limit under our Fair Usage Limits Policy. Next review available in: 39 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (13)
Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 45bd370f07
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Done, | ||
| } | ||
|
|
||
| impl<State: Send + Sync, Ctx: Send + Sync + 'static> AgentHarness<State, Ctx> { |
There was a problem hiding this comment.
When an application uses a borrowed per-run context type (for example AgentHarness<State, &RequestCtx>), this new API is unavailable because the impl requires Ctx: 'static, even though the stream owns ctx_data, is lifetime-bound to &self/state, and is not spawned. The existing invoke paths accept non-'static contexts, so this makes the caller-consumable streaming path unusable for otherwise valid request-scoped contexts; relaxing this to a method-level Ctx: 'a bound (or equivalent) would preserve the intended lifetime without forcing 'static.
Useful? React with 👍 / 👎.
1 participant
Summary
Adds a caller-consumable streaming API to the agent harness and the sub-agent
propagation + mock fidelity that make it useful.
invoke_stream(Step 1)AgentHarness::invoke_stream(...)returns an asyncStream<Item = AgentStreamItem>— live
Event(EventRecord)items in emission order (model/reasoning deltas,ToolStarted/ToolCompletedlifecycle, sub-agent lifecycle, usage, …), thenexactly one terminal
Completed(run)/Failed(err). It's a thin projectionover the existing ordered
EventSink::emitfan-out — aChannelListenerforwards each record into an unbounded channel the stream drains — not a
parallel event system. The run advances only while the stream is polled.
Sub-agent model-delta propagation (Step 2)
A streaming parent now sees a child sub-agent's own model/reasoning deltas, not
just its start/complete bracket. The drive mode is threaded down:
RunContextgains a
streamingflag (set by the loop driver),ToolExecutionContextcarries it, and
SubAgentToolruns the child through the streaming path whenthe parent is streaming. Because sub-agents already share the parent's
EventSink, child deltas land on the parent stream stamped with the child'sown
run_idanddepth. A non-streaming parent is unchanged (flag staysfalse).MockModel::streaming_script(Step 4)A
MockBehavior::StreamScript(Vec<ModelStreamItem>)variant + constructor:stream()emits the scripted items verbatim so tests exercise trulyincremental streaming (fine-grained text/reasoning/tool-call deltas), while
invoke()folds them through aStreamAccumulatorinto the equivalent unaryresponse.
Tool-call lifecycle events (Step 3 in the plan) already exist upstream and
surface through the stream unchanged.
Tests
invoke_streamevent-then-terminal ordering, tool-lifecycle surfacing, Failedterminal; nested sub-agent asserts child deltas (child run id) arrive before
the parent completes, and that a non-streaming parent emits none; scripted
incremental deltas surface reasoning + multiple text deltas; invoke-fold.
Commands run
cargo fmt --checkcargo clippy --all-targets -- -D warningscargo test(939 lib + integration, green)Ref: OpenHuman
docs/tinyagents-vendor-improvement-plan/03-streaming.md(V3, Steps 1/2/4).