fix(downloader): clean local data only on task delete

cmd/internal/worker/worker.go

@@ -24,6 +24,7 @@ import (
 const Version = "0.1.0"
 const maxTaskErrorMessageLength = 1000
 const localResultRemovedRuntimeState = "local_result_removed"
+const deleteRequestedRuntimeState = "delete_requested"
 
 var errTaskPausing = errors.New("task pausing")
 var errTaskCanceling = errors.New("task canceling")
@@ -298,7 +299,6 @@ func (w *Worker) downloadThenUpload(
 				return
 			}
 			if errors.Is(context.Cause(ctx), errTaskCanceling) {
-				w.cleanupTerminalTask(context.WithoutCancel(ctx), log, task, "canceled")
 				if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{Status: "canceled"}); updateErr != nil {
 					log.Error("failed to mark task canceled", "error", updateErr)
 				}
@@ -308,7 +308,6 @@ func (w *Worker) downloadThenUpload(
 			if errors.Is(context.Cause(ctx), errTaskSuspended) {
 				// The server already moved the task to suspended (billing); the
 				// poll told us to stop. Don't touch its status.
-				w.cleanupSuspendedTask(context.WithoutCancel(ctx), log, task)
 				log.Info("task stopped because it was suspended")
 				return
 			}
@@ -329,7 +328,6 @@ func (w *Worker) downloadThenUpload(
 		if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{Status: "failed", ErrorMessage: &msg}); updateErr != nil {
 			log.Error("failed to mark task failed", "error", updateErr)
 		}
-		w.cleanupTerminalTask(context.WithoutCancel(ctx), log, task, "failed")
 		return
 	}
 
@@ -484,39 +482,23 @@ func (w *Worker) resetRuntimeTask(ctx context.Context, task client.DownloadTask,
 	return nil
 }
 
-func (w *Worker) cleanupTerminalTask(ctx context.Context, log *slog.Logger, task client.DownloadTask, reason string) bool {
+func (w *Worker) cleanupDeletedTask(ctx context.Context, log *slog.Logger, task client.DownloadTask) {
+	reason := "deleted"
 	w.cleanupRetainedSeedForTask(ctx, task.ID, reason)
 	if w.engine == nil {
 		log.Warn("downloader engine is unavailable for terminal cleanup", "reason", reason)
-		return false
+		return
 	}
 	resetter, ok := w.engine.(engine.TaskResetter)
 	if !ok {
 		log.Warn("downloader engine does not support terminal cleanup", "engine", w.engine.Name(), "reason", reason)
-		return false
+		return
 	}
 	if err := resetter.ResetTask(ctx, task); err != nil {
 		log.Warn("failed to clean terminal downloader task", "reason", reason, "error", err)
-		return false
+		return
 	}
 	log.Info("cleaned terminal downloader task", "reason", reason)
-	return true
-}
-
-func (w *Worker) cleanupSuspendedTask(ctx context.Context, log *slog.Logger, task client.DownloadTask) {
-	runtime := task.Runtime()
-	if runtime != nil && runtime.State == localResultRemovedRuntimeState {
-		log.Debug("suspended task cleanup already recorded")
-		return
-	}
-	if !w.cleanupTerminalTask(ctx, log, task, "suspended") {
-		return
-	}
-	if _, err := w.updateTask(ctx, task.ID, client.TaskPatch{
-		Runtime: localResultRemovedRuntime(runtime),
-	}); err != nil {
-		log.Error("failed to record suspended task cleanup", "error", err)
-	}
 }
 
 func (w *Worker) uploadAndComplete(
@@ -548,27 +530,13 @@ func (w *Worker) uploadAndComplete(
 				return
 			}
 			if errors.Is(context.Cause(ctx), errTaskCanceling) {
-				if cleanupErr := cleanupDownloadedResult(context.WithoutCancel(ctx), task, result); cleanupErr != nil {
-					log.Warn("failed to remove canceled local downloaded result", "path", result.Path, "error", cleanupErr)
-				}
-				if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{
-					Status:  "canceled",
-					Runtime: localResultRemovedRuntime(currentDetail),
-				}); updateErr != nil {
+				if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{Status: "canceled"}); updateErr != nil {
 					log.Error("failed to mark task canceled during upload", "error", updateErr)
 				}
 				log.Info("task upload canceled by control action")
 				return
 			}
 			if errors.Is(context.Cause(ctx), errTaskSuspended) {
-				if cleanupErr := cleanupDownloadedResult(context.WithoutCancel(ctx), task, result); cleanupErr != nil {
-					log.Warn("failed to remove suspended local downloaded result", "path", result.Path, "error", cleanupErr)
-				}
-				if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{
-					Runtime: localResultRemovedRuntime(currentDetail),
-				}); updateErr != nil {
-					log.Error("failed to record suspended upload cleanup", "error", updateErr)
-				}
 				log.Info("task upload stopped because it was suspended")
 				return
 			}
@@ -593,7 +561,6 @@ func (w *Worker) uploadAndComplete(
 		}
 		msg := taskErrorMessage(err)
 		log.Error("failed to upload result", "error", err)
-		failedDetail := localResultRemovedRuntime(currentDetail)
 		if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{
 			Status:       "failed",
 			ErrorMessage: &msg,
@@ -601,13 +568,10 @@ func (w *Worker) uploadAndComplete(
 				Download: transferProgress(downloadedBytes, &downloadedBytes, zero),
 				Upload:   transferProgress(0, &downloadedBytes, zero),
 			},
-			Runtime: failedDetail,
+			Runtime: currentDetail,
 		}); updateErr != nil {
 			log.Error("failed to mark task failed", "error", updateErr)
 		}
-		if cleanupErr := cleanupDownloadedResult(context.WithoutCancel(ctx), task, result); cleanupErr != nil {
-			log.Warn("failed to remove failed local downloaded result", "path", result.Path, "error", cleanupErr)
-		}
 		return
 	}
 	uploadedBytes := result.Size
@@ -667,17 +631,6 @@ func interruptedRuntime(runtime *client.DownloadTaskRuntime) *client.DownloadTas
 	return runtime
 }
 
-func localResultRemovedRuntime(runtime *client.DownloadTaskRuntime) *client.DownloadTaskRuntime {
-	if runtime == nil {
-		runtime = &client.DownloadTaskRuntime{}
-	}
-	runtime.State = localResultRemovedRuntimeState
-	runtime.Phase = "error"
-	runtime.Seeding = nil
-	runtime.ETASeconds = nil
-	return runtime
-}
-
 func downloadProgressPatch(downloaded int64, total *int64, bps int64) *client.DownloadTaskProgressPatch {
 	return &client.DownloadTaskProgressPatch{Download: transferProgress(downloaded, total, bps)}
 }
@@ -879,7 +832,9 @@ func (w *Worker) ackStoppedControlTask(ctx context.Context, task client.Download
 		return
 	}
 	if task.State() == "canceling" {
-		w.cleanupTerminalTask(ctx, log, task, "canceled")
+		if runtime := task.Runtime(); runtime != nil && runtime.State == deleteRequestedRuntimeState {
+			w.cleanupDeletedTask(ctx, log, task)
+		}
 		if _, err := w.updateTask(ctx, task.ID, client.TaskPatch{Status: "canceled"}); err != nil {
 			log.Error("failed to acknowledge canceled task without local process", "error", err)
 			return
@@ -888,7 +843,7 @@ func (w *Worker) ackStoppedControlTask(ctx context.Context, task client.Download
 		return
 	}
 	if task.State() == "suspended" {
-		w.cleanupSuspendedTask(ctx, log, task)
+		log.Debug("suspended task is stopped without local cleanup")
 	}
 }
 

cmd/internal/worker/worker_test.go

@@ -64,7 +64,7 @@ func TestDownloadThenUploadStopsWhenSuspendedAtStart(t *testing.T) {
 	}
 }
 
-func TestCanceledDownloadCleansRuntimeAndMarksCanceled(t *testing.T) {
+func TestCanceledDownloadPreservesRuntimeAndMarksCanceled(t *testing.T) {
 	api := &recordingAPI{}
 	eng := &recordingEngine{downloadErr: context.Canceled}
 	w := NewWithAPI(config.Config{}, api)
@@ -75,8 +75,8 @@ func TestCanceledDownloadCleansRuntimeAndMarksCanceled(t *testing.T) {
 
 	w.downloadThenUpload(ctx, w.logger, clientTaskWithStatus("task-1", "downloading"), nil)
 
-	if eng.resetCalls != 1 {
-		t.Fatalf("expected canceled task cleanup to reset runtime once, got %d", eng.resetCalls)
+	if eng.resetCalls != 0 {
+		t.Fatalf("expected canceled task to preserve runtime, got %d reset calls", eng.resetCalls)
 	}
 	patch := lastPatchWithStatus(t, api.patches, "canceled")
 	if patch.State() != "canceled" {
@@ -84,7 +84,7 @@ func TestCanceledDownloadCleansRuntimeAndMarksCanceled(t *testing.T) {
 	}
 }
 
-func TestSuspendedDownloadCleansRuntimeWithoutStatusChange(t *testing.T) {
+func TestSuspendedDownloadPreservesRuntimeWithoutStatusChange(t *testing.T) {
 	api := &recordingAPI{}
 	eng := &recordingEngine{downloadErr: context.Canceled}
 	w := NewWithAPI(config.Config{}, api)
@@ -95,19 +95,20 @@ func TestSuspendedDownloadCleansRuntimeWithoutStatusChange(t *testing.T) {
 
 	w.downloadThenUpload(ctx, w.logger, clientTaskWithStatus("task-1", "downloading"), nil)
 
-	if eng.resetCalls != 1 {
-		t.Fatalf("expected suspended task cleanup to reset runtime once, got %d", eng.resetCalls)
+	if eng.resetCalls != 0 {
+		t.Fatalf("expected suspended task to preserve runtime, got %d reset calls", eng.resetCalls)
 	}
 	if _, ok := findPatchWithStatus(api.patches, "suspended"); ok {
 		t.Fatalf("expected worker not to overwrite server-owned suspended status, got %#v", api.patches)
 	}
-	patch := api.patches[len(api.patches)-1]
-	if patch.Runtime == nil || patch.Runtime.State != localResultRemovedRuntimeState {
-		t.Fatalf("expected suspended cleanup marker, got %#v", patch.Runtime)
+	for _, patch := range api.patches {
+		if patch.Runtime != nil && patch.Runtime.State == localResultRemovedRuntimeState {
+			t.Fatalf("expected suspended task not to mark local result removed, got %#v", patch.Runtime)
+		}
 	}
 }
 
-func TestTickSuspendedControlTaskCleansRuntimeOnlyOnce(t *testing.T) {
+func TestTickSuspendedControlTaskPreservesRuntime(t *testing.T) {
 	api := &recordingAPI{
 		controlTasks: []client.DownloadTask{clientTaskWithStatus("task-1", "suspended")},
 	}
@@ -119,37 +120,32 @@ func TestTickSuspendedControlTaskCleansRuntimeOnlyOnce(t *testing.T) {
 	if err := w.tick(context.Background()); err != nil {
 		t.Fatalf("first tick: %v", err)
 	}
-	if eng.resetCalls != 1 {
-		t.Fatalf("expected first suspended control poll to clean once, got %d", eng.resetCalls)
+	if eng.resetCalls != 0 {
+		t.Fatalf("expected suspended control poll to preserve runtime, got %d reset calls", eng.resetCalls)
 	}
-	if len(api.patches) != 1 {
-		t.Fatalf("expected first suspended control cleanup to record runtime once, got %#v", api.patches)
-	}
-	patch := api.patches[0]
-	if patch.State() != "" {
-		t.Fatalf("expected suspended control cleanup to preserve server-owned status, got %#v", patch)
-	}
-	if patch.Runtime == nil || patch.Runtime.State != localResultRemovedRuntimeState {
-		t.Fatalf("expected suspended control cleanup marker, got %#v", patch.Runtime)
+	if len(api.patches) != 0 {
+		t.Fatalf("expected suspended control poll not to patch task, got %#v", api.patches)
 	}
 
 	if err := w.tick(context.Background()); err != nil {
 		t.Fatalf("second tick: %v", err)
 	}
-	if eng.resetCalls != 1 {
-		t.Fatalf("expected repeated suspended control polls to avoid duplicate cleanup, got %d", eng.resetCalls)
+	if eng.resetCalls != 0 {
+		t.Fatalf("expected repeated suspended control polls not to clean runtime, got %d reset calls", eng.resetCalls)
 	}
-	if len(api.patches) != 1 {
-		t.Fatalf("expected repeated suspended control polls not to rewrite cleanup marker, got %#v", api.patches)
+	if len(api.patches) != 0 {
+		t.Fatalf("expected repeated suspended control polls not to patch task, got %#v", api.patches)
 	}
 	if got := api.controlTasks[0].State(); got != "suspended" {
 		t.Fatalf("expected recorded control task status to stay suspended, got %q", got)
 	}
 }
 
-func TestSuspendedControlTaskResumeAfterCleanupRedownloads(t *testing.T) {
+func TestDeleteRequestedControlTaskCleansRuntimeAndAcksCanceled(t *testing.T) {
 	api := &recordingAPI{
-		controlTasks: []client.DownloadTask{clientTaskWithStatus("task-1", "suspended")},
+		controlTasks: []client.DownloadTask{
+			withRuntime(clientTaskWithStatus("task-1", "canceling"), &client.DownloadTaskRuntime{State: deleteRequestedRuntimeState}),
+		},
 	}
 	eng := &recordingEngine{}
 	w := NewWithAPI(config.Config{}, api)
@@ -159,31 +155,37 @@ func TestSuspendedControlTaskResumeAfterCleanupRedownloads(t *testing.T) {
 	if err := w.tick(context.Background()); err != nil {
 		t.Fatalf("tick: %v", err)
 	}
-	marker := api.controlTasks[0].Runtime()
-	if marker == nil || marker.State != localResultRemovedRuntimeState {
-		t.Fatalf("expected suspended control task to record cleanup marker, got %#v", marker)
+	if eng.resetCalls != 1 {
+		t.Fatalf("expected delete-requested task to clean runtime once, got %d reset calls", eng.resetCalls)
 	}
-
-	total := int64(100)
-	eng.downloadErr = errors.New("redownload missing local result")
-	w.process(context.Background(), withRuntime(
-		withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), total, &total),
-		marker,
-	))
-
-	if eng.inspectCalls != 0 {
-		t.Fatalf("expected cleaned suspended result to skip runtime upload inspection, got %d inspect calls", eng.inspectCalls)
-	}
-	if eng.downloadCalls != 1 {
-		t.Fatalf("expected cleaned suspended result to resume via download path, got %d download calls", eng.downloadCalls)
-	}
-	failed := lastPatchWithStatus(t, api.patches, "failed")
-	if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "redownload missing local result") {
-		t.Fatalf("expected resumed redownload failure to be reported, got %#v", failed.ErrorMessage)
+	patch := lastPatchWithStatus(t, api.patches, "canceled")
+	if patch.State() != "canceled" {
+		t.Fatalf("expected delete-requested cleanup to ack canceled, got %#v", patch)
 	}
 }
 
-func TestFailedDownloadCleansRuntimeAndMarksFailed(t *testing.T) {
+func TestCancelingControlTaskWithoutDeleteRequestPreservesRuntime(t *testing.T) {
+	api := &recordingAPI{
+		controlTasks: []client.DownloadTask{clientTaskWithStatus("task-1", "canceling")},
+	}
+	eng := &recordingEngine{}
+	w := NewWithAPI(config.Config{}, api)
+	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	w.engine = eng
+
+	if err := w.tick(context.Background()); err != nil {
+		t.Fatalf("tick: %v", err)
+	}
+	if eng.resetCalls != 0 {
+		t.Fatalf("expected canceling task without delete request to preserve runtime, got %d reset calls", eng.resetCalls)
+	}
+	patch := lastPatchWithStatus(t, api.patches, "canceled")
+	if patch.State() != "canceled" {
+		t.Fatalf("expected canceling task to ack canceled, got %#v", patch)
+	}
+}
+
+func TestFailedDownloadPreservesRuntimeAndMarksFailed(t *testing.T) {
 	api := &recordingAPI{}
 	eng := &recordingEngine{downloadErr: errors.New("disk write failed")}
 	w := NewWithAPI(config.Config{}, api)
@@ -192,8 +194,8 @@ func TestFailedDownloadCleansRuntimeAndMarksFailed(t *testing.T) {
 
 	w.downloadThenUpload(context.Background(), w.logger, clientTaskWithStatus("task-1", "downloading"), nil)
 
-	if eng.resetCalls != 1 {
-		t.Fatalf("expected failed task cleanup to reset runtime once, got %d", eng.resetCalls)
+	if eng.resetCalls != 0 {
+		t.Fatalf("expected failed task to preserve runtime, got %d reset calls", eng.resetCalls)
 	}
 	failed := lastPatchWithStatus(t, api.patches, "failed")
 	if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "disk write failed") {
@@ -201,7 +203,7 @@ func TestFailedDownloadCleansRuntimeAndMarksFailed(t *testing.T) {
 	}
 }
 
-func TestTerminalDownloadStopsCleanPartialFiles(t *testing.T) {
+func TestTerminalDownloadStopsPreservePartialFiles(t *testing.T) {
 	cases := []struct {
 		name          string
 		cancelCause   error
@@ -278,8 +280,8 @@ func TestTerminalDownloadStopsCleanPartialFiles(t *testing.T) {
 				lastPatchWithStatus(t, api.patches, tc.wantStatus)
 			}
 			taskDir := filepath.Join(downloadDir, task.ID)
-			if _, err := os.Stat(taskDir); !os.IsNotExist(err) {
-				t.Fatalf("expected %s cleanup after %s stop, got err=%v", taskDir, tc.name, err)
+			if _, err := os.Stat(taskDir); err != nil {
+				t.Fatalf("expected %s to remain after %s stop, got err=%v", taskDir, tc.name, err)
 			}
 			if tc.requireStop {
 				for _, forbidden := range []string{"failed", "interrupted", "canceled", "paused"} {
@@ -735,15 +737,15 @@ func TestUploadFailurePersistsDownloadCheckpoint(t *testing.T) {
 	if failed.Progress.Download.TotalBytes == nil || *failed.Progress.Download.TotalBytes != result.Size {
 		t.Fatalf("expected total bytes %d, got %#v", result.Size, failed.Progress.Download.TotalBytes)
 	}
-	if failed.Runtime == nil || failed.Runtime.Phase != "error" || failed.Runtime.State != localResultRemovedRuntimeState {
-		t.Fatalf("expected local-result-removed runtime, got %#v", failed.Runtime)
+	if failed.Runtime != nil && failed.Runtime.State == localResultRemovedRuntimeState {
+		t.Fatalf("expected upload failure to preserve local result runtime, got %#v", failed.Runtime)
 	}
-	if _, err := os.Stat(resultPath); !os.IsNotExist(err) {
-		t.Fatalf("expected upload failure to remove local result path, stat err=%v", err)
+	if _, err := os.Stat(resultPath); err != nil {
+		t.Fatalf("expected upload failure to preserve local result path, stat err=%v", err)
 	}
 }
 
-func TestWorkerLifecycleUploadFailureCleansLocalResult(t *testing.T) {
+func TestWorkerLifecycleUploadFailurePreservesLocalResult(t *testing.T) {
 	payloadPath := writeTempFile(t, "downloaded payload")
 	payloadSize := int64(len("downloaded payload"))
 	uploadRequests := 0
@@ -777,18 +779,18 @@ func TestWorkerLifecycleUploadFailureCleansLocalResult(t *testing.T) {
 	if failed.Progress == nil || failed.Progress.Download == nil || failed.Progress.Download.Bytes != payloadSize {
 		t.Fatalf("expected failed task to persist downloaded bytes %d, got %#v", payloadSize, failed.Progress)
 	}
-	if failed.Runtime == nil || failed.Runtime.State != localResultRemovedRuntimeState {
-		t.Fatalf("expected failed task to mark local result removed, got %#v", failed.Runtime)
+	if failed.Runtime != nil && failed.Runtime.State == localResultRemovedRuntimeState {
+		t.Fatalf("expected failed task to preserve local result runtime, got %#v", failed.Runtime)
 	}
 	if uploadRequests != 1 {
 		t.Fatalf("expected one upload attempt, got %d", uploadRequests)
 	}
-	if _, err := os.Stat(payloadPath); !os.IsNotExist(err) {
-		t.Fatalf("expected failed upload to remove local payload, stat err=%v", err)
+	if _, err := os.Stat(payloadPath); err != nil {
+		t.Fatalf("expected failed upload to preserve local payload, stat err=%v", err)
 	}
 }
 
-func TestWorkerLifecycleHTTPUploadFailureCleansLocalResult(t *testing.T) {
+func TestWorkerLifecycleHTTPUploadFailurePreservesLocalResult(t *testing.T) {
 	payload := "downloaded payload"
 	downloadRequests := 0
 	downloadServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -829,8 +831,8 @@ func TestWorkerLifecycleHTTPUploadFailureCleansLocalResult(t *testing.T) {
 	if uploadRequests != 1 {
 		t.Fatalf("expected one upload attempt, got %d", uploadRequests)
 	}
-	if _, err := os.Stat(filepath.Join(downloadDir, "task-1")); !os.IsNotExist(err) {
-		t.Fatalf("expected failed upload to remove local task directory, stat err=%v", err)
+	if _, err := os.Stat(filepath.Join(downloadDir, "task-1")); err != nil {
+		t.Fatalf("expected failed upload to preserve local task directory, stat err=%v", err)
 	}
 
 	second := NewWithAPI(config.Config{}, api)
@@ -841,11 +843,11 @@ func TestWorkerLifecycleHTTPUploadFailureCleansLocalResult(t *testing.T) {
 		failedRuntime,
 	))
 
-	if downloadRequests != 2 {
-		t.Fatalf("expected retry after cleanup to redownload, got %d requests", downloadRequests)
+	if downloadRequests != 1 {
+		t.Fatalf("expected retry after upload failure to reuse local result, got %d download requests", downloadRequests)
 	}
 	if uploadRequests != 2 {
-		t.Fatalf("expected retry to upload redownloaded file, got %d upload requests", uploadRequests)
+		t.Fatalf("expected retry to upload preserved file, got %d upload requests", uploadRequests)
 	}
 	last := api.patches[len(api.patches)-1]
 	if last.State() != "completed" {
@@ -996,7 +998,7 @@ func TestUploadShutdownMarksTaskInterrupted(t *testing.T) {
 	}
 }
 
-func TestSuspendedUploadCleansLocalResultAndForcesRedownload(t *testing.T) {
+func TestSuspendedUploadPreservesLocalResult(t *testing.T) {
 	downloadDir := t.TempDir()
 	taskDir := filepath.Join(downloadDir, "task-1")
 	if err := os.MkdirAll(taskDir, 0o755); err != nil {
@@ -1026,19 +1028,13 @@ func TestSuspendedUploadCleansLocalResultAndForcesRedownload(t *testing.T) {
 	if _, ok := findPatchWithStatus(api.patches, "suspended"); ok {
 		t.Fatalf("expected worker not to overwrite server-owned suspended status, got %#v", api.patches)
 	}
-	patch := api.patches[len(api.patches)-1]
-	if patch.Runtime == nil || patch.Runtime.State != localResultRemovedRuntimeState {
-		t.Fatalf("expected suspended upload cleanup marker, got %#v", patch.Runtime)
+	for _, patch := range api.patches {
+		if patch.Runtime != nil && patch.Runtime.State == localResultRemovedRuntimeState {
+			t.Fatalf("expected suspended upload not to mark local result removed, got %#v", patch.Runtime)
+		}
 	}
-	if _, err := os.Stat(taskDir); !os.IsNotExist(err) {
-		t.Fatalf("expected suspended upload to remove local task directory, stat err=%v", err)
-	}
-	resumed := withRuntime(
-		withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), payloadSize, &payloadSize),
-		patch.Runtime,
-	)
-	if got := nextTaskWorkStage(resumed); got != taskWorkStageDownload {
-		t.Fatalf("expected resumed cleaned upload to redownload, got stage %v", got)
+	if _, err := os.Stat(taskDir); err != nil {
+		t.Fatalf("expected suspended upload to preserve local task directory, stat err=%v", err)
 	}
 }
 

server/http/downloads/download-tasks.integration.test.ts

@@ -1332,6 +1332,29 @@ describe('Download tasks API integration', () => {
       headers: { ...user, 'Content-Type': 'application/json' },
     })
     expect(deleteRes.status).toBe(204)
+
+    const deleteControlRes = await app.request('/api/downloads/tasks?assignedTo=me&status=canceling', {
+      headers: { Authorization: `Bearer ${createdDownloader.token}` },
+    })
+    expect(deleteControlRes.status).toBe(200)
+    const deleteControl = (await deleteControlRes.json()) as DownloadTaskList
+    const deleteControlTask = deleteControl.items.find((item) => item.id === createdTask.id)
+    expect(deleteControlTask).toMatchObject({
+      status: {
+        state: 'canceling',
+        runtime: { state: 'delete_requested' },
+      },
+    })
+
+    const deleteAckRes = await app.request(`/api/downloads/tasks/${createdTask.id}`, {
+      method: 'PATCH',
+      headers: downloaderHeaders,
+      body: JSON.stringify({ status: 'canceled' }),
+    })
+    expect(deleteAckRes.status).toBe(200)
+
+    const deletedTaskRes = await app.request(`/api/downloads/tasks/${createdTask.id}`, { headers: user })
+    expect(deletedTaskRes.status).toBe(404)
   })
 
   it('lets the assigned downloader recover interrupted tasks without resuming user-paused tasks [spec: download-tasks/recover-interrupted]', async () => {

server/usecases/downloads/downloads.ts

@@ -72,6 +72,7 @@ const RESTARTABLE_TASK_STATUSES = [
   'completed',
 ] as const
 const DOWNLOADER_TOKEN_TASK_STATUSES = ['assigned', 'downloading', 'uploading', 'interrupted'] as const
+const DELETE_REQUESTED_RUNTIME_STATE = 'delete_requested'
 const DELETE_DOWNLOADER_REQUEUE_STATUSES = [
   'queued',
   'assigned',
@@ -256,6 +257,16 @@ export async function updateDownloadTask(
     return deps.downloadTasks.get(task.orgId, id)
   }
   if (actor.downloaderId && task.status === 'canceling' && input.status === 'canceled') {
+    if (parseTaskRuntime(task.runtime)?.state === DELETE_REQUESTED_RUNTIME_STATE) {
+      await deps.downloadTasks.delete(id)
+      return downloadTaskFromRecord({
+        ...task,
+        status: 'canceled',
+        runtime: null,
+        finishedAt: task.finishedAt ?? now,
+        updatedAt: now,
+      })
+    }
     await deps.downloadTasks.setFields(id, {
       status: 'canceled',
       runtime: serializeTaskRuntime(stoppedRuntime(task.runtime)),
@@ -380,16 +391,27 @@ export async function performDownloadTaskAction(
   action: DownloadTaskActionInput['action'],
 ): Promise<DownloadTask | { id: string; deleted: true }> {
   const task = await deps.downloadTasks.getRecord(orgId, id)
+  const now = new Date()
 
   if (action === 'delete') {
     if (!TERMINAL_TASK_STATUSES.includes(task.status as (typeof TERMINAL_TASK_STATUSES)[number])) {
       throw new DownloadError('invalid_state', 'Only completed, failed, or canceled tasks can be deleted')
     }
+    if (task.assignedDownloaderId) {
+      await deps.downloadTasks.setFields(id, {
+        status: 'canceling',
+        runtime: serializeTaskRuntime({
+          ...(parseTaskRuntime(task.runtime) ?? {}),
+          state: DELETE_REQUESTED_RUNTIME_STATE,
+        }),
+        updatedAt: now,
+      })
+      return { id, deleted: true }
+    }
     await deps.downloadTasks.delete(id)
     return { id, deleted: true }
   }
 
-  const now = new Date()
   if (action === 'pause') {
     if (task.status === 'paused') return deps.downloadTasks.get(orgId, id)
     if (!PAUSABLE_TASK_STATUSES.includes(task.status as (typeof PAUSABLE_TASK_STATUSES)[number])) {
@@ -677,6 +699,62 @@ function clearTaskRuntimeMessage(runtime: DownloadTaskRuntime | null): DownloadT
   return Object.keys(rest).length > 0 ? rest : null
 }
 
+function downloadTaskFromRecord(row: DownloadTaskRecord): DownloadTask {
+  const runtime = parseTaskRuntime(row.runtime)
+  return {
+    id: row.id,
+    orgId: row.orgId,
+    createdBy: row.createdByUserId,
+    spec: {
+      source: {
+        type: row.sourceType as DownloadTask['spec']['source']['type'],
+        uri: row.sourceUri,
+      },
+      destination: {
+        folder: row.targetFolder,
+        name: row.displayName,
+      },
+      labels: {
+        category: row.category,
+        tags: parseStringArray(row.tags),
+      },
+    },
+    status: {
+      state: row.status as DownloadTask['status']['state'],
+      attempt: row.attempt,
+      assignment: row.assignedDownloaderId
+        ? { downloaderId: row.assignedDownloaderId, assignedAt: row.assignedAt?.toISOString() ?? null }
+        : null,
+      progress: runtime?.progress ?? {
+        download: { bytes: 0, totalBytes: null, bytesPerSecond: 0 },
+        upload: { bytes: 0, totalBytes: null, bytesPerSecond: 0 },
+      },
+      billing: {
+        state: row.billingStatus as DownloadTask['status']['billing']['state'],
+        authorizedBytes: row.billingAuthorizedBytes,
+        chargedBytes: row.billingChargedBytes,
+        chargedCredits: row.billingChargedCredits,
+      },
+      output: row.resultObjectId ? { objectId: row.resultObjectId } : null,
+      runtime,
+      error: row.errorMessage ? { code: row.errorCode, message: row.errorMessage } : null,
+      startedAt: row.startedAt?.toISOString() ?? null,
+      finishedAt: row.finishedAt?.toISOString() ?? null,
+      updatedAt: row.updatedAt.toISOString(),
+    },
+    createdAt: row.createdAt.toISOString(),
+  }
+}
+
+function parseStringArray(value: string): string[] {
+  try {
+    const parsed = JSON.parse(value)
+    return Array.isArray(parsed) ? parsed.filter((item): item is string => typeof item === 'string') : []
+  } catch {
+    return []
+  }
+}
+
 function parseTaskRuntime(value: string | null): DownloadTaskRuntime | null {
   if (!value) return null
   return downloadTaskRuntimeSchema.parse(JSON.parse(value))