pub use kcode_k1_codex_conversations::{
Config, DynamicTool, Error, ErrorKind, Event, ToolCall, ToolResult,
};
use kcode_k1_codex_runtime_actor::{Handle, Started, WeakHandle};
pub use kcode_k1_codex_runtime_actor::{RuntimeObservation, UsageObservation};
use std::{fmt, sync::Arc};
use tokio::sync::mpsc;
#[derive(Clone)]
pub struct Adapter {
handle: Handle,
profile: Arc<Config>,
}
impl Adapter {
pub async fn open(config: Config) -> Result<Self, Error> {
config.validate()?;
let handle = kcode_k1_codex_runtime_actor::open(&config).await?;
Ok(Self {
handle,
profile: Arc::new(config),
})
}
pub fn with_config(&self, config: Config) -> Result<Adapter, Error> {
if let Err(mut error) = validate_derived(&self.profile, &config) {
error.diagnostics = self.handle.diagnostics();
return Err(error);
}
Ok(Adapter {
handle: self.handle.clone(),
profile: Arc::new(config),
})
}
pub async fn start_turn(
&self,
conversation_key: impl Into<String>,
input: impl Into<String>,
) -> Result<Turn, Error> {
let key = conversation_key.into();
let Started { serial, events } = self
.handle
.start(key.clone(), input.into(), Arc::clone(&self.profile))
.await?;
Ok(Turn {
key,
serial,
handle: self.handle.downgrade(),
events: EventStream::new(events),
diagnostics: self.handle.diagnostics(),
})
}
pub async fn steer(
&self,
conversation_key: impl Into<String>,
input: impl Into<String>,
) -> Result<(), Error> {
self.handle
.steer(
conversation_key.into(),
input.into(),
Arc::clone(&self.profile),
)
.await
}
pub async fn subscribe_usage(
&self,
key: String,
) -> Result<mpsc::UnboundedReceiver<RuntimeObservation>, Error> {
self.handle.subscribe_usage(key).await
}
pub async fn close_conversation(&self, key: impl Into<String>) -> Result<(), Error> {
self.handle
.close(key.into(), Arc::clone(&self.profile))
.await
}
pub fn diagnostics(&self) -> Vec<u8> {
self.handle.diagnostics()
}
}
pub struct Turn {
key: String,
serial: u64,
handle: WeakHandle,
events: EventStream,
diagnostics: Vec<u8>,
}
impl fmt::Debug for Turn {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("Turn")
.field("key", &self.key)
.field("serial", &self.serial)
.finish_non_exhaustive()
}
}
impl Turn {
pub async fn next_event(&mut self) -> Option<Event> {
self.events.next().await
}
pub fn try_next_event(&mut self) -> Result<Option<Event>, Error> {
self.events.try_next().map_err(|()| self.unavailable())
}
pub async fn respond(
&self,
call_id: impl Into<String>,
result: ToolResult,
) -> Result<(), Error> {
let handle = self.handle.upgrade().ok_or_else(|| self.unavailable())?;
handle
.respond(self.key.clone(), self.serial, call_id.into(), result)
.await
}
fn unavailable(&self) -> Error {
let diagnostics = self
.handle
.upgrade()
.map_or_else(|| self.diagnostics.clone(), |handle| handle.diagnostics());
Error {
kind: ErrorKind::Unavailable,
message: "Codex app-server is unavailable".into(),
diagnostics,
}
}
}
impl Drop for Turn {
fn drop(&mut self) {
if !self.events.terminal
&& let Some(handle) = self.handle.upgrade()
{
handle.abandon(self.key.clone(), self.serial);
}
}
}
struct EventStream {
receiver: mpsc::UnboundedReceiver<Event>,
terminal: bool,
}
impl EventStream {
fn new(receiver: mpsc::UnboundedReceiver<Event>) -> Self {
Self {
receiver,
terminal: false,
}
}
async fn next(&mut self) -> Option<Event> {
let event = self.receiver.recv().await;
self.observe(event.as_ref());
event
}
fn try_next(&mut self) -> Result<Option<Event>, ()> {
match self.receiver.try_recv() {
Ok(event) => {
self.observe(Some(&event));
Ok(Some(event))
}
Err(mpsc::error::TryRecvError::Empty) => Ok(None),
Err(mpsc::error::TryRecvError::Disconnected) => {
self.terminal = true;
Err(())
}
}
}
fn observe(&mut self, event: Option<&Event>) {
if event.is_none_or(|event| matches!(event, Event::Done | Event::Error(_))) {
self.terminal = true;
}
}
}
fn validate_derived(base: &Config, derived: &Config) -> Result<(), Error> {
derived.validate()?;
if derived.executable != base.executable || derived.working_directory != base.working_directory
{
return Err(Error::new(
ErrorKind::Protocol,
"derived configuration must use the same executable and working directory",
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn profile() -> Config {
Config {
executable: "codex".into(),
working_directory: "/workspace".into(),
model: "model-a".into(),
reasoning_effort: Some("high".into()),
base_instructions: "base instructions".into(),
tools: Vec::new(),
}
}
#[test]
fn derived_profiles_validate_process_identity_and_retain_all_other_fields() {
let base = profile();
let mut derived = profile();
derived.model = "model-b".into();
derived.reasoning_effort = None;
derived.base_instructions = "other instructions".into();
assert!(validate_derived(&base, &derived).is_ok());
let mut different_executable = derived.clone();
different_executable.executable = "other-codex".into();
let error = validate_derived(&base, &different_executable).unwrap_err();
assert_eq!(error.kind, ErrorKind::Protocol);
assert_eq!(
error.message,
"derived configuration must use the same executable and working directory"
);
let mut different_directory = derived;
different_directory.working_directory = "/other".into();
assert!(validate_derived(&base, &different_directory).is_err());
}
#[test]
fn try_next_event_distinguishes_empty_terminal_and_disconnected_states() {
let (sender, receiver) = mpsc::unbounded_channel();
let mut events = EventStream::new(receiver);
assert!(matches!(events.try_next(), Ok(None)));
assert!(!events.terminal);
sender.send(Event::TextDelta("hello".into())).unwrap();
assert!(matches!(
events.try_next(),
Ok(Some(Event::TextDelta(text))) if text == "hello"
));
assert!(!events.terminal);
sender.send(Event::Done).unwrap();
drop(sender);
assert!(matches!(events.try_next(), Ok(Some(Event::Done))));
assert!(events.terminal);
assert!(events.try_next().is_err());
}
}