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
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
//! Crate-wide error type for `agent-mesh-bus`.
//!
//! The bus composes failures from the underlying transport (handshake,
//! envelope verification), the application protocol (correlation lost,
//! handler missing), and replay defense (nonce + sequence). One enum
//! surfaces them so callers can match on intent without dragging
//! transport types into their signatures.
use thiserror::Error;
/// Errors that can arise from the high-level bus API.
#[derive(Debug, Error)]
pub enum BusError {
/// The bus's accept loop is no longer running — typically because
/// the endpoint was closed.
#[error("not running")]
NotRunning,
/// No peer has been registered as a subscriber/handler for the
/// named topic. Returned by `publish` when nobody is listening.
#[error("no peer for topic {0}")]
NoSubscriber(String),
/// A peer fingerprint did not resolve via mDNS (or the resolver
/// timed out before the announcement arrived).
#[error("peer not reachable: {0}")]
Unreachable(String),
/// An in-flight request exceeded its caller-supplied timeout.
#[error("request timed out after {0:?}")]
Timeout(std::time::Duration),
/// The waiter for an in-flight request was dropped before the
/// reply arrived. Usually means the bus shut down mid-request.
#[error("reply correlation lost")]
LostReply,
/// An incoming envelope's nonce was already in the cache.
#[error("replay rejected: nonce already seen")]
Replay,
/// An incoming envelope's sequence number was not strictly greater
/// than the highest seen from this peer.
#[error("sequence error: peer={peer_fp}, expected {expected}, got {actual}")]
BadSequence {
/// Sender's agent fingerprint, hex-encoded for log lines.
peer_fp: String,
/// Next sequence number we'd have accepted.
expected: u64,
/// Sequence number that arrived instead.
actual: u64,
},
/// The caller supplied a local user identity that did not issue the local
/// agent's verified certificate chain.
#[error(
"local identity mismatch: supplied user {supplied_user_fp} does not match certified user {certified_user_fp}"
)]
LocalIdentityMismatch {
/// Fingerprint of the explicitly supplied user key.
supplied_user_fp: String,
/// Verified user root from the local agent certificate.
certified_user_fp: String,
},
/// The transport supplied no authenticated peer for an inbound delivery.
#[error("inbound delivery has no authenticated carrier")]
UnboundDelivery,
/// A direct delivery arrived over a carrier session authenticated as a
/// different agent from the envelope's original signer.
#[error(
"direct-delivery carrier agent {carrier_agent_fp} does not match envelope signer {signer_agent_fp}"
)]
CarrierAgentMismatch {
/// Transport-authenticated carrier agent fingerprint.
carrier_agent_fp: String,
/// Cryptographically verified envelope signer fingerprint.
signer_agent_fp: String,
},
/// A direct delivery's carrier and original signer root at different users.
#[error(
"direct-delivery carrier user {carrier_user_fp} does not match envelope signer user {signer_user_fp}"
)]
CarrierUserMismatch {
/// Transport-authenticated carrier user fingerprint.
carrier_user_fp: String,
/// Cryptographically verified envelope signer user fingerprint.
signer_user_fp: String,
},
/// A correctly carrier-bound peer belongs to a different user root from
/// the local bus, violating the current same-user policy.
#[error("peer user {peer_user_fp} does not match local user {local_user_fp}")]
ForeignPeer {
/// Verified peer user fingerprint.
peer_user_fp: String,
/// Local bus user fingerprint.
local_user_fp: String,
},
/// A direct envelope was signed for a different local agent.
#[error("direct envelope recipient {recipient_agent_fp} does not match local agent {local_agent_fp}")]
WrongRecipient {
/// Agent fingerprint named by `Recipient::Direct`.
recipient_agent_fp: String,
/// Agent fingerprint of the receiving bus.
local_agent_fp: String,
},
/// Anything bubbling up from the transport layer (handshake, dial,
/// envelope I/O, auto-team rejection).
#[error("transport: {0}")]
Transport(#[from] agent_mesh_transport::TransportError),
/// A non-Iroh [`crate::Transport`] backend failed. The backend keeps its
/// structured error at its public API boundary; the bus preserves an
/// actionable rendering without depending on every optional carrier
/// crate (which would create dependency cycles).
#[error("transport backend: {0}")]
TransportBackend(String),
/// Anything bubbling up from agent-mesh-protocol (cert verify,
/// envelope decode, etc.).
#[error("core: {0}")]
Core(#[from] agent_mesh_protocol::MeshError),
/// The responder refused to open a session, for the reason it gave.
#[error("session refused: {0}")]
SessionRefused(String),
/// A session stream carried a frame that breaks the session protocol,
/// or was used after it ended. The session is over.
#[error("session protocol: {0}")]
SessionProtocol(String),
/// A session's peer went away without closing it.
#[error("session peer disconnected")]
PeerDisconnected,
/// A bus could not read or durably reserve its envelope sequences
/// ([`crate::sequence::SequenceReservations`]).
#[error("sequence reservation: {0}")]
SequenceReservation(String),
/// JSON encode/decode failure on the bus's `BusMessage` framing.
#[error("json: {0}")]
Json(#[from] serde_json::Error),
}
/// Convenience alias for the crate's `Result` type.
pub type Result<T> = std::result::Result<T, BusError>;
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn replay_renders_useful_message() {
let e = BusError::Replay;
assert!(format!("{e}").contains("replay"));
}
#[test]
fn bad_sequence_includes_peer_and_numbers() {
let e = BusError::BadSequence {
peer_fp: "abc123".into(),
expected: 5,
actual: 3,
};
let msg = format!("{e}");
assert!(msg.contains("abc123"));
assert!(msg.contains('5'));
assert!(msg.contains('3'));
}
#[test]
fn admission_errors_name_carrier_signer_and_local_roles() {
let unbound = BusError::UnboundDelivery;
assert!(format!("{unbound}").contains("authenticated carrier"));
let agent = BusError::CarrierAgentMismatch {
carrier_agent_fp: "carrier".into(),
signer_agent_fp: "signer".into(),
};
let msg = format!("{agent}");
assert!(msg.contains("carrier"));
assert!(msg.contains("signer"));
let foreign = BusError::ForeignPeer {
peer_user_fp: "peer".into(),
local_user_fp: "local".into(),
};
let msg = format!("{foreign}");
assert!(msg.contains("peer"));
assert!(msg.contains("local"));
}
#[test]
fn local_identity_mismatch_names_both_roots() {
let error = BusError::LocalIdentityMismatch {
supplied_user_fp: "supplied".into(),
certified_user_fp: "certified".into(),
};
let message = error.to_string();
assert!(message.contains("supplied"));
assert!(message.contains("certified"));
}
#[test]
fn timeout_renders_duration() {
let e = BusError::Timeout(std::time::Duration::from_secs(2));
assert!(format!("{e}").contains('2'));
}
#[test]
fn transport_error_converts_via_from() {
let t = agent_mesh_transport::TransportError::Iroh("bind failed".into());
let e: BusError = t.into();
assert!(matches!(e, BusError::Transport(_)));
}
#[test]
fn backend_error_preserves_actionable_context() {
let error = BusError::TransportBackend("ssh exited 255: permission denied".into());
let rendered = error.to_string();
assert!(rendered.contains("ssh exited 255"));
assert!(rendered.contains("permission denied"));
}
#[test]
fn core_error_converts_via_from() {
let c = agent_mesh_protocol::MeshError::BadSignature;
let e: BusError = c.into();
assert!(matches!(e, BusError::Core(_)));
}
#[test]
fn json_error_converts_via_from() {
let r: std::result::Result<serde_json::Value, _> = serde_json::from_str("not-json");
let je = r.unwrap_err();
let e: BusError = je.into();
assert!(matches!(e, BusError::Json(_)));
}
}