feat(drive): watch resilience, conflict reconciliation, and incremental sync

src/handwritten/commands/drive/api.ts

@@ -1,15 +1,24 @@
 import { loadSdkClientWithAuthedFetch } from "../../auth/load-sdk-client.js"
 import {
   driveFileDelete,
+  driveFileMove,
   driveLibraryGet,
   driveManifestGet,
 } from "../../../generated/sdk/index.js"
-import type { DeleteDriveFileResponse, DriveLibrary, DriveManifestResponse, UploadDriveFileResponse } from "../../../generated/sdk/index.js"
+import type {
+  DeleteDriveFileResponse,
+  DriveLibrary,
+  DriveManifestResponse,
+  MoveDriveFileResponse,
+  UploadDriveFileResponse,
+} from "../../../generated/sdk/index.js"
 import type { ConfigStore } from "../../config/index.js"
 
 export interface DriveApiOptions {
   store?: ConfigStore
   fetchImpl?: typeof fetch
+  // Watch session client id; lets the server tag events for echo suppression.
+  clientId?: string
 }
 
 type JsonResult<T> = {
@@ -31,7 +40,9 @@ async function expectJsonResult<T>(result: JsonResult<T>): Promise<T> {
 
 // Lets the server attribute drive file operations to the sync loop instead
 // of generic API traffic (drive_file_operation actor: "sync" vs "api").
-const SYNC_CLIENT_HEADERS = { "x-wspc-client": "drive-sync" } as const
+function syncClientHeaders(clientId?: string): { "x-wspc-client": string } {
+  return { "x-wspc-client": clientId === undefined ? "drive-sync" : `drive-sync/${clientId}` }
+}
 
 function driveContentUrl(baseUrl: string, id: string): URL {
   const baseWithTrailingSlash = baseUrl.endsWith("/") ? baseUrl : `${baseUrl}/`
@@ -41,6 +52,7 @@ function driveContentUrl(baseUrl: string, id: string): URL {
 export async function createDriveApi(opts: DriveApiOptions = {}) {
   const client = await loadSdkClientWithAuthedFetch(opts)
   const rawClient = client._rawClient as never
+  const clientHeaders = syncClientHeaders(opts.clientId)
 
   return {
     async getLibrary(id: string): Promise<DriveLibrary> {
@@ -50,26 +62,53 @@ export async function createDriveApi(opts: DriveApiOptions = {}) {
       })
       return expectJsonResult(result)
     },
-    async getManifest(id: string, cursor?: string): Promise<DriveManifestResponse> {
+    async getManifest(
+      id: string,
+      cursor?: string,
+      sinceCursor?: string,
+    ): Promise<DriveManifestResponse & { latest_cursor?: string; resync_required?: boolean }> {
+      const query: Record<string, string> = {}
+      if (cursor !== undefined) query.cursor = cursor
+      // since_cursor / latest_cursor / resync_required exist server-side but are
+      // not in the generated SDK types yet; widen until the spec is regenerated.
+      if (sinceCursor !== undefined) query.since_cursor = sinceCursor
       const result = await driveManifestGet({
         client: rawClient,
         path: { id },
-        ...(cursor ? { query: { cursor } } : {}),
+        ...(Object.keys(query).length > 0 ? { query: query as never } : {}),
       })
       return expectJsonResult(result)
     },
     async deleteFile(id: string, path: string, expectedEntryVersion: number) {
       const result = await driveFileDelete({
         client: rawClient,
         path: { id },
-        headers: SYNC_CLIENT_HEADERS,
+        headers: clientHeaders,
         body: {
           path,
           expected_entry_version: expectedEntryVersion,
         },
       })
       return expectJsonResult<DeleteDriveFileResponse>(result)
     },
+    async moveFile(
+      id: string,
+      fromPath: string,
+      toPath: string,
+      expectedEntryVersion?: number,
+    ): Promise<MoveDriveFileResponse> {
+      const result = await driveFileMove({
+        client: rawClient,
+        path: { id },
+        headers: clientHeaders,
+        body: {
+          from_path: fromPath,
+          to_path: toPath,
+          ...(expectedEntryVersion === undefined ? {} : { expected_entry_version: expectedEntryVersion }),
+        },
+      })
+      return expectJsonResult<MoveDriveFileResponse>(result)
+    },
     async uploadFile(
       id: string,
       path: string,
@@ -88,7 +127,7 @@ export async function createDriveApi(opts: DriveApiOptions = {}) {
         headers: {
           "content-type": "application/octet-stream",
           "x-drive-content-sha256": sha256,
-          ...SYNC_CLIENT_HEADERS,
+          ...clientHeaders,
         },
         body,
       })

src/handwritten/commands/drive/decision.ts

@@ -41,7 +41,7 @@ export function decideDriveAction(
 
   if (!remote) return decideRemoteMissing(localStatus)
   if (!local) return decideLocalMissing(entry, remoteStatus)
-  return decideBothPresent(entry, localStatus, remoteStatus)
+  return decideBothPresent(entry, localStatus, remoteStatus, local, remote)
 }
 
 function decideWithoutBase(local: DecisionLocal | undefined, remote: DecisionRemote | undefined): DriveAction {
@@ -75,23 +75,29 @@ function decideBothPresent(
   entry: DecisionEntry,
   localStatus: LocalStatus,
   remoteStatus: RemoteStatus,
+  local: DecisionLocal,
+  remote: DecisionRemote,
 ): DriveAction {
+  // Identical bytes on both sides means the conflict is already converged,
+  // whatever the base says; only the state entry needs to catch up.
+  const converged = remote.content_sha256 !== undefined && local.sha256 === remote.content_sha256
+
   if (localStatus === "unchanged" && remoteStatus === "content_same_new_version") {
     return { type: "state_only" }
   }
   if (remoteStatus === "missing_version") {
     return { type: "conflict", reason: "remote_missing_entry_version" }
   }
   if (localStatus === "unknown" && remoteStatus !== "unchanged") {
-    return { type: "conflict", reason: "unknown_local_base_remote_changed" }
+    return converged ? { type: "state_only" } : { type: "conflict", reason: "unknown_local_base_remote_changed" }
   }
   if (localStatus === "unknown") return { type: "conflict", reason: "unknown_local_base" }
   if (localStatus === "unchanged" && remoteStatus === "changed") return { type: "download" }
   if (localStatus === "changed" && remoteStatus === "unchanged") {
     return { type: "upload_update", expectedEntryVersion: entry.entry_version }
   }
   if (localStatus !== "unchanged" && remoteStatus !== "unchanged") {
-    return { type: "conflict", reason: "local_and_remote_changed" }
+    return converged ? { type: "state_only" } : { type: "conflict", reason: "local_and_remote_changed" }
   }
 
   return { type: "unchanged" }

src/handwritten/commands/drive/merge.ts

@@ -72,8 +72,8 @@ export function mergeText3(base: string, local: string, remote: string): MergeTe
 
 export function conflictCopyPath(path: string, side: ConflictSide, timestamp: string, versionId: string): string {
   const parsed = pathPosix.parse(path)
-  const shortVersionId = safeShortVersionId(versionId)
-  const fileName = `${parsed.name}.${side}-conflict-${timestamp}-${shortVersionId}${parsed.ext}`
+  // Full version id: truncated ids collide across conflicts on the same file.
+  const fileName = `${parsed.name}.${side}-conflict-${timestamp}-${safeVersionId(versionId)}${parsed.ext}`
 
   if (parsed.dir === "") {
     return fileName
@@ -107,7 +107,7 @@ function normalizeLines(text: string): string[] {
   return text.replace(/\r\n/g, "\n").replace(/\r/g, "\n").split("\n")
 }
 
-function safeShortVersionId(versionId: string): string {
+function safeVersionId(versionId: string): string {
   const safeVersionId = versionId.replace(/[^A-Za-z0-9_-]/g, "_")
-  return (safeVersionId.length > 0 ? safeVersionId : "unknown").slice(0, 8)
+  return safeVersionId.length > 0 ? safeVersionId : "unknown"
 }

src/handwritten/commands/drive/realtime.ts

@@ -4,7 +4,7 @@ import type { DriveRealtimeSource } from "./watch.js"
 
 export type DriveRealtimeMessage =
   | { type: "ready"; cursor?: string; replayed: number }
-  | { type: "library_changed"; cursor?: string; path?: string }
+  | { type: "library_changed"; cursor?: string; path?: string; origin_client_id?: string }
   | { type: "resync_required"; cursor?: string; reason?: string }
   | { type: "error"; code?: string; message?: string }
   | { type: "unknown"; message_type?: string }
@@ -138,12 +138,17 @@ export function createDriveRealtimeSource(args: {
       return
     }
     if (message.type === "library_changed") {
-      handlers?.onEvent({
-        debounce_ms: 2000,
-        reason: "library_changed",
-        ...(message.cursor === undefined ? {} : { cursor: message.cursor }),
-        ...(message.path === undefined ? {} : { path: message.path }),
-      })
+      // Own echo: this client caused the change, so a sync would be a no-op;
+      // only the cursor needs to advance.
+      const ownEcho = message.origin_client_id !== undefined && message.origin_client_id === currentRealtime.client_id
+      if (!ownEcho) {
+        handlers?.onEvent({
+          debounce_ms: 2000,
+          reason: "library_changed",
+          ...(message.cursor === undefined ? {} : { cursor: message.cursor }),
+          ...(message.path === undefined ? {} : { path: message.path }),
+        })
+      }
       if (message.cursor !== undefined) {
         await persistBestEffort({ ...currentRealtime, last_cursor: message.cursor, last_event_at: driveIsoTimestamp(clock) })
       }
@@ -235,6 +240,7 @@ export function parseDriveRealtimeMessage(raw: string): DriveRealtimeMessage {
       type: "library_changed",
       ...(cursor === undefined ? {} : { cursor }),
       ...(typeof value.path === "string" ? { path: value.path } : {}),
+      ...(typeof value.origin_client_id === "string" ? { origin_client_id: value.origin_client_id } : {}),
     }
   }
   if (messageType === "resync_required") {