From 8436cc421bc5c9b876937f156b9b23cd74ad79fc Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Fri, 16 Oct 2026 20:59:34 -0700 Subject: [PATCH 1/4] fix(T13842): docs fetch serves the merged head, not the last writer's raw input updateDocBySlug wrote the docs_attachments row (the bytes docs fetch and docs view read) from the incoming text, then projected the edit through the section CRDT. After a stale-base merge the conduit head held both writers' edits but the row held only the last writer's input, so readers silently lost the concurrent edit even where the merge was correct (probe on main: an append merge reported staleBaseMerged:true, the head had both lines, the fetched body had only B's). The projection now runs before the row write, and the row is written from the resulting head: mergedText when this write advanced the head, the head version's text when it lost the head race (an explicit lower HLC), and the incoming bytes only when the projection fails. Conflicts are unchanged: the head excludes a conflicting incoming edit, which stays in its own version. Tests (docs-update-concurrency.test.ts, both fail on main): a stale-base append merge's fetched body carries both writers' lines and equals the head; a write that loses the head race leaves the fetched body equal to the head. Co-Authored-By: Claude Opus 5.5 --- .changeset/docs-fetch-merged-head.md | 15 ++ .../__tests__/docs-update-concurrency.test.ts | 101 +++++++- packages/core/src/docs/docs-update.ts | 221 ++++++++++-------- 3 files changed, 240 insertions(+), 97 deletions(-) create mode 100644 .changeset/docs-fetch-merged-head.md diff --git a/.changeset/docs-fetch-merged-head.md b/.changeset/docs-fetch-merged-head.md new file mode 100644 index 0000000000..86da1b49f1 --- /dev/null +++ b/.changeset/docs-fetch-merged-head.md @@ -0,0 +1,15 @@ +--- +id: docs-fetch-merged-head +tasks: [T13842] +kind: fix +summary: docs fetch now returns the merged head after concurrent edits, not the last writer's raw input +--- + +`docs update` wrote the doc row (what `docs fetch` and `docs view` read) +from the incoming bytes, then projected the edit through the section CRDT. +After a stale-base merge the conduit head held both writers' edits, but the +row held only the last writer's input, so readers lost the other edit even +when the merge itself was correct. The projection now runs first and the row +is written from the resulting head text: the merged text when the write +advanced the head, the current head version's text when it lost the head +race. The incoming bytes are written only when the projection fails. diff --git a/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts b/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts index 80fe8612cd..39c9d17788 100644 --- a/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts +++ b/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts @@ -61,7 +61,28 @@ import { readFileSync, writeFileSync } from 'node:fs'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +/** + * T13842: when set, the real projection runs with this explicit HLC, so a + * write can lose the head race (a minted HLC always advances the head). + */ +let forcedHlc: string | undefined; +vi.mock('../section-write-path.js', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + projectDocUpdateThroughSections: ( + root: string, + input: Parameters[1], + ) => + actual.projectDocUpdateThroughSections( + root, + forcedHlc !== undefined ? { ...input, hlc: forcedHlc } : input, + ), + }; +}); + import { getConduitAttachmentVersion, getConduitDocHead, @@ -679,3 +700,81 @@ async function seedLegacyDoc(root: string, slug: string): Promise { }) .run(); } + +describe("docs fetch serves the merged head, not the last writer's raw input (T13842)", () => { + const SEED = '# Plan\n\npre.\n\n## Alpha\n\nalpha body\n\n## Beta\n\nbeta body'; + + async function withStore(fn: (root: string) => Promise): Promise { + const root = await mkdtemp(join(tmpdir(), 'cleo-docs-fetch-head-')); + process.env['CLEO_DIR'] = join(root, '.cleo'); + closeDb(); + try { + await seedLegacyDoc(root, SLUG); + expect((await updateDocBySlug(root, { slug: SLUG, content: SEED })).ok).toBe(true); + await fn(root); + } finally { + forcedHlc = undefined; + closeDb(); + delete process.env['CLEO_DIR']; + await rm(root, { recursive: true, force: true }); + } + } + + async function fetchedBody(root: string): Promise { + const model = new DocsReadModel({ projectRoot: root }); + const doc = await model.resolveBySlug(SLUG); + return doc === null ? null : model.fetchContent(doc); + } + + async function headText(root: string): Promise { + const docUid = (await resolveConduitDocUid(root, SLUG)) as string; + const head = await getConduitDocHead(root, docUid); + const version = await getConduitAttachmentVersion(root, head?.headVersionUid ?? ''); + return readFileSync( + join(root, '.cleo', 'attachments', 'sha256', version?.storageKey ?? ''), + 'utf-8', + ); + } + + it("a stale-base append merge: the fetched body carries both writers' lines", async () => { + await withStore(async (root) => { + const docUid = (await resolveConduitDocUid(root, SLUG)) as string; + const base = (await getConduitDocHead(root, docUid))?.headVersionUid; + const a = await updateDocBySlug(root, { + slug: SLUG, + content: SEED.replace('alpha body', 'alpha body\n\nA line'), + }); + expect(a.ok).toBe(true); + const b = await updateDocBySlug(root, { + slug: SLUG, + content: SEED.replace('beta body', 'beta body\n\nB line'), + baseVersionUid: base, + }); + expect(b.ok && b.result.staleBaseMerged).toBe(true); + + const body = await fetchedBody(root); + expect(body).toContain('A line'); + expect(body).toContain('B line'); + expect(body).toBe(await headText(root)); + }); + }); + + it('a write that loses the head race: the fetched body is the head, not the losing input', async () => { + await withStore(async (root) => { + const winner = SEED.replace('alpha body', 'alpha body\n\nwinner line'); + forcedHlc = '9999999999999-000000-f4d5b7e5-4a36-4c39-9f8b-8e2d6a1f2c40'; + expect((await updateDocBySlug(root, { slug: SLUG, content: winner })).ok).toBe(true); + // A later write with a lower HLC loses the head race; its version stays in history. + forcedHlc = '0000000000001-000000-f4d5b7e5-4a36-4c39-9f8b-8e2d6a1f2c40'; + const loser = await updateDocBySlug(root, { + slug: SLUG, + content: SEED.replace('beta body', 'beta body\n\nloser line'), + }); + expect(loser.ok).toBe(true); + + const head = await headText(root); + expect(head).toContain('winner line'); + expect(await fetchedBody(root)).toBe(head); + }); + }); +}); diff --git a/packages/core/src/docs/docs-update.ts b/packages/core/src/docs/docs-update.ts index 50d6adef75..7f12a522ba 100644 --- a/packages/core/src/docs/docs-update.ts +++ b/packages/core/src/docs/docs-update.ts @@ -588,7 +588,7 @@ export async function updateDocBySlug( const previousAttachmentId = oldRow.id; const previousSha256 = oldRow.sha256; - const newSha256 = buf === null ? previousSha256 : sha256Of(buf); + let newSha256 = buf === null ? previousSha256 : sha256Of(buf); const wouldChange = newSha256 !== previousSha256 || oldRow.lifecycleStatus !== status; // The body schema is checked on new bytes; a lifecycle-only change keeps @@ -697,6 +697,114 @@ export async function updateDocBySlug( }; } + // T13355: project the update through the section CRDT merge layer (spec + // `cleodocs-rearch-design` §2 option b). Text bodies only — inline content + // or a UTF-8 text file (T13688); binary files stay on the whole-doc path. + // T13842: it runs BEFORE the row write, and the row (what `docs fetch` + // reads) is written from the resulting HEAD text, not the incoming bytes — + // otherwise a merged concurrent edit lives only in the conduit head and + // readers get the last writer's raw input. Best-effort: on a projection + // failure the incoming bytes are written, as before. + // T13604: a recorded conflict is threaded into the result envelope and a + // stderr warning names the sections — the update reports changed:true, so + // without this the caller would never learn their edit is not in the head. + let sectionConflicts: DocsSectionConflict[] | undefined; + let leaseNotices: DocsSectionLeaseNotice[] | undefined; + let staleBaseMerged = false; + if (text !== null) { + try { + const { projectDocUpdateThroughSections } = await import('./section-write-path.js'); + // T13356 AC4: the real session id drives lease acquire/renew, so a + // lease dies honestly with its session (TTL bound, holder-only renew). + // Gate 16: the bound session is resolved via resolveBoundSession — + // never a bare getActiveSession(). + const boundSession = await resolveBoundSession(projectRoot); + const projection = await projectDocUpdateThroughSections(projectRoot, { + slug, + authorAgentId: attachedBy, + content: text, + ...(boundSession !== null ? { sessionId: boundSession.session.id } : {}), + ...(params.message !== undefined ? { changeSummary: params.message } : {}), + // T13674: the declared base replaces the previous head as the + // single-writer/concurrent discriminator, so a stale base merges + // (append) or conflicts (non-append) instead of re-seeding over a + // concurrent writer's landed edit. + ...(baseSections !== undefined ? { baseSections } : {}), + }); + // The head is the merged text when this write advanced it; when it lost + // the head race (an explicit lower HLC) the head stays at the other + // version, so the row mirrors that version's text instead. + const headText = projection.head.advanced + ? projection.mergedText + : await readVersionText(projectRoot, projection.head.head.headVersionUid); + if (headText !== null && headText !== text) { + buf = Buffer.from(headText, 'utf-8'); + newSha256 = sha256Of(buf); + } + // T13674: the declared base was superseded AND at least one edited + // section composed into the head — its text is the merge of the + // caller's edit and a concurrent writer's. A section that only + // conflicted merged nothing, so it is reported as a conflict alone. + const conflicted = new Set(projection.conflicts.map((c) => c.sectionId)); + staleBaseMerged = + params.baseVersionUid !== undefined && + projection.previousHeadVersionUid !== null && + params.baseVersionUid !== projection.previousHeadVersionUid && + projection.sectionsModified.some((id) => !conflicted.has(id)); + if (staleBaseMerged) { + pushWarning({ + // @sync-invariant none:input-shape the warning only names a merge the CRDT projection already performed (docs.head.max-hlc); it rejects nothing and writes nothing + code: 'W_DOC_STALE_BASE', + message: + `slug '${slug}' was updated against a superseded base — a concurrent writer's edit was merged ` + + 'into the head. Re-fetch (`cleo docs fetch ' + + slug + + '`) before your next edit so it is authored against the current head.', + }); + } + if (projection.conflicts.length > 0) { + sectionConflicts = projection.conflicts.map((c) => ({ + sectionId: c.sectionId, + incomingVersionUid: c.incomingVersionUid, + projectionVersionUid: c.projectionVersionUid, + })); + pushWarning({ + // @sync-invariant none:input-shape the warning only NAMES a conflict the CRDT projection already recorded (docs.head.max-hlc); it rejects nothing and writes nothing + code: 'W_DOC_SECTION_CONFLICT', + message: + `concurrent section edit(s) on slug '${slug}' did not merge into the conduit head: ` + + projection.conflicts + .map( + (c) => + `section '${c.sectionId}' (incoming version ${c.incomingVersionUid} kept in history; head keeps projection version ${c.projectionVersionUid})`, + ) + .join('; '), + }); + } + if (projection.leaseNotices.length > 0) { + leaseNotices = projection.leaseNotices.map((n) => ({ + sectionId: n.sectionId, + holder: n.holder, + sessionId: n.sessionId, + expiresAt: n.expiresAt, + })); + pushWarning({ + // @sync-invariant none:input-shape the warning only NAMES an advisory lease held by another session (T13356); it rejects nothing and writes nothing + code: 'W_DOC_SECTION_LEASED', + message: + `section lease(s) held by another session on slug '${slug}' (advisory — your edit still landed): ` + + projection.leaseNotices + .map((n) => `section '${n.sectionId}' held by ${n.holder} until ${n.expiresAt}`) + .join('; '), + }); + } + } catch (projectionErr) { + process.emitWarning( + `docs section CRDT projection failed for update of slug '${slug}' (the incoming bytes are written): ${projectionErr instanceof Error ? projectionErr.message : String(projectionErr)}`, + ); + } + } + // Write the new blob to disk BEFORE the transaction so failure here // never leaves committed DB state pointing at missing bytes (AC3). // Orphaned blob files are harmless — they're content-addressed, so a @@ -948,101 +1056,6 @@ export async function updateDocBySlug( ); } - // T13355: project the update through the section CRDT merge layer (spec - // `cleodocs-rearch-design` §2 option b). Text bodies only — inline content - // or a UTF-8 text file (T13688); binary files stay on the whole-doc path. - // Best-effort: a projection failure must not fail the completed update — - // the whole-row write above is the fallback when no section state exists. - // T13604: a recorded conflict is threaded into the result envelope and a - // stderr warning names the sections — the update reports changed:true, so - // without this the caller would never learn their edit is not in the head. - let sectionConflicts: DocsSectionConflict[] | undefined; - let leaseNotices: DocsSectionLeaseNotice[] | undefined; - let staleBaseMerged = false; - if (text !== null) { - try { - const { projectDocUpdateThroughSections } = await import('./section-write-path.js'); - // T13356 AC4: the real session id drives lease acquire/renew, so a - // lease dies honestly with its session (TTL bound, holder-only renew). - // Gate 16: the bound session is resolved via resolveBoundSession — - // never a bare getActiveSession(). - const boundSession = await resolveBoundSession(projectRoot); - const projection = await projectDocUpdateThroughSections(projectRoot, { - slug, - authorAgentId: attachedBy, - content: text, - ...(boundSession !== null ? { sessionId: boundSession.session.id } : {}), - ...(params.message !== undefined ? { changeSummary: params.message } : {}), - // T13674: the declared base replaces the previous head as the - // single-writer/concurrent discriminator, so a stale base merges - // (append) or conflicts (non-append) instead of re-seeding over a - // concurrent writer's landed edit. - ...(baseSections !== undefined ? { baseSections } : {}), - }); - // T13674: the declared base was superseded AND at least one edited - // section composed into the head — its text is the merge of the - // caller's edit and a concurrent writer's. A section that only - // conflicted merged nothing, so it is reported as a conflict alone. - const conflicted = new Set(projection.conflicts.map((c) => c.sectionId)); - staleBaseMerged = - params.baseVersionUid !== undefined && - projection.previousHeadVersionUid !== null && - params.baseVersionUid !== projection.previousHeadVersionUid && - projection.sectionsModified.some((id) => !conflicted.has(id)); - if (staleBaseMerged) { - pushWarning({ - // @sync-invariant none:input-shape the warning only names a merge the CRDT projection already performed (docs.head.max-hlc); it rejects nothing and writes nothing - code: 'W_DOC_STALE_BASE', - message: - `slug '${slug}' was updated against a superseded base — a concurrent writer's edit was merged ` + - 'into the head. Re-fetch (`cleo docs fetch ' + - slug + - '`) before your next edit so it is authored against the current head.', - }); - } - if (projection.conflicts.length > 0) { - sectionConflicts = projection.conflicts.map((c) => ({ - sectionId: c.sectionId, - incomingVersionUid: c.incomingVersionUid, - projectionVersionUid: c.projectionVersionUid, - })); - pushWarning({ - // @sync-invariant none:input-shape the warning only NAMES a conflict the CRDT projection already recorded (docs.head.max-hlc); it rejects nothing and writes nothing - code: 'W_DOC_SECTION_CONFLICT', - message: - `concurrent section edit(s) on slug '${slug}' did not merge into the conduit head: ` + - projection.conflicts - .map( - (c) => - `section '${c.sectionId}' (incoming version ${c.incomingVersionUid} kept in history; head keeps projection version ${c.projectionVersionUid})`, - ) - .join('; '), - }); - } - if (projection.leaseNotices.length > 0) { - leaseNotices = projection.leaseNotices.map((n) => ({ - sectionId: n.sectionId, - holder: n.holder, - sessionId: n.sessionId, - expiresAt: n.expiresAt, - })); - pushWarning({ - // @sync-invariant none:input-shape the warning only NAMES an advisory lease held by another session (T13356); it rejects nothing and writes nothing - code: 'W_DOC_SECTION_LEASED', - message: - `section lease(s) held by another session on slug '${slug}' (advisory — your edit still landed): ` + - projection.leaseNotices - .map((n) => `section '${n.sectionId}' held by ${n.holder} until ${n.expiresAt}`) - .join('; '), - }); - } - } catch (projectionErr) { - process.emitWarning( - `docs section CRDT projection failed after update of slug '${slug}': ${projectionErr instanceof Error ? projectionErr.message : String(projectionErr)}`, - ); - } - } - const { readDocHeadVersionUid } = await import('./fetch-result.js'); const headVersionUid = await readDocHeadVersionUid(projectRoot, slug); return { @@ -1070,6 +1083,22 @@ export async function updateDocBySlug( }; } +/** + * A conduit version's text, read from its content-addressed blob (T13842). + * `null` when the version or its bytes cannot be read; the caller then keeps + * the incoming bytes. + */ +async function readVersionText(projectRoot: string, versionUid: string): Promise { + const { getConduitAttachmentVersion } = await import('../store/conduit-doc-history.js'); + const version = await getConduitAttachmentVersion(projectRoot, versionUid); + if (!version?.storageKey) return null; + const { readFile } = await import('node:fs/promises'); + return readFile( + join(getCleoDirAbsolute(projectRoot), 'attachments', 'sha256', version.storageKey), + 'utf-8', + ).catch(() => null); +} + /** The bytes as text when they are valid UTF-8 without NUL bytes; else null. */ function utf8TextOf(buf: Buffer): string | null { if (buf.includes(0)) return null; From bfb7b8939e2990c991ceb88265d77ba43ed57242 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 11:26:14 -0700 Subject: [PATCH 2/4] fix(T13842): preview the merge, write the row, then commit the projection review-2253 (#2338): running the projection before the blob/row write meant a later blob, read-back or transaction failure left the conduit head advanced over an old row and reported a failure, and a retry re-projected as a duplicate. The merge is now previewed with previewDocUpdateThroughSections (#2364: the real projection inside its transaction, rolled back). The row bytes are the preview's head text (mergedText when it would advance, the current head's text when it would lose the head race); wouldChange and the body schema check run on those bytes. The blob and row are written, and only then is the projection committed. A failed preview writes the incoming bytes and leaves the head unchanged (W_DOC_PROJECTION_FAILED); a failed commit after the row write is reported with the head uid; a head that diverged between preview and commit is reported (W_DOC_HEAD_DIVERGED). The noop check stays keyed on the incoming bytes. #2326's prior-version link is computed once and passed to both preview and commit. Tests (docs-update-concurrency.test.ts): a blob write failing after the merge leaves head and history untouched (E_FILE_ERROR); a failed preview writes the incoming bytes with the head unchanged; a failed commit keeps the merged row and the old head. All three fail on the previous ordering; the two earlier T13842 tests still pass. Co-Authored-By: Claude Opus 5.5 --- .changeset/docs-fetch-merged-head.md | 13 +- .../__tests__/docs-update-concurrency.test.ts | 95 +++++- packages/core/src/docs/docs-update.ts | 287 +++++++++++------- 3 files changed, 262 insertions(+), 133 deletions(-) diff --git a/.changeset/docs-fetch-merged-head.md b/.changeset/docs-fetch-merged-head.md index 86da1b49f1..ab15d57421 100644 --- a/.changeset/docs-fetch-merged-head.md +++ b/.changeset/docs-fetch-merged-head.md @@ -9,7 +9,12 @@ summary: docs fetch now returns the merged head after concurrent edits, not the from the incoming bytes, then projected the edit through the section CRDT. After a stale-base merge the conduit head held both writers' edits, but the row held only the last writer's input, so readers lost the other edit even -when the merge itself was correct. The projection now runs first and the row -is written from the resulting head text: the merged text when the write -advanced the head, the current head version's text when it lost the head -race. The incoming bytes are written only when the projection fails. +when the merge itself was correct. The merge is now previewed first (the +real projection, rolled back), the row is written from the resulting head +text (the merged text when the write advances the head, the current head's +text when it loses the head race), and the projection is committed only +after the row write succeeds, so a failed blob or row write never leaves the +head advanced over an old row. A failed preview writes the incoming bytes and +leaves the head unchanged; a failed commit after the row write is reported +with the head uid (`W_DOC_PROJECTION_FAILED`). The body schema is checked on +the merged text. diff --git a/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts b/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts index 302460c54c..979c55f680 100644 --- a/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts +++ b/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts @@ -57,29 +57,35 @@ */ import { createHash } from 'node:crypto'; -import { readFileSync, writeFileSync } from 'node:fs'; +import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; /** - * T13842: when set, the real projection runs with this explicit HLC, so a - * write can lose the head race (a minted HLC always advances the head). + * T13842 test seams over the REAL section write path: `forcedHlc` runs the + * preview and the projection with an explicit HLC (a minted one always + * advances the head); `failPreview` / `failCommit` make that step throw. */ let forcedHlc: string | undefined; +let failPreview = false; +let failCommit = false; vi.mock('../section-write-path.js', async (importOriginal) => { const actual = await importOriginal(); + type Input = Parameters[1]; + const withHlc = (input: Input): Input => + forcedHlc !== undefined ? { ...input, hlc: forcedHlc } : input; return { ...actual, - projectDocUpdateThroughSections: ( - root: string, - input: Parameters[1], - ) => - actual.projectDocUpdateThroughSections( - root, - forcedHlc !== undefined ? { ...input, hlc: forcedHlc } : input, - ), + previewDocUpdateThroughSections: (root: string, input: Input) => { + if (failPreview) throw new Error('injected preview failure'); + return actual.previewDocUpdateThroughSections(root, withHlc(input)); + }, + projectDocUpdateThroughSections: (root: string, input: Input) => { + if (failCommit) throw new Error('injected commit failure'); + return actual.projectDocUpdateThroughSections(root, withHlc(input)); + }, }; }); @@ -714,6 +720,8 @@ describe("docs fetch serves the merged head, not the last writer's raw input (T1 await fn(root); } finally { forcedHlc = undefined; + failPreview = false; + failCommit = false; closeDb(); delete process.env['CLEO_DIR']; await rm(root, { recursive: true, force: true }); @@ -777,4 +785,69 @@ describe("docs fetch serves the merged head, not the last writer's raw input (T1 expect(await fetchedBody(root)).toBe(head); }); }); + + /** Land A's Alpha append on the seed head and return the base B fetched before it. */ + async function landA(root: string): Promise<{ docUid: string; base: string; aText: string }> { + const docUid = (await resolveConduitDocUid(root, SLUG)) as string; + const base = (await getConduitDocHead(root, docUid))?.headVersionUid as string; + const aText = SEED.replace('alpha body', 'alpha body\n\nA line'); + expect((await updateDocBySlug(root, { slug: SLUG, content: aText })).ok).toBe(true); + return { docUid, base, aText }; + } + + const B_TEXT = SEED.replace('beta body', 'beta body\n\nB line'); + + it('a blob write that fails after the merge leaves the head and history untouched', async () => { + await withStore(async (root) => { + const { docUid, base, aText } = await landA(root); + const headBefore = (await getConduitDocHead(root, docUid))?.headVersionUid; + const versionsBefore = (await listConduitAttachmentVersions(root, docUid)).length; + // Block the row blob of the merged text: a directory where its file goes. + const merged = aText.replace('beta body', 'beta body\n\nB line'); + const sha = createHash('sha256').update(merged, 'utf-8').digest('hex'); + mkdirSync( + join(root, '.cleo', 'attachments', 'sha256', sha.slice(0, 2), `${sha.slice(2)}.txt`), + { + recursive: true, + }, + ); + + const b = await updateDocBySlug(root, { slug: SLUG, content: B_TEXT, baseVersionUid: base }); + + expect(b.ok).toBe(false); + if (b.ok) return; + expect(b.error.code).toBe('E_FILE_ERROR'); + expect((await getConduitDocHead(root, docUid))?.headVersionUid).toBe(headBefore); + expect((await listConduitAttachmentVersions(root, docUid)).length).toBe(versionsBefore); + expect(await fetchedBody(root)).toBe(aText); + }); + }); + + it('a failed merge preview writes the incoming bytes and leaves the head unchanged', async () => { + await withStore(async (root) => { + const { docUid, base } = await landA(root); + const headBefore = (await getConduitDocHead(root, docUid))?.headVersionUid; + failPreview = true; + + const b = await updateDocBySlug(root, { slug: SLUG, content: B_TEXT, baseVersionUid: base }); + + expect(b.ok).toBe(true); + expect(await fetchedBody(root)).toBe(B_TEXT); + expect((await getConduitDocHead(root, docUid))?.headVersionUid).toBe(headBefore); + }); + }); + + it('a failed projection commit after the row write keeps the merged row and the old head', async () => { + await withStore(async (root) => { + const { docUid, base, aText } = await landA(root); + const headBefore = (await getConduitDocHead(root, docUid))?.headVersionUid; + failCommit = true; + + const b = await updateDocBySlug(root, { slug: SLUG, content: B_TEXT, baseVersionUid: base }); + + expect(b.ok).toBe(true); + expect(await fetchedBody(root)).toBe(aText.replace('beta body', 'beta body\n\nB line')); + expect((await getConduitDocHead(root, docUid))?.headVersionUid).toBe(headBefore); + }); + }); }); diff --git a/packages/core/src/docs/docs-update.ts b/packages/core/src/docs/docs-update.ts index 833e7044d6..6df45fc204 100644 --- a/packages/core/src/docs/docs-update.ts +++ b/packages/core/src/docs/docs-update.ts @@ -43,6 +43,7 @@ import { attachmentRefs, docsAttachments } from '../store/tasks-schema.js'; import { assertTwinCollapseWritable } from '../store/twin-collapse.js'; import { writeFileAtomic } from '../tools/fs.js'; import { deriveDocLinks, linksJsonOrNull } from './derive-links.js'; +import type { ProjectDocUpdateInput } from './section-write-path.js'; import { validateDocBody } from './validate-body.js'; import { getCanonicalCleoVersion } from './version-ssot.js'; @@ -588,7 +589,70 @@ export async function updateDocBySlug( const previousAttachmentId = oldRow.id; const previousSha256 = oldRow.sha256; - let newSha256 = buf === null ? previousSha256 : sha256Of(buf); + const incomingSha256 = buf === null ? previousSha256 : sha256Of(buf); + + // T13842: the row written below is what `docs fetch` reads, so it must hold + // the HEAD text, not the incoming bytes — after a stale-base merge the head + // carries a concurrent writer's edit the incoming bytes lack. The merge is + // PREVIEWED here (the real projection, rolled back), the row is written from + // its result, and the projection is committed only after the row write + // succeeded, so a failed blob or row write never leaves the head advanced + // over an old row. Text bodies only (T13688); binary files stay whole-doc. + // A failed preview writes the incoming bytes and leaves the head unchanged. + let projectionInput: ProjectDocUpdateInput | null = null; + if ( + text !== null && + buf !== null && + params.dryRun !== true && + incomingSha256 !== previousSha256 + ) { + try { + const { previewDocUpdateThroughSections } = await import('./section-write-path.js'); + // T13734: link the pre-update bytes to the doc before the first projection. + const priorVersionUid = await recordPriorVersionOnAdopt( + projectRoot, + slug, + oldRow, + attachedBy, + ); + // T13356 AC4: the real session id drives lease acquire/renew, so a + // lease dies honestly with its session (TTL bound, holder-only renew). + // Gate 16: the bound session is resolved via resolveBoundSession — + // never a bare getActiveSession(). + const boundSession = await resolveBoundSession(projectRoot); + const input: ProjectDocUpdateInput = { + slug, + authorAgentId: attachedBy, + content: text, + ...(boundSession !== null ? { sessionId: boundSession.session.id } : {}), + ...(params.message !== undefined ? { changeSummary: params.message } : {}), + ...(priorVersionUid !== null ? { parentVersionUid: priorVersionUid } : {}), + // T13674: the declared base replaces the previous head as the + // single-writer/concurrent discriminator, so a stale base merges + // (append) or conflicts (non-append) instead of re-seeding over a + // concurrent writer's landed edit. + ...(baseSections !== undefined ? { baseSections } : {}), + }; + const preview = await previewDocUpdateThroughSections(projectRoot, input); + // The head is the merged text when this write would advance it; when it + // loses the head race (an explicit lower HLC) the head stays where it is. + const headText = preview.wouldAdvance + ? preview.mergedText + : await readCurrentHeadText(projectRoot, slug); + if (headText === null) throw new Error('the current conduit head text is unreadable'); + if (headText !== text) buf = Buffer.from(headText, 'utf-8'); + projectionInput = input; + } catch (previewErr) { + pushWarning({ + // @sync-invariant none:input-shape the warning only reports that the section projection was skipped; the incoming bytes are written as before and nothing is rejected + code: 'W_DOC_PROJECTION_FAILED', + message: + `section merge preview failed for slug '${slug}' — the incoming bytes are written and the conduit head is unchanged: ` + + (previewErr instanceof Error ? previewErr.message : String(previewErr)), + }); + } + } + const newSha256 = buf === null ? previousSha256 : sha256Of(buf); const wouldChange = newSha256 !== previousSha256 || oldRow.lifecycleStatus !== status; // The body schema is checked on new bytes; a lifecycle-only change keeps @@ -646,7 +710,7 @@ export async function updateDocBySlug( // update if the caller asked for one, and still write an audit entry so // operators can see the noop attempt. A lifecycle-only update (no new // bytes, T12654) always takes this path. - if (buf === null || newSha256 === previousSha256) { + if (buf === null || incomingSha256 === previousSha256) { // Only touch lifecycle_status if it actually changes. if (oldRow.lifecycleStatus !== status) { await db @@ -697,122 +761,6 @@ export async function updateDocBySlug( }; } - // T13355: project the update through the section CRDT merge layer (spec - // `cleodocs-rearch-design` §2 option b). Text bodies only — inline content - // or a UTF-8 text file (T13688); binary files stay on the whole-doc path. - // T13842: it runs BEFORE the row write, and the row (what `docs fetch` - // reads) is written from the resulting HEAD text, not the incoming bytes — - // otherwise a merged concurrent edit lives only in the conduit head and - // readers get the last writer's raw input. Best-effort: on a projection - // failure the incoming bytes are written, as before. - // T13604: a recorded conflict is threaded into the result envelope and a - // stderr warning names the sections — the update reports changed:true, so - // without this the caller would never learn their edit is not in the head. - let sectionConflicts: DocsSectionConflict[] | undefined; - let leaseNotices: DocsSectionLeaseNotice[] | undefined; - let staleBaseMerged = false; - if (text !== null) { - try { - const { projectDocUpdateThroughSections } = await import('./section-write-path.js'); - // T13734: link the pre-update bytes to the doc before the first projection. - const priorVersionUid = await recordPriorVersionOnAdopt( - projectRoot, - slug, - oldRow, - attachedBy, - ); - // T13356 AC4: the real session id drives lease acquire/renew, so a - // lease dies honestly with its session (TTL bound, holder-only renew). - // Gate 16: the bound session is resolved via resolveBoundSession — - // never a bare getActiveSession(). - const boundSession = await resolveBoundSession(projectRoot); - const projection = await projectDocUpdateThroughSections(projectRoot, { - slug, - authorAgentId: attachedBy, - content: text, - ...(boundSession !== null ? { sessionId: boundSession.session.id } : {}), - ...(params.message !== undefined ? { changeSummary: params.message } : {}), - ...(priorVersionUid !== null ? { parentVersionUid: priorVersionUid } : {}), - // T13674: the declared base replaces the previous head as the - // single-writer/concurrent discriminator, so a stale base merges - // (append) or conflicts (non-append) instead of re-seeding over a - // concurrent writer's landed edit. - ...(baseSections !== undefined ? { baseSections } : {}), - }); - // The head is the merged text when this write advanced it; when it lost - // the head race (an explicit lower HLC) the head stays at the other - // version, so the row mirrors that version's text instead. - const headText = projection.head.advanced - ? projection.mergedText - : await readVersionText(projectRoot, projection.head.head.headVersionUid); - if (headText !== null && headText !== text) { - buf = Buffer.from(headText, 'utf-8'); - newSha256 = sha256Of(buf); - } - // T13674: the declared base was superseded AND at least one edited - // section composed into the head — its text is the merge of the - // caller's edit and a concurrent writer's. A section that only - // conflicted merged nothing, so it is reported as a conflict alone. - const conflicted = new Set(projection.conflicts.map((c) => c.sectionId)); - staleBaseMerged = - params.baseVersionUid !== undefined && - projection.previousHeadVersionUid !== null && - params.baseVersionUid !== projection.previousHeadVersionUid && - projection.sectionsModified.some((id) => !conflicted.has(id)); - if (staleBaseMerged) { - pushWarning({ - // @sync-invariant none:input-shape the warning only names a merge the CRDT projection already performed (docs.head.max-hlc); it rejects nothing and writes nothing - code: 'W_DOC_STALE_BASE', - message: - `slug '${slug}' was updated against a superseded base — a concurrent writer's edit was merged ` + - 'into the head. Re-fetch (`cleo docs fetch ' + - slug + - '`) before your next edit so it is authored against the current head.', - }); - } - if (projection.conflicts.length > 0) { - sectionConflicts = projection.conflicts.map((c) => ({ - sectionId: c.sectionId, - incomingVersionUid: c.incomingVersionUid, - projectionVersionUid: c.projectionVersionUid, - })); - pushWarning({ - // @sync-invariant none:input-shape the warning only NAMES a conflict the CRDT projection already recorded (docs.head.max-hlc); it rejects nothing and writes nothing - code: 'W_DOC_SECTION_CONFLICT', - message: - `concurrent section edit(s) on slug '${slug}' did not merge into the conduit head: ` + - projection.conflicts - .map( - (c) => - `section '${c.sectionId}' (incoming version ${c.incomingVersionUid} kept in history; head keeps projection version ${c.projectionVersionUid})`, - ) - .join('; '), - }); - } - if (projection.leaseNotices.length > 0) { - leaseNotices = projection.leaseNotices.map((n) => ({ - sectionId: n.sectionId, - holder: n.holder, - sessionId: n.sessionId, - expiresAt: n.expiresAt, - })); - pushWarning({ - // @sync-invariant none:input-shape the warning only NAMES an advisory lease held by another session (T13356); it rejects nothing and writes nothing - code: 'W_DOC_SECTION_LEASED', - message: - `section lease(s) held by another session on slug '${slug}' (advisory — your edit still landed): ` + - projection.leaseNotices - .map((n) => `section '${n.sectionId}' held by ${n.holder} until ${n.expiresAt}`) - .join('; '), - }); - } - } catch (projectionErr) { - process.emitWarning( - `docs section CRDT projection failed for update of slug '${slug}' (the incoming bytes are written): ${projectionErr instanceof Error ? projectionErr.message : String(projectionErr)}`, - ); - } - } - // Write the new blob to disk BEFORE the transaction so failure here // never leaves committed DB state pointing at missing bytes (AC3). // Orphaned blob files are harmless — they're content-addressed, so a @@ -1064,6 +1012,99 @@ export async function updateDocBySlug( ); } + // T13842: commit the previewed projection now that the row holds its text. + // T13604: a recorded conflict is threaded into the result envelope and a + // stderr warning names the sections — the update reports changed:true, so + // without this the caller would never learn their edit is not in the head. + let sectionConflicts: DocsSectionConflict[] | undefined; + let leaseNotices: DocsSectionLeaseNotice[] | undefined; + let staleBaseMerged = false; + if (projectionInput !== null) { + try { + const { projectDocUpdateThroughSections } = await import('./section-write-path.js'); + const projection = await projectDocUpdateThroughSections(projectRoot, projectionInput); + const committedText = projection.head.advanced + ? projection.mergedText + : await readVersionText(projectRoot, projection.head.head.headVersionUid); + if (committedText !== buf.toString('utf-8')) { + pushWarning({ + // @sync-invariant none:input-shape the warning only reports that another writer landed between the preview and the commit; both versions are kept and nothing is rejected + code: 'W_DOC_HEAD_DIVERGED', + message: + `slug '${slug}': another writer landed between the merge preview and its commit — the doc row holds the previewed text, ` + + `the conduit head is ${projection.head.head.headVersionUid}. Re-fetch (\`cleo docs fetch ${slug}\`) and re-apply if needed.`, + }); + } + // T13674: the declared base was superseded AND at least one edited + // section composed into the head — its text is the merge of the + // caller's edit and a concurrent writer's. A section that only + // conflicted merged nothing, so it is reported as a conflict alone. + const conflicted = new Set(projection.conflicts.map((c) => c.sectionId)); + staleBaseMerged = + params.baseVersionUid !== undefined && + projection.previousHeadVersionUid !== null && + params.baseVersionUid !== projection.previousHeadVersionUid && + projection.sectionsModified.some((id) => !conflicted.has(id)); + if (staleBaseMerged) { + pushWarning({ + // @sync-invariant none:input-shape the warning only names a merge the CRDT projection already performed (docs.head.max-hlc); it rejects nothing and writes nothing + code: 'W_DOC_STALE_BASE', + message: + `slug '${slug}' was updated against a superseded base — a concurrent writer's edit was merged ` + + 'into the head. Re-fetch (`cleo docs fetch ' + + slug + + '`) before your next edit so it is authored against the current head.', + }); + } + if (projection.conflicts.length > 0) { + sectionConflicts = projection.conflicts.map((c) => ({ + sectionId: c.sectionId, + incomingVersionUid: c.incomingVersionUid, + projectionVersionUid: c.projectionVersionUid, + })); + pushWarning({ + // @sync-invariant none:input-shape the warning only NAMES a conflict the CRDT projection already recorded (docs.head.max-hlc); it rejects nothing and writes nothing + code: 'W_DOC_SECTION_CONFLICT', + message: + `concurrent section edit(s) on slug '${slug}' did not merge into the conduit head: ` + + projection.conflicts + .map( + (c) => + `section '${c.sectionId}' (incoming version ${c.incomingVersionUid} kept in history; head keeps projection version ${c.projectionVersionUid})`, + ) + .join('; '), + }); + } + if (projection.leaseNotices.length > 0) { + leaseNotices = projection.leaseNotices.map((n) => ({ + sectionId: n.sectionId, + holder: n.holder, + sessionId: n.sessionId, + expiresAt: n.expiresAt, + })); + pushWarning({ + // @sync-invariant none:input-shape the warning only NAMES an advisory lease held by another session (T13356); it rejects nothing and writes nothing + code: 'W_DOC_SECTION_LEASED', + message: + `section lease(s) held by another session on slug '${slug}' (advisory — your edit still landed): ` + + projection.leaseNotices + .map((n) => `section '${n.sectionId}' held by ${n.holder} until ${n.expiresAt}`) + .join('; '), + }); + } + } catch (projectionErr) { + const { readDocHeadVersionUid: readHead } = await import('./fetch-result.js'); + const head = await readHead(projectRoot, slug); + pushWarning({ + // @sync-invariant none:input-shape the warning only reports a failed projection commit after the row write; the row and the version history are intact and nothing is rejected + code: 'W_DOC_PROJECTION_FAILED', + message: + `slug '${slug}': the doc row was written from the previewed merge, but committing the section projection failed ` + + `(${projectionErr instanceof Error ? projectionErr.message : String(projectionErr)}); the conduit head stays at ${head ?? 'none'} — re-run the update to land it`, + }); + } + } + const { readDocHeadVersionUid } = await import('./fetch-result.js'); const headVersionUid = await readDocHeadVersionUid(projectRoot, slug); return { @@ -1107,6 +1148,16 @@ async function readVersionText(projectRoot: string, versionUid: string): Promise ).catch(() => null); } +/** The current conduit head's text for a slug (T13842), or `null` when it has none or it is unreadable. */ +async function readCurrentHeadText(projectRoot: string, slug: string): Promise { + const { getConduitDocHead, resolveConduitDocUid } = await import( + '../store/conduit-doc-history.js' + ); + const docUid = await resolveConduitDocUid(projectRoot, slug); + const head = docUid !== null ? await getConduitDocHead(projectRoot, docUid) : null; + return head !== null ? readVersionText(projectRoot, head.headVersionUid) : null; +} + /** The bytes as text when they are valid UTF-8 without NUL bytes; else null. */ function utf8TextOf(buf: Buffer): string | null { if (buf.includes(0)) return null; From b94560b566668630da2667111331be3579818f0a Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 11:28:45 -0700 Subject: [PATCH 3/4] chore(T13842): tag the caught head-unreadable throw for gate 38 Co-Authored-By: Claude Opus 5.5 --- packages/core/src/docs/docs-update.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/packages/core/src/docs/docs-update.ts b/packages/core/src/docs/docs-update.ts index 6df45fc204..25eea01ef6 100644 --- a/packages/core/src/docs/docs-update.ts +++ b/packages/core/src/docs/docs-update.ts @@ -639,7 +639,10 @@ export async function updateDocBySlug( const headText = preview.wouldAdvance ? preview.mergedText : await readCurrentHeadText(projectRoot, slug); - if (headText === null) throw new Error('the current conduit head text is unreadable'); + if (headText === null) { + // @sync-invariant none:input-shape caught just below: an unreadable head skips the projection and the incoming bytes are written; nothing is rejected + throw new Error('the current conduit head text is unreadable'); + } if (headText !== text) buf = Buffer.from(headText, 'utf-8'); projectionInput = input; } catch (previewErr) { From 1ec71e436ea7620f4cd183e35f8f337c967ff244 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Sat, 17 Oct 2026 11:32:00 -0700 Subject: [PATCH 4/4] fix(T13881): no in-process writer can land on the shared handle during the merge preview previewDocUpdateThroughSections rolls back on the process-wide shared native handle, so any write another in-process writer put on that handle during the preview's awaits would be rolled back with it. docs-update now (1) settles every tracked background op first (bounded; the preview is skipped, incoming bytes written, when one is still running) and (2) runs the preview on the handle's writer queue via scheduleTaskBackground, so accessor transactions and committed background effects queued meanwhile start only after the rollback. The settle does not consume the shutdown receipt's ledger. Co-Authored-By: Claude Opus 5.5 --- packages/core/src/docs/docs-update.ts | 39 +++++++++++++++++++++++++-- 1 file changed, 37 insertions(+), 2 deletions(-) diff --git a/packages/core/src/docs/docs-update.ts b/packages/core/src/docs/docs-update.ts index 25eea01ef6..9f7fa4875b 100644 --- a/packages/core/src/docs/docs-update.ts +++ b/packages/core/src/docs/docs-update.ts @@ -43,7 +43,7 @@ import { attachmentRefs, docsAttachments } from '../store/tasks-schema.js'; import { assertTwinCollapseWritable } from '../store/twin-collapse.js'; import { writeFileAtomic } from '../tools/fs.js'; import { deriveDocLinks, linksJsonOrNull } from './derive-links.js'; -import type { ProjectDocUpdateInput } from './section-write-path.js'; +import type { DocUpdatePreview, ProjectDocUpdateInput } from './section-write-path.js'; import { validateDocBody } from './validate-body.js'; import { getCanonicalCleoVersion } from './version-ssot.js'; @@ -633,7 +633,35 @@ export async function updateDocBySlug( // concurrent writer's landed edit. ...(baseSections !== undefined ? { baseSections } : {}), }; - const preview = await previewDocUpdateThroughSections(projectRoot, input); + // T13881: the preview ROLLS BACK on the process-wide shared native handle, + // so a write another in-process writer lands on that handle during its + // awaits would be rolled back with it. Two guards: (1) every tracked + // background op (graph nodes, hook dispatches, mirrors) settles first, + // and the preview is skipped if one is still running; (2) the preview + // runs on the handle's writer queue (scheduleTaskBackground), so accessor + // transactions and committed background effects queued meanwhile start + // only after its rollback. + const { settleBackgroundOps } = await import('../store/background-ops.js'); + const pending = await settleBackgroundOps(PREVIEW_SETTLE_BUDGET_MS); + if (pending > 0) { + // @sync-invariant none:input-shape caught just below: the preview is skipped and the incoming bytes are written; nothing is rejected + throw new Error(`${pending} background write(s) still running; the preview is skipped`); + } + const native = getNativeTasksDb(projectRoot); + if (!native) { + // @sync-invariant none:input-shape caught just below: without the store handle the preview is skipped and the incoming bytes are written; nothing is rejected + throw new Error('tasks database is not initialised'); + } + const { scheduleTaskBackground } = await import('../store/sqlite-data-accessor.js'); + const queued: { preview?: DocUpdatePreview } = {}; + await scheduleTaskBackground(native, async () => { + queued.preview = await previewDocUpdateThroughSections(projectRoot, input); + }); + const preview = queued.preview; + if (preview === undefined) { + // @sync-invariant none:input-shape caught just below: an empty preview skips the projection and the incoming bytes are written; nothing is rejected + throw new Error('the merge preview returned nothing'); + } // The head is the merged text when this write would advance it; when it // loses the head race (an explicit lower HLC) the head stays where it is. const headText = preview.wouldAdvance @@ -1135,6 +1163,13 @@ export async function updateDocBySlug( }; } +/** + * How long a docs update waits for tracked background writes to settle before + * it previews a merge (T13881). A write still running after this is not + * rolled back by the preview: the preview is skipped instead. + */ +const PREVIEW_SETTLE_BUDGET_MS = 5_000; + /** * A conduit version's text, read from its content-addressed blob (T13842). * `null` when the version or its bytes cannot be read; the caller then keeps