fix: improve recovery failure diagnostics (#3869)
This commit is contained in:
@@ -263,6 +263,28 @@ func (m *Manager) persistExecutorRunning(ctx context.Context, execution *AgentEx
|
||||
_ = m.persistExecutorRunningResult(ctx, execution)
|
||||
}
|
||||
|
||||
// buildRunningForPersistence reads a tracked execution while the execution
|
||||
// store's read lock is held. Status transitions use that store lock, so taking
|
||||
// the same lock here prevents persistence from racing with an asynchronous
|
||||
// readiness failure. Callers that are persisting an execution before it is
|
||||
// tracked still use the mapper directly.
|
||||
func (m *Manager) buildRunningForPersistence(
|
||||
execution *AgentExecution,
|
||||
prior *models.ExecutorRunning,
|
||||
) *models.ExecutorRunning {
|
||||
if execution == nil || m.executionStore == nil {
|
||||
return buildRunningFromExecution(execution, prior)
|
||||
}
|
||||
|
||||
var running *models.ExecutorRunning
|
||||
if err := m.executionStore.WithRLock(execution.ID, func(tracked *AgentExecution) {
|
||||
running = buildRunningFromExecution(tracked, prior)
|
||||
}); err == nil {
|
||||
return running
|
||||
}
|
||||
return buildRunningFromExecution(execution, prior)
|
||||
}
|
||||
|
||||
func (m *Manager) persistExecutorRunningResult(ctx context.Context, execution *AgentExecution) error {
|
||||
if m.runningWriter == nil {
|
||||
// Permitted in tests that don't exercise persistence; logged so a
|
||||
@@ -305,7 +327,7 @@ func (m *Manager) persistExecutorRunningResult(ctx context.Context, execution *A
|
||||
execution.AgentProfileID = prior.ExecutionProfileID
|
||||
}
|
||||
|
||||
running := buildRunningFromExecution(execution, prior)
|
||||
running := m.buildRunningForPersistence(execution, prior)
|
||||
// Attach the host-local liveness handle for local/standalone rows. Kept out
|
||||
// of buildRunningFromExecution (a pure mapper) because the PID lives on the
|
||||
// manager, wired from the agentctl launcher at DI. resolveLocalPID returns 0
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"github.com/kandev/kandev/internal/agentruntime"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
v1 "github.com/kandev/kandev/pkg/api/v1"
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
type captureExecutorRunningWriter struct {
|
||||
@@ -172,6 +173,22 @@ func TestPersistExecutorRunningRestoresRecoveredExecutionProfile(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestPersistExecutorRunningUsesTrackedExecutionStatus(t *testing.T) {
|
||||
writer := &captureExecutorRunningWriter{}
|
||||
mgr := newTestManager(t)
|
||||
mgr.SetExecutorRunningWriter(writer)
|
||||
execution := &AgentExecution{
|
||||
ID: "exec-tracked-status", TaskID: "task-1", SessionID: "session-1",
|
||||
Status: v1.AgentStatusRunning,
|
||||
}
|
||||
require.NoError(t, mgr.executionStore.Add(execution))
|
||||
mgr.executionStore.UpdateError(execution.ID, "agentctl not ready")
|
||||
|
||||
require.NoError(t, mgr.persistExecutorRunningResult(context.Background(), execution))
|
||||
require.NotNil(t, writer.running)
|
||||
require.Equal(t, models.ExecutorRunningStatusFailed, writer.running.Status)
|
||||
}
|
||||
|
||||
func TestPersistExecutorRunningReturnsUpsertFailure(t *testing.T) {
|
||||
writer := &captureExecutorRunningWriter{upsertErr: errors.New("database is locked")}
|
||||
mgr := newTestManager(t)
|
||||
|
||||
@@ -63,7 +63,7 @@ func resolvePreparerSetupScript(req *EnvPrepareRequest, workspacePath string) (s
|
||||
|
||||
resolved := resolver.Resolve(script)
|
||||
if isScriptEffectivelyEmpty(resolved) {
|
||||
prepareScriptLogger.Warn("setup script is comment-only after resolution, skipping",
|
||||
prepareScriptLogger.Debug("setup script is comment-only after resolution, skipping",
|
||||
zap.String("task_id", req.TaskID),
|
||||
zap.String("executor_type", string(req.ExecutorType)),
|
||||
zap.Bool("use_worktree", req.UseWorktree),
|
||||
|
||||
@@ -1,13 +1,136 @@
|
||||
package lifecycle
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/kandev/kandev/internal/agent/executor"
|
||||
commonlogger "github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zapcore"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
)
|
||||
|
||||
// @covers AC-PLATFORM-DIAGNOSTIC-LOGGING-001.12
|
||||
func TestResolvePreparerSetupScriptDiagnosticSeverity(t *testing.T) {
|
||||
core, logs := observer.New(zapcore.DebugLevel)
|
||||
log, err := commonlogger.NewFromZap(zap.New(core))
|
||||
if err != nil {
|
||||
t.Fatalf("create observer logger: %v", err)
|
||||
}
|
||||
previousLogger := prepareScriptLogger
|
||||
prepareScriptLogger = log
|
||||
t.Cleanup(func() { prepareScriptLogger = previousLogger })
|
||||
|
||||
cases := []struct {
|
||||
name string
|
||||
request EnvPrepareRequest
|
||||
wantExplicit bool
|
||||
wantEmptyDiagnostic bool
|
||||
secret string
|
||||
}{
|
||||
{
|
||||
name: "default comment-only script",
|
||||
request: EnvPrepareRequest{
|
||||
TaskID: "task-default",
|
||||
ExecutorType: executor.NameStandalone,
|
||||
RepositoryPath: "/tmp/my-repo",
|
||||
},
|
||||
wantEmptyDiagnostic: true,
|
||||
},
|
||||
{
|
||||
name: "explicit comment-only script",
|
||||
request: EnvPrepareRequest{
|
||||
TaskID: "task-explicit",
|
||||
ExecutorType: executor.NameStandalone,
|
||||
RepositoryPath: "/tmp/my-repo",
|
||||
SetupScript: "# secret-comment\n# another comment",
|
||||
},
|
||||
wantExplicit: true,
|
||||
wantEmptyDiagnostic: true,
|
||||
secret: "secret-comment",
|
||||
},
|
||||
{
|
||||
name: "blank and shebang-only script",
|
||||
request: EnvPrepareRequest{
|
||||
TaskID: "task-shebang",
|
||||
ExecutorType: executor.NameStandalone,
|
||||
RepositoryPath: "/tmp/my-repo",
|
||||
SetupScript: "\n#!/bin/sh\n\n",
|
||||
},
|
||||
wantExplicit: true,
|
||||
wantEmptyDiagnostic: true,
|
||||
},
|
||||
{
|
||||
name: "executable script",
|
||||
request: EnvPrepareRequest{
|
||||
TaskID: "task-command",
|
||||
ExecutorType: executor.NameStandalone,
|
||||
RepositoryPath: "/tmp/my-repo",
|
||||
SetupScript: "echo setup-command",
|
||||
},
|
||||
wantExplicit: true,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range cases {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
before := logs.Len()
|
||||
got, err := resolvePreparerSetupScript(&tt.request, "/tmp/my-repo")
|
||||
if err != nil {
|
||||
t.Fatalf("resolvePreparerSetupScript() error = %v", err)
|
||||
}
|
||||
if tt.wantEmptyDiagnostic && got != "" {
|
||||
t.Fatalf("resolved script = %q, want empty", got)
|
||||
}
|
||||
if !tt.wantEmptyDiagnostic && got == "" {
|
||||
t.Fatal("resolved executable script is empty")
|
||||
}
|
||||
|
||||
entries := logs.All()[before:]
|
||||
assertSetupScriptDiagnostic(t, entries, tt.request, tt.wantExplicit, tt.wantEmptyDiagnostic, tt.secret)
|
||||
if !tt.wantEmptyDiagnostic && len(entries) != 0 {
|
||||
t.Fatalf("executable script emitted omission diagnostics: %v", entries)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
for _, entry := range logs.All() {
|
||||
if entry.Level == zap.WarnLevel {
|
||||
t.Fatalf("comment-only setup emitted warning: %#v", entry)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func assertSetupScriptDiagnostic(
|
||||
t *testing.T, entries []observer.LoggedEntry, req EnvPrepareRequest, wantExplicit, wantEmpty bool, secret string,
|
||||
) {
|
||||
t.Helper()
|
||||
if !wantEmpty {
|
||||
return
|
||||
}
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("diagnostic entries = %d, want 1", len(entries))
|
||||
}
|
||||
entry := entries[0]
|
||||
if entry.Message != "setup script is comment-only after resolution, skipping" {
|
||||
t.Fatalf("diagnostic message = %q", entry.Message)
|
||||
}
|
||||
if entry.Level != zap.DebugLevel {
|
||||
t.Fatalf("diagnostic level = %s, want debug", entry.Level)
|
||||
}
|
||||
fields := entry.ContextMap()
|
||||
if fields["task_id"] != req.TaskID || fields["executor_type"] != string(req.ExecutorType) ||
|
||||
fields["use_worktree"] != req.UseWorktree || fields["has_explicit_script"] != wantExplicit {
|
||||
t.Fatalf("diagnostic fields = %v", fields)
|
||||
}
|
||||
if secret != "" && (strings.Contains(entry.Message, secret) || strings.Contains(fmt.Sprint(fields), secret)) {
|
||||
t.Fatalf("diagnostic exposed script content: %#v", entry)
|
||||
}
|
||||
}
|
||||
|
||||
func TestResolvePreparerSetupScript_LocalFallbackCommentOnly(t *testing.T) {
|
||||
req := &EnvPrepareRequest{
|
||||
ExecutorType: executor.NameStandalone,
|
||||
|
||||
@@ -226,7 +226,7 @@ func TestTaskEventBroadcaster_NoDuplicateSubscriptions(t *testing.T) {
|
||||
//
|
||||
// Update this number when adding or removing event subscriptions in
|
||||
// RegisterTaskNotifications — it is intentionally exact.
|
||||
const wantSubscriptions = 76
|
||||
const wantSubscriptions = 77
|
||||
if got := len(b.subscriptions); got != wantSubscriptions {
|
||||
t.Errorf("RegisterTaskNotifications created %d subscriptions, want %d — "+
|
||||
"did an event get subscribed twice?", got, wantSubscriptions)
|
||||
|
||||
@@ -676,6 +676,13 @@ func classifyPRDiscoveryError(err error) PRDiscoveryHealthCategory {
|
||||
return PRDiscoveryHealthUnavailable
|
||||
}
|
||||
|
||||
// ClassifyPRDiscoveryError returns the bounded category used for PR discovery
|
||||
// diagnostics. It deliberately exposes the category contract without exposing
|
||||
// provider error text or payloads to callers.
|
||||
func ClassifyPRDiscoveryError(err error) PRDiscoveryHealthCategory {
|
||||
return classifyPRDiscoveryError(err)
|
||||
}
|
||||
|
||||
func containsRateLimitMarker(value string) bool {
|
||||
lower := strings.ToLower(value)
|
||||
for _, marker := range []string{"rate limit", "too many requests", "429", "abuse detection", "secondary limit"} {
|
||||
|
||||
@@ -781,6 +781,9 @@ func (s *Service) detectPushAndAssociatePRWithIdentity(
|
||||
if identity.owner == "" || identity.name == "" {
|
||||
return
|
||||
}
|
||||
if prDiscoveryContextCanceled(ctx) {
|
||||
return
|
||||
}
|
||||
|
||||
// Check if we already have a watch for this (session, repo, branch).
|
||||
// Multi-branch: keying by branch as well means a secondary branch's
|
||||
@@ -790,6 +793,9 @@ func (s *Service) detectPushAndAssociatePRWithIdentity(
|
||||
// If the watch has pr_number=0, it's still searching — do an immediate
|
||||
// search (faster than waiting for the 1-minute poller).
|
||||
existing, err := s.githubService.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, identity.repositoryID, branch)
|
||||
if prDiscoveryContextCanceled(ctx) {
|
||||
return
|
||||
}
|
||||
if err == nil && existing != nil {
|
||||
if existing.PRNumber > 0 {
|
||||
return // PR already found and being monitored
|
||||
@@ -798,45 +804,133 @@ func (s *Service) detectPushAndAssociatePRWithIdentity(
|
||||
return
|
||||
}
|
||||
|
||||
// Try to find a PR immediately, then retry after delays
|
||||
delays := []time.Duration{0, 30 * time.Second, 60 * time.Second}
|
||||
for _, delay := range delays {
|
||||
if delay > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-time.After(delay):
|
||||
}
|
||||
// Re-check if a watch was created in the meantime (e.g. by CreatePR callback)
|
||||
if ex, err := s.githubService.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, identity.repositoryID, branch); err == nil && ex != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
foundPR, findErr := s.githubService.FindPRByBranchForWorkspace(
|
||||
ctx, workspaceID, identity.owner, identity.name, branch,
|
||||
)
|
||||
if findErr != nil || foundPR == nil {
|
||||
s.logger.Debug("no PR found for branch (will retry)",
|
||||
zap.String("branch", branch),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("repository_name", repositoryName),
|
||||
zap.Duration("delay", delay))
|
||||
continue
|
||||
}
|
||||
s.logger.Info("PR found after push, associating with task",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("repository_name", repositoryName),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.String("branch", branch))
|
||||
s.associatePRFromPushScoped(ctx, workspaceID, sessionID, taskID, identity.owner, identity.name, identity.repositoryID, branch, foundPR)
|
||||
foundPR := s.discoverPRAfterPush(ctx, workspaceID, sessionID, taskID, identity, branch)
|
||||
if foundPR == nil {
|
||||
return
|
||||
}
|
||||
s.logger.Warn("exhausted all retries, no PR found after push",
|
||||
s.logger.Info("PR found after push, associating with task",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("repository_name", repositoryName),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.String("branch", branch))
|
||||
s.associatePRFromPushScoped(ctx, workspaceID, sessionID, taskID, identity.owner, identity.name, identity.repositoryID, branch, foundPR)
|
||||
}
|
||||
|
||||
func (s *Service) discoverPRAfterPush(
|
||||
ctx context.Context,
|
||||
workspaceID, sessionID, taskID string,
|
||||
identity pushRepositoryIdentity,
|
||||
branch string,
|
||||
) *github.PR {
|
||||
const prDiscoveryOutcomeFailed = "failed"
|
||||
delays := []time.Duration{0, 30 * time.Second, 60 * time.Second}
|
||||
attemptCount := 0
|
||||
errorCount := 0
|
||||
emptyCount := 0
|
||||
lastOutcome := ""
|
||||
for attempt, delay := range delays {
|
||||
if delay > 0 {
|
||||
if !s.waitForPRDiscoveryRetry(ctx, delay) {
|
||||
return nil
|
||||
}
|
||||
// Re-check if a watch was created in the meantime (e.g. by CreatePR callback).
|
||||
if ex, err := s.githubService.GetPRWatchBySessionRepoAndBranch(ctx, sessionID, identity.repositoryID, branch); prDiscoveryContextCanceled(ctx) {
|
||||
return nil
|
||||
} else if err == nil && ex != nil {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
if prDiscoveryContextCanceled(ctx) {
|
||||
return nil
|
||||
}
|
||||
attemptCount++
|
||||
foundPR, findErr := s.githubService.FindPRByBranchForWorkspace(
|
||||
ctx, workspaceID, identity.owner, identity.name, branch,
|
||||
)
|
||||
if prDiscoveryContextCanceled(ctx) {
|
||||
return nil
|
||||
}
|
||||
if findErr != nil {
|
||||
errorCount++
|
||||
lastOutcome = prDiscoveryOutcomeFailed
|
||||
s.logger.Warn("PR discovery lookup failed after push",
|
||||
zap.String("operation", "post_push_branch_lookup"),
|
||||
zap.String("workspace_id", workspaceID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("repository_id", identity.repositoryID),
|
||||
zap.String("owner", identity.owner),
|
||||
zap.String("repo", identity.name),
|
||||
zap.String("branch", branch),
|
||||
zap.String("category", string(github.ClassifyPRDiscoveryError(findErr))),
|
||||
zap.Int("attempt", attempt+1))
|
||||
continue
|
||||
}
|
||||
if foundPR == nil {
|
||||
emptyCount++
|
||||
lastOutcome = "empty"
|
||||
s.logger.Debug("no PR found for branch (will retry)",
|
||||
zap.String("operation", "post_push_branch_lookup"),
|
||||
zap.String("workspace_id", workspaceID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("repository_id", identity.repositoryID),
|
||||
zap.String("owner", identity.owner),
|
||||
zap.String("repo", identity.name),
|
||||
zap.String("branch", branch),
|
||||
zap.Int("attempt", attempt+1),
|
||||
zap.Duration("delay", delay))
|
||||
continue
|
||||
}
|
||||
return foundPR
|
||||
}
|
||||
s.logPRDiscoveryExhausted(workspaceID, sessionID, taskID, identity, branch,
|
||||
attemptCount, errorCount, emptyCount, lastOutcome)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) logPRDiscoveryExhausted(
|
||||
workspaceID, sessionID, taskID string,
|
||||
identity pushRepositoryIdentity,
|
||||
branch string,
|
||||
attemptCount, errorCount, emptyCount int,
|
||||
lastOutcome string,
|
||||
) {
|
||||
s.logger.Debug("exhausted all retries, no PR found after push",
|
||||
zap.String("operation", "post_push_branch_lookup"),
|
||||
zap.String("workspace_id", workspaceID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("repository_id", identity.repositoryID),
|
||||
zap.String("owner", identity.owner),
|
||||
zap.String("repo", identity.name),
|
||||
zap.String("branch", branch),
|
||||
zap.Int("attempt_count", attemptCount),
|
||||
zap.Int("error_count", errorCount),
|
||||
zap.Int("empty_count", emptyCount),
|
||||
zap.String("last_outcome", lastOutcome))
|
||||
}
|
||||
|
||||
func (s *Service) waitForPRDiscoveryRetry(ctx context.Context, delay time.Duration) bool {
|
||||
if s.prDiscoveryWait != nil {
|
||||
return s.prDiscoveryWait(ctx, delay)
|
||||
}
|
||||
timer := time.NewTimer(delay)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-timer.C:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
func prDiscoveryContextCanceled(ctx context.Context) bool {
|
||||
// Provider clients can use their own timeout context and return a wrapped
|
||||
// context error while the caller remains active. Only the caller context
|
||||
// decides whether discovery should stop quietly.
|
||||
return ctx.Err() != nil
|
||||
}
|
||||
|
||||
// resolvePushRepo returns (owner, name, repository_id) for the per-repo push
|
||||
@@ -946,43 +1040,69 @@ func (s *Service) searchPRForExistingWatch(
|
||||
foundPR, findErr := s.githubService.FindPRByBranchForWorkspace(
|
||||
ctx, workspaceID, watch.Owner, watch.Repo, branch,
|
||||
)
|
||||
if findErr == nil && foundPR != nil {
|
||||
if foundPR.RepoOwner != "" && foundPR.RepoName != "" &&
|
||||
(!strings.EqualFold(watch.Owner, foundPR.RepoOwner) || !strings.EqualFold(watch.Repo, foundPR.RepoName)) {
|
||||
if err := s.githubService.UpdatePRWatchRepository(ctx, watch.ID, foundPR.RepoOwner, foundPR.RepoName); err != nil {
|
||||
s.logger.Warn("failed to rebind PR watch repository",
|
||||
zap.String("watch_id", watch.ID),
|
||||
zap.String("owner", foundPR.RepoOwner),
|
||||
zap.String("repo", foundPR.RepoName),
|
||||
zap.Error(err))
|
||||
return
|
||||
}
|
||||
watch.Owner = foundPR.RepoOwner
|
||||
watch.Repo = foundPR.RepoName
|
||||
}
|
||||
if err := s.githubService.UpdatePRWatchPRNumber(ctx, watch.ID, foundPR.Number); err != nil {
|
||||
s.logger.Warn("failed to update PR watch number",
|
||||
zap.String("watch_id", watch.ID),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.Error(err))
|
||||
}
|
||||
// Use the watch's own repository_id so the association lands on the
|
||||
// correct per-repo TaskPR row (matters once multi-repo watches exist;
|
||||
// for legacy single-repo watches this is empty and matches the old
|
||||
// "delete all" behavior).
|
||||
if _, err := s.githubService.AssociatePRWithTaskForWorkspace(
|
||||
ctx, workspaceID, taskID, watch.RepositoryID, foundPR,
|
||||
); err != nil {
|
||||
s.logger.Error("failed to associate PR with task",
|
||||
zap.String("task_id", taskID),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.Error(err))
|
||||
}
|
||||
s.logger.Info("auto-detected PR from push (existing watch)",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.String("branch", branch))
|
||||
if prDiscoveryContextCanceled(ctx) {
|
||||
return
|
||||
}
|
||||
if findErr != nil {
|
||||
s.logger.Warn("PR discovery lookup failed for existing watch",
|
||||
zap.String("operation", "existing_watch_branch_lookup"),
|
||||
zap.String("workspace_id", workspaceID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("repository_id", watch.RepositoryID),
|
||||
zap.String("owner", watch.Owner),
|
||||
zap.String("repo", watch.Repo),
|
||||
zap.String("branch", branch),
|
||||
zap.String("category", string(github.ClassifyPRDiscoveryError(findErr))))
|
||||
return
|
||||
}
|
||||
if foundPR == nil {
|
||||
s.logger.Debug("no PR found for existing watch",
|
||||
zap.String("operation", "existing_watch_branch_lookup"),
|
||||
zap.String("workspace_id", workspaceID),
|
||||
zap.String("session_id", sessionID),
|
||||
zap.String("task_id", taskID),
|
||||
zap.String("repository_id", watch.RepositoryID),
|
||||
zap.String("owner", watch.Owner),
|
||||
zap.String("repo", watch.Repo),
|
||||
zap.String("branch", branch))
|
||||
return
|
||||
}
|
||||
if foundPR.RepoOwner != "" && foundPR.RepoName != "" &&
|
||||
(!strings.EqualFold(watch.Owner, foundPR.RepoOwner) || !strings.EqualFold(watch.Repo, foundPR.RepoName)) {
|
||||
if err := s.githubService.UpdatePRWatchRepository(ctx, watch.ID, foundPR.RepoOwner, foundPR.RepoName); err != nil {
|
||||
s.logger.Warn("failed to rebind PR watch repository",
|
||||
zap.String("watch_id", watch.ID),
|
||||
zap.String("owner", foundPR.RepoOwner),
|
||||
zap.String("repo", foundPR.RepoName),
|
||||
zap.Error(err))
|
||||
return
|
||||
}
|
||||
watch.Owner = foundPR.RepoOwner
|
||||
watch.Repo = foundPR.RepoName
|
||||
}
|
||||
if err := s.githubService.UpdatePRWatchPRNumber(ctx, watch.ID, foundPR.Number); err != nil {
|
||||
s.logger.Warn("failed to update PR watch number",
|
||||
zap.String("watch_id", watch.ID),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.Error(err))
|
||||
}
|
||||
// Use the watch's own repository_id so the association lands on the
|
||||
// correct per-repo TaskPR row (matters once multi-repo watches exist;
|
||||
// for legacy single-repo watches this is empty and matches the old
|
||||
// "delete all" behavior).
|
||||
if _, err := s.githubService.AssociatePRWithTaskForWorkspace(
|
||||
ctx, workspaceID, taskID, watch.RepositoryID, foundPR,
|
||||
); err != nil {
|
||||
s.logger.Error("failed to associate PR with task",
|
||||
zap.String("task_id", taskID),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.Error(err))
|
||||
}
|
||||
s.logger.Info("auto-detected PR from push (existing watch)",
|
||||
zap.String("session_id", sessionID),
|
||||
zap.Int("pr_number", foundPR.Number),
|
||||
zap.String("branch", branch))
|
||||
}
|
||||
|
||||
// resolveSessionRepo looks up the repository owner and name for a session.
|
||||
|
||||
@@ -0,0 +1,382 @@
|
||||
package orchestrator
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
commonlogger "github.com/kandev/kandev/internal/common/logger"
|
||||
"github.com/kandev/kandev/internal/github"
|
||||
"github.com/kandev/kandev/internal/task/models"
|
||||
"go.uber.org/zap"
|
||||
"go.uber.org/zap/zapcore"
|
||||
"go.uber.org/zap/zaptest/observer"
|
||||
)
|
||||
|
||||
type pushDiscoveryLookup struct {
|
||||
pr *github.PR
|
||||
err error
|
||||
}
|
||||
|
||||
type pushDiscoveryCall struct {
|
||||
WorkspaceID string
|
||||
Owner string
|
||||
Repo string
|
||||
Branch string
|
||||
}
|
||||
|
||||
type pushDiagnosticsGitHubService struct {
|
||||
*mockGitHubService
|
||||
lookups []pushDiscoveryLookup
|
||||
calls []pushDiscoveryCall
|
||||
}
|
||||
|
||||
func (s *pushDiagnosticsGitHubService) FindPRByBranchForWorkspace(
|
||||
_ context.Context, workspaceID, owner, repo, branch string,
|
||||
) (*github.PR, error) {
|
||||
s.calls = append(s.calls, pushDiscoveryCall{
|
||||
WorkspaceID: workspaceID,
|
||||
Owner: owner,
|
||||
Repo: repo,
|
||||
Branch: branch,
|
||||
})
|
||||
if len(s.lookups) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
lookup := s.lookups[0]
|
||||
s.lookups = s.lookups[1:]
|
||||
return lookup.pr, lookup.err
|
||||
}
|
||||
|
||||
func newPushDiagnosticsService(
|
||||
t *testing.T, repositoryID, repositoryName, owner, repoName string, watch *github.PRWatch,
|
||||
) (*Service, *pushDiagnosticsGitHubService, *observer.ObservedLogs) {
|
||||
t.Helper()
|
||||
ctx := context.Background()
|
||||
repository := setupTestRepo(t)
|
||||
seedSession(t, repository, "t1", "s1", "step1")
|
||||
now := time.Now().UTC()
|
||||
if err := repository.CreateRepository(ctx, &models.Repository{
|
||||
ID: repositoryID,
|
||||
WorkspaceID: "ws1",
|
||||
Name: repositoryName,
|
||||
SourceType: "provider",
|
||||
Provider: "github",
|
||||
ProviderOwner: owner,
|
||||
ProviderName: repoName,
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create repository: %v", err)
|
||||
}
|
||||
if err := repository.CreateTaskRepository(ctx, &models.TaskRepository{
|
||||
ID: "task-repository-1",
|
||||
TaskID: "t1",
|
||||
RepositoryID: repositoryID,
|
||||
Position: 0,
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}); err != nil {
|
||||
t.Fatalf("create task repository: %v", err)
|
||||
}
|
||||
session, err := repository.GetTaskSession(ctx, "s1")
|
||||
if err != nil {
|
||||
t.Fatalf("get session: %v", err)
|
||||
}
|
||||
session.RepositoryID = repositoryID
|
||||
if err := repository.UpdateTaskSession(ctx, session); err != nil {
|
||||
t.Fatalf("update session: %v", err)
|
||||
}
|
||||
|
||||
log, logs := newPushDiagnosticsLogger(t)
|
||||
svc := createTestService(repository, newMockStepGetter(), newMockTaskRepo())
|
||||
svc.logger = log
|
||||
gh := &pushDiagnosticsGitHubService{
|
||||
mockGitHubService: &mockGitHubService{prWatch: watch},
|
||||
}
|
||||
svc.SetGitHubService(gh)
|
||||
return svc, gh, logs
|
||||
}
|
||||
|
||||
func newPushDiagnosticsLogger(t *testing.T) (*commonlogger.Logger, *observer.ObservedLogs) {
|
||||
t.Helper()
|
||||
core, logs := observer.New(zapcore.DebugLevel)
|
||||
log, err := commonlogger.NewFromZap(zap.New(core))
|
||||
if err != nil {
|
||||
t.Fatalf("create observer logger: %v", err)
|
||||
}
|
||||
return log, logs
|
||||
}
|
||||
|
||||
func installPushDiagnosticsWait(svc *Service, waits *[]time.Duration) {
|
||||
svc.prDiscoveryWait = func(ctx context.Context, delay time.Duration) bool {
|
||||
*waits = append(*waits, delay)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
default:
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func assertPushDiagnosticLogsContainNoRawError(t *testing.T, logs *observer.ObservedLogs, raw string) {
|
||||
t.Helper()
|
||||
for _, entry := range logs.All() {
|
||||
if strings.Contains(entry.Message, raw) || strings.Contains(fmt.Sprint(entry.ContextMap()), raw) {
|
||||
t.Fatalf("log exposed raw provider error %q: %#v", raw, entry)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestPushDiscoveryDiagnostics(t *testing.T) {
|
||||
providerErr := errors.New("GET https://token:push-secret@example.invalid/repos/myorg/myrepo: 503 service unavailable")
|
||||
providerDeadlineErr := fmt.Errorf("provider request: %w", context.DeadlineExceeded)
|
||||
foundPR := &github.PR{
|
||||
Number: 17,
|
||||
RepoOwner: "myorg",
|
||||
RepoName: "myrepo",
|
||||
HeadBranch: "feature-branch",
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
lookups []pushDiscoveryLookup
|
||||
wantCalls int
|
||||
wantWaits []time.Duration
|
||||
wantWarns int
|
||||
wantSummary bool
|
||||
wantLastState string
|
||||
wantAssoc int
|
||||
}{
|
||||
{
|
||||
name: "all empty",
|
||||
lookups: []pushDiscoveryLookup{{}, {}, {}},
|
||||
wantCalls: 3,
|
||||
wantWaits: []time.Duration{30 * time.Second, 60 * time.Second},
|
||||
wantSummary: true,
|
||||
wantLastState: "empty",
|
||||
},
|
||||
{
|
||||
name: "all errors",
|
||||
lookups: []pushDiscoveryLookup{{err: providerErr}, {err: providerErr}, {err: providerErr}},
|
||||
wantCalls: 3,
|
||||
wantWaits: []time.Duration{30 * time.Second, 60 * time.Second},
|
||||
wantWarns: 3,
|
||||
wantSummary: true,
|
||||
wantLastState: "failed",
|
||||
},
|
||||
{
|
||||
name: "wrapped provider deadline remains retryable",
|
||||
lookups: []pushDiscoveryLookup{{err: providerDeadlineErr}, {err: providerDeadlineErr}, {err: providerDeadlineErr}},
|
||||
wantCalls: 3,
|
||||
wantWaits: []time.Duration{30 * time.Second, 60 * time.Second},
|
||||
wantWarns: 3,
|
||||
wantSummary: true,
|
||||
wantLastState: "failed",
|
||||
},
|
||||
{
|
||||
name: "error then empty",
|
||||
lookups: []pushDiscoveryLookup{{err: providerErr}, {}, {}},
|
||||
wantCalls: 3,
|
||||
wantWaits: []time.Duration{30 * time.Second, 60 * time.Second},
|
||||
wantWarns: 1,
|
||||
wantSummary: true,
|
||||
wantLastState: "empty",
|
||||
},
|
||||
{
|
||||
name: "empty then error",
|
||||
lookups: []pushDiscoveryLookup{{}, {err: providerErr}, {}},
|
||||
wantCalls: 3,
|
||||
wantWaits: []time.Duration{30 * time.Second, 60 * time.Second},
|
||||
wantWarns: 1,
|
||||
wantSummary: true,
|
||||
wantLastState: "empty",
|
||||
},
|
||||
{
|
||||
name: "error then found",
|
||||
lookups: []pushDiscoveryLookup{{err: providerErr}, {pr: foundPR}},
|
||||
wantCalls: 2,
|
||||
wantWaits: []time.Duration{30 * time.Second},
|
||||
wantWarns: 1,
|
||||
wantAssoc: 1,
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
svc, gh, logs := newPushDiagnosticsService(t, "repo1", "myrepo", "myorg", "myrepo", nil)
|
||||
gh.lookups = append([]pushDiscoveryLookup(nil), tt.lookups...)
|
||||
var waits []time.Duration
|
||||
installPushDiagnosticsWait(svc, &waits)
|
||||
|
||||
svc.detectPushAndAssociatePR(context.Background(), "s1", "t1", "", "feature-branch")
|
||||
|
||||
if len(gh.calls) != tt.wantCalls {
|
||||
t.Fatalf("lookup calls = %d, want %d", len(gh.calls), tt.wantCalls)
|
||||
}
|
||||
if len(waits) != len(tt.wantWaits) {
|
||||
t.Fatalf("waits = %v, want %v", waits, tt.wantWaits)
|
||||
}
|
||||
for i := range waits {
|
||||
if waits[i] != tt.wantWaits[i] {
|
||||
t.Fatalf("waits = %v, want %v", waits, tt.wantWaits)
|
||||
}
|
||||
}
|
||||
if gh.associateCalls != tt.wantAssoc {
|
||||
t.Fatalf("association calls = %d, want %d", gh.associateCalls, tt.wantAssoc)
|
||||
}
|
||||
failed := logs.FilterMessage("PR discovery lookup failed after push").All()
|
||||
if len(failed) != tt.wantWarns {
|
||||
t.Fatalf("failed lookup warnings = %d, want %d; logs = %v", len(failed), tt.wantWarns, logs.All())
|
||||
}
|
||||
for _, entry := range failed {
|
||||
fields := entry.ContextMap()
|
||||
if fields["workspace_id"] != "ws1" || fields["task_id"] != "t1" ||
|
||||
fields["repository_id"] != "repo1" || fields["owner"] != "myorg" ||
|
||||
fields["repo"] != "myrepo" || fields["branch"] != "feature-branch" ||
|
||||
fields["category"] != string(github.PRDiscoveryHealthUnavailable) {
|
||||
t.Fatalf("failed lookup fields = %v", fields)
|
||||
}
|
||||
}
|
||||
summary := logs.FilterMessage("exhausted all retries, no PR found after push").All()
|
||||
if len(summary) != boolToInt(tt.wantSummary) {
|
||||
t.Fatalf("retry summaries = %d, want %d; logs = %v", len(summary), boolToInt(tt.wantSummary), logs.All())
|
||||
}
|
||||
if tt.wantSummary {
|
||||
if summary[0].Level != zap.DebugLevel {
|
||||
t.Fatalf("retry summary level = %s, want debug", summary[0].Level)
|
||||
}
|
||||
fields := summary[0].ContextMap()
|
||||
if fields["attempt_count"] != int64(3) || fields["error_count"] != int64(countLookupErrors(tt.lookups)) ||
|
||||
fields["empty_count"] != int64(3-countLookupErrors(tt.lookups)) || fields["last_outcome"] != tt.wantLastState {
|
||||
t.Fatalf("retry summary fields = %v", fields)
|
||||
}
|
||||
}
|
||||
assertPushDiagnosticLogsContainNoRawError(t, logs, "push-secret")
|
||||
})
|
||||
}
|
||||
|
||||
t.Run("cancellation before lookup", func(t *testing.T) {
|
||||
svc, gh, logs := newPushDiagnosticsService(t, "repo1", "myrepo", "myorg", "myrepo", nil)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
svc.detectPushAndAssociatePR(ctx, "s1", "t1", "", "feature-branch")
|
||||
if len(gh.calls) != 0 {
|
||||
t.Fatalf("lookup calls = %d, want 0", len(gh.calls))
|
||||
}
|
||||
if logs.FilterMessage("PR discovery lookup failed after push").Len() != 0 ||
|
||||
logs.FilterMessage("exhausted all retries, no PR found after push").Len() != 0 {
|
||||
t.Fatalf("cancellation emitted failure logs: %v", logs.All())
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("cancellation during delay", func(t *testing.T) {
|
||||
svc, gh, logs := newPushDiagnosticsService(t, "repo1", "myrepo", "myorg", "myrepo", nil)
|
||||
gh.lookups = []pushDiscoveryLookup{{err: providerErr}}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
var waits []time.Duration
|
||||
svc.prDiscoveryWait = func(_ context.Context, delay time.Duration) bool {
|
||||
waits = append(waits, delay)
|
||||
cancel()
|
||||
return false
|
||||
}
|
||||
svc.detectPushAndAssociatePR(ctx, "s1", "t1", "", "feature-branch")
|
||||
if len(gh.calls) != 1 || len(waits) != 1 || waits[0] != 30*time.Second {
|
||||
t.Fatalf("calls/waits = %d/%v, want 1/[30s]", len(gh.calls), waits)
|
||||
}
|
||||
if logs.FilterMessage("exhausted all retries, no PR found after push").Len() != 0 {
|
||||
t.Fatalf("cancellation emitted exhaustion warning: %v", logs.All())
|
||||
}
|
||||
assertPushDiagnosticLogsContainNoRawError(t, logs, "push-secret")
|
||||
})
|
||||
}
|
||||
|
||||
func TestExistingWatchDiscoveryDiagnostics(t *testing.T) {
|
||||
providerErr := errors.New("GET https://token:watch-secret@example.invalid/repos/acme/secondary: 503 service unavailable")
|
||||
providerDeadlineErr := fmt.Errorf("provider request: %w", context.DeadlineExceeded)
|
||||
foundPR := &github.PR{
|
||||
Number: 23,
|
||||
RepoOwner: "acme",
|
||||
RepoName: "secondary",
|
||||
HeadBranch: "feature-branch",
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
lookup pushDiscoveryLookup
|
||||
wantWarn bool
|
||||
wantAssoc int
|
||||
}{
|
||||
{name: "provider error", lookup: pushDiscoveryLookup{err: providerErr}, wantWarn: true},
|
||||
{name: "wrapped provider deadline", lookup: pushDiscoveryLookup{err: providerDeadlineErr}, wantWarn: true},
|
||||
{name: "empty result", lookup: pushDiscoveryLookup{}, wantAssoc: 0},
|
||||
{name: "found result", lookup: pushDiscoveryLookup{pr: foundPR}, wantAssoc: 1},
|
||||
{name: "provider cancellation", lookup: pushDiscoveryLookup{err: context.Canceled}, wantWarn: true, wantAssoc: 0},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
watch := &github.PRWatch{
|
||||
ID: "watch-secondary",
|
||||
SessionID: "s1",
|
||||
TaskID: "t1",
|
||||
RepositoryID: "repo-secondary",
|
||||
Owner: "acme",
|
||||
Repo: "secondary",
|
||||
PRNumber: 0,
|
||||
Branch: "feature-branch",
|
||||
}
|
||||
svc, gh, logs := newPushDiagnosticsService(t, "repo-secondary", "secondary", "acme", "secondary", watch)
|
||||
gh.lookups = []pushDiscoveryLookup{tt.lookup}
|
||||
|
||||
svc.detectPushAndAssociatePR(context.Background(), "s1", "t1", "", "feature-branch")
|
||||
|
||||
if len(gh.calls) != 1 || gh.calls[0].WorkspaceID != "ws1" || gh.calls[0].Owner != "acme" ||
|
||||
gh.calls[0].Repo != "secondary" || gh.calls[0].Branch != "feature-branch" {
|
||||
t.Fatalf("lookup identity = %#v, want ws1/acme/secondary/feature-branch", gh.calls)
|
||||
}
|
||||
if gh.associateCalls != tt.wantAssoc {
|
||||
t.Fatalf("association calls = %d, want %d", gh.associateCalls, tt.wantAssoc)
|
||||
}
|
||||
failed := logs.FilterMessage("PR discovery lookup failed for existing watch").All()
|
||||
if len(failed) != boolToInt(tt.wantWarn) {
|
||||
t.Fatalf("existing-watch failure warnings = %d, want %d; logs = %v", len(failed), boolToInt(tt.wantWarn), logs.All())
|
||||
}
|
||||
if tt.wantWarn {
|
||||
fields := failed[0].ContextMap()
|
||||
if fields["workspace_id"] != "ws1" || fields["task_id"] != "t1" ||
|
||||
fields["repository_id"] != "repo-secondary" || fields["owner"] != "acme" ||
|
||||
fields["repo"] != "secondary" || fields["category"] != string(github.PRDiscoveryHealthUnavailable) {
|
||||
t.Fatalf("existing-watch failure fields = %v", fields)
|
||||
}
|
||||
}
|
||||
if logs.FilterMessage("no PR found for existing watch").Len() > 0 && tt.lookup.err != nil {
|
||||
t.Fatalf("failed existing-watch lookup was logged as empty: %v", logs.All())
|
||||
}
|
||||
if tt.lookup.err == nil && tt.lookup.pr == nil && logs.FilterMessage("no PR found for existing watch").Len() != 1 {
|
||||
t.Fatalf("empty existing-watch result did not emit debug log: %v", logs.All())
|
||||
}
|
||||
assertPushDiagnosticLogsContainNoRawError(t, logs, "watch-secret")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func countLookupErrors(lookups []pushDiscoveryLookup) int {
|
||||
count := 0
|
||||
for _, lookup := range lookups {
|
||||
if lookup.err != nil {
|
||||
count++
|
||||
}
|
||||
}
|
||||
return count
|
||||
}
|
||||
|
||||
func boolToInt(value bool) int {
|
||||
if value {
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
@@ -1018,6 +1018,9 @@ type Service struct {
|
||||
|
||||
// GitHub service for PR auto-detection on push
|
||||
githubService GitHubService
|
||||
// prDiscoveryWait is nil in production and overridable by package tests so
|
||||
// retry diagnostics can be exercised without real-time delays.
|
||||
prDiscoveryWait func(context.Context, time.Duration) bool
|
||||
// ciAutomationInFlight serializes each PR's evaluation and coalesces one
|
||||
// follow-up request instead of dropping an event that arrives mid-run.
|
||||
ciAutomationInFlight ciAutomationCoordinator
|
||||
|
||||
@@ -531,6 +531,7 @@ func (m *Manager) handleBaseFetchFailure(
|
||||
onProgress SyncProgressCallback,
|
||||
) (string, string, error) {
|
||||
reason := classifyGitFallbackReason(err, string(output), execCtxErr)
|
||||
m.logRefreshDiagnostic("fetch", repoPath, baseBranch, reason, output, err, execCtxErr)
|
||||
if required {
|
||||
fallback := strings.TrimSpace(fallbackBaseBranch)
|
||||
if reason == gitFallbackReasonMissingRemoteRef && fallback != "" && fallback != baseBranch {
|
||||
@@ -701,6 +702,7 @@ func (m *Manager) pullCurrentBranchOrFallback(
|
||||
output, err, execCtxErr := m.runGitCombinedAfterAcquire(ctx, m.pullTimeout, repoPath, "pull", "--ff-only", "origin", baseBranch)
|
||||
if err != nil {
|
||||
reason := classifyGitFallbackReason(err, string(output), execCtxErr)
|
||||
m.logRefreshDiagnostic("pull", repoPath, baseBranch, reason, output, err, execCtxErr)
|
||||
resolved, selectErr := m.selectContainingRef(ctx, repoPath, baseBranch, remoteRef)
|
||||
if selectErr != nil {
|
||||
if !localBaseExists {
|
||||
@@ -770,9 +772,8 @@ func (m *Manager) selectContainingRef(
|
||||
}
|
||||
|
||||
// syncFailureCause intentionally suppresses cmdErr because Git output can
|
||||
// contain credentials. Callers expose only a bounded failure class and keep
|
||||
// raw command output in internal logs where the existing redaction policy
|
||||
// applies.
|
||||
// contain credentials. Callers expose only a bounded failure class and never
|
||||
// include the command output in errors or progress.
|
||||
func syncFailureCause(reason string, _ error, contextErr error) error {
|
||||
if contextErr != nil {
|
||||
return contextErr
|
||||
|
||||
@@ -0,0 +1,127 @@
|
||||
package worktree
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"os/exec"
|
||||
"strings"
|
||||
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
const (
|
||||
refreshDiagnosticAuthentication = "authentication_failed"
|
||||
refreshDiagnosticSSHKey = "ssh_public_key_rejected"
|
||||
refreshDiagnosticSSHHost = "ssh_host_verification_failed"
|
||||
refreshDiagnosticDNS = "dns_resolution_failed"
|
||||
refreshDiagnosticConnection = "connection_failed"
|
||||
refreshDiagnosticTLS = "tls_verification_failed"
|
||||
refreshDiagnosticMissingRef = "missing_remote_ref"
|
||||
refreshDiagnosticNonFastForward = "non_fast_forward"
|
||||
refreshDiagnosticRepository = "repository_access_failed"
|
||||
refreshDiagnosticLock = "lock_contention"
|
||||
refreshDiagnosticTimeout = "timeout"
|
||||
refreshDiagnosticCanceled = "canceled"
|
||||
refreshDiagnosticUnknown = "unknown"
|
||||
)
|
||||
|
||||
type refreshDiagnostic struct {
|
||||
code string
|
||||
detail string
|
||||
exitCode int
|
||||
hasExitCode bool
|
||||
}
|
||||
|
||||
func classifyRefreshDiagnostic(output string, runErr, contextErr error) refreshDiagnostic {
|
||||
if errors.Is(contextErr, context.Canceled) || errors.Is(runErr, context.Canceled) {
|
||||
return newRefreshDiagnostic(refreshDiagnosticCanceled, "The Git refresh was canceled.", runErr)
|
||||
}
|
||||
if errors.Is(contextErr, context.DeadlineExceeded) || errors.Is(runErr, context.DeadlineExceeded) {
|
||||
return newRefreshDiagnostic(refreshDiagnosticTimeout, "The Git refresh timed out.", runErr)
|
||||
}
|
||||
|
||||
code, detail := classifyRefreshDiagnosticOutput(output)
|
||||
return newRefreshDiagnostic(code, detail, runErr)
|
||||
}
|
||||
|
||||
func classifyRefreshDiagnosticOutput(output string) (string, string) {
|
||||
lower := strings.ToLower(output)
|
||||
switch {
|
||||
case strings.Contains(lower, "permission denied (publickey)"),
|
||||
strings.Contains(lower, "no supported authentication methods available"):
|
||||
return refreshDiagnosticSSHKey, "The SSH server rejected the configured public key."
|
||||
case strings.Contains(lower, "host key verification failed"),
|
||||
strings.Contains(lower, "remote host identification has changed"):
|
||||
return refreshDiagnosticSSHHost, "The SSH host could not be verified."
|
||||
case strings.Contains(lower, "could not resolve host"),
|
||||
strings.Contains(lower, "could not resolve hostname"),
|
||||
strings.Contains(lower, "name or service not known"),
|
||||
strings.Contains(lower, "temporary failure in name resolution"):
|
||||
return refreshDiagnosticDNS, "The remote host name could not be resolved."
|
||||
case strings.Contains(lower, "ssl certificate problem"),
|
||||
strings.Contains(lower, "certificate verify failed"),
|
||||
strings.Contains(lower, "server certificate verification failed"):
|
||||
return refreshDiagnosticTLS, "The remote TLS certificate could not be verified."
|
||||
case strings.Contains(lower, "connection refused"),
|
||||
strings.Contains(lower, "connection reset"),
|
||||
strings.Contains(lower, "network is unreachable"),
|
||||
strings.Contains(lower, "no route to host"),
|
||||
strings.Contains(lower, "failed to connect"),
|
||||
strings.Contains(lower, "couldn't connect"):
|
||||
return refreshDiagnosticConnection, "The connection to the Git remote failed."
|
||||
case strings.Contains(lower, "couldn't find remote ref"),
|
||||
strings.Contains(lower, "could not find remote ref"),
|
||||
strings.Contains(lower, "remote ref does not exist"),
|
||||
strings.Contains(lower, "no such ref"):
|
||||
return refreshDiagnosticMissingRef, "The requested remote branch was not found."
|
||||
case strings.Contains(lower, "not possible to fast-forward"):
|
||||
return refreshDiagnosticNonFastForward, "The local branch could not be fast-forwarded."
|
||||
case strings.Contains(lower, "non-fast-forward"):
|
||||
return refreshDiagnosticNonFastForward, "The Git refresh encountered a non-fast-forward update."
|
||||
case strings.Contains(lower, "index.lock"),
|
||||
strings.Contains(lower, "cannot lock ref"),
|
||||
strings.Contains(lower, "another git process seems to be running"),
|
||||
// Require both fragments so a generic "unable to create" error stays unknown.
|
||||
strings.Contains(lower, "unable to create") && strings.Contains(lower, "lock"):
|
||||
return refreshDiagnosticLock, "Git could not acquire the repository lock."
|
||||
case strings.Contains(lower, "repository not found"),
|
||||
strings.Contains(lower, "does not appear to be a git repository"),
|
||||
strings.Contains(lower, "could not read from remote repository"):
|
||||
return refreshDiagnosticRepository, "The Git repository could not be accessed."
|
||||
case strings.Contains(lower, "could not read username"),
|
||||
strings.Contains(lower, "authentication failed"),
|
||||
strings.Contains(lower, "terminal prompts disabled"),
|
||||
strings.Contains(lower, "authentication required"):
|
||||
return refreshDiagnosticAuthentication, "Git could not authenticate to the remote."
|
||||
default:
|
||||
return refreshDiagnosticUnknown, "The Git refresh failed for an unknown reason."
|
||||
}
|
||||
}
|
||||
|
||||
func newRefreshDiagnostic(code, detail string, runErr error) refreshDiagnostic {
|
||||
diagnostic := refreshDiagnostic{code: code, detail: detail}
|
||||
var exitErr *exec.ExitError
|
||||
if errors.As(runErr, &exitErr) && exitErr.ExitCode() >= 0 {
|
||||
diagnostic.exitCode = exitErr.ExitCode()
|
||||
diagnostic.hasExitCode = true
|
||||
}
|
||||
return diagnostic
|
||||
}
|
||||
|
||||
func (m *Manager) logRefreshDiagnostic(
|
||||
operation, repoPath, branch, reason string, output []byte, runErr, contextErr error,
|
||||
) {
|
||||
diagnostic := classifyRefreshDiagnostic(string(output), runErr, contextErr)
|
||||
fields := []zap.Field{
|
||||
zap.String("operation", operation),
|
||||
zap.String("repository_path", repoPath),
|
||||
zap.String("branch", branch),
|
||||
zap.String("reason", reason),
|
||||
zap.String("diagnostic_code", diagnostic.code),
|
||||
zap.String("detail", diagnostic.detail),
|
||||
}
|
||||
if diagnostic.hasExitCode {
|
||||
fields = append(fields, zap.Int("exit_code", diagnostic.exitCode))
|
||||
}
|
||||
m.logger.Warn("git refresh failed", fields...)
|
||||
}
|
||||
@@ -0,0 +1,338 @@
|
||||
package worktree
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// @covers AC-WORKSPACES-WORKTREE-BASE-REFRESH-001.14
|
||||
func TestRefreshDiagnosticClassification(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
output string
|
||||
runErr error
|
||||
contextErr error
|
||||
wantCode string
|
||||
wantDetail string
|
||||
wantNoText string
|
||||
}{
|
||||
{name: "SSH public key", output: "Permission denied (publickey).", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticSSHKey, wantDetail: "public key"},
|
||||
{name: "SSH host key", output: "Host key verification failed.", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticSSHHost, wantDetail: "host"},
|
||||
{name: "DNS", output: "Could not resolve host: example.invalid", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticDNS, wantDetail: "host name"},
|
||||
{name: "connection", output: "Connection refused", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticConnection, wantDetail: "connection"},
|
||||
{name: "TLS", output: "SSL certificate problem: unable to get local issuer certificate", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticTLS, wantDetail: "TLS"},
|
||||
{name: "missing ref", output: "couldn't find remote ref feature/missing", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticMissingRef, wantDetail: "remote branch"},
|
||||
{name: "non-fast-forward", output: "rejected: non-fast-forward", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticNonFastForward, wantDetail: "non-fast-forward"},
|
||||
{name: "pull fast-forward refusal", output: "fatal: Not possible to fast-forward, aborting.", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticNonFastForward, wantDetail: "local branch", wantNoText: "remote"},
|
||||
{name: "tag clobber is not branch refusal", output: "! [rejected] v1 -> v1 (would clobber existing tag)", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticUnknown, wantDetail: "unknown reason"},
|
||||
{name: "repository access", output: "Repository not found.", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticRepository, wantDetail: "accessed"},
|
||||
{name: "lock", output: "fatal: Unable to create '.git/index.lock': File exists.", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticLock, wantDetail: "repository lock"},
|
||||
{name: "generic lock", output: "fatal: unable to create '.git/refs/heads/main.lock': File exists.", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticLock, wantDetail: "repository lock"},
|
||||
{name: "authentication", output: "terminal prompts disabled", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticAuthentication, wantDetail: "authenticate"},
|
||||
{name: "unknown", output: "unexpected backend detail: token-secret", runErr: errors.New("exit status 128"), wantCode: refreshDiagnosticUnknown, wantDetail: "unknown reason", wantNoText: "token-secret"},
|
||||
{name: "timeout wins", output: "Permission denied (publickey)", runErr: context.DeadlineExceeded, contextErr: context.DeadlineExceeded, wantCode: refreshDiagnosticTimeout, wantDetail: "timed out"},
|
||||
{name: "cancellation wins", output: "Permission denied (publickey)", runErr: context.Canceled, contextErr: context.Canceled, wantCode: refreshDiagnosticCanceled, wantDetail: "canceled"},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
got := classifyRefreshDiagnostic(tt.output, tt.runErr, tt.contextErr)
|
||||
if got.code != tt.wantCode {
|
||||
t.Fatalf("diagnostic code = %q, want %q", got.code, tt.wantCode)
|
||||
}
|
||||
if !strings.Contains(strings.ToLower(got.detail), strings.ToLower(tt.wantDetail)) {
|
||||
t.Fatalf("diagnostic detail = %q, want it to contain %q", got.detail, tt.wantDetail)
|
||||
}
|
||||
if tt.wantNoText != "" && strings.Contains(got.detail, tt.wantNoText) {
|
||||
t.Fatalf("diagnostic detail contains raw output %q: %q", tt.wantNoText, got.detail)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// @covers AC-WORKSPACES-WORKTREE-BASE-REFRESH-001.14
|
||||
func TestBaseRefreshDiagnosticEmission(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
gitScript string
|
||||
wantRef string
|
||||
wantCode string
|
||||
wantOperation string
|
||||
secret string
|
||||
}{
|
||||
{
|
||||
name: "fetch authentication failure",
|
||||
gitScript: `
|
||||
case "${1:-}" in
|
||||
fetch)
|
||||
echo "fatal: could not read Username for 'https://token:fetch-secret@example.invalid/repo.git': terminal prompts disabled" >&2
|
||||
exit 128
|
||||
;;
|
||||
rev-parse)
|
||||
exit 0
|
||||
;;
|
||||
*)
|
||||
exit 0
|
||||
;;
|
||||
esac
|
||||
`,
|
||||
wantRef: "main",
|
||||
wantCode: "authentication_failed",
|
||||
wantOperation: "fetch",
|
||||
secret: "fetch-secret",
|
||||
},
|
||||
{
|
||||
name: "fetch non-fast-forward update",
|
||||
gitScript: `
|
||||
case "${1:-}" in
|
||||
rev-parse)
|
||||
exit 0
|
||||
;;
|
||||
fetch)
|
||||
echo "From https://example.invalid/repo.git" >&2
|
||||
echo " ! [rejected] main -> main (non-fast-forward)" >&2
|
||||
echo "error: some local refs could not be updated" >&2
|
||||
exit 128
|
||||
;;
|
||||
*)
|
||||
exit 0
|
||||
;;
|
||||
esac
|
||||
`,
|
||||
wantRef: "main",
|
||||
wantCode: "non_fast_forward",
|
||||
wantOperation: "fetch",
|
||||
},
|
||||
{
|
||||
name: "pull TLS failure",
|
||||
gitScript: `
|
||||
case "${1:-}" in
|
||||
fetch)
|
||||
exit 0
|
||||
;;
|
||||
pull)
|
||||
echo "fatal: unable to access 'https://token:pull-secret@example.invalid/repo.git': SSL certificate problem" >&2
|
||||
exit 128
|
||||
;;
|
||||
rev-parse)
|
||||
if [ "${2:-}" = "--abbrev-ref" ]; then
|
||||
echo "main"
|
||||
fi
|
||||
exit 0
|
||||
;;
|
||||
merge-base)
|
||||
exit 0
|
||||
;;
|
||||
*)
|
||||
exit 0
|
||||
;;
|
||||
esac
|
||||
`,
|
||||
wantRef: "origin/main",
|
||||
wantCode: "tls_verification_failed",
|
||||
wantOperation: "pull",
|
||||
secret: "pull-secret",
|
||||
},
|
||||
{
|
||||
name: "pull fast-forward refusal",
|
||||
gitScript: `
|
||||
case "${1:-}" in
|
||||
fetch)
|
||||
exit 0
|
||||
;;
|
||||
pull)
|
||||
echo "fatal: Not possible to fast-forward, aborting." >&2
|
||||
exit 128
|
||||
;;
|
||||
rev-parse)
|
||||
if [ "${2:-}" = "--abbrev-ref" ]; then
|
||||
echo "main"
|
||||
fi
|
||||
exit 0
|
||||
;;
|
||||
merge-base)
|
||||
exit 0
|
||||
;;
|
||||
*)
|
||||
exit 0
|
||||
;;
|
||||
esac
|
||||
`,
|
||||
wantRef: "origin/main",
|
||||
wantCode: "non_fast_forward",
|
||||
wantOperation: "pull",
|
||||
},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
scriptDir := writeFakeGitScript(t, tt.gitScript)
|
||||
t.Setenv("PATH", scriptDir+string(os.PathListSeparator)+os.Getenv("PATH"))
|
||||
|
||||
log, logs := newObservedWorktreeLogger(t)
|
||||
mgr, err := NewManager(newTestConfig(t), newMockStore(), log)
|
||||
if err != nil {
|
||||
t.Fatalf("NewManager() error = %v", err)
|
||||
}
|
||||
|
||||
ref, err := mgr.pullBaseBranch(context.Background(), t.TempDir(), "main", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("pullBaseBranch() error = %v", err)
|
||||
}
|
||||
if ref != tt.wantRef {
|
||||
t.Fatalf("pullBaseBranch() ref = %q, want %q", ref, tt.wantRef)
|
||||
}
|
||||
|
||||
entries := logs.FilterMessage("git refresh failed").All()
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("observed %d refresh diagnostics, want 1; logs = %v", len(entries), logs.All())
|
||||
}
|
||||
fields := entries[0].ContextMap()
|
||||
if fields["operation"] != tt.wantOperation {
|
||||
t.Fatalf("operation = %v, want %q", fields["operation"], tt.wantOperation)
|
||||
}
|
||||
if fields["branch"] != "main" {
|
||||
t.Fatalf("branch = %v, want main", fields["branch"])
|
||||
}
|
||||
if fields["diagnostic_code"] != tt.wantCode {
|
||||
t.Fatalf("diagnostic_code = %v, want %q", fields["diagnostic_code"], tt.wantCode)
|
||||
}
|
||||
if fields["detail"] == "" {
|
||||
t.Fatal("diagnostic detail is empty")
|
||||
}
|
||||
for _, entry := range logs.All() {
|
||||
if strings.Contains(entry.Message, "https://token:") ||
|
||||
(tt.secret != "" && strings.Contains(entry.Message, tt.secret)) {
|
||||
t.Fatalf("log message contains secret material: %q", entry.Message)
|
||||
}
|
||||
for key, value := range entry.ContextMap() {
|
||||
rendered := fmt.Sprint(value)
|
||||
if strings.Contains(key, "https://token:") ||
|
||||
strings.Contains(rendered, "https://token:") ||
|
||||
(tt.secret != "" &&
|
||||
(strings.Contains(key, tt.secret) || strings.Contains(rendered, tt.secret))) {
|
||||
t.Fatalf("log field %q contains secret material: %v", key, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// @covers AC-WORKSPACES-WORKTREE-BASE-REFRESH-001.14
|
||||
func TestRefreshDiagnosticsPreservePolicy(t *testing.T) {
|
||||
scriptDir := writeFakeGitScript(t, `
|
||||
case "${1:-}" in
|
||||
rev-parse)
|
||||
exit 1
|
||||
;;
|
||||
fetch)
|
||||
echo "fatal: unexpected backend detail token-secret" >&2
|
||||
exit 128
|
||||
;;
|
||||
*)
|
||||
exit 0
|
||||
;;
|
||||
esac
|
||||
`)
|
||||
t.Setenv("PATH", scriptDir+string(os.PathListSeparator)+os.Getenv("PATH"))
|
||||
|
||||
log, logs := newObservedWorktreeLogger(t)
|
||||
mgr, err := NewManager(newTestConfig(t), newMockStore(), log)
|
||||
if err != nil {
|
||||
t.Fatalf("NewManager() error = %v", err)
|
||||
}
|
||||
|
||||
var events []SyncProgressEvent
|
||||
_, _, err = mgr.pullBaseBranchWithPolicy(
|
||||
context.Background(), t.TempDir(), "main", "", true, captureSyncProgress(&events),
|
||||
)
|
||||
if !errors.Is(err, ErrGitCommandFailed) {
|
||||
t.Fatalf("pullBaseBranchWithPolicy() error = %v, want ErrGitCommandFailed", err)
|
||||
}
|
||||
if strings.Contains(err.Error(), "token-secret") {
|
||||
t.Fatalf("returned error exposed raw Git output: %v", err)
|
||||
}
|
||||
if len(events) != 2 || events[1].Status != SyncProgressFailed {
|
||||
t.Fatalf("sync progress = %#v, want running then failed", events)
|
||||
}
|
||||
if events[1].Error != "git_command_failed" || strings.Contains(events[1].Output, "token-secret") {
|
||||
t.Fatalf("sync progress exposed raw failure details: %#v", events[1])
|
||||
}
|
||||
|
||||
entries := logs.FilterMessage("git refresh failed").All()
|
||||
if len(entries) != 1 {
|
||||
t.Fatalf("observed %d refresh diagnostics, want 1; logs = %v", len(entries), logs.All())
|
||||
}
|
||||
fields := entries[0].ContextMap()
|
||||
if fields["operation"] != "fetch" || fields["branch"] != "main" || fields["reason"] != "git_command_failed" {
|
||||
t.Fatalf("diagnostic fields = %v, want fetch/main/git_command_failed", fields)
|
||||
}
|
||||
if fields["diagnostic_code"] != refreshDiagnosticUnknown {
|
||||
t.Fatalf("diagnostic_code = %v, want %q", fields["diagnostic_code"], refreshDiagnosticUnknown)
|
||||
}
|
||||
for _, entry := range logs.All() {
|
||||
if strings.Contains(entry.Message, "token-secret") || strings.Contains(fmt.Sprint(entry.ContextMap()), "token-secret") {
|
||||
t.Fatalf("log exposed raw Git output: %#v", entry)
|
||||
}
|
||||
}
|
||||
|
||||
t.Run("configured fallback failure", func(t *testing.T) {
|
||||
scriptDir := writeFakeGitScript(t, `
|
||||
case "${1:-}" in
|
||||
rev-parse)
|
||||
exit 1
|
||||
;;
|
||||
fetch)
|
||||
case "${4:-}" in
|
||||
feature/missing)
|
||||
echo "fatal: couldn't find remote ref feature/missing" >&2
|
||||
;;
|
||||
main)
|
||||
echo "fatal: could not read Username for 'https://token:fallback-secret@example.invalid/repo.git': terminal prompts disabled" >&2
|
||||
;;
|
||||
esac
|
||||
exit 128
|
||||
;;
|
||||
*)
|
||||
exit 0
|
||||
;;
|
||||
esac
|
||||
`)
|
||||
t.Setenv("PATH", scriptDir+string(os.PathListSeparator)+os.Getenv("PATH"))
|
||||
log, logs := newObservedWorktreeLogger(t)
|
||||
mgr, err := NewManager(newTestConfig(t), newMockStore(), log)
|
||||
if err != nil {
|
||||
t.Fatalf("NewManager() error = %v", err)
|
||||
}
|
||||
var events []SyncProgressEvent
|
||||
_, _, err = mgr.pullBaseBranchWithFallback(
|
||||
context.Background(), t.TempDir(), "feature/missing", "main", captureSyncProgress(&events),
|
||||
)
|
||||
if !errors.Is(err, ErrAuthFailed) {
|
||||
t.Fatalf("pullBaseBranchWithFallback() error = %v, want ErrAuthFailed", err)
|
||||
}
|
||||
if strings.Contains(err.Error(), "fallback-secret") || strings.Contains(fmt.Sprint(events), "fallback-secret") {
|
||||
t.Fatalf("fallback failure exposed raw Git output: error=%v events=%#v", err, events)
|
||||
}
|
||||
entries := logs.FilterMessage("git refresh failed").All()
|
||||
if len(entries) != 2 {
|
||||
t.Fatalf("observed %d refresh diagnostics, want primary and fallback failures", len(entries))
|
||||
}
|
||||
if entries[0].ContextMap()["branch"] != "feature/missing" || entries[0].ContextMap()["diagnostic_code"] != refreshDiagnosticMissingRef {
|
||||
t.Fatalf("primary diagnostic fields = %v", entries[0].ContextMap())
|
||||
}
|
||||
if entries[1].ContextMap()["branch"] != "main" || entries[1].ContextMap()["diagnostic_code"] != refreshDiagnosticAuthentication {
|
||||
t.Fatalf("fallback diagnostic fields = %v", entries[1].ContextMap())
|
||||
}
|
||||
for _, entry := range logs.All() {
|
||||
if strings.Contains(fmt.Sprint(entry.ContextMap()), "fallback-secret") {
|
||||
t.Fatalf("fallback diagnostic exposed raw Git output: %#v", entry)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user