1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
//! Broker wire protocol: client ↔ broker frames, JSON over WebSocket text.
//!
//! Message payloads reuse `bamboo-subagent`'s [`InboxMessage`] / [`MsgId`] /
//! [`AgentRef`] verbatim — the broker is a transport for those, it does not
//! reinterpret them.
use bamboo_subagent::{AgentRef, InboxMessage, MsgId};
use serde::{Deserialize, Serialize};
/// Client → broker.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ClientFrame {
/// First frame on a connection: authenticate (`token`) and bind this
/// connection to a mailbox key (`agent.session_id`).
Hello { agent: AgentRef, token: String },
/// Durably enqueue `message` into the mailbox of session `to`.
Deliver { to: String, message: InboxMessage },
/// Start receiving this client's own mailbox (push). Backlog (incl. crash
/// leftovers) is delivered first, then new messages as they arrive.
Subscribe,
/// Acknowledge a processed message so the broker deletes it (at-least-once;
/// an unacked message is re-pushed on the next subscribe).
Ack { id: MsgId },
/// Out-of-band cancel: ask the broker to signal session `to` to abort the
/// in-flight run correlated to `correlation_id` (the timed-out ask's id).
/// Ephemeral and fire-and-forget — NOT durable, never enters a mailbox, never
/// acked. A control signal, deliberately off the at-least-once work path so a
/// cancel can never queue behind the very work it cancels. #50.
Cancel { to: String, correlation_id: MsgId },
/// Ask the broker which actors are currently connected serving `role` — the
/// bus IS the live-actor registry (presence is connection-truth). The broker
/// answers with [`BrokerFrame::Connected`]. Replaces the HTTP `/v1/agents`
/// registry discover for schedulable worker selection (Phase 3).
ListConnected { role: String },
}
/// Broker → client.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum BrokerFrame {
/// Handshake accepted.
Welcome,
/// Handshake or request rejected; the broker closes the connection after
/// an auth error.
///
/// `id` correlates the rejection back to the [`ClientFrame::Deliver`]
/// that caused it (e.g. `MailboxFull`) — [`Some`] the message's own
/// [`MsgId`] when the broker rejected a specific `Deliver`, `None` for
/// rejections with nothing to correlate to (bad handshake, a malformed
/// frame). Lets `BrokerClient::deliver` route the rejection back to the
/// waiting caller instead of the caller only ever seeing a generic
/// receipt timeout (review finding on #491/#53).
Error {
reason: String,
#[serde(default)]
id: Option<MsgId>,
},
/// A message pushed from the subscriber's mailbox.
Message { message: InboxMessage },
/// Receipt that a [`ClientFrame::Deliver`] was durably enqueued.
Delivered { id: MsgId },
/// Out-of-band cancel pushed to a live subscriber: abort the in-flight run
/// correlated to `correlation_id`. Ephemeral — never persisted/acked (#50).
Cancel { correlation_id: MsgId },
/// Answer to [`ClientFrame::ListConnected`]: the mailbox ids of every actor
/// currently connected serving the requested role.
Connected { ids: Vec<String> },
}
impl ClientFrame {
pub fn to_text(&self) -> String {
serde_json::to_string(self).unwrap_or_else(|_| "{}".to_string())
}
pub fn from_text(s: &str) -> serde_json::Result<Self> {
serde_json::from_str(s)
}
}
impl BrokerFrame {
pub fn to_text(&self) -> String {
serde_json::to_string(self).unwrap_or_else(|_| "{}".to_string())
}
pub fn from_text(s: &str) -> serde_json::Result<Self> {
serde_json::from_str(s)
}
}
#[cfg(test)]
mod tests {
use super::*;
use bamboo_subagent::{AskBody, AskMode, InboxKind};
use chrono::Utc;
fn ask_msg() -> InboxMessage {
InboxMessage {
id: MsgId::new(),
from: AgentRef {
session_id: "parent".into(),
role: None,
},
kind: InboxKind::Ask,
body: serde_json::to_value(AskBody {
question: "status?".into(),
mode: AskMode::Query,
})
.unwrap(),
created_at: Utc::now(),
correlation_id: None,
}
}
#[test]
fn client_frames_round_trip() {
let frames = [
ClientFrame::Hello {
agent: AgentRef {
session_id: "p".into(),
role: Some("root".into()),
},
token: "t".into(),
},
ClientFrame::Deliver {
to: "child".into(),
message: ask_msg(),
},
ClientFrame::Subscribe,
ClientFrame::Ack { id: MsgId::new() },
ClientFrame::Cancel {
to: "child".into(),
correlation_id: MsgId::new(),
},
ClientFrame::ListConnected {
role: "gpu-pool".into(),
},
];
for f in frames {
assert_eq!(ClientFrame::from_text(&f.to_text()).unwrap(), f);
}
// tag stability
let v: serde_json::Value = serde_json::from_str(&ClientFrame::Subscribe.to_text()).unwrap();
assert_eq!(v["kind"], "subscribe");
let c: serde_json::Value = serde_json::from_str(
&ClientFrame::Cancel {
to: "child".into(),
correlation_id: MsgId::new(),
}
.to_text(),
)
.unwrap();
assert_eq!(c["kind"], "cancel");
}
#[test]
fn broker_frames_round_trip() {
let frames = [
BrokerFrame::Welcome,
BrokerFrame::Error {
reason: "bad token".into(),
id: None,
},
BrokerFrame::Error {
reason: "mailbox 'child' is full (2 pending messages)".into(),
id: Some(MsgId::new()),
},
BrokerFrame::Message { message: ask_msg() },
BrokerFrame::Delivered { id: MsgId::new() },
BrokerFrame::Cancel {
correlation_id: MsgId::new(),
},
BrokerFrame::Connected {
ids: vec!["w-1".into(), "w-2".into()],
},
];
for f in frames {
assert_eq!(BrokerFrame::from_text(&f.to_text()).unwrap(), f);
}
}
/// A legacy `Error` frame serialized without the `id` field (as every
/// `Error` was before this correlation id existed) still parses, with
/// `id` defaulting to `None` — so an older broker (or a captured/replayed
/// frame) never fails a newer client's deserialization.
#[test]
fn error_frame_without_id_defaults_to_none() {
let legacy = serde_json::json!({ "kind": "error", "reason": "bad token" });
let parsed: BrokerFrame = serde_json::from_value(legacy).unwrap();
assert_eq!(
parsed,
BrokerFrame::Error {
reason: "bad token".into(),
id: None,
}
);
}
}