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

src/anonymous_clients/anonymous_client_connection.rs

@@ -8,6 +8,7 @@ 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;
@@ -221,13 +222,9 @@ impl AnonymousClientConnection {
                     .await;
             }
 
-            // Presence is account-scoped and anonymous sessions have no
+            // Handle client status changes
-            // persisted account preference to change.
+            if cv.is_type(CommunicationType::ClientChanged) {
-            if cv.is_type(CommunicationType::SetUserState)
+                self.handle_client_changed(cv).await;
-                || cv.is_type(CommunicationType::ClientChanged)
-            {
-                self.send_error_response(&cv.get_id(), CommunicationType::ErrorNoUserId)
-                    .await;
                 return;
             }
 
@@ -302,6 +299,7 @@ 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;
@@ -323,6 +321,22 @@ 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,12 +1,10 @@
-use std::{env, time::Duration};
+use std::env;
 
 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 {
@@ -21,11 +19,6 @@ 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>,
 }
 
@@ -45,15 +38,6 @@ 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()
@@ -71,8 +55,6 @@ impl Config {
             omega_host,
             omega_port,
             omikron_id,
-            omega_sync_timeout,
-            omega_sync_retries,
             livekit: livekit_from_environment()?,
         })
     }

src/data/user.rs

@@ -1,4 +1,7 @@
-#[derive(Debug, Clone, PartialEq, Eq)]
+use strum::IntoEnumIterator;
+use strum_macros::EnumIter;
+
+#[derive(Debug, Clone, PartialEq, EnumIter, Eq)]
 #[allow(unused, non_camel_case_types)]
 pub enum UserStatus {
     user_offline,
@@ -17,40 +20,12 @@ impl UserStatus {
     pub fn to_string(&self) -> String {
         format!("{:?}", self)
     }
-    pub fn from_client_preference(s: &str) -> Option<Self> {
+    pub fn from_str(s: &str) -> Option<UserStatus> {
-        match s {
+        for sel in UserStatus::iter() {
-            "user_online" => Some(Self::user_online),
+            if &sel.to_string() == s {
-            "user_idle" => Some(Self::user_idle),
+                return Some(sel);
-            "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

@@ -1,145 +0,0 @@
-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,2 +1 @@
-pub mod capabilities;
 pub mod omega_connection;

src/omega/omega_connection.rs

@@ -1,9 +1,6 @@
-use super::capabilities::PeerCapabilities;
 use crate::{
-    config::Config,
+    config::Config, data::user::UserStatus, log_cv_in, log_cv_out, log_err, log_in, log_out,
-    log_cv_in, log_cv_out, log_err, log_in, log_out,
+    rho::rho_manager::RhoManager, util::logger::PrintType,
-    rho::rho_manager::RhoManager,
-    util::{data_type_id, logger::PrintType},
 };
 use dashmap::DashMap;
 use mtp::client::{Client, MTPConnection, Sender};
@@ -16,7 +13,7 @@ use mtp::{
 use mtp_transport::ConnectionHandle;
 use std::{sync::Arc, time::Duration};
 use tokio::{
-    sync::{Mutex, RwLock, mpsc, oneshot, watch},
+    sync::{Mutex, RwLock, mpsc, watch},
     task::JoinHandle,
     time::{Instant, sleep},
 };
@@ -26,7 +23,6 @@ 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);
@@ -34,37 +30,6 @@ 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
 // ============================================================================
@@ -120,33 +85,24 @@ pub fn start_task_cleanup_loop(omega: Arc<OmegaConnection>) {
 pub enum ConnectionState {
     Disconnected,
     Connecting,
-    Identifying,
+    Connected { identified: bool },
-    SynchronizingRoutes,
-    SynchronizingSubscriptions,
-    Ready,
 }
 
 #[allow(unused_variables)]
 impl ConnectionState {
     pub fn is_connected(&self) -> bool {
-        !matches!(
+        match self {
-            self,
+            ConnectionState::Connected { identified } => true,
-            ConnectionState::Disconnected | ConnectionState::Connecting
+            _ => false,
-        )
+        }
     }
 
     #[allow(dead_code)]
     pub fn is_identified(&self) -> bool {
-        matches!(
+        match self {
-            self,
+            ConnectionState::Connected { identified: true } => true,
-            ConnectionState::SynchronizingRoutes
+            _ => false,
-                | ConnectionState::SynchronizingSubscriptions
+        }
-                | ConnectionState::Ready
-        )
-    }
-
-    pub fn is_ready(&self) -> bool {
-        matches!(self, ConnectionState::Ready)
     }
 }
 
@@ -161,8 +117,6 @@ 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
@@ -172,14 +126,6 @@ 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 {
@@ -197,8 +143,6 @@ 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)),
@@ -210,8 +154,6 @@ impl OmegaConnection {
             keyring,
             rho,
             waiting_tasks: DashMap::new(),
-            session_presence: DashMap::new(),
-            peer_capabilities: Arc::new(RwLock::new(PeerCapabilities::default())),
         }
     }
 
@@ -325,7 +267,6 @@ 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)
@@ -361,43 +302,18 @@ impl OmegaConnection {
         // Store sender
         let sender_arc = Arc::new(connection.sender.clone());
         *self.sender.write().await = Some(sender_arc.clone());
-        // A new authenticated transport gets a fresh request circuit. A
+        *self.state.write().await = ConnectionState::Connected { identified: false };
-        // 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_self.read_loop(connection, sender_handle).await;
-                .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.
@@ -438,10 +354,8 @@ impl OmegaConnection {
         }
     }
 
-    async fn sync_client_iota_status(&self) {
+    async fn sync_client_iota_status(self: Arc<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;
@@ -453,120 +367,21 @@ impl OmegaConnection {
 
         for rho in rho_connections {
             for client_conn in rho.get_client_connections().await {
-                let type_map = mtp::codec::TypeMap::latest();
+                connected_user_ids.push(DataValue::SignedNumber(
-                let user_id = DataValue::SignedNumber(client_conn.get_user_id().await.into());
+                    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 peer_capabilities = self.peer_capabilities.read().await.clone();
+        let sync_msg = CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
-        let sync_msg = if peer_capabilities.session_snapshot_v1 {
+            .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids))
-            CommunicationValue::new(CommunicationType::SyncClientIotaStatus)
+            .add_typed_default(DataType::UserIds, DataValue::Array(connected_user_ids))
-                .add_typed_default(DataType::IotaIds, DataValue::Array(connected_iota_ids))
+            .add_typed_default(
-                .add_typed_default(DataType::UserStates, DataValue::Array(connected_sessions))
+                DataType::RhoConnections,
-                .add_typed_default(
+                DataValue::SignedNumber(self.rho.connection_count().await as i128),
-                    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),
-                )
-        };
 
-        if !peer_capabilities.session_snapshot_v1 {
+        self.send_message(&sync_msg).await;
-            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;
-        }
     }
 
     // -------------------------------------------------------------------------
@@ -577,7 +392,6 @@ 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();
@@ -589,14 +403,6 @@ 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()) {
@@ -624,36 +430,6 @@ 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);
@@ -712,14 +488,6 @@ 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(());
@@ -839,371 +607,72 @@ 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) -> Result<(), String> {
+    pub async fn close_iota(&self, iota_id: i64) {
         let cv = CommunicationValue::new(CommunicationType::IotaDisconnected)
             .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()));
-        let request_id = cv.get_id();
+        self.send_message(&cv).await;
-        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_disconnected(&self, user_id: i64, session_id: i64) -> Result<(), String> {
+    pub async fn client_changed(
-        self.session_presence.remove(&(user_id, session_id));
+        &self,
+        _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()),
             );
-        let request_id = cv.get_id();
+        self.send_message(&cv).await;
-        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_connected(
+    pub async fn user_states(&self, user_id: i64, user_ids: Vec<i64>) {
-        &self,
+        let user_ids = user_ids
-        user_id: i64,
+            .iter()
-        session_id: i64,
+            .map(|v| DataValue::SignedNumber((*v).into()))
-        iota_id: i64,
+            .collect();
-    ) -> Result<(), String> {
+
-        let request = CommunicationValue::new(CommunicationType::UserConnected)
+        let cv = CommunicationValue::new(CommunicationType::GetStates)
-            .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
+            .add_typed_default(DataType::UserIds, DataValue::Array(user_ids));
-            .add_typed_default(
+        let msg_id = cv.get_id();
-                DataType::SessionId,
+
-                DataValue::SignedNumber(session_id.into()),
+        let rho_manager = self.rho.clone();
-            )
+        self.waiting_tasks.insert(
-            .add_typed_default(DataType::IotaId, DataValue::SignedNumber(iota_id.into()))
+            msg_id,
-            .add_typed_default(
+            WaitingTask {
-                DataType::UserState,
+                task: Box::new(
-                DataValue::Str("user_online".to_string()),
+                    move |_: Arc<OmegaConnection>, response: CommunicationValue| {
-            )
+                        let rho_manager = rho_manager.clone();
-            .add_typed_default(
+                        tokio::spawn(async move {
-                DataType::UpdatedAt,
+                            let rho = rho_manager.get_for_user(user_id).await;
-                DataValue::SignedNumber(
+                            if let Some(rho) = rho {
-                    std::time::SystemTime::now()
+                                for client in rho.get_client_connections_for_user(user_id).await {
-                        .duration_since(std::time::UNIX_EPOCH)
+                                    client.send_message(&response).await;
-                        .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,
             },
         );
 
-        if !self.is_ready().await {
+        self.send_message(&cv).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,6 +24,7 @@ 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>,
 }
@@ -34,6 +35,7 @@ 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(),
@@ -77,12 +79,6 @@ 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 {
@@ -108,26 +104,9 @@ impl ClientConnection {
 
             let mut cv = cv;
 
-            if cv.is_type(CommunicationType::SetUserState) {
+            // Handle client status changes
-                self.handle_set_user_state(cv).await;
+            if cv.is_type(CommunicationType::ClientChanged) {
-                return;
+                self.handle_client_changed(cv).await;
-            }
-
-            // 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;
             }
 
@@ -212,46 +191,6 @@ 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;
@@ -351,91 +290,34 @@ impl ClientConnection {
         });
     }
     async fn handle_omega_forward(self: Arc<Self>, cv: CommunicationValue) {
-        let request_id = cv.get_id();
+        let client_for_closure = self.clone();
-        match self.await_omega_response(cv).await {
+        let response_cv = self
-            Ok(response_cv) => self.send_message(&response_cv).await,
+            .state
-            Err(_) => {
-                self.send_error_response(request_id, CommunicationType::ErrorInternal)
-                    .await;
-            }
-        }
-    }
-
-    async fn await_omega_response(&self, cv: CommunicationValue) -> Result<CommunicationValue, ()> {
-        self.state
             .omega
             .clone()
             .await_response(&cv.with_sender(self.user_id), Some(Duration::from_secs(20)))
-            .await
+            .await;
-            .map_err(|_| ())
+        if let Ok(response_cv) = response_cv {
+            client_for_closure.send_message(&response_cv).await;
+        }
     }
 
-    async fn request_set_user_state(
+    /// Handle client status change
-        &self,
+    async fn handle_client_changed(self: Arc<Self>, cv: CommunicationValue) {
-        cv: CommunicationValue,
+        let user_id = self.get_user_id().await;
-    ) -> Result<CommunicationValue, CommunicationType> {
+        if let DataValue::Str(status_str) = cv.get_data(DataType::UserState) {
-        if !self.state.omega.is_ready().await {
+            let user_status = UserStatus::from_str(&status_str).unwrap_or(UserStatus::user_online);
-            return Err(CommunicationType::ErrorInternal);
+            if let Some(rho_conn) = self.get_rho_connection().await {
-        }
+                self.state
-        let Some(state) = cv
+                    .omega
-            .get_data(DataType::UserState)
+                    .client_changed(
-            .as_str()
+                        rho_conn.get_iota_id().await as i64,
-            .and_then(UserStatus::from_client_preference)
+                        user_id as i64,
-        else {
+                        self.session_id as i64,
-            return Err(CommunicationType::ErrorInvalidData);
+                        user_status,
-        };
+                    )
-        let Some(rho) = self.get_rho_connection().await else {
+                    .await;
-            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,
         }
     }
 
@@ -948,6 +830,35 @@ 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;
@@ -969,6 +880,7 @@ 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,23 +136,10 @@ impl GeneralConnection {
         if let Some(rho_conn) = rho {
             rho_conn.bind_user_id(user_id).await;
             rho_conn.add_client_connection(client.clone()).await;
-            match self
+            self.notify_user_connected(user_id, rho_conn.get_iota_id().await as i64)
-                .notify_user_connected(user_id, rho_conn.get_iota_id().await as i64)
+                .await;
-                .await
+            self.send_initial_client_state_request(&rho_conn, user_id)
-            {
+                .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,
@@ -168,13 +155,29 @@ impl GeneralConnection {
         true
     }
 
-    async fn notify_user_connected(&self, user_id: i64, iota_id: i64) -> Result<(), String> {
+    async fn notify_user_connected(&self, user_id: i64, iota_id: i64) {
         let session_id = *self.session_id.read().await as i64;
-        self.state
+        let notify = CommunicationValue::new(CommunicationType::UserConnected)
-            .omega
+            .add_typed_default(DataType::UserId, DataValue::SignedNumber(user_id.into()))
-            .clone()
+            .add_typed_default(
-            .user_connected(user_id, session_id, iota_id)
+                DataType::SessionId,
-            .await
+                DataValue::SignedNumber(session_id.into()),
+            )
+            .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>> {
@@ -224,6 +227,9 @@ 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;
@@ -240,14 +246,6 @@ 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,47 +23,6 @@ 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>,
@@ -213,29 +172,6 @@ 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);
@@ -270,34 +206,6 @@ 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;
@@ -442,12 +350,6 @@ 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;
 
@@ -615,20 +517,7 @@ impl IotaConnection {
         }
 
         let mut interested_ids: Vec<i64> = Vec::new();
-        let Some(session_id) = cv
+        let session_id = cv.get_data(DataType::SessionId).as_signed_number();
-            .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
@@ -769,18 +658,20 @@ impl IotaConnection {
         // ============================
         // Notify Omega
         // ============================
-        if let Err(error) = self
+        self.state
-            .replace_state_subscription(user_id as i64, session_id, interested_ids.clone())
+            .omega
-            .await
+            .user_states(user_id as i64, interested_ids.clone())
-        {
+            .await;
-            log_err!(
+
-                self.iota_id as i64,
+        // ============================
-                PrintType::Omega,
+        // Notify Rho
-                "Failed to replace presence subscription for user {} session {}: {}",
+        // ============================
-                user_id,
+        if let Some(rho_conn) = self.get_rho_connection().await {
-                session_id,
+            if let Some(session_id) = session_id.and_then(|id| i64::try_from(id).ok()) {
-                error
+                rho_conn
-            );
+                    .set_interested(user_id as i64, session_id, interested_ids)
+                    .await;
+            }
         }
 
         // ============================
@@ -888,48 +779,3 @@ 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::{log_err, rho::app_connection::AppConnection};
+use crate::{data::user::UserStatus, rho::app_connection::AppConnection};
 use dashmap::DashMap;
 use mtp::codec::{CommunicationValue, DataType};
 use std::sync::Arc;
@@ -86,14 +86,6 @@ 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,
@@ -186,21 +178,11 @@ impl RhoConnection {
             return;
         }
         self.client_connections.remove(&key);
-        if let Err(error) = self
+        self.iota_connection
-            .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()
@@ -212,6 +194,16 @@ 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;
         }
     }
 
@@ -236,14 +228,7 @@ impl RhoConnection {
         }
 
         // Notify OmegaConnection
-        if let Err(error) = self.iota_connection.state.omega.close_iota(iota_id).await {
+        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
@@ -270,6 +255,25 @@ 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,11 +66,7 @@ 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)
@@ -139,17 +135,4 @@ 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
-    }
 }