1pub use kcode_k1_codex_conversations::{
2 Config, DynamicTool, Error, ErrorKind, Event, ToolCall, ToolResult,
3};
4use kcode_k1_codex_runtime_actor::{Handle, Started, WeakHandle};
5use std::{fmt, sync::Arc};
6use tokio::sync::mpsc;
7
8#[derive(Clone)]
9pub struct Adapter {
10 handle: Handle,
11 profile: Arc<Config>,
12}
13
14impl Adapter {
15 pub async fn open(config: Config) -> Result<Self, Error> {
16 config.validate()?;
17 let handle = kcode_k1_codex_runtime_actor::open(&config).await?;
18 Ok(Self {
19 handle,
20 profile: Arc::new(config),
21 })
22 }
23
24 pub fn with_config(&self, config: Config) -> Result<Adapter, Error> {
25 if let Err(mut error) = validate_derived(&self.profile, &config) {
26 error.diagnostics = self.handle.diagnostics();
27 return Err(error);
28 }
29 Ok(Adapter {
30 handle: self.handle.clone(),
31 profile: Arc::new(config),
32 })
33 }
34
35 pub async fn start_turn(
36 &self,
37 conversation_key: impl Into<String>,
38 input: impl Into<String>,
39 ) -> Result<Turn, Error> {
40 let key = conversation_key.into();
41 let Started { serial, events } = self
42 .handle
43 .start(key.clone(), input.into(), Arc::clone(&self.profile))
44 .await?;
45 Ok(Turn {
46 key,
47 serial,
48 handle: self.handle.downgrade(),
49 events: EventStream::new(events),
50 diagnostics: self.handle.diagnostics(),
51 })
52 }
53
54 pub async fn steer(
55 &self,
56 conversation_key: impl Into<String>,
57 input: impl Into<String>,
58 ) -> Result<(), Error> {
59 self.handle
60 .steer(
61 conversation_key.into(),
62 input.into(),
63 Arc::clone(&self.profile),
64 )
65 .await
66 }
67
68 pub async fn close_conversation(&self, key: impl Into<String>) -> Result<(), Error> {
69 self.handle
70 .close(key.into(), Arc::clone(&self.profile))
71 .await
72 }
73
74 pub fn diagnostics(&self) -> Vec<u8> {
75 self.handle.diagnostics()
76 }
77}
78
79pub struct Turn {
80 key: String,
81 serial: u64,
82 handle: WeakHandle,
83 events: EventStream,
84 diagnostics: Vec<u8>,
85}
86
87impl fmt::Debug for Turn {
88 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
89 formatter
90 .debug_struct("Turn")
91 .field("key", &self.key)
92 .field("serial", &self.serial)
93 .finish_non_exhaustive()
94 }
95}
96
97impl Turn {
98 pub async fn next_event(&mut self) -> Option<Event> {
99 self.events.next().await
100 }
101
102 pub fn try_next_event(&mut self) -> Result<Option<Event>, Error> {
103 self.events.try_next().map_err(|()| self.unavailable())
104 }
105
106 pub async fn respond(
107 &self,
108 call_id: impl Into<String>,
109 result: ToolResult,
110 ) -> Result<(), Error> {
111 let handle = self.handle.upgrade().ok_or_else(|| self.unavailable())?;
112 handle
113 .respond(self.key.clone(), self.serial, call_id.into(), result)
114 .await
115 }
116
117 fn unavailable(&self) -> Error {
118 let diagnostics = self
119 .handle
120 .upgrade()
121 .map_or_else(|| self.diagnostics.clone(), |handle| handle.diagnostics());
122 Error {
123 kind: ErrorKind::Unavailable,
124 message: "Codex app-server is unavailable".into(),
125 diagnostics,
126 }
127 }
128}
129
130impl Drop for Turn {
131 fn drop(&mut self) {
132 if !self.events.terminal
133 && let Some(handle) = self.handle.upgrade()
134 {
135 handle.abandon(self.key.clone(), self.serial);
136 }
137 }
138}
139
140struct EventStream {
141 receiver: mpsc::UnboundedReceiver<Event>,
142 terminal: bool,
143}
144
145impl EventStream {
146 fn new(receiver: mpsc::UnboundedReceiver<Event>) -> Self {
147 Self {
148 receiver,
149 terminal: false,
150 }
151 }
152
153 async fn next(&mut self) -> Option<Event> {
154 let event = self.receiver.recv().await;
155 self.observe(event.as_ref());
156 event
157 }
158
159 fn try_next(&mut self) -> Result<Option<Event>, ()> {
160 match self.receiver.try_recv() {
161 Ok(event) => {
162 self.observe(Some(&event));
163 Ok(Some(event))
164 }
165 Err(mpsc::error::TryRecvError::Empty) => Ok(None),
166 Err(mpsc::error::TryRecvError::Disconnected) => {
167 self.terminal = true;
168 Err(())
169 }
170 }
171 }
172
173 fn observe(&mut self, event: Option<&Event>) {
174 if event.is_none_or(|event| matches!(event, Event::Done | Event::Error(_))) {
175 self.terminal = true;
176 }
177 }
178}
179
180fn validate_derived(base: &Config, derived: &Config) -> Result<(), Error> {
181 derived.validate()?;
182 if derived.executable != base.executable || derived.working_directory != base.working_directory
183 {
184 return Err(Error::new(
185 ErrorKind::Protocol,
186 "derived configuration must use the same executable and working directory",
187 ));
188 }
189 Ok(())
190}
191
192#[cfg(test)]
193mod tests {
194 use super::*;
195
196 fn profile() -> Config {
197 Config {
198 executable: "codex".into(),
199 working_directory: "/workspace".into(),
200 model: "model-a".into(),
201 reasoning_effort: Some("high".into()),
202 base_instructions: "base instructions".into(),
203 tools: Vec::new(),
204 }
205 }
206
207 #[test]
208 fn derived_profiles_validate_process_identity_and_retain_all_other_fields() {
209 let base = profile();
210 let mut derived = profile();
211 derived.model = "model-b".into();
212 derived.reasoning_effort = None;
213 derived.base_instructions = "other instructions".into();
214 assert!(validate_derived(&base, &derived).is_ok());
215
216 let mut different_executable = derived.clone();
217 different_executable.executable = "other-codex".into();
218 let error = validate_derived(&base, &different_executable).unwrap_err();
219 assert_eq!(error.kind, ErrorKind::Protocol);
220 assert_eq!(
221 error.message,
222 "derived configuration must use the same executable and working directory"
223 );
224
225 let mut different_directory = derived;
226 different_directory.working_directory = "/other".into();
227 assert!(validate_derived(&base, &different_directory).is_err());
228 }
229
230 #[test]
231 fn try_next_event_distinguishes_empty_terminal_and_disconnected_states() {
232 let (sender, receiver) = mpsc::unbounded_channel();
233 let mut events = EventStream::new(receiver);
234 assert!(matches!(events.try_next(), Ok(None)));
235 assert!(!events.terminal);
236
237 sender.send(Event::TextDelta("hello".into())).unwrap();
238 assert!(matches!(
239 events.try_next(),
240 Ok(Some(Event::TextDelta(text))) if text == "hello"
241 ));
242 assert!(!events.terminal);
243
244 sender.send(Event::Done).unwrap();
245 drop(sender);
246 assert!(matches!(events.try_next(), Ok(Some(Event::Done))));
247 assert!(events.terminal);
248 assert!(events.try_next().is_err());
249 }
250}