use kcode_k1_codex_conversations::{Error, ErrorKind, is_model_reroute};
use serde_json::{Map, Number, Value, json};
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
pub enum RpcId {
String(String),
Number(Number),
}
impl RpcId {
pub fn as_value(&self) -> Value {
match self {
Self::String(value) => Value::String(value.clone()),
Self::Number(value) => Value::Number(value.clone()),
}
}
}
impl TryFrom<&Value> for RpcId {
type Error = Error;
fn try_from(value: &Value) -> Result<Self, Self::Error> {
match value {
Value::String(value) => Ok(Self::String(value.clone())),
Value::Number(value) => Ok(Self::Number(value.clone())),
_ => Err(protocol("server request id must be a string or number")),
}
}
}
impl From<RpcId> for Value {
fn from(id: RpcId) -> Self {
id.as_value()
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct ServerError {
pub details: Value,
}
impl ServerError {
pub fn new(details: Value) -> Self {
Self { details }
}
pub fn to_error(&self, diagnostics: Vec<u8>) -> Error {
let detail = self
.details
.get("message")
.and_then(Value::as_str)
.or_else(|| {
self.details
.pointer("/error/message")
.and_then(Value::as_str)
});
let message = detail.map_or_else(
|| "Codex app-server error".to_owned(),
|detail| format!("Codex app-server error: {detail}"),
);
Error {
kind: ErrorKind::Server,
message,
diagnostics,
}
}
}
#[derive(Clone, Debug, PartialEq)]
pub struct ClientResponse {
pub id: u64,
pub outcome: ResponseOutcome,
}
#[derive(Clone, Debug, PartialEq)]
pub enum ResponseOutcome {
Result(Value),
Error(ServerError),
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Scope {
pub thread_id: String,
pub turn_id: String,
}
#[derive(Clone, Debug, PartialEq)]
pub struct ScopedEvent {
pub scope: Scope,
pub kind: ScopedKind,
}
#[derive(Clone, Debug, PartialEq)]
pub enum ScopedKind {
AgentTextDelta(String),
DynamicToolCall(DynamicToolCall),
TurnCompleted(TurnCompleted),
TurnStarted,
Error(ServerError),
}
#[derive(Clone, Debug, PartialEq)]
pub struct DynamicToolCall {
pub rpc_id: RpcId,
pub call_id: String,
pub name: String,
pub arguments: Value,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum TurnStatus {
Completed,
Interrupted,
Failed,
}
#[derive(Clone, Debug, PartialEq)]
pub struct TurnCompleted {
pub status: TurnStatus,
pub failure: Option<ServerError>,
}
#[derive(Clone, Debug, PartialEq)]
pub struct ServerRequest {
pub method: String,
pub id: RpcId,
}
#[derive(Clone, Debug, PartialEq)]
pub struct ModelReroute {
pub method: String,
pub id: Option<RpcId>,
}
#[derive(Clone, Debug, PartialEq)]
pub enum Inbound {
ClientResponse(ClientResponse),
Scoped(ScopedEvent),
ResolvedRequest(RpcId),
GlobalServerError(ServerError),
ModelReroute(ModelReroute),
UnsupportedServerRequest(ServerRequest),
IgnoredNotification { method: String },
Malformed(Error),
}
pub fn decode(message: Value) -> Inbound {
decode_inner(&message).unwrap_or_else(Inbound::Malformed)
}
pub fn rejection_payload(id: &RpcId, code: i64, message: &str) -> Value {
json!({"id": id.as_value(), "error": {"code": code, "message": message}})
}
fn decode_inner(message: &Value) -> Result<Inbound, Error> {
let object = message
.as_object()
.ok_or_else(|| protocol("app-server message must be an object"))?;
let method = match object.get("method") {
Some(Value::String(method)) => Some(method.as_str()),
Some(_) => return Err(protocol("app-server method must be a string")),
None => None,
};
match method {
Some(method) => decode_method(method, object, message),
None => decode_response(object),
}
}
fn decode_method(
method: &str,
object: &Map<String, Value>,
message: &Value,
) -> Result<Inbound, Error> {
if is_model_reroute(method) {
let id = object.get("id").map(RpcId::try_from).transpose()?;
return Ok(Inbound::ModelReroute(ModelReroute {
method: method.to_owned(),
id,
}));
}
match method {
"item/agentMessage/delta" | "item/tool/call" | "turn/completed" | "turn/started" => {
decode_scoped(method, object).map(Inbound::Scoped)
}
"error" if has_scope_fields(object.get("params")) => {
decode_scoped(method, object).map(Inbound::Scoped)
}
"error" => Ok(Inbound::GlobalServerError(ServerError::new(
object.get("params").unwrap_or(message).clone(),
))),
"serverRequest/resolved" => {
let params = required_object(object, "params", "resolved request omitted params")?;
let id = params
.get("requestId")
.ok_or_else(|| protocol("resolved request omitted requestId"))?;
Ok(Inbound::ResolvedRequest(RpcId::try_from(id)?))
}
_ => match object.get("id") {
Some(id) => Ok(Inbound::UnsupportedServerRequest(ServerRequest {
method: method.to_owned(),
id: RpcId::try_from(id)?,
})),
None => Ok(Inbound::IgnoredNotification {
method: method.to_owned(),
}),
},
}
}
fn decode_response(object: &Map<String, Value>) -> Result<Inbound, Error> {
let id = object
.get("id")
.and_then(Value::as_u64)
.ok_or_else(|| protocol("app-server response id was not an unsigned integer"))?;
let outcome = match (object.get("result"), object.get("error")) {
(Some(result), None) => ResponseOutcome::Result(result.clone()),
(None, Some(error)) => ResponseOutcome::Error(ServerError::new(error.clone())),
(Some(_), Some(_)) => return Err(protocol("app-server response had result and error")),
(None, None) => return Err(protocol("app-server response omitted result and error")),
};
Ok(Inbound::ClientResponse(ClientResponse { id, outcome }))
}
fn decode_scoped(method: &str, object: &Map<String, Value>) -> Result<ScopedEvent, Error> {
let params = required_object(object, "params", "scoped message omitted params")?;
let scope = decode_scope(params)?;
let kind = match method {
"item/agentMessage/delta" => ScopedKind::AgentTextDelta(
required_string(params, "delta", "agent message delta omitted delta")?.to_owned(),
),
"item/tool/call" => {
let rpc_id = object
.get("id")
.ok_or_else(|| protocol("dynamic tool request omitted id"))?;
let call_id = required_string(params, "callId", "dynamic tool request omitted callId")?;
let name = required_string(params, "tool", "dynamic tool request omitted tool")?;
let arguments = params
.get("arguments")
.ok_or_else(|| protocol("dynamic tool request omitted arguments"))?;
ScopedKind::DynamicToolCall(DynamicToolCall {
rpc_id: RpcId::try_from(rpc_id)?,
call_id: call_id.to_owned(),
name: name.to_owned(),
arguments: arguments.clone(),
})
}
"turn/completed" => {
let turn = required_object(params, "turn", "turn/completed omitted turn")?;
let status = match required_string(turn, "status", "turn/completed omitted status")? {
"completed" => TurnStatus::Completed,
"interrupted" => TurnStatus::Interrupted,
"failed" => TurnStatus::Failed,
_ => return Err(protocol("turn/completed had an invalid status")),
};
let failure = turn.get("error").cloned().map(ServerError::new);
ScopedKind::TurnCompleted(TurnCompleted { status, failure })
}
"turn/started" => ScopedKind::TurnStarted,
"error" => ScopedKind::Error(ServerError::new(
params.get("error").unwrap_or(&Value::Null).clone(),
)),
_ => return Err(protocol("unsupported scoped method")),
};
Ok(ScopedEvent { scope, kind })
}
fn decode_scope(params: &Map<String, Value>) -> Result<Scope, Error> {
let thread_id = required_string(params, "threadId", "scoped message omitted threadId")?;
let direct = optional_string(params, "turnId", "scoped message turnId must be a string")?;
let nested = match params.get("turn") {
Some(Value::Object(turn)) => Some(required_string(
turn,
"id",
"scoped message turn omitted id",
)?),
Some(_) => return Err(protocol("scoped message turn must be an object")),
None => None,
};
let turn_id = match (direct, nested) {
(Some(direct), Some(nested)) if direct != nested => {
return Err(protocol("scoped message had conflicting turn ids"));
}
(Some(turn), _) | (None, Some(turn)) => turn,
(None, None) => return Err(protocol("scoped message omitted turn id")),
};
Ok(Scope {
thread_id: thread_id.to_owned(),
turn_id: turn_id.to_owned(),
})
}
fn has_scope_fields(params: Option<&Value>) -> bool {
params.and_then(Value::as_object).is_some_and(|params| {
params.contains_key("threadId")
|| params.contains_key("turnId")
|| params.contains_key("turn")
})
}
fn required_object<'a>(
object: &'a Map<String, Value>,
field: &str,
message: &'static str,
) -> Result<&'a Map<String, Value>, Error> {
object
.get(field)
.and_then(Value::as_object)
.ok_or_else(|| protocol(message))
}
fn required_string<'a>(
object: &'a Map<String, Value>,
field: &str,
message: &'static str,
) -> Result<&'a str, Error> {
object
.get(field)
.and_then(Value::as_str)
.ok_or_else(|| protocol(message))
}
fn optional_string<'a>(
object: &'a Map<String, Value>,
field: &str,
message: &'static str,
) -> Result<Option<&'a str>, Error> {
object
.get(field)
.map(|value| value.as_str().ok_or_else(|| protocol(message)))
.transpose()
}
fn protocol(message: impl Into<String>) -> Error {
Error::new(ErrorKind::Protocol, message)
}
#[cfg(test)]
mod tests {
use super::*;
fn parsed(json: &str) -> Inbound {
decode(serde_json::from_str(json).expect("valid test JSON"))
}
fn malformed(json: &str) {
assert!(matches!(
parsed(json),
Inbound::Malformed(Error {
kind: ErrorKind::Protocol,
..
})
));
}
#[test]
fn decodes_client_response_outcomes_and_rejects_wrong_ids() {
assert_eq!(
parsed(r#"{"id":7,"result":{"ok":true}}"#),
Inbound::ClientResponse(ClientResponse {
id: 7,
outcome: ResponseOutcome::Result(json!({"ok": true})),
})
);
assert!(matches!(
parsed(r#"{"id":8,"error":{"message":"no"}}"#),
Inbound::ClientResponse(ClientResponse {
id: 8,
outcome: ResponseOutcome::Error(_),
})
));
for value in [
r#"{"id":"7","result":{}}"#,
r#"{"id":-1,"result":{}}"#,
r#"{"id":1.5,"result":{}}"#,
] {
malformed(value);
}
malformed(r#"{"id":7}"#);
malformed(r#"{"id":7,"result":{},"error":{}}"#);
}
#[test]
fn decodes_delta_and_enforces_exact_scope() {
assert_eq!(
parsed(
r#"{"method":"item/agentMessage/delta","params":{"threadId":"th","turnId":"tu","delta":"hi"}}"#
),
Inbound::Scoped(ScopedEvent {
scope: Scope {
thread_id: "th".into(),
turn_id: "tu".into()
},
kind: ScopedKind::AgentTextDelta("hi".into()),
})
);
malformed(r#"{"method":"turn/started","params":{"threadId":1,"turnId":"tu"}}"#);
malformed(r#"{"method":"turn/started","params":{"threadId":"th","turnId":1}}"#);
malformed(
r#"{"method":"turn/started","params":{"threadId":"th","turnId":"a","turn":{"id":"b"}}}"#,
);
}
#[test]
fn decodes_tool_calls_and_preserves_rpc_id_kind() {
let string = parsed(
r#"{"method":"item/tool/call","id":"9","params":{"threadId":"th","turnId":"tu","callId":"c","tool":"search","arguments":{"q":1}}}"#,
);
let number = parsed(
r#"{"method":"item/tool/call","id":9,"params":{"threadId":"th","turnId":"tu","callId":"c2","tool":"search","arguments":{}}}"#,
);
let Inbound::Scoped(ScopedEvent {
kind: ScopedKind::DynamicToolCall(first),
..
}) = string
else {
panic!("tool call expected")
};
let Inbound::Scoped(ScopedEvent {
kind: ScopedKind::DynamicToolCall(second),
..
}) = number
else {
panic!("tool call expected")
};
assert_ne!(first.rpc_id, second.rpc_id);
assert_eq!(first.call_id, "c");
assert_eq!(first.name, "search");
assert_eq!(first.arguments, json!({"q": 1}));
assert_eq!(rejection_payload(&first.rpc_id, -32602, "bad")["id"], "9");
malformed(
r#"{"method":"item/tool/call","id":null,"params":{"threadId":"th","turnId":"tu","callId":"c","tool":"t","arguments":{}}}"#,
);
malformed(
r#"{"method":"item/tool/call","id":1,"params":{"threadId":"th","turnId":"tu","callId":2,"tool":"t","arguments":{}}}"#,
);
malformed(
r#"{"method":"item/tool/call","id":1,"params":{"threadId":"th","turnId":"tu","callId":"c","tool":"t"}}"#,
);
}
#[test]
fn decodes_completion_statuses_and_failures() {
for (status, expected) in [
("completed", TurnStatus::Completed),
("interrupted", TurnStatus::Interrupted),
("failed", TurnStatus::Failed),
] {
let value = format!(
r#"{{"method":"turn/completed","params":{{"threadId":"th","turn":{{"id":"tu","status":"{status}","error":{{"message":"why"}}}}}}}}"#
);
let Inbound::Scoped(ScopedEvent {
kind: ScopedKind::TurnCompleted(completed),
..
}) = parsed(&value)
else {
panic!("completion expected")
};
assert_eq!(completed.status, expected);
assert_eq!(
completed.failure.expect("failure").details["message"],
"why"
);
}
malformed(
r#"{"method":"turn/completed","params":{"threadId":"th","turn":{"id":"tu","status":"other"}}}"#,
);
malformed(
r#"{"method":"turn/completed","params":{"threadId":"th","turn":{"id":"tu","status":1}}}"#,
);
}
#[test]
fn decodes_turn_started_scoped_error_and_global_error() {
assert!(matches!(
parsed(r#"{"method":"turn/started","params":{"threadId":"th","turnId":"tu"}}"#),
Inbound::Scoped(ScopedEvent {
kind: ScopedKind::TurnStarted,
..
})
));
assert!(matches!(
parsed(
r#"{"method":"error","params":{"threadId":"th","turnId":"tu","error":{"message":"scoped"}}}"#
),
Inbound::Scoped(ScopedEvent {
kind: ScopedKind::Error(_),
..
})
));
let Inbound::GlobalServerError(error) =
parsed(r#"{"method":"error","params":{"message":"global"}}"#)
else {
panic!("global error expected")
};
assert_eq!(
error.to_error(vec![1]).message,
"Codex app-server error: global"
);
malformed(r#"{"method":"error","params":{"threadId":"th","turnId":3}}"#);
}
#[test]
fn classifies_reroutes_requests_notifications_and_resolutions() {
assert!(matches!(
parsed(r#"{"method":"model/rerouted","id":"r"}"#),
Inbound::ModelReroute(ModelReroute { id: Some(RpcId::String(id)), .. }) if id == "r"
));
assert!(matches!(
parsed(r#"{"method":"future/request","id":4}"#),
Inbound::UnsupportedServerRequest(ServerRequest {
id: RpcId::Number(_),
..
})
));
assert_eq!(
parsed(r#"{"method":"serverRequest/resolved","params":{"requestId":"4"}}"#),
Inbound::ResolvedRequest(RpcId::String("4".into()))
);
assert_eq!(
parsed(r#"{"method":"future/notification"}"#),
Inbound::IgnoredNotification {
method: "future/notification".into()
}
);
malformed(r#"{"method":"serverRequest/resolved","params":{"requestId":null}}"#);
malformed(r#"{"method":"future/request","id":null}"#);
malformed(r#"{"method":1,"id":1}"#);
malformed("[]");
}
}