feat: add Slack status feedback as thread replies

When a Slack message triggers a Task, the bot now posts thread replies
to the originating message as the Task progresses:
- "Working on your request..." when accepted
- "Done!" when succeeded
- "Failed." when failed

Implementation:
- SlackReporter posts/updates thread replies via slack-go
- SlackTaskReporter mirrors GitHubTaskReporter pattern using
  Slack-specific annotations for channel, thread_ts, reply_ts
- sourceAnnotations() stamps Slack metadata on Tasks at creation
- reportingEnabled() returns true for Slack sources (always-on)
- buildWorkItem() now carries channel ID for annotation stamping

Towards AIE-17

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: jkahuja

cmd/kelos-spawner/main.go

@@ -186,6 +186,23 @@ func runReportingCycle(ctx context.Context, cl client.Client, key types.Namespac
 	return nil
 }
 
+func runSlackReportingCycle(ctx context.Context, cl client.Client, key types.NamespacedName, reporter *reporting.SlackTaskReporter) error {
+	var taskList kelosv1alpha1.TaskList
+	if err := cl.List(ctx, &taskList,
+		client.InNamespace(key.Namespace),
+		client.MatchingLabels{"kelos.dev/taskspawner": key.Name},
+	); err != nil {
+		return fmt.Errorf("listing tasks for Slack reporting: %w", err)
+	}
+
+	for i := range taskList.Items {
+		if err := reporter.ReportTaskStatus(ctx, &taskList.Items[i]); err != nil {
+			ctrl.Log.WithName("spawner").Error(err, "Reporting Slack task status", "task", taskList.Items[i].Name)
+		}
+	}
+	return nil
+}
+
 func runCycle(ctx context.Context, cl client.Client, key types.NamespacedName, githubOwner, githubRepo, githubAPIBaseURL, githubTokenFile, jiraBaseURL, jiraProject, jiraJQL, slackTriggerCommand, slackChannels, slackAllowedUsers string, httpClient *http.Client) error {
 	start := time.Now()
 	err := runCycleCore(ctx, cl, key, githubOwner, githubRepo, githubAPIBaseURL, githubTokenFile, jiraBaseURL, jiraProject, jiraJQL, slackTriggerCommand, slackChannels, slackAllowedUsers, httpClient)
@@ -526,10 +543,18 @@ func renderTaskTemplateMetadata(ts *kelosv1alpha1.TaskSpawner, item source.WorkI
 	return labels, annotations, nil
 }
 
-// sourceAnnotations returns annotations that stamp GitHub source metadata
-// onto a spawned Task. These annotations enable downstream consumers (such
-// as the reporting watcher) to identify the originating issue or PR.
+// sourceAnnotations returns annotations that stamp source metadata onto a
+// spawned Task. These annotations enable downstream consumers (such as the
+// reporting watcher) to identify the originating issue, PR, or Slack message.
 func sourceAnnotations(ts *kelosv1alpha1.TaskSpawner, item source.WorkItem) map[string]string {
+	if ts.Spec.When.Slack != nil && len(item.Labels) >= 2 {
+		return map[string]string{
+			reporting.AnnotationSlackReporting: "enabled",
+			reporting.AnnotationSlackChannel:   item.Labels[1],
+			reporting.AnnotationSlackThreadTS:  item.ID,
+		}
+	}
+
 	if ts.Spec.When.GitHubIssues == nil && ts.Spec.When.GitHubPullRequests == nil {
 		return nil
 	}
@@ -551,9 +576,12 @@ func sourceAnnotations(ts *kelosv1alpha1.TaskSpawner, item source.WorkItem) map[
 	return annotations
 }
 
-// reportingEnabled returns true when GitHub reporting is configured and enabled
+// reportingEnabled returns true when reporting is configured and enabled
 // on the TaskSpawner.
 func reportingEnabled(ts *kelosv1alpha1.TaskSpawner) bool {
+	if ts.Spec.When.Slack != nil {
+		return true
+	}
 	if ts.Spec.When.GitHubIssues != nil && ts.Spec.When.GitHubIssues.Reporting != nil {
 		return ts.Spec.When.GitHubIssues.Reporting.Enabled
 	}

cmd/kelos-spawner/reconciler.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"net/http"
+	"os"
 	"time"
 
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
@@ -82,25 +83,36 @@ func runOnce(ctx context.Context, cl client.Client, key types.NamespacedName, cf
 	}
 
 	if reportingEnabled(&ts) {
-		token, err := readGitHubToken(cfg.GitHubTokenFile)
-		if err != nil {
-			return 0, fmt.Errorf("reading GitHub token for reporting: %w", err)
-		}
+		if ts.Spec.When.Slack != nil {
+			botToken := os.Getenv("SLACK_BOT_TOKEN")
+			slackReporter := &reporting.SlackTaskReporter{
+				Client:   cl,
+				Reporter: &reporting.SlackReporter{BotToken: botToken},
+			}
+			if err := runSlackReportingCycle(ctx, cl, key, slackReporter); err != nil {
+				return 0, err
+			}
+		} else {
+			token, err := readGitHubToken(cfg.GitHubTokenFile)
+			if err != nil {
+				return 0, fmt.Errorf("reading GitHub token for reporting: %w", err)
+			}
 
-		// Reporting always uses the direct API base URL (writes bypass the proxy).
-		reporter := &reporting.TaskReporter{
-			Client: cl,
-			Reporter: &reporting.GitHubReporter{
-				Owner:     cfg.GitHubOwner,
-				Repo:      cfg.GitHubRepo,
-				Token:     token,
-				TokenFile: cfg.GitHubTokenFile,
-				BaseURL:   cfg.GitHubAPIBaseURL,
-				Client:    cfg.HTTPClient,
-			},
-		}
-		if err := runReportingCycle(ctx, cl, key, reporter); err != nil {
-			return 0, err
+			// Reporting always uses the direct API base URL (writes bypass the proxy).
+			reporter := &reporting.TaskReporter{
+				Client: cl,
+				Reporter: &reporting.GitHubReporter{
+					Owner:     cfg.GitHubOwner,
+					Repo:      cfg.GitHubRepo,
+					Token:     token,
+					TokenFile: cfg.GitHubTokenFile,
+					BaseURL:   cfg.GitHubAPIBaseURL,
+					Client:    cfg.HTTPClient,
+				},
+			}
+			if err := runReportingCycle(ctx, cl, key, reporter); err != nil {
+				return 0, err
+			}
 		}
 	}
 

internal/reporting/slack.go

@@ -0,0 +1,55 @@
+package reporting
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/slack-go/slack"
+)
+
+// SlackReporter posts and updates thread replies in Slack channels.
+type SlackReporter struct {
+	// BotToken is the Bot User OAuth Token (xoxb-...).
+	BotToken string
+}
+
+// PostThreadReply posts a new message as a thread reply and returns the
+// reply's message timestamp.
+func (r *SlackReporter) PostThreadReply(ctx context.Context, channel, threadTS, text string) (string, error) {
+	api := slack.New(r.BotToken)
+	_, ts, err := api.PostMessageContext(ctx, channel,
+		slack.MsgOptionText(text, false),
+		slack.MsgOptionTS(threadTS),
+	)
+	if err != nil {
+		return "", fmt.Errorf("posting Slack thread reply: %w", err)
+	}
+	return ts, nil
+}
+
+// UpdateMessage updates an existing Slack message in place.
+func (r *SlackReporter) UpdateMessage(ctx context.Context, channel, messageTS, text string) error {
+	api := slack.New(r.BotToken)
+	_, _, _, err := api.UpdateMessageContext(ctx, channel, messageTS,
+		slack.MsgOptionText(text, false),
+	)
+	if err != nil {
+		return fmt.Errorf("updating Slack message: %w", err)
+	}
+	return nil
+}
+
+// FormatSlackAccepted returns the thread reply text for an accepted task.
+func FormatSlackAccepted(taskName string) string {
+	return fmt.Sprintf("Working on your request... (Task: %s)", taskName)
+}
+
+// FormatSlackSucceeded returns the thread reply text for a succeeded task.
+func FormatSlackSucceeded(taskName string) string {
+	return fmt.Sprintf("Done! (Task: %s)", taskName)
+}
+
+// FormatSlackFailed returns the thread reply text for a failed task.
+func FormatSlackFailed(taskName string) string {
+	return fmt.Sprintf("Failed. (Task: %s)", taskName)
+}

internal/reporting/slack_test.go

@@ -0,0 +1,66 @@
+package reporting
+
+import (
+	"context"
+	"testing"
+)
+
+func TestFormatSlackMessages(t *testing.T) {
+	tests := []struct {
+		name     string
+		fn       func(string) string
+		taskName string
+		want     string
+	}{
+		{
+			name:     "accepted",
+			fn:       FormatSlackAccepted,
+			taskName: "spawner-1234567890.123456",
+			want:     "Working on your request... (Task: spawner-1234567890.123456)",
+		},
+		{
+			name:     "succeeded",
+			fn:       FormatSlackSucceeded,
+			taskName: "spawner-1234567890.123456",
+			want:     "Done! (Task: spawner-1234567890.123456)",
+		},
+		{
+			name:     "failed",
+			fn:       FormatSlackFailed,
+			taskName: "spawner-1234567890.123456",
+			want:     "Failed. (Task: spawner-1234567890.123456)",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.fn(tt.taskName)
+			if got != tt.want {
+				t.Errorf("got %q, want %q", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestSlackReporterConstruction(t *testing.T) {
+	reporter := &SlackReporter{BotToken: "xoxb-test-token"}
+	if reporter.BotToken != "xoxb-test-token" {
+		t.Errorf("BotToken = %q, want %q", reporter.BotToken, "xoxb-test-token")
+	}
+}
+
+func TestSlackReporter_PostThreadReplyError(t *testing.T) {
+	reporter := &SlackReporter{BotToken: "xoxb-invalid"}
+	_, err := reporter.PostThreadReply(context.Background(), "C123", "1234.5678", "test")
+	if err == nil {
+		t.Error("expected error with invalid token, got nil")
+	}
+}
+
+func TestSlackReporter_UpdateMessageError(t *testing.T) {
+	reporter := &SlackReporter{BotToken: "xoxb-invalid"}
+	err := reporter.UpdateMessage(context.Background(), "C123", "1234.5678", "test")
+	if err == nil {
+		t.Error("expected error with invalid token, got nil")
+	}
+}

internal/reporting/watcher.go

@@ -32,6 +32,26 @@ const (
 	// AnnotationGitHubReportPhase records the last Task phase that was
 	// reported to GitHub, preventing duplicate API calls on re-list.
 	AnnotationGitHubReportPhase = "kelos.dev/github-report-phase"
+
+	// AnnotationSlackReporting indicates that Slack reporting is enabled
+	// for this Task.
+	AnnotationSlackReporting = "kelos.dev/slack-reporting"
+
+	// AnnotationSlackChannel records the Slack channel ID where the
+	// originating message was posted.
+	AnnotationSlackChannel = "kelos.dev/slack-channel"
+
+	// AnnotationSlackThreadTS records the originating message timestamp,
+	// used as thread_ts for posting replies.
+	AnnotationSlackThreadTS = "kelos.dev/slack-thread-ts"
+
+	// AnnotationSlackReplyTS stores the message timestamp of the status
+	// reply so subsequent updates edit the same message.
+	AnnotationSlackReplyTS = "kelos.dev/slack-reply-ts"
+
+	// AnnotationSlackReportPhase records the last Task phase that was
+	// reported to Slack, preventing duplicate API calls on re-list.
+	AnnotationSlackReportPhase = "kelos.dev/slack-report-phase"
 )
 
 // TaskReporter watches Tasks and reports status changes to GitHub.
@@ -156,3 +176,107 @@ func (tr *TaskReporter) persistReportingState(ctx context.Context, task *kelosv1
 
 	return nil
 }
+
+// SlackTaskReporter watches Tasks and reports status changes to Slack
+// as thread replies on the originating message.
+type SlackTaskReporter struct {
+	Client   client.Client
+	Reporter *SlackReporter
+}
+
+// ReportTaskStatus checks a Task's current phase against its last reported
+// phase and creates or updates the Slack thread reply as needed.
+func (tr *SlackTaskReporter) ReportTaskStatus(ctx context.Context, task *kelosv1alpha1.Task) error {
+	log := ctrl.Log.WithName("slack-reporter")
+
+	annotations := task.Annotations
+	if annotations == nil {
+		return nil
+	}
+
+	if annotations[AnnotationSlackReporting] != "enabled" {
+		return nil
+	}
+
+	channel := annotations[AnnotationSlackChannel]
+	threadTS := annotations[AnnotationSlackThreadTS]
+	if channel == "" || threadTS == "" {
+		return nil
+	}
+
+	var desiredPhase string
+	switch task.Status.Phase {
+	case kelosv1alpha1.TaskPhasePending, kelosv1alpha1.TaskPhaseRunning, kelosv1alpha1.TaskPhaseWaiting:
+		desiredPhase = "accepted"
+	case kelosv1alpha1.TaskPhaseSucceeded:
+		desiredPhase = "succeeded"
+	case kelosv1alpha1.TaskPhaseFailed:
+		desiredPhase = "failed"
+	default:
+		return nil
+	}
+
+	if annotations[AnnotationSlackReportPhase] == desiredPhase {
+		return nil
+	}
+
+	var body string
+	switch desiredPhase {
+	case "accepted":
+		body = FormatSlackAccepted(task.Name)
+	case "succeeded":
+		body = FormatSlackSucceeded(task.Name)
+	case "failed":
+		body = FormatSlackFailed(task.Name)
+	}
+
+	replyTS := annotations[AnnotationSlackReplyTS]
+	if replyTS == "" {
+		log.Info("Posting Slack thread reply", "task", task.Name, "channel", channel, "phase", desiredPhase)
+		newTS, err := tr.Reporter.PostThreadReply(ctx, channel, threadTS, body)
+		if err != nil {
+			return fmt.Errorf("posting Slack reply for task %s: %w", task.Name, err)
+		}
+		replyTS = newTS
+	} else {
+		log.Info("Updating Slack thread reply", "task", task.Name, "channel", channel, "phase", desiredPhase)
+		if err := tr.Reporter.UpdateMessage(ctx, channel, replyTS, body); err != nil {
+			return fmt.Errorf("updating Slack reply for task %s: %w", task.Name, err)
+		}
+	}
+
+	if err := tr.persistSlackReportingState(ctx, task, replyTS, desiredPhase); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (tr *SlackTaskReporter) persistSlackReportingState(ctx context.Context, task *kelosv1alpha1.Task, replyTS, desiredPhase string) error {
+	if err := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		var current kelosv1alpha1.Task
+		if err := tr.Client.Get(ctx, client.ObjectKeyFromObject(task), &current); err != nil {
+			return err
+		}
+
+		if current.Annotations == nil {
+			current.Annotations = make(map[string]string)
+		}
+		current.Annotations[AnnotationSlackReplyTS] = replyTS
+		current.Annotations[AnnotationSlackReportPhase] = desiredPhase
+
+		if err := tr.Client.Update(ctx, &current); err != nil {
+			return err
+		}
+
+		task.Annotations = current.Annotations
+		return nil
+	}); err != nil {
+		if apierrors.IsNotFound(err) {
+			return fmt.Errorf("persisting Slack reporting annotations on task %s: task no longer exists", task.Name)
+		}
+		return fmt.Errorf("persisting Slack reporting annotations on task %s: %w", task.Name, err)
+	}
+
+	return nil
+}