use serde::{Deserialize, Deserializer, Serialize, Serializer};
use serde_json::{Map, Value};
use super::{
AssistantMessage, CommandLifecycle, ControlCancelRequest, ControlRequest, ControlResponse,
Message, RateLimitEvent, ResultMessage, StreamEvent, System, UserMessage, from_fields, str_of,
tagged,
};
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
pub enum Inbound {
User(UserMessage),
Assistant(AssistantMessage),
System(System),
Result(ResultMessage),
StreamEvent(StreamEvent),
RateLimitEvent(RateLimitEvent),
CommandLifecycle(CommandLifecycle),
ControlRequest(ControlRequest),
ControlResponse(ControlResponse),
ControlCancelRequest(ControlCancelRequest),
Other {
type_: String,
raw: Value,
},
Malformed {
type_: String,
error: String,
raw: Value,
},
}
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ParseError {
#[error("invalid JSON: {0}")]
Json(#[from] serde_json::Error),
#[error("not a JSON object")]
NotAnObject,
}
#[derive(Debug, Clone, PartialEq)]
#[non_exhaustive]
#[expect(
clippy::large_enum_variant,
reason = "one per line read, moved once: boxing would only add an allocation"
)]
pub enum Control {
Request(ControlRequest),
Response(ControlResponse),
CancelRequest(ControlCancelRequest),
}
#[derive(Debug, Clone, PartialEq)]
#[allow(
clippy::large_enum_variant,
reason = "one per line read, moved once; and its size depends on serde_json's features \
(a build with `preserve_order` makes it lopsided), so not an `expect`"
)]
pub enum Frame {
Message(Message),
Control(Control),
}
impl Inbound {
pub fn from_line(line: &str) -> Result<Self, ParseError> {
Self::from_value(serde_json::from_str(line.trim())?)
}
pub fn from_value(v: Value) -> Result<Self, ParseError> {
match v {
Value::Object(obj) => Ok(Self::from_object(obj)),
_ => Err(ParseError::NotAnObject),
}
}
pub fn from_object(obj: Map<String, Value>) -> Self {
let Some(type_) = str_of(&obj, "type") else {
super::stats::other("");
return Self::Other {
type_: String::new(),
raw: Value::Object(obj),
};
};
let skip = &["type"];
let parsed = match type_ {
"user" => from_fields(&obj, skip).map(Self::User),
"assistant" => from_fields(&obj, skip).map(Self::Assistant),
"system" => System::from_object(&obj).map(Self::System),
"result" => from_fields(&obj, skip).map(Self::Result),
"stream_event" => from_fields(&obj, skip).map(Self::StreamEvent),
"rate_limit_event" => from_fields(&obj, skip).map(Self::RateLimitEvent),
"command_lifecycle" => from_fields(&obj, skip).map(Self::CommandLifecycle),
"control_request" => from_fields(&obj, skip).map(Self::ControlRequest),
"control_response" => from_fields(&obj, skip).map(Self::ControlResponse),
"control_cancel_request" => from_fields(&obj, skip).map(Self::ControlCancelRequest),
_ => {
super::stats::other(type_);
return Self::Other {
type_: type_.to_owned(),
raw: Value::Object(obj),
};
}
};
match parsed {
Ok(v) => v,
Err(e) => {
let error = e.to_string();
super::stats::malformed(type_, &error);
Self::Malformed {
type_: type_.to_owned(),
error,
raw: Value::Object(obj),
}
}
}
}
pub fn type_name(&self) -> &str {
match self {
Self::User(_) => "user",
Self::Assistant(_) => "assistant",
Self::System(_) => "system",
Self::Result(_) => "result",
Self::StreamEvent(_) => "stream_event",
Self::RateLimitEvent(_) => "rate_limit_event",
Self::CommandLifecycle(_) => "command_lifecycle",
Self::ControlRequest(_) => "control_request",
Self::ControlResponse(_) => "control_response",
Self::ControlCancelRequest(_) => "control_cancel_request",
Self::Other { type_, .. } | Self::Malformed { type_, .. } => type_,
}
}
pub fn ends_turn(&self) -> bool {
match self {
Self::Result(_) => true,
Self::Malformed { type_, .. } => type_ == "result",
_ => false,
}
}
pub fn to_value(&self) -> Result<Value, serde_json::Error> {
let t = self.type_name();
match self {
Self::User(m) => tagged("type", t, m),
Self::Assistant(m) => tagged("type", t, m),
Self::System(m) => tagged("type", t, m),
Self::Result(m) => tagged("type", t, m),
Self::StreamEvent(m) => tagged("type", t, m),
Self::RateLimitEvent(m) => tagged("type", t, m),
Self::CommandLifecycle(m) => tagged("type", t, m),
Self::ControlRequest(m) => tagged("type", t, m),
Self::ControlResponse(m) => tagged("type", t, m),
Self::ControlCancelRequest(m) => tagged("type", t, m),
Self::Other { raw, .. } | Self::Malformed { raw, .. } => Ok(raw.clone()),
}
}
pub fn split(self) -> Frame {
match self {
Self::User(m) => Frame::Message(Message::User(m)),
Self::Assistant(m) => Frame::Message(Message::Assistant(m)),
Self::System(m) => Frame::Message(Message::System(m)),
Self::Result(m) => Frame::Message(Message::Result(m)),
Self::StreamEvent(m) => Frame::Message(Message::StreamEvent(m)),
Self::RateLimitEvent(m) => Frame::Message(Message::RateLimitEvent(m)),
Self::CommandLifecycle(m) => Frame::Message(Message::CommandLifecycle(m)),
Self::Other { type_, raw } => Frame::Message(Message::Other { type_, raw }),
Self::Malformed { type_, error, raw } => {
Frame::Message(Message::Malformed { type_, error, raw })
}
Self::ControlRequest(c) => Frame::Control(Control::Request(c)),
Self::ControlResponse(c) => Frame::Control(Control::Response(c)),
Self::ControlCancelRequest(c) => Frame::Control(Control::CancelRequest(c)),
}
}
}
impl TryFrom<Inbound> for Message {
type Error = Control;
fn try_from(i: Inbound) -> Result<Self, Control> {
match i.split() {
Frame::Message(m) => Ok(m),
Frame::Control(c) => Err(c),
}
}
}
impl From<Message> for Inbound {
fn from(m: Message) -> Self {
match m {
Message::User(m) => Self::User(m),
Message::Assistant(m) => Self::Assistant(m),
Message::System(m) => Self::System(m),
Message::Result(m) => Self::Result(m),
Message::StreamEvent(m) => Self::StreamEvent(m),
Message::RateLimitEvent(m) => Self::RateLimitEvent(m),
Message::CommandLifecycle(m) => Self::CommandLifecycle(m),
Message::Other { type_, raw } => Self::Other { type_, raw },
Message::Malformed { type_, error, raw } => Self::Malformed { type_, error, raw },
}
}
}
impl From<Control> for Inbound {
fn from(c: Control) -> Self {
match c {
Control::Request(c) => Self::ControlRequest(c),
Control::Response(c) => Self::ControlResponse(c),
Control::CancelRequest(c) => Self::ControlCancelRequest(c),
}
}
}
impl From<Frame> for Inbound {
fn from(f: Frame) -> Self {
match f {
Frame::Message(m) => m.into(),
Frame::Control(c) => c.into(),
}
}
}
impl Serialize for Inbound {
fn serialize<S: Serializer>(&self, s: S) -> Result<S::Ok, S::Error> {
self.to_value()
.map_err(serde::ser::Error::custom)?
.serialize(s)
}
}
impl<'de> Deserialize<'de> for Inbound {
fn deserialize<D: Deserializer<'de>>(d: D) -> Result<Self, D::Error> {
let obj = super::object(d, "Inbound")?;
Ok(Self::from_object(obj))
}
}
impl Message {
pub fn ends_turn(&self) -> bool {
match self {
Self::Result(_) => true,
Self::Malformed { type_, .. } => type_ == "result",
_ => false,
}
}
pub fn type_name(&self) -> &str {
match self {
Self::User(_) => "user",
Self::Assistant(_) => "assistant",
Self::System(_) => "system",
Self::Result(_) => "result",
Self::StreamEvent(_) => "stream_event",
Self::RateLimitEvent(_) => "rate_limit_event",
Self::CommandLifecycle(_) => "command_lifecycle",
Self::Other { type_, .. } | Self::Malformed { type_, .. } => type_,
}
}
}