diff --git a/CHANGELOG.md b/CHANGELOG.md
index 6df49a17..c92b0e28 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,8 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added `EventKindJobInterrupted`, emitted when a running job is interrupted because its client is shutting down, the job was cancelled, and has been made immediately available to be worked again. [PR #1290](https://github.com/riverqueue/river/pull/1290).
 - Added `JobListParams.TagsAll` and `JobListParams.TagsAny` for filtering jobs that match every or any exact tag, respectively. [PR #1339](https://github.com/riverqueue/river/pull/1339).
 
+### Changed
+
+- Jobs that didn't finish in time organically while a client was stopping and had to have their context cancelled no longer have this cancellation counted as an error. `attempt` is reset to the number it was before the job started working, `errors` is left unchanged, and `state` is made `available` so jobs are eligible to be retried immediately. [PR #1290](https://github.com/riverqueue/river/pull/1290)
+
 ## [0.42.0] - 2026-07-31
 
 ### Added
diff --git a/client_test.go b/client_test.go
index 3c57db3d..cda5eb6b 100644
--- a/client_test.go
+++ b/client_test.go
@@ -342,6 +342,7 @@ func subscribe[TTx any](t *testing.T, client *Client[TTx]) <-chan *Event {
 		EventKindJobCancelled,
 		EventKindJobCompleted,
 		EventKindJobFailed,
+		EventKindJobInterrupted,
 		EventKindJobSnoozed,
 		EventKindQueuePaused,
 		EventKindQueueResumed,
@@ -2766,7 +2767,6 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		}))
 
 		client := runNewTestClient(ctx, t, config)
-
 		_, err := client.Insert(ctx, JobArgs{}, nil)
 		require.NoError(t, err)
 
@@ -2784,6 +2784,76 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		}
 	})
 
+	t.Run("ErroringJobGetsFreshAttempt", func(t *testing.T) {
+		t.Parallel()
+
+		config := newTestConfig(t, "")
+		config.SoftStopTimeout = 100 * time.Millisecond
+
+		firstRunDoneChan := make(chan struct{})
+		jobStartedChan := make(chan int64, 2)
+		var runCount atomic.Int32
+		AddWorker(config.Workers, WorkFunc(func(ctx context.Context, job *Job[JobArgs]) error {
+			jobStartedChan <- job.ID
+			switch runCount.Add(1) {
+			case 1:
+				<-ctx.Done()
+				close(firstRunDoneChan)
+				return ctx.Err()
+			default:
+				return errors.New("real job error")
+			}
+		}))
+
+		client := runNewTestClient(ctx, t, config)
+		subscribeChan, cancelSubscribe := client.Subscribe(EventKindJobInterrupted)
+		t.Cleanup(cancelSubscribe)
+
+		insertRes, err := client.Insert(ctx, JobArgs{}, &InsertOpts{MaxAttempts: 2})
+		require.NoError(t, err)
+
+		jobID := riversharedtest.WaitOrTimeout(t, jobStartedChan)
+		require.Equal(t, insertRes.Job.ID, jobID)
+
+		require.NoError(t, client.Stop(ctx))
+		riversharedtest.WaitOrTimeout(t, firstRunDoneChan)
+
+		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+		require.NotNil(t, event)
+		require.Equal(t, EventKindJobInterrupted, event.Kind)
+		require.Equal(t, insertRes.Job.ID, event.Job.ID)
+		require.Equal(t, rivertype.JobStateAvailable, event.Job.State)
+		require.NotNil(t, event.JobStats)
+
+		jobAfter, err := client.driver.GetExecutor().JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: jobID, Schema: client.config.Schema})
+		require.NoError(t, err)
+		require.Equal(t, 0, jobAfter.Attempt)
+		require.Nil(t, jobAfter.FinalizedAt)
+		require.Equal(t, 2, jobAfter.MaxAttempts)
+		require.Equal(t, rivertype.JobStateAvailable, jobAfter.State)
+		require.WithinDuration(t, time.Now(), jobAfter.ScheduledAt, 2*time.Second)
+		require.Empty(t, jobAfter.Errors)
+
+		require.NoError(t, client.Start(ctx))
+
+		jobID = riversharedtest.WaitOrTimeout(t, jobStartedChan)
+		require.Equal(t, insertRes.Job.ID, jobID)
+
+		var jobAfterRealError *rivertype.JobRow
+		require.Eventually(t, func() bool {
+			var err error
+			jobAfterRealError, err = client.driver.GetExecutor().JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: jobID, Schema: client.config.Schema})
+			require.NoError(t, err)
+			return jobAfterRealError.State != rivertype.JobStateRunning
+		}, 5*time.Second, 10*time.Millisecond)
+
+		require.Equal(t, 1, jobAfterRealError.Attempt)
+		require.Equal(t, rivertype.JobStateRetryable, jobAfterRealError.State)
+		require.Len(t, jobAfterRealError.Errors, 1)
+		require.Equal(t, "real job error", jobAfterRealError.Errors[0].Error)
+		require.Less(t, time.Until(jobAfterRealError.ScheduledAt), 3*time.Second)
+	})
+
 	t.Run("SoftStopSucceedsBeforeTimeout", func(t *testing.T) {
 		t.Parallel()
 
@@ -2820,7 +2890,7 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 			close(jobStartedChan)
 			<-ctx.Done()
 			close(jobDoneChan)
-			return nil
+			return ctx.Err()
 		}))
 
 		var (
@@ -2832,6 +2902,8 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 
 		client, err := NewClient(driver, config)
 		require.NoError(t, err)
+		subscribeChan, cancelSubscribe := client.Subscribe(EventKindJobInterrupted)
+		t.Cleanup(cancelSubscribe)
 
 		startCtx, startCtxCancel := context.WithCancel(ctx)
 		defer startCtxCancel()
@@ -2854,6 +2926,12 @@ func Test_Client_SoftStopTimeout(t *testing.T) {
 		default:
 			t.Fatal("expected job to have been cancelled by soft stop timeout")
 		}
+
+		event := riversharedtest.WaitOrTimeout(t, subscribeChan)
+		require.NotNil(t, event)
+		require.Equal(t, EventKindJobInterrupted, event.Kind)
+		require.Equal(t, rivertype.JobStateAvailable, event.Job.State)
+		require.NotNil(t, event.JobStats)
 	})
 }
 
diff --git a/event.go b/event.go
index b1694b88..15565357 100644
--- a/event.go
+++ b/event.go
@@ -23,6 +23,11 @@ const (
 	// differentiate each type of occurrence.
 	EventKindJobFailed EventKind = "job_failed"
 
+	// EventKindJobInterrupted occurs when a running job is interrupted because
+	// its client is shutting down and is made immediately available to be worked
+	// again. An interruption does not consume an attempt or add an attempt error.
+	EventKindJobInterrupted EventKind = "job_interrupted"
+
 	// EventKindJobSnoozed occurs when a job is snoozed.
 	EventKindJobSnoozed EventKind = "job_snoozed"
 
@@ -37,12 +42,13 @@ const (
 // exported because end users should have no way of subscribing to all known
 // kinds for forward compatibility reasons.
 var allKinds = map[EventKind]struct{}{ //nolint:gochecknoglobals
-	EventKindJobCancelled: {},
-	EventKindJobCompleted: {},
-	EventKindJobFailed:    {},
-	EventKindJobSnoozed:   {},
-	EventKindQueuePaused:  {},
-	EventKindQueueResumed: {},
+	EventKindJobCancelled:   {},
+	EventKindJobCompleted:   {},
+	EventKindJobFailed:      {},
+	EventKindJobInterrupted: {},
+	EventKindJobSnoozed:     {},
+	EventKindQueuePaused:    {},
+	EventKindQueueResumed:   {},
 }
 
 // Event wraps an event that occurred within a River client, like a job being
diff --git a/internal/jobcompleter/job_completer.go b/internal/jobcompleter/job_completer.go
index 5e6c7c74..b796a8c6 100644
--- a/internal/jobcompleter/job_completer.go
+++ b/internal/jobcompleter/job_completer.go
@@ -45,7 +45,7 @@ type SubscribeFunc func(update CompleterJobUpdated)
 type CompleterJobUpdated struct {
 	Job      *rivertype.JobRow
 	JobStats *jobstats.JobStatistics
-	Snoozed  bool
+	Reason   riverdriver.JobSetStateReason
 }
 
 type InlineCompleter struct {
@@ -104,7 +104,7 @@ func (c *InlineCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobst
 	c.subscribeCh <- []CompleterJobUpdated{{
 		Job:      jobs[0],
 		JobStats: stats,
-		Snoozed:  params.Snoozed,
+		Reason:   params.Reason,
 	}}
 
 	return nil
@@ -219,7 +219,7 @@ func (c *AsyncCompleter) JobSetStateIfRunning(ctx context.Context, stats *jobsta
 		c.subscribeCh <- []CompleterJobUpdated{{
 			Job:      jobs[0],
 			JobStats: stats,
-			Snoozed:  params.Snoozed,
+			Reason:   params.Reason,
 		}}
 
 		return nil
@@ -503,19 +503,21 @@ func (c *BatchCompleter) handleBatch(ctx context.Context) error {
 
 	var (
 		completeTime = c.Time.Now()
-		events       = make([]CompleterJobUpdated, len(jobRows))
+		events       = make([]CompleterJobUpdated, 0, len(jobRows))
 	)
-	for i, jobRow := range jobRows {
+	for _, jobRow := range jobRows {
 		setState := setStateBatch[jobRow.ID]
 		setState.Stats.CompleteDuration = completeTime.Sub(setState.StartTime)
-		events[i] = CompleterJobUpdated{
+		events = append(events, CompleterJobUpdated{
 			Job:      jobRow,
 			JobStats: setState.Stats,
-			Snoozed:  setState.Params.Snoozed,
-		}
+			Reason:   setState.Params.Reason,
+		})
 	}
 
-	c.subscribeCh <- events
+	if len(events) > 0 {
+		c.subscribeCh <- events
+	}
 
 	func() {
 		c.setStateParamsMu.Lock()
diff --git a/internal/jobcompleter/job_completer_test.go b/internal/jobcompleter/job_completer_test.go
index fc499dbf..77203933 100644
--- a/internal/jobcompleter/job_completer_test.go
+++ b/internal/jobcompleter/job_completer_test.go
@@ -326,9 +326,9 @@ func testCompleterSubscribe(t *testing.T, constructor func(schema string, exec r
 	completer.Stop() // closes subscribeChan
 
 	updates := riversharedtest.WaitOrTimeoutN(t, jobUpdateChan, 4)
-	for range 4 {
-		require.Equal(t, rivertype.JobStateCompleted, updates[0].Job.State)
-		require.False(t, updates[0].Snoozed)
+	for _, update := range updates {
+		require.Equal(t, rivertype.JobStateCompleted, update.Job.State)
+		require.Equal(t, riverdriver.JobSetStateReasonCompleted, update.Reason)
 	}
 	go completer.Stop()
 	// drain all remaining jobs
@@ -1008,6 +1008,7 @@ func testCompleter[TCompleter JobCompleter](
 			job5 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 			job6 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 			job7 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
+			job8 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 		)
 
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateCancelled(job1.ID, time.Now(), []byte("{}"), nil)))
@@ -1015,8 +1016,9 @@ func testCompleter[TCompleter JobCompleter](
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateDiscarded(job3.ID, time.Now(), []byte("{}"), nil)))
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateErrorAvailable(job4.ID, time.Now(), []byte("{}"), nil)))
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateErrorRetryable(job5.ID, time.Now(), []byte("{}"), nil)))
-		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozed(job6.ID, time.Now(), 10, nil)))
-		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job7.ID, time.Now(), 10, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateInterrupted(job6.ID, time.Now(), job6.Attempt, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozed(job7.ID, time.Now(), 10, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job8.ID, time.Now(), 10, nil)))
 
 		completer.Stop()
 
@@ -1025,8 +1027,9 @@ func testCompleter[TCompleter JobCompleter](
 		requireState(t, bundle, job3.ID, rivertype.JobStateDiscarded)
 		requireState(t, bundle, job4.ID, rivertype.JobStateAvailable)
 		requireState(t, bundle, job5.ID, rivertype.JobStateRetryable)
-		requireState(t, bundle, job6.ID, rivertype.JobStateScheduled)
-		requireState(t, bundle, job7.ID, rivertype.JobStateAvailable)
+		requireState(t, bundle, job6.ID, rivertype.JobStateAvailable)
+		requireState(t, bundle, job7.ID, rivertype.JobStateScheduled)
+		requireState(t, bundle, job8.ID, rivertype.JobStateAvailable)
 	})
 
 	t.Run("Subscription", func(t *testing.T) {
@@ -1037,23 +1040,23 @@ func testCompleter[TCompleter JobCompleter](
 		var (
 			job1 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 			job2 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
+			job3 = testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{Schema: bundle.schema, State: ptrutil.Ptr(rivertype.JobStateRunning)})
 		)
 
 		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateCompleted(job1.ID, time.Now(), nil)))
-		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job2.ID, time.Now(), job2.Attempt, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateInterrupted(job2.ID, time.Now(), job2.Attempt, nil)))
+		require.NoError(t, completer.JobSetStateIfRunning(ctx, &jobstats.JobStatistics{}, riverdriver.JobSetStateSnoozedAvailable(job3.ID, time.Now(), job3.Attempt, nil)))
 
 		completer.Stop()
 
 		// Unfortunately we have to do this awkward loop to wait for all updates
 		// because updates are sent through the channel as batches. The sync and
-		// async completer don't work in batches and therefore send batches of
-		// one item each while the batch completer will send both. Put
-		// otherwise, expect the sync and async completers to iterate this loop
-		// twice and batch completer to iterate it once.
+		// async completers send batches of one item each, while the batch
+		// completer may group updates.
 		var jobUpdates []CompleterJobUpdated
 		for {
 			jobUpdates = append(jobUpdates, riversharedtest.WaitOrTimeout(t, bundle.subscribeCh)...)
-			if len(jobUpdates) >= 2 {
+			if len(jobUpdates) >= 3 {
 				break
 			}
 		}
@@ -1070,10 +1073,13 @@ func testCompleter[TCompleter JobCompleter](
 
 		job1Update := findUpdate(job1.ID)
 		require.Equal(t, rivertype.JobStateCompleted, job1Update.Job.State)
-		require.False(t, job1Update.Snoozed)
+		require.Equal(t, riverdriver.JobSetStateReasonCompleted, job1Update.Reason)
 		job2Update := findUpdate(job2.ID)
 		require.Equal(t, rivertype.JobStateAvailable, job2Update.Job.State)
-		require.True(t, job2Update.Snoozed)
+		require.Equal(t, riverdriver.JobSetStateReasonInterrupted, job2Update.Reason)
+		job3Update := findUpdate(job3.ID)
+		require.Equal(t, rivertype.JobStateAvailable, job3Update.Job.State)
+		require.Equal(t, riverdriver.JobSetStateReasonSnoozed, job3Update.Reason)
 	})
 
 	t.Run("MultipleCycles", func(t *testing.T) {
diff --git a/internal/jobexecutor/job_executor.go b/internal/jobexecutor/job_executor.go
index c6680014..adaa6265 100644
--- a/internal/jobexecutor/job_executor.go
+++ b/internal/jobexecutor/job_executor.go
@@ -18,6 +18,7 @@ import (
 	"github.com/riverqueue/river/internal/jobcompleter"
 	"github.com/riverqueue/river/internal/jobstats"
 	"github.com/riverqueue/river/internal/pluginlookup"
+	"github.com/riverqueue/river/internal/rivercommon"
 	"github.com/riverqueue/river/internal/workunit"
 	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/rivershared/baseservice"
@@ -450,6 +451,7 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 		cancelJob bool
 		cancelErr *rivertype.JobCancelError
 	)
+	softStopped := isSoftStopCancelError(ctx, res.Err)
 
 	logAttrs := []any{
 		slog.String("error", res.ErrorStr()),
@@ -461,6 +463,8 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 	case errors.As(res.Err, &cancelErr):
 		cancelJob = true
 		e.Logger.DebugContext(ctx, e.Name+": Job cancelled explicitly", logAttrs...)
+	case softStopped:
+		e.Logger.InfoContext(ctx, e.Name+": Job stopped due to client shutdown; retrying", logAttrs...)
 	case res.Err != nil:
 		if jobRow.Attempt >= jobRow.MaxAttempts {
 			e.Logger.InfoContext(ctx, e.Name+": Job errored", logAttrs...)
@@ -471,11 +475,21 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 		e.Logger.InfoContext(ctx, e.Name+": Job panicked", logAttrs...)
 	}
 
-	if e.ErrorHandler != nil && !cancelJob {
+	if e.ErrorHandler != nil && !cancelJob && !softStopped {
 		// Error handlers also have an opportunity to cancel the job.
 		cancelJob = e.invokeErrorHandler(ctx, res)
 	}
 
+	now := e.Time.Now()
+
+	if softStopped {
+		params := riverdriver.JobSetStateInterrupted(jobRow.ID, now, max(jobRow.Attempt-1, 0), metadataUpdates)
+		if err := e.Completer.JobSetStateIfRunning(ctx, e.stats, params); err != nil {
+			e.Logger.ErrorContext(ctx, e.Name+": Failed to make soft-stopped job available", logAttrs...)
+		}
+		return
+	}
+
 	attemptErr := rivertype.AttemptError{
 		At:      e.start,
 		Attempt: jobRow.Attempt,
@@ -489,8 +503,6 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 		return
 	}
 
-	now := e.Time.Now()
-
 	if cancelJob {
 		if err := e.Completer.JobSetStateIfRunning(ctx, e.stats, riverdriver.JobSetStateCancelled(jobRow.ID, now, errData, metadataUpdates)); err != nil {
 			e.Logger.ErrorContext(ctx, e.Name+": Failed to cancel job and report error", logAttrs...)
@@ -539,6 +551,15 @@ func (e *JobExecutor) reportError(ctx context.Context, jobRow *rivertype.JobRow,
 	}
 }
 
+// isSoftStopCancelError reports whether a worker returned because the client
+// was stopping and cancelled its job context. The context cause distinguishes
+// client stop cancellation from ordinary worker cancellation or timeouts.
+func isSoftStopCancelError(ctx context.Context, err error) bool {
+	return err != nil &&
+		errors.Is(context.Cause(ctx), rivercommon.ErrStop) &&
+		(errors.Is(err, context.Canceled) || errors.Is(err, rivercommon.ErrStop))
+}
+
 type withJobsAndErrorsByID interface {
 	ErrorsByID() map[int64]error
 	Jobs() []*rivertype.JobRow
diff --git a/internal/jobexecutor/job_executor_test.go b/internal/jobexecutor/job_executor_test.go
index 4890bc42..6f5f02c4 100644
--- a/internal/jobexecutor/job_executor_test.go
+++ b/internal/jobexecutor/job_executor_test.go
@@ -362,6 +362,44 @@ func TestJobExecutor_Execute(t *testing.T) {
 		require.Equal(t, rivertype.JobStateDiscarded, job.State)
 	})
 
+	// "Decrements attempt" means restoring attempt to the value it had before
+	// JobGetAvailable incremented it for this run.
+	t.Run("SoftStopCancelMakesJobAvailableAndDecrementsAttempt", func(t *testing.T) {
+		t.Parallel()
+
+		executor, bundle := setup(t)
+
+		bundle.jobRow.Attempt = bundle.jobRow.MaxAttempts
+		_, err := bundle.exec.JobUpdateFull(ctx, &riverdriver.JobUpdateFullParams{
+			ID:              bundle.jobRow.ID,
+			AttemptDoUpdate: true,
+			Attempt:         bundle.jobRow.Attempt,
+		})
+		require.NoError(t, err)
+
+		workCtx, cancel := context.WithCancelCause(ctx)
+		cancel(rivercommon.ErrStop)
+
+		executor.WorkUnit = newWorkUnitFactoryWithCustomRetry(func() error { return context.Canceled }, nil).MakeUnit(bundle.jobRow)
+
+		executor.Execute(workCtx)
+		updates := riversharedtest.WaitOrTimeout(t, bundle.updateCh)
+		require.Len(t, updates, 1)
+		require.Equal(t, riverdriver.JobSetStateReasonInterrupted, updates[0].Reason)
+
+		job, err := bundle.exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{
+			ID:     bundle.jobRow.ID,
+			Schema: "",
+		})
+		require.NoError(t, err)
+		require.Nil(t, job.FinalizedAt)
+		require.Equal(t, bundle.jobRow.Attempt-1, job.Attempt)
+		require.Equal(t, bundle.jobRow.MaxAttempts, job.MaxAttempts)
+		require.Equal(t, rivertype.JobStateAvailable, job.State)
+		require.WithinDuration(t, time.Now(), job.ScheduledAt, 2*time.Second)
+		require.Empty(t, job.Errors)
+	})
+
 	t.Run("JobCancelErrorCancelsJobEvenWithRemainingAttempts", func(t *testing.T) {
 		t.Parallel()
 
diff --git a/riverdriver/river_driver_interface.go b/riverdriver/river_driver_interface.go
index 6ab9e812..d4b87f31 100644
--- a/riverdriver/river_driver_interface.go
+++ b/riverdriver/river_driver_interface.go
@@ -563,12 +563,32 @@ type JobSetStateIfRunningParams struct {
 	FinalizedAt     *time.Time
 	MetadataDoMerge bool
 	MetadataUpdates []byte
+	Reason          JobSetStateReason
 	ScheduledAt     *time.Time
 	Schema          string // added by completer
-	Snoozed         bool
 	State           rivertype.JobState
 }
 
+// JobSetStateReason describes why a running job's state was changed. It lets
+// callers distinguish transitions that result in the same state, like a failed
+// job being retried and a job interrupted by client shutdown both becoming
+// available.
+type JobSetStateReason string
+
+const (
+	// JobSetStateReasonCancelled indicates that a job was cancelled.
+	JobSetStateReasonCancelled JobSetStateReason = "cancelled"
+	// JobSetStateReasonCompleted indicates that a job completed successfully.
+	JobSetStateReasonCompleted JobSetStateReason = "completed"
+	// JobSetStateReasonFailed indicates that a job failed.
+	JobSetStateReasonFailed JobSetStateReason = "failed"
+	// JobSetStateReasonInterrupted indicates that a job was interrupted by
+	// client shutdown and made available to be worked again.
+	JobSetStateReasonInterrupted JobSetStateReason = "interrupted"
+	// JobSetStateReasonSnoozed indicates that a job was snoozed.
+	JobSetStateReasonSnoozed JobSetStateReason = "snoozed"
+)
+
 func JobSetStateCancelled(id int64, finalizedAt time.Time, errData []byte, metadataUpdates []byte) *JobSetStateIfRunningParams {
 	return &JobSetStateIfRunningParams{
 		ID:              id,
@@ -576,6 +596,7 @@ func JobSetStateCancelled(id int64, finalizedAt time.Time, errData []byte, metad
 		MetadataDoMerge: len(metadataUpdates) > 0,
 		MetadataUpdates: metadataUpdates,
 		FinalizedAt:     &finalizedAt,
+		Reason:          JobSetStateReasonCancelled,
 		State:           rivertype.JobStateCancelled,
 	}
 }
@@ -586,6 +607,7 @@ func JobSetStateCompleted(id int64, finalizedAt time.Time, metadataUpdates []byt
 		ID:              id,
 		MetadataDoMerge: len(metadataUpdates) > 0,
 		MetadataUpdates: metadataUpdates,
+		Reason:          JobSetStateReasonCompleted,
 		State:           rivertype.JobStateCompleted,
 	}
 }
@@ -597,16 +619,19 @@ func JobSetStateDiscarded(id int64, finalizedAt time.Time, errData []byte, metad
 		MetadataDoMerge: len(metadataUpdates) > 0,
 		MetadataUpdates: metadataUpdates,
 		FinalizedAt:     &finalizedAt,
+		Reason:          JobSetStateReasonFailed,
 		State:           rivertype.JobStateDiscarded,
 	}
 }
 
+// JobSetStateErrorAvailable makes an errored job immediately available.
 func JobSetStateErrorAvailable(id int64, scheduledAt time.Time, errData []byte, metadataUpdates []byte) *JobSetStateIfRunningParams {
 	return &JobSetStateIfRunningParams{
 		ID:              id,
 		ErrData:         errData,
 		MetadataDoMerge: len(metadataUpdates) > 0,
 		MetadataUpdates: metadataUpdates,
+		Reason:          JobSetStateReasonFailed,
 		ScheduledAt:     &scheduledAt,
 		State:           rivertype.JobStateAvailable,
 	}
@@ -618,19 +643,34 @@ func JobSetStateErrorRetryable(id int64, scheduledAt time.Time, errData []byte,
 		ErrData:         errData,
 		MetadataDoMerge: len(metadataUpdates) > 0,
 		MetadataUpdates: metadataUpdates,
+		Reason:          JobSetStateReasonFailed,
 		ScheduledAt:     &scheduledAt,
 		State:           rivertype.JobStateRetryable,
 	}
 }
 
+// JobSetStateInterrupted makes a job that was interrupted by client shutdown
+// immediately available without recording an error.
+func JobSetStateInterrupted(id int64, scheduledAt time.Time, attempt int, metadataUpdates []byte) *JobSetStateIfRunningParams {
+	return &JobSetStateIfRunningParams{
+		Attempt:         &attempt,
+		ID:              id,
+		MetadataDoMerge: len(metadataUpdates) > 0,
+		MetadataUpdates: metadataUpdates,
+		Reason:          JobSetStateReasonInterrupted,
+		ScheduledAt:     &scheduledAt,
+		State:           rivertype.JobStateAvailable,
+	}
+}
+
 func JobSetStateSnoozed(id int64, scheduledAt time.Time, attempt int, metadataUpdates []byte) *JobSetStateIfRunningParams {
 	return &JobSetStateIfRunningParams{
 		Attempt:         &attempt,
 		ID:              id,
 		MetadataDoMerge: len(metadataUpdates) > 0,
 		MetadataUpdates: metadataUpdates,
+		Reason:          JobSetStateReasonSnoozed,
 		ScheduledAt:     &scheduledAt,
-		Snoozed:         true,
 		State:           rivertype.JobStateScheduled,
 	}
 }
@@ -641,8 +681,8 @@ func JobSetStateSnoozedAvailable(id int64, scheduledAt time.Time, attempt int, m
 		ID:              id,
 		MetadataDoMerge: len(metadataUpdates) > 0,
 		MetadataUpdates: metadataUpdates,
+		Reason:          JobSetStateReasonSnoozed,
 		ScheduledAt:     &scheduledAt,
-		Snoozed:         true,
 		State:           rivertype.JobStateAvailable,
 	}
 }
diff --git a/riverdriver/river_driver_interface_test.go b/riverdriver/river_driver_interface_test.go
index 978e299c..c885db43 100644
--- a/riverdriver/river_driver_interface_test.go
+++ b/riverdriver/river_driver_interface_test.go
@@ -25,7 +25,7 @@ func TestJobSetStateCancelled(t *testing.T) {
 		require.Nil(t, result.MetadataUpdates)
 		require.False(t, result.MetadataDoMerge)
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonCancelled, result.Reason)
 		require.Equal(t, rivertype.JobStateCancelled, result.State)
 	})
 
@@ -44,7 +44,7 @@ func TestJobSetStateCancelled(t *testing.T) {
 		require.Equal(t, metadata, result.MetadataUpdates)
 		require.True(t, result.MetadataDoMerge)
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonCancelled, result.Reason)
 		require.Equal(t, rivertype.JobStateCancelled, result.State)
 	})
 }
@@ -65,7 +65,7 @@ func TestJobSetStateCompleted(t *testing.T) {
 		require.False(t, result.MetadataDoMerge)
 		require.Nil(t, result.MetadataUpdates)
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonCompleted, result.Reason)
 		require.Equal(t, rivertype.JobStateCompleted, result.State)
 	})
 
@@ -82,7 +82,7 @@ func TestJobSetStateCompleted(t *testing.T) {
 		require.True(t, result.MetadataDoMerge)
 		require.Equal(t, metadata, result.MetadataUpdates)
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonCompleted, result.Reason)
 		require.Equal(t, rivertype.JobStateCompleted, result.State)
 	})
 }
@@ -104,7 +104,7 @@ func TestJobSetStateDiscarded(t *testing.T) {
 		require.False(t, result.MetadataDoMerge)
 		require.Nil(t, result.MetadataUpdates)
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonFailed, result.Reason)
 		require.Equal(t, rivertype.JobStateDiscarded, result.State)
 	})
 
@@ -123,7 +123,7 @@ func TestJobSetStateDiscarded(t *testing.T) {
 		require.Equal(t, metadata, result.MetadataUpdates)
 		require.True(t, result.MetadataDoMerge)
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonFailed, result.Reason)
 		require.Equal(t, rivertype.JobStateDiscarded, result.State)
 	})
 }
@@ -139,13 +139,14 @@ func TestJobSetStateErrorAvailable(t *testing.T) {
 		errData := []byte("error available")
 		result := JobSetStateErrorAvailable(id, scheduledAt, errData, nil)
 		require.Equal(t, id, result.ID)
+		require.Nil(t, result.Attempt)
 		require.Equal(t, errData, result.ErrData)
 		require.False(t, result.MetadataDoMerge)
 		require.Nil(t, result.MetadataUpdates)
 		require.NotNil(t, result.ScheduledAt)
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonFailed, result.Reason)
 		require.Equal(t, rivertype.JobStateAvailable, result.State)
 	})
 
@@ -158,12 +159,13 @@ func TestJobSetStateErrorAvailable(t *testing.T) {
 		metadata := []byte(`{"key": "value"}`)
 		result := JobSetStateErrorAvailable(id, scheduledAt, errData, metadata)
 		require.Equal(t, id, result.ID)
+		require.Nil(t, result.Attempt)
 		require.True(t, result.MetadataDoMerge)
 		require.Equal(t, metadata, result.MetadataUpdates)
 		require.NotNil(t, result.ScheduledAt)
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonFailed, result.Reason)
 		require.Equal(t, errData, result.ErrData)
 	})
 }
@@ -185,7 +187,7 @@ func TestJobSetStateErrorRetryable(t *testing.T) {
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Equal(t, errData, result.ErrData)
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonFailed, result.Reason)
 		require.Equal(t, rivertype.JobStateRetryable, result.State)
 	})
 
@@ -203,11 +205,56 @@ func TestJobSetStateErrorRetryable(t *testing.T) {
 		require.NotNil(t, result.ScheduledAt)
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Empty(t, result.Schema)
-		require.False(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonFailed, result.Reason)
 		require.Equal(t, errData, result.ErrData)
 	})
 }
 
+func TestJobSetStateInterrupted(t *testing.T) {
+	t.Parallel()
+
+	t.Run("EmptyMetadata", func(t *testing.T) {
+		t.Parallel()
+
+		id := int64(6)
+		scheduledAt := time.Now().Truncate(time.Second)
+		attempt := 2
+		result := JobSetStateInterrupted(id, scheduledAt, attempt, nil)
+		require.Equal(t, id, result.ID)
+		require.NotNil(t, result.Attempt)
+		require.Equal(t, attempt, *result.Attempt)
+		require.Nil(t, result.ErrData)
+		require.False(t, result.MetadataDoMerge)
+		require.Nil(t, result.MetadataUpdates)
+		require.Equal(t, JobSetStateReasonInterrupted, result.Reason)
+		require.NotNil(t, result.ScheduledAt)
+		require.True(t, result.ScheduledAt.Equal(scheduledAt))
+		require.Empty(t, result.Schema)
+		require.Equal(t, rivertype.JobStateAvailable, result.State)
+	})
+
+	t.Run("NonEmptyMetadata", func(t *testing.T) {
+		t.Parallel()
+
+		id := int64(6)
+		scheduledAt := time.Now().Truncate(time.Second)
+		attempt := 2
+		metadata := []byte("interrupted metadata")
+		result := JobSetStateInterrupted(id, scheduledAt, attempt, metadata)
+		require.Equal(t, id, result.ID)
+		require.NotNil(t, result.Attempt)
+		require.Equal(t, attempt, *result.Attempt)
+		require.Nil(t, result.ErrData)
+		require.True(t, result.MetadataDoMerge)
+		require.Equal(t, metadata, result.MetadataUpdates)
+		require.Equal(t, JobSetStateReasonInterrupted, result.Reason)
+		require.NotNil(t, result.ScheduledAt)
+		require.True(t, result.ScheduledAt.Equal(scheduledAt))
+		require.Empty(t, result.Schema)
+		require.Equal(t, rivertype.JobStateAvailable, result.State)
+	})
+}
+
 func TestJobSetStateSnoozed(t *testing.T) { //nolint:dupl
 	t.Parallel()
 
@@ -226,7 +273,7 @@ func TestJobSetStateSnoozed(t *testing.T) { //nolint:dupl
 		require.NotNil(t, result.ScheduledAt)
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Empty(t, result.Schema)
-		require.True(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonSnoozed, result.Reason)
 		require.Equal(t, rivertype.JobStateScheduled, result.State)
 	})
 
@@ -245,7 +292,7 @@ func TestJobSetStateSnoozed(t *testing.T) { //nolint:dupl
 		require.NotNil(t, result.ScheduledAt)
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Empty(t, result.Schema)
-		require.True(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonSnoozed, result.Reason)
 		require.Equal(t, rivertype.JobStateScheduled, result.State)
 	})
 }
@@ -281,7 +328,7 @@ func TestJobSetStateSnoozedAvailable(t *testing.T) { //nolint:dupl
 		require.NotNil(t, result.ScheduledAt)
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Empty(t, result.Schema)
-		require.True(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonSnoozed, result.Reason)
 		require.Equal(t, rivertype.JobStateAvailable, result.State)
 	})
 
@@ -301,7 +348,7 @@ func TestJobSetStateSnoozedAvailable(t *testing.T) { //nolint:dupl
 		require.NotNil(t, result.ScheduledAt)
 		require.True(t, result.ScheduledAt.Equal(scheduledAt))
 		require.Empty(t, result.Schema)
-		require.True(t, result.Snoozed)
+		require.Equal(t, JobSetStateReasonSnoozed, result.Reason)
 		require.Equal(t, rivertype.JobStateAvailable, result.State)
 	})
 }
diff --git a/riverdriver/riverdrivertest/driver_client_test.go b/riverdriver/riverdrivertest/driver_client_test.go
index 7081f042..9c9839dc 100644
--- a/riverdriver/riverdrivertest/driver_client_test.go
+++ b/riverdriver/riverdrivertest/driver_client_test.go
@@ -212,6 +212,7 @@ func subscribe[TTx any](t *testing.T, client *river.Client[TTx]) <-chan *river.E
 		river.EventKindJobCancelled,
 		river.EventKindJobCompleted,
 		river.EventKindJobFailed,
+		river.EventKindJobInterrupted,
 		river.EventKindJobSnoozed,
 		river.EventKindQueuePaused,
 		river.EventKindQueueResumed,
diff --git a/riverdriver/riverdrivertest/job_update.go b/riverdriver/riverdrivertest/job_update.go
index 5c1f6a9a..e354ae31 100644
--- a/riverdriver/riverdrivertest/job_update.go
+++ b/riverdriver/riverdrivertest/job_update.go
@@ -698,6 +698,40 @@ func exerciseJobUpdate[TTx any](ctx context.Context, t *testing.T, executorWithT
 			require.Equal(t, "foo.go:123\nbar.go:456", jobAfter.Errors[0].Trace)
 		})
 
+		t.Run("SetsAnInterruptedRunningJobToAvailableWithUpdatedAttempt", func(t *testing.T) {
+			t.Parallel()
+
+			exec, _ := setup(ctx, t)
+
+			now := time.Now().UTC()
+			attempt := 2
+
+			job := testfactory.Job(ctx, t, exec, &testfactory.JobOpts{
+				Attempt:     ptrutil.Ptr(3),
+				MaxAttempts: ptrutil.Ptr(3),
+				State:       ptrutil.Ptr(rivertype.JobStateRunning),
+				UniqueKey:   []byte("unique-key"),
+			})
+
+			params := riverdriver.JobSetStateInterrupted(job.ID, now, attempt, nil)
+			jobsAfter, err := exec.JobSetStateIfRunningMany(ctx, setStateManyParams(params))
+			require.NoError(t, err)
+			jobAfter := jobsAfter[0]
+			require.Equal(t, attempt, jobAfter.Attempt)
+			require.Equal(t, rivertype.JobStateAvailable, jobAfter.State)
+			require.Equal(t, 3, jobAfter.MaxAttempts)
+			require.WithinDuration(t, now, jobAfter.ScheduledAt, time.Microsecond)
+
+			jobUpdated, err := exec.JobGetByID(ctx, &riverdriver.JobGetByIDParams{ID: job.ID, Schema: ""})
+			require.NoError(t, err)
+			require.Equal(t, attempt, jobUpdated.Attempt)
+			require.Equal(t, rivertype.JobStateAvailable, jobUpdated.State)
+			require.Equal(t, 3, jobUpdated.MaxAttempts)
+			require.Equal(t, "unique-key", string(jobUpdated.UniqueKey))
+
+			require.Empty(t, jobAfter.Errors)
+		})
+
 		t.Run("DoesNotTouchAlreadyRetryableJobWithNoMetadataUpdates", func(t *testing.T) {
 			t.Parallel()
 
diff --git a/rivertest/worker.go b/rivertest/worker.go
index 414f846b..d39ad8f9 100644
--- a/rivertest/worker.go
+++ b/rivertest/worker.go
@@ -252,24 +252,19 @@ func completerResultToWorkResult(tb testing.TB, completerResult jobcompleter.Com
 	tb.Helper()
 
 	var kind river.EventKind
-	if completerResult.Snoozed {
+	switch completerResult.Reason {
+	case riverdriver.JobSetStateReasonCancelled:
+		kind = river.EventKindJobCancelled
+	case riverdriver.JobSetStateReasonCompleted:
+		kind = river.EventKindJobCompleted
+	case riverdriver.JobSetStateReasonFailed:
+		kind = river.EventKindJobFailed
+	case riverdriver.JobSetStateReasonInterrupted:
+		kind = river.EventKindJobInterrupted
+	case riverdriver.JobSetStateReasonSnoozed:
 		kind = river.EventKindJobSnoozed
-	} else {
-		switch completerResult.Job.State {
-		case rivertype.JobStateCancelled:
-			kind = river.EventKindJobCancelled
-		case rivertype.JobStateCompleted:
-			kind = river.EventKindJobCompleted
-		case rivertype.JobStateScheduled:
-			kind = river.EventKindJobSnoozed
-		case rivertype.JobStateAvailable, rivertype.JobStateDiscarded, rivertype.JobStateRetryable, rivertype.JobStateRunning:
-			kind = river.EventKindJobFailed
-		case rivertype.JobStatePending:
-			panic("test worker internal error: completion subscriber unexpectedly received job in pending state, river bug")
-		default:
-			// linter exhaustive rule prevents this from being reached
-			panic("test worker internal error: unreachable state to distribute, river bug")
-		}
+	default:
+		panic(fmt.Sprintf("test worker internal error: completion subscriber unexpectedly received reason %q, river bug", completerResult.Reason))
 	}
 
 	return &WorkResult{
diff --git a/subscription_manager.go b/subscription_manager.go
index c74ef4d4..7bcb226c 100644
--- a/subscription_manager.go
+++ b/subscription_manager.go
@@ -9,6 +9,7 @@ import (
 
 	"github.com/riverqueue/river/internal/jobcompleter"
 	"github.com/riverqueue/river/internal/jobstats"
+	"github.com/riverqueue/river/riverdriver"
 	"github.com/riverqueue/river/rivershared/baseservice"
 	"github.com/riverqueue/river/rivershared/startstop"
 	"github.com/riverqueue/river/rivershared/util/sliceutil"
@@ -142,7 +143,7 @@ func (sm *subscriptionManager) distributeJobUpdates(ctx context.Context, updates
 	}
 
 	for _, update := range updates {
-		sm.distributeJobEvent(ctx, update.Job, jobStatisticsFromInternal(update.JobStats), update.Snoozed)
+		sm.distributeJobEvent(ctx, update.Job, jobStatisticsFromInternal(update.JobStats), update.Reason)
 	}
 }
 
@@ -152,27 +153,23 @@ func (sm *subscriptionManager) distributeJobUpdates(ctx context.Context, updates
 // the queue.
 //
 // MUST be called with sm.mu already held.
-func (sm *subscriptionManager) distributeJobEvent(ctx context.Context, job *rivertype.JobRow, stats *JobStatistics, snoozed bool) {
-	var event *Event
-	if snoozed {
-		event = &Event{Kind: EventKindJobSnoozed, Job: job, JobStats: stats}
-	} else {
-		switch job.State {
-		case rivertype.JobStateCancelled:
-			event = &Event{Kind: EventKindJobCancelled, Job: job, JobStats: stats}
-		case rivertype.JobStateCompleted:
-			event = &Event{Kind: EventKindJobCompleted, Job: job, JobStats: stats}
-		case rivertype.JobStateAvailable, rivertype.JobStateDiscarded, rivertype.JobStateRetryable, rivertype.JobStateRunning:
-			event = &Event{Kind: EventKindJobFailed, Job: job, JobStats: stats}
-		case rivertype.JobStatePending, rivertype.JobStateScheduled:
-			// job state may be set to scheduled, but only for snoozed jobs, so
-			// the case at the top should always take precedence before this
-			panic(fmt.Sprintf("completion subscriber unexpectedly received job in %s state, river bug", job.State))
-		default:
-			// linter exhaustive rule prevents this from being reached
-			panic("unreachable state to distribute, river bug")
-		}
+func (sm *subscriptionManager) distributeJobEvent(ctx context.Context, job *rivertype.JobRow, stats *JobStatistics, reason riverdriver.JobSetStateReason) {
+	var kind EventKind
+	switch reason {
+	case riverdriver.JobSetStateReasonCancelled:
+		kind = EventKindJobCancelled
+	case riverdriver.JobSetStateReasonCompleted:
+		kind = EventKindJobCompleted
+	case riverdriver.JobSetStateReasonFailed:
+		kind = EventKindJobFailed
+	case riverdriver.JobSetStateReasonInterrupted:
+		kind = EventKindJobInterrupted
+	case riverdriver.JobSetStateReasonSnoozed:
+		kind = EventKindJobSnoozed
+	default:
+		panic(fmt.Sprintf("completion subscriber unexpectedly received reason %q, river bug", reason))
 	}
+	event := &Event{Kind: kind, Job: job, JobStats: stats}
 
 	// All subscription channels are non-blocking so this is always fast and
 	// there's no risk of falling behind what producers are sending.
diff --git a/subscription_manager_test.go b/subscription_manager_test.go
index 57d4a133..02156dc9 100644
--- a/subscription_manager_test.go
+++ b/subscription_manager_test.go
@@ -60,7 +60,7 @@ func Test_SubscriptionManager(t *testing.T) {
 		manager, bundle := setup(t)
 		t.Cleanup(func() { close(bundle.subscribeCh) })
 
-		sub, cancelSub := manager.SubscribeConfig(&SubscribeConfig{ChanSize: 10, Kinds: []EventKind{EventKindJobCompleted, EventKindJobSnoozed}})
+		sub, cancelSub := manager.SubscribeConfig(&SubscribeConfig{ChanSize: 10, Kinds: []EventKind{EventKindJobCompleted, EventKindJobInterrupted, EventKindJobSnoozed}})
 		t.Cleanup(cancelSub)
 
 		// Send some events
@@ -68,6 +68,7 @@ func Test_SubscriptionManager(t *testing.T) {
 		job2 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateCancelled), FinalizedAt: ptrutil.Ptr(time.Now())})
 		job3 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateRetryable)})
 		job4 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateScheduled)})
+		job5 := testfactory.Job(ctx, t, bundle.exec, &testfactory.JobOpts{State: ptrutil.Ptr(rivertype.JobStateAvailable)})
 
 		makeStats := func(complete, wait, run time.Duration) *jobstats.JobStatistics {
 			return &jobstats.JobStatistics{
@@ -78,25 +79,43 @@ func Test_SubscriptionManager(t *testing.T) {
 		}
 
 		bundle.subscribeCh <- []jobcompleter.CompleterJobUpdated{
-			{Job: job1, JobStats: makeStats(101, 102, 103)}, // completed, should be sent
-			{Job: job2, JobStats: makeStats(201, 202, 203)}, // cancelled, should be skipped
+			{Job: job1, JobStats: makeStats(101, 102, 103), Reason: riverdriver.JobSetStateReasonCompleted}, // completed, should be sent
+			{Job: job2, JobStats: makeStats(201, 202, 203), Reason: riverdriver.JobSetStateReasonCancelled}, // cancelled, should be skipped
 		}
 		bundle.subscribeCh <- []jobcompleter.CompleterJobUpdated{
-			{Job: job3, JobStats: makeStats(301, 302, 303)},                // retryable, should be skipped
-			{Job: job4, JobStats: makeStats(401, 402, 403), Snoozed: true}, // snoozed/scheduled, should be sent
+			{Job: job3, JobStats: makeStats(301, 302, 303), Reason: riverdriver.JobSetStateReasonFailed},      // retryable, should be skipped
+			{Job: job4, JobStats: makeStats(401, 402, 403), Reason: riverdriver.JobSetStateReasonSnoozed},     // snoozed/scheduled, should be sent
+			{Job: job5, JobStats: makeStats(501, 502, 503), Reason: riverdriver.JobSetStateReasonInterrupted}, // interrupted/available, should be sent
 		}
 
-		received := riversharedtest.WaitOrTimeoutN(t, sub, 2)
+		received := riversharedtest.WaitOrTimeoutN(t, sub, 3)
+		require.Equal(t, EventKindJobCompleted, received[0].Kind)
 		require.Equal(t, job1.ID, received[0].Job.ID)
 		require.Equal(t, rivertype.JobStateCompleted, received[0].Job.State)
 		require.Equal(t, time.Duration(101), received[0].JobStats.CompleteDuration)
 		require.Equal(t, time.Duration(102), received[0].JobStats.QueueWaitDuration)
 		require.Equal(t, time.Duration(103), received[0].JobStats.RunDuration)
+		require.Equal(t, EventKindJobSnoozed, received[1].Kind)
 		require.Equal(t, job4.ID, received[1].Job.ID)
 		require.Equal(t, rivertype.JobStateScheduled, received[1].Job.State)
 		require.Equal(t, time.Duration(401), received[1].JobStats.CompleteDuration)
 		require.Equal(t, time.Duration(402), received[1].JobStats.QueueWaitDuration)
 		require.Equal(t, time.Duration(403), received[1].JobStats.RunDuration)
+		require.Equal(t, EventKindJobInterrupted, received[2].Kind)
+		require.Equal(t, job5.ID, received[2].Job.ID)
+		require.Equal(t, rivertype.JobStateAvailable, received[2].Job.State)
+		require.Equal(t, time.Duration(501), received[2].JobStats.CompleteDuration)
+		require.Equal(t, time.Duration(502), received[2].JobStats.QueueWaitDuration)
+		require.Equal(t, time.Duration(503), received[2].JobStats.RunDuration)
+
+		manager.statsMu.Lock()
+		statsAggregate := manager.statsAggregate
+		statsNumJobs := manager.statsNumJobs
+		manager.statsMu.Unlock()
+		require.Equal(t, 5, statsNumJobs)
+		require.Equal(t, time.Duration(101+201+301+401+501), statsAggregate.CompleteDuration)
+		require.Equal(t, time.Duration(102+202+302+402+502), statsAggregate.QueueWaitDuration)
+		require.Equal(t, time.Duration(103+203+303+403+503), statsAggregate.RunDuration)
 
 		cancelSub()
 		select {