chore(deps): update rust crate livekit-api to 0.6.0

Cargo.lock

@@ -14,18 +14,18 @@ dependencies = [
 
 [[package]]
 name = "aho-corasick"
-version = "1.1.4"
+version = "1.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ddd31a130427c27518df266943a5308ed92d4b226cc639f5a8f1002816174301"
+checksum = "c982642fa9e8606056828ee9a8505737230110bb1099153c79efe865c59d12ba"
 dependencies = [
  "memchr",
 ]
 
 [[package]]
 name = "android_system_properties"
-version = "0.1.5"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311"
+checksum = "ae221649c9976a6f6c56ae1facf410f3ddb33cc661c4b7b61020a912d4237fbc"
 dependencies = [
  "libc",
 ]
@@ -109,9 +109,9 @@ checksum = "f2032f911046de80f0a198e0901378627c33f59ea0ac00e363d481118bd70a53"
 
 [[package]]
 name = "aws-lc-rs"
-version = "1.17.3"
+version = "1.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "00bdb5da18dac48ca2cc7cd4a98e533e8635a58e2361d13a1a4ee3888e0d72f1"
+checksum = "ce2b2dcc879c3bae0d371e77c99f2238400ef24ec001394befa67b6e543add9e"
 dependencies = [
  "aws-lc-sys",
  "untrusted 0.7.1",
@@ -120,9 +120,9 @@ dependencies = [
 
 [[package]]
 name = "aws-lc-sys"
-version = "0.43.0"
+version = "0.44.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43103168cc76fe62678a375e722fc9cb3a0146159ac5828bc4f0dfd755c2224c"
+checksum = "f09fae7be8bb3174e05c6afdb34199e6dc0c7c04ba9fa237b1967adfbde27483"
 dependencies = [
  "cc",
  "cmake",
@@ -143,6 +143,12 @@ version = "0.22.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6"
 
+[[package]]
+name = "base64"
+version = "0.23.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5"
+
 [[package]]
 name = "base64ct"
 version = "1.8.3"
@@ -211,9 +217,9 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04"
 
 [[package]]
 name = "cc"
-version = "1.4.0"
+version = "1.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5add81bb678e6cb321aff7fa0dc7689ad82b112dbc032cea19f91d6b8e3582b9"
+checksum = "9066c49992464636f92905fa096ec58baaa4d57ec19a5c096c68d3e25ef3d136"
 dependencies = [
  "find-msvc-tools",
  "jobserver",
@@ -319,12 +325,6 @@ dependencies = [
  "memchr",
 ]
 
-[[package]]
-name = "const-oid"
-version = "0.9.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
-
 [[package]]
 name = "const-oid"
 version = "0.10.2"
@@ -411,8 +411,23 @@ dependencies = [
  "cfg-if",
  "cpufeatures 0.2.17",
  "curve25519-dalek-derive",
- "digest 0.10.7",
+ "fiat-crypto 0.2.9",
- "fiat-crypto",
+ "rustc_version",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "curve25519-dalek"
+version = "5.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b5eed333089e2e1c1ac8c6c0398e5e2497b4c9926ca6d0365ed1e099afa5bc23"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.3.0",
+ "curve25519-dalek-derive",
+ "digest 0.11.3",
+ "fiat-crypto 0.3.0",
  "rustc_version",
  "subtle",
  "zeroize",
@@ -445,20 +460,9 @@ dependencies = [
 
 [[package]]
 name = "data-encoding"
-version = "2.11.0"
+version = "2.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8"
+checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06"
-
-[[package]]
-name = "der"
-version = "0.7.10"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb"
-dependencies = [
- "const-oid 0.9.6",
- "pem-rfc7468",
- "zeroize",
-]
 
 [[package]]
 name = "der"
@@ -466,7 +470,8 @@ version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a69dedd701da44b0536442edf09c81a64b0ab97a7a4a5e3d1971f00027cbc63d"
 dependencies = [
- "const-oid 0.10.2",
+ "const-oid",
+ "pem-rfc7468",
  "zeroize",
 ]
 
@@ -523,7 +528,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2"
 dependencies = [
  "block-buffer 0.12.1",
- "const-oid 0.10.2",
+ "const-oid",
  "crypto-common 0.2.2",
  "ctutils",
 ]
@@ -563,24 +568,25 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813"
 
 [[package]]
 name = "ed25519"
-version = "2.2.3"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53"
+checksum = "29fcf32e6c73d1079f83ab4d782de2d81620346a5f38c6237a86a22f8368980a"
 dependencies = [
- "pkcs8 0.10.2",
+ "pkcs8",
- "signature 2.2.0",
+ "signature 3.0.0",
 ]
 
 [[package]]
 name = "ed25519-dalek"
-version = "2.2.0"
+version = "3.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9"
+checksum = "6ebaa1a2bf1290ab3bfe5a7b771d050ebffab2711c19a81691c683a5144a25de"
 dependencies = [
- "curve25519-dalek",
+ "curve25519-dalek 5.0.0",
  "ed25519",
  "serde",
- "sha2 0.10.9",
+ "sha2 0.11.0",
+ "signature 3.0.0",
  "subtle",
  "zeroize",
 ]
@@ -616,7 +622,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.59.0",
+<<<<<<< HEAD
+ "windows-sys 0.61.2",
+=======
+ "windows-sys 0.52.0",
+>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
 ]
 
 [[package]]
@@ -644,10 +654,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d"
 
 [[package]]
-name = "find-msvc-tools"
+name = "fiat-crypto"
-version = "0.1.9"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582"
+checksum = "64cd1e32ddd350061ae6edb1b082d7c54915b5c672c389143b9a63403a109f24"
+
+[[package]]
+name = "find-msvc-tools"
+version = "0.1.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de"
 
 [[package]]
 name = "fixedbitset"
@@ -953,9 +969,9 @@ checksum = "1a9fcbcc408c5526c3ab80d534e5c86e7967c1fb7aa0a8c76abd1edc27deb877"
 
 [[package]]
 name = "http"
-version = "1.4.2"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6970f50e31d6fc17d3fa27329444bfa74e196cf62e95052a3f6fee181dba6425"
+checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0"
 dependencies = [
  "bytes",
  "itoa",
@@ -998,9 +1014,9 @@ checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
 
 [[package]]
 name = "hybrid-array"
-version = "0.4.13"
+version = "0.4.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "818356c5132c1fede50f837ca96afbe78ff42413047f4abb886217845e1b6c8c"
+checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b"
 dependencies = [
  "ctutils",
  "typenum",
@@ -1214,9 +1230,9 @@ dependencies = [
 
 [[package]]
 name = "ipnet"
-version = "2.12.0"
+version = "2.12.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2"
+checksum = "6a756c3fac73139e83f14c2d742155dd2b78d3ee56597b419a0579b7bdd6dd78"
 
 [[package]]
 name = "itertools"
@@ -1363,9 +1379,9 @@ dependencies = [
 
 [[package]]
 name = "keccak"
-version = "0.2.0"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9e24a010dd405bd7ed803e5253182815b41bf2e6a80cc3bfc066658e03a198aa"
+checksum = "ffd9697dc4a9a62e2da93389f34400b77a28f0287711263cabb203b3ccb9c0e4"
 dependencies = [
  "cfg-if",
  "cpufeatures 0.3.0",
@@ -1452,9 +1468,15 @@ dependencies = [
 
 [[package]]
 name = "livekit-protocol"
-version = "0.7.11"
+<<<<<<< HEAD
+version = "0.7.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "735a237deeff60124edd7831fcabce32b4b9f667d5c4c5688c9d18a2808422c4"
+checksum = "4d26880e94e2f9bab298445e7d86a3794453d211a12ddbd051bd9991a343f9ff"
+=======
+version = "0.7.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "526f22bddf409e5f15449d55cf341647d7c16a1f43df9db9b05be106e4208e1c"
+>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
 dependencies = [
  "pbjson",
  "pbjson-types",
@@ -1531,12 +1553,12 @@ version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "add6b9d92e496f16f4526d68ff29da1483aba4b119baeab8bed3b9e3544a6f3d"
 dependencies = [
- "const-oid 0.10.2",
+ "const-oid",
  "crypto-common 0.2.2",
  "ctutils",
  "hybrid-array",
  "module-lattice",
- "pkcs8 0.11.0",
+ "pkcs8",
  "shake",
  "signature 3.0.0",
 ]
@@ -1581,7 +1603,7 @@ dependencies = [
 [[package]]
 name = "mtp"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "mtp-client",
  "mtp-codec",
@@ -1597,7 +1619,7 @@ dependencies = [
 [[package]]
 name = "mtp-client"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "mtp-codec",
  "mtp-common",
@@ -1610,9 +1632,9 @@ dependencies = [
 [[package]]
 name = "mtp-codec"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
- "base64 0.22.1",
+ "base64 0.23.1",
  "byteorder",
  "mtp-common",
  "mtp-crypto",
@@ -1623,7 +1645,7 @@ dependencies = [
 [[package]]
 name = "mtp-common"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "quinn",
  "rustls",
@@ -1634,9 +1656,9 @@ dependencies = [
 [[package]]
 name = "mtp-crypto"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
- "base64 0.22.1",
+ "base64 0.23.1",
  "chacha20poly1305",
  "ed25519-dalek",
  "getrandom 0.4.3",
@@ -1648,7 +1670,7 @@ dependencies = [
  "rustls",
  "serde",
  "sha2 0.11.0",
- "thiserror 1.0.69",
+ "thiserror 2.0.19",
  "tokio",
  "zeroize",
 ]
@@ -1656,24 +1678,24 @@ dependencies = [
 [[package]]
 name = "mtp-files"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "mtp-crypto",
  "rand 0.10.2",
- "thiserror 1.0.69",
+ "thiserror 2.0.19",
  "zeroize",
 ]
 
 [[package]]
 name = "mtp-host"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "mtp-codec",
  "mtp-common",
  "mtp-crypto",
  "mtp-transport",
- "rand 0.8.7",
+ "rand 0.10.2",
  "tokio",
  "tracing",
  "wtransport",
@@ -1682,7 +1704,7 @@ dependencies = [
 [[package]]
 name = "mtp-transport"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "async-trait",
  "mtp-codec",
@@ -1700,7 +1722,7 @@ dependencies = [
 [[package]]
 name = "mtp-type-map"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "serde",
  "serde_yaml",
@@ -1709,7 +1731,7 @@ dependencies = [
 [[package]]
 name = "mtp-webserver"
 version = "0.2.0"
-source = "git+https://git.methanium.net/Methanium/mtp.git#a692bed326dbfc8eac1a05825f4a287cbab6fd3e"
+source = "git+https://git.methanium.net/Methanium/mtp.git#b067614a684eb1856bc5db7b3fd82148c036ce6b"
 dependencies = [
  "async-trait",
  "bytes",
@@ -2102,9 +2124,9 @@ dependencies = [
 
 [[package]]
 name = "pem-rfc7468"
-version = "0.7.0"
+version = "1.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412"
+checksum = "a6305423e0e7738146434843d1694d621cce767262b2a86910beab705e4493d9"
 dependencies = [
  "base64ct",
 ]
@@ -2131,24 +2153,14 @@ version = "0.2.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
 
-[[package]]
-name = "pkcs8"
-version = "0.10.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7"
-dependencies = [
- "der 0.7.10",
- "spki 0.7.3",
-]
-
 [[package]]
 name = "pkcs8"
 version = "0.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7"
 dependencies = [
- "der 0.8.1",
+ "der",
- "spki 0.8.0",
+ "spki",
 ]
 
 [[package]]
@@ -2327,7 +2339,11 @@ dependencies = [
  "once_cell",
  "socket2",
  "tracing",
- "windows-sys 0.59.0",
+<<<<<<< HEAD
+ "windows-sys 0.61.2",
+=======
+ "windows-sys 0.52.0",
+>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
 ]
 
 [[package]]
@@ -2474,9 +2490,9 @@ dependencies = [
 
 [[package]]
 name = "regex-automata"
-version = "0.4.16"
+version = "0.4.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fcfdb36bda0c880c5931cdc7a2bcdc8ba4556847b9d912bca70bc94708711ad"
+checksum = "ad8553b9b26413251cbf30e620595c7a41b3887f03da04579c0e6b0d6a06b4b2"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -2581,14 +2597,18 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.59.0",
+<<<<<<< HEAD
+ "windows-sys 0.61.2",
+=======
+ "windows-sys 0.52.0",
+>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
 ]
 
 [[package]]
 name = "rustls"
-version = "0.23.42"
+version = "0.23.43"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138"
+checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06"
 dependencies = [
  "aws-lc-rs",
  "log",
@@ -2640,7 +2660,11 @@ dependencies = [
  "security-framework",
  "security-framework-sys",
  "webpki-root-certs",
- "windows-sys 0.59.0",
+<<<<<<< HEAD
+ "windows-sys 0.61.2",
+=======
+ "windows-sys 0.52.0",
+>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
 ]
 
 [[package]]
@@ -2844,7 +2868,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "09057cb2149ad4cbd2da1e26b351f9a4c354219421229c69c3063e6f61947c4a"
 dependencies = [
  "digest 0.11.3",
- "keccak 0.2.0",
+ "keccak 0.2.1",
  "sponge-cursor",
 ]
 
@@ -2927,16 +2951,6 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
-[[package]]
-name = "spki"
-version = "0.7.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d"
-dependencies = [
- "base64ct",
- "der 0.7.10",
-]
-
 [[package]]
 name = "spki"
 version = "0.8.0"
@@ -2944,7 +2958,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1d9efca8738c78ee9484207732f728b1ef517bbb1833d6fc0879ca898a522f6f"
 dependencies = [
  "base64ct",
- "der 0.8.1",
+ "der",
 ]
 
 [[package]]
@@ -3032,10 +3046,14 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.3.4",
+ "getrandom 0.4.3",
  "once_cell",
  "rustix",
- "windows-sys 0.59.0",
+<<<<<<< HEAD
+ "windows-sys 0.61.2",
+=======
+ "windows-sys 0.52.0",
+>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
 ]
 
 [[package]]
@@ -3080,9 +3098,9 @@ dependencies = [
 
 [[package]]
 name = "time"
-version = "0.3.54"
+version = "0.3.55"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3e1d5e639ff6bab73cb6885cc7e7b1de96c3f32c68ec55f3952614bec1092244"
+checksum = "cdb87b95ec50ddfa440816d227a17b2ccbdda963a316a727fda0fc4334f7d134"
 dependencies = [
  "deranged",
  "num-conv",
@@ -3152,13 +3170,13 @@ dependencies = [
 
 [[package]]
 name = "tokio-macros"
-version = "2.7.1"
+version = "2.7.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6328af13490e73a9b4694030fafd93f8c8c6a9dede33e821c3fc63eddf8042ba"
+checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.119",
+ "syn 3.0.3",
 ]
 
 [[package]]
@@ -3596,7 +3614,11 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.59.0",
+<<<<<<< HEAD
+ "windows-sys 0.61.2",
+=======
+ "windows-sys 0.52.0",
+>>>>>>> e0dd9a085fd88168e714af0d6af88769af3aa644
 ]
 
 [[package]]
@@ -3851,7 +3873,7 @@ version = "2.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c7e468321c81fb07fa7f4c636c3972b9100f0346e5b6a9f2bd0603a52f7ed277"
 dependencies = [
- "curve25519-dalek",
+ "curve25519-dalek 4.1.3",
  "rand_core 0.6.4",
  "serde",
  "zeroize",
@@ -3911,18 +3933,18 @@ dependencies = [
 
 [[package]]
 name = "zerocopy"
-version = "0.8.55"
+version = "0.8.56"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5a105cd7b140f6eeec8acff2ea38135d3cab283ada58540f629fe51e46696eb"
+checksum = "556764e583adb45a9f8d413c2a147fa7e8d821e48e12b14fd560b607998b75eb"
 dependencies = [
  "zerocopy-derive",
 ]
 
 [[package]]
 name = "zerocopy-derive"
-version = "0.8.55"
+version = "0.8.56"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0fe976fb70c78cd64cccfe3a6fc142244e8a77b70959b30faf9d0ac37ee228eb"
+checksum = "f2ab42fc20575779bd240faa45f94a74256f755c0fa9e89f0ede20d91d0cdfc1"
 dependencies = [
  "proc-macro2",
  "quote",

Cargo.toml

@@ -30,7 +30,7 @@ dotenv = "0.15.0"
 strum = "0.28.0"
 strum_macros = "0.28.0"
 livekit-api = { version = "0.6.0", features = ["rustls-tls-native-roots"] }
-livekit-protocol = "0.7.10"
+livekit-protocol = "=0.7.10"
 thiserror = "2.0.19"
 trust-dns-resolver = "0.23.2"
 serde_json = "1.0.151"

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;
             }
 
@@ -299,7 +302,6 @@ impl AnonymousClientConnection {
 
             if cv.is_type(CommunicationType::GetUserData)
                 || cv.is_type(CommunicationType::GetIotaData)
-                || cv.is_type(CommunicationType::DeleteUser)
             {
                 self.handle_omega_forward(cv).await;
                 return;
@@ -321,22 +323,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,36 @@ impl OmegaConnection {
                                     }
                                 }
                             }
+                            if cv.is_type(CommunicationType::EraseHostedUserData) {
+                                let Some(iota_id) = cv
+                                    .get_data(DataType::IotaId)
+                                    .as_number()
+                                    .and_then(|id| i64::try_from(id).ok())
+                                else {
+                                    log_err!(0, PrintType::Omega, "Discarded hosted-data erasure without IotaId");
+                                    continue;
+                                };
+                                if let Some(rho) = self.rho.get_by_iota(iota_id).await {
+                                    rho.get_iota_connection().send_message(&cv).await;
+                                } else {
+                                    log_err!(iota_id, PrintType::Omega, "Hosted-data erasure is pending because the Iota is offline");
+                                }
+                                continue;
+                            }
+                            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 +712,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 +839,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(),
@@ -79,6 +77,12 @@ impl ClientConnection {
         *self.rho_connection.write().await = Some(rho_connection);
     }
 
+    /// A membership snapshot can remove an account from an Iota while keeping
+    /// its authenticated Omega connection alive.
+    pub async fn clear_rho_connection(&self) {
+        *self.rho_connection.write().await = None;
+    }
+
     /// Send a CommunicationValue to the client
     pub async fn send_message(self: Arc<Self>, cv: &CommunicationValue) {
         if !*self.is_open.read().await {
@@ -104,9 +108,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 +212,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 +351,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 +948,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 +969,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;
@@ -350,6 +442,12 @@ impl IotaConnection {
             || cv.is_type(CommunicationType::GetUserData)
             || cv.is_type(CommunicationType::GetIotaData)
             || cv.is_type(CommunicationType::DeleteIota)
+            || cv.is_type(CommunicationType::AttachUserBegin)
+            || cv.is_type(CommunicationType::AttachUserComplete)
+            || cv.is_type(CommunicationType::ReleaseUserFromIota)
+            || cv.is_type(CommunicationType::DeleteUserCredentialBegin)
+            || cv.is_type(CommunicationType::DeleteUserCredentialComplete)
+            || cv.is_type(CommunicationType::EraseHostedUserDataAck)
         {
             let sender = self.get_iota_id().await;
 
@@ -517,7 +615,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 +769,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 +888,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;
@@ -86,6 +86,14 @@ impl RhoConnection {
             .collect()
     }
 
+    pub async fn detach_user_clients(&self, user_id: i64) {
+        let clients = self.get_client_connections_for_user(user_id).await;
+        for client in clients {
+            client.clear_rho_connection().await;
+        }
+        self.client_connections.retain(|(id, _), _| *id != user_id as u64);
+    }
+
     #[allow(unused)]
     pub async fn get_app_connections(
         &self,
@@ -178,11 +186,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 +212,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 +236,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 +270,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

@@ -66,7 +66,11 @@ impl RhoManager {
             )
             .await;
 
+        let replacement = user_ids.iter().copied().collect::<std::collections::HashSet<_>>();
         for user_id in previous_users {
+            if !replacement.contains(&user_id) {
+                rho.detach_user_clients(user_id).await;
+            }
             if self
                 .users
                 .get(&user_id)
@@ -135,4 +139,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
+    }
 }