Skip to main content

lash_core/
direct.rs

1use crate::llm::transport::LlmTransportError;
2use crate::llm::types::{
3    LlmAttachment, LlmContentBlock, LlmEventSender, LlmJsonSchema, LlmMessage, LlmOutputSpec,
4    LlmRequest, LlmRequestScope, LlmResponse, LlmRole, LlmStreamEvent, LlmTerminalReason,
5    LlmToolChoice,
6};
7use crate::provider::{ModelCapability, ModelEffortValidationCategory, ProviderHandle};
8use crate::{LashSchema, SchemaContract};
9use lash_trace::{TraceContext, TraceError, TraceEvent, TraceSink};
10use std::sync::Arc;
11
12#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
13#[serde(rename_all = "snake_case")]
14pub enum DirectRole {
15    System,
16    User,
17    Assistant,
18}
19
20#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
21pub enum DirectPart {
22    Text(String),
23    Image(usize),
24}
25
26#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
27pub struct DirectMessage {
28    pub role: DirectRole,
29    pub parts: Vec<DirectPart>,
30}
31
32#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
33pub struct DirectJsonSchema {
34    pub name: String,
35    pub schema: SchemaContract,
36    pub strict: bool,
37}
38
39#[derive(Clone, Debug, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
40pub enum DirectOutputSpec {
41    #[default]
42    Text,
43    JsonObject,
44    JsonSchema(DirectJsonSchema),
45}
46
47#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)]
48pub struct DirectRequest {
49    pub model: String,
50    #[serde(default, skip_serializing_if = "Option::is_none")]
51    pub model_variant: Option<String>,
52    #[serde(default, skip_serializing_if = "ModelCapability::is_empty")]
53    pub model_capability: ModelCapability,
54    #[serde(default)]
55    pub messages: Vec<DirectMessage>,
56    #[serde(default)]
57    pub attachments: Vec<LlmAttachment>,
58    #[serde(default)]
59    pub output: DirectOutputSpec,
60    #[serde(default)]
61    pub generation: crate::GenerationOptions,
62    #[serde(default, skip)]
63    pub stream_events: Option<LlmEventSender>,
64    #[serde(default, skip_serializing_if = "Option::is_none")]
65    pub session_id: Option<String>,
66    #[serde(default, skip_serializing_if = "Option::is_none")]
67    pub caused_by: Option<crate::CausalRef>,
68    #[serde(default, skip_serializing_if = "Option::is_none")]
69    pub replay: Option<crate::RuntimeReplay>,
70}
71
72impl DirectRequest {
73    pub fn text(model: impl Into<String>, prompt: impl Into<String>) -> Self {
74        Self {
75            model: model.into(),
76            model_variant: None,
77            model_capability: ModelCapability::default(),
78            messages: vec![DirectMessage {
79                role: DirectRole::User,
80                parts: vec![DirectPart::Text(prompt.into())],
81            }],
82            attachments: Vec::new(),
83            output: DirectOutputSpec::Text,
84            generation: crate::GenerationOptions::default(),
85            stream_events: None,
86            session_id: None,
87            caused_by: None,
88            replay: None,
89        }
90    }
91
92    pub fn json(model: impl Into<String>, prompt: impl Into<String>) -> Self {
93        Self {
94            output: DirectOutputSpec::JsonObject,
95            ..Self::text(model, prompt)
96        }
97    }
98
99    pub fn json_schema(
100        model: impl Into<String>,
101        prompt: impl Into<String>,
102        schema: DirectJsonSchema,
103    ) -> Self {
104        Self {
105            output: DirectOutputSpec::JsonSchema(schema),
106            ..Self::text(model, prompt)
107        }
108    }
109
110    pub fn with_replay_key(mut self, key: impl Into<String>) -> Self {
111        self.replay = Some(crate::RuntimeReplay { key: key.into() });
112        self
113    }
114
115    pub fn with_caused_by(mut self, caused_by: crate::CausalRef) -> Self {
116        self.caused_by = Some(caused_by);
117        self
118    }
119}
120
121#[derive(Debug, thiserror::Error, Clone)]
122pub enum DirectLlmError {
123    #[error("invalid request: {message}")]
124    InvalidRequest {
125        category: ModelEffortValidationCategory,
126        message: String,
127    },
128    #[error("invalid response: {0}")]
129    InvalidResponse(String),
130    #[error("transport error: {0}")]
131    Transport(#[from] Box<LlmTransportError>),
132}
133
134pub struct DirectLlmClient {
135    provider: ProviderHandle,
136    trace_sink: Option<Arc<dyn TraceSink>>,
137    trace_context: TraceContext,
138    clock: Arc<dyn crate::Clock>,
139}
140
141impl DirectLlmClient {
142    pub fn new(provider: ProviderHandle) -> Self {
143        Self {
144            provider,
145            trace_sink: None,
146            trace_context: TraceContext::default(),
147            clock: Arc::new(crate::SystemClock),
148        }
149    }
150
151    pub fn with_trace_sink(mut self, sink: Option<Arc<dyn TraceSink>>) -> Self {
152        self.trace_sink = sink;
153        self
154    }
155
156    pub fn with_trace_context(mut self, context: TraceContext) -> Self {
157        self.trace_context = context;
158        self
159    }
160
161    pub fn with_clock(mut self, clock: Arc<dyn crate::Clock>) -> Self {
162        self.clock = clock;
163        self
164    }
165
166    pub fn provider(&self) -> &ProviderHandle {
167        &self.provider
168    }
169
170    pub fn provider_mut(&mut self) -> &mut ProviderHandle {
171        &mut self.provider
172    }
173
174    pub async fn complete(
175        &mut self,
176        mut request: DirectRequest,
177    ) -> Result<LlmResponse, DirectLlmError> {
178        // Validate the requested effort against the capability that travels
179        // with the request, and write the resolved (alias-normalized) effort
180        // back so the provider never sees an un-clamped value.
181        request.model_variant = request
182            .model_capability
183            .validate_effort(
184                &request.model,
185                self.provider.kind(),
186                request.model_variant.as_deref(),
187            )
188            .map_err(|error| DirectLlmError::InvalidRequest {
189                category: error.category,
190                message: error.message,
191            })?;
192
193        let output_for_validation = request.output.clone();
194        let model = request.model.clone();
195        let llm_request = build_llm_request(&self.provider, request, model);
196        let llm_call_id = if self.trace_sink.is_some() {
197            let id = uuid::Uuid::new_v4().to_string();
198            crate::trace::emit_trace(
199                &self.trace_sink,
200                &self.trace_context,
201                TraceContext::default().for_llm_call(id.clone()),
202                TraceEvent::LlmCallStarted {
203                    request: crate::trace::trace_llm_request(&llm_request),
204                },
205                self.clock.as_ref(),
206            );
207            Some(id)
208        } else {
209            None
210        };
211        match self.provider.complete(llm_request).await {
212            Ok(response) => {
213                if let Err(error) = validate_direct_output(&output_for_validation, &response) {
214                    if let Some(llm_call_id) = llm_call_id {
215                        crate::trace::emit_trace(
216                            &self.trace_sink,
217                            &self.trace_context,
218                            TraceContext::default().for_llm_call(llm_call_id),
219                            TraceEvent::LlmCallFailed {
220                                error: TraceError {
221                                    message: error.to_string(),
222                                    retryable: false,
223                                    terminal_reason: Some(
224                                        LlmTerminalReason::ProviderError.code().to_string(),
225                                    ),
226                                    code: Some("invalid_structured_output".to_string()),
227                                    raw: None,
228                                },
229                                stream_summary: None,
230                            },
231                            self.clock.as_ref(),
232                        );
233                    }
234                    return Err(error);
235                }
236                if let Some(llm_call_id) = llm_call_id {
237                    crate::trace::emit_trace(
238                        &self.trace_sink,
239                        &self.trace_context,
240                        TraceContext::default().for_llm_call(llm_call_id),
241                        TraceEvent::LlmCallCompleted {
242                            response: crate::trace::trace_llm_response(
243                                response.full_text.clone(),
244                                0,
245                                Some(response.terminal_reason),
246                                crate::trace::trace_output_parts(&response.parts),
247                            ),
248                            usage: Some(crate::trace::trace_usage_from_llm(&response.usage)),
249                            provider_usage: response.provider_usage.clone(),
250                            stream_summary: None,
251                        },
252                        self.clock.as_ref(),
253                    );
254                }
255                Ok(response)
256            }
257            Err(error) => {
258                if let Some(llm_call_id) = llm_call_id {
259                    crate::trace::emit_trace(
260                        &self.trace_sink,
261                        &self.trace_context,
262                        TraceContext::default().for_llm_call(llm_call_id),
263                        TraceEvent::LlmCallFailed {
264                            error: TraceError {
265                                message: error.message.clone(),
266                                retryable: error.retryable,
267                                terminal_reason: Some(error.terminal_reason.code().to_string()),
268                                code: error.code.clone(),
269                                raw: error.raw.clone(),
270                            },
271                            stream_summary: None,
272                        },
273                        self.clock.as_ref(),
274                    );
275                }
276                Err(DirectLlmError::from(Box::new(error)))
277            }
278        }
279    }
280}
281
282pub(crate) fn build_llm_request(
283    provider: &ProviderHandle,
284    request: DirectRequest,
285    model: String,
286) -> LlmRequest {
287    let stream_events = transport_stream_events_for_direct(provider, request.stream_events);
288    let DirectRequest {
289        model: _,
290        model_variant,
291        model_capability,
292        messages,
293        attachments,
294        output,
295        generation,
296        stream_events: _,
297        session_id,
298        caused_by: _,
299        replay: _,
300    } = request;
301
302    let output_spec = match output {
303        DirectOutputSpec::Text => None,
304        DirectOutputSpec::JsonObject => Some(LlmOutputSpec::JsonObject),
305        DirectOutputSpec::JsonSchema(schema) => Some(LlmOutputSpec::JsonSchema(LlmJsonSchema {
306            name: schema.name,
307            schema: schema.schema,
308            strict: schema.strict,
309        })),
310    };
311
312    let mut llm_messages = Vec::new();
313    for message in messages {
314        let role = match message.role {
315            DirectRole::System => LlmRole::System,
316            DirectRole::User => LlmRole::User,
317            DirectRole::Assistant => LlmRole::Assistant,
318        };
319        let mut blocks: Vec<LlmContentBlock> = Vec::new();
320        for part in message.parts {
321            match part {
322                DirectPart::Text(text) => {
323                    if !text.is_empty() {
324                        blocks.push(LlmContentBlock::Text {
325                            text: text.into(),
326                            response_meta: None,
327                            cache_breakpoint: false,
328                        });
329                    }
330                }
331                DirectPart::Image(idx) => {
332                    blocks.push(LlmContentBlock::Image {
333                        attachment_idx: idx,
334                    });
335                }
336            }
337        }
338        if !blocks.is_empty() {
339            llm_messages.push(LlmMessage::new(role, blocks));
340        }
341    }
342
343    let scope = match session_id {
344        Some(session_id) => LlmRequestScope::new(
345            session_id.clone(),
346            format!("{session_id}:frame:direct"),
347            format!("{session_id}:direct"),
348        ),
349        None => {
350            let request_id = uuid::Uuid::new_v4().to_string();
351            LlmRequestScope::new(
352                format!("direct:{request_id}"),
353                format!("direct:{request_id}:frame"),
354                request_id,
355            )
356        }
357    };
358
359    LlmRequest {
360        model,
361        messages: llm_messages,
362        attachments,
363        tools: Vec::new().into(),
364        tool_choice: LlmToolChoice::None,
365        model_variant,
366        model_capability,
367        generation,
368        scope,
369        output_spec,
370        stream_events,
371        provider_trace: None,
372    }
373}
374
375fn validate_direct_output(
376    output: &DirectOutputSpec,
377    response: &LlmResponse,
378) -> Result<(), DirectLlmError> {
379    let DirectOutputSpec::JsonSchema(schema) = output else {
380        return Ok(());
381    };
382    let parsed: serde_json::Value = serde_json::from_str(response.full_text.trim())
383        .map_err(|err| DirectLlmError::InvalidResponse(format!("expected JSON: {err}")))?;
384    LashSchema::new(schema.schema.canonical().clone())
385        .validate(&parsed)
386        .map_err(DirectLlmError::InvalidResponse)
387}
388
389fn transport_stream_events_for_direct(
390    provider: &ProviderHandle,
391    requested: Option<LlmEventSender>,
392) -> Option<LlmEventSender> {
393    if requested.is_some() {
394        return requested;
395    }
396    if provider.requires_streaming() {
397        Some(LlmEventSender::new(|_event: LlmStreamEvent| {}))
398    } else {
399        None
400    }
401}
402
403#[cfg(test)]
404mod tests {
405    use super::*;
406    use crate::llm::types::{LlmOutputPart, LlmTerminalReason, LlmUsage};
407    use crate::provider::{ProviderOptions, ProviderReliability};
408    use crate::testing::TestProvider;
409    use serde_json::json;
410    use std::sync::{Arc, Mutex};
411
412    #[test]
413    fn json_schema_request_preserves_output_schema() {
414        let schema = DirectJsonSchema {
415            name: "answer_shape".to_string(),
416            schema: json!({
417                "type": "object",
418                "properties": {
419                    "answer": { "type": "string" }
420                },
421                "required": ["answer"]
422            })
423            .into(),
424            strict: true,
425        };
426
427        let request = DirectRequest::json_schema("model-a", "return json", schema.clone());
428
429        assert_eq!(
430            request.output,
431            DirectOutputSpec::JsonSchema(schema),
432            "DirectRequest::json_schema must carry the requested output schema"
433        );
434    }
435
436    #[test]
437    fn direct_client_provider_accessors_expose_owned_provider_handle() {
438        let provider = TestProvider::builder()
439            .kind("direct-accessor-provider")
440            .serialize_config(|| json!({"provider": "owned"}))
441            .build()
442            .into_handle();
443        let mut client = DirectLlmClient::new(provider);
444
445        assert_eq!(client.provider().kind(), "direct-accessor-provider");
446        assert_eq!(
447            client.provider().to_spec().config,
448            json!({"provider": "owned"})
449        );
450
451        let options = ProviderOptions {
452            reliability: ProviderReliability::default().max_attempts(7),
453            max_output_tokens: Some(123),
454            ..Default::default()
455        };
456        client.provider_mut().set_options(options.clone());
457
458        assert_eq!(client.provider().options(), options);
459    }
460
461    #[tokio::test]
462    async fn direct_client_complete_delegates_to_provider_and_returns_response() {
463        let captured_request: Arc<Mutex<Option<LlmRequest>>> = Arc::new(Mutex::new(None));
464        let captured_for_provider = Arc::clone(&captured_request);
465        let provider = TestProvider::builder()
466            .kind("direct-complete-provider")
467            .complete(move |request| {
468                let captured_for_provider = Arc::clone(&captured_for_provider);
469                async move {
470                    *captured_for_provider.lock().expect("capture lock") = Some(request);
471                    Ok(LlmResponse {
472                        full_text: "provider delegated response".to_string(),
473                        parts: vec![LlmOutputPart::Text {
474                            text: "provider delegated response".to_string(),
475                            response_meta: None,
476                        }],
477                        usage: LlmUsage {
478                            input_tokens: 11,
479                            output_tokens: 3,
480                            ..Default::default()
481                        },
482                        terminal_reason: LlmTerminalReason::Stop,
483                        ..Default::default()
484                    })
485                }
486            })
487            .build()
488            .into_handle();
489        let mut client = DirectLlmClient::new(provider);
490        let mut request = DirectRequest::json("direct-model", "answer as json");
491        request.session_id = Some("direct-session".to_string());
492
493        let response = client
494            .complete(request)
495            .await
496            .expect("direct completion should delegate");
497
498        assert_eq!(response.full_text, "provider delegated response");
499        let captured = captured_request
500            .lock()
501            .expect("capture lock")
502            .clone()
503            .expect("provider should receive a request");
504        assert_eq!(captured.model, "direct-model");
505        assert_eq!(captured.scope.session_id, "direct-session");
506        assert_eq!(captured.scope.agent_frame_id, "direct-session:frame:direct");
507        assert_eq!(captured.scope.request_id, "direct-session:direct");
508        assert!(matches!(
509            captured.output_spec,
510            Some(LlmOutputSpec::JsonObject)
511        ));
512        assert_eq!(captured.messages.len(), 1);
513    }
514
515    #[tokio::test]
516    async fn direct_client_validates_json_schema_output_against_canonical_schema() {
517        let provider = TestProvider::builder()
518            .kind("direct-validation-provider")
519            .complete(|_request| async {
520                Ok(LlmResponse {
521                    full_text: r#"{"items":[]}"#.to_string(),
522                    terminal_reason: LlmTerminalReason::Stop,
523                    ..Default::default()
524                })
525            })
526            .build()
527            .into_handle();
528        let mut client = DirectLlmClient::new(provider);
529        let request = DirectRequest::json_schema(
530            "direct-model",
531            "return items",
532            DirectJsonSchema {
533                name: "items_result".to_string(),
534                schema: json!({
535                    "type": "object",
536                    "required": ["items"],
537                    "properties": {
538                        "items": {
539                            "type": "array",
540                            "minItems": 1,
541                            "items": { "type": "string" }
542                        }
543                    }
544                })
545                .into(),
546                strict: true,
547            },
548        );
549
550        let err = client
551            .complete(request)
552            .await
553            .expect_err("empty items must fail canonical validation");
554
555        assert!(matches!(err, DirectLlmError::InvalidResponse(_)));
556        assert!(err.to_string().contains("items >= 1"));
557    }
558
559    fn reasoning_capability() -> ModelCapability {
560        ModelCapability {
561            reasoning: Some(crate::ReasoningCapability {
562                efforts: ["low", "medium", "high", "max"]
563                    .into_iter()
564                    .map(String::from)
565                    .collect(),
566                aliases: std::collections::BTreeMap::from([(
567                    "xhigh".to_string(),
568                    "max".to_string(),
569                )]),
570                ..Default::default()
571            }),
572        }
573    }
574
575    #[tokio::test]
576    async fn direct_client_rejects_unsupported_effort_before_provider_call() {
577        let called = Arc::new(Mutex::new(false));
578        let called_for_provider = Arc::clone(&called);
579        let provider = TestProvider::builder()
580            .kind("direct-reject")
581            .complete(move |_request| {
582                let called = Arc::clone(&called_for_provider);
583                async move {
584                    *called.lock().expect("called lock") = true;
585                    Ok(LlmResponse::default())
586                }
587            })
588            .build()
589            .into_handle();
590        let mut client = DirectLlmClient::new(provider);
591
592        let mut request = DirectRequest::text("direct-model", "hi");
593        request.model_variant = Some("turbo".to_string());
594        request.model_capability = reasoning_capability();
595
596        let err = client
597            .complete(request)
598            .await
599            .expect_err("unsupported effort must be rejected");
600        assert!(matches!(
601            err,
602            DirectLlmError::InvalidRequest {
603                category: ModelEffortValidationCategory::UnsupportedEffort,
604                ..
605            }
606        ));
607        assert!(err.to_string().contains("Unsupported effort `turbo`"));
608        assert!(
609            !*called.lock().expect("called lock"),
610            "the provider must not be called when the effort is rejected"
611        );
612    }
613
614    #[tokio::test]
615    async fn direct_client_normalizes_alias_effort_into_outgoing_request() {
616        let captured: Arc<Mutex<Option<Option<String>>>> = Arc::new(Mutex::new(None));
617        let captured_for_provider = Arc::clone(&captured);
618        let provider = TestProvider::builder()
619            .kind("direct-alias")
620            .complete(move |request| {
621                let captured = Arc::clone(&captured_for_provider);
622                async move {
623                    *captured.lock().expect("capture lock") = Some(request.model_variant.clone());
624                    Ok(LlmResponse {
625                        full_text: "ok".to_string(),
626                        terminal_reason: LlmTerminalReason::Stop,
627                        ..Default::default()
628                    })
629                }
630            })
631            .build()
632            .into_handle();
633        let mut client = DirectLlmClient::new(provider);
634
635        let mut request = DirectRequest::text("direct-model", "hi");
636        request.model_variant = Some("XHigh".to_string());
637        request.model_capability = reasoning_capability();
638
639        client.complete(request).await.expect("completion");
640        let seen = captured
641            .lock()
642            .expect("capture lock")
643            .clone()
644            .expect("provider must be called");
645        assert_eq!(
646            seen.as_deref(),
647            Some("max"),
648            "alias `XHigh` must clamp to canonical `max` before the provider sees the request"
649        );
650    }
651
652    #[tokio::test]
653    async fn direct_client_rejects_effort_when_model_is_not_configurable() {
654        let provider = TestProvider::builder()
655            .kind("direct-not-configurable")
656            .complete(|_request| async { Ok(LlmResponse::default()) })
657            .build()
658            .into_handle();
659        let mut client = DirectLlmClient::new(provider);
660
661        let mut request = DirectRequest::text("direct-model", "hi");
662        request.model_variant = Some("high".to_string());
663        // No capability: the model exposes no configurable effort.
664
665        let err = client
666            .complete(request)
667            .await
668            .expect_err("effort on a non-configurable model must be rejected");
669        assert!(matches!(
670            err,
671            DirectLlmError::InvalidRequest {
672                category: ModelEffortValidationCategory::EffortNotConfigurable,
673                ..
674            }
675        ));
676    }
677
678    #[tokio::test]
679    async fn direct_client_rejects_missing_mandatory_effort() {
680        let provider = TestProvider::builder()
681            .kind("direct-mandatory")
682            .complete(|_request| async { Ok(LlmResponse::default()) })
683            .build()
684            .into_handle();
685        let mut client = DirectLlmClient::new(provider);
686
687        let mut capability = reasoning_capability();
688        capability.reasoning.as_mut().expect("reasoning").mandatory = true;
689        let mut request = DirectRequest::text("direct-model", "hi");
690        request.model_capability = capability;
691        // No model_variant supplied, but the model requires one.
692
693        let err = client
694            .complete(request)
695            .await
696            .expect_err("missing mandatory effort must be rejected");
697        assert!(matches!(
698            err,
699            DirectLlmError::InvalidRequest {
700                category: ModelEffortValidationCategory::EffortRequired,
701                ..
702            }
703        ));
704    }
705
706    #[test]
707    fn build_llm_request_preserves_nonempty_content_and_drops_empty_messages() {
708        let provider = TestProvider::default().into_handle();
709        let request = DirectRequest {
710            model: "input-model".to_string(),
711            messages: vec![
712                DirectMessage {
713                    role: DirectRole::System,
714                    parts: vec![DirectPart::Text(String::new())],
715                },
716                DirectMessage {
717                    role: DirectRole::User,
718                    parts: vec![
719                        DirectPart::Text("hello".to_string()),
720                        DirectPart::Text(String::new()),
721                    ],
722                },
723                DirectMessage {
724                    role: DirectRole::Assistant,
725                    parts: vec![DirectPart::Image(2)],
726                },
727            ],
728            attachments: Vec::new(),
729            output: DirectOutputSpec::Text,
730            generation: crate::GenerationOptions::default(),
731            stream_events: None,
732            session_id: None,
733            model_variant: None,
734            model_capability: ModelCapability::default(),
735            caused_by: None,
736            replay: None,
737        };
738
739        let llm_request = build_llm_request(&provider, request, "transport-model".to_string());
740
741        assert_eq!(llm_request.model, "transport-model");
742        assert_eq!(
743            llm_request.messages.len(),
744            2,
745            "empty normalized messages must be dropped"
746        );
747        assert_eq!(llm_request.messages[0].role, LlmRole::User);
748        assert_eq!(llm_request.messages[0].blocks.len(), 1);
749        assert!(matches!(
750            &llm_request.messages[0].blocks[0],
751            LlmContentBlock::Text { text, .. } if text.as_ref() == "hello"
752        ));
753        assert_eq!(llm_request.messages[1].role, LlmRole::Assistant);
754        assert!(matches!(
755            &llm_request.messages[1].blocks[0],
756            LlmContentBlock::Image { attachment_idx: 2 }
757        ));
758    }
759
760    #[test]
761    fn build_llm_request_preserves_direct_stream_sender_and_adds_required_noop_sender() {
762        let captured_events: Arc<Mutex<Vec<LlmStreamEvent>>> = Arc::new(Mutex::new(Vec::new()));
763        let captured_for_sender = Arc::clone(&captured_events);
764        let requested_sender = LlmEventSender::new(move |event| {
765            captured_for_sender
766                .lock()
767                .expect("stream event lock")
768                .push(event);
769        });
770        let mut request = DirectRequest::text("model", "prompt");
771        request.stream_events = Some(requested_sender);
772        let provider = TestProvider::default().into_handle();
773
774        let llm_request = build_llm_request(&provider, request, "model".to_string());
775        let sender = llm_request
776            .stream_events
777            .expect("explicit direct stream sender must be preserved");
778        sender.send(LlmStreamEvent::Delta("delta".to_string()));
779        assert_eq!(captured_events.lock().expect("stream event lock").len(), 1);
780
781        let streaming_provider = TestProvider::builder()
782            .requires_streaming(true)
783            .build()
784            .into_handle();
785        let llm_request = build_llm_request(
786            &streaming_provider,
787            DirectRequest::text("model", "prompt"),
788            "model".to_string(),
789        );
790        assert!(
791            llm_request.stream_events.is_some(),
792            "providers that require streaming need a no-op sender even when direct caller did not request one"
793        );
794    }
795}