feat: reconciliation plan creates pre_start hook runners

pre_start hook containers become first-class resources of the
reconciliation plan. The reconciler plans one OpCreateHookContainer per
declared hook — gated on the same predicate the start phase uses to run
the hooks: a replica to start and no running replica SURVIVING the plan
(a recreated replica is not running at start time, so its runners are
planned too). Runner creation depends on every replica node (the
executor resolves the VolumesFrom target against a final live view),
on the infrastructure, and on the purge of previously observed runners:
names are deterministic (<project>-<service>-pre_start-<i>) so repeated
plans converge on the same container instead of accumulating anonymous
ones, and a new HookIndexLabel ties each runner to its hook.

runPreStart becomes pure execution: it looks up the created-state
runner for each hook by label and never creates one. A declared hook
without a prepared runner is an actionable error naming the
reconciliation command (docker compose up <service>) — the accepted
consequence is that stop-then-start of a hooked service errors, since
runners are consumed on success (locked by e2e). Success removal,
failure retention and cancellation cleanup are unchanged.

Runners stay invisible to the start/ps container listings — they carry
no ConfigHashLabel, which getDefaultFilters requires — and down keeps
removing them by hook label.

Rebased onto main, which diverged substantially since this branch was
opened (the unrelated container-spec-layering refactor, #14093,
restructured compose-go's ServiceConfig/PreStartHook types this PR also
touches). Reconciling pre_start.go/pre_start_test.go against that
refactor surfaced a real bug: createPreStartContainer builds the hook's
config via the shared getCreateConfigs, which unconditionally stamps a
ConfigHashLabel — silently contradicting this same commit's own
"invisible to ps/start" guarantee above, since that label is exactly
what getDefaultFilters requires. Fixed by deleting the label after
getCreateConfigs returns, with a test asserting its absence (an
independent local review, via the GitHub Copilot CLI, and a
fresh-context review agent both converged on catching this during
rebase verification). Also restored two security/label tests that only
existed on one side of the rebase (SecuritySensitiveFieldsHonored,
HookLabelsMerged), ported to call createPreStartContainer directly.

Also addresses glours' three review comments on this PR: hook-container
create operations for a service no longer chain serially (each depends
on the same shared containerNodes+infraDeps+purges set, letting the
executor create them in parallel), the corresponding plan-string test
assertion, and a clearer missing-runner error message hinting this
isn't necessarily a planning bug.

Signed-off-by: Nicolas De Loof <[email protected]>

pkg/api/labels.go

@@ -69,10 +69,16 @@ const (
 	// ContainerEngineLabel stores the name of the engine that runs the container
 	ContainerEngineLabel = "com.docker.compose.engine"
 	// HookLabel identifies an ephemeral lifecycle-hook container and stores the
-	// hook type, e.g. "pre_start". It is applied to every container created by
-	// runPreStartHook so orphan hook containers from a previous failed run can
-	// be found and removed by project+service+hook label filters.
+	// hook type, e.g. "pre_start". It is applied to every runner container
+	// created by createPreStartContainer so stale runners from a previous
+	// failed run can be found and removed by project+service+hook label
+	// filters, and down can sweep them all by project+hook label.
 	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

pkg/compose/create.go

@@ -50,6 +50,11 @@ type createOptions struct {
 	AttachStdin       bool
 	UseNetworkAliases bool
 	Labels            types.Labels
+	// NoConfigHash skips the ConfigHashLabel stamp: hook runners have no
+	// spec-drift concept and must stay invisible to every listing that
+	// filters on the label's mere presence (ps/start's default listing,
+	// down's normal teardown path).
+	NoConfigHash bool
 }
 
 type createConfigs struct {
@@ -262,11 +267,13 @@ func (s *composeService) getCreateConfigs(ctx context.Context,
 	options createOptions,
 ) (createConfigs, error) {
 	labels := options.Labels
-	hash, err := ServiceHash(service)
-	if err != nil {
-		return createConfigs{}, err
+	if !options.NoConfigHash {
+		hash, err := ServiceHash(service)
+		if err != nil {
+			return createConfigs{}, err
+		}
+		labels[api.ConfigHashLabel] = hash
 	}
-	labels[api.ConfigHashLabel] = hash
 	if number > 0 {
 		// One-off containers are not indexed
 		labels[api.ContainerNumberLabel] = strconv.Itoa(number)

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:

pkg/compose/executor_ops.go

@@ -168,6 +168,42 @@ 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 {
+	// A dependency's done-channel also closes on failure: when a replica
+	// create failed and canceled the run, bail out on the cancellation
+	// instead of reporting the empty live view as an internal error.
+	if err := ctx.Err(); err != nil {
+		return err
+	}
+	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
+	}
+	// Recorded for uniformity with execCreateContainer — every create
+	// operation publishes its result. No plan operation consumes a hook
+	// runner's entry today: the start phase rediscovers runners through
+	// listPreStartRunners, by design.
+	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 {

pkg/compose/executor_test.go

@@ -23,6 +23,7 @@ import (
 	"testing"
 
 	"github.com/compose-spec/compose-go/v2/types"
+	"github.com/docker/cli/cli/config/configfile"
 	"github.com/moby/moby/api/types/container"
 	"github.com/moby/moby/api/types/network"
 	"github.com/moby/moby/client"
@@ -46,6 +47,8 @@ func newTestService(t *testing.T) (*composeService, *mocks.MockAPIClient) {
 	cli := mocks.NewMockCli(mockCtrl)
 	apiClient := mocks.NewMockAPIClient(mockCtrl)
 	cli.EXPECT().Client().Return(apiClient).AnyTimes()
+	cli.EXPECT().ConfigFile().Return(&configfile.ConfigFile{}).AnyTimes()
+	apiClient.EXPECT().DaemonHost().Return("unix:///var/run/docker.sock").AnyTimes()
 
 	svc, err := NewComposeService(cli, WithEventProcessor(noopEventProcessor{}))
 	assert.NilError(t, err)
@@ -518,3 +521,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",
+		ContainerSpec: types.ContainerSpec{Image: "alpine"},
+		PreStart: []types.PreStartHook{
+			{ContainerSpec: types.ContainerSpec{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",
+		ContainerSpec: types.ContainerSpec{Image: "alpine"},
+		PreStart:      []types.PreStartHook{{ContainerSpec: types.ContainerSpec{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`)
+}

pkg/compose/observed_state.go

@@ -47,9 +47,11 @@ type ObservedState struct {
 	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. Any observed at collection time is stale by definition —
-	// a previous run failed before removing it — and the reconciler plans
-	// its purge before re-running the hooks.
+	// 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
 }
 

pkg/compose/plan.go

@@ -58,6 +58,9 @@ const (
 	OpWaitCondition OperationType = 40
 	OpRunPreStart   OperationType = 41
 	OpRunPostStart  OperationType = 42
+
+	// Hook operations
+	OpCreateHookContainer OperationType = 43
 )
 
 // PlanPhase situates a node in the plan lifecycle. The Create phase converges
@@ -105,6 +108,8 @@ func (o OperationType) String() string {
 		return "RunPreStart"
 	case OpRunPostStart:
 		return "RunPostStart"
+	case OpCreateHookContainer:
+		return "CreateHookContainer"
 	default:
 		return fmt.Sprintf("Unknown(%d)", int(o))
 	}
@@ -127,6 +132,7 @@ type Operation struct {
 	Timeout      *time.Duration       // for stop operations
 	CreateNodeID int                  // for OpRenameContainer/start-phase ops: ID of the CreateContainer node whose result to target
 	Condition    string               // for OpWaitCondition: depends_on condition to wait for (service_healthy, ...)
+	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

pkg/compose/pre_start.go

@@ -58,6 +58,14 @@ func (s *composeService) resolveHookServiceReferences(ctx context.Context, proje
 	return resolveServiceReferences(hookService, byService)
 }
 
+// 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.
@@ -75,54 +83,96 @@ 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 {
-		created, err := s.createPreStartContainer(ctx, project, service, ctr, hook)
-		if err != nil {
-			return err
+	for i := range service.PreStart {
+		runner, ok := runners[i]
+		if !ok {
+			return fmt.Errorf("service %q pre_start[%d]: no hook runner container found — either its runner "+
+				"was already consumed by a previous start, or the service's container state changed after "+
+				"reconciliation planned this run; run `docker compose up %s` to prepare fresh runners",
+				service.Name, i, service.Name)
 		}
-		if err := s.execPreStartHook(ctx, service, i, created.ID, listener); err != nil {
+		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, created.ID, client.ContainerRemoveOptions{RemoveVolumes: true}); removeErr != nil {
-			logrus.Warnf("service %q pre_start[%d]: failed to remove hook container %s: %v", service.Name, i, created.ID, removeErr)
+		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
 }
 
+// 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))
+	f.Add("status", "created")
+	res, err := s.apiClient().ContainerList(ctx, client.ContainerListOptions{
+		All:     true,
+		Filters: f,
+	})
+	if err != nil {
+		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
+		}
+		if prev, exists := runners[index]; exists {
+			// Deterministic runner names make this impossible for runners
+			// created by this version; a leftover from a version predating
+			// deterministic naming could still collide on the index label.
+			// Surface the unexpected daemon state instead of masking it.
+			logrus.Warnf("service %q pre_start[%d]: multiple created hook runner containers found (%s, %s); using the latter",
+				serviceName, index, prev.ID, ctr.ID)
+		}
+		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 container's lifecycle
-// belongs to whoever created it (the imperative runPreStart loop today, the
-// reconciliation plan once the executor runs hook nodes).
+// 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,
@@ -188,10 +238,19 @@ func (s *composeService) execPreStartHook(
 	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.PreStartHook,
+	ctr container.Summary, index int, name string,
 ) (client.ContainerCreateResult, error) {
+	if index < 0 || index >= len(service.PreStart) {
+		return client.ContainerCreateResult{}, fmt.Errorf("service %q: hook index %d out of range (have %d hooks)", service.Name, index, len(service.PreStart))
+	}
+	hook := service.PreStart[index]
 	// A pre_start hook is a full container specification (compose-spec#656),
 	// already resolved by compose-go at load time: every service attribute
 	// the hook doesn't override — resources, capabilities, dns, sysctls,
@@ -215,7 +274,7 @@ func (s *composeService) createPreStartContainer(
 	}
 	cfgs, err := s.getCreateConfigs(ctx, project, hookService, 0, nil, createOptions{
 		// AutoRemove is intentionally false: a failed hook container is
-		// retained so the operator can inspect its logs. runPreStartHook
+		// retained so the operator can inspect its logs. execPreStartHook
 		// removes it explicitly on success, and the orphan sweep catches
 		// leftovers on the next run.
 		AutoRemove:        false,
@@ -224,19 +283,27 @@ func (s *composeService) createPreStartContainer(
 		// belongs to so `compose down` and label-scoped tooling can find it.
 		// HookLabel distinguishes hook containers from the real service
 		// container; no container-number: tooling telling replicas apart must
-		// not count hook containers. The hook's own labels — declared or
-		// inherited, like any other ContainerSpec attribute — merge in, the
-		// runtime set winning on conflicts.
+		// not count hook containers. HookIndexLabel lets the start phase and
+		// the purge match each runner to its declared hook. The hook's own
+		// labels — declared or inherited, like any other ContainerSpec
+		// attribute — merge in, the runtime set winning on conflicts.
 		Labels: mergeLabels(spec.Labels, types.Labels{
-			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),
 		}),
 	})
 	if err != nil {
 		return client.ContainerCreateResult{}, err
 	}
+	// getCreateConfigs unconditionally stamps a ConfigHashLabel (service
+	// containers use it to detect spec drift) -- a hook runner has no such
+	// concept and must stay invisible to ps/start's default listing and to
+	// down's normal teardown path, both of which filter on this label's mere
+	// presence (see removePreStartHookContainers).
+	delete(cfgs.Container.Labels, api.ConfigHashLabel)
 	// Mounts inherit from the live service container: volumes_from carries
 	// its anonymous and image volumes too. The hook's own mounts, already in
 	// the host config from the merged spec, take precedence on shared
@@ -252,6 +319,7 @@ func (s *composeService) createPreStartContainer(
 	}
 
 	created, err := s.apiClient().ContainerCreate(ctx, client.ContainerCreateOptions{
+		Name:             name,
 		Config:           cfg,
 		HostConfig:       hostCfg,
 		NetworkingConfig: networkingConfig,
@@ -298,32 +366,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:

pkg/compose/reconcile.go

@@ -125,6 +125,11 @@ type reconciler struct {
 	// stop+remove): the start phase must never plan a start for them — the
 	// imperative engine only ever starts what survives the convergence.
 	removedByPlan map[string]bool
+	// hookRunnerNodes records, per service, the create-phase nodes that
+	// materialize the pre_start hook runners, so the start-phase RunPreStart
+	// node can depend on them — the executor must never run the hooks before
+	// their runners exist.
+	hookRunnerNodes map[string][]*PlanNode
 	// stoppedByPlan records containers already stopped by an earlier stage
 	// of the plan (typically planRecreateNetwork) so that downstream stages
 	// can chain on the existing OpStopContainer instead of emitting a second
@@ -176,6 +181,7 @@ func reconcile(_ context.Context, project *types.Project, observed *ObservedStat
 		waitNodes:                   map[string]*PlanNode{},
 		removedByPlan:               map[string]bool{},
 		stoppedByPlan:               map[string]*PlanNode{},
+		hookRunnerNodes:             map[string][]*PlanNode{},
 		connectNodes:                map[string][]*PlanNode{},
 		recreatedServices:           map[string]bool{},
 		observedContainersByService: observed.containersByService(),
@@ -769,8 +775,6 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 		return err
 	}
 
-	r.planPurgeStaleHookRunners(service, expected)
-
 	containers := r.observed.Containers[service.Name]
 	actual := len(containers)
 
@@ -797,6 +801,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 {
@@ -819,12 +831,14 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 				Container:  &containers[i].Summary,
 			}, "", stopNode)
 			r.removedByPlan[resID] = true
+			containerNodes = append(containerNodes, lastNode)
 			continue
 		}
 
 		if r.mustRecreate(service, expectedHash, parentRecreated, oc, strategy) {
 			lastNode = r.planRecreateContainer(service, &containers[i], infraDeps)
 			r.setContainerNode(service.Name, oc.Number, lastNode)
+			containerNodes = append(containerNodes, lastNode)
 			r.recreatedServices[service.Name] = true
 			continue
 		}
@@ -837,6 +851,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{
@@ -846,6 +863,10 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 				Container:  &containers[i].Summary,
 			}, "", infraDeps...)
 			r.setContainerNode(service.Name, oc.Number, lastNode)
+			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
 		}
 	}
 
@@ -864,40 +885,46 @@ func (r *reconciler) reconcileService(service types.ServiceConfig) error {
 			Name:       name,
 		}, "", infraDeps...)
 		r.setContainerNode(service.Name, number, lastNode)
+		containerNodes = append(containerNodes, lastNode)
 	}
 
+	r.planPreStartHookRunners(service, expected, keptRunning, containerNodes, infraDeps)
+
 	if lastNode != nil {
 		r.serviceNodes[service.Name] = lastNode
 	}
 	return nil
 }
 
-// mustRecreate decides whether oc must be recreated to match expected. The
-// expectedHash and parentRecreated inputs are precomputed once per service by
-// planPurgeStaleHookRunners plans the removal of hook-runner containers left
-// behind by a previous run that failed before removing them. It mirrors the
-// imperative purge living inside the gated runPreStart call: emitted only when
-// pre_start is going to run again — hooks declared, a replica to start
-// (scale > 0: the imperative start path returns before the hooks for a
-// scale-0 service) and no replica running at observation — so a genuinely
-// failed hook container stays retained for inspection as long as its service
-// is otherwise up. Removals are best-effort (the imperative purge is
-// warn-only) and independent of every other node.
-func (r *reconciler) planPurgeStaleHookRunners(service types.ServiceConfig, expectedScale int) {
-	stale := r.observed.HookContainers[service.Name]
-	if len(stale) == 0 || len(service.PreStart) == 0 || expectedScale == 0 {
+// planPreStartHookRunners plans the create-phase side of the pre_start hook
+// lifecycle. With 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 — when there is also a replica to start (scale > 0)
+// — creates one fresh runner per declared hook. The purge requires neither
+// hooks to still be declared nor a positive scale: a service that dropped its
+// pre_start declarations or was scaled down to zero must not keep runners of
+// the prior generation accumulating until down — the same rationale as the
+// per_replica return below. 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 keptRunning {
 		return
 	}
-	for _, oc := range r.observed.Containers[service.Name] {
-		if oc.State == container.StateRunning {
-			return
-		}
-	}
 	serviceCopy := service
-	stale = slices.Clone(stale)
+	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 {
-		r.plan.addNode(Operation{
+		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",
@@ -906,9 +933,29 @@ func (r *reconciler) planPurgeStaleHookRunners(service types.ServiceConfig, expe
 			RemoveVolumes: true,
 			BestEffort:    true,
 		}, "")
+		deps = append(deps, node)
+	}
+	if expectedScale == 0 {
+		// No replica will start, so there is no hook to prepare — but the
+		// stale-runner purge above already ran: scaling a service down to
+		// zero must not leave runners of the prior generation behind.
+		return
+	}
+	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...)
+		r.hookRunnerNodes[service.Name] = append(r.hookRunnerNodes[service.Name], 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
 // the rationale (issue #13878).
 func (r *reconciler) mustRecreate(expected types.ServiceConfig, expectedHash string, parentRecreated bool, oc ObservedContainer, policy string) bool {
@@ -1323,6 +1370,9 @@ func (r *reconciler) planServiceStart(service types.ServiceConfig) error {
 		if first.after != nil {
 			deps = append(deps, first.after)
 		}
+		// the hooks execute the runners the create phase prepared: they must
+		// exist before RunPreStart runs
+		deps = append(deps, r.hookRunnerNodes[service.Name]...)
 		preStart := r.plan.addNode(op, startGroupID(first.resID), deps...)
 		preStart.Phase = PhaseStart
 		prev = []*PlanNode{preStart}

pkg/compose/reconcile_start_test.go

@@ -155,11 +155,12 @@ func TestPlanStart_HooksAndReplicaChain(t *testing.T) {
 	assert.Equal(t, plan.String(), strings.TrimSpace(`
 [] -> #1 service:app:1, CreateContainer, no existing container
 [] -> #2 service:app:2, CreateContainer, no existing container
-[1] -> #3 service:app:1, RunPreStart, pre_start hooks [start:app:1] {start}
-[3] -> #4 service:app:1, StartContainer, start [start:app:1] {start}
-[4] -> #5 service:app:1, RunPostStart, post_start hooks [start:app:1] {start}
-[2,5] -> #6 service:app:2, StartContainer, start [start:app:2] {start}
-[6] -> #7 service:app:2, RunPostStart, post_start hooks [start:app:2] {start}
+[1,2] -> #3 hook:app:pre_start:0, CreateHookContainer, pre_start hook
+[1,3] -> #4 service:app:1, RunPreStart, pre_start hooks [start:app:1] {start}
+[4] -> #5 service:app:1, StartContainer, start [start:app:1] {start}
+[5] -> #6 service:app:1, RunPostStart, post_start hooks [start:app:1] {start}
+[2,6] -> #7 service:app:2, StartContainer, start [start:app:2] {start}
+[7] -> #8 service:app:2, RunPostStart, post_start hooks [start:app:2] {start}
 `)+"\n")
 }
 

pkg/compose/reconcile_test.go

@@ -1363,10 +1363,13 @@ func TestReconcileContainers_StaleHookRunnersPurged(t *testing.T) {
 	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 hook:app:stale:stale-a-id, RemoveContainer, stale pre_start hook container
-[] -> #2 hook:app:stale:stale-b-id, RemoveContainer, stale pre_start hook container
-[] -> #3 service:app:1, CreateContainer, no existing container
+[] -> #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 {
@@ -1378,9 +1381,54 @@ func TestReconcileContainers_StaleHookRunnersPurged(t *testing.T) {
 	}
 }
 
-// A scale-0 service never reaches its pre_start hooks (the imperative start
-// path returns before them), so its stale runners are not purged either.
-func TestReconcileContainers_StaleHookRunnersKeptWhenScaleZero(t *testing.T) {
+// A service that dropped ALL its pre_start declarations still gets its stale
+// runners purged (they are observed whenever any service in the model keeps
+// hooks — see mergeHookContainers): without the purge they would accumulate
+// until down, exactly what the per_replica path already guards against. No
+// runner is created, of course.
+func TestReconcileContainers_HooksRemovedStillPurgesStaleRunners(t *testing.T) {
+	project := &types.Project{
+		Name: "myproject",
+		Services: types.Services{
+			// app dropped its hooks; keeper still declares one, so the
+			// observed state lists hook containers at all
+			"app":    {Name: "app", Scale: intPtr(1)},
+			"keeper": {Name: "keeper", Scale: intPtr(1), PreStart: []types.PreStartHook{{}}},
+		},
+	}
+	observed := &ObservedState{
+		ProjectName: "myproject",
+		Containers:  map[string][]ObservedContainer{},
+		HookContainers: map[string][]ObservedContainer{
+			"app": {
+				{ID: "stale-app-id", Summary: container.Summary{ID: "stale-app-id"}},
+			},
+		},
+		Networks: map[string][]ObservedNetwork{},
+		Volumes:  map[string][]ObservedVolume{},
+	}
+
+	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
+	assert.NilError(t, err)
+
+	purged, created := false, 0
+	for _, n := range plan.Nodes {
+		if n.Operation.Type == OpRemoveContainer && n.Operation.Container.ID == "stale-app-id" {
+			purged = true
+		}
+		if n.Operation.Type == OpCreateHookContainer && n.Operation.Service.Name == "app" {
+			created++
+		}
+	}
+	assert.Assert(t, purged, "app's stale runner must be purged despite app declaring no hook anymore:\n%s", plan)
+	assert.Equal(t, created, 0, "no runner must be created for a hook-less service")
+}
+
+// A service scaled down to zero still gets its stale runners purged: hook
+// containers live in HookContainers, not Containers, so the regular
+// scale-down removal never touches them — without the purge they would
+// accumulate until down. No runner is created, since no replica will start.
+func TestReconcileContainers_ScaleZeroStillPurgesStaleRunners(t *testing.T) {
 	project := &types.Project{
 		Name: "myproject",
 		Services: types.Services{
@@ -1391,7 +1439,9 @@ func TestReconcileContainers_StaleHookRunnersKeptWhenScaleZero(t *testing.T) {
 		ProjectName: "myproject",
 		Containers:  map[string][]ObservedContainer{},
 		HookContainers: map[string][]ObservedContainer{
-			"app": {{ID: "stale-a-id", Summary: container.Summary{ID: "stale-a-id"}}},
+			"app": {
+				{ID: "stale-app-id", Summary: container.Summary{ID: "stale-app-id"}},
+			},
 		},
 		Networks: map[string][]ObservedNetwork{},
 		Volumes:  map[string][]ObservedVolume{},
@@ -1399,13 +1449,98 @@ func TestReconcileContainers_StaleHookRunnersKeptWhenScaleZero(t *testing.T) {
 
 	plan, err := reconcile(t.Context(), project, observed, defaultReconcileOptions(), noPrompt)
 	assert.NilError(t, err)
-	assert.Assert(t, plan.IsEmpty(), "unexpected plan:\n%s", plan)
+
+	purged, created := false, 0
+	for _, n := range plan.Nodes {
+		if n.Operation.Type == OpRemoveContainer && n.Operation.Container.ID == "stale-app-id" {
+			purged = true
+		}
+		if n.Operation.Type == OpCreateHookContainer && n.Operation.Service.Name == "app" {
+			created++
+		}
+	}
+	assert.Assert(t, purged, "app's stale runner must be purged despite scale=0:\n%s", plan)
+	assert.Equal(t, created, 0, "no runner must be created for a scale-0 service")
 }
 
-// A running replica gates pre_start off, so the stale runner stays: the
-// imperative purge lives inside the gated runPreStart call and would not run
-// either — a genuinely failed hook container stays retained for inspection as
-// long as its service is otherwise up.
+// A cold start of a hooked service plans one runner per declared hook,
+// independently after the replica create — not chained to each other, since
+// each targets the same already-live replica and has no reason to wait on a
+// sibling hook's create.
+func TestReconcileContainers_HookRunnersPlannedOnColdStart(t *testing.T) {
+	project := &types.Project{
+		Name: "myproject",
+		Services: types.Services{
+			"app": {Name: "app", Scale: intPtr(1), PreStart: []types.PreStartHook{{}, {}}},
+		},
+	}
+	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
+[1] -> #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.PreStartHook{{}}}
+	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 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.PreStartHook{{}}}
 	hash := mustServiceHash(t, svc)
@@ -1436,32 +1571,6 @@ func TestReconcileContainers_StaleHookRunnersKeptWhenReplicaRunning(t *testing.T
 	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",

pkg/compose/service_containers.go

@@ -606,11 +606,12 @@ 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. Pick the replica with the lowest
+	// container-number so the choice is deterministic regardless of the
+	// order the daemon returns containers in.
 	if candidate := lowestNumberedContainer(toStart); len(service.PreStart) > 0 && len(serviceContainers) == len(toStart) && !isRelayContainer(candidate) {
-		if err := s.runPreStart(ctx, project, service, candidate, listener); err != nil {
+		if err := s.runPreStart(ctx, project, service, listener); err != nil {
 			return err
 		}
 	}

pkg/compose/start_test.go

@@ -190,11 +190,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"}
@@ -209,21 +209,13 @@ func TestStartService_PreStartOnLowestReplica(t *testing.T) {
 	// Listed out of order on purpose: replica 2 first.
 	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)
+		Return(client.ContainerListResult{Items: []container.Summary{runnerSummary("hook-1", 0)}}, 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
-		})
 	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()).

pkg/dryrun/dryrunclient.go

@@ -95,6 +95,20 @@ func (d *DryRunClient) resolve(ctx context.Context, ref string) error {
 	return err
 }
 
+// matchesLabelFilters reports whether labels satisfy every requested
+// "key" or "key=value" label filter, mirroring the daemon's label filtering
+// for the faked containers the cache serves.
+func matchesLabelFilters(labels map[string]string, wanted map[string]bool) bool {
+	for f := range wanted {
+		k, v, hasValue := strings.Cut(f, "=")
+		actual, ok := labels[k]
+		if !ok || (hasValue && actual != v) {
+			return false
+		}
+	}
+	return true
+}
+
 func getCallingFunction() string {
 	pc, _, _, _ := runtime.Caller(2)
 	fullName := runtime.FuncForPC(pc).Name()
@@ -122,6 +136,9 @@ func (d *DryRunClient) ContainerCreate(ctx context.Context, options client.Conta
 		ID:     options.Name,
 		Names:  []string{options.Name},
 		Labels: options.Config.Labels,
+		// a container just created is in created state; listings that
+		// filter on state (listPreStartRunners) must see the faked ones
+		State: containerType.StateCreated,
 		HostConfig: struct {
 			NetworkMode string            `json:",omitempty"`
 			Annotations map[string]string `json:",omitempty"`
@@ -179,6 +196,20 @@ func (d *DryRunClient) ContainerList(ctx context.Context, options client.Contain
 				Items: d.containers,
 			}, err
 		}
+	case "listPreStartRunners":
+		// Hook runners created under dry-run exist only in the cache: answer
+		// from it, honoring the caller's label filters (project, service,
+		// hook type). The status=created filter is implicit — every faked
+		// container is in created state by construction, and a real runner
+		// seeded into the cache in any other state is discarded by the
+		// caller's own state check.
+		var items []containerType.Summary
+		for _, ctr := range d.containers {
+			if matchesLabelFilters(ctr.Labels, options.Filters["label"]) {
+				items = append(items, ctr)
+			}
+		}
+		return client.ContainerListResult{Items: items}, nil
 	}
 	return d.apiClient.ContainerList(ctx, options)
 }

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",

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:

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: