feat(up): interactive up runs on a single plan

pkg/api/api.go

@@ -364,7 +364,8 @@ type CreateOptions struct {
 
 // StartOptions group options of the Start API
 type StartOptions struct {
-	// Project is the compose project used to define this app. Might be nil if user ran command just with project name
+	// Project is the compose project used to define this app. Might be nil if user ran command just with project name.
+	// Ignored by Up, which works on the project it is given.
 	Project *types.Project
 	// Attach receives the containers' logs during Up's foreground session.
 	// It doubles as the mode switch: when nil, Up returns once containers
@@ -382,7 +383,9 @@ type StartOptions struct {
 	// ExitCodeFrom reports the exit code of the specified service. Honored
 	// by Up's foreground session only; ignored by Start.
 	ExitCodeFrom string
-	// Wait won't return until containers reached the running|healthy state
+	// Wait won't return until containers reached the running|healthy state.
+	// Up honors it in detached mode only (Attach nil): combined with Attach it
+	// is rejected, see UpOptions.Validate.
 	Wait        bool
 	WaitTimeout time.Duration
 	// Services names the services the user explicitly targeted; it only

pkg/compose/apiSocket.go

@@ -40,6 +40,13 @@ func (s *composeService) useAPISocket(project *types.Project) (*types.Project, e
 	if !useAPISocket {
 		return project, nil
 	}
+	// The project already went through here, as the plan's own project does
+	// when it is handed to a later plan (a watch rebuild): transforming it a
+	// second time would append the socket volume and the config entry again,
+	// duplicate entries in the model that also change the service's config hash.
+	if _, applied := project.Configs["#apisocket"]; applied {
+		return project, nil
+	}
 
 	if s.getContextInfo().ServerOSType() == "windows" {
 		return nil, errors.New("use_api_socket can't be used with a Windows Docker Engine")

pkg/compose/apiSocket_test.go

@@ -0,0 +1,58 @@
+/*
+   Copyright 2026 Docker Compose CLI authors
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.
+*/
+
+package compose
+
+import (
+	"testing"
+
+	"github.com/compose-spec/compose-go/v2/types"
+	"gotest.tools/v3/assert"
+)
+
+// linuxContextInfo is the context of a Linux engine.
+type linuxContextInfo struct{}
+
+func (linuxContextInfo) CurrentContext() string         { return "default" }
+func (linuxContextInfo) ServerOSType() string           { return "linux" }
+func (linuxContextInfo) BuildKitEnabled() (bool, error) { return false, nil }
+
+// TestUseAPISocketIsIdempotent verifies that a project which already went
+// through useAPISocket -- the plan's own project, handed to a later plan such
+// as a watch rebuild -- is not transformed a second time: the socket and the
+// config file would be mounted twice.
+func TestUseAPISocketIsIdempotent(t *testing.T) {
+	svc, _ := newTestService(t, WithContextInfo(linuxContextInfo{}))
+	project := &types.Project{
+		Name: "test",
+		Services: types.Services{
+			"app": {Name: "app", ContainerSpec: types.ContainerSpec{UseAPISocket: true, Environment: types.MappingWithEquals{}}},
+		},
+		Configs: types.Configs{},
+	}
+
+	once, err := svc.useAPISocket(project)
+	assert.NilError(t, err)
+	service := once.Services["app"]
+	assert.Equal(t, len(service.Volumes), 1, "volumes: %v", service.Volumes)
+	assert.Equal(t, len(service.Configs), 1, "configs: %v", service.Configs)
+
+	twice, err := svc.useAPISocket(once)
+	assert.NilError(t, err)
+	service = twice.Services["app"]
+	assert.Equal(t, len(service.Volumes), 1, "volumes: %v", service.Volumes)
+	assert.Equal(t, len(service.Configs), 1, "configs: %v", service.Configs)
+}

pkg/compose/executor_start_test.go

@@ -820,8 +820,7 @@ func TestExecRunPreStart_BailsOutWhenContextCanceled(t *testing.T) {
 // TestExecRunPostStart_RunsHookAndStreamsToListener verifies that
 // execRunPostStart resolves the replica the start chain just brought up and
 // runs its post_start hooks, forwarding their output to the listener the
-// executor was constructed with — the plumbing an attached `up` will use
-// once the interactive-up convergence wires a real one in.
+// executor was constructed with.
 func TestExecRunPostStart_RunsHookAndStreamsToListener(t *testing.T) {
 	svc, apiClient, _ := newStartPhaseTestService(t)
 

pkg/compose/monitor.go

@@ -225,8 +225,9 @@ func (c *monitor) Start(ctx context.Context) error {
 // as they were before the replay existed, with a warning: this is a missing
 // capability, not a reason to fail. A daemon that answers with no time or one
 // that can't be parsed is an error: the replay would have no sound starting
-// point, and nothing explains the answer. An interruption is an error too,
-// for the caller to end quietly.
+// point, and nothing explains the answer. An interruption is an error too, the
+// client's, which carries the context's: callers decide how to report it (Start
+// ends quietly, an interactive up returns it).
 func (c *monitor) readDaemonTime(ctx context.Context) error {
 	if c.sinceRead {
 		return nil

pkg/compose/monitor_test.go

@@ -518,6 +518,8 @@ type fakeEventLog struct {
 	entries []loggedEvent
 	// since records the Since of every subscription, in order
 	since []string
+	// subscribers receive the events emitted after they subscribed
+	subscribers []chan events.Message
 
 	// skew is how far the engine's clock is ahead of the local one
 	skew time.Duration
@@ -566,17 +568,22 @@ func (l *fakeEventLog) emit(msg events.Message) {
 	now := l.now()
 	msg.TimeNano = now.UnixNano()
 	l.entries = append(l.entries, loggedEvent{at: now, msg: msg})
+	for _, sub := range l.subscribers {
+		sub <- msg
+	}
 }
 
 // subscribe opens a stream replaying the buffered events newer than
-// opts.Since (none when it is empty), followed by the given live events.
+// opts.Since (none when it is empty), followed by the given live events, then
+// by whatever is emitted from now on.
 func (l *fakeEventLog) subscribe(t *testing.T, opts client.EventsListOptions, live ...events.Message) client.EventsResult {
 	t.Helper()
 	l.mu.Lock()
 	defer l.mu.Unlock()
 	l.since = append(l.since, opts.Since)
 
-	messages := make(chan events.Message, len(l.entries)+len(live))
+	messages := make(chan events.Message, len(l.entries)+len(live)+64)
+	l.subscribers = append(l.subscribers, messages)
 	if opts.Since != "" {
 		since := parseEventsSince(t, opts.Since)
 		for _, e := range l.entries {

pkg/compose/run.go

@@ -315,7 +315,7 @@ func (s *composeService) startDependencies(ctx context.Context, project *types.P
 	if len(project.Services) > 0 {
 		return s.start(ctx, project.Name, api.StartOptions{
 			Project: project,
-		}, nil)
+		})
 	}
 	return nil
 }

pkg/compose/start.go

@@ -29,7 +29,7 @@ import (
 
 func (s *composeService) Start(ctx context.Context, projectName string, options api.StartOptions) error {
 	return Run(ctx, func(ctx context.Context) error {
-		return s.start(ctx, strings.ToLower(projectName), options, nil)
+		return s.start(ctx, strings.ToLower(projectName), options)
 	}, "start", s.events)
 }
 
@@ -39,12 +39,10 @@ func (s *composeService) Start(ctx context.Context, projectName string, options
 // other lifecycle command already uses. ScopeStart's planStartPhase works
 // straight off the observed containers (see TestPlanStart_StartOnlyScope) --
 // it never converges anything, matching start()'s own historical contract of
-// starting what exists, never creating. listener is non-nil only when called
-// from interactive up's own start phase (today's create()-then-start()
-// sequence, unchanged by this PR -- see upDetached's doc comment): it streams
-// pre_start/post_start hook logs into the attached session, exactly what
-// newPlanExecutor's listener parameter exists for.
-func (s *composeService) start(ctx context.Context, projectName string, options api.StartOptions, listener api.ContainerEventListener) error {
+// starting what exists, never creating. The Start phase of an interactive up
+// does not go through here: it runs from the up's own plan, and hands its
+// session's listener to runStart (see runInteractiveUp).
+func (s *composeService) start(ctx context.Context, projectName string, options api.StartOptions) error {
 	project := options.Project
 	if project == nil {
 		containers, err := s.getContainers(ctx, projectName, oneOffExclude, true)
@@ -88,7 +86,7 @@ func (s *composeService) start(ctx context.Context, projectName string, options
 	// combination nothing rejects). exec.waitTimeout reproduces that for the
 	// plan's own OpWaitCondition nodes, each with its own fresh window
 	// starting when that wait begins -- see upDetached's identical reasoning.
-	exec := s.newPlanExecutor(project, observed, listener)
+	exec := s.newPlanExecutor(project, observed, nil)
 	exec.waitTimeout = options.WaitTimeout
 
 	if err := exec.run(ctx, plan); err != nil {

pkg/compose/start_test.go

@@ -144,7 +144,7 @@ func TestStart_NoOpStartIsSilentAndOnlyListsContainers(t *testing.T) {
 			stoppedServiceContainer("web", 1, container.StateRunning),
 		}}, nil)
 
-	err := svc.start(t.Context(), "prj", api.StartOptions{Project: project}, nil)
+	err := svc.start(t.Context(), "prj", api.StartOptions{Project: project})
 	assert.NilError(t, err)
 	assert.Equal(t, len(rec.summary()), 0, "a no-op start must be silent: %v", rec.summary())
 }
@@ -184,7 +184,7 @@ func TestStart_WaitTimeoutBoundsDependencyWaits(t *testing.T) {
 	// failure instead of a hang
 	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
 	defer cancel()
-	err := svc.start(ctx, "prj", api.StartOptions{Project: project, WaitTimeout: 200 * time.Millisecond}, nil)
+	err := svc.start(ctx, "prj", api.StartOptions{Project: project, WaitTimeout: 200 * time.Millisecond})
 	assert.Error(t, err, "timeout waiting for dependencies")
 }
 

pkg/compose/up.go

@@ -44,11 +44,14 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options
 	if err := options.Validate(); err != nil {
 		return err
 	}
+	var created *createdUp
 	err := Run(ctx, tracing.SpanWrapFunc("project/up", tracing.ProjectOptions(ctx, project), func(ctx context.Context) error {
 		if options.Start.Attach == nil {
 			return s.upDetached(ctx, project, options)
 		}
-		return s.create(ctx, project, options.Create)
+		var err error
+		created, err = s.upCreatePhase(ctx, project, options)
+		return err
 	}), "up", s.events)
 	if err != nil {
 		return err
@@ -61,20 +64,46 @@ func (s *composeService) Up(ctx context.Context, project *types.Project, options
 		_, _ = fmt.Fprintln(s.stdout(), "end of 'compose up' output, interactive run is not supported in dry-run mode")
 		return err
 	}
-	return s.runInteractiveUp(ctx, project, options)
+	return s.runInteractiveUp(ctx, created, options)
 }
 
-// upDetached runs detached `up`'s Create and Start phases as a single plan:
-// the semantic switchover from create() + start() -- two separate daemon
-// snapshots, the second blind to what the first just did -- to one
-// preparePlan/executePlan pass covering both phases (epic #14081, lot 2).
-// Interactive up keeps the create()+start() sequence for now: its own
-// create/start phase boundary (attach/printer/monitor setup in between) is a
-// separate, riskier item of the same lot.
-func (s *composeService) upDetached(ctx context.Context, project *types.Project, options api.UpOptions) error {
+// createdUp is what an interactive up carries from its Create phase to its
+// Start phase: the canonical project the plan was built for, and the plan run
+// whose Start phase is still to dispatch. Both phases share the one daemon
+// snapshot and the one plan the Create phase was prepared from.
+type createdUp struct {
+	project *types.Project
+	run     *planRun
+}
+
+// upCreatePhase prepares interactive up's single Create+Start plan and runs
+// its Create phase only, under the "up" progress scope the caller wraps it in.
+// The Start phase is deliberately left out of that scope: it runs once
+// runInteractiveUp has set up the session streaming logs to the terminal, and
+// the progress display's refresh would repaint over them.
+func (s *composeService) upCreatePhase(ctx context.Context, project *types.Project, options api.UpOptions) (*createdUp, error) {
+	project, exec, plan, err := s.prepareUp(ctx, project, options)
+	if err != nil {
+		return nil, err
+	}
+
+	run, err := exec.begin(plan)
+	if err != nil {
+		return nil, err
+	}
+	if err := run.runCreate(ctx); err != nil {
+		return nil, err
+	}
+	return &createdUp{project: project, run: run}, nil
+}
+
+// prepareUp prepares the single Create+Start plan of an up, detached or
+// interactive, and the executor that runs it. It returns the plan's own
+// project, the one carrying what preparation resolved (see useAPISocket).
+func (s *composeService) prepareUp(ctx context.Context, project *types.Project, options api.UpOptions) (*types.Project, *planExecutor, *Plan, error) {
 	project, observed, plan, err := s.preparePlan(ctx, project, options.Create, ScopeCreateStart)
 	if err != nil {
-		return err
+		return nil, nil, nil, err
 	}
 
 	// Must run against the pre-execution snapshot: observed only labels a
@@ -98,6 +127,20 @@ func (s *composeService) upDetached(ctx context.Context, project *types.Project,
 	// per-wait-independent behavior never had that failure mode.
 	exec := s.newPlanExecutor(project, observed, nil)
 	exec.waitTimeout = options.Start.WaitTimeout
+	return project, exec, plan, nil
+}
+
+// upDetached runs detached `up`'s Create and Start phases as a single plan:
+// the semantic switchover from create() + start() -- two separate daemon
+// snapshots, the second blind to what the first just did -- to one
+// preparePlan/executePlan pass covering both phases (epic #14081, lot 2).
+// Interactive up prepares the same plan but drives its two phases itself,
+// with the attach/printer/monitor setup in between (see upCreatePhase).
+func (s *composeService) upDetached(ctx context.Context, project *types.Project, options api.UpOptions) error {
+	project, exec, plan, err := s.prepareUp(ctx, project, options)
+	if err != nil {
+		return err
+	}
 
 	if err := exec.run(ctx, plan); err != nil {
 		return translateWaitTimeout(err)
@@ -181,7 +224,18 @@ type upSession struct {
 	exitCode     int
 }
 
-func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Project, options api.UpOptions) error {
+// runInteractiveUp sets up the attach/printer/monitor session around the
+// Start phase of created's plan, runs it, then follows the application until
+// it terminates. The Create phase already ran: the session's own signal
+// handling only starts after it, so an interruption of the Create phase is the
+// caller's, through ctx. An interruption met while the session attaches or
+// reads the daemon's time starts nothing; one arriving later does not stop the
+// Start phase, which runs on a context that outlives it.
+func (s *composeService) runInteractiveUp(ctx context.Context, created *createdUp, options api.UpOptions) error {
+	// the plan's own project, not the caller's: the one the Start phase runs
+	// on, so the session, the watcher and the teardown follow the same model
+	project := created.project
+
 	// if we get a second signal during shutdown, we kill the services
 	// immediately, so the channel needs to have sufficient capacity or
 	// we might miss a signal while setting up the second channel read
@@ -283,6 +337,17 @@ func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Pr
 	}
 	monitor.withListener(u.followStartedContainers(attached))
 
+	// The monitor replays events from the daemon's time. Read it now, before
+	// the Start phase: the point precedes the first start, and a daemon that
+	// answers with a time that can't be used fails the command without any
+	// container started. A daemon that can't be asked doesn't (see
+	// readDaemonTime): the monitor then follows live events.
+	if err := monitor.readDaemonTime(globalCtx); err != nil {
+		cancel()
+		_ = u.eg.Wait()
+		return err
+	}
+
 	u.eg.Go(func() error {
 		err := monitor.Start(globalCtx)
 		// The monitor returning means every watched container is gone for
@@ -310,7 +375,8 @@ func (s *composeService) runInteractiveUp(ctx context.Context, project *types.Pr
 	})
 
 	// We use the parent context without cancellation as we manage sigterm to stop the stack
-	err = s.start(context.WithoutCancel(ctx), project.Name, options.Start, u.printer.HandleEvent)
+	err = created.run.runStart(context.WithoutCancel(ctx), u.printer.HandleEvent)
+	err = translateWaitTimeout(err)
 	if err != nil && !u.isTerminated.Load() { // Ignore error if the process is terminated
 		cancel()
 		_ = u.eg.Wait()