Better sync: defer hot paths, rebase before park

src/cli/main.ts

@@ -1684,7 +1684,7 @@ function buildProgram(): Command {
     .option("--debounce <ms>", "burst debounce in ms (daemon; default 5000)", parsePositiveInt)
     .option(
       "--settle <ms>",
-      "skip files younger than this many ms (partial saves)",
+      "minimum mtime age in ms before sync touches a path (default 0)",
       parsePositiveInt,
     )
     .option(

src/sync/converge.ts

@@ -0,0 +1,76 @@
+/**
+ * Shared disk↔kernel write helpers used by push, feed, and reconcile so
+ * inbound rebase and outbound ack stay byte-identical (better-sync.plan).
+ */
+
+import type { KernelClient } from "../client/kernel-client.js";
+import { withVersionSuffix } from "../kernel/deletion.js";
+import type { Version } from "../kernel/wire.js";
+import { extractSystemProperties, split } from "../markdown/frontmatter.js";
+import { renderIgnoredSibling, renderMaterialized, stampProvenance } from "./intrinsics.js";
+import type { FileStore } from "./reconcile.js";
+
+export type UserContent = { frontmatter_raw: string; body: string };
+
+/** Split a file into stored-shape fields, dropping all `$*` intrinsic lines. */
+export function stripToUserContent(text: string): UserContent {
+  const lf = text.replace(/\r\n/g, "\n");
+  const { frontmatter_raw, body } = split(lf);
+  return { frontmatter_raw: extractSystemProperties(frontmatter_raw).raw, body };
+}
+
+/**
+ * After a successful kernel write, restamp the local file. If the editor saved
+ * again during the round-trip, keep those bytes and only update provenance —
+ * never write the snapshot we just pushed over newer typing.
+ */
+export async function ackLocalWrite(
+  store: FileStore,
+  path: string,
+  v: Version,
+  pushed: UserContent,
+): Promise<void> {
+  const now = await store.read(path);
+  if (now === null) return;
+  const current = stripToUserContent(now);
+  if (current.body === pushed.body && current.frontmatter_raw === pushed.frontmatter_raw) {
+    await store.write(path, renderMaterialized(v), { preserveMtime: true });
+    return;
+  }
+  await store.write(path, stampProvenance(now, v.version_id, v.content_hash), {
+    preserveMtime: true,
+  });
+}
+
+/** Optimistic put of local user bytes, then ack provenance on disk. */
+export async function putAndAck(
+  client: KernelClient,
+  store: FileStore,
+  repo: string,
+  path: string,
+  prevVersionId: string,
+  user: UserContent,
+): Promise<Version> {
+  const v = await client.docs.put(repo, prevVersionId, path, user);
+  await ackLocalWrite(store, path, v, user);
+  return v;
+}
+
+/** Park remote as `<name>-<version_id>.md` with `$sync: ignore` (§4.8). */
+export async function parkIgnoredSibling(
+  store: FileStore,
+  path: string,
+  version: Version,
+): Promise<void> {
+  await store.write(withVersionSuffix(path, version.version_id), renderIgnoredSibling(version));
+}
+
+/** Write remote version bytes at the canonical path (fast-forward / materialize). */
+export async function materializeAt(
+  store: FileStore,
+  path: string,
+  version: Version,
+  opts?: { preserveMtime?: boolean },
+): Promise<void> {
+  await store.write(path, renderMaterialized(version), opts);
+}

src/sync/daemon.ts

@@ -9,6 +9,10 @@
  *     first so unlink+add of a rename is a move, not a delete (§4.7); each
  *     path is still stated, not trusted by event type (§4.6).
  *
+ * better-sync.plan: `--settle` is an mtime-age gate on both directions; inbound
+ * conflicts on hot files are held in an in-memory deferred map and retried on
+ * later polls (cursor still advances).
+ *
  * chokidar is confined to this module (§4.4). Echo suppression falls out of
  * self-description (§4.5): our own pushes come back on the feed but the local
  * file already embeds that version+hash, and our own writes trip the watcher
@@ -21,6 +25,11 @@ import type { KernelClient } from "../client/kernel-client.js";
 import { type SyncCursor, readCursor, sourceFields, writeCursor } from "./cursor.js";
 import { applyFeed } from "./feed.js";
 import { createFsStore } from "./fs-store.js";
+import {
+  type DeferredMap,
+  pathIsHot,
+  retryDeferredEntry,
+} from "./hot-path.js";
 import { isIgnored, readFileIntrinsics } from "./intrinsics.js";
 import { SYNC_DIR, type ScopeFilter, makeScopeFilter, toDocPath } from "./paths.js";
 import { type RemoteMap, pushBurst, pushPath } from "./push.js";
@@ -55,8 +64,10 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon {
   const store = createFsStore(opts.root);
   const scope = makeScopeFilter({ include: opts.include, exclude: opts.exclude });
   const map: RemoteMap = new Map();
+  const deferred: DeferredMap = new Map();
   const intervalMs = opts.intervalMs ?? 5000;
   const debounceMs = opts.debounceMs ?? 5000;
+  const settleMs = opts.settleMs ?? 0;
 
   let stopped = false;
   let watcher: FSWatcher | undefined;
@@ -99,34 +110,73 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon {
     }
   }
 
+  /** Retry in-memory deferred inbound holds before draining new feed pages. */
+  async function retryDeferred(): Promise<void> {
+    if (deferred.size === 0) return;
+    for (const [path, entry] of [...deferred.entries()]) {
+      if (!scope.matches(path) && entry.ref.op !== "delete") {
+        deferred.delete(path);
+        continue;
+      }
+      try {
+        const { done } = await retryDeferredEntry(client, store, opts.repo, path, entry, {
+          settleMs,
+          map,
+          deferred,
+        });
+        if (done) deferred.delete(path);
+      } catch (err) {
+        log(`defer retry error\t${path}\t${(err as Error).message}`);
+      }
+    }
+  }
+
   async function pollFeedOnce(): Promise<void> {
+    await retryDeferred();
     const { cursor: next } = await applyFeed(client, store, scope, {
       repo: opts.repo,
       since: cursor,
       log,
       map,
+      settleMs,
+      deferred,
     });
     if (next !== cursor) {
       cursor = next;
       await persistCursor();
     }
   }
 
-  function schedulePush(docPath: string): void {
-    if (!scope.matches(docPath)) return;
-    pending.add(docPath);
+  function armBurstTimer(): void {
     if (burstTimer) clearTimeout(burstTimer);
     burstTimer = setTimeout(() => {
       burstTimer = undefined;
       const batch = [...pending];
       pending.clear();
       void serialize(async () => {
         if (stopped) return;
-        await pushBurst(batch, { client, store, repo: opts.repo, map, log });
+        const ready: string[] = [];
+        for (const path of batch) {
+          if (settleMs > 0 && (await pathIsHot(store, path, settleMs))) {
+            pending.add(path);
+            continue;
+          }
+          ready.push(path);
+        }
+        if (pending.size > 0) armBurstTimer();
+        if (ready.length > 0) {
+          await pushBurst(ready, { client, store, repo: opts.repo, map, log });
+        }
       });
     }, debounceMs);
   }
 
+  function schedulePush(docPath: string): void {
+    if (!scope.matches(docPath)) return;
+    pending.add(docPath);
+    armBurstTimer();
+  }
+
   const ready = (async () => {
     // 1. Startup is deterministic on the cursor marker (§4.9, §7).
     const existing = await readCursor(opts.root);
@@ -160,8 +210,13 @@ export function startDaemon(client: KernelClient, opts: DaemonOptions): Daemon {
       ignoreInitial: true,
       // Prune the sync state dir; the scope filter still guards everything else.
       ignored: (p: string) => p.includes(`/${SYNC_DIR}/`) || p.endsWith(`/${SYNC_DIR}`),
-      ...(opts.settleMs
-        ? { awaitWriteFinish: { stabilityThreshold: opts.settleMs, pollInterval: 100 } }
+      ...(settleMs > 0
+        ? {
+            awaitWriteFinish: {
+              stabilityThreshold: Math.min(settleMs, 2000),
+              pollInterval: 100,
+            },
+          }
         : {}),
     });
     const onEvent = (abs: string): void => {