feat: reconciler plans the pre_start hook lifecycle

pre_start hooks run in ephemeral containers whose lifecycle — purge of
stale runners left by a previous failed run, creation, removal after
success — used to live entirely inside the imperative primitive, opaque
to the reconciliation plan. The plan now describes it: when pre_start is
going to run, each hook chains CreateHookContainer → RunPreStart →
RemoveContainer, preceded by best-effort removals of observed stale
runners. Only the run node stays imperative (start, wait, log streaming,
retain-on-failure) and no longer manages the container it executes: a
failed hook aborts before its remove node, so the container is retained
for inspection and purged as stale on the next run — the imperative
semantics, expressed as graph structure.

Observed state learns to tell hook containers apart: they carry no
container-number label and previously classified as a service replica
numbered 0. They now land in a dedicated HookContainers bucket the
reconciler plans purges from.

The imperative path is recomposed from the same pieces
(createPreStartContainer + execPreStartHook + explicit removal) and
behaves identically, as locked by the existing characterization tests.

Signed-off-by: Nicolas De Loof <nicolas.deloof@gmail.com>

Author: ndeloof

pkg/compose/observed_state.go

@@ -46,6 +46,11 @@ 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. 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.
+	HookContainers map[string][]ObservedContainer // service name → hook containers
 }
 
 // selectNetwork picks, among the live networks recorded for a compose key, the
@@ -148,6 +153,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 +177,15 @@ func (s *composeService) collectObservedState(ctx context.Context, project *type
 
 	for _, ctr := range raw {
 		svcName := ctr.Labels[api.ServiceLabel]
+		if ctr.Labels[api.HookLabel] != "" {
+			// 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.
+			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) {

pkg/compose/observed_state_test.go

@@ -160,6 +160,20 @@ 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",
+				},
+			},
 		},
 	}, nil)
 
@@ -202,6 +216,11 @@ func TestCollectObservedState(t *testing.T) {
 	assert.Equal(t, len(state.Containers["db"]), 1)
 	assert.Equal(t, state.Containers["db"][0].ID, "c2")
 
+	// 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: 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.

pkg/compose/plan.go

@@ -58,6 +58,12 @@ const (
 	OpWaitCondition OperationType = 40
 	OpRunPreStart   OperationType = 41
 	OpRunPostStart  OperationType = 42
+	// OpCreateHookContainer creates the ephemeral container an OpRunPreStart
+	// node then executes: the hook's lifecycle (stale-runner purge, creation,
+	// post-success removal) belongs to the plan, only its execution — start,
+	// wait, log streaming, retain-on-failure — stays with the imperative
+	// primitive.
+	OpCreateHookContainer OperationType = 43
 )
 
 // PlanPhase situates a node in the plan lifecycle. The Create phase converges
@@ -105,6 +111,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 +135,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/OpRunPreStart: index into the service's PreStart hooks
 	// 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

pkg/compose/pre_start.go

@@ -79,26 +79,38 @@ func (s *composeService) runPreStart(ctx context.Context, project *types.Project
 		logrus.Warnf("service %q: failed to remove stale pre_start hook containers: %v", service.Name, err)
 	}
 	for i, hook := range service.PreStart {
-		if err := s.runPreStartHook(ctx, project, service, ctr, i, hook, listener); err != nil {
+		created, err := s.createPreStartContainer(ctx, project, service, ctr, hook)
+		if err != nil {
+			return err
+		}
+		if err := s.execPreStartHook(ctx, service, i, created.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)
+		}
 	}
 	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,
+// 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).
+func (s *composeService) execPreStartHook(
+	ctx context.Context, service types.ServiceConfig,
+	index int, containerID string, listener api.ContainerEventListener,
 ) error {
-	created, err := s.createPreStartContainer(ctx, project, service, ctr, hook)
-	if err != nil {
-		return err
-	}
-
 	// 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 +120,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 +148,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,12 +165,6 @@ 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
 }
 

pkg/compose/reconcile.go

@@ -1187,25 +1187,7 @@ func (r *reconciler) planServiceStart(service types.ServiceConfig) error {
 	// observation — the imperative gating (initial up, force-recreate, or
 	// spec change), decided at plan time
 	if len(service.PreStart) > 0 && !anyRunning {
-		serviceCopy := service
-		first := replicas[0]
-		op := Operation{
-			Type:         OpRunPreStart,
-			ResourceID:   first.resID,
-			Cause:        "pre_start hooks",
-			Service:      &serviceCopy,
-			Container:    first.container,
-			CreateNodeID: first.createNodeID,
-		}
-		// clone: appending into prev's backing array would silently mutate
-		// depNodes' hidden capacity (they share it)
-		deps := slices.Clone(prev)
-		if first.after != nil {
-			deps = append(deps, first.after)
-		}
-		preStart := r.plan.addNode(op, startGroupID(first.resID), deps...)
-		preStart.Phase = PhaseStart
-		prev = []*PlanNode{preStart}
+		prev = r.planPreStart(service, replicas[0], prev)
 	}
 
 	// the replica chain: inject+start then post_start of replica n+1 waits
@@ -1247,6 +1229,86 @@ func (r *reconciler) planServiceStart(service types.ServiceConfig) error {
 	return nil
 }
 
+// planPreStart plans the pre_start sequence gating the first replica start.
+// The hook containers' lifecycle belongs to the plan: stale runners left by a
+// previous run that failed before removing them are purged first (best-effort,
+// the imperative warn-only removal), then each hook chains create → run →
+// remove. Only the run stays an imperative primitive — start, wait, log
+// streaming, retain-on-failure — it no longer manages the container it
+// executes: a failed hook aborts the chain before its remove node, so the
+// container is retained for inspection and purged as stale on the next run.
+// Returns the node(s) the first replica start must wait for.
+func (r *reconciler) planPreStart(service types.ServiceConfig, first startReplica, prev []*PlanNode) []*PlanNode {
+	serviceCopy := service
+	group := startGroupID(first.resID)
+	// clone: appending into prev's backing array would silently mutate
+	// depNodes' hidden capacity (they share it)
+	deps := slices.Clone(prev)
+	if first.after != nil {
+		deps = append(deps, first.after)
+	}
+
+	// observed stale runners are purged in parallel; the first hook waits for
+	// all of them, like the imperative engine purges before running any hook
+	stale := slices.Clone(r.observed.HookContainers[service.Name])
+	slices.SortFunc(stale, func(a, b ObservedContainer) int { return cmp.Compare(a.ID, b.ID) })
+	purges := make([]*PlanNode, 0, len(stale))
+	for i := range stale {
+		purge := 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,
+			BestEffort: true,
+		}, group, deps...)
+		purge.Phase = PhaseStart
+		purges = append(purges, purge)
+	}
+	deps = append(deps, purges...)
+
+	for i := range service.PreStart {
+		resID := fmt.Sprintf("hook:%s:pre_start:%d", service.Name, i)
+		create := r.plan.addNode(Operation{
+			Type:       OpCreateHookContainer,
+			ResourceID: resID,
+			Cause:      "pre_start hook",
+			Service:    &serviceCopy,
+			HookIndex:  i,
+			// the target replica whose volumes the hook shares (VolumesFrom):
+			// observed, or resolved from the create-phase node's result
+			Container:    first.container,
+			CreateNodeID: first.createNodeID,
+		}, group, deps...)
+		create.Phase = PhaseStart
+		run := r.plan.addNode(Operation{
+			Type:       OpRunPreStart,
+			ResourceID: resID,
+			Cause:      "pre_start hook",
+			Service:    &serviceCopy,
+			HookIndex:  i,
+			// the container to execute is the one the create node materialized
+			CreateNodeID: create.ID,
+		}, group, create)
+		run.Phase = PhaseStart
+		remove := r.plan.addNode(Operation{
+			Type:         OpRemoveContainer,
+			ResourceID:   resID,
+			Cause:        "hook completed",
+			Service:      &serviceCopy,
+			CreateNodeID: create.ID,
+			// a removal failure does not gate service start: the hook already
+			// succeeded (imperative parity — warn-only removal)
+			BestEffort: true,
+		}, group, run)
+		remove.Phase = PhaseStart
+		// hooks run sequentially, in declared order: the next hook (and
+		// ultimately the first replica start) waits for this one's removal
+		deps = []*PlanNode{remove}
+	}
+	return deps
+}
+
 // planRecreateContainer decomposes container recreation into 4 atomic operations:
 // CreateContainer(tmpName) → StopContainer → RemoveContainer → RenameContainer
 func (r *reconciler) planRecreateContainer(service types.ServiceConfig, oc *ObservedContainer, infraDeps []*PlanNode) *PlanNode {