fix: coalesce ci auto-fix queue rounds (#1552)
* fix: coalesce ci auto-fix queue rounds * fix: centralize ci auto-fix round cap * fix: keep ci auto-fix state fresh * fix: refine ci auto-fix round status * fix: handle ci auto-fix exhaustion edge cases * fix: reduce ci auto-fix dispatch complexity * fix: refresh pending ci auto-fix before direct prompt * fix: harden ci auto-fix queue replacement --------- Co-authored-by: root <root@kandev.ts.net>
This commit is contained in:
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 (
|
||||
<div
|
||||
data-testid="ci-auto-fix-round-explanation"
|
||||
className="rounded-md border border-border/70 bg-muted/30 px-2.5 py-2 text-[11px] leading-relaxed text-muted-foreground"
|
||||
>
|
||||
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.
|
||||
</div>
|
||||
);
|
||||
}
|
||||
|
||||
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 && (
|
||||
<CIAutoFixRoundExplanation
|
||||
state={automationState}
|
||||
maxRounds={options.auto_fix_max_rounds}
|
||||
/>
|
||||
)}
|
||||
<CIAutomationRow
|
||||
id={`task-ci-auto-merge-${pr.task_id}`}
|
||||
label="Auto-merge when ready"
|
||||
|
||||
@@ -0,0 +1,216 @@
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import type { ReactNode } from "react";
|
||||
import { act, cleanup, fireEvent, render, screen } from "@testing-library/react";
|
||||
import { TooltipProvider } from "@kandev/ui/tooltip";
|
||||
import { StateProvider } from "@/components/state-provider";
|
||||
import { ToastProvider } from "@/components/toast-provider";
|
||||
import { PRStatusChip } from "./pr-status-chip";
|
||||
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 CHIP_TESTID = "pr-status-chip";
|
||||
const ROUND_EXPLANATION_TESTID = "ci-auto-fix-round-explanation";
|
||||
|
||||
const testConstants = vi.hoisted(() => ({
|
||||
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<typeof import("@/lib/api/domains/github-api")>();
|
||||
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<AppState>, ui: ReactNode) {
|
||||
return render(
|
||||
<StateProvider initialState={initialState}>
|
||||
<ToastProvider>
|
||||
<TooltipProvider>{ui}</TooltipProvider>
|
||||
</ToastProvider>
|
||||
</StateProvider>,
|
||||
);
|
||||
}
|
||||
|
||||
function makePR(overrides: Partial<TaskPR> = {}): 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> = {}): 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<AppState> {
|
||||
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), <PRStatusChip taskId="task-1" />);
|
||||
|
||||
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), <PRStatusChip taskId="task-1" />);
|
||||
|
||||
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), <PRStatusChip taskId="task-1" />);
|
||||
|
||||
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), <PRStatusChip taskId="task-1" />);
|
||||
|
||||
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), <PRStatusChip taskId="task-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), <PRStatusChip taskId="task-1" />);
|
||||
|
||||
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");
|
||||
});
|
||||
});
|
||||
@@ -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<TaskCIAutomationOptions> = {}): 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", () => {
|
||||
},
|
||||
<PRStatusChip taskId="task-1" />,
|
||||
);
|
||||
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", () => {
|
||||
},
|
||||
<PRStatusChip taskId="task-1" />,
|
||||
);
|
||||
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",
|
||||
);
|
||||
});
|
||||
|
||||
|
||||
@@ -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 <PRStatusChipInner pr={openPRs[0]} automation={automationFlags} />;
|
||||
return <PRStatusChipMultiInner prs={openPRs} automation={automationFlags} />;
|
||||
return (
|
||||
<PRStatusChipInner
|
||||
pr={openPRs[0]}
|
||||
automation={automationForPR(automationOptions, openPRs[0])}
|
||||
/>
|
||||
);
|
||||
return (
|
||||
<PRStatusChipMultiInner
|
||||
prs={openPRs}
|
||||
automation={automationForPRs(automationOptions, openPRs)}
|
||||
/>
|
||||
);
|
||||
}
|
||||
|
||||
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 && (
|
||||
<span
|
||||
data-testid="pr-status-auto-fix-chip"
|
||||
className="rounded-sm bg-emerald-500/15 px-1 py-0.5 text-[9px] font-medium leading-none text-emerald-500"
|
||||
data-auto-fix-round={`${autoFixRound.current}/${autoFixRound.max}`}
|
||||
data-auto-fix-exhausted={autoFixRound.exhausted ? "true" : "false"}
|
||||
className={`rounded-sm px-1 py-0.5 text-[9px] font-medium leading-none ${
|
||||
autoFixRound.exhausted
|
||||
? "bg-yellow-500/15 text-yellow-500"
|
||||
: "bg-emerald-500/15 text-emerald-500"
|
||||
}`}
|
||||
>
|
||||
Auto-fix
|
||||
Auto-fix {autoFixRound.current}/{autoFixRound.max}
|
||||
</span>
|
||||
)}
|
||||
{automation.autoMerge && (
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
@@ -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)));
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user