fleet: peer roster sync — heartbeat pushes rows to peers via SSH

packages/extension/src/extension.ts

@@ -576,6 +576,39 @@ export async function activate(ctx: vscode.ExtensionContext): Promise<void> {
       return amicodeService?.authHeader ?? "";
     },
     now: () => Date.now(),
+    // Peer roster sync: push this machine's heartbeat row to every known
+    // peer's hub service via SSH. Each peer listens on loopback, so we
+    // exec `ssh <alias> curl ...` to POST the row. Failures are swallowed
+    // independently — an unreachable peer never blocks the local update.
+    pushToPeers: async (rowJson: string) => {
+      try {
+        const rosterFile = (await import("./amicode_service/roster")).rosterFilePath();
+        const raw = fs.readFileSync(rosterFile, "utf8");
+        const doc = JSON.parse(raw);
+        const rows: Array<{ machine_id: string; sshAlias?: string }> = doc?.rows ?? [];
+        const topology = readFleetTopology();
+        const localAlias = (topology.kind === "ok" ? topology.canonical?.sshAlias : undefined) ?? "";
+        const localId = (topology.kind === "ok" ? topology.canonical?.host : undefined) ?? "";
+        const port = (topology.kind === "ok" ? topology.canonical?.port : undefined) ?? 4096;
+        const { execFile } = await import("node:child_process");
+        for (const row of rows) {
+          const alias = row.sshAlias;
+          if (!alias) continue;
+          // Skip self — don't SSH to ourselves.
+          if (alias === localAlias || row.machine_id === localId) continue;
+          // Fire-and-forget SSH to the peer's hub service.
+          const escaped = rowJson.replace(/'/g, "'\\''");
+          execFile("ssh", [
+            "-o", "BatchMode=yes",
+            "-o", "ConnectTimeout=5",
+            alias,
+            `curl -sf -X POST http://127.0.0.1:${port}/amicode/roster -H 'Content-Type: application/json' -d '${escaped}'`,
+          ], { timeout: 10_000 }, () => { /* swallow */ });
+        }
+      } catch {
+        // swallow — peer sync is best-effort
+      }
+    },
   });
   fleetHeartbeat.start();
   ctx.subscriptions.push({ dispose: () => { fleetHeartbeat?.dispose(); fleetHeartbeat = undefined; } });

packages/extension/src/fleet_heartbeat.ts

@@ -39,6 +39,11 @@ export interface FleetHeartbeatDeps {
     setInterval: (fn: () => void, ms: number) => unknown;
     clearInterval: (handle: unknown) => void;
   };
+  /** Peer roster sync — push this machine's row to each peer's hub service.
+   *  Each peer listens on loopback, so we go through SSH. The callback
+   *  receives the JSON-serialized row and pushes it to all known peers.
+   *  Absent → local-only (no cross-machine sync). */
+  pushToPeers?: (rowJson: string) => Promise<void>;
 }
 
 export class FleetHeartbeat {
@@ -80,7 +85,8 @@ export class FleetHeartbeat {
     this.stop();
   }
 
-  /** One heartbeat tick: resolve identity, build roster row, POST it. */
+  /** One heartbeat tick: resolve identity, build roster row, POST it locally
+   *  AND push to peers (when wired). */
   async tick(): Promise<void> {
     try {
       const identity = this.deps.resolveIdentity();
@@ -97,14 +103,22 @@ export class FleetHeartbeat {
         health: "reachable" as const,
         last_report: new Date(now).toISOString(),
       };
+      const rowJson = JSON.stringify(row);
+      // Local POST — updates this machine's own roster.
       await this.deps.fetchImpl(`${this.deps.serviceUrl}/amicode/roster`, {
         method: "POST",
         headers: {
           "Content-Type": "application/json",
           Authorization: this.deps.authHeader,
         },
-        body: JSON.stringify(row),
+        body: rowJson,
       });
+      // Peer sync — push the same row to every known peer's hub service.
+      // Swallowed independently: a peer being unreachable never blocks the
+      // local update or other peers.
+      if (this.deps.pushToPeers) {
+        await this.deps.pushToPeers(rowJson).catch(() => {});
+      }
     } catch {
       // Swallow silently — next tick is the retry.
     }