diff --git a/pool/scheduler.go b/pool/scheduler.go
index 56a4519..2d9320d 100644
--- a/pool/scheduler.go
+++ b/pool/scheduler.go
@@ -275,7 +275,10 @@ func (sched *scheduler) startJobs(ctx context.Context, fence string, jobs []*Job
 		if err != nil {
 			return fmt.Errorf("read job %q ownership: %w", job.Key, err)
 		}
-		if dispatchID == "" {
+		// A pre-existing record proves this scheduler already dispatched the
+		// key; a fresh proposal means any occupant of the key is foreign.
+		owned := dispatchID != ""
+		if !owned {
 			proposed := "scheduler-" + ulid.Make().String()
 			previous, err := sched.claimJobOwnership(ctx, fence, field, proposed)
 			if err != nil {
@@ -284,6 +287,7 @@ func (sched *scheduler) startJobs(ctx context.Context, fence string, jobs []*Job
 			dispatchID = proposed
 			if previous != "" {
 				dispatchID = previous
+				owned = true
 			}
 		}
 		dispatched := &Job{
@@ -294,16 +298,36 @@ func (sched *scheduler) startJobs(ctx context.Context, fence string, jobs []*Job
 			dispatchID: dispatchID,
 		}
 		if _, err := sched.dispatchJob(ctx, fence, dispatchID, dispatched); err != nil {
-			release := errors.Is(err, ErrJobExists)
-			if !release {
-				identity, identityErr := dispatchIdentity(job.Key, job.Payload)
-				if identityErr != nil {
-					return fmt.Errorf("encode job %q identity: %w", job.Key, identityErr)
+			if errors.Is(err, ErrJobExists) {
+				// The scheduled key is already running, which is the state
+				// this transition wanted. An occupant this scheduler
+				// dispatched keeps its ownership so later transitions retry
+				// the same exact dispatch instead of proposing a new one; a
+				// foreign occupant must never become scheduler-owned, so the
+				// speculative claim is dropped and the key is retried on the
+				// next transition. Either way the remaining planned jobs
+				// still run.
+				if owned {
+					continue
+				}
+				if releaseErr := sched.releaseOwnership(ctx, fence, field, dispatchID); releaseErr != nil {
+					return fmt.Errorf("release foreign job %q ownership: %w", job.Key, releaseErr)
 				}
-				record, readErr := sched.node.readDispatchRecord(ctx, dispatchID, identity)
-				release = readErr == nil && record.status == dispatchTerminal
+				sched.logger.Debug(
+					"scheduled job key held by a foreign job",
+					"job", job.Key,
+					"scheduler", sched.name,
+				)
+				continue
+			}
+			identity, identityErr := dispatchIdentity(job.Key, job.Payload)
+			if identityErr != nil {
+				return fmt.Errorf("encode job %q identity: %w", job.Key, identityErr)
 			}
-			if release {
+			record, readErr := sched.node.readDispatchRecord(ctx, dispatchID, identity)
+			if readErr == nil && record.status == dispatchTerminal {
+				// The run this scheduler owns already settled: drop ownership
+				// so the next transition dispatches a fresh run.
 				if releaseErr := sched.releaseOwnership(ctx, fence, field, dispatchID); releaseErr != nil {
 					return errors.Join(
 						fmt.Errorf("dispatch job %q as %q: %w", job.Key, dispatchID, err),
diff --git a/pool/scheduler_test.go b/pool/scheduler_test.go
index 7b7ad7c..ded34d4 100644
--- a/pool/scheduler_test.go
+++ b/pool/scheduler_test.go
@@ -372,14 +372,72 @@ func TestSchedulerCollisionNeverOwnsForeignJob(t *testing.T) {
 	}
 	fence := claimTestSchedulerTransition(t, ctx, sched)
 
-	err := sched.startJobs(ctx, fence, []*JobParam{{Key: "collision", Payload: []byte("scheduled")}})
-	require.ErrorIs(t, err, ErrJobExists)
-	ownership, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"collision")
+	// A foreign job already holds the key: the transition is satisfied, so it
+	// reports no error and plans the remaining jobs, but it must never take
+	// ownership of a job it did not dispatch. Repeated transitions converge on
+	// the same outcome instead of erroring every interval.
+	plan := []*JobParam{
+		{Key: "collision", Payload: []byte("scheduled")},
+		{Key: "independent", Payload: []byte("independent")},
+	}
+	for range 2 {
+		require.NoError(t, sched.startJobs(ctx, fence, plan))
+		ownership, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"collision")
+		require.NoError(t, err)
+		require.Empty(t, ownership, "a foreign job must never become scheduler-owned")
+	}
+	// The job planned after the collision still ran and is scheduler-owned.
+	independent, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"independent")
 	require.NoError(t, err)
-	require.Empty(t, ownership)
+	require.NotEmpty(t, independent, "a collision must not abort the remaining plan")
+
 	require.NoError(t, sched.clearJobs(ctx, fence))
+	require.Eventually(t, func() bool {
+		jobs := worker.Jobs()
+		return len(jobs) == 1 && string(jobs[0].Payload) == "foreign"
+	}, time.Second, time.Millisecond, "clearing schedules must not stop the foreign job")
+	require.NoError(t, node.Shutdown(ctx))
+}
+
+func TestSchedulerKeepsOwnershipWhenItsOwnJobStillRuns(t *testing.T) {
+	rdb := ptesting.NewRedisClient(t)
+	defer ptesting.CleanupRedis(t, rdb, false, "")
+	ctx := ptesting.NewTestContext(t)
+	node := newTestNode(t, ctx, rdb, ulid.Make().String())
+	worker := newTestWorker(t, ctx, node)
+	producer := newTestProducer("schedule", func() (*JobPlan, error) { return &JobPlan{}, nil })
+	encodedName := hex.EncodeToString([]byte(producer.Name()))
+	sched := &scheduler{
+		name:             node.PoolName + ":schedule",
+		interval:         20 * time.Millisecond,
+		producer:         producer,
+		node:             node,
+		keyPrefix:        encodedName + ":",
+		transitionPrefix: "=transition:" + encodedName + ":",
+		owner:            "test-" + ulid.Make().String(),
+		lease:            node.workerTTL,
+		logger:           node.logger,
+	}
+	fence := claimTestSchedulerTransition(t, ctx, sched)
+	plan := []*JobParam{{Key: "recurring", Payload: []byte("scheduled")}}
+
+	require.NoError(t, sched.startJobs(ctx, fence, plan))
+	require.Eventually(t, func() bool { return len(worker.Jobs()) == 1 }, time.Second, time.Millisecond)
+	first, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"recurring")
+	require.NoError(t, err)
+	require.NotEmpty(t, first)
+
+	// Later transitions observe their own job still running: the exact
+	// dispatch identity is preserved instead of proposing a new one, so the
+	// scheduler converges silently rather than re-dispatching every interval.
+	for range 3 {
+		require.NoError(t, sched.startJobs(ctx, fence, plan))
+		again, err := sched.schedulerOwnership(ctx, sched.keyPrefix+"recurring")
+		require.NoError(t, err)
+		require.Equal(t, first, again)
+	}
 	require.Len(t, worker.Jobs(), 1)
-	require.Equal(t, []byte("foreign"), worker.Jobs()[0].Payload)
+	require.NoError(t, sched.clearJobs(ctx, fence))
 	require.NoError(t, node.Shutdown(ctx))
 }