feat(task): 任务队列支持取消运行中与等待中的任务

任务执行接入 context 并放入独立进程组,取消时整组杀死;
新增可选 CancelShell 清理钩子,远程下载、压缩、备份任务
取消后自动清理残留文件,备份改用独享临时目录避免误伤并发备份;
防重改为按语言无关的任务标识 Key 匹配

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

internal/apps/prometheus/app.go

@@ -263,6 +263,7 @@ func (s *App) InstallExporter(w http.ResponseWriter, r *http.Request) {
 	cmd := fmt.Sprintf(`curl -sSLm 10 --retry 3 'https://%s/prometheus/exporters/exporter.sh' | bash -s -- 'install' '%s'`, s.conf.App.DownloadEndpoint, url.PathEscape(req.Slug))
 
 	task := new(biz.Task)
+	task.Key = "prometheus:exporter:" + req.Slug
 	task.Name = s.t.Get("Install Prometheus exporter %s", req.Slug)
 	task.Status = biz.TaskStatusWaiting
 	task.Shell = cmd
@@ -290,6 +291,7 @@ func (s *App) UninstallExporter(w http.ResponseWriter, r *http.Request) {
 	cmd := fmt.Sprintf(`curl -sSLm 10 --retry 3 'https://%s/prometheus/exporters/exporter.sh' | bash -s -- 'uninstall' '%s'`, s.conf.App.DownloadEndpoint, url.PathEscape(req.Slug))
 
 	task := new(biz.Task)
+	task.Key = "prometheus:exporter:" + req.Slug
 	task.Name = s.t.Get("Uninstall Prometheus exporter %s", req.Slug)
 	task.Status = biz.TaskStatusWaiting
 	task.Shell = cmd

internal/biz/app.go

@@ -213,6 +213,7 @@ func (uc *AppUsecase) Install(channel, slug string) error {
 	}
 
 	task := new(Task)
+	task.Key = "app:" + slug
 	task.Name = uc.t.Get("Install app %s", item.Name)
 	task.Status = TaskStatusWaiting
 	task.Shell = script
@@ -253,6 +254,7 @@ func (uc *AppUsecase) UnInstall(slug string) error {
 	}
 
 	task := new(Task)
+	task.Key = "app:" + slug
 	task.Name = uc.t.Get("Uninstall app %s", item.Name)
 	task.Status = TaskStatusWaiting
 	task.Shell = script
@@ -295,6 +297,7 @@ func (uc *AppUsecase) Update(slug string) error {
 	}
 
 	task := new(Task)
+	task.Key = "app:" + slug
 	task.Name = uc.t.Get("Update app %s", item.Name)
 	task.Status = TaskStatusWaiting
 	task.Shell = script

internal/biz/environment.go

@@ -157,6 +157,7 @@ func (uc *EnvironmentUsecase) do(typ, slug, action string) error {
 	}
 
 	task := new(Task)
+	task.Key = fmt.Sprintf("environment:%s:%s", typ, slug)
 	task.Name = name
 	task.Status = TaskStatusWaiting
 	task.Shell = cmd

internal/biz/task.go

@@ -5,20 +5,23 @@ import "time"
 type TaskStatus string
 
 const (
-	TaskStatusWaiting TaskStatus = "waiting"
-	TaskStatusRunning TaskStatus = "running"
-	TaskStatusSuccess TaskStatus = "finished"
-	TaskStatusFailed  TaskStatus = "failed"
+	TaskStatusWaiting  TaskStatus = "waiting"
+	TaskStatusRunning  TaskStatus = "running"
+	TaskStatusSuccess  TaskStatus = "finished"
+	TaskStatusFailed   TaskStatus = "failed"
+	TaskStatusCanceled TaskStatus = "canceled"
 )
 
 type Task struct {
-	ID        uint       `gorm:"primaryKey" json:"id"`
-	Name      string     `gorm:"not null;default:'';index" json:"name"`
-	Status    TaskStatus `gorm:"not null;default:'waiting'" json:"status"`
-	Shell     string     `gorm:"not null;default:''" json:"-"`
-	Log       string     `gorm:"not null;default:''" json:"log"`
-	CreatedAt time.Time  `json:"created_at"`
-	UpdatedAt time.Time  `json:"updated_at"`
+	ID          uint       `gorm:"primaryKey" json:"id"`
+	Key         string     `gorm:"not null;default:'';index" json:"-"` // 任务标识（域:动作:标的），用于防重
+	Name        string     `gorm:"not null;default:'';index" json:"name"`
+	Status      TaskStatus `gorm:"not null;default:'waiting'" json:"status"`
+	Shell       string     `gorm:"not null;default:''" json:"-"`
+	CancelShell string     `gorm:"not null;default:''" json:"-"` // 运行中被取消后执行的清理命令（可选）
+	Log         string     `gorm:"not null;default:''" json:"log"`
+	CreatedAt   time.Time  `json:"created_at"`
+	UpdatedAt   time.Time  `json:"updated_at"`
 }
 
 type TaskRepo interface {
@@ -26,6 +29,7 @@ type TaskRepo interface {
 	List(page, limit uint) ([]*Task, int64, error)
 	Get(id uint) (*Task, error)
 	Delete(id uint) error
+	Cancel(id uint) error
 	UpdateStatus(id uint, status TaskStatus) error
 	UpdateLog(id uint, log string) error
 	Push(task *Task) error
@@ -55,6 +59,10 @@ func (uc *TaskUsecase) Delete(id uint) error {
 	return uc.repo.Delete(id)
 }
 
+func (uc *TaskUsecase) Cancel(id uint) error {
+	return uc.repo.Cancel(id)
+}
+
 func (uc *TaskUsecase) UpdateStatus(id uint, status TaskStatus) error {
 	return uc.repo.UpdateStatus(id, status)
 }

internal/data/task.go

@@ -3,6 +3,7 @@ package data
 import (
 	"errors"
 	"log/slog"
+	"os"
 
 	"github.com/leonelquinteros/gotext"
 	"github.com/samber/do/v2"
@@ -48,7 +49,45 @@ func (r *taskRepo) Get(id uint) (*biz.Task, error) {
 }
 
 func (r *taskRepo) Delete(id uint) error {
-	return r.db.Model(&biz.Task{}).Where("id = ?", id).Delete(&biz.Task{}).Error
+	task, err := r.Get(id)
+	if err != nil {
+		return err
+	}
+	if task.Status == biz.TaskStatusWaiting || task.Status == biz.TaskStatusRunning {
+		return errors.New(r.t.Get("please cancel the task first"))
+	}
+
+	// 清理任务日志文件
+	if task.Log != "" {
+		_ = os.Remove(task.Log)
+	}
+
+	return r.db.Where("id = ?", id).Delete(&biz.Task{}).Error
+}
+
+func (r *taskRepo) Cancel(id uint) error {
+	// 等待中的任务直接原子标记取消，避免与运行器取任务竞争
+	result := r.db.Model(&biz.Task{}).Where("id = ? AND status = ?", id, biz.TaskStatusWaiting).Update("status", biz.TaskStatusCanceled)
+	if result.Error != nil {
+		return result.Error
+	}
+	if result.RowsAffected > 0 {
+		return nil
+	}
+
+	task, err := r.Get(id)
+	if err != nil {
+		return err
+	}
+	if task.Status != biz.TaskStatusRunning {
+		return errors.New(r.t.Get("task is not waiting or running"))
+	}
+
+	// 运行中的任务交由运行器杀死进程组
+	if !r.runner.Cancel(id) {
+		return errors.New(r.t.Get("task has already finished"))
+	}
+	return nil
 }
 
 func (r *taskRepo) UpdateStatus(id uint, status biz.TaskStatus) error {
@@ -60,13 +99,15 @@ func (r *taskRepo) UpdateLog(id uint, log string) error {
 }
 
 func (r *taskRepo) Push(task *biz.Task) error {
-	// 防止有人喜欢酒吧点炒饭
-	var count int64
-	if err := r.db.Model(&biz.Task{}).Where("shell = ? and (status = ? or status = ?)", task.Shell, biz.TaskStatusWaiting, biz.TaskStatusRunning).Count(&count).Error; err != nil {
-		return err
-	}
-	if count > 0 {
-		return errors.New(r.t.Get("duplicate submission, please wait for the previous task to end"))
+	// 防止有人喜欢酒吧点炒饭，按语言无关的任务标识去重
+	if task.Key != "" {
+		var count int64
+		if err := r.db.Model(&biz.Task{}).Where("`key` = ? and (status = ? or status = ?)", task.Key, biz.TaskStatusWaiting, biz.TaskStatusRunning).Count(&count).Error; err != nil {
+			return err
+		}
+		if count > 0 {
+			return errors.New(r.t.Get("duplicate submission, please wait for the previous task to end"))
+		}
 	}
 
 	if err := r.db.Create(task).Error; err != nil {

internal/data/task_test.go

@@ -0,0 +1,55 @@
+package data
+
+import (
+	"context"
+	"testing"
+
+	"github.com/leonelquinteros/gotext"
+	"github.com/libtnb/sqlite"
+	"gorm.io/gorm"
+
+	"github.com/acepanel/panel/v3/internal/biz"
+)
+
+type stubRunner struct{}
+
+func (stubRunner) Run(context.Context) {}
+func (stubRunner) Notify()             {}
+func (stubRunner) Cancel(uint) bool    { return false }
+
+func newTaskRepoForTest(t *testing.T) *taskRepo {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open("file::memory:"), &gorm.Config{SkipDefaultTransaction: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&biz.Task{}); err != nil {
+		t.Fatal(err)
+	}
+	return &taskRepo{t: gotext.NewLocale("", "en"), db: db, runner: stubRunner{}}
+}
+
+func TestTaskPushDedup(t *testing.T) {
+	repo := newTaskRepoForTest(t)
+
+	// shell 含随机片段也按 key 去重
+	if err := repo.Push(&biz.Task{Key: "backup:website:a", Name: "备份", Status: biz.TaskStatusWaiting, Shell: "echo 1"}); err != nil {
+		t.Fatalf("first Push: %v", err)
+	}
+	if err := repo.Push(&biz.Task{Key: "backup:website:a", Name: "Backup", Status: biz.TaskStatusWaiting, Shell: "echo 2"}); err == nil {
+		t.Fatal("duplicate key should be rejected")
+	}
+
+	// 不同 key 放行
+	if err := repo.Push(&biz.Task{Key: "backup:website:b", Name: "Backup", Status: biz.TaskStatusWaiting, Shell: "echo 3"}); err != nil {
+		t.Fatalf("different key should pass: %v", err)
+	}
+
+	// 终态后同 key 放行
+	if err := repo.db.Model(&biz.Task{}).Where("`key` = ?", "backup:website:a").Update("status", biz.TaskStatusCanceled).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := repo.Push(&biz.Task{Key: "backup:website:a", Name: "Backup", Status: biz.TaskStatusWaiting, Shell: "echo 4"}); err != nil {
+		t.Fatalf("Push after terminal status should pass: %v", err)
+	}
+}

internal/migration/v1.go

@@ -140,4 +140,13 @@ func init() {
 			return tx.Migrator().DropTable(&biz.FileShare{})
 		},
 	})
+	Migrations = append(Migrations, &gormigrate.Migration{
+		ID: "20260720-update-task-fields",
+		Migrate: func(tx *gorm.DB) error {
+			return tx.AutoMigrate(&biz.Task{})
+		},
+		Rollback: func(tx *gorm.DB) error {
+			return nil
+		},
+	})
 }

internal/route/task.go

@@ -19,5 +19,6 @@ func TaskRoutes(i do.Injector) (Endpoints, error) {
 		{Method: http.MethodGet, Path: "/api/task", Handler: svc.List, Summary: "获取任务列表", Tags: []string{"任务"}, Request: request.Paginate{}, Response: service.Envelope[service.Page[*biz.Task]]{}},
 		{Method: http.MethodGet, Path: "/api/task/{id}", Handler: svc.Get, Summary: "获取任务详情", Tags: []string{"任务"}, Request: request.ID{}, Response: service.Envelope[biz.Task]{}},
 		{Method: http.MethodDelete, Path: "/api/task/{id}", Handler: svc.Delete, Summary: "删除任务", Tags: []string{"任务"}, Request: request.ID{}},
+		{Method: http.MethodPost, Path: "/api/task/{id}/cancel", Handler: svc.Cancel, Summary: "取消任务", Tags: []string{"任务"}, Request: request.ID{}},
 	}, nil
 }

internal/service/backup.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/leonelquinteros/gotext"
 	"github.com/libtnb/chix/v2"
+	"github.com/libtnb/utils/str"
 	"github.com/samber/do/v2"
 
 	"github.com/acepanel/panel/v3/internal/biz"
@@ -61,17 +62,28 @@ func (s *BackupService) Create(w http.ResponseWriter, r *http.Request) {
 
 	// 备份可能耗时较长（大库），提交到后台任务队列异步执行
 	pathEnv := "export PATH=/bin:/sbin:/usr/bin:/usr/sbin:/usr/local/bin:/usr/local/sbin:$PATH\n"
-	var cmd string
+	var backupCmd string
 	if req.Type == "website" {
-		cmd = fmt.Sprintf("%sacepanel backup website -n '%s' -s '%d'", pathEnv, req.Target, req.Storage)
+		backupCmd = fmt.Sprintf("acepanel backup website -n '%s' -s '%d'", req.Target, req.Storage)
 	} else {
-		cmd = fmt.Sprintf("%sacepanel backup database -t '%s' -n '%s' -s '%d'", pathEnv, req.Type, req.Target, req.Storage)
+		backupCmd = fmt.Sprintf("acepanel backup database -t '%s' -n '%s' -s '%d'", req.Type, req.Target, req.Storage)
 	}
 
+	// 备份进程被杀后无法自清临时目录，包一层独享 TMPDIR 供取消时精确清理，不误伤并发的其他备份
+	tmpDir := fmt.Sprintf(`${TMPDIR:-/tmp}/ace-backup-task-%s`, str.Random(16))
+	cmd := fmt.Sprintf(`%sexport TMPDIR="%s"
+mkdir -p "$TMPDIR"
+%s
+rc=$?
+rm -rf "$TMPDIR"
+exit $rc`, pathEnv, tmpDir, backupCmd)
+
 	task := &biz.Task{
-		Name:   s.t.Get("Backup %s: %s", req.Type, req.Target),
-		Status: biz.TaskStatusWaiting,
-		Shell:  cmd,
+		Key:         fmt.Sprintf("backup:%s:%s", req.Type, req.Target),
+		Name:        s.t.Get("Backup %s: %s", req.Type, req.Target),
+		Status:      biz.TaskStatusWaiting,
+		Shell:       cmd,
+		CancelShell: fmt.Sprintf(`rm -rf "%s"`, tmpDir),
 	}
 	if err = s.taskRepo.Push(task); err != nil {
 		Error(w, http.StatusInternalServerError, "%v", err)

internal/service/environment_php.go

@@ -307,6 +307,7 @@ func (s *EnvironmentPHPService) InstallModule(w http.ResponseWriter, r *http.Req
 	}
 
 	task := new(biz.Task)
+	task.Key = fmt.Sprintf("php:module:%d:%s", req.Version, req.Slug)
 	task.Name = s.t.Get("Install PHP-%d %s module", req.Version, req.Slug)
 	task.Status = biz.TaskStatusWaiting
 	task.Shell = cmd
@@ -341,6 +342,7 @@ func (s *EnvironmentPHPService) UninstallModule(w http.ResponseWriter, r *http.R
 	}
 
 	task := new(biz.Task)
+	task.Key = fmt.Sprintf("php:module:%d:%s", req.Version, req.Slug)
 	task.Name = s.t.Get("Uninstall PHP-%d %s module", req.Version, req.Slug)
 	task.Status = biz.TaskStatusWaiting
 	task.Shell = cmd

internal/service/file.go

@@ -457,9 +457,11 @@ func (s *FileService) RemoteDownload(w http.ResponseWriter, r *http.Request) {
 	}
 
 	task := new(biz.Task)
+	task.Key = "download:" + req.Path
 	task.Name = s.t.Get("Download remote file %v", filepath.Base(req.Path))
 	task.Status = biz.TaskStatusWaiting
 	task.Shell = fmt.Sprintf(`aria2c -c --file-allocation=falloc --allow-overwrite=true --auto-file-renaming=false --check-certificate=false --retry-wait=5 --max-tries=5 -x 16 -s 16 -k 1M -d '%s' -o '%s' '%s' && chmod 0755 '%s' && chown www:www '%s'`, filepath.Dir(req.Path), filepath.Base(req.Path), req.URL, req.Path, req.Path)
+	task.CancelShell = fmt.Sprintf(`rm -f '%s' '%s.aria2'`, req.Path, req.Path)
 
 	if err = s.taskRepo.Push(task); err != nil {
 		Error(w, http.StatusInternalServerError, "%v", err)
@@ -585,9 +587,11 @@ func (s *FileService) Compress(w http.ResponseWriter, r *http.Request) {
 	}
 
 	task := new(biz.Task)
+	task.Key = "compress:" + req.File
 	task.Name = s.t.Get("Compress %v", filepath.Base(req.File))
 	task.Status = biz.TaskStatusWaiting
 	task.Shell = fmt.Sprintf(`%s && chmod 0755 '%s' && chown www:www '%s'`, cmd, req.File, req.File)
+	task.CancelShell = fmt.Sprintf(`rm -f '%s'`, req.File)
 
 	if err = s.taskRepo.Push(task); err != nil {
 		Error(w, http.StatusInternalServerError, "%v", err)
@@ -611,6 +615,7 @@ func (s *FileService) UnCompress(w http.ResponseWriter, r *http.Request) {
 	}
 
 	task := new(biz.Task)
+	task.Key = fmt.Sprintf("uncompress:%s:%s", req.File, req.Path)
 	task.Name = s.t.Get("Uncompress %v", filepath.Base(req.File))
 	task.Status = biz.TaskStatusWaiting
 	task.Shell = fmt.Sprintf(`%s && chmod -R 0755 '%s' && chown -R www:www '%s'`, cmd, req.Path, req.Path)

internal/service/task.go

@@ -76,3 +76,18 @@ func (s *TaskService) Delete(w http.ResponseWriter, r *http.Request) {
 
 	Success(w, nil)
 }
+
+func (s *TaskService) Cancel(w http.ResponseWriter, r *http.Request) {
+	req, err := Bind[request.ID](r)
+	if err != nil {
+		Error(w, http.StatusUnprocessableEntity, "%v", err)
+		return
+	}
+
+	if err = s.taskRepo.Cancel(req.ID); err != nil {
+		Error(w, http.StatusInternalServerError, "%v", err)
+		return
+	}
+
+	Success(w, nil)
+}

internal/taskqueue/runner.go

@@ -5,7 +5,9 @@ import (
 	"fmt"
 	"log/slog"
 	"os"
+	"os/exec"
 	"path/filepath"
+	"sync"
 	"time"
 
 	"gorm.io/gorm"
@@ -19,6 +21,10 @@ type Runner struct {
 	db     *gorm.DB
 	log    *slog.Logger
 	notify chan struct{}
+
+	mu            sync.Mutex
+	currentID     uint               // 当前运行的任务 ID
+	currentCancel context.CancelFunc // 取消当前任务
 }
 
 // NewRunner 创建任务运行器
@@ -38,6 +44,17 @@ func (r *Runner) Notify() {
 	}
 }
 
+// Cancel 取消正在运行的任务，返回是否命中
+func (r *Runner) Cancel(id uint) bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.currentID != id || r.currentCancel == nil {
+		return false
+	}
+	r.currentCancel()
+	return true
+}
+
 // Run 启动运行器
 func (r *Runner) Run(ctx context.Context) {
 	go func() {
@@ -70,7 +87,7 @@ func (r *Runner) drain(ctx context.Context) {
 			return
 		default:
 		}
-		if !r.processNext() {
+		if !r.processNext(ctx) {
 			return
 		}
 	}
@@ -84,20 +101,25 @@ func (r *Runner) clearZombie() {
 }
 
 // processNext 取一条 waiting 任务执行，返回是否有任务被处理
-func (r *Runner) processNext() bool {
+func (r *Runner) processNext(ctx context.Context) bool {
 	task := new(biz.Task)
 	if err := r.db.Where("status = ?", biz.TaskStatusWaiting).Order("id asc").First(task).Error; err != nil {
 		return false
 	}
 
-	r.execute(task)
+	r.execute(ctx, task)
 	return true
 }
 
 // execute 执行单个任务
-func (r *Runner) execute(task *biz.Task) {
-	if err := r.db.Model(task).Update("status", biz.TaskStatusRunning).Error; err != nil {
-		r.log.Error("failed to update task status to running", slog.Any("task_id", task.ID), slog.Any("err", err))
+func (r *Runner) execute(ctx context.Context, task *biz.Task) {
+	// 原子抢占，任务可能在取出后被取消
+	result := r.db.Model(task).Where("status = ?", biz.TaskStatusWaiting).Update("status", biz.TaskStatusRunning)
+	if result.Error != nil {
+		r.log.Error("failed to update task status to running", slog.Any("task_id", task.ID), slog.Any("err", result.Error))
+		return
+	}
+	if result.RowsAffected == 0 {
 		return
 	}
 
@@ -110,9 +132,27 @@ func (r *Runner) execute(task *biz.Task) {
 		return
 	}
 
-	if err := shell.ExecWithLog(task.Shell, logFile); err != nil {
-		r.log.Warn("failed to execute background task", slog.Any("task_id", task.ID), slog.Any("err", err))
-		_ = r.db.Model(task).Update("status", biz.TaskStatusFailed).Error
+	// 登记当前任务，供 Cancel 定位
+	taskCtx, cancel := context.WithCancel(ctx)
+	r.mu.Lock()
+	r.currentID, r.currentCancel = task.ID, cancel
+	r.mu.Unlock()
+	defer func() {
+		r.mu.Lock()
+		r.currentID, r.currentCancel = 0, nil
+		r.mu.Unlock()
+		cancel()
+	}()
+
+	if err := shell.ExecWithLog(taskCtx, task.Shell, logFile); err != nil {
+		// 用户取消标记为 canceled，面板停机保持 failed 由下次启动清理语义兜底
+		status := biz.TaskStatusFailed
+		if taskCtx.Err() != nil && ctx.Err() == nil {
+			status = biz.TaskStatusCanceled
+			r.runCancelShell(task, logFile)
+		}
+		r.log.Warn("background task did not finish", slog.Any("task_id", task.ID), slog.Any("status", status), slog.Any("err", err))
+		_ = r.db.Model(task).Update("status", status).Error
 		return
 	}
 
@@ -120,3 +160,27 @@ func (r *Runner) execute(task *biz.Task) {
 		r.log.Error("failed to update task status to success", slog.Any("task_id", task.ID), slog.Any("err", err))
 	}
 }
+
+// runCancelShell 任务被取消后执行清理命令，输出追加到任务日志
+func (r *Runner) runCancelShell(task *biz.Task, logFile string) {
+	if task.CancelShell == "" {
+		return
+	}
+
+	cleanupCtx, cancel := context.WithTimeout(context.Background(), time.Minute)
+	defer cancel()
+
+	f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
+	if err != nil {
+		r.log.Warn("failed to open task log for cancel shell", slog.Any("task_id", task.ID), slog.Any("err", err))
+		return
+	}
+	defer func(f *os.File) { _ = f.Close() }(f)
+
+	cmd := exec.CommandContext(cleanupCtx, "bash", "-c", task.CancelShell)
+	cmd.Stdout = f
+	cmd.Stderr = f
+	if err = cmd.Run(); err != nil {
+		r.log.Warn("failed to run task cancel shell", slog.Any("task_id", task.ID), slog.Any("err", err))
+	}
+}

internal/taskqueue/runner_test.go

@@ -0,0 +1,108 @@
+package taskqueue
+
+import (
+	"log/slog"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/libtnb/sqlite"
+	"gorm.io/gorm"
+
+	"github.com/acepanel/panel/v3/internal/biz"
+)
+
+func newRunnerForTest(t *testing.T) *Runner {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open("file::memory:"), &gorm.Config{SkipDefaultTransaction: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&biz.Task{}); err != nil {
+		t.Fatal(err)
+	}
+	return NewRunner(db, slog.New(slog.NewTextHandler(os.Stderr, nil)))
+}
+
+// 等待任务进入指定状态
+func waitStatus(t *testing.T, db *gorm.DB, id uint, status biz.TaskStatus, timeout time.Duration) *biz.Task {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	task := new(biz.Task)
+	for time.Now().Before(deadline) {
+		if err := db.First(task, id).Error; err == nil && task.Status == status {
+			return task
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatalf("task %d did not reach status %s, current: %s", id, status, task.Status)
+	return nil
+}
+
+func TestRunnerCancelRunning(t *testing.T) {
+	r := newRunnerForTest(t)
+	dir := t.TempDir()
+	marker := filepath.Join(dir, "cleanup.done")
+
+	task := &biz.Task{
+		Name:        "sleep",
+		Status:      biz.TaskStatusWaiting,
+		Shell:       "sleep 60",
+		CancelShell: "touch " + marker,
+	}
+	if err := r.db.Create(task).Error; err != nil {
+		t.Fatal(err)
+	}
+
+	r.Run(t.Context())
+
+	// 等待任务进入运行状态后取消
+	waitStatus(t, r.db, task.ID, biz.TaskStatusRunning, 3*time.Second)
+	if !r.Cancel(task.ID) {
+		t.Fatal("Cancel should hit the running task")
+	}
+
+	got := waitStatus(t, r.db, task.ID, biz.TaskStatusCanceled, 3*time.Second)
+	if got.Log == "" {
+		t.Fatal("task log path should be set")
+	}
+
+	// 取消清理命令应已执行
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if _, err := os.Stat(marker); err == nil {
+			return
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+	t.Fatal("cancel shell was not executed")
+}
+
+func TestRunnerCancelMiss(t *testing.T) {
+	r := newRunnerForTest(t)
+	if r.Cancel(1) {
+		t.Fatal("Cancel should miss when nothing is running")
+	}
+}
+
+func TestRunnerWaitingCanceledNotExecuted(t *testing.T) {
+	r := newRunnerForTest(t)
+
+	// 已被标记取消的任务不应被运行器抢占执行
+	task := &biz.Task{Name: "noop", Status: biz.TaskStatusCanceled, Shell: "true"}
+	if err := r.db.Create(task).Error; err != nil {
+		t.Fatal(err)
+	}
+
+	r.Run(t.Context())
+
+	time.Sleep(200 * time.Millisecond)
+	got := new(biz.Task)
+	if err := r.db.First(got, task.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if got.Status != biz.TaskStatusCanceled {
+		t.Fatalf("canceled task should stay canceled, got %s", got.Status)
+	}
+}

mocks/biz/TaskRepo.go

@@ -20,6 +20,52 @@ func (_m *TaskRepo) EXPECT() *TaskRepo_Expecter {
 	return &TaskRepo_Expecter{mock: &_m.Mock}
 }
 
+// Cancel provides a mock function with given fields: id
+func (_m *TaskRepo) Cancel(id uint) error {
+	ret := _m.Called(id)
+
+	if len(ret) == 0 {
+		panic("no return value specified for Cancel")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(uint) error); ok {
+		r0 = rf(id)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// TaskRepo_Cancel_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Cancel'
+type TaskRepo_Cancel_Call struct {
+	*mock.Call
+}
+
+// Cancel is a helper method to define mock.On call
+//   - id uint
+func (_e *TaskRepo_Expecter) Cancel(id interface{}) *TaskRepo_Cancel_Call {
+	return &TaskRepo_Cancel_Call{Call: _e.mock.On("Cancel", id)}
+}
+
+func (_c *TaskRepo_Cancel_Call) Run(run func(id uint)) *TaskRepo_Cancel_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(uint))
+	})
+	return _c
+}
+
+func (_c *TaskRepo_Cancel_Call) Return(_a0 error) *TaskRepo_Cancel_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *TaskRepo_Cancel_Call) RunAndReturn(run func(uint) error) *TaskRepo_Cancel_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Delete provides a mock function with given fields: id
 func (_m *TaskRepo) Delete(id uint) error {
 	ret := _m.Called(id)

pkg/shell/exec.go

@@ -10,6 +10,7 @@ import (
 	"os/exec"
 	"slices"
 	"strings"
+	"syscall"
 	"time"
 
 	"github.com/creack/pty"
@@ -32,7 +33,8 @@ func Exec(shell string) (string, error) {
 }
 
 // ExecWithLog 执行 shell 命令并将输出写入指定的日志文件
-func ExecWithLog(shell string, logFile string) error {
+// ctx 取消时会杀死整个进程组
+func ExecWithLog(ctx context.Context, shell string, logFile string) error {
 	_ = os.Setenv("LC_ALL", "C")
 
 	f, err := os.OpenFile(logFile, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0644)
@@ -41,11 +43,19 @@ func ExecWithLog(shell string, logFile string) error {
 	}
 	defer func(f *os.File) { _ = f.Close() }(f)
 
-	cmd := exec.Command("bash", "-c", shell)
+	cmd := exec.CommandContext(ctx, "bash", "-c", shell)
 	cmd.Stdout = f
 	cmd.Stderr = f
+	// 命令会派生子进程（下载、压缩等），放入独立进程组以便取消时整组杀死
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error {
+		return syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+	}
 
 	if err = cmd.Run(); err != nil {
+		if ctx.Err() != nil {
+			return ctx.Err()
+		}
 		return fmt.Errorf("run %s failed, err: %w", shell, err)
 	}
 

pkg/types/task.go

@@ -6,4 +6,5 @@ import "context"
 type TaskRunner interface {
 	Run(ctx context.Context)
 	Notify()
+	Cancel(id uint) bool
 }

web/src/api/panel/task/index.ts

@@ -9,4 +9,6 @@ export default {
   get: (id: number): any => http.Get(`/task/${id}`),
   // 删除任务
   delete: (id: number): any => http.Delete(`/task/${id}`),
+  // 取消任务
+  cancel: (id: number): any => http.Post(`/task/${id}/cancel`),
 }

web/src/views/task/TaskView.vue

@@ -9,7 +9,7 @@ import { useConfirm } from '@/components/system/composables/useConfirm'
 import { formatDateTime } from '@/utils'
 
 const { $gettext } = useGettext()
-const { confirmDelete } = useConfirm()
+const { confirmDelete, confirmAction } = useConfirm()
 const logModal = ref(false)
 const logPath = ref('')
 const logModalRef = ref<{ clear: () => void } | null>(null)
@@ -42,7 +42,9 @@ const columns: any = [
           ? $gettext('Waiting')
           : row.status === 'failed'
             ? $gettext('Failed')
-            : $gettext('Running')
+            : row.status === 'canceled'
+              ? $gettext('Canceled')
+              : $gettext('Running')
     },
   },
   {
@@ -70,7 +72,7 @@ const columns: any = [
     hideInExcel: true,
     render(row: any) {
       const items: any[] = []
-      if (row.status != 'waiting') {
+      if (row.log) {
         items.push(
           h(
             NButton,
@@ -87,7 +89,28 @@ const columns: any = [
           ),
         )
       }
-      if (row.status != 'waiting' && row.status != 'running') {
+      if (row.status == 'waiting' || row.status == 'running') {
+        items.push(
+          h(
+            NButton,
+            {
+              size: 'small',
+              type: 'error',
+              secondary: true,
+              onClick: async () => {
+                const ok = await confirmAction({
+                  title: $gettext('Cancel Task'),
+                  content: $gettext('Are you sure you want to cancel task %{ name }?', {
+                    name: row.name,
+                  }),
+                })
+                if (ok) handleCancel(row.id)
+              },
+            },
+            { default: () => $gettext('Cancel') },
+          ),
+        )
+      } else {
         items.push(
           h(
             NButton,
@@ -127,6 +150,13 @@ const handleDelete = (id: number) => {
   })
 }
 
+const handleCancel = (id: number) => {
+  useRequest(task.cancel(id)).onSuccess(() => {
+    refresh()
+    window.$message.success($gettext('Canceled successfully'))
+  })
+}
+
 onMounted(() => {
   refresh()
 })