diff --git a/src/harness/agent_loop/stream.rs b/src/harness/agent_loop/stream.rs index c52e88e..29ab94b 100644 --- a/src/harness/agent_loop/stream.rs +++ b/src/harness/agent_loop/stream.rs @@ -97,7 +97,7 @@ fn terminal_item(result: Result) -> AgentStreamItem { enum Phase<'a> { /// The run future is still executing. Running { - run_fut: Pin> + 'a>>, + run_fut: Pin> + Send + 'a>>, listener_guard: ChannelListenerGuard, }, /// The run has finished; drain any buffered events, then emit `terminal`. @@ -130,7 +130,7 @@ impl AgentHarness { ctx_data: Ctx, config: RunConfig, input: Vec, - ) -> impl futures::Stream + 'a { + ) -> impl futures::Stream + Send + 'a { let ctx = RunContext::new(config, ctx_data); self.invoke_stream_in_context(state, ctx, input) } @@ -147,7 +147,7 @@ impl AgentHarness { state: &'a State, ctx: RunContext, input: Vec, - ) -> impl futures::Stream + 'a { + ) -> impl futures::Stream + Send + 'a { 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 @@ -163,7 +163,7 @@ impl AgentHarness { // 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> + 'a>> = + let run_fut: Pin> + Send + 'a>> = Box::pin(self.invoke_streaming_in_context_with_status(state, ctx, input)); futures::stream::unfold( diff --git a/src/harness/agent_loop/test.rs b/src/harness/agent_loop/test.rs index 1035d81..9627f7a 100644 --- a/src/harness/agent_loop/test.rs +++ b/src/harness/agent_loop/test.rs @@ -2083,6 +2083,17 @@ async fn invoke_stream_in_context_unsubscribes_channel_listener() { assert_eq!(events.listener_count(), 0); } +#[test] +fn invoke_stream_in_context_stream_is_send() { + fn assert_send(_value: T) {} + + let mut harness: AgentHarness<()> = AgentHarness::new(); + harness.register_model("mock", Arc::new(MockModel::constant("hello there"))); + let ctx = RunContext::new(RunConfig::new("send-stream-run"), ()); + + assert_send(harness.invoke_stream_in_context(&(), ctx, vec![Message::user("hi")])); +} + #[tokio::test] async fn invoke_stream_surfaces_tool_lifecycle() { let mut harness: AgentHarness<()> = AgentHarness::new();