Add reliable cross-tab sync for canvas PUT updates

apps/server/src/modules/canvas/broadcast-canvas-put.test.ts

@@ -0,0 +1,165 @@
+/**
+ * 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 { CanvasSyncEvent } from '@huabu/shared';
+import type { CanvasNode } from '@huabu/shared/canvas-engine';
+
+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();
+  });
+});

apps/server/src/modules/canvas/canvas-executor.ts

@@ -1269,3 +1269,95 @@ 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;
+}

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';