use serde::{Deserialize, Deserializer, Serialize};
/// communication lanes separating distinct semantic streams
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Lane {
Scratchpad,
Cell,
Speech,
Input,
Lifecycle,
}
impl Lane {
pub const fn as_str(&self) -> &'static str {
match self {
Self::Scratchpad => "scratchpad",
Self::Cell => "cell",
Self::Speech => "speech",
Self::Input => "input",
Self::Lifecycle => "lifecycle",
}
}
}
/// origin provenance for diagnostics
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OriginKind {
Human,
Agent,
Checker,
System,
External,
}
/// capability health status
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CapabilityHealthStatus {
Healthy,
Degraded,
Unavailable,
Unknown,
Stale,
}
/// work item scheduling status
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkItemStatus {
Pending,
Selected,
Deferred,
Executing,
Completed,
Failed,
Cancelled,
Claimed,
}
/// capability health observation
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CapabilityHealthObservation {
pub id: String,
pub name: String,
pub scope: String,
pub status: CapabilityHealthStatus,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub evidence: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub impact: Option<String>,
pub observed_ms: i64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub valid_until_ms: Option<i64>,
pub origin: OriginKind,
}
/// work item diagnostic
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct WorkItemDiagnostic {
pub id: String,
pub title: String,
pub purpose: String,
pub status: WorkItemStatus,
pub origin: OriginKind,
pub created_ms: i64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub selection_reason: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub deferred_reason: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub dependencies: Option<Vec<String>>,
}
/// candidate revision diagnostic
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct CandidateRevisionDiagnostic {
pub candidate_id: String,
pub base_revision: String,
pub target_revision: String,
pub status: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub epoch: Option<i64>,
}
/// combined diagnostics payload for lifecycle events
#[derive(Debug, Clone, PartialEq, Default, Serialize, Deserialize)]
pub struct DiagnosticsPayload {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub health: Vec<CapabilityHealthObservation>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub work_items: Vec<WorkItemDiagnostic>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub candidate: Option<CandidateRevisionDiagnostic>,
}
impl DiagnosticsPayload {
pub fn is_empty(&self) -> bool {
self.health.is_empty() && self.work_items.is_empty() && self.candidate.is_none()
}
}
/// Model scratchpad prose with provider reasoning carried separately.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ScratchpadPayload {
pub text: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub reasoning: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub execution_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub delta: Option<bool>,
}
/// status of cell execution
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CellState {
Ok,
Error,
Yielded,
Interrupted,
}
/// cell stream channels
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum CellStreamKind {
Stdout,
Stderr,
Result,
}
/// diagnostic or output chunk produced by code cell execution
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CellStreamOutput {
pub stream: CellStreamKind,
pub text: String,
}
/// cell execution payload (started or completed)
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum CellPayload {
Started {
execution_id: String,
generation: String,
revision: String,
code: String,
},
Completed {
execution_id: String,
state: CellState,
#[serde(default)]
output: Vec<CellStreamOutput>,
#[serde(default)]
truncated_bytes: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
error: Option<String>,
},
}
/// speech / external effect payload (sent or held)
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum SpeechPayload {
Sent {
effect_id: String,
text: String,
event_id: i64,
status: String,
},
Held {
effect_id: String,
text: String,
status: String,
reason: String,
},
}
/// user or external interrupt input payload
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct InputPayload {
pub source: String,
pub text: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub author: Option<String>,
}
/// lifecycle events (kernel restart, compaction, session start, etc.)
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct LifecyclePayload {
pub kind: String,
#[serde(default)]
pub data: serde_json::Value,
}
impl LifecyclePayload {
/// create a lifecycle payload with diagnostics
pub fn diagnostics(kind: impl Into<String>, diag: DiagnosticsPayload) -> Self {
Self {
kind: kind.into(),
data: serde_json::to_value(diag).unwrap_or_default(),
}
}
/// create a capability.checked lifecycle payload
pub fn capability_checked(diag: DiagnosticsPayload) -> Self {
Self::diagnostics("capability.checked", diag)
}
/// create a health update lifecycle payload
pub fn health_update(health: Vec<CapabilityHealthObservation>) -> Self {
Self::diagnostics(
"capability.health",
DiagnosticsPayload {
health,
..Default::default()
},
)
}
/// create a work update lifecycle payload
pub fn work_update(work_items: Vec<WorkItemDiagnostic>) -> Self {
Self::diagnostics(
"work.updated",
DiagnosticsPayload {
work_items,
..Default::default()
},
)
}
/// create a candidate update lifecycle payload
pub fn candidate_update(candidate: Option<CandidateRevisionDiagnostic>) -> Self {
Self::diagnostics(
"candidate.updated",
DiagnosticsPayload {
candidate,
..Default::default()
},
)
}
}
/// union of lane payloads
#[derive(Debug, Clone, PartialEq, Serialize)]
#[serde(untagged)]
pub enum LanePayload {
Scratchpad(ScratchpadPayload),
Cell(CellPayload),
Speech(SpeechPayload),
Input(InputPayload),
Lifecycle(LifecyclePayload),
}
impl LanePayload {
pub const fn lane(&self) -> Lane {
match self {
Self::Scratchpad(_) => Lane::Scratchpad,
Self::Cell(_) => Lane::Cell,
Self::Speech(_) => Lane::Speech,
Self::Input(_) => Lane::Input,
Self::Lifecycle(_) => Lane::Lifecycle,
}
}
}
/// sequence-stamped event record representing one immutable event in a lane
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct V2EventRecord {
pub seq: i64,
pub session_id: String,
pub lane: Lane,
pub payload: LanePayload,
pub created_ms: i64,
}
#[derive(Deserialize)]
struct RawEventRecord {
seq: i64,
session_id: String,
lane: Lane,
payload: serde_json::Value,
created_ms: i64,
}
impl<'de> Deserialize<'de> for V2EventRecord {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let raw = RawEventRecord::deserialize(deserializer)?;
let payload = match raw.lane {
Lane::Scratchpad => serde_json::from_value(raw.payload)
.map(LanePayload::Scratchpad)
.map_err(serde::de::Error::custom)?,
Lane::Cell => serde_json::from_value(raw.payload)
.map(LanePayload::Cell)
.map_err(serde::de::Error::custom)?,
Lane::Speech => serde_json::from_value(raw.payload)
.map(LanePayload::Speech)
.map_err(serde::de::Error::custom)?,
Lane::Input => serde_json::from_value(raw.payload)
.map(LanePayload::Input)
.map_err(serde::de::Error::custom)?,
Lane::Lifecycle => serde_json::from_value(raw.payload)
.map(LanePayload::Lifecycle)
.map_err(serde::de::Error::custom)?,
};
Ok(Self {
seq: raw.seq,
session_id: raw.session_id,
lane: raw.lane,
payload,
created_ms: raw.created_ms,
})
}
}
impl V2EventRecord {
pub fn new(
seq: i64,
session_id: impl Into<String>,
payload: LanePayload,
created_ms: i64,
) -> Self {
Self {
seq,
session_id: session_id.into(),
lane: payload.lane(),
payload,
created_ms,
}
}
pub fn scratchpad(
seq: i64,
session_id: impl Into<String>,
payload: ScratchpadPayload,
created_ms: i64,
) -> Self {
Self::new(
seq,
session_id,
LanePayload::Scratchpad(payload),
created_ms,
)
}
pub fn cell(
seq: i64,
session_id: impl Into<String>,
payload: CellPayload,
created_ms: i64,
) -> Self {
Self::new(seq, session_id, LanePayload::Cell(payload), created_ms)
}
pub fn speech(
seq: i64,
session_id: impl Into<String>,
payload: SpeechPayload,
created_ms: i64,
) -> Self {
Self::new(seq, session_id, LanePayload::Speech(payload), created_ms)
}
pub fn input(
seq: i64,
session_id: impl Into<String>,
payload: InputPayload,
created_ms: i64,
) -> Self {
Self::new(seq, session_id, LanePayload::Input(payload), created_ms)
}
pub fn lifecycle(
seq: i64,
session_id: impl Into<String>,
payload: LifecyclePayload,
created_ms: i64,
) -> Self {
Self::new(seq, session_id, LanePayload::Lifecycle(payload), created_ms)
}
pub fn as_scratchpad(&self) -> Option<&ScratchpadPayload> {
match &self.payload {
LanePayload::Scratchpad(p) => Some(p),
_ => None,
}
}
pub fn as_cell(&self) -> Option<&CellPayload> {
match &self.payload {
LanePayload::Cell(p) => Some(p),
_ => None,
}
}
pub fn as_speech(&self) -> Option<&SpeechPayload> {
match &self.payload {
LanePayload::Speech(p) => Some(p),
_ => None,
}
}
pub fn as_input(&self) -> Option<&InputPayload> {
match &self.payload {
LanePayload::Input(p) => Some(p),
_ => None,
}
}
pub fn as_lifecycle(&self) -> Option<&LifecyclePayload> {
match &self.payload {
LanePayload::Lifecycle(p) => Some(p),
_ => None,
}
}
}
/// client -> daemon v2 messages
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum V2ClientMsg {
/// A subscription, and the only way to ask for history.
///
/// `after_seq: None` is a live-only subscription: no replay, the broadcast alone is the answer.
/// `Some(0)` is "I hold nothing": the daemon answers with the newest window it serves, bounded by
/// the session's context fold and — when `limit` is given — by that many records. A daemon that does
/// not know about windows ignores `limit` and replays the session, which is what it did before.
/// `Some(n)` is a client catching up: it gets every record after `n` and no cap.
Subscribe {
session_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
after_seq: Option<i64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
limit: Option<usize>,
},
/// The records behind a window the client already holds: one page of the past, asked for by a reader
/// who pressed for it. Answered with the same replay shape a subscribe is.
History {
session_id: String,
before_seq: i64,
limit: usize,
},
SendInput {
session_id: String,
source: String,
text: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
client_request_id: Option<String>,
},
Interrupt {
session_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
execution_id: Option<String>,
},
ResetKernel {
session_id: String,
},
Compact {
session_id: String,
},
Stop {
session_id: String,
},
Drain {
session_id: String,
},
Activate {
session_id: String,
expected_epoch: i64,
release_path: String,
},
SessionStatus {
session_id: String,
},
SessionList,
Ack {
session_id: String,
seq: i64,
},
Ping {
#[serde(default, skip_serializing_if = "Option::is_none")]
timestamp: Option<i64>,
},
}
/// daemon -> client v2 messages
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum V2ServerMsg {
/// The range a replay is sending, and what the session's history looks like under it.
///
/// The bounds arrive with the range rather than at the end of it: a client that learned about the
/// history behind its window only when the replay finished would grow a control at the top of a
/// transcript the reader is already reading.
SyncStarted {
session_id: String,
from_seq: i64,
to_seq: i64,
/// The oldest record the daemon can still serve. Absent means it cannot serve a page at all.
#[serde(default, skip_serializing_if = "Option::is_none")]
earliest_seq: Option<i64>,
/// The record the session's active context fold drained through, when it has one: everything at
/// or below it is carried by the fold's summary rather than replayed, which is why a window
/// starts above it.
#[serde(default, skip_serializing_if = "Option::is_none")]
fold_seq: Option<i64>,
},
Event {
record: V2EventRecord,
},
SyncCompleted {
session_id: String,
latest_seq: i64,
/// The oldest record the daemon can still serve, repeated here so a client that missed the start
/// of the replay still learns what it can ask for.
#[serde(default, skip_serializing_if = "Option::is_none")]
earliest_seq: Option<i64>,
},
LiveChunk {
session_id: String,
lane: Lane,
#[serde(default, skip_serializing_if = "Option::is_none")]
execution_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
stream: Option<String>,
delta: String,
},
Ack {
session_id: String,
seq: i64,
#[serde(default, skip_serializing_if = "Option::is_none")]
client_request_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
duplicate: Option<bool>,
},
Pong {
#[serde(default, skip_serializing_if = "Option::is_none")]
timestamp: Option<i64>,
},
Error {
code: String,
message: String,
},
OperatorAck {
action: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
session_id: Option<String>,
success: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
message: Option<String>,
},
SessionStatus {
session_id: String,
state: serde_json::Value,
},
SessionList {
sessions: Vec<(String, serde_json::Value)>,
},
}
/// The window frames, pinned to the bytes a client parses them from.
///
/// This boundary is crossed by a JavaScript client that reads these exact field names, and both sides are
/// otherwise free to drift in silence: a rename here is a console that shows an empty session, with
/// nothing red anywhere. The shapes below are asserted as JSON, and the console's own suite replays the
/// same frames through its parser.
#[cfg(test)]
mod window_frames {
use super::*;
use serde_json::json;
#[test]
fn a_range_arrives_with_its_floor_and_its_fold() {
let started = V2ServerMsg::SyncStarted {
session_id: "default".into(),
from_seq: 26,
to_seq: 31,
earliest_seq: Some(1),
fold_seq: Some(25),
};
assert_eq!(
serde_json::to_value(&started).unwrap(),
json!({
"type": "sync_started",
"session_id": "default",
"from_seq": 26,
"to_seq": 31,
"earliest_seq": 1,
"fold_seq": 25,
})
);
// A daemon that cannot page says nothing rather than saying zero: an absent floor is what tells a
// client not to offer to load more.
let no_page = V2ServerMsg::SyncCompleted {
session_id: "default".into(),
latest_seq: 31,
earliest_seq: None,
};
assert_eq!(
serde_json::to_value(&no_page).unwrap(),
json!({ "type": "sync_completed", "session_id": "default", "latest_seq": 31 })
);
}
#[test]
fn a_window_request_and_a_page_read_the_way_a_client_writes_them() {
let subscribe = V2ClientMsg::Subscribe {
session_id: "default".into(),
after_seq: Some(0),
limit: Some(2000),
};
assert_eq!(
serde_json::to_value(&subscribe).unwrap(),
json!({ "type": "subscribe", "session_id": "default", "after_seq": 0, "limit": 2000 })
);
// No cursor at all is a live-only subscription, and it stays that way.
let live_only = V2ClientMsg::Subscribe {
session_id: "default".into(),
after_seq: None,
limit: None,
};
assert_eq!(
serde_json::to_value(&live_only).unwrap(),
json!({ "type": "subscribe", "session_id": "default" })
);
let page = V2ClientMsg::History {
session_id: "default".into(),
before_seq: 26,
limit: 500,
};
assert_eq!(
serde_json::to_value(&page).unwrap(),
json!({ "type": "history", "session_id": "default", "before_seq": 26, "limit": 500 })
);
// And the same shapes are read back: the wire is symmetric, so a logged frame is a usable request.
assert_eq!(
serde_json::from_value::<V2ClientMsg>(serde_json::to_value(&subscribe).unwrap())
.unwrap(),
subscribe
);
assert_eq!(
serde_json::from_value::<V2ClientMsg>(serde_json::to_value(&page).unwrap()).unwrap(),
page
);
}
}