From 89dc9d0f7e79b55f7218b45dc5fe8b08e364fdaf Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Fri, 24 Jul 2026 04:48:48 +0000
Subject: [PATCH 1/9] feat(sync): add clientId/originatorClientId wire fields
 for P2 Plan A

---
 packages/shared/src/types/api/canvas-sync.ts | 8 ++++++++
 packages/shared/src/types/api/canvas.ts      | 6 ++++++
 2 files changed, 14 insertions(+)

diff --git a/packages/shared/src/types/api/canvas-sync.ts b/packages/shared/src/types/api/canvas-sync.ts
index a6c5567ee..4b37cc5da 100644
--- a/packages/shared/src/types/api/canvas-sync.ts
+++ b/packages/shared/src/types/api/canvas-sync.ts
@@ -62,6 +62,14 @@ export const canvasSyncEventSchema = z.discriminatedUnion('type', [
        * the API layer; the client casts.
        */
       changes: z.array(z.unknown()).optional(),
+      /**
+       * Opaque id of the client (tab) that originated this update (P2 /
+       * Plan A user-edit broadcast). The originating tab filters its own
+       * echo by comparing this against its local client id; absent for
+       * server-originated writes (e.g. headless `/execute`) that no tab
+       * needs to de-dupe.
+       */
+      originatorClientId: z.string().optional(),
     }),
   }),
 ]);
diff --git a/packages/shared/src/types/api/canvas.ts b/packages/shared/src/types/api/canvas.ts
index 095663137..c3fd348a8 100644
--- a/packages/shared/src/types/api/canvas.ts
+++ b/packages/shared/src/types/api/canvas.ts
@@ -96,6 +96,12 @@ export const putCanvasBodySchema = z.object({
   version: z.number().int().nonnegative(),
   state: z.unknown(),
   title: z.string().min(1).optional(),
+  /**
+   * Opaque per-tab client id (P2 / Plan A). Echoed back on the sync
+   * broadcast as `originatorClientId` so the originating tab can skip
+   * its own PUT echo instead of re-applying its own change.
+   */
+  clientId: z.string().optional(),
 });
 export type PutCanvasRequest = z.infer<typeof putCanvasBodySchema>;
 

From 34a1c6507b9f5fa9f9b6f1c71c75af56a82b3f8f Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Fri, 24 Jul 2026 04:53:30 +0000
Subject: [PATCH 2/9] feat(sync): broadcast user PUT edits via diff + wrap PUT
 in canvas mutex (P2 Plan A)

---
 .../src/modules/canvas/canvas-executor.ts     |  91 +++++++
 apps/server/src/modules/canvas/canvas-sync.ts |  11 +-
 .../server/src/modules/canvas/canvas.route.ts | 230 ++++++++++--------
 3 files changed, 231 insertions(+), 101 deletions(-)

diff --git a/apps/server/src/modules/canvas/canvas-executor.ts b/apps/server/src/modules/canvas/canvas-executor.ts
index 626cb8566..98d73d0ef 100644
--- a/apps/server/src/modules/canvas/canvas-executor.ts
+++ b/apps/server/src/modules/canvas/canvas-executor.ts
@@ -1269,3 +1269,94 @@ export async function applyDeltasOnServer(input: {
     };
   });
 }
+
+/**
+ * P2 / Plan A — broadcast a user structure edit to other live tabs.
+ *
+ * The `PUT /:canvasId` autosave persists the *whole* (slim) canvas state
+ * and, unlike `/execute`, produces no deltas of its own. To propagate a
+ * hand-edit to other tabs we diff the pre- vs post-write topology here
+ * and publish the resulting structural deltas on the same sync channel
+ * the headless executor uses.
+ *
+ * Both sides are hydrated from the *current* `.md` sidecars before
+ * diffing: a structure PUT never touches sidecar content (that rides the
+ * separate per-node content endpoint), so hydrating both with today's
+ * sidecars yields `REPLACE_NODE` rows that differ only in geometry /
+ * parenthood — never a spurious content wipe on the receiver.
+ *
+ * `pendingEffects` is intentionally minimal (no `mutatedNodes` /
+ * `contentEditedNodeIds`) so receivers apply the structure but do NOT
+ * run preprocessing or frame-fit for a plain geometry move — only
+ * `deletedNodeIds` is surfaced so they can forget the removed sidecars.
+ *
+ * The caller MUST already hold the per-canvas mutex (the PUT route wraps
+ * its read → version-check → write → this call in `withCanvasMutex`), so
+ * this function does not re-lock. Returns the deltas it broadcast (empty
+ * when the write was a no-op / geometry-identical).
+ */
+export function broadcastCanvasStatePut(input: {
+  canvasId: string;
+  fromVersion: number;
+  toVersion: number;
+  prevNodes: readonly CanvasNode[];
+  prevEdges: readonly CanvasEdge[];
+  nextNodes: readonly CanvasNode[];
+  nextEdges: readonly CanvasEdge[];
+  clientId?: string;
+}): Delta[] {
+  const {
+    canvasId,
+    fromVersion,
+    toVersion,
+    prevNodes,
+    prevEdges,
+    nextNodes,
+    nextEdges,
+    clientId,
+  } = input;
+
+  const store = getCanvasStore(canvasId);
+  const hydratedPrev = hydrateNodes(store, prevNodes);
+  const hydratedNext = hydrateNodes(store, nextNodes);
+
+  const deltas = diffCanvasState(
+    { nodes: hydratedPrev, edges: prevEdges as CanvasEdge[] },
+    { nodes: hydratedNext, edges: nextEdges as CanvasEdge[] },
+  );
+
+  if (deltas.length === 0) return deltas;
+
+  const deletedNodeIds = deltas
+    .filter((d): d is Extract<Delta, { type: 'DELETE_NODE' }> =>
+      d.type === 'DELETE_NODE',
+    )
+    .map((d) => d.node.id);
+
+  const logEntry: DeltaLogEntry = {
+    version: toVersion,
+    ts: Date.now(),
+    commands: [],
+    deltas: deltas as unknown[],
+    originator: { source: 'ui', ...(clientId ? { tabId: clientId } : {}) },
+  };
+  store.appendDeltaLogEntry(logEntry);
+
+  publishCanvasUpdate(canvasId, {
+    type: 'update',
+    data: {
+      fromVersion,
+      toVersion,
+      deltas,
+      pendingEffects: {
+        mutatedNodes: [],
+        deletedNodeIds,
+        contentEditedNodeIds: [],
+        deferredFitFrameIds: [],
+      },
+      ...(clientId ? { originatorClientId: clientId } : {}),
+    },
+  });
+
+  return deltas;
+}
diff --git a/apps/server/src/modules/canvas/canvas-sync.ts b/apps/server/src/modules/canvas/canvas-sync.ts
index a18978b46..b2d7cf515 100644
--- a/apps/server/src/modules/canvas/canvas-sync.ts
+++ b/apps/server/src/modules/canvas/canvas-sync.ts
@@ -12,11 +12,12 @@
  *
  * ALL canvas writes broadcast — the out-of-band HTTP `/execute` route
  * (ACP / headless) AND the built-in / question-node agents that mutate
- * in-process via `executeOnServer` (C2). The chat SSE tool result no
- * longer applies canvas state, so the initiating tab is a plain receiver
- * that applies its own change once, from this broadcast. There is no
- * per-client echo filtering yet (`clientId` is deferred to P2, needed
- * only once user hand-edits also broadcast).
+ * in-process via `executeOnServer` (C2), plus user hand-edits via the
+ * autosave PUT (P2 / Plan A, `broadcastCanvasStatePut`). The chat SSE
+ * tool result no longer applies canvas state, so the initiating tab is a
+ * plain receiver that applies its own change once, from this broadcast.
+ * User-edit broadcasts carry `originatorClientId` so the originating tab
+ * skips its own PUT echo.
  */
 
 import type { CanvasSyncEvent } from '@huabu/shared';
diff --git a/apps/server/src/modules/canvas/canvas.route.ts b/apps/server/src/modules/canvas/canvas.route.ts
index cf1e0e6c1..bfcc8b9c9 100644
--- a/apps/server/src/modules/canvas/canvas.route.ts
+++ b/apps/server/src/modules/canvas/canvas.route.ts
@@ -24,13 +24,21 @@ import {
   setPortalNodePinsCommandSchema,
 } from '@huabu/shared';
 import { nodeRevisionOf } from '@huabu/shared/canvas-engine';
+import type {
+  CanvasEdge,
+  CanvasNode,
+} from '@huabu/shared/canvas-engine';
 
 import {
   CanvasCommandRoutingError,
   executeCanvasCommandsOnHost,
   MissingWorldPortalError,
 } from './canvas-command-router.js';
-import { CanvasNotFoundError, applyDeltasOnServer } from './canvas-executor.js';
+import {
+  CanvasNotFoundError,
+  applyDeltasOnServer,
+  broadcastCanvasStatePut,
+} from './canvas-executor.js';
 import { searchCanvas } from './canvas-search.js';
 import { publishCanvasUpdate } from './canvas-sync.js';
 import {
@@ -60,6 +68,7 @@ import {
   getStructuredStore,
   listCanvases,
   updateNode,
+  withCanvasMutex,
   type CanvasFile,
   type UpdateNodeOutcome,
 } from '../storage/index.js';
@@ -1169,115 +1178,144 @@ const canvasRoutes: FastifyPluginAsync = async (fastify) => {
       return reply.code(400).send({ message: 'Invalid request body' });
     }
 
-    const { version: clientVersion, state, title } = parsed.data;
-    const incomingState = state as {
-      nodes?: NodeLike[];
-      edges?: unknown[];
-      [key: string]: unknown;
-    };
-
-    const store = getCanvasStore(canvasId);
-    const existing = store.read();
-    const serverVersion = existing?.version ?? 0;
-    if (clientVersion !== serverVersion) {
-      return reply.code(409).send({
-        code: 'CANVAS_VERSION_CONFLICT',
-        message: 'Canvas version mismatch',
-        serverVersion,
-      } satisfies CanvasConflictResponse);
-    }
-
-    try {
-      assertWorldPortalTopologyAllowed(
-        canvasId,
-        (existing?.state.nodes ?? []) as NodeLike[],
-        incomingState.nodes ?? [],
-      );
-    } catch (error) {
-      if (error instanceof WorldPortalMutationError) {
-        return reply.code(409).send({ message: error.message });
-      }
-      throw error;
-    }
+    const { version: clientVersion, state, title, clientId } = parsed.data;
+
+    // Serialize the whole read → version-check → rename → write → broadcast
+    // against concurrent `/execute` batches (which hold the same per-canvas
+    // mutex), so a hand-edit PUT can no longer interleave with an agent
+    // write and silently lose an update. P2 / Plan A additionally diffs the
+    // pre- vs post-write topology inside this critical section and
+    // broadcasts the hand-edit to other live tabs.
+    return await withCanvasMutex(canvasId, async () => {
+      const incomingState = state as {
+        nodes?: NodeLike[];
+        edges?: unknown[];
+        [key: string]: unknown;
+      };
 
-    // Title rename (and the directory rename it implies) happens
-    // before any node persistence so a 409 doesn't half-apply changes.
-    const previousTitle = existing?.title ?? null;
-    const nextTitle = title ?? previousTitle;
-    if (typeof title === 'string' && title !== previousTitle) {
-      // Release any handle held inside this Space's directory across the
-      // rename: on Windows a live `fs.watch` handle makes `renameSync` fail
-      // with EPERM (see `withSpaceDirHandlesReleased`).
-      const renameResult = await withSpaceDirHandlesReleased(canvasId, () =>
-        store.renameSelf(title),
-      );
-      if (!renameResult.ok && renameResult.reason === 'conflict') {
+      const store = getCanvasStore(canvasId);
+      const existing = store.read();
+      const serverVersion = existing?.version ?? 0;
+      if (clientVersion !== serverVersion) {
         return reply.code(409).send({
-          code: 'CANVAS_TITLE_CONFLICT',
-          message: `Another canvas already uses the directory name "${renameResult.conflictWith}"`,
-          conflictWith: renameResult.conflictWith,
+          code: 'CANVAS_VERSION_CONFLICT',
+          message: 'Canvas version mismatch',
+          serverVersion,
         } satisfies CanvasConflictResponse);
       }
-      if (!renameResult.ok && renameResult.reason === 'forbidden') {
-        return reply
-          .code(403)
-          .send({ message: 'World canvas cannot be renamed' });
+
+      try {
+        assertWorldPortalTopologyAllowed(
+          canvasId,
+          (existing?.state.nodes ?? []) as NodeLike[],
+          incomingState.nodes ?? [],
+        );
+      } catch (error) {
+        if (error instanceof WorldPortalMutationError) {
+          return reply.code(409).send({ message: error.message });
+        }
+        throw error;
       }
-      if (!renameResult.ok && renameResult.reason === 'fs-error') {
-        request.log.error(
-          { canvasId, err: renameResult.message },
-          'Failed to rename canvas directory',
+
+      // Title rename (and the directory rename it implies) happens
+      // before any node persistence so a 409 doesn't half-apply changes.
+      const previousTitle = existing?.title ?? null;
+      const nextTitle = title ?? previousTitle;
+      if (typeof title === 'string' && title !== previousTitle) {
+        // Release any handle held inside this Space's directory across the
+        // rename: on Windows a live `fs.watch` handle makes `renameSync` fail
+        // with EPERM (see `withSpaceDirHandlesReleased`).
+        const renameResult = await withSpaceDirHandlesReleased(canvasId, () =>
+          store.renameSelf(title),
         );
-        return reply.code(500).send({ message: 'Failed to rename canvas' });
+        if (!renameResult.ok && renameResult.reason === 'conflict') {
+          return reply.code(409).send({
+            code: 'CANVAS_TITLE_CONFLICT',
+            message: `Another canvas already uses the directory name "${renameResult.conflictWith}"`,
+            conflictWith: renameResult.conflictWith,
+          } satisfies CanvasConflictResponse);
+        }
+        if (!renameResult.ok && renameResult.reason === 'forbidden') {
+          return reply
+            .code(403)
+            .send({ message: 'World canvas cannot be renamed' });
+        }
+        if (!renameResult.ok && renameResult.reason === 'fs-error') {
+          request.log.error(
+            { canvasId, err: renameResult.message },
+            'Failed to rename canvas directory',
+          );
+          return reply.code(500).send({ message: 'Failed to rename canvas' });
+        }
       }
-    }
 
-    const timestamp = nowMs();
-    const nextVersion = serverVersion + 1;
+      const timestamp = nowMs();
+      const nextVersion = serverVersion + 1;
 
-    const rawState = incomingState;
+      const rawState = incomingState;
 
-    const slimNodes = stripNodesForCanvas(
-      (rawState?.nodes ?? []) as NodeLike[],
-    );
+      const slimNodes = stripNodesForCanvas(
+        (rawState?.nodes ?? []) as NodeLike[],
+      );
 
-    const canvasFile: CanvasFile = {
-      canvasId,
-      title: nextTitle,
-      version: nextVersion,
-      state: {
-        ...rawState,
-        nodes: slimNodes,
-        edges: rawState?.edges ?? [],
-      },
-      createdAt: existing?.createdAt ?? timestamp,
-      updatedAt: timestamp,
-    };
+      // Capture pre-write topology BEFORE overwriting so Plan A can diff
+      // old → new. The stored state is already slim; the broadcast helper
+      // hydrates both sides from the current sidecars before diffing.
+      const prevNodes = (existing?.state?.nodes ?? []) as CanvasNode[];
+      const prevEdges = (existing?.state?.edges ?? []) as CanvasEdge[];
 
-    // A title rename may have yielded while an async Space deletion or a
-    // competing PUT completed. Recheck immediately before the synchronous
-    // write so a stale request that initially observed a real Space cannot
-    // recreate it after deletion (or overwrite a newer version). The legacy
-    // implicit-create path remains only for requests whose initial read was
-    // genuinely absent.
-    if (existing) {
-      const current = store.read();
-      if (!current) {
-        return reply.code(404).send({ message: 'Canvas not found' });
-      }
-      if (current.version !== existing.version) {
-        return reply.code(409).send({
-          code: 'CANVAS_VERSION_CONFLICT',
-          message: 'Canvas version mismatch',
-          serverVersion: current.version,
-        } satisfies CanvasConflictResponse);
+      const canvasFile: CanvasFile = {
+        canvasId,
+        title: nextTitle,
+        version: nextVersion,
+        state: {
+          ...rawState,
+          nodes: slimNodes,
+          edges: rawState?.edges ?? [],
+        },
+        createdAt: existing?.createdAt ?? timestamp,
+        updatedAt: timestamp,
+      };
+
+      // A title rename may have yielded while an async Space deletion or a
+      // competing PUT completed. Recheck immediately before the synchronous
+      // write so a stale request that initially observed a real Space cannot
+      // recreate it after deletion (or overwrite a newer version). The legacy
+      // implicit-create path remains only for requests whose initial read was
+      // genuinely absent.
+      if (existing) {
+        const current = store.read();
+        if (!current) {
+          return reply.code(404).send({ message: 'Canvas not found' });
+        }
+        if (current.version !== existing.version) {
+          return reply.code(409).send({
+            code: 'CANVAS_VERSION_CONFLICT',
+            message: 'Canvas version mismatch',
+            serverVersion: current.version,
+          } satisfies CanvasConflictResponse);
+        }
       }
-    }
-    store.write(canvasFile);
+      store.write(canvasFile);
 
-    return reply.send({
-      canvasId,
-      version: nextVersion,
+      // Plan A: broadcast the structural diff to other live tabs. No-op
+      // safe (empty diff → no publish); skips the originating tab's own
+      // echo via `originatorClientId`.
+      broadcastCanvasStatePut({
+        canvasId,
+        fromVersion: serverVersion,
+        toVersion: nextVersion,
+        prevNodes,
+        prevEdges,
+        nextNodes: slimNodes as CanvasNode[],
+        nextEdges: (rawState?.edges ?? []) as CanvasEdge[],
+        ...(clientId ? { clientId } : {}),
+      });
+
+      return reply.send({
+        canvasId,
+        version: nextVersion,
+      });
     });
   });
 

From ac6deb30275ad706ac554a638ab2c299998344bc Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Fri, 24 Jul 2026 04:56:56 +0000
Subject: [PATCH 3/9] feat(sync): mint per-tab clientId, send on PUT, filter
 self-echo (P2 Plan A)

---
 apps/web/src/api/canvas.ts            |  6 +++++-
 apps/web/src/store/canvasSyncStore.ts | 15 +++++++++++++--
 apps/web/src/store/clientId.ts        | 14 ++++++++++++++
 3 files changed, 32 insertions(+), 3 deletions(-)
 create mode 100644 apps/web/src/store/clientId.ts

diff --git a/apps/web/src/api/canvas.ts b/apps/web/src/api/canvas.ts
index edf00dbec..2968981bf 100644
--- a/apps/web/src/api/canvas.ts
+++ b/apps/web/src/api/canvas.ts
@@ -3,6 +3,7 @@
 
 import { ApiError, apiFetch, apiUrl } from './_client';
 import { routes } from './_routes';
+import { CLIENT_ID } from '../store/clientId';
 
 import type {
   ApiErrorBody,
@@ -170,7 +171,10 @@ export async function putCanvas(
     headers: {
       'Content-Type': 'application/json',
     },
-    body: JSON.stringify(request),
+    // Stamp this tab's client id so the server can echo it back on the
+    // sync broadcast (`originatorClientId`) and this tab skips its own PUT
+    // echo (P2 / Plan A). Any explicit `request.clientId` wins.
+    body: JSON.stringify({ clientId: CLIENT_ID, ...request }),
     keepalive: options?.keepalive ?? false,
   });
 
diff --git a/apps/web/src/store/canvasSyncStore.ts b/apps/web/src/store/canvasSyncStore.ts
index 3d4a23dc5..881fe75a1 100644
--- a/apps/web/src/store/canvasSyncStore.ts
+++ b/apps/web/src/store/canvasSyncStore.ts
@@ -8,6 +8,7 @@ import { canvasSyncStreamUrl } from '@/api/canvasSync';
 import { dismissToast, toast } from '@/components/Common/Toast';
 import { useAcpThreadChangesStore } from '@/store/acpThreadChangesStore';
 import useCanvasStore from '@/store/canvasStore';
+import { CLIENT_ID } from '@/store/clientId';
 import { usePanelStore } from '@/store/panelStore';
 import { openPreviewNode } from '@/store/previewWorkspace/actions';
 
@@ -90,10 +91,15 @@ function notifySkippedAgentWrites(
     return typeof label === 'string' && label.trim() ? label : 'a note';
   };
   const names = skippedNodeIds.map(labelOf);
+  // User hand-edit broadcasts (P2 / Plan A) carry no `threadId`; an agent
+  // batch does. Word the notice accordingly so a dropped edit from another
+  // window doesn't read as “the agent”.
+  const actor = threadId ? 'The agent’s' : 'A';
+  const source = threadId ? '' : ' from another window';
   const message =
     names.length === 1
-      ? `The agent's change to “${names[0]}” was skipped because you were editing it — your version was kept.`
-      : `The agent's changes to ${names.length} nodes were skipped because you were editing them — your versions were kept.`;
+      ? `${actor} change to “${names[0]}”${source} was skipped because you were editing it — your version was kept.`
+      : `${actor} changes to ${names.length} nodes${source} were skipped because you were editing them — your versions were kept.`;
   if (conflictToastId) dismissToast(conflictToastId);
   conflictToastId = toast(message, {
     tone: 'warning',
@@ -150,6 +156,11 @@ export const useCanvasSyncStore = create<CanvasSyncState>((set, get) => ({
             }
 
             // event.type === 'update'
+            // Skip our own PUT echo (P2 / Plan A): the originating tab has
+            // already applied this edit optimistically, so re-applying the
+            // broadcast would be redundant (and could fight a still-pending
+            // local edit). Server-originated writes carry no id and pass.
+            if (event.data.originatorClientId === CLIENT_ID) return;
             const { fromVersion, toVersion, deltas, pendingEffects } =
               event.data;
             let skippedNodeIds: string[] = [];
diff --git a/apps/web/src/store/clientId.ts b/apps/web/src/store/clientId.ts
new file mode 100644
index 000000000..1f5fdefa1
--- /dev/null
+++ b/apps/web/src/store/clientId.ts
@@ -0,0 +1,14 @@
+/**
+ * Per-tab client identity (P2 / Plan A).
+ *
+ * A stable, opaque id minted once per page load. It rides the autosave
+ * `PUT /api/canvas/:canvasId` body as `clientId` and is echoed back on the
+ * sync broadcast as `originatorClientId`, so this tab can skip its *own*
+ * PUT echo instead of re-applying a change it already rendered
+ * optimistically. Not persisted — a reload deliberately starts a new
+ * session id.
+ */
+export const CLIENT_ID: string =
+  typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function'
+    ? crypto.randomUUID()
+    : `client-${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`;

From 7ce689e707d3e267dfdb851ac693cd494c560b96 Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Fri, 24 Jul 2026 04:59:59 +0000
Subject: [PATCH 4/9] feat(sync): SSE heartbeat + client auto-reconnect with
 backoff (P2 Plan A)

---
 apps/server/src/modules/canvas/sync.route.ts |  13 ++
 apps/web/src/store/canvasSyncStore.ts        | 185 +++++++++++--------
 2 files changed, 117 insertions(+), 81 deletions(-)

diff --git a/apps/server/src/modules/canvas/sync.route.ts b/apps/server/src/modules/canvas/sync.route.ts
index 9c0ec0a87..83949c789 100644
--- a/apps/server/src/modules/canvas/sync.route.ts
+++ b/apps/server/src/modules/canvas/sync.route.ts
@@ -50,7 +50,20 @@ const syncRoutes: FastifyPluginAsync = async (fastify): Promise<void> => {
         writeSSE(reply.raw, event);
       });
 
+      // Periodic comment-line keep-alive so idle connections aren't reaped
+      // by intermediary proxies / load balancers and the client can detect
+      // a dead stream. `:` lines are ignored by the EventSource parser.
+      const heartbeat = setInterval(() => {
+        try {
+          reply.raw.write(': ping\n\n');
+        } catch {
+          /* stream already closed — the close handler will clean up */
+        }
+      }, 25_000);
+      heartbeat.unref?.();
+
       request.raw.on('close', () => {
+        clearInterval(heartbeat);
         unsubscribe();
         try {
           reply.raw.end();
diff --git a/apps/web/src/store/canvasSyncStore.ts b/apps/web/src/store/canvasSyncStore.ts
index 881fe75a1..fd2646d1a 100644
--- a/apps/web/src/store/canvasSyncStore.ts
+++ b/apps/web/src/store/canvasSyncStore.ts
@@ -126,89 +126,112 @@ export const useCanvasSyncStore = create<CanvasSyncState>((set, get) => ({
     set({ canvasId });
 
     void (async () => {
-      try {
-        const response = await fetch(canvasSyncStreamUrl(canvasId), { signal });
-        if (!response.ok) return;
-        await readTypedSSEStream<CanvasSyncEvent>(
-          response,
-          (event) => {
-            // Ignore late frames after a canvas switch / disconnect.
-            if (get().canvasId !== canvasId) return;
-            const canvasStore = useCanvasStore.getState();
-            if (canvasStore.canvasId !== canvasId) return;
-
-            if (event.type === 'snapshot') {
-              // Skip the catch-up reload while an initial/primary load is
-              // still in flight. On fresh open the CanvasPage mount load
-              // is already fetching the latest state, but it hasn't set
-              // `version` yet when this snapshot arrives — without this
-              // guard the stale-version comparison below fires a second,
-              // redundant `loadCanvas` that races the mount load and
-              // leaks a spurious structure PUT (resetting `updatedAt` to
-              // the open time). The in-flight load already brings the
-              // freshest state, so a snapshot-driven reload is only
-              // meaningful once we've settled.
-              if (canvasStore.isLoading) return;
-              if (event.data.version !== canvasStore.version) {
-                void canvasStore.loadCanvas(canvasId, { resetHistory: true });
-              }
-              return;
-            }
-
-            // event.type === 'update'
-            // Skip our own PUT echo (P2 / Plan A): the originating tab has
-            // already applied this edit optimistically, so re-applying the
-            // broadcast would be redundant (and could fight a still-pending
-            // local edit). Server-originated writes carry no id and pass.
-            if (event.data.originatorClientId === CLIENT_ID) return;
-            const { fromVersion, toVersion, deltas, pendingEffects } =
-              event.data;
-            let skippedNodeIds: string[] = [];
-            if (fromVersion === canvasStore.version) {
-              skippedNodeIds = canvasStore.applyDeltasFromAgent(
-                deltas as Delta[],
-                toVersion,
-                pendingEffects as SyncPendingEffects,
-              );
-            } else if (toVersion > canvasStore.version) {
-              // Gap (missed an earlier update). A blind `loadCanvas` would
-              // clobber un-persisted local edits, so skip it while the user
-              // is mid-editing and let autosave's 409 path arbitrate (C3).
-              // Incremental gap-heal (delta-log backfill) is deferred to P2.
-              if (canvasStore.pendingContentNodeIds().length === 0) {
-                void canvasStore.loadCanvas(canvasId, { resetHistory: true });
-              }
-            }
-            // else: stale/older update — ignore.
-
-            // Draw the user's attention *at the moment* an agent write was
-            // dropped because they were editing that node — a passive card
-            // badge alone is easy to miss mid-edit.
-            notifySkippedAgentWrites(
-              skippedNodeIds,
+      const handleEvent = (event: CanvasSyncEvent): void => {
+        // Ignore late frames after a canvas switch / disconnect.
+        if (get().canvasId !== canvasId) return;
+        const canvasStore = useCanvasStore.getState();
+        if (canvasStore.canvasId !== canvasId) return;
+
+        if (event.type === 'snapshot') {
+          // Skip the catch-up reload while an initial/primary load is
+          // still in flight. On fresh open the CanvasPage mount load
+          // is already fetching the latest state, but it hasn't set
+          // `version` yet when this snapshot arrives — without this
+          // guard the stale-version comparison below fires a second,
+          // redundant `loadCanvas` that races the mount load and
+          // leaks a spurious structure PUT (resetting `updatedAt` to
+          // the open time). The in-flight load already brings the
+          // freshest state, so a snapshot-driven reload is only
+          // meaningful once we've settled.
+          if (canvasStore.isLoading) return;
+          if (event.data.version !== canvasStore.version) {
+            void canvasStore.loadCanvas(canvasId, { resetHistory: true });
+          }
+          return;
+        }
+
+        // event.type === 'update'
+        // Skip our own PUT echo (P2 / Plan A): the originating tab has
+        // already applied this edit optimistically, so re-applying the
+        // broadcast would be redundant (and could fight a still-pending
+        // local edit). Server-originated writes carry no id and pass.
+        if (event.data.originatorClientId === CLIENT_ID) return;
+        const { fromVersion, toVersion, deltas, pendingEffects } = event.data;
+        let skippedNodeIds: string[] = [];
+        if (fromVersion === canvasStore.version) {
+          skippedNodeIds = canvasStore.applyDeltasFromAgent(
+            deltas as Delta[],
+            toVersion,
+            pendingEffects as SyncPendingEffects,
+          );
+        } else if (toVersion > canvasStore.version) {
+          // Gap (missed an earlier update). A blind `loadCanvas` would
+          // clobber un-persisted local edits, so skip it while the user
+          // is mid-editing and let autosave's 409 path arbitrate (C3).
+          // Incremental gap-heal (delta-log backfill) is deferred to P2.
+          if (canvasStore.pendingContentNodeIds().length === 0) {
+            void canvasStore.loadCanvas(canvasId, { resetHistory: true });
+          }
+        }
+        // else: stale/older update — ignore.
+
+        // Draw the user's attention *at the moment* an agent write was
+        // dropped because they were editing that node — a passive card
+        // badge alone is easy to miss mid-edit.
+        notifySkippedAgentWrites(skippedNodeIds, event.data.threadId, canvasId);
+
+        // Attribute change-review records to the originating ACP
+        // conversation's card. `skippedNodeIds` marks the rows whose
+        // agent write was blocked by a local edit, so the card
+        // can flag them as conflicts instead of silently listing them
+        // as applied.
+        if (event.data.threadId && Array.isArray(event.data.changes)) {
+          useAcpThreadChangesStore
+            .getState()
+            .replaceFromBroadcast(
               event.data.threadId,
-              canvasId,
+              event.data.changes as CanvasChangeRecord[],
+              skippedNodeIds,
             );
-
-            // Attribute change-review records to the originating ACP
-            // conversation's card. `skippedNodeIds` marks the rows whose
-            // agent write was blocked by a local edit, so the card
-            // can flag them as conflicts instead of silently listing them
-            // as applied.
-            if (event.data.threadId && Array.isArray(event.data.changes)) {
-              useAcpThreadChangesStore
-                .getState()
-                .replaceFromBroadcast(
-                  event.data.threadId,
-                  event.data.changes as CanvasChangeRecord[],
-                  skippedNodeIds,
-                );
-            }
-          },
-          signal,
-        );
-      } catch {
-        /* aborted or network error — ignore */
+        }
+      };
+
+      // Reconnect loop: SSE streams drop (proxy idle-reap, network blips,
+      // server restart). Each reconnect replays the `snapshot` handshake,
+      // which re-runs the version reconcile above and heals any gap missed
+      // while disconnected. `disconnect()` aborts the signal to break out.
+      let backoffMs = 1_000;
+      const MAX_BACKOFF_MS = 30_000;
+      while (!signal.aborted) {
+        try {
+          const response = await fetch(canvasSyncStreamUrl(canvasId), {
+            signal,
+          });
+          if (!response.ok) throw new Error(`sync stream ${response.status}`);
+          backoffMs = 1_000; // healthy connection — reset backoff
+          await readTypedSSEStream<CanvasSyncEvent>(
+            response,
+            handleEvent,
+            signal,
+          );
+        } catch {
+          /* aborted or network error — fall through to backoff + retry */
+        }
+        if (signal.aborted) break;
+        // Wait before reconnecting; an abort interrupts the wait so a
+        // canvas switch / unmount tears down promptly.
+        await new Promise<void>((resolve) => {
+          const timer = setTimeout(resolve, backoffMs);
+          signal.addEventListener(
+            'abort',
+            () => {
+              clearTimeout(timer);
+              resolve();
+            },
+            { once: true },
+          );
+        });
+        backoffMs = Math.min(backoffMs * 2, MAX_BACKOFF_MS);
       }
     })();
   },

From 06facae6f8ad04371353dd79f5fbe76df37588a7 Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Fri, 24 Jul 2026 05:01:31 +0000
Subject: [PATCH 5/9] test(sync): cover broadcastCanvasStatePut
 diff/publish/echo (P2 Plan A)

---
 .../canvas/broadcast-canvas-put.test.ts       | 164 ++++++++++++++++++
 1 file changed, 164 insertions(+)
 create mode 100644 apps/server/src/modules/canvas/broadcast-canvas-put.test.ts

diff --git a/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts b/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
new file mode 100644
index 000000000..47c0729d6
--- /dev/null
+++ b/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
@@ -0,0 +1,164 @@
+/**
+ * P2 / Plan A — `broadcastCanvasStatePut` tests.
+ *
+ * The autosave PUT diffs pre- vs post-write topology and publishes the
+ * structural deltas on the sync channel so *other* tabs learn about a
+ * user hand-edit. These tests exercise that diff+publish in isolation:
+ * geometry moves broadcast one `REPLACE_NODE`, deletes surface
+ * `deletedNodeIds`, no-op writes publish nothing, and the originating
+ * tab's `clientId` is echoed as `originatorClientId` for self-echo
+ * filtering.
+ */
+
+import { mkdtempSync, rmSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+
+import { afterEach, beforeEach, describe, expect, it } from 'vitest';
+
+import { broadcastCanvasStatePut } from './canvas-executor.js';
+import { subscribeCanvasUpdates } from './canvas-sync.js';
+import { getCanvasStore } from '../storage/index.js';
+import { setWorkspacePath } from '../workspace.js';
+
+import type { CanvasNode, CanvasSyncEvent } from '@sediment/shared';
+
+let tmp: string;
+
+beforeEach(() => {
+  tmp = mkdtempSync(join(tmpdir(), 'sediment-planA-'));
+  setWorkspacePath(tmp);
+});
+
+afterEach(() => {
+  rmSync(tmp, { recursive: true, force: true });
+});
+
+function note(id: string, x: number, y: number): CanvasNode {
+  return {
+    id,
+    type: 'note',
+    position: { x, y },
+    data: { label: id },
+  } as unknown as CanvasNode;
+}
+
+/** Seed a canvas at version 1 with the given nodes. */
+function seed(canvasId: string, nodes: CanvasNode[]): void {
+  getCanvasStore(canvasId).write({
+    canvasId,
+    title: null,
+    version: 1,
+    state: { nodes, edges: [] },
+    createdAt: Date.now(),
+    updatedAt: Date.now(),
+  });
+}
+
+/** Collect every sync event published for `canvasId` during `fn`. */
+function captureEvents(canvasId: string, fn: () => void): CanvasSyncEvent[] {
+  const events: CanvasSyncEvent[] = [];
+  const unsubscribe = subscribeCanvasUpdates(canvasId, (e) => events.push(e));
+  try {
+    fn();
+  } finally {
+    unsubscribe();
+  }
+  return events;
+}
+
+describe('broadcastCanvasStatePut — P2 Plan A', () => {
+  it('broadcasts a REPLACE_NODE for a geometry move, echoing clientId', () => {
+    seed('c1', [note('n1', 0, 0)]);
+
+    const events = captureEvents('c1', () => {
+      broadcastCanvasStatePut({
+        canvasId: 'c1',
+        fromVersion: 1,
+        toVersion: 2,
+        prevNodes: [note('n1', 0, 0)],
+        prevEdges: [],
+        nextNodes: [note('n1', 100, 200)],
+        nextEdges: [],
+        clientId: 'tab-A',
+      });
+    });
+
+    expect(events).toHaveLength(1);
+    const [event] = events;
+    expect(event.type).toBe('update');
+    if (event.type !== 'update') throw new Error('expected update');
+    expect(event.data.fromVersion).toBe(1);
+    expect(event.data.toVersion).toBe(2);
+    expect(event.data.originatorClientId).toBe('tab-A');
+    expect(event.data.deltas).toHaveLength(1);
+    const [delta] = event.data.deltas as Array<{ type: string }>;
+    expect(delta.type).toBe('REPLACE_NODE');
+    // A plain geometry move must not schedule preprocessing / fit.
+    expect(event.data.pendingEffects.mutatedNodes).toHaveLength(0);
+    expect(event.data.pendingEffects.contentEditedNodeIds).toHaveLength(0);
+  });
+
+  it('publishes nothing when topology is unchanged (no-op write)', () => {
+    seed('c1', [note('n1', 0, 0)]);
+
+    const events = captureEvents('c1', () => {
+      const deltas = broadcastCanvasStatePut({
+        canvasId: 'c1',
+        fromVersion: 1,
+        toVersion: 2,
+        prevNodes: [note('n1', 0, 0)],
+        prevEdges: [],
+        nextNodes: [note('n1', 0, 0)],
+        nextEdges: [],
+        clientId: 'tab-A',
+      });
+      expect(deltas).toHaveLength(0);
+    });
+
+    expect(events).toHaveLength(0);
+  });
+
+  it('surfaces a removed node via deletedNodeIds', () => {
+    seed('c1', [note('n1', 0, 0), note('n2', 10, 10)]);
+
+    const events = captureEvents('c1', () => {
+      broadcastCanvasStatePut({
+        canvasId: 'c1',
+        fromVersion: 1,
+        toVersion: 2,
+        prevNodes: [note('n1', 0, 0), note('n2', 10, 10)],
+        prevEdges: [],
+        nextNodes: [note('n1', 0, 0)],
+        nextEdges: [],
+        clientId: 'tab-A',
+      });
+    });
+
+    expect(events).toHaveLength(1);
+    const [event] = events;
+    if (event.type !== 'update') throw new Error('expected update');
+    expect(event.data.pendingEffects.deletedNodeIds).toEqual(['n2']);
+  });
+
+  it('omits originatorClientId when no clientId is supplied', () => {
+    seed('c1', [note('n1', 0, 0)]);
+
+    const events = captureEvents('c1', () => {
+      broadcastCanvasStatePut({
+        canvasId: 'c1',
+        fromVersion: 1,
+        toVersion: 2,
+        prevNodes: [note('n1', 0, 0)],
+        prevEdges: [],
+        nextNodes: [note('n1', 5, 5)],
+        nextEdges: [],
+      });
+    });
+
+    expect(events).toHaveLength(1);
+    const [event] = events;
+    if (event.type !== 'update') throw new Error('expected update');
+    expect(event.data.originatorClientId).toBeUndefined();
+  });
+});

From b07afaac8c6bb67fa606f97a3ebdd1bcc5adb5f4 Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Fri, 24 Jul 2026 05:04:16 +0000
Subject: [PATCH 6/9] style(sync): fix import order in canvas.route

---
 apps/server/src/modules/canvas/canvas.route.ts | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/apps/server/src/modules/canvas/canvas.route.ts b/apps/server/src/modules/canvas/canvas.route.ts
index bfcc8b9c9..acadf0b77 100644
--- a/apps/server/src/modules/canvas/canvas.route.ts
+++ b/apps/server/src/modules/canvas/canvas.route.ts
@@ -24,10 +24,6 @@ import {
   setPortalNodePinsCommandSchema,
 } from '@huabu/shared';
 import { nodeRevisionOf } from '@huabu/shared/canvas-engine';
-import type {
-  CanvasEdge,
-  CanvasNode,
-} from '@huabu/shared/canvas-engine';
 
 import {
   CanvasCommandRoutingError,
@@ -111,6 +107,10 @@ import type {
   PutNodeContentResponse,
   RevealNodesFolderResponse,
 } from '@huabu/shared';
+import type {
+  CanvasEdge,
+  CanvasNode,
+} from '@huabu/shared/canvas-engine';
 import type { FastifyPluginAsync } from 'fastify';
 
 /**

From 5f981562e3eb08e026b626ee639bba79d38b5ce8 Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Fri, 24 Jul 2026 05:05:13 +0000
Subject: [PATCH 7/9] fix(sync): import CanvasNode from canvas-engine
 entrypoint in test

---
 apps/server/src/modules/canvas/broadcast-canvas-put.test.ts | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts b/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
index 47c0729d6..2c1117089 100644
--- a/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
+++ b/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
@@ -21,7 +21,8 @@ import { subscribeCanvasUpdates } from './canvas-sync.js';
 import { getCanvasStore } from '../storage/index.js';
 import { setWorkspacePath } from '../workspace.js';
 
-import type { CanvasNode, CanvasSyncEvent } from '@sediment/shared';
+import type { CanvasSyncEvent } from '@sediment/shared';
+import type { CanvasNode } from '@sediment/shared/canvas-engine';
 
 let tmp: string;
 

From 0e953ca73203d1e305721720283f0be845a2c7d9 Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Wed, 12 Aug 2026 08:52:25 +0000
Subject: [PATCH 8/9] fix(sync): use Huabu package imports in broadcast test

---
 apps/server/src/modules/canvas/broadcast-canvas-put.test.ts | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts b/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
index 2c1117089..c2a1515eb 100644
--- a/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
+++ b/apps/server/src/modules/canvas/broadcast-canvas-put.test.ts
@@ -21,8 +21,8 @@ import { subscribeCanvasUpdates } from './canvas-sync.js';
 import { getCanvasStore } from '../storage/index.js';
 import { setWorkspacePath } from '../workspace.js';
 
-import type { CanvasSyncEvent } from '@sediment/shared';
-import type { CanvasNode } from '@sediment/shared/canvas-engine';
+import type { CanvasSyncEvent } from '@huabu/shared';
+import type { CanvasNode } from '@huabu/shared/canvas-engine';
 
 let tmp: string;
 

From 81e11ba7a9360be3b428cbda9001ffb3ebd54f41 Mon Sep 17 00:00:00 2001
From: cxxxxxn <christy05.chen@gmail.com>
Date: Wed, 12 Aug 2026 09:05:53 +0000
Subject: [PATCH 9/9] style(sync): apply Huabu formatting

---
 apps/server/src/modules/canvas/canvas-executor.ts | 5 +++--
 apps/server/src/modules/canvas/canvas.route.ts    | 5 +----
 2 files changed, 4 insertions(+), 6 deletions(-)

diff --git a/apps/server/src/modules/canvas/canvas-executor.ts b/apps/server/src/modules/canvas/canvas-executor.ts
index 98d73d0ef..1fc4737d0 100644
--- a/apps/server/src/modules/canvas/canvas-executor.ts
+++ b/apps/server/src/modules/canvas/canvas-executor.ts
@@ -1328,8 +1328,9 @@ export function broadcastCanvasStatePut(input: {
   if (deltas.length === 0) return deltas;
 
   const deletedNodeIds = deltas
-    .filter((d): d is Extract<Delta, { type: 'DELETE_NODE' }> =>
-      d.type === 'DELETE_NODE',
+    .filter(
+      (d): d is Extract<Delta, { type: 'DELETE_NODE' }> =>
+        d.type === 'DELETE_NODE',
     )
     .map((d) => d.node.id);
 
diff --git a/apps/server/src/modules/canvas/canvas.route.ts b/apps/server/src/modules/canvas/canvas.route.ts
index acadf0b77..705599b17 100644
--- a/apps/server/src/modules/canvas/canvas.route.ts
+++ b/apps/server/src/modules/canvas/canvas.route.ts
@@ -107,10 +107,7 @@ import type {
   PutNodeContentResponse,
   RevealNodesFolderResponse,
 } from '@huabu/shared';
-import type {
-  CanvasEdge,
-  CanvasNode,
-} from '@huabu/shared/canvas-engine';
+import type { CanvasEdge, CanvasNode } from '@huabu/shared/canvas-engine';
 import type { FastifyPluginAsync } from 'fastify';
 
 /**