kcode-k1-codex-adapter 0.2.0

use std::{collections::VecDeque, future::Future, pin::Pin};

use crate::{Adapter, Error, ErrorKind, Event, ToolCall, ToolResult};

/// Fixed successful response used to release every Codex dynamic-tool request.
pub 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 the next user turn.";

/// Future returned while handing one canonical tool-call box to K1.
pub type ToolLaunchFuture<'a> = Pin<Box<dyn Future<Output = Result<(), String>> + Send + 'a>>;

/// Accepts canonical tool-call boxes for asynchronous execution.
pub trait ToolCallLauncher<B>: Send {
    /// Accepts, starts, or queues one tool call without waiting for completion.
    fn launch<'a>(&'a mut self, box_: &'a B) -> ToolLaunchFuture<'a>;
}

/// Converts between a chat engine's canonical box type and text visible to Codex.
///
/// A shim calls `tool_call_box` exactly once for each dynamic tool call it
/// receives. `box_text` must return the complete representation that should be
/// placed into Codex context. Implementations must not add ordering of their
/// own: the shim preserves insertion order.
pub trait BoxCodec {
    /// Canonical box type owned by the chat engine.
    type Box: Clone;

    /// Converts one typed Codex dynamic-tool request into a canonical box.
    fn tool_call_box(&mut self, call: &ToolCall) -> Self::Box;

    /// Returns the exact textual representation of one box for Codex.
    fn box_text<'a>(&self, box_: &'a Self::Box) -> &'a str;
}

/// One ordered item produced by a completed native Codex turn.
#[derive(Clone, Debug, PartialEq)]
pub enum ShimItem<B> {
    /// Adjacent streamed assistant-text deltas, coalesced.
    Text(String),
    /// A canonical dynamic-tool-call box.
    Box(B),
}

/// Atomic terminal output from one native Codex turn.
#[derive(Clone, Debug, PartialEq)]
pub struct ShimOutput<B> {
    /// Assistant text and call boxes in exact provider event order.
    pub items: Vec<ShimItem<B>>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Health {
    Ready,
    Unusable,
}

/// A single-owner, per-conversation bridge between K1 boxes and Codex turns.
///
/// `Shim` deliberately contains no locking. A caller creates one shim for one
/// conversation and invokes it sequentially. Different shims may share clones
/// of the same [`Adapter`], whose actor owns transport concurrency.
pub struct Shim<C: BoxCodec> {
    adapter: Adapter,
    conversation_key: String,
    codec: C,
    launcher: Box<dyn ToolCallLauncher<C::Box>>,
    health: Health,
    pending_boxes: VecDeque<C::Box>,
}

impl<C: BoxCodec> Shim<C> {
    /// Creates a ready shim for one conversation.
    pub fn new(
        adapter: Adapter,
        conversation_key: impl Into<String>,
        codec: C,
        launcher: Box<dyn ToolCallLauncher<C::Box>>,
    ) -> Self {
        Self {
            adapter,
            conversation_key: conversation_key.into(),
            codec,
            launcher,
            health: Health::Ready,
            pending_boxes: VecDeque::new(),
        }
    }

    /// Appends one externally produced box in canonical chat-engine order.
    ///
    /// Call boxes returned by [`Shim::infer`] are output only and are not added
    /// to this pending queue automatically.
    pub fn record_box(&mut self, box_: C::Box) {
        self.pending_boxes.push_back(box_);
    }

    /// Appends externally produced boxes without changing their order.
    pub fn record_boxes(&mut self, boxes: impl IntoIterator<Item = C::Box>) {
        self.pending_boxes.extend(boxes);
    }

    /// Number of boxes waiting to be submitted to a fresh Codex turn.
    pub fn pending_box_count(&self) -> usize {
        self.pending_boxes.len()
    }

    /// Runs exactly one fresh native Codex turn through terminal completion.
    ///
    /// Pending K1 boxes prefix ordinary input and are removed only after the
    /// adapter accepts the turn. Every dynamic call is converted once, launched,
    /// acknowledged with [`ASYNC_TOOL_ACKNOWLEDGEMENT`], and retained as an
    /// ordered output box. No partial output is returned if the active turn
    /// fails.
    pub async fn infer(&mut self, input: impl Into<String>) -> Result<ShimOutput<C::Box>, Error> {
        if self.health == Health::Unusable {
            return Err(self.unusable());
        }

        let submitted_box_count = self.pending_boxes.len();
        let input = append_section(self.render_pending_boxes(), &input.into());

        // Cancellation anywhere after this point fails closed. A definite
        // start rejection restores readiness because no native turn exists.
        self.health = Health::Unusable;
        let mut turn = match self
            .adapter
            .start_turn(self.conversation_key.clone(), input)
            .await
        {
            Ok(turn) => turn,
            Err(error) => {
                self.health = Health::Ready;
                return Err(error);
            }
        };

        for _ in 0..submitted_box_count {
            let removed = self.pending_boxes.pop_front();
            debug_assert!(removed.is_some());
        }

        let mut items = Vec::new();
        loop {
            match turn.next_event().await {
                Some(Event::TextDelta(delta)) => push_text(&mut items, delta),
                Some(Event::ToolCall(call)) => {
                    let box_ = self.codec.tool_call_box(&call);
                    if let Err(message) = self.launcher.launch(&box_).await {
                        return Err(Error {
                            kind: ErrorKind::LaunchRejected,
                            message,
                            diagnostics: self.adapter.diagnostics(),
                        });
                    }
                    turn.respond(
                        call.call_id,
                        ToolResult {
                            success: true,
                            output: ASYNC_TOOL_ACKNOWLEDGEMENT.to_owned(),
                        },
                    )
                    .await?;
                    items.push(ShimItem::Box(box_));
                }
                Some(Event::Done) => {
                    self.health = Health::Ready;
                    return Ok(ShimOutput { items });
                }
                Some(Event::Error(error)) => return Err(error),
                None => return Err(self.adapter.unavailable()),
            }
        }
    }

    fn render_pending_boxes(&self) -> String {
        let mut output = String::new();
        for box_ in &self.pending_boxes {
            output = append_section(output, self.codec.box_text(box_));
        }
        output
    }

    fn unusable(&self) -> Error {
        let mut error = self.adapter.unavailable();
        error.message =
            "Codex shim cannot be reused after an active turn failed or was cancelled".to_owned();
        error
    }
}

fn push_text<B>(items: &mut Vec<ShimItem<B>>, delta: String) {
    match items.last_mut() {
        Some(ShimItem::Text(text)) => text.push_str(&delta),
        _ => items.push(ShimItem::Text(delta)),
    }
}

fn append_section(mut output: String, section: &str) -> String {
    if section.is_empty() {
        return output;
    }
    if !output.is_empty() && !output.ends_with('\n') {
        output.push('\n');
    }
    output.push_str(section);
    output
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Config, DynamicTool};
    use serde_json::{Value, json};
    use std::sync::{Arc, Mutex};

    #[derive(Clone, Debug, PartialEq, Eq)]
    struct TestBox(String);

    #[derive(Default)]
    struct TestCodec {
        conversions: Vec<String>,
    }

    impl BoxCodec for TestCodec {
        type Box = TestBox;

        fn tool_call_box(&mut self, call: &ToolCall) -> Self::Box {
            self.conversions.push(call.call_id.clone());
            TestBox(call.call_id.clone())
        }

        fn box_text<'a>(&self, box_: &'a Self::Box) -> &'a str {
            &box_.0
        }
    }

    #[derive(Clone, Default)]
    struct LauncherState {
        attempts: Arc<Mutex<Vec<String>>>,
        accepted: Arc<Mutex<Vec<String>>>,
    }

    impl LauncherState {
        fn attempts(&self) -> Vec<String> {
            self.attempts
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .clone()
        }

        fn accepted(&self) -> Vec<String> {
            self.accepted
                .lock()
                .unwrap_or_else(|poisoned| poisoned.into_inner())
                .clone()
        }
    }

    struct TestLauncher {
        state: LauncherState,
        rejection: Option<(String, String)>,
        sequence_path: Option<std::path::PathBuf>,
    }

    impl TestLauncher {
        fn accepting(state: LauncherState) -> Self {
            Self {
                state,
                rejection: None,
                sequence_path: None,
            }
        }

        fn rejecting(state: LauncherState, call_id: &str, message: &str) -> Self {
            Self {
                state,
                rejection: Some((call_id.to_owned(), message.to_owned())),
                sequence_path: None,
            }
        }

        fn with_sequence_path(mut self, path: std::path::PathBuf) -> Self {
            self.sequence_path = Some(path);
            self
        }
    }

    impl ToolCallLauncher<TestBox> for TestLauncher {
        fn launch<'a>(&'a mut self, box_: &'a TestBox) -> ToolLaunchFuture<'a> {
            let call_id = box_.0.clone();
            let state = self.state.clone();
            let rejection = self
                .rejection
                .as_ref()
                .filter(|(rejected, _)| rejected == &call_id)
                .map(|(_, message)| message.clone());
            let sequence_path = self.sequence_path.clone();

            Box::pin(async move {
                state
                    .attempts
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .push(call_id.clone());
                if let Some(message) = rejection {
                    return Err(message);
                }
                state
                    .accepted
                    .lock()
                    .unwrap_or_else(|poisoned| poisoned.into_inner())
                    .push(call_id.clone());
                if let Some(path) = sequence_path {
                    use std::io::Write;

                    let mut file = std::fs::OpenOptions::new()
                        .create(true)
                        .append(true)
                        .open(path)
                        .unwrap();
                    writeln!(file, "launch-{call_id}").unwrap();
                }
                Ok(())
            })
        }
    }

    #[cfg(unix)]
    struct TestApp {
        directory: std::path::PathBuf,
        executable: std::path::PathBuf,
    }

    #[cfg(unix)]
    impl TestApp {
        fn new(script: &str) -> Self {
            use std::{
                os::unix::fs::PermissionsExt,
                sync::atomic::{AtomicU64, Ordering},
                time::{SystemTime, UNIX_EPOCH},
            };

            static NEXT: AtomicU64 = AtomicU64::new(0);
            let nonce = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos();
            let directory = std::env::temp_dir().join(format!(
                "kcode-k1-codex-adapter-{}-{nonce}-{}",
                std::process::id(),
                NEXT.fetch_add(1, Ordering::Relaxed)
            ));
            std::fs::create_dir(&directory).unwrap();
            let executable = directory.join("codex");
            std::fs::write(&executable, script).unwrap();
            let mut permissions = std::fs::metadata(&executable).unwrap().permissions();
            permissions.set_mode(0o700);
            std::fs::set_permissions(&executable, permissions).unwrap();
            Self {
                directory,
                executable,
            }
        }

        fn path(&self, name: &str) -> std::path::PathBuf {
            self.directory.join(name)
        }

        async fn adapter(&self) -> Adapter {
            Adapter::open(Config {
                executable: self.executable.clone(),
                working_directory: self.directory.to_string_lossy().into_owned(),
                model: "test-model".into(),
                reasoning_effort: None,
                base_instructions: String::new(),
                tools: vec![DynamicTool {
                    name: "lookup".into(),
                    description: "Lookup a value".into(),
                    input_schema: json!({"type":"object"}),
                }],
            })
            .await
            .unwrap()
        }
    }

    #[cfg(unix)]
    impl Drop for TestApp {
        fn drop(&mut self) {
            let _ = std::fs::remove_dir_all(&self.directory);
        }
    }

    #[cfg(unix)]
    async fn wait_for_lines(path: &std::path::Path, minimum: usize) {
        tokio::time::timeout(std::time::Duration::from_secs(3), async {
            loop {
                let count = std::fs::read_to_string(path)
                    .map(|text| text.lines().count())
                    .unwrap_or(0);
                if count >= minimum {
                    return;
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("timed out waiting for app-server log");
    }

    #[cfg(unix)]
    async fn wait_for_diagnostics(adapter: &Adapter, expected: &[u8]) {
        tokio::time::timeout(std::time::Duration::from_secs(3), async {
            loop {
                let diagnostics = adapter.diagnostics();
                if diagnostics
                    .windows(expected.len())
                    .any(|window| window == expected)
                {
                    return;
                }
                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            }
        })
        .await
        .expect("timed out waiting for app-server diagnostics");
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn launches_ordered_call_waves_before_exact_acknowledgements_and_waits_for_done() {
        let app = TestApp::new(
            r#"#!/bin/sh
set -eu
IFS= read -r initialize
echo '{"id":0,"result":{}}'
IFS= read -r initialized
IFS= read -r thread_start
echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
IFS= read -r turn_start
printf '%s\n' "$turn_start" > turn-start.log
echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"pre"}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"face"}}'
echo '{"id":77,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"A","tool":"lookup","arguments":{"n":1}}}'
IFS= read -r response
printf '%s\n' "$response" >> responses.log
printf 'ack-A\n' >> sequence.log
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"mid"}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"dle"}}'
echo '{"id":78,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"B","tool":"lookup","arguments":{"n":2}}}'
echo '{"id":79,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"C","tool":"lookup","arguments":{"n":3}}}'
IFS= read -r response
printf '%s\n' "$response" >> responses.log
printf 'ack-B\n' >> sequence.log
IFS= read -r response
printf '%s\n' "$response" >> responses.log
printf 'ack-C\n' >> sequence.log
while [ ! -e release ]; do sleep 0.01; done
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"tail"}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"end"}}'
echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'
"#,
        );
        let state = LauncherState::default();
        let launcher =
            TestLauncher::accepting(state.clone()).with_sequence_path(app.path("sequence.log"));
        let mut shim = Shim::new(
            app.adapter().await,
            "conversation-1",
            TestCodec::default(),
            Box::new(launcher),
        );
        shim.record_boxes([TestBox("<pending-1>".into()), TestBox("<pending-2>".into())]);

        let task = tokio::spawn(async move {
            let result = shim.infer("request").await;
            (shim, result)
        });
        wait_for_lines(&app.path("sequence.log"), 6).await;
        assert!(!task.is_finished(), "infer returned before Event::Done");
        std::fs::write(app.path("release"), "").unwrap();
        let (shim, output) = tokio::time::timeout(std::time::Duration::from_secs(3), task)
            .await
            .unwrap()
            .unwrap();
        let output = output.unwrap();

        assert_eq!(
            output.items,
            vec![
                ShimItem::Text("preface".into()),
                ShimItem::Box(TestBox("A".into())),
                ShimItem::Text("middle".into()),
                ShimItem::Box(TestBox("B".into())),
                ShimItem::Box(TestBox("C".into())),
                ShimItem::Text("tailend".into()),
            ]
        );
        assert_eq!(shim.codec.conversions, ["A", "B", "C"]);
        assert_eq!(state.attempts(), ["A", "B", "C"]);
        assert_eq!(state.accepted(), ["A", "B", "C"]);
        assert_eq!(shim.pending_box_count(), 0);

        let start: Value =
            serde_json::from_str(&std::fs::read_to_string(app.path("turn-start.log")).unwrap())
                .unwrap();
        assert_eq!(
            start
                .pointer("/params/input/0/text")
                .and_then(Value::as_str),
            Some("<pending-1>\n<pending-2>\nrequest")
        );

        let responses = std::fs::read_to_string(app.path("responses.log")).unwrap();
        for (line, id) in responses.lines().zip([77, 78, 79]) {
            let response: Value = serde_json::from_str(line).unwrap();
            assert_eq!(response["id"], id);
            assert_eq!(response["result"]["success"], true);
            assert_eq!(
                response
                    .pointer("/result/contentItems/0/text")
                    .and_then(Value::as_str),
                Some(ASYNC_TOOL_ACKNOWLEDGEMENT)
            );
        }

        let sequence: Vec<_> = std::fs::read_to_string(app.path("sequence.log"))
            .unwrap()
            .lines()
            .map(str::to_owned)
            .collect();
        assert_eq!(sequence.len(), 6);
        for entry in [
            "launch-A", "launch-B", "launch-C", "ack-A", "ack-B", "ack-C",
        ] {
            assert_eq!(
                sequence
                    .iter()
                    .filter(|observed| observed.as_str() == entry)
                    .count(),
                1,
                "unexpected sequence: {sequence:?}"
            );
        }
        let position = |entry: &str| {
            sequence
                .iter()
                .position(|observed| observed == entry)
                .unwrap()
        };
        assert!(position("launch-A") < position("launch-B"));
        assert!(position("launch-B") < position("launch-C"));
        assert!(position("launch-A") < position("ack-A"));
        assert!(position("launch-B") < position("ack-B"));
        assert!(position("launch-C") < position("ack-C"));
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn no_tool_inference_leaves_launcher_untouched_and_uses_fresh_turns() {
        let app = TestApp::new(
            r#"#!/bin/sh
set -eu
read initialize
echo '{"id":0,"result":{}}'
read initialized
read thread_start
echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
read first_start
printf '%s\n' "$first_start" >> starts.log
echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"hel"}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"lo"}}'
echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'
read second_start
printf '%s\n' "$second_start" >> starts.log
echo '{"id":3,"result":{"turn":{"id":"turn-2"}}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-2","delta":"again"}}'
echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-2","status":"completed"}}}'
"#,
        );
        let state = LauncherState::default();
        let mut shim = Shim::new(
            app.adapter().await,
            "conversation-1",
            TestCodec::default(),
            Box::new(TestLauncher::accepting(state.clone())),
        );

        assert_eq!(
            shim.infer("first").await.unwrap().items,
            [ShimItem::Text("hello".into())]
        );
        assert_eq!(
            shim.infer("second").await.unwrap().items,
            [ShimItem::Text("again".into())]
        );
        assert!(state.attempts().is_empty());
        assert!(state.accepted().is_empty());
        let starts = std::fs::read_to_string(app.path("starts.log")).unwrap();
        assert_eq!(starts.lines().count(), 2);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn finite_thousand_call_sequence_launches_once_each_without_truncation() {
        let app = TestApp::new(
            r#"#!/bin/sh
set -eu
read initialize
echo '{"id":0,"result":{}}'
read initialized
read thread_start
echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
read turn_start
echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
i=1
while [ "$i" -le 1000 ]; do
  echo "{\"id\":$((1000 + i)),\"method\":\"item/tool/call\",\"params\":{\"threadId\":\"thread-1\",\"turnId\":\"turn-1\",\"callId\":\"call-$i\",\"tool\":\"lookup\",\"arguments\":{\"n\":$i}}}"
  read response
  i=$((i + 1))
done
echo '{"method":"turn/completed","params":{"threadId":"thread-1","turn":{"id":"turn-1","status":"completed"}}}'
"#,
        );
        let state = LauncherState::default();
        let mut shim = Shim::new(
            app.adapter().await,
            "conversation-1",
            TestCodec::default(),
            Box::new(TestLauncher::accepting(state.clone())),
        );

        let output = shim.infer("burst").await.unwrap();
        assert_eq!(output.items.len(), 1000);
        for (index, item) in output.items.iter().enumerate() {
            assert_eq!(item, &ShimItem::Box(TestBox(format!("call-{}", index + 1))));
        }
        let expected: Vec<_> = (1..=1000).map(|index| format!("call-{index}")).collect();
        assert_eq!(shim.codec.conversions, expected);
        assert_eq!(state.attempts(), expected);
        assert_eq!(state.accepted(), expected);
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn launcher_rejection_returns_diagnostics_without_success_and_poisons_reuse() {
        let app = TestApp::new(
            r#"#!/bin/sh
set -eu
read initialize
echo '{"id":0,"result":{}}'
read initialized
printf 'launcher diagnostic\n' >&2
read thread_start
echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
read turn_start
echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"partial"}}'
echo '{"id":77,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"A","tool":"lookup","arguments":{}}}'
IFS= read -r response
printf '%s\n' "$response" > rejection-response.log
IFS= read -r interrupt || true
"#,
        );
        let adapter = app.adapter().await;
        wait_for_diagnostics(&adapter, b"launcher diagnostic\n").await;
        let state = LauncherState::default();
        let mut shim = Shim::new(
            adapter,
            "conversation-1",
            TestCodec::default(),
            Box::new(TestLauncher::rejecting(
                state.clone(),
                "A",
                "launcher refused A",
            )),
        );

        let error = shim.infer("request").await.unwrap_err();
        assert_eq!(error.kind, ErrorKind::LaunchRejected);
        assert_eq!(error.message, "launcher refused A");
        assert!(
            error
                .diagnostics
                .windows(b"launcher diagnostic\n".len())
                .any(|window| window == b"launcher diagnostic\n")
        );
        assert_eq!(shim.codec.conversions, ["A"]);
        assert_eq!(state.attempts(), ["A"]);
        assert!(state.accepted().is_empty());

        let reuse = shim.infer("must reject").await.unwrap_err();
        assert!(
            reuse.message.contains("cannot be reused"),
            "unexpected reuse error: {reuse}"
        );

        wait_for_lines(&app.path("rejection-response.log"), 1).await;
        let response: Value = serde_json::from_str(
            &std::fs::read_to_string(app.path("rejection-response.log")).unwrap(),
        )
        .unwrap();
        assert_eq!(response["id"], 77);
        assert!(response.get("result").is_none());
        assert!(response.get("error").is_some());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn response_failure_preserves_accepted_launch_and_poisons_reuse() {
        let app = TestApp::new(
            r#"#!/bin/sh
set -eu
read initialize
echo '{"id":0,"result":{}}'
read initialized
read thread_start
echo '{"id":1,"result":{"thread":{"id":"thread-1"}}}'
read turn_start
echo '{"id":2,"result":{"turn":{"id":"turn-1"}}}'
exec 0<&-
echo '{"method":"item/agentMessage/delta","params":{"threadId":"thread-1","turnId":"turn-1","delta":"partial"}}'
echo '{"id":77,"method":"item/tool/call","params":{"threadId":"thread-1","turnId":"turn-1","callId":"A","tool":"lookup","arguments":{}}}'
sleep 1
"#,
        );
        let state = LauncherState::default();
        let mut shim = Shim::new(
            app.adapter().await,
            "conversation-1",
            TestCodec::default(),
            Box::new(TestLauncher::accepting(state.clone())),
        );

        let error = shim.infer("request").await.unwrap_err();
        assert_eq!(error.kind, ErrorKind::Unavailable);
        assert_eq!(state.attempts(), ["A"]);
        assert_eq!(state.accepted(), ["A"]);

        let reuse = shim.infer("must reject").await.unwrap_err();
        assert!(
            reuse.message.contains("cannot be reused"),
            "unexpected reuse error: {reuse}"
        );
    }
}