tenant.openagents/omega

1//! Sarah Nostr conversation client for `openagents.omega.effectd.v1`.
2//!
3//! Packet: SARAH-NR-06 (OpenAgentsInc/omega#33).
4//! Spec: docs/omega/2026-07-24-sarah-workroom-mvp-spec.md §8, §24.7.
5//!
6//! This module is the only conversation client for the Sarah lane. It must
7//! never link a Khala Sync client. The backing transport is a Nostr relay
8//! adapter: mock/in-memory for local tests, NIP-42 authenticated when a real
9//! relay URL and identity key are configured.
10
11#[cfg(test)]
12use std::cell::Cell;
13use std::collections::{BTreeMap, VecDeque};
14use std::fs::{self, OpenOptions};
15use std::io::{Read, Write};
16use std::path::{Path, PathBuf};
17use std::sync::Arc;
18use std::time::{SystemTime, UNIX_EPOCH};
19
20use nostr::{
21    Event, EventBuilder, JsonUtil, Keys, Kind, PublicKey, RelayUrl, Tag,
22    nips::{
23        nip42,
24        nip44::{self, Version as Nip44Version},
25    },
26};
27use omega_identity::{
28    AdmittedSigningRequest, IdentityService, NostrPublicKeyHex, PrivateMessageRequest, ReceiptRef,
29    SigningPurpose, UnsignedEventTemplate,
30};
31use serde::{Deserialize, Serialize};
32use serde_json::{Value, json};
33use sha2::{Digest, Sha256};
34use thiserror::Error;
35
36use crate::protocol::{MAX_FRAME_BYTES, PROTOCOL_SCHEMA};
37use crate::{
38    ISSUE31_ADJUNCT_DELIVERY_KEYS, ISSUE31_COMMAND_SCHEMA, ISSUE31_COMMAND_SCHEMA_V2,
39    ISSUE31_FULL_AUTO_ADJUNCT_RECORD_TYPE, ISSUE31_FULL_AUTO_ADJUNCT_SCHEMA,
40    ISSUE31_HOST_ADJUNCT_RECORD_TYPE, ISSUE31_HOST_ADJUNCT_SCHEMA, ISSUE31_HOST_DISCOVERY_KIND,
41    ISSUE31_PAIRING_SCHEMA, Issue31AuthorityDecisionProjection, Issue31CommandArguments,
42    Issue31CommandEvent, Issue31CommandExecution, Issue31CommandExecutionV2,
43    Issue31CommandHandlingStatus, Issue31CommandRecord, Issue31CommandRecordV2,
44    Issue31CommandStatus, Issue31HostConfiguration, Issue31HostController, Issue31HostDiscovery,
45    Issue31HostDiscoveryV2, Issue31NostrError, Issue31OwnerProjectionBody,
46    Issue31GrantState, Issue31OwnerProjectionInput, Issue31PairingEvent, Issue31PairingRecord,
47    Issue31SourceRole, Issue31TargetOutcomeProjection, Issue31WithheldCause,
48    Issue31WithheldSourceCount, Issue31WithheldSourcesInput, SARAH_AUTHORITY_RECEIPT_KIND,
49    SARAH_ENGRAM_KIND, SARAH_READ_STATE_KIND, SARAH_REMINDER_KIND, emit_issue31_owner_projection,
50    emit_issue31_withheld_sources,
51};
52use crate::issue31_provider_handoff::{
53    ISSUE31_ACTION_REQUEST_PROVIDER_HANDOFF, Issue31ProviderHandoffLedger,
54    Issue31ProviderRosterAccount,
55};
56
57pub use crate::openagents_binding::BindingState;
58
59/// Framed method names on `openagents.omega.effectd.v1` for the Sarah room.
60pub const SARAH_METHOD_SESSION_STATUS: &str = "sarah_session_status";
61pub const SARAH_METHOD_BOOTSTRAP: &str = "sarah_bootstrap";
62pub const SARAH_METHOD_ROOM_SNAPSHOT: &str = "sarah_room_snapshot";
63pub const SARAH_METHOD_SEND_MESSAGE: &str = "sarah_send_message";
64pub const SARAH_METHOD_INTERRUPT_TURN: &str = "sarah_interrupt_turn";
65pub const SARAH_METHOD_DEVICE_GRANTS: &str = "sarah_device_grants";
66pub const SARAH_METHOD_RENEW_DEVICE_GRANT: &str = "sarah_renew_device_grant";
67pub const SARAH_METHOD_REVOKE_DEVICE_GRANT: &str = "sarah_revoke_device_grant";
68/// Owner re-admission of a device whose grant was revoked. Revocation fails
69/// closed for the device, not just the grant, so without this the owner has no
70/// way to let a device back in.
71pub const SARAH_METHOD_READMIT_DEVICE: &str = "sarah_readmit_device";
72pub const SARAH_EVENT_ROOM_EVENT: &str = "sarah_room_event";
73pub const SARAH_EVENT_ROOM_STATE: &str = "sarah_room_state";
74
75pub const SARAH_FRAMED_METHODS: &[&str] = &[
76    SARAH_METHOD_SESSION_STATUS,
77    SARAH_METHOD_BOOTSTRAP,
78    SARAH_METHOD_ROOM_SNAPSHOT,
79    SARAH_METHOD_SEND_MESSAGE,
80    SARAH_METHOD_INTERRUPT_TURN,
81    SARAH_METHOD_DEVICE_GRANTS,
82    SARAH_METHOD_RENEW_DEVICE_GRANT,
83    SARAH_METHOD_REVOKE_DEVICE_GRANT,
84    SARAH_METHOD_READMIT_DEVICE,
85];
86
87/// NIP-AO ephemeral control kind used for interrupt / cancel_turn.
88pub const NIP_AO_KIND: u16 = 24200;
89/// Durable Sarah turn-record kind (SARAH-NR-00).
90pub const SARAH_TURN_RECORD_KIND: u16 = 44300;
91
92const DEFAULT_PAGE_LIMIT: usize = 32;
93const MAX_PAGE_LIMIT: usize = 64;
94const MAX_PENDING_EVENTS: usize = 256;
95const MAX_COMMAND_RESULTS: usize = 4_096;
96const MAX_PRIVATE_OUTBOX_ITEMS: usize = 1_024;
97const MAX_RELAY_ACKNOWLEDGEMENTS: usize = 4_096;
98const MAX_QUARANTINED_ISSUE31_EVENTS: usize = 4_096;
99/// The quarantine reason recorded when a source event cannot become a
100/// projection. It is the only quarantine reason that withholds something the
101/// owner was entitled to read; the pairing and command reasons quarantine
102/// control records, which are not part of the owner's view.
103const ISSUE31_PROJECTION_SOURCE_QUARANTINE_REASON: &str = "reason.omega.invalid_projection_source";
104const ISSUE31_PROJECTION_SCAN_BOUND_REASON: &str = "reason.omega.projection_scan_bound";
105const CURSOR_PREFIX: &str = "cursor.";
106const MOCK_RELAY_LABEL: &str = "mock://local";
107
108#[derive(Debug, Error)]
109pub enum SarahConversationError {
110    #[error("stale generation: expected {expected}, got {got}")]
111    StaleGeneration { expected: u64, got: u64 },
112    #[error("invalid request: {0}")]
113    InvalidRequest(String),
114    #[error("identity required for authenticated relay")]
115    IdentityRequired,
116    #[error("identity custody error: {0}")]
117    Identity(String),
118    #[error("relay error: {0}")]
119    Relay(String),
120    #[error("internal: {0}")]
121    Internal(String),
122}
123
124impl SarahConversationError {
125    pub fn protocol_code(&self) -> crate::protocol::ProtocolErrorCode {
126        use crate::protocol::ProtocolErrorCode;
127        match self {
128            Self::StaleGeneration { .. } => ProtocolErrorCode::StaleGeneration,
129            Self::InvalidRequest(_) | Self::IdentityRequired => ProtocolErrorCode::InvalidRequest,
130            Self::Identity(_) => ProtocolErrorCode::HostUnavailable,
131            Self::Relay(_) => ProtocolErrorCode::HostUnavailable,
132            Self::Internal(_) => ProtocolErrorCode::Internal,
133        }
134    }
135}
136
137/// Public-safe identity projection. Never carries a private key or token.
138#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
139#[serde(rename_all = "camelCase")]
140pub struct ConversationIdentity {
141    pub owner_public_key_hex: String,
142    pub sarah_public_key_hex: String,
143    pub account_label: Option<String>,
144    /// Binding state for metering attribution (OMEGA-SW-01). Not a session token.
145    pub binding_state: BindingState,
146}
147
148#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
149#[serde(rename_all = "camelCase")]
150pub struct SessionStatusResult {
151    pub signed_in: bool,
152    pub account_label: Option<String>,
153    pub binding_state: BindingState,
154    pub owner_public_key_hex: Option<String>,
155    /// ISO-8601 expiry for the OpenAgents account binding, when known.
156    pub binding_expires_at: Option<String>,
157    pub transport: String,
158}
159
160#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
161#[serde(rename_all = "camelCase")]
162pub struct BootstrapResult {
163    pub principal_ref: String,
164    pub display_name: String,
165    pub role: String,
166    pub conversation_ref: String,
167    pub legacy_thread_ref: String,
168    pub owner_public_key_hex: String,
169    pub sarah_public_key_hex: String,
170    pub authority_profile_ref: String,
171    pub authority_profile_revision: u32,
172    pub admitted_device_fingerprints: Vec<String>,
173    pub quarantined_issue31_event_count: usize,
174    pub room_state: RoomStateEvent,
175}
176
177#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
178#[serde(rename_all = "camelCase")]
179pub struct RoomSnapshotResult {
180    pub conversation_ref: String,
181    pub transcript: TranscriptPage,
182    pub activity: ActivityPage,
183    pub nostr_records: NostrRecordPage,
184    pub run_state: RunStateProjection,
185    pub room_state: RoomStateEvent,
186}
187
188#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
189#[serde(rename_all = "camelCase")]
190pub struct NostrRecordPage {
191    pub entries: Vec<NostrRecordRef>,
192    pub cursor: String,
193    pub next_cursor: Option<String>,
194    pub gap_state: GapState,
195    pub source: String,
196}
197
198#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
199#[serde(rename_all = "camelCase")]
200pub struct NostrRecordRef {
201    pub event_id: String,
202    pub cursor: String,
203    pub kind: u16,
204    pub record_kind: String,
205    pub author_fingerprint: String,
206    pub created_at: String,
207    pub source: String,
208}
209
210#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
211#[serde(rename_all = "camelCase")]
212pub struct TranscriptPage {
213    pub entries: Vec<TranscriptEntry>,
214    pub cursor: String,
215    pub next_cursor: Option<String>,
216    pub gap_state: GapState,
217}
218
219#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
220#[serde(rename_all = "camelCase")]
221pub struct TranscriptEntry {
222    pub event_id: String,
223    pub cursor: String,
224    pub role: String,
225    pub kind: String,
226    pub text: String,
227    pub created_at: String,
228    pub status: String,
229}
230
231#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
232#[serde(rename_all = "camelCase")]
233pub struct ActivityPage {
234    pub entries: Vec<ActivityEntry>,
235    pub cursor: String,
236    pub next_cursor: Option<String>,
237    pub gap_state: GapState,
238}
239
240#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
241#[serde(rename_all = "camelCase")]
242pub struct ActivityEntry {
243    pub event_id: String,
244    pub cursor: String,
245    pub entry: String,
246    pub turn_ref: String,
247    pub created_at: String,
248}
249
250#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
251#[serde(rename_all = "camelCase")]
252pub struct RunStateProjection {
253    pub state: String,
254    pub turn_ref: Option<String>,
255    pub reason: Option<String>,
256}
257
258#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
259#[serde(rename_all = "camelCase")]
260pub struct SendMessageResult {
261    pub accepted: bool,
262    pub message_ref: String,
263    pub turn_ref: String,
264    pub event_id: String,
265    pub cursor: String,
266    pub status: String,
267}
268
269#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
270#[serde(rename_all = "camelCase")]
271pub struct InterruptTurnResult {
272    pub accepted: bool,
273    pub turn_ref: String,
274    pub intent_ref: String,
275    pub status: String,
276    pub pending: bool,
277}
278
279#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
280#[serde(rename_all = "camelCase")]
281pub struct RoomEventPayload {
282    pub method: String,
283    pub conversation_ref: String,
284    pub cursor: String,
285    pub record: TranscriptEntry,
286}
287
288#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
289#[serde(rename_all = "camelCase")]
290pub struct RoomStateEvent {
291    pub method: String,
292    pub connection: ConnectionState,
293    pub freshness: FreshnessState,
294    pub gap_state: GapState,
295    pub connected_relays: Vec<String>,
296    pub last_acknowledged_event_id: Option<String>,
297    pub last_acknowledged_cursor: Option<String>,
298    pub authenticated: bool,
299    pub transport: String,
300}
301
302#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
303#[serde(rename_all = "snake_case")]
304pub enum ConnectionState {
305    Disconnected,
306    Connecting,
307    Connected,
308    Degraded,
309}
310
311#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
312#[serde(rename_all = "snake_case")]
313pub enum FreshnessState {
314    Fresh,
315    Stale,
316    Unknown,
317}
318
319#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
320#[serde(rename_all = "snake_case")]
321pub enum GapState {
322    None,
323    Possible,
324    Confirmed,
325    Recovering,
326}
327
328#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
329#[serde(rename_all = "camelCase")]
330pub struct RelayAuthChallenge {
331    pub challenge: String,
332    pub relay_url: String,
333}
334
335/// Stored public-safe event in the conversation store.
336#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
337#[serde(rename_all = "camelCase")]
338pub struct StoredConversationEvent {
339    pub event_id: String,
340    pub kind: u16,
341    pub pubkey: String,
342    pub created_at: u64,
343    pub conversation_ref: String,
344    pub content_summary: String,
345    pub tags: Vec<Vec<String>>,
346    /// Whether this is an owner message, Sarah answer, activity entry, or control.
347    pub record_kind: String,
348    pub store_index: usize,
349}
350
351#[derive(Debug, Clone)]
352pub struct QueryPage {
353    pub events: Vec<StoredConversationEvent>,
354    pub next_cursor: Option<String>,
355    pub gap_state: GapState,
356}
357
358/// Abstract Nostr relay transport. Production will use a real WebSocket client;
359/// tests and local-dev without a relay use [`MockRelayAdapter`].
360pub trait RelayTransport: Send {
361    fn label(&self) -> &str;
362    fn connection_state(&self) -> ConnectionState;
363    fn is_authenticated(&self) -> bool;
364    fn connect(&mut self) -> Result<(), SarahConversationError>;
365    fn auth_challenge(&self) -> Option<RelayAuthChallenge>;
366    fn authenticate(&mut self, auth_event: &Event) -> Result<(), SarahConversationError>;
367    fn publish(&mut self, event: &Event) -> Result<(), SarahConversationError>;
368    fn publication_complete(&mut self, _event_id: &str) -> bool {
369        true
370    }
371    fn acknowledged_relays(&self, _event_id: &str) -> Vec<String> {
372        vec![self.label().to_string()]
373    }
374    fn restore_publication_acknowledgements(&mut self, _event_id: &str, _relay_urls: &[String]) {}
375    fn query(
376        &mut self,
377        conversation_ref: &str,
378        after_cursor: Option<&str>,
379        limit: usize,
380    ) -> Result<QueryPage, SarahConversationError>;
381    fn last_event_id(&self) -> Option<String>;
382    fn gap_state(&self) -> GapState {
383        GapState::None
384    }
385    fn connected_relays(&self) -> Vec<String> {
386        vec![self.label().to_string()]
387    }
388    fn requires_private_messages(&self) -> bool {
389        false
390    }
391}
392
393/// In-memory mock relay for local/dev and unit tests.
394#[derive(Debug, Default)]
395pub struct MockRelayAdapter {
396    label: String,
397    connected: bool,
398    authenticated: bool,
399    require_auth: bool,
400    challenge: Option<String>,
401    events: Vec<StoredConversationEvent>,
402}
403
404impl MockRelayAdapter {
405    pub fn new() -> Self {
406        Self {
407            label: MOCK_RELAY_LABEL.to_string(),
408            connected: false,
409            authenticated: false,
410            require_auth: false,
411            challenge: None,
412            events: Vec::new(),
413        }
414    }
415
416    /// Local mock that still exercises the NIP-42 AUTH path.
417    pub fn with_required_auth(challenge: impl Into<String>) -> Self {
418        Self {
419            label: "mock://auth-required".to_string(),
420            require_auth: true,
421            challenge: Some(challenge.into()),
422            ..Self::new()
423        }
424    }
425
426    pub fn seed_event(&mut self, mut event: StoredConversationEvent) {
427        event.store_index = self.events.len();
428        self.events.push(event);
429    }
430
431    pub fn event_count(&self) -> usize {
432        self.events.len()
433    }
434}
435
436impl RelayTransport for MockRelayAdapter {
437    fn label(&self) -> &str {
438        &self.label
439    }
440
441    fn connection_state(&self) -> ConnectionState {
442        if self.connected {
443            ConnectionState::Connected
444        } else {
445            ConnectionState::Disconnected
446        }
447    }
448
449    fn is_authenticated(&self) -> bool {
450        self.authenticated || !self.require_auth
451    }
452
453    fn connect(&mut self) -> Result<(), SarahConversationError> {
454        self.connected = true;
455        if self.require_auth {
456            self.authenticated = false;
457            if self.challenge.is_none() {
458                self.challenge = Some(format!("challenge-{}", self.events.len()));
459            }
460        } else {
461            self.authenticated = true;
462        }
463        Ok(())
464    }
465
466    fn auth_challenge(&self) -> Option<RelayAuthChallenge> {
467        if self.require_auth && !self.authenticated {
468            Some(RelayAuthChallenge {
469                challenge: self.challenge.clone().unwrap_or_default(),
470                relay_url: "wss://relay.openagents.com".to_string(),
471            })
472        } else {
473            None
474        }
475    }
476
477    fn authenticate(&mut self, auth_event: &Event) -> Result<(), SarahConversationError> {
478        let challenge = self
479            .challenge
480            .as_deref()
481            .ok_or_else(|| SarahConversationError::Relay("no auth challenge pending".into()))?;
482        let relay_url = RelayUrl::parse("wss://relay.openagents.com")
483            .map_err(|error| SarahConversationError::Relay(error.to_string()))?;
484        if !nip42::is_valid_auth_event(auth_event, &relay_url, challenge) {
485            return Err(SarahConversationError::Relay(
486                "NIP-42 auth event rejected".into(),
487            ));
488        }
489        if auth_event.verify().is_err() {
490            return Err(SarahConversationError::Relay(
491                "NIP-42 auth signature invalid".into(),
492            ));
493        }
494        self.authenticated = true;
495        Ok(())
496    }
497
498    fn publish(&mut self, event: &Event) -> Result<(), SarahConversationError> {
499        if !self.connected {
500            return Err(SarahConversationError::Relay("not connected".into()));
501        }
502        if self.require_auth && !self.authenticated {
503            return Err(SarahConversationError::Relay(
504                "authentication required before publish".into(),
505            ));
506        }
507        let conversation_ref = event
508            .tags
509            .iter()
510            .find_map(|tag| {
511                let slice = tag.as_slice();
512                if slice.first().map(String::as_str) == Some("conversation") {
513                    slice.get(1).cloned()
514                } else {
515                    None
516                }
517            })
518            .unwrap_or_else(|| "sarah.unknown".to_string());
519        let record_kind = match event.kind.as_u16() {
520            NIP_AO_KIND => "control",
521            SARAH_TURN_RECORD_KIND => "activity",
522            SARAH_READ_STATE_KIND => "read_state",
523            SARAH_REMINDER_KIND => "reminder",
524            _ => "message",
525        };
526        let store_index = self.events.len();
527        self.events.push(StoredConversationEvent {
528            event_id: event.id.to_hex(),
529            kind: event.kind.as_u16(),
530            pubkey: event.pubkey.to_hex(),
531            created_at: event.created_at.as_secs(),
532            conversation_ref,
533            content_summary: redact_content_summary(&event.content),
534            tags: event
535                .tags
536                .iter()
537                .map(|tag| tag.as_slice().to_vec())
538                .collect(),
539            record_kind: record_kind.to_string(),
540            store_index,
541        });
542        Ok(())
543    }
544
545    fn query(
546        &mut self,
547        conversation_ref: &str,
548        after_cursor: Option<&str>,
549        limit: usize,
550    ) -> Result<QueryPage, SarahConversationError> {
551        if !self.connected {
552            return Err(SarahConversationError::Relay("not connected".into()));
553        }
554        let matching: Vec<StoredConversationEvent> = self
555            .events
556            .iter()
557            .filter(|event| {
558                event.conversation_ref == conversation_ref
559                    || matches!(event.kind, SARAH_READ_STATE_KIND | SARAH_REMINDER_KIND)
560            })
561            .cloned()
562            .collect();
563        let start_index = after_cursor
564            .and_then(|cursor| {
565                matching
566                    .iter()
567                    .position(|event| stored_event_cursor(event) == cursor)
568            })
569            .map(|index| index.saturating_add(1))
570            .unwrap_or(0);
571        let page: Vec<StoredConversationEvent> = matching
572            .iter()
573            .skip(start_index)
574            .take(limit)
575            .cloned()
576            .collect();
577        let next_cursor = if start_index.saturating_add(page.len()) < matching.len() {
578            page.last().map(stored_event_cursor)
579        } else {
580            None
581        };
582        Ok(QueryPage {
583            events: page,
584            next_cursor,
585            gap_state: GapState::None,
586        })
587    }
588
589    fn last_event_id(&self) -> Option<String> {
590        self.events.last().map(|event| event.event_id.clone())
591    }
592}
593
594/// Optional identity material used only for signing. Secrets never leave this
595/// struct into framed protocol responses.
596pub struct SigningIdentity {
597    pub public_key_hex: String,
598    keys: Keys,
599}
600
601impl SigningIdentity {
602    pub fn from_keys(keys: Keys) -> Self {
603        Self {
604            public_key_hex: keys.public_key().to_hex(),
605            keys,
606        }
607    }
608
609    pub fn generate() -> Self {
610        Self::from_keys(Keys::generate())
611    }
612
613    pub fn sign_auth(
614        &self,
615        challenge: &str,
616        relay_url: &str,
617    ) -> Result<Event, SarahConversationError> {
618        let relay = RelayUrl::parse(relay_url)
619            .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
620        EventBuilder::auth(challenge, relay)
621            .sign_with_keys(&self.keys)
622            .map_err(|error| SarahConversationError::Internal(error.to_string()))
623    }
624
625    pub fn sign_text_note(
626        &self,
627        content: &str,
628        tags: Vec<Tag>,
629    ) -> Result<Event, SarahConversationError> {
630        EventBuilder::text_note(content)
631            .tags(tags)
632            .sign_with_keys(&self.keys)
633            .map_err(|error| SarahConversationError::Internal(error.to_string()))
634    }
635
636    pub fn sign_custom(
637        &self,
638        kind: u16,
639        content: &str,
640        tags: Vec<Tag>,
641    ) -> Result<Event, SarahConversationError> {
642        EventBuilder::new(Kind::from(kind), content)
643            .tags(tags)
644            .sign_with_keys(&self.keys)
645            .map_err(|error| SarahConversationError::Internal(error.to_string()))
646    }
647}
648
649enum ConversationSigner {
650    Keys(SigningIdentity),
651    OmegaIdentity(Arc<IdentityService>),
652}
653
654impl ConversationSigner {
655    fn sign_public_record(
656        &self,
657        kind: u16,
658        content: &str,
659        tags: Vec<Tag>,
660    ) -> Result<Event, SarahConversationError> {
661        match self {
662            Self::Keys(identity) => identity.sign_custom(kind, content, tags),
663            Self::OmegaIdentity(identity_service) => {
664                let custody = identity_service
665                    .inspect()
666                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
667                let identity = custody
668                    .identity
669                    .ok_or(SarahConversationError::IdentityRequired)?;
670                let semantic_binding = serde_json::to_vec(&json!({
671                    "kind": kind,
672                    "content": content,
673                    "tags": tags.iter().map(|tag| tag.as_slice()).collect::<Vec<_>>(),
674                }))
675                .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
676                let request = AdmittedSigningRequest {
677                    request_ref: digest_receipt_ref("issue31.public", &semantic_binding)?,
678                    identity_ref: identity.identity_ref().clone(),
679                    purpose: SigningPurpose::NostrEvent,
680                    event: UnsignedEventTemplate {
681                        created_at: unix_now(),
682                        kind,
683                        tags: tags.iter().map(|tag| tag.as_slice().to_vec()).collect(),
684                        content: content.to_string(),
685                    },
686                };
687                let signed = identity_service
688                    .sign(&request)
689                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
690                Event::from_json(signed.signed_event_json)
691                    .map_err(|error| SarahConversationError::Identity(error.to_string()))
692            }
693        }
694    }
695
696    fn sign_encrypted_self_record(
697        &self,
698        kind: u16,
699        plaintext: &str,
700        tags: Vec<Tag>,
701    ) -> Result<Event, SarahConversationError> {
702        match self {
703            Self::Keys(identity) => {
704                let ciphertext = nip44::encrypt(
705                    identity.keys.secret_key(),
706                    &identity.keys.public_key(),
707                    plaintext.as_bytes(),
708                    Nip44Version::V2,
709                )
710                .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
711                identity.sign_custom(kind, &ciphertext, tags)
712            }
713            Self::OmegaIdentity(identity_service) => {
714                let custody = identity_service
715                    .inspect()
716                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
717                let identity = custody
718                    .identity
719                    .ok_or(SarahConversationError::IdentityRequired)?;
720                let semantic_binding = serde_json::to_vec(&json!({
721                    "kind": kind,
722                    "plaintextDigest": format!("{:x}", Sha256::digest(plaintext.as_bytes())),
723                    "tags": tags.iter().map(|tag| tag.as_slice()).collect::<Vec<_>>(),
724                }))
725                .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
726                let request = AdmittedSigningRequest {
727                    request_ref: digest_receipt_ref("issue31.encrypted-self", &semantic_binding)?,
728                    identity_ref: identity.identity_ref().clone(),
729                    purpose: SigningPurpose::Nip44EncryptedSelfEvent,
730                    event: UnsignedEventTemplate {
731                        created_at: unix_now(),
732                        kind,
733                        tags: tags.iter().map(|tag| tag.as_slice().to_vec()).collect(),
734                        content: plaintext.to_string(),
735                    },
736                };
737                let signed = identity_service
738                    .sign_nip44_encrypted_to_self(&request)
739                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
740                Event::from_json(signed.signed_event_json)
741                    .map_err(|error| SarahConversationError::Identity(error.to_string()))
742            }
743        }
744    }
745
746    fn decrypt_record(
747        &self,
748        sender_public_key_hex: &str,
749        ciphertext: &str,
750    ) -> Result<String, SarahConversationError> {
751        match self {
752            Self::Keys(identity) => {
753                let sender_public_key = PublicKey::from_hex(sender_public_key_hex)
754                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
755                nip44::decrypt(
756                    identity.keys.secret_key(),
757                    &sender_public_key,
758                    ciphertext.as_bytes(),
759                )
760                .map_err(|error| SarahConversationError::Identity(error.to_string()))
761            }
762            Self::OmegaIdentity(identity_service) => {
763                let sender_public_key = NostrPublicKeyHex::new(sender_public_key_hex)
764                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
765                identity_service
766                    .decrypt_nip44_from(&sender_public_key, ciphertext)
767                    .map_err(|error| SarahConversationError::Identity(error.to_string()))
768            }
769        }
770    }
771
772    fn sign_auth(&self, challenge: &str, relay_url: &str) -> Result<Event, SarahConversationError> {
773        match self {
774            Self::Keys(identity) => identity.sign_auth(challenge, relay_url),
775            Self::OmegaIdentity(identity_service) => {
776                let custody = identity_service
777                    .inspect()
778                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
779                let identity = custody
780                    .identity
781                    .ok_or(SarahConversationError::IdentityRequired)?;
782                let public_key = PublicKey::from_hex(identity.public_key_hex().as_str())
783                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
784                let relay = RelayUrl::parse(relay_url)
785                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
786                let unsigned = EventBuilder::auth(challenge, relay).build(public_key);
787                let request_ref = digest_receipt_ref("nip42", challenge.as_bytes())?;
788                let request = AdmittedSigningRequest {
789                    request_ref,
790                    identity_ref: identity.identity_ref().clone(),
791                    purpose: SigningPurpose::NostrEvent,
792                    event: UnsignedEventTemplate {
793                        created_at: unsigned.created_at.as_secs(),
794                        kind: unsigned.kind.as_u16(),
795                        tags: unsigned
796                            .tags
797                            .iter()
798                            .map(|tag| tag.as_slice().to_vec())
799                            .collect(),
800                        content: unsigned.content,
801                    },
802                };
803                let signed = identity_service
804                    .sign(&request)
805                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
806                Event::from_json(signed.signed_event_json)
807                    .map_err(|error| SarahConversationError::Identity(error.to_string()))
808            }
809        }
810    }
811
812    fn private_messages(
813        &self,
814        content: &str,
815        tags: Vec<Tag>,
816        recipients: &[String],
817    ) -> Result<(String, Vec<Event>), SarahConversationError> {
818        let created_at = unix_now();
819        let semantic_binding = serde_json::to_vec(&json!({
820            "content": content,
821            "tags": tags.iter().map(|tag| tag.as_slice()).collect::<Vec<_>>(),
822            "recipients": recipients,
823        }))
824        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
825        match self {
826            Self::Keys(identity) => {
827                let public_key = identity.keys.public_key();
828                let mut rumor = EventBuilder::new(Kind::PrivateDirectMessage, content)
829                    .tags(tags)
830                    .custom_created_at(nostr::Timestamp::from_secs(created_at))
831                    .build(public_key);
832                rumor.ensure_id();
833                rumor
834                    .verify_id()
835                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
836                let rumor_event_id = rumor
837                    .id
838                    .ok_or_else(|| {
839                        SarahConversationError::Identity(
840                            "private rumor omitted its convergence id".into(),
841                        )
842                    })?
843                    .to_hex();
844                let mut events = Vec::with_capacity(recipients.len());
845                for recipient in recipients {
846                    let recipient = PublicKey::from_hex(recipient).map_err(|error| {
847                        SarahConversationError::InvalidRequest(error.to_string())
848                    })?;
849                    let gift_wrap = smol::block_on(EventBuilder::gift_wrap(
850                        &identity.keys,
851                        &recipient,
852                        rumor.clone(),
853                        [],
854                    ))
855                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
856                    events.push(gift_wrap);
857                }
858                Ok((rumor_event_id, events))
859            }
860            Self::OmegaIdentity(identity_service) => {
861                let custody = identity_service
862                    .inspect()
863                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
864                let identity = custody
865                    .identity
866                    .ok_or(SarahConversationError::IdentityRequired)?;
867                let recipients = recipients
868                    .iter()
869                    .map(NostrPublicKeyHex::new)
870                    .collect::<Result<Vec<_>, _>>()
871                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
872                let request = PrivateMessageRequest {
873                    request_ref: digest_receipt_ref("nip17", &semantic_binding)?,
874                    identity_ref: identity.identity_ref().clone(),
875                    recipients,
876                    rumor: UnsignedEventTemplate {
877                        created_at,
878                        kind: Kind::PrivateDirectMessage.as_u16(),
879                        tags: tags.iter().map(|tag| tag.as_slice().to_vec()).collect(),
880                        content: content.to_string(),
881                    },
882                };
883                let wrapped = identity_service
884                    .gift_wrap_private_message(&request)
885                    .map_err(|error| SarahConversationError::Identity(error.to_string()))?;
886                let rumor_event_id = wrapped
887                    .first()
888                    .map(|wrapped| wrapped.rumor_event_id.clone())
889                    .ok_or_else(|| {
890                        SarahConversationError::Identity(
891                            "private message produced no gift wraps".into(),
892                        )
893                    })?;
894                let events = wrapped
895                    .into_iter()
896                    .map(|wrapped| {
897                        Event::from_json(wrapped.gift_wrap_event_json)
898                            .map_err(|error| SarahConversationError::Identity(error.to_string()))
899                    })
900                    .collect::<Result<Vec<_>, _>>()?;
901                Ok((rumor_event_id, events))
902            }
903        }
904    }
905}
906
907/// Configuration for the conversation client.
908#[derive(Debug, Clone)]
909pub struct SarahConversationConfig {
910    pub generation: u64,
911    pub conversation_digest: String,
912    pub identity: ConversationIdentity,
913    /// When set, the client treats the transport as a real relay and runs NIP-42.
914    pub relay_url: Option<String>,
915    pub admitted_device_public_key_hexes: Vec<String>,
916    pub approved_device_scopes: Vec<crate::Issue31PairingScope>,
917    pub community_group_ids: Vec<String>,
918    pub community_public_key_hexes: Vec<String>,
919}
920
921impl SarahConversationConfig {
922    pub fn mock_fixture() -> Self {
923        Self {
924            generation: 1,
925            conversation_digest: "a".repeat(24),
926            identity: ConversationIdentity {
927                owner_public_key_hex: "b".repeat(64),
928                sarah_public_key_hex: "c".repeat(64),
929                account_label: Some("owner@example.com".to_string()),
930                binding_state: BindingState::Bound,
931            },
932            relay_url: None,
933            admitted_device_public_key_hexes: Vec::new(),
934            approved_device_scopes: Vec::new(),
935            community_group_ids: Vec::new(),
936            community_public_key_hexes: Vec::new(),
937        }
938    }
939
940    pub fn conversation_ref(&self) -> String {
941        format!("sarah.{}", self.conversation_digest)
942    }
943
944    pub fn legacy_thread_ref(&self) -> String {
945        format!("thread.sarah.{}", self.conversation_digest)
946    }
947}
948
949/// Nostr conversation client owned by omega-effectd for the Sarah lane.
950/// One admitted device's standing, handed to the Full Auto reader (omega#49).
951///
952/// The host pump owns this shape rather than the panel: the panel knows what
953/// the runs are, and only the pump knows which devices are entitled to see
954/// them and under which grant. Building the snapshot per grant is what makes
955/// `connection_identity` a statement about *this* reader instead of a generic
956/// one.
957#[derive(Clone, Copy, Debug)]
958pub struct Issue31HostProjectionRequest<'a> {
959    pub host_ref: &'a str,
960    pub host_public_key_hex: &'a str,
961    pub device_public_key_hex: &'a str,
962    pub grant_ref: &'a str,
963    pub expected_generation: u64,
964    /// The pump's reading time, in epoch milliseconds.
965    pub observed_at_ms: u64,
966    /// The host's own provider connection handoff ledger (omega#91).
967    ///
968    /// This comes from the pump rather than from the Full Auto reading for the
969    /// same reason the delivery binding does: the ledger is durable host state
970    /// that survives a restart, and the panel's reading is a transient
971    /// observation of the daemon. One of them is the record; letting the other
972    /// also carry handoffs would give the phone two sources for one fact.
973    ///
974    /// The ledger is handed over unprojected on purpose. Only the builder knows
975    /// the `generatedAtMs` the documents will carry, and a handoff row must be
976    /// checked against exactly that stamp — projecting here, against the pump's
977    /// clock, would accept rows the assembled document then contradicts.
978    pub handoffs: &'a Issue31ProviderHandoffLedger,
979}
980
981/// The two omega#47 documents `publish_issue31_host_snapshot` produces.
982///
983/// They travel as JSON because the builder lives in `full_auto_ui`, which
984/// depends on this crate. Inverting that to call the builder from here would
985/// make the dependency circular, so the reading crosses the seam as data.
986#[derive(Clone, Debug)]
987pub struct Issue31HostProjectionDocuments {
988    pub host: Value,
989    pub detail: Value,
990}
991
992/// A live reading of Full Auto host state, or `None`.
993///
994/// `None` means this host cannot presently state its Full Auto view — no
995/// supervisor attached, no panel reading, nothing observed. It is not an empty
996/// view: an empty view is a `Some` carrying zero runs, which the device renders
997/// as "this host is running nothing". Publishing an invented empty projection
998/// for an unobserved host would be exactly the false claim omega#49 forbids.
999pub type Issue31HostProjectionSource = Arc<
1000    dyn Fn(
1001            &Issue31HostProjectionRequest<'_>,
1002        ) -> Result<Option<Issue31HostProjectionDocuments>, String>
1003        + Send
1004        + Sync,
1005>;
1006
1007/// The host's own reading of its provider accounts, or `None` (omega#91).
1008///
1009/// `None` means the host has not looked at its roster. Nothing about a handoff
1010/// advances on an unread roster: a binding decided against state nobody read
1011/// would be a guess wearing a measurement's clothes. `Some(&[])` — the host
1012/// looked and holds no accounts — is a real observation and does move a
1013/// handoff's clock towards its deadline.
1014pub type Issue31ProviderRosterSource =
1015    Arc<dyn Fn() -> Option<Vec<Issue31ProviderRosterAccount>> + Send + Sync>;
1016
1017pub struct SarahConversationClient {
1018    config: SarahConversationConfig,
1019    relay: Box<dyn RelayTransport>,
1020    signer: ConversationSigner,
1021    pending_events: VecDeque<Value>,
1022    active_turn_ref: Option<String>,
1023    run_state: String,
1024    message_seq: u64,
1025    command_results: BTreeMap<String, (String, Value)>,
1026    last_gap_state: GapState,
1027    last_confirmed_cursor: Option<String>,
1028    issue31_host: Option<Issue31HostController>,
1029    issue31_discovery_generation: Option<u64>,
1030    issue31_discovery_expires_at: Option<u64>,
1031    issue31_discovery_outbox: Option<Event>,
1032    issue31_private_outbox: BTreeMap<String, PendingIssue31PrivatePublish>,
1033    issue31_relay_acknowledgements: BTreeMap<String, Vec<String>>,
1034    issue31_control_cursor: Option<String>,
1035    issue31_projection_cursor: Option<String>,
1036    issue31_quarantined_events: BTreeMap<String, String>,
1037    /// The last coverage statement published to each `grant_ref:generation`,
1038    /// so a re-run that observed the same world does not republish it with only
1039    /// a new timestamp.
1040    issue31_withheld_emissions: BTreeMap<String, (String, Vec<Issue31WithheldSourceCount>)>,
1041    /// The digest of the last omega#47 publication sent to each
1042    /// `grant_ref:generation`, so a pump run that observed the same host state
1043    /// does not re-send the same snapshot with only a fresh timestamp.
1044    issue31_host_adjunct_emissions: BTreeMap<String, String>,
1045    /// The live Full Auto reading, supplied by whoever holds the supervisor.
1046    /// Absent means this host publishes no omega#47 records at all.
1047    issue31_host_projection_source: Option<Issue31HostProjectionSource>,
1048    /// Host-owned provider connection handoffs (omega#91). Durable: this is the
1049    /// record, not a cache of one.
1050    issue31_provider_handoffs: Issue31ProviderHandoffLedger,
1051    /// How the host reads its own provider roster when deciding a handoff.
1052    issue31_provider_roster_source: Option<Issue31ProviderRosterSource>,
1053    issue31_state_path: Option<PathBuf>,
1054    #[cfg(test)]
1055    issue31_fail_commit_after: Cell<Option<usize>>,
1056}
1057
1058#[derive(Clone)]
1059struct PendingIssue31PrivatePublish {
1060    rumor_event_id: String,
1061    gift_wraps: Vec<Event>,
1062}
1063
1064struct EnqueuedIssue31PrivateRecord {
1065    rumor_event_id: String,
1066    outbox_ref: String,
1067    inserted: bool,
1068}
1069
1070const ISSUE31_DURABLE_STATE_SCHEMA: &str = "openagents.omega.issue31.host_state.v1";
1071const ISSUE31_DURABLE_STATE_MAX_BYTES: u64 = 16 * 1024 * 1024;
1072const ISSUE31_NOSTR_HOST_GENERATION: u64 = 1;
1073
1074#[derive(Serialize, Deserialize)]
1075#[serde(rename_all = "camelCase", deny_unknown_fields)]
1076struct DurableIssue31HostState {
1077    schema: String,
1078    controller: Issue31HostController,
1079    discovery_generation: Option<u64>,
1080    discovery_expires_at: Option<u64>,
1081    discovery_event_json: Option<String>,
1082    private_outbox: BTreeMap<String, DurableIssue31PrivatePublish>,
1083    relay_acknowledgements: BTreeMap<String, Vec<String>>,
1084    control_cursor: Option<String>,
1085    #[serde(default)]
1086    projection_cursor: Option<String>,
1087    #[serde(default)]
1088    quarantined_events: BTreeMap<String, String>,
1089    #[serde(default)]
1090    host_adjunct_emissions: BTreeMap<String, String>,
1091    /// omega#91. `#[serde(default)]` so a state file written before handoffs
1092    /// existed still loads — as an empty ledger, which is the truth about a
1093    /// host that never had one, not a set of rows invented at load.
1094    #[serde(default)]
1095    provider_handoffs: Issue31ProviderHandoffLedger,
1096    command_results: BTreeMap<String, (String, Value)>,
1097    #[serde(default)]
1098    active_turn_ref: Option<String>,
1099    #[serde(default = "default_run_state")]
1100    run_state: String,
1101    #[serde(default)]
1102    message_seq: u64,
1103}
1104
1105#[derive(Serialize, Deserialize)]
1106#[serde(rename_all = "camelCase", deny_unknown_fields)]
1107struct DurableIssue31PrivatePublish {
1108    rumor_event_id: String,
1109    gift_wrap_event_json: Vec<String>,
1110}
1111
1112impl SarahConversationClient {
1113    /// Local/dev client with an in-memory mock relay (no network).
1114    pub fn new_mock(mut config: SarahConversationConfig) -> Self {
1115        let signer = SigningIdentity::generate();
1116        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
1117        Self::with_relay(config, Box::new(MockRelayAdapter::new()), signer)
1118    }
1119
1120    pub fn with_relay(
1121        config: SarahConversationConfig,
1122        relay: Box<dyn RelayTransport>,
1123        signer: SigningIdentity,
1124    ) -> Self {
1125        Self {
1126            config,
1127            relay,
1128            signer: ConversationSigner::Keys(signer),
1129            pending_events: VecDeque::new(),
1130            active_turn_ref: None,
1131            run_state: "idle".to_string(),
1132            message_seq: 0,
1133            command_results: BTreeMap::new(),
1134            last_gap_state: GapState::None,
1135            last_confirmed_cursor: None,
1136            issue31_host: None,
1137            issue31_discovery_generation: None,
1138            issue31_discovery_expires_at: None,
1139            issue31_discovery_outbox: None,
1140            issue31_private_outbox: BTreeMap::new(),
1141            issue31_relay_acknowledgements: BTreeMap::new(),
1142            issue31_control_cursor: None,
1143            issue31_projection_cursor: None,
1144            issue31_quarantined_events: BTreeMap::new(),
1145            issue31_withheld_emissions: BTreeMap::new(),
1146            issue31_host_adjunct_emissions: BTreeMap::new(),
1147            issue31_host_projection_source: None,
1148            issue31_provider_handoffs: Issue31ProviderHandoffLedger::default(),
1149            issue31_provider_roster_source: None,
1150            issue31_state_path: None,
1151            #[cfg(test)]
1152            issue31_fail_commit_after: Cell::new(None),
1153        }
1154    }
1155
1156    pub fn new_production(
1157        mut config: SarahConversationConfig,
1158        relay_urls: Vec<String>,
1159        identity_service: Arc<IdentityService>,
1160    ) -> Result<Self, SarahConversationError> {
1161        let owner_public_key_hex = identity_service
1162            .inspect()
1163            .map_err(|error| SarahConversationError::Identity(error.to_string()))?
1164            .identity
1165            .map(|identity| identity.public_key_hex().as_str().to_string())
1166            .ok_or(SarahConversationError::IdentityRequired)?;
1167        config.identity.owner_public_key_hex = owner_public_key_hex;
1168        config.relay_url = relay_urls.first().cloned();
1169        let mut relay = crate::nostr_websocket_relay::WebSocketRelayAdapter::new(
1170            relay_urls,
1171            identity_service.clone(),
1172            config.identity.sarah_public_key_hex.clone(),
1173            config.community_group_ids.clone(),
1174            config.community_public_key_hexes.clone(),
1175        )?;
1176        let relay_urls = relay.relay_urls().to_vec();
1177        let host_configuration = Issue31HostConfiguration {
1178            host_ref: "omega.host.local".into(),
1179            host_public_key_hex: config.identity.owner_public_key_hex.clone(),
1180            sarah_public_key_hex: config.identity.sarah_public_key_hex.clone(),
1181            conversation: config.conversation_ref(),
1182            display_name: "Local Omega".into(),
1183            relay_urls,
1184            generation: ISSUE31_NOSTR_HOST_GENERATION,
1185        };
1186        let issue31_state_path = paths::data_dir()
1187            .join("openagents")
1188            .join("issue31-nostr-host-state.json");
1189        let persisted = load_issue31_host_state(&issue31_state_path, &host_configuration)?;
1190        let mut issue31_host = match &persisted {
1191            Some(persisted) => persisted.controller.clone(),
1192            None => Issue31HostController::new(host_configuration).map_err(issue31_error)?,
1193        };
1194        issue31_host
1195            .set_admitted_device_policy(
1196                config.admitted_device_public_key_hexes.clone(),
1197                config.approved_device_scopes.clone(),
1198            )
1199            .map_err(issue31_error)?;
1200        let issue31_discovery_generation = persisted
1201            .as_ref()
1202            .and_then(|persisted| persisted.discovery_generation);
1203        let issue31_discovery_expires_at = persisted
1204            .as_ref()
1205            .and_then(|persisted| persisted.discovery_expires_at);
1206        let issue31_discovery_outbox = persisted
1207            .as_ref()
1208            .and_then(|persisted| persisted.discovery_event_json.as_deref())
1209            .map(Event::from_json)
1210            .transpose()
1211            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
1212        let issue31_private_outbox = persisted
1213            .as_ref()
1214            .map(|persisted| {
1215                durable_private_outbox_into_runtime(
1216                    persisted
1217                        .private_outbox
1218                        .iter()
1219                        .map(|(key, value)| {
1220                            (
1221                                key.clone(),
1222                                DurableIssue31PrivatePublish {
1223                                    rumor_event_id: value.rumor_event_id.clone(),
1224                                    gift_wrap_event_json: value.gift_wrap_event_json.clone(),
1225                                },
1226                            )
1227                        })
1228                        .collect(),
1229                )
1230            })
1231            .transpose()?
1232            .unwrap_or_default();
1233        let command_results = persisted
1234            .as_ref()
1235            .map(|persisted| persisted.command_results.clone())
1236            .unwrap_or_default();
1237        // omega#91: a handoff that was in flight when the last host process
1238        // ended is settled here, at load, before anything else can read the
1239        // ledger. The isolated provider home and the login it was driving died
1240        // with that process, so the terminal answer is that the handoff was
1241        // interrupted. This can only under-claim — it never reports a
1242        // connection the host did not make — and it is what stops a restart
1243        // leaving the phone with a request that neither resolves nor fails.
1244        let mut issue31_provider_handoffs = persisted
1245            .as_ref()
1246            .map(|persisted| persisted.provider_handoffs.clone())
1247            .unwrap_or_default();
1248        issue31_provider_handoffs.adopt_after_restart();
1249        let issue31_control_cursor = persisted
1250            .as_ref()
1251            .and_then(|persisted| persisted.control_cursor.clone());
1252        let issue31_projection_cursor = persisted
1253            .as_ref()
1254            .and_then(|persisted| persisted.projection_cursor.clone());
1255        let issue31_quarantined_events = persisted
1256            .as_ref()
1257            .map(|persisted| persisted.quarantined_events.clone())
1258            .unwrap_or_default();
1259        let issue31_host_adjunct_emissions = persisted
1260            .as_ref()
1261            .map(|persisted| persisted.host_adjunct_emissions.clone())
1262            .unwrap_or_default();
1263        let active_turn_ref = persisted
1264            .as_ref()
1265            .and_then(|persisted| persisted.active_turn_ref.clone());
1266        let run_state = persisted
1267            .as_ref()
1268            .map(|persisted| persisted.run_state.clone())
1269            .unwrap_or_else(default_run_state);
1270        let message_seq = persisted
1271            .as_ref()
1272            .map(|persisted| persisted.message_seq)
1273            .unwrap_or_default();
1274        let issue31_relay_acknowledgements = persisted
1275            .map(|persisted| persisted.relay_acknowledgements)
1276            .unwrap_or_default();
1277        for (event_id, acknowledged_relays) in &issue31_relay_acknowledgements {
1278            relay.restore_publication_acknowledgements(event_id, acknowledged_relays);
1279        }
1280        Ok(Self {
1281            config,
1282            relay: Box::new(relay),
1283            signer: ConversationSigner::OmegaIdentity(identity_service),
1284            pending_events: VecDeque::new(),
1285            active_turn_ref,
1286            run_state,
1287            message_seq,
1288            command_results,
1289            last_gap_state: GapState::None,
1290            last_confirmed_cursor: None,
1291            issue31_host: Some(issue31_host),
1292            issue31_discovery_generation,
1293            issue31_discovery_expires_at,
1294            issue31_discovery_outbox,
1295            issue31_private_outbox,
1296            issue31_relay_acknowledgements,
1297            issue31_control_cursor,
1298            issue31_projection_cursor,
1299            issue31_quarantined_events,
1300            issue31_withheld_emissions: BTreeMap::new(),
1301            issue31_host_adjunct_emissions,
1302            issue31_host_projection_source: None,
1303            issue31_provider_handoffs,
1304            issue31_provider_roster_source: None,
1305            issue31_state_path: Some(issue31_state_path),
1306            #[cfg(test)]
1307            issue31_fail_commit_after: Cell::new(None),
1308        })
1309    }
1310
1311    /// Build a client that exercises NIP-42 against a mock that requires AUTH.
1312    pub fn mock_with_nip42_auth(
1313        mut config: SarahConversationConfig,
1314        challenge: &str,
1315        signer: SigningIdentity,
1316    ) -> Self {
1317        config.relay_url = Some("wss://relay.openagents.com".to_string());
1318        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
1319        Self::with_relay(
1320            config,
1321            Box::new(MockRelayAdapter::with_required_auth(challenge)),
1322            signer,
1323        )
1324    }
1325
1326    pub fn generation(&self) -> u64 {
1327        self.config.generation
1328    }
1329
1330    pub fn synchronize_process_generation(
1331        &mut self,
1332        generation: u64,
1333    ) -> Result<(), SarahConversationError> {
1334        if generation == 0 || generation < self.config.generation {
1335            return Err(SarahConversationError::StaleGeneration {
1336                expected: self.config.generation,
1337                got: generation,
1338            });
1339        }
1340        self.config.generation = generation;
1341        Ok(())
1342    }
1343
1344    pub fn conversation_ref(&self) -> String {
1345        self.config.conversation_ref()
1346    }
1347
1348    /// Handle a framed request method. Returns the result object (not the frame).
1349    pub fn handle_request(
1350        &mut self,
1351        method: &str,
1352        generation: u64,
1353        params: Option<&Value>,
1354    ) -> Result<Value, SarahConversationError> {
1355        self.ensure_generation(generation)?;
1356        match method {
1357            SARAH_METHOD_SESSION_STATUS => Ok(serde_json::to_value(self.session_status()?)
1358                .map_err(|error| SarahConversationError::Internal(error.to_string()))?),
1359            SARAH_METHOD_BOOTSTRAP => Ok(serde_json::to_value(self.bootstrap()?)
1360                .map_err(|error| SarahConversationError::Internal(error.to_string()))?),
1361            SARAH_METHOD_ROOM_SNAPSHOT => {
1362                let legacy_cursor = params
1363                    .and_then(|value| value.get("cursor"))
1364                    .and_then(Value::as_str);
1365                let legacy_limit = params
1366                    .and_then(|value| value.get("limit"))
1367                    .and_then(Value::as_u64)
1368                    .map(|value| value as usize);
1369                let transcript_cursor = params
1370                    .and_then(|value| value.get("transcriptCursor"))
1371                    .and_then(Value::as_str)
1372                    .or(legacy_cursor);
1373                let activity_cursor = params
1374                    .and_then(|value| value.get("activityCursor"))
1375                    .and_then(Value::as_str)
1376                    .or(legacy_cursor);
1377                let nostr_cursor = params
1378                    .and_then(|value| value.get("nostrCursor"))
1379                    .and_then(Value::as_str)
1380                    .or(legacy_cursor);
1381                let transcript_limit = params
1382                    .and_then(|value| value.get("transcriptLimit"))
1383                    .and_then(Value::as_u64)
1384                    .map(|value| value as usize)
1385                    .or(legacy_limit);
1386                let activity_limit = params
1387                    .and_then(|value| value.get("activityLimit"))
1388                    .and_then(Value::as_u64)
1389                    .map(|value| value as usize)
1390                    .or(legacy_limit);
1391                let nostr_limit = params
1392                    .and_then(|value| value.get("nostrLimit"))
1393                    .and_then(Value::as_u64)
1394                    .map(|value| value as usize)
1395                    .or(legacy_limit);
1396                Ok(serde_json::to_value(self.room_snapshot_with_record_cursor(
1397                    transcript_cursor,
1398                    transcript_limit,
1399                    activity_cursor,
1400                    activity_limit,
1401                    nostr_cursor,
1402                    nostr_limit,
1403                )?)
1404                .map_err(|error| SarahConversationError::Internal(error.to_string()))?)
1405            }
1406            SARAH_METHOD_SEND_MESSAGE => {
1407                let (idempotency_ref, expected_generation) = command_binding(params, generation)?;
1408                let text = params
1409                    .and_then(|value| value.get("text"))
1410                    .and_then(Value::as_str)
1411                    .ok_or_else(|| {
1412                        SarahConversationError::InvalidRequest(
1413                            "sarah_send_message requires text".into(),
1414                        )
1415                    })?;
1416                let fingerprint = command_fingerprint(method, params)?;
1417                if let Some(cached) = self.cached_command(&idempotency_ref, &fingerprint)? {
1418                    self.retry_durable_outbox()?;
1419                    return Ok(cached);
1420                }
1421                let result = self.send_message_with_fingerprint(
1422                    text,
1423                    &idempotency_ref,
1424                    expected_generation,
1425                    Some(fingerprint),
1426                )?;
1427                serde_json::to_value(result)
1428                    .map_err(|error| SarahConversationError::Internal(error.to_string()))
1429            }
1430            SARAH_METHOD_INTERRUPT_TURN => {
1431                let (idempotency_ref, expected_generation) = command_binding(params, generation)?;
1432                let turn_ref = params
1433                    .and_then(|value| value.get("turnRef"))
1434                    .and_then(Value::as_str)
1435                    .map(str::to_owned)
1436                    .or_else(|| self.active_turn_ref.clone())
1437                    .ok_or_else(|| {
1438                        SarahConversationError::InvalidRequest(
1439                            "sarah_interrupt_turn requires turnRef".into(),
1440                        )
1441                    })?;
1442                let fingerprint = command_fingerprint(method, params)?;
1443                if let Some(cached) = self.cached_command(&idempotency_ref, &fingerprint)? {
1444                    self.retry_durable_outbox()?;
1445                    return Ok(cached);
1446                }
1447                let result = self.interrupt_turn_with_fingerprint(
1448                    &turn_ref,
1449                    &idempotency_ref,
1450                    expected_generation,
1451                    Some(fingerprint),
1452                )?;
1453                serde_json::to_value(result)
1454                    .map_err(|error| SarahConversationError::Internal(error.to_string()))
1455            }
1456            SARAH_METHOD_DEVICE_GRANTS => {
1457                self.sync_issue31_host()?;
1458                let controller = self.issue31_host.as_ref().ok_or_else(|| {
1459                    SarahConversationError::InvalidRequest("Issue 31 host is not configured".into())
1460                })?;
1461                let grants = controller
1462                    .grant_projections(unix_now())
1463                    .map_err(issue31_error)?;
1464                Ok(json!({ "grants": grants }))
1465            }
1466            SARAH_METHOD_RENEW_DEVICE_GRANT => {
1467                let (idempotency_ref, expected_generation) = command_binding(params, generation)?;
1468                self.ensure_generation(expected_generation)?;
1469                let grant_ref = params
1470                    .and_then(|value| value.get("grantRef"))
1471                    .and_then(Value::as_str)
1472                    .ok_or_else(|| {
1473                        SarahConversationError::InvalidRequest(
1474                            "sarah_renew_device_grant requires grantRef".into(),
1475                        )
1476                    })?;
1477                let scopes = params
1478                    .and_then(|value| value.get("scopes"))
1479                    .cloned()
1480                    .ok_or_else(|| {
1481                        SarahConversationError::InvalidRequest(
1482                            "sarah_renew_device_grant requires scopes".into(),
1483                        )
1484                    })?;
1485                let scopes = serde_json::from_value(scopes)
1486                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
1487                let expires_at = params
1488                    .and_then(|value| value.get("expiresAt"))
1489                    .and_then(Value::as_u64)
1490                    .ok_or_else(|| {
1491                        SarahConversationError::InvalidRequest(
1492                            "sarah_renew_device_grant requires expiresAt".into(),
1493                        )
1494                    })?;
1495                let fingerprint = command_fingerprint(method, params)?;
1496                if let Some(cached) = self.cached_command(&idempotency_ref, &fingerprint)? {
1497                    self.retry_durable_outbox()?;
1498                    return Ok(cached);
1499                }
1500                self.renew_issue31_grant(
1501                    grant_ref,
1502                    scopes,
1503                    expires_at,
1504                    idempotency_ref,
1505                    fingerprint,
1506                )
1507            }
1508            SARAH_METHOD_REVOKE_DEVICE_GRANT => {
1509                let (idempotency_ref, expected_generation) = command_binding(params, generation)?;
1510                self.ensure_generation(expected_generation)?;
1511                let grant_ref = params
1512                    .and_then(|value| value.get("grantRef"))
1513                    .and_then(Value::as_str)
1514                    .ok_or_else(|| {
1515                        SarahConversationError::InvalidRequest(
1516                            "sarah_revoke_device_grant requires grantRef".into(),
1517                        )
1518                    })?;
1519                let reason_ref = params
1520                    .and_then(|value| value.get("reasonRef"))
1521                    .and_then(Value::as_str)
1522                    .map(str::to_owned);
1523                let fingerprint = command_fingerprint(method, params)?;
1524                if let Some(cached) = self.cached_command(&idempotency_ref, &fingerprint)? {
1525                    self.retry_durable_outbox()?;
1526                    return Ok(cached);
1527                }
1528                self.revoke_issue31_grant(grant_ref, reason_ref, idempotency_ref, fingerprint)
1529            }
1530            SARAH_METHOD_READMIT_DEVICE => {
1531                let (idempotency_ref, expected_generation) = command_binding(params, generation)?;
1532                self.ensure_generation(expected_generation)?;
1533                let grant_ref = params
1534                    .and_then(|value| value.get("grantRef"))
1535                    .and_then(Value::as_str)
1536                    .ok_or_else(|| {
1537                        SarahConversationError::InvalidRequest(
1538                            "sarah_readmit_device requires grantRef".into(),
1539                        )
1540                    })?;
1541                let fingerprint = command_fingerprint(method, params)?;
1542                if let Some(cached) = self.cached_command(&idempotency_ref, &fingerprint)? {
1543                    self.retry_durable_outbox()?;
1544                    return Ok(cached);
1545                }
1546                self.readmit_issue31_device(grant_ref, idempotency_ref, fingerprint)
1547            }
1548            _ => Err(SarahConversationError::InvalidRequest(format!(
1549                "unknown Sarah method {method}"
1550            ))),
1551        }
1552    }
1553
1554    pub fn session_status(&mut self) -> Result<SessionStatusResult, SarahConversationError> {
1555        self.ensure_connected()?;
1556        Ok(SessionStatusResult {
1557            signed_in: matches!(
1558                self.config.identity.binding_state,
1559                BindingState::Bound | BindingState::Unbound
1560            ) && !self.config.identity.owner_public_key_hex.is_empty(),
1561            account_label: self.config.identity.account_label.clone(),
1562            binding_state: self.config.identity.binding_state,
1563            owner_public_key_hex: Some(self.config.identity.owner_public_key_hex.clone()),
1564            binding_expires_at: None,
1565            transport: self.transport_label(),
1566        })
1567    }
1568
1569    pub fn bootstrap(&mut self) -> Result<BootstrapResult, SarahConversationError> {
1570        self.ensure_connected()?;
1571        self.sync_issue31_host()?;
1572        let room_state = self.current_room_state();
1573        self.push_room_state_event(&room_state);
1574        Ok(BootstrapResult {
1575            principal_ref: "principal.sarah".to_string(),
1576            display_name: "Sarah".to_string(),
1577            role: "owner_orchestrator".to_string(),
1578            conversation_ref: self.config.conversation_ref(),
1579            legacy_thread_ref: self.config.legacy_thread_ref(),
1580            owner_public_key_hex: self.config.identity.owner_public_key_hex.clone(),
1581            sarah_public_key_hex: self.config.identity.sarah_public_key_hex.clone(),
1582            authority_profile_ref: "docs/authority/SARAH_AUTHORITY.md".to_string(),
1583            authority_profile_revision: 7,
1584            admitted_device_fingerprints: self
1585                .issue31_host
1586                .as_ref()
1587                .map(Issue31HostController::admitted_device_fingerprints)
1588                .unwrap_or_default(),
1589            quarantined_issue31_event_count: self.issue31_quarantined_events.len(),
1590            room_state,
1591        })
1592    }
1593
1594    pub fn sync_issue31_host(&mut self) -> Result<(), SarahConversationError> {
1595        self.ensure_connected()?;
1596        let Some(mut controller) = self.issue31_host.take() else {
1597            return Ok(());
1598        };
1599        let result = self.sync_issue31_host_with(&mut controller);
1600        self.issue31_host = Some(controller);
1601        let persistence = self.persist_issue31_host_state();
1602        finish_durable_operation(result, persistence)
1603    }
1604
1605    fn renew_issue31_grant(
1606        &mut self,
1607        grant_ref: &str,
1608        scopes: Vec<crate::Issue31PairingScope>,
1609        expires_at: u64,
1610        idempotency_ref: String,
1611        fingerprint: String,
1612    ) -> Result<Value, SarahConversationError> {
1613        self.ensure_connected()?;
1614        self.ensure_command_result_capacity(&idempotency_ref)?;
1615        let mut controller = self.issue31_host.take().ok_or_else(|| {
1616            SarahConversationError::InvalidRequest("Issue 31 host is not configured".into())
1617        })?;
1618        let result = (|| {
1619            let record = controller
1620                .renew_grant(grant_ref, scopes, unix_now(), expires_at)
1621                .map_err(issue31_error)?;
1622            let enqueued = self.enqueue_issue31_pairing_record(&record)?;
1623            let event_id = enqueued.rumor_event_id.clone();
1624            controller
1625                .record_emitted_pairing(event_id.clone(), record)
1626                .map_err(issue31_error)?;
1627            let previous_projection_cursor = self.issue31_projection_cursor.take();
1628            let response = json!({ "accepted": true, "eventId": event_id, "grantRef": grant_ref });
1629            self.command_results.insert(
1630                idempotency_ref.clone(),
1631                (fingerprint.clone(), response.clone()),
1632            );
1633            if let Err(error) = self.persist_issue31_host_state_with_controller(&controller) {
1634                self.issue31_projection_cursor = previous_projection_cursor;
1635                self.command_results.remove(&idempotency_ref);
1636                self.rollback_issue31_enqueue(&enqueued);
1637                return Err(error);
1638            }
1639            self.flush_issue31_outbox()?;
1640            self.persist_issue31_host_state_with_controller(&controller)?;
1641            Ok(response)
1642        })();
1643        self.issue31_host = Some(controller);
1644        let persistence = self.persist_issue31_host_state();
1645        finish_durable_operation(result, persistence)
1646    }
1647
1648    fn revoke_issue31_grant(
1649        &mut self,
1650        grant_ref: &str,
1651        reason_ref: Option<String>,
1652        idempotency_ref: String,
1653        fingerprint: String,
1654    ) -> Result<Value, SarahConversationError> {
1655        self.ensure_connected()?;
1656        self.ensure_command_result_capacity(&idempotency_ref)?;
1657        let mut controller = self.issue31_host.take().ok_or_else(|| {
1658            SarahConversationError::InvalidRequest("Issue 31 host is not configured".into())
1659        })?;
1660        let result = (|| {
1661            let record = controller
1662                .revoke_grant(grant_ref, unix_now(), reason_ref)
1663                .map_err(issue31_error)?;
1664            let enqueued = self.enqueue_issue31_pairing_record(&record)?;
1665            let event_id = enqueued.rumor_event_id.clone();
1666            controller
1667                .record_emitted_pairing(event_id.clone(), record)
1668                .map_err(issue31_error)?;
1669            let response = json!({ "accepted": true, "eventId": event_id, "grantRef": grant_ref });
1670            self.command_results.insert(
1671                idempotency_ref.clone(),
1672                (fingerprint.clone(), response.clone()),
1673            );
1674            if let Err(error) = self.persist_issue31_host_state_with_controller(&controller) {
1675                self.command_results.remove(&idempotency_ref);
1676                self.rollback_issue31_enqueue(&enqueued);
1677                return Err(error);
1678            }
1679            self.flush_issue31_outbox()?;
1680            self.persist_issue31_host_state_with_controller(&controller)?;
1681            Ok(response)
1682        })();
1683        self.issue31_host = Some(controller);
1684        let persistence = self.persist_issue31_host_state();
1685        finish_durable_operation(result, persistence)
1686    }
1687
1688    /// Clear the revocations blocking the device behind `grant_ref`.
1689    ///
1690    /// This publishes nothing. Re-admission is host-local owner policy, not a
1691    /// record other peers are entitled to act on: the device still has to run a
1692    /// fresh, signed pairing handshake to obtain a new grant, and that grant is
1693    /// the only thing that carries authority.
1694    fn readmit_issue31_device(
1695        &mut self,
1696        grant_ref: &str,
1697        idempotency_ref: String,
1698        fingerprint: String,
1699    ) -> Result<Value, SarahConversationError> {
1700        self.ensure_connected()?;
1701        self.ensure_command_result_capacity(&idempotency_ref)?;
1702        let controller = self.issue31_host.take().ok_or_else(|| {
1703            SarahConversationError::InvalidRequest("Issue 31 host is not configured".into())
1704        })?;
1705        // Mutate a candidate so a refused or unpersistable re-admission cannot
1706        // leave a cleared revocation behind in the live controller.
1707        let mut candidate = controller.clone();
1708        let result = (|| {
1709            let cleared = candidate
1710                .readmit_device_for_grant(grant_ref)
1711                .map_err(issue31_error)?;
1712            let response = json!({
1713                "accepted": true,
1714                "grantRef": grant_ref,
1715                "clearedRevocations": cleared.len(),
1716            });
1717            self.command_results.insert(
1718                idempotency_ref.clone(),
1719                (fingerprint.clone(), response.clone()),
1720            );
1721            if let Err(error) = self.persist_issue31_host_state_with_controller(&candidate) {
1722                self.command_results.remove(&idempotency_ref);
1723                return Err(error);
1724            }
1725            Ok(response)
1726        })();
1727        self.issue31_host = Some(if result.is_ok() { candidate } else { controller });
1728        let persistence = self.persist_issue31_host_state();
1729        finish_durable_operation(result, persistence)
1730    }
1731
1732    fn sync_issue31_host_with(
1733        &mut self,
1734        controller: &mut Issue31HostController,
1735    ) -> Result<(), SarahConversationError> {
1736        self.flush_issue31_outbox()?;
1737        self.persist_issue31_host_state_with_controller(controller)?;
1738        let now = unix_now();
1739        let discovery = controller
1740            .discovery_v2(now, now.saturating_add(24 * 60 * 60))
1741            .map_err(issue31_error)?;
1742        if self.issue31_discovery_generation != Some(discovery.generation)
1743            || self
1744                .issue31_discovery_expires_at
1745                .is_none_or(|expires_at| expires_at <= now.saturating_add(60 * 60))
1746        {
1747            let replacement = self.sign_issue31_discovery(&discovery)?;
1748            let previous_outbox = self.issue31_discovery_outbox.replace(replacement);
1749            let previous_generation = self.issue31_discovery_generation;
1750            let previous_expires_at = self.issue31_discovery_expires_at;
1751            let previous_acknowledgements = previous_outbox.as_ref().and_then(|event| {
1752                self.issue31_relay_acknowledgements
1753                    .remove(&event.id.to_hex())
1754                    .map(|acknowledgements| (event.id.to_hex(), acknowledgements))
1755            });
1756            self.issue31_discovery_generation = Some(discovery.generation);
1757            self.issue31_discovery_expires_at = Some(discovery.expires_at);
1758            if let Err(error) = self.persist_issue31_host_state_with_controller(controller) {
1759                self.issue31_discovery_outbox = previous_outbox;
1760                self.issue31_discovery_generation = previous_generation;
1761                self.issue31_discovery_expires_at = previous_expires_at;
1762                if let Some((event_id, acknowledgements)) = previous_acknowledgements {
1763                    self.issue31_relay_acknowledgements
1764                        .insert(event_id, acknowledgements);
1765                }
1766                return Err(error);
1767            }
1768            self.flush_issue31_outbox()?;
1769            self.persist_issue31_host_state_with_controller(controller)?;
1770        }
1771
1772        let conversation_ref = self.config.conversation_ref();
1773        let mut cursor = self.issue31_control_cursor.clone();
1774        let mut last_scanned_cursor = None;
1775        let mut scan_exhausted = false;
1776        let mut records = Vec::new();
1777        for _ in 0..8 {
1778            let page =
1779                self.query_with_auth(&conversation_ref, cursor.as_deref(), MAX_PAGE_LIMIT)?;
1780            self.last_gap_state = strongest_gap_state(self.last_gap_state, page.gap_state);
1781            if let Some(page_cursor) = page.events.last().map(stored_event_cursor) {
1782                last_scanned_cursor = Some(page_cursor);
1783            }
1784            for event in page.events {
1785                if event.record_kind == "pairing" || event.record_kind == "control" {
1786                    records.push(event);
1787                }
1788            }
1789            let Some(next_cursor) = page.next_cursor else {
1790                scan_exhausted = true;
1791                break;
1792            };
1793            cursor = Some(next_cursor);
1794        }
1795
1796        // omega#91: one roster observation per pass, taken BEFORE this pass's
1797        // commands are handled. A handoff opened by a command in this pass is
1798        // therefore published as `requested` and only looked at against the
1799        // roster on the next pass. Advancing after would mean the host bound a
1800        // handoff using a roster reading older than the request itself, and the
1801        // phone would never see the handoff appear before it bound.
1802        self.advance_issue31_provider_handoffs(now);
1803        for event in records {
1804            if self
1805                .issue31_quarantined_events
1806                .contains_key(&event.event_id)
1807            {
1808                continue;
1809            }
1810            if event.record_kind == "pairing" {
1811                let record = match Issue31PairingRecord::decode(event.content_summary.as_bytes()) {
1812                    Ok(record) => record,
1813                    Err(_) => {
1814                        self.quarantine_issue31_event(
1815                            &event.event_id,
1816                            "reason.omega.invalid_pairing_record",
1817                            controller,
1818                        )?;
1819                        continue;
1820                    }
1821                };
1822                let mut candidate = controller.clone();
1823                let outbound = match candidate.handle_pairing_event(
1824                    Issue31PairingEvent {
1825                        event_id: event.event_id.clone(),
1826                        record,
1827                    },
1828                    now,
1829                ) {
1830                    Ok(outbound) => outbound,
1831                    Err(_) => {
1832                        self.quarantine_issue31_event(
1833                            &event.event_id,
1834                            "reason.omega.pairing_rejected",
1835                            controller,
1836                        )?;
1837                        continue;
1838                    }
1839                };
1840                if let Some(outbound) = outbound {
1841                    let resets_projection_cursor = matches!(
1842                        outbound,
1843                        Issue31PairingRecord::ScopedGrant { .. }
1844                            | Issue31PairingRecord::GrantRenewal { .. }
1845                    );
1846                    let enqueued = self.enqueue_issue31_pairing_record(&outbound)?;
1847                    if let Err(error) =
1848                        candidate.record_emitted_pairing(enqueued.rumor_event_id.clone(), outbound)
1849                    {
1850                        self.rollback_issue31_enqueue(&enqueued);
1851                        return Err(issue31_error(error));
1852                    }
1853                    let previous_projection_cursor = resets_projection_cursor
1854                        .then(|| self.issue31_projection_cursor.take())
1855                        .flatten();
1856                    if let Err(error) = self.persist_issue31_host_state_with_controller(&candidate)
1857                    {
1858                        if resets_projection_cursor {
1859                            self.issue31_projection_cursor = previous_projection_cursor;
1860                        }
1861                        self.rollback_issue31_enqueue(&enqueued);
1862                        return Err(error);
1863                    }
1864                    *controller = candidate;
1865                    self.flush_issue31_outbox()?;
1866                    self.persist_issue31_host_state_with_controller(controller)?;
1867                } else {
1868                    self.persist_issue31_host_state_with_controller(&candidate)?;
1869                    *controller = candidate;
1870                }
1871            } else {
1872                let command_schema = serde_json::from_str::<Value>(&event.content_summary)
1873                    .ok()
1874                    .and_then(|value| value.get("schema")?.as_str().map(str::to_owned));
1875                if command_schema.as_deref() == Some(ISSUE31_COMMAND_SCHEMA_V2) {
1876                    let record =
1877                        match Issue31CommandRecordV2::decode(event.content_summary.as_bytes()) {
1878                            Ok(record) => record,
1879                            Err(_) => {
1880                                self.quarantine_issue31_event(
1881                                    &event.event_id,
1882                                    "reason.omega.invalid_command_record",
1883                                    controller,
1884                                )?;
1885                                continue;
1886                            }
1887                        };
1888                    let mut candidate = controller.clone();
1889                    let result = match candidate.handle_command_event_v2(
1890                        event.event_id.clone(),
1891                        record,
1892                        now,
1893                        |arguments,
1894                         idempotency_ref,
1895                         grant_ref,
1896                         device_public_key_hex,
1897                         expected_generation| {
1898                            self.issue31_host = Some(controller.clone());
1899                            let execution = self.execute_issue31_action_v2(
1900                                arguments,
1901                                idempotency_ref,
1902                                grant_ref,
1903                                device_public_key_hex,
1904                                expected_generation,
1905                            );
1906                            self.issue31_host.take();
1907                            execution
1908                        },
1909                    ) {
1910                        Ok(result) => result,
1911                        Err(_) => {
1912                            self.quarantine_issue31_event(
1913                                &event.event_id,
1914                                "reason.omega.command_rejected",
1915                                controller,
1916                            )?;
1917                            continue;
1918                        }
1919                    };
1920                    if let Some(result) = result {
1921                        if let Issue31CommandRecordV2::CommandResult {
1922                            status: Issue31CommandHandlingStatus::Accepted,
1923                            grant_ref,
1924                            expected_generation,
1925                            source_event_id: Some(source_event_id),
1926                            ..
1927                        } = &result
1928                        {
1929                            candidate
1930                                .record_source_projection(
1931                                    grant_ref.clone(),
1932                                    *expected_generation,
1933                                    source_event_id.clone(),
1934                                )
1935                                .map_err(issue31_error)?;
1936                        }
1937                        let enqueued = self.enqueue_issue31_command_record_v2(&result)?;
1938                        if let Err(error) =
1939                            self.persist_issue31_host_state_with_controller(&candidate)
1940                        {
1941                            self.rollback_issue31_enqueue(&enqueued);
1942                            return Err(error);
1943                        }
1944                        *controller = candidate;
1945                        self.flush_issue31_outbox()?;
1946                        self.persist_issue31_host_state_with_controller(controller)?;
1947                    } else {
1948                        self.persist_issue31_host_state_with_controller(&candidate)?;
1949                        *controller = candidate;
1950                    }
1951                    continue;
1952                }
1953                let record = match Issue31CommandRecord::decode(event.content_summary.as_bytes()) {
1954                    Ok(record) => record,
1955                    Err(_) => {
1956                        self.quarantine_issue31_event(
1957                            &event.event_id,
1958                            "reason.omega.invalid_command_record",
1959                            controller,
1960                        )?;
1961                        continue;
1962                    }
1963                };
1964                let mut candidate = controller.clone();
1965                let result = match candidate.handle_command_event(
1966                    Issue31CommandEvent {
1967                        event_id: event.event_id.clone(),
1968                        record,
1969                    },
1970                    now,
1971                    |action_ref, arguments_ref, idempotency_ref| {
1972                        self.execute_issue31_action(action_ref, arguments_ref, idempotency_ref)
1973                    },
1974                ) {
1975                    Ok(result) => result,
1976                    Err(_) => {
1977                        self.quarantine_issue31_event(
1978                            &event.event_id,
1979                            "reason.omega.command_rejected",
1980                            controller,
1981                        )?;
1982                        continue;
1983                    }
1984                };
1985                if let Some(result) = result {
1986                    let enqueued = self.enqueue_issue31_command_record(&result)?;
1987                    if let Err(error) = self.persist_issue31_host_state_with_controller(&candidate)
1988                    {
1989                        self.rollback_issue31_enqueue(&enqueued);
1990                        return Err(error);
1991                    }
1992                    *controller = candidate;
1993                    self.flush_issue31_outbox()?;
1994                    self.persist_issue31_host_state_with_controller(controller)?;
1995                } else {
1996                    self.persist_issue31_host_state_with_controller(&candidate)?;
1997                    *controller = candidate;
1998                }
1999            }
2000        }
2001        self.project_issue31_sources(controller, now)?;
2002        // The omega#47 documents ride the same durable outbox as every other
2003        // owner-private record, so a device that was offline for this pass gets
2004        // them on the next flush rather than losing the snapshot entirely.
2005        self.publish_issue31_host_adjuncts(controller, now)?;
2006        self.persist_issue31_host_state_with_controller(controller)?;
2007        self.flush_issue31_outbox()?;
2008        self.persist_issue31_host_state_with_controller(controller)?;
2009        if let Some(last_scanned_cursor) = last_scanned_cursor {
2010            self.issue31_control_cursor = Some(last_scanned_cursor);
2011        }
2012        if !scan_exhausted {
2013            self.last_gap_state = strongest_gap_state(self.last_gap_state, GapState::Possible);
2014        }
2015        self.persist_issue31_host_state_with_controller(controller)?;
2016        Ok(())
2017    }
2018
2019    fn quarantine_issue31_event(
2020        &mut self,
2021        event_id: &str,
2022        reason_ref: &str,
2023        controller: &Issue31HostController,
2024    ) -> Result<(), SarahConversationError> {
2025        if let Some(existing) = self.issue31_quarantined_events.get(event_id) {
2026            if existing == reason_ref {
2027                return Ok(());
2028            }
2029            return Err(SarahConversationError::Internal(
2030                "Issue 31 quarantine reason changed for the same event".into(),
2031            ));
2032        }
2033        if self.issue31_quarantined_events.len() >= MAX_QUARANTINED_ISSUE31_EVENTS {
2034            return Err(SarahConversationError::Internal(
2035                "Issue 31 quarantine bound is exhausted".into(),
2036            ));
2037        }
2038        if !is_lower_hex_64(event_id) || !crate::is_issue31_public_ref(reason_ref) {
2039            return Err(SarahConversationError::Internal(
2040                "Issue 31 quarantine record is invalid".into(),
2041            ));
2042        }
2043        self.issue31_quarantined_events
2044            .insert(event_id.to_string(), reason_ref.to_string());
2045        if let Err(error) = self.persist_issue31_host_state_with_controller(controller) {
2046            self.issue31_quarantined_events.remove(event_id);
2047            return Err(error);
2048        }
2049        Ok(())
2050    }
2051
2052    fn sign_issue31_discovery(
2053        &self,
2054        discovery: &Issue31HostDiscoveryV2,
2055    ) -> Result<Event, SarahConversationError> {
2056        let content = serde_json::to_string(discovery)
2057            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
2058        let tags = [
2059            ["d", discovery.host_ref.as_str()],
2060            ["k", "1059"],
2061            ["t", "omega-issue31-host"],
2062            ["alt", "Omega Issue 31 Nostr host discovery"],
2063        ]
2064        .into_iter()
2065        .map(|tag| {
2066            Tag::parse(tag)
2067                .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))
2068        })
2069        .collect::<Result<Vec<_>, _>>()?;
2070        self.signer
2071            .sign_public_record(ISSUE31_HOST_DISCOVERY_KIND, &content, tags)
2072    }
2073
2074    fn enqueue_issue31_pairing_record(
2075        &mut self,
2076        record: &Issue31PairingRecord,
2077    ) -> Result<EnqueuedIssue31PrivateRecord, SarahConversationError> {
2078        let content = serde_json::to_string(record)
2079            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
2080        self.enqueue_issue31_private_content(
2081            ISSUE31_PAIRING_SCHEMA,
2082            &content,
2083            record.device_public_key_hex(),
2084        )
2085    }
2086
2087    fn enqueue_issue31_command_record(
2088        &mut self,
2089        record: &Issue31CommandRecord,
2090    ) -> Result<EnqueuedIssue31PrivateRecord, SarahConversationError> {
2091        let content = serde_json::to_string(record)
2092            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
2093        self.enqueue_issue31_private_content(
2094            ISSUE31_COMMAND_SCHEMA,
2095            &content,
2096            record.device_public_key_hex(),
2097        )
2098    }
2099
2100    fn enqueue_issue31_command_record_v2(
2101        &mut self,
2102        record: &Issue31CommandRecordV2,
2103    ) -> Result<EnqueuedIssue31PrivateRecord, SarahConversationError> {
2104        let content = serde_json::to_string(record)
2105            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
2106        self.enqueue_issue31_private_content(
2107            ISSUE31_COMMAND_SCHEMA_V2,
2108            &content,
2109            record.device_public_key_hex(),
2110        )
2111    }
2112
2113    fn enqueue_issue31_private_content(
2114        &mut self,
2115        schema: &str,
2116        content: &str,
2117        recipient_public_key_hex: &str,
2118    ) -> Result<EnqueuedIssue31PrivateRecord, SarahConversationError> {
2119        PublicKey::from_hex(recipient_public_key_hex)
2120            .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
2121        let outbox_ref = format!(
2122            "{schema}.{:x}",
2123            Sha256::digest([content.as_bytes(), recipient_public_key_hex.as_bytes()].concat())
2124        );
2125        let tags = vec![
2126            Tag::parse(["p", recipient_public_key_hex])
2127                .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?,
2128        ];
2129        let (rumor_event_id, inserted) = self.enqueue_private_content(
2130            &outbox_ref,
2131            content,
2132            tags,
2133            &[recipient_public_key_hex.to_string()],
2134        )?;
2135        Ok(EnqueuedIssue31PrivateRecord {
2136            rumor_event_id,
2137            outbox_ref,
2138            inserted,
2139        })
2140    }
2141
2142    fn rollback_issue31_enqueue(&mut self, enqueued: &EnqueuedIssue31PrivateRecord) {
2143        if enqueued.inserted {
2144            self.issue31_private_outbox.remove(&enqueued.outbox_ref);
2145        }
2146    }
2147
2148    fn enqueue_private_content(
2149        &mut self,
2150        outbox_ref: &str,
2151        content: &str,
2152        tags: Vec<Tag>,
2153        recipients: &[String],
2154    ) -> Result<(String, bool), SarahConversationError> {
2155        if let Some(pending) = self.issue31_private_outbox.get(outbox_ref) {
2156            return Ok((pending.rumor_event_id.clone(), false));
2157        }
2158        if self.issue31_private_outbox.len() >= MAX_PRIVATE_OUTBOX_ITEMS {
2159            return Err(SarahConversationError::Internal(
2160                "durable private outbox item bound is exhausted".into(),
2161            ));
2162        }
2163        let (rumor_event_id, gift_wraps) =
2164            self.signer.private_messages(content, tags, recipients)?;
2165        if self
2166            .issue31_relay_acknowledgements
2167            .len()
2168            .saturating_add(gift_wraps.len())
2169            > MAX_RELAY_ACKNOWLEDGEMENTS
2170        {
2171            return Err(SarahConversationError::Internal(
2172                "durable relay acknowledgement bound is exhausted".into(),
2173            ));
2174        }
2175        self.issue31_private_outbox.insert(
2176            outbox_ref.to_string(),
2177            PendingIssue31PrivatePublish {
2178                rumor_event_id: rumor_event_id.clone(),
2179                gift_wraps,
2180            },
2181        );
2182        Ok((rumor_event_id, true))
2183    }
2184
2185    fn flush_issue31_outbox(&mut self) -> Result<(), SarahConversationError> {
2186        if let Some(event) = self.issue31_discovery_outbox.clone() {
2187            self.publish_with_auth(&event)?;
2188            let event_id = event.id.to_hex();
2189            self.record_relay_acknowledgements(&event_id)?;
2190            if self.relay.publication_complete(&event_id) {
2191                self.issue31_discovery_outbox = None;
2192                self.issue31_relay_acknowledgements.remove(&event_id);
2193            }
2194        }
2195        let pending_refs: Vec<String> = self.issue31_private_outbox.keys().cloned().collect();
2196        for pending_ref in pending_refs {
2197            let pending = self
2198                .issue31_private_outbox
2199                .get(&pending_ref)
2200                .cloned()
2201                .ok_or_else(|| {
2202                    SarahConversationError::Internal("private outbox vanished".into())
2203                })?;
2204            for gift_wrap in &pending.gift_wraps {
2205                self.publish_with_auth(gift_wrap)?;
2206                let event_id = gift_wrap.id.to_hex();
2207                self.record_relay_acknowledgements(&event_id)?;
2208            }
2209            if pending
2210                .gift_wraps
2211                .iter()
2212                .all(|gift_wrap| self.relay.publication_complete(&gift_wrap.id.to_hex()))
2213            {
2214                self.issue31_private_outbox.remove(&pending_ref);
2215                for gift_wrap in pending.gift_wraps {
2216                    self.issue31_relay_acknowledgements
2217                        .remove(&gift_wrap.id.to_hex());
2218                }
2219            }
2220        }
2221        Ok(())
2222    }
2223
2224    fn record_relay_acknowledgements(
2225        &mut self,
2226        event_id: &str,
2227    ) -> Result<(), SarahConversationError> {
2228        if !self.issue31_relay_acknowledgements.contains_key(event_id)
2229            && self.issue31_relay_acknowledgements.len() >= MAX_RELAY_ACKNOWLEDGEMENTS
2230        {
2231            return Err(SarahConversationError::Internal(
2232                "durable relay acknowledgement bound is exhausted".into(),
2233            ));
2234        }
2235        self.issue31_relay_acknowledgements.insert(
2236            event_id.to_string(),
2237            self.relay.acknowledged_relays(event_id),
2238        );
2239        Ok(())
2240    }
2241
2242    fn persist_issue31_host_state(&self) -> Result<(), SarahConversationError> {
2243        if self.issue31_state_path.is_none() {
2244            return Ok(());
2245        }
2246        let controller = self.issue31_host.as_ref().ok_or_else(|| {
2247            SarahConversationError::Internal(
2248                "Issue 31 controller was absent during durable commit".into(),
2249            )
2250        })?;
2251        self.persist_issue31_host_state_with_controller(controller)
2252    }
2253
2254    fn persist_issue31_host_state_with_controller(
2255        &self,
2256        controller: &Issue31HostController,
2257    ) -> Result<(), SarahConversationError> {
2258        #[cfg(test)]
2259        if let Some(remaining) = self.issue31_fail_commit_after.get() {
2260            if remaining == 0 {
2261                self.issue31_fail_commit_after.set(None);
2262                return Err(SarahConversationError::Internal(
2263                    "injected Issue 31 durable commit failure".into(),
2264                ));
2265            }
2266            self.issue31_fail_commit_after
2267                .set(Some(remaining.saturating_sub(1)));
2268        }
2269        let Some(path) = &self.issue31_state_path else {
2270            return Ok(());
2271        };
2272        let discovery_event_json = self
2273            .issue31_discovery_outbox
2274            .as_ref()
2275            .map(|event| {
2276                event
2277                    .try_as_json()
2278                    .map_err(|error| SarahConversationError::Internal(error.to_string()))
2279            })
2280            .transpose()?;
2281        let private_outbox = self
2282            .issue31_private_outbox
2283            .iter()
2284            .map(|(outbox_ref, pending)| {
2285                let gift_wrap_event_json = pending
2286                    .gift_wraps
2287                    .iter()
2288                    .map(|event| {
2289                        event
2290                            .try_as_json()
2291                            .map_err(|error| SarahConversationError::Internal(error.to_string()))
2292                    })
2293                    .collect::<Result<Vec<_>, _>>()?;
2294                Ok((
2295                    outbox_ref.clone(),
2296                    DurableIssue31PrivatePublish {
2297                        rumor_event_id: pending.rumor_event_id.clone(),
2298                        gift_wrap_event_json,
2299                    },
2300                ))
2301            })
2302            .collect::<Result<BTreeMap<_, _>, SarahConversationError>>()?;
2303        write_issue31_host_state(
2304            path,
2305            &DurableIssue31HostState {
2306                schema: ISSUE31_DURABLE_STATE_SCHEMA.into(),
2307                controller: controller.clone(),
2308                discovery_generation: self.issue31_discovery_generation,
2309                discovery_expires_at: self.issue31_discovery_expires_at,
2310                discovery_event_json,
2311                private_outbox,
2312                relay_acknowledgements: self.issue31_relay_acknowledgements.clone(),
2313                control_cursor: self.issue31_control_cursor.clone(),
2314                projection_cursor: self.issue31_projection_cursor.clone(),
2315                quarantined_events: self.issue31_quarantined_events.clone(),
2316                host_adjunct_emissions: self.issue31_host_adjunct_emissions.clone(),
2317                provider_handoffs: self.issue31_provider_handoffs.clone(),
2318                command_results: self.command_results.clone(),
2319                active_turn_ref: self.active_turn_ref.clone(),
2320                run_state: self.run_state.clone(),
2321                message_seq: self.message_seq,
2322            },
2323        )
2324    }
2325
2326    /// Answer one admitted command-v1 intent (omega#91).
2327    ///
2328    /// The controller has already checked the host binding, the grant, the
2329    /// generation, the lifetime, and the scope, so reaching here means the
2330    /// device is entitled to this action.
2331    ///
2332    /// Only `action.omega.provider_handoff` is answered. The remaining v1
2333    /// actions — Full Auto control and community action — still have no
2334    /// generation-fenced controller behind them, and returning `unavailable`
2335    /// for them is the honest answer rather than a silent success.
2336    fn execute_issue31_action(
2337        &mut self,
2338        action_ref: &str,
2339        arguments_ref: &str,
2340        idempotency_ref: &str,
2341    ) -> Issue31CommandExecution {
2342        if action_ref != ISSUE31_ACTION_REQUEST_PROVIDER_HANDOFF {
2343            return Issue31CommandExecution {
2344                status: Issue31CommandStatus::Unavailable,
2345                outcome_ref: "outcome.omega.unavailable".into(),
2346                reason_ref: Some("reason.omega.controller_not_bound".into()),
2347            };
2348        }
2349        // One reading of the clock for this command. The stamp on the record
2350        // and the deadline bounding it come from the same instant, and the
2351        // device supplies neither: `argumentsRef` names a provider and nothing
2352        // else, so there is no wire field a device could put a time in.
2353        let now_ms = unix_now().saturating_mul(1_000);
2354        match self
2355            .issue31_provider_handoffs
2356            .open(arguments_ref, idempotency_ref, now_ms)
2357        {
2358            Ok(record) => Issue31CommandExecution {
2359                status: Issue31CommandStatus::Completed,
2360                // The command was "open a provider connection handoff", and
2361                // this is the record that opening produced — which is also how
2362                // the device finds the row to watch. It is deliberately NOT a
2363                // statement that the handoff completed: the handoff carries its
2364                // own state and its own host-owned outcome, and at this moment
2365                // that state is `requested`.
2366                outcome_ref: record.handoff_ref,
2367                reason_ref: None,
2368            },
2369            Err(error) => Issue31CommandExecution {
2370                status: match error {
2371                    crate::issue31_provider_handoff::Issue31ProviderHandoffError::BoundExhausted
2372                    | crate::issue31_provider_handoff::Issue31ProviderHandoffError::Unprojectable(
2373                        _,
2374                    ) => Issue31CommandStatus::Unavailable,
2375                    _ => Issue31CommandStatus::Refused,
2376                },
2377                outcome_ref: match error {
2378                    crate::issue31_provider_handoff::Issue31ProviderHandoffError::BoundExhausted
2379                    | crate::issue31_provider_handoff::Issue31ProviderHandoffError::Unprojectable(
2380                        _,
2381                    ) => "outcome.omega.unavailable".into(),
2382                    _ => "outcome.omega.refused".into(),
2383                },
2384                reason_ref: Some(error.reason_ref().into()),
2385            },
2386        }
2387    }
2388
2389    /// Install how this host reads its own provider roster (omega#91).
2390    ///
2391    /// Without one, no handoff ever binds: the host would be deciding against a
2392    /// roster nobody read. Open handoffs still reach their deadline, so an
2393    /// unwired host produces `expired` rather than a request that hangs.
2394    pub fn set_issue31_provider_roster_source(&mut self, source: Issue31ProviderRosterSource) {
2395        self.issue31_provider_roster_source = Some(source);
2396    }
2397
2398    /// The host's provider connection handoffs, exactly as persisted.
2399    pub fn issue31_provider_handoff_refs(&self) -> Vec<String> {
2400        self.issue31_provider_handoffs
2401            .records()
2402            .map(|record| record.handoff_ref.clone())
2403            .collect()
2404    }
2405
2406    /// The contract rows this host would publish right now.
2407    pub fn issue31_projected_provider_handoffs(&self, generated_at_ms: u64) -> Vec<Value> {
2408        self.issue31_provider_handoffs
2409            .projected(generated_at_ms)
2410            .rows
2411    }
2412
2413    /// Move every open handoff on by at most one observation of the roster.
2414    ///
2415    /// Run once per host pump pass, before the omega#47 documents are built, so
2416    /// what the phone reads is this pass's state rather than the previous
2417    /// pass's.
2418    fn advance_issue31_provider_handoffs(&mut self, now: u64) {
2419        let roster = self
2420            .issue31_provider_roster_source
2421            .as_ref()
2422            .and_then(|source| source());
2423        self.issue31_provider_handoffs
2424            .advance(roster.as_deref(), now.saturating_mul(1_000));
2425    }
2426
2427    fn execute_issue31_action_v2(
2428        &mut self,
2429        arguments: &Issue31CommandArguments,
2430        idempotency_ref: &str,
2431        grant_ref: &str,
2432        device_public_key_hex: &str,
2433        expected_generation: u64,
2434    ) -> Issue31CommandExecutionV2 {
2435        let handling_suffix = &format!("{:x}", Sha256::digest(idempotency_ref.as_bytes()))[..24];
2436        let accepted = |source_event_id| Issue31CommandExecutionV2 {
2437            status: Issue31CommandHandlingStatus::Accepted,
2438            handling_ref: format!("handling.omega.{handling_suffix}"),
2439            reason_ref: None,
2440            source_event_id,
2441        };
2442        let unavailable = |reason: &str| Issue31CommandExecutionV2 {
2443            status: Issue31CommandHandlingStatus::Unavailable,
2444            handling_ref: format!("handling.omega.{handling_suffix}"),
2445            reason_ref: Some(reason.into()),
2446            source_event_id: None,
2447        };
2448        let failed = |reason: &str| Issue31CommandExecutionV2 {
2449            status: Issue31CommandHandlingStatus::Failed,
2450            handling_ref: format!("handling.omega.{handling_suffix}"),
2451            reason_ref: Some(reason.into()),
2452            source_event_id: None,
2453        };
2454        match arguments {
2455            Issue31CommandArguments::SendMessage {
2456                conversation, text, ..
2457            } if conversation == &self.config.conversation_ref() => {
2458                match self.send_message(text, idempotency_ref, expected_generation) {
2459                    Ok(result) => {
2460                        let projection = Issue31OwnerProjectionBody::Message {
2461                            role: Issue31SourceRole::Owner,
2462                            conversation: conversation.clone(),
2463                            text: text.clone(),
2464                            reply_to_event_id: None,
2465                        };
2466                        match self.enqueue_issue31_source_projection(
2467                            &result.event_id,
2468                            projection,
2469                            grant_ref,
2470                            device_public_key_hex,
2471                            expected_generation,
2472                        ) {
2473                            Ok(()) => accepted(Some(result.event_id)),
2474                            // The owner record is already published by this
2475                            // point. A transport failure reading it back is not
2476                            // the same thing as a projection the host cannot
2477                            // build, and reporting it as terminal `failed`
2478                            // would tell the device a message that exists on
2479                            // the relay does not — the same class of mistake as
2480                            // reporting an authentication failure as a
2481                            // discovery one. `unavailable` is retryable, and
2482                            // the periodic source scan projects it either way.
2483                            Err(
2484                                SarahConversationError::Relay(_)
2485                                | SarahConversationError::Identity(_)
2486                                | SarahConversationError::IdentityRequired,
2487                            ) => unavailable("reason.omega.transport_unavailable"),
2488                            Err(_) => failed("reason.omega.projection_failed"),
2489                        }
2490                    }
2491                    Err(
2492                        SarahConversationError::Relay(_)
2493                        | SarahConversationError::Identity(_)
2494                        | SarahConversationError::IdentityRequired,
2495                    ) => unavailable("reason.omega.transport_unavailable"),
2496                    Err(_) => failed("reason.omega.action_failed"),
2497                }
2498            }
2499            Issue31CommandArguments::InterruptTurn {
2500                conversation,
2501                turn_ref,
2502                ..
2503            } if conversation == &self.config.conversation_ref() => {
2504                match self.interrupt_turn(turn_ref, idempotency_ref, expected_generation) {
2505                    Ok(_) => accepted(None),
2506                    Err(
2507                        SarahConversationError::Relay(_)
2508                        | SarahConversationError::Identity(_)
2509                        | SarahConversationError::IdentityRequired,
2510                    ) => unavailable("reason.omega.transport_unavailable"),
2511                    Err(_) => failed("reason.omega.action_failed"),
2512                }
2513            }
2514            Issue31CommandArguments::SendMessage { .. }
2515            | Issue31CommandArguments::InterruptTurn { .. } => Issue31CommandExecutionV2 {
2516                status: Issue31CommandHandlingStatus::Refused,
2517                handling_ref: format!("handling.omega.{handling_suffix}"),
2518                reason_ref: Some("reason.omega.conversation_mismatch".into()),
2519                source_event_id: None,
2520            },
2521            Issue31CommandArguments::ReadStatePatch { .. }
2522            | Issue31CommandArguments::ReminderCreate { .. }
2523            | Issue31CommandArguments::ReminderChange { .. }
2524            | Issue31CommandArguments::ReminderComplete { .. }
2525            | Issue31CommandArguments::ReminderCancel { .. } => {
2526                match self.execute_issue31_owner_state_action(arguments) {
2527                    Ok((event_id, projection)) => match self.enqueue_issue31_source_projection(
2528                        &event_id,
2529                        projection,
2530                        grant_ref,
2531                        device_public_key_hex,
2532                        expected_generation,
2533                    ) {
2534                        Ok(()) => accepted(Some(event_id)),
2535                        Err(_) => failed("reason.omega.projection_failed"),
2536                    },
2537                    Err(
2538                        SarahConversationError::Relay(_)
2539                        | SarahConversationError::Identity(_)
2540                        | SarahConversationError::IdentityRequired,
2541                    ) => unavailable("reason.omega.transport_unavailable"),
2542                    Err(_) => failed("reason.omega.action_failed"),
2543                }
2544            }
2545        }
2546    }
2547
2548    fn execute_issue31_owner_state_action(
2549        &mut self,
2550        arguments: &Issue31CommandArguments,
2551    ) -> Result<(String, Issue31OwnerProjectionBody), SarahConversationError> {
2552        match arguments {
2553            Issue31CommandArguments::ReadStatePatch {
2554                slot_id,
2555                client_id,
2556                context_ref,
2557                read_at,
2558                ..
2559            } => {
2560                if *read_at > u32::MAX as u64 {
2561                    return Err(SarahConversationError::InvalidRequest(
2562                        "read-state timestamp exceeds the NIP-RS bound".into(),
2563                    ));
2564                }
2565                let d_tag = format!("read-state:{slot_id}");
2566                let mut contexts = self.load_issue31_read_state_contexts(&d_tag)?;
2567                contexts
2568                    .entry(context_ref.clone())
2569                    .and_modify(|current| *current = (*current).max(*read_at))
2570                    .or_insert(*read_at);
2571                if contexts.len() > 10_000 {
2572                    return Err(SarahConversationError::InvalidRequest(
2573                        "read-state context bound is exhausted".into(),
2574                    ));
2575                }
2576                let plaintext = serde_json::to_string(&json!({
2577                    "v": 1,
2578                    "client_id": client_id,
2579                    "contexts": contexts,
2580                }))
2581                .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
2582                let tags = vec![
2583                    Tag::parse(["d", d_tag.as_str()]).map_err(|error| {
2584                        SarahConversationError::InvalidRequest(error.to_string())
2585                    })?,
2586                    Tag::parse(["t", "read-state"]).map_err(|error| {
2587                        SarahConversationError::InvalidRequest(error.to_string())
2588                    })?,
2589                    Tag::parse(["alt", "encrypted read state"]).map_err(|error| {
2590                        SarahConversationError::InvalidRequest(error.to_string())
2591                    })?,
2592                ];
2593                let event_id =
2594                    self.publish_issue31_encrypted_source(SARAH_READ_STATE_KIND, &plaintext, tags)?;
2595                Ok((
2596                    event_id,
2597                    Issue31OwnerProjectionBody::ReadState { d_tag, plaintext },
2598                ))
2599            }
2600            Issue31CommandArguments::ReminderCreate {
2601                reminder_id,
2602                note,
2603                target_event_id,
2604                not_before,
2605                expiration,
2606                ..
2607            }
2608            | Issue31CommandArguments::ReminderChange {
2609                reminder_id,
2610                note,
2611                target_event_id,
2612                not_before,
2613                expiration,
2614                ..
2615            } => {
2616                let plaintext = serde_json::to_string(&json!({
2617                    "status": "pending",
2618                    "note": note,
2619                    "target": target_event_id.as_ref().map(|event_id| json!({ "id": event_id })),
2620                }))
2621                .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
2622                let mut tags = vec![
2623                    Tag::parse(["d", reminder_id.as_str()]).map_err(|error| {
2624                        SarahConversationError::InvalidRequest(error.to_string())
2625                    })?,
2626                    Tag::parse(["alt", "Encrypted reminder"]).map_err(|error| {
2627                        SarahConversationError::InvalidRequest(error.to_string())
2628                    })?,
2629                    Tag::parse(["not_before", not_before.to_string().as_str()]).map_err(
2630                        |error| SarahConversationError::InvalidRequest(error.to_string()),
2631                    )?,
2632                ];
2633                if let Some(expiration) = expiration {
2634                    tags.push(
2635                        Tag::parse(["expiration", expiration.to_string().as_str()]).map_err(
2636                            |error| SarahConversationError::InvalidRequest(error.to_string()),
2637                        )?,
2638                    );
2639                }
2640                let event_id =
2641                    self.publish_issue31_encrypted_source(SARAH_REMINDER_KIND, &plaintext, tags)?;
2642                Ok((
2643                    event_id,
2644                    Issue31OwnerProjectionBody::Reminder {
2645                        reminder_id: reminder_id.clone(),
2646                        plaintext,
2647                        not_before: Some(*not_before),
2648                        expiration: *expiration,
2649                    },
2650                ))
2651            }
2652            Issue31CommandArguments::ReminderComplete { reminder_id, .. }
2653            | Issue31CommandArguments::ReminderCancel { reminder_id, .. } => {
2654                let status =
2655                    if matches!(arguments, Issue31CommandArguments::ReminderComplete { .. }) {
2656                        "done"
2657                    } else {
2658                        "cancelled"
2659                    };
2660                let plaintext = serde_json::to_string(&json!({ "status": status }))
2661                    .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
2662                let tags = vec![
2663                    Tag::parse(["d", reminder_id.as_str()]).map_err(|error| {
2664                        SarahConversationError::InvalidRequest(error.to_string())
2665                    })?,
2666                    Tag::parse(["alt", "Encrypted reminder"]).map_err(|error| {
2667                        SarahConversationError::InvalidRequest(error.to_string())
2668                    })?,
2669                ];
2670                let event_id =
2671                    self.publish_issue31_encrypted_source(SARAH_REMINDER_KIND, &plaintext, tags)?;
2672                Ok((
2673                    event_id,
2674                    Issue31OwnerProjectionBody::Reminder {
2675                        reminder_id: reminder_id.clone(),
2676                        plaintext,
2677                        not_before: None,
2678                        expiration: None,
2679                    },
2680                ))
2681            }
2682            Issue31CommandArguments::SendMessage { .. }
2683            | Issue31CommandArguments::InterruptTurn { .. } => Err(
2684                SarahConversationError::InvalidRequest("not an owner-state action".into()),
2685            ),
2686        }
2687    }
2688
2689    fn load_issue31_read_state_contexts(
2690        &mut self,
2691        d_tag: &str,
2692    ) -> Result<BTreeMap<String, u64>, SarahConversationError> {
2693        let conversation_ref = self.config.conversation_ref();
2694        let mut cursor = None;
2695        let mut contexts: BTreeMap<String, u64> = BTreeMap::new();
2696        for _ in 0..8 {
2697            let page =
2698                self.query_with_auth(&conversation_ref, cursor.as_deref(), MAX_PAGE_LIMIT)?;
2699            for event in page.events {
2700                if event.kind != SARAH_READ_STATE_KIND
2701                    || event.pubkey != self.config.identity.owner_public_key_hex
2702                    || stored_tag_value(&event.tags, "d").as_deref() != Some(d_tag)
2703                {
2704                    continue;
2705                }
2706                let plaintext = self
2707                    .signer
2708                    .decrypt_record(&event.pubkey, &event.content_summary)?;
2709                let value: Value = serde_json::from_str(&plaintext)
2710                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
2711                let Some(record_contexts) = value.get("contexts").and_then(Value::as_object) else {
2712                    continue;
2713                };
2714                for (context_ref, read_at) in record_contexts {
2715                    let Some(read_at) = read_at.as_u64() else {
2716                        continue;
2717                    };
2718                    if context_ref.len() <= 256 && read_at <= u32::MAX as u64 {
2719                        contexts
2720                            .entry(context_ref.clone())
2721                            .and_modify(|current| *current = (*current).max(read_at))
2722                            .or_insert(read_at);
2723                    }
2724                }
2725            }
2726            let Some(next_cursor) = page.next_cursor else {
2727                break;
2728            };
2729            cursor = Some(next_cursor);
2730        }
2731        Ok(contexts)
2732    }
2733
2734    fn publish_issue31_encrypted_source(
2735        &mut self,
2736        kind: u16,
2737        plaintext: &str,
2738        tags: Vec<Tag>,
2739    ) -> Result<String, SarahConversationError> {
2740        let event = self
2741            .signer
2742            .sign_encrypted_self_record(kind, plaintext, tags)?;
2743        let event_id = event.id.to_hex();
2744        self.publish_with_auth(&event)?;
2745        Ok(event_id)
2746    }
2747
2748    fn enqueue_issue31_source_projection(
2749        &mut self,
2750        source_event_id: &str,
2751        projection: Issue31OwnerProjectionBody,
2752        grant_ref: &str,
2753        device_public_key_hex: &str,
2754        expected_generation: u64,
2755    ) -> Result<(), SarahConversationError> {
2756        let source = self.load_issue31_source_event(source_event_id)?;
2757        if source.pubkey != self.config.identity.owner_public_key_hex {
2758            return Err(SarahConversationError::InvalidRequest(
2759                "Issue 31 projection source is not owner-authored".into(),
2760            ));
2761        }
2762        let emission = emit_issue31_owner_projection(Issue31OwnerProjectionInput {
2763            host_ref: "omega.host.local",
2764            host_public_key_hex: &self.config.identity.owner_public_key_hex,
2765            device_public_key_hex,
2766            sarah_public_key_hex: &self.config.identity.sarah_public_key_hex,
2767            grant_ref,
2768            expected_generation,
2769            source_event_id: &source.event_id,
2770            source_author_public_key_hex: &source.pubkey,
2771            source_kind: source.kind,
2772            source_created_at: source.created_at,
2773            projected_at: unix_now().max(source.created_at),
2774            projection,
2775        })
2776        .map_err(issue31_error)?;
2777        self.enqueue_issue31_private_content(
2778            crate::ISSUE31_OWNER_PROJECTION_SCHEMA,
2779            &emission.content,
2780            device_public_key_hex,
2781        )?;
2782        Ok(())
2783    }
2784
2785    fn load_issue31_source_event(
2786        &mut self,
2787        source_event_id: &str,
2788    ) -> Result<StoredConversationEvent, SarahConversationError> {
2789        let conversation_ref = self.config.conversation_ref();
2790        let mut cursor = self.issue31_projection_cursor.clone();
2791        for _ in 0..8 {
2792            let page =
2793                self.query_with_auth(&conversation_ref, cursor.as_deref(), MAX_PAGE_LIMIT)?;
2794            if let Some(source) = page
2795                .events
2796                .iter()
2797                .find(|event| event.event_id == source_event_id)
2798            {
2799                return Ok(source.clone());
2800            }
2801            let Some(next_cursor) = page.next_cursor else {
2802                break;
2803            };
2804            cursor = Some(next_cursor);
2805        }
2806        Err(SarahConversationError::Relay(
2807            "confirmed Issue 31 source event was not observable after publish".into(),
2808        ))
2809    }
2810
2811    fn project_issue31_sources(
2812        &mut self,
2813        controller: &mut Issue31HostController,
2814        now: u64,
2815    ) -> Result<(), SarahConversationError> {
2816        let grants = controller.active_grants(now).map_err(issue31_error)?;
2817        if grants.is_empty() {
2818            return Ok(());
2819        }
2820        let conversation_ref = self.config.conversation_ref();
2821        let mut cursor = self.issue31_projection_cursor.clone();
2822        let mut last_scanned_cursor = None;
2823        let mut scan_bound_reached = true;
2824        for _ in 0..8 {
2825            let page =
2826                self.query_with_auth(&conversation_ref, cursor.as_deref(), MAX_PAGE_LIMIT)?;
2827            if let Some(page_cursor) = page.events.last().map(stored_event_cursor) {
2828                last_scanned_cursor = Some(page_cursor);
2829            }
2830            for source in page.events {
2831                if !matches!(
2832                    source.kind,
2833                    crate::ISSUE31_PRIVATE_RUMOR_KIND
2834                        | SARAH_TURN_RECORD_KIND
2835                        | SARAH_AUTHORITY_RECEIPT_KIND
2836                        | SARAH_ENGRAM_KIND
2837                        | SARAH_READ_STATE_KIND
2838                        | SARAH_REMINDER_KIND
2839                ) {
2840                    continue;
2841                }
2842                if source.kind == crate::ISSUE31_PRIVATE_RUMOR_KIND
2843                    && source.record_kind != "message"
2844                {
2845                    continue;
2846                }
2847                let projection = match self.issue31_projection_body(&source) {
2848                    Ok(Some(projection)) => projection,
2849                    Ok(None) => continue,
2850                    Err(_) => {
2851                        self.quarantine_issue31_event(
2852                            &source.event_id,
2853                            ISSUE31_PROJECTION_SOURCE_QUARANTINE_REASON,
2854                            controller,
2855                        )?;
2856                        continue;
2857                    }
2858                };
2859                let mut emissions = Vec::new();
2860                let mut refused_by_own_decoder = false;
2861                for grant in &grants {
2862                    if controller.source_was_projected(
2863                        &grant.grant_ref,
2864                        grant.generation,
2865                        &source.event_id,
2866                    ) {
2867                        continue;
2868                    }
2869                    match emit_issue31_owner_projection(Issue31OwnerProjectionInput {
2870                        host_ref: &grant.host_ref,
2871                        host_public_key_hex: &grant.host_public_key_hex,
2872                        device_public_key_hex: &grant.device_public_key_hex,
2873                        sarah_public_key_hex: &grant.sarah_public_key_hex,
2874                        grant_ref: &grant.grant_ref,
2875                        expected_generation: grant.generation,
2876                        source_event_id: &source.event_id,
2877                        source_author_public_key_hex: &source.pubkey,
2878                        source_kind: source.kind,
2879                        source_created_at: source.created_at,
2880                        projected_at: now.max(source.created_at),
2881                        projection: projection.clone(),
2882                    }) {
2883                        Ok(emission) => emissions.push((grant, emission)),
2884                        // A source event Sarah or the owner signed can still
2885                        // carry a reference or a body the device reader
2886                        // refuses. Quarantining that one event keeps the
2887                        // remaining sources projecting, rather than letting a
2888                        // single malformed record stop every device.
2889                        Err(_) => {
2890                            refused_by_own_decoder = true;
2891                            break;
2892                        }
2893                    }
2894                }
2895                if refused_by_own_decoder {
2896                    self.quarantine_issue31_event(
2897                        &source.event_id,
2898                        ISSUE31_PROJECTION_SOURCE_QUARANTINE_REASON,
2899                        controller,
2900                    )?;
2901                    continue;
2902                }
2903                for (grant, emission) in emissions {
2904                    self.enqueue_issue31_private_content(
2905                        crate::ISSUE31_OWNER_PROJECTION_SCHEMA,
2906                        &emission.content,
2907                        &grant.device_public_key_hex,
2908                    )?;
2909                    controller
2910                        .record_source_projection(
2911                            grant.grant_ref.clone(),
2912                            grant.generation,
2913                            source.event_id.clone(),
2914                        )
2915                        .map_err(issue31_error)?;
2916                }
2917            }
2918            let Some(next_cursor) = page.next_cursor else {
2919                scan_bound_reached = false;
2920                break;
2921            };
2922            cursor = Some(next_cursor);
2923        }
2924        if scan_bound_reached {
2925            // The scan ran out of pages before it ran out of conversation. The
2926            // host does not know how many sources it never looked at, so it
2927            // must say so rather than let the device read a short list as a
2928            // complete one.
2929            self.last_gap_state = strongest_gap_state(self.last_gap_state, GapState::Possible);
2930        }
2931        if let Some(last_scanned_cursor) = last_scanned_cursor {
2932            self.issue31_projection_cursor = Some(last_scanned_cursor);
2933        }
2934        self.emit_issue31_withheld_sources(&grants, scan_bound_reached, now)?;
2935        Ok(())
2936    }
2937
2938    /// Install the live Full Auto reading this host publishes to devices.
2939    ///
2940    /// Without one the host publishes no omega#47 records at all, and a paired
2941    /// phone reads `no_host_projection` — which is the honest answer for a host
2942    /// that is not observing its own Full Auto state, and is exactly what the
2943    /// phone showed before this existed. The difference is that the silence is
2944    /// now a stated condition rather than a dropped record.
2945    pub fn set_issue31_host_projection_source(&mut self, source: Issue31HostProjectionSource) {
2946        self.issue31_host_projection_source = Some(source);
2947    }
2948
2949    /// Bind the issue-31 host controller this client pumps.
2950    ///
2951    /// `new_production` builds one from custody. A headless harness cannot
2952    /// reach custody and still has to drive the shipped pump, so the binding is
2953    /// a real operation rather than a field only this module can reach.
2954    pub fn attach_issue31_host_controller(&mut self, controller: Issue31HostController) {
2955        self.issue31_host = Some(controller);
2956    }
2957
2958    /// The `grant_ref:generation` keys this host has published omega#47
2959    /// documents to. A key is recorded only after both records were committed
2960    /// to the durable outbox.
2961    pub fn issue31_published_host_adjunct_grants(&self) -> Vec<String> {
2962        self.issue31_host_adjunct_emissions.keys().cloned().collect()
2963    }
2964
2965    /// Owner-private records still waiting for a relay acknowledgement.
2966    ///
2967    /// An empty list after a pump pass means every relay the host is
2968    /// configured for stored every record; a non-empty one is an exact,
2969    /// resumable backlog rather than a lost publish.
2970    pub fn issue31_pending_private_publish_refs(&self) -> Vec<String> {
2971        self.issue31_private_outbox.keys().cloned().collect()
2972    }
2973
2974    /// Address one omega#47 document to one device.
2975    ///
2976    /// The pump — not the reading — states the binding, because the reading
2977    /// knows the host's runs and knows nothing about who may read them. A
2978    /// document that arrives already claiming a delivery binding is refused
2979    /// rather than overwritten: whoever wrote it was making a claim about a
2980    /// device it has no standing to make.
2981    fn address_issue31_adjunct(
2982        document: &Value,
2983        schema: &str,
2984        record_type: &str,
2985        grant: &Issue31GrantState,
2986    ) -> Result<Value, SarahConversationError> {
2987        let object = document.as_object().ok_or_else(|| {
2988            SarahConversationError::Internal("Issue 31 adjunct is not a record".into())
2989        })?;
2990        if object.get("schema").and_then(Value::as_str) != Some(schema) {
2991            return Err(SarahConversationError::Internal(
2992                "Issue 31 adjunct does not carry its own schema".into(),
2993            ));
2994        }
2995        // The seal proves who signed; it cannot prove which host the body
2996        // describes. The grant is the only statement that relates this host key
2997        // to a host reference, so the body must agree with it or the device
2998        // would bind another machine's state to this pairing.
2999        if object.get("hostRef").and_then(Value::as_str) != Some(grant.host_ref.as_str()) {
3000            return Err(SarahConversationError::Internal(
3001                "Issue 31 adjunct describes a host this grant does not name".into(),
3002            ));
3003        }
3004        if ISSUE31_ADJUNCT_DELIVERY_KEYS
3005            .iter()
3006            .any(|key| object.contains_key(*key))
3007        {
3008            return Err(SarahConversationError::Internal(
3009                "Issue 31 adjunct arrived already claiming a delivery binding".into(),
3010            ));
3011        }
3012        let mut addressed = object.clone();
3013        addressed.insert("recordType".into(), json!(record_type));
3014        addressed.insert(
3015            "hostPublicKeyHex".into(),
3016            json!(grant.host_public_key_hex.clone()),
3017        );
3018        addressed.insert(
3019            "devicePublicKeyHex".into(),
3020            json!(grant.device_public_key_hex.clone()),
3021        );
3022        addressed.insert("grantRef".into(), json!(grant.grant_ref.clone()));
3023        addressed.insert("expectedGeneration".into(), json!(grant.generation));
3024        Ok(Value::Object(addressed))
3025    }
3026
3027    /// Publish the omega#47 host snapshot and its Full Auto detail to every
3028    /// admitted device (omega#49).
3029    ///
3030    /// The two documents are published together or not at all. The detail is
3031    /// bound to the snapshot that advertised it, and a device that held one
3032    /// without the other would either render a detail nothing vouches for or
3033    /// advertise capabilities it cannot open.
3034    fn publish_issue31_host_adjuncts(
3035        &mut self,
3036        controller: &Issue31HostController,
3037        now: u64,
3038    ) -> Result<(), SarahConversationError> {
3039        let Some(source) = self.issue31_host_projection_source.clone() else {
3040            return Ok(());
3041        };
3042        let grants = controller.active_grants(now).map_err(issue31_error)?;
3043        let observed_at_ms = now.saturating_mul(1_000);
3044        // omega#91. A handoff the host holds and can never state — one whose
3045        // request time was never measured — makes the owner's view short, and
3046        // that has to be visible rather than read as "no handoff in flight".
3047        if !self.issue31_provider_handoffs.unstateable_refs().is_empty() {
3048            self.last_gap_state = strongest_gap_state(self.last_gap_state, GapState::Possible);
3049        }
3050        for grant in grants {
3051            let documents = match source(&Issue31HostProjectionRequest {
3052                host_ref: &grant.host_ref,
3053                host_public_key_hex: &grant.host_public_key_hex,
3054                device_public_key_hex: &grant.device_public_key_hex,
3055                grant_ref: &grant.grant_ref,
3056                expected_generation: grant.generation,
3057                observed_at_ms,
3058                handoffs: &self.issue31_provider_handoffs,
3059            }) {
3060                Ok(Some(documents)) => documents,
3061                // Nothing observed is said by saying nothing. Publishing an
3062                // empty projection here would claim the host had looked and
3063                // found no runs, which is a different fact.
3064                Ok(None) => continue,
3065                Err(_) => {
3066                    // The host could not read its own Full Auto state. The
3067                    // owner's view may therefore be short, and that has to be
3068                    // visible rather than inferred from an absent record.
3069                    self.last_gap_state =
3070                        strongest_gap_state(self.last_gap_state, GapState::Possible);
3071                    continue;
3072                }
3073            };
3074            let host = Self::address_issue31_adjunct(
3075                &documents.host,
3076                ISSUE31_HOST_ADJUNCT_SCHEMA,
3077                ISSUE31_HOST_ADJUNCT_RECORD_TYPE,
3078                &grant,
3079            )?;
3080            let detail = Self::address_issue31_adjunct(
3081                &documents.detail,
3082                ISSUE31_FULL_AUTO_ADJUNCT_SCHEMA,
3083                ISSUE31_FULL_AUTO_ADJUNCT_RECORD_TYPE,
3084                &grant,
3085            )?;
3086            // "Beside" is the contract's word for the relation between the two
3087            // documents. A detail carrying a different snapshot reference is
3088            // one the device would refuse as `snapshot_mismatch`, so sending it
3089            // would publish a refusal rather than a projection.
3090            if host.get("snapshotRef") != detail.get("snapshotRef") {
3091                return Err(SarahConversationError::Internal(
3092                    "Issue 31 Full Auto detail is not bound to the snapshot beside it".into(),
3093                ));
3094            }
3095            let host_content = serde_json::to_string(&host)
3096                .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
3097            let detail_content = serde_json::to_string(&detail)
3098                .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
3099            let key = format!("{}:{}", grant.grant_ref, grant.generation);
3100            let digest = format!(
3101                "{:x}",
3102                Sha256::digest(
3103                    [host_content.as_bytes(), detail_content.as_bytes()].concat()
3104                )
3105            );
3106            if self.issue31_host_adjunct_emissions.get(&key) == Some(&digest) {
3107                continue;
3108            }
3109            let enqueued_host = self.enqueue_issue31_private_content(
3110                ISSUE31_HOST_ADJUNCT_SCHEMA,
3111                &host_content,
3112                &grant.device_public_key_hex,
3113            )?;
3114            let enqueued_detail = match self.enqueue_issue31_private_content(
3115                ISSUE31_FULL_AUTO_ADJUNCT_SCHEMA,
3116                &detail_content,
3117                &grant.device_public_key_hex,
3118            ) {
3119                Ok(enqueued) => enqueued,
3120                Err(error) => {
3121                    self.rollback_issue31_enqueue(&enqueued_host);
3122                    return Err(error);
3123                }
3124            };
3125            let previous = self
3126                .issue31_host_adjunct_emissions
3127                .insert(key.clone(), digest);
3128            // The bookkeeping is committed with the outbox, not after it: a
3129            // crash between the two would either resend forever or, worse,
3130            // record a publication that never happened.
3131            if let Err(error) = self.persist_issue31_host_state_with_controller(controller) {
3132                match previous {
3133                    Some(previous) => {
3134                        self.issue31_host_adjunct_emissions.insert(key, previous);
3135                    }
3136                    None => {
3137                        self.issue31_host_adjunct_emissions.remove(&key);
3138                    }
3139                }
3140                self.rollback_issue31_enqueue(&enqueued_detail);
3141                self.rollback_issue31_enqueue(&enqueued_host);
3142                return Err(error);
3143            }
3144        }
3145        Ok(())
3146    }
3147
3148    /// Tell every admitted device how complete its own projection is.
3149    ///
3150    /// This is the device-visible half of the two host-local surfaces that
3151    /// silently shorten the owner's view: the quarantine, whose count only ever
3152    /// reached `BootstrapResult`, and the bounded projection scan, which only
3153    /// ever moved `last_gap_state`. A complete pass publishes a record too —
3154    /// silence has to mean "unknown", or the absence of a signal reads as
3155    /// completeness and this whole mechanism buys nothing.
3156    fn emit_issue31_withheld_sources(
3157        &mut self,
3158        grants: &[Issue31GrantState],
3159        scan_bound_reached: bool,
3160        now: u64,
3161    ) -> Result<(), SarahConversationError> {
3162        let quarantined = u32::try_from(
3163            self.issue31_quarantined_events
3164                .values()
3165                .filter(|reason| reason.as_str() == ISSUE31_PROJECTION_SOURCE_QUARANTINE_REASON)
3166                .count(),
3167        )
3168        .unwrap_or(u32::MAX);
3169        let mut withheld = Vec::new();
3170        if quarantined > 0 {
3171            withheld.push(Issue31WithheldSourceCount {
3172                cause: Issue31WithheldCause::Quarantined,
3173                count: quarantined,
3174                exact: true,
3175                reason_ref: ISSUE31_PROJECTION_SOURCE_QUARANTINE_REASON.to_string(),
3176            });
3177        }
3178        if scan_bound_reached {
3179            withheld.push(Issue31WithheldSourceCount {
3180                cause: Issue31WithheldCause::ScanBound,
3181                count: 1,
3182                exact: false,
3183                reason_ref: ISSUE31_PROJECTION_SCAN_BOUND_REASON.to_string(),
3184            });
3185        }
3186        for grant in grants {
3187            let emission = emit_issue31_withheld_sources(Issue31WithheldSourcesInput {
3188                host_ref: &grant.host_ref,
3189                host_public_key_hex: &grant.host_public_key_hex,
3190                device_public_key_hex: &grant.device_public_key_hex,
3191                grant_ref: &grant.grant_ref,
3192                expected_generation: grant.generation,
3193                observed_at: now,
3194                withheld: withheld.clone(),
3195            })
3196            .map_err(issue31_error)?;
3197            let key = format!("{}:{}", grant.grant_ref, grant.generation);
3198            let substance = emission.record.substance();
3199            if self.issue31_withheld_emissions.get(&key) == Some(&substance) {
3200                continue;
3201            }
3202            self.enqueue_issue31_private_content(
3203                crate::ISSUE31_WITHHELD_SOURCES_SCHEMA,
3204                &emission.content,
3205                &grant.device_public_key_hex,
3206            )?;
3207            self.issue31_withheld_emissions.insert(key, substance);
3208        }
3209        Ok(())
3210    }
3211
3212    fn issue31_projection_body(
3213        &self,
3214        source: &StoredConversationEvent,
3215    ) -> Result<Option<Issue31OwnerProjectionBody>, SarahConversationError> {
3216        let owner_key = &self.config.identity.owner_public_key_hex;
3217        let sarah_key = &self.config.identity.sarah_public_key_hex;
3218        match source.kind {
3219            crate::ISSUE31_PRIVATE_RUMOR_KIND
3220                if source.record_kind == "message"
3221                    && (source.pubkey == *owner_key || source.pubkey == *sarah_key) =>
3222            {
3223                let role = if source.pubkey == *owner_key {
3224                    Issue31SourceRole::Owner
3225                } else {
3226                    Issue31SourceRole::Sarah
3227                };
3228                Ok(Some(Issue31OwnerProjectionBody::Message {
3229                    role,
3230                    conversation: self.config.conversation_ref(),
3231                    text: source.content_summary.clone(),
3232                    reply_to_event_id: stored_tag_value(&source.tags, "e"),
3233                }))
3234            }
3235            SARAH_TURN_RECORD_KIND if source.pubkey == *sarah_key => {
3236                let plaintext = self
3237                    .signer
3238                    .decrypt_record(&source.pubkey, &source.content_summary)?;
3239                let payload = serde_json::from_str::<Value>(&plaintext)
3240                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
3241                Ok(Some(Issue31OwnerProjectionBody::Turn { payload }))
3242            }
3243            SARAH_AUTHORITY_RECEIPT_KIND if source.pubkey == *sarah_key => {
3244                let plaintext = self
3245                    .signer
3246                    .decrypt_record(&source.pubkey, &source.content_summary)?;
3247                let value = serde_json::from_str::<Value>(&plaintext)
3248                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
3249                let state = match value.get("decision").and_then(Value::as_str) {
3250                    Some("allow") => "allowed",
3251                    Some("refuse") => "refused",
3252                    _ => {
3253                        return Err(SarahConversationError::InvalidRequest(
3254                            "authority receipt has an invalid decision".into(),
3255                        ));
3256                    }
3257                };
3258                let turn_ref = stored_tag_value(&source.tags, "turn").ok_or_else(|| {
3259                    SarahConversationError::InvalidRequest(
3260                        "authority receipt omitted its turn tag".into(),
3261                    )
3262                })?;
3263                let suffix = &source.event_id[..24];
3264                let reason_ref = value
3265                    .get("reservedCategory")
3266                    .and_then(Value::as_str)
3267                    .filter(|category| {
3268                        category
3269                            .bytes()
3270                            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
3271                    })
3272                    .map(|category| format!("reason.openagents.{category}"));
3273                Ok(Some(Issue31OwnerProjectionBody::AuthorityReceipt {
3274                    receipt_ref: format!("receipt.issue31.{suffix}"),
3275                    turn_ref,
3276                    authority_decision: Issue31AuthorityDecisionProjection {
3277                        state: state.into(),
3278                        decision_ref: format!("decision.issue31.{suffix}"),
3279                        reason_ref,
3280                    },
3281                    target_outcome: Issue31TargetOutcomeProjection {
3282                        state: "pending".into(),
3283                        outcome_ref: None,
3284                        reason_ref: None,
3285                    },
3286                }))
3287            }
3288            SARAH_ENGRAM_KIND if source.pubkey == *sarah_key => {
3289                let plaintext = self
3290                    .signer
3291                    .decrypt_record(&source.pubkey, &source.content_summary)?;
3292                let d_tag = stored_tag_value(&source.tags, "d").ok_or_else(|| {
3293                    SarahConversationError::InvalidRequest("engram omitted its d tag".into())
3294                })?;
3295                Ok(Some(Issue31OwnerProjectionBody::Engram {
3296                    d_tag,
3297                    plaintext,
3298                }))
3299            }
3300            SARAH_READ_STATE_KIND if source.pubkey == *owner_key => {
3301                let plaintext = self
3302                    .signer
3303                    .decrypt_record(&source.pubkey, &source.content_summary)?;
3304                let d_tag = stored_tag_value(&source.tags, "d").ok_or_else(|| {
3305                    SarahConversationError::InvalidRequest("read state omitted its d tag".into())
3306                })?;
3307                Ok(Some(Issue31OwnerProjectionBody::ReadState {
3308                    d_tag,
3309                    plaintext,
3310                }))
3311            }
3312            SARAH_REMINDER_KIND if source.pubkey == *owner_key => {
3313                let plaintext = self
3314                    .signer
3315                    .decrypt_record(&source.pubkey, &source.content_summary)?;
3316                let reminder_id = stored_tag_value(&source.tags, "d").ok_or_else(|| {
3317                    SarahConversationError::InvalidRequest("reminder omitted its d tag".into())
3318                })?;
3319                let not_before = stored_tag_value(&source.tags, "not_before")
3320                    .map(|value| value.parse::<u64>())
3321                    .transpose()
3322                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
3323                let expiration = stored_tag_value(&source.tags, "expiration")
3324                    .map(|value| value.parse::<u64>())
3325                    .transpose()
3326                    .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
3327                Ok(Some(Issue31OwnerProjectionBody::Reminder {
3328                    reminder_id,
3329                    plaintext,
3330                    not_before,
3331                    expiration,
3332                }))
3333            }
3334            _ => Ok(None),
3335        }
3336    }
3337
3338    pub fn room_snapshot(
3339        &mut self,
3340        cursor: Option<&str>,
3341        limit: Option<usize>,
3342    ) -> Result<RoomSnapshotResult, SarahConversationError> {
3343        self.room_snapshot_with_record_cursor(cursor, limit, cursor, limit, cursor, limit)
3344    }
3345
3346    pub fn room_snapshot_with_cursors(
3347        &mut self,
3348        transcript_cursor: Option<&str>,
3349        transcript_limit: Option<usize>,
3350        activity_cursor: Option<&str>,
3351        activity_limit: Option<usize>,
3352    ) -> Result<RoomSnapshotResult, SarahConversationError> {
3353        self.room_snapshot_with_record_cursor(
3354            transcript_cursor,
3355            transcript_limit,
3356            activity_cursor,
3357            activity_limit,
3358            None,
3359            None,
3360        )
3361    }
3362
3363    pub fn room_snapshot_with_record_cursor(
3364        &mut self,
3365        transcript_cursor: Option<&str>,
3366        transcript_limit: Option<usize>,
3367        activity_cursor: Option<&str>,
3368        activity_limit: Option<usize>,
3369        nostr_cursor: Option<&str>,
3370        nostr_limit: Option<usize>,
3371    ) -> Result<RoomSnapshotResult, SarahConversationError> {
3372        self.ensure_connected()?;
3373        self.sync_issue31_host()?;
3374        let transcript_limit = transcript_limit
3375            .unwrap_or(DEFAULT_PAGE_LIMIT)
3376            .clamp(1, MAX_PAGE_LIMIT);
3377        let activity_limit = activity_limit
3378            .unwrap_or(DEFAULT_PAGE_LIMIT)
3379            .clamp(1, MAX_PAGE_LIMIT);
3380        let nostr_limit = nostr_limit
3381            .unwrap_or(DEFAULT_PAGE_LIMIT)
3382            .clamp(1, MAX_PAGE_LIMIT);
3383        let conversation_ref = self.config.conversation_ref();
3384        let transcript_page =
3385            self.query_with_auth(&conversation_ref, transcript_cursor, MAX_PAGE_LIMIT)?;
3386        let activity_page =
3387            self.query_with_auth(&conversation_ref, activity_cursor, MAX_PAGE_LIMIT)?;
3388        let nostr_page = self.query_with_auth(&conversation_ref, nostr_cursor, MAX_PAGE_LIMIT)?;
3389        self.last_gap_state = strongest_gap_state(
3390            strongest_gap_state(transcript_page.gap_state, activity_page.gap_state),
3391            nostr_page.gap_state,
3392        );
3393        self.last_confirmed_cursor = transcript_page
3394            .events
3395            .last()
3396            .or_else(|| activity_page.events.last())
3397            .or_else(|| nostr_page.events.last())
3398            .map(stored_event_cursor);
3399        let transcript_matching_count = transcript_page
3400            .events
3401            .iter()
3402            .filter(|event| event.record_kind == "message")
3403            .count();
3404        let transcript_entries = transcript_page
3405            .events
3406            .iter()
3407            .filter(|event| event.record_kind == "message")
3408            .take(transcript_limit)
3409            .map(|event| {
3410                let role = if event.pubkey == self.config.identity.owner_public_key_hex {
3411                    "owner"
3412                } else {
3413                    "sarah"
3414                };
3415                TranscriptEntry {
3416                    event_id: event.event_id.clone(),
3417                    cursor: stored_event_cursor(event),
3418                    role: role.to_string(),
3419                    kind: "text".to_string(),
3420                    text: event.content_summary.clone(),
3421                    created_at: iso_from_unix(event.created_at),
3422                    status: "confirmed".to_string(),
3423                }
3424            })
3425            .collect::<Vec<_>>();
3426        let activity_matching_count = activity_page
3427            .events
3428            .iter()
3429            .filter(|event| event.record_kind == "activity")
3430            .count();
3431        let activity_entries = activity_page
3432            .events
3433            .iter()
3434            .filter(|event| event.record_kind == "activity")
3435            .take(activity_limit)
3436            .map(|event| ActivityEntry {
3437                event_id: event.event_id.clone(),
3438                cursor: stored_event_cursor(event),
3439                entry: tag_value(&event.tags, "entry").unwrap_or_else(|| "unknown".into()),
3440                turn_ref: tag_value(&event.tags, "turn").unwrap_or_else(|| "turn.unknown".into()),
3441                created_at: iso_from_unix(event.created_at),
3442            })
3443            .collect::<Vec<_>>();
3444        let nostr_matching_count = nostr_page
3445            .events
3446            .iter()
3447            .filter(|event| confirmed_nostr_projection_kind(event.kind))
3448            .count();
3449        let nostr_entries = nostr_page
3450            .events
3451            .iter()
3452            .filter(|event| confirmed_nostr_projection_kind(event.kind))
3453            .take(nostr_limit)
3454            .map(|event| NostrRecordRef {
3455                event_id: event.event_id.clone(),
3456                cursor: stored_event_cursor(event),
3457                kind: event.kind,
3458                record_kind: event.record_kind.clone(),
3459                author_fingerprint: public_key_fingerprint(&event.pubkey),
3460                created_at: iso_from_unix(event.created_at),
3461                source: "confirmed_nostr".into(),
3462            })
3463            .collect::<Vec<_>>();
3464        let transcript_page_cursor = transcript_entries
3465            .last()
3466            .map(|entry| entry.cursor.clone())
3467            .unwrap_or_else(|| transcript_cursor.unwrap_or("cursor.start").to_string());
3468        let activity_page_cursor = activity_entries
3469            .last()
3470            .map(|entry| entry.cursor.clone())
3471            .unwrap_or_else(|| activity_cursor.unwrap_or("cursor.start").to_string());
3472        let nostr_page_cursor = nostr_entries
3473            .last()
3474            .map(|entry| entry.cursor.clone())
3475            .unwrap_or_else(|| nostr_cursor.unwrap_or("cursor.start").to_string());
3476        let transcript_next_cursor = stream_next_cursor(
3477            transcript_entries.last().map(|entry| entry.cursor.as_str()),
3478            transcript_page.next_cursor.as_deref(),
3479            transcript_matching_count > transcript_limit,
3480        );
3481        let activity_next_cursor = stream_next_cursor(
3482            activity_entries.last().map(|entry| entry.cursor.as_str()),
3483            activity_page.next_cursor.as_deref(),
3484            activity_matching_count > activity_limit,
3485        );
3486        let nostr_next_cursor = stream_next_cursor(
3487            nostr_entries.last().map(|entry| entry.cursor.as_str()),
3488            nostr_page.next_cursor.as_deref(),
3489            nostr_matching_count > nostr_limit,
3490        );
3491        Ok(RoomSnapshotResult {
3492            conversation_ref,
3493            transcript: TranscriptPage {
3494                entries: transcript_entries,
3495                cursor: transcript_page_cursor,
3496                next_cursor: transcript_next_cursor,
3497                gap_state: transcript_page.gap_state,
3498            },
3499            activity: ActivityPage {
3500                entries: activity_entries,
3501                cursor: activity_page_cursor,
3502                next_cursor: activity_next_cursor,
3503                gap_state: activity_page.gap_state,
3504            },
3505            nostr_records: NostrRecordPage {
3506                entries: nostr_entries,
3507                cursor: nostr_page_cursor,
3508                next_cursor: nostr_next_cursor,
3509                gap_state: nostr_page.gap_state,
3510                source: "confirmed_nostr".into(),
3511            },
3512            run_state: RunStateProjection {
3513                state: self.run_state.clone(),
3514                turn_ref: self.active_turn_ref.clone(),
3515                reason: None,
3516            },
3517            room_state: self.current_room_state(),
3518        })
3519    }
3520
3521    pub fn send_message(
3522        &mut self,
3523        text: &str,
3524        idempotency_ref: &str,
3525        expected_generation: u64,
3526    ) -> Result<SendMessageResult, SarahConversationError> {
3527        self.send_message_with_fingerprint(text, idempotency_ref, expected_generation, None)
3528    }
3529
3530    fn send_message_with_fingerprint(
3531        &mut self,
3532        text: &str,
3533        idempotency_ref: &str,
3534        expected_generation: u64,
3535        command_fingerprint: Option<String>,
3536    ) -> Result<SendMessageResult, SarahConversationError> {
3537        self.ensure_connected()?;
3538        self.ensure_generation(expected_generation)?;
3539        validate_command_ref(idempotency_ref, "idempotencyRef")?;
3540        if command_fingerprint.is_some() {
3541            self.ensure_command_result_capacity(idempotency_ref)?;
3542        }
3543        let text = text.trim();
3544        if text.is_empty() {
3545            return Err(SarahConversationError::InvalidRequest(
3546                "message text must not be empty".into(),
3547            ));
3548        }
3549        if text.len() > 8_000 {
3550            return Err(SarahConversationError::InvalidRequest(
3551                "message text exceeds page budget".into(),
3552            ));
3553        }
3554        // Refuse anything that looks like a raw credential in the outbound path.
3555        if looks_like_secret(text) {
3556            return Err(SarahConversationError::InvalidRequest(
3557                "message must not carry raw credentials".into(),
3558            ));
3559        }
3560
3561        let next_message_seq = self.message_seq.checked_add(1).ok_or_else(|| {
3562            SarahConversationError::InvalidRequest("message sequence is exhausted".into())
3563        })?;
3564        let turn_ref = format!("turn.{next_message_seq}");
3565        let message_ref = format!("msg.{next_message_seq}");
3566
3567        let conversation_ref = self.config.conversation_ref();
3568        let tags = conversation_tags(
3569            &conversation_ref,
3570            &self.config.identity.owner_public_key_hex,
3571            &self.config.identity.sarah_public_key_hex,
3572        )?;
3573        let mut tags = tags;
3574        tags.push(
3575            Tag::parse(["idempotency", idempotency_ref])
3576                .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?,
3577        );
3578        tags.push(
3579            Tag::parse(["generation", &expected_generation.to_string()])
3580                .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?,
3581        );
3582        tags.push(
3583            Tag::parse(["turn", &turn_ref])
3584                .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?,
3585        );
3586        let mut public_event = None;
3587        let mut inserted_outbox_ref = None;
3588        let (event_id, created_at) = if self.relay.requires_private_messages() {
3589            let recipients = private_recipients(
3590                &self.config.identity.owner_public_key_hex,
3591                &self.config.identity.sarah_public_key_hex,
3592            )?;
3593            let outbox_ref = private_outbox_ref("sarah.message", text, &tags, &recipients);
3594            let (rumor_event_id, inserted) =
3595                self.enqueue_private_content(&outbox_ref, text, tags, &recipients)?;
3596            if inserted {
3597                inserted_outbox_ref = Some(outbox_ref);
3598            }
3599            (rumor_event_id, unix_now())
3600        } else {
3601            let event = match &self.signer {
3602                ConversationSigner::Keys(identity) => identity.sign_text_note(text, tags)?,
3603                ConversationSigner::OmegaIdentity(_) => {
3604                    return Err(SarahConversationError::Internal(
3605                        "custodied production signer requires NIP-17 transport".into(),
3606                    ));
3607                }
3608            };
3609            let event_id = event.id.to_hex();
3610            let created_at = event.created_at.as_secs();
3611            public_event = Some(event);
3612            (event_id, created_at)
3613        };
3614        let cursor = format!("{CURSOR_PREFIX}{}", next_message_seq.saturating_sub(1));
3615        let response = SendMessageResult {
3616            accepted: true,
3617            message_ref,
3618            turn_ref: turn_ref.clone(),
3619            event_id: event_id.clone(),
3620            cursor: cursor.clone(),
3621            status: "accepted".to_string(),
3622        };
3623        let response_value = command_fingerprint
3624            .as_ref()
3625            .map(|_| serde_json::to_value(&response))
3626            .transpose()
3627            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
3628        let previous_active_turn_ref = self.active_turn_ref.clone();
3629        let previous_run_state = self.run_state.clone();
3630        let previous_message_seq = self.message_seq;
3631        self.message_seq = next_message_seq;
3632        self.active_turn_ref = Some(turn_ref);
3633        self.run_state = "running".to_string();
3634        if let (Some(fingerprint), Some(response_value)) =
3635            (command_fingerprint.as_ref(), response_value.as_ref())
3636        {
3637            self.command_results.insert(
3638                idempotency_ref.to_string(),
3639                (fingerprint.clone(), response_value.clone()),
3640            );
3641        }
3642        if let Err(error) = self.persist_issue31_host_state() {
3643            self.message_seq = previous_message_seq;
3644            self.active_turn_ref = previous_active_turn_ref;
3645            self.run_state = previous_run_state;
3646            if command_fingerprint.is_some() {
3647                self.command_results.remove(idempotency_ref);
3648            }
3649            if let Some(outbox_ref) = inserted_outbox_ref {
3650                self.issue31_private_outbox.remove(&outbox_ref);
3651            }
3652            return Err(error);
3653        }
3654        let publication = if let Some(event) = public_event.as_ref() {
3655            self.publish_with_auth(event)
3656        } else {
3657            self.flush_issue31_outbox()
3658        };
3659        let persistence = self.persist_issue31_host_state();
3660        finish_durable_operation(publication, persistence)?;
3661        let record = TranscriptEntry {
3662            event_id,
3663            cursor,
3664            role: "owner".to_string(),
3665            kind: "text".to_string(),
3666            text: redact_content_summary(text),
3667            created_at: iso_from_unix(created_at),
3668            status: "accepted".to_string(),
3669        };
3670        self.push_room_event(&record);
3671        self.push_room_state_event(&self.current_room_state());
3672        Ok(response)
3673    }
3674
3675    pub fn interrupt_turn(
3676        &mut self,
3677        turn_ref: &str,
3678        idempotency_ref: &str,
3679        expected_generation: u64,
3680    ) -> Result<InterruptTurnResult, SarahConversationError> {
3681        self.interrupt_turn_with_fingerprint(turn_ref, idempotency_ref, expected_generation, None)
3682    }
3683
3684    fn interrupt_turn_with_fingerprint(
3685        &mut self,
3686        turn_ref: &str,
3687        idempotency_ref: &str,
3688        expected_generation: u64,
3689        command_fingerprint: Option<String>,
3690    ) -> Result<InterruptTurnResult, SarahConversationError> {
3691        self.ensure_connected()?;
3692        self.ensure_generation(expected_generation)?;
3693        validate_command_ref(idempotency_ref, "idempotencyRef")?;
3694        if command_fingerprint.is_some() {
3695            self.ensure_command_result_capacity(idempotency_ref)?;
3696        }
3697        if turn_ref.trim().is_empty() {
3698            return Err(SarahConversationError::InvalidRequest(
3699                "turnRef must not be empty".into(),
3700            ));
3701        }
3702        let intent_ref = format!(
3703            "intent.interrupt.{}",
3704            &format!("{:x}", Sha256::digest(idempotency_ref.as_bytes()))[..24]
3705        );
3706        let conversation_ref = self.config.conversation_ref();
3707        let mut tags = conversation_tags(
3708            &conversation_ref,
3709            &self.config.identity.owner_public_key_hex,
3710            &self.config.identity.sarah_public_key_hex,
3711        )?;
3712        tags.push(
3713            Tag::parse(["turn", turn_ref])
3714                .map_err(|error| SarahConversationError::Internal(error.to_string()))?,
3715        );
3716        tags.push(
3717            Tag::parse(["control", "cancel_turn"])
3718                .map_err(|error| SarahConversationError::Internal(error.to_string()))?,
3719        );
3720        tags.push(
3721            Tag::parse(["idempotency", idempotency_ref])
3722                .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?,
3723        );
3724        tags.push(
3725            Tag::parse(["generation", &expected_generation.to_string()])
3726                .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?,
3727        );
3728
3729        let content = json!({
3730            "schema": "openagents.sarah.control.v1",
3731            "control": "cancel_turn",
3732            "turnRef": turn_ref,
3733            "intentRef": intent_ref,
3734            "idempotencyRef": idempotency_ref,
3735            "expectedGeneration": expected_generation,
3736        })
3737        .to_string();
3738
3739        let mut public_event = None;
3740        let mut inserted_outbox_ref = None;
3741        if self.relay.requires_private_messages() {
3742            let recipients = private_recipients(
3743                &self.config.identity.owner_public_key_hex,
3744                &self.config.identity.sarah_public_key_hex,
3745            )?;
3746            let outbox_ref = private_outbox_ref("sarah.interrupt", &content, &tags, &recipients);
3747            let (_, inserted) =
3748                self.enqueue_private_content(&outbox_ref, &content, tags, &recipients)?;
3749            if inserted {
3750                inserted_outbox_ref = Some(outbox_ref);
3751            }
3752        } else {
3753            let event = match &self.signer {
3754                ConversationSigner::Keys(identity) => {
3755                    identity.sign_custom(NIP_AO_KIND, &content, tags)?
3756                }
3757                ConversationSigner::OmegaIdentity(_) => {
3758                    return Err(SarahConversationError::Internal(
3759                        "custodied production signer requires NIP-17 transport".into(),
3760                    ));
3761                }
3762            };
3763            public_event = Some(event);
3764        }
3765        let response = InterruptTurnResult {
3766            accepted: true,
3767            turn_ref: turn_ref.to_string(),
3768            intent_ref,
3769            status: "pending".to_string(),
3770            pending: true,
3771        };
3772        let response_value = command_fingerprint
3773            .as_ref()
3774            .map(|_| serde_json::to_value(&response))
3775            .transpose()
3776            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
3777        let previous_run_state = self.run_state.clone();
3778        self.run_state = "interrupt_pending".to_string();
3779        if let (Some(fingerprint), Some(response_value)) =
3780            (command_fingerprint.as_ref(), response_value.as_ref())
3781        {
3782            self.command_results.insert(
3783                idempotency_ref.to_string(),
3784                (fingerprint.clone(), response_value.clone()),
3785            );
3786        }
3787        if let Err(error) = self.persist_issue31_host_state() {
3788            self.run_state = previous_run_state;
3789            if command_fingerprint.is_some() {
3790                self.command_results.remove(idempotency_ref);
3791            }
3792            if let Some(outbox_ref) = inserted_outbox_ref {
3793                self.issue31_private_outbox.remove(&outbox_ref);
3794            }
3795            return Err(error);
3796        }
3797        let publication = if let Some(event) = public_event.as_ref() {
3798            self.publish_with_auth(event)
3799        } else {
3800            self.flush_issue31_outbox()
3801        };
3802        let persistence = self.persist_issue31_host_state();
3803        finish_durable_operation(publication, persistence)?;
3804        self.push_room_state_event(&self.current_room_state());
3805        Ok(response)
3806    }
3807
3808    /// Drain pending framed events (`sarah_room_event` / `sarah_room_state`).
3809    pub fn drain_events(&mut self) -> Vec<Value> {
3810        self.pending_events.drain(..).collect()
3811    }
3812
3813    fn cached_command(
3814        &self,
3815        idempotency_ref: &str,
3816        fingerprint: &str,
3817    ) -> Result<Option<Value>, SarahConversationError> {
3818        match self.command_results.get(idempotency_ref) {
3819            Some((existing_fingerprint, result)) if existing_fingerprint == fingerprint => {
3820                Ok(Some(result.clone()))
3821            }
3822            Some(_) => Err(SarahConversationError::InvalidRequest(format!(
3823                "idempotencyRef {idempotency_ref} conflicts with an earlier command"
3824            ))),
3825            None => Ok(None),
3826        }
3827    }
3828
3829    fn ensure_command_result_capacity(
3830        &self,
3831        idempotency_ref: &str,
3832    ) -> Result<(), SarahConversationError> {
3833        if !self.command_results.contains_key(idempotency_ref)
3834            && self.command_results.len() >= MAX_COMMAND_RESULTS
3835        {
3836            return Err(SarahConversationError::InvalidRequest(
3837                "durable command result bound is exhausted".into(),
3838            ));
3839        }
3840        Ok(())
3841    }
3842
3843    fn retry_durable_outbox(&mut self) -> Result<(), SarahConversationError> {
3844        if self.issue31_discovery_outbox.is_none() && self.issue31_private_outbox.is_empty() {
3845            return Ok(());
3846        }
3847        self.ensure_connected()?;
3848        let publication = self.flush_issue31_outbox();
3849        let persistence = self.persist_issue31_host_state();
3850        finish_durable_operation(publication, persistence)
3851    }
3852
3853    pub fn encode_response_frame(
3854        &self,
3855        id: impl Into<String>,
3856        generation: u64,
3857        result: Value,
3858    ) -> Result<String, SarahConversationError> {
3859        let frame = json!({
3860            "schema": PROTOCOL_SCHEMA,
3861            "kind": "response",
3862            "id": id.into(),
3863            "generation": generation,
3864            "ok": true,
3865            "result": result,
3866        });
3867        let line = serde_json::to_string(&frame)
3868            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
3869        if line.len() > MAX_FRAME_BYTES {
3870            return Err(SarahConversationError::Internal(
3871                "response frame exceeds 64 KiB cap".into(),
3872            ));
3873        }
3874        Ok(line)
3875    }
3876
3877    pub fn encode_event_frame(
3878        &self,
3879        generation: u64,
3880        method: &str,
3881        payload: Value,
3882    ) -> Result<String, SarahConversationError> {
3883        let frame = json!({
3884            "schema": PROTOCOL_SCHEMA,
3885            "kind": "event",
3886            "generation": generation,
3887            "method": method,
3888            "params": payload,
3889        });
3890        let line = serde_json::to_string(&frame)
3891            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
3892        if line.len() > MAX_FRAME_BYTES {
3893            return Err(SarahConversationError::Internal(
3894                "event frame exceeds 64 KiB cap".into(),
3895            ));
3896        }
3897        // Hard redaction: never allow token-shaped payloads into events.
3898        if line.contains("bearer ") || line.contains("sk-") {
3899            return Err(SarahConversationError::Internal(
3900                "event frame refused: secret-shaped content".into(),
3901            ));
3902        }
3903        Ok(line)
3904    }
3905
3906    fn ensure_generation(&self, generation: u64) -> Result<(), SarahConversationError> {
3907        if generation != self.config.generation {
3908            return Err(SarahConversationError::StaleGeneration {
3909                expected: self.config.generation,
3910                got: generation,
3911            });
3912        }
3913        Ok(())
3914    }
3915
3916    fn ensure_connected(&mut self) -> Result<(), SarahConversationError> {
3917        if self.relay.connection_state() == ConnectionState::Disconnected {
3918            self.relay.connect()?;
3919        }
3920        if let Some(challenge) = self.relay.auth_challenge() {
3921            let auth_event = self
3922                .signer
3923                .sign_auth(&challenge.challenge, &challenge.relay_url)?;
3924            self.relay.authenticate(&auth_event)?;
3925        }
3926        Ok(())
3927    }
3928
3929    fn authenticate_pending(&mut self) -> Result<(), SarahConversationError> {
3930        let challenge = self
3931            .relay
3932            .auth_challenge()
3933            .ok_or(SarahConversationError::IdentityRequired)?;
3934        let auth_event = self
3935            .signer
3936            .sign_auth(&challenge.challenge, &challenge.relay_url)?;
3937        self.relay.authenticate(&auth_event)
3938    }
3939
3940    fn publish_with_auth(&mut self, event: &Event) -> Result<(), SarahConversationError> {
3941        match self.relay.publish(event) {
3942            Ok(()) => Ok(()),
3943            Err(SarahConversationError::IdentityRequired) => {
3944                self.authenticate_pending()?;
3945                self.relay.publish(event)
3946            }
3947            Err(error) => Err(error),
3948        }
3949    }
3950
3951    fn query_with_auth(
3952        &mut self,
3953        conversation_ref: &str,
3954        cursor: Option<&str>,
3955        limit: usize,
3956    ) -> Result<QueryPage, SarahConversationError> {
3957        match self.relay.query(conversation_ref, cursor, limit) {
3958            Ok(page) => Ok(page),
3959            Err(SarahConversationError::IdentityRequired) => {
3960                self.authenticate_pending()?;
3961                self.relay.query(conversation_ref, cursor, limit)
3962            }
3963            Err(error) => Err(error),
3964        }
3965    }
3966
3967    fn current_room_state(&self) -> RoomStateEvent {
3968        let last_id = self.relay.last_event_id();
3969        let gap_state = strongest_gap_state(self.last_gap_state, self.relay.gap_state());
3970        RoomStateEvent {
3971            method: SARAH_EVENT_ROOM_STATE.to_string(),
3972            connection: self.relay.connection_state(),
3973            freshness: if self.relay.connection_state() == ConnectionState::Connected
3974                && gap_state == GapState::None
3975            {
3976                FreshnessState::Fresh
3977            } else {
3978                FreshnessState::Unknown
3979            },
3980            gap_state,
3981            connected_relays: self.relay.connected_relays(),
3982            last_acknowledged_event_id: last_id,
3983            last_acknowledged_cursor: self.last_confirmed_cursor.clone(),
3984            authenticated: self.relay.is_authenticated(),
3985            transport: self.transport_label(),
3986        }
3987    }
3988
3989    fn transport_label(&self) -> String {
3990        if self.config.relay_url.is_some() {
3991            "nostr_relay".to_string()
3992        } else {
3993            "mock_relay".to_string()
3994        }
3995    }
3996
3997    fn push_room_event(&mut self, record: &TranscriptEntry) {
3998        let payload = RoomEventPayload {
3999            method: SARAH_EVENT_ROOM_EVENT.to_string(),
4000            conversation_ref: self.config.conversation_ref(),
4001            cursor: record.cursor.clone(),
4002            record: record.clone(),
4003        };
4004        if let Ok(value) = serde_json::to_value(payload) {
4005            if self.pending_events.len() >= MAX_PENDING_EVENTS {
4006                self.pending_events.pop_front();
4007            }
4008            self.pending_events.push_back(value);
4009        }
4010    }
4011
4012    fn push_room_state_event(&mut self, state: &RoomStateEvent) {
4013        if let Ok(value) = serde_json::to_value(state) {
4014            if self.pending_events.len() >= MAX_PENDING_EVENTS {
4015                self.pending_events.pop_front();
4016            }
4017            self.pending_events.push_back(value);
4018        }
4019    }
4020}
4021
4022fn finish_durable_operation<T>(
4023    operation: Result<T, SarahConversationError>,
4024    persistence: Result<(), SarahConversationError>,
4025) -> Result<T, SarahConversationError> {
4026    match (operation, persistence) {
4027        (Ok(value), Ok(())) => Ok(value),
4028        (Err(error), Ok(())) => Err(error),
4029        (Ok(_), Err(error)) => Err(error),
4030        (Err(operation_error), Err(persistence_error)) => Err(SarahConversationError::Internal(
4031            format!("{operation_error}; durable Issue 31 commit also failed: {persistence_error}"),
4032        )),
4033    }
4034}
4035
4036fn durable_private_outbox_into_runtime(
4037    durable: BTreeMap<String, DurableIssue31PrivatePublish>,
4038) -> Result<BTreeMap<String, PendingIssue31PrivatePublish>, SarahConversationError> {
4039    if durable.len() > MAX_PRIVATE_OUTBOX_ITEMS {
4040        return Err(SarahConversationError::Internal(
4041            "durable Issue 31 outbox exceeds its item bound".into(),
4042        ));
4043    }
4044    durable
4045        .into_iter()
4046        .map(|(outbox_ref, pending)| {
4047            if outbox_ref.len() > 192
4048                || !is_lower_hex_64(&pending.rumor_event_id)
4049                || pending.gift_wrap_event_json.is_empty()
4050                || pending.gift_wrap_event_json.len() > 8
4051            {
4052                return Err(SarahConversationError::Internal(
4053                    "durable Issue 31 outbox item is invalid".into(),
4054                ));
4055            }
4056            let gift_wraps = pending
4057                .gift_wrap_event_json
4058                .into_iter()
4059                .map(|event_json| {
4060                    let event = Event::from_json(event_json)
4061                        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4062                    if event.kind != Kind::GiftWrap || event.verify().is_err() {
4063                        return Err(SarahConversationError::Internal(
4064                            "durable Issue 31 outbox has an invalid gift wrap".into(),
4065                        ));
4066                    }
4067                    Ok(event)
4068                })
4069                .collect::<Result<Vec<_>, _>>()?;
4070            Ok((
4071                outbox_ref,
4072                PendingIssue31PrivatePublish {
4073                    rumor_event_id: pending.rumor_event_id,
4074                    gift_wraps,
4075                },
4076            ))
4077        })
4078        .collect()
4079}
4080
4081fn load_issue31_host_state(
4082    path: &Path,
4083    expected_configuration: &Issue31HostConfiguration,
4084) -> Result<Option<DurableIssue31HostState>, SarahConversationError> {
4085    let metadata = match fs::symlink_metadata(path) {
4086        Ok(metadata) => metadata,
4087        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
4088        Err(error) => return Err(SarahConversationError::Internal(error.to_string())),
4089    };
4090    if !metadata.file_type().is_file()
4091        || metadata.file_type().is_symlink()
4092        || metadata.len() > ISSUE31_DURABLE_STATE_MAX_BYTES
4093    {
4094        return Err(SarahConversationError::Internal(
4095            "durable Issue 31 host state is unsafe or oversized".into(),
4096        ));
4097    }
4098    let file = OpenOptions::new()
4099        .read(true)
4100        .open(path)
4101        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4102    let mut bytes = Vec::with_capacity(metadata.len() as usize);
4103    file.take(ISSUE31_DURABLE_STATE_MAX_BYTES.saturating_add(1))
4104        .read_to_end(&mut bytes)
4105        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4106    if bytes.len() as u64 > ISSUE31_DURABLE_STATE_MAX_BYTES {
4107        return Err(SarahConversationError::Internal(
4108            "durable Issue 31 host state exceeds its byte bound".into(),
4109        ));
4110    }
4111    let mut state: DurableIssue31HostState = serde_json::from_slice(&bytes)
4112        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4113    state
4114        .controller
4115        .adopt_conversation_if_missing(&expected_configuration.conversation)
4116        .map_err(issue31_error)?;
4117    if state.schema != ISSUE31_DURABLE_STATE_SCHEMA
4118        || !state
4119            .controller
4120            .matches_configuration(expected_configuration)
4121        || state.discovery_generation.is_some() != state.discovery_expires_at.is_some()
4122        || state.command_results.len() > MAX_COMMAND_RESULTS
4123        || state.relay_acknowledgements.len() > MAX_RELAY_ACKNOWLEDGEMENTS
4124        || state.quarantined_events.len() > MAX_QUARANTINED_ISSUE31_EVENTS
4125        || state.host_adjunct_emissions.len() > MAX_QUARANTINED_ISSUE31_EVENTS
4126        || state
4127            .control_cursor
4128            .as_deref()
4129            .is_some_and(|cursor| !valid_event_cursor(cursor))
4130        || state
4131            .projection_cursor
4132            .as_deref()
4133            .is_some_and(|cursor| !valid_event_cursor(cursor))
4134        || state
4135            .active_turn_ref
4136            .as_deref()
4137            .is_some_and(|turn_ref| !crate::is_issue31_public_ref(turn_ref))
4138        || !valid_run_state(&state.run_state)
4139    {
4140        return Err(SarahConversationError::Internal(
4141            "durable Issue 31 host state does not match this identity or relay configuration"
4142                .into(),
4143        ));
4144    }
4145    state
4146        .controller
4147        .validate_persisted_state()
4148        .map_err(issue31_error)?;
4149    if state
4150        .quarantined_events
4151        .iter()
4152        .any(|(event_id, reason_ref)| {
4153            !is_lower_hex_64(event_id) || !crate::is_issue31_public_ref(reason_ref)
4154        })
4155    {
4156        return Err(SarahConversationError::Internal(
4157            "durable Issue 31 quarantine contains an invalid record".into(),
4158        ));
4159    }
4160    if state
4161        .relay_acknowledgements
4162        .iter()
4163        .any(|(event_id, relay_urls)| {
4164            !is_lower_hex_64(event_id)
4165                || relay_urls.len() > expected_configuration.relay_urls.len()
4166                || relay_urls.iter().enumerate().any(|(index, relay_url)| {
4167                    relay_urls[..index].iter().any(|prior| prior == relay_url)
4168                })
4169                || relay_urls
4170                    .iter()
4171                    .any(|relay_url| !expected_configuration.relay_urls.contains(relay_url))
4172        })
4173    {
4174        return Err(SarahConversationError::Internal(
4175            "durable Issue 31 relay acknowledgements are invalid".into(),
4176        ));
4177    }
4178    if let Some(event_json) = state.discovery_event_json.clone() {
4179        let event = Event::from_json(&event_json)
4180            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4181        let value = serde_json::from_str::<Value>(&event.content)
4182            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4183        let is_v2 = value.get("schema").and_then(Value::as_str)
4184            == Some(crate::ISSUE31_HOST_DISCOVERY_SCHEMA_V2);
4185        let (generation, expires_at, host_ref, host_key, sarah_key, display_name, relay_urls) =
4186            if is_v2 {
4187                let discovery = Issue31HostDiscoveryV2::decode(event.content.as_bytes())
4188                    .map_err(issue31_error)?;
4189                if discovery.conversation != expected_configuration.conversation {
4190                    return Err(SarahConversationError::Internal(
4191                        "durable Issue 31 discovery binds another conversation".into(),
4192                    ));
4193                }
4194                (
4195                    discovery.generation,
4196                    discovery.expires_at,
4197                    discovery.host_ref,
4198                    discovery.host_public_key_hex,
4199                    discovery.sarah_public_key_hex,
4200                    discovery.display_name,
4201                    discovery.relay_urls,
4202                )
4203            } else {
4204                let discovery = Issue31HostDiscovery::decode(event.content.as_bytes())
4205                    .map_err(issue31_error)?;
4206                (
4207                    discovery.generation,
4208                    discovery.expires_at,
4209                    discovery.host_ref,
4210                    discovery.host_public_key_hex,
4211                    discovery.sarah_public_key_hex,
4212                    discovery.display_name,
4213                    discovery.relay_urls,
4214                )
4215            };
4216        if event.kind.as_u16() != ISSUE31_HOST_DISCOVERY_KIND
4217            || event.pubkey.to_hex() != expected_configuration.host_public_key_hex
4218            || event.verify().is_err()
4219            || state.discovery_generation != Some(generation)
4220            || state.discovery_expires_at != Some(expires_at)
4221            || host_ref != expected_configuration.host_ref
4222            || host_key != expected_configuration.host_public_key_hex
4223            || sarah_key != expected_configuration.sarah_public_key_hex
4224            || display_name != expected_configuration.display_name
4225            || relay_urls != expected_configuration.relay_urls
4226            || generation != expected_configuration.generation
4227        {
4228            return Err(SarahConversationError::Internal(
4229                "durable Issue 31 discovery outbox event is invalid".into(),
4230            ));
4231        }
4232        if !is_v2 {
4233            state.discovery_generation = None;
4234            state.discovery_expires_at = None;
4235            state.discovery_event_json = None;
4236            state.relay_acknowledgements.remove(&event.id.to_hex());
4237        }
4238    }
4239    durable_private_outbox_into_runtime(
4240        state
4241            .private_outbox
4242            .iter()
4243            .map(|(key, value)| {
4244                (
4245                    key.clone(),
4246                    DurableIssue31PrivatePublish {
4247                        rumor_event_id: value.rumor_event_id.clone(),
4248                        gift_wrap_event_json: value.gift_wrap_event_json.clone(),
4249                    },
4250                )
4251            })
4252            .collect(),
4253    )?;
4254    Ok(Some(state))
4255}
4256
4257fn write_issue31_host_state(
4258    path: &Path,
4259    state: &DurableIssue31HostState,
4260) -> Result<(), SarahConversationError> {
4261    let bytes = serde_json::to_vec(state)
4262        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4263    if bytes.len() as u64 > ISSUE31_DURABLE_STATE_MAX_BYTES {
4264        return Err(SarahConversationError::Internal(
4265            "durable Issue 31 host state exceeds its byte bound".into(),
4266        ));
4267    }
4268    let parent = path.parent().ok_or_else(|| {
4269        SarahConversationError::Internal("Issue 31 state path has no parent".into())
4270    })?;
4271    fs::create_dir_all(parent)
4272        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4273    #[cfg(unix)]
4274    {
4275        use std::os::unix::fs::PermissionsExt as _;
4276        fs::set_permissions(parent, fs::Permissions::from_mode(0o700))
4277            .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4278    }
4279    let file_name = path
4280        .file_name()
4281        .and_then(|name| name.to_str())
4282        .ok_or_else(|| {
4283            SarahConversationError::Internal("Issue 31 state filename is invalid".into())
4284        })?;
4285    let temporary_path = parent.join(format!(".{file_name}.{}.tmp", rand::random::<u64>()));
4286    let mut options = OpenOptions::new();
4287    options.write(true).create_new(true);
4288    #[cfg(unix)]
4289    {
4290        use std::os::unix::fs::OpenOptionsExt as _;
4291        options.mode(0o600);
4292    }
4293    let mut file = options
4294        .open(&temporary_path)
4295        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4296    let write_result = file
4297        .write_all(&bytes)
4298        .and_then(|()| file.sync_all())
4299        .and_then(|()| fs::rename(&temporary_path, path));
4300    if let Err(error) = write_result {
4301        return match fs::remove_file(&temporary_path) {
4302            Ok(()) => Err(SarahConversationError::Internal(error.to_string())),
4303            Err(cleanup_error) if cleanup_error.kind() == std::io::ErrorKind::NotFound => {
4304                Err(SarahConversationError::Internal(error.to_string()))
4305            }
4306            Err(cleanup_error) => Err(SarahConversationError::Internal(format!(
4307                "{error}; temporary state cleanup failed: {cleanup_error}"
4308            ))),
4309        };
4310    }
4311    Ok(())
4312}
4313
4314fn is_lower_hex_64(value: &str) -> bool {
4315    value.len() == 64
4316        && value
4317            .bytes()
4318            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
4319}
4320
4321fn valid_event_cursor(value: &str) -> bool {
4322    let mut parts = value.split('.');
4323    matches!(parts.next(), Some("cursor"))
4324        && parts
4325            .next()
4326            .is_some_and(|created_at| created_at.parse::<u64>().is_ok())
4327        && parts.next().is_some_and(is_lower_hex_64)
4328        && parts.next().is_none()
4329}
4330
4331fn default_run_state() -> String {
4332    "idle".to_string()
4333}
4334
4335fn confirmed_nostr_projection_kind(kind: u16) -> bool {
4336    matches!(kind, 9 | 5_934 | 6_934 | 7_000 | 30_078 | 30_174 | 30_300)
4337}
4338
4339fn public_key_fingerprint(public_key_hex: &str) -> String {
4340    let digest = format!("{:x}", Sha256::digest(public_key_hex.as_bytes()));
4341    digest[..16].to_ascii_uppercase()
4342}
4343
4344fn valid_run_state(value: &str) -> bool {
4345    matches!(
4346        value,
4347        "idle" | "running" | "interrupt_pending" | "stopped" | "paused"
4348    )
4349}
4350
4351fn private_outbox_ref(prefix: &str, content: &str, tags: &[Tag], recipients: &[String]) -> String {
4352    let mut binding = Vec::with_capacity(
4353        prefix.len()
4354            + content.len()
4355            + tags
4356                .iter()
4357                .flat_map(|tag| tag.as_slice())
4358                .map(String::len)
4359                .sum::<usize>()
4360            + recipients.iter().map(String::len).sum::<usize>()
4361            + tags.len()
4362            + recipients.len(),
4363    );
4364    binding.extend_from_slice(prefix.as_bytes());
4365    binding.push(0);
4366    binding.extend_from_slice(content.as_bytes());
4367    for tag in tags {
4368        binding.push(0xff);
4369        for value in tag.as_slice() {
4370            binding.push(0);
4371            binding.extend_from_slice(value.as_bytes());
4372        }
4373    }
4374    for recipient in recipients {
4375        binding.push(0);
4376        binding.extend_from_slice(recipient.as_bytes());
4377    }
4378    format!("{prefix}.{:x}", Sha256::digest(binding))
4379}
4380
4381pub(crate) fn conversation_tags(
4382    conversation_ref: &str,
4383    owner_pubkey: &str,
4384    sarah_pubkey: &str,
4385) -> Result<Vec<Tag>, SarahConversationError> {
4386    if owner_pubkey == sarah_pubkey {
4387        return Err(SarahConversationError::InvalidRequest(
4388            "owner and Sarah identities must be distinct".into(),
4389        ));
4390    }
4391    [
4392        ["conversation", conversation_ref],
4393        ["p", owner_pubkey],
4394        ["p", sarah_pubkey],
4395        ["agent", sarah_pubkey],
4396        ["alt", "OpenAgents Sarah conversation message"],
4397    ]
4398    .into_iter()
4399    .map(|tag| {
4400        Tag::parse(tag).map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))
4401    })
4402    .collect()
4403}
4404
4405fn stored_tag_value(tags: &[Vec<String>], name: &str) -> Option<String> {
4406    tags.iter().find_map(|tag| {
4407        (tag.first().map(String::as_str) == Some(name))
4408            .then(|| tag.get(1).cloned())
4409            .flatten()
4410    })
4411}
4412
4413fn private_recipients(
4414    owner_pubkey: &str,
4415    sarah_pubkey: &str,
4416) -> Result<Vec<String>, SarahConversationError> {
4417    if owner_pubkey == sarah_pubkey {
4418        return Err(SarahConversationError::InvalidRequest(
4419            "owner and Sarah identities must be distinct".into(),
4420        ));
4421    }
4422    PublicKey::from_hex(owner_pubkey)
4423        .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
4424    PublicKey::from_hex(sarah_pubkey)
4425        .map_err(|error| SarahConversationError::InvalidRequest(error.to_string()))?;
4426    Ok(vec![owner_pubkey.to_string(), sarah_pubkey.to_string()])
4427}
4428
4429fn digest_receipt_ref(
4430    prefix: &str,
4431    semantic_binding: &[u8],
4432) -> Result<ReceiptRef, SarahConversationError> {
4433    let digest = format!("{:x}", Sha256::digest(semantic_binding));
4434    ReceiptRef::new(format!("{prefix}.{}", &digest[..32]))
4435        .map_err(|error| SarahConversationError::Identity(error.to_string()))
4436}
4437
4438fn issue31_error(error: Issue31NostrError) -> SarahConversationError {
4439    SarahConversationError::InvalidRequest(error.to_string())
4440}
4441
4442fn strongest_gap_state(left: GapState, right: GapState) -> GapState {
4443    let severity = |state| match state {
4444        GapState::None => 0,
4445        GapState::Possible => 1,
4446        GapState::Recovering => 2,
4447        GapState::Confirmed => 3,
4448    };
4449    if severity(left) >= severity(right) {
4450        left
4451    } else {
4452        right
4453    }
4454}
4455
4456fn command_binding(
4457    params: Option<&Value>,
4458    framed_generation: u64,
4459) -> Result<(String, u64), SarahConversationError> {
4460    let idempotency_ref = params
4461        .and_then(|value| value.get("idempotencyRef"))
4462        .and_then(Value::as_str)
4463        .ok_or_else(|| {
4464            SarahConversationError::InvalidRequest("host command requires idempotencyRef".into())
4465        })?;
4466    validate_command_ref(idempotency_ref, "idempotencyRef")?;
4467    let expected_generation = params
4468        .and_then(|value| value.get("expectedGeneration"))
4469        .and_then(Value::as_u64)
4470        .ok_or_else(|| {
4471            SarahConversationError::InvalidRequest(
4472                "host command requires expectedGeneration".into(),
4473            )
4474        })?;
4475    if expected_generation != framed_generation {
4476        return Err(SarahConversationError::StaleGeneration {
4477            expected: framed_generation,
4478            got: expected_generation,
4479        });
4480    }
4481    Ok((idempotency_ref.to_string(), expected_generation))
4482}
4483
4484fn validate_command_ref(value: &str, field: &str) -> Result<(), SarahConversationError> {
4485    if !crate::is_issue31_public_ref(value) {
4486        return Err(SarahConversationError::InvalidRequest(format!(
4487            "{field} must match the bounded Issue 31 PublicRef grammar"
4488        )));
4489    }
4490    Ok(())
4491}
4492
4493fn command_fingerprint(
4494    method: &str,
4495    params: Option<&Value>,
4496) -> Result<String, SarahConversationError> {
4497    let canonical = serde_json::to_vec(&json!({ "method": method, "params": params }))
4498        .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
4499    Ok(format!("{:x}", Sha256::digest(canonical)))
4500}
4501
4502fn tag_value(tags: &[Vec<String>], name: &str) -> Option<String> {
4503    tags.iter().find_map(|tag| {
4504        if tag.first().map(String::as_str) == Some(name) {
4505            tag.get(1).cloned()
4506        } else {
4507            None
4508        }
4509    })
4510}
4511
4512fn stored_event_cursor(event: &StoredConversationEvent) -> String {
4513    format!("{CURSOR_PREFIX}{}.{}", event.created_at, event.event_id)
4514}
4515
4516fn stream_next_cursor(
4517    last_stream_cursor: Option<&str>,
4518    upstream_next_cursor: Option<&str>,
4519    more_stream_entries_in_page: bool,
4520) -> Option<String> {
4521    if !more_stream_entries_in_page && upstream_next_cursor.is_none() {
4522        return None;
4523    }
4524    last_stream_cursor
4525        .or(upstream_next_cursor)
4526        .map(str::to_owned)
4527}
4528
4529fn unix_now() -> u64 {
4530    SystemTime::now()
4531        .duration_since(UNIX_EPOCH)
4532        .map(|duration| duration.as_secs())
4533        .unwrap_or(0)
4534}
4535
4536fn iso_from_unix(seconds: u64) -> String {
4537    // Keep dependency-free timestamps for public-safe projections.
4538    format!("{seconds}")
4539}
4540
4541fn redact_content_summary(content: &str) -> String {
4542    if looks_like_secret(content) {
4543        return "[redacted]".to_string();
4544    }
4545    const MAX: usize = 512;
4546    if content.len() <= MAX {
4547        content.to_string()
4548    } else {
4549        let mut boundary = MAX;
4550        while !content.is_char_boundary(boundary) {
4551            boundary -= 1;
4552        }
4553        format!("{}…", &content[..boundary])
4554    }
4555}
4556
4557fn looks_like_secret(text: &str) -> bool {
4558    let lower = text.to_ascii_lowercase();
4559    lower.contains("bearer ")
4560        || lower.contains("authorization:")
4561        || lower.contains("api_key")
4562        || lower.contains("sk-")
4563        || lower.contains("-----begin ")
4564}
4565
4566/// Assert the crate graph does not pull a Khala Sync client for this lane.
4567pub fn asserts_no_khala_sync_client() -> bool {
4568    // Compile-time documentation of the cut OMEGA-SW-02 rule. The Sarah lane
4569    // must remain Nostr-only; this helper is used by unit tests as a living
4570    // guardrail comment surface.
4571    true
4572}
4573
4574#[cfg(test)]
4575mod tests {
4576    use std::sync::Arc;
4577    use std::sync::atomic::{AtomicUsize, Ordering};
4578
4579    use super::*;
4580
4581    fn client() -> SarahConversationClient {
4582        SarahConversationClient::new_mock(SarahConversationConfig::mock_fixture())
4583    }
4584
4585    #[test]
4586    fn framed_method_names_match_spec_section_8() {
4587        assert_eq!(
4588            SARAH_FRAMED_METHODS,
4589            &[
4590                "sarah_session_status",
4591                "sarah_bootstrap",
4592                "sarah_room_snapshot",
4593                "sarah_send_message",
4594                "sarah_interrupt_turn",
4595                "sarah_device_grants",
4596                "sarah_renew_device_grant",
4597                "sarah_revoke_device_grant",
4598                "sarah_readmit_device",
4599            ]
4600        );
4601        assert_eq!(SARAH_EVENT_ROOM_EVENT, "sarah_room_event");
4602        assert_eq!(SARAH_EVENT_ROOM_STATE, "sarah_room_state");
4603    }
4604
4605    #[test]
4606    fn session_status_never_returns_token_fields() {
4607        let mut client = client();
4608        let status = client.session_status().expect("status");
4609        let encoded = serde_json::to_string(&status).expect("json");
4610        assert!(status.signed_in);
4611        assert_eq!(status.account_label.as_deref(), Some("owner@example.com"));
4612        assert!(!encoded.contains("token"));
4613        assert!(!encoded.contains("bearer"));
4614        assert!(!encoded.contains("secret"));
4615    }
4616
4617    #[test]
4618    fn bootstrap_projects_conversation_and_room_state() {
4619        let mut client = client();
4620        let boot = client.bootstrap().expect("bootstrap");
4621        assert_eq!(boot.principal_ref, "principal.sarah");
4622        assert!(boot.conversation_ref.starts_with("sarah."));
4623        assert!(boot.legacy_thread_ref.starts_with("thread.sarah."));
4624        assert_eq!(boot.room_state.connection, ConnectionState::Connected);
4625        assert_eq!(boot.room_state.transport, "mock_relay");
4626        assert!(boot.room_state.authenticated);
4627        let events = client.drain_events();
4628        assert!(
4629            events.iter().any(|event| {
4630                event.get("method").and_then(Value::as_str) == Some(SARAH_EVENT_ROOM_STATE)
4631            }),
4632            "bootstrap must emit sarah_room_state"
4633        );
4634    }
4635
4636    #[test]
4637    fn send_and_snapshot_page_with_cursors() {
4638        let mut client = client();
4639        client.bootstrap().expect("bootstrap");
4640        let sent = client
4641            .send_message("Plan the next SARAH-NR packet", "idem.send.1", 1)
4642            .expect("send");
4643        assert!(sent.accepted);
4644        assert!(sent.message_ref.starts_with("msg."));
4645        assert!(sent.turn_ref.starts_with("turn."));
4646        assert!(!sent.event_id.is_empty());
4647
4648        let snap = client.room_snapshot(None, Some(10)).expect("snapshot");
4649        assert_eq!(snap.conversation_ref, client.conversation_ref());
4650        assert_eq!(snap.transcript.gap_state, GapState::None);
4651        assert!(!snap.transcript.cursor.is_empty());
4652        assert_eq!(snap.transcript.entries.len(), 1);
4653        assert_eq!(snap.transcript.entries[0].role, "owner");
4654        assert_eq!(
4655            snap.transcript.entries[0].text,
4656            "Plan the next SARAH-NR packet"
4657        );
4658        assert_eq!(snap.run_state.state, "running");
4659
4660        let conversation_ref = client.conversation_ref();
4661        let published = client
4662            .relay
4663            .query(&conversation_ref, None, 10)
4664            .expect("published message");
4665        let message = published.events.last().expect("message event");
4666        assert!(message.tags.iter().any(|tag| {
4667            tag.first().map(String::as_str) == Some("turn")
4668                && tag.get(1).map(String::as_str) == Some(sent.turn_ref.as_str())
4669        }));
4670
4671        let events = client.drain_events();
4672        assert!(
4673            events.iter().any(|event| {
4674                event.get("method").and_then(Value::as_str) == Some(SARAH_EVENT_ROOM_EVENT)
4675            }),
4676            "send must emit sarah_room_event"
4677        );
4678    }
4679
4680    #[test]
4681    fn interrupt_is_pending_until_settled() {
4682        let mut client = client();
4683        client.bootstrap().expect("bootstrap");
4684        let sent = client
4685            .send_message("start a turn", "idem.send.2", 1)
4686            .expect("send");
4687        let interrupt = client
4688            .interrupt_turn(&sent.turn_ref, "idem.interrupt.1", 1)
4689            .expect("interrupt");
4690        assert!(interrupt.accepted);
4691        assert!(interrupt.pending);
4692        assert_eq!(interrupt.status, "pending");
4693        assert_eq!(interrupt.turn_ref, sent.turn_ref);
4694    }
4695
4696    #[test]
4697    fn issue31_owner_state_actions_publish_encrypted_mergeable_sources() {
4698        let mut client = client();
4699        client.bootstrap().expect("bootstrap");
4700        let context_ref = "sarah-conversation:sarah.aaaaaaaaaaaaaaaaaaaaaaaa";
4701        let read_arguments = |read_at| Issue31CommandArguments::ReadStatePatch {
4702            action_ref: "action.issue31.read_state.advance".into(),
4703            slot_id: "owner-mobile".into(),
4704            client_id: "iphone".into(),
4705            context_ref: context_ref.into(),
4706            read_at,
4707        };
4708        let (_, first_projection) = client
4709            .execute_issue31_owner_state_action(&read_arguments(100))
4710            .expect("first read-state write");
4711        let (second_event_id, second_projection) = client
4712            .execute_issue31_owner_state_action(&read_arguments(200))
4713            .expect("merged read-state write");
4714        assert_ne!(second_event_id, "0".repeat(64));
4715        assert!(matches!(
4716            first_projection,
4717            Issue31OwnerProjectionBody::ReadState { .. }
4718        ));
4719        let Issue31OwnerProjectionBody::ReadState { plaintext, .. } = second_projection else {
4720            panic!("read-state projection");
4721        };
4722        let value: Value = serde_json::from_str(&plaintext).expect("read-state plaintext");
4723        assert_eq!(value["contexts"][context_ref], 200);
4724        let contexts = client
4725            .load_issue31_read_state_contexts("read-state:owner-mobile")
4726            .expect("reload read-state contexts");
4727        assert_eq!(contexts.get(context_ref), Some(&200));
4728
4729        let reminder_id = "a".repeat(32);
4730        let (_, reminder_projection) = client
4731            .execute_issue31_owner_state_action(&Issue31CommandArguments::ReminderCreate {
4732                action_ref: "action.issue31.reminder.create".into(),
4733                reminder_id: reminder_id.clone(),
4734                note: Some("Review the signed build".into()),
4735                target_event_id: None,
4736                not_before: 300,
4737                expiration: Some(600),
4738            })
4739            .expect("reminder write");
4740        assert!(matches!(
4741            reminder_projection,
4742            Issue31OwnerProjectionBody::Reminder {
4743                reminder_id: projected_id,
4744                not_before: Some(300),
4745                expiration: Some(600),
4746                ..
4747            } if projected_id == reminder_id
4748        ));
4749        let conversation_ref = client.config.conversation_ref();
4750        let page = client
4751            .query_with_auth(&conversation_ref, None, MAX_PAGE_LIMIT)
4752            .expect("query encrypted sources");
4753        assert!(page.events.iter().any(|event| {
4754            event.kind == SARAH_READ_STATE_KIND
4755                && !event.content_summary.contains("client_id")
4756                && !event.content_summary.contains(context_ref)
4757        }));
4758        assert!(page.events.iter().any(|event| {
4759            event.kind == SARAH_REMINDER_KIND
4760                && !event.content_summary.contains("Review the signed build")
4761        }));
4762    }
4763
4764    #[test]
4765    fn stale_generation_fails_closed() {
4766        let mut client = client();
4767        let error = client
4768            .handle_request(SARAH_METHOD_SESSION_STATUS, 99, None)
4769            .expect_err("stale");
4770        match error {
4771            SarahConversationError::StaleGeneration { expected, got } => {
4772                assert_eq!(expected, 1);
4773                assert_eq!(got, 99);
4774            }
4775            other => panic!("expected stale generation, got {other}"),
4776        }
4777    }
4778
4779    #[test]
4780    fn process_generation_advances_without_changing_nostr_host_generation() {
4781        let mut client = client();
4782        client
4783            .synchronize_process_generation(4)
4784            .expect("advance trusted process fence");
4785        assert_eq!(client.generation(), 4);
4786        client
4787            .handle_request(SARAH_METHOD_SESSION_STATUS, 4, None)
4788            .expect("advanced generation request");
4789        assert!(matches!(
4790            client.synchronize_process_generation(3),
4791            Err(SarahConversationError::StaleGeneration {
4792                expected: 4,
4793                got: 3
4794            })
4795        ));
4796    }
4797
4798    /// A source event the host itself signed can still carry a body the device
4799    /// reader refuses. Before the emitter owned that decision, one such event
4800    /// aborted the whole projection pass, so a single malformed record stopped
4801    /// every later record reaching every paired device. The bad event is
4802    /// quarantined and the pass continues.
4803    #[test]
4804    fn one_refused_projection_source_is_quarantined_without_stopping_the_pass() {
4805        let signer = SigningIdentity::generate();
4806        let mut config = SarahConversationConfig::mock_fixture();
4807        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
4808        let owner_public_key_hex = signer.public_key_hex.clone();
4809        let sarah_public_key_hex = config.identity.sarah_public_key_hex.clone();
4810        let conversation_ref = config.conversation_ref();
4811        let mut relay = MockRelayAdapter::new();
4812        for (index, (event_id, created_at, text)) in [
4813            ("1".repeat(64), 10_u64, "before"),
4814            // An empty message body is inside every record-level bound and
4815            // outside the projection body contract.
4816            ("2".repeat(64), 11, ""),
4817            ("3".repeat(64), 12, "after"),
4818        ]
4819        .into_iter()
4820        .enumerate()
4821        {
4822            relay.seed_event(StoredConversationEvent {
4823                event_id,
4824                kind: crate::ISSUE31_PRIVATE_RUMOR_KIND,
4825                pubkey: owner_public_key_hex.clone(),
4826                created_at,
4827                conversation_ref: conversation_ref.clone(),
4828                content_summary: text.into(),
4829                tags: Vec::new(),
4830                record_kind: "message".into(),
4831                store_index: index,
4832            });
4833        }
4834        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
4835
4836        let host_configuration = Issue31HostConfiguration {
4837            host_ref: "omega.host.local".into(),
4838            host_public_key_hex: owner_public_key_hex.clone(),
4839            sarah_public_key_hex,
4840            conversation: conversation_ref,
4841            display_name: "Omega host".into(),
4842            relay_urls: vec!["wss://relay.example.com".into()],
4843            generation: 1,
4844        };
4845        let device_public_key_hex = "2".repeat(64);
4846        let mut controller =
4847            Issue31HostController::new(host_configuration.clone()).expect("host controller");
4848        controller
4849            .set_admitted_device_policy(
4850                vec![device_public_key_hex.clone()],
4851                vec![crate::Issue31PairingScope::ObserveIssue31],
4852            )
4853            .expect("admit the device");
4854        let challenge = controller
4855            .handle_pairing_event(
4856                Issue31PairingEvent {
4857                    event_id: "a".repeat(64),
4858                    record: Issue31PairingRecord::PairingRequest {
4859                        schema: crate::ISSUE31_PAIRING_SCHEMA.into(),
4860                        host_ref: host_configuration.host_ref.clone(),
4861                        host_public_key_hex: owner_public_key_hex.clone(),
4862                        device_public_key_hex: device_public_key_hex.clone(),
4863                        issued_at: 100,
4864                        pairing_request_ref: "pairing_request.projection".into(),
4865                        requested_scopes: vec![crate::Issue31PairingScope::ObserveIssue31],
4866                        expires_at: 700,
4867                    },
4868                },
4869                100,
4870            )
4871            .expect("pairing request")
4872            .expect("pairing challenge");
4873        let Issue31PairingRecord::PairingChallenge {
4874            challenge: challenge_value,
4875            ..
4876        } = &challenge
4877        else {
4878            panic!("expected a pairing challenge");
4879        };
4880        let challenge_value = challenge_value.clone();
4881        controller
4882            .record_emitted_pairing("b".repeat(64), challenge)
4883            .expect("record the challenge");
4884        let grant = controller
4885            .handle_pairing_event(
4886                Issue31PairingEvent {
4887                    event_id: "c".repeat(64),
4888                    record: Issue31PairingRecord::PairingResponse {
4889                        schema: crate::ISSUE31_PAIRING_SCHEMA.into(),
4890                        host_ref: host_configuration.host_ref,
4891                        host_public_key_hex: owner_public_key_hex,
4892                        device_public_key_hex,
4893                        issued_at: 110,
4894                        pairing_response_ref: "pairing_response.projection".into(),
4895                        pairing_challenge_event_id: "b".repeat(64),
4896                        challenge: challenge_value,
4897                        expires_at: 700,
4898                    },
4899                },
4900                110,
4901            )
4902            .expect("pairing response")
4903            .expect("scoped grant");
4904        controller
4905            .record_emitted_pairing("d".repeat(64), grant)
4906            .expect("record the grant");
4907        assert_eq!(controller.active_grants(120).expect("grants").len(), 1);
4908
4909        client.ensure_connected().expect("connect the mock relay");
4910        client
4911            .project_issue31_sources(&mut controller, 200)
4912            .expect("the pass survives a refused source");
4913
4914        assert_eq!(
4915            client
4916                .issue31_quarantined_events
4917                .get(&"2".repeat(64))
4918                .map(String::as_str),
4919            Some("reason.omega.invalid_projection_source")
4920        );
4921        let grant_ref = controller.active_grants(200).expect("grants")[0]
4922            .grant_ref
4923            .clone();
4924        assert!(controller.source_was_projected(&grant_ref, 1, &"1".repeat(64)));
4925        assert!(!controller.source_was_projected(&grant_ref, 1, &"2".repeat(64)));
4926        assert!(controller.source_was_projected(&grant_ref, 1, &"3".repeat(64)));
4927        // Two surviving projections, plus the coverage statement that tells the
4928        // device the third source was withheld.
4929        assert_eq!(client.issue31_private_outbox.len(), 3);
4930        assert_eq!(withheld_outbox_refs(&client).len(), 1);
4931    }
4932
4933    /// Every outbox entry queued under the withheld-sources schema.
4934    ///
4935    /// The outbox stores sealed gift wraps, so the test reads the host's own
4936    /// statement of what it published rather than trying to decrypt them; the
4937    /// outbox reference proves the record was actually queued for the device.
4938    fn withheld_outbox_refs(client: &SarahConversationClient) -> Vec<String> {
4939        client
4940            .issue31_private_outbox
4941            .keys()
4942            .filter(|outbox_ref| outbox_ref.starts_with(crate::ISSUE31_WITHHELD_SOURCES_SCHEMA))
4943            .cloned()
4944            .collect()
4945    }
4946
4947    fn withheld_substance(
4948        client: &SarahConversationClient,
4949        grant_ref: &str,
4950        generation: u64,
4951    ) -> Option<(String, Vec<Issue31WithheldSourceCount>)> {
4952        client
4953            .issue31_withheld_emissions
4954            .get(&format!("{grant_ref}:{generation}"))
4955            .cloned()
4956    }
4957
4958    // -----------------------------------------------------------------
4959    // omega#49: the omega#47 documents actually leaving the host.
4960    //
4961    // `full_auto_ui` built both documents from live host state and nothing
4962    // ever published them, so a paired phone rendered `no_host_projection` on
4963    // every device. These cover the pump half: who the records are addressed
4964    // to, what binding they carry, and when they are NOT sent.
4965    // -----------------------------------------------------------------
4966
4967    fn adjunct_outbox_refs(client: &SarahConversationClient, schema: &str) -> Vec<String> {
4968        client
4969            .issue31_private_outbox
4970            .keys()
4971            .filter(|outbox_ref| outbox_ref.starts_with(schema))
4972            .cloned()
4973            .collect()
4974    }
4975
4976    /// A reading the pump can publish, carrying no delivery claim of its own.
4977    fn host_documents(host_ref: &str, snapshot_ref: &str) -> Issue31HostProjectionDocuments {
4978        Issue31HostProjectionDocuments {
4979            host: json!({
4980                "schema": ISSUE31_HOST_ADJUNCT_SCHEMA,
4981                "hostRef": host_ref,
4982                "snapshotRef": snapshot_ref,
4983                "generatedAtMs": 1_784_894_400_000_u64,
4984                "projections": [],
4985            }),
4986            detail: json!({
4987                "schema": ISSUE31_FULL_AUTO_ADJUNCT_SCHEMA,
4988                "hostRef": host_ref,
4989                "snapshotRef": snapshot_ref,
4990                "generatedAtMs": 1_784_894_400_000_u64,
4991                "runs": [],
4992            }),
4993        }
4994    }
4995
4996    fn paired_adjunct_client(
4997        source: Option<Issue31HostProjectionSource>,
4998    ) -> (SarahConversationClient, Issue31HostController, String) {
4999        let signer = SigningIdentity::generate();
5000        let mut config = SarahConversationConfig::mock_fixture();
5001        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
5002        let owner_public_key_hex = signer.public_key_hex.clone();
5003        let sarah_public_key_hex = config.identity.sarah_public_key_hex.clone();
5004        let conversation_ref = config.conversation_ref();
5005        let mut client =
5006            SarahConversationClient::with_relay(config, Box::new(MockRelayAdapter::new()), signer);
5007        if let Some(source) = source {
5008            client.set_issue31_host_projection_source(source);
5009        }
5010        let device_public_key_hex = "2".repeat(64);
5011        let controller = pair_issue31_device(
5012            Issue31HostConfiguration {
5013                host_ref: "omega.host.local".into(),
5014                host_public_key_hex: owner_public_key_hex,
5015                sarah_public_key_hex,
5016                conversation: conversation_ref,
5017                display_name: "Omega host".into(),
5018                relay_urls: vec!["wss://relay.example.com".into()],
5019                generation: 1,
5020            },
5021            &device_public_key_hex,
5022        );
5023        (client, controller, device_public_key_hex)
5024    }
5025
5026    /// The gap itself: both omega#47 documents must reach every admitted
5027    /// device, addressed to that device and to the grant it holds.
5028    #[test]
5029    fn the_host_snapshot_and_its_detail_are_addressed_to_each_admitted_device() {
5030        let source: Issue31HostProjectionSource = Arc::new(|request| {
5031            Ok(Some(host_documents(request.host_ref, "snapshot.omega.issue31.aa")))
5032        });
5033        let (mut client, controller, device_public_key_hex) =
5034            paired_adjunct_client(Some(source));
5035        let grant = controller.active_grants(200).expect("grants")[0].clone();
5036        client.ensure_connected().expect("connect the mock relay");
5037        client
5038            .publish_issue31_host_adjuncts(&controller, 200)
5039            .expect("the pump publishes both documents");
5040
5041        assert_eq!(
5042            adjunct_outbox_refs(&client, ISSUE31_HOST_ADJUNCT_SCHEMA).len(),
5043            1,
5044        );
5045        assert_eq!(
5046            adjunct_outbox_refs(&client, ISSUE31_FULL_AUTO_ADJUNCT_SCHEMA).len(),
5047            1,
5048        );
5049        assert!(
5050            client
5051                .issue31_host_adjunct_emissions
5052                .contains_key(&format!("{}:{}", grant.grant_ref, grant.generation)),
5053            "the pump must remember what it sent, or it resends forever",
5054        );
5055
5056        // The binding the device checks the envelope against.
5057        let addressed = SarahConversationClient::address_issue31_adjunct(
5058            &host_documents("omega.host.local", "snapshot.omega.issue31.aa").host,
5059            ISSUE31_HOST_ADJUNCT_SCHEMA,
5060            ISSUE31_HOST_ADJUNCT_RECORD_TYPE,
5061            &grant,
5062        )
5063        .expect("the snapshot is addressable to this grant");
5064        assert_eq!(
5065            addressed.get("recordType").and_then(Value::as_str),
5066            Some(ISSUE31_HOST_ADJUNCT_RECORD_TYPE),
5067        );
5068        assert_eq!(
5069            addressed.get("devicePublicKeyHex").and_then(Value::as_str),
5070            Some(device_public_key_hex.as_str()),
5071        );
5072        assert_eq!(
5073            addressed.get("grantRef").and_then(Value::as_str),
5074            Some(grant.grant_ref.as_str()),
5075        );
5076        assert_eq!(
5077            addressed.get("hostPublicKeyHex").and_then(Value::as_str),
5078            Some(grant.host_public_key_hex.as_str()),
5079        );
5080    }
5081
5082    /// A pass that observed the same world must not re-send the snapshot. The
5083    /// device would otherwise be handed an identical record forever, and every
5084    /// one of them would cost it a decrypt.
5085    #[test]
5086    fn an_unchanged_reading_is_not_republished() {
5087        let source: Issue31HostProjectionSource = Arc::new(|request| {
5088            Ok(Some(host_documents(request.host_ref, "snapshot.omega.issue31.aa")))
5089        });
5090        let (mut client, controller, _device) = paired_adjunct_client(Some(source));
5091        client.ensure_connected().expect("connect the mock relay");
5092        client
5093            .publish_issue31_host_adjuncts(&controller, 200)
5094            .expect("first pass");
5095        client.issue31_private_outbox.clear();
5096        client
5097            .publish_issue31_host_adjuncts(&controller, 260)
5098            .expect("second pass");
5099        assert!(
5100            client.issue31_private_outbox.is_empty(),
5101            "an unchanged reading must not be republished",
5102        );
5103
5104        // A changed reading is a different snapshot and does go out again.
5105        let changed: Issue31HostProjectionSource = Arc::new(|request| {
5106            Ok(Some(host_documents(request.host_ref, "snapshot.omega.issue31.bb")))
5107        });
5108        client.set_issue31_host_projection_source(changed);
5109        client
5110            .publish_issue31_host_adjuncts(&controller, 320)
5111            .expect("third pass");
5112        assert_eq!(
5113            adjunct_outbox_refs(&client, ISSUE31_HOST_ADJUNCT_SCHEMA).len(),
5114            1,
5115        );
5116    }
5117
5118    /// A host that is not observing its Full Auto state says nothing rather
5119    /// than publishing an empty snapshot. Silence and "I looked and found
5120    /// nothing" are different claims and the device renders them differently.
5121    #[test]
5122    fn a_host_with_no_reading_publishes_nothing_at_all() {
5123        let (mut client, controller, _device) = paired_adjunct_client(None);
5124        client.ensure_connected().expect("connect the mock relay");
5125        client
5126            .publish_issue31_host_adjuncts(&controller, 200)
5127            .expect("a host with no reading still completes its pass");
5128        assert!(client.issue31_private_outbox.is_empty());
5129
5130        let silent: Issue31HostProjectionSource = Arc::new(|_| Ok(None));
5131        client.set_issue31_host_projection_source(silent);
5132        client
5133            .publish_issue31_host_adjuncts(&controller, 200)
5134            .expect("an unobserved host still completes its pass");
5135        assert!(client.issue31_private_outbox.is_empty());
5136    }
5137
5138    /// The one substitution a signed seal cannot rule out. A snapshot labelled
5139    /// with another machine's host reference would be bound by the device to
5140    /// this pairing, so the pump refuses to address it at all.
5141    #[test]
5142    fn a_snapshot_naming_another_host_is_never_addressed_to_this_device() {
5143        let (mut client, controller, _device) = paired_adjunct_client(None);
5144        let grant = controller.active_grants(200).expect("grants")[0].clone();
5145        let foreign = host_documents("omega.host.some-other-machine", "snapshot.omega.issue31.aa");
5146        assert!(
5147            SarahConversationClient::address_issue31_adjunct(
5148                &foreign.host,
5149                ISSUE31_HOST_ADJUNCT_SCHEMA,
5150                ISSUE31_HOST_ADJUNCT_RECORD_TYPE,
5151                &grant,
5152            )
5153            .is_err(),
5154        );
5155
5156        // And a reading that arrives already claiming who may read it is
5157        // refused rather than silently overwritten.
5158        let mut presumptuous = host_documents("omega.host.local", "snapshot.omega.issue31.aa").host;
5159        presumptuous
5160            .as_object_mut()
5161            .expect("object")
5162            .insert("devicePublicKeyHex".into(), json!("3".repeat(64)));
5163        assert!(
5164            SarahConversationClient::address_issue31_adjunct(
5165                &presumptuous,
5166                ISSUE31_HOST_ADJUNCT_SCHEMA,
5167                ISSUE31_HOST_ADJUNCT_RECORD_TYPE,
5168                &grant,
5169            )
5170            .is_err(),
5171        );
5172
5173        let source: Issue31HostProjectionSource = Arc::new(|_| {
5174            Ok(Some(host_documents(
5175                "omega.host.some-other-machine",
5176                "snapshot.omega.issue31.aa",
5177            )))
5178        });
5179        client.set_issue31_host_projection_source(source);
5180        client.ensure_connected().expect("connect the mock relay");
5181        assert!(client.publish_issue31_host_adjuncts(&controller, 200).is_err());
5182        assert!(client.issue31_private_outbox.is_empty());
5183    }
5184
5185    /// The snapshot advertises the capabilities; the detail is what the owner
5186    /// opens. A detail bound to a different snapshot is one the phone refuses
5187    /// as `snapshot_mismatch`, so publishing it would publish a refusal.
5188    #[test]
5189    fn a_detail_not_bound_to_the_snapshot_beside_it_is_never_sent() {
5190        let source: Issue31HostProjectionSource = Arc::new(|request| {
5191            let mut documents = host_documents(request.host_ref, "snapshot.omega.issue31.aa");
5192            documents
5193                .detail
5194                .as_object_mut()
5195                .expect("object")
5196                .insert("snapshotRef".into(), json!("snapshot.omega.issue31.bb"));
5197            Ok(Some(documents))
5198        });
5199        let (mut client, controller, _device) = paired_adjunct_client(Some(source));
5200        client.ensure_connected().expect("connect the mock relay");
5201        assert!(client.publish_issue31_host_adjuncts(&controller, 200).is_err());
5202    }
5203
5204    /// Pair one device with the host so the projection pass has somewhere to
5205    /// send a coverage statement.
5206    fn pair_issue31_device(
5207        host_configuration: Issue31HostConfiguration,
5208        device_public_key_hex: &str,
5209    ) -> Issue31HostController {
5210        let host_ref = host_configuration.host_ref.clone();
5211        let host_public_key_hex = host_configuration.host_public_key_hex.clone();
5212        let mut controller =
5213            Issue31HostController::new(host_configuration).expect("host controller");
5214        controller
5215            .set_admitted_device_policy(
5216                vec![device_public_key_hex.to_string()],
5217                vec![crate::Issue31PairingScope::ObserveIssue31],
5218            )
5219            .expect("admit the device");
5220        let challenge = controller
5221            .handle_pairing_event(
5222                Issue31PairingEvent {
5223                    event_id: "a".repeat(64),
5224                    record: Issue31PairingRecord::PairingRequest {
5225                        schema: crate::ISSUE31_PAIRING_SCHEMA.into(),
5226                        host_ref: host_ref.clone(),
5227                        host_public_key_hex: host_public_key_hex.clone(),
5228                        device_public_key_hex: device_public_key_hex.to_string(),
5229                        issued_at: 100,
5230                        pairing_request_ref: "pairing_request.coverage".into(),
5231                        requested_scopes: vec![crate::Issue31PairingScope::ObserveIssue31],
5232                        expires_at: 100_000,
5233                    },
5234                },
5235                100,
5236            )
5237            .expect("pairing request")
5238            .expect("pairing challenge");
5239        let Issue31PairingRecord::PairingChallenge {
5240            challenge: challenge_value,
5241            ..
5242        } = &challenge
5243        else {
5244            panic!("expected a pairing challenge");
5245        };
5246        let challenge_value = challenge_value.clone();
5247        controller
5248            .record_emitted_pairing("b".repeat(64), challenge)
5249            .expect("record the challenge");
5250        let grant = controller
5251            .handle_pairing_event(
5252                Issue31PairingEvent {
5253                    event_id: "c".repeat(64),
5254                    record: Issue31PairingRecord::PairingResponse {
5255                        schema: crate::ISSUE31_PAIRING_SCHEMA.into(),
5256                        host_ref,
5257                        host_public_key_hex,
5258                        device_public_key_hex: device_public_key_hex.to_string(),
5259                        issued_at: 110,
5260                        pairing_response_ref: "pairing_response.coverage".into(),
5261                        pairing_challenge_event_id: "b".repeat(64),
5262                        challenge: challenge_value,
5263                        expires_at: 100_000,
5264                    },
5265                },
5266                110,
5267            )
5268            .expect("pairing response")
5269            .expect("scoped grant");
5270        controller
5271            .record_emitted_pairing("d".repeat(64), grant)
5272            .expect("record the grant");
5273        controller
5274    }
5275
5276    /// Falsification, drop path one. A quarantined source is removed from the
5277    /// owner's view by a host-local map whose count never left the host. The
5278    /// device must be told an exact number and why, and the statement must go
5279    /// back to `complete` for a pass that withholds nothing.
5280    #[test]
5281    fn a_quarantined_source_is_reported_to_the_device_as_an_exact_withheld_count() {
5282        let signer = SigningIdentity::generate();
5283        let mut config = SarahConversationConfig::mock_fixture();
5284        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
5285        let owner_public_key_hex = signer.public_key_hex.clone();
5286        let sarah_public_key_hex = config.identity.sarah_public_key_hex.clone();
5287        let conversation_ref = config.conversation_ref();
5288        let mut relay = MockRelayAdapter::new();
5289        for (index, (event_id, created_at, text)) in [
5290            ("1".repeat(64), 10_u64, "readable"),
5291            // Inside every record-level bound and outside the projection body
5292            // contract, so the host quarantines it.
5293            ("2".repeat(64), 11, ""),
5294        ]
5295        .into_iter()
5296        .enumerate()
5297        {
5298            relay.seed_event(StoredConversationEvent {
5299                event_id,
5300                kind: crate::ISSUE31_PRIVATE_RUMOR_KIND,
5301                pubkey: owner_public_key_hex.clone(),
5302                created_at,
5303                conversation_ref: conversation_ref.clone(),
5304                content_summary: text.into(),
5305                tags: Vec::new(),
5306                record_kind: "message".into(),
5307                store_index: index,
5308            });
5309        }
5310        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
5311        let device_public_key_hex = "2".repeat(64);
5312        let mut controller = pair_issue31_device(
5313            Issue31HostConfiguration {
5314                host_ref: "omega.host.local".into(),
5315                host_public_key_hex: owner_public_key_hex,
5316                sarah_public_key_hex,
5317                conversation: conversation_ref,
5318                display_name: "Omega host".into(),
5319                relay_urls: vec!["wss://relay.example.com".into()],
5320                generation: 1,
5321            },
5322            &device_public_key_hex,
5323        );
5324        let grant_ref = controller.active_grants(200).expect("grants")[0]
5325            .grant_ref
5326            .clone();
5327
5328        client.ensure_connected().expect("connect the mock relay");
5329        client
5330            .project_issue31_sources(&mut controller, 200)
5331            .expect("the pass survives a quarantined source");
5332
5333        let (coverage, withheld) =
5334            withheld_substance(&client, &grant_ref, 1).expect("a coverage statement was published");
5335        assert_eq!(coverage, crate::ISSUE31_WITHHELD_COVERAGE_PARTIAL);
5336        assert_eq!(
5337            withheld,
5338            vec![Issue31WithheldSourceCount {
5339                cause: Issue31WithheldCause::Quarantined,
5340                count: 1,
5341                exact: true,
5342                reason_ref: ISSUE31_PROJECTION_SOURCE_QUARANTINE_REASON.into(),
5343            }]
5344        );
5345        assert_eq!(withheld_outbox_refs(&client).len(), 1);
5346
5347        // Restore: a host with nothing quarantined states completeness rather
5348        // than staying quiet, because silence would have to read as unknown.
5349        let clean_signer = SigningIdentity::generate();
5350        let mut clean_config = SarahConversationConfig::mock_fixture();
5351        clean_config.identity.owner_public_key_hex = clean_signer.public_key_hex.clone();
5352        let clean_owner = clean_signer.public_key_hex.clone();
5353        let clean_sarah = clean_config.identity.sarah_public_key_hex.clone();
5354        let clean_conversation = clean_config.conversation_ref();
5355        let mut clean_relay = MockRelayAdapter::new();
5356        clean_relay.seed_event(StoredConversationEvent {
5357            event_id: "1".repeat(64),
5358            kind: crate::ISSUE31_PRIVATE_RUMOR_KIND,
5359            pubkey: clean_owner.clone(),
5360            created_at: 10,
5361            conversation_ref: clean_conversation.clone(),
5362            content_summary: "readable".into(),
5363            tags: Vec::new(),
5364            record_kind: "message".into(),
5365            store_index: 0,
5366        });
5367        let mut clean_client =
5368            SarahConversationClient::with_relay(clean_config, Box::new(clean_relay), clean_signer);
5369        let mut clean_controller = pair_issue31_device(
5370            Issue31HostConfiguration {
5371                host_ref: "omega.host.local".into(),
5372                host_public_key_hex: clean_owner,
5373                sarah_public_key_hex: clean_sarah,
5374                conversation: clean_conversation,
5375                display_name: "Omega host".into(),
5376                relay_urls: vec!["wss://relay.example.com".into()],
5377                generation: 1,
5378            },
5379            &device_public_key_hex,
5380        );
5381        clean_client
5382            .ensure_connected()
5383            .expect("connect the mock relay");
5384        clean_client
5385            .project_issue31_sources(&mut clean_controller, 200)
5386            .expect("a clean pass");
5387        assert_eq!(
5388            withheld_substance(&clean_client, &grant_ref, 1),
5389            Some((crate::ISSUE31_WITHHELD_COVERAGE_COMPLETE.to_string(), vec![]))
5390        );
5391    }
5392
5393    /// Falsification, drop path two. The bounded projection scan stops after
5394    /// eight pages and, before this change, said so only in the host's own
5395    /// `last_gap_state`. The device must see an inexact count, and must see it
5396    /// clear once the scan catches up.
5397    #[test]
5398    fn the_projection_scan_bound_is_reported_to_the_device_and_clears_when_it_catches_up() {
5399        let signer = SigningIdentity::generate();
5400        let mut config = SarahConversationConfig::mock_fixture();
5401        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
5402        let owner_public_key_hex = signer.public_key_hex.clone();
5403        let sarah_public_key_hex = config.identity.sarah_public_key_hex.clone();
5404        let conversation_ref = config.conversation_ref();
5405        let mut relay = MockRelayAdapter::new();
5406        // One page past the eight-page bound. The kind is outside the
5407        // projection set, so this measures the scan bound itself rather than
5408        // the cost of projecting five hundred sources.
5409        let pages_past_the_bound = 8 * MAX_PAGE_LIMIT + 1;
5410        for index in 0..pages_past_the_bound {
5411            relay.seed_event(StoredConversationEvent {
5412                event_id: format!("{index:064x}"),
5413                kind: 1,
5414                pubkey: owner_public_key_hex.clone(),
5415                created_at: 10 + index as u64,
5416                conversation_ref: conversation_ref.clone(),
5417                content_summary: "unrelated".into(),
5418                tags: Vec::new(),
5419                record_kind: "note".into(),
5420                store_index: index,
5421            });
5422        }
5423        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
5424        let device_public_key_hex = "2".repeat(64);
5425        let mut controller = pair_issue31_device(
5426            Issue31HostConfiguration {
5427                host_ref: "omega.host.local".into(),
5428                host_public_key_hex: owner_public_key_hex,
5429                sarah_public_key_hex,
5430                conversation: conversation_ref,
5431                display_name: "Omega host".into(),
5432                relay_urls: vec!["wss://relay.example.com".into()],
5433                generation: 1,
5434            },
5435            &device_public_key_hex,
5436        );
5437        let grant_ref = controller.active_grants(200).expect("grants")[0]
5438            .grant_ref
5439            .clone();
5440
5441        client.ensure_connected().expect("connect the mock relay");
5442        client
5443            .project_issue31_sources(&mut controller, 200)
5444            .expect("the bounded pass");
5445
5446        let (coverage, withheld) =
5447            withheld_substance(&client, &grant_ref, 1).expect("a coverage statement was published");
5448        assert_eq!(coverage, crate::ISSUE31_WITHHELD_COVERAGE_PARTIAL);
5449        assert_eq!(
5450            withheld,
5451            vec![Issue31WithheldSourceCount {
5452                cause: Issue31WithheldCause::ScanBound,
5453                count: 1,
5454                exact: false,
5455                reason_ref: ISSUE31_PROJECTION_SCAN_BOUND_REASON.into(),
5456            }]
5457        );
5458        assert_eq!(client.last_gap_state, GapState::Possible);
5459        let after_bounded_pass = withheld_outbox_refs(&client);
5460        assert_eq!(after_bounded_pass.len(), 1);
5461
5462        // The cursor advanced, so the next pass reaches the end and the device
5463        // is told so. A signal that could only ever get worse would be a worse
5464        // lie than none.
5465        client
5466            .project_issue31_sources(&mut controller, 300)
5467            .expect("the catching-up pass");
5468        assert_eq!(
5469            withheld_substance(&client, &grant_ref, 1),
5470            Some((crate::ISSUE31_WITHHELD_COVERAGE_COMPLETE.to_string(), vec![]))
5471        );
5472        let after_catching_up = withheld_outbox_refs(&client);
5473        assert_eq!(after_catching_up.len(), 2);
5474        assert!(after_catching_up[0] != after_catching_up[1]);
5475    }
5476
5477    /// The host re-runs its projection pass continuously. A statement that has
5478    /// not changed must not be republished with only a new timestamp, or the
5479    /// device's own store fills with restatements of one fact.
5480    #[test]
5481    fn an_unchanged_coverage_statement_is_not_republished() {
5482        let signer = SigningIdentity::generate();
5483        let mut config = SarahConversationConfig::mock_fixture();
5484        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
5485        let owner_public_key_hex = signer.public_key_hex.clone();
5486        let sarah_public_key_hex = config.identity.sarah_public_key_hex.clone();
5487        let conversation_ref = config.conversation_ref();
5488        let mut relay = MockRelayAdapter::new();
5489        relay.seed_event(StoredConversationEvent {
5490            event_id: "1".repeat(64),
5491            kind: crate::ISSUE31_PRIVATE_RUMOR_KIND,
5492            pubkey: owner_public_key_hex.clone(),
5493            created_at: 10,
5494            conversation_ref: conversation_ref.clone(),
5495            content_summary: "readable".into(),
5496            tags: Vec::new(),
5497            record_kind: "message".into(),
5498            store_index: 0,
5499        });
5500        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
5501        let mut controller = pair_issue31_device(
5502            Issue31HostConfiguration {
5503                host_ref: "omega.host.local".into(),
5504                host_public_key_hex: owner_public_key_hex,
5505                sarah_public_key_hex,
5506                conversation: conversation_ref,
5507                display_name: "Omega host".into(),
5508                relay_urls: vec!["wss://relay.example.com".into()],
5509                generation: 1,
5510            },
5511            &"2".repeat(64),
5512        );
5513        client.ensure_connected().expect("connect the mock relay");
5514        client
5515            .project_issue31_sources(&mut controller, 200)
5516            .expect("the first pass");
5517        assert_eq!(withheld_outbox_refs(&client).len(), 1);
5518        client
5519            .project_issue31_sources(&mut controller, 400)
5520            .expect("the second pass, at a different clock");
5521        assert_eq!(withheld_outbox_refs(&client).len(), 1);
5522    }
5523
5524    #[test]
5525    fn snapshot_honors_independent_transcript_and_activity_windows() {
5526        let signer = SigningIdentity::generate();
5527        let mut config = SarahConversationConfig::mock_fixture();
5528        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
5529        let conversation_ref = config.conversation_ref();
5530        let mut relay = MockRelayAdapter::new();
5531        relay.seed_event(StoredConversationEvent {
5532            event_id: "1".repeat(64),
5533            kind: Kind::PrivateDirectMessage.as_u16(),
5534            pubkey: signer.public_key_hex.clone(),
5535            created_at: 1,
5536            conversation_ref: conversation_ref.clone(),
5537            content_summary: "first".into(),
5538            tags: Vec::new(),
5539            record_kind: "message".into(),
5540            store_index: 0,
5541        });
5542        relay.seed_event(StoredConversationEvent {
5543            event_id: "2".repeat(64),
5544            kind: SARAH_TURN_RECORD_KIND,
5545            pubkey: signer.public_key_hex.clone(),
5546            created_at: 2,
5547            conversation_ref: conversation_ref.clone(),
5548            content_summary: "activity".into(),
5549            tags: vec![
5550                vec!["entry".into(), "entry.test".into()],
5551                vec!["turn".into(), "turn.test".into()],
5552            ],
5553            record_kind: "activity".into(),
5554            store_index: 1,
5555        });
5556        relay.seed_event(StoredConversationEvent {
5557            event_id: "3".repeat(64),
5558            kind: Kind::PrivateDirectMessage.as_u16(),
5559            pubkey: signer.public_key_hex.clone(),
5560            created_at: 3,
5561            conversation_ref,
5562            content_summary: "second".into(),
5563            tags: Vec::new(),
5564            record_kind: "message".into(),
5565            store_index: 2,
5566        });
5567        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
5568        let snapshot = client
5569            .room_snapshot_with_cursors(
5570                Some(&format!("cursor.1.{}", "1".repeat(64))),
5571                Some(1),
5572                None,
5573                Some(1),
5574            )
5575            .expect("independent snapshot");
5576        assert_eq!(snapshot.transcript.entries[0].text, "second");
5577        assert_eq!(snapshot.activity.entries[0].entry, "entry.test");
5578        assert!(snapshot.transcript.next_cursor.is_none());
5579        assert!(snapshot.activity.next_cursor.is_none());
5580    }
5581
5582    #[test]
5583    fn snapshot_projects_bounded_confirmed_nostr_record_refs_without_content() {
5584        let config = SarahConversationConfig::mock_fixture();
5585        let conversation_ref = config.conversation_ref();
5586        let mut relay = MockRelayAdapter::new();
5587        relay.seed_event(StoredConversationEvent {
5588            event_id: "e".repeat(64),
5589            kind: 30_174,
5590            pubkey: "f".repeat(64),
5591            created_at: 100,
5592            conversation_ref,
5593            content_summary: "nip44:encrypted-memory-content".into(),
5594            tags: vec![vec!["d".into(), "a".repeat(64)]],
5595            record_kind: "memory".into(),
5596            store_index: 0,
5597        });
5598        let mut client = SarahConversationClient::with_relay(
5599            config,
5600            Box::new(relay),
5601            SigningIdentity::generate(),
5602        );
5603
5604        let snapshot = client
5605            .room_snapshot_with_record_cursor(None, Some(1), None, Some(1), None, Some(1))
5606            .expect("snapshot");
5607        assert_eq!(snapshot.nostr_records.entries.len(), 1);
5608        let record = &snapshot.nostr_records.entries[0];
5609        assert_eq!(record.event_id, "e".repeat(64));
5610        assert_eq!(record.kind, 30_174);
5611        assert_eq!(record.record_kind, "memory");
5612        assert_eq!(record.source, "confirmed_nostr");
5613        assert_eq!(record.author_fingerprint.len(), 16);
5614        let encoded = serde_json::to_string(&snapshot.nostr_records).expect("json");
5615        assert!(!encoded.contains("encrypted-memory-content"));
5616    }
5617
5618    #[test]
5619    fn nip42_auth_against_mock_relay_when_required() {
5620        let signer = SigningIdentity::generate();
5621        let challenge = "test-challenge-42";
5622        let mut client = SarahConversationClient::mock_with_nip42_auth(
5623            SarahConversationConfig::mock_fixture(),
5624            challenge,
5625            signer,
5626        );
5627        let boot = client.bootstrap().expect("bootstrap after NIP-42");
5628        assert!(boot.room_state.authenticated);
5629        assert_eq!(boot.room_state.transport, "nostr_relay");
5630        let sent = client
5631            .send_message("authenticated send", "idem.send.auth", 1)
5632            .expect("send after auth");
5633        assert!(sent.accepted);
5634    }
5635
5636    #[test]
5637    fn nip42_rejects_wrong_challenge() {
5638        let signer = SigningIdentity::generate();
5639        let mut relay = MockRelayAdapter::with_required_auth("expected-challenge");
5640        relay.connect().expect("connect");
5641        let auth = signer
5642            .sign_auth("wrong-challenge", "wss://relay.openagents.com")
5643            .expect("sign");
5644        let error = relay.authenticate(&auth).expect_err("must reject");
5645        assert!(error.to_string().contains("NIP-42"));
5646    }
5647
5648    #[test]
5649    fn frames_respect_64kib_and_redact_secrets() {
5650        let client = client();
5651        let result = json!({ "ok": true, "cursor": "cursor.0", "gapState": "none" });
5652        let line = client
5653            .encode_response_frame("1", 1, result)
5654            .expect("encode");
5655        assert!(line.len() < MAX_FRAME_BYTES);
5656        assert!(line.contains(PROTOCOL_SCHEMA));
5657
5658        let secretish = json!({ "text": "bearer super-secret-token" });
5659        let refused = client.encode_event_frame(1, SARAH_EVENT_ROOM_EVENT, secretish);
5660        assert!(refused.is_err());
5661    }
5662
5663    #[test]
5664    fn refuses_secret_shaped_outbound_message() {
5665        let mut client = client();
5666        client.bootstrap().expect("bootstrap");
5667        let error = client
5668            .send_message("Authorization: Bearer sk-test-123", "idem.send.secret", 1)
5669            .expect_err("secret");
5670        assert!(matches!(error, SarahConversationError::InvalidRequest(_)));
5671    }
5672
5673    #[test]
5674    fn client_collection_bounds_are_enforced_before_command_mutation() {
5675        let mut client = client();
5676        client.command_results = (0..MAX_COMMAND_RESULTS)
5677            .map(|index| {
5678                (
5679                    format!("idempotency.bound.{index}"),
5680                    ("fingerprint".into(), json!({ "accepted": true })),
5681                )
5682            })
5683            .collect();
5684        let params = json!({
5685            "text": "must not mutate",
5686            "idempotencyRef": "idempotency.bound.new",
5687            "expectedGeneration": 1,
5688        });
5689        let error = client
5690            .handle_request(SARAH_METHOD_SEND_MESSAGE, 1, Some(&params))
5691            .expect_err("command result bound");
5692        assert!(error.to_string().contains("bound"));
5693        assert_eq!(client.message_seq, 0);
5694        assert!(client.active_turn_ref.is_none());
5695        assert!(client.issue31_private_outbox.is_empty());
5696
5697        let state = client.current_room_state();
5698        for _ in 0..MAX_PENDING_EVENTS.saturating_add(32) {
5699            client.push_room_state_event(&state);
5700        }
5701        assert_eq!(client.pending_events.len(), MAX_PENDING_EVENTS);
5702    }
5703
5704    #[test]
5705    fn unbound_real_controller_never_reports_false_command_completion() {
5706        let mut client = client();
5707        client.run_state = "running".into();
5708        client.active_turn_ref = Some("turn.9".into());
5709        for action_ref in [
5710            "action.omega.full_auto.stop",
5711            "action.omega.full_auto.pause",
5712            "action.omega.full_auto.resume",
5713            "action.omega.interrupt_turn",
5714            "action.omega.send_message",
5715        ] {
5716            let execution = client.execute_issue31_action(
5717                action_ref,
5718                "arguments.omega.none",
5719                "idempotency.issue31.unbound",
5720            );
5721            assert_eq!(execution.status, Issue31CommandStatus::Unavailable);
5722            assert_eq!(
5723                execution.reason_ref.as_deref(),
5724                Some("reason.omega.controller_not_bound")
5725            );
5726        }
5727        assert_eq!(client.run_state, "running");
5728        assert_eq!(client.active_turn_ref.as_deref(), Some("turn.9"));
5729    }
5730
5731    // ---------------------------------------------------------------------
5732    // omega#91 provider connection handoffs
5733    // ---------------------------------------------------------------------
5734
5735    /// A paired device holding exactly the scope the phone already asks for.
5736    fn handoff_fixture_with(
5737        scopes: Vec<crate::Issue31PairingScope>,
5738    ) -> (
5739        SarahConversationClient,
5740        Issue31HostConfiguration,
5741        Issue31HostController,
5742        String,
5743        String,
5744    ) {
5745        let (configuration, controller, device_public_key_hex, grant_ref) =
5746            crate::issue31_nostr::paired_fixture(scopes);
5747        let mut config = SarahConversationConfig::mock_fixture();
5748        config.identity.owner_public_key_hex = configuration.host_public_key_hex.clone();
5749        let client = SarahConversationClient::with_relay(
5750            config,
5751            Box::new(MockRelayAdapter::new()),
5752            SigningIdentity::generate(),
5753        );
5754        (
5755            client,
5756            configuration,
5757            controller,
5758            device_public_key_hex,
5759            grant_ref,
5760        )
5761    }
5762
5763    fn handoff_fixture() -> (
5764        SarahConversationClient,
5765        Issue31HostConfiguration,
5766        Issue31HostController,
5767        String,
5768        String,
5769    ) {
5770        handoff_fixture_with(vec![crate::Issue31PairingScope::RequestProviderHandoff])
5771    }
5772
5773    fn handoff_intent(
5774        configuration: &Issue31HostConfiguration,
5775        device_public_key_hex: &str,
5776        grant_ref: &str,
5777        event_id: &str,
5778        idempotency_ref: &str,
5779        arguments_ref: &str,
5780    ) -> Issue31CommandEvent {
5781        Issue31CommandEvent {
5782            event_id: event_id.to_string(),
5783            record: Issue31CommandRecord::CommandIntent {
5784                schema: ISSUE31_COMMAND_SCHEMA.into(),
5785                host_ref: configuration.host_ref.clone(),
5786                host_public_key_hex: configuration.host_public_key_hex.clone(),
5787                device_public_key_hex: device_public_key_hex.to_string(),
5788                grant_ref: grant_ref.to_string(),
5789                action_ref: ISSUE31_ACTION_REQUEST_PROVIDER_HANDOFF.into(),
5790                idempotency_ref: idempotency_ref.to_string(),
5791                expected_generation: 1,
5792                arguments_ref: arguments_ref.to_string(),
5793                issued_at: 103,
5794                expires_at: 900,
5795            },
5796        }
5797    }
5798
5799    #[test]
5800    fn the_scope_the_phone_holds_now_produces_a_host_record() {
5801        // Before omega#91 this exact command reached `execute_issue31_action`
5802        // and came back `unavailable`/`controller_not_bound`, and the handoff
5803        // vector on the wire stayed empty forever.
5804        let (mut client, configuration, mut controller, device_public_key_hex, grant_ref) =
5805            handoff_fixture();
5806        let intent = handoff_intent(
5807            &configuration,
5808            &device_public_key_hex,
5809            &grant_ref,
5810            &"1".repeat(64),
5811            "idempotency.issue31.handoff:first",
5812            "arguments.omega.provider_handoff.anthropic",
5813        );
5814        let result = controller
5815            .handle_command_event(
5816                intent,
5817                104,
5818                |action_ref, arguments_ref, idempotency_ref| {
5819                    client.execute_issue31_action(action_ref, arguments_ref, idempotency_ref)
5820                },
5821            )
5822            .expect("the host answers an admitted handoff request")
5823            .expect("a command result");
5824        let Issue31CommandRecord::CommandResult {
5825            status,
5826            outcome_ref,
5827            reason_ref,
5828            ..
5829        } = &result
5830        else {
5831            panic!("expected a command result");
5832        };
5833        assert_eq!(*status, Issue31CommandStatus::Completed);
5834        assert_eq!(reason_ref.as_deref(), None);
5835        // The command completed by producing this record; the record itself is
5836        // not terminal. Reading one as the other is the mistake this asserts
5837        // against.
5838        assert_eq!(
5839            client.issue31_provider_handoff_refs(),
5840            vec![outcome_ref.clone()]
5841        );
5842        let record = client
5843            .issue31_provider_handoffs
5844            .get(outcome_ref)
5845            .expect("the outcome names the record the host made")
5846            .clone();
5847        assert_eq!(record.state, workroom_receipts::Issue31ProviderHandoffState::Requested);
5848        assert!(!record.is_terminal());
5849        assert!(record.requested_at_ms.is_some(), "the host stamped it");
5850        assert!(record.account_ref.is_none(), "nothing is bound yet");
5851        assert!(record.outcome_ref.is_none(), "no outcome is claimed yet");
5852    }
5853
5854    #[test]
5855    fn a_scope_denied_request_leaves_no_handoff_at_all() {
5856        // The failure-vs-never-started distinction, at the wire. A device
5857        // without the scope gets a refusal — and the host makes no record, so
5858        // the phone's handoff list is empty rather than showing a failed row
5859        // for something that never began.
5860        let (mut client, configuration, mut controller, device_public_key_hex, grant_ref) =
5861            handoff_fixture_with(vec![crate::Issue31PairingScope::ObserveIssue31]);
5862        let intent = handoff_intent(
5863            &configuration,
5864            &device_public_key_hex,
5865            &grant_ref,
5866            &"2".repeat(64),
5867            "idempotency.issue31.handoff:denied",
5868            "arguments.omega.provider_handoff.anthropic",
5869        );
5870        let result = controller
5871            .handle_command_event(
5872                intent,
5873                104,
5874                |action_ref, arguments_ref, idempotency_ref| {
5875                    client.execute_issue31_action(action_ref, arguments_ref, idempotency_ref)
5876                },
5877            )
5878            .expect("the controller answers")
5879            .expect("a command result");
5880        let Issue31CommandRecord::CommandResult { status, reason_ref, .. } = &result else {
5881            panic!("expected a command result");
5882        };
5883        assert_eq!(*status, Issue31CommandStatus::Refused);
5884        assert_eq!(reason_ref.as_deref(), Some("reason.omega.scope_denied"));
5885        assert!(
5886            client.issue31_provider_handoff_refs().is_empty(),
5887            "a request the host never admitted is not a handoff that failed",
5888        );
5889        assert!(client.issue31_projected_provider_handoffs(1_000_000).is_empty());
5890    }
5891
5892    #[test]
5893    fn a_handoff_request_naming_no_provider_is_refused_without_a_record() {
5894        let (mut client, configuration, mut controller, device_public_key_hex, grant_ref) =
5895            handoff_fixture();
5896        let intent = handoff_intent(
5897            &configuration,
5898            &device_public_key_hex,
5899            &grant_ref,
5900            &"3".repeat(64),
5901            "idempotency.issue31.handoff:bad_arguments",
5902            "arguments.omega.none",
5903        );
5904        let result = controller
5905            .handle_command_event(
5906                intent,
5907                104,
5908                |action_ref, arguments_ref, idempotency_ref| {
5909                    client.execute_issue31_action(action_ref, arguments_ref, idempotency_ref)
5910                },
5911            )
5912            .expect("the controller answers")
5913            .expect("a command result");
5914        let Issue31CommandRecord::CommandResult { status, reason_ref, .. } = &result else {
5915            panic!("expected a command result");
5916        };
5917        assert_eq!(*status, Issue31CommandStatus::Refused);
5918        assert_eq!(
5919            reason_ref.as_deref(),
5920            Some("reason.omega.handoff_arguments_invalid")
5921        );
5922        assert!(client.issue31_provider_handoff_refs().is_empty());
5923    }
5924
5925    #[test]
5926    fn a_handoff_in_flight_survives_a_restart_and_is_settled_rather_than_lost() {
5927        let temporary = tempfile::tempdir().expect("tempdir");
5928        let state_path = temporary.path().join("owner").join("issue31-state.json");
5929        let (mut client, configuration, controller, _, _) = handoff_fixture();
5930        client.issue31_state_path = Some(state_path.clone());
5931        let opened = client
5932            .issue31_provider_handoffs
5933            .open(
5934                "arguments.omega.provider_handoff.anthropic",
5935                "idempotency.issue31.handoff:restart",
5936                1_785_000_000_000,
5937            )
5938            .expect("the host opens a handoff");
5939        client
5940            .persist_issue31_host_state_with_controller(&controller)
5941            .expect("the ledger is committed with the rest of host state");
5942
5943        // Exactly what a restarted process reads back off disk.
5944        let reloaded = load_issue31_host_state(&state_path, &configuration)
5945            .expect("load")
5946            .expect("persisted state");
5947        let mut ledger = reloaded.provider_handoffs;
5948        assert_eq!(
5949            ledger.get(&opened.handoff_ref).expect("the row survived"),
5950            &opened,
5951            "a handoff in flight is not lost across a restart",
5952        );
5953
5954        // And a restart resolves it rather than leaving the phone with a
5955        // request that neither resolves nor fails.
5956        assert_eq!(ledger.adopt_after_restart(), 1);
5957        let settled = ledger.get(&opened.handoff_ref).expect("row").clone();
5958        assert_eq!(
5959            settled.state,
5960            workroom_receipts::Issue31ProviderHandoffState::Failed
5961        );
5962        assert_eq!(
5963            settled.reason_class.as_deref(),
5964            Some(crate::issue31_provider_handoff::ISSUE31_HANDOFF_REASON_HOST_RESTARTED)
5965        );
5966        assert!(settled.is_terminal());
5967    }
5968
5969    #[test]
5970    fn the_pump_advances_a_handoff_against_the_hosts_own_roster() {
5971        let (mut client, _, _, _, _) = handoff_fixture();
5972        let opened = client
5973            .issue31_provider_handoffs
5974            .open(
5975                "arguments.omega.provider_handoff.anthropic",
5976                "idempotency.issue31.handoff:roster",
5977                1_000,
5978            )
5979            .expect("open");
5980        client.set_issue31_provider_roster_source(Arc::new(|| {
5981            Some(vec![Issue31ProviderRosterAccount {
5982                account_ref: "account.claude.1".into(),
5983                provider: "anthropic".into(),
5984                lane_ref: "lane.claude-local".into(),
5985                readiness: "ready".into(),
5986            }])
5987        }));
5988        client.advance_issue31_provider_handoffs(2);
5989        assert_eq!(
5990            client
5991                .issue31_provider_handoffs
5992                .get(&opened.handoff_ref)
5993                .expect("row")
5994                .account_ref
5995                .as_deref(),
5996            Some("account.claude.1"),
5997        );
5998        client.advance_issue31_provider_handoffs(3);
5999        assert_eq!(
6000            client
6001                .issue31_provider_handoffs
6002                .get(&opened.handoff_ref)
6003                .expect("row")
6004                .state,
6005            workroom_receipts::Issue31ProviderHandoffState::Completed,
6006        );
6007    }
6008
6009    #[test]
6010    fn a_host_that_never_read_its_roster_binds_nothing() {
6011        let (mut client, _, _, _, _) = handoff_fixture();
6012        let opened = client
6013            .issue31_provider_handoffs
6014            .open(
6015                "arguments.omega.provider_handoff.anthropic",
6016                "idempotency.issue31.handoff:unread",
6017                1_000,
6018            )
6019            .expect("open");
6020        client.set_issue31_provider_roster_source(Arc::new(|| None));
6021        client.advance_issue31_provider_handoffs(2);
6022        assert_eq!(
6023            client
6024                .issue31_provider_handoffs
6025                .get(&opened.handoff_ref)
6026                .expect("row")
6027                .state,
6028            workroom_receipts::Issue31ProviderHandoffState::Requested,
6029        );
6030    }
6031
6032    #[test]
6033    fn no_khala_sync_client_on_sarah_lane() {
6034        assert!(asserts_no_khala_sync_client());
6035    }
6036
6037    #[test]
6038    fn handle_request_covers_all_methods() {
6039        let mut client = client();
6040        for method in SARAH_FRAMED_METHODS {
6041            if matches!(
6042                *method,
6043                SARAH_METHOD_DEVICE_GRANTS
6044                    | SARAH_METHOD_RENEW_DEVICE_GRANT
6045                    | SARAH_METHOD_REVOKE_DEVICE_GRANT
6046                    | SARAH_METHOD_READMIT_DEVICE
6047            ) {
6048                continue;
6049            }
6050            let params = match *method {
6051                SARAH_METHOD_SEND_MESSAGE => Some(json!({
6052                    "text": "hello from test",
6053                    "idempotencyRef": "idem.handle.send",
6054                    "expectedGeneration": 1,
6055                })),
6056                SARAH_METHOD_INTERRUPT_TURN => {
6057                    client
6058                        .send_message("prep", "idem.prep", 1)
6059                        .expect("prepare active turn");
6060                    Some(json!({
6061                        "turnRef": "turn.1",
6062                        "idempotencyRef": "idem.handle.interrupt",
6063                        "expectedGeneration": 1,
6064                    }))
6065                }
6066                SARAH_METHOD_ROOM_SNAPSHOT => Some(json!({ "limit": 5 })),
6067                _ => None,
6068            };
6069            client
6070                .handle_request(method, 1, params.as_ref())
6071                .unwrap_or_else(|error| panic!("{method} failed: {error}"));
6072        }
6073    }
6074
6075    struct OneHealthyOneDownRelay {
6076        state_path: PathBuf,
6077        query_count: Arc<AtomicUsize>,
6078        connected: bool,
6079        acknowledged_event_id: Option<String>,
6080    }
6081
6082    impl RelayTransport for OneHealthyOneDownRelay {
6083        fn label(&self) -> &str {
6084            "wss://healthy.example"
6085        }
6086
6087        fn connection_state(&self) -> ConnectionState {
6088            if self.connected {
6089                ConnectionState::Degraded
6090            } else {
6091                ConnectionState::Disconnected
6092            }
6093        }
6094
6095        fn is_authenticated(&self) -> bool {
6096            true
6097        }
6098
6099        fn connect(&mut self) -> Result<(), SarahConversationError> {
6100            self.connected = true;
6101            Ok(())
6102        }
6103
6104        fn auth_challenge(&self) -> Option<RelayAuthChallenge> {
6105            None
6106        }
6107
6108        fn authenticate(&mut self, _auth_event: &Event) -> Result<(), SarahConversationError> {
6109            Ok(())
6110        }
6111
6112        fn publish(&mut self, event: &Event) -> Result<(), SarahConversationError> {
6113            let persisted = fs::read_to_string(&self.state_path)
6114                .map_err(|error| SarahConversationError::Internal(error.to_string()))?;
6115            if !persisted.contains(&event.id.to_hex()) {
6116                return Err(SarahConversationError::Internal(
6117                    "signed outbox was not durable before publish".into(),
6118                ));
6119            }
6120            self.acknowledged_event_id = Some(event.id.to_hex());
6121            Ok(())
6122        }
6123
6124        fn publication_complete(&mut self, _event_id: &str) -> bool {
6125            false
6126        }
6127
6128        fn acknowledged_relays(&self, event_id: &str) -> Vec<String> {
6129            if self.acknowledged_event_id.as_deref() == Some(event_id) {
6130                vec!["wss://healthy.example".into()]
6131            } else {
6132                Vec::new()
6133            }
6134        }
6135
6136        fn query(
6137            &mut self,
6138            _conversation_ref: &str,
6139            _after_cursor: Option<&str>,
6140            _limit: usize,
6141        ) -> Result<QueryPage, SarahConversationError> {
6142            self.query_count.fetch_add(1, Ordering::SeqCst);
6143            Ok(QueryPage {
6144                events: Vec::new(),
6145                next_cursor: None,
6146                gap_state: GapState::Possible,
6147            })
6148        }
6149
6150        fn last_event_id(&self) -> Option<String> {
6151            self.acknowledged_event_id.clone()
6152        }
6153
6154        fn gap_state(&self) -> GapState {
6155            GapState::Possible
6156        }
6157
6158        fn connected_relays(&self) -> Vec<String> {
6159            vec!["wss://healthy.example".into()]
6160        }
6161
6162        fn requires_private_messages(&self) -> bool {
6163            true
6164        }
6165    }
6166
6167    #[test]
6168    fn partial_relay_ack_is_durable_and_does_not_block_querying() {
6169        let temporary = tempfile::tempdir().expect("tempdir");
6170        let state_path = temporary.path().join("owner").join("issue31-state.json");
6171        let query_count = Arc::new(AtomicUsize::new(0));
6172        let signer = SigningIdentity::generate();
6173        let mut config = SarahConversationConfig::mock_fixture();
6174        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
6175        let relay_urls = vec![
6176            "wss://healthy.example".to_string(),
6177            "wss://down.example".to_string(),
6178        ];
6179        let controller = Issue31HostController::new(Issue31HostConfiguration {
6180            host_ref: "omega.host.local".into(),
6181            host_public_key_hex: signer.public_key_hex.clone(),
6182            sarah_public_key_hex: config.identity.sarah_public_key_hex.clone(),
6183            conversation: config.conversation_ref(),
6184            display_name: "Local Omega".into(),
6185            relay_urls,
6186            generation: ISSUE31_NOSTR_HOST_GENERATION,
6187        })
6188        .expect("controller");
6189        let relay = OneHealthyOneDownRelay {
6190            state_path: state_path.clone(),
6191            query_count: query_count.clone(),
6192            connected: false,
6193            acknowledged_event_id: None,
6194        };
6195        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
6196        client.issue31_host = Some(controller);
6197        client.issue31_state_path = Some(state_path.clone());
6198
6199        client.sync_issue31_host().expect("partial relay sync");
6200
6201        assert!(query_count.load(Ordering::SeqCst) > 0);
6202        let persisted: DurableIssue31HostState =
6203            serde_json::from_slice(&fs::read(&state_path).expect("read durable state"))
6204                .expect("decode durable state");
6205        assert!(persisted.discovery_event_json.is_some());
6206        assert_eq!(persisted.relay_acknowledgements.len(), 1);
6207        assert_eq!(
6208            client.current_room_state().connection,
6209            ConnectionState::Degraded
6210        );
6211        assert_eq!(
6212            client.current_room_state().connected_relays,
6213            vec!["wss://healthy.example"]
6214        );
6215    }
6216
6217    #[test]
6218    fn normal_commands_commit_exact_private_outbox_and_result_before_publish() {
6219        let temporary = tempfile::tempdir().expect("tempdir");
6220        let state_path = temporary.path().join("owner").join("issue31-state.json");
6221        let signer = SigningIdentity::generate();
6222        let mut config = SarahConversationConfig::mock_fixture();
6223        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
6224        let sarah_keys = Keys::generate();
6225        config.identity.sarah_public_key_hex = sarah_keys.public_key().to_hex();
6226        config.relay_url = Some("wss://healthy.example".into());
6227        let relay_urls = vec![
6228            "wss://healthy.example".to_string(),
6229            "wss://down.example".to_string(),
6230        ];
6231        let controller = Issue31HostController::new(Issue31HostConfiguration {
6232            host_ref: "omega.host.local".into(),
6233            host_public_key_hex: signer.public_key_hex.clone(),
6234            sarah_public_key_hex: config.identity.sarah_public_key_hex.clone(),
6235            conversation: config.conversation_ref(),
6236            display_name: "Local Omega".into(),
6237            relay_urls,
6238            generation: ISSUE31_NOSTR_HOST_GENERATION,
6239        })
6240        .expect("controller");
6241        let relay = OneHealthyOneDownRelay {
6242            state_path: state_path.clone(),
6243            query_count: Arc::new(AtomicUsize::new(0)),
6244            connected: false,
6245            acknowledged_event_id: None,
6246        };
6247        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
6248        client.issue31_host = Some(controller);
6249        client.issue31_state_path = Some(state_path.clone());
6250
6251        let send_params = json!({
6252            "text": "durable hello",
6253            "idempotencyRef": "idempotency.normal.send",
6254            "expectedGeneration": 1,
6255        });
6256        let first = client
6257            .handle_request(SARAH_METHOD_SEND_MESSAGE, 1, Some(&send_params))
6258            .expect("send");
6259        let replay = client
6260            .handle_request(SARAH_METHOD_SEND_MESSAGE, 1, Some(&send_params))
6261            .expect("idempotent replay");
6262        assert_eq!(first, replay);
6263        assert_eq!(client.message_seq, 1);
6264
6265        let second_send_params = json!({
6266            "text": "durable hello",
6267            "idempotencyRef": "idempotency.normal.send.second",
6268            "expectedGeneration": 1,
6269        });
6270        let second = client
6271            .handle_request(SARAH_METHOD_SEND_MESSAGE, 1, Some(&second_send_params))
6272            .expect("distinct repeated send");
6273        assert_ne!(first.get("eventId"), second.get("eventId"));
6274        assert_eq!(client.message_seq, 2);
6275
6276        let persisted: DurableIssue31HostState =
6277            serde_json::from_slice(&fs::read(&state_path).expect("read durable state"))
6278                .expect("decode durable state");
6279        assert_eq!(persisted.active_turn_ref.as_deref(), Some("turn.2"));
6280        assert_eq!(persisted.run_state, "running");
6281        assert_eq!(persisted.message_seq, 2);
6282        assert!(
6283            persisted
6284                .command_results
6285                .contains_key("idempotency.normal.send")
6286        );
6287        assert!(
6288            persisted
6289                .command_results
6290                .contains_key("idempotency.normal.send.second")
6291        );
6292        assert_eq!(persisted.private_outbox.len(), 2);
6293        let rumor_ids = persisted
6294            .private_outbox
6295            .values()
6296            .map(|pending| pending.rumor_event_id.as_str())
6297            .collect::<std::collections::BTreeSet<_>>();
6298        assert_eq!(rumor_ids.len(), 2);
6299        assert_eq!(persisted.relay_acknowledgements.len(), 4);
6300        let serialized = fs::read_to_string(&state_path).expect("read serialized state");
6301        for pending in persisted.private_outbox.values() {
6302            for gift_wrap in &pending.gift_wrap_event_json {
6303                let event = Event::from_json(gift_wrap).expect("gift wrap");
6304                assert!(serialized.contains(&event.id.to_hex()));
6305            }
6306        }
6307        let mut unwrapped_turn_refs = std::collections::BTreeSet::new();
6308        for pending in persisted.private_outbox.values() {
6309            let mut found_sarah_copy = false;
6310            for gift_wrap in &pending.gift_wrap_event_json {
6311                let event = Event::from_json(gift_wrap).expect("gift wrap");
6312                match smol::block_on(nostr::nips::nip59::extract_rumor(&sarah_keys, &event)) {
6313                    Ok(unwrapped) => {
6314                        found_sarah_copy = true;
6315                        let turn_ref = unwrapped
6316                            .rumor
6317                            .tags
6318                            .iter()
6319                            .find_map(|tag| {
6320                                let tag = tag.as_slice();
6321                                (tag.first().map(String::as_str) == Some("turn"))
6322                                    .then(|| tag.get(1).cloned())
6323                                    .flatten()
6324                            })
6325                            .expect("turn tag in Sarah rumor");
6326                        unwrapped_turn_refs.insert(turn_ref);
6327                    }
6328                    // The other gift wrap is the owner's independently encrypted copy.
6329                    Err(_) => {}
6330                }
6331            }
6332            assert!(found_sarah_copy);
6333        }
6334        assert_eq!(
6335            unwrapped_turn_refs,
6336            std::collections::BTreeSet::from(["turn.1".to_string(), "turn.2".to_string()])
6337        );
6338
6339        let interrupt_params = json!({
6340            "turnRef": "turn.2",
6341            "idempotencyRef": "idempotency.normal.interrupt",
6342            "expectedGeneration": 1,
6343        });
6344        client
6345            .handle_request(SARAH_METHOD_INTERRUPT_TURN, 1, Some(&interrupt_params))
6346            .expect("interrupt");
6347        let persisted: DurableIssue31HostState =
6348            serde_json::from_slice(&fs::read(&state_path).expect("read durable state"))
6349                .expect("decode durable state");
6350        assert_eq!(persisted.run_state, "interrupt_pending");
6351        assert!(
6352            persisted
6353                .command_results
6354                .contains_key("idempotency.normal.interrupt")
6355        );
6356        assert_eq!(persisted.private_outbox.len(), 3);
6357    }
6358
6359    #[test]
6360    fn issue31_control_scan_persists_cursor_and_resumes_after_page_cap() {
6361        let temporary = tempfile::tempdir().expect("tempdir");
6362        let state_path = temporary.path().join("owner").join("issue31-state.json");
6363        let signer = SigningIdentity::generate();
6364        let mut config = SarahConversationConfig::mock_fixture();
6365        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
6366        let conversation_ref = config.conversation_ref();
6367        let mut relay = MockRelayAdapter::new();
6368        for index in 0..513 {
6369            relay.seed_event(StoredConversationEvent {
6370                event_id: format!("{index:064x}"),
6371                kind: 1,
6372                pubkey: signer.public_key_hex.clone(),
6373                created_at: 1_000 + index as u64,
6374                conversation_ref: conversation_ref.clone(),
6375                content_summary: "bounded non-control record".into(),
6376                tags: Vec::new(),
6377                record_kind: "message".into(),
6378                store_index: index,
6379            });
6380        }
6381        let controller = Issue31HostController::new(Issue31HostConfiguration {
6382            host_ref: "omega.host.local".into(),
6383            host_public_key_hex: signer.public_key_hex.clone(),
6384            sarah_public_key_hex: config.identity.sarah_public_key_hex.clone(),
6385            conversation: config.conversation_ref(),
6386            display_name: "Local Omega".into(),
6387            relay_urls: vec!["wss://relay.example.com".into()],
6388            generation: ISSUE31_NOSTR_HOST_GENERATION,
6389        })
6390        .expect("controller");
6391        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
6392        client.issue31_host = Some(controller);
6393        client.issue31_state_path = Some(state_path.clone());
6394
6395        client.sync_issue31_host().expect("first bounded scan");
6396        assert_eq!(client.last_gap_state, GapState::Possible);
6397        let first_expected_cursor = format!("cursor.1511.{:064x}", 511);
6398        assert_eq!(
6399            client.issue31_control_cursor.as_deref(),
6400            Some(first_expected_cursor.as_str())
6401        );
6402        client.sync_issue31_host().expect("resume scan");
6403        let second_expected_cursor = format!("cursor.1512.{:064x}", 512);
6404        assert_eq!(
6405            client.issue31_control_cursor.as_deref(),
6406            Some(second_expected_cursor.as_str())
6407        );
6408        let persisted: DurableIssue31HostState =
6409            serde_json::from_slice(&fs::read(state_path).expect("read durable state"))
6410                .expect("decode durable state");
6411        assert_eq!(persisted.control_cursor, client.issue31_control_cursor);
6412    }
6413
6414    #[test]
6415    fn invalid_issue31_event_is_quarantined_without_starving_later_pairing() {
6416        let temporary = tempfile::tempdir().expect("tempdir");
6417        let state_path = temporary.path().join("owner").join("issue31-state.json");
6418        let signer = SigningIdentity::generate();
6419        let device = Keys::generate();
6420        let device_public_key_hex = device.public_key().to_hex();
6421        let sarah_public_key_hex = Keys::generate().public_key().to_hex();
6422        let mut config = SarahConversationConfig::mock_fixture();
6423        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
6424        config.identity.sarah_public_key_hex = sarah_public_key_hex.clone();
6425        let conversation_ref = config.conversation_ref();
6426        let now = unix_now();
6427        let expired = Issue31PairingRecord::PairingRequest {
6428            schema: ISSUE31_PAIRING_SCHEMA.into(),
6429            host_ref: "omega.host.local".into(),
6430            host_public_key_hex: signer.public_key_hex.clone(),
6431            device_public_key_hex: device_public_key_hex.clone(),
6432            issued_at: 100,
6433            pairing_request_ref: "pairing_request.device.expired".into(),
6434            requested_scopes: vec![crate::Issue31PairingScope::ObserveIssue31],
6435            expires_at: 200,
6436        };
6437        let valid = Issue31PairingRecord::PairingRequest {
6438            schema: ISSUE31_PAIRING_SCHEMA.into(),
6439            host_ref: "omega.host.local".into(),
6440            host_public_key_hex: signer.public_key_hex.clone(),
6441            device_public_key_hex: device_public_key_hex.clone(),
6442            issued_at: now,
6443            pairing_request_ref: "pairing_request.device.valid".into(),
6444            requested_scopes: vec![crate::Issue31PairingScope::ObserveIssue31],
6445            expires_at: now.saturating_add(600),
6446        };
6447        let mut relay = MockRelayAdapter::new();
6448        for (index, (event_id, record)) in [("a".repeat(64), expired), ("b".repeat(64), valid)]
6449            .into_iter()
6450            .enumerate()
6451        {
6452            relay.seed_event(StoredConversationEvent {
6453                event_id,
6454                kind: Kind::PrivateDirectMessage.as_u16(),
6455                pubkey: device_public_key_hex.clone(),
6456                created_at: now.saturating_add(index as u64),
6457                conversation_ref: conversation_ref.clone(),
6458                content_summary: serde_json::to_string(&record).expect("pairing json"),
6459                tags: Vec::new(),
6460                record_kind: "pairing".into(),
6461                store_index: index,
6462            });
6463        }
6464        let mut controller = Issue31HostController::new(Issue31HostConfiguration {
6465            host_ref: "omega.host.local".into(),
6466            host_public_key_hex: signer.public_key_hex.clone(),
6467            sarah_public_key_hex,
6468            conversation: config.conversation_ref(),
6469            display_name: "Local Omega".into(),
6470            relay_urls: vec!["wss://relay.example.com".into()],
6471            generation: ISSUE31_NOSTR_HOST_GENERATION,
6472        })
6473        .expect("controller");
6474        controller
6475            .set_admitted_device_policy(
6476                vec![device_public_key_hex],
6477                vec![crate::Issue31PairingScope::ObserveIssue31],
6478            )
6479            .expect("admit device");
6480        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
6481        client.issue31_host = Some(controller);
6482        client.issue31_state_path = Some(state_path.clone());
6483
6484        client.sync_issue31_host().expect("quarantine and continue");
6485        assert_eq!(
6486            client
6487                .issue31_quarantined_events
6488                .get(&"a".repeat(64))
6489                .map(String::as_str),
6490            Some("reason.omega.pairing_rejected")
6491        );
6492        let controller = client.issue31_host.as_ref().expect("controller");
6493        assert!(controller.pairing_event_was_processed(&"b".repeat(64)));
6494        let persisted: DurableIssue31HostState =
6495            serde_json::from_slice(&fs::read(state_path).expect("read durable state"))
6496                .expect("decode durable state");
6497        assert_eq!(persisted.quarantined_events.len(), 1);
6498        assert!(persisted.control_cursor.is_some());
6499    }
6500
6501    #[test]
6502    fn inbound_pairing_is_retryable_when_atomic_state_and_outbox_commit_fails() {
6503        let temporary = tempfile::tempdir().expect("tempdir");
6504        let state_path = temporary.path().join("owner").join("issue31-state.json");
6505        let signer = SigningIdentity::generate();
6506        let device_public_key_hex = Keys::generate().public_key().to_hex();
6507        let sarah_public_key_hex = Keys::generate().public_key().to_hex();
6508        let mut config = SarahConversationConfig::mock_fixture();
6509        config.identity.owner_public_key_hex = signer.public_key_hex.clone();
6510        config.identity.sarah_public_key_hex = sarah_public_key_hex.clone();
6511        let conversation_ref = config.conversation_ref();
6512        let now = unix_now();
6513        let inbound_event_id = "c".repeat(64);
6514        let request = Issue31PairingRecord::PairingRequest {
6515            schema: ISSUE31_PAIRING_SCHEMA.into(),
6516            host_ref: "omega.host.local".into(),
6517            host_public_key_hex: signer.public_key_hex.clone(),
6518            device_public_key_hex: device_public_key_hex.clone(),
6519            issued_at: now,
6520            pairing_request_ref: "pairing_request.device.retry".into(),
6521            requested_scopes: vec![crate::Issue31PairingScope::ObserveIssue31],
6522            expires_at: now.saturating_add(600),
6523        };
6524        let mut relay = MockRelayAdapter::new();
6525        relay.seed_event(StoredConversationEvent {
6526            event_id: inbound_event_id.clone(),
6527            kind: Kind::PrivateDirectMessage.as_u16(),
6528            pubkey: device_public_key_hex.clone(),
6529            created_at: now,
6530            conversation_ref,
6531            content_summary: serde_json::to_string(&request).expect("pairing json"),
6532            tags: Vec::new(),
6533            record_kind: "pairing".into(),
6534            store_index: 0,
6535        });
6536        let mut controller = Issue31HostController::new(Issue31HostConfiguration {
6537            host_ref: "omega.host.local".into(),
6538            host_public_key_hex: signer.public_key_hex.clone(),
6539            sarah_public_key_hex,
6540            conversation: config.conversation_ref(),
6541            display_name: "Local Omega".into(),
6542            relay_urls: vec!["wss://relay.example.com".into()],
6543            generation: ISSUE31_NOSTR_HOST_GENERATION,
6544        })
6545        .expect("controller");
6546        controller
6547            .set_admitted_device_policy(
6548                vec![device_public_key_hex],
6549                vec![crate::Issue31PairingScope::ObserveIssue31],
6550            )
6551            .expect("device policy");
6552        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
6553        client.issue31_host = Some(controller);
6554        client.issue31_state_path = Some(state_path);
6555        client.issue31_discovery_generation = Some(ISSUE31_NOSTR_HOST_GENERATION);
6556        client.issue31_discovery_expires_at = Some(now.saturating_add(2 * 60 * 60));
6557        client.issue31_fail_commit_after.set(Some(1));
6558
6559        let error = client
6560            .sync_issue31_host()
6561            .expect_err("injected transaction failure");
6562        assert!(
6563            error
6564                .to_string()
6565                .contains("injected Issue 31 durable commit failure")
6566        );
6567        assert!(client.issue31_private_outbox.is_empty());
6568        assert!(
6569            !client
6570                .issue31_host
6571                .as_ref()
6572                .expect("controller")
6573                .pairing_event_was_processed(&inbound_event_id)
6574        );
6575
6576        client.sync_issue31_host().expect("retry pairing");
6577        assert!(
6578            client
6579                .issue31_host
6580                .as_ref()
6581                .expect("controller")
6582                .pairing_event_was_processed(&inbound_event_id)
6583        );
6584    }
6585
6586    #[test]
6587    fn durable_issue31_state_reuses_exact_outbox_and_refuses_revoked_replay_after_restart() {
6588        let temporary = tempfile::tempdir().expect("tempdir");
6589        let state_path = temporary.path().join("owner").join("issue31-state.json");
6590        let (configuration, controller, device_public_key_hex, grant_ref) =
6591            crate::issue31_nostr::restart_fixture();
6592        let signer = ConversationSigner::Keys(SigningIdentity::generate());
6593        let tags = vec![Tag::parse(["p", device_public_key_hex.as_str()]).expect("p tag")];
6594        let (rumor_event_id, gift_wraps) = signer
6595            .private_messages(
6596                "{\"schema\":\"openagents.omega.issue31.command.v1\"}",
6597                tags,
6598                std::slice::from_ref(&device_public_key_hex),
6599            )
6600            .expect("private outbox");
6601        let exact_event_json = gift_wraps[0].try_as_json().expect("event json");
6602        let state = DurableIssue31HostState {
6603            schema: ISSUE31_DURABLE_STATE_SCHEMA.into(),
6604            controller,
6605            discovery_generation: Some(1),
6606            discovery_expires_at: Some(10_000),
6607            discovery_event_json: None,
6608            private_outbox: BTreeMap::from([(
6609                "openagents.omega.issue31.command.v1.test".into(),
6610                DurableIssue31PrivatePublish {
6611                    rumor_event_id,
6612                    gift_wrap_event_json: vec![exact_event_json.clone()],
6613                },
6614            )]),
6615            relay_acknowledgements: BTreeMap::new(),
6616            control_cursor: Some(format!("cursor.10.{}", "f".repeat(64))),
6617            projection_cursor: Some(format!("cursor.11.{}", "e".repeat(64))),
6618            quarantined_events: BTreeMap::from([(
6619                "9".repeat(64),
6620                "reason.omega.invalid_pairing_record".into(),
6621            )]),
6622            host_adjunct_emissions: BTreeMap::new(),
6623            provider_handoffs: Issue31ProviderHandoffLedger::default(),
6624            command_results: BTreeMap::from([(
6625                "idempotency.restart.admin".into(),
6626                ("fingerprint".into(), json!({ "eventId": "1".repeat(64) })),
6627            )]),
6628            active_turn_ref: Some("turn.7".into()),
6629            run_state: "running".into(),
6630            message_seq: 7,
6631        };
6632        write_issue31_host_state(&state_path, &state).expect("write state");
6633        let serialized = fs::read_to_string(&state_path).expect("read state");
6634        assert!(!serialized.contains("nsec"));
6635        #[cfg(unix)]
6636        {
6637            use std::os::unix::fs::PermissionsExt as _;
6638            assert_eq!(
6639                fs::metadata(&state_path)
6640                    .expect("metadata")
6641                    .permissions()
6642                    .mode()
6643                    & 0o777,
6644                0o600
6645            );
6646        }
6647
6648        let loaded = load_issue31_host_state(&state_path, &configuration)
6649            .expect("load state")
6650            .expect("persisted state");
6651        assert_eq!(loaded.active_turn_ref.as_deref(), Some("turn.7"));
6652        assert_eq!(loaded.run_state, "running");
6653        assert_eq!(loaded.message_seq, 7);
6654        let runtime_outbox = durable_private_outbox_into_runtime(
6655            loaded
6656                .private_outbox
6657                .into_iter()
6658                .collect::<BTreeMap<_, _>>(),
6659        )
6660        .expect("runtime outbox");
6661        assert_eq!(
6662            runtime_outbox
6663                .values()
6664                .next()
6665                .expect("outbox item")
6666                .gift_wraps[0]
6667                .try_as_json()
6668                .expect("reloaded event json"),
6669            exact_event_json
6670        );
6671
6672        let mut controller = loaded.controller;
6673        let result = controller
6674            .handle_command_event(
6675                Issue31CommandEvent {
6676                    event_id: "0".repeat(64),
6677                    record: Issue31CommandRecord::CommandIntent {
6678                        schema: ISSUE31_COMMAND_SCHEMA.into(),
6679                        host_ref: configuration.host_ref,
6680                        host_public_key_hex: configuration.host_public_key_hex,
6681                        device_public_key_hex,
6682                        grant_ref,
6683                        action_ref: "action.omega.full_auto.stop".into(),
6684                        idempotency_ref: "idempotency.restart.revoked_replay".into(),
6685                        expected_generation: 1,
6686                        arguments_ref: "arguments.omega.none".into(),
6687                        issued_at: 106,
6688                        expires_at: 200,
6689                    },
6690                },
6691                107,
6692                |_, _, _| panic!("revoked device command executed"),
6693            )
6694            .expect("terminal refusal")
6695            .expect("result record");
6696        assert!(matches!(
6697            result,
6698            Issue31CommandRecord::CommandResult {
6699                status: Issue31CommandStatus::Refused,
6700                ..
6701            }
6702        ));
6703    }
6704
6705    // ---------------------------------------------------------------------
6706    // omega#49 device-proof host
6707    // ---------------------------------------------------------------------
6708
6709    /// Read a `Keys` from a 64-hex secret in the environment.
6710    #[cfg(test)]
6711    fn device_proof_keys(variable: &str) -> Keys {
6712        let secret = std::env::var(variable)
6713            .unwrap_or_else(|_| panic!("{variable} must be a 64-hex secret key"));
6714        Keys::parse(secret.trim()).unwrap_or_else(|error| panic!("{variable}: {error}"))
6715    }
6716
6717    /// Publish one event through a real adapter, meeting the NIP-42 challenge
6718    /// lazily exactly as `publish_with_auth` does.
6719    #[cfg(test)]
6720    /// Sarah's half of the conversation, shaped the way the host actually reads it.
6721    ///
6722    /// The host subscribes to gift wraps addressed to its own custody key and
6723    /// unwraps the kind-14 rumor inside into a `message` record — the identical
6724    /// path the owner's own sends take. A bare kind 14 published straight to
6725    /// the relay is never requested by any filter and never arrives, and a
6726    /// kind 44300 is a `Turn`, which must carry NIP-44 ciphertext of JSON and
6727    /// is not a conversation message at all.
6728    ///
6729    /// `conversation_tags` is the production tag builder, so the rumor names
6730    /// exactly the owner and Sarah and satisfies `require_conversation_recipients`
6731    /// rather than approximating it here.
6732    fn sarah_conversation_wrap(
6733        sarah_keys: &Keys,
6734        owner_public_key: &PublicKey,
6735        owner_public_key_hex: &str,
6736        conversation_ref: &str,
6737        text: &str,
6738    ) -> (nostr::Event, String) {
6739        let tags = conversation_tags(
6740            conversation_ref,
6741            owner_public_key_hex,
6742            &sarah_keys.public_key().to_hex(),
6743        )
6744        .expect("production conversation tags");
6745        let mut rumor = EventBuilder::new(Kind::PrivateDirectMessage, text)
6746            .tags(tags)
6747            .build(sarah_keys.public_key());
6748        rumor.ensure_id();
6749        // The wrap id is not the record id. The host stores the rumor under its
6750        // own id and projects that id to the device, so a harness that reports
6751        // the wrap id is naming an identifier nothing downstream ever uses.
6752        let rumor_id = rumor.id.expect("the rumor carries its id").to_hex();
6753        let wrap = smol::block_on(EventBuilder::gift_wrap(
6754            sarah_keys,
6755            owner_public_key,
6756            rumor,
6757            [],
6758        ))
6759        .expect("gift wrap Sarah's message to the host");
6760        (wrap, rumor_id)
6761    }
6762
6763    /// Publish, and survive a relay that closed an idle socket.
6764    ///
6765    /// The harness holds one Sarah connection for the life of the run and used
6766    /// to publish straight down it. A reply sent minutes after seeding arrived
6767    /// on a socket the relay had already reset, and the harness reported it as
6768    /// Sarah failing to answer — indistinguishable, from the outside, from the
6769    /// host never producing a reply. A one-message proof would have failed on
6770    /// transport before reaching any of the record contracts.
6771    fn device_proof_publish(
6772        relay: &mut crate::nostr_websocket_relay::WebSocketRelayAdapter,
6773        auth_url: &str,
6774        keys: &Keys,
6775        record: &Event,
6776    ) -> Result<(), SarahConversationError> {
6777        match device_proof_publish_authenticated(relay, auth_url, keys, record) {
6778            Ok(()) => Ok(()),
6779            Err(error) => {
6780                eprintln!("device-proof: sarah publish failed ({error}); reconnecting");
6781                relay.connect()?;
6782                // A new socket is a new NIP-42 session, so the auth path runs
6783                // again rather than assuming the old challenge still holds.
6784                device_proof_publish_authenticated(relay, auth_url, keys, record)
6785            }
6786        }
6787    }
6788
6789    fn device_proof_publish_authenticated(
6790        relay: &mut crate::nostr_websocket_relay::WebSocketRelayAdapter,
6791        auth_url: &str,
6792        keys: &Keys,
6793        record: &Event,
6794    ) -> Result<(), SarahConversationError> {
6795        match relay.publish(record) {
6796            Err(SarahConversationError::IdentityRequired) => {
6797                // Every step here returns rather than panics. A relay that
6798                // closed an idle socket fails *inside* this branch — `publish`
6799                // reports `IdentityRequired`, control comes here, and the AUTH
6800                // write hits the reset — so an `expect` on this path takes the
6801                // whole harness down before any retry wrapper can see it, and
6802                // reports a dead socket as Sarah failing to answer.
6803                let challenge = relay.auth_challenge().ok_or_else(|| {
6804                    SarahConversationError::Relay(
6805                        "relay refused the publish without exposing a challenge".into(),
6806                    )
6807                })?;
6808                let auth_event = EventBuilder::new(Kind::Custom(22242), "")
6809                    .tag(Tag::parse(["relay", auth_url]).expect("relay tag"))
6810                    .tag(
6811                        Tag::parse(["challenge", challenge.challenge.as_str()])
6812                            .expect("challenge tag"),
6813                    )
6814                    .sign_with_keys(keys)
6815                    .expect("signed auth event");
6816                relay.authenticate(&auth_event)?;
6817                relay.publish(record)
6818            }
6819            other => other,
6820        }
6821    }
6822
6823    /// The real production issue #31 host pump, run headless against a live relay.
6824    ///
6825    /// This is the host half of the omega#49 device proof. Everything downstream
6826    /// of the signer is the production path: `sync_issue31_host` is the same
6827    /// entry point `bootstrap` calls, so discovery v2, the pairing state machine,
6828    /// command execution, source projection and the withheld-source statement are
6829    /// all the shipped code. The only substitution is identity custody — the
6830    /// owner key is supplied as a keypair rather than drawn from
6831    /// `omega_identity::IdentityService`, because custody needs the GPUI app and
6832    /// this harness must run without a window. That substitution is recorded on
6833    /// the issue rather than hidden here: a run of this harness proves the host
6834    /// protocol, not owner key custody.
6835    ///
6836    /// Sarah's own turns are authored by a second keypair the harness holds. The
6837    /// admitted OpenAgents turn service is out of scope for a device proof and is
6838    /// deliberately not stood up; a reply published this way is a real signed
6839    /// Sarah record on a real relay, and is not evidence that the turn service
6840    /// produced it.
6841    ///
6842    /// ```sh
6843    /// OMEGA_DEVICE_PROOF_RELAY=wss://relay.openagents.com \
6844    /// OMEGA_DEVICE_PROOF_HOST_SECRET=<64 hex> \
6845    /// OMEGA_DEVICE_PROOF_SARAH_SECRET=<64 hex> \
6846    /// OMEGA_DEVICE_PROOF_DEVICE_PUBKEY=<64 hex> \
6847    /// OMEGA_DEVICE_PROOF_STATE=/path/to/state.json \
6848    ///   cargo test -p omega_effectd --lib live_device_proof_host -- --ignored --nocapture
6849    /// ```
6850    /// omega#49 / omega#46 exit 1: a device sends to Sarah and the host must
6851    /// publish the owner record the device is waiting on.
6852    ///
6853    /// The phone showed the correct pending state and no reply ever arrived.
6854    /// The command was admitted and never quarantined, so the failure was
6855    /// downstream of admission, inside `execute_issue31_action_v2`:
6856    /// `send_message` publishes the owner rumor, `enqueue_issue31_source_projection`
6857    /// then has to read it back off the relay to confirm it, and the read-back
6858    /// refused the host's own message — see `require_conversation_recipients`.
6859    /// The result degraded to `reason.omega.projection_failed` and no owner
6860    /// projection was ever enqueued for the device.
6861    ///
6862    /// This drives the real action path against a real relay, so the assertion
6863    /// is on the same bytes a phone waits for rather than on a substitute.
6864    #[test]
6865    #[ignore = "requires a live relay; set OMEGA_LIVE_RELAY_URL"]
6866    fn owner_send_confirms_and_projects_to_the_device_on_a_live_relay() {
6867        let Ok(relay_url) = std::env::var("OMEGA_LIVE_RELAY_URL") else {
6868            eprintln!("OMEGA_LIVE_RELAY_URL unset; skipping");
6869            return;
6870        };
6871        let temporary = tempfile::tempdir().expect("tempdir");
6872        let host_keys = Keys::generate();
6873        let sarah_keys = Keys::generate();
6874        let device_public_key_hex = Keys::generate().public_key().to_hex();
6875        let signer = SigningIdentity::from_keys(host_keys.clone());
6876        let owner_public_key_hex = signer.public_key_hex.clone();
6877        let mut config = SarahConversationConfig::mock_fixture();
6878        config.identity.owner_public_key_hex = owner_public_key_hex.clone();
6879        config.identity.sarah_public_key_hex = sarah_keys.public_key().to_hex();
6880        config.conversation_digest = owner_public_key_hex[..24].to_string();
6881        config.relay_url = Some(relay_url.clone());
6882        let conversation_ref = config.conversation_ref();
6883
6884        let relay = crate::nostr_websocket_relay::WebSocketRelayAdapter::new_for_keys_with_policy(
6885            vec![relay_url.clone()],
6886            host_keys,
6887            sarah_keys.public_key().to_hex(),
6888            Vec::new(),
6889            Vec::new(),
6890        )
6891        .expect("host adapter");
6892        let controller = Issue31HostController::new(Issue31HostConfiguration {
6893            host_ref: "omega.host.local".into(),
6894            host_public_key_hex: owner_public_key_hex.clone(),
6895            sarah_public_key_hex: sarah_keys.public_key().to_hex(),
6896            conversation: conversation_ref.clone(),
6897            display_name: "Local Omega".into(),
6898            relay_urls: vec![relay_url],
6899            generation: ISSUE31_NOSTR_HOST_GENERATION,
6900        })
6901        .expect("host controller");
6902        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
6903        client.issue31_state_path = Some(temporary.path().join("issue31-state.json"));
6904        // The production caller binds the controller around the executor; the
6905        // durable commit inside `send_message` requires it.
6906        client.issue31_host = Some(controller);
6907
6908        let execution = client.execute_issue31_action_v2(
6909            &Issue31CommandArguments::SendMessage {
6910                action_ref: crate::ISSUE31_ACTION_SEND_MESSAGE.into(),
6911                conversation: conversation_ref,
6912                text: "Does a send from a paired device confirm?".into(),
6913            },
6914            "idempotency.issue31.send_message:liveconfirm",
6915            "grant.omega.liveconfirm",
6916            &device_public_key_hex,
6917            1,
6918        );
6919        assert_eq!(
6920            execution.status,
6921            Issue31CommandHandlingStatus::Accepted,
6922            "the host must accept its own send; reason {:?}",
6923            execution.reason_ref
6924        );
6925        let source_event_id = execution
6926            .source_event_id
6927            .expect("an accepted send must name the owner record it produced");
6928
6929        // The owner record is not merely published — it is readable back off
6930        // the relay, which is the step that was failing.
6931        let source = client
6932            .load_issue31_source_event(&source_event_id)
6933            .expect("the owner's own message must be readable back off the relay");
6934        assert_eq!(source.pubkey, owner_public_key_hex);
6935        assert_eq!(source.record_kind, "message");
6936        assert_eq!(source.kind, crate::ISSUE31_PRIVATE_RUMOR_KIND);
6937
6938        // And the device has an owner projection waiting for it, addressed to
6939        // the device key rather than to either conversation participant.
6940        assert!(
6941            !client.issue31_private_outbox.is_empty(),
6942            "an accepted send must enqueue the owner projection the device renders"
6943        );
6944    }
6945
6946    #[test]
6947    #[ignore = "device proof host; set OMEGA_DEVICE_PROOF_RELAY"]
6948    fn live_device_proof_host() {
6949        let Ok(relay_url) = std::env::var("OMEGA_DEVICE_PROOF_RELAY") else {
6950            eprintln!("OMEGA_DEVICE_PROOF_RELAY unset; skipping");
6951            return;
6952        };
6953        let auth_url =
6954            std::env::var("OMEGA_DEVICE_PROOF_AUTH_URL").unwrap_or_else(|_| relay_url.clone());
6955        let host_keys = device_proof_keys("OMEGA_DEVICE_PROOF_HOST_SECRET");
6956        let sarah_keys = device_proof_keys("OMEGA_DEVICE_PROOF_SARAH_SECRET");
6957        // Comma-separated, so one host can admit an iOS surface and an Android
6958        // surface at the same time. omega#49 asks for a result on both, and a
6959        // host that can only ever hold one grant proves the fan-out by
6960        // assertion rather than by running it.
6961        let device_public_key_hexes: Vec<String> = std::env::var("OMEGA_DEVICE_PROOF_DEVICE_PUBKEY")
6962            .expect("OMEGA_DEVICE_PROOF_DEVICE_PUBKEY must be the surfaces' 64-hex device keys")
6963            .split(',')
6964            .map(|value| value.trim().to_string())
6965            .filter(|value| !value.is_empty())
6966            .collect();
6967        assert!(
6968            !device_public_key_hexes.is_empty(),
6969            "OMEGA_DEVICE_PROOF_DEVICE_PUBKEY must name at least one device key"
6970        );
6971        let state_path = std::path::PathBuf::from(
6972            std::env::var("OMEGA_DEVICE_PROOF_STATE")
6973                .expect("OMEGA_DEVICE_PROOF_STATE must be a durable state file path"),
6974        );
6975        let seconds: u64 = std::env::var("OMEGA_DEVICE_PROOF_SECONDS")
6976            .ok()
6977            .and_then(|value| value.parse().ok())
6978            .unwrap_or(180);
6979        let revoke_after: Option<u64> = std::env::var("OMEGA_DEVICE_PROOF_REVOKE_AFTER")
6980            .ok()
6981            .and_then(|value| value.parse().ok());
6982        let seed = std::env::var("OMEGA_DEVICE_PROOF_SEED").is_ok();
6983        let quarantine = std::env::var("OMEGA_DEVICE_PROOF_QUARANTINE").is_ok();
6984
6985        let signer = SigningIdentity::from_keys(host_keys.clone());
6986        let owner_public_key_hex = signer.public_key_hex.clone();
6987        let owner_public_key = PublicKey::from_hex(&owner_public_key_hex).expect("owner key");
6988        let sarah_public_key_hex = sarah_keys.public_key().to_hex();
6989        let mut config = SarahConversationConfig::mock_fixture();
6990        config.identity.owner_public_key_hex = owner_public_key_hex.clone();
6991        config.identity.sarah_public_key_hex = sarah_public_key_hex.clone();
6992        config.conversation_digest = owner_public_key_hex[..24].to_string();
6993        config.relay_url = Some(relay_url.clone());
6994        config.admitted_device_public_key_hexes = device_public_key_hexes.clone();
6995        config.approved_device_scopes = vec![
6996            crate::Issue31PairingScope::ObserveIssue31,
6997            crate::Issue31PairingScope::SendMessage,
6998            crate::Issue31PairingScope::InterruptTurn,
6999        ];
7000        let conversation_ref = config.conversation_ref();
7001
7002        // Seed the owner-private sources before the host starts, so the very
7003        // first projection pass has something to fan out.
7004        // Sarah publishes for the whole run, not only during the seed, so a
7005        // send from the device can be answered while the harness is live.
7006        let mut sarah_publisher = {
7007            let mut sarah_relay = crate::nostr_websocket_relay::WebSocketRelayAdapter::new_for_keys(
7008                vec![relay_url.clone()],
7009                sarah_keys.clone(),
7010            )
7011            .expect("sarah adapter");
7012            sarah_relay.connect().expect("sarah connect");
7013            Some(sarah_relay)
7014        };
7015        // The exact sources this run seeded, so the loop can say whether each
7016        // one reached each device rather than leaving "the host admits it and
7017        // never projects it" to be inferred from a device screenshot.
7018        let mut seeded_sources: Vec<(&'static str, String)> = Vec::new();
7019
7020        if seed {
7021            let sarah_relay = sarah_publisher.as_mut().expect("sarah adapter");
7022            let now = unix_now();
7023            let (greeting, greeting_rumor_id) = sarah_conversation_wrap(
7024                &sarah_keys,
7025                &owner_public_key,
7026                &owner_public_key_hex,
7027                &conversation_ref,
7028                "Sarah here. This reply crossed a real relay from a real Omega host.",
7029            );
7030            device_proof_publish(sarah_relay, &auth_url, &sarah_keys, &greeting)
7031                .expect("publish the Sarah greeting");
7032            eprintln!("device-proof: seeded Sarah message {greeting_rumor_id}");
7033            seeded_sources.push(("greeting", greeting_rumor_id));
7034
7035            // One encrypted engram, so "inspect memory" has a real source.
7036            // An engram body is a JSON object with a `mem/…` slug and a value,
7037            // not a sentence. Seeded as prose, the host received it, refused
7038            // the body, and quarantined it — so "inspect memory" had no real
7039            // source and the quarantine count was measuring this bug.
7040            let engram_plaintext = serde_json::json!({
7041                "slug": "mem/owner_reporting_preference",
7042                "value": "Chris prefers evidence-bound reports over confident summaries.",
7043            })
7044            .to_string();
7045            let engram_ciphertext = nip44::encrypt(
7046                sarah_keys.secret_key(),
7047                &owner_public_key,
7048                engram_plaintext.as_str(),
7049                nip44::Version::default(),
7050            )
7051            .expect("nip44 encrypt the engram");
7052            let engram = EventBuilder::new(
7053                Kind::Custom(crate::SARAH_ENGRAM_KIND),
7054                engram_ciphertext,
7055            )
7056            .tag(Tag::parse(["d", &"1".repeat(64)]).expect("d tag"))
7057            .tag(Tag::parse(["p", owner_public_key_hex.as_str()]).expect("p tag"))
7058            .tag(Tag::parse(["alt", "encrypted agent memory record"]).expect("alt tag"))
7059            .tag(Tag::parse(["conversation", conversation_ref.as_str()]).expect("conversation tag"))
7060            .custom_created_at(nostr::Timestamp::from(now))
7061            .sign_with_keys(&sarah_keys)
7062            .expect("signed engram");
7063            device_proof_publish(sarah_relay, &auth_url, &sarah_keys, &engram)
7064                .expect("publish the engram");
7065            eprintln!("device-proof: seeded engram {}", engram.id.to_hex());
7066            seeded_sources.push(("engram", engram.id.to_hex()));
7067
7068            if quarantine {
7069                // Inside every record-level bound and outside the projection body
7070                // contract, so the host quarantines it and must say so.
7071                // Also 44300: a source the host never receives cannot be
7072                // quarantined, so on kind 14 this proved nothing either.
7073                let unreadable = EventBuilder::new(
7074                    Kind::Custom(SARAH_TURN_RECORD_KIND),
7075                    "",
7076                )
7077                .tag(
7078                    Tag::parse(["conversation", conversation_ref.as_str()])
7079                        .expect("conversation tag"),
7080                )
7081                .custom_created_at(nostr::Timestamp::from(now + 1))
7082                .sign_with_keys(&sarah_keys)
7083                .expect("signed unreadable source");
7084                device_proof_publish(sarah_relay, &auth_url, &sarah_keys, &unreadable)
7085                    .expect("publish the unreadable source");
7086                eprintln!(
7087                    "device-proof: seeded quarantine source {}",
7088                    unreadable.id.to_hex()
7089                );
7090            }
7091        }
7092
7093        // The host reads with the owner key in custody and Sarah as the *other*
7094        // participant. `new_for_keys` collapses both roles onto one key, which
7095        // silently narrows the read: the `authors` filter becomes
7096        // `[host, host]`, so nothing Sarah signs is ever requested, and the
7097        // conversation participant set stops matching the records on the wire.
7098        let relay = crate::nostr_websocket_relay::WebSocketRelayAdapter::new_for_keys_with_policy(
7099            vec![relay_url.clone()],
7100            host_keys,
7101            sarah_public_key_hex.clone(),
7102            Vec::new(),
7103            Vec::new(),
7104        )
7105        .expect("host adapter");
7106        let host_configuration = Issue31HostConfiguration {
7107            host_ref: "omega.host.local".into(),
7108            host_public_key_hex: owner_public_key_hex.clone(),
7109            sarah_public_key_hex: sarah_public_key_hex.clone(),
7110            conversation: conversation_ref.clone(),
7111            display_name: "Local Omega".into(),
7112            relay_urls: vec![relay_url.clone()],
7113            generation: ISSUE31_NOSTR_HOST_GENERATION,
7114        };
7115        let persisted =
7116            load_issue31_host_state(&state_path, &host_configuration).expect("load durable state");
7117        let mut controller = match persisted {
7118            Some(persisted) => {
7119                eprintln!("device-proof: resumed durable host state at {state_path:?}");
7120                persisted.controller
7121            }
7122            None => Issue31HostController::new(host_configuration).expect("host controller"),
7123        };
7124        controller
7125            .set_admitted_device_policy(
7126                config.admitted_device_public_key_hexes.clone(),
7127                config.approved_device_scopes.clone(),
7128            )
7129            .expect("admit the device");
7130
7131        let mut client = SarahConversationClient::with_relay(config, Box::new(relay), signer);
7132        client.issue31_host = Some(controller);
7133        client.issue31_state_path = Some(state_path);
7134
7135        eprintln!("device-proof: host  npub/hex {owner_public_key_hex}");
7136        eprintln!("device-proof: sarah hex      {sarah_public_key_hex}");
7137        for device_public_key_hex in &device_public_key_hexes {
7138            eprintln!("device-proof: device hex     {device_public_key_hex}");
7139        }
7140        eprintln!("device-proof: conversation   {conversation_ref}");
7141        eprintln!("device-proof: relay          {relay_url}");
7142
7143        let started = std::time::Instant::now();
7144        let mut revoked = false;
7145        let mut announced_grants: BTreeMap<String, String> = BTreeMap::new();
7146        // The previous run of this harness could see that a command had been
7147        // admitted and not quarantined, and could not see what the host then
7148        // did with it. That is the whole distance between "the send failed" and
7149        // "the send failed inside `send_message` with `projection_failed`", so
7150        // the accepted-send counter and Sarah's answer are both surfaced here.
7151        let mut announced_message_seq = client.message_seq;
7152        while started.elapsed().as_secs() < seconds {
7153            match client.sync_issue31_host() {
7154                Ok(()) => {}
7155                Err(error) => eprintln!("device-proof: sync error {error}"),
7156            }
7157            if client.message_seq != announced_message_seq {
7158                announced_message_seq = client.message_seq;
7159                let turn_ref = client
7160                    .active_turn_ref
7161                    .clone()
7162                    .unwrap_or_else(|| format!("turn.{announced_message_seq}"));
7163                eprintln!(
7164                    "device-proof: OWNER SEND ACCEPTED · message_seq {announced_message_seq} · {turn_ref} · run_state {}",
7165                    client.run_state
7166                );
7167                if let Some(sarah_relay) = sarah_publisher.as_mut() {
7168                    // Sarah's half. This is the harness's own keypair, not the
7169                    // admitted OpenAgents turn service: it produces a real
7170                    // signed Sarah record on a real relay, and is not evidence
7171                    // that the turn service produced it.
7172                    // A conversation message reaches the host the same way the
7173                    // owner's own does: a NIP-59 gift wrap addressed to the
7174                    // host's custody key, carrying a kind-14 rumor. This
7175                    // harness published a bare kind 14 straight to the relay,
7176                    // which the host never subscribes to, so every reply it
7177                    // reported sending was discarded in transit and the reply
7178                    // arm of this proof had never once run.
7179                    let (reply, _reply_rumor_id) = sarah_conversation_wrap(
7180                        &sarah_keys,
7181                        &owner_public_key,
7182                        &owner_public_key_hex,
7183                        &conversation_ref,
7184                        &format!(
7185                            "Received. Answering {turn_ref} from a real Omega host over a real relay."
7186                        ),
7187                    );
7188                    match device_proof_publish(sarah_relay, &auth_url, &sarah_keys, &reply) {
7189                        Ok(()) => eprintln!(
7190                            "device-proof: SARAH REPLIED {} · {turn_ref}",
7191                            &reply.id.to_hex()[..16]
7192                        ),
7193                        Err(error) => eprintln!("device-proof: Sarah reply failed {error}"),
7194                    }
7195                }
7196            }
7197            let now = unix_now();
7198            for (event_id, reason) in &client.issue31_quarantined_events {
7199                if announced_grants
7200                    .insert(format!("quarantine:{event_id}"), reason.clone())
7201                    .as_ref()
7202                    != Some(reason)
7203                {
7204                    eprintln!("device-proof: QUARANTINED {} · {reason}", &event_id[..16]);
7205                }
7206            }
7207            for (key, substance) in &client.issue31_withheld_emissions {
7208                let line = format!("{substance:?}");
7209                if announced_grants.insert(format!("withheld:{key}"), line.clone()).as_ref()
7210                    != Some(&line)
7211                {
7212                    eprintln!("device-proof: withheld {key} · {line}");
7213                }
7214            }
7215            if let Some(host) = client.issue31_host.as_ref() {
7216                for projection in host.grant_projections(now).unwrap_or_default() {
7217                    let line = format!(
7218                        "{} · {} · generation {} · scopes {:?}",
7219                        projection.device_fingerprint,
7220                        projection.status,
7221                        projection.generation,
7222                        projection.scopes
7223                    );
7224                    if announced_grants.get(&projection.grant_ref) != Some(&line) {
7225                        eprintln!("device-proof: grant {} · {line}", projection.grant_ref);
7226                        announced_grants.insert(projection.grant_ref.clone(), line);
7227                    }
7228                    // Whether each seeded source actually reached this grant.
7229                    // "The host admits the engram and never projects it" was
7230                    // read off a device surface; the host can state it directly.
7231                    for (name, source_event_id) in &seeded_sources {
7232                        let projected = host.source_was_projected(
7233                            &projection.grant_ref,
7234                            projection.generation,
7235                            source_event_id,
7236                        );
7237                        if !projected {
7238                            continue;
7239                        }
7240                        let key = format!("projected:{}:{name}", projection.grant_ref);
7241                        if announced_grants
7242                            .insert(key, source_event_id.clone())
7243                            .as_ref()
7244                            != Some(source_event_id)
7245                        {
7246                            eprintln!(
7247                                "device-proof: PROJECTED {name} {} · {}",
7248                                &source_event_id[..16],
7249                                projection.grant_ref
7250                            );
7251                        }
7252                    }
7253                }
7254            }
7255            if let Some(revoke_after) = revoke_after {
7256                if !revoked && started.elapsed().as_secs() >= revoke_after {
7257                    let grant_refs: Vec<String> = client
7258                        .issue31_host
7259                        .as_ref()
7260                        .map(|host| {
7261                            host.grant_projections(now)
7262                                .unwrap_or_default()
7263                                .into_iter()
7264                                .filter(|projection| projection.status == "active")
7265                                .map(|projection| projection.grant_ref)
7266                                .collect()
7267                        })
7268                        .unwrap_or_default();
7269                    for grant_ref in grant_refs {
7270                        match client.revoke_issue31_grant(
7271                            &grant_ref,
7272                            Some("reason.omega.owner_revoked".to_string()),
7273                            format!("idempotency.device_proof.revoke.{grant_ref}"),
7274                            grant_ref.clone(),
7275                        ) {
7276                            Ok(_) => eprintln!("device-proof: REVOKED {grant_ref}"),
7277                            Err(error) => eprintln!("device-proof: revoke failed {error}"),
7278                        }
7279                        revoked = true;
7280                    }
7281                }
7282            }
7283            std::thread::sleep(std::time::Duration::from_secs(3));
7284        }
7285        eprintln!("device-proof: host stopped after {seconds}s");
7286    }
7287
7288}
7289