Add Cancelled phase for task cancellation

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

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) {
 

internal/cli/cancel.go

@@ -0,0 +1,84 @@
+package cli
+
+import (
+	"context"
+	"fmt"
+	"os"
+
+	"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"
+)
+
+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 <name>",
+		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]
+
+			// 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 {
+				if err := cl.Get(ctx, client.ObjectKey{Namespace: ns, Name: taskName}, &task); err != nil {
+					return err
+				}
+				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)
+			}
+
+			fmt.Fprintf(os.Stdout, "task/%s cancelled\n", taskName)
+			return nil
+		},
+	}
+
+	return cmd
+}

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 <name>" {
+		t.Errorf("expected Use to be 'task <name>', 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")
+	}
+}

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

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 {

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),