Merge remote-tracking branch 'origin/master' into jsonl-packed-chunk-rows

Master removed the stdio demo (#702c8cc30) — accept the deletion; this branch's packChunks passthrough survives in acp-demo (auto-merged), and cli-demo/tui-demo arrived from master without one (the follow-up snapshot PR decides which demos expose the switch). Generated catalogs regenerated over merged sources; the hand-written session.md durability paragraph re-weaves this branch's lossless-encoding wording with master's invariant- companion sentence.
2026-07-22 15:46:12 +08:00
parent 3869f2ce17 4267076407
commit 3922e956db
1242 changed files with 57560 additions and 15291 deletions
@@ -8,7 +8,7 @@

 import { Context } from 'cordis'
 import z from 'schemastery'
-import { open, mkdir, readFile, readdir, link, rm, truncate } from 'node:fs/promises'
+import { open, mkdir, readFile, readdir, link, rm, stat as fsStat, truncate } from 'node:fs/promises'
 import { dirname, resolve } from 'node:path'
 import { randomBytes } from 'node:crypto'
 import {
@@ -21,6 +21,7 @@ import {
  type JsonlCompression,
 } from './format.ts'
 import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from './zstd.ts'
+import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts'

 export type { JsonlCompression } from './format.ts'

@@ -92,9 +93,6 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
  private coordinator: PersistenceCoordinator<JsonlTornMarker>
  private rootEncodingCheck: Promise<void> | undefined

-  /** Runtime host platform used to decide whether directory sync is supported. */
-  readonly internals: { platform: NodeJS.Platform } = { platform: process.platform }
-
  constructor(ctx: Context, public config: Config) {
    super(ctx)
    // Resolve once so later process.cwd() changes cannot split one backend across roots.
@@ -268,32 +266,36 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi

  // --- materialization / append / repair (file mechanics) ---

-  /** Atomically write the header line + first batch (temp-write, fsync, collision-safe hard-link publish). */
+  /** Atomically write the header line + first batch (temp-write, fsync, publish). */
  private async materialize(meta: SessionHeader, events: readonly SessionEvent[]): Promise<void> {
    const dir = sessionDir(this.root, meta.cwd)
-    await mkdir(this.root, { recursive: true, mode: 0o700 })
-    await this.syncDir(dirname(this.root))
-    await mkdir(dir, { recursive: true, mode: 0o700 })
-    await this.syncDir(this.root)
    const finalPath = logPath(this.root, meta.cwd, meta.id, this.compression)
-    // Materialization is the first write; an existing log is an id collision.
-    /* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
-    if (await this.exists(finalPath)) {
-      throw new Error(`refusing to materialize "${meta.id}": a log already exists on disk (load/resume it instead)`)
-    }
    await this.rejectOppositeArtifact(meta.cwd, meta.id)
    const content = await this.encodeMaterialization(meta, events)
-
-    const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
-    const handle = await open(tmp, 'wx', 0o600)
-    try {
-      await handle.writeFile(content)
-      await handle.sync()
-    } finally {
-      await handle.close()
+    /* v8 ignore next -- native Windows coverage exercises this platform dispatch; Linux covers the POSIX peer */
+    if (process.platform === 'win32') {
+      await this.materializeWin32(dir, finalPath, meta.id, content)
+    } else {
+      await this.materializePosix(dir, finalPath, meta.id, content)
    }
-    // Publish with link()+unlink(): unlike rename(), link fails if another
-    // process materialized the same id first.
+  }
+
+  /* v8 ignore start -- Windows uses the Win32 durable-publish path; POSIX coverage exercises this peer. */
+  private async materializePosix(
+    dir: string,
+    finalPath: string,
+    id: SessionId,
+    content: Buffer | string,
+  ): Promise<void> {
+    await mkdir(this.root, { recursive: true, mode: 0o700 })
+    await this.syncDirPosix(dirname(this.root))
+    await mkdir(dir, { recursive: true, mode: 0o700 })
+    await this.syncDirPosix(this.root)
+    await this.rejectExistingLog(finalPath, id)
+    const tmp = await this.writeSyncedTempFile(finalPath, content)
+    // Publish via link()+unlink(), NOT rename(): link fails with EEXIST if the
+    // final path already exists, so two processes materializing the same id
+    // concurrently cannot clobber each other. rename() would silently overwrite.
    let linked = false
    try {
      await link(tmp, finalPath)
@@ -304,16 +306,64 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
      /* v8 ignore next -- link failure is the TOCTOU/IO race guarded above; not reachable in test */
      if (!linked) await rm(tmp, { force: true })
    }
-    // The published link becomes crash-durable only after its directory fsync.
-    await this.syncDir(dir)
-    // Best-effort temp cleanup: the log is already published and durable, so a failure to
-    // remove the (now-redundant) temp hard link must not reject the append.
+    // link() succeeded — the log is published. fsync the directory so the new
+    // entry survives a power loss: the new link is not crash-durable until the
+    // parent directory's metadata is synced.
+    await this.syncDirPosix(dir)
+    // Best-effort temp cleanup: the log is already published and durable, so a
+    // failure to remove the (now-redundant) temp hard link must NOT reject the
+    // append. Swallow only the rm failure; nothing else of consequence runs here.
    try {
      await rm(tmp, { force: true })
    } catch {
      /* v8 ignore next -- redundant temp link; publish already durable, rm failure is an unreachable IO edge */
    }
  }
+  /* v8 ignore stop */
+
+  /* v8 ignore start -- native Windows coverage exercises this integration path */
+  private async materializeWin32(
+    dir: string,
+    finalPath: string,
+    id: SessionId,
+    content: Buffer | string,
+  ): Promise<void> {
+    await ensureDurableDirectoryWin32(this.root)
+    await ensureDurableDirectoryWin32(dir)
+    await this.rejectExistingLog(finalPath, id)
+    const tmp = await this.writeSyncedTempFile(finalPath, content)
+    try {
+      await publishNewFileWin32(tmp, finalPath)
+    } catch (error) {
+      await rm(tmp, { force: true })
+      throw error
+    }
+  }
+  /* v8 ignore stop */
+
+  private async rejectExistingLog(finalPath: string, id: SessionId): Promise<void> {
+    // Never publish over an existing committed log: materialize is the first
+    // write of a session the backend believes is new. A file here means a
+    // different session shares this id on disk — reject loudly. (createCore
+    // already guards the create path, so this is unreachable-in-practice TOCTOU
+    // defense.)
+    /* v8 ignore next 3 -- createCore guards collisions before materialize; this is a TOCTOU backstop */
+    if (await this.exists(finalPath)) {
+      throw new Error(`refusing to materialize "${id}": a log already exists on disk (load/resume it instead)`)
+    }
+  }
+
+  private async writeSyncedTempFile(finalPath: string, content: Buffer | string): Promise<string> {
+    const tmp = `${finalPath}.${randomBytes(6).toString('hex')}.tmp`
+    const handle = await open(tmp, 'wx', 0o600)
+    try {
+      await handle.writeFile(content)
+      await handle.sync()
+    } finally {
+      await handle.close()
+    }
+    return tmp
+  }

  /** Encode the header and first batch without combining their frame boundaries. */
  private async encodeMaterialization(meta: SessionHeader, events: readonly SessionEvent[]): Promise<Buffer | string> {
@@ -331,22 +381,17 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
    return this.compression === 'zstd' ? compressZstdFrame(body) : body
  }

-  /** fsync a directory when the host exposes that durability primitive. */
-  private async syncDir(dir: string): Promise<void> {
+  /** fsync a POSIX directory so a just-created/renamed entry is crash-durable. */
+  /* v8 ignore start -- Windows uses write-through namespace operations; POSIX coverage exercises directory fsync. */
+  private async syncDirPosix(dir: string): Promise<void> {
    const handle = await open(dir, 'r')
    try {
-      try {
-        await handle.sync()
-      } catch (error: unknown) {
-        const code = (error as NodeJS.ErrnoException | null)?.code
-        // Node opens directories on Windows but its fsync binding rejects them.
-        // File-content fsync remains mandatory; only this unsupported primitive is skipped.
-        if (this.internals.platform !== 'win32' || code !== 'EPERM') throw error
-      }
+      await handle.sync()
    } finally {
      await handle.close()
    }
  }
+  /* v8 ignore stop */

  /**
   * Append and fsync event lines. On a partial write or sync failure, restore the
@@ -357,17 +402,37 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
    const content = await this.encodeEventBatch(events)
    const path = logPath(this.root, meta.cwd, meta.id, this.compression)
    const handle = await open(path, 'a')
+    let closed = false
+    const closeAppendHandle = async (): Promise<void> => {
+      if (closed) return
+      closed = true
+      await handle.close()
+    }
+
    try {
      const { size: before } = await handle.stat()
      try {
        await handle.writeFile(content)
        await handle.sync()
      } catch (error) {
-        // Roll back whatever bytes landed so a retry starts from a clean EOF.
-        await handle.truncate(before)
-        await handle.sync()
+        try {
+          await closeAppendHandle()
+          await this.rollbackAppend(path, before)
+        } catch (rollbackError) {
+          throw new AggregateError([error, rollbackError], `failed to roll back append to "${path}"`)
+        }
        throw error
      }
+    } finally {
+      await closeAppendHandle()
+    }
+  }
+
+  private async rollbackAppend(path: string, size: number): Promise<void> {
+    const handle = await open(path, 'r+')
+    try {
+      await handle.truncate(size)
+      await handle.sync()
    } finally {
      await handle.close()
    }
@@ -519,13 +584,36 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
      await handle.close()
      return true
    } catch (error) {
-      // Only ENOENT means absent. A permission/I/O error must surface, not be
-      // collapsed to `false` — otherwise load() reports "not found" and collision
-      // checks proceed under a false absence assumption.
-      if (isENOENT(error)) return false
+      // Only ENOENT means absent. A permission/I/O error must surface rather
+      // than letting load or collision checks proceed under false absence.
+      // Windows reports ENOENT, not ENOTDIR, for `regular-file/child`; verify
+      // the immediate parent so a blocked cwd bucket remains a storage fault.
+      /* v8 ignore else -- Windows reports file-valued parents as ENOENT; POSIX covers direct ENOTDIR. */
+      if (isENOENT(error)) {
+        await this.assertLogParentAllowsAbsence(path)
+        return false
+      }
+      /* v8 ignore next -- Windows repairs ENOTDIR from ENOENT above; POSIX covers direct ENOTDIR. */
      throw error
    }
  }
+
+  /* v8 ignore start -- native Windows coverage exercises this repair; POSIX open reports ENOTDIR before this point. */
+  private async assertLogParentAllowsAbsence(path: string): Promise<void> {
+    try {
+      const parent = dirname(path)
+      const info = await fsStat(parent)
+      if (info.isDirectory()) return
+      const error = new Error(`ENOTDIR: parent path exists but is not a directory: ${parent}`) as NodeJS.ErrnoException
+      error.code = 'ENOTDIR'
+      error.path = parent
+      throw error
+    } catch (error) {
+      if (isENOENT(error)) return
+      throw error
+    }
+  }
+  /* v8 ignore stop */
 }

 export default SessionPersistenceJsonl
@@ -0,0 +1,30 @@
+/**
+ * Package-owned invariant companion for `@deepseek-ai/dsh-session-persistence-jsonl`.
+ * @module @deepseek-ai/dsh-session-persistence-jsonl/invariant
+ */
+
+/* jscpd:ignore-start */
+import type { Context } from 'cordis'
+import type { InvariantInstaller } from '@deepseek-ai/dsh-invariants'
+
+const PACKAGE_NAME = '@deepseek-ai/dsh-session-persistence-jsonl'
+
+/** Cordis companion plugin name. */
+export const name = 'session-persistence-jsonl-invariant'
+/** Service required before the companion can reserve package ownership. */
+export const inject = ['invariants']
+
+/**
+ * No runtime invariant: persistence correctness requires backend round-trip and crash-tail tests;
+ * this package exposes no continuously observable in-process relation.
+ */
+const install: InvariantInstaller = () => {}
+
+/**
+ * Register this package's invariant companion.
+ * @param ctx - Cordis context carrying the invariant service.
+ * @returns the installed registration's disposer after setup succeeds.
+ */
+export const apply = (ctx: Context): Promise<() => void> =>
+  Promise.resolve(ctx.invariants.register(PACKAGE_NAME, install))
+/* jscpd:ignore-end */
@@ -0,0 +1,150 @@
+/**
+ * Windows durable namespace helpers for the JSONL backend.
+ *
+ * POSIX publishes a newly-created log by creating a directory entry and then
+ * fsyncing the parent directory. Windows does not expose that parent-directory
+ * fsync contract through Node, so the Windows path uses the native durable
+ * namespace primitive instead: create a staging object in the target directory
+ * and publish it with `MoveFileExW(..., MOVEFILE_WRITE_THROUGH)` without
+ * replacement or cross-volume copy fallback.
+ *
+ * @module dsh-session-persistence-jsonl/win32
+ */
+
+import { mkdtemp, rm, stat } from 'node:fs/promises'
+import { basename, join, parse, resolve, toNamespacedPath } from 'node:path'
+
+type MoveFileExW = (existing: string, replacement: string, flags: number) => number
+type GetLastError = () => number
+
+interface Win32Bindings {
+  moveFileExW: MoveFileExW
+  getLastError: GetLastError
+}
+
+interface Win32ErrnoException extends NodeJS.ErrnoException {
+  win32Code: number
+  dest: string
+}
+
+const MOVEFILE_WRITE_THROUGH = 0x00000008
+const ERROR_FILE_NOT_FOUND = 2
+const ERROR_PATH_NOT_FOUND = 3
+const ERROR_ACCESS_DENIED = 5
+const ERROR_NOT_SAME_DEVICE = 17
+const ERROR_FILE_EXISTS = 80
+const ERROR_INVALID_NAME = 123
+const ERROR_ALREADY_EXISTS = 183
+
+let bindings: Win32Bindings | undefined
+
+/** Load the small Win32 surface lazily so non-Windows processes never load Koffi. */
+async function win32(): Promise<Win32Bindings> {
+  if (bindings !== undefined) return bindings
+  const koffi = (await import('koffi')).default
+  const kernel32 = koffi.load('kernel32.dll')
+  bindings = {
+    moveFileExW: kernel32.func('__stdcall', 'MoveFileExW', 'int', ['str16', 'str16', 'uint']) as MoveFileExW,
+    getLastError: kernel32.func('__stdcall', 'GetLastError', 'uint', []) as GetLastError,
+  }
+  return bindings
+}
+
+function errnoCode(win32Code: number): string {
+  switch (win32Code) {
+    case ERROR_FILE_NOT_FOUND:
+    case ERROR_PATH_NOT_FOUND:
+      return 'ENOENT'
+    case ERROR_ACCESS_DENIED:
+      return 'EACCES'
+    case ERROR_NOT_SAME_DEVICE:
+      return 'EXDEV'
+    case ERROR_FILE_EXISTS:
+    case ERROR_ALREADY_EXISTS:
+      return 'EEXIST'
+    case ERROR_INVALID_NAME:
+      return 'EINVAL'
+    default:
+      return 'EIO'
+  }
+}
+
+function win32Error(syscall: string, win32Code: number, path: string, dest: string): Win32ErrnoException {
+  const code = errnoCode(win32Code)
+  const error = new Error(`${syscall} ${code} (Win32 ${win32Code}): ${path} -> ${dest}`) as Win32ErrnoException
+  error.code = code
+  error.errno = win32Code
+  error.syscall = syscall
+  error.path = path
+  error.dest = dest
+  error.win32Code = win32Code
+  return error
+}
+
+function isENOENT(error: unknown): boolean {
+  return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT'
+}
+
+function isEEXIST(error: unknown): boolean {
+  return (error as NodeJS.ErrnoException | null)?.code === 'EEXIST'
+}
+
+async function assertDirectory(path: string): Promise<boolean> {
+  try {
+    const info = await stat(path)
+    if (info.isDirectory()) return true
+    const error = new Error(`path exists but is not a directory: ${path}`) as NodeJS.ErrnoException
+    error.code = 'ENOTDIR'
+    error.path = path
+    throw error
+  } catch (error) {
+    if (isENOENT(error)) return false
+    throw error
+  }
+}
+
+/**
+ * Publish `existing` at `replacement` with Windows write-through rename
+ * semantics. The destination must not already exist; the move must stay within
+ * the volume (no copy fallback flag is set).
+ * @param existing - the synced staging path to move.
+ * @param replacement - the final path, which must not already exist.
+ */
+export async function publishNewFileWin32(existing: string, replacement: string): Promise<void> {
+  const api = await win32()
+  const ok = api.moveFileExW(toNamespacedPath(existing), toNamespacedPath(replacement), MOVEFILE_WRITE_THROUGH)
+  if (ok === 0) throw win32Error('MoveFileExW', api.getLastError(), existing, replacement)
+}
+
+/**
+ * Create `target` and its missing ancestors with durable Windows namespace
+ * publication. Each missing directory is first created as a random staging
+ * sibling, then moved to its final name with `MOVEFILE_WRITE_THROUGH`; races
+ * with another creator are accepted only after verifying the winner is a
+ * directory.
+ * @param target - the absolute directory path to create durably when absent.
+ */
+export async function ensureDurableDirectoryWin32(target: string): Promise<void> {
+  const absolute = resolve(target)
+  const root = parse(absolute).root
+  await assertDirectory(root)
+
+  const segments = absolute.slice(root.length).split(/[\\/]+/).filter(part => part.length > 0)
+  let current = root
+  for (const segment of segments) {
+    const next = join(current, segment)
+    if (!await assertDirectory(next)) await createLeafDirectoryWin32(current, next)
+    current = next
+  }
+}
+
+async function createLeafDirectoryWin32(parent: string, target: string): Promise<void> {
+  const staging = await mkdtemp(join(parent, `.dsh-mkdir-${basename(target)}-`))
+  try {
+    await publishNewFileWin32(staging, target)
+  } catch (error) {
+    await rm(staging, { recursive: true, force: true })
+    if (isEEXIST(error) && await assertDirectory(target)) return
+    throw error
+  }
+}