fix(llm): force structured directive on the streaming path too

Follow-up to the blocking-path fix: providers now override
`complete_streaming_structured` so streaming structured generation also forces
`tool_choice` / sets native `response_format`, instead of falling back to the
non-forcing default.

- anthropic/openai: extract `send_streaming` from `complete_streaming`; the
  trait methods become thin wrappers, and `complete_streaming_structured`
  applies the directive before executing. The large streaming parsers are
  unchanged.
- zhipu: already delegates to its inner client (no change).
- tests: RecordingClient records the streaming directive; new unit test asserts
  `generate_streaming` forces the tool; new `#[ignore]` real-LLM streaming case.
  Validated against gpt-4o: streaming forced tool_choice -> 8 partials, valid
  object (5/5 integration cases pass).

Author: claude

core/src/llm/anthropic.rs

@@ -303,10 +303,34 @@ impl LlmClient for AnthropicClient {
         system: Option<&str>,
         tools: &[ToolDefinition],
         cancel_token: CancellationToken,
+    ) -> Result<mpsc::Receiver<StreamEvent>> {
+        self.send_streaming(self.build_request(messages, system, tools), cancel_token)
+            .await
+    }
+
+    async fn complete_streaming_structured(
+        &self,
+        messages: &[Message],
+        system: Option<&str>,
+        tools: &[ToolDefinition],
+        directive: &structured::StructuredDirective,
+        cancel_token: CancellationToken,
+    ) -> Result<mpsc::Receiver<StreamEvent>> {
+        let mut request_body = self.build_request(messages, system, tools);
+        Self::apply_directive(&mut request_body, directive);
+        self.send_streaming(request_body, cancel_token).await
+    }
+}
+
+impl AnthropicClient {
+    /// Execute a fully-built streaming request body (sets `stream: true`).
+    async fn send_streaming(
+        &self,
+        mut request_body: serde_json::Value,
+        cancel_token: CancellationToken,
     ) -> Result<mpsc::Receiver<StreamEvent>> {
         {
             let request_started_at = Instant::now();
-            let mut request_body = self.build_request(messages, system, tools);
             request_body["stream"] = serde_json::json!(true);
 
             let url = format!("{}/v1/messages", self.base_url);

core/src/llm/openai.rs

@@ -516,37 +516,40 @@ impl LlmClient for OpenAiClient {
         tools: &[ToolDefinition],
         cancel_token: tokio_util::sync::CancellationToken,
     ) -> Result<mpsc::Receiver<StreamEvent>> {
-        {
-            let request_started_at = Instant::now();
-            let mut openai_messages = Vec::new();
-
-            if let Some(sys) = system {
-                openai_messages.push(serde_json::json!({
-                    "role": "system",
-                    "content": sys,
-                }));
-            }
-
-            openai_messages.extend(self.convert_messages(messages));
-
-            let mut request = serde_json::json!({
-                "model": self.model,
-                "messages": openai_messages,
-                "stream": true,
-                "stream_options": { "include_usage": true },
-            });
-
-            if let Some(temp) = self.temperature {
-                request["temperature"] = serde_json::json!(temp);
-            }
-            if let Some(max) = self.max_tokens {
-                request["max_tokens"] = serde_json::json!(max);
-            }
+        self.send_streaming(
+            self.build_chat_request(messages, system, tools, None),
+            cancel_token,
+        )
+        .await
+    }
 
-            if !tools.is_empty() {
-                request["tools"] = serde_json::json!(self.convert_tools(tools));
-            }
+    async fn complete_streaming_structured(
+        &self,
+        messages: &[Message],
+        system: Option<&str>,
+        tools: &[ToolDefinition],
+        directive: &structured::StructuredDirective,
+        cancel_token: tokio_util::sync::CancellationToken,
+    ) -> Result<mpsc::Receiver<StreamEvent>> {
+        self.send_streaming(
+            self.build_chat_request(messages, system, tools, Some(directive)),
+            cancel_token,
+        )
+        .await
+    }
+}
 
+impl OpenAiClient {
+    /// Execute a fully-built streaming chat-completions request (sets `stream`).
+    async fn send_streaming(
+        &self,
+        mut request: serde_json::Value,
+        cancel_token: tokio_util::sync::CancellationToken,
+    ) -> Result<mpsc::Receiver<StreamEvent>> {
+        {
+            request["stream"] = serde_json::json!(true);
+            request["stream_options"] = serde_json::json!({ "include_usage": true });
+            let request_started_at = Instant::now();
             let url = format!("{}{}", self.base_url, self.chat_completions_path);
             let request_headers = self.request_headers();
 

core/src/llm/structured_tests.rs

@@ -1399,6 +1399,52 @@ impl LlmClient for RecordingClient {
         *self.last_tool_names.lock().unwrap() = tools.iter().map(|t| t.name.clone()).collect();
         self.pop()
     }
+
+    async fn complete_streaming_structured(
+        &self,
+        _messages: &[Message],
+        _system: Option<&str>,
+        tools: &[ToolDefinition],
+        directive: &StructuredDirective,
+        _cancel_token: CancellationToken,
+    ) -> anyhow::Result<mpsc::Receiver<StreamEvent>> {
+        self.structured_calls
+            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+        *self.last_directive.lock().unwrap() = Some(directive.clone());
+        *self.last_tool_names.lock().unwrap() = tools.iter().map(|t| t.name.clone()).collect();
+        let response = self.pop()?;
+        let (tx, rx) = mpsc::channel(10);
+        tokio::spawn(async move {
+            for block in &response.message.content {
+                if let ContentBlock::ToolUse { name, input, .. } = block {
+                    tx.send(StreamEvent::ToolUseStart {
+                        id: "call_001".to_string(),
+                        name: name.clone(),
+                    })
+                    .await
+                    .ok();
+                    let json_str = serde_json::to_string(input).unwrap();
+                    for chunk in json_str.as_bytes().chunks(8) {
+                        tx.send(StreamEvent::ToolUseInputDelta(
+                            String::from_utf8_lossy(chunk).to_string(),
+                        ))
+                        .await
+                        .ok();
+                    }
+                } else if let ContentBlock::Text { text } = block {
+                    for chunk in text.as_bytes().chunks(8) {
+                        tx.send(StreamEvent::TextDelta(
+                            String::from_utf8_lossy(chunk).to_string(),
+                        ))
+                        .await
+                        .ok();
+                    }
+                }
+            }
+            tx.send(StreamEvent::Done(response)).await.ok();
+        });
+        Ok(rx)
+    }
 }
 
 fn person_request(mode: StructuredMode) -> StructuredRequest {
@@ -1533,6 +1579,40 @@ async fn test_routing_json_uses_json_object_when_supported() {
     assert!(directive.force_tool.is_none());
 }
 
+#[tokio::test]
+async fn test_streaming_routing_tool_mode_forces_tool_choice() {
+    // The streaming path must also force the directive (via
+    // complete_streaming_structured), not silently drop it.
+    let client = RecordingClient::new(
+        NativeStructuredSupport::ForcedTool,
+        vec![MockStructuredClient::tool_call_response(
+            "emit_person",
+            serde_json::json!({ "name": "Bob" }),
+        )],
+    );
+
+    let partials = Arc::new(Mutex::new(0usize));
+    let partials_cb = partials.clone();
+    let result = generate_streaming(
+        &client,
+        &person_request(StructuredMode::Tool),
+        Box::new(move |_partial| {
+            *partials_cb.lock().unwrap() += 1;
+        }),
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(result.object["name"], "Bob");
+    assert_eq!(result.mode_used, StructuredMode::Tool);
+    let directive = client.last_directive.lock().unwrap().clone().unwrap();
+    assert_eq!(directive.force_tool.as_deref(), Some("emit_person"));
+    assert!(
+        *partials.lock().unwrap() >= 1,
+        "on_partial should fire at least once (final object)"
+    );
+}
+
 // ========================================================================
 // Adversarial JSON extraction edge cases
 // ========================================================================

core/tests/test_structured_json_real_llm.rs

@@ -22,7 +22,8 @@ use std::time::Duration;
 
 use a3s_code_core::config::CodeConfig;
 use a3s_code_core::llm::structured::{
-    generate_blocking, StructuredMode, StructuredRequest, StructuredResult,
+    generate_blocking, generate_streaming, PartialObjectCallback, StructuredMode,
+    StructuredRequest, StructuredResult,
 };
 use a3s_code_core::llm::{create_client_with_config, LlmClient};
 use a3s_code_core::planning::LlmPlanner;
@@ -162,6 +163,39 @@ async fn real_structured_tool_mode_is_stable() {
     );
 }
 
+/// The streaming structured path must also force `tool_choice` and yield a
+/// valid object (the streaming counterpart of the core fix).
+#[tokio::test(flavor = "multi_thread")]
+#[ignore = "requires real provider credentials and network access"]
+async fn real_structured_tool_mode_streaming() {
+    let client = real_client();
+    let partials = std::sync::Arc::new(std::sync::Mutex::new(0usize));
+    let partials_cb = partials.clone();
+    let on_partial: PartialObjectCallback = Box::new(move |_p| {
+        *partials_cb.lock().unwrap() += 1;
+    });
+
+    let result = tokio::time::timeout(
+        CALL_TIMEOUT,
+        generate_streaming(
+            client.as_ref(),
+            &person_request(StructuredMode::Tool),
+            on_partial,
+        ),
+    )
+    .await
+    .expect("streaming call timed out")
+    .expect("streaming tool-mode generation failed");
+
+    assert_eq!(result.mode_used, StructuredMode::Tool);
+    assert_valid_person(&result.object);
+    eprintln!(
+        "[tool-stream] partials={} -> {}",
+        *partials.lock().unwrap(),
+        result.object
+    );
+}
+
 /// Native `response_format: json_object` on the OpenAI-compatible provider.
 #[tokio::test(flavor = "multi_thread")]
 #[ignore = "requires real provider credentials and network access"]