1pub use kcode_k1_codex_conversations::{
2 Config, DynamicTool, Error, ErrorKind, Event, ToolCall, ToolResult,
3};
4use kcode_k1_codex_runtime_actor::{Handle, Started, WeakHandle};
5pub use kcode_k1_codex_runtime_actor::{RuntimeObservation, UsageObservation};
6use std::{fmt, sync::Arc};
7use tokio::sync::mpsc;
8
9#[derive(Clone)]
10pub struct Adapter {
11 handle: Handle,
12 profile: Arc<Config>,
13}
14
15impl Adapter {
16 pub async fn open(config: Config) -> Result<Self, Error> {
17 config.validate()?;
18 let handle = kcode_k1_codex_runtime_actor::open(&config).await?;
19 Ok(Self {
20 handle,
21 profile: Arc::new(config),
22 })
23 }
24
25 pub fn with_config(&self, config: Config) -> Result<Adapter, Error> {
26 if let Err(mut error) = validate_derived(&self.profile, &config) {
27 error.diagnostics = self.handle.diagnostics();
28 return Err(error);
29 }
30 Ok(Adapter {
31 handle: self.handle.clone(),
32 profile: Arc::new(config),
33 })
34 }
35
36 pub async fn start_turn(
37 &self,
38 conversation_key: impl Into<String>,
39 input: impl Into<String>,
40 ) -> Result<Turn, Error> {
41 let key = conversation_key.into();
42 let Started { serial, events } = self
43 .handle
44 .start(key.clone(), input.into(), Arc::clone(&self.profile))
45 .await?;
46 Ok(Turn {
47 key,
48 serial,
49 handle: self.handle.downgrade(),
50 events: EventStream::new(events),
51 diagnostics: self.handle.diagnostics(),
52 })
53 }
54
55 pub async fn start_web_search_turn(
56 &self,
57 key: String,
58 query: String,
59 model: String,
60 reasoning_effort: String,
61 ) -> Result<SearchTurn, Error> {
62 let profile =
63 search_profile(&self.profile, model, reasoning_effort).map_err(|mut error| {
64 error.diagnostics = self.handle.diagnostics();
65 error
66 })?;
67 let Started { serial, events } = self
68 .handle
69 .start_web_search(key.clone(), query, Arc::new(profile))
70 .await?;
71 Ok(SearchTurn {
72 key,
73 serial,
74 handle: self.handle.downgrade(),
75 events: EventStream::new(events),
76 diagnostics: self.handle.diagnostics(),
77 closed: false,
78 })
79 }
80
81 pub async fn steer(
82 &self,
83 conversation_key: impl Into<String>,
84 input: impl Into<String>,
85 ) -> Result<(), Error> {
86 self.handle
87 .steer(
88 conversation_key.into(),
89 input.into(),
90 Arc::clone(&self.profile),
91 )
92 .await
93 }
94
95 pub async fn subscribe_usage(
96 &self,
97 key: String,
98 ) -> Result<mpsc::UnboundedReceiver<RuntimeObservation>, Error> {
99 self.handle.subscribe_usage(key).await
100 }
101
102 pub async fn close_conversation(&self, key: impl Into<String>) -> Result<(), Error> {
103 self.handle
104 .close(key.into(), Arc::clone(&self.profile))
105 .await
106 }
107
108 pub fn diagnostics(&self) -> Vec<u8> {
109 self.handle.diagnostics()
110 }
111}
112
113pub struct Turn {
114 key: String,
115 serial: u64,
116 handle: WeakHandle,
117 events: EventStream,
118 diagnostics: Vec<u8>,
119}
120
121impl fmt::Debug for Turn {
122 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
123 formatter
124 .debug_struct("Turn")
125 .field("key", &self.key)
126 .field("serial", &self.serial)
127 .finish_non_exhaustive()
128 }
129}
130
131impl Turn {
132 pub async fn next_event(&mut self) -> Option<Event> {
133 self.events.next().await
134 }
135
136 pub fn try_next_event(&mut self) -> Result<Option<Event>, Error> {
137 self.events.try_next().map_err(|()| self.unavailable())
138 }
139
140 pub async fn respond(
141 &self,
142 call_id: impl Into<String>,
143 result: ToolResult,
144 ) -> Result<(), Error> {
145 let handle = self.handle.upgrade().ok_or_else(|| self.unavailable())?;
146 handle
147 .respond(self.key.clone(), self.serial, call_id.into(), result)
148 .await
149 }
150
151 fn unavailable(&self) -> Error {
152 unavailable(&self.handle, &self.diagnostics)
153 }
154}
155
156impl Drop for Turn {
157 fn drop(&mut self) {
158 if !self.events.terminal
159 && let Some(handle) = self.handle.upgrade()
160 {
161 handle.abandon(self.key.clone(), self.serial);
162 }
163 }
164}
165
166pub struct SearchTurn {
167 key: String,
168 serial: u64,
169 handle: WeakHandle,
170 events: EventStream,
171 diagnostics: Vec<u8>,
172 closed: bool,
173}
174
175impl fmt::Debug for SearchTurn {
176 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
177 formatter
178 .debug_struct("SearchTurn")
179 .field("key", &self.key)
180 .field("serial", &self.serial)
181 .field("closed", &self.closed)
182 .finish_non_exhaustive()
183 }
184}
185
186impl SearchTurn {
187 pub async fn next_event(&mut self) -> Option<Event> {
188 self.events.next().await
189 }
190
191 pub fn try_next_event(&mut self) -> Result<Option<Event>, Error> {
192 self.events.try_next().map_err(|()| self.unavailable())
193 }
194
195 pub async fn close(mut self) -> Result<(), Error> {
196 let handle = self.handle.upgrade().ok_or_else(|| self.unavailable())?;
197 handle.close_one_shot(self.key.clone(), self.serial).await?;
198 self.closed = true;
199 Ok(())
200 }
201
202 fn unavailable(&self) -> Error {
203 unavailable(&self.handle, &self.diagnostics)
204 }
205}
206
207impl Drop for SearchTurn {
208 fn drop(&mut self) {
209 if !self.closed
210 && let Some(handle) = self.handle.upgrade()
211 {
212 handle.retire_one_shot(self.key.clone(), self.serial);
213 }
214 }
215}
216
217struct EventStream {
218 receiver: mpsc::UnboundedReceiver<Event>,
219 terminal: bool,
220}
221
222impl EventStream {
223 fn new(receiver: mpsc::UnboundedReceiver<Event>) -> Self {
224 Self {
225 receiver,
226 terminal: false,
227 }
228 }
229
230 async fn next(&mut self) -> Option<Event> {
231 let event = self.receiver.recv().await;
232 self.observe(event.as_ref());
233 event
234 }
235
236 fn try_next(&mut self) -> Result<Option<Event>, ()> {
237 match self.receiver.try_recv() {
238 Ok(event) => {
239 self.observe(Some(&event));
240 Ok(Some(event))
241 }
242 Err(mpsc::error::TryRecvError::Empty) => Ok(None),
243 Err(mpsc::error::TryRecvError::Disconnected) => {
244 self.terminal = true;
245 Err(())
246 }
247 }
248 }
249
250 fn observe(&mut self, event: Option<&Event>) {
251 if event.is_none_or(|event| matches!(event, Event::Done | Event::Error(_))) {
252 self.terminal = true;
253 }
254 }
255}
256
257fn search_profile(base: &Config, model: String, reasoning_effort: String) -> Result<Config, Error> {
258 let mut profile = base.clone();
259 profile.model = model;
260 profile.reasoning_effort = Some(reasoning_effort);
261 profile.base_instructions.clear();
262 profile.tools.clear();
263 profile.validate()?;
264 Ok(profile)
265}
266
267fn unavailable(handle: &WeakHandle, fallback: &[u8]) -> Error {
268 let diagnostics = handle
269 .upgrade()
270 .map_or_else(|| fallback.to_vec(), |handle| handle.diagnostics());
271 Error {
272 kind: ErrorKind::Unavailable,
273 message: "Codex app-server is unavailable".into(),
274 diagnostics,
275 }
276}
277
278fn validate_derived(base: &Config, derived: &Config) -> Result<(), Error> {
279 derived.validate()?;
280 if derived.executable != base.executable || derived.working_directory != base.working_directory
281 {
282 return Err(Error::new(
283 ErrorKind::Protocol,
284 "derived configuration must use the same executable and working directory",
285 ));
286 }
287 Ok(())
288}
289
290#[cfg(test)]
291mod tests {
292 use super::*;
293
294 fn profile() -> Config {
295 Config {
296 executable: "codex".into(),
297 working_directory: "/workspace".into(),
298 model: "model-a".into(),
299 reasoning_effort: Some("high".into()),
300 base_instructions: "base instructions".into(),
301 tools: Vec::new(),
302 }
303 }
304
305 #[test]
306 fn derived_profiles_validate_process_identity_and_retain_all_other_fields() {
307 let base = profile();
308 let mut derived = profile();
309 derived.model = "model-b".into();
310 derived.reasoning_effort = None;
311 derived.base_instructions = "other instructions".into();
312 assert!(validate_derived(&base, &derived).is_ok());
313
314 let mut different_executable = derived.clone();
315 different_executable.executable = "other-codex".into();
316 let error = validate_derived(&base, &different_executable).unwrap_err();
317 assert_eq!(error.kind, ErrorKind::Protocol);
318 assert_eq!(
319 error.message,
320 "derived configuration must use the same executable and working directory"
321 );
322
323 let mut different_directory = derived;
324 different_directory.working_directory = "/other".into();
325 assert!(validate_derived(&base, &different_directory).is_err());
326 }
327
328 #[test]
329 fn search_profiles_are_isolated_from_the_adapter_profile() {
330 let mut base = profile();
331 base.tools.push(DynamicTool {
332 name: "tool".into(),
333 description: "description".into(),
334 input_schema: Default::default(),
335 });
336 let search = search_profile(&base, "search-model".into(), "medium".into()).unwrap();
337 assert_eq!(search.executable, base.executable);
338 assert_eq!(search.working_directory, base.working_directory);
339 assert_eq!(search.model, "search-model");
340 assert_eq!(search.reasoning_effort.as_deref(), Some("medium"));
341 assert!(search.base_instructions.is_empty());
342 assert!(search.tools.is_empty());
343 assert_eq!(base.model, "model-a");
344 assert_eq!(base.base_instructions, "base instructions");
345 assert_eq!(base.tools.len(), 1);
346 }
347
348 #[test]
349 fn try_next_event_distinguishes_empty_terminal_and_disconnected_states() {
350 let (sender, receiver) = mpsc::unbounded_channel();
351 let mut events = EventStream::new(receiver);
352 assert!(matches!(events.try_next(), Ok(None)));
353 assert!(!events.terminal);
354
355 sender.send(Event::TextDelta("hello".into())).unwrap();
356 assert!(matches!(
357 events.try_next(),
358 Ok(Some(Event::TextDelta(text))) if text == "hello"
359 ));
360 assert!(!events.terminal);
361
362 sender.send(Event::Done).unwrap();
363 drop(sender);
364 assert!(matches!(events.try_next(), Ok(Some(Event::Done))));
365 assert!(events.terminal);
366 assert!(events.try_next().is_err());
367 }
368}