[Fix] User States

mtp-type-maps

@@ -1 +1 @@
-Subproject commit f753cfd406fc0050a8ccd76a90897205efceacd8
+Subproject commit 486541b9483356ff49ff3ec7016f87d3ecbeaa0e

src/anonymous_clients/anonymous_client_connection.rs

@@ -8,7 +8,6 @@ use uuid::Uuid;
 use crate::anonymous_clients::anonymous_manager::{self, generate_username};
 use crate::app_state::AppState;
 use crate::calls::call_group::call_invite_secret_from_cv;
-use crate::data::user::UserStatus;
 use crate::rho::connection::{GeneralConnection, MtpReceiver, MtpSender};
 use crate::util::data_type_id;
 use crate::util::logger::PrintType;
@@ -222,9 +221,13 @@ impl AnonymousClientConnection {
                     .await;
             }
 
-            // Handle client status changes
+            // Presence is account-scoped and anonymous sessions have no
-            if cv.is_type(CommunicationType::ClientChanged) {
+            // persisted account preference to change.
-                self.handle_client_changed(cv).await;
+            if cv.is_type(CommunicationType::SetUserState)
+                || cv.is_type(CommunicationType::ClientChanged)
+            {
+                self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId)
+                    .await;
                 return;
             }
 
@@ -321,22 +324,6 @@ impl AnonymousClientConnection {
         });
     }
 
-    /// Handle client status change
-    async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
-        if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) {
-            let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
-            self.state
-                .omega
-                .client_changed(
-                    self.user_id as i64,
-                    self.user_id as i64,
-                    self.user_id as i64,
-                    user_status,
-                )
-                .await;
-        }
-    }
-
     /// Handle call invite
     async fn handle_call_invite(self: Arc<Self>, cv: CommunicationValue) {
         let receiver_id: i64 = cv.get_data(DataType::ReceiverId).as_number().unwrap_or(0) as i64;

src/config.rs

@@ -1,10 +1,12 @@
-use std::env;
+use std::{env, time::Duration};
 
 use thiserror::Error;
 
 const DEFAULT_RHO_PORT: u16 = 443;
 const DEFAULT_OMEGA_HOST: &str = "tensamin.net";
 const DEFAULT_OMEGA_PORT: u16 = 9187;
+const DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS: u64 = 20;
+const DEFAULT_OMEGA_SYNC_RETRIES: u32 = 3;
 
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct LiveKitConfig {
@@ -19,6 +21,11 @@ pub struct Config {
     pub omega_host: String,
     pub omega_port: u16,
     pub omikron_id: u64,
+    /// Maximum duration of one route or subscription synchronization request.
+    pub omega_sync_timeout: Duration,
+    /// Number of synchronization requests before the transport is closed and
+    /// the normal reconnect loop starts.
+    pub omega_sync_retries: u32,
     pub livekit: Option<LiveKitConfig>,
 }
 
@@ -38,6 +45,15 @@ impl Config {
         let rho_port = parse_or_default("RHO_PORT", DEFAULT_RHO_PORT)?;
         let omega_port = parse_or_default("OMEGA_PORT", DEFAULT_OMEGA_PORT)?;
         let omikron_id = parse_or_default("ID", 0_u64)?;
+        // Each synchronization request is bounded by this timeout. After
+        // omega_sync_retries attempts, Omikron closes the authenticated
+        // transport so the normal reconnect loop can establish a clean state.
+        let omega_sync_timeout = Duration::from_secs(parse_or_default(
+            "OMEGA_SYNC_TIMEOUT_SECONDS",
+            DEFAULT_OMEGA_SYNC_TIMEOUT_SECONDS,
+        )?);
+        let omega_sync_retries =
+            parse_or_default("OMEGA_SYNC_RETRIES", DEFAULT_OMEGA_SYNC_RETRIES)?.max(1);
         let omega_host = env::var("OMEGA_HOST")
             .unwrap_or_else(|_| DEFAULT_OMEGA_HOST.to_string())
             .trim()
@@ -55,6 +71,8 @@ impl Config {
             omega_host,
             omega_port,
             omikron_id,
+            omega_sync_timeout,
+            omega_sync_retries,
             livekit: livekit_from_environment()?,
         })
     }

src/data/user.rs

@@ -1,7 +1,4 @@
-use strum::IntoEnumIterator;
+#[derive(Debug, Clone, PartialEq, Eq)]
-use strum_macros::EnumIter;
-
-#[derive(Debug, Clone, PartialEq, EnumIter, Eq)]
 #[allow(unused, non_camel_case_types)]
 pub enum UserStatus {
     user_offline,
@@ -20,12 +17,40 @@ impl UserStatus {
     pub fn to_string(&self) -> String {
         format!("{:?}", self)
     }
-    pub fn from_str(s: &str) -> Option<UserStatus> {
+    pub fn from_client_preference(s: &str) -> Option<Self> {
-        for sel in UserStatus::iter() {
+        match s {
-            if &sel.to_string() == s {
+            "user_online" => Some(Self::user_online),
-                return Some(sel);
+            "user_idle" => Some(Self::user_idle),
+            "user_dnd" => Some(Self::user_dnd),
+            "user_wc" => Some(Self::user_wc),
+            "user_invisible" => Some(Self::user_invisible),
+            _ => None,
         }
     }
-        None
+
+    pub fn public_value(&self) -> Self {
+        match self {
+            Self::user_invisible => Self::user_offline,
+            value => value.clone(),
+        }
+    }
+
+    pub fn from_str(s: &str) -> Option<Self> {
+        Self::from_client_preference(s)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::UserStatus;
+
+    #[test]
+    fn network_parser_rejects_derived_states() {
+        assert_eq!(UserStatus::from_str("user_offline"), None);
+        assert_eq!(UserStatus::from_str("iota_offline"), None);
+        assert_eq!(
+            UserStatus::from_str("user_online"),
+            Some(UserStatus::user_online)
+        );
     }
 }

src/omega/capabilities.rs

@@ -0,0 +1,145 @@
+use std::collections::BTreeSet;
+
+const OMIKRON_PREFIX: &str = "omikron;caps=";
+const OMEGA_PREFIX: &str = "omega;caps=";
+const SET_USER_STATE: &str = "set_user_state_v1";
+const STATE_SUBSCRIBE: &str = "state_subscribe_v1";
+const SESSION_SNAPSHOT: &str = "session_snapshot_v1";
+const CLIENT_STATE_PUSH: &str = "client_state_push_v1";
+
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct PeerCapabilities {
+    pub set_user_state_v1: bool,
+    pub state_subscribe_v1: bool,
+    pub session_snapshot_v1: bool,
+    pub client_state_push_v1: bool,
+}
+
+impl PeerCapabilities {
+    pub fn current() -> Self {
+        Self {
+            set_user_state_v1: true,
+            state_subscribe_v1: true,
+            session_snapshot_v1: true,
+            client_state_push_v1: true,
+        }
+    }
+
+    pub fn identification_description(&self) -> String {
+        let mut names = Vec::new();
+        if self.set_user_state_v1 {
+            names.push(SET_USER_STATE);
+        }
+        if self.state_subscribe_v1 {
+            names.push(STATE_SUBSCRIBE);
+        }
+        if self.session_snapshot_v1 {
+            names.push(SESSION_SNAPSHOT);
+        }
+        if self.client_state_push_v1 {
+            names.push(CLIENT_STATE_PUSH);
+        }
+        format!("{OMIKRON_PREFIX}{}", names.join(","))
+    }
+
+    pub fn from_identification_description(description: Option<&str>) -> Result<Self, ()> {
+        parse_capabilities(description, OMIKRON_PREFIX)
+    }
+
+    pub fn from_omega_identification_description(description: Option<&str>) -> Result<Self, ()> {
+        // A missing Omega descriptor selects the documented legacy wire
+        // contracts: tuple route snapshots, GetStates-only subscriptions,
+        // and ClientChanged pushes.
+        parse_capabilities(description, OMEGA_PREFIX)
+    }
+}
+
+fn parse_capabilities(description: Option<&str>, prefix: &str) -> Result<PeerCapabilities, ()> {
+    let Some(description) = description else {
+        return Ok(PeerCapabilities::default());
+    };
+    if description == "omega" {
+        return Ok(PeerCapabilities::default());
+    }
+    let Some(capabilities) = description.strip_prefix(prefix) else {
+        return Err(());
+    };
+    let mut seen = BTreeSet::new();
+    for capability in capabilities.split(',') {
+        if capability.is_empty() || !seen.insert(capability) {
+            return Err(());
+        }
+    }
+    if seen.iter().any(|capability| {
+        !matches!(
+            *capability,
+            SET_USER_STATE | STATE_SUBSCRIBE | SESSION_SNAPSHOT | CLIENT_STATE_PUSH
+        )
+    }) {
+        return Err(());
+    }
+    Ok(PeerCapabilities {
+        set_user_state_v1: seen.contains(SET_USER_STATE),
+        state_subscribe_v1: seen.contains(STATE_SUBSCRIBE),
+        session_snapshot_v1: seen.contains(SESSION_SNAPSHOT),
+        client_state_push_v1: seen.contains(CLIENT_STATE_PUSH),
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn capability_description_round_trips() {
+        let capabilities = PeerCapabilities::current();
+        assert_eq!(
+            PeerCapabilities::from_identification_description(Some(
+                &capabilities.identification_description()
+            )),
+            Ok(capabilities)
+        );
+    }
+
+    #[test]
+    fn malformed_capabilities_are_rejected() {
+        assert!(
+            PeerCapabilities::from_identification_description(Some("omikron;caps=unknown_v9"))
+                .is_err()
+        );
+        assert!(
+            PeerCapabilities::from_identification_description(Some(
+                "omikron;caps=state_subscribe_v1,state_subscribe_v1"
+            ))
+            .is_err()
+        );
+        assert!(PeerCapabilities::from_identification_description(Some("omikron;caps=")).is_err());
+    }
+
+    #[test]
+    fn missing_capabilities_mean_legacy_peer() {
+        assert_eq!(
+            PeerCapabilities::from_identification_description(None),
+            Ok(PeerCapabilities::default())
+        );
+        assert_eq!(
+            PeerCapabilities::from_omega_identification_description(Some("omega")),
+            Ok(PeerCapabilities::default())
+        );
+    }
+
+    #[test]
+    fn omega_capabilities_use_the_omega_namespace() {
+        assert!(
+            PeerCapabilities::from_omega_identification_description(Some(
+                "omega;caps=set_user_state_v1"
+            ))
+            .unwrap()
+            .set_user_state_v1
+        );
+        assert!(
+            PeerCapabilities::from_identification_description(Some("omega;caps=set_user_state_v1"))
+                .is_err()
+        );
+    }
+}

src/omega/mod.rs

@@ -1 +1,2 @@
+pub mod capabilities;
 pub mod omega_connection;

src/omega/omega_connection.rs

@@ -1,6 +1,9 @@
+use super::capabilities::PeerCapabilities;
 use crate::{
-    config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, log_out,
+    config::Config,
-    rho::rho_manager::RhoManager, util::logger::PrintType,
+    log_cv_in, log_cv_out, log_err, log_in, log_out,
+    rho::rho_manager::RhoManager,
+    util::{data_type_id, logger::PrintType},
 };
 use dashmap::DashMap;
 use mtp::client::{Client, MTPConnection, Sender};
@@ -13,7 +16,7 @@ use mtp::{
 use mtp_transport::ConnectionHandle;
 use std::{sync::Arc, time::Duration};
 use tokio::{
-    sync::{Mutex, RwLock, mpsc, watch},
+    sync::{Mutex, RwLock, mpsc, oneshot, watch},
     task::JoinHandle,
     time::{Instant, sleep},
 };
@@ -23,6 +26,7 @@ use uuid::Uuid;
 const RECONNECT_DELAY: Duration = Duration::from_secs(5);
 const MAX_RECONNECT_DELAY: Duration = Duration::from_secs(300);
 const CONNECTION_TIMEOUT: Duration = Duration::from_secs(10);
+const CAPABILITY_NEGOTIATION_TIMEOUT: Duration = Duration::from_secs(1);
 const PING_INTERVAL: Duration = Duration::from_secs(5);
 const TASK_CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
 const TASK_MAX_AGE: Duration = Duration::from_secs(60);
@@ -30,6 +34,37 @@ const MAX_CONCURRENT_REQUESTS: usize = 128;
 const CIRCUIT_BREAKER_FAILURE_THRESHOLD: u32 = 3;
 const CIRCUIT_BREAKER_COOLDOWN: Duration = Duration::from_secs(30);
 
+fn client_changed_target(value: &CommunicationValue) -> Option<(i64, i64)> {
+    if !value.is_type(CommunicationType::ClientChanged) {
+        return None;
+    }
+    let receiver = i64::try_from(value.get_receiver())
+        .ok()
+        .filter(|id| *id > 0)?;
+    let session_id = value
+        .get_data(DataType::SessionId)
+        .as_signed_number()
+        .and_then(|id| i64::try_from(id).ok())
+        .filter(|id| *id > 0)?;
+    value
+        .get_data(DataType::UserId)
+        .as_signed_number()
+        .and_then(|id| i64::try_from(id).ok())
+        .filter(|id| *id > 0)?;
+    value.get_data(DataType::UserState).as_str()?;
+    Some((receiver, session_id))
+}
+
+fn parse_omega_capability_response(value: &CommunicationValue) -> Result<PeerCapabilities, String> {
+    if !value.is_type(CommunicationType::IdentificationResponse) {
+        return Err("not an identification response".to_string());
+    }
+    PeerCapabilities::from_omega_identification_description(
+        value.get_data(DataType::Description).as_str(),
+    )
+    .map_err(|_| "Omega advertised an invalid capability set".to_string())
+}
+
 // ============================================================================
 // Waiting Task System
 // ============================================================================
@@ -85,24 +120,33 @@ pub fn start_task_cleanup_loop(omega: Arc<OmegaConnection>) {
 pub enum ConnectionState {
     Disconnected,
     Connecting,
-    Connected { identified: bool },
+    Identifying,
+    SynchronizingRoutes,
+    SynchronizingSubscriptions,
+    Ready,
 }
 
 #[allow(unused_variables)]
 impl ConnectionState {
     pub fn is_connected(&self) -> bool {
-        match self {
+        !matches!(
-            ConnectionState::Connected { identified } => true,
+            self,
-            _ => false,
+            ConnectionState::Disconnected | ConnectionState::Connecting
-        }
+        )
     }
 
     #[allow(dead_code)]
     pub fn is_identified(&self) -> bool {
-        match self {
+        matches!(
-            ConnectionState::Connected { identified: true } => true,
+            self,
-            _ => false,
+            ConnectionState::SynchronizingRoutes
+                | ConnectionState::SynchronizingSubscriptions
+                | ConnectionState::Ready
+        )
     }
+
+    pub fn is_ready(&self) -> bool {
+        matches!(self, ConnectionState::Ready)
     }
 }
 
@@ -117,6 +161,8 @@ pub struct OmegaConnection {
     host: String,
     port: u16,
     omikron_id: u64,
+    sync_timeout: Duration,
+    sync_retries: u32,
     pub connection_id: Uuid,
     shutdown_tx: Arc<Mutex<Option<watch::Sender<bool>>>>,
     // Track if we should reconnect on close
@@ -126,6 +172,14 @@ pub struct OmegaConnection {
     keyring: mtp::crypto::Keyring,
     rho: Arc<RhoManager>,
     waiting_tasks: DashMap<u32, WaitingTask>,
+    session_presence: DashMap<(i64, i64), SessionPresenceState>,
+    peer_capabilities: Arc<RwLock<PeerCapabilities>>,
+}
+
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct SessionPresenceState {
+    pub subscribed_user_ids: Vec<i64>,
+    pub pending: bool,
 }
 
 impl OmegaConnection {
@@ -143,6 +197,8 @@ impl OmegaConnection {
             host: config.omega_host.clone(),
             port: config.omega_port,
             omikron_id: config.omikron_id,
+            sync_timeout: config.omega_sync_timeout,
+            sync_retries: config.omega_sync_retries,
             connection_id: Uuid::new_v4(),
             shutdown_tx: Arc::new(Mutex::new(Some(shutdown_tx))),
             reconnect_on_close: Arc::new(RwLock::new(true)),
@@ -154,6 +210,8 @@ impl OmegaConnection {
             keyring,
             rho,
             waiting_tasks: DashMap::new(),
+            session_presence: DashMap::new(),
+            peer_capabilities: Arc::new(RwLock::new(PeerCapabilities::default())),
         }
     }
 
@@ -267,6 +325,7 @@ impl OmegaConnection {
 
         let client_config = ClientConfig::new(format!("https://{}:{}", self.host, self.port))
             .with_client_id(self.omikron_id)
+            .with_description(PeerCapabilities::current().identification_description())
             .with_policy(
                 Policy::default()
                     .with_send_mode(SendMode::SingleStreamPerMessage)
@@ -302,18 +361,43 @@ impl OmegaConnection {
         // Store sender
         let sender_arc = Arc::new(connection.sender.clone());
         *self.sender.write().await = Some(sender_arc.clone());
-        *self.state.write().await = ConnectionState::Connected { identified: false };
+        // A new authenticated transport gets a fresh request circuit. A
+        // previous transport can have failed while synchronizing and must not
+        // prevent this connection from making its initial requests.
+        self.circuit_breaker.lock().await.record_success();
+        *self.state.write().await = ConnectionState::Identifying;
 
         // Get handle for close monitoring
         let sender_handle = sender_arc.handle().clone();
         let connection = Arc::new(connection);
+        let (capability_tx, capability_rx) = oneshot::channel();
 
         // Start read loop
         let read_self = self.clone();
         let read_handle = tokio::spawn(async move {
-            read_self.read_loop(connection, sender_handle).await;
+            read_self
+                .read_loop(connection, sender_handle, Some(capability_tx))
+                .await;
         });
 
+        let negotiated_capabilities =
+            match tokio::time::timeout(CAPABILITY_NEGOTIATION_TIMEOUT, capability_rx).await {
+                Ok(Ok(Ok(capabilities))) => capabilities,
+                Ok(Ok(Err(error))) => {
+                    self.fail_synchronization("capability negotiation", error)
+                        .await;
+                    let _ = read_handle.await;
+                    return Err("Omega capability negotiation failed".to_string());
+                }
+                Ok(Err(_)) | Err(_) => {
+                    // An Omega from before capability negotiation sends no second
+                    // identification response. Fall back to its documented
+                    // legacy contracts rather than guessing modern support.
+                    PeerCapabilities::default()
+                }
+            };
+        *self.peer_capabilities.write().await = negotiated_capabilities;
+
         // Tell omega our current state now that we're actually connected -
         // doing this after teardown (as before) sent into a sender that had
         // already been cleared, silently dropping the sync every time.
@@ -354,8 +438,10 @@ impl OmegaConnection {
         }
     }
 
-    async fn sync_client_iota_status(self: Arc<Self>) {
+    async fn sync_client_iota_status(&self) {
+        *self.state.write().await = ConnectionState::SynchronizingRoutes;
         let mut connected_iota_ids: Vec<DataValue> = Vec::new();
+        let mut connected_sessions: Vec<DataValue> = Vec::new();
         let mut connected_user_ids: Vec<DataValue> = Vec::new();
 
         let iota_ids = self.rho.iota_ids().await;
@@ -367,21 +453,120 @@ impl OmegaConnection {
 
         for rho in rho_connections {
             for client_conn in rho.get_client_connections().await {
-                connected_user_ids.push(DataValue::SignedNumber(
+                let type_map = mtp::codec::TypeMap::latest();
-                    client_conn.get_user_id().await.into(),
+                let user_id = DataValue::SignedNumber(client_conn.get_user_id().await.into());
-                ));
+                let session_id = DataValue::SignedNumber((client_conn.session_id as i128).into());
+                let iota_id = DataValue::SignedNumber((rho.get_iota_id().await as i128).into());
+                let peer_capabilities = self.peer_capabilities.read().await.clone();
+                if peer_capabilities.session_snapshot_v1 {
+                    connected_sessions.push(DataValue::Container(vec![
+                        (data_type_id(DataType::UserId, &type_map), user_id),
+                        (data_type_id(DataType::SessionId, &type_map), session_id),
+                        (data_type_id(DataType::IotaId, &type_map), iota_id),
+                    ]));
+                } else {
+                    connected_user_ids.push(user_id);
+                }
             }
         }
 
-        let sync_msg = CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
+        let peer_capabilities = self.peer_capabilities.read().await.clone();
+        let sync_msg = if peer_capabilities.session_snapshot_v1 {
+            CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
+                .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids))
+                .add_typed_default(DataType::UserStates, DataValue::Array(connected_sessions))
+                .add_typed_default(
+                    DataType::RhoConnections,
+                    DataValue::SignedNumber(self.rho.connection_count().await as i128),
+                )
+        } else {
+            // Legacy Omega has no request/response contract for this message.
+            // Its exact snapshot is IotaIds + UserIds + RhoConnections.
+            CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
                 .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids))
                 .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids))
                 .add_typed_default(
                     DataType::RhoConnections,
                     DataValue::SignedNumber(self.rho.connection_count().await as i128),
-            );
+                )
+        };
 
-        self.send_message(&sync_msg).await;
+        if !peer_capabilities.session_snapshot_v1 {
+            if let Err(error) = self.send_message_result(&sync_msg).await {
+                self.fail_synchronization("legacy route synchronization", error)
+                    .await;
+                return;
+            }
+            *self.state.write().await = ConnectionState::SynchronizingSubscriptions;
+            if let Err(error) = self.restore_presence_subscriptions().await {
+                self.fail_synchronization("subscription restoration", error)
+                    .await;
+                return;
+            }
+            *self.state.write().await = ConnectionState::Ready;
+            return;
+        }
+
+        let mut response = Err("route synchronization did not start".to_string());
+        for attempt in 0..self.sync_retries {
+            response = self
+                .await_response(&sync_msg, Some(self.sync_timeout))
+                .await;
+            match &response {
+                Ok(value) if value.is_type(CommunicationType::Success) => break,
+                Ok(value) if value.is_type(CommunicationType::ErrorInvalidData) => break,
+                _ if attempt + 1 < self.sync_retries => {
+                    sleep(Duration::from_millis(250 * 2u64.pow(attempt))).await
+                }
+                _ => break,
+            }
+        }
+        match response {
+            Ok(response) if response.is_type(CommunicationType::Success) => {
+                *self.state.write().await = ConnectionState::SynchronizingSubscriptions;
+                if let Err(error) = self.restore_presence_subscriptions().await {
+                    log_err!(
+                        0,
+                        PrintType::Omega,
+                        "Presence subscription restoration failed: {}",
+                        error
+                    );
+                    self.fail_synchronization("subscription restoration", error)
+                        .await;
+                    return;
+                }
+                *self.state.write().await = ConnectionState::Ready;
+            }
+            Ok(response) => {
+                let error = format!(
+                    "Omega rejected presence snapshot with {}",
+                    response.get_type()
+                );
+                self.fail_synchronization("route synchronization", error)
+                    .await;
+                return;
+            }
+            Err(error) => {
+                self.fail_synchronization("route synchronization", error)
+                    .await;
+                return;
+            }
+        }
+    }
+
+    async fn fail_synchronization(&self, phase: &str, error: String) {
+        log_err!(
+            0,
+            PrintType::Omega,
+            "Omega {} failed after {} attempts: {}; closing transport for reconnect",
+            phase,
+            self.sync_retries,
+            error
+        );
+        *self.state.write().await = ConnectionState::Disconnected;
+        if let Some(sender) = self.sender.read().await.as_ref() {
+            sender.close().await;
+        }
     }
 
     // -------------------------------------------------------------------------
@@ -392,6 +577,7 @@ impl OmegaConnection {
         self: Arc<Self>,
         connection: Arc<MTPConnection>,
         sender_handle: Arc<ConnectionHandle>,
+        mut capability_tx: Option<oneshot::Sender<Result<PeerCapabilities, String>>>,
     ) {
         // Monitor both receiver and sender handle for close
         let mut close_rx = sender_handle.subscribe_close();
@@ -403,6 +589,14 @@ impl OmegaConnection {
                         Ok(cv) => {
                             log_cv_in!(PrintType::Omega, &cv);
 
+                            if cv.is_type(CommunicationType::IdentificationResponse) {
+                                if let Some(tx) = capability_tx.take() {
+                                    let result = parse_omega_capability_response(&cv);
+                                    let _ = tx.send(result);
+                                }
+                                continue;
+                            }
+
                             let msg_id = cv.get_id();
                             if let Some((_, task)) = self.waiting_tasks.remove(&msg_id) {
                                 if (task.task)(self.clone(), cv.clone()) {
@@ -430,6 +624,20 @@ impl OmegaConnection {
                                     }
                                 }
                             }
+                            if cv.is_type(CommunicationType::ClientChanged) {
+                                let Some((receiver, session_id)) = client_changed_target(&cv)
+                                else {
+                                    continue;
+                                };
+                                if let Some(client) = self
+                                    .rho
+                                    .get_client_connection(receiver, session_id)
+                                    .await
+                                {
+                                    client.send_message(&cv).await;
+                                }
+                                continue;
+                            }
                         }
                         Err(e) => {
                             log_err!(0, PrintType::Omega, "Receive error: {}", e);
@@ -488,6 +696,14 @@ impl OmegaConnection {
         let _ = self.try_send_message(cv).await;
     }
 
+    pub async fn send_message_result(&self, cv: &CommunicationValue) -> Result<(), String> {
+        self.try_send_message(cv).await
+    }
+
+    pub async fn supports_set_user_state(&self) -> bool {
+        self.peer_capabilities.read().await.set_user_state_v1
+    }
+
     pub async fn await_connection(&self, timeout_duration: Option<Duration>) -> Result<(), String> {
         if self.state.read().await.is_connected() {
             return Ok(());
@@ -607,72 +823,371 @@ impl OmegaConnection {
         self.state.read().await.is_identified()
     }
 
+    pub async fn is_ready(&self) -> bool {
+        self.state.read().await.is_ready()
+    }
+
     #[allow(dead_code)]
-    pub async fn close_iota(&self, iota_id: i64) {
+    pub async fn close_iota(&self, iota_id: i64) -> Result<(), String> {
         let cv = CommunicationValue::new(CommunicationType::IotaDisconnected)
             .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
-        self.send_message(&cv).await;
+        let request_id = cv.get_id();
+        let result = self.lifecycle_request(&cv).await;
+        if let Err(error) = &result {
+            self.log_lifecycle_failure("IotaDisconnected", iota_id, None, request_id, error);
+            self.sync_client_iota_status().await;
+        }
+        result
     }
 
-    pub async fn client_changed(
+    pub async fn client_disconnected(&self, user_id: i64, session_id: i64) -> Result<(), String> {
-        &self,
+        self.session_presence.remove(&(user_id, session_id));
-        _iota_id: i64,
-        user_id: i64,
-        session_id: i64,
-        state: UserStatus,
-    ) {
-        let cv = CommunicationValue::new(CommunicationType::ClientChanged)
-            .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
-            .add_typed_default(
-                DataType::SessionId,
-                DataValue::SignedNumber(session_id.into()),
-            )
-            .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()));
-        self.send_message(&cv).await;
-    }
-
-    pub async fn client_disconnected(&self, user_id: i64, session_id: i64) {
         let cv = CommunicationValue::new(CommunicationType::UserDisconnected)
             .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
             .add_typed_default(
                 DataType::SessionId,
                 DataValue::SignedNumber(session_id.into()),
             );
-        self.send_message(&cv).await;
+        let request_id = cv.get_id();
+        let result = self.lifecycle_request(&cv).await;
+        if let Err(error) = &result {
+            self.log_lifecycle_failure(
+                "UserDisconnected",
+                user_id,
+                Some(session_id),
+                request_id,
+                error,
+            );
+            self.sync_client_iota_status().await;
+        }
+        result
     }
 
-    pub async fn user_states(&self, user_id: i64, user_ids: Vec<i64>) {
+    pub async fn user_connected(
-        let user_ids = user_ids
+        &self,
-            .iter()
+        user_id: i64,
-            .map(|v| DataValue::SignedNumber((*v).into()))
+        session_id: i64,
-            .collect();
+        iota_id: i64,
-
+    ) -> Result<(), String> {
-        let cv = CommunicationValue::new(CommunicationType::GetStates)
+        let request = CommunicationValue::new(CommunicationType::UserConnected)
-            .add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
+            .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
-        let msg_id = cv.get_id();
+            .add_typed_default(
-
+                DataType::SessionId,
-        let rho_manager = self.rho.clone();
+                DataValue::SignedNumber(session_id.into()),
-        self.waiting_tasks.insert(
+            )
-            msg_id,
+            .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
-            WaitingTask {
+            .add_typed_default(
-                task: Box::new(
+                DataType::UserState,
-                    move |_: Arc<OmegaConnection>, response: CommunicationValue| {
+                DataValue::Str("user_online".to_string()),
-                        let rho_manager = rho_manager.clone();
+            )
-                        tokio::spawn(async move {
+            .add_typed_default(
-                            let rho = rho_manager.get_for_user(user_id).await;
+                DataType::UpdatedAt,
-                            if let Some(rho) = rho {
+                DataValue::SignedNumber(
-                                for client in rho.get_client_connections_for_user(user_id).await {
+                    std::time::SystemTime::now()
-                                    client.send_message(&response).await;
+                        .duration_since(std::time::UNIX_EPOCH)
-                                }
+                        .unwrap_or_default()
-                            }
+                        .as_millis() as i128,
-                        });
-                        true
-                    },
                 ),
-                inserted_at: Instant::now(),
+            );
+        let request_id = request.get_id();
+        let result = self.lifecycle_request(&request).await;
+        if let Err(error) = &result {
+            self.log_lifecycle_failure(
+                "UserConnected",
+                user_id,
+                Some(session_id),
+                request_id,
+                error,
+            );
+            self.sync_client_iota_status().await;
+        }
+        result
+    }
+
+    pub async fn iota_connected(&self, iota_id: i64) -> Result<(), String> {
+        let request = CommunicationValue::new(CommunicationType::IotaConnected)
+            .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
+        let request_id = request.get_id();
+        let result = self.lifecycle_request(&request).await;
+        if let Err(error) = &result {
+            self.log_lifecycle_failure("IotaConnected", iota_id, None, request_id, error);
+            self.sync_client_iota_status().await;
+        }
+        result
+    }
+
+    pub async fn reconcile_routes(&self) {
+        self.sync_client_iota_status().await;
+    }
+
+    async fn lifecycle_request(&self, request: &CommunicationValue) -> Result<(), String> {
+        if !self.is_ready().await {
+            return Err("Omega is not synchronized".to_string());
+        }
+        if !self.peer_capabilities.read().await.session_snapshot_v1 {
+            // Legacy Omega handles lifecycle messages as notifications and
+            // never returns Success. A transport-level send result is the
+            // only acknowledgement available under that contract.
+            return self.send_message_result(request).await;
+        }
+        let response = self
+            .await_response(request, Some(Duration::from_secs(20)))
+            .await?;
+        if response.is_type(CommunicationType::Success) {
+            Ok(())
+        } else {
+            Err(format!("Omega rejected {}", response.get_type()))
+        }
+    }
+
+    fn log_lifecycle_failure(
+        &self,
+        operation: &str,
+        entity_id: i64,
+        session_id: Option<i64>,
+        request_id: u32,
+        error: &str,
+    ) {
+        log_err!(
+            0,
+            PrintType::Omega,
+            "Lifecycle operation={} entity_id={} session_id={:?} request_id={} failed: {}",
+            operation,
+            entity_id,
+            session_id,
+            request_id,
+            error
+        );
+    }
+
+    pub async fn replace_state_subscription(
+        &self,
+        user_id: i64,
+        session_id: i64,
+        mut user_ids: Vec<i64>,
+    ) -> Result<(), String> {
+        if user_id <= 0 || session_id <= 0 {
+            return Err("invalid user or session id".to_string());
+        }
+
+        user_ids.sort_unstable();
+        user_ids.dedup();
+
+        let key = (user_id, session_id);
+        // Store the latest authoritative snapshot before checking readiness.
+        // A client can synchronize while the transport is reconnecting.
+        self.session_presence.insert(
+            key,
+            SessionPresenceState {
+                subscribed_user_ids: user_ids.clone(),
+                pending: true,
             },
         );
 
-        self.send_message(&cv).await;
+        if !self.is_ready().await {
+            return Err("Omega is not synchronized".to_string());
+        }
+
+        self.deliver_state_subscription(key, user_ids).await
+    }
+
+    async fn deliver_state_subscription(
+        &self,
+        key: (i64, i64),
+        user_ids: Vec<i64>,
+    ) -> Result<(), String> {
+        let (user_id, session_id) = key;
+
+        if !self.peer_capabilities.read().await.state_subscribe_v1 {
+            let delivered_ids = user_ids.clone();
+            let result = self
+                .request_user_states(user_id, session_id, user_ids)
+                .await;
+            if result.is_ok() {
+                self.mark_presence_delivered(key, &delivered_ids).await;
+            }
+            return result;
+        }
+
+        let subscription = CommunicationValue::new(CommunicationType::StateSubscribe)
+            .with_sender(user_id as u64)
+            .add_typed_default(
+                DataType::SessionId,
+                DataValue::SignedNumber(session_id.into()),
+            )
+            .add_typed_default(
+                DataType::UserIds,
+                DataValue::Array(
+                    user_ids
+                        .iter()
+                        .map(|user_id| DataValue::SignedNumber((*user_id).into()))
+                        .collect(),
+                ),
+            );
+        let response = self
+            .await_response(&subscription, Some(Duration::from_secs(20)))
+            .await?;
+        if !response.is_type(CommunicationType::Success) {
+            return Err(format!(
+                "StateSubscribe rejected with {}",
+                response.get_type()
+            ));
+        }
+
+        let delivered_ids = user_ids.clone();
+        self.request_user_states(user_id, session_id, user_ids)
+            .await?;
+        self.mark_presence_delivered(key, &delivered_ids).await;
+        Ok(())
+    }
+
+    async fn restore_presence_subscriptions(&self) -> Result<(), String> {
+        let mut active = Vec::new();
+        for rho in self.rho.connections().await {
+            active.extend(rho.get_client_connections().await);
+        }
+        let active_keys: std::collections::HashSet<_> = active
+            .iter()
+            .map(|client| (client.user_id as i64, client.session_id as i64))
+            .collect();
+
+        // Do not replay subscriptions for sessions that disappeared while the
+        // transport was down.
+        self.session_presence
+            .retain(|key, _| active_keys.contains(key));
+
+        for key in active_keys {
+            let Some(snapshot) = self.session_presence.get(&key).map(|entry| entry.clone()) else {
+                continue;
+            };
+            if !snapshot.pending {
+                continue;
+            }
+            self.deliver_state_subscription(key, snapshot.subscribed_user_ids)
+                .await?;
+        }
+        Ok(())
+    }
+
+    async fn mark_presence_delivered(&self, key: (i64, i64), delivered_ids: &[i64]) {
+        if let Some(mut entry) = self.session_presence.get_mut(&key)
+            && entry.subscribed_user_ids == delivered_ids
+        {
+            entry.pending = false;
+        }
+    }
+
+    async fn request_user_states(
+        &self,
+        user_id: i64,
+        session_id: i64,
+        user_ids: Vec<i64>,
+    ) -> Result<(), String> {
+        let request = CommunicationValue::new(CommunicationType::GetStates)
+            .add_typed_default(
+                DataType::SessionId,
+                DataValue::SignedNumber(session_id.into()),
+            )
+            .add_typed_default(
+                DataType::UserIds,
+                DataValue::Array(
+                    user_ids
+                        .iter()
+                        .map(|user_id| DataValue::SignedNumber((*user_id).into()))
+                        .collect(),
+                ),
+            );
+        let response = self
+            .await_response(&request, Some(Duration::from_secs(20)))
+            .await?;
+        let legacy_peer = !self.peer_capabilities.read().await.client_state_push_v1;
+        let response =
+            if legacy_peer && response.get_data(DataType::SessionId).as_number().is_none() {
+                response.add_typed_default(
+                    DataType::SessionId,
+                    DataValue::SignedNumber(session_id.into()),
+                )
+            } else {
+                response
+            };
+        if !legacy_peer
+            && response.get_data(DataType::SessionId).as_number() != Some(session_id as i128)
+        {
+            return Err("GetStates response had the wrong session".to_string());
+        }
+
+        if let Some(client) = self.rho.get_client_connection(user_id, session_id).await {
+            client.send_message(&response).await;
+        }
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{PeerCapabilities, client_changed_target, parse_omega_capability_response};
+    use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue};
+
+    fn notification() -> CommunicationValue {
+        CommunicationValue::new(CommunicationType::ClientChanged)
+            .with_receiver(7)
+            .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11))
+            .add_typed_default(DataType::UserId, DataValue::SignedNumber(20))
+            .add_typed_default(DataType::UserState, DataValue::Str("user_online".into()))
+    }
+
+    #[test]
+    fn client_changed_target_requires_all_routing_fields() {
+        assert_eq!(client_changed_target(&notification()), Some((7, 11)));
+
+        let mut missing_session = notification();
+        missing_session.remove_data(DataType::SessionId);
+        assert_eq!(client_changed_target(&missing_session), None);
+
+        let mut missing_state = notification();
+        missing_state.remove_data(DataType::UserState);
+        assert_eq!(client_changed_target(&missing_state), None);
+    }
+
+    #[test]
+    fn client_changed_target_rejects_non_positive_ids() {
+        let invalid = CommunicationValue::new(CommunicationType::ClientChanged)
+            .with_receiver(0)
+            .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11))
+            .add_typed_default(DataType::UserId, DataValue::SignedNumber(20))
+            .add_typed_default(DataType::UserState, DataValue::Str("user_online".into()));
+        assert_eq!(client_changed_target(&invalid), None);
+    }
+
+    #[test]
+    fn capability_response_negotiates_new_omega() {
+        let response = CommunicationValue::new(CommunicationType::IdentificationResponse)
+            .add_typed_default(
+                DataType::Description,
+                DataValue::Str(
+                    "omega;caps=set_user_state_v1,state_subscribe_v1,session_snapshot_v1,client_state_push_v1"
+                        .into(),
+                ),
+            );
+        let capabilities = parse_omega_capability_response(&response).unwrap();
+        assert!(capabilities.state_subscribe_v1);
+        assert!(capabilities.client_state_push_v1);
+    }
+
+    #[test]
+    fn missing_capability_response_is_legacy() {
+        let response = CommunicationValue::new(CommunicationType::IdentificationResponse);
+        let capabilities = parse_omega_capability_response(&response).unwrap();
+        assert_eq!(capabilities, PeerCapabilities::default());
+    }
+
+    #[test]
+    fn unsupported_capability_response_is_rejected() {
+        let response = CommunicationValue::new(CommunicationType::IdentificationResponse)
+            .add_typed_default(
+                DataType::Description,
+                DataValue::Str("omega;caps=unsupported_v9".into()),
+            );
+        assert!(parse_omega_capability_response(&response).is_err());
     }
 }

src/rho/client_connection.rs

@@ -24,7 +24,6 @@ pub struct ClientConnection {
     pub receiver: Arc<MtpReceiver>,
     pub_key: Arc<RwLock<Option<Vec<u8>>>>,
     pub rho_connection: Arc<RwLock<Option<Arc<RhoConnection>>>>,
-    pub interested_users: Arc<RwLock<Vec<i64>>>,
     is_open: Arc<RwLock<bool>>,
     message_slots: Arc<tokio::sync::Semaphore>,
 }
@@ -35,7 +34,6 @@ impl ClientConnection {
             state: general.state.clone(),
             pub_key: Arc::new(RwLock::new(None)),
             rho_connection: general.rho_connection.clone(),
-            interested_users: Arc::new(RwLock::new(Vec::new())),
             is_open: Arc::new(RwLock::new(true)),
             sender: general.sender.clone(),
             receiver: general.receiver.clone(),
@@ -104,9 +102,26 @@ impl ClientConnection {
 
             let mut cv = cv;
 
-            // Handle client status changes
+            if cv.is_type(CommunicationType::SetUserState) {
-            if cv.is_type(CommunicationType::ClientChanged) {
+                self.handle_set_user_state(cv).await;
-                self.handle_client_changed(cv).await;
+                return;
+            }
+
+            // Compatibility for clients predating SetUserState.  The target
+            // user fields, if present, are deliberately ignored: an
+            // authenticated connection may only change its own state.
+            if cv.is_type(CommunicationType::ClientChanged)
+                && cv.get_data_opt(DataType::UserState).is_some()
+            {
+                self.handle_set_user_state(
+                    CommunicationValue::new(CommunicationType::SetUserState)
+                        .with_id(cv.get_id())
+                        .add_typed_default(
+                            DataType::UserState,
+                            cv.get_data(DataType::UserState).clone(),
+                        ),
+                )
+                .await;
                 return;
             }
 
@@ -191,6 +206,46 @@ impl ClientConnection {
                 || cv.is_type(CommunicationType::GetIotaData)
                 || cv.is_type(CommunicationType::DeleteUser)
             {
+                if cv.is_type(CommunicationType::ChangeUserData)
+                    && cv.get_data_opt(DataType::OnlineStatus).is_some()
+                {
+                    let mut profile_request = cv.clone();
+                    let preference = profile_request.remove_data(DataType::OnlineStatus);
+                    let state_request = CommunicationValue::new(CommunicationType::SetUserState)
+                        .with_id(cv.get_id())
+                        .add_typed_default(DataType::UserState, preference.unwrap());
+                    let state_response = match self.request_set_user_state(state_request).await {
+                        Ok(response) => response,
+                        Err(error_type) => {
+                            self.send_error_response(cv.get_id(), error_type).await;
+                            return;
+                        }
+                    };
+                    if !state_response.is_type(CommunicationType::Success) {
+                        self.send_message(&state_response).await;
+                        return;
+                    }
+
+                    if profile_request.data_len() == 0 {
+                        self.send_message(&state_response).await;
+                        return;
+                    }
+
+                    match self
+                        .await_omega_response(profile_request.with_sender(self.user_id))
+                        .await
+                    {
+                        Ok(response) if response.is_type(CommunicationType::Success) => {
+                            self.send_message(&response).await;
+                        }
+                        Ok(response) => self.send_message(&response).await,
+                        Err(_) => {
+                            self.send_error_response(cv.get_id(), CommunicationType::ErrorInternal)
+                                .await;
+                        }
+                    }
+                    return;
+                }
                 let sender = self.get_user_id().await;
                 self.handle_omega_forward(cv.with_sender(sender as u64))
                     .await;
@@ -290,34 +345,91 @@ impl ClientConnection {
         });
     }
     async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
-        let client_for_closure = self.clone();
+        let request_id = cv.get_id();
-        let response_cv = self
+        match self.await_omega_response(cv).await {
-            .state
+            Ok(response_cv) => self.send_message(&response_cv).await,
-            .omega
+            Err(_) => {
-            .clone()
+                self.send_error_response(request_id, CommunicationType::ErrorInternal)
-            .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
                     .await;
-        if let Ok(response_cv) = response_cv {
+            }
-            client_for_closure.send_message(&response_cv).await;
         }
     }
 
-    /// Handle client status change
+    async fn await_omega_response(&self, cv: CommunicationValue) -> Result<CommunicationValue, ()> {
-    async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
-        let user_id = self.get_user_id().await;
-        if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) {
-            let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
-            if let Some(rho_conn) = self.get_rho_connection().await {
         self.state
             .omega
-                    .client_changed(
+            .clone()
-                        rho_conn.get_iota_id().await as i64,
+            .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
-                        user_id as i64,
+            .await
-                        self.session_id as i64,
+            .map_err(|_| ())
-                        user_status,
-                    )
-                    .await;
     }
+
+    async fn request_set_user_state(
+        &self,
+        cv: CommunicationValue,
+    ) -> Result<CommunicationValue, CommunicationType> {
+        if !self.state.omega.is_ready().await {
+            return Err(CommunicationType::ErrorInternal);
+        }
+        let Some(state) = cv
+            .get_data(DataType::UserState)
+            .as_str()
+            .and_then(UserStatus::from_client_preference)
+        else {
+            return Err(CommunicationType::ErrorInvalidData);
+        };
+        let Some(rho) = self.get_rho_connection().await else {
+            return Err(CommunicationType::ErrorNoIota);
+        };
+        let request = if self.state.omega.supports_set_user_state().await {
+            CommunicationValue::new(CommunicationType::SetUserState)
+                .with_id(cv.get_id())
+                .with_sender(self.user_id)
+                .add_typed_default(
+                    DataType::UserId,
+                    DataValue::SignedNumber(self.user_id as i128),
+                )
+                .add_typed_default(
+                    DataType::SessionId,
+                    DataValue::SignedNumber(self.session_id as i128),
+                )
+                .add_typed_default(
+                    DataType::IotaId,
+                    DataValue::SignedNumber(rho.get_iota_id().await as i128),
+                )
+                .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))
+        } else {
+            // Legacy Omega accepts the original ClientChanged request. It is
+            // intentionally limited to the authenticated user's ID and a
+            // writable preference, with no derived connectivity state.
+            CommunicationValue::new(CommunicationType::ClientChanged)
+                .with_id(cv.get_id())
+                .with_sender(self.user_id)
+                .add_typed_default(
+                    DataType::UserId,
+                    DataValue::SignedNumber(self.user_id as i128),
+                )
+                .add_typed_default(DataType::UserState, DataValue::Str(state.to_string()))
+        };
+        if !self.state.omega.supports_set_user_state().await {
+            self.state
+                .omega
+                .send_message_result(&request)
+                .await
+                .map_err(|_| CommunicationType::ErrorInternal)?;
+            return Ok(CommunicationValue::new(CommunicationType::Success)
+                .with_id(cv.get_id())
+                .add_typed_default(DataType::UserState, DataValue::Str(state.to_string())));
+        }
+        self.await_omega_response(request)
+            .await
+            .map_err(|_| CommunicationType::ErrorInternal)
+    }
+
+    async fn handle_set_user_state(self: Arc<Self>, cv: CommunicationValue) {
+        match self.request_set_user_state(cv.clone()).await {
+            Ok(response) => self.send_message(&response).await,
+            Err(error_type) => self.send_error_response(cv.get_id(), error_type).await,
         }
     }
 
@@ -830,35 +942,6 @@ impl ClientConnection {
         let _ = self.sender.close();
     }
 
-    /// Set interested users list
-    pub async fn set_interested_users(self: Arc<Self>, interested_ids: Vec<i64>) {
-        let mut interested_guard = self.interested_users.write().await;
-        *interested_guard = interested_ids;
-    }
-    #[allow(dead_code)]
-    pub async fn get_interested_users(self: Arc<Self>) -> Vec<i64> {
-        let interested_guard = self.interested_users.read().await;
-        interested_guard.clone()
-    }
-
-    /// Check if interested in a user and send notification
-    #[allow(dead_code)]
-    pub async fn are_you_interested(self: Arc<Self>, user_id: i64, user_status: &str) {
-        let interested_guard = self.clone().get_interested_users().await;
-        if interested_guard.contains(&user_id) {
-            let status = if user_status == "user_invisible" {
-                "user_offline"
-            } else {
-                user_status
-            };
-            let notification = CommunicationValue::new(CommunicationType::ClientChanged)
-                .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
-                .add_typed_default(DataType::UserState, DataValue::Str(status.to_string()));
-
-            self.send_message(&notification).await;
-        }
-    }
-
     /// Handle connection close
     pub async fn handle_close(self: Arc<Self>) {
         let user_id = self.get_user_id().await;
@@ -880,7 +963,6 @@ impl Clone for ClientConnection {
             client_version: self.client_version.clone(),
             pub_key: Arc::clone(&self.pub_key),
             rho_connection: Arc::clone(&self.rho_connection),
-            interested_users: Arc::clone(&self.interested_users),
             is_open: Arc::clone(&self.is_open),
             message_slots: Arc::clone(&self.message_slots),
         }

src/rho/connection.rs

@@ -136,10 +136,23 @@ impl GeneralConnection {
         if let Some(rho_conn) = rho {
             rho_conn.bind_user_id(user_id).await;
             rho_conn.add_client_connection(client.clone()).await;
-            self.notify_user_connected(user_id, rho_conn.get_iota_id().await as i64)
+            match self
-                .await;
+                .notify_user_connected(user_id, rho_conn.get_iota_id().await as i64)
+                .await
+            {
+                Ok(()) => {
                     self.send_initial_client_state_request(&rho_conn, user_id)
                         .await;
+                }
+                Err(error) => {
+                    log_err!(
+                        user_id,
+                        PrintType::Client,
+                        "UserConnected acknowledgement failed: {}",
+                        error
+                    );
+                }
+            }
         } else {
             log_err!(
                 user_id,
@@ -155,29 +168,13 @@ impl GeneralConnection {
         true
     }
 
-    async fn notify_user_connected(&self, user_id: i64, iota_id: i64) {
+    async fn notify_user_connected(&self, user_id: i64, iota_id: i64) -> Result<(), String> {
         let session_id = *self.session_id.read().await as i64;
-        let notify = CommunicationValue::new(CommunicationType::UserConnected)
+        self.state
-            .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+            .omega
-            .add_typed_default(
+            .clone()
-                DataType::SessionId,
+            .user_connected(user_id, session_id, iota_id)
-                DataValue::SignedNumber(session_id.into()),
+            .await
-            )
-            .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
-            .add_typed_default(
-                DataType::UserState,
-                DataValue::Str("user_online".to_string()),
-            )
-            .add_typed_default(
-                DataType::UpdatedAt,
-                DataValue::SignedNumber(
-                    std::time::SystemTime::now()
-                        .duration_since(std::time::UNIX_EPOCH)
-                        .unwrap_or_default()
-                        .as_millis() as i128,
-                ),
-            );
-        self.state.omega.clone().send_message(&notify).await;
     }
 
     async fn find_user_rho(&self, user_id: i64) -> Option<Arc<RhoConnection>> {
@@ -227,9 +224,6 @@ impl GeneralConnection {
     async fn migrate_iota(self: &Arc<Self>) {
         let id = self.id;
         log_in!(id as i64, PrintType::Iota, "Iota {} connected", id);
-        let notify = CommunicationValue::new(CommunicationType::IotaConnected)
-            .add_typed_default(DataType::IotaId, DataValue::SignedNumber(id.into()));
-        self.state.omega.clone().send_message(&notify).await;
 
         let iota = IotaConnection::from_general(self.clone(), id).await;
         let previous = self.state.rho.get_by_iota(id as i64).await;
@@ -246,6 +240,14 @@ impl GeneralConnection {
             }
         }
         self.state.rho.add(rho).await;
+        if let Err(error) = self.state.omega.clone().iota_connected(id as i64).await {
+            log_err!(
+                id as i64,
+                PrintType::Iota,
+                "IotaConnected acknowledgement failed: {}",
+                error
+            );
+        }
         self.load_iota_users(id).await;
         iota.start();
     }

src/rho/iota_connection.rs

@@ -23,6 +23,47 @@ use tokio::sync::mpsc;
 
 use super::rho_connection::RhoConnection;
 
+fn contact_snapshot(value: &CommunicationValue) -> Option<(i64, i64, Vec<i64>)> {
+    if !value.is_type(CommunicationType::ClientStateSync)
+        && !value.is_type(CommunicationType::AddConversation)
+    {
+        return None;
+    }
+
+    let user_id = i64::try_from(value.get_receiver())
+        .ok()
+        .filter(|id| *id > 0)?;
+    let session_id = value
+        .get_data(DataType::SessionId)
+        .as_signed_number()
+        .and_then(|id| i64::try_from(id).ok())
+        .filter(|id| *id > 0)?;
+    let DataValue::Array(values) = value.get_data(DataType::UserIds) else {
+        return None;
+    };
+
+    let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest());
+    let mut contact_ids = Vec::with_capacity(values.len());
+    for value in values {
+        let contact_id = match value {
+            DataValue::SignedNumber(id) => i64::try_from(*id).ok(),
+            DataValue::Container(entries) => entries.iter().find_map(|(key, value)| {
+                (*key == user_id_type)
+                    .then(|| value.as_signed_number())
+                    .flatten()
+                    .and_then(|id| i64::try_from(id).ok())
+            }),
+            _ => None,
+        }
+        .filter(|id| *id > 0)?;
+        contact_ids.push(contact_id);
+    }
+    contact_ids.sort_unstable();
+    contact_ids.dedup();
+
+    Some((user_id, session_id, contact_ids))
+}
+
 #[allow(dead_code)]
 pub struct IotaConnection {
     pub state: Arc<AppState>,
@@ -172,6 +213,29 @@ impl IotaConnection {
         }
     }
 
+    async fn replace_state_subscription(
+        &self,
+        user_id: i64,
+        session_id: i64,
+        contact_ids: Vec<i64>,
+    ) -> Result<(), String> {
+        let Some(rho_connection) = self.get_rho_connection().await else {
+            return Err("Rho connection is unavailable".to_string());
+        };
+        if rho_connection
+            .get_client_connection(user_id, session_id)
+            .await
+            .is_none()
+        {
+            return Err("target client session is no longer connected".to_string());
+        }
+
+        self.state
+            .omega
+            .replace_state_subscription(user_id, session_id, contact_ids)
+            .await
+    }
+
     /// Send a CommunicationValue to the Iota
     pub async fn send_message(&self, cv: &CommunicationValue) {
         log_cv_out!(PrintType::Iota, cv);
@@ -206,6 +270,34 @@ impl IotaConnection {
             cv
         };
 
+        // Iota's UserIds field is an authoritative replacement snapshot for
+        // the requesting client session, including contact mutations.
+        if let Some((user_id, session_id, contact_ids)) = contact_snapshot(&cv) {
+            if let Err(error) = self
+                .replace_state_subscription(user_id, session_id, contact_ids)
+                .await
+            {
+                log_err!(
+                    self.iota_id as i64,
+                    PrintType::Omega,
+                    "Failed to replace presence subscription for user {} session {}: {}",
+                    user_id,
+                    session_id,
+                    error
+                );
+            }
+        }
+
+        if cv.is_type(CommunicationType::StateSubscribe) {
+            self.send_error_response(
+                cv.get_id(),
+                CommunicationType::ErrorInvalidData,
+                Some("StateSubscribe must come from an authoritative contact snapshot"),
+            )
+            .await;
+            return;
+        }
+
         // Handle GET_CHATS
         if cv.is_type(CommunicationType::GetChats) {
             self.handle_get_chats(cv).await;
@@ -517,7 +609,20 @@ impl IotaConnection {
         }
 
         let mut interested_ids: Vec<i64> = Vec::new();
-        let session_id = cv.get_data(DataType::SessionId).as_signed_number();
+        let Some(session_id) = cv
+            .get_data(DataType::SessionId)
+            .as_signed_number()
+            .and_then(|id| i64::try_from(id).ok())
+            .filter(|id| *id > 0)
+        else {
+            self.forward_to_client(
+                CommunicationValue::new(CommunicationType::ErrorInvalidData)
+                    .with_id(cv.get_id())
+                    .with_receiver(user_id),
+            )
+            .await;
+            return;
+        };
         let tm = TypeMap::latest();
 
         // Presence interest is the complete contact set, independent of
@@ -658,20 +763,18 @@ impl IotaConnection {
         // ============================
         // Notify Omega
         // ============================
-        self.state
+        if let Err(error) = self
-            .omega
+            .replace_state_subscription(user_id as i64, session_id, interested_ids.clone())
-            .user_states(user_id as i64, interested_ids.clone())
+            .await
-            .await;
+        {
-
+            log_err!(
-        // ============================
+                self.iota_id as i64,
-        // Notify Rho
+                PrintType::Omega,
-        // ============================
+                "Failed to replace presence subscription for user {} session {}: {}",
-        if let Some(rho_conn) = self.get_rho_connection().await {
+                user_id,
-            if let Some(session_id) = session_id.and_then(|id| i64::try_from(id).ok()) {
+                session_id,
-                rho_conn
+                error
-                    .set_interested(user_id as i64, session_id, interested_ids)
+            );
-                    .await;
-            }
         }
 
         // ============================
@@ -779,3 +882,48 @@ impl std::fmt::Debug for IotaConnection {
             .finish()
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::contact_snapshot;
+    use crate::util::data_type_id;
+    use mtp::codec::{CommunicationType, CommunicationValue, DataType, DataValue, TypeMap};
+
+    fn snapshot(user_ids: DataValue) -> CommunicationValue {
+        CommunicationValue::new(CommunicationType::ClientStateSync)
+            .with_receiver(7)
+            .add_typed_default(DataType::SessionId, DataValue::SignedNumber(11))
+            .add_typed_default(DataType::UserIds, user_ids)
+    }
+
+    #[test]
+    fn contact_snapshot_deduplicates_flat_and_typed_ids() {
+        let user_id_type = data_type_id(DataType::UserId, &TypeMap::latest());
+        let parsed = contact_snapshot(&snapshot(DataValue::Array(vec![
+            DataValue::SignedNumber(20),
+            DataValue::Container(vec![(user_id_type, DataValue::SignedNumber(21))]),
+            DataValue::SignedNumber(20),
+        ])))
+        .unwrap();
+
+        assert_eq!(parsed, (7, 11, vec![20, 21]));
+    }
+
+    #[test]
+    fn contact_snapshot_accepts_empty_contact_sets() {
+        assert_eq!(
+            contact_snapshot(&snapshot(DataValue::Array(Vec::new()))),
+            Some((7, 11, Vec::new()))
+        );
+    }
+
+    #[test]
+    fn contact_snapshot_rejects_malformed_contact_sets() {
+        assert_eq!(
+            contact_snapshot(&snapshot(DataValue::Array(vec![DataValue::Str(
+                "bad".into(),
+            )]))),
+            None
+        );
+    }
+}

src/rho/rho_connection.rs

@@ -1,6 +1,6 @@
 use super::{client_connection::ClientConnection, iota_connection::IotaConnection};
 
-use crate::{data::user::UserStatus, rho::app_connection::AppConnection};
+use crate::{log_err, rho::app_connection::AppConnection};
 use dashmap::DashMap;
 use mtp::codec::{CommunicationValue, DataType};
 use std::sync::Arc;
@@ -178,11 +178,21 @@ impl RhoConnection {
             return;
         }
         self.client_connections.remove(&key);
-        self.iota_connection
+        if let Err(error) = self
+            .iota_connection
             .state
             .omega
             .client_disconnected(target_user_id, target_session_id)
-            .await;
+            .await
+        {
+            log_err!(
+                target_user_id,
+                crate::util::logger::PrintType::Client,
+                "UserDisconnected acknowledgement failed for session {}: {}",
+                target_session_id,
+                error
+            );
+        }
         let remaining_for_user = self
             .client_connections
             .iter()
@@ -194,16 +204,6 @@ impl RhoConnection {
                 .rho
                 .remove_user_binding(target_user_id, self.get_iota_id().await as i64)
                 .await;
-            self.iota_connection
-                .state
-                .omega
-                .client_changed(
-                    self.get_iota_id().await as i64,
-                    target_user_id,
-                    target_session_id,
-                    UserStatus::user_offline,
-                )
-                .await;
         }
     }
 
@@ -228,7 +228,14 @@ impl RhoConnection {
         }
 
         // Notify OmegaConnection
-        self.iota_connection.state.omega.close_iota(iota_id).await;
+        if let Err(error) = self.iota_connection.state.omega.close_iota(iota_id).await {
+            log_err!(
+                iota_id,
+                crate::util::logger::PrintType::Iota,
+                "IotaDisconnected acknowledgement failed: {}",
+                error
+            );
+        }
     }
 
     /// Send message from Iota to specific client
@@ -255,25 +262,6 @@ impl RhoConnection {
         self.iota_connection.send_message(&cv).await;
     }
 
-    /// Set interested users for a specific client
-    pub async fn set_interested(&self, user_id: i64, session_id: i64, interested_ids: Vec<i64>) {
-        if let Some(connection) = self.get_client_connection(user_id, session_id).await {
-            connection.set_interested_users(interested_ids).await;
-        }
-    }
-
-    /// Check if clients are interested in a user
-    #[allow(dead_code)]
-    pub async fn are_they_interested(&self, user_id: i64, user_status: &str) {
-        let connections = self.get_client_connections().await;
-        for connection in connections.iter() {
-            connection
-                .clone()
-                .are_you_interested(user_id, user_status)
-                .await;
-        }
-    }
-
     /// Check if this RhoConnection contains a specific user ID
     #[allow(dead_code)]
     pub async fn contains_user(&self, user_id: &i64) -> bool {

src/rho/rho_manager.rs

@@ -135,4 +135,17 @@ impl RhoManager {
             .map(|entry| entry.value().clone())
             .collect()
     }
+
+    pub async fn get_client_connection(
+        &self,
+        user_id: i64,
+        session_id: i64,
+    ) -> Option<Arc<super::client_connection::ClientConnection>> {
+        for rho in self.connections().await {
+            if let Some(client) = rho.get_client_connection(user_id, session_id).await {
+                return Some(client);
+            }
+        }
+        None
+    }
 }