From 5252a5519905f2950924e5c06bcbb548e59ce287 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Carlos=20Flor=C3=AAncio?= Date: Sat, 27 Jun 2026 00:01:05 +0100 Subject: [PATCH] fix: improve pr ci automation reliability (#1508) --- apps/backend/internal/github/gh_client.go | 2 + .../backend/internal/github/gh_client_test.go | 96 +++ apps/backend/internal/github/handlers.go | 6 +- .../internal/github/pat_client_test.go | 44 ++ apps/backend/internal/github/poller_test.go | 182 ++++++ apps/backend/internal/github/service.go | 4 +- .../internal/github/service_pr_sync_test.go | 4 +- .../internal/github/service_pr_watch.go | 64 +- .../github/service_pr_watch_batched.go | 11 +- .../service_repo_negative_cache_test.go | 45 ++ apps/backend/internal/github/store.go | 2 + .../github/store_ci_automation_test.go | 10 +- .../orchestrator/event_handlers_github.go | 61 +- .../event_handlers_github_ci_automation.go | 165 ++++- ...vent_handlers_github_ci_automation_test.go | 581 +++++++++++++++++- .../event_handlers_github_test.go | 55 +- docs/specs/ui/ci-pr-automation.md | 6 +- 17 files changed, 1243 insertions(+), 95 deletions(-) diff --git a/apps/backend/internal/github/gh_client.go b/apps/backend/internal/github/gh_client.go index 7d2b452bfd..2cf0fa853e 100644 --- a/apps/backend/internal/github/gh_client.go +++ b/apps/backend/internal/github/gh_client.go @@ -647,6 +647,8 @@ func (c *GHClient) ListCheckRuns(ctx context.Context, owner, repo, ref string) ( return mergeChecks(convertRawCheckRuns(checkRunsRaw), convertRawStatusContexts(statusRaw)), nil } +// decodeGHCheckRuns decodes whitespace-separated JSON check-run objects +// emitted by gh --paginate --jq '.check_runs[]'. func decodeGHCheckRuns(out string) ([]ghCheckRun, error) { if strings.TrimSpace(out) == "" { return nil, nil diff --git a/apps/backend/internal/github/gh_client_test.go b/apps/backend/internal/github/gh_client_test.go index 8a99588e62..378d3eadbc 100644 --- a/apps/backend/internal/github/gh_client_test.go +++ b/apps/backend/internal/github/gh_client_test.go @@ -1,7 +1,10 @@ package github import ( + "context" "errors" + "os" + "path/filepath" "strings" "testing" "time" @@ -124,6 +127,99 @@ func TestParseTimePtrValue(t *testing.T) { } } +func TestGHClient_ListCheckRuns_PaginatesCheckRuns(t *testing.T) { + binDir := t.TempDir() + logPath := filepath.Join(t.TempDir(), "gh-args.log") + ghPath := filepath.Join(binDir, "gh") + script := `#!/bin/sh +printf '%s\n' "$*" >> "$GH_ARGS_LOG" +case "$*" in + *check-runs*) + case "$*" in + *--slurp*) printf '%s\n' 'unsupported slurp' >&2; exit 1 ;; + *--paginate*) printf '%s\n' '{"name":"unit","status":"completed","conclusion":"success","html_url":"https://ci/unit"}' '{"name":"lint","status":"completed","conclusion":"failure","html_url":"https://ci/lint"}' ;; + *) printf '%s\n' '[{"name":"unit","status":"completed","conclusion":"success","html_url":"https://ci/unit"}]' ;; + esac + ;; + *status*) printf '%s\n' '[]' ;; + *) printf '%s\n' '[]' ;; +esac +` + if err := os.WriteFile(ghPath, []byte(script), 0o755); err != nil { + t.Fatalf("write fake gh: %v", err) + } + t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("GH_ARGS_LOG", logPath) + + checks, err := NewGHClient().ListCheckRuns(context.Background(), "acme", "widget", "sha") + if err != nil { + t.Fatalf("ListCheckRuns: %v", err) + } + if len(checks) != 2 { + t.Fatalf("checks = %d, want 2: %+v", len(checks), checks) + } + var lint *CheckRun + for i := range checks { + if checks[i].Name == "lint" { + lint = &checks[i] + break + } + } + if lint == nil || lint.Conclusion != "failure" { + t.Fatalf("expected failed lint check from paginated output, got %+v", checks) + } + logged, err := os.ReadFile(logPath) + if err != nil { + t.Fatalf("read gh args log: %v", err) + } + if !strings.Contains(string(logged), "--paginate") { + t.Fatalf("ListCheckRuns should call gh api with --paginate, got:\n%s", logged) + } + if strings.Contains(string(logged), "--slurp") { + t.Fatalf("ListCheckRuns should not combine gh api --slurp with --jq, got:\n%s", logged) + } +} + +func TestDecodeGHCheckRuns(t *testing.T) { + tests := []struct { + name string + input string + wantNames []string + wantErr bool + }{ + {name: "empty", input: "", wantNames: nil}, + {name: "single object", input: `{"name":"unit","status":"completed","conclusion":"success"}`, wantNames: []string{"unit"}}, + { + name: "multiple objects with whitespace", + input: "\n {\"name\":\"unit\",\"status\":\"completed\",\"conclusion\":\"success\"}\n{\"name\":\"lint\",\"status\":\"completed\",\"conclusion\":\"failure\"}\t", + wantNames: []string{"unit", "lint"}, + }, + {name: "invalid json", input: `{"name":"unit"`, wantErr: true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, err := decodeGHCheckRuns(tt.input) + if tt.wantErr { + if err == nil { + t.Fatal("expected error") + } + return + } + if err != nil { + t.Fatalf("decodeGHCheckRuns: %v", err) + } + if len(got) != len(tt.wantNames) { + t.Fatalf("runs = %d, want %d: %+v", len(got), len(tt.wantNames), got) + } + for i, name := range tt.wantNames { + if got[i].Name != name { + t.Fatalf("run[%d].Name = %q, want %q", i, got[i].Name, name) + } + } + }) + } +} + func TestParseRepoURL(t *testing.T) { tests := []struct { name string diff --git a/apps/backend/internal/github/handlers.go b/apps/backend/internal/github/handlers.go index 2cce5c49e4..a50681f1e3 100644 --- a/apps/backend/internal/github/handlers.go +++ b/apps/backend/internal/github/handlers.go @@ -2,6 +2,7 @@ package github import ( "context" + "errors" "strings" "github.com/gin-gonic/gin" @@ -261,7 +262,10 @@ func wsSyncTaskPR(svc *Service, _ *logger.Logger) func(ctx context.Context, msg return wsWithField("task_id", func(ctx context.Context, taskID string) (interface{}, error) { prs, permanent, err := svc.TriggerPRSyncAllPermanent(ctx, taskID) if err != nil { - return nil, err + var partial *PartialPRSyncError + if !permanent && (!errors.As(err, &partial) || len(prs) == 0) { + return nil, err + } } // Return an envelope so the frontend always gets a deterministic // shape even on empty results (`{prs: []}`); a bare `nil` slice diff --git a/apps/backend/internal/github/pat_client_test.go b/apps/backend/internal/github/pat_client_test.go index f2ea9b6dc9..ade84789a9 100644 --- a/apps/backend/internal/github/pat_client_test.go +++ b/apps/backend/internal/github/pat_client_test.go @@ -196,6 +196,50 @@ func TestConvertPatPR_MergeableState(t *testing.T) { } } +func TestPATClient_ListCheckRuns_PaginatesCheckRuns(t *testing.T) { + var checkRunQueries []string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case "/repos/acme/widget/commits/sha/check-runs": + checkRunQueries = append(checkRunQueries, r.URL.RawQuery) + switch r.URL.Query().Get("page") { + case "", "1": + w.Header().Set("Link", `<`+githubAPIBase+`/repos/acme/widget/commits/sha/check-runs?per_page=100&page=2>; rel="next"`) + _, _ = w.Write([]byte(`{"check_runs":[{"name":"unit","status":"completed","conclusion":"success","html_url":"https://ci/unit"}]}`)) + case "2": + _, _ = w.Write([]byte(`{"check_runs":[{"name":"lint","status":"completed","conclusion":"failure","html_url":"https://ci/lint"}]}`)) + default: + t.Errorf("unexpected check-runs page %q", r.URL.Query().Get("page")) + http.Error(w, "unexpected page", http.StatusInternalServerError) + } + case "/repos/acme/widget/commits/sha/status": + _, _ = w.Write([]byte(`{"statuses":[]}`)) + default: + t.Errorf("unexpected path %q", r.URL.Path) + http.Error(w, "unexpected path", http.StatusInternalServerError) + } + })) + t.Cleanup(srv.Close) + + c := newPATClientPointingAt(t, srv.URL) + checks, err := c.ListCheckRuns(context.Background(), "acme", "widget", "sha") + if err != nil { + t.Fatalf("ListCheckRuns: %v", err) + } + if len(checks) != 2 { + t.Fatalf("checks = %d, want 2: %+v", len(checks), checks) + } + if checks[1].Name != "lint" || checks[1].Conclusion != "failure" { + t.Fatalf("expected failed lint check from page 2, got %+v", checks[1]) + } + if len(checkRunQueries) != 2 { + t.Fatalf("check-runs requests = %d, want 2 (%v)", len(checkRunQueries), checkRunQueries) + } + if !strings.Contains(checkRunQueries[0], "per_page=100") { + t.Fatalf("first check-runs request should request per_page=100, got %q", checkRunQueries[0]) + } +} + func TestPATClient_FindPRByBranch_UsesGraphQLHeadRefName(t *testing.T) { srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/graphql" { diff --git a/apps/backend/internal/github/poller_test.go b/apps/backend/internal/github/poller_test.go index ecd411e6ac..81126c9599 100644 --- a/apps/backend/internal/github/poller_test.go +++ b/apps/backend/internal/github/poller_test.go @@ -14,6 +14,7 @@ import ( _ "github.com/mattn/go-sqlite3" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" ) @@ -642,6 +643,187 @@ func TestTryBatchedPRWatchCheck_NumberedWatch_AppliesStatus(t *testing.T) { } } +func TestTryBatchedPRWatchCheck_PublishesOnPRFeedbackWatermarkChange(t *testing.T) { + poller, _, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + prUpdatedAt := time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC) + gh.prResponses = []string{`{ + "data": { + "repo0": { + "pr0": { + "state": "OPEN", "title": "Test PR", "url": "https://x/1", + "isDraft": false, "mergeable": "MERGEABLE", "mergeStateStatus": "CLEAN", + "headRefName": "feat", "baseRefName": "main", "headRefOid": "abc", + "author": {"login": "alice"}, + "createdAt": "2026-01-01T00:00:00Z", "updatedAt": "2026-01-02T00:00:00Z", + "reviews": {"nodes": []}, + "reviewRequests": {"totalCount": 0}, + "commits": {"nodes": [{"commit": {"statusCheckRollup": {"state": "SUCCESS"}}}]} + } + } + } + }`} + + gotEvent := make(chan *bus.Event, 1) + if _, err := poller.eventBus.Subscribe(events.GitHubPRFeedback, func(_ context.Context, event *bus.Event) error { + gotEvent <- event + return nil + }); err != nil { + t.Fatalf("subscribe PR feedback: %v", err) + } + + watch := &PRWatch{ + SessionID: "s1", + TaskID: "t1", + Owner: "o", + Repo: "r", + PRNumber: 42, + Branch: "feat", + LastCheckStatus: "success", + LastReviewState: "", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + if err := store.CreateTaskPR(ctx, &TaskPR{ + TaskID: "t1", + Owner: "o", + Repo: "r", + PRNumber: 42, + PRURL: "https://github.com/o/r/pull/42", + PRTitle: "Test PR", + HeadBranch: "feat", + BaseBranch: "main", + State: "open", + }); err != nil { + t.Fatalf("create task PR: %v", err) + } + + if !poller.tryBatchedPRWatchCheck(ctx, []*PRWatch{watch}) { + t.Fatalf("expected batched path to succeed") + } + select { + case <-gotEvent: + default: + t.Fatal("expected PR feedback watermark change to publish PR feedback event") + } + updated, err := store.GetPRWatchBySession(ctx, "s1") + if err != nil || updated == nil { + t.Fatalf("get PR watch: err=%v, w=%v", err, updated) + } + if updated.LastCommentAt == nil || !updated.LastCommentAt.Equal(prUpdatedAt) { + t.Fatalf("LastCommentAt = %v, want %v", updated.LastCommentAt, prUpdatedAt) + } +} + +func TestPRWatchFeedbackHelpers(t *testing.T) { + now := time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC) + older := now.Add(-time.Hour) + newer := now.Add(time.Hour) + status := &PRStatus{PR: &PR{UpdatedAt: now}} + if prWatchFeedbackUpdatedSinceWatch(nil, status) { + t.Fatal("nil watch should not report changed") + } + if prWatchFeedbackUpdatedSinceWatch(&PRWatch{}, nil) { + t.Fatal("nil status should not report changed") + } + if prWatchFeedbackUpdatedSinceWatch(&PRWatch{}, &PRStatus{PR: &PR{}}) { + t.Fatal("zero PR updated_at should not report changed") + } + if !prWatchFeedbackUpdatedSinceWatch(&PRWatch{}, status) { + t.Fatal("nil watermark should report changed for non-zero PR updated_at") + } + if !prWatchFeedbackUpdatedSinceWatch(&PRWatch{LastCommentAt: &older}, status) { + t.Fatal("newer PR updated_at should report changed") + } + if prWatchFeedbackUpdatedSinceWatch(&PRWatch{LastCommentAt: &newer}, status) { + t.Fatal("older PR updated_at should not report changed") + } + if got := prWatchFeedbackWatermark(&PRWatch{LastCommentAt: &newer}, status); got == nil || !got.Equal(newer) { + t.Fatalf("watermark should not move backward, got %v want %v", got, newer) + } + if got := prWatchFeedbackWatermark(&PRWatch{LastCommentAt: &older}, status); got == nil || !got.Equal(now) { + t.Fatalf("watermark should advance to PR updated_at, got %v want %v", got, now) + } + if got := prWatchFeedbackWatermark(&PRWatch{LastCommentAt: &older}, &PRStatus{}); got == nil || !got.Equal(older) { + t.Fatalf("watermark should preserve existing value without PR updated_at, got %v want %v", got, older) + } +} + +func TestTriggerPRSyncAll_ReturnsPerWatchFallbackErrors(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + watch := &PRWatch{ + SessionID: "s1", + TaskID: "t1", + Owner: "o", + Repo: "r", + PRNumber: 42, + Branch: "feat", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + + prs, err := svc.TriggerPRSyncAll(ctx, "t1") + if err == nil { + t.Fatal("expected per-watch fallback error") + } + if len(prs) != 0 { + t.Fatalf("prs = %d, want 0: %+v", len(prs), prs) + } + if !strings.Contains(err.Error(), "o/r#42") { + t.Fatalf("expected aggregate error to identify failed PR, got %v", err) + } +} + +func TestTriggerPRSyncAll_ReturnsPartialErrorWithSuccessfulPRs(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + mockClient.AddPR(&PR{ + Number: 1, + Title: "live", + State: "open", + HeadSHA: "sha-live", + HeadBranch: "feat-live", + BaseBranch: "main", + RepoOwner: "o", + RepoName: "r", + UpdatedAt: time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC), + }) + for _, watch := range []*PRWatch{ + {SessionID: "s1", TaskID: "t1", RepositoryID: "repo-live", Owner: "o", Repo: "r", PRNumber: 1, Branch: "feat-live"}, + {SessionID: "s2", TaskID: "t1", RepositoryID: "repo-dead", Owner: "o", Repo: "r", PRNumber: 2, Branch: "feat-dead"}, + } { + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatalf("create PR watch: %v", err) + } + } + if err := store.CreateTaskPR(ctx, &TaskPR{ + TaskID: "t1", + RepositoryID: "repo-live", + Owner: "o", + Repo: "r", + PRNumber: 1, + PRURL: "https://github.com/o/r/pull/1", + PRTitle: "live", + HeadBranch: "feat-live", + BaseBranch: "main", + State: "open", + }); err != nil { + t.Fatalf("create task PR: %v", err) + } + + prs, err := svc.TriggerPRSyncAll(ctx, "t1") + var partial *PartialPRSyncError + if !errors.As(err, &partial) { + t.Fatalf("expected PartialPRSyncError, got %v", err) + } + if len(prs) != 1 || prs[0].RepositoryID != "repo-live" { + t.Fatalf("expected successful live PR result, got %+v", prs) + } +} + func TestTryBatchedPRWatchCheck_SearchingWatch_DetectsPR(t *testing.T) { poller, _, gh, store := setupBatchedPollerTest(t) ctx := context.Background() diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index df46d64ca8..b71d43c463 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -33,8 +33,8 @@ const ( // service's self-approval guard, and tests. const reviewEventApprove = "APPROVE" -// prSyncFreshnessWindow is how long PR data is considered fresh (skip GitHub API). -const prSyncFreshnessWindow = 30 * time.Second +// PRSyncFreshnessWindow is how long PR data is considered fresh. +const PRSyncFreshnessWindow = 30 * time.Second // cleanupFetchFailureThreshold is the number of consecutive GetPRFeedback / // GetIssueState errors a single dedup row may accumulate before the cleanup diff --git a/apps/backend/internal/github/service_pr_sync_test.go b/apps/backend/internal/github/service_pr_sync_test.go index 31420f5841..3017cfd750 100644 --- a/apps/backend/internal/github/service_pr_sync_test.go +++ b/apps/backend/internal/github/service_pr_sync_test.go @@ -118,7 +118,7 @@ func TestTriggerPRSync_NoWatch(t *testing.T) { // last_checked_at. The frontend re-syncs every 5s while no PR is found, so // each repeated sync hit `gh` again and logged a warning. After the fix the // first sync probes once and stamps the watch; the immediate second sync is -// throttled within prSyncFreshnessWindow and must NOT probe again. +// throttled within PRSyncFreshnessWindow and must NOT probe again. func TestTriggerPRSyncAll_ThrottlesDetectionProbe(t *testing.T) { _, svc, mockClient, store := setupPollerTest(t) ctx := context.Background() @@ -144,7 +144,7 @@ func TestTriggerPRSyncAll_ThrottlesDetectionProbe(t *testing.T) { t.Fatalf("expected exactly 1 detection probe on first sync, got %d", got) } - // Second sync immediately after — well within prSyncFreshnessWindow. + // Second sync immediately after — well within PRSyncFreshnessWindow. if _, err := svc.TriggerPRSyncAll(ctx, "t1"); err != nil { t.Fatalf("second TriggerPRSyncAll: %v", err) } diff --git a/apps/backend/internal/github/service_pr_watch.go b/apps/backend/internal/github/service_pr_watch.go index ce24096e67..044b0eca9e 100644 --- a/apps/backend/internal/github/service_pr_watch.go +++ b/apps/backend/internal/github/service_pr_watch.go @@ -2,6 +2,7 @@ package github import ( "context" + "errors" "fmt" "strconv" "strings" @@ -122,16 +123,39 @@ func (s *Service) CheckPRWatch(ctx context.Context, watch *PRWatch) (*PRStatus, // Check for check status or review state changes hasNew := status.ChecksState != watch.LastCheckStatus || status.ReviewState != watch.LastReviewState + commentAt := prWatchFeedbackWatermark(watch, status) + hasNew = hasNew || prWatchFeedbackUpdatedSinceWatch(watch, status) // Update watch timestamps now := time.Now().UTC() - if err := s.store.UpdatePRWatchTimestamps(ctx, watch.ID, now, nil, status.ChecksState, status.ReviewState); err != nil { + if err := s.store.UpdatePRWatchTimestamps(ctx, watch.ID, now, commentAt, status.ChecksState, status.ReviewState); err != nil { s.logger.Error("failed to update PR watch timestamps", zap.String("id", watch.ID), zap.Error(err)) } return status, hasNew, nil } +func prWatchFeedbackUpdatedSinceWatch(watch *PRWatch, status *PRStatus) bool { + if watch == nil || status == nil || status.PR == nil || status.PR.UpdatedAt.IsZero() { + return false + } + return watch.LastCommentAt == nil || status.PR.UpdatedAt.After(*watch.LastCommentAt) +} + +func prWatchFeedbackWatermark(watch *PRWatch, status *PRStatus) *time.Time { + if status != nil && status.PR != nil && !status.PR.UpdatedAt.IsZero() { + updatedAt := status.PR.UpdatedAt + if watch != nil && watch.LastCommentAt != nil && watch.LastCommentAt.After(updatedAt) { + return watch.LastCommentAt + } + return &updatedAt + } + if watch == nil { + return nil + } + return watch.LastCommentAt +} + // EnsurePRWatch creates a PRWatch with pr_number=0 for a // (session, repo, branch) triple if one doesn't already exist. The poller // will detect the PR by searching for the branch on GitHub. `repositoryID` @@ -391,7 +415,7 @@ func (s *Service) ListWorkspaceTaskPRs(ctx context.Context, workspaceID string) staleTasks := make(map[string]struct{}) for taskID, prs := range result { for _, tp := range prs { - if tp.LastSyncedAt == nil || time.Since(*tp.LastSyncedAt) >= prSyncFreshnessWindow { + if tp.LastSyncedAt == nil || time.Since(*tp.LastSyncedAt) >= PRSyncFreshnessWindow { staleTasks[taskID] = struct{}{} break } @@ -726,6 +750,7 @@ func (s *Service) areAllWatchesPermanentlyMissing(watches []*PRWatch) bool { // single bad cycle doesn't leave the UI staring at stale data. func (s *Service) triggerPRSyncAllPerWatch(ctx context.Context, taskID string, watches []*PRWatch) ([]*TaskPR, error) { results := make([]*TaskPR, 0, len(watches)) + var syncErrs []error for _, w := range watches { var tp *TaskPR var syncErr error @@ -746,13 +771,40 @@ func (s *Service) triggerPRSyncAllPerWatch(ctx context.Context, taskID string, w zap.String("repository_id", w.RepositoryID), zap.Int("pr_number", w.PRNumber), zap.Error(syncErr)) + syncErrs = append(syncErrs, fmt.Errorf("%s/%s#%d: %w", w.Owner, w.Repo, w.PRNumber, syncErr)) continue } if tp != nil { results = append(results, tp) } } - return results, nil + if len(syncErrs) == 0 { + return results, nil + } + err := errors.Join(syncErrs...) + if len(results) > 0 { + return results, &PartialPRSyncError{Err: err} + } + return results, err +} + +// PartialPRSyncError reports that some PR watches failed while others synced. +type PartialPRSyncError struct { + Err error +} + +func (e *PartialPRSyncError) Error() string { + if e == nil || e.Err == nil { + return "partial PR sync failure" + } + return e.Err.Error() +} + +func (e *PartialPRSyncError) Unwrap() error { + if e == nil { + return nil + } + return e.Err } func (s *Service) triggerPRDetection(ctx context.Context, watch *PRWatch, taskID string) (*TaskPR, error) { @@ -796,7 +848,7 @@ func (s *Service) detectPRForWatchOnce(ctx context.Context, watch *PRWatch, task // Without this, a branch whose PR never appears (e.g. an unresolvable repo) // re-hits `gh` on every on-demand sync, and the frontend re-syncs every 5s // while no PR is found — flooding the logs with identical failures. - if watch.LastCheckedAt != nil && time.Since(*watch.LastCheckedAt) < prSyncFreshnessWindow { + if watch.LastCheckedAt != nil && time.Since(*watch.LastCheckedAt) < PRSyncFreshnessWindow { return s.store.GetTaskPRByRepository(ctx, taskID, watch.RepositoryID) } @@ -817,7 +869,7 @@ func (s *Service) detectPRForWatchOnce(ctx context.Context, watch *PRWatch, task // flood this fixes IS the error path (an unresolvable repo makes `gh` exit // non-zero), so stamping only on success would leave the bug unfixed. The // tradeoff: a transient GitHub error throttles the next on-demand probe for - // prSyncFreshnessWindow, which is fine — it stops hammering a failing + // PRSyncFreshnessWindow, which is fine — it stops hammering a failing // endpoint, and the 60s background poller still retries. This diverges // intentionally from poller.detectPRForWatch, which stamps only on success. // The empty-string check/review args clear last_check_status / @@ -868,7 +920,7 @@ func (s *Service) triggerPRStatusSync(ctx context.Context, watch *PRWatch, taskI return s.store.GetTaskPR(c, taskID) } if tp, _ := loadTaskPR(ctx); tp != nil && tp.LastSyncedAt != nil { - if time.Since(*tp.LastSyncedAt) < prSyncFreshnessWindow { + if time.Since(*tp.LastSyncedAt) < PRSyncFreshnessWindow { return tp, nil } } diff --git a/apps/backend/internal/github/service_pr_watch_batched.go b/apps/backend/internal/github/service_pr_watch_batched.go index 232d1644e4..a802ba34cd 100644 --- a/apps/backend/internal/github/service_pr_watch_batched.go +++ b/apps/backend/internal/github/service_pr_watch_batched.go @@ -228,17 +228,20 @@ func (s *Service) applyBatchedNumberedWatch( // Bump last_checked_at to suppress retry storms and surface SyncFailed // so the WS handler can flag the result as permanent for the frontend. if s.isRepoCachedAsMissing(w.Owner, w.Repo) { - _ = s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, nil, "", "") + _ = s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, w.LastCommentAt, "", "") return PRWatchSyncResult{Watch: w, SyncFailed: true} } status, ok := statusByKey[prStatusCacheKey(w.Owner, w.Repo, w.PRNumber)] if !ok || status == nil { // Alias missing — best-effort liveness bump so we don't immediately re-probe. - _ = s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, nil, "", "") + _ = s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, w.LastCommentAt, "", "") return PRWatchSyncResult{Watch: w} } - changed := status.ChecksState != w.LastCheckStatus || status.ReviewState != w.LastReviewState - if err := s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, nil, status.ChecksState, status.ReviewState); err != nil { + changed := status.ChecksState != w.LastCheckStatus || + status.ReviewState != w.LastReviewState || + prWatchFeedbackUpdatedSinceWatch(w, status) + commentAt := prWatchFeedbackWatermark(w, status) + if err := s.store.UpdatePRWatchTimestamps(ctx, w.ID, now, commentAt, status.ChecksState, status.ReviewState); err != nil { s.logger.Error("failed to update PR watch timestamps", zap.String("id", w.ID), zap.Error(err)) } // Gap-fill: a numbered watch can exist even when its exact task_pr row was diff --git a/apps/backend/internal/github/service_repo_negative_cache_test.go b/apps/backend/internal/github/service_repo_negative_cache_test.go index ca5146fc15..1222c1dc3a 100644 --- a/apps/backend/internal/github/service_repo_negative_cache_test.go +++ b/apps/backend/internal/github/service_repo_negative_cache_test.go @@ -2,11 +2,14 @@ package github import ( "context" + "errors" "runtime" "strings" "sync" "sync/atomic" "testing" + + ws "github.com/kandev/kandev/pkg/websocket" ) // TestSyncWatchesBatched_NegativeCache_ShortCircuits is the regression @@ -320,3 +323,45 @@ func TestService_TriggerPRSyncAllPermanent_FlagsDeadRepos(t *testing.T) { } } } + +func TestWSSyncTaskPR_ReturnsPermanentEnvelopeWhenSyncErrors(t *testing.T) { + _, svc, gh, store := setupBatchedPollerTest(t) + ctx := context.Background() + + w := &PRWatch{SessionID: "s1", TaskID: "t1", Owner: "Dead", Repo: "Repo", PRNumber: 0, Branch: "x"} + if err := store.CreatePRWatch(ctx, w); err != nil { + t.Fatalf("create watch: %v", err) + } + gh.branchErr = &batchedMissingReposErr{ + Repos: []repoRef{{Owner: w.Owner, Repo: w.Repo}}, + Inner: errors.New("sibling repo unavailable"), + } + + msg, err := ws.NewRequest("req-1", ws.ActionGitHubTaskPRSync, map[string]string{"task_id": "t1"}) + if err != nil { + t.Fatalf("NewRequest: %v", err) + } + resp, err := wsSyncTaskPR(svc, nil)(ctx, msg) + if err != nil { + t.Fatalf("wsSyncTaskPR returned transport error: %v", err) + } + if resp.Type != ws.MessageTypeResponse { + t.Fatalf("response type = %q, want %q; payload=%s", resp.Type, ws.MessageTypeResponse, string(resp.Payload)) + } + var payload struct { + PRs []*TaskPR `json:"prs"` + Permanent bool `json:"permanent"` + } + if err := resp.ParsePayload(&payload); err != nil { + t.Fatalf("ParsePayload: %v", err) + } + if !payload.Permanent { + t.Fatalf("permanent = false, want true") + } + if payload.PRs == nil { + t.Fatalf("prs = nil, want empty slice") + } + if len(payload.PRs) != 0 { + t.Fatalf("prs length = %d, want 0", len(payload.PRs)) + } +} diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index b734dd00d0..bf2519b5cb 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -1121,6 +1121,8 @@ func (s *Store) RefreshTaskCIFixCheckpoint(ctx context.Context, taskID, reposito 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 = NULL, + last_fix_session_id = NULL, last_error = NULL, updated_at = excluded.updated_at`, taskID, repositoryID, prNumber, signature, checkpointJSON, now, now) diff --git a/apps/backend/internal/github/store_ci_automation_test.go b/apps/backend/internal/github/store_ci_automation_test.go index 14ad420fab..0e90eb5e77 100644 --- a/apps/backend/internal/github/store_ci_automation_test.go +++ b/apps/backend/internal/github/store_ci_automation_test.go @@ -131,7 +131,7 @@ func TestStoreTaskCIPRState_RecordAttemptsAndError(t *testing.T) { } } -func TestStoreTaskCIPRState_RefreshCheckpointPreservesPromptDispatchMetadata(t *testing.T) { +func TestStoreTaskCIPRState_RefreshCheckpointClearsPromptDispatchMetadata(t *testing.T) { store := newTestStore(t) ctx := context.Background() enqueuedAt := time.Date(2026, 6, 18, 12, 0, 0, 0, time.UTC) @@ -158,11 +158,11 @@ func TestStoreTaskCIPRState_RefreshCheckpointPreservesPromptDispatchMetadata(t * if state.LastFixSignature != "after" || state.LastFixCheckpointJSON != `{"failed_checks":[]}` { t.Fatalf("checkpoint was not refreshed: %+v", state) } - if state.LastFixSessionID == nil || *state.LastFixSessionID != "session-1" { - t.Fatalf("LastFixSessionID=%v, want session-1", state.LastFixSessionID) + if state.LastFixSessionID != nil { + t.Fatalf("LastFixSessionID=%v, want nil", state.LastFixSessionID) } - if state.LastFixEnqueuedAt == nil || !state.LastFixEnqueuedAt.Equal(enqueuedAt) { - t.Fatalf("LastFixEnqueuedAt=%v, want %v", state.LastFixEnqueuedAt, enqueuedAt) + if state.LastFixEnqueuedAt != nil { + t.Fatalf("LastFixEnqueuedAt=%v, want nil", state.LastFixEnqueuedAt) } } diff --git a/apps/backend/internal/orchestrator/event_handlers_github.go b/apps/backend/internal/orchestrator/event_handlers_github.go index 4cb720d3cc..9ff993fcdb 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github.go +++ b/apps/backend/internal/orchestrator/event_handlers_github.go @@ -42,6 +42,7 @@ type GitHubService interface { AssociatePRWithTask(ctx context.Context, taskID, repositoryID string, pr *github.PR) (*github.TaskPR, error) GetTaskPR(ctx context.Context, taskID string) (*github.TaskPR, error) ListTaskPRs(ctx context.Context, taskIDs []string) (map[string][]*github.TaskPR, error) + TriggerPRSyncAll(ctx context.Context, taskID string) ([]*github.TaskPR, error) GetTaskPRByOwnerRepoNumber(ctx context.Context, taskID, owner, repo string, prNumber int) (*github.TaskPR, error) GetTaskCIOptionsResponse(ctx context.Context, taskID string) (*github.TaskCIOptionsResponse, error) GetTaskCIPRState(ctx context.Context, taskID, repositoryID string, prNumber int) (*github.TaskCIPRAutomationState, error) @@ -192,18 +193,66 @@ func (s *Service) handleTaskCIOptionsUpdated(ctx context.Context, event *bus.Eve } detachedCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), ciAutomationDetachedTimeout) defer cancel() - prsByTask, err := s.githubService.ListTaskPRs(detachedCtx, []string{options.TaskID}) + prs, err := s.githubService.TriggerPRSyncAll(detachedCtx, options.TaskID) if err != nil { - s.logger.Debug("failed to load task PRs for CI automation options update", zap.String("task_id", options.TaskID), zap.Error(err)) - return nil + s.logger.Warn("failed to sync task PRs for CI automation options update", zap.String("task_id", options.TaskID), zap.Error(err)) + s.recordTaskCIOptionsSyncError(detachedCtx, options.TaskID, prs, err) } - for _, pr := range prsByTask[options.TaskID] { - s.startTaskPRCIAutomation(ctx, pr) + for _, pr := range prs { + s.startTaskPRCIAutomationWithoutRefresh(ctx, pr) } return nil } +func (s *Service) recordTaskCIOptionsSyncError(ctx context.Context, taskID string, synced []*github.TaskPR, syncErr error) { + syncedKeys := make(map[ciAutomationTaskPRSyncKey]struct{}, len(synced)) + for _, pr := range synced { + if pr == nil { + continue + } + syncedKeys[ciAutomationTaskPRSyncKeyFor(pr)] = struct{}{} + } + prsByTask, err := s.githubService.ListTaskPRs(ctx, []string{taskID}) + if err != nil { + s.logger.Warn("failed to load task PRs after CI automation sync failure", zap.String("task_id", taskID), zap.Error(err)) + return + } + for _, pr := range prsByTask[taskID] { + if _, ok := syncedKeys[ciAutomationTaskPRSyncKeyFor(pr)]; ok { + continue + } + s.recordCIAutomationError(ctx, pr, fmt.Sprintf("sync PR status: %v", syncErr)) + } +} + +type ciAutomationTaskPRSyncKey struct { + repositoryID string + owner string + repo string + prNumber int +} + +func ciAutomationTaskPRSyncKeyFor(pr *github.TaskPR) ciAutomationTaskPRSyncKey { + if pr == nil { + return ciAutomationTaskPRSyncKey{} + } + return ciAutomationTaskPRSyncKey{ + repositoryID: pr.RepositoryID, + owner: strings.ToLower(pr.Owner), + repo: strings.ToLower(pr.Repo), + prNumber: pr.PRNumber, + } +} + func (s *Service) startTaskPRCIAutomation(ctx context.Context, pr *github.TaskPR) { + s.startTaskPRCIAutomationWithRefresh(ctx, pr, true) +} + +func (s *Service) startTaskPRCIAutomationWithoutRefresh(ctx context.Context, pr *github.TaskPR) { + s.startTaskPRCIAutomationWithRefresh(ctx, pr, false) +} + +func (s *Service) startTaskPRCIAutomationWithRefresh(ctx context.Context, pr *github.TaskPR, refresh bool) { if pr == nil { return } @@ -219,7 +268,7 @@ func (s *Service) startTaskPRCIAutomation(ctx context.Context, pr *github.TaskPR defer s.ciAutomationInFlight.Delete(key) automationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), ciAutomationDetachedTimeout) defer cancel() - if err := s.handleTaskPRCIAutomation(automationCtx, pr); err != nil { + if err := s.handleTaskPRCIAutomationWithRefresh(automationCtx, pr, refresh); err != nil { s.logger.Debug("CI automation handling failed", zap.String("task_id", pr.TaskID), zap.Error(err)) } }() diff --git a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go index 169be6d276..140ee4183f 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go +++ b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation.go @@ -21,10 +21,10 @@ const ( ciAutomationOrigin = "ci_automation" ciAutomationCheckSuccess = "success" ciAutomationCheckFailure = "failure" - ciAutomationCheckSkipped = "skipped" ciAutomationCheckCompleted = "completed" ciAutomationChangesRequested = "changes_requested" ciAutomationPRFeedbackToken = "{{pr.feedback}}" + ciAutomationFixBlockWindow = time.Hour ) var ciAutomationSnapshotFieldReplacer = strings.NewReplacer("\r", " ", "\n", " ", "<", "", ">", "") @@ -49,6 +49,10 @@ type ciAutomationCommentSnapshot struct { } func (s *Service) handleTaskPRCIAutomation(ctx context.Context, pr *github.TaskPR) error { + return s.handleTaskPRCIAutomationWithRefresh(ctx, pr, true) +} + +func (s *Service) handleTaskPRCIAutomationWithRefresh(ctx context.Context, pr *github.TaskPR, refresh bool) error { if s.githubService == nil || pr == nil { return nil } @@ -57,51 +61,110 @@ func (s *Service) handleTaskPRCIAutomation(ctx context.Context, pr *github.TaskP s.logger.Debug("load CI automation options failed", zap.String("task_id", pr.TaskID), zap.Error(err)) return nil } - if options.AutoFixEnabled && ciAutomationShouldAutoFix(pr) { - s.handleTaskPRCIAutoFix(ctx, pr, options) + freshlySynced := ciAutomationHasFreshPRStatus(pr) + if refresh && (options.AutoFixEnabled || options.AutoMergeEnabled) { + refreshed, synced, syncErr := s.refreshTaskPRForCIAutomation(ctx, pr) + if syncErr != nil { + s.recordCIAutomationError(ctx, pr, fmt.Sprintf("sync PR status: %v", syncErr)) + return nil + } + pr = refreshed + freshlySynced = synced + } + autoFixBlockedMerge := false + if options.AutoFixEnabled && ciAutomationCanAutoFixFromFeedback(pr) { + autoFixBlockedMerge = s.handleTaskPRCIAutoFix(ctx, pr, options) + } + if autoFixBlockedMerge { + return nil } if options.AutoMergeEnabled && ciAutomationReadyToMerge(pr) { + if !freshlySynced { + s.recordCIAutomationError(ctx, pr, "PR status is not freshly synced for auto-merge") + return nil + } s.handleTaskPRCIAutoMerge(ctx, pr) } return nil } -func (s *Service) handleTaskPRCIAutoFix(ctx context.Context, pr *github.TaskPR, options *github.TaskCIOptionsResponse) { +func (s *Service) refreshTaskPRForCIAutomation(ctx context.Context, pr *github.TaskPR) (*github.TaskPR, bool, error) { + if pr == nil { + return nil, false, nil + } + prs, err := s.githubService.TriggerPRSyncAll(ctx, pr.TaskID) + if refreshed := ciAutomationFindMatchingPR(prs, pr); refreshed != nil { + return refreshed, ciAutomationHasFreshPRStatus(refreshed), nil + } + if err != nil { + return nil, false, err + } + return pr, false, nil +} + +func ciAutomationFindMatchingPR(prs []*github.TaskPR, target *github.TaskPR) *github.TaskPR { + if target == nil { + return nil + } + for _, pr := range prs { + if pr == nil || pr.TaskID != target.TaskID || pr.PRNumber != target.PRNumber { + continue + } + if target.RepositoryID != "" && pr.RepositoryID == target.RepositoryID { + return pr + } + if target.RepositoryID == "" && pr.Owner == target.Owner && pr.Repo == target.Repo { + return pr + } + } + return nil +} + +func ciAutomationHasFreshPRStatus(pr *github.TaskPR) bool { + return ciAutomationHasFreshPRStatusAt(pr, time.Now()) +} + +func ciAutomationHasFreshPRStatusAt(pr *github.TaskPR, now time.Time) bool { + if pr == nil || pr.LastSyncedAt == nil { + return false + } + return now.Sub(*pr.LastSyncedAt) <= github.PRSyncFreshnessWindow +} + +func (s *Service) handleTaskPRCIAutoFix(ctx context.Context, pr *github.TaskPR, options *github.TaskCIOptionsResponse) bool { 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)) - return + return true } + if !ciAutomationCanAutoFixFromFeedbackPR(feedback) { + 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 + return true } previous := decodeCIAutomationCheckpoint(state) delta := ciAutomationBuildDelta(feedback, previous) checkpoint := ciAutomationCurrentCheckpoint(feedback) - if len(delta.FailedChecks) == 0 && len(delta.Comments) == 0 { - if state != nil && len(previous.FailedChecks)+len(previous.Comments) > 0 { - checkpointJSON, signature := encodeCIAutomationCheckpoint(checkpoint) - if err := s.githubService.RefreshTaskCIFixCheckpoint(context.WithoutCancel(ctx), pr.TaskID, pr.RepositoryID, pr.PRNumber, signature, checkpointJSON); err != nil { - s.logger.Debug("record CI auto-fix checkpoint refresh failed", zap.String("task_id", pr.TaskID), zap.Error(err)) - } - } - return - } checkpointJSON, signature := encodeCIAutomationCheckpoint(checkpoint) + if len(delta.FailedChecks) == 0 && len(delta.Comments) == 0 { + return s.handleTaskPRCIAutoFixEmptyDelta(ctx, pr, state, previous, signature, checkpointJSON) + } if state != nil && state.LastFixSignature == signature { - return + return ciAutomationDuplicateFixAttemptBlocksMerge(state) } prompt := ciAutomationRenderPrompt(options.EffectiveAutoFixPrompt, pr, delta) session, err := s.repo.GetActiveTaskSessionByTaskID(ctx, pr.TaskID) if err != nil || session == nil { s.recordCIAutomationError(ctx, pr, "no promptable task session for CI auto-fix") - return + return true } if err := s.dispatchCIAutomationPrompt(ctx, session, prompt); err != nil { s.recordCIAutomationError(ctx, pr, err.Error()) - return + return true } if err := s.githubService.RecordTaskCIFixAttempt(context.WithoutCancel(ctx), github.TaskCIFixAttempt{ TaskID: pr.TaskID, @@ -114,6 +177,19 @@ func (s *Service) handleTaskPRCIAutoFix(ctx context.Context, pr *github.TaskPR, }); err != nil { s.logger.Debug("record CI auto-fix attempt failed", zap.String("task_id", pr.TaskID), zap.Error(err)) } + return true +} + +func (s *Service) handleTaskPRCIAutoFixEmptyDelta(ctx context.Context, pr *github.TaskPR, state *github.TaskCIPRAutomationState, previous ciAutomationCheckpoint, signature, checkpointJSON string) bool { + if state != nil && state.LastFixSignature == signature && ciAutomationDuplicateFixAttemptBlocksMerge(state) { + return true + } + if state != nil && len(previous.FailedChecks)+len(previous.Comments) > 0 { + if err := s.githubService.RefreshTaskCIFixCheckpoint(context.WithoutCancel(ctx), pr.TaskID, pr.RepositoryID, pr.PRNumber, signature, checkpointJSON); err != nil { + s.logger.Debug("record CI auto-fix checkpoint refresh failed", zap.String("task_id", pr.TaskID), zap.Error(err)) + } + } + return false } func (s *Service) handleTaskPRCIAutoMerge(ctx context.Context, pr *github.TaskPR) { @@ -204,16 +280,41 @@ func ciAutomationChatPrompt(prompt string) string { } func (s *Service) recordCIAutomationError(ctx context.Context, pr *github.TaskPR, message string) { + s.logger.Warn("CI automation error", + zap.String("task_id", pr.TaskID), + zap.String("repository_id", pr.RepositoryID), + zap.String("owner", pr.Owner), + zap.String("repo", pr.Repo), + zap.Int("pr_number", pr.PRNumber), + zap.String("error", message)) if err := s.githubService.RecordTaskCIError(context.WithoutCancel(ctx), pr.TaskID, pr.RepositoryID, pr.PRNumber, message); err != nil { s.logger.Debug("record CI automation error failed", zap.String("task_id", pr.TaskID), zap.Error(err)) } } -func ciAutomationShouldAutoFix(pr *github.TaskPR) bool { - if pr == nil || pr.State == "closed" || pr.State == "merged" { - return false +func ciAutomationCanAutoFixFromFeedback(pr *github.TaskPR) bool { + return pr != nil && pr.State != "closed" && pr.State != "merged" +} + +func ciAutomationCanAutoFixFromFeedbackPR(feedback *github.PRFeedback) bool { + if feedback == nil || feedback.PR == nil { + return true } - return pr.ChecksState == ciAutomationCheckFailure || pr.ReviewState == ciAutomationChangesRequested || pr.UnresolvedReviewThreads > 0 + return feedback.PR.State != "closed" && feedback.PR.State != "merged" +} + +func ciAutomationFilterFeedbackForPR(pr *github.TaskPR, feedback *github.PRFeedback) *github.PRFeedback { + if feedback == nil || pr == nil || pr.UnresolvedReviewThreads > 0 { + return feedback + } + filtered := *feedback + filtered.Comments = make([]github.PRComment, 0, len(feedback.Comments)) + for _, comment := range feedback.Comments { + if comment.Path == "" && comment.Line == 0 { + filtered.Comments = append(filtered.Comments, comment) + } + } + return &filtered } func ciAutomationReadyToMerge(pr *github.TaskPR) bool { @@ -246,7 +347,7 @@ func ciAutomationBuildDelta(feedback *github.PRFeedback, previous ciAutomationCh return delta } for _, check := range feedback.Checks { - if check.Status != ciAutomationCheckCompleted || check.Conclusion == ciAutomationCheckSuccess || check.Conclusion == ciAutomationCheckSkipped { + if check.Status != ciAutomationCheckCompleted || !ciAutomationCheckConclusionNeedsFix(check.Conclusion) { continue } snap := ciAutomationCheckSnapshot{Name: check.Name, Conclusion: check.Conclusion, HTMLURL: check.HTMLURL, Output: check.Output} @@ -266,6 +367,24 @@ func ciAutomationBuildDelta(feedback *github.PRFeedback, previous ciAutomationCh return delta } +func ciAutomationCheckConclusionNeedsFix(conclusion string) bool { + return conclusion == ciAutomationCheckFailure || + conclusion == "timed_out" || + conclusion == "cancelled" || + conclusion == "action_required" +} + +func ciAutomationDuplicateFixAttemptBlocksMerge(state *github.TaskCIPRAutomationState) bool { + return ciAutomationDuplicateFixAttemptBlocksMergeAt(state, time.Now()) +} + +func ciAutomationDuplicateFixAttemptBlocksMergeAt(state *github.TaskCIPRAutomationState, now time.Time) bool { + if state == nil || state.LastFixEnqueuedAt == nil { + return false + } + return now.Sub(*state.LastFixEnqueuedAt) <= ciAutomationFixBlockWindow +} + func ciAutomationCheckKey(check ciAutomationCheckSnapshot) string { return check.Name + "|" + check.Conclusion + "|" + check.HTMLURL + "|" + check.Output } diff --git a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go index 6ba5faae86..7009bbe366 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_github_ci_automation_test.go @@ -15,28 +15,6 @@ import ( "github.com/kandev/kandev/internal/task/models" ) -func TestCIAutomationShouldAutoFix(t *testing.T) { - tests := []struct { - name string - pr *github.TaskPR - want bool - }{ - {name: "failed checks", pr: &github.TaskPR{ChecksState: "failure"}, want: true}, - {name: "changes requested", pr: &github.TaskPR{ReviewState: "changes_requested"}, want: true}, - {name: "unresolved threads", pr: &github.TaskPR{UnresolvedReviewThreads: 1}, want: true}, - {name: "passing approved", pr: &github.TaskPR{ChecksState: "success", ReviewState: "approved"}, want: false}, - {name: "closed ignored", pr: &github.TaskPR{State: "closed", ChecksState: "failure"}, want: false}, - {name: "merged ignored", pr: &github.TaskPR{State: "merged", ReviewState: "changes_requested"}, want: false}, - } - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - if got := ciAutomationShouldAutoFix(tt.pr); got != tt.want { - t.Fatalf("ciAutomationShouldAutoFix=%v, want %v", got, tt.want) - } - }) - } -} - func TestCIAutomationReadyToMerge(t *testing.T) { required := 1 ready := github.TaskPR{ @@ -165,6 +143,23 @@ func TestCIAutomationFeedbackDeltaIncludesEditedComments(t *testing.T) { } } +func TestCIAutomationFilterFeedbackForPRSkipsReviewCommentsWithoutUnresolvedThreads(t *testing.T) { + feedback := &github.PRFeedback{ + Comments: []github.PRComment{ + {ID: 1, Body: "resolved review comment", Path: "main.go", Line: 12}, + {ID: 2, Body: "plain PR comment"}, + }, + } + filtered := ciAutomationFilterFeedbackForPR(&github.TaskPR{}, feedback) + if len(filtered.Comments) != 1 || filtered.Comments[0].ID != 2 { + t.Fatalf("expected only plain PR comment, got %+v", filtered.Comments) + } + withThreads := ciAutomationFilterFeedbackForPR(&github.TaskPR{UnresolvedReviewThreads: 1}, feedback) + if len(withThreads.Comments) != 2 { + t.Fatalf("expected unresolved review threads to keep review comments, got %+v", withThreads.Comments) + } +} + func TestCIAutomationFeedbackDeltaIncludesChangedCheckOutput(t *testing.T) { previous := ciAutomationCheckpoint{ FailedChecks: []ciAutomationCheckSnapshot{{Name: "unit", Conclusion: "failure", HTMLURL: "https://ci/unit", Output: "old"}}, @@ -179,6 +174,37 @@ func TestCIAutomationFeedbackDeltaIncludesChangedCheckOutput(t *testing.T) { } } +func TestCIAutomationFeedbackDeltaIgnoresNeutralChecks(t *testing.T) { + feedback := &github.PRFeedback{ + Checks: []github.CheckRun{ + {Name: "optional", Status: "completed", Conclusion: "neutral", HTMLURL: "https://ci/optional"}, + {Name: "future", Status: "completed", Conclusion: "stale", HTMLURL: "https://ci/future"}, + {Name: "unit", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/unit"}, + }, + } + + delta := ciAutomationBuildDelta(feedback, ciAutomationCheckpoint{}) + if len(delta.FailedChecks) != 1 || delta.FailedChecks[0].Name != "unit" { + t.Fatalf("expected only failing check in delta, got %+v", delta.FailedChecks) + } +} + +func TestCIAutomationFeedbackDeltaIncludesKnownFailingConclusions(t *testing.T) { + feedback := &github.PRFeedback{ + Checks: []github.CheckRun{ + {Name: "failure", Status: "completed", Conclusion: "failure", HTMLURL: "https://ci/failure"}, + {Name: "timed out", Status: "completed", Conclusion: "timed_out", HTMLURL: "https://ci/timed-out"}, + {Name: "cancelled", Status: "completed", Conclusion: "cancelled", HTMLURL: "https://ci/cancelled"}, + {Name: "action required", Status: "completed", Conclusion: "action_required", HTMLURL: "https://ci/action-required"}, + }, + } + + delta := ciAutomationBuildDelta(feedback, ciAutomationCheckpoint{}) + if len(delta.FailedChecks) != 4 { + t.Fatalf("failed checks = %d, want 4: %+v", len(delta.FailedChecks), delta.FailedChecks) + } +} + func TestCIAutomationRenderSnapshotSanitizesUntrustedFields(t *testing.T) { delta := ciAutomationCheckpoint{ FailedChecks: []ciAutomationCheckSnapshot{{Name: "unit\n