fix: clean downloader terminal artifacts (#477)

* fix: clean downloader terminal artifacts

Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29

* test: tolerate stale cloud license cleanup

Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29

* test: retry transient pairing poll failures

Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29

* fix: make suspended downloader cleanup idempotent

Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29

---------

Co-authored-by: Jordan Park <jordan-park@mails.agent-kanban.dev>

cmd/internal/worker/disk_unix.go

@@ -0,0 +1,41 @@
+//go:build !windows
+
+package worker
+
+import (
+	"os"
+	"path/filepath"
+
+	"golang.org/x/sys/unix"
+)
+
+func freeDiskBytes(path string) (int64, error) {
+	statPath, err := existingStatPath(path)
+	if err != nil {
+		return 0, err
+	}
+	var stat unix.Statfs_t
+	if err := unix.Statfs(statPath, &stat); err != nil {
+		return 0, err
+	}
+	return int64(stat.Bavail) * int64(stat.Bsize), nil
+}
+
+func existingStatPath(path string) (string, error) {
+	if path == "" {
+		return ".", nil
+	}
+	path = filepath.Clean(path)
+	for {
+		if _, err := os.Stat(path); err == nil {
+			return path, nil
+		} else if !os.IsNotExist(err) {
+			return "", err
+		}
+		parent := filepath.Dir(path)
+		if parent == path {
+			return "", os.ErrNotExist
+		}
+		path = parent
+	}
+}

cmd/internal/worker/disk_unix_test.go

@@ -0,0 +1,25 @@
+//go:build !windows
+
+package worker
+
+import (
+	"testing"
+
+	"github.com/saltbo/zpan/internal/config"
+	"golang.org/x/sys/unix"
+)
+
+func TestHeartbeatReportsDownloadDirFreeDiskExactly(t *testing.T) {
+	downloadDir := t.TempDir()
+	var stat unix.Statfs_t
+	if err := unix.Statfs(downloadDir, &stat); err != nil {
+		t.Fatalf("statfs %s: %v", downloadDir, err)
+	}
+	want := int64(stat.Bavail) * int64(stat.Bsize)
+
+	w := NewWithAPI(config.Config{DownloadDir: downloadDir}, &recordingAPI{})
+
+	if got := w.heartbeat().FreeDiskBytes; got != want {
+		t.Fatalf("expected heartbeat free disk %d, got %d", want, got)
+	}
+}

cmd/internal/worker/disk_windows.go

@@ -0,0 +1,45 @@
+//go:build windows
+
+package worker
+
+import (
+	"os"
+	"path/filepath"
+
+	"golang.org/x/sys/windows"
+)
+
+func freeDiskBytes(path string) (int64, error) {
+	statPath, err := existingStatPath(path)
+	if err != nil {
+		return 0, err
+	}
+	ptr, err := windows.UTF16PtrFromString(statPath)
+	if err != nil {
+		return 0, err
+	}
+	var freeBytes uint64
+	if err := windows.GetDiskFreeSpaceEx(ptr, &freeBytes, nil, nil); err != nil {
+		return 0, err
+	}
+	return int64(freeBytes), nil
+}
+
+func existingStatPath(path string) (string, error) {
+	if path == "" {
+		return ".", nil
+	}
+	path = filepath.Clean(path)
+	for {
+		if _, err := os.Stat(path); err == nil {
+			return path, nil
+		} else if !os.IsNotExist(err) {
+			return "", err
+		}
+		parent := filepath.Dir(path)
+		if parent == path {
+			return "", os.ErrNotExist
+		}
+		path = parent
+	}
+}

cmd/internal/worker/seeds.go

@@ -30,11 +30,21 @@ type retainedSeed struct {
 	cleanup    func(context.Context) error
 }
 
-func cleanupDownloadedResult(ctx context.Context, result engine.Result) error {
+func cleanupDownloadedResult(ctx context.Context, task client.DownloadTask, result engine.Result) error {
 	if result.Seed != nil && result.Seed.Cleanup != nil {
 		return result.Seed.Cleanup(ctx)
 	}
-	return os.RemoveAll(result.Path)
+	parent := filepath.Dir(result.Path)
+	if filepath.Base(parent) == task.ID {
+		return os.RemoveAll(parent)
+	}
+	if result.IsDir {
+		return os.RemoveAll(result.Path)
+	}
+	if err := os.Remove(result.Path); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	return nil
 }
 
 func (w *Worker) retainSeed(task client.DownloadTask, result engine.Result, log *slog.Logger) bool {

cmd/internal/worker/worker.go

@@ -23,6 +23,7 @@ import (
 
 const Version = "0.1.0"
 const maxTaskErrorMessageLength = 1000
+const localResultRemovedRuntimeState = "local_result_removed"
 
 var errTaskPausing = errors.New("task pausing")
 var errTaskCanceling = errors.New("task canceling")
@@ -297,6 +298,7 @@ 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)
 				}
@@ -306,6 +308,7 @@ 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
 			}
@@ -326,6 +329,7 @@ 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
 	}
 
@@ -393,13 +397,16 @@ func nextTaskWorkStage(task client.DownloadTask) taskWorkStage {
 	if task.State() == "uploading" {
 		return taskWorkStageUploadExistingResult
 	}
+	runtime := task.Runtime()
+	if runtime != nil && runtime.State == localResultRemovedRuntimeState {
+		return taskWorkStageDownload
+	}
 	if task.State() != "assigned" && task.State() != "downloading" && task.State() != "interrupted" {
 		return taskWorkStageDownload
 	}
 	if task.Status.Progress.Upload.Bytes > 0 {
 		return taskWorkStageUploadExistingResult
 	}
-	runtime := task.Runtime()
 	if runtime != nil && (runtime.Phase == "uploading" || runtime.Phase == "completed") {
 		return taskWorkStageUploadExistingResult
 	}
@@ -477,6 +484,41 @@ 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 {
+	w.cleanupRetainedSeedForTask(ctx, task.ID, reason)
+	if w.engine == nil {
+		log.Warn("downloader engine is unavailable for terminal cleanup", "reason", reason)
+		return false
+	}
+	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
+	}
+	if err := resetter.ResetTask(ctx, task); err != nil {
+		log.Warn("failed to clean terminal downloader task", "reason", reason, "error", err)
+		return false
+	}
+	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(
 	ctx context.Context,
 	log *slog.Logger,
@@ -498,6 +540,38 @@ func (w *Worker) uploadAndComplete(
 	if err != nil {
 		downloadedBytes := result.Size
 		if errors.Is(err, context.Canceled) {
+			if errors.Is(context.Cause(ctx), errTaskPausing) {
+				if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{Status: "paused"}); updateErr != nil {
+					log.Error("failed to mark task paused during upload", "error", updateErr)
+				}
+				log.Info("task upload paused by control action")
+				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 {
+					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
+			}
 			uploadingDetail := currentDetail
 			if uploadingDetail == nil {
 				uploadingDetail = &client.DownloadTaskRuntime{}
@@ -519,12 +593,7 @@ func (w *Worker) uploadAndComplete(
 		}
 		msg := taskErrorMessage(err)
 		log.Error("failed to upload result", "error", err)
-		failedDetail := currentDetail
-		if failedDetail == nil {
-			failedDetail = &client.DownloadTaskRuntime{}
-		}
-		failedDetail.Phase = "uploading"
-		failedDetail.Seeding = nil
+		failedDetail := localResultRemovedRuntime(currentDetail)
 		if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{
 			Status:       "failed",
 			ErrorMessage: &msg,
@@ -536,6 +605,9 @@ func (w *Worker) uploadAndComplete(
 		}); 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
@@ -564,7 +636,7 @@ func (w *Worker) uploadAndComplete(
 		w.cleanupRetainedSeeds(ctx)
 		return
 	}
-	if err := cleanupDownloadedResult(ctx, result); err != nil {
+	if err := cleanupDownloadedResult(ctx, task, result); err != nil {
 		log.Warn("failed to remove local downloaded result", "path", result.Path, "error", err)
 	}
 }
@@ -595,6 +667,17 @@ 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)}
 }
@@ -796,11 +879,16 @@ func (w *Worker) ackStoppedControlTask(ctx context.Context, task client.Download
 		return
 	}
 	if task.State() == "canceling" {
+		w.cleanupTerminalTask(ctx, log, task, "canceled")
 		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
 		}
 		log.Info("acknowledged canceled task without local process")
+		return
+	}
+	if task.State() == "suspended" {
+		w.cleanupSuspendedTask(ctx, log, task)
 	}
 }
 
@@ -812,6 +900,10 @@ func (w *Worker) heartbeat() client.Heartbeat {
 		capabilities = w.engine.Capabilities()
 	}
 	speeds := w.currentTransferSpeeds()
+	freeDiskBytes, err := freeDiskBytes(w.cfg.DownloadDir)
+	if err != nil {
+		w.logger.Warn("failed to inspect downloader free disk space", "download_dir", w.cfg.DownloadDir, "error", err)
+	}
 	return client.Heartbeat{
 		Version:            Version,
 		Hostname:           host.DownloaderHostname(),
@@ -823,7 +915,7 @@ func (w *Worker) heartbeat() client.Heartbeat {
 		CurrentTasks:       w.currentTasks(),
 		DownloadBps:        speeds.downloadBps,
 		UploadBps:          speeds.uploadBps,
-		FreeDiskBytes:      0,
+		FreeDiskBytes:      freeDiskBytes,
 	}
 }
 

cmd/internal/worker/worker_test.go

@@ -64,6 +64,234 @@ func TestDownloadThenUploadStopsWhenSuspendedAtStart(t *testing.T) {
 	}
 }
 
+func TestCanceledDownloadCleansRuntimeAndMarksCanceled(t *testing.T) {
+	api := &recordingAPI{}
+	eng := &recordingEngine{downloadErr: context.Canceled}
+	w := NewWithAPI(config.Config{}, api)
+	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	w.engine = eng
+	ctx, cancel := context.WithCancelCause(context.Background())
+	cancel(errTaskCanceling)
+
+	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)
+	}
+	patch := lastPatchWithStatus(t, api.patches, "canceled")
+	if patch.State() != "canceled" {
+		t.Fatalf("expected canceled patch, got %#v", patch)
+	}
+}
+
+func TestSuspendedDownloadCleansRuntimeWithoutStatusChange(t *testing.T) {
+	api := &recordingAPI{}
+	eng := &recordingEngine{downloadErr: context.Canceled}
+	w := NewWithAPI(config.Config{}, api)
+	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	w.engine = eng
+	ctx, cancel := context.WithCancelCause(context.Background())
+	cancel(errTaskSuspended)
+
+	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 _, 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)
+	}
+}
+
+func TestTickSuspendedControlTaskCleansRuntimeOnlyOnce(t *testing.T) {
+	api := &recordingAPI{
+		controlTasks: []client.DownloadTask{clientTaskWithStatus("task-1", "suspended")},
+	}
+	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("first tick: %v", err)
+	}
+	if eng.resetCalls != 1 {
+		t.Fatalf("expected first suspended control poll to clean once, got %d", 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 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 len(api.patches) != 1 {
+		t.Fatalf("expected repeated suspended control polls not to rewrite cleanup marker, 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) {
+	api := &recordingAPI{
+		controlTasks: []client.DownloadTask{clientTaskWithStatus("task-1", "suspended")},
+	}
+	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)
+	}
+	marker := api.controlTasks[0].Runtime()
+	if marker == nil || marker.State != localResultRemovedRuntimeState {
+		t.Fatalf("expected suspended control task to record cleanup marker, got %#v", marker)
+	}
+
+	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)
+	}
+}
+
+func TestFailedDownloadCleansRuntimeAndMarksFailed(t *testing.T) {
+	api := &recordingAPI{}
+	eng := &recordingEngine{downloadErr: errors.New("disk write failed")}
+	w := NewWithAPI(config.Config{}, api)
+	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	w.engine = eng
+
+	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)
+	}
+	failed := lastPatchWithStatus(t, api.patches, "failed")
+	if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "disk write failed") {
+		t.Fatalf("expected failure message to be reported, got %#v", failed.ErrorMessage)
+	}
+}
+
+func TestTerminalDownloadStopsCleanPartialFiles(t *testing.T) {
+	cases := []struct {
+		name          string
+		cancelCause   error
+		failedMessage string
+		wantStatus    string
+		requireStop   bool
+	}{
+		{
+			name:        "canceled",
+			cancelCause: errTaskCanceling,
+			wantStatus:  "canceled",
+		},
+		{
+			name:        "suspended",
+			cancelCause: errTaskSuspended,
+			requireStop: true,
+		},
+		{
+			name:          "failed",
+			failedMessage: "disk write failed",
+			wantStatus:    "failed",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			api := &recordingAPI{}
+			downloadDir := t.TempDir()
+			partialPath := filepath.Join(downloadDir, "task-1", "payload.bin")
+			ready := make(chan struct{}, 1)
+			eng := &recordingEngine{
+				downloadFunc: func(ctx context.Context, task client.DownloadTask, progress engine.Progress) (engine.Result, error) {
+					if err := os.MkdirAll(filepath.Dir(partialPath), 0o755); err != nil {
+						return engine.Result{}, err
+					}
+					if err := os.WriteFile(partialPath, []byte("partial"), 0o644); err != nil {
+						return engine.Result{}, err
+					}
+					ready <- struct{}{}
+					if tc.failedMessage != "" {
+						return engine.Result{}, errors.New(tc.failedMessage)
+					}
+					<-ctx.Done()
+					return engine.Result{}, ctx.Err()
+				},
+				resetTaskFn: func(context.Context, client.DownloadTask) error {
+					return os.RemoveAll(filepath.Join(downloadDir, "task-1"))
+				},
+			}
+			w := NewWithAPI(config.Config{DownloadDir: downloadDir}, api)
+			w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+			w.engine = eng
+
+			task := clientHTTPTask("task-1", "downloading", "https://example.com/payload.bin", "payload.bin")
+			if tc.cancelCause != nil {
+				ctx, cancel := context.WithCancelCause(context.Background())
+				done := make(chan struct{})
+				go func() {
+					w.downloadThenUpload(ctx, w.logger, task, nil)
+					close(done)
+				}()
+				select {
+				case <-ready:
+				case <-time.After(5 * time.Second):
+					t.Fatal("timed out waiting for partial download")
+				}
+				cancel(tc.cancelCause)
+				waitForWorkerTestCompletion(t, done)
+			} else {
+				w.downloadThenUpload(context.Background(), w.logger, task, nil)
+			}
+
+			if tc.wantStatus != "" {
+				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 tc.requireStop {
+				for _, forbidden := range []string{"failed", "interrupted", "canceled", "paused"} {
+					if _, ok := findPatchWithStatus(api.patches, forbidden); ok {
+						t.Fatalf("expected suspended stop not to emit %q, got %#v", forbidden, api.patches)
+					}
+				}
+			}
+		})
+	}
+}
+
 func TestWatchEngineProcessFatalOnUnexpectedExit(t *testing.T) {
 	w := NewWithAPI(config.Config{}, nil)
 	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
@@ -233,7 +461,7 @@ func TestReconcileEngineSeedsAdoptsUntrackedOrphans(t *testing.T) {
 }
 
 func TestHeartbeatReportsAggregateTransferSpeeds(t *testing.T) {
-	w := NewWithAPI(config.Config{Engine: "auto", MaxConcurrentTasks: 5}, &recordingAPI{})
+	w := NewWithAPI(config.Config{Engine: "auto", MaxConcurrentTasks: 5, DownloadDir: t.TempDir()}, &recordingAPI{})
 
 	if _, ok := w.startTask(context.Background(), "task-1"); !ok {
 		t.Fatal("expected task-1 to start")
@@ -251,6 +479,9 @@ func TestHeartbeatReportsAggregateTransferSpeeds(t *testing.T) {
 	if heartbeat.DownloadBps != 400 || heartbeat.UploadBps != 60 {
 		t.Fatalf("expected aggregate speeds 400/60, got %d/%d", heartbeat.DownloadBps, heartbeat.UploadBps)
 	}
+	if heartbeat.FreeDiskBytes <= 0 {
+		t.Fatalf("expected heartbeat to report free disk bytes, got %d", heartbeat.FreeDiskBytes)
+	}
 
 	w.finish("task-1")
 	heartbeat = w.heartbeat()
@@ -445,11 +676,39 @@ func TestCollectDirectoryEntriesSkipsDownloadSidecars(t *testing.T) {
 	}
 }
 
+func TestCleanupDownloadedResultRemovesTaskDirForNestedDirectoryResult(t *testing.T) {
+	downloadDir := t.TempDir()
+	taskDir := filepath.Join(downloadDir, "task-1")
+	resultDir := filepath.Join(taskDir, "payload")
+	if err := os.MkdirAll(resultDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(resultDir, "file.txt"), []byte("payload"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(taskDir, "payload.torrent"), []byte("sidecar"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+
+	err := cleanupDownloadedResult(context.Background(), clientTaskWithStatus("task-1", "downloading"), engine.Result{
+		Path:  resultDir,
+		Name:  "payload",
+		IsDir: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := os.Stat(taskDir); !os.IsNotExist(err) {
+		t.Fatalf("expected nested directory result cleanup to remove task dir, stat err=%v", err)
+	}
+}
+
 func TestUploadFailurePersistsDownloadCheckpoint(t *testing.T) {
 	api := &recordingAPI{createFolderErr: errors.New("unauthorized")}
 	w := NewWithAPI(config.Config{}, api)
+	resultPath := t.TempDir()
 	result := engine.Result{
-		Path:  t.TempDir(),
+		Path:  resultPath,
 		Name:  "album",
 		Size:  1234,
 		IsDir: true,
@@ -476,12 +735,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 != "uploading" {
-		t.Fatalf("expected uploading detail phase, got %#v", failed.Runtime)
+	if failed.Runtime == nil || failed.Runtime.Phase != "error" || failed.Runtime.State != localResultRemovedRuntimeState {
+		t.Fatalf("expected local-result-removed 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)
 	}
 }
 
-func TestWorkerLifecycleRetriesUploadWithoutRedownloading(t *testing.T) {
+func TestWorkerLifecycleUploadFailureCleansLocalResult(t *testing.T) {
 	payloadPath := writeTempFile(t, "downloaded payload")
 	payloadSize := int64(len("downloaded payload"))
 	uploadRequests := 0
@@ -515,33 +777,18 @@ func TestWorkerLifecycleRetriesUploadWithoutRedownloading(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.Phase != "uploading" {
-		t.Fatalf("expected failed task to persist uploading phase, got %#v", failed.Runtime)
+	if failed.Runtime == nil || failed.Runtime.State != localResultRemovedRuntimeState {
+		t.Fatalf("expected failed task to mark local result removed, got %#v", failed.Runtime)
 	}
-
-	second := NewWithAPI(config.Config{}, api)
-	second.engine = eng
-	second.process(context.Background(), withRuntime(
-		withDownloadCheckpoint(clientHTTPTask("task-1", "assigned", "https://example.com/payload.bin", "payload.bin"), payloadSize, &payloadSize),
-		&client.DownloadTaskRuntime{Phase: "uploading"},
-	))
-
-	if eng.downloadCalls != 1 {
-		t.Fatalf("expected retry to avoid a second download, got %d download calls", eng.downloadCalls)
+	if uploadRequests != 1 {
+		t.Fatalf("expected one upload attempt, got %d", uploadRequests)
 	}
-	if eng.inspectCalls != 1 {
-		t.Fatalf("expected retry to inspect the runtime task, got %d inspect calls", eng.inspectCalls)
-	}
-	if uploadRequests != 2 {
-		t.Fatalf("expected both attempts to upload the local result, got %d upload requests", uploadRequests)
-	}
-	last := api.patches[len(api.patches)-1]
-	if last.State() != "completed" {
-		t.Fatalf("expected retry to complete task, got last patch %#v", last)
+	if _, err := os.Stat(payloadPath); !os.IsNotExist(err) {
+		t.Fatalf("expected failed upload to remove local payload, stat err=%v", err)
 	}
 }
 
-func TestWorkerLifecycleRetriesHTTPUploadFromCheckpointWithoutRedownloading(t *testing.T) {
+func TestWorkerLifecycleHTTPUploadFailureCleansLocalResult(t *testing.T) {
 	payload := "downloaded payload"
 	downloadRequests := 0
 	downloadServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
@@ -579,19 +826,26 @@ func TestWorkerLifecycleRetriesHTTPUploadFromCheckpointWithoutRedownloading(t *t
 	if downloadRequests != 1 {
 		t.Fatalf("expected initial attempt to download once, got %d requests", downloadRequests)
 	}
+	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)
+	}
 
 	second := NewWithAPI(config.Config{}, api)
 	second.engine = engine.HTTP{Dir: downloadDir}
+	failedRuntime := failed.Runtime
 	second.process(context.Background(), withRuntime(
 		withDownloadCheckpoint(clientHTTPTask("task-1", "assigned", downloadServer.URL+"/payload.bin", "payload.bin"), payloadSize, &payloadSize),
-		&client.DownloadTaskRuntime{Phase: "uploading"},
+		failedRuntime,
 	))
 
-	if downloadRequests != 1 {
-		t.Fatalf("expected retry not to request download source again, got %d requests", downloadRequests)
+	if downloadRequests != 2 {
+		t.Fatalf("expected retry after cleanup to redownload, got %d requests", downloadRequests)
 	}
 	if uploadRequests != 2 {
-		t.Fatalf("expected both attempts to upload the local file, got %d upload requests", uploadRequests)
+		t.Fatalf("expected retry to upload redownloaded file, got %d upload requests", uploadRequests)
 	}
 	last := api.patches[len(api.patches)-1]
 	if last.State() != "completed" {
@@ -707,6 +961,9 @@ func TestDownloadShutdownMarksTaskInterrupted(t *testing.T) {
 	if patch.Runtime == nil || patch.Runtime.Message == "" {
 		t.Fatalf("expected interrupted detail message, got %#v", patch.Runtime)
 	}
+	if eng.resetCalls != 0 {
+		t.Fatalf("expected interrupted shutdown to preserve resumable runtime data, got %d resets", eng.resetCalls)
+	}
 }
 
 func TestUploadShutdownMarksTaskInterrupted(t *testing.T) {
@@ -739,6 +996,154 @@ func TestUploadShutdownMarksTaskInterrupted(t *testing.T) {
 	}
 }
 
+func TestSuspendedUploadCleansLocalResultAndForcesRedownload(t *testing.T) {
+	downloadDir := t.TempDir()
+	taskDir := filepath.Join(downloadDir, "task-1")
+	if err := os.MkdirAll(taskDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	payloadPath := filepath.Join(taskDir, "payload.bin")
+	payload := "downloaded payload"
+	if err := os.WriteFile(payloadPath, []byte(payload), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	payloadSize := int64(len(payload))
+	api := &recordingAPI{
+		createObjectDraft: client.ObjectDraft{ID: "object-1", Name: "payload.bin", Upload: &client.ObjectUploadInstructions{SessionID: "session-1", PartSize: payloadSize, URLs: []string{"http://127.0.0.1:1"}}},
+	}
+	w := NewWithAPI(config.Config{}, api)
+
+	ctx, cancel := context.WithCancelCause(context.Background())
+	cancel(errTaskSuspended)
+	w.uploadAndComplete(
+		ctx,
+		slog.New(slog.NewTextHandler(io.Discard, nil)),
+		clientTaskWithUploadToken("task-1", "downloading"),
+		engine.Result{Path: payloadPath, Name: "payload.bin", Size: payloadSize},
+		&client.DownloadTaskRuntime{Phase: "uploading"},
+	)
+
+	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)
+	}
+	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)
+	}
+}
+
+func TestProcessRedownloadsWhenLocalResultWasCleaned(t *testing.T) {
+	api := &recordingAPI{}
+	eng := &recordingEngine{downloadErr: errors.New("redownload missing local result")}
+	w := NewWithAPI(config.Config{}, api)
+	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	w.engine = eng
+	total := int64(100)
+
+	w.process(context.Background(), withRuntime(
+		withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), total, &total),
+		&client.DownloadTaskRuntime{Phase: "error", State: localResultRemovedRuntimeState},
+	))
+
+	if eng.inspectCalls != 0 {
+		t.Fatalf("expected cleaned local result to skip runtime upload inspection, got %d inspect calls", eng.inspectCalls)
+	}
+	if eng.downloadCalls != 1 {
+		t.Fatalf("expected cleaned local 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)
+	}
+}
+
+func TestPausedAndInterruptedDownloadsPreservePartialFiles(t *testing.T) {
+	cases := []struct {
+		name       string
+		cancelCtx  func(context.CancelCauseFunc, context.CancelFunc)
+		wantStatus string
+	}{
+		{
+			name: "paused",
+			cancelCtx: func(cancelCause context.CancelCauseFunc, _ context.CancelFunc) {
+				cancelCause(errTaskPausing)
+			},
+			wantStatus: "paused",
+		},
+		{
+			name: "interrupted",
+			cancelCtx: func(_ context.CancelCauseFunc, cancel context.CancelFunc) {
+				cancel()
+			},
+			wantStatus: "interrupted",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			api := &recordingAPI{}
+			downloadDir := t.TempDir()
+			path := filepath.Join(downloadDir, "task-1", "payload.bin")
+			ready := make(chan struct{}, 1)
+			eng := &recordingEngine{
+				downloadFunc: func(ctx context.Context, task client.DownloadTask, progress engine.Progress) (engine.Result, error) {
+					if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+						return engine.Result{}, err
+					}
+					if err := os.WriteFile(path, []byte("partial"), 0o644); err != nil {
+						return engine.Result{}, err
+					}
+					ready <- struct{}{}
+					<-ctx.Done()
+					return engine.Result{}, ctx.Err()
+				},
+				resetTaskFn: func(context.Context, client.DownloadTask) error {
+					return os.RemoveAll(filepath.Join(downloadDir, "task-1"))
+				},
+			}
+
+			w := NewWithAPI(config.Config{DownloadDir: downloadDir}, api)
+			w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+			w.engine = eng
+
+			ctx, cancel := context.WithCancel(context.Background())
+			ctxWithCause, cancelCause := context.WithCancelCause(ctx)
+			done := make(chan struct{})
+			go func() {
+				w.downloadThenUpload(ctxWithCause, w.logger, clientHTTPTask("task-1", "downloading", "https://example.com/payload.bin", "payload.bin"), nil)
+				close(done)
+			}()
+
+			select {
+			case <-ready:
+			case <-time.After(5 * time.Second):
+				t.Fatal("timed out waiting for partial download")
+			}
+			tc.cancelCtx(cancelCause, cancel)
+			waitForWorkerTestCompletion(t, done)
+
+			lastPatchWithStatus(t, api.patches, tc.wantStatus)
+			info, err := os.Stat(path)
+			if err != nil {
+				t.Fatalf("expected resumable file %s to remain, got %v", path, err)
+			}
+			if info.Size() == 0 {
+				t.Fatalf("expected resumable file %s to keep partial content", path)
+			}
+		})
+	}
+}
+
 func TestUploadETARoundsRemainingSeconds(t *testing.T) {
 	eta := uploadETA(&uploadProgress{uploaded: 25, totalBytes: 100}, 20)
 
@@ -838,6 +1243,14 @@ func TestNextTaskWorkStage(t *testing.T) {
 			task: withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), 100, &total),
 			want: taskWorkStageUploadExistingResult,
 		},
+		{
+			name: "assigned with removed local result marker",
+			task: withRuntime(
+				withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), 100, &total),
+				&client.DownloadTaskRuntime{Phase: "error", State: localResultRemovedRuntimeState},
+			),
+			want: taskWorkStageDownload,
+		},
 		{
 			name: "assigned partial download",
 			task: withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), 99, &total),
@@ -977,6 +1390,55 @@ func TestRetainSeedKeepsDownloadedResult(t *testing.T) {
 	}
 }
 
+func TestRetainedSeedExpiresWhenLedgerPersistenceFails(t *testing.T) {
+	stateFile := filepath.Join(t.TempDir(), "state-file")
+	if err := os.WriteFile(stateFile, []byte("not a directory"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	cleaned := false
+	w := NewWithAPI(config.Config{SeedEnabled: true, SeedDuration: time.Hour, StateDir: stateFile}, &recordingAPI{})
+	w.engine = &recordingEngine{}
+
+	retained := w.retainSeed(
+		clientTask("task-1"),
+		engine.Result{
+			Path: filepath.Join(t.TempDir(), "result"),
+			Size: 123,
+			Seed: &engine.Seed{
+				Engine:   "aria2",
+				ID:       "gid",
+				InfoHash: "infohash",
+				Path:     t.TempDir(),
+				Snapshot: func(context.Context) (engine.SeedSnapshot, error) {
+					return engine.SeedSnapshot{}, nil
+				},
+				Cleanup: func(context.Context) error {
+					cleaned = true
+					return nil
+				},
+			},
+		},
+		w.logger,
+	)
+
+	if !retained {
+		t.Fatal("expected seed to remain tracked in memory")
+	}
+	if len(w.retainedSeedSnapshot()) != 1 {
+		t.Fatalf("expected retained seed despite ledger failure, got %d", len(w.retainedSeedSnapshot()))
+	}
+	w.retainedSeeds[0].expiresAt = time.Now().Add(-time.Second)
+
+	w.cleanupRetainedSeeds(context.Background())
+
+	if !cleaned {
+		t.Fatal("expected in-memory retained seed to expire and clean up")
+	}
+	if len(w.retainedSeedSnapshot()) != 0 {
+		t.Fatalf("expected expired seed to be removed from memory, got %d", len(w.retainedSeedSnapshot()))
+	}
+}
+
 func TestReportRetainedSeedsCleansMissingSeed(t *testing.T) {
 	cleaned := false
 	w := NewWithAPI(config.Config{SeedEnabled: true}, &recordingAPI{})
@@ -1317,6 +1779,15 @@ func writeTempFile(t *testing.T, content string) string {
 	return file.Name()
 }
 
+func waitForWorkerTestCompletion(t *testing.T, done <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("timed out waiting for worker completion")
+	}
+}
+
 func clientTask(id string) client.DownloadTask {
 	return client.DownloadTask{
 		ID: id,
@@ -1378,7 +1849,9 @@ func findPatchWithStatus(patches []client.TaskPatch, status string) (client.Task
 type recordingEngine struct {
 	downloadResult engine.Result
 	downloadErr    error
+	downloadFunc   func(context.Context, client.DownloadTask, engine.Progress) (engine.Result, error)
 	resetErr       error
+	resetTaskFn    func(context.Context, client.DownloadTask) error
 	taskSnapshot   engine.TaskSnapshot
 	inspectErr     error
 	inspectPanic   any
@@ -1425,13 +1898,19 @@ func (e *recordingEngine) ListSeeds(context.Context) ([]engine.Seed, error) {
 	return e.listSeeds, nil
 }
 
-func (e *recordingEngine) ResetTask(context.Context, client.DownloadTask) error {
+func (e *recordingEngine) ResetTask(ctx context.Context, task client.DownloadTask) error {
 	e.resetCalls++
+	if e.resetTaskFn != nil {
+		return e.resetTaskFn(ctx, task)
+	}
 	return e.resetErr
 }
 
-func (e *recordingEngine) Download(context.Context, client.DownloadTask, engine.Progress) (engine.Result, error) {
+func (e *recordingEngine) Download(ctx context.Context, task client.DownloadTask, progress engine.Progress) (engine.Result, error) {
 	e.downloadCalls++
+	if e.downloadFunc != nil {
+		return e.downloadFunc(ctx, task, progress)
+	}
 	return e.downloadResult, e.downloadErr
 }
 
@@ -1439,6 +1918,7 @@ type recordingAPI struct {
 	patches            []client.TaskPatch
 	patchedIDs         []string
 	seedingTasks       []client.DownloadTask
+	controlTasks       []client.DownloadTask
 	assignedTasks      []client.DownloadTask
 	suspendDownloading bool
 	createFolderErr    error
@@ -1451,7 +1931,7 @@ func (a *recordingAPI) Heartbeat(context.Context, client.Heartbeat) error {
 }
 
 func (a *recordingAPI) AssignedControlTasks(context.Context) ([]client.DownloadTask, error) {
-	return nil, nil
+	return a.controlTasks, nil
 }
 
 func (a *recordingAPI) AssignedTasks(context.Context) ([]client.DownloadTask, error) {
@@ -1469,8 +1949,32 @@ func (a *recordingAPI) UpdateTask(_ context.Context, id string, patch client.Tas
 	if a.suspendDownloading && state == "downloading" {
 		state = "suspended"
 	}
+	recordedPatch := patch
+	if state != patch.State() {
+		recordedPatch.Status = state
+	}
+	applyRecordedTaskPatch(a.controlTasks, id, recordedPatch)
+	applyRecordedTaskPatch(a.assignedTasks, id, recordedPatch)
 	task := clientTaskWithStatus(id, state)
-	task.Status.Runtime = patch.Runtime
+	task = applyTaskPatch(task, recordedPatch)
+	return task, nil
+}
+
+func applyRecordedTaskPatch(tasks []client.DownloadTask, id string, patch client.TaskPatch) {
+	for i := range tasks {
+		if tasks[i].ID == id {
+			tasks[i] = applyTaskPatch(tasks[i], patch)
+		}
+	}
+}
+
+func applyTaskPatch(task client.DownloadTask, patch client.TaskPatch) client.DownloadTask {
+	if patch.State() != "" {
+		task.Status.State = patch.State()
+	}
+	if patch.Runtime != nil {
+		task.Status.Runtime = patch.Runtime
+	}
 	if patch.Progress != nil {
 		if patch.Progress.Download != nil {
 			task.Status.Progress.Download = *patch.Progress.Download
@@ -1479,7 +1983,7 @@ func (a *recordingAPI) UpdateTask(_ context.Context, id string, patch client.Tas
 			task.Status.Progress.Upload = *patch.Progress.Upload
 		}
 	}
-	return task, nil
+	return task
 }
 
 func (a *recordingAPI) CreateFolder(context.Context, string, string, string) (client.ObjectDraft, error) {