diff --git a/apps/backend/internal/github/controller_test.go b/apps/backend/internal/github/controller_test.go
index b492bcc133..7c64621ed0 100644
--- a/apps/backend/internal/github/controller_test.go
+++ b/apps/backend/internal/github/controller_test.go
@@ -674,6 +674,9 @@ func TestHttpTaskCIOptions_DefaultAndPatch(t *testing.T) {
if !got.UsingDefaultPrompt || got.EffectiveAutoFixPrompt != "resolved default prompt" {
t.Fatalf("unexpected prompt fields: %+v", got)
}
+ if got.AutoFixMaxRounds != TaskCIAutoFixMaxRounds {
+ t.Fatalf("auto-fix max rounds = %d, want %d", got.AutoFixMaxRounds, TaskCIAutoFixMaxRounds)
+ }
if len(got.PRStates) != 1 || got.PRStates[0].PRNumber != 42 {
t.Fatalf("expected synthesized PR state for PR #42, got %+v", got.PRStates)
}
diff --git a/apps/backend/internal/github/models.go b/apps/backend/internal/github/models.go
index def03b73d4..d505355a30 100644
--- a/apps/backend/internal/github/models.go
+++ b/apps/backend/internal/github/models.go
@@ -4,6 +4,9 @@ package github
import "time"
+// TaskCIAutoFixMaxRounds is the server-enforced CI auto-fix loop guard.
+const TaskCIAutoFixMaxRounds = 10
+
// PR represents a GitHub Pull Request.
type PR struct {
Number int `json:"number"`
@@ -219,6 +222,7 @@ type TaskCIOptionsResponse struct {
AutoFixEnabled bool `json:"auto_fix_enabled"`
AutoMergeEnabled bool `json:"auto_merge_enabled"`
AutoFixPromptOverride *string `json:"auto_fix_prompt_override"`
+ AutoFixMaxRounds int `json:"auto_fix_max_rounds"`
EffectiveAutoFixPrompt string `json:"effective_auto_fix_prompt"`
UsingDefaultPrompt bool `json:"using_default_prompt"`
UpdatedAt time.Time `json:"updated_at"`
@@ -234,6 +238,8 @@ type TaskCIPRAutomationState struct {
LastFixCheckpointJSON string `json:"last_fix_checkpoint_json" db:"last_fix_checkpoint_json"`
LastFixEnqueuedAt *time.Time `json:"last_fix_enqueued_at,omitempty" db:"last_fix_enqueued_at"`
LastFixSessionID *string `json:"last_fix_session_id,omitempty" db:"last_fix_session_id"`
+ AutoFixRoundCount int `json:"auto_fix_round_count" db:"auto_fix_round_count"`
+ AutoFixExhaustedAt *time.Time `json:"auto_fix_exhausted_at" db:"auto_fix_exhausted_at"`
LastMergeSignature string `json:"last_merge_signature" db:"last_merge_signature"`
LastMergeAttemptAt *time.Time `json:"last_merge_attempt_at,omitempty" db:"last_merge_attempt_at"`
LastError *string `json:"last_error,omitempty" db:"last_error"`
@@ -250,6 +256,7 @@ type TaskCIFixAttempt struct {
CheckpointJSON string
SessionID string
EnqueuedAt time.Time
+ IncrementRound bool
}
// TaskCIMergeAttempt records an auto-merge attempt for a task PR.
diff --git a/apps/backend/internal/github/service_ci_automation.go b/apps/backend/internal/github/service_ci_automation.go
index c170ae469d..90390c3229 100644
--- a/apps/backend/internal/github/service_ci_automation.go
+++ b/apps/backend/internal/github/service_ci_automation.go
@@ -72,6 +72,14 @@ func (s *Service) RecordTaskCIError(ctx context.Context, taskID, repositoryID st
return s.store.RecordTaskCIError(ctx, taskID, repositoryID, prNumber, message)
}
+// MarkTaskCIAutoFixExhausted records that auto-fix reached its per-PR round cap.
+func (s *Service) MarkTaskCIAutoFixExhausted(ctx context.Context, taskID, repositoryID string, prNumber int, message string) error {
+ if s.store == nil {
+ return errStoreUnavailable
+ }
+ return s.store.MarkTaskCIAutoFixExhausted(ctx, taskID, repositoryID, prNumber, message)
+}
+
// ClearTaskCIError clears a CI automation error.
func (s *Service) ClearTaskCIError(ctx context.Context, taskID, repositoryID string, prNumber int) error {
if s.store == nil {
@@ -91,6 +99,7 @@ func (s *Service) buildTaskCIOptionsResponse(ctx context.Context, opts *TaskCIOp
AutoFixEnabled: opts.AutoFixEnabled,
AutoMergeEnabled: opts.AutoMergeEnabled,
AutoFixPromptOverride: opts.AutoFixPromptOverride,
+ AutoFixMaxRounds: TaskCIAutoFixMaxRounds,
EffectiveAutoFixPrompt: effectivePrompt,
UsingDefaultPrompt: usingDefault,
UpdatedAt: opts.UpdatedAt,
diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go
index bf2519b5cb..302c2982fe 100644
--- a/apps/backend/internal/github/store.go
+++ b/apps/backend/internal/github/store.go
@@ -179,6 +179,8 @@ const createTablesSQL = `
last_fix_checkpoint_json TEXT NOT NULL DEFAULT '',
last_fix_enqueued_at DATETIME,
last_fix_session_id TEXT,
+ auto_fix_round_count INTEGER NOT NULL DEFAULT 0,
+ auto_fix_exhausted_at DATETIME,
last_merge_signature TEXT NOT NULL DEFAULT '',
last_merge_attempt_at DATETIME,
last_error TEXT,
@@ -223,6 +225,9 @@ func (s *Store) initSchema() error {
if err := s.addWatchSelfHealColumns(); err != nil {
return err
}
+ if err := s.addTaskCIRoundColumns(); err != nil {
+ return err
+ }
if err := s.migratePRTablesForMultiRepo(); err != nil {
return fmt.Errorf("migrate PR tables for multi-repo: %w", err)
}
@@ -381,6 +386,24 @@ func (s *Store) addWatchSelfHealColumns() error {
return nil
}
+func (s *Store) addTaskCIRoundColumns() error {
+ cols, err := s.tableColumns("github_task_ci_pr_state")
+ if err != nil {
+ return fmt.Errorf("read github_task_ci_pr_state columns: %w", err)
+ }
+ if _, ok := cols["auto_fix_round_count"]; !ok {
+ if _, err := s.db.Exec("ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_round_count INTEGER NOT NULL DEFAULT 0"); err != nil {
+ return fmt.Errorf("add github_task_ci_pr_state.auto_fix_round_count: %w", err)
+ }
+ }
+ if _, ok := cols["auto_fix_exhausted_at"]; !ok {
+ if _, err := s.db.Exec("ALTER TABLE github_task_ci_pr_state ADD COLUMN auto_fix_exhausted_at DATETIME"); err != nil {
+ return fmt.Errorf("add github_task_ci_pr_state.auto_fix_exhausted_at: %w", err)
+ }
+ }
+ return nil
+}
+
// tableColumns returns the set of column names declared on `table`. Cheap
// SQLite PRAGMA lookup; used by addWatchSelfHealColumns to skip ALTERs on a
// fresh install whose createTablesSQL already includes the columns. Mirrors
@@ -1031,6 +1054,10 @@ func (s *Store) UpdateTaskCIOptions(ctx context.Context, taskID string, patch Ta
taskID, now, now); err != nil {
return nil, err
}
+ var previous TaskCIOptions
+ if err := tx.GetContext(writeCtx, &previous, `SELECT * FROM github_task_ci_options WHERE task_id = ?`, taskID); err != nil {
+ return nil, err
+ }
autoFixSet, autoFixValue := boolPatchValue(patch.AutoFixEnabled)
autoMergeSet, autoMergeValue := boolPatchValue(patch.AutoMergeEnabled)
promptSet := patch.AutoFixPromptOverride != nil
@@ -1051,6 +1078,21 @@ func (s *Store) UpdateTaskCIOptions(ctx context.Context, taskID string, patch Ta
autoFixSet, autoFixValue, autoMergeSet, autoMergeValue, promptSet, promptValue, now, taskID); err != nil {
return nil, err
}
+ if autoFixSet && autoFixValue && !previous.AutoFixEnabled {
+ if _, err := tx.ExecContext(writeCtx, `
+ UPDATE github_task_ci_pr_state
+ SET auto_fix_round_count = 0,
+ last_fix_signature = '',
+ last_fix_checkpoint_json = '',
+ last_fix_enqueued_at = NULL,
+ last_fix_session_id = NULL,
+ last_error = CASE WHEN auto_fix_exhausted_at IS NOT NULL THEN NULL ELSE last_error END,
+ auto_fix_exhausted_at = NULL,
+ updated_at = ?
+ WHERE task_id = ?`, now, taskID); err != nil {
+ return nil, err
+ }
+ }
if err := tx.Commit(); err != nil {
return nil, err
}
@@ -1093,20 +1135,26 @@ func (s *Store) RecordTaskCIFixAttempt(ctx context.Context, attempt TaskCIFixAtt
when = time.Now().UTC()
}
now := time.Now().UTC()
+ roundCount := 0
+ if attempt.IncrementRound {
+ roundCount = 1
+ }
_, err := s.db.ExecContext(ctx, `
INSERT INTO github_task_ci_pr_state (
task_id, repository_id, pr_number, last_fix_signature, last_fix_checkpoint_json,
- last_fix_enqueued_at, last_fix_session_id, created_at, updated_at
- ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+ last_fix_enqueued_at, last_fix_session_id, auto_fix_round_count, auto_fix_exhausted_at,
+ created_at, updated_at
+ ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, NULL, ?, ?)
ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
last_fix_signature = excluded.last_fix_signature,
last_fix_checkpoint_json = excluded.last_fix_checkpoint_json,
last_fix_enqueued_at = excluded.last_fix_enqueued_at,
last_fix_session_id = excluded.last_fix_session_id,
+ auto_fix_round_count = github_task_ci_pr_state.auto_fix_round_count + excluded.auto_fix_round_count,
last_error = NULL,
updated_at = excluded.updated_at`,
attempt.TaskID, attempt.RepositoryID, attempt.PRNumber, attempt.Signature,
- attempt.CheckpointJSON, when, nullableString(attempt.SessionID), now, now)
+ attempt.CheckpointJSON, when, nullableString(attempt.SessionID), roundCount, now, now)
return err
}
@@ -1165,6 +1213,22 @@ func (s *Store) RecordTaskCIError(ctx context.Context, taskID, repositoryID stri
return err
}
+// MarkTaskCIAutoFixExhausted records that auto-fix reached its per-PR round cap.
+func (s *Store) MarkTaskCIAutoFixExhausted(ctx context.Context, taskID, repositoryID string, prNumber int, message string) error {
+ ctx = context.WithoutCancel(ctx)
+ now := time.Now().UTC()
+ _, err := s.db.ExecContext(ctx, `
+ INSERT INTO github_task_ci_pr_state (
+ task_id, repository_id, pr_number, auto_fix_exhausted_at, last_error, created_at, updated_at
+ ) VALUES (?, ?, ?, ?, ?, ?, ?)
+ ON CONFLICT(task_id, repository_id, pr_number) DO UPDATE SET
+ auto_fix_exhausted_at = excluded.auto_fix_exhausted_at,
+ last_error = excluded.last_error,
+ updated_at = excluded.updated_at`,
+ taskID, repositoryID, prNumber, now, strings.TrimSpace(message), now, now)
+ return err
+}
+
// ClearTaskCIError clears the latest CI automation error for a task PR.
func (s *Store) ClearTaskCIError(ctx context.Context, taskID, repositoryID string, prNumber int) error {
ctx = context.WithoutCancel(ctx)
diff --git a/apps/backend/internal/github/store_ci_automation_test.go b/apps/backend/internal/github/store_ci_automation_test.go
index 0e90eb5e77..3771cffdb6 100644
--- a/apps/backend/internal/github/store_ci_automation_test.go
+++ b/apps/backend/internal/github/store_ci_automation_test.go
@@ -75,9 +75,22 @@ func TestStoreTaskCIPRState_RecordAttemptsAndError(t *testing.T) {
CheckpointJSON: `{"checks":["test"]}`,
SessionID: "session-1",
EnqueuedAt: at,
+ IncrementRound: true,
}); err != nil {
t.Fatalf("record fix attempt: %v", err)
}
+ if err := store.RecordTaskCIFixAttempt(ctx, TaskCIFixAttempt{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ Signature: "fix-sig-2",
+ CheckpointJSON: `{"checks":["test","lint"]}`,
+ SessionID: "session-1",
+ EnqueuedAt: at.Add(30 * time.Second),
+ IncrementRound: false,
+ }); err != nil {
+ t.Fatalf("record replacement fix attempt: %v", err)
+ }
if err := store.RecordTaskCIMergeAttempt(ctx, TaskCIMergeAttempt{
TaskID: "task-1",
RepositoryID: "repo-1",
@@ -98,9 +111,12 @@ func TestStoreTaskCIPRState_RecordAttemptsAndError(t *testing.T) {
if state == nil {
t.Fatal("expected state row")
}
- if state.LastFixSignature != "fix-sig" || state.LastFixCheckpointJSON != `{"checks":["test"]}` {
+ if state.LastFixSignature != "fix-sig-2" || state.LastFixCheckpointJSON != `{"checks":["test","lint"]}` {
t.Fatalf("unexpected fix state: %+v", state)
}
+ if state.AutoFixRoundCount != 1 {
+ t.Fatalf("AutoFixRoundCount=%d, want 1", state.AutoFixRoundCount)
+ }
if state.LastFixSessionID == nil || *state.LastFixSessionID != "session-1" {
t.Fatalf("LastFixSessionID=%v, want session-1", state.LastFixSessionID)
}
@@ -131,6 +147,54 @@ func TestStoreTaskCIPRState_RecordAttemptsAndError(t *testing.T) {
}
}
+func TestStoreTaskCIPRState_MarkExhaustedAndResetOnReenable(t *testing.T) {
+ store := newTestStore(t)
+ ctx := context.Background()
+ at := time.Date(2026, 6, 18, 12, 0, 0, 0, time.UTC)
+
+ if err := store.RecordTaskCIFixAttempt(ctx, TaskCIFixAttempt{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ Signature: "fix-sig",
+ CheckpointJSON: `{}`,
+ SessionID: "session-1",
+ EnqueuedAt: at,
+ IncrementRound: true,
+ }); err != nil {
+ t.Fatalf("record fix attempt: %v", err)
+ }
+ if err := store.MarkTaskCIAutoFixExhausted(ctx, "task-1", "repo-1", 42, "CI auto-fix paused after 10 rounds for this PR"); err != nil {
+ t.Fatalf("mark exhausted: %v", err)
+ }
+ state, err := store.GetTaskCIPRState(ctx, "task-1", "repo-1", 42)
+ if err != nil {
+ t.Fatalf("get exhausted state: %v", err)
+ }
+ if state.AutoFixExhaustedAt == nil || state.LastError == nil {
+ t.Fatalf("expected exhausted timestamp and error, got %+v", state)
+ }
+
+ disabled := false
+ if _, err := store.UpdateTaskCIOptions(ctx, "task-1", TaskCIOptionsPatch{AutoFixEnabled: &disabled}); err != nil {
+ t.Fatalf("disable auto-fix: %v", err)
+ }
+ enabled := true
+ if _, err := store.UpdateTaskCIOptions(ctx, "task-1", TaskCIOptionsPatch{AutoFixEnabled: &enabled}); err != nil {
+ t.Fatalf("re-enable auto-fix: %v", err)
+ }
+ state, err = store.GetTaskCIPRState(ctx, "task-1", "repo-1", 42)
+ if err != nil {
+ t.Fatalf("get reset state: %v", err)
+ }
+ if state.AutoFixRoundCount != 0 || state.AutoFixExhaustedAt != nil || state.LastError != nil {
+ t.Fatalf("expected auto-fix round state reset, got %+v", state)
+ }
+ if state.LastFixSignature != "" || state.LastFixCheckpointJSON != "" || state.LastFixEnqueuedAt != nil || state.LastFixSessionID != nil {
+ t.Fatalf("expected auto-fix checkpoint state reset, got %+v", state)
+ }
+}
+
func TestStoreTaskCIPRState_RefreshCheckpointClearsPromptDispatchMetadata(t *testing.T) {
store := newTestStore(t)
ctx := context.Background()
diff --git a/apps/backend/internal/orchestrator/event_handlers_agent.go b/apps/backend/internal/orchestrator/event_handlers_agent.go
index 21bb35de8c..6cbc8d5823 100644
--- a/apps/backend/internal/orchestrator/event_handlers_agent.go
+++ b/apps/backend/internal/orchestrator/event_handlers_agent.go
@@ -82,17 +82,42 @@ func (s *Service) publishQueueStatusEvent(ctx context.Context, sessionID string)
// Preserves the original Metadata (e.g. sender_task_id from message_task_kandev)
// so attribution survives transient failures + retries.
func (s *Service) requeueMessage(ctx context.Context, queuedMsg *messagequeue.QueuedMessage, queuedBy string) {
- requeuedMsg, queueErr := s.messageQueue.QueueMessageWithMetadata(
- ctx,
- queuedMsg.SessionID,
- queuedMsg.TaskID,
- queuedMsg.Content,
- queuedMsg.Model,
- queuedBy,
- queuedMsg.PlanMode,
- queuedMsg.Attachments,
- queuedMsg.Metadata,
+ coalesceKey := messageCoalesceKey(queuedMsg)
+ if queuedMsg.QueuedBy != "" && coalesceKey != "" {
+ queuedBy = queuedMsg.QueuedBy
+ }
+ var (
+ requeuedMsg *messagequeue.QueuedMessage
+ replaced bool
+ queueErr error
)
+ if coalesceKey != "" {
+ requeuedMsg, replaced, queueErr = s.messageQueue.QueueMessageWithCoalesceKey(
+ ctx,
+ queuedMsg.SessionID,
+ queuedMsg.TaskID,
+ queuedMsg.Content,
+ queuedMsg.Model,
+ queuedBy,
+ queuedMsg.PlanMode,
+ queuedMsg.Attachments,
+ queuedMsg.Metadata,
+ coalesceKey,
+ true,
+ )
+ } else {
+ requeuedMsg, queueErr = s.messageQueue.QueueMessageWithMetadata(
+ ctx,
+ queuedMsg.SessionID,
+ queuedMsg.TaskID,
+ queuedMsg.Content,
+ queuedMsg.Model,
+ queuedBy,
+ queuedMsg.PlanMode,
+ queuedMsg.Attachments,
+ queuedMsg.Metadata,
+ )
+ }
if queueErr != nil {
s.logger.Error("failed to requeue message",
zap.String("session_id", queuedMsg.SessionID),
@@ -107,10 +132,27 @@ func (s *Service) requeueMessage(ctx context.Context, queuedMsg *messagequeue.Qu
zap.String("task_id", queuedMsg.TaskID),
zap.String("old_queue_id", queuedMsg.ID),
zap.String("new_queue_id", requeuedMsg.ID),
- zap.String("queued_by", queuedBy))
+ zap.String("queued_by", queuedBy),
+ zap.String("coalesce_key", coalesceKey),
+ zap.Bool("replaced", replaced))
s.publishQueueStatusEvent(ctx, queuedMsg.SessionID)
}
+func messageCoalesceKey(queuedMsg *messagequeue.QueuedMessage) string {
+ if queuedMsg == nil || len(queuedMsg.Metadata) == 0 {
+ return ""
+ }
+ value, ok := queuedMsg.Metadata[messagequeue.MetadataCoalesceKey]
+ if !ok {
+ return ""
+ }
+ key, ok := value.(string)
+ if !ok {
+ return ""
+ }
+ return key
+}
+
// handleAgentBootReady handles the boot signal: an agent's ACP session has
// finished initializing but no turn has run yet. This event is distinct from
// agent.ready (turn-end) so the orchestrator never has to disambiguate the
diff --git a/apps/backend/internal/orchestrator/event_handlers_github.go b/apps/backend/internal/orchestrator/event_handlers_github.go
index 9ff993fcdb..6191b25594 100644
--- a/apps/backend/internal/orchestrator/event_handlers_github.go
+++ b/apps/backend/internal/orchestrator/event_handlers_github.go
@@ -50,6 +50,7 @@ type GitHubService interface {
RefreshTaskCIFixCheckpoint(ctx context.Context, taskID, repositoryID string, prNumber int, signature, checkpointJSON string) error
RecordTaskCIMergeAttempt(ctx context.Context, attempt github.TaskCIMergeAttempt) error
RecordTaskCIError(ctx context.Context, taskID, repositoryID string, prNumber int, message string) error
+ MarkTaskCIAutoFixExhausted(ctx context.Context, taskID, repositoryID string, prNumber int, message string) error
ClearTaskCIError(ctx context.Context, taskID, repositoryID string, prNumber int) error
GetPRFeedback(ctx context.Context, owner, repo string, number int) (*github.PRFeedback, error)
MergePR(ctx context.Context, owner, repo string, number int, mergeMethod string) error
@@ -188,7 +189,7 @@ func (s *Service) handleTaskPRUpdated(ctx context.Context, event *bus.Event) err
func (s *Service) handleTaskCIOptionsUpdated(ctx context.Context, event *bus.Event) error {
options, ok := event.Data.(*github.TaskCIOptionsResponse)
- if !ok || options == nil || (!options.AutoFixEnabled && !options.AutoMergeEnabled) || s.githubService == nil {
+ if !ok || options == nil || event.Source == ciAutomationStateEventSource || (!options.AutoFixEnabled && !options.AutoMergeEnabled) || s.githubService == nil {
return nil
}
detachedCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), ciAutomationDetachedTimeout)
diff --git a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go
index 140ee4183f..836c787402 100644
--- a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go
+++ b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go
@@ -5,12 +5,15 @@ import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
+ "errors"
"fmt"
"strings"
"time"
"go.uber.org/zap"
+ "github.com/kandev/kandev/internal/events"
+ "github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/github"
"github.com/kandev/kandev/internal/orchestrator/messagequeue"
"github.com/kandev/kandev/internal/sysprompt"
@@ -25,10 +28,15 @@ const (
ciAutomationChangesRequested = "changes_requested"
ciAutomationPRFeedbackToken = "{{pr.feedback}}"
ciAutomationFixBlockWindow = time.Hour
+ ciAutomationMaxFixRounds = github.TaskCIAutoFixMaxRounds
+ ciAutomationKindAutoFix = "ci_auto_fix"
+ ciAutomationStateEventSource = "ci_automation_state"
)
var ciAutomationSnapshotFieldReplacer = strings.NewReplacer("\r", " ", "\n", " ", "<", "", ">", "")
+var errCIAutoFixRoundCapReached = errors.New("CI auto-fix round cap reached")
+
type ciAutomationCheckpoint struct {
FailedChecks []ciAutomationCheckSnapshot `json:"failed_checks"`
Comments []ciAutomationCommentSnapshot `json:"comments"`
@@ -48,6 +56,22 @@ type ciAutomationCommentSnapshot struct {
Line int `json:"line,omitempty"`
}
+type ciAutomationDispatchKind string
+
+const (
+ ciAutomationDispatchDirect ciAutomationDispatchKind = "direct"
+ ciAutomationDispatchQueuedInsert ciAutomationDispatchKind = "queued_insert"
+ ciAutomationDispatchQueuedReplace ciAutomationDispatchKind = "queued_replace"
+)
+
+type ciAutomationDispatchResult struct {
+ kind ciAutomationDispatchKind
+}
+
+func (r ciAutomationDispatchResult) consumesRound() bool {
+ return r.kind == ciAutomationDispatchDirect || r.kind == ciAutomationDispatchQueuedInsert
+}
+
func (s *Service) handleTaskPRCIAutomation(ctx context.Context, pr *github.TaskPR) error {
return s.handleTaskPRCIAutomationWithRefresh(ctx, pr, true)
}
@@ -132,6 +156,14 @@ func ciAutomationHasFreshPRStatusAt(pr *github.TaskPR, now time.Time) bool {
}
func (s *Service) handleTaskPRCIAutoFix(ctx context.Context, pr *github.TaskPR, options *github.TaskCIOptionsResponse) bool {
+ state, err := s.githubService.GetTaskCIPRState(ctx, pr.TaskID, pr.RepositoryID, pr.PRNumber)
+ if err != nil {
+ s.recordCIAutomationError(ctx, pr, fmt.Sprintf("load CI automation state: %v", err))
+ return true
+ }
+ if state != nil && state.AutoFixExhaustedAt != nil {
+ return !ciAutomationReadyToMerge(pr)
+ }
feedback, err := s.githubService.GetPRFeedback(ctx, pr.Owner, pr.Repo, pr.PRNumber)
if err != nil {
s.recordCIAutomationError(ctx, pr, fmt.Sprintf("fetch PR feedback: %v", err))
@@ -141,11 +173,6 @@ func (s *Service) handleTaskPRCIAutoFix(ctx context.Context, pr *github.TaskPR,
return false
}
feedback = ciAutomationFilterFeedbackForPR(pr, feedback)
- state, err := s.githubService.GetTaskCIPRState(ctx, pr.TaskID, pr.RepositoryID, pr.PRNumber)
- if err != nil {
- s.recordCIAutomationError(ctx, pr, fmt.Sprintf("load CI automation state: %v", err))
- return true
- }
previous := decodeCIAutomationCheckpoint(state)
delta := ciAutomationBuildDelta(feedback, previous)
checkpoint := ciAutomationCurrentCheckpoint(feedback)
@@ -156,13 +183,23 @@ func (s *Service) handleTaskPRCIAutoFix(ctx context.Context, pr *github.TaskPR,
if state != nil && state.LastFixSignature == signature {
return ciAutomationDuplicateFixAttemptBlocksMerge(state)
}
+ allowNewRound := !ciAutomationFixRoundsExhausted(state)
prompt := ciAutomationRenderPrompt(options.EffectiveAutoFixPrompt, pr, delta)
session, err := s.repo.GetActiveTaskSessionByTaskID(ctx, pr.TaskID)
if err != nil || session == nil {
+ if !allowNewRound {
+ s.markCIAutoFixExhausted(ctx, pr)
+ return true
+ }
s.recordCIAutomationError(ctx, pr, "no promptable task session for CI auto-fix")
return true
}
- if err := s.dispatchCIAutomationPrompt(ctx, session, prompt); err != nil {
+ result, err := s.dispatchCIAutomationPromptForPR(ctx, session, pr, prompt, signature, allowNewRound)
+ if errors.Is(err, errCIAutoFixRoundCapReached) {
+ s.markCIAutoFixExhausted(ctx, pr)
+ return true
+ }
+ if err != nil {
s.recordCIAutomationError(ctx, pr, err.Error())
return true
}
@@ -174,8 +211,11 @@ func (s *Service) handleTaskPRCIAutoFix(ctx context.Context, pr *github.TaskPR,
CheckpointJSON: checkpointJSON,
SessionID: session.ID,
EnqueuedAt: time.Now().UTC(),
+ IncrementRound: result.consumesRound(),
}); err != nil {
s.logger.Debug("record CI auto-fix attempt failed", zap.String("task_id", pr.TaskID), zap.Error(err))
+ } else {
+ s.publishTaskCIOptionsState(ctx, pr.TaskID)
}
return true
}
@@ -215,33 +255,70 @@ func (s *Service) handleTaskPRCIAutoMerge(ctx context.Context, pr *github.TaskPR
_ = s.githubService.ClearTaskCIError(context.WithoutCancel(ctx), pr.TaskID, pr.RepositoryID, pr.PRNumber)
}
-func (s *Service) dispatchCIAutomationPrompt(ctx context.Context, session *models.TaskSession, prompt string) error {
+func (s *Service) dispatchCIAutomationPromptForPR(ctx context.Context, session *models.TaskSession, pr *github.TaskPR, prompt, signature string, allowNewRound bool) (ciAutomationDispatchResult, error) {
chatPrompt := ciAutomationChatPrompt(prompt)
switch session.State {
case models.TaskSessionStateRunning, models.TaskSessionStateStarting:
- if s.messageQueue == nil {
- return fmt.Errorf("message queue is not configured")
- }
- metadata := ciAutomationMessageMetadata()
- _, err := s.messageQueue.QueueMessageWithMetadata(ctx, session.ID, session.TaskID, chatPrompt, "", messagequeue.QueuedByWorkflow, false, nil, metadata)
- if err != nil {
- return err
- }
- s.publishQueueStatusEvent(ctx, session.ID)
- return nil
+ return s.queueOrReplaceCIAutomationPromptForPR(ctx, session, pr, chatPrompt, signature, allowNewRound)
case models.TaskSessionStateWaitingForInput, models.TaskSessionStateIdle:
+ result, replaced, err := s.replacePendingCIAutomationPromptForPR(ctx, session, pr, chatPrompt, signature)
+ if err != nil {
+ return ciAutomationDispatchResult{}, err
+ }
+ if replaced {
+ if !s.drainQueuedMessageForPromptableSession(ctx, session.ID) {
+ return ciAutomationDispatchResult{}, fmt.Errorf("failed to dispatch replaced CI automation prompt")
+ }
+ return result, nil
+ }
+ if !allowNewRound {
+ return ciAutomationDispatchResult{}, errCIAutoFixRoundCapReached
+ }
if !s.recordCIAutomationUserMessage(ctx, session.TaskID, session.ID, chatPrompt) {
- return fmt.Errorf("failed to record CI automation user message")
+ return ciAutomationDispatchResult{}, fmt.Errorf("failed to record CI automation user message")
}
if _, err := s.PromptTask(ctx, session.TaskID, session.ID, chatPrompt, "", false, nil, true); err != nil {
- return err
+ return ciAutomationDispatchResult{}, err
}
- return nil
+ return ciAutomationDispatchResult{kind: ciAutomationDispatchDirect}, nil
default:
- return fmt.Errorf("session is not promptable: %s", session.State)
+ return ciAutomationDispatchResult{}, fmt.Errorf("session is not promptable: %s", session.State)
}
}
+func (s *Service) replacePendingCIAutomationPromptForPR(ctx context.Context, session *models.TaskSession, pr *github.TaskPR, chatPrompt, signature string) (ciAutomationDispatchResult, bool, error) {
+ if s.messageQueue == nil {
+ return ciAutomationDispatchResult{}, false, nil
+ }
+ result, err := s.queueOrReplaceCIAutomationPromptForPR(ctx, session, pr, chatPrompt, signature, false)
+ if err == nil {
+ return result, true, nil
+ }
+ if errors.Is(err, errCIAutoFixRoundCapReached) {
+ return ciAutomationDispatchResult{}, false, nil
+ }
+ return ciAutomationDispatchResult{}, false, err
+}
+
+func (s *Service) queueOrReplaceCIAutomationPromptForPR(ctx context.Context, session *models.TaskSession, pr *github.TaskPR, chatPrompt, signature string, allowInsert bool) (ciAutomationDispatchResult, error) {
+ if s.messageQueue == nil {
+ return ciAutomationDispatchResult{}, fmt.Errorf("message queue is not configured")
+ }
+ metadata := ciAutomationMessageMetadataForPR(pr, signature)
+ _, replaced, err := s.messageQueue.QueueMessageWithCoalesceKey(ctx, session.ID, session.TaskID, chatPrompt, "", messagequeue.QueuedByWorkflow, false, nil, metadata, ciAutomationCoalesceKey(pr), allowInsert)
+ if err != nil {
+ if errors.Is(err, messagequeue.ErrEntryNotFound) && !allowInsert {
+ return ciAutomationDispatchResult{}, errCIAutoFixRoundCapReached
+ }
+ return ciAutomationDispatchResult{}, err
+ }
+ s.publishQueueStatusEvent(ctx, session.ID)
+ if replaced {
+ return ciAutomationDispatchResult{kind: ciAutomationDispatchQueuedReplace}, nil
+ }
+ return ciAutomationDispatchResult{kind: ciAutomationDispatchQueuedInsert}, nil
+}
+
func (s *Service) recordCIAutomationUserMessage(ctx context.Context, taskID, sessionID, prompt string) bool {
if s.messageCreator == nil || prompt == "" {
return false
@@ -271,6 +348,27 @@ func ciAutomationMessageMetadata() map[string]interface{} {
return meta
}
+func ciAutomationMessageMetadataForPR(pr *github.TaskPR, signature string) map[string]interface{} {
+ meta := ciAutomationMessageMetadata()
+ meta["automation_kind"] = ciAutomationKindAutoFix
+ meta["ci_auto_fix_key"] = ciAutomationCoalesceKey(pr)
+ meta["feedback_signature"] = signature
+ if pr != nil {
+ meta["repository_id"] = pr.RepositoryID
+ meta["owner"] = pr.Owner
+ meta["repo"] = pr.Repo
+ meta["pr_number"] = pr.PRNumber
+ }
+ return meta
+}
+
+func ciAutomationCoalesceKey(pr *github.TaskPR) string {
+ if pr == nil {
+ return "ci-auto-fix||0"
+ }
+ return fmt.Sprintf("ci-auto-fix|%s|%s|%d", pr.TaskID, pr.RepositoryID, pr.PRNumber)
+}
+
func ciAutomationChatPrompt(prompt string) string {
prompt = strings.TrimSpace(prompt)
if prompt == "" {
@@ -378,6 +476,45 @@ func ciAutomationDuplicateFixAttemptBlocksMerge(state *github.TaskCIPRAutomation
return ciAutomationDuplicateFixAttemptBlocksMergeAt(state, time.Now())
}
+func ciAutomationFixRoundsExhausted(state *github.TaskCIPRAutomationState) bool {
+ if state == nil {
+ return false
+ }
+ return state.AutoFixExhaustedAt != nil || state.AutoFixRoundCount >= ciAutomationMaxFixRounds
+}
+
+func (s *Service) markCIAutoFixExhausted(ctx context.Context, pr *github.TaskPR) {
+ if pr == nil {
+ return
+ }
+ message := fmt.Sprintf("CI auto-fix paused after %d rounds for this PR", ciAutomationMaxFixRounds)
+ s.logger.Warn("CI automation auto-fix round cap reached",
+ zap.String("task_id", pr.TaskID),
+ zap.String("repository_id", pr.RepositoryID),
+ zap.Int("pr_number", pr.PRNumber),
+ zap.Int("max_rounds", ciAutomationMaxFixRounds))
+ if err := s.githubService.MarkTaskCIAutoFixExhausted(context.WithoutCancel(ctx), pr.TaskID, pr.RepositoryID, pr.PRNumber, message); err != nil {
+ s.logger.Debug("record CI auto-fix exhaustion failed", zap.String("task_id", pr.TaskID), zap.Error(err))
+ return
+ }
+ s.publishTaskCIOptionsState(ctx, pr.TaskID)
+}
+
+func (s *Service) publishTaskCIOptionsState(ctx context.Context, taskID string) {
+ if s.githubService == nil || s.eventBus == nil || taskID == "" {
+ return
+ }
+ resp, err := s.githubService.GetTaskCIOptionsResponse(context.WithoutCancel(ctx), taskID)
+ if err != nil {
+ s.logger.Debug("load task CI options for state publish failed", zap.String("task_id", taskID), zap.Error(err))
+ return
+ }
+ event := bus.NewEvent(events.GitHubTaskCIOptionsUpdated, ciAutomationStateEventSource, resp)
+ if err := s.eventBus.Publish(context.WithoutCancel(ctx), events.GitHubTaskCIOptionsUpdated, event); err != nil {
+ s.logger.Debug("publish task CI options state failed", zap.String("task_id", taskID), zap.Error(err))
+ }
+}
+
func ciAutomationDuplicateFixAttemptBlocksMergeAt(state *github.TaskCIPRAutomationState, now time.Time) bool {
if state == nil || state.LastFixEnqueuedAt == nil {
return false
diff --git a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go
index 7009bbe366..0b4e622705 100644
--- a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go
+++ b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go
@@ -1,5 +1,7 @@
package orchestrator
+//revive:disable:file-length-limit // CI automation regression coverage is intentionally scenario-heavy.
+
import (
"context"
"errors"
@@ -7,6 +9,7 @@ import (
"testing"
"time"
+ "github.com/kandev/kandev/internal/events"
"github.com/kandev/kandev/internal/events/bus"
"github.com/kandev/kandev/internal/github"
"github.com/kandev/kandev/internal/orchestrator/executor"
@@ -254,6 +257,15 @@ func TestHandleTaskPRCIAutomationQueuesFixDedupesAndMerges(t *testing.T) {
},
}
svc.SetGitHubService(ghSvc)
+ svc.eventBus = bus.NewMemoryEventBus(testLogger())
+ var ciOptionsEvents []*bus.Event
+ _, err := svc.eventBus.Subscribe(events.GitHubTaskCIOptionsUpdated, func(_ context.Context, event *bus.Event) error {
+ ciOptionsEvents = append(ciOptionsEvents, event)
+ return nil
+ })
+ if err != nil {
+ t.Fatalf("subscribe CI options events: %v", err)
+ }
if err := svc.handleTaskPRCIAutomation(ctx, pr); err != nil {
t.Fatalf("handle auto-fix: %v", err)
@@ -271,6 +283,9 @@ func TestHandleTaskPRCIAutomationQueuesFixDedupesAndMerges(t *testing.T) {
if len(ghSvc.fixAttempts) != 1 {
t.Fatalf("expected one fix attempt, got %d", len(ghSvc.fixAttempts))
}
+ if len(ciOptionsEvents) != 1 || ciOptionsEvents[0].Source != ciAutomationStateEventSource {
+ t.Fatalf("expected one CI options state refresh event, got %+v", ciOptionsEvents)
+ }
_, signature := encodeCIAutomationCheckpoint(ciAutomationCurrentCheckpoint(ghSvc.prFeedback))
ghSvc.ciPRState = &github.TaskCIPRAutomationState{LastFixSignature: signature, LastFixCheckpointJSON: ghSvc.fixAttempts[0].CheckpointJSON}
@@ -638,6 +653,462 @@ func TestHandleTaskPRCIAutomationDuplicateFixAttemptBlocksMerge(t *testing.T) {
}
}
+func TestHandleTaskPRCIAutomationCoalescesQueuedAutoFixForRunningSession(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
+ svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+ now := time.Now().UTC()
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ LastSyncedAt: &now,
+ }
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ EffectiveAutoFixPrompt: "Fix the PR\n\n{{pr.feedback}}",
+ },
+ triggerPRSyncAllPRs: []*github.TaskPR{pr},
+ prFeedback: &github.PRFeedback{
+ Checks: []github.CheckRun{{Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}},
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ if err := svc.handleTaskPRCIAutomation(ctx, pr); err != nil {
+ t.Fatalf("handle first auto-fix: %v", err)
+ }
+ if len(ghSvc.fixAttempts) != 1 {
+ t.Fatalf("expected first fix attempt, got %+v", ghSvc.fixAttempts)
+ }
+ ghSvc.ciPRState = &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ LastFixSignature: ghSvc.fixAttempts[0].Signature,
+ LastFixCheckpointJSON: ghSvc.fixAttempts[0].CheckpointJSON,
+ LastFixEnqueuedAt: &now,
+ LastFixSessionID: ptrString("session-1"),
+ }
+ ghSvc.prFeedback = &github.PRFeedback{
+ Checks: []github.CheckRun{
+ {Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"},
+ {Name: "lint", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/lint"},
+ },
+ }
+
+ if err := svc.handleTaskPRCIAutomation(ctx, pr); err != nil {
+ t.Fatalf("handle second auto-fix: %v", err)
+ }
+ status := svc.messageQueue.GetStatus(ctx, "session-1")
+ if status.Count != 1 {
+ t.Fatalf("expected queued CI auto-fix to be replaced, got %+v", status)
+ }
+ if !strings.Contains(status.Entries[0].Content, "lint") {
+ t.Fatalf("expected queued prompt to contain latest CI feedback, got %q", status.Entries[0].Content)
+ }
+ if strings.Contains(status.Entries[0].Content, "https://ci/unit") {
+ t.Fatalf("expected replacement to avoid stale appended feedback, got %q", status.Entries[0].Content)
+ }
+}
+
+func TestHandleTaskPRCIAutomationStopsBeforeEleventhAutoFixRound(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
+ svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+ now := time.Now().UTC()
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ LastSyncedAt: &now,
+ }
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ EffectiveAutoFixPrompt: "Fix the PR\n\n{{pr.feedback}}",
+ },
+ triggerPRSyncAllPRs: []*github.TaskPR{pr},
+ prFeedback: &github.PRFeedback{
+ Checks: []github.CheckRun{{Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}},
+ },
+ ciPRState: &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ AutoFixRoundCount: ciAutomationMaxFixRounds,
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ if err := svc.handleTaskPRCIAutomation(ctx, pr); err != nil {
+ t.Fatalf("handle capped auto-fix: %v", err)
+ }
+ if status := svc.messageQueue.GetStatus(ctx, "session-1"); status.Count != 0 {
+ t.Fatalf("expected no 11th auto-fix prompt, got %+v", status)
+ }
+ if len(ghSvc.fixAttempts) != 0 {
+ t.Fatalf("expected no fix attempt after cap, got %+v", ghSvc.fixAttempts)
+ }
+ if len(ghSvc.ciExhausted) != 1 || ghSvc.ciExhausted[0].LastError == nil || !strings.Contains(*ghSvc.ciExhausted[0].LastError, "10 rounds") {
+ t.Fatalf("expected exhausted CI state, got %+v", ghSvc.ciExhausted)
+ }
+}
+
+func TestHandleTaskPRCIAutomationSkipsAlreadyExhaustedPRBeforeFeedbackFetch(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
+ svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+ exhaustedAt := time.Now().UTC()
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ ChecksState: "failure",
+ }
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ EffectiveAutoFixPrompt: "Fix the PR\n\n{{pr.feedback}}",
+ },
+ ciPRState: &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ AutoFixRoundCount: ciAutomationMaxFixRounds,
+ AutoFixExhaustedAt: &exhaustedAt,
+ },
+ prFeedback: &github.PRFeedback{
+ Checks: []github.CheckRun{{Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}},
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ if err := svc.handleTaskPRCIAutomationWithRefresh(ctx, pr, false); err != nil {
+ t.Fatalf("handle exhausted auto-fix: %v", err)
+ }
+ if ghSvc.prFeedbackCalls != 0 {
+ t.Fatalf("expected exhausted PR to skip full feedback fetch, got %d calls", ghSvc.prFeedbackCalls)
+ }
+ if len(ghSvc.ciExhausted) != 0 {
+ t.Fatalf("expected no repeated exhaustion write, got %+v", ghSvc.ciExhausted)
+ }
+ if status := svc.messageQueue.GetStatus(ctx, "session-1"); status.Count != 0 {
+ t.Fatalf("expected no queued prompt for exhausted PR, got %+v", status)
+ }
+}
+
+func TestHandleTaskPRCIAutomationAutoMergeRunsAfterAutoFixExhaustion(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
+ svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+ now := time.Now().UTC()
+ exhaustedAt := now.Add(-time.Minute)
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ ChecksState: "success",
+ ReviewState: "approved",
+ MergeableState: "clean",
+ LastSyncedAt: &now,
+ }
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ AutoMergeEnabled: true,
+ },
+ ciPRState: &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ AutoFixRoundCount: ciAutomationMaxFixRounds,
+ AutoFixExhaustedAt: &exhaustedAt,
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ if err := svc.handleTaskPRCIAutomationWithRefresh(ctx, pr, false); err != nil {
+ t.Fatalf("handle exhausted auto-merge: %v", err)
+ }
+ if ghSvc.prFeedbackCalls != 0 {
+ t.Fatalf("expected exhausted auto-fix to skip feedback fetch, got %d calls", ghSvc.prFeedbackCalls)
+ }
+ if ghSvc.mergeCalls != 1 {
+ t.Fatalf("expected auto-merge after exhausted auto-fix, got %d calls", ghSvc.mergeCalls)
+ }
+}
+
+func TestDispatchCIAutomationPromptForPRIdleRoundCapUsesDedicatedError(t *testing.T) {
+ ctx := context.Background()
+ svc := &Service{}
+ session := &models.TaskSession{
+ ID: "session-1",
+ TaskID: "task-1",
+ State: models.TaskSessionStateIdle,
+ }
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ }
+
+ _, err := svc.dispatchCIAutomationPromptForPR(ctx, session, pr, "Fix the PR", "signature", false)
+ if !errors.Is(err, errCIAutoFixRoundCapReached) {
+ t.Fatalf("expected auto-fix round cap error, got %v", err)
+ }
+ if errors.Is(err, messagequeue.ErrEntryNotFound) {
+ t.Fatalf("round cap should not reuse queue entry-not-found sentinel")
+ }
+}
+
+func TestHandleTaskPRCIAutomationAtRoundCapReplacesPendingAutoFix(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
+ svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+ now := time.Now().UTC()
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ LastSyncedAt: &now,
+ }
+ _, _, err := svc.messageQueue.QueueMessageWithCoalesceKey(ctx, "session-1", "task-1", "@ci-auto-fix\n\nold feedback", "", messagequeue.QueuedByWorkflow, false, nil, ciAutomationMessageMetadataForPR(pr, "old"), ciAutomationCoalesceKey(pr), true)
+ if err != nil {
+ t.Fatalf("seed pending auto-fix: %v", err)
+ }
+ previous := ciAutomationCurrentCheckpoint(&github.PRFeedback{
+ Checks: []github.CheckRun{{Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}},
+ })
+ previousJSON, previousSignature := encodeCIAutomationCheckpoint(previous)
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ EffectiveAutoFixPrompt: "Fix the PR\n\n{{pr.feedback}}",
+ },
+ triggerPRSyncAllPRs: []*github.TaskPR{pr},
+ prFeedback: &github.PRFeedback{
+ Checks: []github.CheckRun{
+ {Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"},
+ {Name: "lint", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/lint"},
+ },
+ },
+ ciPRState: &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ LastFixSignature: previousSignature,
+ LastFixCheckpointJSON: previousJSON,
+ LastFixEnqueuedAt: &now,
+ AutoFixRoundCount: ciAutomationMaxFixRounds,
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ if err := svc.handleTaskPRCIAutomation(ctx, pr); err != nil {
+ t.Fatalf("handle capped replacement: %v", err)
+ }
+ status := svc.messageQueue.GetStatus(ctx, "session-1")
+ if status.Count != 1 || !strings.Contains(status.Entries[0].Content, "lint") || strings.Contains(status.Entries[0].Content, "old feedback") {
+ t.Fatalf("expected pending auto-fix replacement with latest feedback, got %+v", status)
+ }
+ if len(ghSvc.fixAttempts) != 1 || ghSvc.fixAttempts[0].IncrementRound {
+ t.Fatalf("expected replacement checkpoint without another round, got %+v", ghSvc.fixAttempts)
+ }
+ if len(ghSvc.ciExhausted) != 0 {
+ t.Fatalf("expected pending round replacement not exhaustion, got %+v", ghSvc.ciExhausted)
+ }
+}
+
+func TestHandleTaskPRCIAutomationAtRoundCapReplacesPendingAutoFixForWaitingSession(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateWaitingForInput)
+ agentMgr := &mockAgentManager{isAgentRunning: true, repoForExecutionLookup: repo, promptDone: make(chan struct{})}
+ svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
+ svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+ seedExecutorRunning(t, repo, "session-1", "task-1", "exec-1")
+ session, err := repo.GetTaskSession(ctx, "session-1")
+ if err != nil {
+ t.Fatalf("load session: %v", err)
+ }
+ session.AgentExecutionID = "exec-1"
+ if err := repo.UpdateTaskSession(ctx, session); err != nil {
+ t.Fatalf("update session execution: %v", err)
+ }
+ now := time.Now().UTC()
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ LastSyncedAt: &now,
+ }
+ _, _, err = svc.messageQueue.QueueMessageWithCoalesceKey(ctx, "session-1", "task-1", "@ci-auto-fix\n\nold feedback", "", messagequeue.QueuedByWorkflow, false, nil, ciAutomationMessageMetadataForPR(pr, "old"), ciAutomationCoalesceKey(pr), true)
+ if err != nil {
+ t.Fatalf("seed pending auto-fix: %v", err)
+ }
+ previous := ciAutomationCurrentCheckpoint(&github.PRFeedback{
+ Checks: []github.CheckRun{{Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}},
+ })
+ previousJSON, previousSignature := encodeCIAutomationCheckpoint(previous)
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ EffectiveAutoFixPrompt: "Fix the PR\n\n{{pr.feedback}}",
+ },
+ triggerPRSyncAllPRs: []*github.TaskPR{pr},
+ prFeedback: &github.PRFeedback{
+ Checks: []github.CheckRun{
+ {Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"},
+ {Name: "lint", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/lint"},
+ },
+ },
+ ciPRState: &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ LastFixSignature: previousSignature,
+ LastFixCheckpointJSON: previousJSON,
+ LastFixEnqueuedAt: &now,
+ AutoFixRoundCount: ciAutomationMaxFixRounds,
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ if err := svc.handleTaskPRCIAutomation(ctx, pr); err != nil {
+ t.Fatalf("handle capped waiting replacement: %v", err)
+ }
+ select {
+ case <-agentMgr.promptDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for replaced auto-fix prompt dispatch")
+ }
+ status := svc.messageQueue.GetStatus(ctx, "session-1")
+ if status.Count != 0 {
+ t.Fatalf("expected replaced auto-fix prompt to drain immediately, got %+v", status)
+ }
+ if len(agentMgr.capturedPrompts) != 1 || !strings.Contains(agentMgr.capturedPrompts[0], "lint") || strings.Contains(agentMgr.capturedPrompts[0], "old feedback") {
+ t.Fatalf("expected latest replaced auto-fix prompt to dispatch, got %+v", agentMgr.capturedPrompts)
+ }
+ if len(ghSvc.fixAttempts) != 1 || ghSvc.fixAttempts[0].IncrementRound {
+ t.Fatalf("expected replacement checkpoint without another round, got %+v", ghSvc.fixAttempts)
+ }
+ if len(ghSvc.ciExhausted) != 0 {
+ t.Fatalf("expected pending round replacement not exhaustion, got %+v", ghSvc.ciExhausted)
+ }
+}
+
+func TestHandleTaskPRCIAutomationReplacesPendingAutoFixBeforeDirectPrompt(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateWaitingForInput)
+ agentMgr := &mockAgentManager{isAgentRunning: true, repoForExecutionLookup: repo, promptDone: make(chan struct{})}
+ svc := createTestServiceWithAgent(repo, newMockStepGetter(), newMockTaskRepo(), agentMgr)
+ svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+ seedExecutorRunning(t, repo, "session-1", "task-1", "exec-1")
+ session, err := repo.GetTaskSession(ctx, "session-1")
+ if err != nil {
+ t.Fatalf("load session: %v", err)
+ }
+ session.AgentExecutionID = "exec-1"
+ if err := repo.UpdateTaskSession(ctx, session); err != nil {
+ t.Fatalf("update session execution: %v", err)
+ }
+ now := time.Now().UTC()
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ LastSyncedAt: &now,
+ }
+ _, _, err = svc.messageQueue.QueueMessageWithCoalesceKey(ctx, "session-1", "task-1", "@ci-auto-fix\n\nold feedback", "", messagequeue.QueuedByWorkflow, false, nil, ciAutomationMessageMetadataForPR(pr, "old"), ciAutomationCoalesceKey(pr), true)
+ if err != nil {
+ t.Fatalf("seed pending auto-fix: %v", err)
+ }
+ previous := ciAutomationCurrentCheckpoint(&github.PRFeedback{
+ Checks: []github.CheckRun{{Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}},
+ })
+ previousJSON, previousSignature := encodeCIAutomationCheckpoint(previous)
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ EffectiveAutoFixPrompt: "Fix the PR\n\n{{pr.feedback}}",
+ },
+ triggerPRSyncAllPRs: []*github.TaskPR{pr},
+ prFeedback: &github.PRFeedback{
+ Checks: []github.CheckRun{
+ {Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"},
+ {Name: "lint", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/lint"},
+ },
+ },
+ ciPRState: &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ LastFixSignature: previousSignature,
+ LastFixCheckpointJSON: previousJSON,
+ LastFixEnqueuedAt: &now,
+ AutoFixRoundCount: 3,
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ if err := svc.handleTaskPRCIAutomation(ctx, pr); err != nil {
+ t.Fatalf("handle waiting replacement: %v", err)
+ }
+ select {
+ case <-agentMgr.promptDone:
+ case <-time.After(time.Second):
+ t.Fatal("timed out waiting for replaced auto-fix prompt dispatch")
+ }
+ status := svc.messageQueue.GetStatus(ctx, "session-1")
+ if status.Count != 0 {
+ t.Fatalf("expected replaced auto-fix prompt to drain before direct prompt, got %+v", status)
+ }
+ if len(agentMgr.capturedPrompts) != 1 || !strings.Contains(agentMgr.capturedPrompts[0], "lint") || strings.Contains(agentMgr.capturedPrompts[0], "old feedback") {
+ t.Fatalf("expected latest replaced auto-fix prompt to dispatch, got %+v", agentMgr.capturedPrompts)
+ }
+ if len(ghSvc.fixAttempts) != 1 || ghSvc.fixAttempts[0].IncrementRound {
+ t.Fatalf("expected replacement checkpoint without consuming another round, got %+v", ghSvc.fixAttempts)
+ }
+}
+
func TestCIAutomationFindMatchingPRRequiresRepositoryIDWhenPresent(t *testing.T) {
target := &github.TaskPR{
TaskID: "task-1",
@@ -672,7 +1143,33 @@ func TestCIAutomationFindMatchingPRRequiresRepositoryIDWhenPresent(t *testing.T)
}
}
-func TestDispatchCIAutomationPromptDoesNotRecordUserMessageWhenQueueFails(t *testing.T) {
+func TestDispatchCIAutomationPromptForPRRunningRoundCapUsesDedicatedError(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
+ svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+ session, err := repo.GetTaskSession(ctx, "session-1")
+ if err != nil {
+ t.Fatalf("load session: %v", err)
+ }
+ pr := &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ }
+
+ _, err = svc.dispatchCIAutomationPromptForPR(ctx, session, pr, "Fix the PR", "signature", false)
+ if !errors.Is(err, errCIAutoFixRoundCapReached) {
+ t.Fatalf("expected auto-fix round cap error, got %v", err)
+ }
+ if errors.Is(err, messagequeue.ErrEntryNotFound) {
+ t.Fatalf("round cap should not expose queue entry-not-found sentinel")
+ }
+}
+
+func TestDispatchCIAutomationPromptForPRDoesNotRecordUserMessageWhenQueueFails(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
@@ -680,12 +1177,13 @@ func TestDispatchCIAutomationPromptDoesNotRecordUserMessageWhenQueueFails(t *tes
svc.messageQueue = nil
messageCreator := &mockMessageCreator{}
svc.messageCreator = messageCreator
+ pr := &github.TaskPR{TaskID: "task-1", RepositoryID: "repo-1", Owner: "acme", Repo: "widget", PRNumber: 42}
- err := svc.dispatchCIAutomationPrompt(ctx, &models.TaskSession{
+ _, err := svc.dispatchCIAutomationPromptForPR(ctx, &models.TaskSession{
ID: "session-1",
TaskID: "task-1",
State: models.TaskSessionStateRunning,
- }, "Fix the PR")
+ }, pr, "Fix the PR", "signature", true)
if err == nil {
t.Fatal("expected queue failure")
}
@@ -694,7 +1192,7 @@ func TestDispatchCIAutomationPromptDoesNotRecordUserMessageWhenQueueFails(t *tes
}
}
-func TestDispatchCIAutomationPromptQueuesWhenRunningUserMessageCannotBeRecorded(t *testing.T) {
+func TestDispatchCIAutomationPromptForPRQueuesWhenRunningUserMessageCannotBeRecorded(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateRunning)
@@ -706,11 +1204,15 @@ func TestDispatchCIAutomationPromptQueuesWhenRunningUserMessageCannotBeRecorded(
if err != nil {
t.Fatalf("load session: %v", err)
}
+ pr := &github.TaskPR{TaskID: "task-1", RepositoryID: "repo-1", Owner: "acme", Repo: "widget", PRNumber: 42}
- err = svc.dispatchCIAutomationPrompt(ctx, session, "Fix the PR")
+ result, err := svc.dispatchCIAutomationPromptForPR(ctx, session, pr, "Fix the PR", "signature", true)
if err != nil {
t.Fatalf("expected queued CI automation prompt, got %v", err)
}
+ if !result.consumesRound() {
+ t.Fatalf("expected new queued prompt to consume a round, got %+v", result)
+ }
status := svc.messageQueue.GetStatus(ctx, "session-1")
if status.Count != 1 {
t.Fatalf("expected queued prompt when user message cannot be recorded yet, got %d", status.Count)
@@ -720,7 +1222,7 @@ func TestDispatchCIAutomationPromptQueuesWhenRunningUserMessageCannotBeRecorded(
}
}
-func TestDispatchCIAutomationPromptRecordsUserMessageBeforeDirectPrompt(t *testing.T) {
+func TestDispatchCIAutomationPromptForPRRecordsUserMessageBeforeDirectPrompt(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateWaitingForInput)
@@ -739,10 +1241,15 @@ func TestDispatchCIAutomationPromptRecordsUserMessageBeforeDirectPrompt(t *testi
if err := repo.UpdateTaskSession(ctx, session); err != nil {
t.Fatalf("update session execution: %v", err)
}
+ pr := &github.TaskPR{TaskID: "task-1", RepositoryID: "repo-1", Owner: "acme", Repo: "widget", PRNumber: 42}
- if err := svc.dispatchCIAutomationPrompt(ctx, session, "Fix the PR"); err != nil {
+ result, err := svc.dispatchCIAutomationPromptForPR(ctx, session, pr, "Fix the PR", "signature", true)
+ if err != nil {
t.Fatalf("dispatch direct prompt: %v", err)
}
+ if !result.consumesRound() {
+ t.Fatalf("expected direct prompt to consume a round, got %+v", result)
+ }
if len(messageCreator.userMessages) != 1 {
t.Fatalf("expected one visible CI automation user message, got %d", len(messageCreator.userMessages))
}
@@ -757,7 +1264,7 @@ func TestDispatchCIAutomationPromptRecordsUserMessageBeforeDirectPrompt(t *testi
}
}
-func TestDispatchCIAutomationPromptRecordsUserMessageBeforeDirectPromptFailure(t *testing.T) {
+func TestDispatchCIAutomationPromptForPRRecordsUserMessageBeforeDirectPromptFailure(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateWaitingForInput)
@@ -780,8 +1287,9 @@ func TestDispatchCIAutomationPromptRecordsUserMessageBeforeDirectPromptFailure(t
if err := repo.UpdateTaskSession(ctx, session); err != nil {
t.Fatalf("update session execution: %v", err)
}
+ pr := &github.TaskPR{TaskID: "task-1", RepositoryID: "repo-1", Owner: "acme", Repo: "widget", PRNumber: 42}
- err = svc.dispatchCIAutomationPrompt(ctx, session, "Fix the PR")
+ _, err = svc.dispatchCIAutomationPromptForPR(ctx, session, pr, "Fix the PR", "signature", true)
if err == nil || !strings.Contains(err.Error(), "agent rejected prompt") {
t.Fatalf("expected prompt failure, got %v", err)
}
@@ -868,6 +1376,52 @@ func TestHandleTaskPRCIAutomationRecordsErrorWhenNoPromptableSession(t *testing.
}
}
+func TestHandleTaskPRCIAutomationMarksExhaustedWithoutPromptableSession(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedTaskAndSession(t, repo, "task-1", "session-1", models.TaskSessionStateCompleted)
+ svc := createTestService(repo, newMockStepGetter(), newMockTaskRepo())
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ EffectiveAutoFixPrompt: "Fix the PR",
+ },
+ prFeedback: &github.PRFeedback{
+ Checks: []github.CheckRun{{Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}},
+ },
+ ciPRState: &github.TaskCIPRAutomationState{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ PRNumber: 42,
+ AutoFixRoundCount: ciAutomationMaxFixRounds,
+ },
+ }
+ svc.SetGitHubService(ghSvc)
+
+ err := svc.handleTaskPRCIAutomation(ctx, &github.TaskPR{
+ TaskID: "task-1",
+ RepositoryID: "repo-1",
+ Owner: "acme",
+ Repo: "widget",
+ PRNumber: 42,
+ State: "open",
+ ChecksState: "failure",
+ })
+ if err != nil {
+ t.Fatalf("handle auto-fix: %v", err)
+ }
+ if len(ghSvc.ciExhausted) != 1 {
+ t.Fatalf("expected exhausted CI state without promptable session, got %+v", ghSvc.ciExhausted)
+ }
+ if len(ghSvc.ciErrors) != 0 {
+ t.Fatalf("expected cap exhaustion instead of no-session error, got %+v", ghSvc.ciErrors)
+ }
+ if len(ghSvc.fixAttempts) != 0 {
+ t.Fatalf("expected no fix attempt without promptable session, got %+v", ghSvc.fixAttempts)
+ }
+}
+
func TestHandleTaskPRCIAutomationMergesWhenStateReadFails(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
@@ -1033,6 +1587,29 @@ func TestHandleTaskCIOptionsUpdatedStartsAutomationForTaskPRs(t *testing.T) {
}
}
+func TestHandleTaskCIOptionsUpdatedIgnoresStateRefreshEvents(t *testing.T) {
+ ctx := context.Background()
+ svc := createTestService(setupTestRepo(t), newMockStepGetter(), newMockTaskRepo())
+ ghSvc := &mockGitHubService{
+ ciOptionsResp: &github.TaskCIOptionsResponse{TaskID: "task-1"},
+ }
+ svc.SetGitHubService(ghSvc)
+
+ err := svc.handleTaskCIOptionsUpdated(ctx, &bus.Event{
+ Source: ciAutomationStateEventSource,
+ Data: &github.TaskCIOptionsResponse{
+ TaskID: "task-1",
+ AutoFixEnabled: true,
+ },
+ })
+ if err != nil {
+ t.Fatalf("handle CI options state refresh: %v", err)
+ }
+ if ghSvc.triggerPRSyncAllCalls != 0 {
+ t.Fatalf("state refresh should not start automation sync, got %d calls", ghSvc.triggerPRSyncAllCalls)
+ }
+}
+
func TestHandleTaskCIOptionsUpdatedRecordsSyncFailureForLinkedPRs(t *testing.T) {
ctx := context.Background()
svc := createTestService(setupTestRepo(t), newMockStepGetter(), newMockTaskRepo())
@@ -1216,3 +1793,7 @@ func waitForCIAutomationIdle(t *testing.T, svc *Service, key string, timeout tim
}
}
}
+
+func ptrString(value string) *string {
+ return &value
+}
diff --git a/apps/backend/internal/orchestrator/event_handlers_github_test.go b/apps/backend/internal/orchestrator/event_handlers_github_test.go
index c5cd85869a..6c0c24fd5b 100644
--- a/apps/backend/internal/orchestrator/event_handlers_github_test.go
+++ b/apps/backend/internal/orchestrator/event_handlers_github_test.go
@@ -76,12 +76,14 @@ type mockGitHubService struct {
ciPRState *github.TaskCIPRAutomationState
ciPRStateErr error
prFeedback *github.PRFeedback
+ prFeedbackCalls int
fixAttempts []github.TaskCIFixAttempt
fixCheckpointRefresh []github.TaskCIFixAttempt
mergeAttempts []github.TaskCIMergeAttempt
mergeCalls int
mergeErr error
ciErrors []github.TaskCIPRAutomationState
+ ciExhausted []github.TaskCIPRAutomationState
}
func (m *mockGitHubService) Client() github.Client { return m.client }
@@ -181,10 +183,22 @@ func (m *mockGitHubService) RecordTaskCIError(_ context.Context, taskID, reposit
})
return nil
}
+func (m *mockGitHubService) MarkTaskCIAutoFixExhausted(_ context.Context, taskID, repositoryID string, prNumber int, message string) error {
+ now := time.Now().UTC()
+ m.ciExhausted = append(m.ciExhausted, github.TaskCIPRAutomationState{
+ TaskID: taskID,
+ RepositoryID: repositoryID,
+ PRNumber: prNumber,
+ AutoFixExhaustedAt: &now,
+ LastError: &message,
+ })
+ return nil
+}
func (m *mockGitHubService) ClearTaskCIError(context.Context, string, string, int) error {
return nil
}
func (m *mockGitHubService) GetPRFeedback(context.Context, string, string, int) (*github.PRFeedback, error) {
+ m.prFeedbackCalls++
if m.prFeedback != nil {
return m.prFeedback, nil
}
diff --git a/apps/backend/internal/orchestrator/event_handlers_test.go b/apps/backend/internal/orchestrator/event_handlers_test.go
index 46f7931f77..a287dfbee6 100644
--- a/apps/backend/internal/orchestrator/event_handlers_test.go
+++ b/apps/backend/internal/orchestrator/event_handlers_test.go
@@ -827,6 +827,59 @@ func TestExecuteQueuedMessage_RequeuesTransientPromptFailure(t *testing.T) {
}
}
+func TestExecuteQueuedMessage_RequeuesCoalescedMessageWithOriginalSender(t *testing.T) {
+ ctx := context.Background()
+ repo := setupTestRepo(t)
+ seedSession(t, repo, "t1", "s1", "step1")
+
+ session, err := repo.GetTaskSession(ctx, "s1")
+ if err != nil {
+ t.Fatalf("failed to get session: %v", err)
+ }
+ session.State = models.TaskSessionStateWaitingForInput
+ session.AgentExecutionID = "exec-1"
+ seedExecutorRunning(t, repo, session.ID, session.TaskID, "exec-1")
+ if err := repo.UpdateTaskSession(ctx, session); err != nil {
+ t.Fatalf("failed to update session: %v", err)
+ }
+
+ taskRepo := newMockTaskRepo()
+ agentMgr := &mockAgentManager{
+ isAgentRunning: true,
+ promptErr: errors.New("agent stream disconnected: read tcp [::1]:56463->[::1]:10002: use of closed network connection"),
+ }
+ svc := createTestServiceWithAgent(repo, newMockStepGetter(), taskRepo, agentMgr)
+ svc.executor = executor.NewExecutor(agentMgr, repo, testLogger(), executor.ExecutorConfig{})
+
+ queuedMsg := &messagequeue.QueuedMessage{
+ ID: "q1",
+ SessionID: "s1",
+ TaskID: "t1",
+ Content: "hello",
+ Metadata: map[string]interface{}{messagequeue.MetadataCoalesceKey: "ci-key"},
+ QueuedBy: messagequeue.QueuedByWorkflow,
+ }
+ if _, _, err := svc.messageQueue.QueueMessageWithCoalesceKey(ctx, "s1", "t1", "newer ci feedback", "", messagequeue.QueuedByWorkflow, false, nil, map[string]interface{}{messagequeue.MetadataCoalesceKey: "ci-key"}, "ci-key", true); err != nil {
+ t.Fatalf("seed newer coalesced message: %v", err)
+ }
+
+ svc.executeQueuedMessage("s1", queuedMsg)
+
+ status := svc.messageQueue.GetStatus(ctx, "s1")
+ if status.Count != 1 {
+ t.Fatalf("expected coalesced retry to replace existing pending entry, count=%d", status.Count)
+ }
+ if status.Entries[0].Content != "hello" {
+ t.Fatalf("expected transient retry to remain coalesced, got %q", status.Entries[0].Content)
+ }
+ if status.Entries[0].QueuedBy != messagequeue.QueuedByWorkflow {
+ t.Fatalf("expected original queued_by to be preserved, got %q", status.Entries[0].QueuedBy)
+ }
+ if status.Entries[0].Metadata[messagequeue.MetadataCoalesceKey] != "ci-key" {
+ t.Fatalf("expected coalesce metadata to be preserved, got %+v", status.Entries[0].Metadata)
+ }
+}
+
func TestExecuteQueuedMessage_FiresOnTurnStart(t *testing.T) {
ctx := context.Background()
repo := setupTestRepo(t)
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository.go b/apps/backend/internal/orchestrator/messagequeue/repository.go
index d59251586b..3789e6b4b7 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository.go
@@ -15,6 +15,12 @@ type Repository interface {
// or inserted (false). Honors maxPerSession when inserting.
AppendOrInsertTail(ctx context.Context, sessionID, taskID, content, model, queuedBy string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, maxPerSession int) (*QueuedMessage, bool, error)
+ // InsertOrReplaceByCoalesceKey replaces an existing queued entry with the
+ // same session, queued_by, and metadata coalesce key. If no matching entry
+ // exists, it inserts a new tail entry when allowInsert is true. Returns
+ // ErrEntryNotFound when allowInsert is false and no matching entry exists.
+ InsertOrReplaceByCoalesceKey(ctx context.Context, msg *QueuedMessage, coalesceKey string, maxPerSession int, allowInsert bool) (*QueuedMessage, bool, error)
+
// ListBySession returns all entries for a session ordered by position ascending.
ListBySession(ctx context.Context, sessionID string) ([]QueuedMessage, error)
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go
index 257e10e819..a2c1db4db9 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_memory.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_memory.go
@@ -86,6 +86,39 @@ func (r *memoryRepository) AppendOrInsertTail(_ context.Context, sessionID, task
return msg, false, nil
}
+func (r *memoryRepository) InsertOrReplaceByCoalesceKey(_ context.Context, msg *QueuedMessage, coalesceKey string, maxPerSession int, allowInsert bool) (*QueuedMessage, bool, error) {
+ r.mu.Lock()
+ defer r.mu.Unlock()
+
+ for _, existing := range r.entries[msg.SessionID] {
+ if existing.QueuedBy != msg.QueuedBy {
+ continue
+ }
+ if metadataString(existing.Metadata, MetadataCoalesceKey) != coalesceKey {
+ continue
+ }
+ if msg.QueuedAt.IsZero() {
+ msg.QueuedAt = time.Now().UTC()
+ }
+ existing.TaskID = msg.TaskID
+ existing.Content = msg.Content
+ existing.Model = msg.Model
+ existing.PlanMode = msg.PlanMode
+ existing.Attachments = msg.Attachments
+ existing.Metadata = msg.Metadata
+ existing.QueuedAt = msg.QueuedAt
+ out := *existing
+ return &out, true, nil
+ }
+ if !allowInsert {
+ return nil, false, ErrEntryNotFound
+ }
+ if err := r.insertLocked(msg, maxPerSession); err != nil {
+ return nil, false, err
+ }
+ return msg, false, nil
+}
+
func (r *memoryRepository) ListBySession(_ context.Context, sessionID string) ([]QueuedMessage, error) {
r.mu.Lock()
defer r.mu.Unlock()
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
index b8c9a7a0c2..3c74f89ff3 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite.go
@@ -184,6 +184,131 @@ func (r *sqliteRepository) AppendOrInsertTail(ctx context.Context, sessionID, ta
return msg, false, nil
}
+func (r *sqliteRepository) InsertOrReplaceByCoalesceKey(ctx context.Context, msg *QueuedMessage, coalesceKey string, maxPerSession int, allowInsert bool) (*QueuedMessage, bool, error) {
+ tx, err := r.db.BeginTxx(ctx, nil)
+ if err != nil {
+ return nil, false, fmt.Errorf("begin coalesce tx: %w", err)
+ }
+ defer func() { _ = tx.Rollback() }()
+
+ existing, err := r.findCoalesced(ctx, tx, msg.SessionID, msg.QueuedBy, coalesceKey)
+ if err != nil {
+ return nil, false, err
+ }
+ if existing != nil {
+ updated, err := r.replaceCoalesced(ctx, tx, existing, msg)
+ if err != nil {
+ return nil, false, err
+ }
+ if err := tx.Commit(); err != nil {
+ return nil, false, err
+ }
+ return updated, true, nil
+ }
+ if !allowInsert {
+ return nil, false, ErrEntryNotFound
+ }
+ if err := r.insertCoalesced(ctx, tx, msg, maxPerSession); err != nil {
+ return nil, false, err
+ }
+ if err := tx.Commit(); err != nil {
+ return nil, false, err
+ }
+ return msg, false, nil
+}
+
+func (r *sqliteRepository) replaceCoalesced(ctx context.Context, tx *sqlx.Tx, existing, msg *QueuedMessage) (*QueuedMessage, error) {
+ if msg.QueuedAt.IsZero() {
+ msg.QueuedAt = time.Now().UTC()
+ }
+ attachmentsJSON, err := marshalAttachments(msg.Attachments)
+ if err != nil {
+ return nil, err
+ }
+ metadataJSON, err := marshalMetadata(msg.Metadata)
+ if err != nil {
+ return nil, err
+ }
+ res, err := tx.ExecContext(ctx, r.db.Rebind(`
+ UPDATE queued_messages
+ SET task_id = ?, content = ?, model = ?, plan_mode = ?,
+ attachments_json = ?, metadata_json = ?, queued_at = ?
+ WHERE id = ? AND session_id = ? AND queued_by = ?
+ `),
+ msg.TaskID, msg.Content, msg.Model, boolToInt(msg.PlanMode),
+ attachmentsJSON, metadataJSON, msg.QueuedAt,
+ existing.ID, msg.SessionID, msg.QueuedBy,
+ )
+ if err != nil {
+ return nil, fmt.Errorf("replace coalesced queued: %w", err)
+ }
+ rows, err := res.RowsAffected()
+ if err != nil {
+ return nil, fmt.Errorf("replace coalesced rows affected: %w", err)
+ }
+ if rows == 0 {
+ return nil, ErrEntryNotFound
+ }
+ existing.TaskID = msg.TaskID
+ existing.Content = msg.Content
+ existing.Model = msg.Model
+ existing.PlanMode = msg.PlanMode
+ existing.Attachments = msg.Attachments
+ existing.Metadata = msg.Metadata
+ existing.QueuedAt = msg.QueuedAt
+ return existing, nil
+}
+
+func (r *sqliteRepository) insertCoalesced(ctx context.Context, tx *sqlx.Tx, msg *QueuedMessage, maxPerSession int) error {
+ if err := r.ensureQueueCapacity(ctx, tx, msg.SessionID, maxPerSession); err != nil {
+ return err
+ }
+ var maxPos sql.NullInt64
+ if err := tx.GetContext(ctx, &maxPos, r.db.Rebind(`SELECT MAX(position) FROM queued_messages WHERE session_id = ?`), msg.SessionID); err != nil {
+ return fmt.Errorf("max position: %w", err)
+ }
+ msg.Position = maxPos.Int64 + 1
+ if msg.ID == "" {
+ msg.ID = uuid.New().String()
+ }
+ if msg.QueuedAt.IsZero() {
+ msg.QueuedAt = time.Now().UTC()
+ }
+ attachmentsJSON, err := marshalAttachments(msg.Attachments)
+ if err != nil {
+ return err
+ }
+ metadataJSON, err := marshalMetadata(msg.Metadata)
+ if err != nil {
+ return err
+ }
+ if _, err := tx.ExecContext(ctx, r.db.Rebind(`
+ INSERT INTO queued_messages
+ (id, session_id, task_id, position, content, model, plan_mode, attachments_json, metadata_json, queued_at, queued_by)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+ `),
+ msg.ID, msg.SessionID, msg.TaskID, msg.Position, msg.Content, msg.Model,
+ boolToInt(msg.PlanMode), attachmentsJSON, metadataJSON, msg.QueuedAt, msg.QueuedBy,
+ ); err != nil {
+ return fmt.Errorf("insert coalesced queued: %w", err)
+ }
+ return nil
+}
+
+func (r *sqliteRepository) ensureQueueCapacity(ctx context.Context, tx *sqlx.Tx, sessionID string, maxPerSession int) error {
+ if maxPerSession <= 0 {
+ return nil
+ }
+ var count int
+ if err := tx.GetContext(ctx, &count, r.db.Rebind(`SELECT COUNT(*) FROM queued_messages WHERE session_id = ?`), sessionID); err != nil {
+ return fmt.Errorf("count: %w", err)
+ }
+ if count >= maxPerSession {
+ return ErrQueueFull
+ }
+ return nil
+}
+
func (r *sqliteRepository) ListBySession(ctx context.Context, sessionID string) ([]QueuedMessage, error) {
rows, err := r.ro.QueryxContext(ctx, r.ro.Rebind(`
SELECT id, session_id, task_id, position, content, model, plan_mode,
@@ -428,6 +553,33 @@ func (r *sqliteRepository) scanTail(ctx context.Context, tx *sqlx.Tx, sessionID
return msg, nil
}
+func (r *sqliteRepository) findCoalesced(ctx context.Context, tx *sqlx.Tx, sessionID, queuedBy, coalesceKey string) (*QueuedMessage, error) {
+ rows, err := tx.QueryxContext(ctx, r.db.Rebind(`
+ SELECT id, session_id, task_id, position, content, model, plan_mode,
+ attachments_json, metadata_json, queued_at, queued_by
+ FROM queued_messages
+ WHERE session_id = ? AND queued_by = ?
+ ORDER BY position ASC
+ `), sessionID, queuedBy)
+ if err != nil {
+ return nil, fmt.Errorf("scan coalesced queued: %w", err)
+ }
+ defer func() { _ = rows.Close() }()
+ for rows.Next() {
+ msg, err := scanQueuedRow(rows)
+ if err != nil {
+ return nil, err
+ }
+ if metadataString(msg.Metadata, MetadataCoalesceKey) == coalesceKey {
+ return msg, nil
+ }
+ }
+ if err := rows.Err(); err != nil {
+ return nil, err
+ }
+ return nil, nil
+}
+
// scanQueuedRow scans a single queued_messages row from any sqlx-compatible row source.
func scanQueuedRow(scanner interface{ Scan(dest ...any) error }) (*QueuedMessage, error) {
var (
@@ -483,3 +635,11 @@ func boolToInt(b bool) int {
}
return 0
}
+
+func metadataString(meta map[string]interface{}, key string) string {
+ if meta == nil {
+ return ""
+ }
+ value, _ := meta[key].(string)
+ return value
+}
diff --git a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go
index 5f21600ee9..e97c8fcc9e 100644
--- a/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go
+++ b/apps/backend/internal/orchestrator/messagequeue/repository_sqlite_test.go
@@ -187,6 +187,30 @@ func TestSQLiteRepository_UpdateContent(t *testing.T) {
}
}
+func TestSQLiteRepository_ReplaceCoalescedDetectsMissingRow(t *testing.T) {
+ repo := newTestSQLiteRepo(t).(*sqliteRepository)
+ ctx := context.Background()
+ tx, err := repo.db.BeginTxx(ctx, nil)
+ if err != nil {
+ t.Fatalf("begin tx: %v", err)
+ }
+ defer func() { _ = tx.Rollback() }()
+
+ _, err = repo.replaceCoalesced(ctx, tx,
+ &QueuedMessage{ID: "missing", SessionID: "s1", QueuedBy: QueuedByWorkflow},
+ &QueuedMessage{
+ SessionID: "s1",
+ TaskID: "t1",
+ Content: "new",
+ QueuedBy: QueuedByWorkflow,
+ Metadata: map[string]interface{}{MetadataCoalesceKey: "ci-key"},
+ },
+ )
+ if !errors.Is(err, ErrEntryNotFound) {
+ t.Fatalf("expected ErrEntryNotFound for vanished coalesced row, got %v", err)
+ }
+}
+
func TestSQLiteRepository_DeleteByID(t *testing.T) {
repo := newTestSQLiteRepo(t)
ctx := context.Background()
diff --git a/apps/backend/internal/orchestrator/messagequeue/service.go b/apps/backend/internal/orchestrator/messagequeue/service.go
index 4dfbc994ee..2d1bc51371 100644
--- a/apps/backend/internal/orchestrator/messagequeue/service.go
+++ b/apps/backend/internal/orchestrator/messagequeue/service.go
@@ -54,6 +54,7 @@ func (s *Service) QueueMessage(ctx context.Context, sessionID, taskID, content,
// is propagated to the resulting Message row when the queued message is
// drained (e.g. sender_task_id for messages sent via message_task_kandev).
func (s *Service) QueueMessageWithMetadata(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}) (*QueuedMessage, error) {
+ metadataCopy := copyMessageMetadata(metadata, 0)
msg := &QueuedMessage{
SessionID: sessionID,
TaskID: taskID,
@@ -61,7 +62,7 @@ func (s *Service) QueueMessageWithMetadata(ctx context.Context, sessionID, taskI
Model: model,
PlanMode: planMode,
Attachments: attachments,
- Metadata: metadata,
+ Metadata: metadataCopy,
QueuedBy: userID,
}
if err := s.repo.Insert(ctx, msg, s.maxPerSession); err != nil {
@@ -81,6 +82,54 @@ func (s *Service) QueueMessageWithMetadata(ctx context.Context, sessionID, taskI
return msg, nil
}
+// QueueMessageWithCoalesceKey replaces an existing pending entry with the same
+// coalesce key, session, and queued_by value. When no matching entry exists it
+// inserts a new tail entry if allowInsert is true; otherwise ErrEntryNotFound is
+// returned. The returned bool is true when an existing entry was replaced.
+func (s *Service) QueueMessageWithCoalesceKey(ctx context.Context, sessionID, taskID, content, model, userID string, planMode bool, attachments []MessageAttachment, metadata map[string]interface{}, coalesceKey string, allowInsert bool) (*QueuedMessage, bool, error) {
+ metadataCopy := copyMessageMetadata(metadata, 1)
+ metadataCopy[MetadataCoalesceKey] = coalesceKey
+ msg := &QueuedMessage{
+ SessionID: sessionID,
+ TaskID: taskID,
+ Content: content,
+ Model: model,
+ PlanMode: planMode,
+ Attachments: attachments,
+ Metadata: metadataCopy,
+ QueuedBy: userID,
+ }
+ queued, replaced, err := s.repo.InsertOrReplaceByCoalesceKey(ctx, msg, coalesceKey, s.maxPerSession, allowInsert)
+ if err != nil {
+ if errors.Is(err, ErrQueueFull) {
+ s.logger.Info("queue full",
+ zap.String("session_id", sessionID),
+ zap.Int("max", s.maxPerSession))
+ }
+ return nil, false, err
+ }
+ s.logger.Info("message queued with coalesce key",
+ zap.String("session_id", sessionID),
+ zap.String("task_id", taskID),
+ zap.String("entry_id", queued.ID),
+ zap.String("coalesce_key", coalesceKey),
+ zap.Bool("replaced", replaced),
+ zap.Int64("position", queued.Position),
+ zap.Int("content_length", len(content)))
+ return queued, replaced, nil
+}
+
+func copyMessageMetadata(metadata map[string]interface{}, extraCapacity int) map[string]interface{} {
+ if len(metadata) == 0 && extraCapacity == 0 {
+ return nil
+ }
+ out := make(map[string]interface{}, len(metadata)+extraCapacity)
+ for key, value := range metadata {
+ out[key] = value
+ }
+ return out
+}
+
// AppendContent appends content onto the session's tail entry when the tail's
// queued_by matches userID. Otherwise inserts a new entry. Returns
// ErrQueueFull when an insert would exceed the cap.
diff --git a/apps/backend/internal/orchestrator/messagequeue/service_test.go b/apps/backend/internal/orchestrator/messagequeue/service_test.go
index c35b966978..b027674a32 100644
--- a/apps/backend/internal/orchestrator/messagequeue/service_test.go
+++ b/apps/backend/internal/orchestrator/messagequeue/service_test.go
@@ -126,6 +126,64 @@ func TestAppendContent(t *testing.T) {
})
}
+func TestQueueMessageWithCoalesceKey(t *testing.T) {
+ t.Run("replaces matching entry without changing FIFO position", func(t *testing.T) {
+ svc := setupService(t)
+ ctx := context.Background()
+
+ first, err := svc.QueueMessage(ctx, "s", "t", "first", "", "user", false, nil)
+ require.NoError(t, err)
+ ci, replaced, err := svc.QueueMessageWithCoalesceKey(ctx, "s", "t", "old ci", "", QueuedByWorkflow, false, nil, map[string]interface{}{"origin": "ci"}, "ci-key", true)
+ require.NoError(t, err)
+ require.False(t, replaced)
+ _, err = svc.QueueMessage(ctx, "s", "t", "tail", "", "user", false, nil)
+ require.NoError(t, err)
+
+ updated, replaced, err := svc.QueueMessageWithCoalesceKey(ctx, "s", "t", "new ci", "", QueuedByWorkflow, false, nil, map[string]interface{}{"origin": "ci-new"}, "ci-key", true)
+ require.NoError(t, err)
+ require.True(t, replaced)
+ require.Equal(t, ci.ID, updated.ID)
+
+ status := svc.GetStatus(ctx, "s")
+ require.Equal(t, 3, status.Count)
+ assert.Equal(t, first.ID, status.Entries[0].ID)
+ assert.Equal(t, ci.ID, status.Entries[1].ID)
+ assert.Equal(t, "new ci", status.Entries[1].Content)
+ assert.Equal(t, "ci-new", status.Entries[1].Metadata["origin"])
+ assert.Equal(t, "tail", status.Entries[2].Content)
+ })
+
+ t.Run("does not mutate caller metadata or retag existing entries", func(t *testing.T) {
+ svc := setupService(t)
+ ctx := context.Background()
+ metadata := map[string]interface{}{"origin": "ci"}
+
+ first, replaced, err := svc.QueueMessageWithCoalesceKey(ctx, "s", "t", "first ci", "", QueuedByWorkflow, false, nil, metadata, "ci-key", true)
+ require.NoError(t, err)
+ require.False(t, replaced)
+ second, replaced, err := svc.QueueMessageWithCoalesceKey(ctx, "s", "t", "second ci", "", QueuedByWorkflow, false, nil, metadata, "other-key", true)
+ require.NoError(t, err)
+ require.False(t, replaced)
+
+ status := svc.GetStatus(ctx, "s")
+ require.Equal(t, 2, status.Count)
+ assert.Equal(t, first.ID, status.Entries[0].ID)
+ assert.Equal(t, second.ID, status.Entries[1].ID)
+ assert.Equal(t, "ci-key", status.Entries[0].Metadata[MetadataCoalesceKey])
+ assert.Equal(t, "other-key", status.Entries[1].Metadata[MetadataCoalesceKey])
+ assert.NotContains(t, metadata, MetadataCoalesceKey)
+ })
+
+ t.Run("returns entry not found when insert disabled and no match exists", func(t *testing.T) {
+ svc := setupService(t)
+ ctx := context.Background()
+
+ _, _, err := svc.QueueMessageWithCoalesceKey(ctx, "s", "t", "ci", "", QueuedByWorkflow, false, nil, nil, "ci-key", false)
+ assert.ErrorIs(t, err, ErrEntryNotFound)
+ assert.Equal(t, 0, svc.GetStatus(ctx, "s").Count)
+ })
+}
+
func TestTakeQueued(t *testing.T) {
t.Run("returns entries in FIFO order", func(t *testing.T) {
svc := setupService(t)
diff --git a/apps/backend/internal/orchestrator/messagequeue/types.go b/apps/backend/internal/orchestrator/messagequeue/types.go
index 72566c9f04..5457fffa13 100644
--- a/apps/backend/internal/orchestrator/messagequeue/types.go
+++ b/apps/backend/internal/orchestrator/messagequeue/types.go
@@ -20,6 +20,10 @@ const (
QueuedByWorkflow = "workflow"
)
+// MetadataCoalesceKey identifies queued entries that should be replaced rather
+// than appended when a newer pending message supersedes an older one.
+const MetadataCoalesceKey = "coalesce_key"
+
// QueueFullErrorCode is the well-known WS / MCP error code surfaced when an
// insert would exceed the per-session cap. Shared between the user-side WS
// handlers and the inter-task MCP handler so the wire contract stays in sync.
diff --git a/apps/web/components/github/pr-ci-automation-controls.tsx b/apps/web/components/github/pr-ci-automation-controls.tsx
index 8be3326f3c..f9d420df32 100644
--- a/apps/web/components/github/pr-ci-automation-controls.tsx
+++ b/apps/web/components/github/pr-ci-automation-controls.tsx
@@ -17,7 +17,8 @@ import { Textarea } from "@kandev/ui/textarea";
import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip";
import { useToast } from "@/components/toast-provider";
import { useTaskCIAutomationOptions } from "@/hooks/domains/github/use-task-ci-options";
-import type { TaskCIAutomationPatch, TaskPR } from "@/lib/types/github";
+import { autoFixRoundForState, findCIAutomationStateForPR } from "@/lib/github/ci-automation";
+import type { TaskCIAutomationPatch, TaskCIPRAutomationState, TaskPR } from "@/lib/types/github";
const PR_FEEDBACK_PLACEHOLDER = "{{pr.feedback}}";
@@ -207,12 +208,35 @@ function CIAutomationErrorRow({
);
}
+function CIAutoFixRoundExplanation({
+ state,
+ maxRounds,
+}: {
+ state: TaskCIPRAutomationState | undefined;
+ maxRounds: number | null | undefined;
+}) {
+ const round = autoFixRoundForState(state, maxRounds);
+ return (
+
+ Auto-fix has used {round.current} of {round.max} rounds for this PR. A round is counted when
+ Kandev sends or queues a CI auto-fix message. Updating an already queued auto-fix message does
+ not use another round. When this is at {round.max}/{round.max} and there is no pending
+ auto-fix message left to update, Kandev pauses auto-fix for this PR so it cannot loop forever.
+ Disable and re-enable auto-fix after manual review to start over.
+
+ );
+}
+
export function PRCIAutomationControls({ pr }: { pr: TaskPR }) {
const { options, loading, saving, error, refresh, update, resetPrompt } =
useTaskCIAutomationOptions(pr.task_id);
const { toast } = useToast();
const [promptOpen, setPromptOpen] = useState(false);
const [promptDraft, setPromptDraft] = useState("");
+ const automationState = findCIAutomationStateForPR(options?.pr_states, pr);
const openPromptEditor = useCallback(() => {
setPromptDraft(options?.auto_fix_prompt_override ?? options?.effective_auto_fix_prompt ?? "");
@@ -281,6 +305,12 @@ export function PRCIAutomationControls({ pr }: { pr: TaskPR }) {
disabled={disabled}
onCheckedChange={(checked) => patchOption({ auto_fix_enabled: checked })}
/>
+ {options?.auto_fix_enabled && (
+
+ )}
({
+ defaultCIFixPrompt: "Default CI fix prompt",
+}));
+
+const responsiveMock = vi.hoisted(() => ({
+ breakpoint: "desktop" as "mobile" | "tablet" | "compactDesktop" | "desktop",
+ isFinePointer: true,
+}));
+
+vi.mock("@/hooks/use-responsive-breakpoint", () => ({
+ useResponsiveBreakpoint: () => ({
+ breakpoint: responsiveMock.breakpoint,
+ isMobile: responsiveMock.breakpoint === "mobile",
+ isTablet: responsiveMock.breakpoint === "tablet",
+ isDesktop:
+ responsiveMock.breakpoint === "compactDesktop" || responsiveMock.breakpoint === "desktop",
+ isCompactDesktop: responsiveMock.breakpoint === "compactDesktop",
+ isFullDesktop: responsiveMock.breakpoint === "desktop",
+ isFinePointer: responsiveMock.isFinePointer,
+ usesDesktopWorkbench:
+ responsiveMock.breakpoint === "compactDesktop" || responsiveMock.breakpoint === "desktop",
+ }),
+}));
+
+vi.mock("@/lib/api/domains/github-api", async (importOriginal) => {
+ const actual = await importOriginal();
+ return {
+ ...actual,
+ getPRFeedback: vi.fn().mockResolvedValue(null),
+ getTaskCIAutomationOptions: vi.fn().mockResolvedValue(makeCIOptions()),
+ listWorkspaceTaskPRs: vi.fn().mockResolvedValue({ task_prs: {} }),
+ };
+});
+
+vi.mock("@/lib/ws/connection", () => ({
+ getWebSocketClient: vi.fn(() => null),
+}));
+
+function renderWithStore(initialState: Partial, ui: ReactNode) {
+ return render(
+
+
+ {ui}
+
+ ,
+ );
+}
+
+function makePR(overrides: Partial = {}): TaskPR {
+ return {
+ id: "pr-id",
+ task_id: "task-1",
+ owner: "acme",
+ repo: "demo",
+ pr_number: 42,
+ pr_url: "https://github.com/acme/demo/pull/42",
+ pr_title: "Test PR",
+ head_branch: "feat",
+ base_branch: "main",
+ author_login: "alice",
+ state: "open",
+ review_state: "approved",
+ checks_state: "success",
+ mergeable_state: "clean",
+ review_count: 1,
+ pending_review_count: 0,
+ comment_count: 0,
+ unresolved_review_threads: 0,
+ checks_total: 2,
+ checks_passing: 2,
+ additions: 0,
+ deletions: 0,
+ created_at: "",
+ merged_at: null,
+ closed_at: null,
+ last_synced_at: null,
+ updated_at: "",
+ ...overrides,
+ };
+}
+
+function makeCIOptions(overrides: Partial = {}): TaskCIAutomationOptions {
+ return {
+ task_id: "task-1",
+ auto_fix_enabled: false,
+ auto_merge_enabled: false,
+ auto_fix_prompt_override: null,
+ effective_auto_fix_prompt: testConstants.defaultCIFixPrompt,
+ using_default_prompt: true,
+ updated_at: "2026-06-18T10:00:00Z",
+ auto_fix_max_rounds: 10,
+ pr_states: [],
+ ...overrides,
+ };
+}
+
+function makeCIPrState(roundCount: number, exhausted = false) {
+ return {
+ task_id: "task-1",
+ repository_id: "",
+ pr_number: 42,
+ last_fix_signature: "",
+ last_fix_checkpoint_json: "",
+ last_fix_enqueued_at: null,
+ last_fix_session_id: null,
+ auto_fix_round_count: roundCount,
+ auto_fix_exhausted_at: exhausted ? "2026-06-18T11:00:00Z" : null,
+ last_merge_signature: "",
+ last_merge_attempt_at: null,
+ last_error: exhausted ? "CI auto-fix paused after 10 rounds for this PR" : null,
+ created_at: "2026-06-18T10:00:00Z",
+ updated_at: "2026-06-18T10:00:00Z",
+ };
+}
+
+function stateWithAutoFix(
+ roundCount: number,
+ exhausted = false,
+ maxRounds = 10,
+): Partial {
+ return {
+ taskPRs: { byTaskId: { "task-1": [makePR()] } },
+ taskCIAutomation: {
+ byTaskId: {
+ "task-1": makeCIOptions({
+ auto_fix_enabled: true,
+ auto_fix_max_rounds: maxRounds,
+ pr_states: [makeCIPrState(roundCount, exhausted)],
+ }),
+ },
+ loading: {},
+ saving: {},
+ errors: {},
+ },
+ };
+}
+
+beforeEach(() => {
+ responsiveMock.breakpoint = "desktop";
+ responsiveMock.isFinePointer = true;
+});
+
+afterEach(() => {
+ cleanup();
+});
+
+describe("PRStatusChip auto-fix round display", () => {
+ it("shows the CI auto-fix round count in the enabled chip", () => {
+ renderWithStore(stateWithAutoFix(3), );
+
+ expect(screen.getByTestId(AUTO_FIX_BADGE_TESTID).textContent).toBe("Auto-fix 3/10");
+ });
+
+ it("shows exhausted auto-fix state as a warning chip", () => {
+ renderWithStore(stateWithAutoFix(10, true), );
+
+ const chip = screen.getByTestId(AUTO_FIX_BADGE_TESTID);
+ expect(chip.textContent).toBe("Auto-fix 10/10");
+ expect(chip.getAttribute("data-auto-fix-exhausted")).toBe("true");
+ });
+
+ it("does not mark 10/10 as exhausted until the backend pauses auto-fix", () => {
+ renderWithStore(stateWithAutoFix(10), );
+
+ const chip = screen.getByTestId(AUTO_FIX_BADGE_TESTID);
+ expect(chip.textContent).toBe("Auto-fix 10/10");
+ expect(chip.getAttribute("data-auto-fix-exhausted")).toBe("false");
+ });
+
+ it("uses the max round count returned by the backend", () => {
+ renderWithStore(stateWithAutoFix(3, false, 12), );
+
+ expect(screen.getByTestId(AUTO_FIX_BADGE_TESTID).textContent).toBe("Auto-fix 3/12");
+ });
+
+ it("explains the auto-fix round count from the hover popover", async () => {
+ renderWithStore(stateWithAutoFix(1), );
+
+ fireEvent.mouseEnter(screen.getByTestId(CHIP_TESTID));
+ const explanation = await screen.findByTestId(ROUND_EXPLANATION_TESTID);
+ expect(explanation.textContent).toContain("Auto-fix has used 1 of 10 rounds");
+ expect(explanation.textContent).toContain(
+ "Updating an already queued auto-fix message does not use another round",
+ );
+ expect(explanation.textContent).toContain(
+ "pauses auto-fix for this PR so it cannot loop forever",
+ );
+ });
+
+ it("shows the auto-fix round explanation in the mobile drawer", () => {
+ responsiveMock.breakpoint = "mobile";
+ responsiveMock.isFinePointer = false;
+ renderWithStore(stateWithAutoFix(2), );
+
+ act(() => {
+ fireEvent.click(screen.getByTestId(CHIP_TESTID));
+ });
+
+ const explanation = screen.getByTestId(ROUND_EXPLANATION_TESTID);
+ expect(explanation.textContent).toContain("Auto-fix has used 2 of 10 rounds");
+ });
+});
diff --git a/apps/web/components/github/pr-status-chip.test.tsx b/apps/web/components/github/pr-status-chip.test.tsx
index 1217dd50ce..d6e5ab94e2 100644
--- a/apps/web/components/github/pr-status-chip.test.tsx
+++ b/apps/web/components/github/pr-status-chip.test.tsx
@@ -9,6 +9,12 @@ import { PR_CI_DESKTOP_POPOVER_SCROLL_CLASS } from "./pr-ci-popover";
import type { AppState } from "@/lib/state/store";
import type { TaskCIAutomationOptions, TaskPR } from "@/lib/types/github";
+const AUTO_FIX_BADGE_TESTID = "pr-status-auto-fix-chip";
+
+const testConstants = vi.hoisted(() => ({
+ defaultCIFixPrompt: "Default CI fix prompt",
+}));
+
const responsiveMock = vi.hoisted(() => ({
breakpoint: "desktop" as "mobile" | "tablet" | "compactDesktop" | "desktop",
isFinePointer: true,
@@ -38,7 +44,7 @@ vi.mock("@/lib/api/domains/github-api", async (importOriginal) => {
auto_fix_enabled: false,
auto_merge_enabled: false,
auto_fix_prompt_override: null,
- effective_auto_fix_prompt: "Default CI fix prompt",
+ effective_auto_fix_prompt: testConstants.defaultCIFixPrompt,
using_default_prompt: true,
updated_at: "2026-06-18T10:00:00Z",
pr_states: [],
@@ -100,7 +106,7 @@ function makeCIOptions(overrides: Partial = {}): TaskCI
auto_fix_enabled: false,
auto_merge_enabled: false,
auto_fix_prompt_override: null,
- effective_auto_fix_prompt: "Default CI fix prompt",
+ effective_auto_fix_prompt: testConstants.defaultCIFixPrompt,
using_default_prompt: true,
updated_at: "2026-06-18T10:00:00Z",
pr_states: [],
@@ -229,10 +235,10 @@ describe("PRStatusChip desktop branch", () => {
},
,
);
- expect(screen.getByTestId("pr-status-auto-fix-chip").textContent).toBe("Auto-fix");
+ expect(screen.getByTestId(AUTO_FIX_BADGE_TESTID).textContent).toBe("Auto-fix 0/10");
expect(screen.getByTestId("pr-status-auto-merge-chip").textContent).toBe("Auto-merge");
expect(screen.getByTestId(CHIP_TESTID).getAttribute("aria-label")).toBe(
- "Pull request #42 CI status, auto-fix enabled, auto-merge enabled",
+ "Pull request #42 CI status, auto-fix enabled 0 of 10 rounds used, auto-merge enabled",
);
});
});
@@ -293,10 +299,10 @@ describe("PRStatusChip mobile branch", () => {
},
,
);
- expect(screen.getByTestId("pr-status-auto-fix-chip").textContent).toBe("Auto-fix");
+ expect(screen.getByTestId(AUTO_FIX_BADGE_TESTID).textContent).toBe("Auto-fix 0/10");
expect(screen.queryByTestId("pr-status-auto-merge-chip")).toBeNull();
expect(screen.getByTestId(CHIP_TESTID).getAttribute("aria-label")).toBe(
- "Pull request #42 CI status, auto-fix enabled",
+ "Pull request #42 CI status, auto-fix enabled 0 of 10 rounds used",
);
});
diff --git a/apps/web/components/github/pr-status-chip.tsx b/apps/web/components/github/pr-status-chip.tsx
index a480491571..48f6cab8c5 100644
--- a/apps/web/components/github/pr-status-chip.tsx
+++ b/apps/web/components/github/pr-status-chip.tsx
@@ -37,7 +37,9 @@ import {
pickDefaultPR,
} from "@/components/github/pr-task-icon";
import { useTouchDrawer } from "@/hooks/use-compact-task-chrome";
-import type { TaskPR } from "@/lib/types/github";
+import { autoFixRoundForState, findCIAutomationStateForPR } from "@/lib/github/ci-automation";
+import type { AutoFixRoundInfo } from "@/lib/github/ci-automation";
+import type { TaskCIAutomationOptions, TaskPR } from "@/lib/types/github";
const HOVER_OPEN_DELAY_MS = 150;
const HOVER_CLOSE_DELAY_MS = 150;
@@ -56,6 +58,7 @@ type ChipStatus =
type AutomationFlags = {
autoFix: boolean;
autoMerge: boolean;
+ autoFixRound: AutoFixRoundInfo | null;
};
function chipStatus(pr: TaskPR): ChipStatus {
@@ -160,10 +163,6 @@ function useChipPopoverInteractions() {
export function PRStatusChip({ taskId }: { taskId: string | null }) {
const { prs } = useTaskPR(taskId);
const { options: automationOptions } = useTaskCIAutomationOptions(taskId);
- const automationFlags: AutomationFlags = {
- autoFix: Boolean(automationOptions?.auto_fix_enabled),
- autoMerge: Boolean(automationOptions?.auto_merge_enabled),
- };
// Defensive Array.isArray: a partial hydration can briefly seed the store
// with a non-array value (same guard as PRTaskIcon).
// Only open PRs are worth a CI chip — terminal PRs (merged/closed) are
@@ -180,8 +179,62 @@ export function PRStatusChip({ taskId }: { taskId: string | null }) {
usePRFeedbackBackgroundSync(pickDefaultPR(openPRs));
if (openPRs.length === 0) return null;
if (openPRs.length === 1)
- return ;
- return ;
+ return (
+
+ );
+ return (
+
+ );
+}
+
+function automationForPR(
+ options: TaskCIAutomationOptions | null | undefined,
+ pr: TaskPR,
+): AutomationFlags {
+ return {
+ autoFix: Boolean(options?.auto_fix_enabled),
+ autoMerge: Boolean(options?.auto_merge_enabled),
+ autoFixRound: options?.auto_fix_enabled
+ ? autoFixRoundForState(
+ findCIAutomationStateForPR(options.pr_states, pr),
+ options.auto_fix_max_rounds,
+ )
+ : null,
+ };
+}
+
+function automationForPRs(
+ options: TaskCIAutomationOptions | null | undefined,
+ prs: TaskPR[],
+): AutomationFlags {
+ const roundInfos = options?.auto_fix_enabled
+ ? prs.map((pr) =>
+ autoFixRoundForState(
+ findCIAutomationStateForPR(options.pr_states, pr),
+ options.auto_fix_max_rounds,
+ ),
+ )
+ : [];
+ return {
+ autoFix: Boolean(options?.auto_fix_enabled),
+ autoMerge: Boolean(options?.auto_merge_enabled),
+ autoFixRound: pickAttentionRound(roundInfos),
+ };
+}
+
+function pickAttentionRound(roundInfos: AutoFixRoundInfo[]): AutoFixRoundInfo | null {
+ if (roundInfos.length === 0) return null;
+ return roundInfos.reduce((best, next) => {
+ if (next.exhausted && !best.exhausted) return next;
+ if (next.exhausted === best.exhausted && next.current > best.current) return next;
+ return best;
+ });
}
type ChipButtonAttrs = {
@@ -196,7 +249,9 @@ type ChipButtonAttrs = {
function automationAriaSuffix(automation: AutomationFlags): string {
const flags = [
- automation.autoFix ? "auto-fix enabled" : null,
+ automation.autoFix
+ ? `auto-fix enabled${automation.autoFixRound ? ` ${automation.autoFixRound.current} of ${automation.autoFixRound.max} rounds used` : ""}`
+ : null,
automation.autoMerge ? "auto-merge enabled" : null,
].filter(Boolean);
return flags.length > 0 ? `, ${flags.join(", ")}` : "";
@@ -220,14 +275,21 @@ function chipButtonAttrs(
function AutomationFlagBadges({ automation }: { automation: AutomationFlags }) {
if (!automation.autoFix && !automation.autoMerge) return null;
+ const autoFixRound = automation.autoFixRound;
return (
<>
- {automation.autoFix && (
+ {automation.autoFix && autoFixRound && (
- Auto-fix
+ Auto-fix {autoFixRound.current}/{autoFixRound.max}
)}
{automation.autoMerge && (
diff --git a/apps/web/lib/github/ci-automation.test.ts b/apps/web/lib/github/ci-automation.test.ts
new file mode 100644
index 0000000000..378841751d
--- /dev/null
+++ b/apps/web/lib/github/ci-automation.test.ts
@@ -0,0 +1,75 @@
+import { describe, expect, it } from "vitest";
+import {
+ autoFixRoundForState,
+ clampAutoFixRound,
+ findCIAutomationStateForPR,
+ normalizeAutoFixMaxRounds,
+} from "./ci-automation";
+import type { TaskCIPRAutomationState, TaskPR } from "@/lib/types/github";
+
+function makePR(repositoryID: string, prNumber = 42): TaskPR {
+ return {
+ repository_id: repositoryID,
+ pr_number: prNumber,
+ } as TaskPR;
+}
+
+function makeState(
+ repositoryID: string,
+ prNumber: number,
+ roundCount: number,
+ exhaustedAt: string | null = null,
+): TaskCIPRAutomationState {
+ return {
+ repository_id: repositoryID,
+ pr_number: prNumber,
+ auto_fix_round_count: roundCount,
+ auto_fix_exhausted_at: exhaustedAt,
+ } as TaskCIPRAutomationState;
+}
+
+describe("CI automation helpers", () => {
+ it("finds PR automation state by repository id and PR number", () => {
+ const states = [makeState("", 42, 1), makeState("repo-1", 42, 2), makeState("repo-1", 7, 3)];
+
+ expect(findCIAutomationStateForPR(states, makePR("repo-1", 42))).toBe(states[1]);
+ expect(findCIAutomationStateForPR(states, makePR("", 42))).toBe(states[0]);
+ expect(findCIAutomationStateForPR(states, makePR("repo-missing", 42))).toBeUndefined();
+ });
+
+ it("returns an empty round state when no PR state exists", () => {
+ expect(autoFixRoundForState(undefined, 12)).toEqual({
+ current: 0,
+ max: 12,
+ exhausted: false,
+ });
+ });
+
+ it("treats only backend exhaustion timestamps as paused", () => {
+ expect(autoFixRoundForState(makeState("repo-1", 42, 10), 10)).toEqual({
+ current: 10,
+ max: 10,
+ exhausted: false,
+ });
+ expect(autoFixRoundForState(makeState("repo-1", 42, 3, "2026-06-18T11:00:00Z"), 10)).toEqual({
+ current: 3,
+ max: 10,
+ exhausted: true,
+ });
+ });
+
+ it("normalizes max rounds and clamps current rounds", () => {
+ expect(normalizeAutoFixMaxRounds(undefined)).toBe(10);
+ expect(normalizeAutoFixMaxRounds(null)).toBe(10);
+ expect(normalizeAutoFixMaxRounds(Number.NaN)).toBe(10);
+ expect(normalizeAutoFixMaxRounds(0)).toBe(1);
+ expect(normalizeAutoFixMaxRounds(-4)).toBe(1);
+ expect(normalizeAutoFixMaxRounds(12.8)).toBe(12);
+
+ expect(clampAutoFixRound(undefined, 10)).toBe(0);
+ expect(clampAutoFixRound(Number.NaN, 10)).toBe(0);
+ expect(clampAutoFixRound(-2, 10)).toBe(0);
+ expect(clampAutoFixRound(4.9, 10)).toBe(4);
+ expect(clampAutoFixRound(14, 10)).toBe(10);
+ });
+});
diff --git a/apps/web/lib/github/ci-automation.ts b/apps/web/lib/github/ci-automation.ts
new file mode 100644
index 0000000000..6a3805ae70
--- /dev/null
+++ b/apps/web/lib/github/ci-automation.ts
@@ -0,0 +1,42 @@
+import type { TaskCIPRAutomationState, TaskPR } from "@/lib/types/github";
+
+const DEFAULT_AUTO_FIX_MAX_ROUNDS = 10;
+
+export type AutoFixRoundInfo = {
+ current: number;
+ max: number;
+ exhausted: boolean;
+};
+
+export function findCIAutomationStateForPR(
+ states: TaskCIPRAutomationState[] | undefined,
+ pr: TaskPR,
+): TaskCIPRAutomationState | undefined {
+ const repositoryID = pr.repository_id ?? "";
+ return states?.find(
+ (state) => state.pr_number === pr.pr_number && state.repository_id === repositoryID,
+ );
+}
+
+export function autoFixRoundForState(
+ state: TaskCIPRAutomationState | undefined,
+ maxRounds: number | null | undefined,
+): AutoFixRoundInfo {
+ const max = normalizeAutoFixMaxRounds(maxRounds);
+ const current = clampAutoFixRound(state?.auto_fix_round_count, max);
+ return {
+ current,
+ max,
+ exhausted: Boolean(state?.auto_fix_exhausted_at),
+ };
+}
+
+export function normalizeAutoFixMaxRounds(value: number | null | undefined) {
+ if (typeof value !== "number" || !Number.isFinite(value)) return DEFAULT_AUTO_FIX_MAX_ROUNDS;
+ return Math.max(1, Math.trunc(value));
+}
+
+export function clampAutoFixRound(value: number | null | undefined, maxRounds: number) {
+ if (typeof value !== "number" || !Number.isFinite(value)) return 0;
+ return Math.min(maxRounds, Math.max(0, Math.trunc(value)));
+}
diff --git a/apps/web/lib/types/github.ts b/apps/web/lib/types/github.ts
index 755f39b6e0..7e820cad9b 100644
--- a/apps/web/lib/types/github.ts
+++ b/apps/web/lib/types/github.ts
@@ -192,6 +192,8 @@ export type TaskCIPRAutomationState = {
last_fix_checkpoint_json: string;
last_fix_enqueued_at: string | null;
last_fix_session_id: string | null;
+ auto_fix_round_count: number;
+ auto_fix_exhausted_at: string | null;
last_merge_signature: string;
last_merge_attempt_at: string | null;
last_error: string | null;
@@ -204,6 +206,7 @@ export type TaskCIAutomationOptions = {
auto_fix_enabled: boolean;
auto_merge_enabled: boolean;
auto_fix_prompt_override: string | null;
+ auto_fix_max_rounds?: number;
effective_auto_fix_prompt: string;
using_default_prompt: boolean;
updated_at: string;
diff --git a/docs/specs/ui/ci-pr-automation.md b/docs/specs/ui/ci-pr-automation.md
index db6c94dcd8..488f22dd33 100644
--- a/docs/specs/ui/ci-pr-automation.md
+++ b/docs/specs/ui/ci-pr-automation.md
@@ -32,6 +32,11 @@ Users can already see pull request CI/review status above the task chat input, b
- Saving CI automation options while `Auto-fix CI & address comments` or `Auto-merge when ready` is enabled immediately evaluates the task's current linked PRs instead of waiting for the next PR watch poll. This includes prompt edits made while automation is already enabled; unchanged feedback is still deduped by the per-PR checkpoint.
- Every auto-fix attempt records the latest feedback snapshot it used, including non-actionable snapshots that were intentionally no-ops. Later fix rounds include only new or materially changed CI/review feedback since the last recorded round, with enough summary context for the agent to understand the PR.
- Automation must not repeatedly prompt for the same failure/comment snapshot or repeatedly retry the same failed merge attempt on every poll.
+- When auto-fix is enabled and the task session is busy, Kandev keeps at most one pending CI auto-fix queue entry per task/repository/PR. Newer feedback replaces that pending entry instead of appending another queued `@ci-auto-fix` message.
+- Auto-fix is capped at 10 accepted rounds per task/repository/PR. A round is counted when Kandev sends a prompt directly or inserts a new queued auto-fix prompt. Replacing an already queued auto-fix prompt does not count as another round.
+- The auto-fix enabled chip above the chat input shows round progress as `Auto-fix N/10`; PRs paused by the backend after the cap is reached show `Auto-fix 10/10` with warning/paused styling.
+- Hovering the chip on desktop, or opening the same PR CI drawer on mobile, explains in plain language how many rounds have been used, what counts as a round, that queue replacement does not count again, and that Kandev pauses when 10/10 has no pending auto-fix message left to update.
+- Accepted round-count changes and exhausted-state changes are broadcast to open clients through the task CI options update event so the chip stays current without a reload.
- Automation controls persist across Kandev restarts.
## Data model
@@ -55,6 +60,8 @@ Users can already see pull request CI/review status above the task chat input, b
- `last_fix_checkpoint_json` string nullable. JSON snapshot of feedback used in the last fix round. This includes non-actionable no-op snapshots so identical bot summaries/status updates are not repeatedly sent.
- `last_fix_enqueued_at` timestamp nullable.
- `last_fix_session_id` string nullable.
+- `auto_fix_round_count` integer, default `0`. Counts accepted auto-fix rounds for this task/repository/PR.
+- `auto_fix_exhausted_at` timestamp nullable. Set when Kandev pauses auto-fix after the 10-round cap.
- `last_merge_signature` string nullable. Deterministic hash of the last readiness state used for a merge attempt.
- `last_merge_attempt_at` timestamp nullable.
- `last_error` string nullable. Latest user-visible automation error for this task/PR pair.
@@ -86,6 +93,7 @@ Response:
"auto_fix_enabled": false,
"auto_merge_enabled": false,
"auto_fix_prompt_override": null,
+ "auto_fix_max_rounds": 10,
"effective_auto_fix_prompt": "Fix the PR feedback...",
"using_default_prompt": true,
"updated_at": "2026-06-18T00:00:00Z",
@@ -94,6 +102,8 @@ Response:
"repository_id": "repo-123",
"pr_number": 42,
"last_fix_enqueued_at": null,
+ "auto_fix_round_count": 0,
+ "auto_fix_exhausted_at": null,
"last_merge_attempt_at": null,
"last_error": null
}
@@ -148,6 +158,8 @@ Auto-fix cycle for one task/PR:
it to reply with a fix summary and resolve the addressed PR review threads so they do not keep
the PR blocked.
8. Once the prompt is queued or accepted by the agent runtime, Kandev records the new signature/checkpoint and attempt metadata for the latest feedback snapshot, actionable or non-actionable, so identical snapshots are not sent repeatedly while the agent is still working.
+9. If the task session is busy and a pending auto-fix entry for this task/repository/PR already exists, Kandev replaces that queued entry with the latest rendered prompt instead of appending a second queued message. The round count is unchanged.
+10. If a new prompt would require an 11th accepted auto-fix round for the same task/repository/PR, Kandev does not send or queue the prompt. It records a paused error and keeps the chip visible as `Auto-fix 10/10`. Disabling and re-enabling auto-fix resets the round count and paused state for the task's PR automation rows.
Auto-merge cycle for one task/PR:
@@ -173,7 +185,9 @@ Auto-merge cycle for one task/PR:
| Full PR feedback fetch fails | Auto-fix does not prompt; per-PR automation state records the error and the next materially changed lightweight status may retry. |
| Task has no promptable session | Auto-fix records an error instead of creating a surprising new session. |
| Task session is busy | Auto-fix queues the rendered prompt with workflow/automation metadata for later delivery; the visible `@ci-auto-fix` chat message, including PR snapshot details, is created when the queued prompt is delivered and before the agent's response for that turn. |
+| Task session is busy and a pending auto-fix already exists for that PR | Kandev replaces the pending queued prompt with the latest feedback snapshot; it does not append a second queued message or increment the round count. |
| Same feedback snapshot repeats | Auto-fix does not send another prompt. |
+| Auto-fix reaches 10 rounds for a PR | Kandev pauses auto-fix for that task/repository/PR, records a visible error, and does not create an 11th round. Already exhausted PRs skip full feedback fetching on later watcher wakes. |
| GitHub merge fails | Auto-merge records the error and does not retry until the readiness signature changes. |
| Default prompt row is missing | Backend falls back to the embedded `ci-auto-fix.md` content. |
| Kandev restarts while an automation prompt is queued | Queued message and automation options/checkpoints persist according to the existing message queue and new CI automation tables. |
@@ -199,6 +213,10 @@ Auto-merge cycle for one task/PR:
- **GIVEN** auto-fix is enabled and a watched PR transitions from passing to failing CI, **WHEN** the 1-minute PR watch poll observes the failure, **THEN** Kandev fetches full PR feedback and sends or queues one auto-fix prompt for that failure snapshot.
- **GIVEN** auto-fix already prompted for a failure snapshot, **WHEN** the same failure is observed again on a later poll, **THEN** no duplicate prompt is sent.
- **GIVEN** auto-fix already prompted for a failure snapshot, **WHEN** a new failed check or new unresolved review comment appears, **THEN** Kandev sends or queues a new prompt containing the new or materially changed feedback.
+- **GIVEN** auto-fix is enabled and the task session is running, **WHEN** changed CI feedback appears multiple times for the same PR before the queue drains, **THEN** Kandev keeps one queued `@ci-auto-fix` entry for that PR and updates it with the latest feedback.
+- **GIVEN** auto-fix has used 1 of 10 rounds for a PR, **WHEN** the user views the auto-fix chip above the chat input, **THEN** the chip reads `Auto-fix 1/10` and the hover/drawer explanation states that one round out of ten has been used.
+- **GIVEN** auto-fix has already used 10 rounds for a PR and no pending auto-fix queue entry exists, **WHEN** new actionable feedback appears, **THEN** Kandev does not send or queue another prompt and records the PR as paused at `Auto-fix 10/10`.
+- **GIVEN** auto-fix has already used 10 rounds for a PR and the 10th round is still queued, **WHEN** new actionable feedback appears, **THEN** Kandev replaces that pending queued prompt without incrementing the round count.
- **GIVEN** auto-fix sends a prompt for a snapshot that contains only non-actionable automation summaries or status updates, **WHEN** the agent reviews that prompt, **THEN** the agent does not modify files, commit, or push and only reports that there is nothing actionable to address.
- **GIVEN** auto-fix is enabled and the task session is running, **WHEN** new actionable PR feedback appears, **THEN** the prompt is queued and delivered after the current turn rather than interrupting the running session, and the chat history shows the `@ci-auto-fix` user message with visible PR snapshot details before the agent output for the queued turn.
- **GIVEN** auto-merge is enabled and the PR has passing checks, required reviews, no unresolved threads, and clean mergeability, **WHEN** the PR watch poll observes the ready state, **THEN** Kandev merges the PR with the existing backend merge-method selection.