feat(transfers): progress and cancel for SSH uploads and all remote downloads

src/main/ipc/filesystem-mutations-runtime-upload.test.ts

@@ -38,8 +38,18 @@ const request = {
   expectedEnvironmentRuntimeId: 'rt-1'
 }
 
-function fakeSender(): EventEmitter {
-  return new EventEmitter()
+let nextSenderId = 0
+
+function fakeSender(): EventEmitter & {
+  id: number
+  isDestroyed: () => boolean
+  send: ReturnType<typeof vi.fn>
+} {
+  return Object.assign(new EventEmitter(), {
+    id: ++nextSenderId,
+    isDestroyed: () => false,
+    send: vi.fn()
+  })
 }
 
 function listenerCount(sender: EventEmitter): number {
@@ -147,6 +157,29 @@ describe('fs:uploadExternalFileToRuntime', () => {
     expect(sweepMock).toHaveBeenCalledTimes(1)
   })
 
+  it("does not let another renderer cancel this renderer's upload id", async () => {
+    const owner = fakeSender()
+    const other = fakeSender()
+    streamMock.mockImplementation(async ({ cancelSignal }: { cancelSignal: AbortSignal }) => {
+      await handlers.get('fs:cancelRuntimeUpload')!({ sender: other }, { uploadId: 'same-id' })
+      expect(cancelSignal.aborted).toBe(false)
+
+      await handlers.get('fs:cancelRuntimeUpload')!({ sender: owner }, { uploadId: 'same-id' })
+      expect(cancelSignal.aborted).toBe(true)
+      await handlers.get('fs:releaseRuntimeUpload')!({ sender: owner }, { uploadId: 'same-id' })
+      return { byteLength: 0 }
+    })
+
+    await expect(
+      handlers.get('fs:uploadExternalFileToRuntime')!(
+        { sender: owner },
+        { ...request, uploadId: 'same-id' }
+      )
+    ).resolves.toEqual({ byteLength: 0 })
+    owner.emit('destroyed')
+    other.emit('destroyed')
+  })
+
   it('does not sweep when the stream fails while the renderer is still alive', async () => {
     const sender = fakeSender()
     streamMock.mockRejectedValue(new Error("File changed since it was staged: 'file.bin'"))

src/main/ipc/filesystem-mutations.ts

@@ -30,6 +30,12 @@ import {
 import { streamExternalFileToRuntime } from './runtime-upload-file-stream'
 import { abortWhenRendererGone } from './renderer-lifetime-abort'
 import { sweepAbandonedRuntimeUploadTempPath } from './runtime-upload-temp-sweep'
+import {
+  RUNTIME_UPLOAD_PROGRESS_CHANNEL,
+  throttleRuntimeUploadProgress
+} from './runtime-upload-progress'
+import { registerRuntimeUploadCancelHandlers } from './runtime-upload-cancel-ipc'
+import { registerCancellableUpload, scopeRuntimeUploadId } from './runtime-upload-cancellation'
 import type { RuntimeUploadFileStreamRequest } from '../../shared/runtime-upload-staging-contract'
 import { resolveEnvironment } from '../../shared/runtime-environment-store'
 
@@ -255,11 +261,30 @@ export function registerFilesystemMutationHandlers(store: Store): void {
       // move in main, a reload or close has to stop the transfer explicitly,
       // or a multi-GB upload outlives the window that asked for it.
       const lifetime = abortWhenRendererGone(event.sender)
+      const uploadId = args.uploadId
+      // Why: replies to the frame that asked, so a second window's drop cannot
+      // move this one's progress bar.
+      const emit = uploadId
+        ? throttleRuntimeUploadProgress((progress) => {
+            if (!event.sender.isDestroyed()) {
+              event.sender.send(RUNTIME_UPLOAD_PROGRESS_CHANNEL, progress)
+            }
+          })
+        : null
+      const cancellation = uploadId
+        ? registerCancellableUpload(scopeRuntimeUploadId(event.sender.id, uploadId))
+        : null
       try {
         return await streamExternalFileToRuntime({
           ...request,
           userDataPath,
-          signal: lifetime.signal
+          signal: lifetime.signal,
+          cancelSignal: cancellation?.signal,
+          onProgress:
+            emit && uploadId
+              ? ({ sentBytes, totalBytes }) =>
+                  emit({ uploadId, sentBytes, totalBytes, fileSequence: args.fileSequence })
+              : undefined
         })
       } catch (error) {
         if (lifetime.signal.aborted) {
@@ -269,11 +294,14 @@ export function registerFilesystemMutationHandlers(store: Store): void {
         }
         throw error
       } finally {
+        cancellation?.release()
         lifetime.dispose()
       }
     }
   )
 
+  registerRuntimeUploadCancelHandlers()
+
   // Why: terminal drag-and-drop resolver. Local worktrees pass paths through
   // unchanged (reference-in-place; preserves zero-latency drop). SSH worktrees
   // upload each path into `${worktreePath}/.orca/drops/` and return remote

src/main/ipc/runtime-upload-cancel-ipc.test.ts

@@ -0,0 +1,83 @@
+import { EventEmitter } from 'node:events'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+const handlers = new Map<
+  string,
+  (event: { sender: EventEmitter & { id: number } }, args: unknown) => unknown
+>()
+
+vi.mock('electron', () => ({
+  ipcMain: {
+    handle: (
+      channel: string,
+      handler: (event: { sender: EventEmitter & { id: number } }, args: unknown) => unknown
+    ) => handlers.set(channel, handler)
+  }
+}))
+
+import { registerRuntimeUploadCancelHandlers } from './runtime-upload-cancel-ipc'
+import {
+  forgetRuntimeUploadCancellation,
+  isUploadCancelled,
+  scopeRuntimeUploadId
+} from './runtime-upload-cancellation'
+
+registerRuntimeUploadCancelHandlers()
+
+function renderer(id: number): EventEmitter & { id: number } {
+  return Object.assign(new EventEmitter(), { id })
+}
+
+function invoke(channel: string, sender: EventEmitter & { id: number }, args: unknown): unknown {
+  return handlers.get(channel)!({ sender }, args)
+}
+
+afterEach(() => {
+  for (const senderId of [1, 2, 3]) {
+    for (const uploadId of ['u', 'a', 'b']) {
+      forgetRuntimeUploadCancellation(scopeRuntimeUploadId(senderId, uploadId))
+    }
+  }
+})
+
+describe('runtime upload cancel IPC', () => {
+  it('records and releases a cancel only for the renderer that sent it', () => {
+    const first = renderer(1)
+    invoke('fs:cancelRuntimeUpload', first, { uploadId: 'u' })
+
+    expect(isUploadCancelled(scopeRuntimeUploadId(1, 'u'))).toBe(true)
+    expect(isUploadCancelled(scopeRuntimeUploadId(2, 'u'))).toBe(false)
+
+    invoke('fs:releaseRuntimeUpload', first, { uploadId: 'u' })
+    expect(isUploadCancelled(scopeRuntimeUploadId(1, 'u'))).toBe(false)
+  })
+
+  it('rejects malformed upload ids without changing cancellation state', () => {
+    const sender = renderer(1)
+    for (const args of [
+      null,
+      undefined,
+      {},
+      { uploadId: '' },
+      { uploadId: '  ' },
+      { uploadId: 3 }
+    ]) {
+      expect(() => invoke('fs:cancelRuntimeUpload', sender, args)).not.toThrow()
+      expect(() => invoke('fs:releaseRuntimeUpload', sender, args)).not.toThrow()
+    }
+
+    expect(isUploadCancelled(scopeRuntimeUploadId(1, ''))).toBe(false)
+    expect(isUploadCancelled(scopeRuntimeUploadId(1, '  '))).toBe(false)
+  })
+
+  it('drops cancels when the renderer goes away and re-arms after reload', () => {
+    const sender = renderer(3)
+    invoke('fs:cancelRuntimeUpload', sender, { uploadId: 'a' })
+    sender.emit('did-navigate')
+    expect(isUploadCancelled(scopeRuntimeUploadId(3, 'a'))).toBe(false)
+
+    invoke('fs:cancelRuntimeUpload', sender, { uploadId: 'b' })
+    sender.emit('render-process-gone')
+    expect(isUploadCancelled(scopeRuntimeUploadId(3, 'b'))).toBe(false)
+  })
+})

src/main/ipc/runtime-upload-cancel-ipc.ts

@@ -0,0 +1,58 @@
+import { ipcMain, type WebContents } from 'electron'
+import { abortWhenRendererGone } from './renderer-lifetime-abort'
+import {
+  cancelRuntimeUpload,
+  forgetRuntimeUploadCancellation,
+  forgetRuntimeUploadCancellationsForSender,
+  scopeRuntimeUploadId
+} from './runtime-upload-cancellation'
+
+type CancelSender = Pick<WebContents, 'id' | 'once' | 'removeListener'>
+
+const sendersWithCleanup = new WeakSet<CancelSender>()
+
+function readUploadId(args: unknown): string | undefined {
+  if (!args || typeof args !== 'object' || !('uploadId' in args)) {
+    return undefined
+  }
+  return typeof args.uploadId === 'string' && args.uploadId.trim() !== ''
+    ? args.uploadId
+    : undefined
+}
+
+/** A reloaded or gone renderer can no longer release its remembered cancels. */
+function forgetCancelsWhenRendererGoes(sender: CancelSender): void {
+  if (sendersWithCleanup.has(sender)) {
+    return
+  }
+  sendersWithCleanup.add(sender)
+  const senderId = sender.id
+  const lifetime = abortWhenRendererGone(sender)
+  lifetime.signal.addEventListener(
+    'abort',
+    () => {
+      lifetime.dispose()
+      forgetRuntimeUploadCancellationsForSender(senderId)
+      // A reload reuses WebContents, so its next cancel must arm cleanup again.
+      sendersWithCleanup.delete(sender)
+    },
+    { once: true }
+  )
+}
+
+export function registerRuntimeUploadCancelHandlers(): void {
+  ipcMain.handle('fs:cancelRuntimeUpload', (event, args: unknown): void => {
+    const uploadId = readUploadId(args)
+    if (uploadId) {
+      forgetCancelsWhenRendererGoes(event.sender)
+      cancelRuntimeUpload(scopeRuntimeUploadId(event.sender.id, uploadId))
+    }
+  })
+
+  ipcMain.handle('fs:releaseRuntimeUpload', (event, args: unknown): void => {
+    const uploadId = readUploadId(args)
+    if (uploadId) {
+      forgetRuntimeUploadCancellation(scopeRuntimeUploadId(event.sender.id, uploadId))
+    }
+  })
+}

src/main/ipc/runtime-upload-cancellation.test.ts

@@ -0,0 +1,112 @@
+import { afterEach, describe, expect, it } from 'vitest'
+import {
+  cancelRuntimeUpload,
+  forgetRuntimeUploadCancellation,
+  forgetRuntimeUploadCancellationsForSender,
+  isRuntimeUploadCancelled,
+  isUploadCancelled,
+  registerCancellableUpload,
+  RuntimeUploadCancelledError,
+  scopeRuntimeUploadId
+} from './runtime-upload-cancellation'
+
+afterEach(() => {
+  forgetRuntimeUploadCancellation('u')
+  forgetRuntimeUploadCancellation('v')
+})
+
+describe('runtime upload cancellation', () => {
+  it('aborts an upload that is already running', () => {
+    const upload = registerCancellableUpload('u')
+    expect(upload.signal.aborted).toBe(false)
+
+    cancelRuntimeUpload('u')
+
+    expect(upload.signal.aborted).toBe(true)
+    upload.release()
+  })
+
+  it('aborts the next file when cancel lands between two files of one drop', () => {
+    const first = registerCancellableUpload('u')
+    first.release()
+    cancelRuntimeUpload('u')
+
+    const second = registerCancellableUpload('u')
+
+    expect(second.signal.aborted).toBe(true)
+    second.release()
+  })
+
+  it('leaves other uploads running', () => {
+    const target = registerCancellableUpload('u')
+    const bystander = registerCancellableUpload('v')
+
+    cancelRuntimeUpload('u')
+
+    expect(target.signal.aborted).toBe(true)
+    expect(bystander.signal.aborted).toBe(false)
+    target.release()
+    bystander.release()
+  })
+
+  it('stops applying a cancel once the drop is forgotten', () => {
+    cancelRuntimeUpload('u')
+    forgetRuntimeUploadCancellation('u')
+
+    const fresh = registerCancellableUpload('u')
+    expect(isUploadCancelled('u')).toBe(false)
+    expect(fresh.signal.aborted).toBe(false)
+    fresh.release()
+  })
+
+  it('releasing a superseded registration does not evict the live one', () => {
+    const stale = registerCancellableUpload('u')
+    const live = registerCancellableUpload('u')
+    stale.release()
+
+    cancelRuntimeUpload('u')
+
+    expect(live.signal.aborted).toBe(true)
+    live.release()
+  })
+
+  it('keeps windows apart even when they mint the same id', () => {
+    expect(scopeRuntimeUploadId(1, 'u')).not.toBe(scopeRuntimeUploadId(2, 'u'))
+  })
+
+  it("forgets a gone renderer's remembered cancels and only that renderer's", () => {
+    cancelRuntimeUpload(scopeRuntimeUploadId(1, 'u'))
+    cancelRuntimeUpload(scopeRuntimeUploadId(2, 'u'))
+
+    forgetRuntimeUploadCancellationsForSender(1)
+
+    expect(isUploadCancelled(scopeRuntimeUploadId(1, 'u'))).toBe(false)
+    expect(isUploadCancelled(scopeRuntimeUploadId(2, 'u'))).toBe(true)
+    forgetRuntimeUploadCancellation(scopeRuntimeUploadId(2, 'u'))
+  })
+
+  it('keeps a live upload aborted if its drop is released mid-flight', () => {
+    const live = registerCancellableUpload('u')
+    cancelRuntimeUpload('u')
+    forgetRuntimeUploadCancellation('u')
+
+    expect(live.signal.aborted).toBe(true)
+    live.release()
+  })
+
+  it('keeps a live registration cancellable after its remembered cancel is released', () => {
+    const live = registerCancellableUpload('u')
+    forgetRuntimeUploadCancellation('u')
+
+    cancelRuntimeUpload('u')
+
+    expect(live.signal.aborted).toBe(true)
+    live.release()
+  })
+
+  it('recognises its own error and nothing else', () => {
+    expect(isRuntimeUploadCancelled(new RuntimeUploadCancelledError())).toBe(true)
+    expect(isRuntimeUploadCancelled(new Error('disk full'))).toBe(false)
+    expect(isRuntimeUploadCancelled('cancelled')).toBe(false)
+  })
+})