chimera_opencode/
session.rs

1use std::future::Future;
2use std::pin::Pin;
3use std::sync::{Arc, OnceLock};
4use std::time::Duration;
5
6use chimera_core::*;
7
8use chimera_core::ToolChoice;
9
10use crate::config::{OpenCodeConfig, OpenCodeProvider};
11use crate::translate;
12use crate::wire::*;
13
14const DEFAULT_TIMEOUT_SECS: u64 = 300;
15const DEFAULT_MAX_RETRIES: u32 = 2;
16const DEFAULT_MODEL: &str = "glm-5";
17const BASE_RETRY_DELAY: Duration = Duration::from_secs(5);
18const MAX_RETRY_DELAY: Duration = Duration::from_secs(120);
19
20pub struct HttpSession {
21    api_key: Option<String>,
22    provider: OpenCodeProvider,
23    session_id: Arc<OnceLock<String>>,
24    active_interrupt: Arc<tokio::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>>,
25    config: SessionConfig<OpenCodeConfig>,
26    history: Arc<tokio::sync::Mutex<Vec<WireMessage>>>,
27    client: Option<reqwest::Client>,
28}
29
30impl HttpSession {
31    pub(crate) fn new(
32        api_key: Option<String>,
33        provider: OpenCodeProvider,
34        config: SessionConfig<OpenCodeConfig>,
35    ) -> Self {
36        Self {
37            api_key,
38            provider,
39            session_id: Arc::new(OnceLock::new()),
40            active_interrupt: Arc::new(tokio::sync::Mutex::new(None)),
41            config,
42            history: Arc::new(tokio::sync::Mutex::new(Vec::new())),
43            client: None,
44        }
45    }
46
47    pub(crate) fn with_session_id(self, id: String) -> Self {
48        let _ = self.session_id.set(id);
49        self
50    }
51
52    #[cfg(test)]
53    pub(crate) fn api_key(&self) -> Option<&str> {
54        self.api_key.as_deref()
55    }
56
57    fn model(&self) -> &str {
58        let m = self.config.model.as_deref().unwrap_or(DEFAULT_MODEL);
59        // Strip "provider-id/" prefix (e.g. "opencode-go/glm-5" -> "glm-5")
60        m.split_once('/').map(|(_, model)| model).unwrap_or(m)
61    }
62
63    fn timeout(&self) -> Duration {
64        Duration::from_secs(
65            self.config
66                .backend
67                .timeout_secs
68                .unwrap_or(DEFAULT_TIMEOUT_SECS),
69        )
70    }
71
72    fn max_retries(&self) -> u32 {
73        self.config
74            .backend
75            .max_retries
76            .unwrap_or(DEFAULT_MAX_RETRIES)
77    }
78
79    fn base_url(&self) -> &str {
80        self.provider.base_url()
81    }
82
83    fn get_or_build_client(&mut self) -> Result<reqwest::Client> {
84        if let Some(ref client) = self.client {
85            return Ok(client.clone());
86        }
87        let client = reqwest::Client::builder()
88            .timeout(self.timeout())
89            .build()
90            .map_err(|e| AgentError::Other {
91                message: "failed to build HTTP client".into(),
92                source: Some(Box::new(e)),
93            })?;
94        self.client = Some(client.clone());
95        Ok(client)
96    }
97
98    fn build_request(
99        &self,
100        input: &Input,
101        options: &TurnOptions,
102        history: &[WireMessage],
103    ) -> Result<ChatCompletionRequest> {
104        let mut messages = Vec::new();
105
106        if let Some(sp) = &self.config.system_prompt {
107            messages.push(WireMessage::system(sp.clone()));
108        }
109
110        messages.extend_from_slice(history);
111
112        let prompt_text = prompt_text_from_input(input)?;
113        messages.push(WireMessage::user(prompt_text));
114
115        let response_format = options.output_schema.as_ref().map(|schema| ResponseFormat {
116            format_type: "json_schema".into(),
117            json_schema: Some(serde_json::json!({
118                "name": "response",
119                "strict": true,
120                "schema": schema,
121            })),
122        });
123
124        let stream = if self.config.backend.stream {
125            Some(true)
126        } else {
127            None
128        };
129
130        let tools: Vec<WireTool> = options
131            .tools
132            .iter()
133            .map(|t| WireTool {
134                kind: "function".into(),
135                function: WireToolFunction {
136                    name: t.name.clone(),
137                    description: t.description.clone(),
138                    parameters: t.parameters.clone(),
139                },
140            })
141            .collect();
142
143        let tool_choice = options.tool_choice.as_ref().map(|tc| match tc {
144            ToolChoice::Auto => serde_json::json!("auto"),
145            ToolChoice::Required => serde_json::json!("required"),
146            ToolChoice::None => serde_json::json!("none"),
147            ToolChoice::Function(name) => {
148                serde_json::json!({"type": "function", "function": {"name": name}})
149            }
150        });
151
152        Ok(ChatCompletionRequest {
153            model: self.model().to_string(),
154            messages,
155            max_tokens: self.config.backend.max_tokens,
156            temperature: self.config.backend.temperature,
157            stream,
158            response_format,
159            tools,
160            tool_choice,
161        })
162    }
163
164    /// Make a single-turn completion where the model is forced to call `tool`.
165    ///
166    /// Returns the raw JSON arguments string from the first tool call.
167    pub async fn complete_with_tool(
168        &mut self,
169        system_prompt: &str,
170        user_message: &str,
171        tool: chimera_core::ToolDefinition,
172    ) -> chimera_core::Result<String> {
173        use chimera_core::{Input, Session, ToolChoice, TurnOptions};
174        // Override system prompt for this call.
175        let original_system = self.config.system_prompt.clone();
176        if !system_prompt.is_empty() {
177            self.config.system_prompt = Some(system_prompt.to_string());
178        }
179        // Force non-streaming for structured extraction.
180        let original_stream = self.config.backend.stream;
181        self.config.backend.stream = false;
182
183        let options = TurnOptions {
184            tools: vec![tool],
185            tool_choice: Some(ToolChoice::Required),
186            ..TurnOptions::default()
187        };
188        let output = self
189            .turn(Input::Text(user_message.to_string()), options)
190            .await;
191
192        // Restore config.
193        self.config.system_prompt = original_system;
194        self.config.backend.stream = original_stream;
195
196        let output = output?;
197        output
198            .tool_calls
199            .into_iter()
200            .next()
201            .map(|tc| tc.arguments)
202            .ok_or_else(|| chimera_core::AgentError::TurnFailed {
203                message: "model did not produce a tool call".into(),
204            })
205    }
206}
207
208fn prompt_text_from_input(input: &Input) -> Result<String> {
209    match input {
210        Input::Text(s) => Ok(s.clone()),
211        Input::Structured(parts) => {
212            let parts = parts
213                .iter()
214                .map(|p| match p {
215                    InputPart::Text(t) => Ok(t.as_str()),
216                    InputPart::Image(path) => Err(AgentError::Other {
217                        message: format!(
218                            "opencode HTTP mode does not support image input yet: {}",
219                            path.display()
220                        ),
221                        source: None,
222                    }),
223                    _ => Err(AgentError::Other {
224                        message: "opencode HTTP mode received an unsupported structured input part"
225                            .into(),
226                        source: None,
227                    }),
228                })
229                .collect::<Result<Vec<_>>>()?;
230            Ok(parts.join("\n\n"))
231        }
232        _ => Err(AgentError::Other {
233            message: "opencode HTTP mode received an unsupported input shape".into(),
234            source: None,
235        }),
236    }
237}
238
239/// Public session type -- dispatches between HTTP (direct API) and Agent (subprocess) modes.
240pub enum OpenCodeSession {
241    /// HTTP mode: calls the OpenCode Zen API (or compatible endpoint) directly.
242    Http(Box<HttpSession>),
243    /// Agent mode: spawns `opencode serve` and communicates via the local HTTP API.
244    Agent(Box<crate::agent_session::OpenCodeAgentSession>),
245}
246
247impl Session for OpenCodeSession {
248    fn turn_stream(
249        &mut self,
250        input: Input,
251        options: TurnOptions,
252    ) -> Pin<Box<dyn Future<Output = Result<EventStream>> + Send + '_>> {
253        match self {
254            OpenCodeSession::Http(s) => s.turn_stream(input, options),
255            OpenCodeSession::Agent(s) => s.turn_stream(input, options),
256        }
257    }
258
259    fn session_id(&self) -> Option<&str> {
260        match self {
261            OpenCodeSession::Http(s) => s.session_id(),
262            OpenCodeSession::Agent(s) => s.session_id(),
263        }
264    }
265
266    fn interrupt(&mut self) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
267        match self {
268            OpenCodeSession::Http(s) => s.interrupt(),
269            OpenCodeSession::Agent(s) => s.interrupt(),
270        }
271    }
272}
273
274impl Session for HttpSession {
275    fn turn_stream(
276        &mut self,
277        input: Input,
278        options: TurnOptions,
279    ) -> Pin<Box<dyn Future<Output = Result<EventStream>> + Send + '_>> {
280        Box::pin(async move {
281            if self.active_interrupt.lock().await.is_some() {
282                return Err(AgentError::Other {
283                    message: "turn already in progress".into(),
284                    source: None,
285                });
286            }
287
288            let client = self.get_or_build_client()?;
289            let history_snapshot = self.history.lock().await.clone();
290            let request = self.build_request(&input, &options, &history_snapshot)?;
291            let url = format!("{}/chat/completions", self.base_url());
292            let api_key = self.api_key.clone();
293            let max_retries = self.max_retries();
294            let is_streaming = request.stream == Some(true);
295
296            // Append user message to history for multi-turn.
297            let prompt_text = prompt_text_from_input(&input)?;
298            self.history
299                .lock()
300                .await
301                .push(WireMessage::user(prompt_text));
302
303            let (tx, rx) =
304                tokio::sync::mpsc::channel::<std::result::Result<AgentEvent, AgentError>>(128);
305            let session_id = Arc::clone(&self.session_id);
306            let (interrupt_tx, interrupt_rx) = tokio::sync::oneshot::channel();
307            {
308                let mut slot = self.active_interrupt.lock().await;
309                *slot = Some(interrupt_tx);
310            }
311            let active_interrupt = Arc::clone(&self.active_interrupt);
312            let history = Arc::clone(&self.history);
313            let timeout_duration = options.timeout;
314
315            tokio::spawn(async move {
316                execute_turn(
317                    client,
318                    url,
319                    api_key,
320                    request,
321                    max_retries,
322                    is_streaming,
323                    tx,
324                    interrupt_rx,
325                    session_id,
326                    active_interrupt,
327                    history,
328                    timeout_duration,
329                )
330                .await;
331            });
332
333            Ok(EventStream::from_receiver(rx))
334        })
335    }
336
337    fn session_id(&self) -> Option<&str> {
338        self.session_id.get().map(|s| s.as_str())
339    }
340
341    fn interrupt(&mut self) -> Pin<Box<dyn Future<Output = Result<()>> + Send + '_>> {
342        let active_interrupt = Arc::clone(&self.active_interrupt);
343        Box::pin(async move {
344            if let Some(tx) = active_interrupt.lock().await.take() {
345                let _ = tx.send(());
346            }
347            Ok(())
348        })
349    }
350}
351
352#[allow(clippy::too_many_arguments)]
353async fn execute_turn(
354    client: reqwest::Client,
355    url: String,
356    api_key: Option<String>,
357    request: ChatCompletionRequest,
358    max_retries: u32,
359    is_streaming: bool,
360    tx: tokio::sync::mpsc::Sender<std::result::Result<AgentEvent, AgentError>>,
361    mut interrupt_rx: tokio::sync::oneshot::Receiver<()>,
362    session_id: Arc<OnceLock<String>>,
363    active_interrupt: Arc<tokio::sync::Mutex<Option<tokio::sync::oneshot::Sender<()>>>>,
364    history: Arc<tokio::sync::Mutex<Vec<WireMessage>>>,
365    timeout_duration: Option<Duration>,
366) {
367    let timeout_sleep = tokio::time::sleep(timeout_duration.unwrap_or(Duration::MAX));
368    tokio::pin!(timeout_sleep);
369    let has_timeout = timeout_duration.is_some();
370
371    let _ = tx.send(Ok(AgentEvent::TurnStarted)).await;
372
373    // Retry loop for the HTTP request.
374    let mut attempt = 0u32;
375    let response = loop {
376        let mut req = client.post(&url).json(&request);
377        if let Some(ref key) = api_key {
378            req = req.bearer_auth(key);
379        }
380
381        tokio::select! {
382            _ = &mut interrupt_rx => {
383                let _ = tx.send(Err(AgentError::Interrupted)).await;
384                active_interrupt.lock().await.take();
385                return;
386            }
387            _ = &mut timeout_sleep, if has_timeout => {
388                let _ = tx.send(Err(AgentError::Timeout {
389                    duration: timeout_duration.unwrap(),
390                })).await;
391                active_interrupt.lock().await.take();
392                return;
393            }
394            result = req.send() => {
395                match result {
396                    Ok(resp) if resp.status().is_success() => break resp,
397                    Ok(resp) => {
398                        let status = resp.status();
399                        let retry_after = resp.headers()
400                            .get("retry-after")
401                            .and_then(|v| v.to_str().ok())
402                            .and_then(|s| s.parse::<u64>().ok())
403                            .unwrap_or(60);
404                        let body = resp.text().await.unwrap_or_default();
405
406                        if attempt >= max_retries {
407                            let _ = tx.send(Ok(AgentEvent::TurnFailed {
408                                message: format!("HTTP {status}: {body}"),
409                            })).await;
410                            active_interrupt.lock().await.take();
411                            return;
412                        }
413
414                        let delay = if status.as_u16() == 429 {
415                            Duration::from_secs(retry_after.min(MAX_RETRY_DELAY.as_secs()))
416                        } else if status.is_server_error() {
417                            (BASE_RETRY_DELAY * (1 << attempt)).min(MAX_RETRY_DELAY)
418                        } else {
419                            // Client error (4xx other than 429) -- don't retry.
420                            let _ = tx.send(Ok(AgentEvent::TurnFailed {
421                                message: format!("HTTP {status}: {body}"),
422                            })).await;
423                            active_interrupt.lock().await.take();
424                            return;
425                        };
426
427                        attempt += 1;
428                        tokio::time::sleep(delay).await;
429                        continue;
430                    }
431                    Err(e) => {
432                        if attempt >= max_retries {
433                            let _ = tx.send(Err(AgentError::Other {
434                                message: format!("HTTP request failed: {e}"),
435                                source: Some(Box::new(e)),
436                            })).await;
437                            active_interrupt.lock().await.take();
438                            return;
439                        }
440                        attempt += 1;
441                        let delay = (BASE_RETRY_DELAY * (1 << attempt)).min(MAX_RETRY_DELAY);
442                        tokio::time::sleep(delay).await;
443                        continue;
444                    }
445                }
446            }
447        }
448    };
449
450    if is_streaming {
451        handle_sse_stream(
452            response,
453            &tx,
454            interrupt_rx,
455            &session_id,
456            &history,
457            timeout_duration,
458        )
459        .await;
460    } else {
461        handle_non_streaming(response, &tx, &session_id, &history).await;
462    }
463
464    active_interrupt.lock().await.take();
465}
466
467async fn handle_non_streaming(
468    response: reqwest::Response,
469    tx: &tokio::sync::mpsc::Sender<std::result::Result<AgentEvent, AgentError>>,
470    session_id: &Arc<OnceLock<String>>,
471    history: &Arc<tokio::sync::Mutex<Vec<WireMessage>>>,
472) {
473    match response.json::<ChatCompletionResponse>().await {
474        Ok(resp) => {
475            // Append assistant response to history for multi-turn.
476            if let Some(choice) = resp.choices.first() {
477                let content = choice.message.content.as_deref().unwrap_or("");
478                history.lock().await.push(WireMessage::assistant(content));
479            }
480
481            let events = translate::translate_response(resp);
482            for event in events {
483                if let AgentEvent::TurnCompleted {
484                    result: Some(ref r),
485                    ..
486                } = event
487                    && let Some(ref sid) = r.session_id
488                {
489                    let _ = session_id.set(sid.clone());
490                }
491                if tx.send(Ok(event)).await.is_err() {
492                    return;
493                }
494            }
495        }
496        Err(e) => {
497            let _ = tx
498                .send(Err(AgentError::Other {
499                    message: format!("failed to parse response: {e}"),
500                    source: Some(Box::new(e)),
501                }))
502                .await;
503        }
504    }
505}
506
507#[allow(clippy::too_many_arguments)]
508/// Accumulated state for a tool call being built across SSE chunks.
509#[derive(Default)]
510struct PendingToolCall {
511    id: String,
512    name: String,
513    arguments: String,
514}
515
516async fn handle_sse_stream(
517    response: reqwest::Response,
518    tx: &tokio::sync::mpsc::Sender<std::result::Result<AgentEvent, AgentError>>,
519    mut interrupt_rx: tokio::sync::oneshot::Receiver<()>,
520    session_id: &Arc<OnceLock<String>>,
521    history: &Arc<tokio::sync::Mutex<Vec<WireMessage>>>,
522    timeout_duration: Option<Duration>,
523) {
524    let timeout_sleep = tokio::time::sleep(timeout_duration.unwrap_or(Duration::MAX));
525    tokio::pin!(timeout_sleep);
526    let has_timeout = timeout_duration.is_some();
527    let mut buf = String::new();
528    let mut full_content = String::new();
529    // Tool call accumulator: index -> PendingToolCall
530    let mut pending_tool_calls: std::collections::BTreeMap<usize, PendingToolCall> =
531        std::collections::BTreeMap::new();
532    let mut response = response;
533
534    loop {
535        tokio::select! {
536            _ = &mut interrupt_rx => {
537                let _ = tx.send(Err(AgentError::Interrupted)).await;
538                return;
539            }
540            _ = &mut timeout_sleep, if has_timeout => {
541                let _ = tx.send(Err(AgentError::Timeout {
542                    duration: timeout_duration.unwrap(),
543                })).await;
544                return;
545            }
546            chunk_result = response.chunk() => {
547                match chunk_result {
548                    Ok(Some(bytes)) => {
549                        buf.push_str(&String::from_utf8_lossy(&bytes));
550
551                        while let Some(pos) = buf.find('\n') {
552                            let line = buf[..pos].to_string();
553                            buf = buf[pos + 1..].to_string();
554
555                            let line = line.trim();
556                            if line.is_empty() {
557                                continue;
558                            }
559
560                            if let Some(data) = line.strip_prefix("data: ") {
561                                if data == "[DONE]" {
562                                    // Append accumulated content to history.
563                                    if !full_content.is_empty() {
564                                        history
565                                            .lock()
566                                            .await
567                                            .push(WireMessage::assistant(&full_content));
568                                    }
569                                    // Emit accumulated tool calls.
570                                    for (_, tc) in pending_tool_calls {
571                                        if tx
572                                            .send(Ok(AgentEvent::ToolCall {
573                                                id: tc.id,
574                                                name: tc.name,
575                                                arguments: tc.arguments,
576                                            }))
577                                            .await
578                                            .is_err()
579                                        {
580                                            return;
581                                        }
582                                    }
583                                    return;
584                                }
585
586                                if let Ok(chunk) =
587                                    serde_json::from_str::<ChatCompletionChunk>(data)
588                                {
589                                    // Accumulate content for history.
590                                    for choice in &chunk.choices {
591                                        if let Some(ref c) = choice.delta.content {
592                                            full_content.push_str(c);
593                                        }
594                                        // Accumulate tool call deltas.
595                                        for tc_delta in &choice.delta.tool_calls {
596                                            let entry = pending_tool_calls
597                                                .entry(tc_delta.index)
598                                                .or_default();
599                                            if let Some(ref id) = tc_delta.id {
600                                                entry.id.clone_from(id);
601                                            }
602                                            if let Some(ref f) = tc_delta.function {
603                                                if let Some(ref name) = f.name {
604                                                    entry.name.clone_from(name);
605                                                }
606                                                if let Some(ref args) = f.arguments {
607                                                    entry.arguments.push_str(args);
608                                                }
609                                            }
610                                        }
611                                    }
612
613                                    let events = translate::translate_chunk(chunk);
614                                    for event in events {
615                                        if let AgentEvent::TurnCompleted {
616                                            result: Some(ref r),
617                                            ..
618                                        } = event
619                                            && let Some(ref sid) = r.session_id
620                                        {
621                                            let _ = session_id.set(sid.clone());
622                                        }
623                                        if tx.send(Ok(event)).await.is_err() {
624                                            return;
625                                        }
626                                    }
627                                }
628                            }
629                        }
630                    }
631                    Ok(None) => {
632                        // Stream ended without [DONE]. Append any accumulated content.
633                        if !full_content.is_empty() {
634                            history.lock().await.push(
635                                WireMessage::assistant(&full_content),
636                            );
637                        }
638                        return;
639                    }
640                    Err(e) => {
641                        let _ = tx.send(Err(AgentError::Other {
642                            message: format!("SSE stream error: {e}"),
643                            source: Some(Box::new(e)),
644                        })).await;
645                        return;
646                    }
647                }
648            }
649        }
650    }
651}
652
653#[cfg(test)]
654mod tests {
655    use super::*;
656
657    fn make_session(config: SessionConfig<OpenCodeConfig>) -> HttpSession {
658        HttpSession::new(Some("test-key".into()), OpenCodeProvider::Zen, config)
659    }
660
661    fn default_config() -> SessionConfig<OpenCodeConfig> {
662        SessionConfig::builder()
663            .backend(OpenCodeConfig::default())
664            .build()
665    }
666
667    #[test]
668    fn session_id_none_initially() {
669        let session = make_session(default_config());
670        assert!(session.session_id().is_none());
671    }
672
673    #[test]
674    fn session_id_after_set() {
675        let session = make_session(default_config()).with_session_id("sess-1".into());
676        assert_eq!(session.session_id(), Some("sess-1"));
677    }
678
679    #[test]
680    fn model_defaults_to_glm5() {
681        let session = make_session(default_config());
682        assert_eq!(session.model(), "glm-5");
683    }
684
685    #[test]
686    fn model_from_config() {
687        let config = SessionConfig::builder()
688            .model("claude-sonnet-4-20250514")
689            .backend(OpenCodeConfig::default())
690            .build();
691        let session = make_session(config);
692        assert_eq!(session.model(), "claude-sonnet-4-20250514");
693    }
694
695    #[test]
696    fn model_strips_provider_prefix() {
697        let config = SessionConfig::builder()
698            .model("opencode-go/glm-5")
699            .backend(OpenCodeConfig::default())
700            .build();
701        let session = make_session(config);
702        assert_eq!(session.model(), "glm-5");
703    }
704
705    #[tokio::test]
706    async fn build_request_basic() {
707        let session = make_session(default_config());
708        let history = session.history.lock().await.clone();
709        let req = session
710            .build_request(
711                &Input::Text("hello".into()),
712                &TurnOptions::default(),
713                &history,
714            )
715            .unwrap();
716
717        assert_eq!(req.model, "glm-5");
718        assert_eq!(req.messages.len(), 1);
719        assert_eq!(req.messages[0].role, "user");
720        assert_eq!(req.messages[0].content, "hello");
721        assert_eq!(req.stream, Some(true));
722        assert!(req.response_format.is_none());
723    }
724
725    #[tokio::test]
726    async fn build_request_with_system_prompt_and_schema() {
727        let config = SessionConfig::builder()
728            .system_prompt("Be helpful")
729            .backend(OpenCodeConfig::builder().stream(false).build())
730            .build();
731        let session = make_session(config);
732        let options = TurnOptions {
733            output_schema: Some(serde_json::json!({"type": "object"})),
734            ..Default::default()
735        };
736        let history = session.history.lock().await.clone();
737        let req = session
738            .build_request(&Input::Text("test".into()), &options, &history)
739            .unwrap();
740
741        assert_eq!(req.messages[0].role, "system");
742        assert_eq!(req.messages[0].content, "Be helpful");
743        assert_eq!(req.messages[1].role, "user");
744        assert!(req.stream.is_none());
745        assert!(req.response_format.is_some());
746        assert_eq!(
747            req.response_format.as_ref().unwrap().format_type,
748            "json_schema"
749        );
750    }
751
752    #[tokio::test]
753    async fn build_request_includes_history() {
754        let config = default_config();
755        let session = make_session(config);
756        {
757            let mut h = session.history.lock().await;
758            h.push(WireMessage::user("first"));
759            h.push(WireMessage::assistant("reply"));
760        }
761        let history = session.history.lock().await.clone();
762        let req = session
763            .build_request(
764                &Input::Text("second".into()),
765                &TurnOptions::default(),
766                &history,
767            )
768            .unwrap();
769
770        assert_eq!(req.messages.len(), 3);
771        assert_eq!(req.messages[0].content, "first");
772        assert_eq!(req.messages[1].content, "reply");
773        assert_eq!(req.messages[2].content, "second");
774    }
775
776    #[tokio::test]
777    async fn turn_stream_rejects_when_turn_active() {
778        let mut session = make_session(default_config());
779        let (tx, _rx) = tokio::sync::oneshot::channel();
780        *session.active_interrupt.lock().await = Some(tx);
781
782        let err = session
783            .turn_stream(Input::Text("hi".into()), TurnOptions::default())
784            .await;
785        match err {
786            Err(AgentError::Other { message, .. }) => {
787                assert_eq!(message, "turn already in progress")
788            }
789            _ => panic!("expected Other error"),
790        }
791    }
792
793    #[test]
794    fn prompt_text_rejects_images() {
795        let err = prompt_text_from_input(&Input::Structured(vec![InputPart::Image(
796            std::path::PathBuf::from("/tmp/example.png"),
797        )]))
798        .unwrap_err();
799
800        assert!(
801            err.to_string()
802                .contains("opencode HTTP mode does not support image input yet")
803        );
804    }
805}
chimera_opencode/session.rs

chimera_opencode/
session.rs
