Add Docker controlled-session controller adapter

docs/CONTROLLED_SESSION_DESIGN.md

@@ -33,7 +33,12 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta
   workload plan inert, establishes the Engine attachment before start, applies
   the initial and later dimensions through the Engine API, preserves exact
   input and output bytes, independently observes the container exit code, and
-  exposes graceful and forced stop operations. Controller launch, protocol/PTY
+  exposes graceful and forced stop operations. The Docker controller adapter
+  is also implemented: it requires the exact lease-private channel socket
+  before container creation, creates the frozen controller plan inert, starts
+  it at most once, independently observes its exit, exposes graceful and forced
+  stop operations, captures the full container ID returned by creation, and
+  pins all later lifecycle operations to that exact container. Protocol/PTY
   bridging, controlled-session networking, and lifecycle orchestration remain
   later slices.
 - Initial runtime: Linux containers under Docker

internal/dockerdeploy/controlled_session_controller.go

@@ -0,0 +1,312 @@
+package dockerdeploy
+
+import (
+	"bytes"
+	"context"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"sync"
+
+	"github.com/omry/reploy/internal/controlledsession"
+)
+
+type dockerControllerBackendV1 struct {
+	run                 commandRunner
+	observe             func(context.Context, CommandSpec, string) (int, error)
+	requireReadyChannel func(ControlledSessionContainerPlanV1) error
+}
+
+type dockerControllerWaitResultV1 struct {
+	status controlledsession.ProcessStatusV1
+	err    error
+}
+
+// DockerControllerV1 is the narrow Docker boundary for the trusted controller
+// container. Creation uses the immutable planned name; every later operation
+// uses the exact full container ID returned by that successful create.
+type DockerControllerV1 struct {
+	plan        ControlledSessionContainerPlanV1
+	containerID string
+	backend     dockerControllerBackendV1
+
+	operationMu sync.Mutex
+	stateMu     sync.Mutex
+	started     bool
+	startTried  bool
+	cleaned     bool
+	waitDone    chan struct{}
+	waitResult  dockerControllerWaitResultV1
+}
+
+// PrepareDockerControllerV1 verifies that the private channel is ready and
+// creates the exact controller container without starting it.
+func PrepareDockerControllerV1(
+	ctx context.Context,
+	plan ControlledSessionContainerPlanV1,
+) (*DockerControllerV1, error) {
+	return prepareDockerControllerV1(ctx, plan, dockerControllerBackendV1{
+		run:                 runDockerCommand,
+		observe:             observeDockerContainerExitV1,
+		requireReadyChannel: requirePreparedControlledSessionControllerChannelV1,
+	})
+}
+
+func prepareDockerControllerV1(
+	ctx context.Context,
+	plan ControlledSessionContainerPlanV1,
+	backend dockerControllerBackendV1,
+) (*DockerControllerV1, error) {
+	plan = cloneControlledSessionContainerPlanV1(plan)
+	if err := ValidateControlledSessionContainerPlanV1(plan); err != nil {
+		return nil, fmt.Errorf("prepare controlled-session controller: %w", err)
+	}
+	if plan.Role != ControlledSessionRoleControllerV1 {
+		return nil, fmt.Errorf("prepare controlled-session controller: container role must be %q", ControlledSessionRoleControllerV1)
+	}
+	if backend.run == nil || backend.observe == nil || backend.requireReadyChannel == nil {
+		return nil, fmt.Errorf("prepare controlled-session controller: backend is incomplete")
+	}
+	if err := backend.requireReadyChannel(plan); err != nil {
+		return nil, fmt.Errorf("prepare controlled-session controller channel: %w", err)
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	var createOutput bytes.Buffer
+	var createErrorOutput bytes.Buffer
+	if err := backend.run(controlledSessionCommandSpecV1(plan.Create), RunOptions{
+		Context: ctx, Stdout: &createOutput, Stderr: &createErrorOutput,
+	}); err != nil {
+		if output := trimmedCommandOutput(createErrorOutput.String()); output != "" {
+			err = fmt.Errorf("%w\ncommand output:\n%s", err, output)
+		}
+		createErr := fmt.Errorf("create controlled-session controller container %q: %w", plan.Container, err)
+		cleanupErr := rollbackAmbiguousControlledSessionControllerCreateV1(backend, plan)
+		if cleanupErr != nil {
+			cleanupErr = fmt.Errorf("reconcile controlled-session controller container %q after ambiguous create failure: %w", plan.Container, cleanupErr)
+		}
+		return nil, errors.Join(createErr, cleanupErr)
+	}
+	containerID, err := parseDockerControllerContainerIDV1(createOutput.String())
+	if err != nil {
+		return nil, fmt.Errorf("create controlled-session controller container %q: %w; refusing name-based cleanup because the created container identity is unknown", plan.Container, err)
+	}
+	return &DockerControllerV1{
+		plan: plan, containerID: containerID, backend: backend, waitDone: make(chan struct{}),
+	}, nil
+}
+
+// Start starts the inert controller and begins independent Docker exit
+// observation. A failed start is terminal because Docker may have started the
+// process before its response was lost; the exact container is force-removed.
+func (controller *DockerControllerV1) Start(ctx context.Context) error {
+	controller.operationMu.Lock()
+	defer controller.operationMu.Unlock()
+
+	controller.stateMu.Lock()
+	if controller.cleaned {
+		controller.stateMu.Unlock()
+		return fmt.Errorf("controlled-session controller container %q is already cleaned", controller.plan.Container)
+	}
+	if controller.startTried {
+		controller.stateMu.Unlock()
+		return fmt.Errorf("controlled-session controller container %q start was already attempted", controller.plan.Container)
+	}
+	controller.startTried = true
+	controller.stateMu.Unlock()
+
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	start := CommandSpec{Name: controller.plan.Start.Name, Args: []string{"start", controller.containerID}}
+	if err := controller.backend.run(start, RunOptions{Context: ctx}); err != nil {
+		startErr := fmt.Errorf("start controlled-session controller container %q: %w", controller.plan.Container, err)
+		cleanupErr := rollbackControlledSessionControllerContainerV1(controller.backend, controller.plan, controller.containerID)
+		if cleanupErr == nil {
+			controller.stateMu.Lock()
+			controller.cleaned = true
+			controller.stateMu.Unlock()
+		} else {
+			cleanupErr = fmt.Errorf("remove controlled-session controller container %q after ambiguous start failure: %w", controller.plan.Container, cleanupErr)
+		}
+		return errors.Join(startErr, cleanupErr)
+	}
+
+	controller.stateMu.Lock()
+	controller.started = true
+	controller.stateMu.Unlock()
+	go controller.observeExit()
+	return nil
+}
+
+// Wait returns the immutable Docker-observed controller exit status. Caller
+// cancellation stops only this wait; independent host observation continues.
+func (controller *DockerControllerV1) Wait(ctx context.Context) (controlledsession.ProcessStatusV1, error) {
+	controller.stateMu.Lock()
+	started := controller.started
+	controller.stateMu.Unlock()
+	if !started {
+		return controlledsession.ProcessStatusV1{}, fmt.Errorf("controlled-session controller container %q is not started", controller.plan.Container)
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	select {
+	case <-controller.waitDone:
+		return controller.waitResult.status, controller.waitResult.err
+	case <-ctx.Done():
+		return controlledsession.ProcessStatusV1{}, fmt.Errorf("wait for controlled-session controller container %q: %w", controller.plan.Container, ctx.Err())
+	}
+}
+
+// RequestGracefulStop delivers SIGTERM to the controller. The lifecycle
+// supervisor owns the grace deadline and follows with ForceStop when needed.
+func (controller *DockerControllerV1) RequestGracefulStop(ctx context.Context) error {
+	return controller.signal(ctx, "TERM", "request graceful stop for")
+}
+
+// ForceStop delivers SIGKILL to the controller.
+func (controller *DockerControllerV1) ForceStop(ctx context.Context) error {
+	return controller.signal(ctx, "KILL", "force stop")
+}
+
+func (controller *DockerControllerV1) signal(ctx context.Context, signal string, action string) error {
+	controller.operationMu.Lock()
+	defer controller.operationMu.Unlock()
+	if err := controller.requireStarted(); err != nil {
+		return err
+	}
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	command := CommandSpec{Name: controller.plan.Start.Name, Args: []string{
+		"kill", "--signal", signal, controller.containerID,
+	}}
+	if err := controller.backend.run(command, RunOptions{Context: ctx}); err != nil {
+		return fmt.Errorf("%s controlled-session controller container %q: %w", action, controller.plan.Container, err)
+	}
+	return nil
+}
+
+// Cleanup force-removes the exact created controller by its full container ID.
+// Successful cleanup is idempotent within this adapter; a failed attempt may
+// be retried.
+func (controller *DockerControllerV1) Cleanup(ctx context.Context) error {
+	controller.operationMu.Lock()
+	defer controller.operationMu.Unlock()
+
+	controller.stateMu.Lock()
+	if controller.cleaned {
+		controller.stateMu.Unlock()
+		return nil
+	}
+	controller.stateMu.Unlock()
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	cleanup := CommandSpec{Name: controller.plan.Cleanup.Name, Args: []string{"container", "rm", "--force", controller.containerID}}
+	if err := controller.backend.run(cleanup, RunOptions{Context: ctx}); err != nil {
+		return fmt.Errorf("remove controlled-session controller container %q: %w", controller.plan.Container, err)
+	}
+	controller.stateMu.Lock()
+	controller.cleaned = true
+	controller.stateMu.Unlock()
+	return nil
+}
+
+func (controller *DockerControllerV1) observeExit() {
+	code, err := controller.backend.observe(
+		context.Background(),
+		controlledSessionCommandSpecV1(controller.plan.Start),
+		controller.containerID,
+	)
+	if err != nil {
+		controller.waitResult = dockerControllerWaitResultV1{
+			status: controlledsession.ProcessStatusV1{
+				Kind: controlledsession.ProcessStatusUnavailableV1, Reason: "Docker controller observation was lost",
+			},
+			err: fmt.Errorf("observe controlled-session controller container %q exit: %w", controller.plan.Container, err),
+		}
+	} else {
+		exitCode := code
+		controller.waitResult = dockerControllerWaitResultV1{
+			status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &exitCode},
+		}
+	}
+	close(controller.waitDone)
+}
+
+func (controller *DockerControllerV1) requireStarted() error {
+	controller.stateMu.Lock()
+	defer controller.stateMu.Unlock()
+	if !controller.started {
+		return fmt.Errorf("controlled-session controller container %q is not started", controller.plan.Container)
+	}
+	if controller.cleaned {
+		return fmt.Errorf("controlled-session controller container %q is already cleaned", controller.plan.Container)
+	}
+	return nil
+}
+
+func rollbackControlledSessionControllerContainerV1(
+	backend dockerControllerBackendV1,
+	plan ControlledSessionContainerPlanV1,
+	containerID string,
+) error {
+	cleanupCtx, cancel := context.WithTimeout(context.Background(), defaultDockerPreflightTimeout)
+	defer cancel()
+	cleanup := CommandSpec{Name: plan.Cleanup.Name, Args: []string{"container", "rm", "--force", containerID}}
+	return backend.run(cleanup, RunOptions{Context: cleanupCtx})
+}
+
+func parseDockerControllerContainerIDV1(output string) (string, error) {
+	containerID := string(bytes.TrimSpace([]byte(output)))
+	if len(containerID) != 64 {
+		return "", fmt.Errorf("Docker create returned invalid full container ID %q", containerID)
+	}
+	if _, err := hex.DecodeString(containerID); err != nil {
+		return "", fmt.Errorf("Docker create returned invalid full container ID %q", containerID)
+	}
+	return containerID, nil
+}
+
+func rollbackAmbiguousControlledSessionControllerCreateV1(
+	backend dockerControllerBackendV1,
+	plan ControlledSessionContainerPlanV1,
+) error {
+	inspectCtx, cancel := context.WithTimeout(context.Background(), defaultDockerPreflightTimeout)
+	defer cancel()
+	var output bytes.Buffer
+	inspect := CommandSpec{Name: plan.Create.Name, Args: []string{
+		"container", "inspect", "--format", "{{json .Config.Labels}} {{json .State.Status}}", plan.Container,
+	}}
+	if err := backend.run(inspect, RunOptions{Context: inspectCtx, Stdout: &output, Stderr: &output}); err != nil {
+		return fmt.Errorf("inspect possible controller container before removal: %w", err)
+	}
+	decoder := json.NewDecoder(bytes.NewReader(output.Bytes()))
+	labels := map[string]string{}
+	if err := decoder.Decode(&labels); err != nil {
+		return fmt.Errorf("decode possible controller container labels: %w", err)
+	}
+	var status string
+	if err := decoder.Decode(&status); err != nil {
+		return fmt.Errorf("decode possible controller container state: %w", err)
+	}
+	for _, expected := range plan.Labels {
+		if labels[expected.Name] != expected.Value {
+			return fmt.Errorf("refuse to remove container because its controlled-session ownership labels do not match the immutable plan")
+		}
+	}
+	if status != "created" {
+		return fmt.Errorf("refuse to remove matching controlled-session container because its Docker state is %q, not %q", status, "created")
+	}
+	removeCtx, removeCancel := context.WithTimeout(context.Background(), defaultDockerPreflightTimeout)
+	defer removeCancel()
+	remove := CommandSpec{Name: plan.Cleanup.Name, Args: []string{"container", "rm", plan.Container}}
+	if err := backend.run(remove, RunOptions{Context: removeCtx}); err != nil {
+		return fmt.Errorf("remove inert matching controlled-session container: %w", err)
+	}
+	return nil
+}

internal/dockerdeploy/controlled_session_controller_channel_linux.go

@@ -0,0 +1,43 @@
+//go:build linux
+
+package dockerdeploy
+
+import (
+	"fmt"
+	"os"
+	"path/filepath"
+
+	"github.com/omry/reploy/internal/deploy"
+)
+
+func requirePreparedControlledSessionControllerChannelV1(plan ControlledSessionContainerPlanV1) error {
+	var source string
+	for _, mount := range plan.Mounts {
+		if mount.Name == "session-channel" {
+			if mount.Type != "bind" || mount.SourceKind != deploy.RuntimeMountSourceDirectory || mount.Target != controlledSessionChannelRootV1 || !mount.ReadOnly {
+				return fmt.Errorf("controller session-channel mount does not match the private read-only bind contract")
+			}
+			source = mount.Source
+			break
+		}
+	}
+	if source == "" {
+		return fmt.Errorf("controller plan does not contain the private session-channel mount")
+	}
+	directory, err := os.Lstat(source)
+	if err != nil {
+		return fmt.Errorf("inspect private channel directory %q: %w", source, err)
+	}
+	if !directory.IsDir() || directory.Mode()&os.ModeSymlink != 0 {
+		return fmt.Errorf("private channel source %q is not a real directory", source)
+	}
+	socket := filepath.Join(source, controlledSessionChannelSocketNameV1)
+	info, err := os.Lstat(socket)
+	if err != nil {
+		return fmt.Errorf("inspect private channel socket %q: %w", socket, err)
+	}
+	if info.Mode()&os.ModeSocket == 0 {
+		return fmt.Errorf("private channel path %q is not a Unix socket", socket)
+	}
+	return nil
+}