diff --git a/pkg/compose/rebuild_scheduler.go b/pkg/compose/rebuild_scheduler.go
new file mode 100644
index 0000000000..e934105aab
--- /dev/null
+++ b/pkg/compose/rebuild_scheduler.go
@@ -0,0 +1,145 @@
+/*
+
+   Copyright 2020 Docker Compose CLI authors
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package compose
+
+import (
+	"context"
+	"sync"
+
+	"github.com/docker/compose/v5/pkg/utils"
+)
+
+// rebuildFunc performs a rebuild for the given services. The context is the
+// run's own: the scheduler cancels it when the run is interrupted because a
+// newer request made its outcome stale, and when the watch shuts down.
+type rebuildFunc func(ctx context.Context, services []string) error
+
+// rebuildScheduler coalesces rebuild requests received while a rebuild is
+// already running into a single trailing rebuild, so that a burst of file
+// change events never queues up more than one extra rebuild.
+//
+// Its convergence invariant: every request ends up covered by a rebuild
+// whose build-context snapshot was taken after the request. A run made
+// stale by a newer request for one of its own services is interrupted
+// rather than left to finish a result that would be replaced anyway.
+type rebuildScheduler struct {
+	ctx     context.Context // watch lifecycle: no run outlives it
+	rebuild rebuildFunc
+
+	mu        sync.Mutex
+	building  bool
+	active    utils.Set[string]  // services being rebuilt by the current run, if any
+	pending   utils.Set[string]  // services queued for the next trailing rebuild
+	interrupt context.CancelFunc // cancels the current run; nil outside a run
+	wg        sync.WaitGroup
+}
+
+func newRebuildScheduler(ctx context.Context, rebuild rebuildFunc) *rebuildScheduler {
+	return &rebuildScheduler{
+		ctx:     ctx,
+		rebuild: rebuild,
+		pending: utils.NewSet[string](),
+	}
+}
+
+// Request asks for a rebuild of services. It never blocks on the rebuild
+// itself: the services are always merged into the pending set, and the run
+// loop is (re)started only if it isn't already draining it.
+//
+// The decision must be made synchronously, under the lock, so that requests
+// racing with the run loop's completion are never lost nor cause two
+// rebuilds to run concurrently.
+//
+// A request naming a service the current run is already rebuilding makes
+// that run stale — its build context was snapshotted before the change
+// behind this request — so the run is interrupted. Interruption kills the
+// whole run, so its complete active set is folded back into pending: the
+// trailing rebuild redoes every affected service against a fresh snapshot.
+func (s *rebuildScheduler) Request(services []string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.pending.AddAll(services...)
+	if !s.building {
+		s.building = true
+		s.wg.Add(1)
+		go s.run()
+		return
+	}
+	for _, service := range services {
+		if s.active.Has(service) {
+			s.pending.AddAll(s.active.Elements()...)
+			s.interrupt()
+			return
+		}
+	}
+}
+
+// Pending reports whether a rebuild for the given service is queued for the
+// next trailing rebuild. Such a rebuild will snapshot the build context
+// after whatever change the caller is reacting to: recreating and starting
+// the service from it makes a separate restart redundant.
+func (s *rebuildScheduler) Pending(service string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.pending.Has(service)
+}
+
+// InFlight reports whether the given service is being rebuilt by the
+// current run. That run's snapshot predates the caller's change: acting on
+// the service's container would race the run's create/start, and the run's
+// outcome is already stale — fold the action into a Request instead.
+func (s *rebuildScheduler) InFlight(service string) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.active.Has(service)
+}
+
+// run drains the pending set into a rebuild, looping to pick up whatever
+// accumulated (or was folded back by an interruption) while that rebuild
+// was in flight, until nothing is left or the watch shuts down.
+func (s *rebuildScheduler) run() {
+	defer s.wg.Done()
+	for {
+		s.mu.Lock()
+		if len(s.pending) == 0 || s.ctx.Err() != nil {
+			s.building = false
+			s.mu.Unlock()
+			return
+		}
+		runCtx, cancel := context.WithCancel(s.ctx)
+		s.interrupt = cancel
+		s.active = s.pending
+		s.pending = utils.NewSet[string]()
+		services := s.active.Elements()
+		s.mu.Unlock()
+
+		// The error is deliberately not handled here: rebuild reports every
+		// failure to the user log itself, and an interrupted run's services
+		// are already folded back into pending by Request.
+		_ = s.rebuild(runCtx, services)
+
+		s.mu.Lock()
+		s.active = nil
+		s.interrupt = nil
+		s.mu.Unlock()
+		cancel()
+	}
+}
+
+// Wait blocks until the scheduler is idle: no rebuild is running and nothing
+// is pending. The caller must have stopped issuing Requests.
+func (s *rebuildScheduler) Wait() {
+	s.wg.Wait()
+}
diff --git a/pkg/compose/rebuild_scheduler_test.go b/pkg/compose/rebuild_scheduler_test.go
new file mode 100644
index 0000000000..f4f17426c9
--- /dev/null
+++ b/pkg/compose/rebuild_scheduler_test.go
@@ -0,0 +1,385 @@
+/*
+
+   Copyright 2020 Docker Compose CLI authors
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+       http://www.apache.org/licenses/LICENSE-2.0
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package compose
+
+import (
+	"context"
+	"errors"
+	"sort"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"gotest.tools/v3/assert"
+)
+
+// fakeRebuilder is a deterministic, channel-driven stand-in for the real
+// `s.rebuild` call. Every invocation:
+//   - records the requested services (sorted, for easy comparison),
+//   - blocks until the test sends a value (nil or an error) on `release`,
+//   - fails the test via t.Errorf (safe to call from any goroutine) if it
+//     is ever entered while another invocation hasn't returned yet, which
+//     is exactly the "no concurrent rebuild" property under test.
+type fakeRebuilder struct {
+	t *testing.T
+
+	running int32 // atomic: 1 while an invocation is in flight
+
+	started chan []string // signaled with sorted services each time rebuild() is entered
+	release chan error    // test sends here to let the current invocation return
+
+	mu    sync.Mutex
+	calls [][]string // sorted services for every completed invocation, in order
+}
+
+func newFakeRebuilder(t *testing.T) *fakeRebuilder {
+	t.Helper()
+	return &fakeRebuilder{
+		t:       t,
+		started: make(chan []string),
+		release: make(chan error),
+	}
+}
+
+func (f *fakeRebuilder) rebuild(ctx context.Context, services []string) error {
+	if !atomic.CompareAndSwapInt32(&f.running, 0, 1) {
+		f.t.Errorf("rebuild() invoked while a previous rebuild was still in progress: services=%v", services)
+	}
+	defer atomic.StoreInt32(&f.running, 0)
+
+	sorted := append([]string(nil), services...)
+	sort.Strings(sorted)
+
+	f.mu.Lock()
+	f.calls = append(f.calls, sorted)
+	f.mu.Unlock()
+
+	f.started <- sorted
+	select {
+	case err := <-f.release:
+		return err
+	case <-ctx.Done():
+		// the run was interrupted (or the watch shut down): return like the
+		// real rebuild would once its build context is cancelled
+		return ctx.Err()
+	}
+}
+
+func (f *fakeRebuilder) callCount() int {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return len(f.calls)
+}
+
+// awaitStarted waits (with a bounded, generous timeout so a genuine bug
+// hangs the test instead of the suite) for the next rebuild invocation to
+// begin, returning the sorted services it was called with.
+func awaitStarted(t *testing.T, f *fakeRebuilder) []string {
+	t.Helper()
+	select {
+	case services := <-f.started:
+		return services
+	case <-time.After(2 * time.Second):
+		t.Fatal("timed out waiting for rebuild to start")
+		return nil
+	}
+}
+
+// assertNoRebuildStarts asserts that no new rebuild invocation begins
+// within a short grace window. Since a correctly implemented scheduler
+// decides synchronously (under its own lock) whether to start a rebuild or
+// merely record it as pending, this is not a race against a slow producer:
+// if the scheduler doesn't start a rebuild by the time this is called, it
+// never will for the requests already issued.
+func assertNoRebuildStarts(t *testing.T, f *fakeRebuilder) {
+	t.Helper()
+	select {
+	case services := <-f.started:
+		t.Fatalf("unexpected rebuild started with services=%v", services)
+	case <-time.After(50 * time.Millisecond):
+		// expected: nothing started
+	}
+}
+
+// TestRebuildScheduler_FirstRequestStartsImmediately covers behavior (1):
+// a request issued while idle must trigger a rebuild without being delayed
+// or batched -- latency for an isolated edit must be unchanged.
+func TestRebuildScheduler_FirstRequestStartsImmediately(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	scheduler.Request([]string{"web"})
+
+	services := awaitStarted(t, f)
+	assert.DeepEqual(t, services, []string{"web"})
+
+	f.release <- nil
+	scheduler.Wait()
+
+	assert.Equal(t, f.callCount(), 1)
+}
+
+// TestRebuildScheduler_RequestsDuringBuildAreCoalescedIntoPending covers
+// behavior (2): requests arriving while a rebuild is already running must
+// not start a new rebuild; they accumulate into a deduplicated pending set.
+func TestRebuildScheduler_RequestsDuringBuildAreCoalescedIntoPending(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	scheduler.Request([]string{"web"})
+	first := awaitStarted(t, f)
+	assert.DeepEqual(t, first, []string{"web"})
+
+	// These arrive while the first rebuild is still in flight (we haven't
+	// released it yet), name no active service (a request for one would
+	// interrupt the run — covered by the interruption tests), and must not
+	// trigger immediate rebuilds.
+	scheduler.Request([]string{"api"})
+	scheduler.Request([]string{"api", "worker"})
+
+	assertNoRebuildStarts(t, f)
+	assert.Equal(t, f.callCount(), 1)
+
+	f.release <- nil
+
+	// The pending requests above must still produce exactly one trailing
+	// rebuild (verified by TestRebuildScheduler_TrailingRebuildConsolidatesPendingServices);
+	// drain it here so Wait() isn't left blocking on it forever.
+	awaitStarted(t, f)
+	f.release <- nil
+	scheduler.Wait()
+}
+
+// TestRebuildScheduler_TrailingRebuildConsolidatesPendingServices covers
+// behavior (3): once the in-progress rebuild completes, exactly one
+// trailing rebuild starts automatically, covering the union of every
+// service requested while the first rebuild was running, and the pending
+// set is cleared.
+func TestRebuildScheduler_TrailingRebuildConsolidatesPendingServices(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	scheduler.Request([]string{"web"})
+	awaitStarted(t, f)
+
+	scheduler.Request([]string{"api"})
+	scheduler.Request([]string{"api", "worker"})
+
+	f.release <- nil // let the first rebuild finish
+
+	trailing := awaitStarted(t, f)
+	assert.DeepEqual(t, trailing, []string{"api", "worker"})
+
+	f.release <- nil
+	scheduler.Wait()
+
+	// Exactly one trailing rebuild: nothing else was requested during it.
+	assert.Equal(t, f.callCount(), 2)
+}
+
+// TestRebuildScheduler_TrailingRebuildsChainUntilPendingEmpty covers
+// behavior (4): if new requests arrive while a trailing rebuild is
+// running, another trailing rebuild fires when it completes, and this
+// repeats until a rebuild finishes with nothing pending, at which point the
+// scheduler goes idle.
+func TestRebuildScheduler_TrailingRebuildsChainUntilPendingEmpty(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	scheduler.Request([]string{"web"})
+	assert.DeepEqual(t, awaitStarted(t, f), []string{"web"})
+
+	scheduler.Request([]string{"api"})
+	f.release <- nil // first rebuild done, "api" is pending -> triggers 2nd rebuild
+
+	assert.DeepEqual(t, awaitStarted(t, f), []string{"api"})
+
+	// A request arrives *during* the trailing rebuild: must chain into a 3rd.
+	scheduler.Request([]string{"db"})
+	f.release <- nil // 2nd rebuild done, "db" is pending -> triggers 3rd rebuild
+
+	assert.DeepEqual(t, awaitStarted(t, f), []string{"db"})
+
+	// Nothing requested during the 3rd rebuild: scheduler must return to idle.
+	f.release <- nil
+	scheduler.Wait()
+
+	assertNoRebuildStarts(t, f)
+	assert.Equal(t, f.callCount(), 3)
+}
+
+// TestRebuildScheduler_RebuildErrorDoesNotBlockPendingProcessing covers
+// behavior (6): a failed rebuild must not prevent a subsequently pending
+// rebuild from being processed.
+func TestRebuildScheduler_RebuildErrorDoesNotBlockPendingProcessing(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	scheduler.Request([]string{"web"})
+	awaitStarted(t, f)
+
+	scheduler.Request([]string{"api"})
+	f.release <- errors.New("build failed") // first rebuild fails
+
+	// Despite the error, the pending "api" request must still be processed.
+	trailing := awaitStarted(t, f)
+	assert.DeepEqual(t, trailing, []string{"api"})
+
+	f.release <- nil
+	scheduler.Wait()
+
+	assert.Equal(t, f.callCount(), 2)
+}
+
+// TestRebuildScheduler_NoConcurrentRebuilds covers behavior (5): at most
+// one rebuild must ever be running at a time, and the scheduler's internal
+// state (pending set, "building" flag) must not race under concurrent
+// requests. Run with `go test -race` to make both properties meaningful.
+func TestRebuildScheduler_NoConcurrentRebuilds(t *testing.T) {
+	var running int32
+	var maxObservedConcurrency int32
+
+	rebuild := func(_ context.Context, _ []string) error {
+		n := atomic.AddInt32(&running, 1)
+		defer atomic.AddInt32(&running, -1)
+
+		for {
+			max := atomic.LoadInt32(&maxObservedConcurrency)
+			if n <= max || atomic.CompareAndSwapInt32(&maxObservedConcurrency, max, n) {
+				break
+			}
+		}
+
+		// Widen the window during which a concurrency bug would be
+		// observable. This does not make the test's pass/fail outcome
+		// depend on timing: it only increases the odds of *catching* a
+		// bug, it can never cause a false failure.
+		time.Sleep(2 * time.Millisecond)
+		return nil
+	}
+
+	scheduler := newRebuildScheduler(t.Context(), rebuild)
+
+	const goroutines = 50
+	var wg sync.WaitGroup
+	wg.Add(goroutines)
+	for i := 0; i < goroutines; i++ {
+		go func() {
+			defer wg.Done()
+			scheduler.Request([]string{"svc"})
+		}()
+	}
+	wg.Wait()
+
+	scheduler.Wait()
+
+	assert.Equal(t, atomic.LoadInt32(&maxObservedConcurrency), int32(1))
+}
+
+// TestRebuildScheduler_PendingAndInFlight covers the queries callers use to
+// decide what to do with a restart racing a rebuild of the same service:
+// drop it (pending — the coming rebuild converges on its own) or fold it
+// into a Request (in flight — the running rebuild is stale).
+func TestRebuildScheduler_PendingAndInFlight(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	assert.Equal(t, scheduler.Pending("web"), false)
+	assert.Equal(t, scheduler.InFlight("web"), false)
+
+	scheduler.Request([]string{"web"})
+	awaitStarted(t, f)
+	assert.Equal(t, scheduler.InFlight("web"), true)
+	assert.Equal(t, scheduler.Pending("web"), false)
+
+	// Queued for the trailing rebuild while "web" is still building.
+	scheduler.Request([]string{"api"})
+	assert.Equal(t, scheduler.Pending("api"), true)
+	assert.Equal(t, scheduler.InFlight("api"), false)
+
+	f.release <- nil // "web" finishes, "api" starts as the trailing rebuild
+	awaitStarted(t, f)
+	assert.Equal(t, scheduler.InFlight("api"), true)
+	assert.Equal(t, scheduler.InFlight("web"), false)
+
+	f.release <- nil
+	scheduler.Wait()
+
+	assert.Equal(t, scheduler.Pending("api"), false)
+	assert.Equal(t, scheduler.InFlight("api"), false)
+}
+
+// A request naming a service the current run is rebuilding makes that run
+// stale: the run must be interrupted, and the trailing rebuild must cover
+// the interrupted run's whole active set — interruption killed the other
+// services' rebuild too — so every service converges on a fresh snapshot.
+func TestRebuildScheduler_RequestForActiveServiceInterruptsRun(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	scheduler.Request([]string{"web", "api"})
+	assert.DeepEqual(t, awaitStarted(t, f), []string{"api", "web"})
+
+	// Interrupts the run (no f.release send: the fake returns through its
+	// cancelled context, like the real rebuild would).
+	scheduler.Request([]string{"web"})
+
+	trailing := awaitStarted(t, f)
+	assert.DeepEqual(t, trailing, []string{"api", "web"})
+
+	f.release <- nil
+	scheduler.Wait()
+	assert.Equal(t, f.callCount(), 2)
+}
+
+// A request for a service the current run is NOT rebuilding must not
+// interrupt it: the run's outcome is still wanted, the new service just
+// waits its turn in the trailing rebuild.
+func TestRebuildScheduler_RequestForOtherServiceDoesNotInterrupt(t *testing.T) {
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(t.Context(), f.rebuild)
+
+	scheduler.Request([]string{"web"})
+	awaitStarted(t, f)
+
+	scheduler.Request([]string{"api"})
+	assertNoRebuildStarts(t, f) // first run still in flight, not interrupted
+
+	f.release <- nil // it completes normally...
+	assert.DeepEqual(t, awaitStarted(t, f), []string{"api"})
+
+	f.release <- nil
+	scheduler.Wait()
+	assert.Equal(t, f.callCount(), 2)
+}
+
+// Once the watch context is cancelled the scheduler must stop draining:
+// no trailing rebuild fires with a dead context, and Wait returns.
+func TestRebuildScheduler_ShutdownStopsDraining(t *testing.T) {
+	ctx, cancel := context.WithCancel(t.Context())
+	f := newFakeRebuilder(t)
+	scheduler := newRebuildScheduler(ctx, f.rebuild)
+
+	scheduler.Request([]string{"web"})
+	awaitStarted(t, f)
+	scheduler.Request([]string{"api"}) // pending when the shutdown hits
+
+	cancel() // the in-flight run unwinds through its context
+
+	scheduler.Wait()
+	assertNoRebuildStarts(t, f)
+	assert.Equal(t, f.callCount(), 1)
+}
diff --git a/pkg/compose/watch.go b/pkg/compose/watch.go
index baba7707d6..9529c006df 100644
--- a/pkg/compose/watch.go
+++ b/pkg/compose/watch.go
@@ -373,11 +373,19 @@ func isSync(trigger types.Trigger) bool {
 
 func (s *composeService) watchEvents(ctx context.Context, project *types.Project, options api.WatchOptions, watcher watch.Notify, syncer sync.Syncer, rules []watchRule) error {
 	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
 
 	// debounce and group filesystem events so that we capture IDE saving many files as one "batch" event
 	batchEvents := watch.BatchDebounceEvents(ctx, watcher.Events())
 
+	scheduler := newRebuildScheduler(ctx, func(ctx context.Context, services []string) error {
+		return s.rebuild(ctx, project, services, options)
+	})
+	// Rebuilds run asynchronously in their own goroutine(s); cancel first so
+	// an in-flight one can unwind, then wait for it to settle so none
+	// outlive this function.
+	defer scheduler.Wait()
+	defer cancel()
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -406,7 +414,7 @@ func (s *composeService) watchEvents(ctx context.Context, project *types.Project
 			}
 			start := time.Now()
 			logrus.Debugf("batch start: count[%d]", len(batch))
-			err := s.handleWatchBatch(ctx, project, options, batch, rules, syncer)
+			err := s.handleWatchBatch(ctx, project, options, batch, rules, syncer, scheduler)
 			if err != nil {
 				logrus.Warnf("Error handling changed files: %v", err)
 				// If context was canceled, exit immediately
@@ -557,7 +565,9 @@ func (t tarDockerClient) Untar(ctx context.Context, id string, archive io.ReadCl
 	return err
 }
 
-func (s *composeService) handleWatchBatch(ctx context.Context, project *types.Project, options api.WatchOptions, batch []watch.FileEvent, rules []watchRule, syncer sync.Syncer) error {
+func (s *composeService) handleWatchBatch(ctx context.Context, project *types.Project, options api.WatchOptions,
+	batch []watch.FileEvent, rules []watchRule, syncer sync.Syncer, scheduler *rebuildScheduler,
+) error {
 	var (
 		restart   = map[string]bool{}
 		syncfiles = map[string][]*sync.PathMapping{}
@@ -590,15 +600,38 @@ func (s *composeService) handleWatchBatch(ctx context.Context, project *types.Pr
 		}
 	}
 
-	logrus.Debugf("watch actions: rebuild %d sync %d restart %d", len(rebuild), len(syncfiles), len(restart))
-
 	if len(rebuild) > 0 {
-		err := s.rebuild(ctx, project, utils.MapKeys(rebuild), options)
-		if err != nil {
-			return err
+		scheduler.Request(utils.MapKeys(rebuild))
+	}
+
+	// A rebuild already recreates and starts the service, so a separate
+	// restart is redundant at best and races the rebuild's create/start at
+	// worst. Which rebuild matters:
+	//   - pending: it will snapshot the build context after this change —
+	//     drop the restart, the rebuild converges on its own;
+	//   - in flight: its snapshot predates this change, so its outcome is
+	//     stale — fold the restart into a Request, which interrupts the
+	//     doomed run and rebuilds from a fresh snapshot (recreate + start:
+	//     the restart intent, converged).
+	for service := range restart {
+		switch {
+		case scheduler.Pending(service):
+			logrus.Debugf("skipping restart for service %q: rebuild pending", service)
+			delete(restart, service)
+		case scheduler.InFlight(service):
+			logrus.Debugf("turning restart for service %q into a rebuild: stale rebuild in flight", service)
+			scheduler.Request([]string{service})
+			delete(restart, service)
 		}
 	}
 
+	logrus.Debugf("watch actions: rebuild %d sync %d restart %d", len(rebuild), len(syncfiles), len(restart))
+
+	// A sync or exec below may still target a service whose container is
+	// being replaced by an in-flight asynchronous rebuild (from this batch or
+	// an earlier one): unlike restart, this is a conscious tradeoff, since
+	// syncer.Sync/exec resolve containers by lookup and simply fail loudly
+	// (rather than racing the container lifecycle) if the container is gone.
 	for serviceName, pathMappings := range syncfiles {
 		writeWatchSyncMessage(options.LogTo, serviceName, pathMappings)
 		err := syncer.Sync(ctx, serviceName, pathMappings)
diff --git a/pkg/compose/watch_test.go b/pkg/compose/watch_test.go
index 963b5180b5..e217d627e8 100644
--- a/pkg/compose/watch_test.go
+++ b/pkg/compose/watch_test.go
@@ -21,6 +21,7 @@ import (
 	"os"
 	"path/filepath"
 	"slices"
+	"sync/atomic"
 	"testing"
 	"testing/synctest"
 	"time"
@@ -159,19 +160,77 @@ func TestWatch_Sync(t *testing.T) {
 		})
 		assert.DeepEqual(t, expected, actual)
 
-		// Rebuild fails before sync actions from the same batch are processed.
+		// The rebuild triggered by "/rebuild" now runs asynchronously, so it no
+		// longer blocks the sync of "/sync/changed" from the same batch.
+		// synctest.Wait() only returns once the rebuild's goroutine has
+		// settled (it runs to completion here, exercising the mocked
+		// ImageList/ImageRemove prune calls), so the mock expectations above
+		// are already satisfied by the time we get here.
 		watcher.Events() <- watch.NewFileEvent("/rebuild")
 		watcher.Events() <- watch.NewFileEvent("/sync/changed")
 		time.Sleep(watch.QuietPeriod)
 		synctest.Wait()
-		select {
-		case batch := <-syncer.synced:
-			t.Fatalf("received unexpected events: %v", batch)
-		default:
-			// expected
+		actual = <-syncer.synced
+		expected = []*sync.PathMapping{
+			{HostPath: "/sync/changed", ContainerPath: "/work/changed"},
+		}
+		assert.DeepEqual(t, expected, actual)
+	})
+}
+
+// A rebuild running in the background from an earlier batch must not be
+// raced by a plain restart of the same service triggered by a later batch:
+// the restart is folded into the scheduler instead — the stale run is
+// interrupted and a fresh rebuild (recreate + start, the restart intent)
+// converges on a context snapshot taken after the change.
+func TestHandleWatchBatch_RestartDuringInFlightRebuildConverges(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	cli := mocks.NewMockCli(mockCtrl)
+	cli.EXPECT().Err().Return(streams.NewOut(os.Stderr)).AnyTimes()
+	// A correctly folded restart never touches the Docker client at all.
+	cli.EXPECT().Client().Times(0)
+	service := composeService{dockerCli: cli}
+
+	proj := types.Project{
+		Name: "myProjectName",
+		Services: types.Services{
+			"test": {Name: "test"},
+		},
+	}
+
+	rules, err := getWatchRules(&types.DevelopConfig{
+		Watch: []types.Trigger{
+			{Path: "/restart", Action: "restart"},
+		},
+	}, types.ServiceConfig{Name: "test"})
+	assert.NilError(t, err)
+
+	var runs int32
+	started := make(chan struct{}, 2)
+	release := make(chan error)
+	scheduler := newRebuildScheduler(t.Context(), func(ctx context.Context, _ []string) error {
+		n := atomic.AddInt32(&runs, 1)
+		started <- struct{}{}
+		if n == 1 {
+			// the stale run only ever ends by interruption
+			<-ctx.Done()
+			return ctx.Err()
 		}
-		// TODO: there's not a great way to assert that the rebuild attempt happened
+		return <-release
 	})
+	scheduler.Request([]string{"test"}) // simulate a rebuild still running from an earlier batch
+	<-started
+
+	err = service.handleWatchBatch(t.Context(), &proj, api.WatchOptions{LogTo: stdLogger{}},
+		[]watch.FileEvent{watch.NewFileEvent("/restart")}, rules, newFakeSyncer(), scheduler)
+	assert.NilError(t, err)
+
+	// the trailing rebuild starting at all proves the stale run was
+	// interrupted and the restart folded into a fresh rebuild
+	<-started
+	release <- nil
+	scheduler.Wait()
+	assert.Equal(t, atomic.LoadInt32(&runs), int32(2))
 }
 
 type fakeSyncer struct {