kcode_k1_codex_adapter/
shim.rs

1use std::{collections::VecDeque, future::Future, pin::Pin};
2
3use crate::{Adapter, Error, ErrorKind, Event, ToolCall, ToolResult};
4
5/// Fixed successful response used to release every Codex dynamic-tool request.
6pub const ASYNC_TOOL_ACKNOWLEDGEMENT: &str = "The tool was launched asynchronously. Its result will not be available during this turn. Do not wait for or poll this call; continue the turn without its result. The result will be provided in a subsequent turn when it becomes available.";
7
8/// Future returned while handing one canonical tool-call box to K1.
9pub type ToolLaunchFuture<'a> = Pin<Box<dyn Future<Output = Result<(), String>> + Send + 'a>>;
10
11/// Accepts canonical tool-call boxes for asynchronous execution.
12pub trait ToolCallLauncher<B>: Send {
13    /// Accepts, starts, or queues one tool call without waiting for completion.
14    fn launch<'a>(&'a mut self, box_: &'a B) -> ToolLaunchFuture<'a>;
15}
16
17/// Converts between a chat engine's canonical box type and text visible to Codex.
18///
19/// A shim calls `tool_call_box` exactly once for each dynamic tool call it
20/// receives. `box_text` must return the complete representation that should be
21/// placed into Codex context. Implementations must not add ordering of their
22/// own: the shim preserves insertion order.
23pub trait BoxCodec {
24    /// Canonical box type owned by the chat engine.
25    type Box: Clone;
26
27    /// Converts one typed Codex dynamic-tool request into a canonical box.
28    fn tool_call_box(&mut self, call: &ToolCall) -> Self::Box;
29
30    /// Returns the exact textual representation of one box for Codex.
31    fn box_text<'a>(&self, box_: &'a Self::Box) -> &'a str;
32}
33
34/// One ordered item produced by a completed native Codex turn.
35#[derive(Clone, Debug, PartialEq)]
36pub enum ShimItem<B> {
37    /// Adjacent streamed assistant-text deltas, coalesced.
38    Text(String),
39    /// A canonical dynamic-tool-call box.
40    Box(B),
41}
42
43/// Atomic terminal output from one native Codex turn.
44#[derive(Clone, Debug, PartialEq)]
45pub struct ShimOutput<B> {
46    /// Assistant text and call boxes in exact provider event order.
47    pub items: Vec<ShimItem<B>>,
48}
49
50#[derive(Clone, Copy, Debug, PartialEq, Eq)]
51enum Health {
52    Ready,
53    Unusable,
54}
55
56/// A single-owner, per-conversation bridge between K1 boxes and Codex turns.
57///
58/// `Shim` deliberately contains no locking. A caller creates one shim for one
59/// conversation and invokes it sequentially. Different shims may share clones
60/// of the same [`Adapter`], whose actor owns transport concurrency.
61pub struct Shim<C: BoxCodec> {
62    adapter: Adapter,
63    conversation_key: String,
64    codec: C,
65    launcher: Box<dyn ToolCallLauncher<C::Box>>,
66    health: Health,
67    pending_boxes: VecDeque<C::Box>,
68}
69
70impl<C: BoxCodec> Shim<C> {
71    /// Creates a ready shim for one conversation.
72    pub fn new(
73        adapter: Adapter,
74        conversation_key: impl Into<String>,
75        codec: C,
76        launcher: Box<dyn ToolCallLauncher<C::Box>>,
77    ) -> Self {
78        Self {
79            adapter,
80            conversation_key: conversation_key.into(),
81            codec,
82            launcher,
83            health: Health::Ready,
84            pending_boxes: VecDeque::new(),
85        }
86    }
87
88    /// Appends one externally produced box in canonical chat-engine order.
89    ///
90    /// Call boxes returned by [`Shim::infer`] are output only and are not added
91    /// to this pending queue automatically.
92    pub fn record_box(&mut self, box_: C::Box) {
93        self.pending_boxes.push_back(box_);
94    }
95
96    /// Appends externally produced boxes without changing their order.
97    pub fn record_boxes(&mut self, boxes: impl IntoIterator<Item = C::Box>) {
98        self.pending_boxes.extend(boxes);
99    }
100
101    /// Number of boxes waiting to be submitted to a fresh Codex turn.
102    pub fn pending_box_count(&self) -> usize {
103        self.pending_boxes.len()
104    }
105
106    /// Runs exactly one fresh native Codex turn through terminal completion.
107    ///
108    /// Pending K1 boxes prefix ordinary input and are removed only after the
109    /// adapter accepts the turn. Every dynamic call is converted once, launched,
110    /// acknowledged with [`ASYNC_TOOL_ACKNOWLEDGEMENT`], and retained as an
111    /// ordered output box. No partial output is returned if the active turn
112    /// fails.
113    pub async fn infer(&mut self, input: impl Into<String>) -> Result<ShimOutput<C::Box>, Error> {
114        if self.health == Health::Unusable {
115            return Err(self.unusable());
116        }
117
118        let submitted_box_count = self.pending_boxes.len();
119        let input = append_section(self.render_pending_boxes(), &input.into());
120
121        // Cancellation anywhere after this point fails closed. A definite
122        // start rejection restores readiness because no native turn exists.
123        self.health = Health::Unusable;
124        let mut turn = match self
125            .adapter
126            .start_turn(self.conversation_key.clone(), input)
127            .await
128        {
129            Ok(turn) => turn,
130            Err(error) => {
131                self.health = Health::Ready;
132                return Err(error);
133            }
134        };
135
136        for _ in 0..submitted_box_count {
137            let removed = self.pending_boxes.pop_front();
138            debug_assert!(removed.is_some());
139        }
140
141        let mut items = Vec::new();
142        loop {
143            match turn.next_event().await {
144                Some(Event::TextDelta(delta)) => push_text(&mut items, delta),
145                Some(Event::ToolCall(call)) => {
146                    let box_ = self.codec.tool_call_box(&call);
147                    if let Err(message) = self.launcher.launch(&box_).await {
148                        return Err(Error {
149                            kind: ErrorKind::LaunchRejected,
150                            message,
151                            diagnostics: self.adapter.diagnostics(),
152                        });
153                    }
154                    turn.respond(
155                        call.call_id,
156                        ToolResult {
157                            success: true,
158                            output: ASYNC_TOOL_ACKNOWLEDGEMENT.to_owned(),
159                        },
160                    )
161                    .await?;
162                    items.push(ShimItem::Box(box_));
163                }
164                Some(Event::Done) => {
165                    self.health = Health::Ready;
166                    return Ok(ShimOutput { items });
167                }
168                Some(Event::Error(error)) => return Err(error),
169                None => return Err(self.adapter.unavailable()),
170            }
171        }
172    }
173
174    fn render_pending_boxes(&self) -> String {
175        let mut output = String::new();
176        for box_ in &self.pending_boxes {
177            output = append_section(output, self.codec.box_text(box_));
178        }
179        output
180    }
181
182    fn unusable(&self) -> Error {
183        let mut error = self.adapter.unavailable();
184        error.message =
185            "Codex shim cannot be reused after an active turn failed or was cancelled".to_owned();
186        error
187    }
188}
189
190fn push_text<B>(items: &mut Vec<ShimItem<B>>, delta: String) {
191    match items.last_mut() {
192        Some(ShimItem::Text(text)) => text.push_str(&delta),
193        _ => items.push(ShimItem::Text(delta)),
194    }
195}
196
197fn append_section(mut output: String, section: &str) -> String {
198    if section.is_empty() {
199        return output;
200    }
201    if !output.is_empty() && !output.ends_with('\n') {
202        output.push('\n');
203    }
204    output.push_str(section);
205    output
206}
207
208#[cfg(test)]
209mod tests {
210    use super::*;
211    use crate::{Config, DynamicTool};
212    use serde_json::{Value, json};
213    use std::sync::{Arc, Mutex};
214
215    #[derive(Clone, Debug, PartialEq, Eq)]
216    struct TestBox(String);
217
218    #[derive(Default)]
219    struct TestCodec {
220        conversions: Vec<String>,
221    }
222
223    impl BoxCodec for TestCodec {
224        type Box = TestBox;
225
226        fn tool_call_box(&mut self, call: &ToolCall) -> Self::Box {
227            self.conversions.push(call.call_id.clone());
228            TestBox(call.call_id.clone())
229        }
230
231        fn box_text<'a>(&self, box_: &'a Self::Box) -> &'a str {
232            &box_.0
233        }
234    }
235
236    #[derive(Clone, Default)]
237    struct LauncherState {
238        attempts: Arc<Mutex<Vec<String>>>,
239        accepted: Arc<Mutex<Vec<String>>>,
240    }
241
242    impl LauncherState {
243        fn attempts(&self) -> Vec<String> {
244            self.attempts
245                .lock()
246                .unwrap_or_else(|poisoned| poisoned.into_inner())
247                .clone()
248        }
249
250        fn accepted(&self) -> Vec<String> {
251            self.accepted
252                .lock()
253                .unwrap_or_else(|poisoned| poisoned.into_inner())
254                .clone()
255        }
256    }
257
258    struct TestLauncher {
259        state: LauncherState,
260        rejection: Option<(String, String)>,
261        sequence_path: Option<std::path::PathBuf>,
262    }
263
264    impl TestLauncher {
265        fn accepting(state: LauncherState) -> Self {
266            Self {
267                state,
268                rejection: None,
269                sequence_path: None,
270            }
271        }
272
273        fn rejecting(state: LauncherState, call_id: &str, message: &str) -> Self {
274            Self {
275                state,
276                rejection: Some((call_id.to_owned(), message.to_owned())),
277                sequence_path: None,
278            }
279        }
280
281        fn with_sequence_path(mut self, path: std::path::PathBuf) -> Self {
282            self.sequence_path = Some(path);
283            self
284        }
285    }
286
287    impl ToolCallLauncher<TestBox> for TestLauncher {
288        fn launch<'a>(&'a mut self, box_: &'a TestBox) -> ToolLaunchFuture<'a> {
289            let call_id = box_.0.clone();
290            let state = self.state.clone();
291            let rejection = self
292                .rejection
293                .as_ref()
294                .filter(|(rejected, _)| rejected == &call_id)
295                .map(|(_, message)| message.clone());
296            let sequence_path = self.sequence_path.clone();
297
298            Box::pin(async move {
299                state
300                    .attempts
301                    .lock()
302                    .unwrap_or_else(|poisoned| poisoned.into_inner())
303                    .push(call_id.clone());
304                if let Some(message) = rejection {
305                    return Err(message);
306                }
307                state
308                    .accepted
309                    .lock()
310                    .unwrap_or_else(|poisoned| poisoned.into_inner())
311                    .push(call_id.clone());
312                if let Some(path) = sequence_path {
313                    use std::io::Write;
314
315                    let mut file = std::fs::OpenOptions::new()
316                        .create(true)
317                        .append(true)
318                        .open(path)
319                        .unwrap();
320                    writeln!(file, "launch-{call_id}").unwrap();
321                }
322                Ok(())
323            })
324        }
325    }
326
327    #[cfg(unix)]
328    struct TestApp {
329        directory: std::path::PathBuf,
330        executable: std::path::PathBuf,
331    }
332
333    #[cfg(unix)]
334    impl TestApp {
335        fn new(script: &str) -> Self {
336            use std::{
337                os::unix::fs::PermissionsExt,
338                sync::atomic::{AtomicU64, Ordering},
339                time::{SystemTime, UNIX_EPOCH},
340            };
341
342            static NEXT: AtomicU64 = AtomicU64::new(0);
343            let nonce = SystemTime::now()
344                .duration_since(UNIX_EPOCH)
345                .unwrap()
346                .as_nanos();
347            let directory = std::env::temp_dir().join(format!(
348                "kcode-k1-codex-adapter-{}-{nonce}-{}",
349                std::process::id(),
350                NEXT.fetch_add(1, Ordering::Relaxed)
351            ));
352            std::fs::create_dir(&directory).unwrap();
353            let executable = directory.join("codex");
354            std::fs::write(&executable, script).unwrap();
355            let mut permissions = std::fs::metadata(&executable).unwrap().permissions();
356            permissions.set_mode(0o700);
357            std::fs::set_permissions(&executable, permissions).unwrap();
358            Self {
359                directory,
360                executable,
361            }
362        }
363
364        fn path(&self, name: &str) -> std::path::PathBuf {
365            self.directory.join(name)
366        }
367
368        async fn adapter(&self) -> Adapter {
369            Adapter::open(Config {
370                executable: self.executable.clone(),
371                working_directory: self.directory.to_string_lossy().into_owned(),
372                model: "test-model".into(),
373                reasoning_effort: None,
374                base_instructions: String::new(),
375                tools: vec![DynamicTool {
376                    name: "lookup".into(),
377                    description: "Lookup a value".into(),
378                    input_schema: json!({"type":"object"}),
379                }],
380            })
381            .await
382            .unwrap()
383        }
384    }
385
386    #[cfg(unix)]
387    impl Drop for TestApp {
388        fn drop(&mut self) {
389            let _ = std::fs::remove_dir_all(&self.directory);
390        }
391    }
392
393    #[cfg(unix)]
394    async fn wait_for_lines(path: &std::path::Path, minimum: usize) {
395        tokio::time::timeout(std::time::Duration::from_secs(3), async {
396            loop {
397                let count = std::fs::read_to_string(path)
398                    .map(|text| text.lines().count())
399                    .unwrap_or(0);
400                if count >= minimum {
401                    return;
402                }
403                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
404            }
405        })
406        .await
407        .expect("timed out waiting for app-server log");
408    }
409
410    #[cfg(unix)]
411    async fn wait_for_diagnostics(adapter: &Adapter, expected: &[u8]) {
412        tokio::time::timeout(std::time::Duration::from_secs(3), async {
413            loop {
414                let diagnostics = adapter.diagnostics();
415                if diagnostics
416                    .windows(expected.len())
417                    .any(|window| window == expected)
418                {
419                    return;
420                }
421                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
422            }
423        })
424        .await
425        .expect("timed out waiting for app-server diagnostics");
426    }
427
428    #[cfg(unix)]
429    #[tokio::test]
430    async fn launches_ordered_call_waves_before_exact_acknowledgements_and_waits_for_done() {
431        let app = TestApp::new(
432            r#"#!/bin/sh
433set -eu
434IFS= read -r initialize
435echo '{"id":0,"result":{}}'
436IFS= read -r initialized
437IFS= read -r thread_start
438echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
439IFS= read -r turn_start
440printf '%s\n' "$turn_start" > turn-start.log
441echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
442echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"pre"}}'
443echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"face"}}'
444echo '{"id":77,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"A","tool":"lookup","arguments":{"n":1}}}'
445IFS= read -r response
446printf '%s\n' "$response" >> responses.log
447printf 'ack-A\n' >> sequence.log
448echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"mid"}}'
449echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"dle"}}'
450echo '{"id":78,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"B","tool":"lookup","arguments":{"n":2}}}'
451echo '{"id":79,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"C","tool":"lookup","arguments":{"n":3}}}'
452IFS= read -r response
453printf '%s\n' "$response" >> responses.log
454printf 'ack-B\n' >> sequence.log
455IFS= read -r response
456printf '%s\n' "$response" >> responses.log
457printf 'ack-C\n' >> sequence.log
458while [ ! -e release ]; do sleep 0.01; done
459echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"tail"}}'
460echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"end"}}'
461echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'
462"#,
463        );
464        let state = LauncherState::default();
465        let launcher =
466            TestLauncher::accepting(state.clone()).with_sequence_path(app.path("sequence.log"));
467        let mut shim = Shim::new(
468            app.adapter().await,
469            "conversation-1",
470            TestCodec::default(),
471            Box::new(launcher),
472        );
473        shim.record_boxes([TestBox("<pending-1>".into()), TestBox("<pending-2>".into())]);
474
475        let task = tokio::spawn(async move {
476            let result = shim.infer("request").await;
477            (shim, result)
478        });
479        wait_for_lines(&app.path("sequence.log"), 6).await;
480        assert!(!task.is_finished(), "infer returned before Event::Done");
481        std::fs::write(app.path("release"), "").unwrap();
482        let (shim, output) = tokio::time::timeout(std::time::Duration::from_secs(3), task)
483            .await
484            .unwrap()
485            .unwrap();
486        let output = output.unwrap();
487
488        assert_eq!(
489            output.items,
490            vec![
491                ShimItem::Text("preface".into()),
492                ShimItem::Box(TestBox("A".into())),
493                ShimItem::Text("middle".into()),
494                ShimItem::Box(TestBox("B".into())),
495                ShimItem::Box(TestBox("C".into())),
496                ShimItem::Text("tailend".into()),
497            ]
498        );
499        assert_eq!(shim.codec.conversions, ["A", "B", "C"]);
500        assert_eq!(state.attempts(), ["A", "B", "C"]);
501        assert_eq!(state.accepted(), ["A", "B", "C"]);
502        assert_eq!(shim.pending_box_count(), 0);
503
504        let start: Value =
505            serde_json::from_str(&std::fs::read_to_string(app.path("turn-start.log")).unwrap())
506                .unwrap();
507        assert_eq!(
508            start
509                .pointer("/params/input/0/text")
510                .and_then(Value::as_str),
511            Some("<pending-1>\n<pending-2>\nrequest")
512        );
513
514        let responses = std::fs::read_to_string(app.path("responses.log")).unwrap();
515        for (line, id) in responses.lines().zip([77, 78, 79]) {
516            let response: Value = serde_json::from_str(line).unwrap();
517            assert_eq!(response["id"], id);
518            assert_eq!(response["result"]["success"], true);
519            assert_eq!(
520                response
521                    .pointer("/result/contentItems/0/text")
522                    .and_then(Value::as_str),
523                Some(ASYNC_TOOL_ACKNOWLEDGEMENT)
524            );
525        }
526
527        let sequence: Vec<_> = std::fs::read_to_string(app.path("sequence.log"))
528            .unwrap()
529            .lines()
530            .map(str::to_owned)
531            .collect();
532        assert_eq!(sequence.len(), 6);
533        for entry in [
534            "launch-A", "launch-B", "launch-C", "ack-A", "ack-B", "ack-C",
535        ] {
536            assert_eq!(
537                sequence
538                    .iter()
539                    .filter(|observed| observed.as_str() == entry)
540                    .count(),
541                1,
542                "unexpected sequence: {sequence:?}"
543            );
544        }
545        let position = |entry: &str| {
546            sequence
547                .iter()
548                .position(|observed| observed == entry)
549                .unwrap()
550        };
551        assert!(position("launch-A") < position("launch-B"));
552        assert!(position("launch-B") < position("launch-C"));
553        assert!(position("launch-A") < position("ack-A"));
554        assert!(position("launch-B") < position("ack-B"));
555        assert!(position("launch-C") < position("ack-C"));
556    }
557
558    #[cfg(unix)]
559    #[tokio::test]
560    async fn no_tool_inference_leaves_launcher_untouched_and_uses_fresh_turns() {
561        let app = TestApp::new(
562            r#"#!/bin/sh
563set -eu
564read initialize
565echo '{"id":0,"result":{}}'
566read initialized
567read thread_start
568echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
569read first_start
570printf '%s\n' "$first_start" >> starts.log
571echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
572echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"hel"}}'
573echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"lo"}}'
574echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'
575read second_start
576printf '%s\n' "$second_start" >> starts.log
577echo '{"id":3,"result":{"turn":{"id":"turn-2"}}}'
578echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-2","delta":"again"}}'
579echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-2","status":"completed"}}}'
580"#,
581        );
582        let state = LauncherState::default();
583        let mut shim = Shim::new(
584            app.adapter().await,
585            "conversation-1",
586            TestCodec::default(),
587            Box::new(TestLauncher::accepting(state.clone())),
588        );
589
590        assert_eq!(
591            shim.infer("first").await.unwrap().items,
592            [ShimItem::Text("hello".into())]
593        );
594        assert_eq!(
595            shim.infer("second").await.unwrap().items,
596            [ShimItem::Text("again".into())]
597        );
598        assert!(state.attempts().is_empty());
599        assert!(state.accepted().is_empty());
600        let starts = std::fs::read_to_string(app.path("starts.log")).unwrap();
601        assert_eq!(starts.lines().count(), 2);
602    }
603
604    #[cfg(unix)]
605    #[tokio::test]
606    async fn finite_thousand_call_sequence_launches_once_each_without_truncation() {
607        let app = TestApp::new(
608            r#"#!/bin/sh
609set -eu
610read initialize
611echo '{"id":0,"result":{}}'
612read initialized
613read thread_start
614echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
615read turn_start
616echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
617i=1
618while [ "$i" -le 1000 ]; do
619  echo "{\"id\":$((1000 + i)),\"method\":\"item/tool/call\",\"params\":{\"threadId\":\"thread-1\",\"turnId\":\"turn-1\",\"callId\":\"call-$i\",\"tool\":\"lookup\",\"arguments\":{\"n\":$i}}}"
620  read response
621  i=$((i + 1))
622done
623echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'
624"#,
625        );
626        let state = LauncherState::default();
627        let mut shim = Shim::new(
628            app.adapter().await,
629            "conversation-1",
630            TestCodec::default(),
631            Box::new(TestLauncher::accepting(state.clone())),
632        );
633
634        let output = shim.infer("burst").await.unwrap();
635        assert_eq!(output.items.len(), 1000);
636        for (index, item) in output.items.iter().enumerate() {
637            assert_eq!(item, &ShimItem::Box(TestBox(format!("call-{}", index + 1))));
638        }
639        let expected: Vec<_> = (1..=1000).map(|index| format!("call-{index}")).collect();
640        assert_eq!(shim.codec.conversions, expected);
641        assert_eq!(state.attempts(), expected);
642        assert_eq!(state.accepted(), expected);
643    }
644
645    #[cfg(unix)]
646    #[tokio::test]
647    async fn launcher_rejection_returns_diagnostics_without_success_and_poisons_reuse() {
648        let app = TestApp::new(
649            r#"#!/bin/sh
650set -eu
651read initialize
652echo '{"id":0,"result":{}}'
653read initialized
654printf 'launcher diagnostic\n' >&2
655read thread_start
656echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
657read turn_start
658echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
659echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"partial"}}'
660echo '{"id":77,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"A","tool":"lookup","arguments":{}}}'
661IFS= read -r response
662printf '%s\n' "$response" > rejection-response.log
663IFS= read -r interrupt || true
664"#,
665        );
666        let adapter = app.adapter().await;
667        wait_for_diagnostics(&adapter, b"launcher diagnostic\n").await;
668        let state = LauncherState::default();
669        let mut shim = Shim::new(
670            adapter,
671            "conversation-1",
672            TestCodec::default(),
673            Box::new(TestLauncher::rejecting(
674                state.clone(),
675                "A",
676                "launcher refused A",
677            )),
678        );
679
680        let error = shim.infer("request").await.unwrap_err();
681        assert_eq!(error.kind, ErrorKind::LaunchRejected);
682        assert_eq!(error.message, "launcher refused A");
683        assert!(
684            error
685                .diagnostics
686                .windows(b"launcher diagnostic\n".len())
687                .any(|window| window == b"launcher diagnostic\n")
688        );
689        assert_eq!(shim.codec.conversions, ["A"]);
690        assert_eq!(state.attempts(), ["A"]);
691        assert!(state.accepted().is_empty());
692
693        let reuse = shim.infer("must reject").await.unwrap_err();
694        assert!(
695            reuse.message.contains("cannot be reused"),
696            "unexpected reuse error: {reuse}"
697        );
698
699        wait_for_lines(&app.path("rejection-response.log"), 1).await;
700        let response: Value = serde_json::from_str(
701            &std::fs::read_to_string(app.path("rejection-response.log")).unwrap(),
702        )
703        .unwrap();
704        assert_eq!(response["id"], 77);
705        assert!(response.get("result").is_none());
706        assert!(response.get("error").is_some());
707    }
708
709    #[cfg(unix)]
710    #[tokio::test]
711    async fn response_failure_preserves_accepted_launch_and_poisons_reuse() {
712        let app = TestApp::new(
713            r#"#!/bin/sh
714set -eu
715read initialize
716echo '{"id":0,"result":{}}'
717read initialized
718read thread_start
719echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
720read turn_start
721echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
722exec 0<&-
723echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"partial"}}'
724echo '{"id":77,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"A","tool":"lookup","arguments":{}}}'
725sleep 1
726"#,
727        );
728        let state = LauncherState::default();
729        let mut shim = Shim::new(
730            app.adapter().await,
731            "conversation-1",
732            TestCodec::default(),
733            Box::new(TestLauncher::accepting(state.clone())),
734        );
735
736        let error = shim.infer("request").await.unwrap_err();
737        assert_eq!(error.kind, ErrorKind::Unavailable);
738        assert_eq!(state.attempts(), ["A"]);
739        assert_eq!(state.accepted(), ["A"]);
740
741        let reuse = shim.infer("must reject").await.unwrap_err();
742        assert!(
743            reuse.message.contains("cannot be reused"),
744            "unexpected reuse error: {reuse}"
745        );
746    }
747}
kcode_k1_codex_adapter/shim.rs

kcode_k1_codex_adapter/
shim.rs
