diff --git a/crates/agen/src/engine.rs b/crates/agen/src/engine.rs index 737fac9a1..45a28293d 100644 --- a/crates/agen/src/engine.rs +++ b/crates/agen/src/engine.rs @@ -3401,12 +3401,14 @@ impl Engine { })); } } - None if response_completed => break, + None if !early_dispatch || response_completed => break, None => { - // A clean transport EOF is not an authoritative provider - // response boundary. Treat it exactly like an interrupted - // stream so any early side effect is terminalized and never - // followed by automatic continuation/regeneration. + // Providers that support only response-complete tool calls use + // stream EOF as their established completion boundary. Early + // dispatch is never enabled for them. Once per-block dispatch is + // active, however, transport EOF alone cannot prove that the + // assistant response completed after a side effect may have begun. + // Require the provider's explicit terminal status in that mode. self.timeline.flush_usage(); return Ok(StreamResponseOutput { completion: StreamCompletion::Interrupted { diff --git a/crates/agen/tests/parallel_execution_test.rs b/crates/agen/tests/parallel_execution_test.rs index 3638f48d5..279c65eeb 100644 --- a/crates/agen/tests/parallel_execution_test.rs +++ b/crates/agen/tests/parallel_execution_test.rs @@ -1679,10 +1679,6 @@ async fn unsupported_provider_visibly_defers_opt_in_dispatch() { warning.contains("provider does not expose a trustworthy per-call completion boundary") })); - tx.send(Ok(Event::Status(StatusEvent { - status: ResponseStatus::Completed, - }))) - .unwrap(); drop(tx); tokio::time::timeout(Duration::from_secs(1), probe.started.notified()) .await diff --git a/crates/agen/tests/response_completion_compatibility_test.rs b/crates/agen/tests/response_completion_compatibility_test.rs new file mode 100644 index 000000000..c7b7d8fac --- /dev/null +++ b/crates/agen/tests/response_completion_compatibility_test.rs @@ -0,0 +1,142 @@ +use std::sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, +}; + +use agen::llm_client::{ + ClientError, LlmClient, Request, ResponseStream, ToolCallCompletionSupport, + scheme::{Scheme, gemini::GeminiScheme, openai_chat::OpenAIScheme}, +}; +use agen::{Engine, EngineRunExit, History, ToolCallDispatchMode}; +use async_trait::async_trait; + +#[derive(Clone)] +struct SseFrame { + event_type: &'static str, + data: &'static str, +} + +#[derive(Clone)] +struct SchemeReplayClient { + scheme: S, + responses: Arc>>, + stream_calls: Arc, +} + +impl SchemeReplayClient { + fn new(scheme: S, responses: Vec>) -> Self { + Self { + scheme, + responses: Arc::new(responses), + stream_calls: Arc::new(AtomicUsize::new(0)), + } + } + + fn stream_count(&self) -> usize { + self.stream_calls.load(Ordering::SeqCst) + } +} + +#[async_trait] +impl LlmClient for SchemeReplayClient { + async fn stream(&self, _request: Request) -> Result { + let response_index = self.stream_calls.fetch_add(1, Ordering::SeqCst); + let frames = self + .responses + .get(response_index) + .ok_or_else(|| ClientError::Config("scheme replay responses exhausted".to_string()))?; + let mut state = S::State::default(); + let mut output = Vec::new(); + for frame in frames { + match self + .scheme + .parse_sse(frame.event_type, frame.data, &mut state) + { + Ok(events) => output.extend(events.into_iter().map(Ok)), + Err(error) => { + output.push(Err(error)); + break; + } + } + } + Ok(Box::pin(futures::stream::iter(output))) + } + + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + fn tool_call_completion_support(&self) -> ToolCallCompletionSupport { + self.scheme.tool_call_completion_support() + } +} + +async fn assert_response_complete_scheme_accepts_clean_eof( + scheme: S, + frames: Vec, +) { + assert_eq!( + scheme.tool_call_completion_support(), + ToolCallCompletionSupport::ResponseComplete + ); + + for mode in [ + ToolCallDispatchMode::AfterResponse, + ToolCallDispatchMode::OnToolCallComplete, + ] { + let client = SchemeReplayClient::new(scheme.clone(), vec![frames.clone()]); + let probe = client.clone(); + let mut engine = Engine::new(client); + engine.set_tool_call_dispatch_mode(mode); + let mut history = History::new(); + + let output = engine.run(&mut history, "hello").await; + + assert!( + matches!(output.result, EngineRunExit::Finished), + "{mode:?} should accept the response-complete scheme's clean EOF, got {:?}", + output.result + ); + assert_eq!( + probe.stream_count(), + 1, + "{mode:?} must not open a continuation request after the adapter's normal EOF" + ); + } +} + +#[tokio::test] +async fn openai_chat_clean_eof_completes_in_default_and_opt_in_fallback_modes() { + assert_response_complete_scheme_accepts_clean_eof( + OpenAIScheme::new(), + vec![ + SseFrame { + event_type: "message", + data: r#"{"id":"chatcmpl-123","object":"chat.completion.chunk","created":1694268190,"model":"gpt-4o","choices":[{"index":0,"delta":{"content":"Hello"},"finish_reason":"stop"}]}"#, + }, + SseFrame { + event_type: "message", + data: "[DONE]", + }, + ], + ) + .await; +} + +#[tokio::test] +async fn gemini_clean_eof_completes_in_default_and_opt_in_fallback_modes() { + assert_response_complete_scheme_accepts_clean_eof( + GeminiScheme::new(), + vec![ + SseFrame { + event_type: "message", + data: r#"{"candidates":[{"content":{"parts":[{"text":"Hello"}],"role":"model"},"finishReason":"STOP","index":0}]}"#, + }, + SseFrame { + event_type: "message", + data: "[DONE]", + }, + ], + ) + .await; +}