1use kcode_k1_codex_conversations::{Error, ErrorKind, is_model_reroute};
2use serde_json::{Map, Number, Value, json};
3#[derive(Clone, Debug, PartialEq, Eq, Hash)]
4pub enum RpcId {
5 String(String),
6 Number(Number),
7}
8impl RpcId {
9 pub fn as_value(&self) -> Value {
10 match self {
11 Self::String(value) => Value::String(value.clone()),
12 Self::Number(value) => Value::Number(value.clone()),
13 }
14 }
15}
16impl TryFrom<&Value> for RpcId {
17 type Error = Error;
18 fn try_from(value: &Value) -> Result<Self, Self::Error> {
19 match value {
20 Value::String(value) => Ok(Self::String(value.clone())),
21 Value::Number(value) => Ok(Self::Number(value.clone())),
22 _ => Err(protocol("server request id must be a string or number")),
23 }
24 }
25}
26impl From<RpcId> for Value {
27 fn from(id: RpcId) -> Self {
28 id.as_value()
29 }
30}
31#[derive(Clone, Debug, PartialEq)]
32pub struct ServerError {
33 pub details: Value,
34}
35impl ServerError {
36 pub fn new(details: Value) -> Self {
37 Self { details }
38 }
39 pub fn to_error(&self, diagnostics: Vec<u8>) -> Error {
40 let detail = self
41 .details
42 .get("message")
43 .and_then(Value::as_str)
44 .or_else(|| {
45 self.details
46 .pointer("/error/message")
47 .and_then(Value::as_str)
48 });
49 let message = detail.map_or_else(
50 || "Codex app-server error".to_owned(),
51 |detail| format!("Codex app-server error: {detail}"),
52 );
53 Error {
54 kind: ErrorKind::Server,
55 message,
56 diagnostics,
57 }
58 }
59}
60#[derive(Clone, Debug, PartialEq)]
61pub struct ClientResponse {
62 pub id: u64,
63 pub outcome: ResponseOutcome,
64}
65#[derive(Clone, Debug, PartialEq)]
66pub enum ResponseOutcome {
67 Result(Value),
68 Error(ServerError),
69}
70#[derive(Clone, Debug, PartialEq, Eq)]
71pub struct Scope {
72 pub thread_id: String,
73 pub turn_id: String,
74}
75#[derive(Clone, Debug, PartialEq)]
76pub struct ScopedEvent {
77 pub scope: Scope,
78 pub kind: ScopedKind,
79}
80#[derive(Clone, Debug, PartialEq)]
81pub enum ScopedKind {
82 AgentTextDelta(String),
83 DynamicToolCall(DynamicToolCall),
84 TurnCompleted(TurnCompleted),
85 TurnStarted,
86 Error(ServerError),
87}
88#[derive(Clone, Debug, PartialEq)]
89pub struct DynamicToolCall {
90 pub rpc_id: RpcId,
91 pub call_id: String,
92 pub name: String,
93 pub arguments: Value,
94}
95#[derive(Clone, Copy, Debug, PartialEq, Eq)]
96pub enum TurnStatus {
97 Completed,
98 Interrupted,
99 Failed,
100}
101#[derive(Clone, Debug, PartialEq)]
102pub struct TurnCompleted {
103 pub status: TurnStatus,
104 pub failure: Option<ServerError>,
105}
106#[derive(Clone, Debug, PartialEq)]
107pub struct ServerRequest {
108 pub method: String,
109 pub id: RpcId,
110}
111#[derive(Clone, Debug, PartialEq)]
112pub struct ModelReroute {
113 pub method: String,
114 pub id: Option<RpcId>,
115}
116#[derive(Clone, Debug, PartialEq)]
117pub enum Inbound {
118 ClientResponse(ClientResponse),
119 Scoped(ScopedEvent),
120 ResolvedRequest(RpcId),
121 GlobalServerError(ServerError),
122 ModelReroute(ModelReroute),
123 UnsupportedServerRequest(ServerRequest),
124 IgnoredNotification { method: String },
125 Malformed(Error),
126}
127pub fn decode(message: Value) -> Inbound {
128 decode_inner(&message).unwrap_or_else(Inbound::Malformed)
129}
130pub fn rejection_payload(id: &RpcId, code: i64, message: &str) -> Value {
131 json!({"id": id.as_value(), "error": {"code": code, "message": message}})
132}
133fn decode_inner(message: &Value) -> Result<Inbound, Error> {
134 let object = message
135 .as_object()
136 .ok_or_else(|| protocol("app-server message must be an object"))?;
137 let method = match object.get("method") {
138 Some(Value::String(method)) => Some(method.as_str()),
139 Some(_) => return Err(protocol("app-server method must be a string")),
140 None => None,
141 };
142 match method {
143 Some(method) => decode_method(method, object, message),
144 None => decode_response(object),
145 }
146}
147fn decode_method(
148 method: &str,
149 object: &Map<String, Value>,
150 message: &Value,
151) -> Result<Inbound, Error> {
152 if is_model_reroute(method) {
153 let id = object.get("id").map(RpcId::try_from).transpose()?;
154 return Ok(Inbound::ModelReroute(ModelReroute {
155 method: method.to_owned(),
156 id,
157 }));
158 }
159 match method {
160 "item/agentMessage/delta" | "item/tool/call" | "turn/completed" | "turn/started" => {
161 decode_scoped(method, object).map(Inbound::Scoped)
162 }
163 "error" if has_scope_fields(object.get("params")) => {
164 decode_scoped(method, object).map(Inbound::Scoped)
165 }
166 "error" => Ok(Inbound::GlobalServerError(ServerError::new(
167 object.get("params").unwrap_or(message).clone(),
168 ))),
169 "serverRequest/resolved" => {
170 let params = required_object(object, "params", "resolved request omitted params")?;
171 let id = params
172 .get("requestId")
173 .ok_or_else(|| protocol("resolved request omitted requestId"))?;
174 Ok(Inbound::ResolvedRequest(RpcId::try_from(id)?))
175 }
176 _ => match object.get("id") {
177 Some(id) => Ok(Inbound::UnsupportedServerRequest(ServerRequest {
178 method: method.to_owned(),
179 id: RpcId::try_from(id)?,
180 })),
181 None => Ok(Inbound::IgnoredNotification {
182 method: method.to_owned(),
183 }),
184 },
185 }
186}
187fn decode_response(object: &Map<String, Value>) -> Result<Inbound, Error> {
188 let id = object
189 .get("id")
190 .and_then(Value::as_u64)
191 .ok_or_else(|| protocol("app-server response id was not an unsigned integer"))?;
192 let outcome = match (object.get("result"), object.get("error")) {
193 (Some(result), None) => ResponseOutcome::Result(result.clone()),
194 (None, Some(error)) => ResponseOutcome::Error(ServerError::new(error.clone())),
195 (Some(_), Some(_)) => return Err(protocol("app-server response had result and error")),
196 (None, None) => return Err(protocol("app-server response omitted result and error")),
197 };
198 Ok(Inbound::ClientResponse(ClientResponse { id, outcome }))
199}
200fn decode_scoped(method: &str, object: &Map<String, Value>) -> Result<ScopedEvent, Error> {
201 let params = required_object(object, "params", "scoped message omitted params")?;
202 let scope = decode_scope(params)?;
203 let kind = match method {
204 "item/agentMessage/delta" => ScopedKind::AgentTextDelta(
205 required_string(params, "delta", "agent message delta omitted delta")?.to_owned(),
206 ),
207 "item/tool/call" => {
208 let rpc_id = object
209 .get("id")
210 .ok_or_else(|| protocol("dynamic tool request omitted id"))?;
211 let call_id = required_string(params, "callId", "dynamic tool request omitted callId")?;
212 let name = required_string(params, "tool", "dynamic tool request omitted tool")?;
213 let arguments = params
214 .get("arguments")
215 .ok_or_else(|| protocol("dynamic tool request omitted arguments"))?;
216 ScopedKind::DynamicToolCall(DynamicToolCall {
217 rpc_id: RpcId::try_from(rpc_id)?,
218 call_id: call_id.to_owned(),
219 name: name.to_owned(),
220 arguments: arguments.clone(),
221 })
222 }
223 "turn/completed" => {
224 let turn = required_object(params, "turn", "turn/completed omitted turn")?;
225 let status = match required_string(turn, "status", "turn/completed omitted status")? {
226 "completed" => TurnStatus::Completed,
227 "interrupted" => TurnStatus::Interrupted,
228 "failed" => TurnStatus::Failed,
229 _ => return Err(protocol("turn/completed had an invalid status")),
230 };
231 let failure = turn.get("error").cloned().map(ServerError::new);
232 ScopedKind::TurnCompleted(TurnCompleted { status, failure })
233 }
234 "turn/started" => ScopedKind::TurnStarted,
235 "error" => ScopedKind::Error(ServerError::new(
236 params.get("error").unwrap_or(&Value::Null).clone(),
237 )),
238 _ => return Err(protocol("unsupported scoped method")),
239 };
240 Ok(ScopedEvent { scope, kind })
241}
242fn decode_scope(params: &Map<String, Value>) -> Result<Scope, Error> {
243 let thread_id = required_string(params, "threadId", "scoped message omitted threadId")?;
244 let direct = optional_string(params, "turnId", "scoped message turnId must be a string")?;
245 let nested = match params.get("turn") {
246 Some(Value::Object(turn)) => Some(required_string(
247 turn,
248 "id",
249 "scoped message turn omitted id",
250 )?),
251 Some(_) => return Err(protocol("scoped message turn must be an object")),
252 None => None,
253 };
254 let turn_id = match (direct, nested) {
255 (Some(direct), Some(nested)) if direct != nested => {
256 return Err(protocol("scoped message had conflicting turn ids"));
257 }
258 (Some(turn), _) | (None, Some(turn)) => turn,
259 (None, None) => return Err(protocol("scoped message omitted turn id")),
260 };
261 Ok(Scope {
262 thread_id: thread_id.to_owned(),
263 turn_id: turn_id.to_owned(),
264 })
265}
266fn has_scope_fields(params: Option<&Value>) -> bool {
267 params.and_then(Value::as_object).is_some_and(|params| {
268 params.contains_key("threadId")
269 || params.contains_key("turnId")
270 || params.contains_key("turn")
271 })
272}
273fn required_object<'a>(
274 object: &'a Map<String, Value>,
275 field: &str,
276 message: &'static str,
277) -> Result<&'a Map<String, Value>, Error> {
278 object
279 .get(field)
280 .and_then(Value::as_object)
281 .ok_or_else(|| protocol(message))
282}
283fn required_string<'a>(
284 object: &'a Map<String, Value>,
285 field: &str,
286 message: &'static str,
287) -> Result<&'a str, Error> {
288 object
289 .get(field)
290 .and_then(Value::as_str)
291 .ok_or_else(|| protocol(message))
292}
293fn optional_string<'a>(
294 object: &'a Map<String, Value>,
295 field: &str,
296 message: &'static str,
297) -> Result<Option<&'a str>, Error> {
298 object
299 .get(field)
300 .map(|value| value.as_str().ok_or_else(|| protocol(message)))
301 .transpose()
302}
303fn protocol(message: impl Into<String>) -> Error {
304 Error::new(ErrorKind::Protocol, message)
305}
306#[cfg(test)]
307mod tests {
308 use super::*;
309 fn parsed(json: &str) -> Inbound {
310 decode(serde_json::from_str(json).expect("valid test JSON"))
311 }
312 fn malformed(json: &str) {
313 assert!(matches!(
314 parsed(json),
315 Inbound::Malformed(Error {
316 kind: ErrorKind::Protocol,
317 ..
318 })
319 ));
320 }
321 #[test]
322 fn decodes_client_response_outcomes_and_rejects_wrong_ids() {
323 assert_eq!(
324 parsed(r#"{"id":7,"result":{"ok":true}}"#),
325 Inbound::ClientResponse(ClientResponse {
326 id: 7,
327 outcome: ResponseOutcome::Result(json!({"ok": true})),
328 })
329 );
330 assert!(matches!(
331 parsed(r#"{"id":8,"error":{"message":"no"}}"#),
332 Inbound::ClientResponse(ClientResponse {
333 id: 8,
334 outcome: ResponseOutcome::Error(_),
335 })
336 ));
337 for value in [
338 r#"{"id":"7","result":{}}"#,
339 r#"{"id":-1,"result":{}}"#,
340 r#"{"id":1.5,"result":{}}"#,
341 ] {
342 malformed(value);
343 }
344 malformed(r#"{"id":7}"#);
345 malformed(r#"{"id":7,"result":{},"error":{}}"#);
346 }
347 #[test]
348 fn decodes_delta_and_enforces_exact_scope() {
349 assert_eq!(
350 parsed(
351 r#"{"method":"item/agentMessage/delta","params":{"threadId":"th","turnId":"tu","delta":"hi"}}"#
352 ),
353 Inbound::Scoped(ScopedEvent {
354 scope: Scope {
355 thread_id: "th".into(),
356 turn_id: "tu".into()
357 },
358 kind: ScopedKind::AgentTextDelta("hi".into()),
359 })
360 );
361 malformed(r#"{"method":"turn/started","params":{"threadId":1,"turnId":"tu"}}"#);
362 malformed(r#"{"method":"turn/started","params":{"threadId":"th","turnId":1}}"#);
363 malformed(
364 r#"{"method":"turn/started","params":{"threadId":"th","turnId":"a","turn":{"id":"b"}}}"#,
365 );
366 }
367 #[test]
368 fn decodes_tool_calls_and_preserves_rpc_id_kind() {
369 let string = parsed(
370 r#"{"method":"item/tool/call","id":"9","params":{"threadId":"th","turnId":"tu","callId":"c","tool":"search","arguments":{"q":1}}}"#,
371 );
372 let number = parsed(
373 r#"{"method":"item/tool/call","id":9,"params":{"threadId":"th","turnId":"tu","callId":"c2","tool":"search","arguments":{}}}"#,
374 );
375 let Inbound::Scoped(ScopedEvent {
376 kind: ScopedKind::DynamicToolCall(first),
377 ..
378 }) = string
379 else {
380 panic!("tool call expected")
381 };
382 let Inbound::Scoped(ScopedEvent {
383 kind: ScopedKind::DynamicToolCall(second),
384 ..
385 }) = number
386 else {
387 panic!("tool call expected")
388 };
389 assert_ne!(first.rpc_id, second.rpc_id);
390 assert_eq!(first.call_id, "c");
391 assert_eq!(first.name, "search");
392 assert_eq!(first.arguments, json!({"q": 1}));
393 assert_eq!(rejection_payload(&first.rpc_id, -32602, "bad")["id"], "9");
394 malformed(
395 r#"{"method":"item/tool/call","id":null,"params":{"threadId":"th","turnId":"tu","callId":"c","tool":"t","arguments":{}}}"#,
396 );
397 malformed(
398 r#"{"method":"item/tool/call","id":1,"params":{"threadId":"th","turnId":"tu","callId":2,"tool":"t","arguments":{}}}"#,
399 );
400 malformed(
401 r#"{"method":"item/tool/call","id":1,"params":{"threadId":"th","turnId":"tu","callId":"c","tool":"t"}}"#,
402 );
403 }
404 #[test]
405 fn decodes_completion_statuses_and_failures() {
406 for (status, expected) in [
407 ("completed", TurnStatus::Completed),
408 ("interrupted", TurnStatus::Interrupted),
409 ("failed", TurnStatus::Failed),
410 ] {
411 let value = format!(
412 r#"{{"method":"turn/completed","params":{{"threadId":"th","turn":{{"id":"tu","status":"{status}","error":{{"message":"why"}}}}}}}}"#
413 );
414 let Inbound::Scoped(ScopedEvent {
415 kind: ScopedKind::TurnCompleted(completed),
416 ..
417 }) = parsed(&value)
418 else {
419 panic!("completion expected")
420 };
421 assert_eq!(completed.status, expected);
422 assert_eq!(
423 completed.failure.expect("failure").details["message"],
424 "why"
425 );
426 }
427 malformed(
428 r#"{"method":"turn/completed","params":{"threadId":"th","turn":{"id":"tu","status":"other"}}}"#,
429 );
430 malformed(
431 r#"{"method":"turn/completed","params":{"threadId":"th","turn":{"id":"tu","status":1}}}"#,
432 );
433 }
434 #[test]
435 fn decodes_turn_started_scoped_error_and_global_error() {
436 assert!(matches!(
437 parsed(r#"{"method":"turn/started","params":{"threadId":"th","turnId":"tu"}}"#),
438 Inbound::Scoped(ScopedEvent {
439 kind: ScopedKind::TurnStarted,
440 ..
441 })
442 ));
443 assert!(matches!(
444 parsed(
445 r#"{"method":"error","params":{"threadId":"th","turnId":"tu","error":{"message":"scoped"}}}"#
446 ),
447 Inbound::Scoped(ScopedEvent {
448 kind: ScopedKind::Error(_),
449 ..
450 })
451 ));
452 let Inbound::GlobalServerError(error) =
453 parsed(r#"{"method":"error","params":{"message":"global"}}"#)
454 else {
455 panic!("global error expected")
456 };
457 assert_eq!(
458 error.to_error(vec![1]).message,
459 "Codex app-server error: global"
460 );
461 malformed(r#"{"method":"error","params":{"threadId":"th","turnId":3}}"#);
462 }
463 #[test]
464 fn classifies_reroutes_requests_notifications_and_resolutions() {
465 assert!(matches!(
466 parsed(r#"{"method":"model/rerouted","id":"r"}"#),
467 Inbound::ModelReroute(ModelReroute { id: Some(RpcId::String(id)), .. }) if id == "r"
468 ));
469 assert!(matches!(
470 parsed(r#"{"method":"future/request","id":4}"#),
471 Inbound::UnsupportedServerRequest(ServerRequest {
472 id: RpcId::Number(_),
473 ..
474 })
475 ));
476 assert_eq!(
477 parsed(r#"{"method":"serverRequest/resolved","params":{"requestId":"4"}}"#),
478 Inbound::ResolvedRequest(RpcId::String("4".into()))
479 );
480 assert_eq!(
481 parsed(r#"{"method":"future/notification"}"#),
482 Inbound::IgnoredNotification {
483 method: "future/notification".into()
484 }
485 );
486 malformed(r#"{"method":"serverRequest/resolved","params":{"requestId":null}}"#);
487 malformed(r#"{"method":"future/request","id":null}"#);
488 malformed(r#"{"method":1,"id":1}"#);
489 malformed("[]");
490 }
491}