use crate::*;
use serde_json::{Value, json};
use std::{
collections::{HashMap, HashSet},
process::Stdio,
time::Duration,
};
use tokio::{
io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader, Lines},
process::{Child, ChildStdin, ChildStdout, Command as ProcessCommand},
sync::mpsc,
task::JoinHandle,
time::timeout,
};
const INITIALIZE_TIMEOUT: Duration = Duration::from_secs(30);
const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(2);
const EARLY_EVENT_LIMIT: usize = 1024;
type ToolToken = (String, u64, String);
struct Active {
serial: u64,
turn: Option<String>,
events: Option<Events>,
early: Vec<Value>,
cancelled: bool,
interrupted: bool,
failure: Option<Value>,
}
#[derive(Default)]
struct Conversation {
thread: Option<String>,
active: Option<Active>,
}
enum Pending {
Thread {
key: String,
serial: u64,
input: String,
},
Turn {
key: String,
serial: u64,
},
}
struct PendingTool {
id: Value,
rpc_key: String,
}
struct Actor {
child: Child,
writer: ChildStdin,
lines: Lines<BufReader<ChildStdout>>,
stderr_task: Option<JoinHandle<()>>,
commands: mpsc::UnboundedReceiver<crate::Command>,
config: Config,
diagnostics: Diagnostics,
conversations: HashMap<String, Conversation>,
by_thread: HashMap<String, String>,
pending: HashMap<u64, Pending>,
tools: HashMap<ToolToken, PendingTool>,
rpc_ids: HashSet<String>,
next_id: u64,
next_turn: u64,
}
pub(crate) async fn open(config: Config) -> Result<Adapter, Error> {
validate_config(&config)?;
let diagnostics = Diagnostics::default();
let mut command = ProcessCommand::new(&config.executable);
command.arg("app-server");
for setting in [
r#"web_search="disabled""#,
"mcp_servers={}",
"agents.enabled=false",
"features.shell_tool=false",
"features.unified_exec=false",
"features.apps=false",
"apps._default.enabled=false",
"features.multi_agent=false",
"features.goals=false",
"features.hooks=false",
"features.memories=false",
"features.remote_plugin=false",
"features.code_mode.enabled=false",
"features.skill_mcp_dependency_install=false",
"tools.view_image=false",
r#"history.persistence="none""#,
] {
command.args(["-c", setting]);
}
if let Some(effort) = &config.reasoning_effort {
let setting = format!("model_reasoning_effort={}", json!(effort));
command.args(["-c", &setting]);
}
let mut child = command
.current_dir(&config.working_directory)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true)
.spawn()
.map_err(|error| {
fault(
ErrorKind::Unavailable,
format!("failed to start Codex app-server: {error}"),
&diagnostics,
)
})?;
let mut writer = child.stdin.take().expect("stdin was configured as a pipe");
let stdout = child
.stdout
.take()
.expect("stdout was configured as a pipe");
let mut stderr = child
.stderr
.take()
.expect("stderr was configured as a pipe");
let captured = diagnostics.clone();
let mut stderr_task = Some(tokio::spawn(async move {
let mut buffer = [0_u8; 8192];
loop {
match stderr.read(&mut buffer).await {
Ok(0) | Err(_) => break,
Ok(length) => captured
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.extend_from_slice(&buffer[..length]),
}
}
}));
let mut lines = BufReader::new(stdout).lines();
let initialized = timeout(
INITIALIZE_TIMEOUT,
initialize(&mut writer, &mut lines, &diagnostics),
)
.await
.unwrap_or_else(|_| {
Err(fault(
ErrorKind::Unavailable,
"Codex app-server initialization timed out",
&diagnostics,
))
});
if let Err(mut error) = initialized {
stop(&mut child, &mut writer, &mut stderr_task).await;
error.diagnostics = snapshot(&diagnostics);
return Err(error);
}
let (commands, receiver) = mpsc::unbounded_channel();
let adapter = Adapter {
client: Arc::new(Client {
commands,
diagnostics: diagnostics.clone(),
}),
};
tokio::spawn(async move {
let mut actor = Actor {
child,
writer,
lines,
stderr_task,
commands: receiver,
config,
diagnostics,
conversations: HashMap::new(),
by_thread: HashMap::new(),
pending: HashMap::new(),
tools: HashMap::new(),
rpc_ids: HashSet::new(),
next_id: 1,
next_turn: 1,
};
let outcome = actor.run().await;
let _ = actor.reject_all("adapter stopped").await;
let sinks = actor.take_sinks();
actor.shutdown().await;
let mut error = outcome
.err()
.unwrap_or_else(|| actor.error(ErrorKind::Unavailable, "Codex app-server was closed"));
error.diagnostics = snapshot(&actor.diagnostics);
for sink in sinks {
let _ = sink.send(Event::Error(error.clone()));
}
});
Ok(adapter)
}
fn validate_config(config: &Config) -> Result<(), Error> {
let diagnostics = Diagnostics::default();
let mut names = HashSet::new();
for tool in &config.tools {
if tool.name.is_empty() {
return Err(fault(
ErrorKind::Protocol,
"dynamic tool names must not be empty",
&diagnostics,
));
}
if !names.insert(&tool.name) {
return Err(fault(
ErrorKind::Protocol,
format!("duplicate dynamic tool name: {}", tool.name),
&diagnostics,
));
}
}
Ok(())
}
async fn initialize(
writer: &mut ChildStdin,
lines: &mut Lines<BufReader<ChildStdout>>,
diagnostics: &Diagnostics,
) -> Result<(), Error> {
write_json(
writer,
&json!({
"method": "initialize",
"id": 0,
"params": {
"clientInfo": {
"name": "kcode-k1-codex-adapter",
"title": "K1 Codex Adapter",
"version": env!("CARGO_PKG_VERSION")
},
"capabilities": { "experimentalApi": true }
}
}),
)
.await
.map_err(|message| fault(ErrorKind::Unavailable, message, diagnostics))?;
loop {
let line = lines
.next_line()
.await
.map_err(|error| fault(ErrorKind::Unavailable, error.to_string(), diagnostics))?
.ok_or_else(|| {
fault(
ErrorKind::Unavailable,
"Codex app-server closed during initialization",
diagnostics,
)
})?;
let message: Value = serde_json::from_str(&line)
.map_err(|error| fault(ErrorKind::Protocol, error.to_string(), diagnostics))?;
if message.get("id").and_then(Value::as_u64) != Some(0) {
continue;
}
if let Some(error) = message.get("error") {
return Err(server_fault(error, diagnostics));
}
if message.get("result").is_some() {
break;
}
return Err(fault(
ErrorKind::Protocol,
"initialize response omitted result",
diagnostics,
));
}
write_json(writer, &json!({"method": "initialized", "params": {}}))
.await
.map_err(|message| fault(ErrorKind::Unavailable, message, diagnostics))
}
impl Actor {
async fn run(&mut self) -> Result<(), Error> {
loop {
tokio::select! {
command = self.commands.recv() => match command {
Some(command) => self.command(command).await?,
None => return Ok(()),
},
line = self.lines.next_line() => {
let line = line
.map_err(|error| self.error(ErrorKind::Unavailable, error.to_string()))?
.ok_or_else(|| self.error(ErrorKind::Unavailable, "Codex app-server closed stdout"))?;
let message = serde_json::from_str(&line)
.map_err(|error| self.error(ErrorKind::Protocol, error.to_string()))?;
self.message(message).await?;
}
}
}
}
async fn command(&mut self, command: crate::Command) -> Result<(), Error> {
match command {
crate::Command::Start {
key,
input,
events,
reply,
} => {
if self
.conversations
.get(&key)
.and_then(|conversation| conversation.active.as_ref())
.is_some()
{
let _ = reply.send(Err(
self.error(ErrorKind::Busy, "conversation already has an active turn")
));
return Ok(());
}
let serial = self.next_turn;
self.next_turn = self.next_turn.checked_add(1).ok_or_else(|| {
self.error(ErrorKind::Protocol, "turn serial space was exhausted")
})?;
let thread = self
.conversations
.get(&key)
.and_then(|conversation| conversation.thread.clone());
self.conversations.entry(key.clone()).or_default().active = Some(Active {
serial,
turn: None,
events: Some(events),
early: Vec::new(),
cancelled: false,
interrupted: false,
failure: None,
});
let (id, pending) = if let Some(thread) = thread {
(
self.request("turn/start", turn_params(&thread, input))
.await?,
Pending::Turn {
key: key.clone(),
serial,
},
)
} else {
(
self.request("thread/start", thread_params(&self.config))
.await?,
Pending::Thread {
key: key.clone(),
serial,
input,
},
)
};
self.pending.insert(id, pending);
if reply.send(Ok(serial)).is_err() {
self.cancel(&key, serial).await?;
}
}
crate::Command::ToolResult {
key,
turn,
call_id,
result,
reply,
} => {
let Some(pending) = self.tools.remove(&(key, turn, call_id)) else {
let _ = reply.send(Err(self.error(
ErrorKind::InvalidToolResult,
"tool call is not pending on this turn",
)));
return Ok(());
};
self.rpc_ids.remove(&pending.rpc_key);
let response = json!({
"id": pending.id,
"result": {
"contentItems": [{"type": "inputText", "text": result.output}],
"success": result.success
}
});
match write_json(&mut self.writer, &response).await {
Ok(()) => {
let _ = reply.send(Ok(()));
}
Err(message) => {
let error = self.error(ErrorKind::Unavailable, message);
let _ = reply.send(Err(error.clone()));
return Err(error);
}
}
}
crate::Command::Abandon { key, turn } => {
self.cancel(&key, turn).await?;
}
}
Ok(())
}
async fn message(&mut self, message: Value) -> Result<(), Error> {
if let Some(method) = message
.get("method")
.and_then(Value::as_str)
.map(str::to_owned)
{
match method.as_str() {
"item/agentMessage/delta" | "item/tool/call" | "turn/completed" => {
self.scoped(message).await?;
}
"turn/started" | "error" if has_scope(message.get("params")) => {
self.scoped(message).await?;
}
"serverRequest/resolved" => {
self.resolved(message.get("params").unwrap_or(&Value::Null));
}
"turn/started" => {}
"error" => {
let details = message.get("params").unwrap_or(&message);
return Err(server_fault(details, &self.diagnostics));
}
_ if message.get("id").is_some() => {
let (id, _) = valid_rpc_id(message.get("id").expect("id was checked"))
.map_err(|error| self.error(ErrorKind::Protocol, error))?;
self.rpc_error(id, -32601, "unsupported server request")
.await?;
}
_ => {}
}
return Ok(());
}
if message.get("id").is_some() {
return self.response(message).await;
}
Err(self.error(
ErrorKind::Protocol,
"app-server message omitted method and id",
))
}
async fn response(&mut self, message: Value) -> Result<(), Error> {
let id = message.get("id").and_then(Value::as_u64).ok_or_else(|| {
self.error(
ErrorKind::Protocol,
"app-server response id was not an unsigned integer",
)
})?;
let Some(pending) = self.pending.remove(&id) else {
return Ok(());
};
let (key, serial) = match &pending {
Pending::Thread { key, serial, .. } | Pending::Turn { key, serial } => {
(key.clone(), *serial)
}
};
if let Some(details) = message.get("error") {
self.prestart_fail(&key, serial, server_fault(details, &self.diagnostics))
.await?;
return Ok(());
}
if message.get("result").is_none() {
let error = self.error(
ErrorKind::Protocol,
"app-server response omitted result and error",
);
self.prestart_fail(&key, serial, error).await?;
return Ok(());
}
match pending {
Pending::Thread { key, serial, input } => {
let Some(thread) = message
.pointer("/result/thread/id")
.and_then(Value::as_str)
.map(str::to_owned)
else {
let error = self.error(
ErrorKind::Protocol,
"thread/start response omitted thread.id",
);
self.prestart_fail(&key, serial, error).await?;
return Ok(());
};
if self
.by_thread
.get(&thread)
.is_some_and(|owner| owner != &key)
{
let error = self.error(
ErrorKind::Protocol,
"thread/start reused another conversation's thread.id",
);
self.prestart_fail(&key, serial, error).await?;
return Ok(());
}
self.by_thread.insert(thread.clone(), key.clone());
if let Some(conversation) = self.conversations.get_mut(&key) {
conversation.thread = Some(thread.clone());
}
let cancelled = self
.conversations
.get(&key)
.and_then(|conversation| conversation.active.as_ref())
.is_none_or(|active| active.serial != serial || active.cancelled);
if cancelled {
self.take_active(&key, serial);
} else {
let id = self
.request("turn/start", turn_params(&thread, input))
.await?;
self.pending.insert(id, Pending::Turn { key, serial });
}
}
Pending::Turn { key, serial } => {
let Some(turn) = message
.pointer("/result/turn/id")
.and_then(Value::as_str)
.map(str::to_owned)
else {
let error =
self.error(ErrorKind::Protocol, "turn/start response omitted turn.id");
self.prestart_fail(&key, serial, error).await?;
return Ok(());
};
let Some(active) = self
.conversations
.get_mut(&key)
.and_then(|conversation| conversation.active.as_mut())
.filter(|active| active.serial == serial)
else {
self.interrupt_orphan(&key, &turn).await?;
return Ok(());
};
active.turn = Some(turn);
let early = std::mem::take(&mut active.early);
for event in early {
let cancelled = self
.conversations
.get(&key)
.and_then(|conversation| conversation.active.as_ref())
.is_none_or(|active| active.serial != serial || active.cancelled);
if cancelled {
self.reject_if_request(&event, "turn was abandoned").await?;
} else {
self.scoped(event).await?;
}
}
let cancelled = self
.conversations
.get(&key)
.and_then(|conversation| conversation.active.as_ref())
.is_some_and(|active| active.serial == serial && active.cancelled);
if cancelled {
self.interrupt(&key, serial).await?;
}
}
}
Ok(())
}
async fn scoped(&mut self, message: Value) -> Result<(), Error> {
let (thread, turn) =
scope(message.get("params")).map_err(|error| self.error(ErrorKind::Protocol, error))?;
let Some(key) = self.by_thread.get(thread).cloned() else {
self.reject_if_request(&message, "tool call has an unknown thread")
.await?;
return Ok(());
};
let Some(active) = self
.conversations
.get_mut(&key)
.and_then(|conversation| conversation.active.as_mut())
else {
self.reject_if_request(&message, "tool call does not belong to an active turn")
.await?;
return Ok(());
};
let serial = active.serial;
match active.turn.as_deref() {
Some(current) if current != turn => {
self.reject_if_request(&message, "tool call does not belong to the active turn")
.await?;
return Ok(());
}
None => {
if active.early.len() == EARLY_EVENT_LIMIT {
return Err(self.error(
ErrorKind::Protocol,
"too many events arrived before turn/start completed",
));
}
active.early.push(message);
return Ok(());
}
_ => {}
}
self.dispatch(key, serial, message).await
}
async fn dispatch(&mut self, key: String, serial: u64, message: Value) -> Result<(), Error> {
let method = message
.get("method")
.and_then(Value::as_str)
.expect("only method-bearing messages are dispatched");
let params = message
.get("params")
.expect("scoped messages contain params");
match method {
"item/agentMessage/delta" => {
let delta = params
.get("delta")
.and_then(Value::as_str)
.ok_or_else(|| {
self.error(ErrorKind::Protocol, "agent-message delta omitted delta")
})?
.to_owned();
if !self.emit(&key, serial, Event::TextDelta(delta)) {
self.cancel(&key, serial).await?;
}
}
"item/tool/call" => self.tool_call(&key, serial, &message).await?,
"turn/completed" => self.complete(&key, serial, params).await?,
"error" => {
if let Some(active) = self
.conversations
.get_mut(&key)
.and_then(|conversation| conversation.active.as_mut())
.filter(|active| active.serial == serial)
{
active.failure = params.get("error").cloned();
}
}
"turn/started" => {}
_ => unreachable!("message filtering guarantees a known method"),
}
Ok(())
}
async fn tool_call(&mut self, key: &str, serial: u64, message: &Value) -> Result<(), Error> {
let (id, rpc_key) =
valid_rpc_id(message.get("id").ok_or_else(|| {
self.error(ErrorKind::Protocol, "dynamic tool request omitted id")
})?)
.map_err(|error| self.error(ErrorKind::Protocol, error))?;
if !self.rpc_ids.insert(rpc_key.clone()) {
return Err(self.error(ErrorKind::Protocol, "duplicate app-server request id"));
}
let params = message
.get("params")
.expect("scoped messages contain params");
let fields = params
.get("callId")
.and_then(Value::as_str)
.zip(params.get("tool").and_then(Value::as_str))
.zip(params.get("arguments"));
let Some(((call_id, name), arguments)) = fields else {
self.rpc_ids.remove(&rpc_key);
self.rpc_error(id, -32602, "malformed dynamic tool request")
.await?;
let error = self.error(ErrorKind::Protocol, "malformed dynamic tool request");
return self.fail_turn(key, serial, error).await;
};
if !self.config.tools.iter().any(|tool| tool.name == name) {
self.rpc_ids.remove(&rpc_key);
self.rpc_error(id, -32602, "unconfigured dynamic tool requested")
.await?;
let error = self.error(ErrorKind::Protocol, "unconfigured dynamic tool requested");
return self.fail_turn(key, serial, error).await;
}
let token = (key.to_owned(), serial, call_id.to_owned());
if self.tools.contains_key(&token) {
self.rpc_ids.remove(&rpc_key);
self.rpc_error(id, -32602, "duplicate dynamic tool call id")
.await?;
let error = self.error(ErrorKind::Protocol, "duplicate dynamic tool call id");
return self.fail_turn(key, serial, error).await;
}
self.tools.insert(token, PendingTool { id, rpc_key });
if !self.emit(
key,
serial,
Event::ToolCall(ToolCall {
call_id: call_id.to_owned(),
name: name.to_owned(),
arguments: arguments.clone(),
}),
) {
self.cancel(key, serial).await?;
}
Ok(())
}
async fn complete(&mut self, key: &str, serial: u64, params: &Value) -> Result<(), Error> {
let pending_count = self
.tools
.keys()
.filter(|(tool_key, tool_serial, _)| tool_key == key && *tool_serial == serial)
.count();
if pending_count != 0 {
self.reject_tools(
key,
serial,
"turn completed before its tool calls were answered",
)
.await?;
}
let Some(active) = self.take_active(key, serial) else {
return Ok(());
};
if active.cancelled {
return Ok(());
}
let event = match params.pointer("/turn/status").and_then(Value::as_str) {
Some("completed") if pending_count == 0 => Event::Done,
Some("completed") => Event::Error(self.error(
ErrorKind::Protocol,
"turn completed with pending tool calls",
)),
Some("interrupted") => {
Event::Error(self.error(ErrorKind::Interrupted, "Codex turn was interrupted"))
}
Some("failed") => Event::Error(server_fault(
params
.pointer("/turn/error")
.or(active.failure.as_ref())
.unwrap_or(params),
&self.diagnostics,
)),
_ => Event::Error(
self.error(ErrorKind::Protocol, "turn/completed had an invalid status"),
),
};
if let Some(events) = active.events {
let _ = events.send(event);
}
Ok(())
}
async fn fail_turn(&mut self, key: &str, serial: u64, error: Error) -> Result<(), Error> {
if let Some(active) = self
.conversations
.get_mut(key)
.and_then(|conversation| conversation.active.as_mut())
.filter(|active| active.serial == serial)
{
if let Some(events) = active.events.take() {
let _ = events.send(Event::Error(error));
}
active.cancelled = true;
}
self.reject_tools(key, serial, "turn failed").await?;
self.interrupt(key, serial).await
}
async fn prestart_fail(&mut self, key: &str, serial: u64, error: Error) -> Result<(), Error> {
let active = self.take_active(key, serial);
self.reject_tools(key, serial, "turn failed to start")
.await?;
if let Some(events) = active.and_then(|active| active.events) {
let _ = events.send(Event::Error(error));
}
Ok(())
}
async fn cancel(&mut self, key: &str, serial: u64) -> Result<(), Error> {
if let Some(active) = self
.conversations
.get_mut(key)
.and_then(|conversation| conversation.active.as_mut())
.filter(|active| active.serial == serial)
{
active.events.take();
active.cancelled = true;
}
self.reject_tools(key, serial, "turn was abandoned").await?;
self.interrupt(key, serial).await
}
async fn interrupt(&mut self, key: &str, serial: u64) -> Result<(), Error> {
let target = self.conversations.get_mut(key).and_then(|conversation| {
let thread = conversation.thread.clone()?;
let active = conversation.active.as_mut()?;
if active.serial != serial || active.interrupted {
return None;
}
let turn = active.turn.clone()?;
active.interrupted = true;
Some((thread, turn))
});
if let Some((thread, turn)) = target {
self.request(
"turn/interrupt",
json!({"threadId": thread, "turnId": turn}),
)
.await?;
}
Ok(())
}
async fn interrupt_orphan(&mut self, key: &str, turn: &str) -> Result<(), Error> {
if let Some(thread) = self
.conversations
.get(key)
.and_then(|conversation| conversation.thread.clone())
{
self.request(
"turn/interrupt",
json!({"threadId": thread, "turnId": turn}),
)
.await?;
}
Ok(())
}
fn emit(&mut self, key: &str, serial: u64, event: Event) -> bool {
self.conversations
.get_mut(key)
.and_then(|conversation| conversation.active.as_mut())
.filter(|active| active.serial == serial)
.and_then(|active| active.events.as_ref())
.is_some_and(|events| events.send(event).is_ok())
}
fn resolved(&mut self, params: &Value) {
let Some(rpc_key) = params
.get("requestId")
.and_then(|id| valid_rpc_id(id).ok())
.map(|(_, key)| key)
else {
return;
};
if let Some(token) = self
.tools
.iter()
.find_map(|(token, pending)| (pending.rpc_key == rpc_key).then(|| token.clone()))
{
self.tools.remove(&token);
self.rpc_ids.remove(&rpc_key);
}
}
async fn reject_if_request(&mut self, message: &Value, reason: &str) -> Result<(), Error> {
if message.get("method").and_then(Value::as_str) == Some("item/tool/call") {
let (id, _) = valid_rpc_id(message.get("id").ok_or_else(|| {
self.error(ErrorKind::Protocol, "dynamic tool request omitted id")
})?)
.map_err(|error| self.error(ErrorKind::Protocol, error))?;
self.rpc_error(id, -32602, reason).await?;
}
Ok(())
}
async fn reject_tools(&mut self, key: &str, serial: u64, reason: &str) -> Result<(), Error> {
let tokens: Vec<_> = self
.tools
.keys()
.filter(|(tool_key, tool_serial, _)| tool_key == key && *tool_serial == serial)
.cloned()
.collect();
for token in tokens {
if let Some(pending) = self.tools.remove(&token) {
self.rpc_ids.remove(&pending.rpc_key);
self.rpc_error(pending.id, -32800, reason).await?;
}
}
Ok(())
}
async fn reject_all(&mut self, reason: &str) -> Result<(), Error> {
let tokens: Vec<_> = self.tools.keys().cloned().collect();
for token in tokens {
if let Some(pending) = self.tools.remove(&token) {
self.rpc_ids.remove(&pending.rpc_key);
self.rpc_error(pending.id, -32800, reason).await?;
}
}
Ok(())
}
async fn request(&mut self, method: &str, params: Value) -> Result<u64, Error> {
let id = self.next_id;
self.next_id = self.next_id.checked_add(1).ok_or_else(|| {
self.error(ErrorKind::Protocol, "client request id space was exhausted")
})?;
write_json(
&mut self.writer,
&json!({"method": method, "id": id, "params": params}),
)
.await
.map_err(|message| self.error(ErrorKind::Unavailable, message))?;
Ok(id)
}
async fn rpc_error(&mut self, id: Value, code: i64, message: &str) -> Result<(), Error> {
write_json(
&mut self.writer,
&json!({"id": id, "error": {"code": code, "message": message}}),
)
.await
.map_err(|error| self.error(ErrorKind::Unavailable, error))
}
fn take_active(&mut self, key: &str, serial: u64) -> Option<Active> {
let conversation = self.conversations.get_mut(key)?;
if conversation
.active
.as_ref()
.is_some_and(|active| active.serial == serial)
{
conversation.active.take()
} else {
None
}
}
fn take_sinks(&mut self) -> Vec<Events> {
self.conversations
.values_mut()
.filter_map(|conversation| conversation.active.take().and_then(|active| active.events))
.collect()
}
async fn shutdown(&mut self) {
stop(&mut self.child, &mut self.writer, &mut self.stderr_task).await;
}
fn error(&self, kind: ErrorKind, message: impl Into<String>) -> Error {
fault(kind, message, &self.diagnostics)
}
}
fn has_scope(params: Option<&Value>) -> bool {
params.and_then(|params| params.get("threadId")).is_some()
&& params.and_then(native_turn).is_some()
}
fn scope(params: Option<&Value>) -> Result<(&str, &str), String> {
let params = params.ok_or_else(|| "omitted params".to_owned())?;
let thread = params
.get("threadId")
.and_then(Value::as_str)
.ok_or_else(|| "omitted threadId".to_owned())?;
let turn = native_turn(params).ok_or_else(|| "omitted turn id".to_owned())?;
Ok((thread, turn))
}
fn native_turn(params: &Value) -> Option<&str> {
params
.get("turnId")
.and_then(Value::as_str)
.or_else(|| params.pointer("/turn/id").and_then(Value::as_str))
}
fn valid_rpc_id(id: &Value) -> Result<(Value, String), String> {
match id {
Value::String(value) => Ok((id.clone(), format!("s:{value}"))),
Value::Number(value) => Ok((id.clone(), format!("n:{value}"))),
_ => Err("server request id must be a string or number".to_owned()),
}
}
fn thread_params(config: &Config) -> Value {
let tools: Vec<_> = config
.tools
.iter()
.map(|tool| {
json!({
"name": tool.name,
"description": tool.description,
"inputSchema": tool.input_schema
})
})
.collect();
json!({
"model": config.model,
"cwd": config.working_directory,
"approvalPolicy": "never",
"sandbox": "readOnly",
"baseInstructions": config.base_instructions,
"serviceName": "kcode-k1-codex-adapter",
"ephemeral": true,
"dynamicTools": tools
})
}
fn turn_params(thread: &str, input: String) -> Value {
json!({
"threadId": thread,
"input": [{"type": "text", "text": input}]
})
}
async fn write_json(writer: &mut ChildStdin, value: &Value) -> Result<(), String> {
let mut bytes = serde_json::to_vec(value).map_err(|error| error.to_string())?;
bytes.push(b'\n');
writer
.write_all(&bytes)
.await
.map_err(|error| error.to_string())?;
writer.flush().await.map_err(|error| error.to_string())
}
async fn stop(
child: &mut Child,
writer: &mut ChildStdin,
stderr_task: &mut Option<JoinHandle<()>>,
) {
let _ = writer.shutdown().await;
if timeout(SHUTDOWN_TIMEOUT, child.wait()).await.is_err() {
let _ = child.start_kill();
let _ = child.wait().await;
}
if let Some(task) = stderr_task.take() {
let _ = timeout(SHUTDOWN_TIMEOUT, task).await;
}
}
fn server_fault(value: &Value, diagnostics: &Diagnostics) -> Error {
let details = value.to_string();
let message = value
.get("message")
.and_then(Value::as_str)
.map(|message| format!("{message}: {details}"))
.unwrap_or(details);
fault(ErrorKind::Server, message, diagnostics)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn rpc_id_keys_preserve_json_type() {
assert_eq!(valid_rpc_id(&json!(7)).unwrap().1, "n:7");
assert_eq!(valid_rpc_id(&json!("7")).unwrap().1, "s:7");
assert!(valid_rpc_id(&Value::Null).is_err());
}
#[test]
fn thread_configuration_is_locked_down() {
let config = Config {
executable: "codex".into(),
working_directory: "/tmp".to_owned(),
model: "example-model".to_owned(),
reasoning_effort: None,
base_instructions: "base".to_owned(),
tools: vec![DynamicTool {
name: "lookup".to_owned(),
description: "Look something up".to_owned(),
input_schema: json!({"type": "object"}),
}],
};
let params = thread_params(&config);
assert_eq!(params["approvalPolicy"], "never");
assert_eq!(params["sandbox"], "readOnly");
assert_eq!(params["ephemeral"], true);
assert_eq!(params["dynamicTools"][0]["name"], "lookup");
}
}