diff --git a/src/handwritten/commands/drive/sync.ts b/src/handwritten/commands/drive/sync.ts index 90d97c0..7f35e4c 100644 --- a/src/handwritten/commands/drive/sync.ts +++ b/src/handwritten/commands/drive/sync.ts @@ -75,7 +75,21 @@ function emptySummary(): DriveSyncSummary { } } -export async function runDriveSyncOnce(root: string, api?: DriveSyncApi, clock: DriveClock = systemDriveClock): Promise { +export type DriveSyncProgress = (processed: number, total: number) => void + +// Progress counts actionable paths only (transfers and conflict handling); +// counting unchanged/state-only paths would make incremental syncs jump to +// ~100% instantly and stall there. +function isActionableAction(action: DriveAction): boolean { + return action.type !== "unchanged" && action.type !== "state_only" && action.type !== "remove_state" +} + +export async function runDriveSyncOnce( + root: string, + api?: DriveSyncApi, + clock: DriveClock = systemDriveClock, + onProgress?: DriveSyncProgress, +): Promise { return withDriveLock(root, async () => { let state = await readDriveState(root) const syncApi = api ?? (await createDriveApi()) @@ -93,11 +107,23 @@ export async function runDriveSyncOnce(root: string, api?: DriveSyncApi, clock: .filter((path) => !blockedPaths.has(path)) .sort((left, right) => left.localeCompare(right)) + // decideDriveAction is pure and reads only this path's slices of the + // initial state, so this pre-pass total matches the loop's actions. + const total = paths.filter((path) => + isActionableAction(decideDriveAction(state.entries[path], localFiles[path], remoteFiles[path])), + ).length + let processed = 0 + onProgress?.(processed, total) + for (const path of paths) { const remote = remoteFiles[path] const action = decideDriveAction(state.entries[path], localFiles[path], remote) const result = await processPath({ root, state, api: syncApi, path, action, remote, local: localFiles[path], summary, clock }) state = result.state + if (isActionableAction(action)) { + processed += 1 + onProgress?.(processed, total) + } if (result.stop) break } diff --git a/src/handwritten/commands/drive/watch.ts b/src/handwritten/commands/drive/watch.ts index a096a07..80c6afd 100644 --- a/src/handwritten/commands/drive/watch.ts +++ b/src/handwritten/commands/drive/watch.ts @@ -6,7 +6,7 @@ import { loadRealtimeAuthHeaders } from "../../auth/load-sdk-client.js" import { render, shouldOutputJson } from "../../output/render.js" import { createDriveRealtimeSource } from "./realtime.js" import { DRIVE_DIR, ensureDriveRealtimeState, readDriveState, writeDriveRealtimeState } from "./state.js" -import { runDriveSyncOnce, type DriveSyncSummary } from "./sync.js" +import { runDriveSyncOnce, type DriveSyncProgress, type DriveSyncSummary } from "./sync.js" export interface DriveWatchSource { onChange(handler: (path: string) => void): void @@ -34,7 +34,7 @@ export interface DriveWatchOptions { source?: DriveWatchSource realtimeSource?: DriveRealtimeSource readState?: typeof readDriveState - runSync?: (root: string) => Promise + runSync?: (root: string, onProgress?: DriveSyncProgress) => Promise once?: boolean debounceMs?: number remoteDebounceMs?: number @@ -42,7 +42,9 @@ export interface DriveWatchOptions { } export async function runDriveWatch(root: string, options: DriveWatchOptions = {}): Promise { - const runSync = options.runSync ?? runDriveSyncOnce + const runSync = + options.runSync ?? + ((syncRoot: string, onProgress?: DriveSyncProgress) => runDriveSyncOnce(syncRoot, undefined, undefined, onProgress)) const debounceMs = options.debounceMs ?? 500 const remoteDebounceMs = options.remoteDebounceMs ?? 2000 // Watch is a long-lived event stream: json mode must emit NDJSON (one event @@ -83,7 +85,9 @@ export async function runDriveWatch(root: string, options: DriveWatchOptions = { // A large first upload can run for minutes; without this the app // shows a stale badge and users assume sync never started. emit({ kind: "drive_sync_start" }) - const summary = await runSync(root) + const summary = await runSync(root, (processed, total) => { + emit({ kind: "drive_sync_progress", processed, total }) + }) emit({ kind: "drive_sync_once", ...summary }) backoffMs = 1000 } catch (error) { diff --git a/test/handwritten/drive/sync.test.ts b/test/handwritten/drive/sync.test.ts index 6343b46..52d4fe9 100644 --- a/test/handwritten/drive/sync.test.ts +++ b/test/handwritten/drive/sync.test.ts @@ -201,6 +201,32 @@ describe("drive sync once", () => { }) }) + it("reports progress over actionable paths only, excluding unchanged", async () => { + const root = await mkdtemp(join(tmpdir(), "wspc-drive-sync-progress-")) + const state = await initDriveState(root, "lib_1") + state.entries["same.txt"] = stateEntry("same.txt", "same", 2) + await writeDriveState(root, state) + await writeFile(join(root, "same.txt"), "same") + await writeFile(join(root, "new.txt"), "hello") + const remote = entry("remote.txt", "remote", 3) + const api = mkApi([{ entries: [entry("same.txt", "same", 2), remote] }]) + api.downloads.set("remote.txt", "remote") + const progress: Array<[number, number]> = [] + + const result = await runDriveSyncOnce(root, api, undefined, (processed, total) => { + progress.push([processed, total]) + }) + + expect(result.uploaded).toBe(1) + expect(result.downloaded).toBe(1) + expect(result.unchanged).toBe(1) + expect(progress).toEqual([ + [0, 2], + [1, 2], + [2, 2], + ]) + }) + it("includes merged count and conflict copy metadata in sync summaries", async () => { const root = await mkdtemp(join(tmpdir(), "wspc-drive-sync-summary-shape-")) await initDriveState(root, "lib_1") diff --git a/test/handwritten/drive/watch.test.ts b/test/handwritten/drive/watch.test.ts index 1dd1fd9..f0c4910 100644 --- a/test/handwritten/drive/watch.test.ts +++ b/test/handwritten/drive/watch.test.ts @@ -127,7 +127,33 @@ describe("drive watch", () => { await runDriveWatch("/tmp/root", { source, runSync, readState, onEvent, once: true }) expect(runSync).toHaveBeenCalledTimes(1) - expect(runSync).toHaveBeenCalledWith("/tmp/root") + expect(runSync).toHaveBeenCalledWith("/tmp/root", expect.any(Function)) + }) + + it("emits drive_sync_progress events from the sync progress callback", async () => { + const source = fakeSource() + const events: Array<{ kind?: string; processed?: number; total?: number }> = [] + const runSync = vi.fn(async (_root: string, onProgress?: (processed: number, total: number) => void) => { + onProgress?.(0, 2) + onProgress?.(1, 2) + onProgress?.(2, 2) + return syncSummary() + }) + + await runDriveWatch("/tmp/root", { + source, + runSync, + readState, + onEvent: (event) => events.push(event as { kind?: string; processed?: number; total?: number }), + once: true, + }) + + const progressEvents = events.filter((event) => event.kind === "drive_sync_progress") + expect(progressEvents).toEqual([ + { kind: "drive_sync_progress", processed: 0, total: 2 }, + { kind: "drive_sync_progress", processed: 1, total: 2 }, + { kind: "drive_sync_progress", processed: 2, total: 2 }, + ]) }) it("emits drive_sync_start before each sync round", async () => {