From b4765461a0bf06b10eb1a8a6b102b3e4281be605 Mon Sep 17 00:00:00 2001 From: Irina Truong Date: Tue, 2 Jun 2026 19:03:59 -0700 Subject: [PATCH 1/3] Add Cancelled phase for task cancellation Add TaskPhaseCancelled as a terminal phase and CancelledBy field in TaskStatus. When a task is set to Cancelled, the controller deletes its Job, releases branch locks, and handles TTL cleanup. Dependent tasks that reference a Cancelled dependency are failed immediately. Add `kelos cancel task ` CLI command that patches a task's status to Cancelled with cancelledBy=user. Update all terminal-phase checks across spawner, taskspawner controller, CLI (get --phase, logs), and reporting (GitHub comments, Check Runs, Slack) to include the new Cancelled phase. Ref: kelos-dev/kelos#765 Co-Authored-By: Claude Opus 4.6 --- api/v1alpha1/task_types.go | 7 + cmd/kelos-spawner/main.go | 8 +- internal/cli/cancel.go | 78 ++++++ internal/cli/cancel_test.go | 34 +++ internal/cli/get.go | 5 +- internal/cli/logs.go | 2 +- internal/cli/root.go | 1 + internal/controller/cancel_test.go | 256 ++++++++++++++++++ internal/controller/task_controller.go | 52 +++- internal/controller/taskspawner_controller.go | 4 +- .../charts/kelos/templates/crds/task-crd.yaml | 5 + internal/manifests/install-crd.yaml | 5 + internal/reporting/github.go | 5 + internal/reporting/slack.go | 6 +- internal/reporting/watcher.go | 18 +- 15 files changed, 470 insertions(+), 16 deletions(-) create mode 100644 internal/cli/cancel.go create mode 100644 internal/cli/cancel_test.go create mode 100644 internal/controller/cancel_test.go diff --git a/api/v1alpha1/task_types.go b/api/v1alpha1/task_types.go index 75c2b4be3..8c7147347 100644 --- a/api/v1alpha1/task_types.go +++ b/api/v1alpha1/task_types.go @@ -32,6 +32,8 @@ const ( TaskPhaseFailed TaskPhase = "Failed" // TaskPhaseWaiting means the Task is waiting for dependencies or branch lock. TaskPhaseWaiting TaskPhase = "Waiting" + // TaskPhaseCancelled means the Task was cancelled before completing. + TaskPhaseCancelled TaskPhase = "Cancelled" ) // SecretReference refers to a Secret containing credentials. @@ -245,6 +247,11 @@ type TaskStatus struct { // Results contains structured key-value outputs produced by the agent. // +optional Results map[string]string `json:"results,omitempty"` + + // CancelledBy records what initiated cancellation (e.g., "user", + // "spawner:my-spawner"). Only set when Phase is Cancelled. + // +optional + CancelledBy string `json:"cancelledBy,omitempty"` } // +genclient diff --git a/cmd/kelos-spawner/main.go b/cmd/kelos-spawner/main.go index e31adfe39..dc9eee4e2 100644 --- a/cmd/kelos-spawner/main.go +++ b/cmd/kelos-spawner/main.go @@ -293,7 +293,9 @@ func runCycleWithSourceCore(ctx context.Context, cl client.Client, key types.Nam for i := range existingTaskList.Items { t := &existingTaskList.Items[i] existingTaskMap[t.Name] = t - if t.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && t.Status.Phase != kelosv1alpha1.TaskPhaseFailed { + if t.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && + t.Status.Phase != kelosv1alpha1.TaskPhaseFailed && + t.Status.Phase != kelosv1alpha1.TaskPhaseCancelled { activeTasks++ } } @@ -314,7 +316,9 @@ func runCycleWithSourceCore(ctx context.Context, cl client.Client, key types.Nam // the item will be picked up as new on the next cycle since the old task // no longer exists. if !item.TriggerTime.IsZero() && - (existing.Status.Phase == kelosv1alpha1.TaskPhaseSucceeded || existing.Status.Phase == kelosv1alpha1.TaskPhaseFailed) && + (existing.Status.Phase == kelosv1alpha1.TaskPhaseSucceeded || + existing.Status.Phase == kelosv1alpha1.TaskPhaseFailed || + existing.Status.Phase == kelosv1alpha1.TaskPhaseCancelled) && existing.Status.CompletionTime != nil && item.TriggerTime.After(existing.Status.CompletionTime.Time) { diff --git a/internal/cli/cancel.go b/internal/cli/cancel.go new file mode 100644 index 000000000..555a3c942 --- /dev/null +++ b/internal/cli/cancel.go @@ -0,0 +1,78 @@ +package cli + +import ( + "context" + "fmt" + "os" + + "github.com/spf13/cobra" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + kelosv1alpha1 "github.com/kelos-dev/kelos/api/v1alpha1" +) + +func newCancelCommand(cfg *ClientConfig) *cobra.Command { + cmd := &cobra.Command{ + Use: "cancel", + Short: "Cancel resources", + RunE: func(cmd *cobra.Command, args []string) error { + cmd.Help() + return fmt.Errorf("must specify a resource type") + }, + } + + cmd.AddCommand(newCancelTaskCommand(cfg)) + + return cmd +} + +func newCancelTaskCommand(cfg *ClientConfig) *cobra.Command { + cmd := &cobra.Command{ + Use: "task ", + Short: "Cancel a running task", + Args: func(cmd *cobra.Command, args []string) error { + if len(args) == 0 { + return fmt.Errorf("task name is required\nUsage: %s", cmd.Use) + } + if len(args) > 1 { + return fmt.Errorf("too many arguments: expected 1 task name, got %d\nUsage: %s", len(args), cmd.Use) + } + return nil + }, + RunE: func(cmd *cobra.Command, args []string) error { + cl, ns, err := cfg.NewClient() + if err != nil { + return err + } + + ctx := context.Background() + taskName := args[0] + + var task kelosv1alpha1.Task + if err := cl.Get(ctx, client.ObjectKey{Namespace: ns, Name: taskName}, &task); err != nil { + return fmt.Errorf("getting task %s: %w", taskName, err) + } + + switch task.Status.Phase { + case kelosv1alpha1.TaskPhaseSucceeded, kelosv1alpha1.TaskPhaseFailed, kelosv1alpha1.TaskPhaseCancelled: + fmt.Fprintf(os.Stdout, "task/%s is already in terminal phase %s\n", taskName, task.Status.Phase) + return nil + } + + now := metav1.Now() + task.Status.Phase = kelosv1alpha1.TaskPhaseCancelled + task.Status.Message = "Cancelled by user" + task.Status.CancelledBy = "user" + task.Status.CompletionTime = &now + if err := cl.Status().Update(ctx, &task); err != nil { + return fmt.Errorf("cancelling task %s: %w", taskName, err) + } + + fmt.Fprintf(os.Stdout, "task/%s cancelled\n", taskName) + return nil + }, + } + + return cmd +} diff --git a/internal/cli/cancel_test.go b/internal/cli/cancel_test.go new file mode 100644 index 000000000..a01441684 --- /dev/null +++ b/internal/cli/cancel_test.go @@ -0,0 +1,34 @@ +package cli + +import ( + "testing" +) + +func TestCancelCommandRegistered(t *testing.T) { + root := NewRootCommand() + cmd := findSubcommand(t, root, []string{"cancel", "task"}) + if cmd == nil { + t.Fatal("expected cancel task subcommand to be registered") + } + if cmd.Use != "task " { + t.Errorf("expected Use to be 'task ', got %q", cmd.Use) + } +} + +func TestCancelTaskCommand_RequiresName(t *testing.T) { + root := NewRootCommand() + cmd := findSubcommand(t, root, []string{"cancel", "task"}) + err := cmd.Args(cmd, []string{}) + if err == nil { + t.Error("Expected error when no task name provided") + } +} + +func TestCancelTaskCommand_RejectsTooManyArgs(t *testing.T) { + root := NewRootCommand() + cmd := findSubcommand(t, root, []string{"cancel", "task"}) + err := cmd.Args(cmd, []string{"task1", "task2"}) + if err == nil { + t.Error("Expected error when too many args provided") + } +} diff --git a/internal/cli/get.go b/internal/cli/get.go index 3d253e39e..1af69241b 100644 --- a/internal/cli/get.go +++ b/internal/cli/get.go @@ -192,7 +192,7 @@ func newGetTaskCommand(cfg *ClientConfig, allNamespaces *bool) *cobra.Command { cmd.Flags().StringVarP(&output, "output", "o", "", "Output format (yaml or json)") cmd.Flags().BoolVarP(&detail, "detail", "d", false, "Show detailed information for a specific task") - cmd.Flags().StringSliceVar(&phases, "phase", nil, "Filter tasks by phase (Pending, Running, Waiting, Succeeded, Failed)") + cmd.Flags().StringSliceVar(&phases, "phase", nil, "Filter tasks by phase (Pending, Running, Waiting, Succeeded, Failed, Cancelled)") cmd.ValidArgsFunction = completeTaskNames(cfg) _ = cmd.RegisterFlagCompletionFunc("output", cobra.FixedCompletions([]string{"yaml", "json"}, cobra.ShellCompDirectiveNoFileComp)) @@ -288,12 +288,13 @@ var validTaskPhases = map[kelosv1alpha1.TaskPhase]bool{ kelosv1alpha1.TaskPhaseWaiting: true, kelosv1alpha1.TaskPhaseSucceeded: true, kelosv1alpha1.TaskPhaseFailed: true, + kelosv1alpha1.TaskPhaseCancelled: true, } func validatePhases(phases []string) error { for _, p := range phases { if !validTaskPhases[kelosv1alpha1.TaskPhase(p)] { - return fmt.Errorf("unknown phase %q: must be one of Pending, Running, Waiting, Succeeded, Failed", p) + return fmt.Errorf("unknown phase %q: must be one of Pending, Running, Waiting, Succeeded, Failed, Cancelled", p) } } return nil diff --git a/internal/cli/logs.go b/internal/cli/logs.go index 61765dc86..10c4c2fc5 100644 --- a/internal/cli/logs.go +++ b/internal/cli/logs.go @@ -156,7 +156,7 @@ func resolveTaskPodName(ctx context.Context, cl client.Client, namespace string, } func isTerminalTaskPhase(phase kelosv1alpha1.TaskPhase) bool { - return phase == kelosv1alpha1.TaskPhaseSucceeded || phase == kelosv1alpha1.TaskPhaseFailed + return phase == kelosv1alpha1.TaskPhaseSucceeded || phase == kelosv1alpha1.TaskPhaseFailed || phase == kelosv1alpha1.TaskPhaseCancelled } func streamLogs(ctx context.Context, cs *kubernetes.Clientset, namespace, podName, container string, follow bool) error { diff --git a/internal/cli/root.go b/internal/cli/root.go index d73d5eca9..e8db1c634 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -44,6 +44,7 @@ func NewRootCommand() *cobra.Command { newGetCommand(cfg), newLogsCommand(cfg), newDeleteCommand(cfg), + newCancelCommand(cfg), newSuspendCommand(cfg), newResumeCommand(cfg), newInitCommand(cfg), diff --git a/internal/controller/cancel_test.go b/internal/controller/cancel_test.go new file mode 100644 index 000000000..dbd5878e4 --- /dev/null +++ b/internal/controller/cancel_test.go @@ -0,0 +1,256 @@ +package controller + +import ( + "context" + "testing" + + batchv1 "k8s.io/api/batch/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/tools/record" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + kelosv1alpha1 "github.com/kelos-dev/kelos/api/v1alpha1" +) + +func TestTaskReconciler_CancelledTaskDeletesJob(t *testing.T) { + scheme := newTestScheme() + + task := &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cancel-me", + Namespace: "default", + Finalizers: []string{taskFinalizer}, + }, + Spec: kelosv1alpha1.TaskSpec{ + Type: "claude-code", + Prompt: "do something", + Credentials: kelosv1alpha1.Credentials{ + Type: kelosv1alpha1.CredentialTypeNone, + }, + }, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseCancelled, + CancelledBy: "user", + }, + } + + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cancel-me", + Namespace: "default", + }, + } + + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(task, job). + WithStatusSubresource(task). + Build() + + r := &TaskReconciler{ + Client: fakeClient, + Scheme: scheme, + BranchLocker: NewBranchLocker(), + Recorder: record.NewFakeRecorder(10), + } + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "cancel-me", Namespace: "default"}, + }) + if err != nil { + t.Fatalf("Reconcile() returned error: %v", err) + } + + // Verify Job was deleted. + var updatedJob batchv1.Job + err = fakeClient.Get(context.Background(), types.NamespacedName{Name: "cancel-me", Namespace: "default"}, &updatedJob) + if err == nil { + t.Error("Expected Job to be deleted after task cancellation") + } +} + +func TestTaskReconciler_CancelledTaskReleasesBranchLock(t *testing.T) { + scheme := newTestScheme() + + workspace := &kelosv1alpha1.Workspace{ + ObjectMeta: metav1.ObjectMeta{ + Name: "my-workspace", + Namespace: "default", + }, + Spec: kelosv1alpha1.WorkspaceSpec{ + Repo: "https://github.com/org/repo.git", + }, + } + + task := &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cancel-branch", + Namespace: "default", + Finalizers: []string{taskFinalizer}, + }, + Spec: kelosv1alpha1.TaskSpec{ + Type: "claude-code", + Prompt: "do something", + Branch: "feat/test", + Credentials: kelosv1alpha1.Credentials{ + Type: kelosv1alpha1.CredentialTypeNone, + }, + WorkspaceRef: &kelosv1alpha1.WorkspaceReference{ + Name: "my-workspace", + }, + }, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseCancelled, + CancelledBy: "user", + }, + } + + job := &batchv1.Job{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cancel-branch", + Namespace: "default", + }, + } + + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(task, job, workspace). + WithStatusSubresource(task). + Build() + + bl := NewBranchLocker() + lockKey := branchLockKey(task) + bl.TryAcquire(lockKey, task.Name) + + r := &TaskReconciler{ + Client: fakeClient, + Scheme: scheme, + BranchLocker: bl, + Recorder: record.NewFakeRecorder(10), + } + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "cancel-branch", Namespace: "default"}, + }) + if err != nil { + t.Fatalf("Reconcile() returned error: %v", err) + } + + // Verify branch lock was released — another task should be able to acquire it. + acquired, _ := bl.TryAcquire(lockKey, "another-task") + if !acquired { + t.Error("Expected branch lock to be released after task cancellation") + } +} + +func TestTaskReconciler_CancelledTaskWithNoJob(t *testing.T) { + scheme := newTestScheme() + + now := metav1.Now() + task := &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "already-cancelled", + Namespace: "default", + Finalizers: []string{taskFinalizer}, + }, + Spec: kelosv1alpha1.TaskSpec{ + Type: "claude-code", + Prompt: "do something", + Credentials: kelosv1alpha1.Credentials{ + Type: kelosv1alpha1.CredentialTypeNone, + }, + }, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseCancelled, + CancelledBy: "user", + CompletionTime: &now, + }, + } + + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(task). + WithStatusSubresource(task). + Build() + + r := &TaskReconciler{ + Client: fakeClient, + Scheme: scheme, + BranchLocker: NewBranchLocker(), + Recorder: record.NewFakeRecorder(10), + } + + // Should not error even when no Job exists. + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "already-cancelled", Namespace: "default"}, + }) + if err != nil { + t.Fatalf("Reconcile() returned error: %v", err) + } +} + +func TestTTLExpired_IncludesCancelled(t *testing.T) { + scheme := newTestScheme() + r := &TaskReconciler{Scheme: scheme} + + ttl := int32(0) + now := metav1.Now() + task := &kelosv1alpha1.Task{ + Spec: kelosv1alpha1.TaskSpec{ + TTLSecondsAfterFinished: &ttl, + }, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseCancelled, + CompletionTime: &now, + }, + } + + expired, _ := r.ttlExpired(task) + if !expired { + t.Error("Expected TTL to be expired for Cancelled task with TTL=0") + } +} + +// Verify that the reconciler handles a Cancelled task that is still Pending +// (i.e., the Job never started). This covers the case where a user cancels +// before the agent even begins. +func TestTaskReconciler_CancelsPendingTask(t *testing.T) { + scheme := newTestScheme() + + task := &kelosv1alpha1.Task{ + ObjectMeta: metav1.ObjectMeta{ + Name: "pending-cancel", + Namespace: "default", + Finalizers: []string{taskFinalizer}, + }, + Spec: kelosv1alpha1.TaskSpec{ + Type: "claude-code", + Prompt: "do something", + Credentials: kelosv1alpha1.Credentials{ + Type: kelosv1alpha1.CredentialTypeNone, + }, + }, + Status: kelosv1alpha1.TaskStatus{ + Phase: kelosv1alpha1.TaskPhaseCancelled, + CancelledBy: "user", + }, + } + + fakeClient := fake.NewClientBuilder().WithScheme(scheme). + WithObjects(task). + WithStatusSubresource(task). + Build() + + r := &TaskReconciler{ + Client: fakeClient, + Scheme: scheme, + BranchLocker: NewBranchLocker(), + Recorder: record.NewFakeRecorder(10), + } + + _, err := r.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "pending-cancel", Namespace: "default"}, + }) + if err != nil { + t.Fatalf("Reconcile() returned error: %v", err) + } +} diff --git a/internal/controller/task_controller.go b/internal/controller/task_controller.go index de48a5c02..bef1dbc6e 100644 --- a/internal/controller/task_controller.go +++ b/internal/controller/task_controller.go @@ -103,6 +103,42 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. } } + // Handle externally-cancelled tasks: delete the Job and clean up. + if task.Status.Phase == kelosv1alpha1.TaskPhaseCancelled { + if jobExists { + logger.Info("Task cancelled, deleting Job", "task", task.Name) + propagationPolicy := metav1.DeletePropagationBackground + if err := r.Delete(ctx, &job, &client.DeleteOptions{ + PropagationPolicy: &propagationPolicy, + }); err != nil && !apierrors.IsNotFound(err) { + return ctrl.Result{}, err + } + taskCompletedTotal.WithLabelValues(task.Namespace, task.Spec.Type, string(kelosv1alpha1.TaskPhaseCancelled)).Inc() + r.recordEvent(&task, corev1.EventTypeNormal, "TaskCancelled", "Task was cancelled") + } else if task.Status.JobName == "" { + // Task was cancelled before a Job was ever created (e.g., while + // Waiting on dependencies). Record metric on first reconcile only — + // JobName being empty distinguishes this from a post-deletion reconcile. + taskCompletedTotal.WithLabelValues(task.Namespace, task.Spec.Type, string(kelosv1alpha1.TaskPhaseCancelled)).Inc() + r.recordEvent(&task, corev1.EventTypeNormal, "TaskCancelled", "Task was cancelled") + } + if task.Spec.Branch != "" { + r.BranchLocker.Release(branchLockKey(&task), task.Name) + } + // Check TTL expiration for cancelled tasks. + if expired, requeueAfter := r.ttlExpired(&task); expired { + logger.Info("Deleting Task due to TTL expiration", "task", task.Name) + r.recordEvent(&task, corev1.EventTypeNormal, "TaskExpired", "Deleting Task due to TTL expiration") + if err := r.Delete(ctx, &task); err != nil && !apierrors.IsNotFound(err) { + return ctrl.Result{}, err + } + return ctrl.Result{}, nil + } else if requeueAfter > 0 { + return ctrl.Result{RequeueAfter: requeueAfter}, nil + } + return ctrl.Result{}, nil + } + // Create Job if it doesn't exist if !jobExists { if len(task.Spec.DependsOn) > 0 { @@ -634,7 +670,9 @@ func (r *TaskReconciler) ttlExpired(task *kelosv1alpha1.Task) (bool, time.Durati if task.Spec.TTLSecondsAfterFinished == nil { return false, 0 } - if task.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && task.Status.Phase != kelosv1alpha1.TaskPhaseFailed { + if task.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && + task.Status.Phase != kelosv1alpha1.TaskPhaseFailed && + task.Status.Phase != kelosv1alpha1.TaskPhaseCancelled { return false, 0 } if task.Status.CompletionTime == nil { @@ -726,14 +764,14 @@ func (r *TaskReconciler) checkDependencies(ctx context.Context, task *kelosv1alp return false, ctrl.Result{}, err } - if depTask.Status.Phase == kelosv1alpha1.TaskPhaseFailed { - logger.Info("Dependency failed", "dependency", depName) + if depTask.Status.Phase == kelosv1alpha1.TaskPhaseFailed || depTask.Status.Phase == kelosv1alpha1.TaskPhaseCancelled { + logger.Info("Dependency failed or cancelled", "dependency", depName, "phase", depTask.Status.Phase) updateErr := retry.RetryOnConflict(retry.DefaultRetry, func() error { if getErr := r.Get(ctx, client.ObjectKeyFromObject(task), task); getErr != nil { return getErr } task.Status.Phase = kelosv1alpha1.TaskPhaseFailed - task.Status.Message = fmt.Sprintf("Dependency %q failed", depName) + task.Status.Message = fmt.Sprintf("Dependency %q is %s", depName, string(depTask.Status.Phase)) now := metav1.Now() task.Status.CompletionTime = &now return r.Status().Update(ctx, task) @@ -741,7 +779,7 @@ func (r *TaskReconciler) checkDependencies(ctx context.Context, task *kelosv1alp if updateErr != nil { logger.Error(updateErr, "Unable to update Task status") } - r.recordEvent(task, corev1.EventTypeWarning, "DependencyFailed", "Dependency %q failed", depName) + r.recordEvent(task, corev1.EventTypeWarning, "DependencyFailed", "Dependency %q is %s", depName, string(depTask.Status.Phase)) taskCompletedTotal.WithLabelValues(task.Namespace, task.Spec.Type, string(kelosv1alpha1.TaskPhaseFailed)).Inc() return false, ctrl.Result{}, nil } @@ -947,7 +985,9 @@ func (r *TaskReconciler) enqueueDependentTasks(ctx context.Context, obj client.O } // Only trigger when a task reaches a terminal phase - if task.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && task.Status.Phase != kelosv1alpha1.TaskPhaseFailed { + if task.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && + task.Status.Phase != kelosv1alpha1.TaskPhaseFailed && + task.Status.Phase != kelosv1alpha1.TaskPhaseCancelled { return nil } diff --git a/internal/controller/taskspawner_controller.go b/internal/controller/taskspawner_controller.go index bfe92a6b9..18650e744 100644 --- a/internal/controller/taskspawner_controller.go +++ b/internal/controller/taskspawner_controller.go @@ -179,7 +179,9 @@ func (r *TaskSpawnerReconciler) countActiveTasks(ctx context.Context, taskSpawne for i := range taskList.Items { task := &taskList.Items[i] // Count tasks that are not in terminal phases - if task.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && task.Status.Phase != kelosv1alpha1.TaskPhaseFailed { + if task.Status.Phase != kelosv1alpha1.TaskPhaseSucceeded && + task.Status.Phase != kelosv1alpha1.TaskPhaseFailed && + task.Status.Phase != kelosv1alpha1.TaskPhaseCancelled { activeTasks++ } } diff --git a/internal/manifests/charts/kelos/templates/crds/task-crd.yaml b/internal/manifests/charts/kelos/templates/crds/task-crd.yaml index 372b812bf..7826082ee 100644 --- a/internal/manifests/charts/kelos/templates/crds/task-crd.yaml +++ b/internal/manifests/charts/kelos/templates/crds/task-crd.yaml @@ -3843,6 +3843,11 @@ spec: status: description: TaskStatus defines the observed state of Task. properties: + cancelledBy: + description: |- + CancelledBy records what initiated cancellation (e.g., "user", + "spawner:my-spawner"). Only set when Phase is Cancelled. + type: string completionTime: description: CompletionTime is when the Task completed. format: date-time diff --git a/internal/manifests/install-crd.yaml b/internal/manifests/install-crd.yaml index 24b0a14e9..5babc65cb 100644 --- a/internal/manifests/install-crd.yaml +++ b/internal/manifests/install-crd.yaml @@ -4067,6 +4067,11 @@ spec: status: description: TaskStatus defines the observed state of Task. properties: + cancelledBy: + description: |- + CancelledBy records what initiated cancellation (e.g., "user", + "spawner:my-spawner"). Only set when Phase is Cancelled. + type: string completionTime: description: CompletionTime is when the Task completed. format: date-time diff --git a/internal/reporting/github.go b/internal/reporting/github.go index 7d5ee473a..6b78b80d9 100644 --- a/internal/reporting/github.go +++ b/internal/reporting/github.go @@ -144,3 +144,8 @@ func FormatSucceededComment(taskName string) string { func FormatFailedComment(taskName string) string { return fmt.Sprintf("🤖 **Kelos Task Status**\n\nTask `%s` has **failed**. ❌", taskName) } + +// FormatCancelledComment returns the comment body for a cancelled task. +func FormatCancelledComment(taskName string) string { + return fmt.Sprintf("🤖 **Kelos Task Status**\n\nTask `%s` has been **cancelled**. 🚫", taskName) +} diff --git a/internal/reporting/slack.go b/internal/reporting/slack.go index dfcaced01..adffd1410 100644 --- a/internal/reporting/slack.go +++ b/internal/reporting/slack.go @@ -88,8 +88,9 @@ func contextBlock(taskName string) *slack.ContextBlock { // phaseHeaderText maps each phase to its leading Block Kit section text. // Phases without an entry (e.g. "succeeded") get no header block. var phaseHeaderText = map[string]string{ - "accepted": ":hourglass_flowing_sand: *Working on your request...*", - "failed": ":x: *Something went wrong*", + "accepted": ":hourglass_flowing_sand: *Working on your request...*", + "failed": ":x: *Something went wrong*", + "cancelled": ":stop_button: *Task cancelled*", } // phaseFallbackText maps each phase to its default fallback text (before the @@ -98,6 +99,7 @@ var phaseFallbackText = map[string]string{ "accepted": "Working on your request...", "succeeded": "Done!", "failed": "Failed.", + "cancelled": "Cancelled.", } // FormatProgressMessage returns a Slack message with Block Kit blocks for diff --git a/internal/reporting/watcher.go b/internal/reporting/watcher.go index f3835d60c..b86a17c46 100644 --- a/internal/reporting/watcher.go +++ b/internal/reporting/watcher.go @@ -215,6 +215,8 @@ func (tr *TaskReporter) reportViaComment(ctx context.Context, task *kelosv1alpha desiredPhase = "succeeded" case kelosv1alpha1.TaskPhaseFailed: desiredPhase = "failed" + case kelosv1alpha1.TaskPhaseCancelled: + desiredPhase = "cancelled" default: return nil } @@ -266,6 +268,8 @@ func (tr *TaskReporter) reportViaComment(ctx context.Context, task *kelosv1alpha body = FormatSucceededComment(task.Name) case "failed": body = FormatFailedComment(task.Name) + case "cancelled": + body = FormatCancelledComment(task.Name) } if commentID == 0 { @@ -337,6 +341,14 @@ func (tr *TaskReporter) reportViaCheckRun(ctx context.Context, task *kelosv1alph Title: checkName + " — Failed", Summary: fmt.Sprintf("Agent task `%s` has failed", task.Name), } + case kelosv1alpha1.TaskPhaseCancelled: + desiredPhase = "cancelled" + status = "completed" + conclusion = "cancelled" + output = &checkRunOutput{ + Title: checkName + " — Cancelled", + Summary: fmt.Sprintf("Agent task `%s` has been cancelled", task.Name), + } default: return nil } @@ -509,6 +521,8 @@ func (tr *SlackTaskReporter) ReportTaskStatus(ctx context.Context, task *kelosv1 desiredPhase = "succeeded" case kelosv1alpha1.TaskPhaseFailed: desiredPhase = "failed" + case kelosv1alpha1.TaskPhaseCancelled: + desiredPhase = "cancelled" default: return nil } @@ -529,7 +543,7 @@ func (tr *SlackTaskReporter) ReportTaskStatus(ctx context.Context, task *kelosv1 // thread compact. When the response was split into multiple messages, // replace the progress message with the first part and post the rest // as new replies. - if desiredPhase == "succeeded" || desiredPhase == "failed" { + if desiredPhase == "succeeded" || desiredPhase == "failed" || desiredPhase == "cancelled" { if progressTS := tr.getProgressTS(task.UID); progressTS != "" { log.Info("Updating Slack progress message with final result", "task", task.Name, "channel", channel, "phase", desiredPhase) if err := tr.Reporter.UpdateMessage(ctx, channel, progressTS, msgs[0]); err != nil { @@ -567,7 +581,7 @@ func (tr *SlackTaskReporter) ReportTaskStatus(ctx context.Context, task *kelosv1 } // Clean up caches when reporting a terminal phase. - if desiredPhase == "succeeded" || desiredPhase == "failed" { + if desiredPhase == "succeeded" || desiredPhase == "failed" || desiredPhase == "cancelled" { tr.clearProgressCache(task.UID) tr.clearActivityState(task.UID) } From 48f65921bd779ddfe086fc6121aa4d6749c201a2 Mon Sep 17 00:00:00 2001 From: Irina Truong Date: Tue, 2 Jun 2026 19:11:49 -0700 Subject: [PATCH 2/3] fix: address review feedback on cancel command - Use apierrors.IsNotFound for strict Job deletion assertion in test - Add retry.RetryOnConflict to CLI cancel status update for conflict resilience Co-Authored-By: Claude Opus 4.6 --- internal/cli/cancel.go | 33 ++++++++++++++++-------------- internal/controller/cancel_test.go | 5 +++-- 2 files changed, 21 insertions(+), 17 deletions(-) diff --git a/internal/cli/cancel.go b/internal/cli/cancel.go index 555a3c942..21053d67c 100644 --- a/internal/cli/cancel.go +++ b/internal/cli/cancel.go @@ -7,6 +7,7 @@ import ( "github.com/spf13/cobra" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/retry" "sigs.k8s.io/controller-runtime/pkg/client" kelosv1alpha1 "github.com/kelos-dev/kelos/api/v1alpha1" @@ -49,23 +50,25 @@ func newCancelTaskCommand(cfg *ClientConfig) *cobra.Command { ctx := context.Background() taskName := args[0] - var task kelosv1alpha1.Task - if err := cl.Get(ctx, client.ObjectKey{Namespace: ns, Name: taskName}, &task); err != nil { - return fmt.Errorf("getting task %s: %w", taskName, err) - } + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { + var task kelosv1alpha1.Task + if err := cl.Get(ctx, client.ObjectKey{Namespace: ns, Name: taskName}, &task); err != nil { + return err + } - switch task.Status.Phase { - case kelosv1alpha1.TaskPhaseSucceeded, kelosv1alpha1.TaskPhaseFailed, kelosv1alpha1.TaskPhaseCancelled: - fmt.Fprintf(os.Stdout, "task/%s is already in terminal phase %s\n", taskName, task.Status.Phase) - return nil - } + switch task.Status.Phase { + case kelosv1alpha1.TaskPhaseSucceeded, kelosv1alpha1.TaskPhaseFailed, kelosv1alpha1.TaskPhaseCancelled: + fmt.Fprintf(os.Stdout, "task/%s is already in terminal phase %s\n", taskName, task.Status.Phase) + return nil + } - now := metav1.Now() - task.Status.Phase = kelosv1alpha1.TaskPhaseCancelled - task.Status.Message = "Cancelled by user" - task.Status.CancelledBy = "user" - task.Status.CompletionTime = &now - if err := cl.Status().Update(ctx, &task); err != nil { + now := metav1.Now() + task.Status.Phase = kelosv1alpha1.TaskPhaseCancelled + task.Status.Message = "Cancelled by user" + task.Status.CancelledBy = "user" + task.Status.CompletionTime = &now + return cl.Status().Update(ctx, &task) + }); err != nil { return fmt.Errorf("cancelling task %s: %w", taskName, err) } diff --git a/internal/controller/cancel_test.go b/internal/controller/cancel_test.go index dbd5878e4..dd1704447 100644 --- a/internal/controller/cancel_test.go +++ b/internal/controller/cancel_test.go @@ -5,6 +5,7 @@ import ( "testing" batchv1 "k8s.io/api/batch/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" "k8s.io/client-go/tools/record" @@ -65,8 +66,8 @@ func TestTaskReconciler_CancelledTaskDeletesJob(t *testing.T) { // Verify Job was deleted. var updatedJob batchv1.Job err = fakeClient.Get(context.Background(), types.NamespacedName{Name: "cancel-me", Namespace: "default"}, &updatedJob) - if err == nil { - t.Error("Expected Job to be deleted after task cancellation") + if !apierrors.IsNotFound(err) { + t.Errorf("Expected NotFound error for deleted Job, got: %v", err) } } From 7513929e6d4c5dc0123c43d3e37cb8cddcb0443b Mon Sep 17 00:00:00 2001 From: Irina Truong Date: Tue, 2 Jun 2026 19:21:59 -0700 Subject: [PATCH 3/3] fix: terminal-phase check before retry loop in cancel command Move the already-terminal check before the retry loop so it returns early without printing the misleading "cancelled" success message. Co-Authored-By: Claude Opus 4.6 --- internal/cli/cancel.go | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/internal/cli/cancel.go b/internal/cli/cancel.go index 21053d67c..012828015 100644 --- a/internal/cli/cancel.go +++ b/internal/cli/cancel.go @@ -50,18 +50,21 @@ func newCancelTaskCommand(cfg *ClientConfig) *cobra.Command { ctx := context.Background() taskName := args[0] + // Check terminal state before attempting cancellation. + var task kelosv1alpha1.Task + if err := cl.Get(ctx, client.ObjectKey{Namespace: ns, Name: taskName}, &task); err != nil { + return fmt.Errorf("getting task %s: %w", taskName, err) + } + switch task.Status.Phase { + case kelosv1alpha1.TaskPhaseSucceeded, kelosv1alpha1.TaskPhaseFailed, kelosv1alpha1.TaskPhaseCancelled: + fmt.Fprintf(os.Stdout, "task/%s is already in terminal phase %s\n", taskName, task.Status.Phase) + return nil + } + if err := retry.RetryOnConflict(retry.DefaultRetry, func() error { - var task kelosv1alpha1.Task if err := cl.Get(ctx, client.ObjectKey{Namespace: ns, Name: taskName}, &task); err != nil { return err } - - switch task.Status.Phase { - case kelosv1alpha1.TaskPhaseSucceeded, kelosv1alpha1.TaskPhaseFailed, kelosv1alpha1.TaskPhaseCancelled: - fmt.Fprintf(os.Stdout, "task/%s is already in terminal phase %s\n", taskName, task.Status.Phase) - return nil - } - now := metav1.Now() task.Status.Phase = kelosv1alpha1.TaskPhaseCancelled task.Status.Message = "Cancelled by user"