diff --git a/pkg/api/labels.go b/pkg/api/labels.go
index 9122fb5a83..53e863e31d 100644
--- a/pkg/api/labels.go
+++ b/pkg/api/labels.go
@@ -67,6 +67,11 @@ const (
 	// runPreStartHook so orphan hook containers from a previous failed run can
 	// be found and removed by project+service+hook label filters.
 	HookLabel = "com.docker.compose.hook"
+	// HookIndexLabel stores the position of the hook in its service's hook
+	// list (e.g. pre_start[2] → "2"), so the start phase can match each
+	// declared hook with the runner container the reconciliation plan
+	// prepared for it.
+	HookIndexLabel = "com.docker.compose.hook-index"
 )
 
 // ComposeVersion is the compose tool version as declared by label VersionLabel
diff --git a/pkg/compose/executor.go b/pkg/compose/executor.go
index c23b3f7b21..5be4beab1a 100644
--- a/pkg/compose/executor.go
+++ b/pkg/compose/executor.go
@@ -160,6 +160,8 @@ func (exec *planExecutor) executeNode(ctx context.Context, node *PlanNode) error
 		return exec.execRemoveContainer(ctx, op)
 	case OpRenameContainer:
 		return exec.execRenameContainer(ctx, node)
+	case OpCreateHookContainer:
+		return exec.execCreateHookContainer(ctx, node)
 	case OpRunProvider:
 		return exec.compose.runPlugin(ctx, exec.project, *op.Service, "up")
 	default:
diff --git a/pkg/compose/executor_ops.go b/pkg/compose/executor_ops.go
index 13738d045b..2b515a0a34 100644
--- a/pkg/compose/executor_ops.go
+++ b/pkg/compose/executor_ops.go
@@ -142,8 +142,15 @@ func (exec *planExecutor) execStopContainer(ctx context.Context, op Operation) e
 }
 
 func (exec *planExecutor) execRemoveContainer(ctx context.Context, op Operation) error {
-	_, err := exec.compose.apiClient().ContainerRemove(ctx, op.Container.ID, client.ContainerRemoveOptions{Force: true})
+	_, err := exec.compose.apiClient().ContainerRemove(ctx, op.Container.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: op.RemoveVolumes})
 	if err != nil {
+		if op.BestEffort {
+			// warn-only removal (stale pre_start hook runner): the container
+			// stays visible to the operator, the plan carries on — and the
+			// live view below keeps it, since it was not removed
+			logrus.Warnf("failed to remove %s: %v", op.ResourceID, err)
+			return nil
+		}
 		return err
 	}
 	// Why: a dependent service's create may resolve `network_mode: service:X`
@@ -161,6 +168,32 @@ func (exec *planExecutor) execRemoveContainer(ctx context.Context, op Operation)
 	return nil
 }
 
+// execCreateHookContainer creates the runner container for one pre_start hook.
+// The target replica — whose volumes the hook shares via VolumesFrom — is
+// resolved from the live view at execution time: the node depends on every
+// container operation of its service, so the view is final here, and the
+// lowest-numbered replica matches the one the start phase hands the hooks.
+func (exec *planExecutor) execCreateHookContainer(ctx context.Context, node *PlanNode) error {
+	op := node.Operation
+	service := *op.Service
+	exec.containersMu.Lock()
+	replicas := slices.Clone(exec.containersByService[service.Name])
+	exec.containersMu.Unlock()
+	if len(replicas) == 0 {
+		return fmt.Errorf("internal: no %q container to attach pre_start hook %d to", service.Name, op.HookIndex)
+	}
+	target := lowestNumberedContainer(replicas)
+	created, err := exec.compose.createPreStartContainer(ctx, exec.project, service, target, op.HookIndex, op.Name)
+	if err != nil {
+		return err
+	}
+	exec.pctx.set(node.ID, operationResult{
+		ContainerID:   created.ID,
+		ContainerName: op.Name,
+	})
+	return nil
+}
+
 func (exec *planExecutor) execRenameContainer(ctx context.Context, node *PlanNode) error {
 	op := node.Operation
 	if op.CreateNodeID == 0 {
diff --git a/pkg/compose/executor_test.go b/pkg/compose/executor_test.go
index 9e711c22f3..ea60259503 100644
--- a/pkg/compose/executor_test.go
+++ b/pkg/compose/executor_test.go
@@ -174,6 +174,38 @@ func emptyObservedState(project string) *ObservedState {
 // Goes through newPlanExecutor + run (i.e. the same code path executePlan
 // uses in production) so the test exercises the errgroup, done-channel
 // wiring and group tracker — not a hand-rolled loop over executeNode.
+// A best-effort removal failure (stale pre_start hook runner purge) is
+// warn-only: the plan carries on, and the removal passes RemoveVolumes so the
+// runner's anonymous volumes go with it — the imperative purge semantics.
+func TestExecutePlanBestEffortRemoveContainerFailureTolerated(t *testing.T) {
+	svc, apiClient := newTestService(t)
+
+	ctr := container.Summary{
+		ID:     "hook1",
+		Names:  []string{"/some-hook-runner"},
+		Labels: map[string]string{api.ServiceLabel: "web"},
+	}
+
+	apiClient.EXPECT().ContainerRemove(gomock.Any(), "hook1", gomock.Any()).
+		DoAndReturn(func(_ context.Context, _ string, opts client.ContainerRemoveOptions) (client.ContainerRemoveResult, error) {
+			assert.Assert(t, opts.RemoveVolumes, "hook-runner purge must drop anonymous volumes")
+			return client.ContainerRemoveResult{}, errors.New("device or resource busy")
+		})
+
+	plan := &Plan{}
+	plan.addNode(Operation{
+		Type:          OpRemoveContainer,
+		ResourceID:    "hook:web:stale:hook1",
+		Cause:         "stale pre_start hook container",
+		Container:     &ctr,
+		RemoveVolumes: true,
+		BestEffort:    true,
+	}, "")
+
+	err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, emptyObservedState("test"), plan)
+	assert.NilError(t, err)
+}
+
 func TestExecutePlanRemoveContainerDropsFromCache(t *testing.T) {
 	svc, apiClient := newTestService(t)
 
@@ -486,3 +518,83 @@ type conflictError struct{}
 
 func (conflictError) Error() string { return "conflict" }
 func (conflictError) Conflict()     {}
+
+// TestExecutePlanCreateHookContainer verifies the hook-runner create op: the
+// target replica is resolved from the executor's live view (lowest
+// container-number, mirroring what the start phase hands the hooks) and the
+// runner is created under the deterministic name carried by the operation.
+func TestExecutePlanCreateHookContainer(t *testing.T) {
+	svc, apiClient := newTestService(t)
+	apiClient.EXPECT().Ping(gomock.Any(), client.PingOptions{NegotiateAPIVersion: true}).
+		Return(client.PingResult{APIVersion: "1.44"}, nil).AnyTimes()
+	apiClient.EXPECT().ClientVersion().Return("1.44").AnyTimes()
+
+	service := types.ServiceConfig{
+		Name:  "web",
+		Image: "alpine",
+		PreStart: []types.ServiceHook{
+			{Command: types.ShellCommand{"init"}},
+		},
+	}
+	observed := &ObservedState{
+		ProjectName: "test",
+		Containers: map[string][]ObservedContainer{
+			"web": {
+				// Deliberately listed out of order: replica 2 first.
+				{ID: "c2", Summary: container.Summary{ID: "c2", Labels: map[string]string{api.ContainerNumberLabel: "2"}}},
+				{ID: "c1", Summary: container.Summary{ID: "c1", Labels: map[string]string{api.ContainerNumberLabel: "1"}}},
+			},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	var gotOpts client.ContainerCreateOptions
+	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
+		DoAndReturn(func(_ context.Context, opts client.ContainerCreateOptions) (client.ContainerCreateResult, error) {
+			gotOpts = opts
+			return client.ContainerCreateResult{ID: "hook-1"}, nil
+		})
+
+	plan := &Plan{}
+	plan.addNode(Operation{
+		Type:       OpCreateHookContainer,
+		ResourceID: "hook:web:pre_start:0",
+		Cause:      "pre_start hook",
+		Service:    &service,
+		HookIndex:  0,
+		Name:       getHookContainerName("test", "web", 0),
+	}, "")
+
+	err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, observed, plan)
+	assert.NilError(t, err)
+	assert.Equal(t, gotOpts.Name, "test-web-pre_start-0")
+	assert.DeepEqual(t, gotOpts.HostConfig.VolumesFrom, []string{"c1"})
+	assert.Equal(t, gotOpts.Config.Labels[api.HookIndexLabel], "0")
+}
+
+// TestExecutePlanCreateHookContainerNoReplica: a hook-create node scheduled
+// with no replica in the live view is an internal planning error — the
+// reconciler guarantees the node depends on the replica's create.
+func TestExecutePlanCreateHookContainerNoReplica(t *testing.T) {
+	svc, _ := newTestService(t)
+
+	service := types.ServiceConfig{
+		Name:     "web",
+		Image:    "alpine",
+		PreStart: []types.ServiceHook{{Command: types.ShellCommand{"init"}}},
+	}
+
+	plan := &Plan{}
+	plan.addNode(Operation{
+		Type:       OpCreateHookContainer,
+		ResourceID: "hook:web:pre_start:0",
+		Cause:      "pre_start hook",
+		Service:    &service,
+		HookIndex:  0,
+		Name:       getHookContainerName("test", "web", 0),
+	}, "")
+
+	err := svc.executePlan(t.Context(), &types.Project{Name: "test"}, emptyObservedState("test"), plan)
+	assert.ErrorContains(t, err, `no "web" container`)
+}
diff --git a/pkg/compose/observed_state.go b/pkg/compose/observed_state.go
index 25d0e3da6c..32a272ff3a 100644
--- a/pkg/compose/observed_state.go
+++ b/pkg/compose/observed_state.go
@@ -46,6 +46,13 @@ type ObservedState struct {
 	// others as orphans (see selectNetwork/selectVolume).
 	Networks map[string][]ObservedNetwork // compose network key → observed
 	Volumes  map[string][]ObservedVolume  // compose volume key → observed
+	// HookContainers are ephemeral lifecycle-hook runners (HookLabel set),
+	// per service: fresh runners prepared by a previous plan and not yet
+	// consumed, or leftovers of a run that failed before removing them.
+	// Whenever the hooks are going to run again, the reconciler purges every
+	// observed runner and plans fresh ones (see planPreStartHookRunners), so
+	// the set always converges to the current service definition.
+	HookContainers map[string][]ObservedContainer // service name → hook containers
 }
 
 // selectNetwork picks, among the live networks recorded for a compose key, the
@@ -148,6 +155,8 @@ func (s *composeService) collectObservedState(ctx context.Context, project *type
 		Containers:  map[string][]ObservedContainer{},
 		Networks:    map[string][]ObservedNetwork{},
 		Volumes:     map[string][]ObservedVolume{},
+
+		HookContainers: map[string][]ObservedContainer{},
 	}
 
 	// --- Containers ---
@@ -170,6 +179,18 @@ func (s *composeService) collectObservedState(ctx context.Context, project *type
 
 	for _, ctr := range raw {
 		svcName := ctr.Labels[api.ServiceLabel]
+		if ctr.Labels[api.HookLabel] != "" && knownServices[svcName] {
+			// lifecycle-hook containers (ephemeral pre_start runners) are
+			// neither service replicas nor one-offs: classified apart, so
+			// they never masquerade as a replica (they carry no
+			// container-number label and would otherwise read as number 0)
+			// and the reconciler can plan purging stale ones. A hook
+			// container whose service left the model falls through to the
+			// orphan check below instead — nothing plans purges for an
+			// unknown service, and --remove-orphans must keep cleaning it.
+			state.HookContainers[svcName] = append(state.HookContainers[svcName], toObservedContainer(ctr))
+			continue
+		}
 		if isNotOneOff(ctr) && knownServices[svcName] {
 			state.Containers[svcName] = append(state.Containers[svcName], toObservedContainer(ctr))
 		} else if isOrphaned(project)(ctr) {
diff --git a/pkg/compose/observed_state_test.go b/pkg/compose/observed_state_test.go
index cc24868de4..a09a092018 100644
--- a/pkg/compose/observed_state_test.go
+++ b/pkg/compose/observed_state_test.go
@@ -160,6 +160,33 @@ func TestCollectObservedState(t *testing.T) {
 					api.OneoffLabel:  "True",
 				},
 			},
+			{
+				// Stale lifecycle-hook runner (a previous run failed before
+				// removing it): neither a replica (it has no container-number
+				// label and must not read as number 0) nor a one-off —
+				// classified apart so the reconciler can plan its purge.
+				ID:    "c5",
+				Names: []string{"/hook-runner"},
+				State: container.StateExited,
+				Labels: map[string]string{
+					api.ServiceLabel: "web",
+					api.ProjectLabel: "myproject",
+					api.HookLabel:    "pre_start",
+				},
+			},
+			{
+				// Hook runner whose service left the model: nothing plans
+				// purges for an unknown service, so it must keep flowing to
+				// the orphan path --remove-orphans cleans.
+				ID:    "c6",
+				Names: []string{"/old-hook-runner"},
+				State: container.StateExited,
+				Labels: map[string]string{
+					api.ServiceLabel: "old",
+					api.ProjectLabel: "myproject",
+					api.HookLabel:    "pre_start",
+				},
+			},
 		},
 	}, nil)
 
@@ -202,11 +229,20 @@ func TestCollectObservedState(t *testing.T) {
 	assert.Equal(t, len(state.Containers["db"]), 1)
 	assert.Equal(t, state.Containers["db"][0].ID, "c2")
 
-	// Orphans: only the model-absent service "old". The running one-off c4 is
-	// absent everywhere — not in the "web" bucket (asserted above: 1 replica),
-	// not an orphan: up leaves live `compose run` sessions alone.
-	assert.Equal(t, len(state.Orphans), 1)
+	// The hook runner is classified apart — not a "web" replica (asserted
+	// above: 1 replica), not an orphan
+	assert.Equal(t, len(state.HookContainers["web"]), 1)
+	assert.Equal(t, state.HookContainers["web"][0].ID, "c5")
+
+	// Orphans: the model-absent service "old" — its replica AND its hook
+	// runner (c6), which must not hide in HookContainers where nothing would
+	// ever purge it. The running one-off c4 is absent everywhere — not in the
+	// "web" bucket (asserted above: 1 replica), not an orphan: up leaves live
+	// `compose run` sessions alone.
+	assert.Equal(t, len(state.Orphans), 2)
 	assert.Equal(t, state.Orphans[0].ID, "c3")
+	assert.Equal(t, state.Orphans[1].ID, "c6")
+	assert.Equal(t, len(state.HookContainers["old"]), 0)
 
 	// Networks
 	assert.Equal(t, len(state.Networks), 1)
diff --git a/pkg/compose/plan.go b/pkg/compose/plan.go
index 6a118f12e0..fa6bbc1f6c 100644
--- a/pkg/compose/plan.go
+++ b/pkg/compose/plan.go
@@ -53,6 +53,10 @@ const (
 
 	// Provider operations
 	OpRunProvider OperationType = 30
+
+	// Hook operations. 40-42 are reserved for the start-phase operations
+	// (wait condition, pre_start run, post_start run).
+	OpCreateHookContainer OperationType = 43
 )
 
 // String returns the human-readable name of an OperationType.
@@ -82,6 +86,8 @@ func (o OperationType) String() string {
 		return "RenameContainer"
 	case OpRunProvider:
 		return "RunProvider"
+	case OpCreateHookContainer:
+		return "CreateHookContainer"
 	default:
 		return fmt.Sprintf("Unknown(%d)", int(o))
 	}
@@ -103,11 +109,16 @@ type Operation struct {
 	Volume       *types.VolumeConfig  // for volume operations
 	Timeout      *time.Duration       // for stop operations
 	CreateNodeID int                  // for OpRenameContainer: ID of the CreateContainer node whose result to rename
-	// BestEffort marks an operation whose failure must not abort the plan. It is
-	// used for the optional removal of the old network on a rename: if the
-	// network is still in use (by non-Compose containers) the removal is skipped
-	// with a warning instead of failing — the new network already carries a
-	// different name, so the migration does not depend on the old one going away.
+	HookIndex    int                  // for OpCreateHookContainer: position of the hook in the service's hook list
+	// RemoveVolumes asks OpRemoveContainer to also remove the container's
+	// anonymous volumes — the imperative semantics for hook-runner containers.
+	RemoveVolumes bool
+	// BestEffort marks an operation whose failure must not abort the plan.
+	// Used for the optional removal of the old network on a rename (if the
+	// network is still in use by non-Compose containers the removal is skipped
+	// with a warning — the new network already carries a different name), and
+	// for purging stale pre_start hook runners (the imperative purge is
+	// warn-only: a failed removal leaves the container visible, never blocks).
 	BestEffort bool
 }
 
diff --git a/pkg/compose/pre_start.go b/pkg/compose/pre_start.go
index 4d70c54bbd..97b04e651a 100644
--- a/pkg/compose/pre_start.go
+++ b/pkg/compose/pre_start.go
@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"io"
 	"strconv"
+	"strings"
 
 	"github.com/compose-spec/compose-go/v2/types"
 	"github.com/moby/moby/api/pkg/stdcopy"
@@ -38,6 +39,14 @@ import (
 // by a project+service+hook label filter.
 const preStartHookType = "pre_start"
 
+// getHookContainerName builds the deterministic name of a pre_start runner
+// container, e.g. "myproject-db-pre_start-0". A stable name makes the runner
+// an addressable resource of the reconciliation plan: repeated plans converge
+// on the same container instead of accumulating anonymous ones.
+func getHookContainerName(projectName, serviceName string, index int) string {
+	return strings.Join([]string{projectName, serviceName, preStartHookType, strconv.Itoa(index)}, api.Separator)
+}
+
 // lowestNumberedContainer returns the container with the lowest
 // com.docker.compose.container-number label, so pre_start always targets the
 // same replica regardless of the order the daemon returned them in.
@@ -55,50 +64,93 @@ func lowestNumberedContainer(containers Containers) container.Summary {
 }
 
 // runPreStart executes the service's pre_start hooks sequentially, in declared
-// order. Each hook runs as an ephemeral container that shares the service
-// container's volumes via VolumesFrom and is attached to the same networks.
-// A non-zero exit gates service start.
+// order. Each hook runs in a runner container prepared by the reconciliation
+// plan (see planPreStartHookRunners) that shares the service container's
+// volumes via VolumesFrom and is attached to the same networks. A non-zero
+// exit gates service start.
+//
+// runPreStart never creates a runner itself: a declared hook without a fresh
+// runner is an error telling the user to reconcile — the runners were either
+// consumed by a previous start (e.g. `stop` then `start`) or never prepared.
 //
 // With per_replica: false (the only currently supported mode), the hook sees
 // the volumes of the first non-running replica only — anonymous volumes and
 // tmpfs mounts are per-replica and not shared. Use named volumes or bind
 // mounts for data the hook produces.
-func (s *composeService) runPreStart(ctx context.Context, project *types.Project, service types.ServiceConfig, ctr container.Summary, listener api.ContainerEventListener) error {
+func (s *composeService) runPreStart(ctx context.Context, project *types.Project, service types.ServiceConfig, listener api.ContainerEventListener) error {
 	// Validate every hook up front so an unsupported entry never triggers any I/O.
 	for i, hook := range service.PreStart {
 		if hook.PerReplica {
 			return fmt.Errorf("service %q pre_start[%d]: per_replica is not yet supported; remove per_replica or set it to false", service.Name, i)
 		}
 	}
-	// Remove any hook containers left behind by a previous failed run so they do
-	// not accumulate. Only one orphan can exist per service (failure gates the
-	// remaining hooks), but we clean the whole set in case the service definition
-	// changed between runs. Removal failures are non-fatal: they are logged so
-	// the operator can identify the stale container.
-	if err := s.removeOrphanPreStartContainers(ctx, project.Name, service.Name); err != nil {
-		logrus.Warnf("service %q: failed to remove stale pre_start hook containers: %v", service.Name, err)
+	runners, err := s.listPreStartRunners(ctx, project.Name, service.Name)
+	if err != nil {
+		return err
 	}
-	for i, hook := range service.PreStart {
-		if err := s.runPreStartHook(ctx, project, service, ctr, i, hook, listener); err != nil {
+	for i := range service.PreStart {
+		runner, ok := runners[i]
+		if !ok {
+			return fmt.Errorf("service %q pre_start[%d]: no hook runner container found — runners are prepared when the service is created and consumed when its hooks run; run %q to prepare them again",
+				service.Name, i, "docker compose up "+service.Name)
+		}
+		if err := s.execPreStartHook(ctx, service, i, runner.ID, listener); err != nil {
 			return err
 		}
+		// Success: remove the hook container, mirroring the old AutoRemove behaviour
+		// (including its anonymous volumes). A removal failure is logged but does not
+		// gate service start — the hook already succeeded.
+		if _, removeErr := s.apiClient().ContainerRemove(ctx, runner.ID, client.ContainerRemoveOptions{RemoveVolumes: true}); removeErr != nil {
+			logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s: %v", service.Name, i, runner.ID, removeErr)
+		}
 	}
 	return nil
 }
 
-func (s *composeService) runPreStartHook(
-	ctx context.Context, project *types.Project, service types.ServiceConfig,
-	ctr container.Summary, index int, hook types.ServiceHook, listener api.ContainerEventListener,
-) error {
-	created, err := s.createPreStartContainer(ctx, project, service, ctr, hook)
+// listPreStartRunners returns the service's created-state pre_start hook
+// runner containers, indexed by their HookIndexLabel. Runners in any other
+// state are ignored — a failed hook retained for inspection or an
+// old-generation container without an index label is not executable; only a
+// fresh runner prepared by the reconciliation plan is.
+func (s *composeService) listPreStartRunners(ctx context.Context, projectName, serviceName string) (map[int]container.Summary, error) {
+	f := projectFilter(projectName)
+	f.Add("label", serviceFilter(serviceName))
+	f.Add("label", hookFilter(preStartHookType))
+	res, err := s.apiClient().ContainerList(ctx, client.ContainerListOptions{
+		All:     true,
+		Filters: f,
+	})
 	if err != nil {
-		return err
+		return nil, err
+	}
+	runners := map[int]container.Summary{}
+	for _, ctr := range res.Items {
+		if ctr.State != container.StateCreated {
+			continue
+		}
+		index, err := strconv.Atoi(ctr.Labels[api.HookIndexLabel])
+		if err != nil {
+			continue
+		}
+		runners[index] = ctr
 	}
+	return runners, nil
+}
 
+// execPreStartHook starts an already-created hook container, streams its logs
+// and waits for its exit. It owns only execution-failure handling: a container
+// that never started or a run cancelled by the user is removed, a genuinely
+// failed hook is retained for post-mortem inspection. Removing the container
+// after a successful run is the caller's job — the runner was prepared by the
+// reconciliation plan and is consumed (removed) by runPreStart on success.
+func (s *composeService) execPreStartHook(
+	ctx context.Context, service types.ServiceConfig,
+	index int, containerID string, listener api.ContainerEventListener,
+) error {
 	// Subscribe to wait before start to avoid missing the exit event for short-lived hooks.
 	// WaitConditionNotRunning would match immediately because the container is still in
 	// "created" state, so use WaitConditionNextExit to block until the run actually finishes.
-	waitRes := s.apiClient().ContainerWait(ctx, created.ID, client.ContainerWaitOptions{
+	waitRes := s.apiClient().ContainerWait(ctx, containerID, client.ContainerWaitOptions{
 		Condition: container.WaitConditionNextExit,
 	})
 
@@ -108,13 +160,13 @@ func (s *composeService) runPreStartHook(
 	// open cannot deadlock `<-logsDone`.
 	logCtx, cancelLogs := context.WithCancel(ctx)
 	defer cancelLogs()
-	logsDone, getTail := s.streamPreStartLogs(logCtx, created.ID, service, index, listener)
+	logsDone, getTail := s.streamPreStartLogs(logCtx, containerID, service, index, listener)
 
-	if _, err := s.apiClient().ContainerStart(ctx, created.ID, client.ContainerStartOptions{}); err != nil {
+	if _, err := s.apiClient().ContainerStart(ctx, containerID, client.ContainerStartOptions{}); err != nil {
 		// AutoRemove is false, so we must remove the never-started container
 		// explicitly. A failed removal is logged so the orphan is visible.
-		if _, removeErr := s.apiClient().ContainerRemove(ctx, created.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
-			logrus.Warnf("service %q pre_start[%d]: failed to remove orphan hook container %s: %v", service.Name, index, created.ID, removeErr)
+		if _, removeErr := s.apiClient().ContainerRemove(ctx, containerID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
+			logrus.Warnf("service %q pre_start[%d]: failed to remove orphan hook container %s: %v", service.Name, index, containerID, removeErr)
 		}
 		// Drain waitRes so the client's wait goroutine exits without having to
 		// wait for the parent context to be canceled.
@@ -136,15 +188,15 @@ func (s *composeService) runPreStartHook(
 		// and return the raw context error without decorating it with the tail or
 		// retaining the container for post-mortem inspection.
 		if ctx.Err() != nil {
-			if _, removeErr := s.apiClient().ContainerRemove(context.Background(), created.ID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
-				logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s after cancellation: %v", service.Name, index, created.ID, removeErr)
+			if _, removeErr := s.apiClient().ContainerRemove(context.Background(), containerID, client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}); removeErr != nil {
+				logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s after cancellation: %v", service.Name, index, containerID, removeErr)
 			}
 			return waitErr
 		}
 		// Genuine hook failure: retain the container so the operator can run
 		// `docker logs <id>` and `docker inspect <id>` to diagnose the failure.
 		// Include the short container ID in the error to make it actionable.
-		shortID := created.ID
+		shortID := containerID
 		if len(shortID) > 12 {
 			shortID = shortID[:12]
 		}
@@ -153,19 +205,19 @@ func (s *composeService) runPreStartHook(
 		}
 		return fmt.Errorf("%w (hook container %s retained for inspection)", waitErr, shortID)
 	}
-	// Success: remove the hook container, mirroring the old AutoRemove behaviour
-	// (including its anonymous volumes). A removal failure is logged but does not
-	// gate service start — the hook already succeeded.
-	if _, removeErr := s.apiClient().ContainerRemove(ctx, created.ID, client.ContainerRemoveOptions{RemoveVolumes: true}); removeErr != nil {
-		logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s: %v", service.Name, index, created.ID, removeErr)
-	}
 	return nil
 }
 
+// createPreStartContainer creates the runner container for the index-th
+// pre_start hook of service, named after getHookContainerName and stamped
+// with the hook labels so both the start phase (by index) and the purge (by
+// hook type) can find it. It only creates: the runner is left in created
+// state for the start phase to execute.
 func (s *composeService) createPreStartContainer(
 	ctx context.Context, project *types.Project, service types.ServiceConfig,
-	ctr container.Summary, hook types.ServiceHook,
+	ctr container.Summary, index int, name string,
 ) (client.ContainerCreateResult, error) {
+	hook := service.PreStart[index]
 	image := hook.Image
 	if image == "" {
 		image = api.GetImageNameOrDefault(service, project.Name)
@@ -182,10 +234,11 @@ func (s *composeService) createPreStartContainer(
 		// HookLabel also distinguishes hook containers from the real service
 		// container (which shares ProjectLabel and ServiceLabel).
 		Labels: map[string]string{
-			api.ProjectLabel: project.Name,
-			api.ServiceLabel: service.Name,
-			api.VersionLabel: api.ComposeVersion,
-			api.HookLabel:    preStartHookType,
+			api.ProjectLabel:   project.Name,
+			api.ServiceLabel:   service.Name,
+			api.VersionLabel:   api.ComposeVersion,
+			api.HookLabel:      preStartHookType,
+			api.HookIndexLabel: strconv.Itoa(index),
 		},
 	}
 	hostCfg := &container.HostConfig{
@@ -209,6 +262,7 @@ func (s *composeService) createPreStartContainer(
 	hostCfg.NetworkMode = networkMode
 
 	created, err := s.apiClient().ContainerCreate(ctx, client.ContainerCreateOptions{
+		Name:             name,
 		Config:           cfg,
 		HostConfig:       hostCfg,
 		NetworkingConfig: networkingConfig,
@@ -253,32 +307,6 @@ func (s *composeService) connectPreStartExtraNetworks(ctx context.Context, proje
 	return nil
 }
 
-// removeOrphanPreStartContainers finds and force-removes any hook containers
-// left behind by a previous failed run of this service's pre_start hooks.
-// Containers are identified by project + service + HookLabel=pre_start.
-// Removal failures are logged at warn level and do not abort the run.
-func (s *composeService) removeOrphanPreStartContainers(ctx context.Context, projectName, serviceName string) error {
-	f := projectFilter(projectName)
-	f.Add("label", serviceFilter(serviceName))
-	f.Add("label", hookFilter(preStartHookType))
-	res, err := s.apiClient().ContainerList(ctx, client.ContainerListOptions{
-		All:     true,
-		Filters: f,
-	})
-	if err != nil {
-		return err
-	}
-	for _, ctr := range res.Items {
-		if _, removeErr := s.apiClient().ContainerRemove(ctx, ctr.ID, client.ContainerRemoveOptions{
-			Force:         true,
-			RemoveVolumes: true,
-		}); removeErr != nil {
-			logrus.Warnf("failed to remove stale pre_start hook container %s: %v", ctr.ID, removeErr)
-		}
-	}
-	return nil
-}
-
 func waitPreStart(ctx context.Context, serviceName string, index int, waitRes client.ContainerWaitResult) error {
 	// ContainerWait can deliver on Result and Error at the same instant. Two
 	// races have to be closed deterministically here:
diff --git a/pkg/compose/pre_start_test.go b/pkg/compose/pre_start_test.go
index 635eef6739..6da00e7335 100644
--- a/pkg/compose/pre_start_test.go
+++ b/pkg/compose/pre_start_test.go
@@ -21,6 +21,7 @@ import (
 	"context"
 	"errors"
 	"io"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -69,16 +70,29 @@ func emptyLogs() client.ContainerLogsResult {
 	return io.NopCloser(bytes.NewReader(nil))
 }
 
-// expectEmptyOrphanScan sets up the ContainerList expectation for the orphan
-// pre_start cleanup that happens once at the start of every runPreStart call
-// (after the per_replica validation loop). It returns the empty list.
-func expectEmptyOrphanScan(apiClient *mocks.MockAPIClient) *gomock.Call {
+// runnerSummary builds the container.Summary of a hook runner as prepared by
+// the reconciliation plan: created state, hook labels carrying the index.
+func runnerSummary(id string, index int) container.Summary {
+	return container.Summary{
+		ID:    id,
+		State: container.StateCreated,
+		Labels: map[string]string{
+			api.HookLabel:      preStartHookType,
+			api.HookIndexLabel: strconv.Itoa(index),
+		},
+	}
+}
+
+// expectRunnerScan sets up the ContainerList expectation for the runner lookup
+// that happens once at the start of every runPreStart call (after the
+// per_replica validation loop). It returns the given runners.
+func expectRunnerScan(apiClient *mocks.MockAPIClient, runners ...container.Summary) *gomock.Call {
 	return apiClient.EXPECT().
 		ContainerList(gomock.Any(), gomock.Any()).
-		Return(client.ContainerListResult{}, nil)
+		Return(client.ContainerListResult{Items: runners}, nil)
 }
 
-// expectSuccessRemove sets up the ContainerRemove call that runPreStartHook
+// expectSuccessRemove sets up the ContainerRemove call that runPreStart
 // makes after a successful hook run (mirrors old AutoRemove behaviour).
 func expectSuccessRemove(apiClient *mocks.MockAPIClient, hookID string) *gomock.Call {
 	return apiClient.EXPECT().
@@ -98,34 +112,29 @@ func TestPreStart_SuccessTwoHooksInOrder(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"echo", "second"}},
 		},
 	}
-	ctr := container.Summary{ID: "service-ctr-id"}
 
-	// Orphan cleanup runs once before the first hook.
-	scan := expectEmptyOrphanScan(apiClient)
+	// Both runners were prepared by the plan; the scan runs once up front.
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0), runnerSummary("hook-2", 1))
 
-	// Hook 1: create → wait (subscribe) → logs (subscribe) → start → remove.
-	create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	// Hook 1: wait (subscribe) → logs (subscribe) → start → remove.
 	wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0)).After(create1)
+		Return(waitResultExit(0)).After(scan)
 	logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil).After(wait1)
 	start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
 		Return(client.ContainerStartResult{}, nil).After(logs1)
 	remove1 := expectSuccessRemove(apiClient, "hook-1").After(start1)
 
-	// Hook 2 is only created after hook 1 has been removed.
-	create2 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-2"}, nil).After(remove1)
+	// Hook 2 only runs after hook 1 has been removed.
 	wait2 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-2", gomock.Any()).
-		Return(waitResultExit(0)).After(create2)
+		Return(waitResultExit(0)).After(remove1)
 	logs2 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-2", gomock.Any()).
 		Return(emptyLogs(), nil).After(wait2)
 	start2 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-2", gomock.Any()).
 		Return(client.ContainerStartResult{}, nil).After(logs2)
 	expectSuccessRemove(apiClient, "hook-2").After(start2)
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	err := tested.runPreStart(t.Context(), project, service, func(api.ContainerEvent) {})
 	assert.NilError(t, err)
 }
 
@@ -141,20 +150,18 @@ func TestPreStart_FirstHookFailsStopsExecution(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"echo", "never"}},
 		},
 	}
-	ctr := container.Summary{ID: "service-ctr-id"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0), runnerSummary("hook-2", 1))
 	wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(42)).After(create1)
+		Return(waitResultExit(42)).After(scan)
 	logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil).After(wait1)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
 		Return(client.ContainerStartResult{}, nil).After(logs1)
-	// Hook container is retained on failure — no ContainerRemove expected.
+	// Hook container is retained on failure — no ContainerRemove expected, and
+	// hook-2's runner is never touched.
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	err := tested.runPreStart(t.Context(), project, service, func(api.ContainerEvent) {})
 	assert.ErrorContains(t, err, `service "web" pre_start[0]`)
 	assert.ErrorContains(t, err, "42")
 }
@@ -170,13 +177,147 @@ func TestPreStart_PerReplicaRejected(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}, PerReplica: true},
 		},
 	}
-	ctr := container.Summary{ID: "service-ctr-id"}
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	err := tested.runPreStart(t.Context(), project, service, func(api.ContainerEvent) {})
 	assert.ErrorContains(t, err, `service "web" pre_start[0]`)
 	assert.ErrorContains(t, err, "per_replica is not yet supported")
 }
 
+// ---------------------------------------------------------------------------
+// Pure-execution contract: runPreStart never creates a runner
+// ---------------------------------------------------------------------------
+
+// TestPreStart_MissingRunnerFails pins the no-fallback contract: a declared
+// hook without a prepared runner is an error pointing the user at the
+// reconciliation command, not an implicit creation. This is what a user sees
+// on `stop` then `start`: the runners were consumed by the first start.
+func TestPreStart_MissingRunnerFails(t *testing.T) {
+	tested, apiClient := newPreStartTestService(t)
+
+	project := &types.Project{Name: "demo"}
+	service := types.ServiceConfig{
+		Name:  "web",
+		Image: "alpine",
+		PreStart: []types.ServiceHook{
+			{Image: "alpine", Command: types.ShellCommand{"true"}},
+		},
+	}
+
+	expectRunnerScan(apiClient)
+
+	err := tested.runPreStart(t.Context(), project, service, nil)
+	assert.ErrorContains(t, err, `service "web" pre_start[0]`)
+	assert.ErrorContains(t, err, "docker compose up web")
+}
+
+// TestPreStart_ConsumedRunnerNotReused verifies that a runner in any state but
+// created (here: exited, e.g. retained after a failure) is not re-executed —
+// the hook reports the runner as missing instead of re-running a stale one.
+func TestPreStart_ConsumedRunnerNotReused(t *testing.T) {
+	tested, apiClient := newPreStartTestService(t)
+
+	project := &types.Project{Name: "demo"}
+	service := types.ServiceConfig{
+		Name:  "web",
+		Image: "alpine",
+		PreStart: []types.ServiceHook{
+			{Image: "alpine", Command: types.ShellCommand{"true"}},
+		},
+	}
+
+	consumed := runnerSummary("hook-old", 0)
+	consumed.State = container.StateExited
+	expectRunnerScan(apiClient, consumed)
+
+	err := tested.runPreStart(t.Context(), project, service, nil)
+	assert.ErrorContains(t, err, `service "web" pre_start[0]`)
+	assert.ErrorContains(t, err, "docker compose up web")
+}
+
+// TestPreStart_OldGenerationRunnerIgnored verifies that a hook container
+// without a HookIndexLabel (created by an older compose version) is never
+// matched to a declared hook: the purge planned by the reconciler is the only
+// consumer of those containers.
+func TestPreStart_OldGenerationRunnerIgnored(t *testing.T) {
+	tested, apiClient := newPreStartTestService(t)
+
+	project := &types.Project{Name: "demo"}
+	service := types.ServiceConfig{
+		Name:  "web",
+		Image: "alpine",
+		PreStart: []types.ServiceHook{
+			{Image: "alpine", Command: types.ShellCommand{"true"}},
+		},
+	}
+
+	legacy := container.Summary{
+		ID:     "legacy-hook",
+		State:  container.StateCreated,
+		Labels: map[string]string{api.HookLabel: preStartHookType},
+	}
+	expectRunnerScan(apiClient, legacy)
+
+	err := tested.runPreStart(t.Context(), project, service, nil)
+	assert.ErrorContains(t, err, `service "web" pre_start[0]`)
+	assert.ErrorContains(t, err, "docker compose up web")
+}
+
+// TestPreStart_SecondRunnerMissing verifies that a missing runner is only
+// reported when its hook is reached: the first hook runs (and is consumed)
+// before pre_start[1] fails.
+func TestPreStart_SecondRunnerMissing(t *testing.T) {
+	tested, apiClient := newPreStartTestService(t)
+
+	project := &types.Project{Name: "demo"}
+	service := types.ServiceConfig{
+		Name:  "web",
+		Image: "alpine",
+		PreStart: []types.ServiceHook{
+			{Image: "alpine", Command: types.ShellCommand{"true"}},
+			{Image: "alpine", Command: types.ShellCommand{"true"}},
+		},
+	}
+
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
+	wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
+		Return(waitResultExit(0)).After(scan)
+	logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
+		Return(emptyLogs(), nil).After(wait1)
+	start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
+		Return(client.ContainerStartResult{}, nil).After(logs1)
+	expectSuccessRemove(apiClient, "hook-1").After(start1)
+
+	err := tested.runPreStart(t.Context(), project, service, nil)
+	assert.ErrorContains(t, err, `service "web" pre_start[1]`)
+	assert.ErrorContains(t, err, "docker compose up web")
+}
+
+// TestPreStart_RunnerScanFails verifies that a ContainerList failure during
+// the runner lookup is fatal: without the runner set runPreStart cannot tell
+// prepared hooks from missing ones.
+func TestPreStart_RunnerScanFails(t *testing.T) {
+	tested, apiClient := newPreStartTestService(t)
+
+	project := &types.Project{Name: "proj"}
+	service := types.ServiceConfig{
+		Name:  "web",
+		Image: "alpine",
+		PreStart: []types.ServiceHook{
+			{Image: "alpine", Command: types.ShellCommand{"true"}},
+		},
+	}
+
+	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
+		Return(client.ContainerListResult{}, errors.New("daemon unavailable"))
+
+	err := tested.runPreStart(t.Context(), project, service, nil)
+	assert.ErrorContains(t, err, "daemon unavailable")
+}
+
+// ---------------------------------------------------------------------------
+// Runner creation (createPreStartContainer, called by the plan executor)
+// ---------------------------------------------------------------------------
+
 func TestPreStart_ImageFallsBackToBuiltImage(t *testing.T) {
 	tested, apiClient := newPreStartTestService(t)
 
@@ -191,22 +332,15 @@ func TestPreStart_ImageFallsBackToBuiltImage(t *testing.T) {
 	ctr := container.Summary{ID: "service-ctr-id"}
 
 	var gotImage string
-	scan := expectEmptyOrphanScan(apiClient)
 	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
 		DoAndReturn(func(_ any, opts client.ContainerCreateOptions) (client.ContainerCreateResult, error) {
 			gotImage = opts.Config.Image
 			return client.ContainerCreateResult{ID: "hook-1"}, nil
-		}).After(scan)
-	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerStartResult{}, nil)
-	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
-		Return(emptyLogs(), nil)
-	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
-	expectSuccessRemove(apiClient, "hook-1")
+		})
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	created, err := tested.createPreStartContainer(t.Context(), project, service, ctr, 0, "demo-web-pre_start-0")
 	assert.NilError(t, err)
+	assert.Equal(t, created.ID, "hook-1")
 	assert.Equal(t, gotImage, api.GetImageNameOrDefault(service, project.Name))
 }
 
@@ -224,26 +358,22 @@ func TestPreStart_ExplicitHookImageUsed(t *testing.T) {
 	ctr := container.Summary{ID: "service-ctr-id"}
 
 	var gotImage string
-	scan := expectEmptyOrphanScan(apiClient)
 	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
 		DoAndReturn(func(_ any, opts client.ContainerCreateOptions) (client.ContainerCreateResult, error) {
 			gotImage = opts.Config.Image
 			return client.ContainerCreateResult{ID: "hook-1"}, nil
-		}).After(scan)
-	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerStartResult{}, nil)
-	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
-		Return(emptyLogs(), nil)
-	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
-	expectSuccessRemove(apiClient, "hook-1")
+		})
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	_, err := tested.createPreStartContainer(t.Context(), project, service, ctr, 0, "demo-web-pre_start-0")
 	assert.NilError(t, err)
 	assert.Equal(t, gotImage, "custom-hook-image:1.2.3")
 }
 
-func TestPreStart_VolumesFromServiceContainer(t *testing.T) {
+// TestPreStart_CreateNameAndLabels pins the runner's plan-visible identity:
+// deterministic container name, VolumesFrom on the target replica, AutoRemove
+// off (retention is managed explicitly) and the hook labels — type for the
+// purge, index for the start-phase lookup.
+func TestPreStart_CreateNameAndLabels(t *testing.T) {
 	tested, apiClient := newPreStartTestService(t)
 
 	project := &types.Project{Name: "demo"}
@@ -252,37 +382,26 @@ func TestPreStart_VolumesFromServiceContainer(t *testing.T) {
 		Image: "alpine",
 		PreStart: []types.ServiceHook{
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
+			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
 	ctr := container.Summary{ID: "service-ctr-id"}
 
-	var gotVolumesFrom []string
-	var gotAutoRemove bool
-	var gotLabels map[string]string
-	scan := expectEmptyOrphanScan(apiClient)
+	var gotOpts client.ContainerCreateOptions
 	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
 		DoAndReturn(func(_ any, opts client.ContainerCreateOptions) (client.ContainerCreateResult, error) {
-			gotVolumesFrom = opts.HostConfig.VolumesFrom
-			gotAutoRemove = opts.HostConfig.AutoRemove
-			gotLabels = opts.Config.Labels
+			gotOpts = opts
 			return client.ContainerCreateResult{ID: "hook-1"}, nil
-		}).After(scan)
-	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerStartResult{}, nil)
-	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
-		Return(emptyLogs(), nil)
-	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
-	expectSuccessRemove(apiClient, "hook-1")
+		})
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	name := getHookContainerName(project.Name, service.Name, 1)
+	_, err := tested.createPreStartContainer(t.Context(), project, service, ctr, 1, name)
 	assert.NilError(t, err)
-	assert.DeepEqual(t, gotVolumesFrom, []string{"service-ctr-id"})
-	// AutoRemove must be false: the hook container is retained on failure for
-	// post-mortem and explicitly removed on success by runPreStartHook.
-	assert.Assert(t, !gotAutoRemove, "AutoRemove must be false")
-	// HookLabel must be set so orphan cleanup can identify the container.
-	assert.Equal(t, gotLabels[api.HookLabel], preStartHookType)
+	assert.Equal(t, gotOpts.Name, "demo-web-pre_start-1")
+	assert.DeepEqual(t, gotOpts.HostConfig.VolumesFrom, []string{"service-ctr-id"})
+	assert.Assert(t, !gotOpts.HostConfig.AutoRemove, "AutoRemove must be false")
+	assert.Equal(t, gotOpts.Config.Labels[api.HookLabel], preStartHookType)
+	assert.Equal(t, gotOpts.Config.Labels[api.HookIndexLabel], "1")
 }
 
 func TestPreStart_ContainerCreateFailurePropagates(t *testing.T) {
@@ -294,16 +413,14 @@ func TestPreStart_ContainerCreateFailurePropagates(t *testing.T) {
 		Image: "alpine",
 		PreStart: []types.ServiceHook{
 			{Image: "missing:latest", Command: types.ShellCommand{"true"}},
-			{Image: "alpine", Command: types.ShellCommand{"never"}},
 		},
 	}
 	ctr := container.Summary{ID: "service-ctr-id"}
 
-	scan := expectEmptyOrphanScan(apiClient)
 	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{}, errors.New("no such image: missing:latest")).After(scan)
+		Return(client.ContainerCreateResult{}, errors.New("no such image: missing:latest"))
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	_, err := tested.createPreStartContainer(t.Context(), project, service, ctr, 0, "demo-web-pre_start-0")
 	assert.ErrorContains(t, err, "no such image")
 }
 
@@ -318,13 +435,10 @@ func TestPreStart_ContainerStartFailurePropagates(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "service-ctr-id"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0)).After(create1)
+		Return(waitResultExit(0)).After(scan)
 	logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil).After(wait1)
 	start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
@@ -335,7 +449,7 @@ func TestPreStart_ContainerStartFailurePropagates(t *testing.T) {
 	apiClient.EXPECT().ContainerRemove(gomock.Any(), "hook-1", client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}).
 		Return(client.ContainerRemoveResult{}, nil).After(start1)
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	err := tested.runPreStart(t.Context(), project, service, func(api.ContainerEvent) {})
 	assert.ErrorContains(t, err, "container start failed")
 }
 
@@ -356,7 +470,6 @@ func TestPreStart_WaitResultPreferredOverNilError(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "service-ctr-id"}
 
 	// Both channels are buffered and pre-populated so the outer select in
 	// waitPreStart sees them ready at the same instant.
@@ -365,18 +478,16 @@ func TestPreStart_WaitResultPreferredOverNilError(t *testing.T) {
 	resultC <- container.WaitResponse{StatusCode: 0}
 	errC <- nil
 
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerWaitResult{Result: resultC, Error: errC})
+		Return(client.ContainerWaitResult{Result: resultC, Error: errC}).After(scan)
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
 		Return(client.ContainerStartResult{}, nil)
 	expectSuccessRemove(apiClient, "hook-1")
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, func(api.ContainerEvent) {})
+	err := tested.runPreStart(t.Context(), project, service, func(api.ContainerEvent) {})
 	assert.NilError(t, err)
 }
 
@@ -440,13 +551,10 @@ func TestPreStart_DetachedModeAttachesLogs(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "service-ctr-id"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0)).After(create1)
+		Return(waitResultExit(0)).After(scan)
 	// ContainerLogs MUST be called even with a nil listener.
 	logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil).After(wait1)
@@ -454,7 +562,7 @@ func TestPreStart_DetachedModeAttachesLogs(t *testing.T) {
 		Return(client.ContainerStartResult{}, nil).After(logs1)
 	expectSuccessRemove(apiClient, "hook-1")
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	assert.NilError(t, err)
 }
 
@@ -472,7 +580,6 @@ func TestPreStart_FailureIncludesTail(t *testing.T) {
 			{Image: "postgres", Command: types.ShellCommand{"migrate"}},
 		},
 	}
-	ctr := container.Summary{ID: "service-ctr-id"}
 
 	// Build a stdcopy-multiplexed log stream with a stderr error line.
 	logContent := append(
@@ -480,18 +587,16 @@ func TestPreStart_FailureIncludesTail(t *testing.T) {
 		stdcopyFrame(2, "table 'sites' doesn't exist\n")..., // stderr error
 	)
 
-	scan := expectEmptyOrphanScan(apiClient)
-	create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	wait1 := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(1)).After(create1)
+		Return(waitResultExit(1)).After(scan)
 	logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(io.NopCloser(bytes.NewReader(logContent)), nil).After(wait1)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
 		Return(client.ContainerStartResult{}, nil).After(logs1)
 	// Hook container is retained on failure — no ContainerRemove expected.
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	assert.Assert(t, err != nil)
 	assert.ErrorContains(t, err, "pre_start[0]")
 	// Stderr content must be in the error (stderr bias).
@@ -504,7 +609,7 @@ func TestPreStart_FailureIncludesTail(t *testing.T) {
 
 // TestPreStart_SuccessRemovesContainer verifies that a successful pre_start hook
 // triggers an explicit ContainerRemove (with RemoveVolumes: true to mirror the
-// old AutoRemove behaviour) and that AutoRemove is false at create time.
+// old AutoRemove behaviour).
 func TestPreStart_SuccessRemovesContainer(t *testing.T) {
 	tested, apiClient := newPreStartTestService(t)
 
@@ -516,17 +621,10 @@ func TestPreStart_SuccessRemovesContainer(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
-	var gotAutoRemove bool
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		DoAndReturn(func(_ any, opts client.ContainerCreateOptions) (client.ContainerCreateResult, error) {
-			gotAutoRemove = opts.HostConfig.AutoRemove
-			return client.ContainerCreateResult{ID: "hook-1"}, nil
-		}).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
+		Return(waitResultExit(0)).After(scan)
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
@@ -536,9 +634,8 @@ func TestPreStart_SuccessRemovesContainer(t *testing.T) {
 		ContainerRemove(gomock.Any(), "hook-1", client.ContainerRemoveOptions{RemoveVolumes: true}).
 		Return(client.ContainerRemoveResult{}, nil)
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	assert.NilError(t, err)
-	assert.Assert(t, !gotAutoRemove, "AutoRemove must be false; retention is managed explicitly")
 }
 
 // TestPreStart_FailureRetainsContainer verifies that a pre_start hook that exits
@@ -555,22 +652,19 @@ func TestPreStart_FailureRetainsContainer(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"migrate"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
 	logContent := stdcopyFrame(2, "migration failed: table missing\n")
 
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(1))
+		Return(waitResultExit(1)).After(scan)
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(io.NopCloser(bytes.NewReader(logContent)), nil)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
 		Return(client.ContainerStartResult{}, nil)
 	// No ContainerRemove expectation: gomock fails on unexpected calls.
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	assert.ErrorContains(t, err, "pre_start[0]")
 	assert.ErrorContains(t, err, "table missing")
 	// Short container ID must appear in the error so the operator can run
@@ -594,17 +688,14 @@ func TestPreStart_CancellationRemovesContainer(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"long-running-op"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-cancel-123"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-cancel-123", 0))
 	// ContainerWait channel: cancel the context before delivering any result so
 	// waitPreStart returns ctx.Err().
 	resultC := make(chan container.WaitResponse)
 	errC := make(chan error)
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-cancel-123", gomock.Any()).
-		Return(client.ContainerWaitResult{Result: resultC, Error: errC})
+		Return(client.ContainerWaitResult{Result: resultC, Error: errC}).After(scan)
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-cancel-123", gomock.Any()).
 		Return(emptyLogs(), nil)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-cancel-123", gomock.Any()).
@@ -616,7 +707,7 @@ func TestPreStart_CancellationRemovesContainer(t *testing.T) {
 
 	errCh := make(chan error, 1)
 	go func() {
-		errCh <- tested.runPreStart(ctx, project, service, ctr, nil)
+		errCh <- tested.runPreStart(ctx, project, service, nil)
 	}()
 	// Cancel after the hook container has started.
 	cancel()
@@ -627,43 +718,6 @@ func TestPreStart_CancellationRemovesContainer(t *testing.T) {
 	assert.Assert(t, !strings.Contains(err.Error(), "retained"), "cancelled hook must not be retained; got: %s", err)
 }
 
-// TestPreStart_RemovesOrphanBeforeRun verifies that a hook container left behind
-// by a previous failed run is force-removed before the new container is created.
-func TestPreStart_RemovesOrphanBeforeRun(t *testing.T) {
-	tested, apiClient := newPreStartTestService(t)
-
-	project := &types.Project{Name: "proj"}
-	service := types.ServiceConfig{
-		Name:  "web",
-		Image: "alpine",
-		PreStart: []types.ServiceHook{
-			{Image: "alpine", Command: types.ShellCommand{"migrate"}},
-		},
-	}
-	ctr := container.Summary{ID: "svc-ctr"}
-
-	// ContainerList returns the stale orphan from a previous run.
-	orphan := container.Summary{ID: "orphan-hook-123", State: "exited"}
-	scan := apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
-		Return(client.ContainerListResult{Items: []container.Summary{orphan}}, nil)
-	// Orphan must be removed before the new hook container is created.
-	removeOrphan := apiClient.EXPECT().
-		ContainerRemove(gomock.Any(), "orphan-hook-123", client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}).
-		Return(client.ContainerRemoveResult{}, nil).After(scan)
-	create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(removeOrphan)
-	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0)).After(create1)
-	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
-		Return(emptyLogs(), nil)
-	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerStartResult{}, nil)
-	expectSuccessRemove(apiClient, "hook-1")
-
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
-	assert.NilError(t, err)
-}
-
 // TestPreStart_SuccessRemoveFailureIsNonFatal verifies that a ContainerRemove
 // failure on the success path is logged but does not fail runPreStart.
 func TestPreStart_SuccessRemoveFailureIsNonFatal(t *testing.T) {
@@ -677,13 +731,10 @@ func TestPreStart_SuccessRemoveFailureIsNonFatal(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
+		Return(waitResultExit(0)).After(scan)
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
@@ -694,7 +745,7 @@ func TestPreStart_SuccessRemoveFailureIsNonFatal(t *testing.T) {
 		Return(client.ContainerRemoveResult{}, errors.New("already removed"))
 
 	// The hook succeeded; the service must start even if removal failed.
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	assert.NilError(t, err)
 }
 
@@ -769,13 +820,10 @@ func TestPreStart_StreamLogsError_NilListener(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
+		Return(waitResultExit(0)).After(scan)
 	// ContainerLogs fails; nil listener → no warning event, done closed immediately.
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(nil, errors.New("logs: connection refused"))
@@ -783,7 +831,7 @@ func TestPreStart_StreamLogsError_NilListener(t *testing.T) {
 		Return(client.ContainerStartResult{}, nil)
 	expectSuccessRemove(apiClient, "hook-1")
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	assert.NilError(t, err)
 }
 
@@ -801,13 +849,10 @@ func TestPreStart_StreamLogsError_WithListener(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
+		Return(waitResultExit(0)).After(scan)
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(nil, errors.New("logs: daemon unavailable"))
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
@@ -818,7 +863,7 @@ func TestPreStart_StreamLogsError_WithListener(t *testing.T) {
 	listener := func(ev api.ContainerEvent) {
 		gotWarning = ev.Line
 	}
-	err := tested.runPreStart(t.Context(), project, service, ctr, listener)
+	err := tested.runPreStart(t.Context(), project, service, listener)
 	assert.NilError(t, err)
 	assert.Assert(t, gotWarning != "", "listener must receive a warning when ContainerLogs fails")
 	assert.Assert(t, bytes.Contains([]byte(gotWarning), []byte("warning")), "expected 'warning' in: %q", gotWarning)
@@ -827,7 +872,7 @@ func TestPreStart_StreamLogsError_WithListener(t *testing.T) {
 // TestPreStart_OldAPIVersion covers the versions.LessThan(apiVersion, "1.44")
 // branch in createPreStartContainer: on a pre-1.44 daemon the extra-networks
 // path runs via connectPreStartExtraNetworks. With only one (primary) network
-// no NetworkConnect call is issued and the hook succeeds normally.
+// no NetworkConnect call is issued and the create succeeds normally.
 func TestPreStart_OldAPIVersion(t *testing.T) {
 	tested, apiClient := newPreStartTestServiceWithVersion(t, "1.43")
 
@@ -849,20 +894,13 @@ func TestPreStart_OldAPIVersion(t *testing.T) {
 	}
 	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
 	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
-	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
-	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
-		Return(emptyLogs(), nil)
-	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerStartResult{}, nil)
+		Return(client.ContainerCreateResult{ID: "hook-1"}, nil)
 	// Single network = primary only; no NetworkConnect expected.
-	expectSuccessRemove(apiClient, "hook-1")
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	created, err := tested.createPreStartContainer(t.Context(), project, service, ctr, 0, "proj-web-pre_start-0")
 	assert.NilError(t, err)
+	assert.Equal(t, created.ID, "hook-1")
 }
 
 // TestPreStart_ConnectExtraNetworksSuccess covers connectPreStartExtraNetworks
@@ -924,7 +962,7 @@ func TestPreStart_ConnectExtraNetworksFails(t *testing.T) {
 }
 
 // TestPreStart_ContainerStartFailureAndRemoveFails covers the Warnf path in
-// runPreStartHook when ContainerStart fails AND the subsequent ContainerRemove
+// execPreStartHook when ContainerStart fails AND the subsequent ContainerRemove
 // also fails (the orphan is unremovable but the caller still gets the start error).
 func TestPreStart_ContainerStartFailureAndRemoveFails(t *testing.T) {
 	tested, apiClient := newPreStartTestService(t)
@@ -937,13 +975,10 @@ func TestPreStart_ContainerStartFailureAndRemoveFails(t *testing.T) {
 			{Image: "alpine", Command: types.ShellCommand{"true"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
-	create1 := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0)).After(create1)
+		Return(waitResultExit(0)).After(scan)
 	logs1 := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(emptyLogs(), nil)
 	start1 := apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
@@ -952,82 +987,11 @@ func TestPreStart_ContainerStartFailureAndRemoveFails(t *testing.T) {
 	apiClient.EXPECT().ContainerRemove(gomock.Any(), "hook-1", client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}).
 		Return(client.ContainerRemoveResult{}, errors.New("removal failed")).After(start1)
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	// The original start error must be returned, not the removal error.
 	assert.ErrorContains(t, err, "start failed")
 }
 
-// TestPreStart_OrphanScanFails verifies that when ContainerList fails during
-// orphan cleanup the Warnf path in runPreStart is hit, but execution continues
-// and the hook runs successfully.
-func TestPreStart_OrphanScanFails(t *testing.T) {
-	tested, apiClient := newPreStartTestService(t)
-
-	project := &types.Project{Name: "proj"}
-	service := types.ServiceConfig{
-		Name:  "web",
-		Image: "alpine",
-		PreStart: []types.ServiceHook{
-			{Image: "alpine", Command: types.ShellCommand{"true"}},
-		},
-	}
-	ctr := container.Summary{ID: "svc-ctr"}
-
-	// ContainerList fails → Warnf in runPreStart; hook still proceeds.
-	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
-		Return(client.ContainerListResult{}, errors.New("daemon unavailable"))
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil)
-	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
-	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
-		Return(emptyLogs(), nil)
-	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerStartResult{}, nil)
-	expectSuccessRemove(apiClient, "hook-1")
-
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
-	assert.NilError(t, err)
-}
-
-// TestPreStart_OrphanRemovalFails verifies the Warnf path in
-// removeOrphanPreStartContainers when an individual stale container cannot be
-// removed. The hook must still proceed normally.
-func TestPreStart_OrphanRemovalFails(t *testing.T) {
-	tested, apiClient := newPreStartTestService(t)
-
-	project := &types.Project{Name: "proj"}
-	service := types.ServiceConfig{
-		Name:  "web",
-		Image: "alpine",
-		PreStart: []types.ServiceHook{
-			{Image: "alpine", Command: types.ShellCommand{"true"}},
-		},
-	}
-	ctr := container.Summary{ID: "svc-ctr"}
-
-	orphan := container.Summary{ID: "orphan-123"}
-	// ContainerList finds the orphan; its removal fails → Warnf then continue.
-	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
-		Return(client.ContainerListResult{Items: []container.Summary{orphan}}, nil)
-	apiClient.EXPECT().
-		ContainerRemove(gomock.Any(), "orphan-123", client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}).
-		Return(client.ContainerRemoveResult{}, errors.New("permission denied"))
-	// Hook still runs and succeeds.
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil)
-	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0))
-	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
-		Return(emptyLogs(), nil)
-	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
-		Return(client.ContainerStartResult{}, nil)
-	expectSuccessRemove(apiClient, "hook-1")
-
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
-	assert.NilError(t, err)
-}
-
 // TestPreStart_OldAPINetworkConnectFails covers the createPreStartContainer path
 // for API < 1.44 with a secondary network: when NetworkConnect fails the function
 // must clean up the created-but-never-started container and return the error.
@@ -1054,9 +1018,8 @@ func TestPreStart_OldAPINetworkConnectFails(t *testing.T) {
 	}
 	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
 	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+		Return(client.ContainerCreateResult{ID: "hook-1"}, nil)
 	// NetworkConnect for the secondary network fails.
 	apiClient.EXPECT().NetworkConnect(gomock.Any(), "proj_extra", gomock.Any()).
 		Return(client.NetworkConnectResult{}, errors.New("network not found"))
@@ -1065,7 +1028,7 @@ func TestPreStart_OldAPINetworkConnectFails(t *testing.T) {
 		ContainerRemove(gomock.Any(), "hook-1", client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}).
 		Return(client.ContainerRemoveResult{}, nil)
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	_, err := tested.createPreStartContainer(t.Context(), project, service, ctr, 0, "proj-web-pre_start-0")
 	assert.ErrorContains(t, err, "network not found")
 }
 
@@ -1094,9 +1057,8 @@ func TestPreStart_OldAPINetworkConnectAndRemoveFails(t *testing.T) {
 	}
 	ctr := container.Summary{ID: "svc-ctr"}
 
-	scan := expectEmptyOrphanScan(apiClient)
 	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+		Return(client.ContainerCreateResult{ID: "hook-1"}, nil)
 	apiClient.EXPECT().NetworkConnect(gomock.Any(), "proj_extra", gomock.Any()).
 		Return(client.NetworkConnectResult{}, errors.New("network not found"))
 	// Cleanup removal also fails → Warnf; original error is still returned.
@@ -1104,7 +1066,7 @@ func TestPreStart_OldAPINetworkConnectAndRemoveFails(t *testing.T) {
 		ContainerRemove(gomock.Any(), "hook-1", client.ContainerRemoveOptions{Force: true, RemoveVolumes: true}).
 		Return(client.ContainerRemoveResult{}, errors.New("removal also failed"))
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	_, err := tested.createPreStartContainer(t.Context(), project, service, ctr, 0, "proj-web-pre_start-0")
 	assert.ErrorContains(t, err, "network not found")
 }
 
@@ -1133,15 +1095,12 @@ func TestPreStart_RuntimeAPIVersionError(t *testing.T) {
 	}
 	ctr := container.Summary{ID: "svc-ctr"}
 
-	// Orphan scan succeeds (ContainerList does not need Ping).
-	apiClient.EXPECT().ContainerList(gomock.Any(), gomock.Any()).
-		Return(client.ContainerListResult{}, nil)
 	// createPreStartContainer calls RuntimeAPIVersion → Ping fails.
 	apiClient.EXPECT().
 		Ping(gomock.Any(), client.PingOptions{NegotiateAPIVersion: true}).
 		Return(client.PingResult{}, errors.New("daemon unreachable"))
 
-	err = s.runPreStart(t.Context(), project, service, ctr, nil)
+	_, err = s.createPreStartContainer(t.Context(), project, service, ctr, 0, "proj-web-pre_start-0")
 	assert.ErrorContains(t, err, "daemon unreachable")
 }
 
@@ -1159,23 +1118,27 @@ func TestPreStart_FailureStdoutOnlyTail(t *testing.T) {
 			{Image: "postgres", Command: types.ShellCommand{"migrate"}},
 		},
 	}
-	ctr := container.Summary{ID: "svc-ctr"}
 
 	// stdout-only content: stderr frame absent so getTail falls back to stdout.
 	logContent := stdcopyFrame(1, "migration failed: schema mismatch\n")
 
-	scan := expectEmptyOrphanScan(apiClient)
-	apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		Return(client.ContainerCreateResult{ID: "hook-1"}, nil).After(scan)
+	scan := expectRunnerScan(apiClient, runnerSummary("hook-1", 0))
 	apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(1))
+		Return(waitResultExit(1)).After(scan)
 	apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
 		Return(io.NopCloser(bytes.NewReader(logContent)), nil)
 	apiClient.EXPECT().ContainerStart(gomock.Any(), "hook-1", gomock.Any()).
 		Return(client.ContainerStartResult{}, nil)
 
-	err := tested.runPreStart(t.Context(), project, service, ctr, nil)
+	err := tested.runPreStart(t.Context(), project, service, nil)
 	assert.ErrorContains(t, err, "pre_start[0]")
 	// Stdout fallback: no stderr → stdout content appears in the error.
 	assert.ErrorContains(t, err, "schema mismatch")
 }
+
+// TestGetHookContainerName pins the deterministic runner naming scheme the
+// reconciliation plan relies on for convergence.
+func TestGetHookContainerName(t *testing.T) {
+	assert.Equal(t, getHookContainerName("demo", "web", 0), "demo-web-pre_start-0")
+	assert.Equal(t, getHookContainerName("demo", "web", 12), "demo-web-pre_start-12")
+}
diff --git a/pkg/compose/reconcile.go b/pkg/compose/reconcile.go
index d22f035617..6f7e45c8ef 100644
--- a/pkg/compose/reconcile.go
+++ b/pkg/compose/reconcile.go
@@ -688,6 +688,14 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 	infraDeps := r.infrastructureDeps(service)
 
 	var lastNode *PlanNode
+	// containerNodes collects every node planned for this service's replicas,
+	// so the pre_start hook-runner creation below can wait for the executor's
+	// live container view to be final before resolving its target replica.
+	var containerNodes []*PlanNode
+	// keptRunning reports whether a running replica survives the plan
+	// untouched — the start phase will then skip pre_start, so no hook
+	// runner is needed.
+	keptRunning := false
 
 	// Process existing containers
 	for i, oc := range containers {
@@ -708,11 +716,13 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 				Cause:      "scale down",
 				Container:  &containers[i].Summary,
 			}, "", stopNode)
+			containerNodes = append(containerNodes, lastNode)
 			continue
 		}
 
 		if r.mustRecreate(service, expectedHash, parentRecreated, oc, strategy) {
 			lastNode = r.planRecreateContainer(service, &containers[i], infraDeps)
+			containerNodes = append(containerNodes, lastNode)
 			r.recreatedServices[service.Name] = true
 			continue
 		}
@@ -725,6 +735,9 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 			// (start.go), which lists containers again and starts them in
 			// dependency order. Exited containers are deliberately left as-is
 			// here so that phase (or the user) decides.
+			if oc.State == container.StateRunning {
+				keptRunning = true
+			}
 		default:
 			// Any other state (paused, dead, ...): attempt to (re)start
 			lastNode = r.plan.addNode(Operation{
@@ -733,6 +746,10 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 				Cause:      "not running",
 				Container:  &containers[i].Summary,
 			}, "", infraDeps...)
+			containerNodes = append(containerNodes, lastNode)
+			// The replica is running once this op executes, so the start
+			// phase will skip pre_start for the service.
+			keptRunning = true
 		}
 	}
 
@@ -750,14 +767,66 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 			Number:     number,
 			Name:       name,
 		}, "", infraDeps...)
+		containerNodes = append(containerNodes, lastNode)
 	}
 
+	r.planPreStartHookRunners(service, expected, keptRunning, containerNodes, infraDeps)
+
 	if lastNode != nil {
 		r.serviceNodes[service.Name] = lastNode
 	}
 	return nil
 }
 
+// planPreStartHookRunners plans the create-phase side of the pre_start hook
+// lifecycle. When the start phase is going to run the hooks — hooks declared,
+// a replica to start (scale > 0) and no running replica surviving the plan —
+// it purges every observed hook-runner container (leftovers of a previous run
+// that failed, or was never started) and creates one fresh runner per declared
+// hook. The start phase only executes runners prepared here; it never creates
+// one itself.
+//
+// Purges are best-effort (mirroring the historical imperative purge, which was
+// warn-only) but the creates depend on them: runner names are deterministic,
+// so a genuinely stuck old runner surfaces as a name conflict on the create.
+// The creates also depend on every replica node and on the infrastructure, so
+// the executor resolves the target replica (VolumesFrom) against a final live
+// view with networks in place. When a running replica survives the plan, the
+// start phase skips pre_start entirely: nothing is created, and a hook
+// container retained after a failure stays available for inspection.
+func (r *reconciler) planPreStartHookRunners(service types.ServiceConfig, expectedScale int, keptRunning bool, containerNodes, infraDeps []*PlanNode) {
+	if len(service.PreStart) == 0 || expectedScale == 0 || keptRunning {
+		return
+	}
+	serviceCopy := service
+	deps := slices.Concat(containerNodes, infraDeps)
+	stale := slices.Clone(r.observed.HookContainers[service.Name])
+	slices.SortFunc(stale, func(a, b ObservedContainer) int { return strings.Compare(a.ID, b.ID) })
+	for i := range stale {
+		node := r.plan.addNode(Operation{
+			Type:          OpRemoveContainer,
+			ResourceID:    fmt.Sprintf("hook:%s:stale:%s", service.Name, stale[i].ID[:min(12, len(stale[i].ID))]),
+			Cause:         "stale pre_start hook container",
+			Service:       &serviceCopy,
+			Container:     &stale[i].Summary,
+			RemoveVolumes: true,
+			BestEffort:    true,
+		}, "")
+		deps = append(deps, node)
+	}
+	for i := range service.PreStart {
+		node := r.plan.addNode(Operation{
+			Type:       OpCreateHookContainer,
+			ResourceID: fmt.Sprintf("hook:%s:%s:%d", service.Name, preStartHookType, i),
+			Cause:      "pre_start hook",
+			Service:    &serviceCopy,
+			HookIndex:  i,
+			Name:       getHookContainerName(r.project.Name, service.Name, i),
+		}, "", deps...)
+		deps = []*PlanNode{node}
+	}
+}
+
 // mustRecreate decides whether oc must be recreated to match expected. The
 // expectedHash and parentRecreated inputs are precomputed once per service by
 // reconcileService — see expectedConfigHash and parentNamespaceRecreated for
diff --git a/pkg/compose/reconcile_test.go b/pkg/compose/reconcile_test.go
index c0336ce162..674d5f6f7e 100644
--- a/pkg/compose/reconcile_test.go
+++ b/pkg/compose/reconcile_test.go
@@ -1251,6 +1251,209 @@ func TestReconcileContainers_ExitedIsNoop(t *testing.T) {
 // container creation depends on the last plan node of the service it depends
 // on (via reconciler.serviceNodes). Without this, services declared in
 // depends_on could start before their dependencies' operations complete.
+// Stale pre_start hook runners (left by a previous run that failed before
+// removing them) are purged by the plan when pre_start is going to run again:
+// hooks declared and no replica running at observation — the imperative
+// gating. Removals are best-effort and drop the runner's anonymous volumes,
+// like the warn-only imperative purge they mirror.
+func TestReconcileContainers_StaleHookRunnersPurged(t *testing.T) {
+	project := &types.Project{
+		Name: "myproject",
+		Services: types.Services{
+			"app": {Name: "app", Scale: intPtr(1), PreStart: []types.ServiceHook{{}}},
+		},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers:  map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {
+				// deliberately out of ID order: the plan sorts for determinism
+				{ID: "stale-b-id", Summary: container.Summary{ID: "stale-b-id"}},
+				{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}},
+			},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+
+	// The fresh runner (#4) waits for the replica create (live view final) and
+	// for both purges (its deterministic name must be free).
+	assert.Equal(t, plan.String(), strings.TrimSpace(`
+[] -> #1 service:app:1, CreateContainer, no existing container
+[] -> #2 hook:app:stale:stale-a-id, RemoveContainer, stale pre_start hook container
+[] -> #3 hook:app:stale:stale-b-id, RemoveContainer, stale pre_start hook container
+[1,2,3] -> #4 hook:app:pre_start:0, CreateHookContainer, pre_start hook
+`)+"\n")
+
+	for _, n := range plan.Nodes {
+		if n.Operation.Type != OpRemoveContainer {
+			continue
+		}
+		assert.Assert(t, n.Operation.BestEffort, "purge #%d must not abort the plan on failure", n.ID)
+		assert.Assert(t, n.Operation.RemoveVolumes, "purge #%d must drop the runner's anonymous volumes", n.ID)
+	}
+}
+
+// A cold start of a hooked service plans one runner per declared hook, chained
+// in declaration order after the replica create.
+func TestReconcileContainers_HookRunnersPlannedOnColdStart(t *testing.T) {
+	project := &types.Project{
+		Name: "myproject",
+		Services: types.Services{
+			"app": {Name: "app", Scale: intPtr(1), PreStart: []types.ServiceHook{{}, {}}},
+		},
+	}
+	observed := &ObservedState{
+		ProjectName:    "myproject",
+		Containers:     map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{},
+		Networks:       map[string][]ObservedNetwork{},
+		Volumes:        map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+
+	assert.Equal(t, plan.String(), strings.TrimSpace(`
+[] -> #1 service:app:1, CreateContainer, no existing container
+[1] -> #2 hook:app:pre_start:0, CreateHookContainer, pre_start hook
+[2] -> #3 hook:app:pre_start:1, CreateHookContainer, pre_start hook
+`)+"\n")
+
+	for _, n := range plan.Nodes {
+		if n.Operation.Type != OpCreateHookContainer {
+			continue
+		}
+		assert.Equal(t, n.Operation.Name, getHookContainerName("myproject", "app", n.Operation.HookIndex))
+	}
+}
+
+// A running replica whose config diverged is recreated, so it will not be
+// running when the start phase evaluates pre_start: the runner must be
+// planned. The observed-running gate applies to replicas that SURVIVE the
+// plan, not to the observation itself.
+func TestReconcileContainers_HookRunnersPlannedOnRecreate(t *testing.T) {
+	svc := types.ServiceConfig{Name: "app", Scale: intPtr(1), PreStart: []types.ServiceHook{{}}}
+	project := &types.Project{
+		Name:     "myproject",
+		Services: types.Services{"app": svc},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers: map[string][]ObservedContainer{
+			"app": {{
+				ID: "c1", Number: 1, State: container.StateRunning, ConfigHash: "outdated",
+				Summary: container.Summary{
+					ID: "c1", State: container.StateRunning,
+					Labels: map[string]string{api.ServiceLabel: "app", api.ContainerNumberLabel: "1", api.ConfigHashLabel: "outdated"},
+				},
+			}},
+		},
+		HookContainers: map[string][]ObservedContainer{},
+		Networks:       map[string][]ObservedNetwork{},
+		Volumes:        map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+
+	var hookCreates []*PlanNode
+	for _, n := range plan.Nodes {
+		if n.Operation.Type == OpCreateHookContainer {
+			hookCreates = append(hookCreates, n)
+		}
+	}
+	assert.Equal(t, len(hookCreates), 1, "recreated service must get its hook runner:\n%s", plan)
+}
+
+// A scale-0 service never reaches its pre_start hooks (the start phase
+// returns before them), so nothing is planned: no runner, no purge.
+func TestReconcileContainers_StaleHookRunnersKeptWhenScaleZero(t *testing.T) {
+	project := &types.Project{
+		Name: "myproject",
+		Services: types.Services{
+			"app": {Name: "app", Scale: intPtr(0), PreStart: []types.ServiceHook{{}}},
+		},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers:  map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}}},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+	assert.Assert(t, plan.IsEmpty(), "unexpected plan:\n%s", plan)
+}
+
+// A running replica surviving the plan gates pre_start off, so nothing is
+// planned for the hooks: no fresh runner, and the stale runner stays — a
+// genuinely failed hook container is retained for inspection as long as its
+// service is otherwise up.
+func TestReconcileContainers_StaleHookRunnersKeptWhenReplicaRunning(t *testing.T) {
+	svc := types.ServiceConfig{Name: "app", Scale: intPtr(1), PreStart: []types.ServiceHook{{}}}
+	hash := mustServiceHash(t, svc)
+	project := &types.Project{
+		Name:     "myproject",
+		Services: types.Services{"app": svc},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers: map[string][]ObservedContainer{
+			"app": {{
+				ID: "c1", Number: 1, State: container.StateRunning, ConfigHash: hash,
+				Summary: container.Summary{
+					ID: "c1", State: container.StateRunning,
+					Labels: map[string]string{api.ServiceLabel: "app", api.ContainerNumberLabel: "1", api.ConfigHashLabel: hash},
+				},
+			}},
+		},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}}},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+	assert.Assert(t, plan.IsEmpty())
+}
+
+// Without pre_start hooks in the model, stale runners are not the plan's to
+// purge: runPreStart never runs, so the imperative engine leaves them alone
+// too.
+func TestReconcileContainers_StaleHookRunnersKeptWithoutHooks(t *testing.T) {
+	project := &types.Project{
+		Name:     "myproject",
+		Services: types.Services{"app": {Name: "app", Scale: intPtr(1)}},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers:  map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}}},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+
+	assert.Equal(t, plan.String(), strings.TrimSpace(`
+[] -> #1 service:app:1, CreateContainer, no existing container
+`)+"\n")
+}
+
 func TestReconcileContainers_DependsOnChain(t *testing.T) {
 	project := &types.Project{
 		Name: "myproject",
diff --git a/pkg/compose/service_containers.go b/pkg/compose/service_containers.go
index 716acd0a11..add93f346f 100644
--- a/pkg/compose/service_containers.go
+++ b/pkg/compose/service_containers.go
@@ -599,11 +599,10 @@ func (s *composeService) startService(ctx context.Context,
 
 	// pre_start runs once per service, only when no replica is already running
 	// (e.g. initial up, force-recreate, or spec change). per_replica: false is
-	// the only currently supported mode. Pick the replica with the lowest
-	// container-number so the choice is deterministic regardless of the order
-	// the daemon returns containers in.
+	// the only currently supported mode. The hooks execute in runner containers
+	// prepared by the reconciliation plan.
 	if len(service.PreStart) > 0 && len(serviceContainers) == len(toStart) {
-		if err := s.runPreStart(ctx, project, service, lowestNumberedContainer(toStart), listener); err != nil {
+		if err := s.runPreStart(ctx, project, service, listener); err != nil {
 			return err
 		}
 	}
diff --git a/pkg/compose/start_test.go b/pkg/compose/start_test.go
index 859aa274cc..0b67e96f2e 100644
--- a/pkg/compose/start_test.go
+++ b/pkg/compose/start_test.go
@@ -185,11 +185,11 @@ func TestStartService_StartsOnlyStoppedReplicas(t *testing.T) {
 	})
 }
 
-// TestStartService_PreStartOnLowestReplica locks the pre_start gating: with no
-// replica running, the hooks run exactly once, against the replica with the
-// lowest container-number — regardless of the order the daemon listed them in
-// — and before any service container is started.
-func TestStartService_PreStartOnLowestReplica(t *testing.T) {
+// TestStartService_PreStartRunsBeforeReplicas locks the pre_start gating: with
+// no replica running, the hooks run exactly once — executing the runner
+// container the reconciliation plan prepared — and before any service
+// container is started.
+func TestStartService_PreStartRunsBeforeReplicas(t *testing.T) {
 	svc, apiClient, _ := newStartTestService(t)
 
 	project := &types.Project{Name: "prj"}
@@ -203,21 +203,13 @@ func TestStartService_PreStartOnLowestReplica(t *testing.T) {
 	replica1 := serviceContainer("web", 1, container.StateExited)
 	containers := Containers{replica2, replica1}
 
-	// runPreStart sweeps orphan hook containers from any previous failed run
-	// before creating the new one.
-	orphanScan := apiClient.EXPECT().
+	// runPreStart looks up the runner containers prepared by the plan.
+	runnerScan := apiClient.EXPECT().
 		ContainerList(gomock.Any(), gomock.Any()).
-		Return(client.ContainerListResult{}, nil)
-
-	// The hook container shares the volumes of the lowest-numbered replica.
-	hookCreate := apiClient.EXPECT().ContainerCreate(gomock.Any(), gomock.Any()).
-		After(orphanScan).
-		DoAndReturn(func(_ context.Context, opts client.ContainerCreateOptions) (client.ContainerCreateResult, error) {
-			assert.DeepEqual(t, opts.HostConfig.VolumesFrom, []string{replica1.ID})
-			return client.ContainerCreateResult{ID: "hook-1"}, nil
-		})
+		Return(client.ContainerListResult{Items: []container.Summary{runnerSummary("hook-1", 0)}}, nil)
+
 	hookWait := apiClient.EXPECT().ContainerWait(gomock.Any(), "hook-1", gomock.Any()).
-		Return(waitResultExit(0)).After(hookCreate)
+		Return(waitResultExit(0)).After(runnerScan)
 	// streamPreStartLogs always opens ContainerLogs (even with nil listener) so
 	// the tail is available for failure error messages.
 	hookLogs := apiClient.EXPECT().ContainerLogs(gomock.Any(), "hook-1", gomock.Any()).
diff --git a/pkg/e2e/hooks_test.go b/pkg/e2e/hooks_test.go
index e7db829f33..96bfff69b6 100644
--- a/pkg/e2e/hooks_test.go
+++ b/pkg/e2e/hooks_test.go
@@ -200,6 +200,39 @@ func TestPreStartHookNotReRunOnScaleUp(t *testing.T) {
 			OutputContains("1 /mnt/tokens.txt"))
 }
 
+func TestPreStartHookCreateThenStart(t *testing.T) {
+	s := NewScenario(t, "create must prepare the pre_start runners so a later start executes them")
+	s.Step("create leaves the service created, hooks not yet run",
+		ComposeCmd("create"),
+		ServiceState("sample", "created")).
+		Step("nothing in the volume before start",
+			probeVolume(s, "sh", "-c", "wc -l < /mnt/tokens.txt || echo missing"),
+			OutputContains("missing")).
+		Step("start runs the hook then the service",
+			ComposeCmd("start"),
+			ServiceState("sample", "running")).
+		Step("the hook ran exactly once",
+			probeVolume(s, "wc", "-l", "/mnt/tokens.txt"),
+			OutputContains("1 /mnt/tokens.txt"))
+}
+
+func TestPreStartHookStopThenStartFails(t *testing.T) {
+	NewScenario(t, "start after stop must fail on a hooked service: the runners were consumed, only a reconciliation (up) prepares new ones").
+		Step("up runs the hook and starts the service",
+			ComposeCmd("up", "-d", "--wait").Within(60*time.Second)).
+		Step("stop leaves the service exited",
+			ComposeCmd("stop"),
+			ServiceState("sample", "exited")).
+		Step("start fails, pointing at the reconciliation command",
+			ComposeCmd("start").MayFail(),
+			ExitCode(1),
+			OutputContains("pre_start[0]"),
+			OutputContains("docker compose up")).
+		Step("up prepares fresh runners and recovers",
+			ComposeCmd("up", "-d", "--wait").Within(60*time.Second),
+			ServiceState("sample", "running"))
+}
+
 func TestPreStartHookRunsOnceForScaledService(t *testing.T) {
 	s := NewScenario(t, "with the default per_replica: false, a pre_start hook must run once for the whole service")
 	s.Step("up starts both replicas",
diff --git a/pkg/e2e/testdata/TestPreStartHookCreateThenStart/compose.yaml b/pkg/e2e/testdata/TestPreStartHookCreateThenStart/compose.yaml
new file mode 100644
index 0000000000..e00ab31f58
--- /dev/null
+++ b/pkg/e2e/testdata/TestPreStartHookCreateThenStart/compose.yaml
@@ -0,0 +1,11 @@
+services:
+  sample:
+    image: alpine
+    command: sh -c 'sleep 30'
+    volumes:
+      - data:/shared
+    pre_start:
+      - image: alpine
+        command: sh -c 'echo $(cat /proc/sys/kernel/random/uuid) >> /shared/tokens.txt'
+volumes:
+  data:
diff --git a/pkg/e2e/testdata/TestPreStartHookStopThenStartFails/compose.yaml b/pkg/e2e/testdata/TestPreStartHookStopThenStartFails/compose.yaml
new file mode 100644
index 0000000000..e00ab31f58
--- /dev/null
+++ b/pkg/e2e/testdata/TestPreStartHookStopThenStartFails/compose.yaml
@@ -0,0 +1,11 @@
+services:
+  sample:
+    image: alpine
+    command: sh -c 'sleep 30'
+    volumes:
+      - data:/shared
+    pre_start:
+      - image: alpine
+        command: sh -c 'echo $(cat /proc/sys/kernel/random/uuid) >> /shared/tokens.txt'
+volumes:
+  data: