fix(coderd/x/chatd): log retry errors and add a task timeout (#26412)

This PR adds logging when the chat runner retries and exits because of
an error. It also adds a 15-minute task timeout to ensure that stuck
tasks do not hang forever.

coderd/x/chatd/chatd_retry_test.go

@@ -3,13 +3,16 @@ package chatd_test
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/google/uuid"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/stretchr/testify/require"
 
+	"cdr.dev/slog/v3"
 	"github.com/coder/coder/v2/coderd/database"
 	"github.com/coder/coder/v2/coderd/database/dbgen"
 	"github.com/coder/coder/v2/coderd/database/dbtestutil"
@@ -26,6 +29,7 @@ func TestActiveServer_RetryStatePersistedDuringBackoff(t *testing.T) {
 	ctx := testutil.Context(t, testutil.WaitLong)
 	db, ps := dbtestutil.NewDB(t)
 	clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
+	sink := testutil.NewFakeSink(t)
 	var calls atomic.Int32
 	openAIURL := chattest.NewOpenAI(t, func(req *chattest.OpenAIRequest) chattest.OpenAIResponse {
 		if !req.Stream {
@@ -39,6 +43,7 @@ func TestActiveServer_RetryStatePersistedDuringBackoff(t *testing.T) {
 	user, org, model := seedChatDependenciesWithProvider(t, db, "openai", openAIURL)
 	server := newActiveTestServer(t, db, ps, func(cfg *chatd.Config) {
 		cfg.Clock = clock
+		cfg.Logger = sink.Logger()
 	})
 
 	chat := createChatThroughServer(ctx, t, db, server, org.ID, user.ID, model.ID, "hello")
@@ -65,6 +70,12 @@ func TestActiveServer_RetryStatePersistedDuringBackoff(t *testing.T) {
 	latest, err := db.GetChatByID(ctx, chat.ID)
 	require.NoError(t, err)
 	require.False(t, latest.RetryState.Valid)
+	entries := retryEntriesWithMessage(sink, "chat generation retrying")
+	require.Len(t, entries, 1)
+	require.Equal(t, "generate_assistant", retrySinkFieldValue(t, entries[0].Fields, "action"))
+	require.Equal(t, "openai", retrySinkFieldValue(t, entries[0].Fields, "provider"))
+	require.Equal(t, "429", retrySinkFieldValue(t, entries[0].Fields, "status_code"))
+	require.Equal(t, "false", retrySinkFieldValue(t, entries[0].Fields, "chain_broken"))
 	require.Greater(t, latest.RetryStateVersion, withRetry.RetryStateVersion)
 	messages, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ChatID: chat.ID})
 	require.NoError(t, err)
@@ -116,13 +127,18 @@ func TestActiveServer_RetryStreamSilenceTimeoutAndClassification(t *testing.T) {
 		})
 	})
 
-	t.Run("stream silence timeout retry recovers", func(t *testing.T) {
+	t.Run("silent stream generation retry recovers", func(t *testing.T) {
 		t.Parallel()
 
 		ctx := testutil.Context(t, testutil.WaitLong)
 		db, ps := dbtestutil.NewDB(t)
-		clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
 		reg := prometheus.NewRegistry()
+		clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
+		streamGuardTrap := clock.Trap().AfterFunc("streamSilenceGuard")
+		defer streamGuardTrap.Close()
+		retryTrap := clock.Trap().NewTimer("chatworker", "generation-retry")
+		defer retryTrap.Close()
+		sink := testutil.NewFakeSink(t)
 		var calls atomic.Int32
 		openAIURL := chattest.NewOpenAI(t, func(req *chattest.OpenAIRequest) chattest.OpenAIResponse {
 			if !req.Stream {
@@ -137,27 +153,54 @@ func TestActiveServer_RetryStreamSilenceTimeoutAndClassification(t *testing.T) {
 		user, org, model := seedChatDependenciesWithProvider(t, db, "openai", openAIURL)
 		server := newActiveTestServer(t, db, ps, func(cfg *chatd.Config) {
 			cfg.Clock = clock
+			cfg.Logger = sink.Logger()
 			cfg.PrometheusRegistry = reg
+			cfg.PendingChatAcquireInterval = 30 * time.Minute
+			cfg.ChatHeartbeatInterval = 30 * time.Minute
 		})
 
 		chat := createChatThroughServer(ctx, t, db, server, org.ID, user.ID, model.ID, "hello")
-		advanceUntilProviderCall(ctx, clock, &calls, 1)
-		advanceToNextTimer(ctx, clock)
-		advanceUntilProviderCall(ctx, clock, &calls, 2)
+		firstGuard := streamGuardTrap.MustWait(ctx)
+		firstGuard.MustRelease(ctx)
+		waitUntilProviderCall(ctx, t, &calls, 1)
+		advanceMockClockBy(ctx, t, clock, firstGuard.Duration)
+		retryTimer := retryTrap.MustWait(ctx)
+		retryTimer.MustRelease(ctx)
+		advanceMockClockBy(ctx, t, clock, retryTimer.Duration)
+		secondGuard := streamGuardTrap.MustWait(ctx)
+		secondGuard.MustRelease(ctx)
+		waitUntilProviderCall(ctx, t, &calls, 2)
 		waitForChatStatus(ctx, t, db, chat.ID, database.ChatStatusWaiting)
 		require.Equal(t, int32(2), calls.Load())
 		messages, err := db.GetChatMessagesByChatID(ctx, database.GetChatMessagesByChatIDParams{ChatID: chat.ID})
 		require.NoError(t, err)
 		requireTextPart(t, messages[len(messages)-1], "recovered")
+		require.Empty(t, retryEntriesWithMessage(sink, "chatworker task retrying"))
+		entries := retryEntriesWithMessage(sink, "chat generation retrying")
+		require.NotEmpty(t, entries)
+		require.Equal(t, "generate_assistant", retrySinkFieldValue(t, entries[0].Fields, "action"))
+		require.Equal(t, string(codersdk.ChatErrorKindStreamSilenceTimeout), retrySinkFieldValue(t, entries[0].Fields, "error_kind"))
+		require.Equal(t, "openai", retrySinkFieldValue(t, entries[0].Fields, "provider"))
 		requireRetryCounter(t, reg, "coderd_chatd_stream_retries_total", 1, map[string]string{
 			"provider":     "openai",
-			"model":        "gpt-4o-mini",
+			"model":        model.Model,
 			"kind":         string(codersdk.ChatErrorKindStreamSilenceTimeout),
 			"chain_broken": "false",
 		})
 	})
 }
 
+func retryEntriesWithMessage(sink *testutil.FakeSink, message string) []slog.SinkEntry {
+	return sink.Entries(func(e slog.SinkEntry) bool { return e.Message == message })
+}
+
+func retrySinkFieldValue(t *testing.T, fields slog.Map, name string) string {
+	t.Helper()
+	value, ok := sinkFieldValue(fields, name)
+	require.True(t, ok, "missing log field %q", name)
+	return fmt.Sprint(value)
+}
+
 func requireRetryCounter(t *testing.T, reg *prometheus.Registry, name string, wantValue float64, wantLabels map[string]string) {
 	t.Helper()
 	require.True(t, hasRetryCounter(t, reg, name, wantValue, wantLabels), "metric %s not found", name)
@@ -210,6 +253,28 @@ func waitForChatRetryState(ctx context.Context, t *testing.T, db database.Store,
 	return chat
 }
 
+func waitUntilProviderCall(ctx context.Context, t *testing.T, calls *atomic.Int32, want int32) {
+	t.Helper()
+	testutil.Eventually(ctx, t, func(context.Context) bool {
+		return calls.Load() >= want
+	}, testutil.IntervalFast)
+}
+
+func advanceMockClockBy(ctx context.Context, t *testing.T, clock *quartz.Mock, d time.Duration) {
+	t.Helper()
+	for remaining := d; remaining > 0; {
+		next, ok := clock.Peek()
+		require.True(t, ok, "no pending clock event while advancing %s", remaining)
+		if next > remaining {
+			clock.Advance(remaining).MustWait(ctx)
+			return
+		}
+		_, waiter := clock.AdvanceNext()
+		waiter.MustWait(ctx)
+		remaining -= next
+	}
+}
+
 func advanceUntilProviderCall(ctx context.Context, clock *quartz.Mock, calls *atomic.Int32, want int32) {
 	for calls.Load() < want {
 		advanceToNextTimer(ctx, clock)

coderd/x/chatd/generation.go

@@ -360,17 +360,17 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS
 			Messages:          messages,
 			ChainModeDisabled: chainModeDisabled,
 		}
-		prepared, err := retryGenerationPhase(ctx, s.waitGenerationPhaseBackoff, func() (generationPrepared, error) {
+		prepared, err := retryGenerationPhase(ctx, s, "prepare", func() (generationPrepared, error) {
 			return s.server.prepareGeneration(ctx, prepareInput)
 		})
 		if err != nil {
-			if errors.Is(err, errTaskExpectedExit) {
-				return errTaskExpectedExit
+			if errors.Is(err, errTaskExpectedExit) || errors.Is(err, errTaskRetryable) {
+				return xerrors.Errorf("prepare generation: %w", err)
 			}
 			return s.finishGenerationError(ctx, machine, input, 0, err, generationAttemptNotRequired)
 		}
 		cleanup := prepared.Cleanup
-		decision, err := retryGenerationPhase(ctx, s.waitGenerationPhaseBackoff, func() (generationDecision, error) {
+		decision, err := retryGenerationPhase(ctx, s, "decide", func() (generationDecision, error) {
 			return decideGenerationAction(generationDecisionInput{
 				chat:                       prepared.Chat,
 				messages:                   prepared.Messages,
@@ -387,8 +387,8 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS
 		})
 		if err != nil {
 			cleanup()
-			if errors.Is(err, errTaskExpectedExit) {
-				return errTaskExpectedExit
+			if errors.Is(err, errTaskExpectedExit) || errors.Is(err, errTaskRetryable) {
+				return xerrors.Errorf("decide generation: %w", err)
 			}
 			if errors.Is(err, errCompactionStillOverLimit) && prepared.Compaction != nil {
 				s.server.metrics.RecordCompaction(
@@ -424,19 +424,33 @@ func (s *taskStarter) StartGeneration(ctx context.Context, input chatWorkerTaskS
 		if actionErr == nil {
 			return nil
 		}
-		if errors.Is(actionErr, errTaskExpectedExit) || errors.Is(actionErr, chatloop.ErrInterrupted) {
-			return nil
+		// Task cancellation is handled by the runner, not here.
+		if ctx.Err() != nil && errors.Is(actionErr, context.Canceled) {
+			return errors.Join(errTaskExpectedExit, xerrors.Errorf("generation action: %w", actionErr), ctx.Err())
 		}
-		if errors.Is(actionErr, context.Canceled) && ctx.Err() != nil {
+		if errors.Is(actionErr, errTaskExpectedExit) || errors.Is(actionErr, chatloop.ErrInterrupted) {
 			return nil
 		}
 		classified := chaterror.Classify(actionErr)
 		if classified.Retryable {
+			action := decision.kind
 			decision, err := s.recordGenerationRetry(ctx, machine, input, classified)
 			if err != nil {
 				return err
 			}
 			if decision.retry {
+				s.opts.Logger.Warn(ctx, "chat generation retrying",
+					slog.F("chat_id", input.ChatID),
+					slog.F("worker_id", input.WorkerID),
+					slog.F("action", action),
+					slog.F("generation_attempt", decision.generationAttempt),
+					slog.F("delay", decision.delay),
+					slog.F("error_kind", classified.Kind),
+					slog.F("provider", classified.Provider),
+					slog.F("status_code", classified.StatusCode),
+					slog.F("chain_broken", classified.ChainBroken),
+					slogError(actionErr),
+				)
 				if classified.ChainBroken {
 					chainModeDisabled = true
 				}
@@ -548,7 +562,7 @@ func (s *taskStarter) waitGenerationRetry(ctx context.Context, delay time.Durati
 	case <-timer.C:
 		return nil
 	case <-ctx.Done():
-		return errTaskExpectedExit
+		return errors.Join(errTaskExpectedExit, xerrors.Errorf("wait generation retry: %w", ctx.Err()))
 	}
 }
 
@@ -576,12 +590,9 @@ func generationPhaseBackoff(attempt int) time.Duration {
 // returns early on success or on a terminal error (see terminalGeneration).
 // Non-terminal errors are retried with exponential backoff. Context
 // cancellation returns errTaskExpectedExit so shutdown does not write an
-// error state. When every attempt fails, the last error is returned.
-func retryGenerationPhase[T any](
-	ctx context.Context,
-	wait func(context.Context, time.Duration) error,
-	fn func() (T, error),
-) (T, error) {
+// error state. Task timeouts return a retryable task error so the runner can
+// start a fresh attempt. When every attempt fails, the last error is returned.
+func retryGenerationPhase[T any](ctx context.Context, starter *taskStarter, phase string, fn func() (T, error)) (T, error) {
 	var zero T
 	var lastErr error
 	for attempt := 0; attempt < generationPhaseMaxAttempts; attempt++ {
@@ -590,14 +601,22 @@ func retryGenerationPhase[T any](
 			return result, nil
 		}
 		if isTerminalGeneration(err) {
-			return zero, err
+			return zero, xerrors.Errorf("retryGenerationPhase terminal error: %w", err)
 		}
 		if ctx.Err() != nil {
-			return zero, errTaskExpectedExit
+			return zero, errors.Join(errTaskExpectedExit, xerrors.Errorf("retryGenerationPhase %s: %w", phase, ctx.Err()))
 		}
 		lastErr = err
 		if attempt < generationPhaseMaxAttempts-1 {
-			if waitErr := wait(ctx, generationPhaseBackoff(attempt)); waitErr != nil {
+			delay := generationPhaseBackoff(attempt)
+			starter.opts.Logger.Warn(ctx, "chat generation phase retrying",
+				slog.F("phase", phase),
+				slog.F("attempt", attempt+1),
+				slog.F("max_attempts", generationPhaseMaxAttempts),
+				slog.F("delay", delay),
+				slogError(err),
+			)
+			if waitErr := starter.waitGenerationPhaseBackoff(ctx, delay); waitErr != nil {
 				return zero, waitErr
 			}
 		}
@@ -612,7 +631,7 @@ func (s *taskStarter) waitGenerationPhaseBackoff(ctx context.Context, delay time
 	case <-timer.C:
 		return nil
 	case <-ctx.Done():
-		return errTaskExpectedExit
+		return errors.Join(errTaskExpectedExit, xerrors.Errorf("wait generation phase backoff: %w", ctx.Err()))
 	}
 }
 

coderd/x/chatd/generation_retry_internal_test.go

@@ -7,6 +7,9 @@ import (
 
 	"github.com/stretchr/testify/require"
 	"golang.org/x/xerrors"
+
+	"github.com/coder/coder/v2/testutil"
+	"github.com/coder/quartz"
 )
 
 func TestTerminalGeneration(t *testing.T) {
@@ -36,92 +39,100 @@ func TestGenerationPhaseBackoff(t *testing.T) {
 func TestRetryGenerationPhase(t *testing.T) {
 	t.Parallel()
 
-	noopWait := func(context.Context, time.Duration) error { return nil }
-
 	t.Run("SuccessFirstTry", func(t *testing.T) {
 		t.Parallel()
+		starter := newGenerationPhaseTestStarter(t, quartz.NewMock(t).WithLogger(quartz.NoOpLogger))
 		calls := 0
-		waits := 0
-		wait := func(context.Context, time.Duration) error {
-			waits++
-			return nil
-		}
-		got, err := retryGenerationPhase(context.Background(), wait, func() (int, error) {
+		got, err := retryGenerationPhase(context.Background(), starter, "prepare", func() (int, error) {
 			calls++
 			return 42, nil
 		})
 		require.NoError(t, err)
 		require.Equal(t, 42, got)
 		require.Equal(t, 1, calls)
-		require.Equal(t, 0, waits)
 	})
 
 	t.Run("RetryThenSuccess", func(t *testing.T) {
 		t.Parallel()
+		clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
+		timerTrap := clock.Trap().NewTimer("chatworker", "generation-phase-retry")
+		defer timerTrap.Close()
+		sink := testutil.NewFakeSink(t)
+		starter := newGenerationPhaseTestStarter(t, clock)
+		starter.opts.Logger = sink.Logger()
+		ctx := testutil.Context(t, testutil.WaitLong)
 		calls := 0
-		waits := 0
-		var delays []time.Duration
-		wait := func(_ context.Context, d time.Duration) error {
-			waits++
-			delays = append(delays, d)
-			return nil
-		}
-		got, err := retryGenerationPhase(context.Background(), wait, func() (string, error) {
-			calls++
-			if calls < 2 {
-				return "", xerrors.New("transient")
-			}
-			return "ok", nil
-		})
-		require.NoError(t, err)
-		require.Equal(t, "ok", got)
+		done := make(chan phaseRetryResult[string], 1)
+		go func() {
+			got, err := retryGenerationPhase(ctx, starter, "prepare", func() (string, error) {
+				calls++
+				if calls < 2 {
+					return "", xerrors.New("transient")
+				}
+				return "ok", nil
+			})
+			done <- phaseRetryResult[string]{value: got, err: err}
+		}()
+
+		timerTrap.MustWait(ctx).MustRelease(ctx)
+		clock.Advance(generationPhaseBackoff(0)).MustWait(ctx)
+		result := <-done
+		require.NoError(t, result.err)
+		require.Equal(t, "ok", result.value)
 		require.Equal(t, 2, calls)
-		require.Equal(t, 1, waits)
-		require.Equal(t, []time.Duration{generationPhaseBackoff(0)}, delays)
+		entries := entriesWithMessage(sink, "chat generation phase retrying")
+		require.Len(t, entries, 1)
+		require.Equal(t, "prepare", sinkFieldValue(t, entries[0].Fields, "phase"))
+		require.Equal(t, "1", sinkFieldValue(t, entries[0].Fields, "attempt"))
+		require.Equal(t, generationPhaseBackoff(0).String(), sinkFieldValue(t, entries[0].Fields, "delay"))
 	})
 
 	t.Run("ExhaustsAndReturnsLastError", func(t *testing.T) {
 		t.Parallel()
+		clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
+		timerTrap := clock.Trap().NewTimer("chatworker", "generation-phase-retry")
+		defer timerTrap.Close()
+		starter := newGenerationPhaseTestStarter(t, clock)
+		ctx := testutil.Context(t, testutil.WaitLong)
 		calls := 0
-		waits := 0
-		wait := func(context.Context, time.Duration) error {
-			waits++
-			return nil
-		}
-		_, err := retryGenerationPhase(context.Background(), wait, func() (int, error) {
-			calls++
-			return 0, xerrors.Errorf("attempt %d", calls)
-		})
-		require.EqualError(t, err, "attempt 3")
+		done := make(chan error, 1)
+		go func() {
+			_, err := retryGenerationPhase(ctx, starter, "prepare", func() (int, error) {
+				calls++
+				return 0, xerrors.Errorf("attempt %d", calls)
+			})
+			done <- err
+		}()
+
+		timerTrap.MustWait(ctx).MustRelease(ctx)
+		clock.Advance(generationPhaseBackoff(0)).MustWait(ctx)
+		timerTrap.MustWait(ctx).MustRelease(ctx)
+		clock.Advance(generationPhaseBackoff(1)).MustWait(ctx)
+		require.EqualError(t, <-done, "attempt 3")
 		require.Equal(t, generationPhaseMaxAttempts, calls)
-		require.Equal(t, generationPhaseMaxAttempts-1, waits)
 	})
 
 	t.Run("TerminalShortCircuits", func(t *testing.T) {
 		t.Parallel()
+		starter := newGenerationPhaseTestStarter(t, quartz.NewMock(t).WithLogger(quartz.NoOpLogger))
 		calls := 0
-		waits := 0
-		wait := func(context.Context, time.Duration) error {
-			waits++
-			return nil
-		}
 		cause := xerrors.New("deterministic")
-		_, err := retryGenerationPhase(context.Background(), wait, func() (int, error) {
+		_, err := retryGenerationPhase(context.Background(), starter, "prepare", func() (int, error) {
 			calls++
 			return 0, terminalGeneration(cause)
 		})
 		require.ErrorIs(t, err, cause)
 		require.True(t, isTerminalGeneration(err))
 		require.Equal(t, 1, calls)
-		require.Equal(t, 0, waits)
 	})
 
 	t.Run("ContextCanceledExitsCleanly", func(t *testing.T) {
 		t.Parallel()
+		starter := newGenerationPhaseTestStarter(t, quartz.NewMock(t).WithLogger(quartz.NoOpLogger))
 		ctx, cancel := context.WithCancel(context.Background())
 		cancel()
 		calls := 0
-		_, err := retryGenerationPhase(ctx, noopWait, func() (int, error) {
+		_, err := retryGenerationPhase(ctx, starter, "prepare", func() (int, error) {
 			calls++
 			return 0, xerrors.New("transient")
 		})
@@ -131,18 +142,40 @@ func TestRetryGenerationPhase(t *testing.T) {
 
 	t.Run("WaitCancellationExitsCleanly", func(t *testing.T) {
 		t.Parallel()
+		clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
+		timerTrap := clock.Trap().NewTimer("chatworker", "generation-phase-retry")
+		defer timerTrap.Close()
+		starter := newGenerationPhaseTestStarter(t, clock)
+		ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitLong))
 		calls := 0
-		waits := 0
-		wait := func(context.Context, time.Duration) error {
-			waits++
-			return errTaskExpectedExit
-		}
-		_, err := retryGenerationPhase(context.Background(), wait, func() (int, error) {
-			calls++
-			return 0, xerrors.New("transient")
-		})
-		require.ErrorIs(t, err, errTaskExpectedExit)
+		done := make(chan error, 1)
+		go func() {
+			_, err := retryGenerationPhase(ctx, starter, "prepare", func() (int, error) {
+				calls++
+				return 0, xerrors.New("transient")
+			})
+			done <- err
+		}()
+
+		timerTrap.MustWait(ctx).MustRelease(ctx)
+		cancel()
+		require.ErrorIs(t, <-done, errTaskExpectedExit)
 		require.Equal(t, 1, calls)
-		require.Equal(t, 1, waits)
 	})
 }
+
+type phaseRetryResult[T any] struct {
+	value T
+	err   error
+}
+
+func newGenerationPhaseTestStarter(t *testing.T, clock quartz.Clock) *taskStarter {
+	t.Helper()
+	require.NotNil(t, clock)
+	return &taskStarter{opts: chatWorkerOptions{
+		Clock:                   clock,
+		Logger:                  testutil.NewFakeSink(t).Logger(),
+		TaskRetryInitialBackoff: time.Millisecond,
+		TaskRetryMaxBackoff:     time.Millisecond,
+	}}
+}

coderd/x/chatd/runner.go

@@ -232,11 +232,11 @@ func (r *runner) runTask(
 	err := runTaskWithRetry(ctx, r.opts.retryOptions(), kind, func(ctx context.Context) error {
 		unlock, ok := r.localLocks.acquire(ctx, key)
 		if !ok {
-			return errTaskExpectedExit
+			return errors.Join(errTaskExpectedExit, xerrors.Errorf("runTask acquire local lock: %w", ctx.Err()))
 		}
 		defer unlock()
 		if ctx.Err() != nil {
-			return errTaskExpectedExit
+			return errors.Join(errTaskExpectedExit, xerrors.Errorf("runTask context canceled: %w", ctx.Err()))
 		}
 
 		switch kind {

coderd/x/chatd/runner_test.go

@@ -1,6 +1,7 @@
 package chatd //nolint:testpackage // Uses unexported chatworker helpers.
 
 import (
+	"context"
 	"testing"
 	"time"
 
@@ -38,6 +39,7 @@ func TestRunner_CancelsActiveTaskWhenHistoryChanges(t *testing.T) {
 	updated := commitAssistantStep(t, f, chat.ID, "first step")
 	require.Greater(t, updated.HistoryVersion, first.input.HistoryVersion)
 	requireTaskCanceled(t, first)
+	require.NotErrorIs(t, context.Cause(first.ctx), errTaskTimeout)
 	second := starter.waitCall(t, taskKindGeneration, chat.ID)
 	require.Equal(t, updated.HistoryVersion, second.input.HistoryVersion)
 }
@@ -104,11 +106,39 @@ func TestRunner_AllowsReplacementForDifferentHistoryOrStatus(t *testing.T) {
 	require.Equal(t, updated.HistoryVersion, second.input.HistoryVersion)
 }
 
+func TestRunner_TaskTimeoutRetries(t *testing.T) {
+	t.Parallel()
+	f := newWorkerTestFixture(t)
+	chat := f.createRunningChat(t)
+	clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
+	timeoutTrap := clock.Trap().AfterFunc("chatworker", "task-timeout-generation")
+	starter := newBlockingTaskStarter(false)
+	opts := testOptions(t, f, starter)
+	opts.Clock = clock
+	opts.TaskRetryInitialBackoff = time.Minute
+	opts.TaskRetryMaxBackoff = time.Minute
+	startWorker(t, opts)
+
+	timeoutTrap.MustWait(testutil.Context(t, testutil.WaitLong)).MustRelease(testutil.Context(t, testutil.WaitLong))
+	timeoutTrap.Close()
+	first := starter.waitCall(t, taskKindGeneration, chat.ID)
+	retryTrap := clock.Trap().NewTimer("chatworker", "task-retry-generation")
+	defer retryTrap.Close()
+
+	ctx := testutil.Context(t, testutil.WaitLong)
+	clock.Advance(defaultTaskTimeout).MustWait(ctx)
+	retryTrap.MustWait(ctx).MustRelease(ctx)
+	require.ErrorIs(t, context.Cause(first.ctx), errTaskTimeout)
+	clock.Advance(time.Minute).MustWait(ctx)
+	second := starter.waitCall(t, taskKindGeneration, chat.ID)
+	require.Equal(t, first.input.HistoryVersion, second.input.HistoryVersion)
+}
+
 func TestWorker_RoutesDatabaseSyncStateToActiveRunner(t *testing.T) {
 	t.Parallel()
 	f := newWorkerTestFixture(t)
 	chat := f.createRunningChat(t)
-	clock := quartz.NewMock(t)
+	clock := quartz.NewMock(t).WithLogger(quartz.NoOpLogger)
 	starter := newBlockingTaskStarter(false)
 	opts := testOptions(t, f, starter)
 	opts.Clock = clock

coderd/x/chatd/tasks.go

@@ -11,6 +11,7 @@ import (
 	"github.com/google/uuid"
 	"golang.org/x/xerrors"
 
+	"cdr.dev/slog/v3"
 	"github.com/coder/coder/v2/coderd/database"
 	coderdpubsub "github.com/coder/coder/v2/coderd/pubsub"
 	"github.com/coder/coder/v2/coderd/x/chatd/chatdebug"
@@ -21,11 +22,18 @@ import (
 	"github.com/coder/quartz"
 )
 
-const postCommitWatchPublishTimeout = 10 * time.Second
+const (
+	postCommitWatchPublishTimeout = 10 * time.Second
+	// defaultTaskTimeout must exceed chatloop's stream-silence guard so
+	// silent provider streams fail through chat-specific retry handling
+	// before the runner retries the whole task.
+	defaultTaskTimeout = 15 * time.Minute
+)
 
 var (
 	errTaskExpectedExit = xerrors.New("chatworker task expected exit")
 	errTaskRetryable    = xerrors.New("chatworker task retryable error")
+	errTaskTimeout      = xerrors.New("chatworker task timeout")
 )
 
 type taskRetryableError struct {
@@ -48,10 +56,15 @@ func (e taskRetryableError) Unwrap() error {
 
 type retryWrapperOptions struct {
 	clock        quartz.Clock
+	logger       slog.Logger
 	initialDelay time.Duration
 	maxDelay     time.Duration
 }
 
+// runTaskWithRetry ensures that a task doesn't exit until it completes
+// successfully or gets canceled. It retries the task in case of any ephemeral errors.
+// It's critical for the correct operation of the chat runner:
+// this function is THE place that ensures task liveness within the runner.
 func runTaskWithRetry(
 	ctx context.Context,
 	opts retryWrapperOptions,
@@ -70,19 +83,46 @@ func runTaskWithRetry(
 	if opts.maxDelay < opts.initialDelay {
 		opts.maxDelay = opts.initialDelay
 	}
-
 	delay := opts.initialDelay
 	for {
-		err := executeTaskSafely(ctx, fn)
-		switch {
-		case err == nil:
-			return nil
-		case errors.Is(err, errTaskExpectedExit):
-			return nil
-		case ctx.Err() != nil:
+		attemptCtx, cancelAttempt := taskAttemptContext(ctx, opts.clock, kind)
+		err := executeTaskSafely(attemptCtx, fn)
+		timedOut := errors.Is(context.Cause(attemptCtx), errTaskTimeout)
+		cancelAttempt()
+		if timedOut && err != nil {
+			if !errors.Is(err, errTaskExpectedExit) ||
+				errors.Is(err, context.Canceled) ||
+				errors.Is(err, context.DeadlineExceeded) ||
+				errors.Is(err, errTaskTimeout) {
+				err = taskRetryableError{err: errors.Join(errTaskTimeout, err)}
+			}
+		}
+		if err == nil {
+			// no log on success to avoid noise
 			return nil
 		}
 
+		exitReason := ""
+		switch {
+		case ctx.Err() != nil:
+			exitReason = "context_canceled"
+		case errors.Is(err, errTaskExpectedExit) && !errors.Is(err, errTaskRetryable):
+			exitReason = "expected_non_retryable_exit"
+		}
+		if exitReason != "" {
+			opts.logger.Debug(ctx, "chatworker task exited",
+				slog.F("task_kind", kind),
+				slog.F("reason", exitReason),
+				slogError(err),
+			)
+			return nil
+		}
+
+		opts.logger.Warn(ctx, "chatworker task retrying",
+			slog.F("task_kind", kind),
+			slog.F("delay", delay),
+			slogError(err),
+		)
 		timer := opts.clock.NewTimer(delay, "chatworker", "task-retry-"+string(kind))
 		select {
 		case <-timer.C:
@@ -100,6 +140,17 @@ func runTaskWithRetry(
 	}
 }
 
+func taskAttemptContext(ctx context.Context, clock quartz.Clock, kind taskKind) (context.Context, func()) {
+	attemptCtx, cancelCause := context.WithCancelCause(ctx)
+	timer := clock.AfterFunc(defaultTaskTimeout, func() {
+		cancelCause(errTaskTimeout)
+	}, "chatworker", "task-timeout-"+string(kind))
+	return attemptCtx, func() {
+		timer.Stop()
+		cancelCause(nil)
+	}
+}
+
 func executeTaskSafely(ctx context.Context, fn func(context.Context) error) (err error) {
 	defer func() {
 		if recovered := recover(); recovered != nil {
@@ -167,6 +218,7 @@ func newTaskStarter(
 func (o chatWorkerOptions) retryOptions() retryWrapperOptions {
 	return retryWrapperOptions{
 		clock:        o.Clock,
+		logger:       o.Logger,
 		initialDelay: o.TaskRetryInitialBackoff,
 		maxDelay:     o.TaskRetryMaxBackoff,
 	}
@@ -200,7 +252,7 @@ func (s *taskStarter) StartInterrupt(ctx context.Context, input chatWorkerTaskSt
 	}
 	if err := s.opts.MessagePartBuffer.CloseEpisode(key); err != nil {
 		if ctx.Err() != nil {
-			return errTaskExpectedExit
+			return errors.Join(errTaskExpectedExit, xerrors.Errorf("close message part episode: %w", err), ctx.Err())
 		}
 		return taskRetryableError{err: xerrors.Errorf("close message part episode: %w", err)}
 	}
@@ -211,7 +263,7 @@ func (s *taskStarter) StartInterrupt(ctx context.Context, input chatWorkerTaskSt
 	}
 	if err != nil {
 		if ctx.Err() != nil {
-			return errTaskExpectedExit
+			return errors.Join(errTaskExpectedExit, xerrors.Errorf("get message part episode: %w", err), ctx.Err())
 		}
 		return taskRetryableError{err: xerrors.Errorf("get message part episode: %w", err)}
 	}
@@ -362,7 +414,7 @@ func (s *taskStarter) waitUntil(ctx context.Context, deadline time.Time) error {
 	case <-timer.C:
 		return nil
 	case <-ctx.Done():
-		return errTaskExpectedExit
+		return errors.Join(errTaskExpectedExit, xerrors.Errorf("wait until: %w", ctx.Err()))
 	}
 }
 
@@ -491,14 +543,14 @@ func (s *taskStarter) publishWatchWithRetry(
 		if err := publishChatWatchEvent(s.opts.Pubsub, chat, kind); err == nil {
 			return nil
 		} else if ctx.Err() != nil {
-			return errTaskExpectedExit
+			return errors.Join(errTaskExpectedExit, xerrors.Errorf("publishChatWatchEvent: %w", ctx.Err()))
 		}
 		timer := s.opts.Clock.NewTimer(delay, "chatworker", "watch-publish-retry")
 		select {
 		case <-timer.C:
 		case <-ctx.Done():
 			timer.Stop()
-			return errTaskExpectedExit
+			return errors.Join(errTaskExpectedExit, xerrors.Errorf("watch publish retry context done: %w", ctx.Err()))
 		}
 		timer.Stop()
 		if delay < s.opts.TaskRetryMaxBackoff {
@@ -560,8 +612,11 @@ func normalizeTaskInfrastructureError(err error, action string) error {
 	if err == nil {
 		return nil
 	}
-	if errors.Is(err, errTaskExpectedExit) || errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) {
-		return errTaskExpectedExit
+	if errors.Is(err, errTaskExpectedExit) {
+		return err
+	}
+	if errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) {
+		return errors.Join(errTaskExpectedExit, xerrors.Errorf("%s: %w", action, err))
 	}
 	return taskRetryableError{err: xerrors.Errorf("%s: %w", action, err)}
 }
@@ -570,8 +625,11 @@ func normalizeTaskTransitionError(err error, action string) error {
 	if err == nil {
 		return nil
 	}
-	if errors.Is(err, errTaskExpectedExit) || errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) {
-		return errTaskExpectedExit
+	if errors.Is(err, errTaskExpectedExit) {
+		return err
+	}
+	if errors.Is(err, chatstate.ErrChatNotFound) || errors.Is(err, sql.ErrNoRows) || errors.Is(err, context.Canceled) {
+		return errors.Join(errTaskExpectedExit, xerrors.Errorf("%s: %w", action, err))
 	}
 	if errors.Is(err, chatstate.ErrTransitionNotAllowed) || errors.Is(err, chatstate.ErrInvalidState) {
 		return xerrors.Errorf("%s: %w", action, err)

coderd/x/chatd/tasks_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"database/sql"
 	"encoding/json"
+	"fmt"
 	"sync"
 	"testing"
 	"time"
@@ -34,9 +35,11 @@ func TestRetryWrapper_ExpectedExitsDoNotRetry(t *testing.T) {
 	t.Parallel()
 
 	ctx := testutil.Context(t, testutil.WaitShort)
+	sink := testutil.NewFakeSink(t)
 	calls := 0
 	err := runTaskWithRetry(ctx, retryWrapperOptions{
 		clock:        quartz.NewMock(t),
+		logger:       sink.Logger(),
 		initialDelay: time.Second,
 		maxDelay:     time.Second,
 	}, taskKindInterrupt, func(context.Context) error {
@@ -45,6 +48,7 @@ func TestRetryWrapper_ExpectedExitsDoNotRetry(t *testing.T) {
 	})
 	require.NoError(t, err)
 	require.Equal(t, 1, calls)
+	require.Empty(t, entriesWithMessage(sink, "chatworker task retrying"))
 }
 
 func TestRetryWrapper_UnexpectedErrorsRetry(t *testing.T) {
@@ -54,11 +58,13 @@ func TestRetryWrapper_UnexpectedErrorsRetry(t *testing.T) {
 	trap := clock.Trap().NewTimer("chatworker", "task-retry-requires_action_timeout")
 	defer trap.Close()
 	ctx := testutil.Context(t, testutil.WaitLong)
+	sink := testutil.NewFakeSink(t)
 	calls := 0
 	done := make(chan error, 1)
 	go func() {
 		done <- runTaskWithRetry(ctx, retryWrapperOptions{
 			clock:        clock,
+			logger:       sink.Logger(),
 			initialDelay: time.Minute,
 			maxDelay:     time.Minute,
 		}, taskKindRequiresActionTimeout, func(context.Context) error {
@@ -74,6 +80,11 @@ func TestRetryWrapper_UnexpectedErrorsRetry(t *testing.T) {
 	clock.Advance(time.Minute).MustWait(ctx)
 	require.NoError(t, <-done)
 	require.Equal(t, 2, calls)
+	entries := entriesWithMessage(sink, "chatworker task retrying")
+	require.Len(t, entries, 1)
+	require.Equal(t, string(taskKindRequiresActionTimeout), sinkFieldValue(t, entries[0].Fields, "task_kind"))
+	require.Equal(t, time.Minute.String(), sinkFieldValue(t, entries[0].Fields, "delay"))
+	require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), "database unavailable")
 }
 
 func TestRetryWrapper_PanicsRetry(t *testing.T) {
@@ -83,11 +94,13 @@ func TestRetryWrapper_PanicsRetry(t *testing.T) {
 	trap := clock.Trap().NewTimer("chatworker", "task-retry-generation")
 	defer trap.Close()
 	ctx := testutil.Context(t, testutil.WaitLong)
+	sink := testutil.NewFakeSink(t)
 	calls := 0
 	done := make(chan error, 1)
 	go func() {
 		done <- runTaskWithRetry(ctx, retryWrapperOptions{
 			clock:        clock,
+			logger:       sink.Logger(),
 			initialDelay: time.Minute,
 			maxDelay:     time.Minute,
 		}, taskKindGeneration, func(context.Context) error {
@@ -103,6 +116,118 @@ func TestRetryWrapper_PanicsRetry(t *testing.T) {
 	clock.Advance(time.Minute).MustWait(ctx)
 	require.NoError(t, <-done)
 	require.Equal(t, 2, calls)
+	entries := entriesWithMessage(sink, "chatworker task retrying")
+	require.Len(t, entries, 1)
+	require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), "chatworker task panic: database unavailable")
+}
+
+// database/sql returns ctx.Err() from ctxDriverQuery, not
+// context.Cause(ctx). This test checks that the retry logic
+// doesn't classify such an error as an expected exit when
+// task timeout is the cause of the cancellation.
+func TestRetryWrapper_TaskTimeoutDBQueryCancellationRetries(t *testing.T) {
+	t.Parallel()
+
+	f := newTaskTestFixture(t)
+	clock := quartz.NewMock(t)
+	timeoutTrap := clock.Trap().AfterFunc("chatworker", "task-timeout-generation")
+	retryTrap := clock.Trap().NewTimer("chatworker", "task-retry-generation")
+	defer retryTrap.Close()
+	ctx := testutil.Context(t, testutil.WaitLong)
+	sink := testutil.NewFakeSink(t)
+	calls := 0
+	firstCallStarted := make(chan struct{})
+	var firstQueryErr error
+	var firstQueryCause error
+	done := make(chan error, 1)
+	go func() {
+		done <- runTaskWithRetry(ctx, retryWrapperOptions{
+			clock:        clock,
+			logger:       sink.Logger(),
+			initialDelay: time.Minute,
+			maxDelay:     time.Minute,
+		}, taskKindGeneration, func(ctx context.Context) error {
+			calls++
+			if calls == 1 {
+				close(firstCallStarted)
+				<-ctx.Done()
+				_, err := f.db.GetDatabaseNow(ctx)
+				firstQueryErr = err
+				firstQueryCause = context.Cause(ctx)
+				return normalizeTaskTransitionError(err, "db query")
+			}
+			return nil
+		})
+	}()
+
+	timeoutTrap.MustWait(ctx).MustRelease(ctx)
+	timeoutTrap.Close()
+	<-firstCallStarted
+	clock.Advance(defaultTaskTimeout).MustWait(ctx)
+	retryTrap.MustWait(ctx).MustRelease(ctx)
+	clock.Advance(time.Minute).MustWait(ctx)
+	require.NoError(t, <-done)
+	require.Equal(t, 2, calls)
+	require.ErrorIs(t, firstQueryErr, context.Canceled)
+	require.NotErrorIs(t, firstQueryErr, errTaskTimeout)
+	require.ErrorIs(t, firstQueryCause, errTaskTimeout)
+	entries := entriesWithMessage(sink, "chatworker task retrying")
+	require.Len(t, entries, 1)
+	require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), errTaskTimeout.Error())
+	require.Contains(t, sinkFieldValue(t, entries[0].Fields, "error"), context.Canceled.Error())
+}
+
+func TestRetryWrapper_ContextCancellationDoesNotRetryOrLog(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(testutil.Context(t, testutil.WaitLong))
+	cancel()
+	sink := testutil.NewFakeSink(t)
+	calls := 0
+	original := xerrors.New("database unavailable")
+	err := runTaskWithRetry(ctx, retryWrapperOptions{
+		clock:        quartz.NewMock(t),
+		logger:       sink.Logger(),
+		initialDelay: time.Second,
+		maxDelay:     time.Second,
+	}, taskKindGeneration, func(context.Context) error {
+		calls++
+		return original
+	})
+	require.NoError(t, err)
+	require.Equal(t, 1, calls)
+	require.Empty(t, entriesWithMessage(sink, "chatworker task retrying"))
+}
+
+func TestNormalizeTaskErrors_ContextCancellationIsExpectedExit(t *testing.T) {
+	t.Parallel()
+
+	err := normalizeTaskInfrastructureError(context.Canceled, "lock chat")
+	require.ErrorIs(t, err, errTaskExpectedExit)
+	require.ErrorIs(t, err, context.Canceled)
+	require.NotErrorIs(t, err, errTaskRetryable)
+	require.NotErrorIs(t, err, errTaskTimeout)
+
+	err = normalizeTaskTransitionError(context.Canceled, "commit chat")
+	require.ErrorIs(t, err, errTaskExpectedExit)
+	require.ErrorIs(t, err, context.Canceled)
+	require.NotErrorIs(t, err, errTaskRetryable)
+	require.NotErrorIs(t, err, errTaskTimeout)
+}
+
+func entriesWithMessage(sink *testutil.FakeSink, message string) []slog.SinkEntry {
+	return sink.Entries(func(e slog.SinkEntry) bool { return e.Message == message })
+}
+
+func sinkFieldValue(t *testing.T, fields slog.Map, name string) string {
+	t.Helper()
+	for _, f := range fields {
+		if f.Name == name {
+			return fmt.Sprint(f.Value)
+		}
+	}
+	t.Fatalf("missing log field %q", name)
+	return ""
 }
 
 func TestInterruptTask_FinishInterruptionOnly(t *testing.T) {