fix(git-sync): converge git-ingest with open editor sessions — stop silent revert/data-loss on live pages

A git push to a page with an OPEN editor was silently reverted: the git
commit landed and the DB body updated, but the page in the browser stayed
on the old content and the editor's next autosave overwrote the git change.

Root cause (distributed, not in the merge): writeBody applied the body
merge via collabGateway.openDirectConnection on whichever instance/process
runs git-sync (the api/worker). When an editor is connected to a DIFFERENT
collab instance/process, that opens a SEPARATE, detached Y.Doc. The merge
landed in the detached doc + DB, but the live editor's Y.Doc never received
the Yjs update; its debounced autosave then persisted its STALE state over
the DB, reverting the git change (and, for concurrent edits to different
paragraphs, losing the git side). In one process the bug is invisible
because the direct connection already shares the editor's doc.

Fix: route the body write through the existing custom-event channel (the
same mechanism comment-marks and updatePageContent use) so the merge runs
on the instance that OWNS the live doc. Its update is then broadcast to
every connection (Document.handleUpdate) and the editor's CRDT converges on
the merged result. New CollaborationGateway.writePageBody dispatches to a
new gitSyncWriteBody handler (builds incoming/base docs before opening the
connection — crash-safe — then 3-way/2-way merges into the live fragment);
without redis it runs locally on the single (owning) instance. writeBody
now just forwards the converted ProseMirror bodies + service userId.

Evidence:
- git-ingest-convergence.spec.ts: deterministic two-Y.Doc repro. PATH B
  (undelivered update) asserts the LOSS (the bug); PATH A (update delivered,
  as the owner-routed write does) asserts the git change SURVIVES and that
  concurrent edits to different paragraphs both survive.
- collaboration.handler.git-sync.spec.ts: exercises the real gitSyncWriteBody
  against a shared doc wired to a connected "editor" doc (models the
  owning-instance broadcast) — editor converges, concurrent edit preserved,
  crash-safe on transform failure.
- gitmost-datasource.service.spec.ts: writeBody now routes via writePageBody
  (RED before this change — it called openDirectConnection).

Honest scope: the failure is cross-instance; full multi-instance convergence
needs a live Hocuspocus + redis and is not provable in a unit test, so the
convergence invariant is captured at the Yjs update-exchange level.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

apps/server/src/integrations/git-sync/services/git-ingest-convergence.spec.ts

@@ -0,0 +1,180 @@
+import * as Y from 'yjs';
+
+import { mergeXmlFragments3Way } from './yjs-body-merge';
+
+/**
+ * Convergence repro for the git-ingest "silent revert" data-loss bug.
+ *
+ * ROOT CAUSE (confirmed): the merge logic itself is correct, but the git-ingest
+ * write was applied via `openDirectConnection` on whichever instance/process
+ * runs git-sync (the api/worker). When an editor is connected to a DIFFERENT
+ * collab instance/process, that opens a SEPARATE, detached Y.Doc. The merge
+ * lands in that detached doc (and the DB), but the live editor's Y.Doc never
+ * receives the Yjs update — so its next debounced autosave overwrites the DB
+ * with its STALE state and silently reverts the git change.
+ *
+ * These tests reproduce the invariant deterministically at the Yjs level (two
+ * Y.Docs exchanging updates), because the real failure is DISTRIBUTED — it only
+ * manifests when the write and the editor live on different instances, which a
+ * single in-process Hocuspocus cannot reproduce (in one process the direct
+ * connection already shares the editor's doc). HONEST SCOPE: this models the two
+ * outcomes; full cross-instance convergence is not (and cannot be) proven in a
+ * unit test without a live multi-instance Hocuspocus + redis.
+ *
+ *   PATH B (the BUG): the git update is NOT delivered to the editor's doc — the
+ *     editor's later autosave reverts the change. Asserts the LOSS.
+ *   PATH A (the FIX): the git update IS delivered to the editor's doc as a Yjs
+ *     update — which is exactly what running the merge on the OWNING instance's
+ *     shared Document does (its update is broadcast to every connection). The
+ *     editor's CRDT converges and a later autosave preserves the git change.
+ *
+ * The fix routes git-sync's body write through CollaborationGateway.writePageBody
+ * (the custom-event channel) so it executes on the owning instance — turning
+ * PATH B into PATH A.
+ */
+
+type Spec = { text: string; id?: string };
+
+// Build a Y.XmlFragment('default'). `id` is set only when provided, mirroring
+// the live doc (block UniqueIDs present) vs a git-parsed body (ids absent).
+function buildFragment(doc: Y.Doc, specs: Spec[]): Y.XmlFragment {
+  const frag = doc.getXmlFragment('default');
+  const blocks = specs.map((s) => {
+    const el = new Y.XmlElement('paragraph');
+    if (s.id) el.setAttribute('id', s.id);
+    const t = new Y.XmlText();
+    if (s.text) t.insert(0, s.text);
+    el.insert(0, [t]);
+    return el;
+  });
+  if (blocks.length) frag.insert(0, blocks);
+  return frag;
+}
+
+const texts = (frag: Y.XmlFragment): string[] =>
+  frag.toArray().map((el) =>
+    (el as Y.XmlElement)
+      .toArray()
+      .map((c) => (c as Y.XmlText).toString())
+      .join(''),
+  );
+
+// Append '!' to the end of the given block's text — a tiny human edit that
+// stands in for a connected editor's autosave-triggering keystroke.
+function humanEdit(doc: Y.Doc, blockIndex: number, mark = '!'): void {
+  const frag = doc.getXmlFragment('default');
+  const el = frag.get(blockIndex) as Y.XmlElement;
+  const t = el.get(0) as Y.XmlText;
+  doc.transact(() => t.insert(t.length, mark));
+}
+
+describe('git-ingest convergence with an open editor', () => {
+  // Shared setup: the page is persisted with two blocks (live ids), and BOTH the
+  // server-side ingest doc (S) and the connected editor's doc (C) load that same
+  // state — they start fully synced, exactly like two instances that each loaded
+  // the page from the DB.
+  function setup() {
+    const db = new Y.Doc();
+    buildFragment(db, [
+      { text: 'alpha', id: 'p1' },
+      { text: 'beta', id: 'p2' },
+    ]);
+    const state0 = Y.encodeStateAsUpdate(db);
+
+    const server = new Y.Doc(); // where the git merge is applied
+    Y.applyUpdate(server, state0);
+    const editor = new Y.Doc(); // the browser's live in-memory doc
+    Y.applyUpdate(editor, state0);
+
+    // base (last-synced, from git markdown — no ids) == the pre-change content.
+    const baseDoc = new Y.Doc();
+    const baseFrag = buildFragment(baseDoc, [{ text: 'alpha' }, { text: 'beta' }]);
+    return { state0, server, editor, baseFrag };
+  }
+
+  // git changed the SECOND block alpha/beta -> beta2; the editor is idle on it.
+  function applyGitMerge(server: Y.Doc, baseFrag: Y.XmlFragment): Uint8Array {
+    const targetDoc = new Y.Doc();
+    const targetFrag = buildFragment(targetDoc, [
+      { text: 'alpha' },
+      { text: 'beta2' },
+    ]);
+    let captured: Uint8Array | null = null;
+    const onUpdate = (u: Uint8Array) => {
+      // Accumulate (the merge emits one update per op when unwrapped); here a
+      // single transact yields one update covering the whole merge.
+      captured = captured ? Y.mergeUpdates([captured, u]) : u;
+    };
+    server.on('update', onUpdate);
+    server.transact(() =>
+      mergeXmlFragments3Way(
+        server.getXmlFragment('default'),
+        targetFrag,
+        baseFrag,
+      ),
+    );
+    server.off('update', onUpdate);
+    return captured!;
+  }
+
+  it('PATH B (the BUG): undelivered git update is reverted by the editor autosave — DATA LOSS', () => {
+    const { server, editor, baseFrag } = setup();
+
+    // git merge lands on the server doc only.
+    applyGitMerge(server, baseFrag);
+    expect(texts(server.getXmlFragment('default'))).toEqual(['alpha', 'beta2']);
+
+    // The editor NEVER receives the update (detached doc on another instance).
+    // It makes an unrelated edit on block 0 and autosaves its full state.
+    humanEdit(editor, 0);
+    const persisted = new Y.Doc();
+    Y.applyUpdate(persisted, Y.encodeStateAsUpdate(editor));
+
+    // git's 'beta2' is gone — the page reverted to 'beta'. This is the bug.
+    expect(texts(persisted.getXmlFragment('default'))).toEqual([
+      'alpha!',
+      'beta',
+    ]);
+  });
+
+  it('PATH A (the FIX): delivering the git update to the editor converges — git change SURVIVES', () => {
+    const { server, editor, baseFrag } = setup();
+
+    // git merge on the server doc, capturing the broadcastable Yjs update.
+    const gitUpdate = applyGitMerge(server, baseFrag);
+
+    // Running on the OWNING instance broadcasts the update to the connected
+    // editor (Document.handleUpdate). Model that: the editor applies it.
+    Y.applyUpdate(editor, gitUpdate);
+    expect(texts(editor.getXmlFragment('default'))).toEqual(['alpha', 'beta2']);
+
+    // The editor now autosaves (unrelated edit on block 0). Its full state still
+    // carries git's change — no revert.
+    humanEdit(editor, 0);
+    const persisted = new Y.Doc();
+    Y.applyUpdate(persisted, Y.encodeStateAsUpdate(editor));
+    expect(texts(persisted.getXmlFragment('default'))).toEqual([
+      'alpha!',
+      'beta2',
+    ]);
+  });
+
+  it('PATH A — concurrent edits to DIFFERENT paragraphs both survive (finding #2)', () => {
+    const { server, editor, baseFrag } = setup();
+
+    // The editor is actively editing block 0 (concurrent with the push).
+    humanEdit(editor, 0, ' EDIT');
+
+    // git changes block 1; merge on the server, broadcast to the editor.
+    const gitUpdate = applyGitMerge(server, baseFrag);
+    Y.applyUpdate(editor, gitUpdate);
+
+    // Both sides preserved: the human's block-0 edit AND git's block-1 change.
+    const persisted = new Y.Doc();
+    Y.applyUpdate(persisted, Y.encodeStateAsUpdate(editor));
+    expect(texts(persisted.getXmlFragment('default'))).toEqual([
+      'alpha EDIT',
+      'beta2',
+    ]);
+  });
+});

apps/server/src/integrations/git-sync/services/gitmost-datasource.service.spec.ts

@@ -46,13 +46,7 @@ jest.mock('../git-sync.loader', () => ({
   })),
 }));
 
-import * as Y from 'yjs';
 import { GitmostDataSourceService } from './gitmost-datasource.service';
-// The body-write seam picks 2-way vs 3-way merge based on whether a base doc was
-// built. We spy on the real module exports (ts-jest CJS output references them
-// through the namespace object, so the spies intercept the SUT's calls) and let
-// them call through, so we assert WHICH merge ran without mocking the behaviour.
-import * as bodyMerge from './yjs-body-merge';
 
 // Focused unit/contract test for the native GitSyncClient adapter.
 // No DB, no real collab server: the repos/services/gateway are mocked and we
@@ -73,16 +67,9 @@ interface Mocks {
     movePage: AnyMock;
     removePage: AnyMock;
   };
-  collabGateway: { openDirectConnection: AnyMock };
+  collabGateway: { writePageBody: AnyMock };
   // Minimal Kysely-ish chainable mock for the direct-query paths.
   db: any;
-  // Captured collab connection (the fake conn the gateway returns).
-  conn: {
-    transact: AnyMock;
-    disconnect: AnyMock;
-    context?: any;
-    capturedFn?: (doc: any) => void;
-  };
 }
 
 function makeQueryBuilder(rows: any[]) {
@@ -99,13 +86,6 @@ function build(rows: any[] = []): {
   service: GitmostDataSourceService;
   mocks: Mocks;
 } {
-  const conn: Mocks['conn'] = {
-    transact: jest.fn(async (fn: (doc: any) => void) => {
-      conn.capturedFn = fn;
-    }),
-    disconnect: jest.fn(async () => undefined),
-  };
-
   const mocks: Mocks = {
     pageRepo: {
       findById: jest.fn(),
@@ -120,15 +100,11 @@ function build(rows: any[] = []): {
       removePage: jest.fn(async () => undefined),
     },
     collabGateway: {
-      openDirectConnection: jest.fn(async (_name: string, ctx: any) => {
-        conn.context = ctx;
-        return conn;
-      }),
+      writePageBody: jest.fn(async () => undefined),
     },
     db: {
       selectFrom: jest.fn(() => makeQueryBuilder(rows)),
     },
-    conn,
   };
 
   const service = new GitmostDataSourceService(
@@ -232,7 +208,7 @@ describe('GitmostDataSourceService', () => {
   });
 
   describe('importPageMarkdown', () => {
-    it('parses md, converts to ProseMirror, and writes body via collab', async () => {
+    it('parses md, converts to ProseMirror, and routes the body write to the owning instance', async () => {
       const { service, mocks } = build();
       mocks.pageRepo.findById.mockResolvedValue({
         id: 'p1',
@@ -243,77 +219,50 @@ describe('GitmostDataSourceService', () => {
         .bind(CTX)
         .importPageMarkdown('p1', '# Hello\n\nworld');
 
-      // writeBody opened a collab connection tagged git-sync + service user.
-      expect(mocks.collabGateway.openDirectConnection).toHaveBeenCalledTimes(1);
-      const [docName, ctx] = mocks.collabGateway.openDirectConnection.mock
-        .calls[0];
+      // writeBody routes through writePageBody (NOT openDirectConnection): the
+      // merge must run on the instance that owns the live doc so a connected
+      // editor converges instead of silently reverting the change. The service
+      // user rides on the payload as the responsible author.
+      expect(mocks.collabGateway.writePageBody).toHaveBeenCalledTimes(1);
+      const [docName, payload] = mocks.collabGateway.writePageBody.mock.calls[0];
       expect(docName).toBe('page.p1');
-      expect(ctx.actor).toBe('git-sync');
-      expect(ctx.user).toEqual({ id: 'svc-user' });
-
-      // transact ran and connection was disconnected (finally).
-      expect(mocks.conn.transact).toHaveBeenCalledTimes(1);
-      expect(mocks.conn.disconnect).toHaveBeenCalledTimes(1);
-      expect(typeof mocks.conn.capturedFn).toBe('function');
+      expect(payload.userId).toBe('svc-user');
+      // A converted ProseMirror doc was passed; no base on a plain import.
+      expect(payload.prosemirrorJson).toEqual(
+        expect.objectContaining({ type: 'doc' }),
+      );
+      expect(payload.baseProsemirrorJson).toBeUndefined();
 
       expect(res.updatedAt).toBe('2026-06-20T11:00:00.000Z');
     });
 
-    it('crash-safe: the incoming doc is built before the connection opens, and the captured merge applies content', async () => {
-      // The incoming Yjs doc is built BEFORE the connection opens, so a transform
-      // failure can never mutate the live body. Here we run the captured transact
-      // callback (the block-level merge) against a REAL empty Y.Doc and confirm it
-      // ends up with content — the write produces a non-empty body, never wipes it.
-      const { service, mocks } = build();
-      mocks.pageRepo.findById.mockResolvedValue({
-        id: 'p1',
-        updatedAt: new Date('2026-06-20T11:00:00.000Z'),
-      });
-      await service.bind(CTX).importPageMarkdown('p1', '# Hi\n\nthere');
-
-      const realDoc = new Y.Doc();
-      expect(() => mocks.conn.capturedFn?.(realDoc)).not.toThrow();
-      // The body fragment is non-empty: the incoming block was merged in.
-      expect(realDoc.getXmlFragment('default').length).toBeGreaterThan(0);
-    });
-
     // The 2-way path (no base) is covered above; this exercises the THREE-WAY
-    // branch that only fires when a `baseMarkdown` is supplied (review #5).
+    // branch that only fires when a `baseMarkdown` is supplied (review #5). The
+    // merge dispatch itself now lives in the collab handler (gitSyncWriteBody);
+    // here we assert the datasource forwards the base so the owning instance can
+    // run the 3-way reconcile.
     describe('with a baseMarkdown (three-way merge)', () => {
-      afterEach(() => jest.restoreAllMocks());
-
-      it('builds a base doc and dispatches to mergeXmlFragments3Way (not the 2-way merge)', async () => {
+      it('forwards the parsed base body so the owning instance can three-way merge', async () => {
         const { service, mocks } = build();
         mocks.pageRepo.findById.mockResolvedValue({
           id: 'p1',
           updatedAt: new Date('2026-06-20T11:00:00.000Z'),
         });
-        // Spy through to the real implementations so we observe the dispatch.
-        const merge3 = jest.spyOn(bodyMerge, 'mergeXmlFragments3Way');
-        const merge2 = jest.spyOn(bodyMerge, 'mergeXmlFragments');
 
         await service
           .bind(CTX)
           .importPageMarkdown('p1', '# Full\n\ngit', '# Base\n\nbase');
 
-        // The body write was staged through collab as before.
-        expect(mocks.conn.transact).toHaveBeenCalledTimes(1);
-        expect(typeof mocks.conn.capturedFn).toBe('function');
-
-        // Running the captured merge against a real live doc takes the 3-way path:
-        // the base was parsed/built and the 3-way helper is invoked with three
-        // fragments; the 2-way fallback is NOT used.
-        const liveDoc = new Y.Doc();
-        expect(() => mocks.conn.capturedFn?.(liveDoc)).not.toThrow();
-
-        expect(merge3).toHaveBeenCalledTimes(1);
-        expect(merge2).not.toHaveBeenCalled();
-        const [liveFrag, gitFrag, baseFrag] = merge3.mock.calls[0];
-        expect(liveFrag).toBeInstanceOf(Y.XmlFragment);
-        expect(gitFrag).toBeInstanceOf(Y.XmlFragment);
-        // The third arg is the BASE fragment — proof the base markdown was parsed
-        // and converted into its own doc for the common-ancestor comparison.
-        expect(baseFrag).toBeInstanceOf(Y.XmlFragment);
+        expect(mocks.collabGateway.writePageBody).toHaveBeenCalledTimes(1);
+        const [, payload] = mocks.collabGateway.writePageBody.mock.calls[0];
+        // Both the incoming body AND the last-synced base were converted and
+        // forwarded — proof the 3-way common-ancestor is plumbed through.
+        expect(payload.prosemirrorJson).toEqual(
+          expect.objectContaining({ type: 'doc' }),
+        );
+        expect(payload.baseProsemirrorJson).toEqual(
+          expect.objectContaining({ type: 'doc' }),
+        );
       });
     });
   });
@@ -337,9 +286,9 @@ describe('GitmostDataSourceService', () => {
         { spaceId: 'space-1', title: 'Title', parentPageId: 'parent-1' },
         { actor: 'git-sync', aiChatId: null },
       );
-      expect(mocks.collabGateway.openDirectConnection).toHaveBeenCalledWith(
+      expect(mocks.collabGateway.writePageBody).toHaveBeenCalledWith(
         'page.new-id',
-        expect.objectContaining({ actor: 'git-sync' }),
+        expect.objectContaining({ userId: 'svc-user' }),
       );
       expect(res).toEqual({
         data: { id: 'new-id' },

apps/server/src/integrations/git-sync/services/gitmost-datasource.service.ts

@@ -1,5 +1,4 @@
 import { Injectable, Logger, NotFoundException } from '@nestjs/common';
-import { TiptapTransformer } from '@hocuspocus/transformer';
 import { generateJitteredKeyBetween } from 'fractional-indexing-jittered';
 import type {
   GitSyncClient,
@@ -12,8 +11,6 @@ import { InjectKysely } from 'nestjs-kysely';
 import { KyselyDB } from '@docmost/db/types/kysely.types';
 import { PageService } from '../../../core/page/services/page.service';
 import { CollaborationGateway } from '../../../collaboration/collaboration.gateway';
-import { tiptapExtensions } from '../../../collaboration/collaboration.util';
-import { mergeXmlFragments, mergeXmlFragments3Way } from './yjs-body-merge';
 import { AuthProvenanceData } from '../../../common/decorators/auth-provenance.decorator';
 
 /**
@@ -387,15 +384,27 @@ export class GitmostDataSourceService {
   // --- linchpin: native body write (§3.3) -----------------------------------
 
   /**
-   * In-process body write — no loopback websocket, no service-user token. Mirrors
-   * the collab handler's 'replace' operation exactly: open a direct connection,
-   * drop the existing fragment, apply the converted doc, then disconnect.
+   * In-process body write — no loopback websocket, no service-user token.
    *
-   * The `{ actor: 'git-sync', user: { id: userId } }` context flows into
+   * Routes the write through `CollaborationGateway.writePageBody`, which applies
+   * the block-level MERGE on the instance that OWNS the live Y.Doc (via the
+   * custom-event channel) rather than opening a direct connection on this
+   * (api/worker) instance. That distinction is load-bearing: when an editor is
+   * connected to a different collab instance/process, a direct connection here
+   * mutates a SEPARATE, detached doc the editor never sees — the editor's next
+   * autosave then silently REVERTS the git change (data loss). Running on the
+   * owning instance broadcasts the merge as a Yjs update so the editor converges
+   * (see CollaborationGateway.writePageBody for the full rationale).
+   *
+   * The merge itself stays a block-level reconcile, not a full-body replace
+   * (review #5): only changed blocks are touched, concurrently-edited blocks are
+   * left untouched, and an unchanged resync is a 0-op write. With a `base` (the
+   * last-synced version) it is a THREE-WAY merge so a block ONLY the human
+   * changed is kept and a block ONLY git changed is taken (conflicts -> git);
+   * without a base (e.g. createPage) it falls back to the 2-way merge. The
+   * `{ actor: 'git-sync', user: { id: userId } }` context flows into
    * PersistenceExtension.onStoreDocument, which persists ydoc+content+textContent,
-   * stamps `lastUpdatedSource = 'git-sync'`, and broadcasts `page.updated`. The
-   * service user (`user.id`) stays the responsible `lastUpdatedById`; the actor
-   * marks provenance.
+   * stamps `lastUpdatedSource = 'git-sync'`, and broadcasts `page.updated`.
    */
   private async writeBody(
     pageId: string,
@@ -404,51 +413,10 @@ export class GitmostDataSourceService {
     baseProsemirrorJson?: unknown,
   ): Promise<void> {
     const documentName = `page.${pageId}`;
-
-    // Build the incoming (and base) Yjs docs BEFORE opening the connection /
-    // touching the live doc. If a transform throws (a malformed/unsupported doc)
-    // we must NOT have mutated the live body — otherwise a conversion failure
-    // could leave the page empty (review #5 — crash-safe conversion).
-    const targetDoc = TiptapTransformer.toYdoc(
+    await this.collabGateway.writePageBody(documentName, {
       prosemirrorJson,
-      'default',
-      tiptapExtensions,
-    );
-    const baseDoc =
-      baseProsemirrorJson != null
-        ? TiptapTransformer.toYdoc(baseProsemirrorJson, 'default', tiptapExtensions)
-        : null;
-
-    const conn = await this.collabGateway.openDirectConnection(documentName, {
-      actor: 'git-sync',
-      // PersistenceExtension reads `context.user.id` for lastUpdatedById, so the
-      // service user is required on the context (unlike the bare `{ actor }`
-      // sketch in issue #194).
-      user: { id: userId },
+      baseProsemirrorJson,
+      userId,
     });
-    try {
-      await conn.transact((doc) => {
-        const liveFrag = doc.getXmlFragment('default');
-        const targetFrag = targetDoc.getXmlFragment('default');
-        // Block-level MERGE rather than a full-body replace (review #5): diff the
-        // live body against the incoming git body and apply only the blocks that
-        // actually changed; concurrently-edited blocks are left untouched and an
-        // unchanged resync is a 0-op write. With a `base` (the last-synced
-        // version) do a THREE-WAY merge so a block ONLY the human changed is kept
-        // and a block ONLY git changed is taken (conflicts -> git). Without a base
-        // (e.g. createPage), fall back to the 2-way merge.
-        if (baseDoc) {
-          mergeXmlFragments3Way(
-            liveFrag,
-            targetFrag,
-            baseDoc.getXmlFragment('default'),
-          );
-        } else {
-          mergeXmlFragments(liveFrag, targetFrag);
-        }
-      });
-    } finally {
-      await conn.disconnect();
-    }
   }
 }