diff --git a/agent/dispatcher.go b/agent/dispatcher.go
new file mode 100644
index 0000000..9f99a57
--- /dev/null
+++ b/agent/dispatcher.go
@@ -0,0 +1,200 @@
+package agent
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"time"
+
+	"github.com/mbertschler/squirrel/config"
+	"github.com/mbertschler/squirrel/store"
+)
+
+// defaultMaxParallelSyncs bounds how many automatic syncs run at once
+// across all destinations. Per-destination serialisation already caps
+// parallelism at the number of configured destinations; this is the
+// overall ceiling so a host with many destinations can't spawn one rclone
+// per destination simultaneously and thrash its uplink and CPU. Four
+// covers the reference household — a NAS pushing photos/docs/media to
+// cloudbox + s3archive + kopia-mirror + htpc — without a LAN pair ever
+// queueing behind a slow offsite, while still bounding a pathological
+// config.
+const defaultMaxParallelSyncs = 4
+
+// syncDispatcher runs (volume, destination) syncs off the scheduler's tick
+// goroutine so a slow or wedged destination can never delay the tick loop,
+// index runs, peer syncs, or syncs to other destinations (#160, F25).
+//
+// The concurrency unit is the destination: each destination has a FIFO
+// queue drained by a single worker, so at most one sync per destination is
+// in flight while a second volume due to the same destination (photos and
+// docs both push to cloudbox in the reference setup) waits its turn instead
+// of being skipped and starved — issue #160 option 4b is per-destination
+// queues, not skip-on-busy. An overall semaphore bounds how many
+// destinations transfer at once.
+//
+// The pre-sync index ordering the scheduler guarantees is unaffected: the
+// tick body still runs a volume's index to completion before it hands that
+// volume's due syncs to the dispatcher.
+type syncDispatcher struct {
+	run    SyncRunner
+	logger *slog.Logger
+	now    func() time.Time
+
+	sem chan struct{}
+
+	mu    sync.Mutex
+	dests map[string]*destQueue
+	wg    sync.WaitGroup
+}
+
+// destQueue is one destination's serial pipeline: at most one sync active,
+// the rest waiting in FIFO order. queued holds every volume name active or
+// pending on this destination so a re-dispatch of a pair already in the
+// pipeline — the scheduler re-evaluates a still-unfinished pair on every
+// tick — is a quiet no-op rather than a duplicate enqueue.
+type destQueue struct {
+	pending []*config.Volume
+	queued  map[string]bool
+	active  bool
+}
+
+func newSyncDispatcher(run SyncRunner, logger *slog.Logger, now func() time.Time, maxParallel int) *syncDispatcher {
+	if maxParallel <= 0 {
+		maxParallel = defaultMaxParallelSyncs
+	}
+	return &syncDispatcher{
+		run:    run,
+		logger: logger,
+		now:    now,
+		sem:    make(chan struct{}, maxParallel),
+		dests:  make(map[string]*destQueue),
+	}
+}
+
+// dispatch enqueues the sync for (vol, destName) on destName's FIFO queue.
+// An idle destination starts its worker immediately; a busy one takes the
+// pair onto its queue to run when the current transfer finishes (deduped so
+// re-evaluation on later ticks can't queue the same pair twice). It never
+// blocks on the transfer — the tick loop returns at once, so a slow
+// destination cannot delay any other.
+func (d *syncDispatcher) dispatch(ctx context.Context, vol *config.Volume, destName string) {
+	if d.run == nil {
+		d.logger.Info("scheduler.skipped",
+			"kind", "sync", "volume", vol.Name, "destination", destName,
+			"reason", "sync runner not configured")
+		return
+	}
+	d.mu.Lock()
+	q := d.dests[destName]
+	if q == nil {
+		q = &destQueue{queued: make(map[string]bool)}
+		d.dests[destName] = q
+	}
+	if q.queued[vol.Name] {
+		d.mu.Unlock() // already active or queued for this destination
+		return
+	}
+	q.queued[vol.Name] = true
+	if q.active {
+		q.pending = append(q.pending, vol)
+		d.mu.Unlock()
+		return
+	}
+	q.active = true
+	d.mu.Unlock()
+	d.wg.Add(1)
+	go d.worker(ctx, destName, vol)
+}
+
+// worker drains destName's queue one sync at a time, so the destination
+// never has two transfers in flight. It exits — freeing the destination —
+// when the queue empties; on shutdown (ctx cancelled) it drains without
+// running, leaving the unfinished pairs to be re-evaluated next start.
+func (d *syncDispatcher) worker(ctx context.Context, destName string, vol *config.Volume) {
+	defer d.wg.Done()
+	for vol != nil {
+		if ctx.Err() == nil {
+			d.runOne(ctx, vol, destName)
+		}
+		vol = d.next(destName, vol.Name)
+	}
+}
+
+// inFlight reports whether (destName, volName) is currently active or queued
+// on destName's pipeline. The scheduler consults it alongside the runs table
+// so a pair whose dispatch from an earlier tick is still working — but whose
+// run row is not yet visible — is recognised as in flight and not dragged
+// through a redundant pre-sync index every tick (#160). Safe for concurrent
+// use; a nil-runner dispatcher (no dests) always reports false.
+func (d *syncDispatcher) inFlight(destName, volName string) bool {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	q := d.dests[destName]
+	if q == nil {
+		return false
+	}
+	return q.queued[volName]
+}
+
+// next removes the just-finished volume from destName's pipeline and returns
+// the next queued volume, or nil (dropping the now-idle destination) when
+// none remain.
+func (d *syncDispatcher) next(destName, done string) *config.Volume {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	q := d.dests[destName]
+	if q == nil {
+		return nil
+	}
+	delete(q.queued, done)
+	if len(q.pending) == 0 {
+		delete(d.dests, destName)
+		return nil
+	}
+	next := q.pending[0]
+	q.pending = q.pending[1:]
+	return next
+}
+
+// runOne takes an overall parallelism slot (or bails on shutdown), runs the
+// sync, and emits the kicked/finished/error logs. A sync that stalls is
+// failed by the runner's transfer timeout (rclone is killed); its
+// diagnosable error lands here and in the run row.
+func (d *syncDispatcher) runOne(ctx context.Context, vol *config.Volume, destName string) {
+	select {
+	case d.sem <- struct{}{}:
+		defer func() { <-d.sem }()
+	case <-ctx.Done():
+		return
+	}
+	if ctx.Err() != nil {
+		return
+	}
+	d.logger.Info("scheduler.kicked",
+		"kind", "sync", "volume", vol.Name, "destination", destName)
+	start := d.now()
+	rep := d.run(ctx, vol, destName)
+	duration := d.now().Sub(start)
+	status := rep.Status
+	if status == "" {
+		status = store.RunStatusFailed
+	}
+	d.logger.Info("scheduler.finished",
+		"kind", "sync", "volume", vol.Name, "destination", destName,
+		"run_id", rep.RunID, "status", status,
+		"duration_ms", duration.Milliseconds(),
+	)
+	if rep.Err != nil {
+		d.logger.Error("scheduler.error",
+			"kind", "sync", "volume", vol.Name, "destination", destName,
+			"run_id", rep.RunID, "err", rep.Err.Error())
+	}
+}
+
+// wait blocks until every destination worker has drained. The scheduler
+// calls it during shutdown, after the run context is cancelled (which kills
+// any in-flight rclone child), so it returns promptly.
+func (d *syncDispatcher) wait() {
+	d.wg.Wait()
+}
diff --git a/agent/dispatcher_test.go b/agent/dispatcher_test.go
new file mode 100644
index 0000000..c795341
--- /dev/null
+++ b/agent/dispatcher_test.go
@@ -0,0 +1,151 @@
+package agent
+
+import (
+	"bytes"
+	"context"
+	"log/slog"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/mbertschler/squirrel/config"
+	"github.com/mbertschler/squirrel/store"
+)
+
+func discardLogger() *slog.Logger { return slog.New(slog.DiscardHandler) }
+
+// TestSyncDispatcherRunsDestinationsConcurrently proves the core F25 fix:
+// two destinations run at the same time, so a slow offsite can never delay
+// a LAN pair. Both runner invocations must be in flight before either is
+// released.
+func TestSyncDispatcherRunsDestinationsConcurrently(t *testing.T) {
+	var active, maxActive atomic.Int32
+	entered := make(chan struct{}, 2)
+	release := make(chan struct{})
+	run := func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		n := active.Add(1)
+		for {
+			old := maxActive.Load()
+			if n <= old || maxActive.CompareAndSwap(old, n) {
+				break
+			}
+		}
+		entered <- struct{}{}
+		<-release
+		active.Add(-1)
+		return SyncRunReport{Status: store.RunStatusSuccess}
+	}
+	d := newSyncDispatcher(run, discardLogger(), time.Now, 4)
+	vol := &config.Volume{Name: "photos"}
+	d.dispatch(context.Background(), vol, "cloudbox")
+	d.dispatch(context.Background(), vol, "htpc")
+	<-entered // both must be running before we release either
+	<-entered
+	close(release)
+	d.wait()
+	if got := maxActive.Load(); got != 2 {
+		t.Fatalf("max concurrent syncs = %d; want 2 (different destinations run concurrently)", got)
+	}
+}
+
+// TestSyncDispatcherSerializesSameDestination proves the concurrency unit
+// is the destination: two volumes due to the same destination on one tick
+// both run — the second is queued behind the first, never starved — while
+// at most one is ever in flight against that destination.
+func TestSyncDispatcherSerializesSameDestination(t *testing.T) {
+	var active, maxActive atomic.Int32
+	var photosRan, docsRan atomic.Bool
+	entered := make(chan string, 2)
+	release := make(chan struct{}) // one send per sync, in completion order
+	run := func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		n := active.Add(1)
+		for {
+			old := maxActive.Load()
+			if n <= old || maxActive.CompareAndSwap(old, n) {
+				break
+			}
+		}
+		switch vol.Name {
+		case "photos":
+			photosRan.Store(true)
+		case "docs":
+			docsRan.Store(true)
+		}
+		entered <- vol.Name
+		<-release
+		active.Add(-1)
+		return SyncRunReport{Status: store.RunStatusSuccess}
+	}
+	d := newSyncDispatcher(run, discardLogger(), time.Now, 4)
+	d.dispatch(context.Background(), &config.Volume{Name: "photos"}, "cloudbox")
+	d.dispatch(context.Background(), &config.Volume{Name: "docs"}, "cloudbox")
+
+	first := <-entered // FIFO: photos was dispatched first
+	if first != "photos" {
+		t.Fatalf("first sync = %q; want photos (FIFO order)", first)
+	}
+	time.Sleep(20 * time.Millisecond) // give a wrongly-concurrent docs a chance to start
+	if got := active.Load(); got != 1 {
+		t.Fatalf("concurrent syncs to one destination = %d; want 1 (serialized)", got)
+	}
+	release <- struct{}{} // let photos finish; the worker then dequeues docs
+	if second := <-entered; second != "docs" {
+		t.Fatalf("second sync = %q; want docs (queued, not starved)", second)
+	}
+	release <- struct{}{} // let docs finish
+	d.wait()
+
+	if !photosRan.Load() || !docsRan.Load() {
+		t.Fatalf("both volumes must run to the shared destination; photos=%v docs=%v",
+			photosRan.Load(), docsRan.Load())
+	}
+	if got := maxActive.Load(); got != 1 {
+		t.Fatalf("max concurrent syncs to one destination = %d; want 1", got)
+	}
+}
+
+// TestSyncDispatcherBoundsOverallParallelism proves the semaphore ceiling:
+// with a single slot, a second destination waits for the first to free the
+// slot rather than running immediately.
+func TestSyncDispatcherBoundsOverallParallelism(t *testing.T) {
+	var bCalled atomic.Bool
+	aEntered := make(chan struct{})
+	release := make(chan struct{})
+	run := func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		if destName == "a" {
+			close(aEntered)
+			<-release
+		} else {
+			bCalled.Store(true)
+		}
+		return SyncRunReport{Status: store.RunStatusSuccess}
+	}
+	d := newSyncDispatcher(run, discardLogger(), time.Now, 1)
+	vol := &config.Volume{Name: "v"}
+	d.dispatch(context.Background(), vol, "a")
+	<-aEntered // a holds the only slot
+	d.dispatch(context.Background(), vol, "b")
+	time.Sleep(20 * time.Millisecond) // give b's worker time to reach the slot wait
+	if bCalled.Load() {
+		t.Fatal("b ran while the single parallelism slot was held by a")
+	}
+	close(release)
+	d.wait()
+	if !bCalled.Load() {
+		t.Fatal("b never ran after the slot was freed")
+	}
+}
+
+// TestSyncDispatcherNilRunnerSkips covers the index-only host: no runner
+// wired, so a dispatch logs a skip and starts no worker.
+func TestSyncDispatcherNilRunnerSkips(t *testing.T) {
+	buf := &bytes.Buffer{}
+	logger := slog.New(slog.NewTextHandler(buf, nil))
+	d := newSyncDispatcher(nil, logger, time.Now, 4)
+	d.dispatch(context.Background(), &config.Volume{Name: "v"}, "dest")
+	d.wait()
+	if !strings.Contains(buf.String(), "sync runner not configured") {
+		t.Fatalf("expected a nil-runner skip log, got:\n%s", buf.String())
+	}
+}
diff --git a/agent/scheduler.go b/agent/scheduler.go
index 17d3883..8524404 100644
--- a/agent/scheduler.go
+++ b/agent/scheduler.go
@@ -30,7 +30,7 @@ const DefaultSchedulerTick = 30 * time.Second
 type scheduler struct {
 	store     *store.Store
 	volumes   map[string]*config.Volume
-	syncRun   SyncRunner
+	dispatch  *syncDispatcher
 	logger    *slog.Logger
 	locks     lockHolder
 	tickEvery time.Duration
@@ -60,7 +60,7 @@ func newScheduler(srv *Server, tickEvery time.Duration, now func() time.Time) *s
 	return &scheduler{
 		store:     srv.store,
 		volumes:   srv.cfg.Volumes,
-		syncRun:   srv.cfg.SyncRunner,
+		dispatch:  newSyncDispatcher(srv.cfg.SyncRunner, srv.cfg.Logger, now, defaultMaxParallelSyncs),
 		logger:    srv.cfg.Logger,
 		locks:     srv.router,
 		tickEvery: tickEvery,
@@ -103,9 +103,12 @@ func (s *scheduler) run(ctx context.Context) {
 	for {
 		select {
 		case <-ctx.Done():
-			// Drain in-flight hooks before returning so Serve's shutdown
-			// wait doesn't race a hook goroutine writing its outcome.
+			// Drain in-flight hooks and per-destination sync workers before
+			// returning so Serve's shutdown wait doesn't race a goroutine
+			// writing its outcome. ctx is already cancelled, so any rclone
+			// child has been killed and the workers return promptly.
 			s.hooks.wait()
+			s.dispatch.wait()
 			return
 		case <-t.C:
 			s.tick(ctx)
@@ -231,42 +234,87 @@ func (s *scheduler) resolveVolume(ctx context.Context, name, absPath string) (st
 	return created, nil
 }
 
-// maybeRunSync evaluates the sync_every cadence and, if any destination
-// is due, runs a pre-sync index and then the per-destination syncs.
-// Returns true when the scheduler took any sync-related action (kicked,
-// skipped, or errored) so the caller can suppress a redundant
-// standalone-index check.
+// maybeRunSync evaluates the sync_every cadence and, for the destinations
+// that are due and not already syncing, runs a pre-sync index and then hands
+// each pair to the per-destination dispatcher. Returns true when the
+// scheduler took any sync-related action (kicked, skipped, or errored) so the
+// caller can suppress a redundant standalone-index check.
 //
 // Invariant: a scheduled sync always runs an index immediately before
-// pushing. If the pre-sync index fails or is skipped, no syncs run
-// this tick — the next tick re-evaluates.
+// pushing. If the pre-sync index fails or is skipped, no syncs run this tick
+// — the next tick re-evaluates.
+//
+// A due destination whose (volume, destination) sync from an earlier tick is
+// still in flight is dropped before the pre-sync index runs. Dispatch is
+// asynchronous (#160), so without this the tick loop would re-index the source
+// on every 30s tick for the entire duration of a slow multi-hour push —
+// burning I/O and flooding the never-pruned runs audit trail with a
+// kind='index' row per tick. When every due destination is already syncing,
+// the pre-sync index is skipped entirely: there is no new push for it to
+// precede, and the in-flight pushes were each already preceded by their own.
 func (s *scheduler) maybeRunSync(ctx context.Context, vol *config.Volume, volumeID int64) bool {
 	if vol.SyncEvery == 0 {
 		return false
 	}
-	var dueDests []string
+	dueDests := s.dueSyncDests(ctx, vol, volumeID)
+	if len(dueDests) == 0 {
+		return false
+	}
+	dispatchable := s.dispatchableSyncDests(ctx, vol, volumeID, dueDests)
+	if len(dispatchable) == 0 {
+		// Every due destination is already syncing (each skip logged per
+		// pair). Report the sync action so the standalone-index branch stays
+		// suppressed — otherwise its own re-index would reintroduce the exact
+		// per-tick flood this guard exists to prevent.
+		return true
+	}
+	if !s.maybeRunIndex(ctx, vol, volumeID, "pre-sync", 0) {
+		return true
+	}
+	for _, destName := range dispatchable {
+		s.dispatch.dispatch(ctx, vol, destName)
+	}
+	return true
+}
+
+// dueSyncDests returns the destinations on the volume's sync_to whose
+// sync_every cadence has elapsed. A per-destination lookup error is logged and
+// that destination is dropped from this tick's evaluation (the next tick
+// re-evaluates), never aborting the others.
+func (s *scheduler) dueSyncDests(ctx context.Context, vol *config.Volume, volumeID int64) []string {
+	var due []string
 	for _, destName := range vol.SyncTo {
-		due, err := s.syncDue(ctx, volumeID, destName, vol.SyncEvery)
+		ok, err := s.syncDue(ctx, volumeID, destName, vol.SyncEvery)
 		if err != nil {
 			s.logger.Error("scheduler.error",
 				"kind", "sync", "volume", vol.Name, "destination", destName,
 				"err", err.Error())
 			continue
 		}
-		if due {
-			dueDests = append(dueDests, destName)
+		if ok {
+			due = append(due, destName)
 		}
 	}
-	if len(dueDests) == 0 {
-		return false
-	}
-	if !s.maybeRunIndex(ctx, vol, volumeID, "pre-sync", 0) {
-		return true
-	}
+	return due
+}
+
+// dispatchableSyncDests filters dueDests to the pairs the dispatcher can start
+// now, dropping any whose sync is already in flight and logging the same
+// per-pair scheduler.skipped the serial scheduler emitted — so an operator
+// still sees why a due destination isn't moving, it just no longer drags a
+// redundant pre-sync index along with it.
+func (s *scheduler) dispatchableSyncDests(ctx context.Context, vol *config.Volume, volumeID int64, dueDests []string) []string {
+	var dispatchable []string
 	for _, destName := range dueDests {
-		s.runSync(ctx, vol, volumeID, destName)
+		if s.syncInFlight(ctx, vol, volumeID, destName) {
+			s.logger.Info("scheduler.skipped",
+				"kind", "sync", "volume", vol.Name, "destination", destName,
+				"reason", "in-flight sync run")
+			continue
+		}
+		dispatchable = append(dispatchable, destName)
 	}
-	return true
+	return dispatchable
 }
 
 // syncDue computes (now - last_finished) ≥ cadence for the (volume,
@@ -460,54 +508,28 @@ func indexRunStatus(rep index.Report, err error) (string, bool) {
 	return store.RunStatusSuccess, true
 }
 
-// runSync kicks one (volume, destination) sync via the configured
-// SyncRunner. The pre-check against the runs table is the in-flight
-// detection the issue specifies; the SyncRunner's downstream call
-// (sync.RunPair) adds its own atomic BeginSyncRunIfClear gate, so a
-// race that sneaks past the pre-check still produces a clean skip
-// rather than a duplicate run.
-//
-// A nil SyncRunner surfaces as a clean skip rather than an error so
-// an agent running pure index-only schedules can omit the sync wiring
-// (and the rclone dependency that comes with it) without the
-// scheduler logging at error level on each tick.
-func (s *scheduler) runSync(ctx context.Context, vol *config.Volume, volumeID int64, destName string) {
+// syncInFlight reports whether a sync of this exact (volume, destination) pair
+// is already active — a stale row, a concurrent CLI `squirrel sync`, or (the
+// common case) a dispatch from an earlier tick still working. It consults two
+// sources so neither a not-yet-visible run row nor a just-drained queue slot
+// can fool it: the dispatcher's in-memory queue (set the instant a pair is
+// enqueued, before its run row exists) and the runs table (authoritative once
+// the row is written, and the signal that also catches a stale row or a
+// concurrent CLI sync). The downstream dispatch and sync.RunPair's atomic
+// BeginSyncRunIfClear gate remain the final guard against a duplicate run, so
+// this check only needs to be good enough to suppress a redundant pre-sync
+// index. A store error is treated as in-flight so a transient DB hiccup skips
+// this tick rather than racing a duplicate push.
+func (s *scheduler) syncInFlight(ctx context.Context, vol *config.Volume, volumeID int64, destName string) bool {
+	if s.dispatch.inFlight(destName, vol.Name) {
+		return true
+	}
 	running, err := s.store.HasRunningRun(ctx, store.RunKindSync, volumeID, destName)
 	if err != nil {
 		s.logger.Error("scheduler.error",
 			"kind", "sync", "volume", vol.Name, "destination", destName,
 			"err", err.Error())
-		return
-	}
-	if running {
-		s.logger.Info("scheduler.skipped",
-			"kind", "sync", "volume", vol.Name, "destination", destName,
-			"reason", "in-flight sync run")
-		return
-	}
-	if s.syncRun == nil {
-		s.logger.Info("scheduler.skipped",
-			"kind", "sync", "volume", vol.Name, "destination", destName,
-			"reason", "sync runner not configured")
-		return
-	}
-	s.logger.Info("scheduler.kicked",
-		"kind", "sync", "volume", vol.Name, "destination", destName)
-	start := s.now()
-	rep := s.syncRun(ctx, vol, destName)
-	duration := s.now().Sub(start)
-	status := rep.Status
-	if status == "" {
-		status = store.RunStatusFailed
-	}
-	s.logger.Info("scheduler.finished",
-		"kind", "sync", "volume", vol.Name, "destination", destName,
-		"run_id", rep.RunID, "status", status,
-		"duration_ms", duration.Milliseconds(),
-	)
-	if rep.Err != nil {
-		s.logger.Error("scheduler.error",
-			"kind", "sync", "volume", vol.Name, "destination", destName,
-			"run_id", rep.RunID, "err", rep.Err.Error())
+		return true
 	}
+	return running
 }
diff --git a/agent/scheduler_test.go b/agent/scheduler_test.go
index 9a4a8d6..617a9a9 100644
--- a/agent/scheduler_test.go
+++ b/agent/scheduler_test.go
@@ -124,10 +124,14 @@ func (f *fakeSyncRunner) Calls() []syncCall {
 // + fake sync runner. The scheduler's tick method drives behaviour
 // directly; the goroutine-driven run() loop is only used in the
 // integration test below.
-func newSchedulerFixture(t *testing.T, volumeCfg *config.Volume) *schedulerFixture {
+func newSchedulerFixture(t *testing.T, vols ...*config.Volume) *schedulerFixture {
 	t.Helper()
-	if volumeCfg.Path == "" {
-		volumeCfg.Path = t.TempDir()
+	volumes := make(map[string]*config.Volume, len(vols))
+	for _, v := range vols {
+		if v.Path == "" {
+			v.Path = t.TempDir()
+		}
+		volumes[v.Name] = v
 	}
 	dbPath := filepath.Join(t.TempDir(), "index.db")
 	s, err := store.Open(dbPath)
@@ -144,7 +148,7 @@ func newSchedulerFixture(t *testing.T, volumeCfg *config.Volume) *schedulerFixtu
 		Listen:     "127.0.0.1:0",
 		Token:      "tok",
 		Version:    "test",
-		Volumes:    map[string]*config.Volume{volumeCfg.Name: volumeCfg},
+		Volumes:    volumes,
 		Logger:     logger,
 		SyncRunner: syncRunner.Runner(),
 	}, s)
@@ -174,7 +178,7 @@ func (f *schedulerFixture) scheduler() *scheduler {
 	return &scheduler{
 		store:     f.srv.store,
 		volumes:   f.srv.cfg.Volumes,
-		syncRun:   f.srv.cfg.SyncRunner,
+		dispatch:  newSyncDispatcher(f.srv.cfg.SyncRunner, f.srv.cfg.Logger, f.clock.Now, defaultMaxParallelSyncs),
 		logger:    f.srv.cfg.Logger,
 		locks:     f.srv.router,
 		tickEvery: time.Second,
@@ -194,7 +198,6 @@ func (f *schedulerFixture) seedFile() {
 		if err := os.WriteFile(p, []byte("x"), 0o644); err != nil {
 			f.t.Fatalf("write %s: %v", p, err)
 		}
-		return
 	}
 }
 
@@ -310,6 +313,7 @@ func TestSchedulerSyncRunsPreSyncIndexFirst(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx)
+	sch.dispatch.wait() // syncs run on per-destination workers; drain them
 
 	runs, err := f.store.ListRuns(ctx, store.ListRunsOpts{})
 	if err != nil {
@@ -412,9 +416,12 @@ func TestSchedulerSkipsWhenInFlightIndexRun(t *testing.T) {
 	}
 }
 
-// TestSchedulerSkipsWhenInFlightSyncRun plants a stale running sync
-// row and asserts the scheduler skips the sync (but, per the
-// invariant, the pre-sync index still ran in this tick). We use a
+// TestSchedulerSkipsWhenInFlightSyncRun plants a stale running sync row and
+// asserts the scheduler skips the sync — and, since that pair is the volume's
+// only due destination, skips the pre-sync index too (#160). Re-indexing the
+// source on every tick for the whole duration of an in-flight push would burn
+// I/O and flood the never-pruned runs audit trail with a kind='index' row per
+// tick; the in-flight push was already preceded by its own index. We use a
 // fake sync runner so we can verify it never gets invoked.
 func TestSchedulerSkipsWhenInFlightSyncRun(t *testing.T) {
 	f := newSchedulerFixture(t, &config.Volume{
@@ -437,6 +444,7 @@ func TestSchedulerSkipsWhenInFlightSyncRun(t *testing.T) {
 
 	sch := f.scheduler()
 	sch.tick(context.Background())
+	sch.dispatch.wait()
 
 	if got := len(f.syncLog.Calls()); got != 0 {
 		t.Fatalf("sync invoked %d times despite running sync row; want 0", got)
@@ -444,25 +452,47 @@ func TestSchedulerSkipsWhenInFlightSyncRun(t *testing.T) {
 	if !f.containsLogLine("scheduler.skipped", "kind=sync", "in-flight") {
 		t.Fatalf("missing in-flight sync skip log:\n%s", f.logBuf.String())
 	}
-	// Pre-sync index still ran — the invariant says sync always
-	// indexes first, and the only thing skipped here is the sync
-	// itself.
+	// No new pre-sync index: with the sole due destination already in
+	// flight, there is no new push for an index to precede. Only the planted
+	// 'running' sync row should exist.
 	runs, _ := f.store.ListRuns(context.Background(), store.ListRunsOpts{})
-	var sawIndex, sawSync bool
-	for _, r := range runs {
-		switch r.Kind {
-		case store.RunKindIndex:
-			sawIndex = true
-		case store.RunKindSync:
-			sawSync = true
-		}
+	if len(runs) != 1 || runs[0].Kind != store.RunKindSync || runs[0].Status != store.RunStatusRunning {
+		t.Fatalf("runs=%+v; want only the planted running sync row (no redundant pre-sync index)", runs)
 	}
-	if !sawIndex {
-		t.Fatalf("expected a pre-sync index row even when sync is gated: %+v", runs)
+}
+
+// TestSchedulerTwoVolumesSameDestinationBothRun guards the starvation trap
+// (#160): the tick iterates volumes by name, so if two volumes are due to
+// the same destination on one tick the earlier one wins the destination.
+// The dispatcher must queue the later one behind it and run it — not skip
+// it every tick forever. "aaa" sorts before "zzz"; both must produce a sync
+// run to "backup".
+func TestSchedulerTwoVolumesSameDestinationBothRun(t *testing.T) {
+	f := newSchedulerFixture(t,
+		&config.Volume{Name: "aaa", SyncTo: []string{"backup"}, SyncEvery: time.Hour},
+		&config.Volume{Name: "zzz", SyncTo: []string{"backup"}, SyncEvery: time.Hour},
+	)
+	f.seedFile()
+	sch := f.scheduler()
+
+	sch.tick(context.Background())
+	sch.dispatch.wait()
+
+	var aaaRan, zzzRan bool
+	for _, c := range f.syncLog.Calls() {
+		if c.Destination != "backup" {
+			continue
+		}
+		switch c.Volume {
+		case "aaa":
+			aaaRan = true
+		case "zzz":
+			zzzRan = true
+		}
 	}
-	if !sawSync {
-		// The planted 'running' sync row counts as a sync; just sanity.
-		t.Fatalf("expected the planted running sync row to remain: %+v", runs)
+	if !aaaRan || !zzzRan {
+		t.Fatalf("both volumes must sync to the shared destination; aaa=%v zzz=%v (calls=%+v)",
+			aaaRan, zzzRan, f.syncLog.Calls())
 	}
 }
 
@@ -517,6 +547,7 @@ func TestSchedulerCadenceUsesLastFinished(t *testing.T) {
 
 	// Tick 1: nothing prior → fires.
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if got := len(f.syncLog.Calls()); got != 1 {
 		t.Fatalf("after tick 1: sync calls = %d; want 1", got)
 	}
@@ -524,6 +555,7 @@ func TestSchedulerCadenceUsesLastFinished(t *testing.T) {
 	// Advance 30min, still under cadence → does not fire.
 	f.clock.Add(30 * time.Minute)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if got := len(f.syncLog.Calls()); got != 1 {
 		t.Fatalf("after tick 2 (under cadence): sync calls = %d; want 1", got)
 	}
@@ -531,11 +563,157 @@ func TestSchedulerCadenceUsesLastFinished(t *testing.T) {
 	// Advance another hour, past cadence → fires again.
 	f.clock.Add(time.Hour)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if got := len(f.syncLog.Calls()); got != 2 {
 		t.Fatalf("after tick 3 (over cadence): sync calls = %d; want 2", got)
 	}
 }
 
+// waitUntil polls cond until it returns true or a short deadline passes,
+// failing the test on timeout. It synchronises with the dispatcher's async
+// workers without a fixed sleep, so a test can wait for one destination's
+// queue to drain while another destination's sync is still (deliberately)
+// blocked.
+func waitUntil(t *testing.T, cond func() bool, what string) {
+	t.Helper()
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(time.Millisecond)
+	}
+	t.Fatalf("timed out waiting for %s", what)
+}
+
+// TestSchedulerInFlightSyncNotReindexedEachTick is the #160 regression guard:
+// once a sync is dispatched and stays in flight across ticks, the scheduler
+// must NOT re-run that volume's pre-sync index every tick — otherwise a slow
+// multi-hour push would flood the never-pruned runs audit trail with a
+// kind='index' row per 30s tick. The pre-sync index runs exactly once (on the
+// tick that dispatches the sync), while a different, not-in-flight
+// (volume, destination) keeps indexing and dispatching normally on the very
+// ticks the wedged one is skipped — proving the guard is per-pair, not global.
+func TestSchedulerInFlightSyncNotReindexedEachTick(t *testing.T) {
+	f := newSchedulerFixture(t,
+		&config.Volume{Name: "photos", SyncTo: []string{"cloudbox"}, SyncEvery: time.Hour},
+		&config.Volume{Name: "docs", SyncTo: []string{"backup"}, SyncEvery: time.Hour},
+	)
+	f.seedFile()
+
+	release := make(chan struct{})
+	photosInFlight := make(chan struct{}, 1)
+	docsFinished := make(chan struct{}, 8)
+	f.syncLog.runFn = func(ctx context.Context, vol *config.Volume, destName string) SyncRunReport {
+		v, err := f.store.GetVolumeByName(ctx, vol.Name)
+		if err != nil {
+			return SyncRunReport{Err: err}
+		}
+		id, blocker, err := f.store.BeginSyncRunIfClear(ctx, store.SyncRunSpec{
+			VolumeID: v.ID, Destination: destName,
+		})
+		if err != nil {
+			return SyncRunReport{Err: err}
+		}
+		if blocker != nil {
+			return SyncRunReport{RunID: blocker.ID, Err: fmt.Errorf("already running")}
+		}
+		if vol.Name == "photos" {
+			// Announce the running row exists, then block — keeping the
+			// photos->cloudbox sync in flight across the following ticks
+			// without ever finishing it (the F25 wedge, minus the kill).
+			photosInFlight <- struct{}{}
+			<-release
+		}
+		if err := f.store.FinishRun(ctx, id, store.RunStatusSuccess, "", 0); err != nil {
+			return SyncRunReport{RunID: id, Err: err}
+		}
+		if vol.Name == "docs" {
+			docsFinished <- struct{}{}
+		}
+		return SyncRunReport{RunID: id, Status: store.RunStatusSuccess}
+	}
+
+	sch := f.scheduler()
+	ctx := context.Background()
+
+	countIndex := func(volName string) int {
+		t.Helper()
+		v, err := f.store.GetVolumeByName(ctx, volName)
+		if err != nil {
+			t.Fatalf("GetVolumeByName(%s): %v", volName, err)
+		}
+		runs, err := f.store.ListRuns(ctx, store.ListRunsOpts{})
+		if err != nil {
+			t.Fatalf("ListRuns: %v", err)
+		}
+		n := 0
+		for _, r := range runs {
+			if r.Kind == store.RunKindIndex && r.VolumeID.Valid && r.VolumeID.Int64 == v.ID {
+				n++
+			}
+		}
+		return n
+	}
+	// drainDocs waits for docs' just-dispatched sync to finish and its queue
+	// slot to clear, so the next tick sees docs as not-in-flight while photos
+	// stays wedged.
+	drainDocs := func(tick int) {
+		<-docsFinished
+		waitUntil(t, func() bool { return !sch.dispatch.inFlight("backup", "docs") },
+			fmt.Sprintf("docs queue to drain after tick %d", tick))
+	}
+
+	// Tick 1: both due, neither in flight -> each indexes once and dispatches.
+	// photos' sync blocks (stays running); docs' sync finishes.
+	sch.tick(ctx)
+	<-photosInFlight
+	drainDocs(1)
+	if got := countIndex("photos"); got != 1 {
+		t.Fatalf("after tick 1: photos index runs = %d; want 1", got)
+	}
+	if got := countIndex("docs"); got != 1 {
+		t.Fatalf("after tick 1: docs index runs = %d; want 1", got)
+	}
+
+	// Ticks 2 and 3: photos is still in flight -> skipped with NO new
+	// pre-sync index. docs comes due again each hour and must keep indexing +
+	// dispatching normally on the very ticks photos is skipped.
+	for tick := 2; tick <= 3; tick++ {
+		f.clock.Add(time.Hour + time.Minute)
+		sch.tick(ctx)
+		drainDocs(tick)
+		if got := countIndex("photos"); got != 1 {
+			t.Fatalf("after tick %d: photos index runs = %d; want 1 (in-flight sync must not be re-indexed)", tick, got)
+		}
+		if got := countIndex("docs"); got != tick {
+			t.Fatalf("after tick %d: docs index runs = %d; want %d (not-in-flight volume indexes each due tick)", tick, got, tick)
+		}
+	}
+
+	var photosCalls, docsCalls int
+	for _, c := range f.syncLog.Calls() {
+		switch c.Volume {
+		case "photos":
+			photosCalls++
+		case "docs":
+			docsCalls++
+		}
+	}
+	if photosCalls != 1 {
+		t.Fatalf("photos sync calls = %d; want 1 (dispatched once, then skipped while in flight)", photosCalls)
+	}
+	if docsCalls != 3 {
+		t.Fatalf("docs sync calls = %d; want 3 (once per due tick)", docsCalls)
+	}
+	if !f.containsLogLine("scheduler.skipped", "kind=sync", "volume=photos", "in-flight") {
+		t.Fatalf("missing in-flight skip log for photos:\n%s", f.logBuf.String())
+	}
+
+	close(release)
+	sch.dispatch.wait()
+}
+
 // TestSchedulerFailedSyncStillConsumesCadence guards the failure
 // policy ("Failed runs aren't specially retried — the next tick
 // re-evaluates"). After a failed sync, the next within-cadence tick
@@ -562,6 +740,7 @@ func TestSchedulerFailedSyncStillConsumesCadence(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if calls.Load() != 1 {
 		t.Fatalf("tick 1 sync calls = %d; want 1", calls.Load())
 	}
@@ -569,6 +748,7 @@ func TestSchedulerFailedSyncStillConsumesCadence(t *testing.T) {
 	// Within cadence → no re-fire even though the prior run failed.
 	f.clock.Add(10 * time.Minute)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if calls.Load() != 1 {
 		t.Fatalf("under-cadence tick after failure called sync %d times; want 1 (no special retry)",
 			calls.Load())
@@ -577,6 +757,7 @@ func TestSchedulerFailedSyncStillConsumesCadence(t *testing.T) {
 	// Past cadence → re-fires (regardless of prior failure).
 	f.clock.Add(time.Hour)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	if calls.Load() != 2 {
 		t.Fatalf("past-cadence tick after failure called sync %d times; want 2", calls.Load())
 	}
@@ -598,6 +779,7 @@ func TestSchedulerBothCadencesShareIndexRun(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx)
+	sch.dispatch.wait()
 
 	runs, _ := f.store.ListRuns(ctx, store.ListRunsOpts{})
 	var indexCount int
@@ -627,11 +809,13 @@ func TestSchedulerStandaloneIndexAfterSyncCadenceWaits(t *testing.T) {
 	ctx := context.Background()
 
 	sch.tick(ctx) // fires sync + pre-sync index.
+	sch.dispatch.wait()
 
 	// 10min later → under index_every from the pre-sync index, no
 	// standalone fires.
 	f.clock.Add(10 * time.Minute)
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	runs, _ := f.store.ListRuns(ctx, store.ListRunsOpts{})
 	var indexCount int
 	for _, r := range runs {
@@ -647,6 +831,7 @@ func TestSchedulerStandaloneIndexAfterSyncCadenceWaits(t *testing.T) {
 	// due yet because sync_every is 1h).
 	f.clock.Add(10 * time.Minute) // total 20min since pre-sync.
 	sch.tick(ctx)
+	sch.dispatch.wait()
 	runs, _ = f.store.ListRuns(ctx, store.ListRunsOpts{})
 	indexCount = 0
 	for _, r := range runs {
@@ -792,9 +977,12 @@ func TestSchedulerDropsSyncWhenRunnerNil(t *testing.T) {
 	})
 	f.seedFile()
 	sch := f.scheduler()
-	sch.syncRun = nil
+	// A dispatcher with no runner is the index-only-host case (no rclone
+	// wired): sync-triggering paths log a skip and create no sync rows.
+	sch.dispatch = newSyncDispatcher(nil, f.srv.cfg.Logger, f.clock.Now, defaultMaxParallelSyncs)
 
 	sch.tick(context.Background())
+	sch.dispatch.wait()
 
 	if !f.containsLogLine("scheduler.skipped", "kind=sync", "sync runner not configured") {
 		t.Fatalf("missing nil-runner skip log:\n%s", f.logBuf.String())
diff --git a/cmd/squirrel/agent.go b/cmd/squirrel/agent.go
index ad18923..872c5df 100644
--- a/cmd/squirrel/agent.go
+++ b/cmd/squirrel/agent.go
@@ -169,6 +169,10 @@ func resolveSchedulerRclone(cmd *cobra.Command, cfg *config.Config) (*sync.Rclon
 	if err != nil {
 		return nil, fmt.Errorf("scheduler needs rclone for scheduled syncs: %w", err)
 	}
+	// Bound every automatic transfer by the no-progress guard so a wedged
+	// endpoint fails its own run instead of hanging forever (#160, F25).
+	// Foreground `squirrel sync` leaves this unset — a human can interrupt.
+	rcl.StallTimeout = sync.DefaultStallTimeout
 	pairs, err := sync.PairsFor(cfg, "", "")
 	if err != nil {
 		return nil, fmt.Errorf("scheduler rclone preflight: %w", err)
diff --git a/sync/rclone.go b/sync/rclone.go
index 9a4ab86..8995da5 100644
--- a/sync/rclone.go
+++ b/sync/rclone.go
@@ -21,6 +21,8 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/mbertschler/squirrel/config"
 	"github.com/mbertschler/squirrel/runevents"
@@ -31,6 +33,26 @@ import (
 // --hash blake3 flag used by sync would be rejected by rclone.
 var MinRcloneVersion = Version{Major: 1, Minor: 66}
 
+// DefaultStallTimeout is the no-progress bound the agent scheduler applies
+// to every automatic rclone transfer (see Rclone.StallTimeout). Ten
+// minutes is comfortably longer than the longest legitimate gap between
+// progress reports — hashing one very large file for a --checksum
+// comparison surfaces no transfer/check advance until it completes — while
+// bounding a wedged endpoint far below the "indefinite" hang F25 recorded.
+// A dark endpoint usually fails sooner via rclone's own --contimeout /
+// --timeout; the guard covers the live-but-stuck case those don't.
+const DefaultStallTimeout = 10 * time.Minute
+
+// rcloneConnectTimeout and rcloneIOTimeout are passed as --contimeout and
+// --timeout on every streaming transfer so a dead or unresponsive endpoint
+// self-terminates instead of hanging. They match rclone's own defaults but
+// are set explicitly so the bound is guaranteed applied and lives in one
+// tunable place (#160, F25).
+const (
+	rcloneConnectTimeout = "1m"
+	rcloneIOTimeout      = "5m"
+)
+
 // Rclone is a configured rclone wrapper. Binary is the resolved executable
 // path; Config is the path to the squirrel-managed rclone.conf written by
 // WriteRcloneConfig. All Run invocations pass --config Config so the user's
@@ -38,6 +60,17 @@ var MinRcloneVersion = Version{Major: 1, Minor: 66}
 type Rclone struct {
 	Binary string
 	Config string
+	// StallTimeout, when > 0, bounds a streaming transfer (RunWithProgress)
+	// by progress rather than by total wall-clock: if rclone reports no
+	// advance in transferred+checked+bytes for this long, the child is
+	// killed and the run fails with a diagnosable "stalled" error. It is the
+	// squirrel-side backstop for a connection that stays alive but stops
+	// making progress — the storage-full S3 hang of F25 that tripped neither
+	// --contimeout nor --timeout. Zero (the default, used by the foreground
+	// CLI where a human can interrupt) disables the guard; the agent
+	// scheduler sets it to DefaultStallTimeout so an unattended cadence can
+	// never wedge forever.
+	StallTimeout time.Duration
 }
 
 // Find locates the rclone binary on PATH. The returned Rclone has Config
@@ -308,23 +341,18 @@ func (r *Rclone) RunWithProgress(ctx context.Context, onProgress func(runevents.
 	if r.Config == "" {
 		return RunResult{}, errors.New("rclone wrapper: Config not set (call WriteRcloneConfig first)")
 	}
-	full := append([]string{
-		"--config", r.Config,
-		"--use-json-log",
-		"--log-level", "INFO",
-		"--stats", "1s",
-	}, args...)
-	cmd := exec.CommandContext(ctx, r.Binary, full...)
-	stderr, err := cmd.StderrPipe()
-	if err != nil {
-		return RunResult{}, fmt.Errorf("rclone stderr pipe: %w", err)
+	// A per-run cancel backs the no-progress guard: rclone's own timeouts
+	// bound a dead connection, the guard bounds a live-but-wedged one (F25).
+	runCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	var guard *stallGuard
+	if r.StallTimeout > 0 {
+		guard = newStallGuard(runCtx, r.StallTimeout, cancel)
 	}
-	stdout, err := cmd.StdoutPipe()
+
+	cmd, stderr, stdout, err := r.startRclone(runCtx, args)
 	if err != nil {
-		return RunResult{}, fmt.Errorf("rclone stdout pipe: %w", err)
-	}
-	if err := cmd.Start(); err != nil {
-		return RunResult{}, fmt.Errorf("rclone start: %w", err)
+		return RunResult{}, err
 	}
 
 	var (
@@ -334,7 +362,7 @@ func (r *Rclone) RunWithProgress(ctx context.Context, onProgress func(runevents.
 	wg.Add(2)
 	go func() {
 		defer wg.Done()
-		parseJSONLog(stderr, &result, onProgress)
+		parseJSONLog(stderr, &result, onProgress, guard.advance)
 	}()
 	go func() {
 		defer wg.Done()
@@ -345,13 +373,136 @@ func (r *Rclone) RunWithProgress(ctx context.Context, onProgress func(runevents.
 	wg.Wait()
 
 	waitErr := cmd.Wait()
+	cancel()
+	guard.wait()
 	if waitErr != nil {
 		result.FatalError = result.Errors == 0
-		return result, rcloneExitError(waitErr, result.Stderr)
+		return result, rcloneRunError(waitErr, result.Stderr, guard, r.StallTimeout)
 	}
 	return result, nil
 }
 
+// startRclone assembles the full argument list — config, JSON logging,
+// per-second stats, and the connect/IO timeouts every transfer carries
+// (#160, F25) — and starts the child on runCtx, returning its stderr and
+// stdout pipes. runCtx backs both the caller's cancellation and the stall
+// guard's kill.
+func (r *Rclone) startRclone(runCtx context.Context, args []string) (*exec.Cmd, io.Reader, io.Reader, error) {
+	full := append([]string{
+		"--config", r.Config,
+		"--use-json-log",
+		"--log-level", "INFO",
+		"--stats", "1s",
+		"--contimeout", rcloneConnectTimeout,
+		"--timeout", rcloneIOTimeout,
+	}, args...)
+	cmd := exec.CommandContext(runCtx, r.Binary, full...)
+	stderr, err := cmd.StderrPipe()
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("rclone stderr pipe: %w", err)
+	}
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return nil, nil, nil, fmt.Errorf("rclone stdout pipe: %w", err)
+	}
+	if err := cmd.Start(); err != nil {
+		return nil, nil, nil, fmt.Errorf("rclone start: %w", err)
+	}
+	return cmd, stderr, stdout, nil
+}
+
+// stallGuard kills a running rclone child when a streaming transfer stops
+// making progress for longer than timeout. rclone's --contimeout/--timeout
+// bound a dead connection; the guard is the squirrel-side backstop for a
+// connection that stays alive yet stops advancing — the storage-full hang
+// of F25 that tripped neither. It is driven off rclone's periodic --stats
+// events (via notifyAdvance): any growth in transferred+checked+bytes
+// resets the timer, so a slow-but-moving transfer of a large volume is
+// never killed while a wedged one is.
+type stallGuard struct {
+	timeout time.Duration
+	cancel  context.CancelFunc
+	poke    chan struct{}
+	done    chan struct{}
+	fired   atomic.Bool
+}
+
+func newStallGuard(ctx context.Context, timeout time.Duration, cancel context.CancelFunc) *stallGuard {
+	g := &stallGuard{
+		timeout: timeout,
+		cancel:  cancel,
+		poke:    make(chan struct{}, 1),
+		done:    make(chan struct{}),
+	}
+	go g.watch(ctx)
+	return g
+}
+
+// advance signals progress from the stats-reader goroutine. Non-blocking:
+// a pending poke already carries the reset, so a dropped signal is
+// harmless. Safe on a nil guard so the disabled path needs no branch.
+func (g *stallGuard) advance() {
+	if g == nil {
+		return
+	}
+	select {
+	case g.poke <- struct{}{}:
+	default:
+	}
+}
+
+// watch resets its timer on every progress poke and, when the timer
+// expires with no intervening progress, records the stall and cancels the
+// run context — which makes exec.CommandContext kill the rclone child. It
+// returns when ctx is done (the run finished or was cancelled elsewhere).
+func (g *stallGuard) watch(ctx context.Context) {
+	defer close(g.done)
+	timer := time.NewTimer(g.timeout)
+	defer timer.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-g.poke:
+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:
+				}
+			}
+			timer.Reset(g.timeout)
+		case <-timer.C:
+			g.fired.Store(true)
+			g.cancel()
+			return
+		}
+	}
+}
+
+// wait blocks until the watch goroutine has returned so the caller can read
+// fired without a race. Safe on a nil guard.
+func (g *stallGuard) wait() {
+	if g == nil {
+		return
+	}
+	<-g.done
+}
+
+// rcloneRunError builds the error for a non-zero rclone exit. When the
+// stall guard fired, the exit is a kill squirrel caused, so we say so
+// plainly (folding in the captured stderr tail, #157) rather than
+// surfacing a bare "signal: killed"; otherwise it defers to
+// rcloneExitError.
+func rcloneRunError(waitErr error, stderrTail string, guard *stallGuard, timeout time.Duration) error {
+	if guard != nil && guard.fired.Load() {
+		if tail := strings.TrimSpace(stderrTail); tail != "" {
+			return fmt.Errorf("rclone stalled: no progress for %s, killed: %w: %s", timeout, waitErr, tail)
+		}
+		return fmt.Errorf("rclone stalled: no progress for %s, killed: %w", timeout, waitErr)
+	}
+	return rcloneExitError(waitErr, stderrTail)
+}
+
 // rcloneExitError wraps a non-zero rclone exit, folding in the captured
 // stderr tail when present so the scheduler log and the run row carry the
 // actual reason (auth, host key, path) rather than a bare "exit status 1"
@@ -554,9 +705,10 @@ func isErrorLevel(level string) bool {
 // that never reaches the JSON log still carries its reason (#157,
 // F6/F15). onProgress, if non-nil, is invoked once per stats event so
 // callers can drive a live UI.
-func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Progress)) {
+func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Progress), onAdvance func()) {
 	scanner := bufio.NewScanner(r)
 	scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024)
+	var lastProgressTotal int64
 	for scanner.Scan() {
 		line := scanner.Bytes()
 		if len(line) == 0 {
@@ -579,6 +731,7 @@ func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Prog
 		}
 		if ev.Stats != nil {
 			applyStatsEvent(result, ev.Stats, onProgress)
+			notifyAdvance(result, &lastProgressTotal, onAdvance)
 			continue
 		}
 		if isErrorLevel(ev.Level) && !isRetrySummary(ev.Msg) {
@@ -596,6 +749,24 @@ func parseJSONLog(r io.Reader, result *RunResult, onProgress func(runevents.Prog
 	}
 }
 
+// notifyAdvance pokes the stall guard whenever the running total of
+// transferred+checked+bytes has grown since the last stats event, so a
+// transfer that is slow but genuinely moving keeps resetting the guard's
+// timer while a wedged one — rclone still emitting per-second stats, but
+// with flat counters — lets it fire. Checks count as progress so a long
+// verification pass over an already-populated destination is not mistaken
+// for a stall.
+func notifyAdvance(result *RunResult, last *int64, onAdvance func()) {
+	if onAdvance == nil {
+		return
+	}
+	total := result.Transferred + result.Checked + result.Bytes
+	if total > *last {
+		*last = total
+		onAdvance()
+	}
+}
+
 // applyStatsEvent folds one rclone stats event into result and drives the
 // optional progress callback. Errors is taken as the running maximum so a
 // per-attempt stats line that reports fewer errors than a prior one (a
diff --git a/sync/rclone_test.go b/sync/rclone_test.go
index efb52e1..340eb2e 100644
--- a/sync/rclone_test.go
+++ b/sync/rclone_test.go
@@ -7,6 +7,7 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -14,6 +15,65 @@ import (
 	"github.com/mbertschler/squirrel/runevents"
 )
 
+// TestStallGuardFiresWithoutProgress proves the no-progress guard cancels
+// the run — and records that it fired — when no advance arrives within the
+// timeout. This is the F25 wedge: rclone alive but transferring nothing.
+// The test waits for the guard rather than racing a deadline, so the exact
+// timeout is not timing-sensitive; a generous value keeps it robust under
+// load.
+func TestStallGuardFiresWithoutProgress(t *testing.T) {
+	var cancelled atomic.Bool
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	g := newStallGuard(ctx, 100*time.Millisecond, func() {
+		cancelled.Store(true)
+		cancel()
+	})
+	g.wait() // blocks until watch returns; with no poke it fires first
+	if !g.fired.Load() {
+		t.Fatal("guard did not fire after the stall timeout elapsed with no progress")
+	}
+	if !cancelled.Load() {
+		t.Fatal("guard fired but never cancelled the run context")
+	}
+}
+
+// TestStallGuardResetsOnProgress proves a transfer that keeps advancing is
+// never killed: pokes driven off a ticker at a fraction of the stall window
+// keep resetting it across a span twice as long as one window, so only a
+// working reset explains the guard staying quiet. The wide ratio between
+// the stall timeout and the poke interval keeps scheduler jitter from
+// faking a stall.
+func TestStallGuardResetsOnProgress(t *testing.T) {
+	var cancelled atomic.Bool
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	const stall = 400 * time.Millisecond
+	g := newStallGuard(ctx, stall, func() {
+		cancelled.Store(true)
+		cancel()
+	})
+	tick := time.NewTicker(stall / 20) // pokes at 20x the resolution of the bound
+	defer tick.Stop()
+	deadline := time.After(2 * stall) // outlast a single window, so a reset is required
+	for done := false; !done; {
+		select {
+		case <-tick.C:
+			g.advance()
+		case <-deadline:
+			done = true
+		}
+	}
+	if g.fired.Load() {
+		t.Fatal("guard fired while progress was still arriving")
+	}
+	cancel() // finishing the run stops the guard cleanly
+	g.wait()
+	if cancelled.Load() {
+		t.Fatal("guard cancelled the run despite steady progress")
+	}
+}
+
 // requireRclone skips the test if rclone is not on PATH. The wrapper tests
 // exercise the real binary against a local-filesystem destination; if a
 // developer wants to run them they need rclone installed.
@@ -58,7 +118,7 @@ func TestParseJSONLogCapturesObjectlessErrors(t *testing.T) {
 		`{"stats":{"errors":1,"fatalError":true,"totalTransfers":0,"totalChecks":0,"bytes":0}}`,
 	}, "\n")
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 
 	if len(r.FailedFiles) != 2 {
 		t.Fatalf("FailedFiles = %+v, want 2 (auth + reading; retry summaries dropped)", r.FailedFiles)
@@ -85,7 +145,7 @@ func TestParseJSONLogCapturesFatalLevelAndRawStderr(t *testing.T) {
 		`{"stats":{"errors":0,"fatalError":true,"totalTransfers":0,"totalChecks":0,"bytes":0}}`,
 	}, "\n")
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 
 	if len(r.FailedFiles) != 1 || !strings.Contains(r.FailedFiles[0].Message, "NoCredentialProviders") {
 		t.Fatalf("FailedFiles = %+v, want the fatal-level message captured", r.FailedFiles)
@@ -105,8 +165,7 @@ func TestParseJSONLogStderrBounded(t *testing.T) {
 		lines = append(lines, fmt.Sprintf("stderr line %04d", i))
 	}
 	var r RunResult
-	parseJSONLog(strings.NewReader(strings.Join(lines, "\n")), &r, nil)
-
+	parseJSONLog(strings.NewReader(strings.Join(lines, "\n")), &r, nil, nil)
 	if len(r.Stderr) == 0 {
 		t.Fatal("Stderr empty, want the tail of the stream")
 	}
@@ -153,7 +212,7 @@ func TestParseJSONLogDetectsHashFallback(t *testing.T) {
 		`{"stats":{"errors":0,"fatalError":false,"totalTransfers":2,"totalChecks":0,"bytes":10}}`,
 	}, "\n")
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 
 	if !r.HashFallback {
 		t.Fatalf("HashFallback = false, want true (no-common-hash notice should be detected)")
@@ -168,7 +227,7 @@ func TestParseJSONLogDetectsHashFallback(t *testing.T) {
 func TestParseJSONLogNoFalseHashFallback(t *testing.T) {
 	stream := `{"stats":{"errors":0,"fatalError":false,"totalTransfers":2,"totalChecks":1,"bytes":10}}`
 	var r RunResult
-	parseJSONLog(strings.NewReader(stream), &r, nil)
+	parseJSONLog(strings.NewReader(stream), &r, nil, nil)
 	if r.HashFallback {
 		t.Fatalf("HashFallback = true on a clean run, want false")
 	}
@@ -187,7 +246,7 @@ func TestParseJSONLogEmitsProgressWithByteTotal(t *testing.T) {
 	var events []runevents.Progress
 	parseJSONLog(strings.NewReader(stream), &r, func(p runevents.Progress) {
 		events = append(events, p)
-	})
+	}, nil)
 	if len(events) != 2 {
 		t.Fatalf("progress events = %d, want 2", len(events))
 	}