Skip to repository content2988 lines · 124.8 KB · rust
tenant.openagents/omega
No repository description is available.
OpenAgents Git authority 2026-07-28T04:14:17.301Z Public web read
NIP-34 coordinate
30617:7649603503856e5148d571eac2766b288a8ff1e9e35d380337a1d2b0015b4f92:omegaMaintainersHidden in public view
References2 branches · 1 tag
Read-only clone
git clone https://openagents.com/git/tenant.openagents/omega.gitBrowse files
nostr_websocket_relay.rs
1use std::collections::{BTreeMap, HashSet};
2use std::sync::Arc;
3use std::time::{Duration, Instant};
4
5use async_io::Timer;
6use async_tungstenite::WebSocketStream;
7use async_tungstenite::async_std::{ConnectStream, connect_async};
8use async_tungstenite::tungstenite::Message;
9use futures::{FutureExt, StreamExt, pin_mut, select};
10use nostr::{Event, EventBuilder, JsonUtil, Kind, PublicKey, RelayUrl};
11#[cfg(any(test, feature = "test-support"))]
12use nostr::Keys;
13#[cfg(any(test, feature = "test-support"))]
14use nostr::nips::nip59;
15
16#[cfg(any(test, feature = "test-support"))]
17async fn nip59_extract_rumor(
18 keys: &Keys,
19 event: &Event,
20) -> Result<nip59::UnwrappedGift, nostr::nips::nip59::Error> {
21 nip59::extract_rumor(keys, event).await
22}
23use omega_identity::{
24 AdmittedSigningRequest, IdentityService, ReceiptRef, SigningPurpose, UnsignedEventTemplate,
25};
26use serde_json::{Value, json};
27use sha2::{Digest, Sha256};
28
29use crate::sarah_conversation::{
30 ConnectionState, GapState, QueryPage, RelayAuthChallenge, RelayTransport,
31 SarahConversationError, StoredConversationEvent,
32};
33use crate::{
34 ISSUE31_COMMAND_SCHEMA, ISSUE31_COMMAND_SCHEMA_V2, ISSUE31_HOST_DISCOVERY_KIND,
35 ISSUE31_HOST_DISCOVERY_SCHEMA_V2, ISSUE31_OWNER_PROJECTION_SCHEMA, ISSUE31_PAIRING_SCHEMA,
36 Issue31CommandRecord, Issue31CommandRecordV2, Issue31HostDiscovery, Issue31HostDiscoveryV2,
37 Issue31OwnerProjectionRecord, Issue31PairingRecord,
38};
39
40const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
41const NETWORK_TIMEOUT: Duration = Duration::from_secs(8);
42const QUERY_TIMEOUT: Duration = Duration::from_secs(12);
43const QUERY_OPERATION_TIMEOUT: Duration = Duration::from_secs(30);
44const MAX_CONNECT_ROUNDS: usize = 2;
45const INITIAL_BACKOFF: Duration = Duration::from_millis(100);
46const MAX_BACKOFF: Duration = Duration::from_millis(800);
47const MAX_PENDING_PUBLICATIONS: usize = 4_096;
48const MAX_CACHED_EVENTS: usize = 8_192;
49const MAX_CACHED_CONTENT_BYTES: usize = 16 * 1024 * 1024;
50const MAX_CONTROL_FRAMES_PER_READ: usize = 64;
51
52#[allow(clippy::disallowed_methods)]
53fn network_timer(duration: Duration) -> Timer {
54 // The standalone omega-effectd child has no GPUI application context to supply a timer.
55 Timer::after(duration)
56}
57
58const NIP_AE_KIND: u16 = 30174;
59const NIP_RS_KIND: u16 = 30078;
60const NIP_ER_KIND: u16 = 30300;
61const NIP_29_GROUP_CHAT_KIND: u16 = 9;
62const LBR_AGENTIC_CODING_REQUEST_KIND: u16 = 5934;
63const LBR_AGENTIC_CODING_RESULT_KIND: u16 = 6934;
64const LBR_FEEDBACK_KIND: u16 = 7000;
65const LBR_KINDS: &[u16] = &[
66 LBR_AGENTIC_CODING_REQUEST_KIND,
67 LBR_AGENTIC_CODING_RESULT_KIND,
68 LBR_FEEDBACK_KIND,
69];
70const ISSUE31_COMMUNITY_KINDS: &[u16] = &[
71 NIP_29_GROUP_CHAT_KIND,
72 LBR_AGENTIC_CODING_REQUEST_KIND,
73 LBR_AGENTIC_CODING_RESULT_KIND,
74 LBR_FEEDBACK_KIND,
75];
76const SARAH_RECORD_KINDS: &[u16] = &[24200, 44200, 44300, 44301];
77
78enum IncomingMessage {
79 Json(Value),
80 TimedOut,
81}
82
83/// A bounded, failover-capable NIP-01 relay transport for the Sarah workroom.
84///
85/// The surrounding effectd protocol is synchronous today, so each operation
86/// drives the non-blocking socket to a bounded completion. Relay `OK` messages
87/// are transport acknowledgements only. They are never projected as command
88/// completion.
89pub struct WebSocketRelayAdapter {
90 relay_urls: Vec<String>,
91 active_relay_index: usize,
92 active_label: String,
93 socket: Option<WebSocketStream<ConnectStream>>,
94 custody: RelayCustody,
95 owner_public_key_hex: String,
96 sarah_public_key_hex: String,
97 community_group_ids: Vec<String>,
98 community_public_key_hexes: Vec<String>,
99 authenticated: bool,
100 auth_challenge: Option<String>,
101 acknowledged_event_id: Option<String>,
102 publish_acknowledgements: BTreeMap<String, HashSet<String>>,
103 healthy_relays: HashSet<String>,
104 events: BTreeMap<String, StoredConversationEvent>,
105 subscription_sequence: u64,
106 gap_state: GapState,
107 event_cache_truncated: bool,
108}
109
110enum RelayCustody {
111 Omega(Arc<IdentityService>),
112 #[cfg(any(test, feature = "test-support"))]
113 Keys(Keys),
114}
115
116impl WebSocketRelayAdapter {
117 pub fn new(
118 relay_urls: Vec<String>,
119 identity_service: Arc<IdentityService>,
120 sarah_public_key_hex: String,
121 community_group_ids: Vec<String>,
122 community_public_key_hexes: Vec<String>,
123 ) -> Result<Self, SarahConversationError> {
124 let owner_public_key_hex = identity_service
125 .inspect()
126 .map_err(|error| SarahConversationError::Identity(error.to_string()))?
127 .identity
128 .map(|identity| identity.public_key_hex().as_str().to_string())
129 .ok_or(SarahConversationError::IdentityRequired)?;
130 Self::with_custody(
131 relay_urls,
132 RelayCustody::Omega(identity_service),
133 owner_public_key_hex,
134 sarah_public_key_hex,
135 community_group_ids,
136 community_public_key_hexes,
137 )
138 }
139
140 /// A relay adapter whose owner key is supplied directly.
141 ///
142 /// Production custody goes through `new`, which reads the key from
143 /// `omega_identity::IdentityService`. This exists for proof harnesses that
144 /// must run without the GPUI app, and is compiled only under `test` or the
145 /// explicit `test-support` feature so it can never reach a shipped build.
146 #[cfg(any(test, feature = "test-support"))]
147 pub fn new_for_keys(
148 relay_urls: Vec<String>,
149 keys: Keys,
150 ) -> Result<Self, SarahConversationError> {
151 let public_key_hex = keys.public_key().to_hex();
152 Self::with_custody(
153 relay_urls,
154 RelayCustody::Keys(keys),
155 public_key_hex.clone(),
156 public_key_hex,
157 Vec::new(),
158 Vec::new(),
159 )
160 }
161
162 /// The same headless adapter, with Sarah and community policy stated.
163 #[cfg(any(test, feature = "test-support"))]
164 pub fn new_for_keys_with_policy(
165 relay_urls: Vec<String>,
166 owner_keys: Keys,
167 sarah_public_key_hex: String,
168 community_group_ids: Vec<String>,
169 community_public_key_hexes: Vec<String>,
170 ) -> Result<Self, SarahConversationError> {
171 let owner_public_key_hex = owner_keys.public_key().to_hex();
172 Self::with_custody(
173 relay_urls,
174 RelayCustody::Keys(owner_keys),
175 owner_public_key_hex,
176 sarah_public_key_hex,
177 community_group_ids,
178 community_public_key_hexes,
179 )
180 }
181
182 fn with_custody(
183 relay_urls: Vec<String>,
184 custody: RelayCustody,
185 owner_public_key_hex: String,
186 sarah_public_key_hex: String,
187 community_group_ids: Vec<String>,
188 community_public_key_hexes: Vec<String>,
189 ) -> Result<Self, SarahConversationError> {
190 let mut unique = HashSet::new();
191 let mut normalized_relay_urls = Vec::new();
192 for relay_url in relay_urls {
193 let relay_url = relay_url.trim();
194 if relay_url.is_empty() {
195 continue;
196 }
197 let relay_url = normalize_relay_url(relay_url)?;
198 if unique.insert(relay_url.clone()) {
199 normalized_relay_urls.push(relay_url);
200 }
201 }
202 let relay_urls = normalized_relay_urls;
203 if relay_urls.is_empty() {
204 return Err(SarahConversationError::InvalidRequest(
205 "at least one relay URL is required".into(),
206 ));
207 }
208 if relay_urls.len() > 8 {
209 return Err(SarahConversationError::InvalidRequest(
210 "at most eight relay URLs are supported".into(),
211 ));
212 }
213 if PublicKey::from_hex(&owner_public_key_hex).is_err()
214 || PublicKey::from_hex(&sarah_public_key_hex).is_err()
215 {
216 return Err(SarahConversationError::InvalidRequest(
217 "relay subscription requires valid owner and Sarah public keys".into(),
218 ));
219 }
220 let community_group_ids = normalized_group_ids(community_group_ids)?;
221 let community_public_key_hexes =
222 normalized_public_keys(community_public_key_hexes, "community author")?;
223 let active_label = relay_urls
224 .first()
225 .cloned()
226 .ok_or_else(|| SarahConversationError::Internal("relay list disappeared".into()))?;
227 Ok(Self {
228 relay_urls,
229 active_relay_index: 0,
230 active_label,
231 socket: None,
232 custody,
233 owner_public_key_hex,
234 sarah_public_key_hex,
235 community_group_ids,
236 community_public_key_hexes,
237 authenticated: false,
238 auth_challenge: None,
239 acknowledged_event_id: None,
240 publish_acknowledgements: BTreeMap::new(),
241 healthy_relays: HashSet::new(),
242 events: BTreeMap::new(),
243 subscription_sequence: 0,
244 gap_state: GapState::None,
245 event_cache_truncated: false,
246 })
247 }
248
249 fn custody_public_key_hex(&self) -> Result<String, SarahConversationError> {
250 match &self.custody {
251 RelayCustody::Omega(identity_service) => identity_service
252 .inspect()
253 .map_err(|error| SarahConversationError::Identity(error.to_string()))?
254 .identity
255 .map(|identity| identity.public_key_hex().as_str().to_string())
256 .ok_or(SarahConversationError::IdentityRequired),
257 #[cfg(any(test, feature = "test-support"))]
258 RelayCustody::Keys(keys) => Ok(keys.public_key().to_hex()),
259 }
260 }
261
262 fn prune_publication_acknowledgements(&mut self, preserved_event_id: &str) {
263 while self.publish_acknowledgements.len() > MAX_PENDING_PUBLICATIONS {
264 let Some(oldest_event_id) = self
265 .publish_acknowledgements
266 .keys()
267 .find(|event_id| event_id.as_str() != preserved_event_id)
268 .cloned()
269 else {
270 break;
271 };
272 self.publish_acknowledgements.remove(&oldest_event_id);
273 }
274 }
275
276 pub fn relay_urls(&self) -> &[String] {
277 &self.relay_urls
278 }
279
280 fn connect_active(&mut self) -> Result<(), SarahConversationError> {
281 self.connect_active_until(Instant::now() + CONNECT_TIMEOUT)
282 }
283
284 fn connect_active_until(&mut self, deadline: Instant) -> Result<(), SarahConversationError> {
285 let relay_url = self
286 .relay_urls
287 .get(self.active_relay_index)
288 .cloned()
289 .ok_or_else(|| SarahConversationError::Internal("active relay disappeared".into()))?;
290 let timeout_duration = deadline.saturating_duration_since(Instant::now());
291 if timeout_duration.is_zero() {
292 return Err(SarahConversationError::Relay(
293 "relay operation deadline elapsed before connect".into(),
294 ));
295 }
296 let connection = smol::block_on(async {
297 let connect = connect_async(relay_url.as_str()).fuse();
298 let timeout = FutureExt::fuse(network_timer(timeout_duration.min(CONNECT_TIMEOUT)));
299 pin_mut!(connect, timeout);
300 select! {
301 result = connect => Some(result),
302 _ = timeout => None,
303 }
304 });
305 let (socket, _) = connection
306 .ok_or_else(|| SarahConversationError::Relay("relay connect timed out".into()))?
307 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
308 self.active_label = relay_url;
309 self.socket = Some(socket);
310 self.authenticated = false;
311 self.auth_challenge = None;
312 Ok(())
313 }
314
315 fn disconnect_and_advance(&mut self) {
316 self.socket = None;
317 self.authenticated = false;
318 self.auth_challenge = None;
319 self.gap_state = GapState::Recovering;
320 self.active_relay_index = (self.active_relay_index + 1) % self.relay_urls.len().max(1);
321 }
322
323 fn send_json(&mut self, payload: Value) -> Result<(), SarahConversationError> {
324 self.send_json_until(payload, Instant::now() + NETWORK_TIMEOUT)
325 }
326
327 fn send_json_until(
328 &mut self,
329 payload: Value,
330 deadline: Instant,
331 ) -> Result<(), SarahConversationError> {
332 let socket = self
333 .socket
334 .as_mut()
335 .ok_or_else(|| SarahConversationError::Relay("not connected".into()))?;
336 let text = serde_json::to_string(&payload)
337 .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
338 let timeout_duration = deadline.saturating_duration_since(Instant::now());
339 if timeout_duration.is_zero() {
340 return Err(SarahConversationError::Relay(
341 "relay operation deadline elapsed before write".into(),
342 ));
343 }
344 let result = smol::block_on(async {
345 let send = socket.send(Message::Text(text.into())).fuse();
346 let timeout = FutureExt::fuse(network_timer(timeout_duration.min(NETWORK_TIMEOUT)));
347 pin_mut!(send, timeout);
348 select! {
349 result = send => Some(result),
350 _ = timeout => None,
351 }
352 });
353 result
354 .ok_or_else(|| SarahConversationError::Relay("relay write timed out".into()))?
355 .map_err(|error| SarahConversationError::Relay(error.to_string()))
356 }
357
358 fn next_message_until(
359 &mut self,
360 deadline: Instant,
361 ) -> Result<IncomingMessage, SarahConversationError> {
362 let mut control_frames = 0_usize;
363 loop {
364 let socket = self
365 .socket
366 .as_mut()
367 .ok_or_else(|| SarahConversationError::Relay("not connected".into()))?;
368 let timeout_duration = deadline.saturating_duration_since(Instant::now());
369 if timeout_duration.is_zero() {
370 return Ok(IncomingMessage::TimedOut);
371 }
372 let message = smol::block_on(async {
373 let read = socket.next().fuse();
374 let timeout = FutureExt::fuse(network_timer(timeout_duration));
375 pin_mut!(read, timeout);
376 select! {
377 result = read => Some(result),
378 _ = timeout => None,
379 }
380 });
381 let Some(message) = message else {
382 return Ok(IncomingMessage::TimedOut);
383 };
384 let message = message
385 .ok_or_else(|| SarahConversationError::Relay("relay closed connection".into()))?
386 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
387 match message {
388 Message::Text(text) => {
389 if text.len() > 2 * 1024 * 1024 {
390 return Err(SarahConversationError::Relay(
391 "relay frame exceeds the inbound budget".into(),
392 ));
393 }
394 let value = serde_json::from_str(text.as_str())
395 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
396 return Ok(IncomingMessage::Json(value));
397 }
398 Message::Binary(bytes) => {
399 if bytes.len() > 2 * 1024 * 1024 {
400 return Err(SarahConversationError::Relay(
401 "relay frame exceeds the inbound budget".into(),
402 ));
403 }
404 let text = std::str::from_utf8(&bytes).map_err(|error| {
405 SarahConversationError::Relay(format!(
406 "relay sent non-UTF-8 binary frame: {error}"
407 ))
408 })?;
409 let value = serde_json::from_str(text)
410 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
411 return Ok(IncomingMessage::Json(value));
412 }
413 Message::Ping(payload) => {
414 record_control_frame(&mut control_frames)?;
415 let socket = self.socket.as_mut().ok_or_else(|| {
416 SarahConversationError::Relay("relay disconnected during ping".into())
417 })?;
418 let result = smol::block_on(async {
419 let send = socket.send(Message::Pong(payload)).fuse();
420 let remaining = deadline.saturating_duration_since(Instant::now());
421 let timeout =
422 FutureExt::fuse(network_timer(remaining.min(NETWORK_TIMEOUT)));
423 pin_mut!(send, timeout);
424 select! {
425 result = send => Some(result),
426 _ = timeout => None,
427 }
428 });
429 result
430 .ok_or_else(|| {
431 SarahConversationError::Relay("relay pong timed out".into())
432 })?
433 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
434 }
435 Message::Pong(_) | Message::Frame(_) => {
436 record_control_frame(&mut control_frames)?;
437 }
438 Message::Close(_) => {
439 return Err(SarahConversationError::Relay(
440 "relay closed connection".into(),
441 ));
442 }
443 }
444 }
445 }
446
447 fn record_auth_challenge(&mut self, value: &Value) -> bool {
448 let Some(array) = value.as_array() else {
449 return false;
450 };
451 if array.first().and_then(Value::as_str) != Some("AUTH") {
452 return false;
453 }
454 self.auth_challenge = array.get(1).and_then(Value::as_str).map(str::to_owned);
455 self.authenticated = false;
456 true
457 }
458
459 fn publish_once(&mut self, event: &Event) -> Result<(), SarahConversationError> {
460 self.send_json(json!(["EVENT", event]))?;
461 let deadline = Instant::now() + NETWORK_TIMEOUT;
462 loop {
463 let remaining = deadline.saturating_duration_since(Instant::now());
464 if remaining.is_zero() {
465 return Err(SarahConversationError::Relay(
466 "relay acknowledgement timed out".into(),
467 ));
468 }
469 match self.next_message_until(deadline)? {
470 IncomingMessage::TimedOut => {
471 return Err(SarahConversationError::Relay(
472 "relay acknowledgement timed out".into(),
473 ));
474 }
475 IncomingMessage::Json(value) => {
476 if self.record_auth_challenge(&value) {
477 return Err(SarahConversationError::IdentityRequired);
478 }
479 let Some(array) = value.as_array() else {
480 continue;
481 };
482 if array.first().and_then(Value::as_str) != Some("OK")
483 || array.get(1).and_then(Value::as_str) != Some(event.id.to_hex().as_str())
484 {
485 continue;
486 }
487 if array.get(2).and_then(Value::as_bool) == Some(true) {
488 self.acknowledged_event_id = Some(event.id.to_hex());
489 return Ok(());
490 }
491 let reason = array
492 .get(3)
493 .and_then(Value::as_str)
494 .unwrap_or("relay rejected event");
495 return Err(SarahConversationError::Relay(reason.to_string()));
496 }
497 }
498 }
499 }
500
501 /// The order in which `publish` walks the configured relays.
502 ///
503 /// Normally left to right. But `publish` is the one operation that hands
504 /// `IdentityRequired` back to its caller, and the caller answers the NIP-42
505 /// challenge on whichever relay raised it and then retries exactly once.
506 /// Restarting that retry at relay 0 would tear down the socket it just
507 /// authenticated in order to re-attempt a relay that is down — and since a
508 /// reconnect earns a fresh challenge, the retry would raise
509 /// `IdentityRequired` again and the publish could never complete. A healthy
510 /// authenticating relay listed after a dead one would be unreachable.
511 ///
512 /// So while an authenticated session is open, that relay is tried first and
513 /// the rest follow. Every relay is still attempted; only the order moves.
514 fn publish_relay_order(&self) -> Vec<usize> {
515 publish_relay_order(
516 self.relay_urls.len(),
517 (self.authenticated && self.socket.is_some()).then_some(self.active_relay_index),
518 )
519 }
520
521 fn query_once_until(
522 &mut self,
523 conversation_ref: &str,
524 deadline: Instant,
525 ) -> Result<GapState, SarahConversationError> {
526 self.subscription_sequence = self.subscription_sequence.saturating_add(1);
527 let subscription_id = format!("omega-issue31-{}", self.subscription_sequence);
528 let mut request = vec![Value::String("REQ".into()), json!(subscription_id)];
529 request.push(json!({
530 "kinds": [Kind::GiftWrap.as_u16()],
531 "#p": [self.custody_public_key_hex()?],
532 "limit": 256
533 }));
534 request.push(json!({
535 "kinds": SARAH_RECORD_KINDS
536 .iter()
537 .copied()
538 .chain([ISSUE31_HOST_DISCOVERY_KIND])
539 .collect::<Vec<_>>(),
540 "authors": [self.owner_public_key_hex.as_str(), self.sarah_public_key_hex.as_str()],
541 "limit": 256
542 }));
543 request.push(json!({
544 "kinds": [NIP_AE_KIND, NIP_ER_KIND],
545 "authors": [self.owner_public_key_hex.as_str(), self.sarah_public_key_hex.as_str()],
546 "limit": 256
547 }));
548 request.push(json!({
549 "kinds": [NIP_RS_KIND],
550 "authors": [self.owner_public_key_hex.as_str()],
551 "limit": 256
552 }));
553 if !self.community_group_ids.is_empty() {
554 request.push(json!({
555 "kinds": [NIP_29_GROUP_CHAT_KIND],
556 "#h": self.community_group_ids,
557 "limit": 256
558 }));
559 }
560 if !self.community_public_key_hexes.is_empty() {
561 request.push(json!({
562 "kinds": LBR_KINDS,
563 "authors": self.community_public_key_hexes,
564 "limit": 256
565 }));
566 }
567 self.send_json_until(Value::Array(request), deadline)?;
568
569 let mut admitted = Vec::new();
570 let mut admission_truncated = false;
571 let mut notice: Option<String> = None;
572 let mut closed_reason: Option<String> = None;
573 let query_result = loop {
574 let remaining = deadline.saturating_duration_since(Instant::now());
575 if remaining.is_zero() {
576 break GapState::Possible;
577 }
578 match self.next_message_until(deadline)? {
579 IncomingMessage::TimedOut => break GapState::Possible,
580 IncomingMessage::Json(value) => {
581 if self.record_auth_challenge(&value) {
582 return Err(SarahConversationError::IdentityRequired);
583 }
584 let Some(array) = value.as_array() else {
585 continue;
586 };
587 match array.first().and_then(Value::as_str) {
588 Some("EVENT")
589 if array.get(1).and_then(Value::as_str)
590 == Some(subscription_id.as_str()) =>
591 {
592 let event_value = array.get(2).ok_or_else(|| {
593 SarahConversationError::Relay(
594 "relay EVENT frame omitted event".into(),
595 )
596 })?;
597 let event =
598 Event::from_json(event_value.to_string()).map_err(|error| {
599 SarahConversationError::Relay(error.to_string())
600 })?;
601 if !event_within_bounds(&event) || event.verify().is_err() {
602 continue;
603 }
604 match self.admit_event(&event, conversation_ref) {
605 Ok(Some(stored)) if admitted.len() < 256 => admitted.push(stored),
606 Ok(Some(_)) => admission_truncated = true,
607 Ok(None) | Err(_) => {}
608 }
609 }
610 Some("EOSE")
611 if array.get(1).and_then(Value::as_str)
612 == Some(subscription_id.as_str()) =>
613 {
614 break query_gap_after_eose(admitted.len(), admission_truncated);
615 }
616 // NIP-01: the relay has ended our subscription. There is
617 // no EOSE coming, so waiting out the deadline would only
618 // turn a clear answer into an opaque timeout.
619 Some("CLOSED")
620 if array.get(1).and_then(Value::as_str)
621 == Some(subscription_id.as_str()) =>
622 {
623 closed_reason = array
624 .get(2)
625 .and_then(Value::as_str)
626 .map(str::to_owned)
627 .or_else(|| Some("relay closed the subscription".into()));
628 break GapState::Possible;
629 }
630 // A NOTICE is not subscription-scoped, so it does not end
631 // the read. But if this query goes on to time out, the
632 // relay's own words are a better explanation than a
633 // generic deadline message.
634 Some("NOTICE") => {
635 notice = array
636 .get(1)
637 .and_then(Value::as_str)
638 .map(|text| text.chars().take(200).collect::<String>());
639 }
640 _ => {}
641 }
642 }
643 }
644 };
645 if let Some(reason) = closed_reason {
646 return Err(SarahConversationError::Relay(format!(
647 "relay closed the query subscription: {reason}"
648 )));
649 }
650 // Closing is courtesy. A relay that has stopped answering must not turn
651 // a completed read into an error, and a query that timed out already
652 // carries its own gap state — reporting a write failure instead would
653 // hide the gap behind a transport message.
654 if self
655 .send_json_until(json!(["CLOSE", subscription_id]), deadline)
656 .is_err()
657 && admitted.is_empty()
658 && query_result != GapState::None
659 && let Some(notice) = notice
660 {
661 return Err(SarahConversationError::Relay(format!(
662 "relay query failed: {notice}"
663 )));
664 }
665
666 for event in admitted {
667 self.events.entry(event.event_id.clone()).or_insert(event);
668 }
669 self.reindex_events();
670 Ok(query_result)
671 }
672
673 fn query_active_relay_until(
674 &mut self,
675 conversation_ref: &str,
676 deadline: Instant,
677 ) -> Result<GapState, SarahConversationError> {
678 let mut authentication_attempted = false;
679 loop {
680 match self.query_once_until(conversation_ref, deadline) {
681 Err(SarahConversationError::IdentityRequired) => {
682 begin_authentication_retry(&mut authentication_attempted)?;
683 let auth_event = self.sign_active_auth_event()?;
684 self.authenticate_until(&auth_event, deadline)?;
685 }
686 result => return result,
687 }
688 }
689 }
690
691 fn sign_active_auth_event(&self) -> Result<Event, SarahConversationError> {
692 let challenge = self.auth_challenge.as_deref().ok_or_else(|| {
693 SarahConversationError::Relay("no relay AUTH challenge is pending".into())
694 })?;
695 let relay = RelayUrl::parse(&self.active_label)
696 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
697 match &self.custody {
698 RelayCustody::Omega(identity_service) => {
699 let custody = identity_service
700 .inspect()
701 .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
702 let identity = custody
703 .identity
704 .ok_or(SarahConversationError::IdentityRequired)?;
705 let public_key = PublicKey::from_hex(identity.public_key_hex().as_str())
706 .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
707 let unsigned = EventBuilder::auth(challenge, relay).build(public_key);
708 let semantic_binding = serde_json::to_vec(&json!({
709 "relay": self.active_label,
710 "challenge": challenge,
711 "createdAt": unsigned.created_at.as_secs(),
712 }))
713 .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
714 let digest = format!("{:x}", Sha256::digest(semantic_binding));
715 let request_ref = ReceiptRef::new(format!("nip42.{}", &digest[..32]))
716 .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
717 let request = AdmittedSigningRequest {
718 request_ref,
719 identity_ref: identity.identity_ref().clone(),
720 purpose: SigningPurpose::NostrEvent,
721 event: UnsignedEventTemplate {
722 created_at: unsigned.created_at.as_secs(),
723 kind: unsigned.kind.as_u16(),
724 tags: unsigned
725 .tags
726 .iter()
727 .map(|tag| tag.as_slice().to_vec())
728 .collect(),
729 content: unsigned.content,
730 },
731 };
732 let signed = identity_service
733 .sign(&request)
734 .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
735 Event::from_json(signed.signed_event_json)
736 .map_err(|error| SarahConversationError::Identity(error.to_string()))
737 }
738 #[cfg(any(test, feature = "test-support"))]
739 RelayCustody::Keys(keys) => EventBuilder::auth(challenge, relay)
740 .sign_with_keys(keys)
741 .map_err(|error| SarahConversationError::Identity(error.to_string())),
742 }
743 }
744
745 fn authenticate_until(
746 &mut self,
747 auth_event: &Event,
748 deadline: Instant,
749 ) -> Result<(), SarahConversationError> {
750 let challenge = self.auth_challenge.as_deref().ok_or_else(|| {
751 SarahConversationError::Relay("no relay AUTH challenge is pending".into())
752 })?;
753 let relay_url = RelayUrl::parse(&self.active_label)
754 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
755 if !nostr::nips::nip42::is_valid_auth_event(auth_event, &relay_url, challenge)
756 || auth_event.verify().is_err()
757 {
758 return Err(SarahConversationError::Relay(
759 "NIP-42 auth event failed local validation".into(),
760 ));
761 }
762 self.send_json_until(json!(["AUTH", auth_event]), deadline)?;
763 loop {
764 let remaining = deadline.saturating_duration_since(Instant::now());
765 if remaining.is_zero() {
766 return Err(SarahConversationError::Relay(
767 "NIP-42 acknowledgement timed out".into(),
768 ));
769 }
770 match self.next_message_until(deadline)? {
771 IncomingMessage::TimedOut => {
772 return Err(SarahConversationError::Relay(
773 "NIP-42 acknowledgement timed out".into(),
774 ));
775 }
776 IncomingMessage::Json(value) => {
777 let Some(array) = value.as_array() else {
778 continue;
779 };
780 if array.first().and_then(Value::as_str) != Some("OK")
781 || array.get(1).and_then(Value::as_str)
782 != Some(auth_event.id.to_hex().as_str())
783 {
784 continue;
785 }
786 if array.get(2).and_then(Value::as_bool) != Some(true) {
787 let reason = array
788 .get(3)
789 .and_then(Value::as_str)
790 .unwrap_or("NIP-42 authentication rejected");
791 return Err(SarahConversationError::Relay(reason.to_string()));
792 }
793 self.authenticated = true;
794 self.auth_challenge = None;
795 return Ok(());
796 }
797 }
798 }
799 }
800
801 fn admit_event(
802 &self,
803 event: &Event,
804 conversation_ref: &str,
805 ) -> Result<Option<StoredConversationEvent>, SarahConversationError> {
806 if event.kind == Kind::GiftWrap {
807 let gift_wrap_event_json = event
808 .try_as_json()
809 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
810 let (rumor_event_id, sender_public_key_hex, created_at, tags, content) =
811 match &self.custody {
812 RelayCustody::Omega(identity_service) => {
813 let gift = identity_service
814 .unwrap_private_message(&gift_wrap_event_json)
815 .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
816 (
817 gift.rumor_event_id,
818 gift.sender_public_key_hex,
819 gift.created_at,
820 gift.tags,
821 gift.content,
822 )
823 }
824 #[cfg(any(test, feature = "test-support"))]
825 RelayCustody::Keys(keys) => {
826 let gift = smol::block_on(crate::nostr_websocket_relay::nip59_extract_rumor(
827 keys, event,
828 ))
829 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
830 if gift.rumor.kind != Kind::PrivateDirectMessage {
831 return Ok(None);
832 }
833 gift.rumor
834 .verify_id()
835 .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
836 let rumor_event_id = gift.rumor.id.ok_or_else(|| {
837 SarahConversationError::Relay(
838 "NIP-59 rumor omitted its convergence id".into(),
839 )
840 })?;
841 (
842 rumor_event_id.to_hex(),
843 gift.sender.to_hex(),
844 gift.rumor.created_at.as_secs(),
845 gift.rumor
846 .tags
847 .iter()
848 .map(|tag| tag.as_slice().to_vec())
849 .collect(),
850 gift.rumor.content,
851 )
852 }
853 };
854 let recipient_public_key_hex = self.custody_public_key_hex()?;
855 let record_kind = private_record_kind(
856 &content,
857 &sender_public_key_hex,
858 &recipient_public_key_hex,
859 &self.owner_public_key_hex,
860 &self.sarah_public_key_hex,
861 &tags,
862 )?;
863 if record_kind == "message" {
864 require_conversation_recipients(
865 &tags,
866 &sender_public_key_hex,
867 &self.owner_public_key_hex,
868 &self.sarah_public_key_hex,
869 )?;
870 if (sender_public_key_hex != self.owner_public_key_hex
871 && sender_public_key_hex != self.sarah_public_key_hex)
872 || tag_value(&tags, "conversation").as_deref() != Some(conversation_ref)
873 {
874 return Ok(None);
875 }
876 }
877 return Ok(Some(StoredConversationEvent {
878 event_id: rumor_event_id,
879 kind: Kind::PrivateDirectMessage.as_u16(),
880 pubkey: sender_public_key_hex,
881 created_at,
882 conversation_ref: conversation_ref.to_string(),
883 content_summary: content,
884 tags,
885 record_kind: record_kind.to_string(),
886 store_index: 0,
887 }));
888 }
889
890 let author = event.pubkey.to_hex();
891 let kind = event.kind.as_u16();
892 let tags: Vec<Vec<String>> = event
893 .tags
894 .iter()
895 .map(|tag| tag.as_slice().to_vec())
896 .collect();
897 if kind == ISSUE31_HOST_DISCOVERY_KIND {
898 let value = serde_json::from_str::<Value>(&event.content)
899 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
900 let (host_ref, host_public_key_hex) = if value.get("schema").and_then(Value::as_str)
901 == Some(ISSUE31_HOST_DISCOVERY_SCHEMA_V2)
902 {
903 let discovery = Issue31HostDiscoveryV2::decode(event.content.as_bytes())
904 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
905 if discovery.conversation != conversation_ref {
906 return Ok(None);
907 }
908 (discovery.host_ref, discovery.host_public_key_hex)
909 } else {
910 let discovery = Issue31HostDiscovery::decode(event.content.as_bytes())
911 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
912 (discovery.host_ref, discovery.host_public_key_hex)
913 };
914 if author != self.owner_public_key_hex
915 || author != self.custody_public_key_hex()?
916 || host_public_key_hex != author
917 || tag_value(&tags, "d").as_deref() != Some(host_ref.as_str())
918 || tag_value(&tags, "k").as_deref() != Some("1059")
919 || tag_value(&tags, "t").as_deref() != Some("omega-issue31-host")
920 || tag_value(&tags, "alt").as_deref() != Some("Omega Issue 31 Nostr host discovery")
921 {
922 return Ok(None);
923 }
924 } else if SARAH_RECORD_KINDS.contains(&kind) {
925 if (author != self.owner_public_key_hex && author != self.sarah_public_key_hex)
926 || tag_value(&tags, "conversation").as_deref() != Some(conversation_ref)
927 {
928 return Ok(None);
929 }
930 } else if kind == NIP_AE_KIND {
931 if author != self.sarah_public_key_hex
932 || !valid_nip_ae_tags(&tags, &self.owner_public_key_hex)
933 {
934 return Ok(None);
935 }
936 } else if kind == NIP_RS_KIND {
937 if author != self.owner_public_key_hex || !valid_nip_rs_tags(&tags) {
938 return Ok(None);
939 }
940 } else if kind == NIP_ER_KIND {
941 if (author != self.owner_public_key_hex && author != self.sarah_public_key_hex)
942 || !valid_nip_er_tags(&tags)
943 {
944 return Ok(None);
945 }
946 } else if kind == NIP_29_GROUP_CHAT_KIND {
947 if !configured_group_tag(&tags, &self.community_group_ids) {
948 return Ok(None);
949 }
950 } else if LBR_KINDS.contains(&kind) {
951 if !self
952 .community_public_key_hexes
953 .iter()
954 .any(|public_key| public_key == &author)
955 {
956 return Ok(None);
957 }
958 } else {
959 return Ok(None);
960 }
961 Ok(Some(StoredConversationEvent {
962 event_id: event.id.to_hex(),
963 kind,
964 pubkey: author,
965 created_at: event.created_at.as_secs(),
966 conversation_ref: conversation_ref.to_string(),
967 content_summary: if SARAH_RECORD_KINDS.contains(&kind)
968 || matches!(kind, NIP_AE_KIND | NIP_RS_KIND | NIP_ER_KIND)
969 {
970 event.content.clone()
971 } else {
972 bounded_summary(&event.content)
973 },
974 tags,
975 record_kind: public_record_kind(event.kind.as_u16()).to_string(),
976 store_index: 0,
977 }))
978 }
979
980 fn reindex_events(&mut self) {
981 let mut events: Vec<StoredConversationEvent> = self.events.values().cloned().collect();
982 events.sort_by(|left, right| {
983 (left.created_at, left.event_id.as_str())
984 .cmp(&(right.created_at, right.event_id.as_str()))
985 });
986 if events.len() > MAX_CACHED_EVENTS {
987 events.drain(..events.len().saturating_sub(MAX_CACHED_EVENTS));
988 self.gap_state = GapState::Possible;
989 self.event_cache_truncated = true;
990 }
991 let mut content_bytes = events
992 .iter()
993 .map(|event| event.content_summary.len())
994 .sum::<usize>();
995 while content_bytes > MAX_CACHED_CONTENT_BYTES && events.len() > 1 {
996 let removed = events.remove(0);
997 content_bytes = content_bytes.saturating_sub(removed.content_summary.len());
998 self.gap_state = GapState::Possible;
999 self.event_cache_truncated = true;
1000 }
1001 self.events.clear();
1002 for (store_index, mut event) in events.into_iter().enumerate() {
1003 event.store_index = store_index;
1004 self.events.insert(event.event_id.clone(), event);
1005 }
1006 }
1007
1008 fn page(
1009 &self,
1010 conversation_ref: &str,
1011 after_cursor: Option<&str>,
1012 limit: usize,
1013 query_gap_state: GapState,
1014 ) -> QueryPage {
1015 let mut matching: Vec<StoredConversationEvent> = self
1016 .events
1017 .values()
1018 .filter(|event| event.conversation_ref == conversation_ref)
1019 .cloned()
1020 .collect();
1021 matching.sort_by(|left, right| {
1022 (left.created_at, left.event_id.as_str())
1023 .cmp(&(right.created_at, right.event_id.as_str()))
1024 });
1025
1026 let mut gap_state = query_gap_state;
1027 let start_index = match after_cursor {
1028 Some(cursor) => match matching
1029 .iter()
1030 .position(|event| event_cursor(event) == cursor)
1031 {
1032 Some(index) => index.saturating_add(1),
1033 None => {
1034 gap_state = GapState::Confirmed;
1035 0
1036 }
1037 },
1038 None => 0,
1039 };
1040 let events: Vec<StoredConversationEvent> = matching
1041 .iter()
1042 .skip(start_index)
1043 .take(limit)
1044 .cloned()
1045 .collect();
1046 let next_cursor = if start_index.saturating_add(events.len()) < matching.len() {
1047 events.last().map(event_cursor)
1048 } else {
1049 None
1050 };
1051 QueryPage {
1052 events,
1053 next_cursor,
1054 gap_state,
1055 }
1056 }
1057}
1058
1059impl RelayTransport for WebSocketRelayAdapter {
1060 fn label(&self) -> &str {
1061 &self.active_label
1062 }
1063
1064 fn connection_state(&self) -> ConnectionState {
1065 if self.socket.is_none() && self.healthy_relays.is_empty() {
1066 ConnectionState::Disconnected
1067 } else if self.gap_state == GapState::None
1068 && self.healthy_relays.len() == self.relay_urls.len()
1069 {
1070 ConnectionState::Connected
1071 } else {
1072 ConnectionState::Degraded
1073 }
1074 }
1075
1076 fn is_authenticated(&self) -> bool {
1077 self.authenticated
1078 }
1079
1080 fn connect(&mut self) -> Result<(), SarahConversationError> {
1081 if self.socket.is_some() {
1082 return Ok(());
1083 }
1084 let attempts = self.relay_urls.len().saturating_mul(MAX_CONNECT_ROUNDS);
1085 let mut last_error = None;
1086 for attempt in 0..attempts {
1087 match self.connect_active() {
1088 Ok(()) => return Ok(()),
1089 Err(error) => last_error = Some(error),
1090 }
1091 self.disconnect_and_advance();
1092 let multiplier = 1_u32 << attempt.min(3);
1093 let backoff = INITIAL_BACKOFF.saturating_mul(multiplier).min(MAX_BACKOFF);
1094 smol::block_on(network_timer(backoff));
1095 }
1096 Err(last_error.unwrap_or_else(|| {
1097 SarahConversationError::Relay("relay connection attempts exhausted".into())
1098 }))
1099 }
1100
1101 fn auth_challenge(&self) -> Option<RelayAuthChallenge> {
1102 self.auth_challenge
1103 .as_ref()
1104 .map(|challenge| RelayAuthChallenge {
1105 challenge: challenge.clone(),
1106 relay_url: self.active_label.clone(),
1107 })
1108 }
1109
1110 fn authenticate(&mut self, auth_event: &Event) -> Result<(), SarahConversationError> {
1111 self.authenticate_until(auth_event, Instant::now() + NETWORK_TIMEOUT)
1112 }
1113
1114 fn publish(&mut self, event: &Event) -> Result<(), SarahConversationError> {
1115 let event_id = event.id.to_hex();
1116 let mut last_error = None;
1117 for relay_index in self.publish_relay_order() {
1118 let relay_url = self.relay_urls[relay_index].clone();
1119 if self
1120 .publish_acknowledgements
1121 .get(&event_id)
1122 .is_some_and(|relays| relays.contains(&relay_url))
1123 {
1124 continue;
1125 }
1126 if self.socket.is_none() || self.active_label != relay_url {
1127 self.socket = None;
1128 self.authenticated = false;
1129 self.auth_challenge = None;
1130 self.active_relay_index = relay_index;
1131 if let Err(error) = self.connect_active() {
1132 last_error = Some(error);
1133 self.healthy_relays.remove(&relay_url);
1134 continue;
1135 }
1136 }
1137 match self.publish_once(event) {
1138 Ok(()) => {
1139 self.healthy_relays.insert(relay_url.clone());
1140 self.publish_acknowledgements
1141 .entry(event_id.clone())
1142 .or_default()
1143 .insert(relay_url);
1144 }
1145 Err(SarahConversationError::IdentityRequired) => {
1146 return Err(SarahConversationError::IdentityRequired);
1147 }
1148 Err(error) => {
1149 last_error = Some(error);
1150 self.healthy_relays.remove(&relay_url);
1151 self.socket = None;
1152 self.authenticated = false;
1153 self.auth_challenge = None;
1154 }
1155 }
1156 }
1157 let acknowledged_relays = self
1158 .publish_acknowledgements
1159 .get(&event_id)
1160 .map_or(0, HashSet::len);
1161 self.prune_publication_acknowledgements(&event_id);
1162 if acknowledged_relays > 0 {
1163 self.acknowledged_event_id = Some(event_id);
1164 if acknowledged_relays < self.relay_urls.len() {
1165 self.gap_state = GapState::Possible;
1166 } else if self.healthy_relays.len() == self.relay_urls.len()
1167 && !self.event_cache_truncated
1168 {
1169 self.gap_state = GapState::None;
1170 }
1171 Ok(())
1172 } else {
1173 self.gap_state = GapState::Possible;
1174 Err(last_error.unwrap_or_else(|| {
1175 SarahConversationError::Relay(
1176 "event has not been acknowledged by every configured relay".into(),
1177 )
1178 }))
1179 }
1180 }
1181
1182 fn publication_complete(&mut self, event_id: &str) -> bool {
1183 let complete = self
1184 .publish_acknowledgements
1185 .get(event_id)
1186 .is_some_and(|relays| relays.len() == self.relay_urls.len());
1187 if complete {
1188 self.publish_acknowledgements.remove(event_id);
1189 }
1190 complete
1191 }
1192
1193 fn acknowledged_relays(&self, event_id: &str) -> Vec<String> {
1194 let mut relay_urls = self
1195 .publish_acknowledgements
1196 .get(event_id)
1197 .map(|relays| relays.iter().cloned().collect::<Vec<_>>())
1198 .unwrap_or_default();
1199 relay_urls.sort();
1200 relay_urls
1201 }
1202
1203 fn restore_publication_acknowledgements(&mut self, event_id: &str, relay_urls: &[String]) {
1204 let configured_relays = &self.relay_urls;
1205 let acknowledgements = self
1206 .publish_acknowledgements
1207 .entry(event_id.to_string())
1208 .or_default();
1209 acknowledgements.extend(
1210 relay_urls
1211 .iter()
1212 .filter(|relay_url| configured_relays.contains(relay_url))
1213 .cloned(),
1214 );
1215 self.prune_publication_acknowledgements(event_id);
1216 }
1217
1218 fn query(
1219 &mut self,
1220 conversation_ref: &str,
1221 after_cursor: Option<&str>,
1222 limit: usize,
1223 ) -> Result<QueryPage, SarahConversationError> {
1224 let operation_deadline = Instant::now() + QUERY_OPERATION_TIMEOUT;
1225 let mut query_gap_state = GapState::None;
1226 let mut successful_relays = 0_usize;
1227 let mut last_error = None;
1228 for relay_index in 0..self.relay_urls.len() {
1229 let relay_url = self.relay_urls[relay_index].clone();
1230 let relay_deadline = operation_deadline.min(Instant::now() + QUERY_TIMEOUT);
1231 if relay_deadline <= Instant::now() {
1232 query_gap_state = GapState::Possible;
1233 last_error = Some(SarahConversationError::Relay(
1234 "multi-relay query deadline elapsed".into(),
1235 ));
1236 break;
1237 }
1238 if self.socket.is_none() || self.active_label != relay_url {
1239 self.socket = None;
1240 self.authenticated = false;
1241 self.auth_challenge = None;
1242 self.active_relay_index = relay_index;
1243 if let Err(error) = self.connect_active_until(relay_deadline) {
1244 last_error = Some(error);
1245 query_gap_state = GapState::Possible;
1246 self.healthy_relays.remove(&relay_url);
1247 continue;
1248 }
1249 }
1250 match self.query_active_relay_until(conversation_ref, relay_deadline) {
1251 Ok(gap_state) => {
1252 successful_relays = successful_relays.saturating_add(1);
1253 self.healthy_relays.insert(relay_url);
1254 if gap_state != GapState::None {
1255 query_gap_state = strongest_gap_state(query_gap_state, gap_state);
1256 }
1257 }
1258 Err(error) => {
1259 last_error = Some(error);
1260 query_gap_state = GapState::Possible;
1261 self.healthy_relays.remove(&relay_url);
1262 self.socket = None;
1263 self.authenticated = false;
1264 self.auth_challenge = None;
1265 }
1266 }
1267 }
1268 if successful_relays == 0 {
1269 return Err(last_error.unwrap_or_else(|| {
1270 SarahConversationError::Relay("all relay queries failed".into())
1271 }));
1272 }
1273 if self.event_cache_truncated
1274 || self
1275 .publish_acknowledgements
1276 .values()
1277 .any(|relays| !relays.is_empty() && relays.len() < self.relay_urls.len())
1278 {
1279 query_gap_state = strongest_gap_state(query_gap_state, GapState::Possible);
1280 }
1281 self.gap_state = query_gap_state;
1282 Ok(self.page(conversation_ref, after_cursor, limit, query_gap_state))
1283 }
1284
1285 fn last_event_id(&self) -> Option<String> {
1286 self.acknowledged_event_id.clone()
1287 }
1288
1289 fn gap_state(&self) -> GapState {
1290 self.gap_state
1291 }
1292
1293 fn connected_relays(&self) -> Vec<String> {
1294 let mut relays = self.healthy_relays.iter().cloned().collect::<Vec<_>>();
1295 relays.sort();
1296 relays
1297 }
1298
1299 fn requires_private_messages(&self) -> bool {
1300 true
1301 }
1302}
1303
1304/// Order the relays a publish should attempt, given an open authenticated
1305/// session on `resume_at` if there is one.
1306///
1307/// Every relay is always attempted exactly once; only the order changes.
1308fn publish_relay_order(relay_count: usize, resume_at: Option<usize>) -> Vec<usize> {
1309 let Some(resume_at) = resume_at.filter(|index| *index < relay_count) else {
1310 return (0..relay_count).collect();
1311 };
1312 std::iter::once(resume_at)
1313 .chain((0..relay_count).filter(move |index| *index != resume_at))
1314 .collect()
1315}
1316
1317fn normalize_relay_url(relay_url: &str) -> Result<String, SarahConversationError> {
1318 let parsed = url::Url::parse(relay_url)
1319 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1320 if !matches!(parsed.scheme(), "ws" | "wss")
1321 || parsed.host_str().is_none()
1322 || !parsed.username().is_empty()
1323 || parsed.password().is_some()
1324 || parsed.query().is_some()
1325 || parsed.fragment().is_some()
1326 {
1327 return Err(SarahConversationError::InvalidRequest(
1328 "relay URL must be a credential-free ws:// or wss:// endpoint".into(),
1329 ));
1330 }
1331 let mut normalized = parsed.to_string();
1332 if normalized.ends_with('/') {
1333 normalized.pop();
1334 }
1335 Ok(normalized)
1336}
1337
1338fn begin_authentication_retry(attempted: &mut bool) -> Result<(), SarahConversationError> {
1339 if std::mem::replace(attempted, true) {
1340 return Err(SarahConversationError::Relay(
1341 "relay repeated NIP-42 challenge after authentication".into(),
1342 ));
1343 }
1344 Ok(())
1345}
1346
1347fn record_control_frame(count: &mut usize) -> Result<(), SarahConversationError> {
1348 *count = count.saturating_add(1);
1349 if *count > MAX_CONTROL_FRAMES_PER_READ {
1350 return Err(SarahConversationError::Relay(
1351 "relay exceeded the control-frame budget".into(),
1352 ));
1353 }
1354 Ok(())
1355}
1356
1357fn strongest_gap_state(left: GapState, right: GapState) -> GapState {
1358 let severity = |state| match state {
1359 GapState::None => 0,
1360 GapState::Possible => 1,
1361 GapState::Recovering => 2,
1362 GapState::Confirmed => 3,
1363 };
1364 if severity(left) >= severity(right) {
1365 left
1366 } else {
1367 right
1368 }
1369}
1370
1371fn event_cursor(event: &StoredConversationEvent) -> String {
1372 format!("cursor.{}.{}", event.created_at, event.event_id)
1373}
1374
1375fn tag_value(tags: &[Vec<String>], name: &str) -> Option<String> {
1376 tags.iter().find_map(|tag| {
1377 if tag.first().map(String::as_str) == Some(name) {
1378 tag.get(1).cloned()
1379 } else {
1380 None
1381 }
1382 })
1383}
1384
1385fn private_record_kind(
1386 content: &str,
1387 sender_public_key_hex: &str,
1388 recipient_public_key_hex: &str,
1389 owner_public_key_hex: &str,
1390 sarah_public_key_hex: &str,
1391 tags: &[Vec<String>],
1392) -> Result<&'static str, SarahConversationError> {
1393 let schema = serde_json::from_str::<Value>(content)
1394 .ok()
1395 .and_then(|value| {
1396 value
1397 .get("schema")
1398 .and_then(Value::as_str)
1399 .map(str::to_owned)
1400 });
1401 match schema.as_deref() {
1402 Some(ISSUE31_COMMAND_SCHEMA) => {
1403 require_single_private_recipient(tags, recipient_public_key_hex)?;
1404 let record = Issue31CommandRecord::decode(content.as_bytes())
1405 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1406 record
1407 .validate_private_binding(sender_public_key_hex, recipient_public_key_hex)
1408 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1409 Ok("control")
1410 }
1411 Some(ISSUE31_COMMAND_SCHEMA_V2) => {
1412 require_single_private_recipient(tags, recipient_public_key_hex)?;
1413 let record = Issue31CommandRecordV2::decode(content.as_bytes())
1414 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1415 record
1416 .validate_private_binding(sender_public_key_hex, recipient_public_key_hex)
1417 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1418 Ok("control")
1419 }
1420 Some(ISSUE31_OWNER_PROJECTION_SCHEMA) => {
1421 require_single_private_recipient(tags, recipient_public_key_hex)?;
1422 let record = Issue31OwnerProjectionRecord::decode(content.as_bytes())
1423 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1424 if record.host_public_key_hex != owner_public_key_hex {
1425 return Err(SarahConversationError::InvalidRequest(
1426 "owner projection targets another host".into(),
1427 ));
1428 }
1429 record
1430 .validate_private_binding(
1431 sender_public_key_hex,
1432 recipient_public_key_hex,
1433 sarah_public_key_hex,
1434 )
1435 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1436 Ok("owner_projection")
1437 }
1438 Some(ISSUE31_PAIRING_SCHEMA) => {
1439 require_single_private_recipient(tags, recipient_public_key_hex)?;
1440 let record = Issue31PairingRecord::decode(content.as_bytes())
1441 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1442 record
1443 .validate_private_binding(sender_public_key_hex, recipient_public_key_hex)
1444 .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1445 Ok("pairing")
1446 }
1447 Some(schema) if schema.starts_with("openagents.omega.issue31.") => Err(
1448 SarahConversationError::InvalidRequest("unknown Issue 31 private schema".into()),
1449 ),
1450 _ => Ok("message"),
1451 }
1452}
1453
1454fn require_single_private_recipient(
1455 tags: &[Vec<String>],
1456 recipient_public_key_hex: &str,
1457) -> Result<(), SarahConversationError> {
1458 let recipients: Vec<&str> = tags
1459 .iter()
1460 .filter(|tag| tag.first().map(String::as_str) == Some("p"))
1461 .filter_map(|tag| tag.get(1).map(String::as_str))
1462 .collect();
1463 if recipients != [recipient_public_key_hex] {
1464 return Err(SarahConversationError::InvalidRequest(
1465 "Issue 31 private rumor must have exactly one local p tag".into(),
1466 ));
1467 }
1468 Ok(())
1469}
1470
1471/// A Sarah conversation message is a two-party record, unlike every Issue 31
1472/// control record around it, and its author is never in its own `p` set.
1473///
1474/// `conversation_tags` writes a `p` tag for the owner *and* one for Sarah, but
1475/// `EventBuilder::build` defaults `allow_self_tagging` to false and strips the
1476/// `p` tag matching the event author. So an owner-authored rumor ships as
1477/// `p: [sarah]` and a Sarah-authored one as `p: [owner]`. That is correct
1478/// NIP-17 — a rumor's `p` tags are its receivers — and it is exactly what the
1479/// old 1:1 `require_single_private_recipient` rule could not express.
1480///
1481/// Holding conversation rumors to that rule refused every message the host
1482/// itself authored. The owner's own send was unreadable back off the relay,
1483/// `load_issue31_source_event` could never confirm it, and the command came
1484/// back `reason.omega.projection_failed` with no owner projection ever
1485/// published — a phone on omega#49 saw the send publish, the host admit the
1486/// command, and no reply arrive, three times. Nothing in the repository caught
1487/// it because every admission test hand-wrote a single-`p` rumor instead of
1488/// building one through the production path.
1489///
1490/// The bound the single-recipient rule was buying is kept. The author plus the
1491/// `p` set must be exactly the conversation's own two participants, so a rumor
1492/// naming any third recipient, or addressed to another pair entirely, is still
1493/// refused. Being the addressed reader is already proven by NIP-59: this custody
1494/// key decrypted the seal.
1495fn require_conversation_recipients(
1496 tags: &[Vec<String>],
1497 sender_public_key_hex: &str,
1498 owner_public_key_hex: &str,
1499 sarah_public_key_hex: &str,
1500) -> Result<(), SarahConversationError> {
1501 let mut named: Vec<&str> = tags
1502 .iter()
1503 .filter(|tag| tag.first().map(String::as_str) == Some("p"))
1504 .filter_map(|tag| tag.get(1).map(String::as_str))
1505 .collect();
1506 named.push(sender_public_key_hex);
1507 named.sort_unstable();
1508 named.dedup();
1509 let mut participants = [owner_public_key_hex, sarah_public_key_hex];
1510 participants.sort_unstable();
1511 if named != participants {
1512 return Err(SarahConversationError::InvalidRequest(
1513 "Issue 31 conversation rumor must name exactly the owner and Sarah".into(),
1514 ));
1515 }
1516 Ok(())
1517}
1518
1519fn configured_group_tag(tags: &[Vec<String>], configured_group_ids: &[String]) -> bool {
1520 let group_ids: Vec<&str> = tags
1521 .iter()
1522 .filter(|tag| tag.first().map(String::as_str) == Some("h"))
1523 .filter_map(|tag| tag.get(1).map(String::as_str))
1524 .collect();
1525 matches!(group_ids.as_slice(), [group_id] if configured_group_ids.iter().any(|configured| configured == group_id))
1526}
1527
1528fn valid_nip_ae_tags(tags: &[Vec<String>], owner_public_key_hex: &str) -> bool {
1529 exact_tag_value(tags, "d").is_some_and(is_lower_hex_64)
1530 && exact_tag_value(tags, "p") == Some(owner_public_key_hex)
1531 && exact_tag_value(tags, "alt") == Some("encrypted agent memory record")
1532}
1533
1534fn valid_nip_rs_tags(tags: &[Vec<String>]) -> bool {
1535 exact_tag_value(tags, "d").is_some_and(|value| {
1536 value
1537 .strip_prefix("read-state:")
1538 .is_some_and(|slot_id| !slot_id.is_empty() && slot_id.len() <= 64 && slot_id.is_ascii())
1539 }) && exact_tag_value(tags, "t") == Some("read-state")
1540 && exact_tag_value(tags, "alt") == Some("encrypted read state")
1541}
1542
1543fn valid_nip_er_tags(tags: &[Vec<String>]) -> bool {
1544 let Some(identifier) = exact_tag_value(tags, "d") else {
1545 return false;
1546 };
1547 if identifier.is_empty() || identifier.len() > 128 {
1548 return false;
1549 }
1550 if exact_tag_value(tags, "alt") != Some("Encrypted reminder") {
1551 return false;
1552 }
1553 let not_before = optional_ascii_u64_tag(tags, "not_before");
1554 let expiration = optional_ascii_u64_tag(tags, "expiration");
1555 if not_before.is_err() || expiration.is_err() {
1556 return false;
1557 }
1558 match (not_before.ok().flatten(), expiration.ok().flatten()) {
1559 (Some(not_before), Some(expiration)) => expiration > not_before,
1560 _ => true,
1561 }
1562}
1563
1564fn exact_tag_value<'a>(tags: &'a [Vec<String>], name: &str) -> Option<&'a str> {
1565 let mut values = tags
1566 .iter()
1567 .filter(|tag| tag.first().map(String::as_str) == Some(name))
1568 .filter_map(|tag| tag.get(1).map(String::as_str));
1569 let value = values.next()?;
1570 if values.next().is_some() {
1571 return None;
1572 }
1573 Some(value)
1574}
1575
1576fn optional_ascii_u64_tag(tags: &[Vec<String>], name: &str) -> Result<Option<u64>, ()> {
1577 let values = tags
1578 .iter()
1579 .filter(|tag| tag.first().map(String::as_str) == Some(name))
1580 .collect::<Vec<_>>();
1581 match values.as_slice() {
1582 [] => Ok(None),
1583 [tag] => {
1584 let value = tag.get(1).ok_or(())?;
1585 if value.is_empty()
1586 || !value.bytes().all(|byte| byte.is_ascii_digit())
1587 || (value.len() > 1 && value.starts_with('0'))
1588 {
1589 return Err(());
1590 }
1591 value.parse().map(Some).map_err(|_| ())
1592 }
1593 _ => Err(()),
1594 }
1595}
1596
1597fn is_lower_hex_64(value: &str) -> bool {
1598 value.len() == 64
1599 && value
1600 .bytes()
1601 .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
1602}
1603
1604fn normalized_group_ids(group_ids: Vec<String>) -> Result<Vec<String>, SarahConversationError> {
1605 if group_ids.len() > 64
1606 || group_ids
1607 .iter()
1608 .any(|group_id| group_id.is_empty() || group_id.len() > 256)
1609 {
1610 return Err(SarahConversationError::InvalidRequest(
1611 "community group policy must contain at most 64 bounded group ids".into(),
1612 ));
1613 }
1614 let unique = group_ids.iter().collect::<HashSet<_>>();
1615 if unique.len() != group_ids.len() {
1616 return Err(SarahConversationError::InvalidRequest(
1617 "community group policy contains duplicates".into(),
1618 ));
1619 }
1620 Ok(group_ids)
1621}
1622
1623fn normalized_public_keys(
1624 public_keys: Vec<String>,
1625 label: &str,
1626) -> Result<Vec<String>, SarahConversationError> {
1627 if public_keys.len() > 64
1628 || public_keys.iter().any(|public_key| {
1629 !is_lower_hex_64(public_key) || PublicKey::from_hex(public_key).is_err()
1630 })
1631 {
1632 return Err(SarahConversationError::InvalidRequest(format!(
1633 "{label} policy must contain at most 64 lowercase public keys"
1634 )));
1635 }
1636 let unique = public_keys.iter().collect::<HashSet<_>>();
1637 if unique.len() != public_keys.len() {
1638 return Err(SarahConversationError::InvalidRequest(format!(
1639 "{label} policy contains duplicates"
1640 )));
1641 }
1642 Ok(public_keys)
1643}
1644
1645fn event_within_bounds(event: &Event) -> bool {
1646 event.content.len() <= 1024 * 1024
1647 && event.tags.len() <= 128
1648 && event.tags.iter().all(|tag| {
1649 let values = tag.as_slice();
1650 values.len() <= 16 && values.iter().all(|value| value.len() <= 1024)
1651 })
1652}
1653
1654fn query_gap_after_eose(admitted_events: usize, admission_truncated: bool) -> GapState {
1655 if admission_truncated || admitted_events == 256 {
1656 GapState::Possible
1657 } else {
1658 GapState::None
1659 }
1660}
1661
1662fn public_record_kind(kind: u16) -> &'static str {
1663 if ISSUE31_COMMUNITY_KINDS.contains(&kind) {
1664 "community"
1665 } else if SARAH_RECORD_KINDS.contains(&kind) {
1666 "activity"
1667 } else {
1668 "memory"
1669 }
1670}
1671
1672fn bounded_summary(content: &str) -> String {
1673 const MAX_SUMMARY_BYTES: usize = 512;
1674 if content.len() <= MAX_SUMMARY_BYTES {
1675 return content.to_string();
1676 }
1677 let mut boundary = MAX_SUMMARY_BYTES;
1678 while !content.is_char_boundary(boundary) {
1679 boundary = boundary.saturating_sub(1);
1680 }
1681 format!("{}…", &content[..boundary])
1682}
1683
1684#[cfg(test)]
1685mod tests {
1686 use nostr::EventBuilder;
1687
1688 use crate::issue31_nostr::{
1689 Issue31HostConfiguration, Issue31HostController, Issue31NostrError, Issue31PairingEvent,
1690 Issue31PairingScope,
1691 };
1692 use crate::sarah_conversation::SARAH_TURN_RECORD_KIND;
1693
1694 use super::*;
1695
1696 // Live proofs against the deployed OpenAgents relay (OMEGA-MOB-31-01).
1697 //
1698 // These four ran green against `wss://relay.openagents.com` on 2026-07-25,
1699 // driving the real `WebSocketRelayAdapter` through connect, NIP-42,
1700 // publish, and read-back. The relay then degraded mid-session: it now
1701 // answers any `REQ` carrying a `#p` filter with
1702 // `["NOTICE","error: Handler error: StorageError"]`, sends no EOSE, and —
1703 // because one bad filter poisons the whole `REQ` — returns nothing for the
1704 // other filters in the same subscription either. Every Omega query includes
1705 // a gift-wrap `#p` filter, so every read is currently blind and the three
1706 // read-dependent proofs below no longer reproduce. The publish-side proof
1707 // still passes. Tracked in OpenAgentsInc/nostr-effect#170.
1708 //
1709 // Worth considering once that lands: issuing one subscription per filter,
1710 // so a filter the relay cannot serve degrades that record family alone
1711 // instead of blinding the client to everything.
1712
1713 /// Live proof against the deployed OpenAgents relay (OMEGA-MOB-31-01).
1714 ///
1715 /// Opt-in, because it needs the network and a real relay:
1716 ///
1717 /// ```sh
1718 /// OMEGA_LIVE_RELAY_URL=wss://openagents-nostr-relay-ezxz4mgdsq-uc.a.run.app \
1719 /// OMEGA_LIVE_RELAY_AUTH_URL=wss://relay.openagents.com \
1720 /// cargo test -p omega_effectd --lib live_relay -- --ignored --nocapture
1721 /// ```
1722 ///
1723 /// `OMEGA_LIVE_RELAY_AUTH_URL` exists because the relay binds
1724 /// `RELAY_PUBLIC_URL` and refuses a NIP-42 auth event whose `relay` tag
1725 /// names any other host with `invalid: relay URL mismatch`. A client may
1726 /// therefore connect through one hostname and still has to tag the
1727 /// canonical one.
1728 ///
1729 /// This exercises the real `WebSocketRelayAdapter` rather than
1730 /// `MockRelayAdapter`: connect, NIP-42 challenge, sign, authenticate,
1731 /// publish a durable turn record, and read it back by conversation ref.
1732 #[test]
1733 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
1734 fn live_relay_round_trip() {
1735 let Ok(url) = std::env::var("OMEGA_LIVE_RELAY_URL") else {
1736 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
1737 return;
1738 };
1739 let auth_url = std::env::var("OMEGA_LIVE_RELAY_AUTH_URL").unwrap_or_else(|_| url.clone());
1740
1741 let keys = Keys::generate();
1742 let mut relay = WebSocketRelayAdapter::new_for_keys(vec![url.clone()], keys.clone())
1743 .expect("adapter for live relay");
1744
1745 relay.connect().expect("connect to live relay");
1746 // A relay is not reported healthy on socket open alone. `healthy_relays`
1747 // is populated by a successful publish or query, so the state is
1748 // deliberately Degraded until an operation proves the relay works.
1749 assert_ne!(relay.connection_state(), ConnectionState::Disconnected);
1750
1751 let conversation_ref = format!("sarah.live.{}", &keys.public_key().to_hex()[..16]);
1752 let record = EventBuilder::new(Kind::Custom(SARAH_TURN_RECORD_KIND), "live round trip")
1753 .tag(nostr::Tag::parse(["conversation", conversation_ref.as_str()]).expect("conversation tag"))
1754 .sign_with_keys(&keys)
1755 .expect("signed turn record");
1756
1757 // Mirror `SarahConversationClient::publish_with_auth`. The adapter does
1758 // not read the relay's proactive AUTH frame during connect, so the
1759 // challenge only becomes visible once an operation meets it.
1760 match relay.publish(&record) {
1761 Ok(()) => {}
1762 Err(SarahConversationError::IdentityRequired) => {
1763 let challenge = relay
1764 .auth_challenge()
1765 .expect("relay must expose a challenge after refusing the publish");
1766 let auth_event = EventBuilder::new(Kind::Custom(22242), "")
1767 .tag(nostr::Tag::parse(["relay", auth_url.as_str()]).expect("relay tag"))
1768 .tag(
1769 nostr::Tag::parse(["challenge", challenge.challenge.as_str()])
1770 .expect("challenge tag"),
1771 )
1772 .sign_with_keys(&keys)
1773 .expect("signed auth event");
1774 relay.authenticate(&auth_event).expect("NIP-42 authenticate");
1775 assert!(relay.is_authenticated(), "relay must accept our auth");
1776 relay.publish(&record).expect("publish after authenticating");
1777 }
1778 Err(error) => panic!("unexpected publish error: {error}"),
1779 }
1780 assert!(
1781 relay.publication_complete(&record.id.to_hex()),
1782 "relay must acknowledge the publication"
1783 );
1784
1785 let page = relay
1786 .query(&conversation_ref, None, 10)
1787 .expect("query the live relay");
1788 assert!(
1789 page.events.iter().any(|event| event.event_id == record.id.to_hex()),
1790 "published event must read back from the live relay"
1791 );
1792 assert_eq!(page.gap_state, GapState::None, "no gap on a fresh conversation");
1793
1794 // Only now, with a publish and a query both acknowledged, is the relay
1795 // proven healthy rather than merely reachable.
1796 assert_eq!(relay.connection_state(), ConnectionState::Connected);
1797 assert_eq!(relay.connected_relays(), vec![url.clone()]);
1798 eprintln!(
1799 "live relay OK: published {} and read it back from {}",
1800 &record.id.to_hex()[..16],
1801 url
1802 );
1803 }
1804
1805 /// Connect, authenticate if the relay asks, and hand back a ready adapter.
1806 ///
1807 /// Shared by the live proofs so each one exercises the identical NIP-42
1808 /// path the production client uses, rather than a per-test approximation.
1809 #[cfg(test)]
1810 fn live_authenticated_adapter(relay_urls: Vec<String>, keys: &Keys) -> WebSocketRelayAdapter {
1811 let mut relay = WebSocketRelayAdapter::new_for_keys(relay_urls, keys.clone())
1812 .expect("adapter for live relay");
1813 relay.connect().expect("connect to live relay");
1814 relay
1815 }
1816
1817 /// Publish through the real adapter, meeting the NIP-42 challenge lazily
1818 /// exactly as `SarahConversationClient::publish_with_auth` does.
1819 #[cfg(test)]
1820 fn live_publish(
1821 relay: &mut WebSocketRelayAdapter,
1822 auth_url: &str,
1823 keys: &Keys,
1824 record: &Event,
1825 ) -> Result<(), SarahConversationError> {
1826 match relay.publish(record) {
1827 Err(SarahConversationError::IdentityRequired) => {
1828 let challenge = relay
1829 .auth_challenge()
1830 .expect("relay must expose a challenge after refusing the publish");
1831 let auth_event = EventBuilder::new(Kind::Custom(22242), "")
1832 .tag(nostr::Tag::parse(["relay", auth_url]).expect("relay tag"))
1833 .tag(
1834 nostr::Tag::parse(["challenge", challenge.challenge.as_str()])
1835 .expect("challenge tag"),
1836 )
1837 .sign_with_keys(keys)
1838 .expect("signed auth event");
1839 relay
1840 .authenticate(&auth_event)
1841 .expect("NIP-42 authenticate");
1842 relay.publish(record)
1843 }
1844 other => other,
1845 }
1846 }
1847
1848 #[cfg(test)]
1849 fn live_turn_record(keys: &Keys, conversation_ref: &str, body: &str, created_at: u64) -> Event {
1850 EventBuilder::new(Kind::Custom(SARAH_TURN_RECORD_KIND), body)
1851 .tag(nostr::Tag::parse(["conversation", conversation_ref]).expect("conversation tag"))
1852 .custom_created_at(nostr::Timestamp::from(created_at))
1853 .sign_with_keys(keys)
1854 .expect("signed turn record")
1855 }
1856
1857 #[cfg(test)]
1858 fn live_relay_env() -> Option<(String, String)> {
1859 let url = std::env::var("OMEGA_LIVE_RELAY_URL").ok()?;
1860 let auth_url = std::env::var("OMEGA_LIVE_RELAY_AUTH_URL").unwrap_or_else(|_| url.clone());
1861 Some((url, auth_url))
1862 }
1863
1864 /// Exit: "Confirmed records remain readable when the application service is
1865 /// unavailable and the relays are reachable."
1866 ///
1867 /// The publishing adapter is dropped outright — the strongest available
1868 /// stand-in for the application service being gone — and a second adapter
1869 /// with no cursors, no cache and no shared state reads the record back from
1870 /// relay storage alone.
1871 #[test]
1872 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
1873 fn live_relay_confirmed_records_outlive_the_application_service() {
1874 let Some((url, auth_url)) = live_relay_env() else {
1875 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
1876 return;
1877 };
1878 let keys = Keys::generate();
1879 let conversation_ref = format!("sarah.live.survive.{}", &keys.public_key().to_hex()[..16]);
1880 let record = live_turn_record(
1881 &keys,
1882 &conversation_ref,
1883 "readable without the application service",
1884 nostr::Timestamp::now().as_secs(),
1885 );
1886 {
1887 let mut publisher = live_authenticated_adapter(vec![url.clone()], &keys);
1888 live_publish(&mut publisher, &auth_url, &keys, &record).expect("publish");
1889 assert!(
1890 publisher.publication_complete(&record.id.to_hex()),
1891 "the relay must acknowledge before we call anything confirmed"
1892 );
1893 } // The application service disappears here, taking all of its state.
1894
1895 let mut reader = live_authenticated_adapter(vec![url.clone()], &keys);
1896 let page = match reader.query(&conversation_ref, None, 10) {
1897 Err(SarahConversationError::IdentityRequired) => {
1898 let challenge = reader.auth_challenge().expect("challenge");
1899 let auth_event = EventBuilder::new(Kind::Custom(22242), "")
1900 .tag(nostr::Tag::parse(["relay", auth_url.as_str()]).expect("relay tag"))
1901 .tag(
1902 nostr::Tag::parse(["challenge", challenge.challenge.as_str()])
1903 .expect("challenge tag"),
1904 )
1905 .sign_with_keys(&keys)
1906 .expect("signed auth event");
1907 reader.authenticate(&auth_event).expect("authenticate");
1908 reader.query(&conversation_ref, None, 10).expect("query")
1909 }
1910 other => other.expect("query"),
1911 };
1912 assert!(
1913 page.events
1914 .iter()
1915 .any(|event| event.event_id == record.id.to_hex()),
1916 "a confirmed record must be readable from a relay with no application service"
1917 );
1918 eprintln!(
1919 "live relay OK: {} survived the application service and read back from {url}",
1920 &record.id.to_hex()[..16]
1921 );
1922 }
1923
1924 /// Exit: "Relay outage and failover do not create false completion."
1925 ///
1926 /// Two halves. With one dead relay and one live relay the publish must
1927 /// succeed and be credited *only* to the relay that actually acknowledged.
1928 /// With every relay dead the publish must fail and completion must stay
1929 /// false — an unreachable relay can never be read as a completed write.
1930 #[test]
1931 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
1932 fn live_relay_outage_and_failover_never_report_false_completion() {
1933 let Some((url, auth_url)) = live_relay_env() else {
1934 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
1935 return;
1936 };
1937 // Port 9 is discard: the connection is refused rather than hanging.
1938 let dead_url = "ws://127.0.0.1:9".to_string();
1939 let keys = Keys::generate();
1940 let conversation_ref = format!("sarah.live.failover.{}", &keys.public_key().to_hex()[..16]);
1941
1942 let record = live_turn_record(
1943 &keys,
1944 &conversation_ref,
1945 "failover",
1946 nostr::Timestamp::now().as_secs(),
1947 );
1948 let mut failover =
1949 live_authenticated_adapter(vec![dead_url.clone(), url.clone()], &keys);
1950 live_publish(&mut failover, &auth_url, &keys, &record).expect("publish must fail over");
1951 let event_id = record.id.to_hex();
1952 let acknowledged = failover.acknowledged_relays(&event_id);
1953 assert!(
1954 acknowledged.contains(&url),
1955 "the live relay acknowledged and must be credited"
1956 );
1957 assert!(
1958 !acknowledged.contains(&dead_url),
1959 "an unreachable relay must never be credited with an acknowledgement"
1960 );
1961 // Completion means every configured relay acknowledged. One relay being
1962 // down is exactly the case where a weaker rule would manufacture a false
1963 // completion, so partial success must not read as complete, and the gap
1964 // must be visible.
1965 assert!(
1966 !failover.publication_complete(&event_id),
1967 "a publish that reached only some relays must not report completion"
1968 );
1969 assert_eq!(failover.gap_state(), GapState::Possible);
1970
1971 // Total outage: no relay can acknowledge, so nothing may be complete.
1972 let outage_record = live_turn_record(
1973 &keys,
1974 &conversation_ref,
1975 "total outage",
1976 nostr::Timestamp::now().as_secs(),
1977 );
1978 let mut outage = WebSocketRelayAdapter::new_for_keys(vec![dead_url], keys.clone())
1979 .expect("adapter for a dead relay");
1980 let _ = outage.connect();
1981 assert!(
1982 outage.publish(&outage_record).is_err(),
1983 "a publish to an unreachable relay must fail rather than succeed silently"
1984 );
1985 assert!(
1986 !outage.publication_complete(&outage_record.id.to_hex()),
1987 "a total relay outage must never report a completed publication"
1988 );
1989 assert!(outage.acknowledged_relays(&outage_record.id.to_hex()).is_empty());
1990 eprintln!("live relay OK: failover credited only {url}; total outage stayed incomplete");
1991 }
1992
1993 /// Exit: "Duplicate, reordered, missing, and stale events converge or show
1994 /// an exact gap."
1995 ///
1996 /// Against the real relay: the same signed event published twice converges
1997 /// to one record; events written newest-first still read back in a stable
1998 /// order; and a cursor the relay has never seen is reported as a confirmed
1999 /// gap rather than silently treated as an empty tail.
2000 #[test]
2001 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
2002 fn live_relay_duplicate_reorder_and_gap_converge_or_report_exactly() {
2003 let Some((url, auth_url)) = live_relay_env() else {
2004 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
2005 return;
2006 };
2007 let keys = Keys::generate();
2008 let conversation_ref = format!("sarah.live.converge.{}", &keys.public_key().to_hex()[..16]);
2009 let base = nostr::Timestamp::now().as_secs();
2010 let mut relay = live_authenticated_adapter(vec![url.clone()], &keys);
2011
2012 // Written newest-first, so a naive reader would surface them reversed.
2013 let newest = live_turn_record(&keys, &conversation_ref, "third", base + 2);
2014 let middle = live_turn_record(&keys, &conversation_ref, "second", base + 1);
2015 let oldest = live_turn_record(&keys, &conversation_ref, "first", base);
2016 for record in [&newest, &middle, &oldest] {
2017 live_publish(&mut relay, &auth_url, &keys, record).expect("publish");
2018 }
2019 // The duplicate: the identical signed event, published a second time.
2020 // A relay may answer OK or reject it as a duplicate; neither may create
2021 // a second record.
2022 let _ = live_publish(&mut relay, &auth_url, &keys, &newest);
2023
2024 let page = relay
2025 .query(&conversation_ref, None, 32)
2026 .expect("query the live relay");
2027 let ids = page
2028 .events
2029 .iter()
2030 .map(|event| event.event_id.clone())
2031 .collect::<Vec<_>>();
2032 let unique = ids.iter().collect::<HashSet<_>>();
2033 assert_eq!(
2034 ids.len(),
2035 unique.len(),
2036 "a republished event must converge to a single record"
2037 );
2038 let ordering = page
2039 .events
2040 .iter()
2041 .map(|event| event.created_at)
2042 .collect::<Vec<_>>();
2043 let mut sorted = ordering.clone();
2044 sorted.sort_unstable();
2045 assert_eq!(
2046 ordering, sorted,
2047 "out-of-order writes must read back in a deterministic order"
2048 );
2049 assert_eq!(page.gap_state, GapState::None);
2050
2051 // A cursor the relay has never issued is an exact, reported gap.
2052 let missing = relay
2053 .query(&conversation_ref, Some("cursor.999999.deadbeef"), 32)
2054 .expect("query with an unknown cursor");
2055 assert_eq!(
2056 missing.gap_state,
2057 GapState::Confirmed,
2058 "an unknown cursor must be reported as an exact gap, not an empty tail"
2059 );
2060 eprintln!(
2061 "live relay OK: {} records converged, ordered, and the unknown cursor reported an exact gap on {url}",
2062 page.events.len()
2063 );
2064 }
2065
2066 /// Exit: "Duplicate, reordered, missing, and STALE events converge or show
2067 /// an exact gap." The sibling test above covers duplicate, reordered and
2068 /// missing. This covers stale.
2069 ///
2070 /// A pairing request that had already expired when it was written is
2071 /// published to the live relay, stored by it, and served back to the host.
2072 /// The relay is transport and storage, so it neither knows nor cares that
2073 /// the record is dead. The host must refuse it for being expired — not for
2074 /// being malformed and not for having been seen before — and must issue
2075 /// neither a challenge nor a grant.
2076 #[test]
2077 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
2078 fn live_relay_stale_records_are_refused_rather_than_acted_on() {
2079 let Some((url, auth_url)) = live_relay_env() else {
2080 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
2081 return;
2082 };
2083 let host_keys = Keys::generate();
2084 let device_keys = Keys::generate();
2085 let host_public_key_hex = host_keys.public_key().to_hex();
2086 let device_public_key_hex = device_keys.public_key().to_hex();
2087 let configuration = Issue31HostConfiguration {
2088 host_ref: "omega.host.stale".into(),
2089 host_public_key_hex: host_public_key_hex.clone(),
2090 sarah_public_key_hex: Keys::generate().public_key().to_hex(),
2091 conversation: format!("sarah.{}", &host_public_key_hex[..24]),
2092 display_name: "Live Omega".into(),
2093 relay_urls: vec![url.clone()],
2094 generation: 1,
2095 };
2096 let mut controller = Issue31HostController::new(configuration.clone()).expect("controller");
2097 controller
2098 .set_admitted_device_policy(
2099 vec![device_public_key_hex.clone()],
2100 vec![Issue31PairingScope::ControlFullAuto],
2101 )
2102 .expect("admit device");
2103
2104 // Issued and expired well before now. The device is admitted and the
2105 // record is otherwise perfectly well formed, so staleness is the only
2106 // thing that can refuse it.
2107 let now = nostr::Timestamp::now().as_secs();
2108 let stale = Issue31PairingRecord::PairingRequest {
2109 schema: ISSUE31_PAIRING_SCHEMA.into(),
2110 host_ref: configuration.host_ref.clone(),
2111 host_public_key_hex: host_public_key_hex.clone(),
2112 device_public_key_hex,
2113 issued_at: now - 7_200,
2114 pairing_request_ref: "pairing_request.live.stale".into(),
2115 requested_scopes: vec![Issue31PairingScope::ControlFullAuto],
2116 expires_at: now - 3_600,
2117 };
2118
2119 let mut device_relay = live_authenticated_adapter(vec![url.clone()], &device_keys);
2120 let host_public_key = PublicKey::from_hex(&host_public_key_hex).expect("host key");
2121 let rumor = EventBuilder::new(
2122 Kind::PrivateDirectMessage,
2123 serde_json::to_string(&stale).expect("record json"),
2124 )
2125 .tag(nostr::Tag::parse(["p", host_public_key_hex.as_str()]).expect("p tag"))
2126 .build(device_keys.public_key());
2127 let gift_wrap = smol::block_on(EventBuilder::gift_wrap(
2128 &device_keys,
2129 &host_public_key,
2130 rumor,
2131 [],
2132 ))
2133 .expect("gift wrap");
2134 live_publish(&mut device_relay, &auth_url, &device_keys, &gift_wrap)
2135 .expect("the relay must store a stale record; staleness is not its business");
2136
2137 let mut host_relay = live_authenticated_adapter(vec![url.clone()], &host_keys);
2138 live_publish(
2139 &mut host_relay,
2140 &auth_url,
2141 &host_keys,
2142 &live_turn_record(&host_keys, &configuration.conversation, "host online", now),
2143 )
2144 .expect("host authenticates");
2145 let page = host_relay
2146 .query(&configuration.conversation, None, 32)
2147 .expect("query");
2148 let served = page
2149 .events
2150 .iter()
2151 .filter_map(|event| {
2152 Issue31PairingRecord::decode(event.content_summary.as_bytes())
2153 .ok()
2154 .map(|record| (event.event_id.clone(), record))
2155 })
2156 .collect::<Vec<_>>();
2157 assert_eq!(
2158 served.len(),
2159 1,
2160 "the relay must actually serve the stale record back, or this proves nothing"
2161 );
2162
2163 for (event_id, record) in served {
2164 assert!(
2165 !controller.pairing_event_was_processed(&event_id),
2166 "the stale record must arrive under an id the host has never seen, so that \
2167 de-duplication is not what refuses it"
2168 );
2169 let refusal = controller
2170 .handle_pairing_event(Issue31PairingEvent { event_id, record }, now)
2171 .expect_err("a stale pairing record must be refused");
2172 assert!(
2173 matches!(&refusal, Issue31NostrError::Invalid(message)
2174 if message.contains("expired")),
2175 "the refusal must be staleness, not dedup or shape: {refusal:?}"
2176 );
2177 }
2178 assert!(
2179 controller
2180 .grant_projections(now)
2181 .expect("grants")
2182 .is_empty(),
2183 "a stale request must not produce a grant"
2184 );
2185 eprintln!(
2186 "live relay OK: a stale pairing record stored by {url} was refused and granted nothing"
2187 );
2188 }
2189
2190 /// Exit: "A revoked device cannot restore its grant from old relay events."
2191 ///
2192 /// Proven end to end through relay storage rather than in memory. The device
2193 /// re-publishes the two records it authors — the pairing request and the
2194 /// pairing response — to the live relay, the host reads them back out of
2195 /// relay storage, and the host must still refuse.
2196 ///
2197 /// Two things make this sharp. The replayed records arrive under the rumor
2198 /// ids the relay returns, so the host's processed-event de-duplication is
2199 /// not what refuses them. And they carry fresh, unexpired timestamps, so the
2200 /// liveness check is not what refuses them either. The only thing left
2201 /// standing between a revoked device and a brand-new `control_full_auto`
2202 /// grant is the revocation itself.
2203 ///
2204 /// This was previously blocked: `relay.openagents.com` answered any filter
2205 /// carrying a `#p` tag with `["NOTICE","error: Handler error: StorageError"]`
2206 /// and never terminated the subscription, and every NIP-17/44/59 private
2207 /// record is `#p`-addressed. That was a double-JSON-encoding defect on both
2208 /// the query and the INSERT paths, fixed in nostr-effect#170 and deployed as
2209 /// `openagents-nostr-relay-00006-7wf`. This test now passes against
2210 /// `wss://relay.openagents.com`. The revocation law itself is additionally
2211 /// proven without the network in `issue31_nostr`.
2212 #[test]
2213 #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
2214 fn live_relay_replay_cannot_restore_a_revoked_device_grant() {
2215 let Some((url, auth_url)) = live_relay_env() else {
2216 eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
2217 return;
2218 };
2219 let host_keys = Keys::generate();
2220 let device_keys = Keys::generate();
2221 let host_public_key_hex = host_keys.public_key().to_hex();
2222 let device_public_key_hex = device_keys.public_key().to_hex();
2223 let configuration = Issue31HostConfiguration {
2224 host_ref: "omega.host.live".into(),
2225 host_public_key_hex: host_public_key_hex.clone(),
2226 sarah_public_key_hex: Keys::generate().public_key().to_hex(),
2227 conversation: format!("sarah.{}", &host_public_key_hex[..24]),
2228 display_name: "Live Omega".into(),
2229 relay_urls: vec![url.clone()],
2230 generation: 1,
2231 };
2232 let mut controller = Issue31HostController::new(configuration.clone()).expect("controller");
2233 controller
2234 .set_admitted_device_policy(
2235 vec![device_public_key_hex.clone()],
2236 vec![Issue31PairingScope::ControlFullAuto],
2237 )
2238 .expect("admit device");
2239
2240 let now = nostr::Timestamp::now().as_secs();
2241 let request = Issue31PairingRecord::PairingRequest {
2242 schema: ISSUE31_PAIRING_SCHEMA.into(),
2243 host_ref: configuration.host_ref.clone(),
2244 host_public_key_hex: host_public_key_hex.clone(),
2245 device_public_key_hex: device_public_key_hex.clone(),
2246 issued_at: now,
2247 pairing_request_ref: "pairing_request.live.replay".into(),
2248 requested_scopes: vec![Issue31PairingScope::ControlFullAuto],
2249 expires_at: now + 600,
2250 };
2251
2252 // Pair for real, then have the owner revoke the grant.
2253 let challenge = controller
2254 .handle_pairing_event(
2255 Issue31PairingEvent {
2256 event_id: "a".repeat(64),
2257 record: request.clone(),
2258 },
2259 now,
2260 )
2261 .expect("request")
2262 .expect("challenge");
2263 let challenge_value = match &challenge {
2264 Issue31PairingRecord::PairingChallenge { challenge, .. } => challenge.clone(),
2265 _ => panic!("challenge"),
2266 };
2267 controller
2268 .record_emitted_pairing("b".repeat(64), challenge)
2269 .expect("record challenge");
2270 let response = Issue31PairingRecord::PairingResponse {
2271 schema: ISSUE31_PAIRING_SCHEMA.into(),
2272 host_ref: configuration.host_ref.clone(),
2273 host_public_key_hex: host_public_key_hex.clone(),
2274 device_public_key_hex,
2275 issued_at: now,
2276 pairing_response_ref: "pairing_response.live.replay".into(),
2277 pairing_challenge_event_id: "b".repeat(64),
2278 challenge: challenge_value,
2279 expires_at: now + 600,
2280 };
2281 let grant = controller
2282 .handle_pairing_event(
2283 Issue31PairingEvent {
2284 event_id: "c".repeat(64),
2285 record: response.clone(),
2286 },
2287 now,
2288 )
2289 .expect("response")
2290 .expect("grant");
2291 let grant_ref = match &grant {
2292 Issue31PairingRecord::ScopedGrant { grant_ref, .. } => grant_ref.clone(),
2293 _ => panic!("grant"),
2294 };
2295 controller
2296 .record_emitted_pairing("d".repeat(64), grant)
2297 .expect("record grant");
2298 let revocation = controller
2299 .revoke_grant(&grant_ref, now + 1, Some("reason.omega.owner_revoked".into()))
2300 .expect("revoke");
2301 controller
2302 .record_emitted_pairing("e".repeat(64), revocation)
2303 .expect("record revocation");
2304
2305 // The device replays what it authored, through the real relay.
2306 let mut device_relay = live_authenticated_adapter(vec![url.clone()], &device_keys);
2307 let host_public_key = PublicKey::from_hex(&host_public_key_hex).expect("host key");
2308 for record in [&request, &response] {
2309 let rumor = EventBuilder::new(
2310 Kind::PrivateDirectMessage,
2311 serde_json::to_string(record).expect("record json"),
2312 )
2313 .tag(nostr::Tag::parse(["p", host_public_key_hex.as_str()]).expect("p tag"))
2314 .build(device_keys.public_key());
2315 let gift_wrap = smol::block_on(EventBuilder::gift_wrap(
2316 &device_keys,
2317 &host_public_key,
2318 rumor,
2319 [],
2320 ))
2321 .expect("gift wrap");
2322 live_publish(&mut device_relay, &auth_url, &device_keys, &gift_wrap)
2323 .expect("publish the replayed pairing record");
2324 }
2325
2326 // The host reads them back out of relay storage. A publish first, so the
2327 // NIP-42 handshake is settled before the query's own deadline starts.
2328 let mut host_relay = live_authenticated_adapter(vec![url.clone()], &host_keys);
2329 live_publish(
2330 &mut host_relay,
2331 &auth_url,
2332 &host_keys,
2333 &live_turn_record(&host_keys, &configuration.conversation, "host online", now),
2334 )
2335 .expect("host authenticates");
2336 let page = host_relay
2337 .query(&configuration.conversation, None, 32)
2338 .expect("query");
2339 let replayed = page
2340 .events
2341 .iter()
2342 .filter_map(|event| {
2343 Issue31PairingRecord::decode(event.content_summary.as_bytes())
2344 .ok()
2345 .map(|record| (event.event_id.clone(), record))
2346 })
2347 .collect::<Vec<_>>();
2348 assert_eq!(
2349 replayed.len(),
2350 2,
2351 "the relay must actually return both replayed records, or this proves nothing"
2352 );
2353
2354 for (event_id, record) in replayed {
2355 assert!(
2356 !controller.pairing_event_was_processed(&event_id),
2357 "the replay must arrive under an id the host has never seen, so that \
2358 de-duplication is not what refuses it"
2359 );
2360 let refusal = controller
2361 .handle_pairing_event(Issue31PairingEvent { event_id, record }, now + 2)
2362 .expect_err("a revoked device must be refused, however its records arrive");
2363 assert!(
2364 matches!(&refusal, Issue31NostrError::Invalid(message)
2365 if message.contains("device admission was revoked")),
2366 "the refusal must be the revocation, not liveness or dedup: {refusal:?}"
2367 );
2368 }
2369 let projections = controller.grant_projections(now + 2).expect("grants");
2370 assert_eq!(projections.len(), 1, "no new grant may exist");
2371 assert_eq!(projections[0].status, "revoked");
2372 eprintln!(
2373 "live relay OK: replayed pairing records from {url} could not restore the revoked grant"
2374 );
2375 }
2376
2377 /// A publish retry must resume on the relay it just authenticated.
2378 ///
2379 /// Restarting at relay 0 after answering a NIP-42 challenge on relay 1 drops
2380 /// the authenticated session to re-attempt a relay that is down. The
2381 /// reconnect that follows earns a fresh challenge, so the caller's single
2382 /// retry raises `IdentityRequired` again and the publish never completes.
2383 /// This is the deterministic form of a failure first caught against the live
2384 /// relay with a dead relay ordered first.
2385 #[test]
2386 fn an_authenticated_publish_retry_resumes_on_the_relay_it_authenticated() {
2387 // No authenticated session: plain left to right.
2388 assert_eq!(publish_relay_order(3, None), vec![0, 1, 2]);
2389 // Authenticated on relay 1 — the case where relay 0 is down and relay 1
2390 // issued the NIP-42 challenge. Relay 1 is retried first, and relay 0 and
2391 // relay 2 are still attempted.
2392 assert_eq!(publish_relay_order(3, Some(1)), vec![1, 0, 2]);
2393 // Every relay appears exactly once, whatever the resume point.
2394 for resume_at in 0..3 {
2395 let order = publish_relay_order(3, Some(resume_at));
2396 assert_eq!(order.len(), 3);
2397 assert_eq!(order.iter().collect::<HashSet<_>>().len(), 3);
2398 assert_eq!(order[0], resume_at);
2399 }
2400 // An out-of-range resume point cannot drop or duplicate a relay.
2401 assert_eq!(publish_relay_order(2, Some(7)), vec![0, 1]);
2402 assert!(publish_relay_order(0, Some(0)).is_empty());
2403 }
2404
2405 /// The adapter's own accessor must agree with the pure ordering rule: an
2406 /// authenticated flag with no open socket is not an authenticated session.
2407 #[test]
2408 fn a_closed_socket_is_not_an_authenticated_session() {
2409 let keys = Keys::generate();
2410 let mut relay = WebSocketRelayAdapter::new_for_keys(
2411 vec![
2412 "wss://down.example.com".to_string(),
2413 "wss://live.example.com".to_string(),
2414 ],
2415 keys,
2416 )
2417 .expect("relay");
2418 relay.active_relay_index = 1;
2419 relay.authenticated = true;
2420 assert!(relay.socket.is_none());
2421 assert_eq!(relay.publish_relay_order(), vec![0, 1]);
2422 }
2423
2424 #[test]
2425 fn relay_urls_are_bounded_deduplicated_and_secret_free() {
2426 let keys = Keys::generate();
2427 let relay = WebSocketRelayAdapter::new_for_keys(
2428 vec![
2429 "wss://relay.example.com".to_string(),
2430 "wss://relay.example.com".to_string(),
2431 "ws://127.0.0.1:7777".to_string(),
2432 ],
2433 keys.clone(),
2434 )
2435 .expect("valid relay list");
2436 assert_eq!(relay.relay_urls().len(), 2);
2437 assert!(
2438 WebSocketRelayAdapter::new_for_keys(
2439 vec!["wss://owner:secret@relay.example.com".to_string()],
2440 keys,
2441 )
2442 .is_err()
2443 );
2444 assert!(normalize_relay_url("wss://relay.example.com/path").is_ok());
2445 assert!(normalize_relay_url("wss://relay.example.com/?token=secret").is_err());
2446 assert!(normalize_relay_url("wss://relay.example.com/#fragment").is_err());
2447 assert_eq!(
2448 normalize_relay_url("wss://relay.example.com/path/sidecar/").expect("normalized path"),
2449 "wss://relay.example.com/path/sidecar"
2450 );
2451 }
2452
2453 #[test]
2454 fn each_relay_authentication_retry_is_single_and_bounded() {
2455 let mut first_relay_attempted = false;
2456 begin_authentication_retry(&mut first_relay_attempted).expect("first relay auth");
2457 assert!(begin_authentication_retry(&mut first_relay_attempted).is_err());
2458
2459 let mut second_relay_attempted = false;
2460 begin_authentication_retry(&mut second_relay_attempted).expect("second relay auth");
2461
2462 let mut control_frames = 0;
2463 for _ in 0..MAX_CONTROL_FRAMES_PER_READ {
2464 record_control_frame(&mut control_frames).expect("bounded control frame");
2465 }
2466 assert!(record_control_frame(&mut control_frames).is_err());
2467 }
2468
2469 #[test]
2470 fn missing_or_unknown_cursor_is_an_exact_gap() {
2471 let keys = Keys::generate();
2472 let mut relay =
2473 WebSocketRelayAdapter::new_for_keys(vec!["wss://relay.example.com".to_string()], keys)
2474 .expect("relay");
2475 relay.events.insert(
2476 "a".repeat(64),
2477 StoredConversationEvent {
2478 event_id: "a".repeat(64),
2479 kind: Kind::PrivateDirectMessage.as_u16(),
2480 pubkey: "b".repeat(64),
2481 created_at: 10,
2482 conversation_ref: "sarah.test".to_string(),
2483 content_summary: "hello".to_string(),
2484 tags: Vec::new(),
2485 record_kind: "message".to_string(),
2486 store_index: 0,
2487 },
2488 );
2489 let first = relay.page("sarah.test", None, 10, GapState::None);
2490 assert_eq!(first.gap_state, GapState::None);
2491 let missing = relay.page("sarah.test", Some("cursor.9.missing"), 10, GapState::None);
2492 assert_eq!(missing.gap_state, GapState::Confirmed);
2493 assert_eq!(missing.events.len(), 1);
2494 }
2495
2496 #[test]
2497 fn duplicate_events_are_idempotent_and_reordered_deterministically() {
2498 let keys = Keys::generate();
2499 let mut relay =
2500 WebSocketRelayAdapter::new_for_keys(vec!["wss://relay.example.com".to_string()], keys)
2501 .expect("relay");
2502 for (id, created_at) in [("b", 20), ("a", 10), ("b", 20)] {
2503 let event_id = id.repeat(64);
2504 relay
2505 .events
2506 .entry(event_id.clone())
2507 .or_insert(StoredConversationEvent {
2508 event_id,
2509 kind: Kind::PrivateDirectMessage.as_u16(),
2510 pubkey: "c".repeat(64),
2511 created_at,
2512 conversation_ref: "sarah.test".to_string(),
2513 content_summary: id.to_string(),
2514 tags: Vec::new(),
2515 record_kind: "message".to_string(),
2516 store_index: 0,
2517 });
2518 }
2519 relay.reindex_events();
2520 let page = relay.page("sarah.test", None, 10, GapState::None);
2521 assert_eq!(page.events.len(), 2);
2522 assert_eq!(page.events[0].content_summary, "a");
2523 assert_eq!(page.events[1].content_summary, "b");
2524 }
2525
2526 #[test]
2527 fn relay_event_and_acknowledgement_caches_are_bounded() {
2528 let keys = Keys::generate();
2529 let mut relay =
2530 WebSocketRelayAdapter::new_for_keys(vec!["wss://relay.example.com".to_string()], keys)
2531 .expect("relay");
2532 for index in 0..=MAX_CACHED_EVENTS {
2533 let event_id = format!("{index:064x}");
2534 relay.events.insert(
2535 event_id.clone(),
2536 StoredConversationEvent {
2537 event_id,
2538 kind: Kind::PrivateDirectMessage.as_u16(),
2539 pubkey: "c".repeat(64),
2540 created_at: index as u64,
2541 conversation_ref: "sarah.test".into(),
2542 content_summary: "bounded".into(),
2543 tags: Vec::new(),
2544 record_kind: "message".into(),
2545 store_index: index,
2546 },
2547 );
2548 }
2549 relay.reindex_events();
2550 assert_eq!(relay.events.len(), MAX_CACHED_EVENTS);
2551 assert_eq!(relay.gap_state, GapState::Possible);
2552
2553 for index in 0..=MAX_PENDING_PUBLICATIONS {
2554 relay.restore_publication_acknowledgements(
2555 &format!("{index:064x}"),
2556 &["wss://relay.example.com".into()],
2557 );
2558 }
2559 assert_eq!(
2560 relay.publish_acknowledgements.len(),
2561 MAX_PENDING_PUBLICATIONS
2562 );
2563 }
2564
2565 #[test]
2566 fn locally_rejects_filter_bypass_and_wrong_private_authors() {
2567 let owner = Keys::generate();
2568 let owner_public_key_hex = owner.public_key().to_hex();
2569 let sarah = Keys::generate();
2570 let community_author = Keys::generate();
2571 let attacker = Keys::generate();
2572 let relay = WebSocketRelayAdapter::new_for_keys_with_policy(
2573 vec!["wss://relay.example.com".to_string()],
2574 owner.clone(),
2575 sarah.public_key().to_hex(),
2576 vec!["community.openagents".into()],
2577 vec![community_author.public_key().to_hex()],
2578 )
2579 .expect("relay");
2580 let public_event = EventBuilder::new(Kind::from(SARAH_TURN_RECORD_KIND), "injected")
2581 .tags(vec![
2582 nostr::Tag::parse(["conversation", "sarah.test"]).expect("conversation tag"),
2583 ])
2584 .sign_with_keys(&attacker)
2585 .expect("signed attacker event");
2586 assert!(
2587 relay
2588 .admit_event(&public_event, "sarah.test")
2589 .expect("local admission")
2590 .is_none()
2591 );
2592 let group_event = EventBuilder::new(Kind::from(NIP_29_GROUP_CHAT_KIND), "group message")
2593 .tags(vec![
2594 nostr::Tag::parse(["h", "community.openagents"]).expect("group tag"),
2595 ])
2596 .sign_with_keys(&owner)
2597 .expect("signed group event");
2598 assert_eq!(
2599 relay
2600 .admit_event(&group_event, "sarah.test")
2601 .expect("group admission")
2602 .expect("admitted group")
2603 .record_kind,
2604 "community"
2605 );
2606 let missing_group = EventBuilder::new(Kind::from(NIP_29_GROUP_CHAT_KIND), "unbound")
2607 .sign_with_keys(&owner)
2608 .expect("signed unbound group event");
2609 assert!(
2610 relay
2611 .admit_event(&missing_group, "sarah.test")
2612 .expect("missing group admission")
2613 .is_none()
2614 );
2615 let wrong_recipient_rumor =
2616 EventBuilder::new(Kind::PrivateDirectMessage, "wrong recipient")
2617 .tags(vec![
2618 nostr::Tag::parse(["p", attacker.public_key().to_hex().as_str()])
2619 .expect("recipient tag"),
2620 nostr::Tag::parse(["conversation", "sarah.test"]).expect("conversation tag"),
2621 ])
2622 .build(sarah.public_key());
2623 let wrong_recipient_gift_wrap = smol::block_on(EventBuilder::gift_wrap(
2624 &sarah,
2625 &owner.public_key(),
2626 wrong_recipient_rumor,
2627 [],
2628 ))
2629 .expect("gift wrap");
2630 assert!(
2631 relay
2632 .admit_event(&wrong_recipient_gift_wrap, "sarah.test")
2633 .is_err()
2634 );
2635 for kind in [
2636 LBR_AGENTIC_CODING_REQUEST_KIND,
2637 LBR_AGENTIC_CODING_RESULT_KIND,
2638 LBR_FEEDBACK_KIND,
2639 ] {
2640 let event = EventBuilder::new(Kind::from(kind), "bounded LBR record")
2641 .sign_with_keys(&community_author)
2642 .expect("signed LBR event");
2643 assert_eq!(
2644 relay
2645 .admit_event(&event, "sarah.test")
2646 .expect("LBR admission")
2647 .expect("admitted LBR")
2648 .record_kind,
2649 "community"
2650 );
2651 let owner_event = EventBuilder::new(Kind::from(kind), "unconfigured author")
2652 .sign_with_keys(&owner)
2653 .expect("signed owner LBR event");
2654 assert!(
2655 relay
2656 .admit_event(&owner_event, "sarah.test")
2657 .expect("owner LBR admission")
2658 .is_none()
2659 );
2660 }
2661
2662 let device = Keys::generate();
2663 let pairing = Issue31PairingRecord::PairingRequest {
2664 schema: ISSUE31_PAIRING_SCHEMA.into(),
2665 host_ref: "omega.host.local".into(),
2666 host_public_key_hex: owner_public_key_hex.clone(),
2667 device_public_key_hex: device.public_key().to_hex(),
2668 issued_at: 100,
2669 pairing_request_ref: "pairing_request.device.test".into(),
2670 requested_scopes: vec![crate::Issue31PairingScope::ObserveIssue31],
2671 expires_at: 200,
2672 };
2673 let content = serde_json::to_string(&pairing).expect("pairing json");
2674 let tags = vec![vec!["p".into(), owner_public_key_hex.clone()]];
2675 assert!(
2676 private_record_kind(
2677 &content,
2678 &attacker.public_key().to_hex(),
2679 &owner_public_key_hex,
2680 &owner_public_key_hex,
2681 &sarah.public_key().to_hex(),
2682 &tags,
2683 )
2684 .is_err()
2685 );
2686 let malformed = json!({
2687 "schema": ISSUE31_PAIRING_SCHEMA,
2688 "recordType": "pairing_request",
2689 })
2690 .to_string();
2691 assert!(
2692 private_record_kind(
2693 &malformed,
2694 &device.public_key().to_hex(),
2695 &owner_public_key_hex,
2696 &owner_public_key_hex,
2697 &sarah.public_key().to_hex(),
2698 &tags,
2699 )
2700 .is_err()
2701 );
2702 let command_v2 = json!({
2703 "schema": ISSUE31_COMMAND_SCHEMA_V2,
2704 "recordType": "command_intent",
2705 "hostRef": "omega.host.local",
2706 "hostPublicKeyHex": owner_public_key_hex,
2707 "devicePublicKeyHex": device.public_key().to_hex(),
2708 "grantRef": "grant.omega.device_1",
2709 "idempotencyRef": "idempotency.issue31.read_1",
2710 "expectedGeneration": 1,
2711 "arguments": {
2712 "kind": "read_state_patch",
2713 "actionRef": "action.issue31.read_state.advance",
2714 "slotId": "mobile",
2715 "clientId": "iphone",
2716 "contextRef": "sarah-conversation:sarah.0123456789abcdef01234567",
2717 "readAt": 150,
2718 },
2719 "issuedAt": 100,
2720 "expiresAt": 200,
2721 })
2722 .to_string();
2723 assert_eq!(
2724 private_record_kind(
2725 &command_v2,
2726 &device.public_key().to_hex(),
2727 &owner_public_key_hex,
2728 &owner_public_key_hex,
2729 &sarah.public_key().to_hex(),
2730 &tags,
2731 )
2732 .expect("command v2 admission"),
2733 "control"
2734 );
2735 let projection = json!({
2736 "schema": ISSUE31_OWNER_PROJECTION_SCHEMA,
2737 "recordType": "owner_projection",
2738 "hostRef": "omega.host.local",
2739 "hostPublicKeyHex": owner_public_key_hex,
2740 "devicePublicKeyHex": device.public_key().to_hex(),
2741 "grantRef": "grant.omega.device_1",
2742 "expectedGeneration": 1,
2743 "sourceEventId": "b".repeat(64),
2744 "sourceAuthorPublicKeyHex": owner_public_key_hex,
2745 "sourceRole": "owner",
2746 "sourceKind": 14,
2747 "sourceCreatedAt": 150,
2748 "projectedAt": 151,
2749 "projection": {
2750 "kind": "message",
2751 "role": "owner",
2752 "conversation": "sarah.0123456789abcdef01234567",
2753 "text": "Ready",
2754 },
2755 })
2756 .to_string();
2757 let projection_tags = vec![vec!["p".into(), device.public_key().to_hex()]];
2758 assert_eq!(
2759 private_record_kind(
2760 &projection,
2761 &owner_public_key_hex,
2762 &device.public_key().to_hex(),
2763 &owner_public_key_hex,
2764 &sarah.public_key().to_hex(),
2765 &projection_tags,
2766 )
2767 .expect("owner projection admission"),
2768 "owner_projection"
2769 );
2770 assert_eq!(query_gap_after_eose(12, false), GapState::None);
2771 assert_eq!(query_gap_after_eose(256, false), GapState::Possible);
2772 }
2773
2774 /// The owner's own send has to be readable back off the relay, because
2775 /// `load_issue31_source_event` will not confirm a source it cannot see and
2776 /// the command result degrades to `reason.omega.projection_failed` with no
2777 /// owner projection ever published — which is exactly what a phone saw on
2778 /// omega#49: the send published, the host admitted the command, and no
2779 /// reply ever arrived.
2780 ///
2781 /// The rumor is built through the production tag builder and the production
2782 /// event builder rather than hand-written. That is the whole point: every
2783 /// prior admission test wrote its own single-`p` rumor and so never saw
2784 /// that `EventBuilder::build` strips the author's own `p` tag, leaving the
2785 /// owner's message addressed only to Sarah and unreadable by its own author.
2786 #[test]
2787 fn owner_authored_conversation_message_survives_its_own_round_trip() {
2788 let owner = Keys::generate();
2789 let sarah = Keys::generate();
2790 let conversation_ref = "sarah.0123456789abcdef01234567";
2791 let tags = crate::sarah_conversation::conversation_tags(
2792 conversation_ref,
2793 &owner.public_key().to_hex(),
2794 &sarah.public_key().to_hex(),
2795 )
2796 .expect("production conversation tags");
2797 let mut rumor = EventBuilder::new(Kind::PrivateDirectMessage, "durable hello")
2798 .tags(tags)
2799 .build(owner.public_key());
2800 rumor.ensure_id();
2801 let rumor_event_id = rumor.id.expect("rumor id").to_hex();
2802
2803 let relay = WebSocketRelayAdapter::new_for_keys_with_policy(
2804 vec!["wss://relay.example.com".to_string()],
2805 owner.clone(),
2806 sarah.public_key().to_hex(),
2807 Vec::new(),
2808 Vec::new(),
2809 )
2810 .expect("relay");
2811
2812 // The host's own copy: sender and recipient are both the owner key,
2813 // which is the copy `load_issue31_source_event` has to find.
2814 let own_copy = smol::block_on(EventBuilder::gift_wrap(
2815 &owner,
2816 &owner.public_key(),
2817 rumor.clone(),
2818 [],
2819 ))
2820 .expect("gift wrap to the owner");
2821 let stored = relay
2822 .admit_event(&own_copy, conversation_ref)
2823 .expect("the owner's own message must be admissible")
2824 .expect("the owner's own message must be stored");
2825 assert_eq!(stored.record_kind, "message");
2826 assert_eq!(stored.event_id, rumor_event_id);
2827 assert_eq!(stored.pubkey, owner.public_key().to_hex());
2828 assert_eq!(stored.kind, crate::ISSUE31_PRIVATE_RUMOR_KIND);
2829 // The wire shape this is defending: the author is not in its own `p`
2830 // set, so a reader that demands to see itself there can never read its
2831 // own send back.
2832 assert_eq!(
2833 stored
2834 .tags
2835 .iter()
2836 .filter(|tag| tag.first().map(String::as_str) == Some("p"))
2837 .filter_map(|tag| tag.get(1).cloned())
2838 .collect::<Vec<_>>(),
2839 vec![sarah.public_key().to_hex()]
2840 );
2841
2842 // The other direction has to keep working: Sarah authors, the owner
2843 // reads, and now it is the owner's `p` tag that survives.
2844 let sarah_tags = crate::sarah_conversation::conversation_tags(
2845 conversation_ref,
2846 &owner.public_key().to_hex(),
2847 &sarah.public_key().to_hex(),
2848 )
2849 .expect("production conversation tags");
2850 let mut sarah_rumor = EventBuilder::new(Kind::PrivateDirectMessage, "Sarah here")
2851 .tags(sarah_tags)
2852 .build(sarah.public_key());
2853 sarah_rumor.ensure_id();
2854 let sarah_wrap = smol::block_on(EventBuilder::gift_wrap(
2855 &sarah,
2856 &owner.public_key(),
2857 sarah_rumor,
2858 [],
2859 ))
2860 .expect("gift wrap to the owner");
2861 let stored_reply = relay
2862 .admit_event(&sarah_wrap, conversation_ref)
2863 .expect("Sarah's message must be admissible")
2864 .expect("Sarah's message must be stored");
2865 assert_eq!(stored_reply.record_kind, "message");
2866 assert_eq!(stored_reply.pubkey, sarah.public_key().to_hex());
2867
2868 // The leak guard the single-recipient rule was buying is unchanged: a
2869 // third recipient on the rumor is still refused outright.
2870 let mut leaky_tags = crate::sarah_conversation::conversation_tags(
2871 conversation_ref,
2872 &owner.public_key().to_hex(),
2873 &sarah.public_key().to_hex(),
2874 )
2875 .expect("production conversation tags");
2876 let stranger = Keys::generate();
2877 leaky_tags
2878 .push(nostr::Tag::parse(["p", stranger.public_key().to_hex().as_str()]).expect("p tag"));
2879 let mut leaky = EventBuilder::new(Kind::PrivateDirectMessage, "leaky")
2880 .tags(leaky_tags)
2881 .build(owner.public_key());
2882 leaky.ensure_id();
2883 let leaky_wrap =
2884 smol::block_on(EventBuilder::gift_wrap(&owner, &owner.public_key(), leaky, []))
2885 .expect("gift wrap");
2886 assert!(
2887 relay.admit_event(&leaky_wrap, conversation_ref).is_err(),
2888 "a conversation rumor naming a third recipient must still be refused"
2889 );
2890
2891 // And a rumor addressed away from this reader entirely is still refused.
2892 let elsewhere_tags = crate::sarah_conversation::conversation_tags(
2893 conversation_ref,
2894 &stranger.public_key().to_hex(),
2895 &sarah.public_key().to_hex(),
2896 )
2897 .expect("production conversation tags");
2898 let mut elsewhere = EventBuilder::new(Kind::PrivateDirectMessage, "elsewhere")
2899 .tags(elsewhere_tags)
2900 .build(owner.public_key());
2901 elsewhere.ensure_id();
2902 let elsewhere_wrap = smol::block_on(EventBuilder::gift_wrap(
2903 &owner,
2904 &owner.public_key(),
2905 elsewhere,
2906 [],
2907 ))
2908 .expect("gift wrap");
2909 assert!(
2910 relay.admit_event(&elsewhere_wrap, conversation_ref).is_err(),
2911 "a conversation rumor addressed to another pair must still be refused"
2912 );
2913 }
2914
2915 #[test]
2916 fn memory_record_families_use_their_wire_contracts_without_conversation_tags() {
2917 let owner = Keys::generate();
2918 let sarah = Keys::generate();
2919 let relay = WebSocketRelayAdapter::new_for_keys_with_policy(
2920 vec!["wss://relay.example.com".to_string()],
2921 owner.clone(),
2922 sarah.public_key().to_hex(),
2923 Vec::new(),
2924 Vec::new(),
2925 )
2926 .expect("relay");
2927
2928 let engram = EventBuilder::new(Kind::from(NIP_AE_KIND), "nip44:ciphertext")
2929 .tags(vec![
2930 nostr::Tag::parse(["d", &"a".repeat(64)]).expect("d tag"),
2931 nostr::Tag::parse(["p", &owner.public_key().to_hex()]).expect("p tag"),
2932 nostr::Tag::parse(["alt", "encrypted agent memory record"]).expect("alt tag"),
2933 ])
2934 .sign_with_keys(&sarah)
2935 .expect("engram");
2936 assert_eq!(
2937 relay
2938 .admit_event(&engram, "sarah.test")
2939 .expect("engram admission")
2940 .expect("engram admitted")
2941 .record_kind,
2942 "memory"
2943 );
2944
2945 let read_state = EventBuilder::new(Kind::from(NIP_RS_KIND), "nip44:ciphertext")
2946 .tags(vec![
2947 nostr::Tag::parse(["d", "read-state:omega-desktop"]).expect("d tag"),
2948 nostr::Tag::parse(["t", "read-state"]).expect("t tag"),
2949 nostr::Tag::parse(["alt", "encrypted read state"]).expect("alt tag"),
2950 ])
2951 .sign_with_keys(&owner)
2952 .expect("read state");
2953 assert!(
2954 relay
2955 .admit_event(&read_state, "sarah.test")
2956 .expect("read state admission")
2957 .is_some()
2958 );
2959
2960 let reminder = EventBuilder::new(Kind::from(NIP_ER_KIND), "nip44:ciphertext")
2961 .tags(vec![
2962 nostr::Tag::parse(["d", "reminder-1"]).expect("d tag"),
2963 nostr::Tag::parse(["alt", "Encrypted reminder"]).expect("alt tag"),
2964 nostr::Tag::parse(["not_before", "100"]).expect("not-before tag"),
2965 nostr::Tag::parse(["expiration", "200"]).expect("expiration tag"),
2966 ])
2967 .sign_with_keys(&sarah)
2968 .expect("reminder");
2969 assert!(
2970 relay
2971 .admit_event(&reminder, "sarah.test")
2972 .expect("reminder admission")
2973 .is_some()
2974 );
2975
2976 let wrong_author = EventBuilder::new(Kind::from(NIP_AE_KIND), "nip44:ciphertext")
2977 .tags(engram.tags)
2978 .sign_with_keys(&owner)
2979 .expect("owner-authored engram");
2980 assert!(
2981 relay
2982 .admit_event(&wrong_author, "sarah.test")
2983 .expect("wrong author admission")
2984 .is_none()
2985 );
2986 }
2987}
2988