feat(ocm-cli): add granular push progress with phase reporting and upload tracking

Author: chriswritescode-dev

ocm-cli/src/manager-api.ts

@@ -52,6 +52,17 @@ export interface MirrorBundleResult {
   created: false
 }
 
+function createByteCounter(onProgress: (bytesSent: number) => void): TransformStream<Uint8Array, Uint8Array> {
+  let bytesSent = 0
+  return new TransformStream<Uint8Array, Uint8Array>({
+    transform(chunk, controller) {
+      bytesSent += chunk.byteLength
+      onProgress(bytesSent)
+      controller.enqueue(chunk)
+    },
+  })
+}
+
 export class ManagerApiError extends Error {
   constructor(
     message: string,
@@ -161,11 +172,16 @@ export class ManagerApi {
     return (await res.json()) as MirrorPatchResult
   }
 
-  async mirrorUploadBundle(repoId: number, bundlePath: string, opts: { branch: string | null; force?: boolean }): Promise<MirrorBundleResult> {
+  async mirrorUploadBundle(
+    repoId: number,
+    bundlePath: string,
+    opts: { branch: string | null; force?: boolean; onProgress?: (bytesSent: number) => void },
+  ): Promise<MirrorBundleResult> {
     const query = opts.force === true ? '?force=1' : ''
     const headers: Record<string, string> = { ...this.headers(), 'Content-Type': 'application/octet-stream' }
     if (opts.branch) headers['X-OCM-Branch'] = opts.branch
-    const body = Readable.toWeb(createReadStream(bundlePath)) as unknown as ReadableStream<Uint8Array>
+    const fileStream = Readable.toWeb(createReadStream(bundlePath)) as unknown as ReadableStream<Uint8Array>
+    const body = opts.onProgress ? fileStream.pipeThrough(createByteCounter(opts.onProgress)) : fileStream
     const res = await fetch(`${this.baseUrl}/api/internal/repos/${repoId}/mirror/bundle${query}`, {
       method: 'POST',
       headers,

ocm-cli/src/mirror.ts

@@ -379,7 +379,17 @@ function applyPatch(repoRoot: string, patch: string): void {
 
 async function createLocalBundle(repoRoot: string): Promise<string> {
   const bundlePath = join(tmpdir(), `ocm-bundle-${Date.now()}-${Math.random().toString(36).slice(2)}.bundle`)
-  runGit(repoRoot, ['bundle', 'create', bundlePath, '--all'])
+  await new Promise<void>((resolve, reject) => {
+    const child = spawn('git', ['bundle', 'create', bundlePath, '--all'], { cwd: repoRoot })
+    const stderrChunks: Buffer[] = []
+    child.stderr.on('data', (chunk: Buffer) => stderrChunks.push(chunk))
+    child.on('error', reject)
+    child.on('close', (code) => {
+      if (code === 0) return resolve()
+      const stderr = Buffer.concat(stderrChunks).toString('utf-8').trim()
+      reject(new Error(`git bundle create failed${stderr ? `: ${stderr}` : ''}`))
+    })
+  })
   return bundlePath
 }
 
@@ -428,14 +438,34 @@ async function writeBundleStream(repoId: number, api: ManagerApi): Promise<strin
   return bundlePath
 }
 
+export type MirrorUpFastPhase =
+  | { kind: 'bundling' }
+  | { kind: 'uploading'; bytesSent: number; totalBytes: number }
+  | { kind: 'processing' }
+  | { kind: 'patching' }
+
 export async function mirrorUpFast(
   plan: MirrorPlan,
-  opts: Pick<MirrorUpOpts, 'api' | 'force'>,
+  opts: Pick<MirrorUpOpts, 'api' | 'force'> & { onPhase?: (phase: MirrorUpFastPhase) => void },
 ): Promise<{ repoId: number; branch: string | null; head: string | null; created: false }> {
   const repoId = plan.matched[0]!.repoId
+  const onPhase = opts.onPhase
+  onPhase?.({ kind: 'bundling' })
   const bundlePath = await createLocalBundle(plan.repoRoot)
   try {
-    await opts.api.mirrorUploadBundle(repoId, bundlePath, { branch: getBranchName(plan.repoRoot), force: opts.force })
+    const { size } = await fsp.stat(bundlePath)
+    onPhase?.({ kind: 'uploading', bytesSent: 0, totalBytes: size })
+    await opts.api.mirrorUploadBundle(repoId, bundlePath, {
+      branch: getBranchName(plan.repoRoot),
+      force: opts.force,
+      onProgress: onPhase
+        ? (bytesSent) => {
+            onPhase({ kind: 'uploading', bytesSent, totalBytes: size })
+            if (bytesSent >= size) onPhase({ kind: 'processing' })
+          }
+        : undefined,
+    })
+    onPhase?.({ kind: 'patching' })
     const patchResult = await mirrorUpPatch(plan, opts)
     return { repoId: patchResult.repoId, branch: patchResult.branch, head: patchResult.head, created: false }
   } finally {

ocm-cli/src/tui-plugin.ts

@@ -4,7 +4,8 @@ import { getToken } from './keychain.js'
 import { fetchRepos, toRemoteRepoSummaries } from './manager-repos.js'
 import { ManagerApi, ManagerApiError } from './manager-api.js'
 import { prepareMirror, checkPushDivergence, mirrorUpFast } from './mirror.js'
-import type { MirrorPlan } from './mirror.js'
+import type { MirrorPlan, MirrorUpFastPhase } from './mirror.js'
+import { formatBytes } from './progress.js'
 import { transferSession, moveReminderText } from './session-move.js'
 import { createManagerReplay, createManagerPromptAsync } from './remote-replay.js'
 import { readSessionEvents } from './local-history.js'
@@ -43,6 +44,31 @@ function showInstallNotice(api: TuiPluginApi): void {
   })
 }
 
+function pushPhaseMessage(phase: MirrorUpFastPhase): string {
+  switch (phase.kind) {
+    case 'bundling':
+      return 'Pushing repo state: creating git bundle…'
+    case 'uploading':
+      return `Pushing repo state: uploading ${formatBytes(phase.bytesSent)} / ${formatBytes(phase.totalBytes)}…`
+    case 'processing':
+      return 'Pushing repo state: waiting for server to import bundle…'
+    case 'patching':
+      return 'Pushing repo state: applying local changes…'
+  }
+}
+
+function createPushPhaseToaster(api: TuiPluginApi): (phase: MirrorUpFastPhase) => void {
+  let lastUploadToastAt = 0
+  return (phase) => {
+    if (phase.kind === 'uploading') {
+      const now = Date.now()
+      if (now - lastUploadToastAt < 1000) return
+      lastUploadToastAt = now
+    }
+    api.ui.toast({ message: pushPhaseMessage(phase) })
+  }
+}
+
 async function runSessionMove(api: TuiPluginApi): Promise<void> {
   try {
     const current = api.route.current
@@ -103,9 +129,12 @@ async function runSessionMove(api: TuiPluginApi): Promise<void> {
       if (!(error instanceof ManagerApiError && error.status === 404)) throw error
     }
 
-    api.ui.toast({ message: 'Pushing repo state…' })
     const selectedPlan: MirrorPlan = { ...plan, matched: [matched] }
-    await mirrorUpFast(selectedPlan, { api: managerApi, force: false })
+    await mirrorUpFast(selectedPlan, {
+      api: managerApi,
+      force: false,
+      onPhase: createPushPhaseToaster(api),
+    })
 
     const result = await transferSession(
       { sessionID, localRoot: plan.repoRoot, remoteDirectory },

ocm-cli/test/mirror.test.ts

@@ -1,10 +1,10 @@
 import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'
-import { mkdtempSync, mkdirSync, rmSync, writeFileSync, existsSync, readdirSync, readFileSync } from 'fs'
+import { mkdtempSync, mkdirSync, rmSync, writeFileSync, existsSync, readdirSync, readFileSync, statSync } from 'fs'
 import { join } from 'path'
 import { tmpdir } from 'os'
 import { randomBytes } from 'crypto'
 import { spawnSync, execSync } from 'child_process'
-import { prepareMirror, MirrorAbort, mirrorDown, mirrorUp, mirrorUpPatch, mirrorUpFast, checkPushDivergence, checkPullDivergence } from '../src/mirror'
+import { prepareMirror, MirrorAbort, mirrorDown, mirrorUp, mirrorUpPatch, mirrorUpFast, checkPushDivergence, checkPullDivergence, type MirrorUpFastPhase } from '../src/mirror'
 import { getBranchName } from '../src/local-repo'
 import { gitRemoteProjectId } from '@opencode-manager/shared/project-id'
 
@@ -818,6 +818,73 @@ describe('mirrorUpFast targets the selected repo', () => {
     expect(result.repoId).toBe(99)
   })
 
+  it('reports bundling, uploading, and patching phases in order', async () => {
+    const repoRoot = join(tmpDir, 'repo-phases')
+    mkdirSync(repoRoot)
+    spawnSync('git', ['init'], { cwd: repoRoot, stdio: 'ignore' })
+    spawnSync('git', ['config', 'user.email', 'test@test.com'], { cwd: repoRoot, stdio: 'ignore' })
+    spawnSync('git', ['config', 'user.name', 'Test'], { cwd: repoRoot, stdio: 'ignore' })
+    writeFileSync(join(repoRoot, 'a.txt'), 'a\n')
+    spawnSync('git', ['add', '.'], { cwd: repoRoot, stdio: 'ignore' })
+    spawnSync('git', ['commit', '-m', 'init'], { cwd: repoRoot, stdio: 'ignore' })
+
+    const api = {
+      mirrorUploadBundle: vi.fn().mockResolvedValue(undefined),
+      mirrorPatch: vi.fn().mockResolvedValue({ repoId: 1, fullPath: '/tmp/x', branch: 'main', head: 'abc', created: false, applied: true }),
+    }
+
+    const plan = {
+      repoRoot,
+      localProjectId: 'proj',
+      matched: [{ repoId: 1, name: 'repo-A', projectId: 'proj', branch: 'main' }],
+    }
+
+    const phases: MirrorUpFastPhase[] = []
+    await mirrorUpFast(plan, { api: api as any, force: false, onPhase: (p) => phases.push(p) })
+
+    expect(phases.map((p) => p.kind)).toEqual(['bundling', 'uploading', 'patching'])
+    const uploading = phases[1] as Extract<MirrorUpFastPhase, { kind: 'uploading' }>
+    expect(uploading.bytesSent).toBe(0)
+    expect(uploading.totalBytes).toBeGreaterThan(0)
+  })
+
+  it('emits cumulative upload progress and a processing phase when the api reports sent bytes', async () => {
+    const repoRoot = join(tmpDir, 'repo-upload-progress')
+    mkdirSync(repoRoot)
+    spawnSync('git', ['init'], { cwd: repoRoot, stdio: 'ignore' })
+    spawnSync('git', ['config', 'user.email', 'test@test.com'], { cwd: repoRoot, stdio: 'ignore' })
+    spawnSync('git', ['config', 'user.name', 'Test'], { cwd: repoRoot, stdio: 'ignore' })
+    writeFileSync(join(repoRoot, 'a.txt'), 'a\n')
+    spawnSync('git', ['add', '.'], { cwd: repoRoot, stdio: 'ignore' })
+    spawnSync('git', ['commit', '-m', 'init'], { cwd: repoRoot, stdio: 'ignore' })
+
+    const api = {
+      mirrorUploadBundle: vi.fn().mockImplementation(
+        async (_repoId: number, bundlePath: string, opts: { onProgress?: (bytesSent: number) => void }) => {
+          const { size } = statSync(bundlePath)
+          opts.onProgress?.(Math.floor(size / 2))
+          opts.onProgress?.(size)
+        },
+      ),
+      mirrorPatch: vi.fn().mockResolvedValue({ repoId: 1, fullPath: '/tmp/x', branch: 'main', head: 'abc', created: false, applied: true }),
+    }
+
+    const plan = {
+      repoRoot,
+      localProjectId: 'proj',
+      matched: [{ repoId: 1, name: 'repo-A', projectId: 'proj', branch: 'main' }],
+    }
+
+    const phases: MirrorUpFastPhase[] = []
+    await mirrorUpFast(plan, { api: api as any, force: false, onPhase: (p) => phases.push(p) })
+
+    expect(phases.map((p) => p.kind)).toEqual(['bundling', 'uploading', 'uploading', 'uploading', 'processing', 'patching'])
+    const sent = phases.filter((p): p is Extract<MirrorUpFastPhase, { kind: 'uploading' }> => p.kind === 'uploading').map((p) => p.bytesSent)
+    expect(sent[0]).toBe(0)
+    expect(sent[1]!).toBeGreaterThan(0)
+    expect(sent[2]!).toBeGreaterThan(sent[1]!)
+  })
+
   it('narrows a multi-match plan so bundle goes to the chosen repo', async () => {
     const repoRoot = join(tmpDir, 'repo-narrow')
     mkdirSync(repoRoot)