Merge pull request #2816 from bobleer/bob/communications-validation

fix(remote-connect): cancel relay tasks and prevent duplicate mobile commands

Author: bobleer

docs/architecture/peer-device-mode.md

@@ -239,6 +239,23 @@ FS) and must not be mixed with Peer Device Mode.
 
 ## Transport
 
+- The shared `services-integrations::remote_connect::relay_client` owns one
+  cancellable connection task. Its read and write futures remain full-duplex,
+  while heartbeat, dial, write deadlines and reconnect belong to that same
+  lifetime. Disconnect joins cancellation; replacing or dropping the client
+  retires the old socket and its reconnect attempts. A generation fence prevents
+  an old connection from publishing state into its replacement. Initial dial
+  failure returns to `Disconnected`. Reconnect restores room/account context,
+  including a server-assigned room id, before admitting new outgoing commands.
+  The outgoing queue holds at most 64 messages and reports saturation explicitly;
+  its failed-socket contents are never replayed. Dial/write deadlines are 15s,
+  heartbeat cadence is 30s, with due heartbeats taking priority over queued
+  commands, and inbound idle detection is 75s. These transport
+  facts do not imply authentication success or application-level acceptance.
+- Mobile delegated-auth recovery retries only a real Relay HTTP 401. An
+  authenticated, encrypted host error mentioning `Unauthorized` or an upstream
+  `HTTP 401` is an application result and must not cause a second mutation.
+  Account-generation fences still apply before and after credential refresh.
 - Controller: `PeerDeviceTransportAdapter` wraps product `invoke` as
   `RemoteCommand::HostInvoke` over `account_device_rpc`.
 - HostInvoke on the controller is **priority-queued** with four requests in

docs/development/communications-e2e.zh-CN.md

@@ -0,0 +1,59 @@
+// Export reviewable coverage worksheets; this is not a test or a second registry.
+import { readFile, mkdir, writeFile } from 'node:fs/promises';
+import { dirname, resolve } from 'node:path';
+import { fileURLToPath } from 'node:url';
+
+const root = resolve(dirname(fileURLToPath(import.meta.url)), '../..');
+const args = process.argv.slice(2);
+if (args.length !== 2 || args[0] !== '--output') {
+  throw new Error('Usage: node scripts/diagnostics/export-communications-matrix.mjs --output <new-directory>');
+}
+const output = resolve(args[1]);
+const registry = JSON.parse(await readFile(resolve(root,
+  'src/crates/contracts/product-domains/src/generated/remote-surface-registry.json'), 'utf8'));
+if (registry.schemaVersion !== 1 || !Array.isArray(registry.operations)) {
+  throw new Error('Unsupported Product Operation Registry shape');
+}
+const source = await readFile(resolve(root,
+  'src/crates/services/services-integrations/src/remote_connect.rs'), 'utf8');
+const commandBody = source.match(/pub enum RemoteCommand \{\n([\s\S]*?)\n\}/)?.[1];
+if (!commandBody) throw new Error('RemoteCommand enum was not found');
+const commands = [...commandBody.matchAll(/^    ([A-Z]\w*)\s*(?:,|\{)/gm)]
+  .map((match) => match[1].replace(/([a-z0-9])([A-Z])/g, '$1_$2').toLowerCase());
+if (commands.length === 0 || new Set(commands).size !== commands.length) {
+  throw new Error('RemoteCommand inventory is empty or contains duplicates');
+}
+const guide = await readFile(resolve(root, 'docs/development/communications-e2e.zh-CN.md'), 'utf8');
+const cases = [...guide.matchAll(/^\| ((?:ENV|RLY|SSH|MOB|BOT|PEER|DSP|COMBO|EXT)-\d+) \| ([^\n]+) \| ([^\n]+) \|$/gm)]
+  .map((match) => [match[1], match[2], match[3]]);
+if (cases.length === 0 || new Set(cases.map(([id]) => id)).size !== cases.length) {
+  throw new Error('Manual case inventory is empty or contains duplicate ids');
+}
+const csv = (rows) => rows.map((row) => row.map((cell) =>
+  `"${String(cell ?? '').replaceAll('"', '""')}"`).join(',')).join('\n') + '\n';
+// A fresh directory prevents a rerun from erasing manually recorded evidence.
+await mkdir(output, { recursive: false });
+await writeFile(resolve(output, 'product-operations.csv'), csv([
+  ['operation', 'remote_workspace_policy', 'peer_policy', 'cli_peer_policy', 'cli_reason',
+    'ssh_result', 'desktop_peer_result', 'cli_peer_result', 'composition_result', 'evidence'],
+  ...registry.operations.map((op) => [op.id, op.remoteWorkspace, op.peer.kind, op.cliPeer.kind,
+    op.cliPeer.reason, 'NOT_RUN', 'NOT_RUN', 'NOT_RUN', 'NOT_RUN', '']),
+]));
+await writeFile(resolve(output, 'remote-commands.csv'), csv([
+  ['command', 'room_mobile_result', 'account_mobile_result', 'feishu_result',
+    'telegram_result', 'weixin_result', 'peer_result', 'evidence'],
+  ...commands.map((command) => [command, ...Array(6).fill('NOT_RUN'), '']),
+]));
+await writeFile(resolve(output, 'manual-cases.csv'), csv([
+  ['case_id', 'steps', 'expected', 'environment', 'result', 'evidence', 'issue'],
+  ...cases.map(([id, steps, expected]) => [id, steps, expected, '', 'NOT_RUN', '', '']),
+]));
+const inventory = {
+  registryDigest: registry.digest,
+  productOperations: registry.operations.length,
+  remoteCommands: commands.length,
+  manualCases: cases.length,
+  note: 'Inventory only. NOT_RUN is not coverage. Preserve environment and evidence per execution.',
+};
+await writeFile(resolve(output, 'inventory.json'), JSON.stringify(inventory, null, 2) + '\n');
+console.log(JSON.stringify(inventory, null, 2));

scripts/diagnostics/export-communications-matrix.mjs

@@ -193,15 +193,21 @@ test(
     });
     const [chunk] = (await once(transport.readable, "data")) as [Buffer];
     const descendantPid = Number.parseInt(chunk.toString("utf8").trim(), 10);
-    if (!transport.readable.readableEnded) {
-      await once(transport.readable, "end");
-    }
-    assert.equal(transport.hasExited(), true);
-
     try {
+      if (!transport.readable.readableEnded) {
+        await once(transport.readable, "end");
+      }
+      // Pipe EOF can arrive before Node delivers ChildProcess's exit event.
+      // Establish actual parent exit before testing orphan-group cleanup.
+      const deadline = Date.now() + 1_000;
+      while (!transport.hasExited() && Date.now() < deadline) {
+        await new Promise((resolve) => setTimeout(resolve, 10));
+      }
+      assert.equal(transport.hasExited(), true);
       await transport.close();
       assert.equal(await processExited(descendantPid), true);
     } finally {
+      await transport.close().catch(() => {});
       if (!(await processExited(descendantPid, 50))) {
         try {
           process.kill(descendantPid);

sdk/typescript/test/managed-host.test.ts

@@ -1779,6 +1779,7 @@ mod tests {
         openbitfun_core::native_hooks::NativeHookOverview {
             enabled: true,
             project_hooks_enabled: true,
+            remote_workspace_unsupported: false,
             files: vec![
                 openbitfun_core::native_hooks::NativeHookFileView {
                     scope: "user",

src/crates/interfaces/app-server/src/management/owner.rs

@@ -111,6 +111,7 @@ cargo check -p openbitfun-services-integrations --no-default-features
 cargo test -p openbitfun-services-integrations --no-default-features --features mcp --test mcp_contracts
 cargo test -p openbitfun-services-integrations --no-default-features --features remote-ssh --test remote_ssh_contracts remote_ssh_disabled_contracts::
 cargo test -p openbitfun-services-integrations --no-default-features --features remote-ssh-concrete --lib remote_ssh::manager::tests::workspace_
+cargo test --locked -p openbitfun-services-integrations --no-default-features --features remote-connect --lib remote_connect::relay_client::tests::
 cargo test -p openbitfun-services-integrations --no-default-features --features file-watch --test file_watch_contracts
 cargo test --locked -p openbitfun-services-integrations --no-default-features --features deep-research --lib deep_research::tests::
 pnpm run check:core-boundaries