fix(remote-connect): isolate host stream readers and recover live sync

src/apps/desktop/src/api/remote_connect_api.rs

@@ -3947,6 +3947,11 @@ pub async fn account_load_older_session(request: UnsubscribeSessionRequest) -> R
 pub async fn account_subscribe_session(request: SubscribeSessionRequest) -> Result<String, String> {
     let generation = account_context_generation();
     let (session, relay) = read_account_context_for_generation(generation).await?;
+    // Events can arrive before this command returns. Give every attempt an
+    // identity so a late error/page from a retired reader cannot poison its successor.
+    let id = uuid::Uuid::new_v4().to_string();
+    let event_subscription_id = id.clone();
+    let error_subscription_id = id.clone();
     let source = request.target_device_id.clone();
     let error_source = source.clone();
     let error_session_id = request.session_id.clone();
@@ -3960,21 +3965,23 @@ pub async fn account_subscribe_session(request: SubscribeSessionRequest) -> Resu
                 if !account_context_is_current(generation) {
                     anyhow::bail!("account subscription retired");
                 }
-                emit_account_event(&event.event, tag_peer_event_source(event.payload, &source));
+                let mut payload = tag_peer_event_source(event.payload, &source);
+                payload["subscriptionId"] = event_subscription_id.clone().into();
+                emit_account_event(&event.event, payload);
                 Ok(())
             }),
             Arc::new(move |error| {
                 if account_context_is_current(generation) {
                     emit_account_event(
                         "account://session-sync-error",
-                        serde_json::json!({"message":error,"sessionId":error_session_id,"targetDeviceId":error_source}),
+                        serde_json::json!({"message":error,"sessionId":error_session_id,"targetDeviceId":error_source,
+                            "subscriptionId":error_subscription_id,"retrying":true}),
                     );
                 }
             }),
         )
         .await
         .map_err(|e| e.to_string())?;
-    let id = uuid::Uuid::new_v4().to_string();
     let mut subscriptions = SESSION_SUBSCRIPTIONS
         .get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
         .lock()

src/apps/mobile/harmonyos/entry/src/main/ets/model/RemoteModels.ets

@@ -188,6 +188,7 @@ export interface RemoteCommand {
   before?: number;
   epoch?: number;
   subscribe?: boolean;
+  subscription_id?: string;
 }
 
 export interface PairChallengeResponse {

src/apps/mobile/harmonyos/entry/src/main/ets/services/CloudAccountClient.ets

@@ -514,22 +514,29 @@ export class CloudAccountClient {
   subscribeSession(relayUrl: string, session: CloudAccountSession, targetId: string,
     streamId: string, callbacks: HostSessionCallbacks): HostSessionStream {
     const target = targetId.trim();
+    const socket = this.connection(relayUrl, session.token);
+    const subscriptionId = Encoding.randomId('stream');
+    let scopedSubscription = false;
     const source: HostSessionSource = {
       target: target,
-      read: (request: HostStreamReadRequest): Promise<HostStreamPage> => {
-        const command: RemoteCommand = { cmd: 'read_stream', stream_id: streamId, subscribe: true };
+      read: async (request: HostStreamReadRequest): Promise<HostStreamPage> => {
+        if (this.realtime !== socket) throw new Error('Account changed');
+        const command: RemoteCommand = { cmd: 'read_stream', stream_id: streamId, subscribe: true, subscription_id: subscriptionId };
         if (request.after !== undefined) command.after = request.after;
         if (request.before !== undefined) command.before = request.before;
         if (request.epoch !== undefined) command.epoch = request.epoch;
-        return this.deviceRpc<HostStreamPage>(relayUrl, session, target, command, 30000);
+        const page = await this.deviceRpc<HostStreamPage>(relayUrl, session, target, command, 30000);
+        scopedSubscription = page.subscription_id === subscriptionId;
+        return page;
       },
       unsubscribe: async (): Promise<void> => {
+        if (this.realtime !== socket || !scopedSubscription) return;
         await this.deviceRpc<CommandStatusResponse>(relayUrl, session, target,
-          { cmd: 'unsubscribe_stream', stream_id: streamId }, 15000);
+          { cmd: 'unsubscribe_stream', stream_id: streamId, subscription_id: subscriptionId }, 15000);
       },
       onHint: (listener: (hint: HostStreamHint) => void): (() => void) =>
         this.observeStreamHints(relayUrl, session, target, listener),
-      onReconnect: (listener: () => void): (() => void) => this.connection(relayUrl, session.token).onReconnect(listener)
+      onReconnect: (listener: () => void): (() => void) => socket.onReconnect(listener)
     };
     return new HostSessionStream(streamId, source, callbacks);
   }

src/apps/mobile/harmonyos/entry/src/main/ets/services/HostSessionStream.ets

@@ -29,6 +29,7 @@ export interface HostSessionEvent { session_id: string; event: string; payload:
 export interface HostStreamEventWire { seq: number; event: string; payload: Object; }
 /** `RemoteResponse::StreamPage`, flattened, as answered to `read_stream`. */
 export interface HostStreamPage extends CommandStatusResponse {
+  subscription_id?: string;
   stream_id: string;
   epoch: number;
   events: HostStreamEventWire[];

src/apps/mobile/harmonyos/tools/tests/host-stream.test.cjs

@@ -304,3 +304,47 @@ test('lazy disk history keeps JS-safe backward cursors separate from live update
     assert.equal(c.errors.length, 0);
   } finally { stream.close(); }
 });
+
+// Use the production account client factory as well as the stream reader: the
+// subscription acknowledgement is a wire capability, not a reader-only flag.
+test('account stream cleanup negotiates independent leases and leaves old hosts to expiry', async () => {
+  const clientSource = fs.readFileSync(path.join(__dirname, '../../entry/src/main/ets/services/CloudAccountClient.ets'), 'utf8');
+  const exported = {}; let nextId = 0;
+  new Function('require', 'exports', ts.transpileModule(clientSource, { compilerOptions: {
+    target: ts.ScriptTarget.ES2022, module: ts.ModuleKind.CommonJS,
+  } }).outputText)(name => ({
+    './HostSessionStream': exportsObject,
+    './Encoding': { Encoding: { randomId: () => 'reader-' + ++nextId } },
+    './RemoteCrypto': { HarmonyRemoteCryptoCipher: class {} },
+  }[name] || {}), exported);
+  for (const scoped of [true, false]) {
+    const client = new exported.CloudAccountClient(false), commands = [];
+    const socket = { onReconnect: () => () => {} };
+    client.realtime = socket;
+    client.connection = () => { client.realtime = socket; return socket; };
+    client.observeStreamHints = () => () => {};
+    client.deviceRpc = async (_url, _session, _target, command) => {
+      commands.push(command);
+      return { resp: 'stream_page', stream_id: 's', epoch: 1, cursor: 0, oldest_seq: 1,
+        events: [], has_more: false, truncated: false,
+        ...(scoped ? { subscription_id: command.subscription_id } : {}) };
+    };
+    const first = callbacks(), second = callbacks();
+    const a = client.subscribeSession('relay', {}, 'desktop', 's', first.hooks);
+    const b = client.subscribeSession('relay', {}, 'desktop', 's', second.hooks);
+    try {
+      await settle(() => first.caught === 1 && second.caught === 1);
+      assert.notEqual(commands[0].subscription_id, commands[1].subscription_id);
+      a.close(); await tick();
+      const cleanup = commands.filter(command => command.cmd === 'unsubscribe_stream');
+      assert.equal(cleanup.length, scoped ? 1 : 0);
+      if (scoped) assert.equal(cleanup[0].subscription_id, commands[0].subscription_id);
+      const newAccountSocket = { onReconnect: () => () => {} };
+      client.realtime = newAccountSocket;
+      const count = commands.length;
+      b.close(); await tick();
+      assert.equal(commands.length, count, 'retired cleanup cannot invoke on a new account');
+      assert.equal(client.realtime, newAccountSocket, 'retired cleanup cannot reopen its old account');
+    } finally { a.close(); b.close(); }
+  }
+});

src/apps/mobile/shared/core-protocol/src/commonMain/kotlin/com/openbitfun/mobile/core/protocol/RemoteCommand.kt

@@ -106,4 +106,5 @@ public data class RemoteCommand(
     @SerialName("before") val before: Long? = null,
     @SerialName("epoch") val epoch: Long? = null,
     @SerialName("subscribe") val subscribe: Boolean? = null,
+    @SerialName("subscription_id") val subscriptionId: String? = null,
 )

src/apps/mobile/shared/core-transport/src/commonMain/kotlin/com/openbitfun/mobile/core/transport/CloudAccountClient.kt

@@ -6,6 +6,8 @@ import kotlinx.coroutines.withContext
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.filter
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.emitAll
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.drop
 import kotlinx.coroutines.flow.onCompletion
@@ -341,12 +343,15 @@ public class CloudAccountClient internal constructor(
      * encrypted device RPC. The relay forwards ciphertext only, and nothing is
      * written on this device: the transcript lives exactly as long as the flow.
      */
+    @OptIn(ExperimentalUuidApi::class)
     public suspend fun subscribeSession(
         relayUrl: String, session: CloudAccountSession, targetDeviceId: String, sessionId: String,
         onError: (Throwable) -> Unit, onCaughtUp: () -> Unit,
-    ): kotlinx.coroutines.flow.Flow<JsonObject> {
+    ): kotlinx.coroutines.flow.Flow<JsonObject> = flow {
         val socket = connection(relayUrl, session.token)
         val target = targetDeviceId.trim()
+        val subscriptionId = Uuid.random().toString()
+        var scopedSubscription = false
         val historyKey = target + ":" + sessionId
         val historyRequests = Channel<CompletableDeferred<Unit>>(Channel.RENDEZVOUS)
         historyReaders[historyKey] = historyRequests
@@ -356,8 +361,9 @@ public class CloudAccountClient internal constructor(
                 check(realtime.value?.socket === socket) { "Account changed" }
                 val startedAt = TimeSource.Monotonic.markNow()
                 val page = deviceRpc(relayUrl, session, target,
-                    RemoteCommand(cmd = "read_stream", streamId = sessionId, after = after, before = before, epoch = epoch, subscribe = true),
+                    RemoteCommand(cmd = "read_stream", streamId = sessionId, after = after, before = before, epoch = epoch, subscribe = true, subscriptionId = subscriptionId),
                     StreamPageWire.serializer(), RELAY_DEFAULT_TIMEOUT_MS)
+                scopedSubscription = page.subscriptionId == subscriptionId
                 val elapsedMs = startedAt.elapsedNow().inWholeMilliseconds
                 // The opening page (`after == null`) is the one the user waits for,
                 // and a slow page is worth naming wherever it happens; a page per
@@ -368,8 +374,8 @@ public class CloudAccountClient internal constructor(
                 return page
             }
             override suspend fun unsubscribe() {
-                if (realtime.value?.socket !== socket) return
-                deviceRpc(relayUrl, session, target, RemoteCommand(cmd = "unsubscribe_stream", streamId = sessionId),
+                if (realtime.value?.socket !== socket || !scopedSubscription) return
+                deviceRpc(relayUrl, session, target, RemoteCommand(cmd = "unsubscribe_stream", streamId = sessionId, subscriptionId = subscriptionId),
                     CommandStatusResponse.serializer(), RELAY_DEFAULT_TIMEOUT_MS)
             }
         }
@@ -379,7 +385,7 @@ public class CloudAccountClient internal constructor(
         // the gap between the two counts is what the receiving device spends.
         val openedAt = TimeSource.Monotonic.markNow()
         var recordsSeen = 0
-        return hostStream(sessionId, target, hints, merge(socket.connections.drop(1), foregroundResumes), reads,
+        emitAll(hostStream(sessionId, target, hints, merge(socket.connections.drop(1), foregroundResumes), reads,
             olderRequests = historyRequests, onError = { error ->
                 log.warn("host stream read failed stream=${sessionId.take(24)} type=${error::class.simpleName} failure=${(error as? CloudAccountException)?.failure} message=${error.message}")
                 onError(error)
@@ -390,7 +396,7 @@ public class CloudAccountClient internal constructor(
                 log.info("host stream ended stream=${sessionId.take(24)} events=$recordsSeen cause=${cause?.let { it::class.simpleName } ?: "none"}")
                 if (historyReaders[historyKey] === historyRequests) historyReaders.remove(historyKey)
                 historyRequests.close()
-            }
+            })
     }
 
     /**

src/apps/mobile/shared/core-transport/src/commonMain/kotlin/com/openbitfun/mobile/core/transport/HostStream.kt

@@ -4,6 +4,7 @@ import com.openbitfun.mobile.core.protocol.CommandStatus
 import com.openbitfun.mobile.core.protocol.isError
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.CoroutineStart
 import kotlinx.coroutines.NonCancellable
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.currentCoroutineContext
@@ -87,6 +88,7 @@ internal data class StreamPageWire(
     val cursor: Long = -1,
     @SerialName("oldest_seq") val oldestSeq: Long = -1,
     val truncated: Boolean = false,
+    @SerialName("subscription_id") val subscriptionId: String? = null,
 ) : CommandStatus
 
 /** A decrypted `host-stream-changed` device event. */
@@ -314,6 +316,14 @@ internal fun hostStream(
     // Requests the lane took off [wakes] and has not served yet. One sweep can
     // hand it several, and every one of them is owed an answer.
     val waiting = ArrayDeque<CompletableDeferred<Unit>>()
+    // Start collection before the opening RPC. SharedFlow does not replay a
+    // final hint that arrives without a collector while that page is in flight.
+    launch(start = CoroutineStart.UNDISPATCHED) {
+        hints.collect { hint -> if (reader.hintIsNew(hint, target)) wakes.trySend(Wake.Dirty(false)) }
+    }
+    launch(start = CoroutineStart.UNDISPATCHED) {
+        reconnects.collect { wakes.trySend(Wake.Dirty(true)) }
+    }
     // Set once a page was read with `subscribe`; before that there is nothing
     // on the host to release.
     var subscribed = false
@@ -336,8 +346,6 @@ internal fun hostStream(
         awaitConsumption()
         onCaughtUp()
 
-        launch { hints.collect { hint -> if (reader.hintIsNew(hint, target)) wakes.send(Wake.Dirty(false)) } }
-        launch { reconnects.collect { wakes.send(Wake.Dirty(true)) } }
         launch { while (currentCoroutineContext().isActive) { delay(keepaliveMs); wakes.send(Wake.Dirty(false)) } }
         launch { for (request in olderRequests) wakes.send(Wake.Older(request)) }
 

src/apps/mobile/shared/core-transport/src/commonTest/kotlin/com/openbitfun/mobile/core/transport/CloudAccountClientTest.kt

@@ -30,6 +30,8 @@ import kotlinx.coroutines.test.advanceUntilIdle
 import kotlinx.coroutines.test.runCurrent
 import kotlinx.coroutines.test.runTest
 import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.JsonPrimitive
+import kotlin.test.assertNotNull
 import kotlinx.serialization.json.JsonObject
 import kotlinx.serialization.json.buildJsonObject
 import kotlinx.serialization.json.put
@@ -368,6 +370,7 @@ class CloudAccountClientTest {
      */
     @Test
     fun hostStreamRecoversRepeatsOnlyTheReadAndUnsubscribesOnClose() = runTest {
+      for (supportsScoped in listOf(true, false)) {
         val commands = mutableListOf<RemoteCommand>()
         val master = ByteArray(32)
         val peerSecret = ByteArray(32) { 11 }
@@ -387,7 +390,11 @@ class CloudAccountClientTest {
                 "unsubscribe_stream" -> """{"resp":"stream_unsubscribed","stream_id":"session"}"""
                 else -> error("unexpected ${command.cmd}")
             }
-            val encrypted = CloudAccountCipher.encrypt(response.encodeToByteArray(), key, nonce)
+            val answer = if (supportsScoped && command.cmd == "read_stream") {
+                JsonObject(RelayJson.parseToJsonElement(response).jsonObject +
+                    ("subscription_id" to JsonPrimitive(command.subscriptionId))).toString()
+            } else response
+            val encrypted = CloudAccountCipher.encrypt(answer.encodeToByteArray(), key, nonce)
             RelayJson.encodeToJsonElement(EncryptedPayload.serializer(),
                 EncryptedPayload(Base64.Default.encode(encrypted), Base64.Default.encode(nonce)))
         }
@@ -399,11 +406,58 @@ class CloudAccountClientTest {
             "desktop-1", "session", {}, {})
             .onEach { received += it.getValue("event").jsonPrimitive.content }
             .first { it.getValue("event").jsonPrimitive.content == STREAM_EVENT_READY }
-        assertEquals(listOf("read_stream", "read_stream", "unsubscribe_stream"), commands.map { it.cmd })
-        assertEquals(RemoteCommand(cmd = "read_stream", streamId = "session", subscribe = true), commands[1])
+        assertEquals(if (supportsScoped) listOf("read_stream", "read_stream", "unsubscribe_stream")
+            else listOf("read_stream", "read_stream"), commands.map { it.cmd })
+        assertNotNull(commands.first().subscriptionId)
+        assertTrue(commands.all { it.subscriptionId == commands.first().subscriptionId })
+        assertEquals(RemoteCommand(cmd = "read_stream", streamId = "session", subscribe = true), commands[1].copy(subscriptionId = null))
         assertEquals(listOf("session-record", STREAM_EVENT_READY), received)
         assertEquals("session", commands.last().streamId)
         assertEquals(1, keyReads, "the peer key is fetched once and reused for every read")
+      }
+    }
+
+    @Test
+    fun collectingTheSameStreamTwiceOwnsIndependentSubscriptions() = runTest {
+        val master = ByteArray(32)
+        val peerSecret = ByteArray(32) { 11 }
+        val key = DeviceIdentity.messageKey(peerSecret, DeviceIdentity.publicKey(master))
+        val commands = mutableListOf<RemoteCommand>()
+        val client = rpcClient(DeviceIdentity.publicKey(peerSecret)) { _, params ->
+            val envelope = RelayJson.decodeFromJsonElement(EncryptedPayload.serializer(), params)
+            val plain = CloudAccountCipher.decrypt(Base64.Default.decode(envelope.encryptedData), key,
+                Base64.Default.decode(envelope.nonce)).decodeToString()
+            val command = RelayJson.decodeFromString(RemoteCommand.serializer(), plain)
+            commands += command
+            val response = if (command.cmd == "read_stream")
+                """{"resp":"stream_page","stream_id":"session","subscription_id":"${command.subscriptionId}","epoch":1,"events":[],"has_more":false,"cursor":0,"oldest_seq":1,"truncated":false}"""
+            else """{"resp":"stream_unsubscribed","stream_id":"session"}"""
+            val nonce = ByteArray(12) { commands.size.toByte() }
+            val encrypted = CloudAccountCipher.encrypt(response.encodeToByteArray(), key, nonce)
+            RelayJson.encodeToJsonElement(EncryptedPayload.serializer(),
+                EncryptedPayload(Base64.Default.encode(encrypted), Base64.Default.encode(nonce)))
+        }
+        val stream = client.subscribeSession("http://192.168.1.2:9700",
+            CloudAccountSession("token-1", "user-1", master), "desktop-1", "session", {}, {})
+        val ready = Channel<Unit>(Channel.UNLIMITED)
+        val first = launch { stream.collect { if (it["event"]?.jsonPrimitive?.content == STREAM_EVENT_READY) ready.send(Unit) } }
+        ready.receive()
+        val second = launch { stream.collect { if (it["event"]?.jsonPrimitive?.content == STREAM_EVENT_READY) ready.send(Unit) } }
+        try {
+            ready.receive()
+            val ids = commands.filter { it.cmd == "read_stream" }.map { it.subscriptionId }
+            assertEquals(2, ids.size)
+            assertNotNull(ids[0])
+            assertNotNull(ids[1])
+            assertTrue(ids[0] != ids[1], "each collector owns its lease and history lane")
+            first.cancelAndJoin()
+            assertEquals(ids[0], commands.last().subscriptionId)
+            assertEquals("unsubscribe_stream", commands.last().cmd)
+            assertTrue(second.isActive)
+        } finally {
+            first.cancelAndJoin()
+            second.cancelAndJoin()
+        }
     }
 
     @Test