diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md
index 14703fb8..4fa1a2b9 100644
--- a/docs/CONTROLLED_SESSION_DESIGN.md
+++ b/docs/CONTROLLED_SESSION_DESIGN.md
@@ -1,6 +1,6 @@
 ---
 status: Active
-updated: 2026-08-09
+updated: 2026-08-10
 summary: Capability-scoped execution sessions that inherit Reploy's global container sandbox.
 ---
 
@@ -55,7 +55,13 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta
   completion and result acknowledgement, and removes both containers and the
   private channel. A workload that starts before a later startup step fails is
   still terminated and its output is finalized through the same barrier.
-  Crash watchdogs and restart reconciliation remain the next ownership phase;
+  Before creating any session resource, the planned controller, workload, and
+  private-channel ownership plus the session, lease, and boot identities are
+  now durably recorded in the existing live-run state. Reploy monotonically
+  fills each exact full container ID after Docker creates it, and both IDs are
+  durable before either process starts. Verified cleanup removes that record;
+  failed or unverifiable partial-preparation cleanup retains it. The watchdog
+  and restart reconciliation remain the next ownership phases, and
   controlled-session networking remains a later phase.
 - Initial runtime: Linux containers under Docker
 - Motivating clients: OmegaFlow recording, sandboxed AI agents, security
diff --git a/internal/deploy/live_run_queue.go b/internal/deploy/live_run_queue.go
index 14e7eb2b..aa790aa2 100644
--- a/internal/deploy/live_run_queue.go
+++ b/internal/deploy/live_run_queue.go
@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"path/filepath"
 	"regexp"
 
 	"github.com/omry/reploy/internal/canonical"
@@ -55,9 +56,28 @@ type LiveRunV1 struct {
 }
 
 type LiveRunQueueV1 struct {
-	Schema  string                      `json:"schema"`
-	Runs    []LiveRunV1                 `json:"runs"`
-	Cleanup []LiveRunContainerCleanupV1 `json:"cleanup,omitempty"`
+	Schema             string                         `json:"schema"`
+	Runs               []LiveRunV1                    `json:"runs"`
+	ControlledSessions []ControlledSessionOwnershipV1 `json:"controlled_sessions,omitempty"`
+	Cleanup            []LiveRunContainerCleanupV1    `json:"cleanup,omitempty"`
+}
+
+type ControlledSessionOwnershipV1 struct {
+	LiveRunID        string                                `json:"live_run_id"`
+	BootSession      string                                `json:"boot_session"`
+	SessionHandle    string                                `json:"session_handle"`
+	ChannelDirectory string                                `json:"channel_directory"`
+	Controller       ControlledSessionContainerOwnershipV1 `json:"controller"`
+	Workload         ControlledSessionContainerOwnershipV1 `json:"workload"`
+}
+
+type ControlledSessionContainerOwnershipV1 struct {
+	Role                string `json:"role"`
+	ID                  string `json:"id"`
+	Name                string `json:"name"`
+	DeploymentID        string `json:"deployment_id"`
+	GenerationReference string `json:"generation_reference"`
+	BuildIdentity       string `json:"build_identity"`
 }
 
 type LiveRunRecoveryReasonV1 string
@@ -98,6 +118,9 @@ var ErrLiveRunConflict = errors.New("another run must finish first")
 
 var liveRunIDPatternV1 = regexp.MustCompile(`^run-[0-9a-f]{16}$`)
 var controlMarkerIDPatternV1 = regexp.MustCompile(`^control-[0-9a-f]{16}$`)
+var controlledSessionHandlePatternV1 = regexp.MustCompile(`^session-[0-9a-f]{64}$`)
+var controlledSessionContainerIDPatternV1 = regexp.MustCompile(`^[0-9a-f]{64}$`)
+var controlledSessionBuildIdentityPatternV1 = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`)
 
 func NewLiveRunQueueV1() LiveRunQueueV1 {
 	return LiveRunQueueV1{Schema: LiveRunQueueSchemaV1, Runs: []LiveRunV1{}}
@@ -189,6 +212,14 @@ func ValidateLiveRunQueueV1(queue LiveRunQueueV1) error {
 	if queue.Runs == nil {
 		return fmt.Errorf("live run queue runs must use an array")
 	}
+	for index, ownership := range queue.ControlledSessions {
+		if err := validateControlledSessionOwnershipV1(ownership); err != nil {
+			return fmt.Errorf("live run queue controlled session %d: %w", index, err)
+		}
+		if index > 0 && queue.ControlledSessions[index-1].LiveRunID >= ownership.LiveRunID {
+			return fmt.Errorf("live run queue controlled sessions must be sorted and unique by live run ID")
+		}
+	}
 	for index, cleanup := range queue.Cleanup {
 		if err := validateLiveRunContainerCleanupV1(cleanup); err != nil {
 			return fmt.Errorf("live run queue cleanup entry %d: %w", index, err)
@@ -247,6 +278,65 @@ func ValidateLiveRunQueueV1(queue LiveRunQueueV1) error {
 	return nil
 }
 
+func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) error {
+	if err := ValidateLiveRunIDV1(ownership.LiveRunID); err != nil {
+		return fmt.Errorf("live run ID: %w", err)
+	}
+	if err := validateBootSessionIDV1(ownership.BootSession); err != nil {
+		return err
+	}
+	if !controlledSessionHandlePatternV1.MatchString(ownership.SessionHandle) {
+		return fmt.Errorf("session handle must use session- followed by 64 lowercase hexadecimal characters")
+	}
+	if !filepath.IsAbs(ownership.ChannelDirectory) || filepath.Clean(ownership.ChannelDirectory) != ownership.ChannelDirectory || !safeRecoveryIdentity(ownership.ChannelDirectory) {
+		return fmt.Errorf("channel directory must be a clean absolute path")
+	}
+	if err := validateControlledSessionContainerOwnershipStateV1(ownership.Controller, "controller"); err != nil {
+		return fmt.Errorf("controller: %w", err)
+	}
+	if err := validateControlledSessionContainerOwnershipStateV1(ownership.Workload, "workload"); err != nil {
+		return fmt.Errorf("workload: %w", err)
+	}
+	if ownership.Controller.ID == "" && ownership.Workload.ID != "" {
+		return fmt.Errorf("workload container ID cannot be recorded before the controller container ID")
+	}
+	if ownership.Controller.ID != "" && ownership.Workload.ID != "" && ownership.Controller.ID == ownership.Workload.ID {
+		return fmt.Errorf("controller and workload must name different containers")
+	}
+	return nil
+}
+
+func validateControlledSessionContainerOwnershipV1(ownership ControlledSessionContainerOwnershipV1, role string) error {
+	if err := validateControlledSessionContainerOwnershipStateV1(ownership, role); err != nil {
+		return err
+	}
+	if ownership.ID == "" {
+		return fmt.Errorf("container ID must use 64 lowercase hexadecimal characters")
+	}
+	return nil
+}
+
+func validateControlledSessionContainerOwnershipStateV1(ownership ControlledSessionContainerOwnershipV1, role string) error {
+	if ownership.Role != role {
+		return fmt.Errorf("role must be %q", role)
+	}
+	if ownership.ID != "" && !controlledSessionContainerIDPatternV1.MatchString(ownership.ID) {
+		return fmt.Errorf("container ID must use 64 lowercase hexadecimal characters")
+	}
+	for label, value := range map[string]string{
+		"name": ownership.Name, "deployment ID": ownership.DeploymentID,
+		"generation reference": ownership.GenerationReference,
+	} {
+		if !safeRecoveryIdentity(value) {
+			return fmt.Errorf("%s must be nonempty safe text", label)
+		}
+	}
+	if !controlledSessionBuildIdentityPatternV1.MatchString(ownership.BuildIdentity) {
+		return fmt.Errorf("build identity must be a sha256 digest")
+	}
+	return nil
+}
+
 func validateLiveRunContainerCleanupV1(cleanup LiveRunContainerCleanupV1) error {
 	if !safeRecoveryIdentity(cleanup.Container) {
 		return fmt.Errorf("cleanup container must be nonempty safe text")
@@ -506,14 +596,19 @@ func ControlMarkersV1(queue LiveRunQueueV1) []ControlMarkerV1 {
 }
 
 func cloneLiveRunQueueV1(queue LiveRunQueueV1) LiveRunQueueV1 {
+	var controlledSessions []ControlledSessionOwnershipV1
+	if queue.ControlledSessions != nil {
+		controlledSessions = append([]ControlledSessionOwnershipV1{}, queue.ControlledSessions...)
+	}
 	var cleanup []LiveRunContainerCleanupV1
 	if queue.Cleanup != nil {
 		cleanup = append([]LiveRunContainerCleanupV1{}, queue.Cleanup...)
 	}
 	return LiveRunQueueV1{
-		Schema:  queue.Schema,
-		Runs:    append([]LiveRunV1{}, queue.Runs...),
-		Cleanup: cleanup,
+		Schema:             queue.Schema,
+		Runs:               append([]LiveRunV1{}, queue.Runs...),
+		ControlledSessions: controlledSessions,
+		Cleanup:            cleanup,
 	}
 }
 
diff --git a/internal/deploy/live_run_queue_file.go b/internal/deploy/live_run_queue_file.go
index ee11810e..9981a325 100644
--- a/internal/deploy/live_run_queue_file.go
+++ b/internal/deploy/live_run_queue_file.go
@@ -128,6 +128,155 @@ func (lock *OperationLock) RecordLiveRunContainerV1(id string, container string)
 	return fmt.Errorf("live run %q is not outstanding", id)
 }
 
+// RecordControlledSessionOwnershipV1 durably binds the planned resources to an
+// active admitted shell and monotonically fills each exact container ID after
+// Docker returns it. The boot identity comes from the admitted run already
+// protected by this lock.
+func (lock *OperationLock) RecordControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) (ControlledSessionOwnershipV1, error) {
+	if lock == nil {
+		return ControlledSessionOwnershipV1{}, fmt.Errorf("record controlled session ownership requires an operation lock")
+	}
+	if err := ValidateLiveRunIDV1(ownership.LiveRunID); err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	lock.mutex.Lock()
+	defer lock.mutex.Unlock()
+	path, err := lock.liveRunQueuePathLockedV1()
+	if err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	queue, _, err := readLiveRunQueuePathV1(path)
+	if err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	var admitted *LiveRunV1
+	for index := range queue.Runs {
+		if queue.Runs[index].ID == ownership.LiveRunID {
+			admitted = &queue.Runs[index]
+			break
+		}
+	}
+	if admitted == nil {
+		return ControlledSessionOwnershipV1{}, fmt.Errorf("live run %q is not outstanding", ownership.LiveRunID)
+	}
+	if admitted.Status != LiveRunStatusActiveV1 || admitted.Kind != LiveRunKindShellV1 {
+		return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session live run %q must be an active shell", ownership.LiveRunID)
+	}
+	if admitted.Container != "" {
+		return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session live run %q already names container %q", ownership.LiveRunID, admitted.Container)
+	}
+	if admitted.GenerationReference != ownership.Workload.GenerationReference {
+		return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session workload generation does not match admitted live run %q", ownership.LiveRunID)
+	}
+	ownership.BootSession = admitted.BootSession
+	if err := validateControlledSessionOwnershipV1(ownership); err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	insert := sort.Search(len(queue.ControlledSessions), func(index int) bool {
+		return queue.ControlledSessions[index].LiveRunID >= ownership.LiveRunID
+	})
+	if insert < len(queue.ControlledSessions) && queue.ControlledSessions[insert].LiveRunID == ownership.LiveRunID {
+		merged, err := mergeControlledSessionOwnershipV1(queue.ControlledSessions[insert], ownership)
+		if err != nil {
+			return ControlledSessionOwnershipV1{}, fmt.Errorf("live run %q already has different controlled-session ownership: %w", ownership.LiveRunID, err)
+		}
+		if merged == queue.ControlledSessions[insert] {
+			return merged, nil
+		}
+		queue.ControlledSessions[insert] = merged
+		if err := commitLiveRunQueuePathV1(path, queue); err != nil {
+			return ControlledSessionOwnershipV1{}, err
+		}
+		return merged, nil
+	}
+	queue.ControlledSessions = append(queue.ControlledSessions, ControlledSessionOwnershipV1{})
+	copy(queue.ControlledSessions[insert+1:], queue.ControlledSessions[insert:])
+	queue.ControlledSessions[insert] = ownership
+	if err := commitLiveRunQueuePathV1(path, queue); err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	return ownership, nil
+}
+
+func mergeControlledSessionOwnershipV1(
+	existing ControlledSessionOwnershipV1,
+	requested ControlledSessionOwnershipV1,
+) (ControlledSessionOwnershipV1, error) {
+	existingPlan := existing
+	requestedPlan := requested
+	existingPlan.Controller.ID = ""
+	existingPlan.Workload.ID = ""
+	requestedPlan.Controller.ID = ""
+	requestedPlan.Workload.ID = ""
+	if existingPlan != requestedPlan {
+		return ControlledSessionOwnershipV1{}, fmt.Errorf("immutable resource plan changed")
+	}
+	merged := existing
+	mergeID := func(current string, next string, role string) (string, error) {
+		if next == "" {
+			return current, nil
+		}
+		if current != "" && current != next {
+			return "", fmt.Errorf("%s container ID changed", role)
+		}
+		return next, nil
+	}
+	var err error
+	merged.Controller.ID, err = mergeID(existing.Controller.ID, requested.Controller.ID, "controller")
+	if err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	merged.Workload.ID, err = mergeID(existing.Workload.ID, requested.Workload.ID, "workload")
+	if err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	if err := validateControlledSessionOwnershipV1(merged); err != nil {
+		return ControlledSessionOwnershipV1{}, err
+	}
+	return merged, nil
+}
+
+// CompleteControlledSessionV1 atomically removes a verified-clean session's
+// ownership record and admitted run. Failed cleanup must not call this method.
+func (lock *OperationLock) CompleteControlledSessionV1(id string) (bool, error) {
+	if lock == nil {
+		return false, fmt.Errorf("complete controlled session requires an operation lock")
+	}
+	if err := ValidateLiveRunIDV1(id); err != nil {
+		return false, err
+	}
+	lock.mutex.Lock()
+	defer lock.mutex.Unlock()
+	path, err := lock.liveRunQueuePathLockedV1()
+	if err != nil {
+		return false, err
+	}
+	queue, _, err := readLiveRunQueuePathV1(path)
+	if err != nil {
+		return false, err
+	}
+	updated, runRemoved, err := RemoveLiveRunV1(queue, id)
+	if err != nil {
+		return false, err
+	}
+	ownershipRemoved := false
+	for index, ownership := range updated.ControlledSessions {
+		if ownership.LiveRunID != id {
+			continue
+		}
+		updated.ControlledSessions = append(updated.ControlledSessions[:index], updated.ControlledSessions[index+1:]...)
+		ownershipRemoved = true
+		break
+	}
+	if !runRemoved && !ownershipRemoved {
+		return false, nil
+	}
+	if err := commitLiveRunQueuePathV1(path, updated); err != nil {
+		return false, err
+	}
+	return true, nil
+}
+
 func (lock *OperationLock) RemoveLiveRunV1(id string) (LiveRunQueueV1, bool, error) {
 	if lock == nil {
 		return LiveRunQueueV1{}, false, fmt.Errorf("remove live run requires an operation lock")
@@ -518,7 +667,7 @@ func commitLiveRunQueuePathV1(path string, queue LiveRunQueueV1) error {
 	if err != nil {
 		return err
 	}
-	if len(queue.Runs) == 0 && len(queue.Cleanup) == 0 {
+	if len(queue.Runs) == 0 && len(queue.ControlledSessions) == 0 && len(queue.Cleanup) == 0 {
 		return removeLiveRunQueuePathV1(path)
 	}
 	if err := writeAtomicStateFile(path, content, 0o600); err != nil {
diff --git a/internal/deploy/live_run_queue_file_test.go b/internal/deploy/live_run_queue_file_test.go
index 1b4a7590..e304796d 100644
--- a/internal/deploy/live_run_queue_file_test.go
+++ b/internal/deploy/live_run_queue_file_test.go
@@ -60,6 +60,155 @@ func TestOperationLockLiveRunQueueFileLifecycle(t *testing.T) {
 	}
 }
 
+func TestOperationLockRecordsExactControlledSessionOwnership(t *testing.T) {
+	dir := t.TempDir()
+	lock, err := AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lock.Unlock()
+	const runID = "run-0000000000000001"
+	const generation = "reploy/env/workload:g-current"
+	status, err := lock.AdmitLiveRunV1(LiveRunV1{
+		ID: runID, Kind: LiveRunKindShellV1, Name: "controlled-session",
+		GenerationReference: generation, Exclusive: true,
+	}, false)
+	if err != nil || status != LiveRunStatusActiveV1 {
+		t.Fatalf("admission = %q, %v", status, err)
+	}
+	ownership := controlledSessionOwnershipFixtureV1(dir, runID, generation)
+	planned := ownership
+	planned.Controller.ID = ""
+	planned.Workload.ID = ""
+	recorded, err := lock.RecordControlledSessionOwnershipV1(planned)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if recorded.BootSession == "" || planned.BootSession != "" || recorded.Controller.ID != "" || recorded.Workload.ID != "" {
+		t.Fatalf("boot identity = recorded %q, input %q", recorded.BootSession, ownership.BootSession)
+	}
+	if err := validateControlledSessionContainerOwnershipV1(recorded.Controller, "controller"); err == nil || !strings.Contains(err.Error(), "container ID") {
+		t.Fatalf("complete container validation accepted planned ownership: %v", err)
+	}
+	workloadFirst := planned
+	workloadFirst.Workload.ID = ownership.Workload.ID
+	if _, err := lock.RecordControlledSessionOwnershipV1(workloadFirst); err == nil || !strings.Contains(err.Error(), "before the controller") {
+		t.Fatalf("workload-first ownership error = %v", err)
+	}
+	controllerPrepared := ownership
+	controllerPrepared.Workload.ID = ""
+	recorded, err = lock.RecordControlledSessionOwnershipV1(controllerPrepared)
+	if err != nil || recorded.Controller.ID != ownership.Controller.ID || recorded.Workload.ID != "" {
+		t.Fatalf("controller ownership = %#v, error=%v", recorded, err)
+	}
+	recorded, err = lock.RecordControlledSessionOwnershipV1(ownership)
+	if err != nil || recorded.Controller.ID != ownership.Controller.ID || recorded.Workload.ID != ownership.Workload.ID {
+		t.Fatalf("complete ownership = %#v, error=%v", recorded, err)
+	}
+	loaded, found, err := lock.ReadLiveRunQueueV1()
+	if err != nil || !found || len(loaded.ControlledSessions) != 1 || loaded.ControlledSessions[0] != recorded {
+		t.Fatalf("controlled-session ownership = %#v, found=%t, error=%v", loaded.ControlledSessions, found, err)
+	}
+	conflict := ownership
+	conflict.Controller.ID = strings.Repeat("c", 64)
+	if _, err := lock.RecordControlledSessionOwnershipV1(conflict); err == nil || !strings.Contains(err.Error(), "different controlled-session ownership") {
+		t.Fatalf("conflicting ownership error = %v", err)
+	}
+	if completed, err := lock.CompleteControlledSessionV1(runID); err != nil || !completed {
+		t.Fatalf("completion = %t, %v", completed, err)
+	}
+	if _, found, err := lock.ReadLiveRunQueueV1(); err != nil || found {
+		t.Fatalf("completed queue found=%t, error=%v", found, err)
+	}
+}
+
+func TestOperationLockControlledSessionOwnershipWriteFailurePreservesQueue(t *testing.T) {
+	dir := t.TempDir()
+	lock, err := AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lock.Unlock()
+	const runID = "run-0000000000000001"
+	const generation = "reploy/env/workload:g-current"
+	if status, err := lock.AdmitLiveRunV1(LiveRunV1{
+		ID: runID, Kind: LiveRunKindShellV1, Name: "controlled-session",
+		GenerationReference: generation, Exclusive: true,
+	}, false); err != nil || status != LiveRunStatusActiveV1 {
+		t.Fatalf("admission = %q, %v", status, err)
+	}
+	planned := controlledSessionOwnershipFixtureV1(dir, runID, generation)
+	planned.Controller.ID = ""
+	planned.Workload.ID = ""
+	if _, err := lock.RecordControlledSessionOwnershipV1(planned); err != nil {
+		t.Fatal(err)
+	}
+	before, _, err := lock.ReadLiveRunQueueV1()
+	if err != nil {
+		t.Fatal(err)
+	}
+	originalReplace := replaceAtomicStateFile
+	replaceAtomicStateFile = func(string, string) error { return errors.New("injected ownership replace failure") }
+	t.Cleanup(func() { replaceAtomicStateFile = originalReplace })
+	controllerPrepared := controlledSessionOwnershipFixtureV1(dir, runID, generation)
+	controllerPrepared.Workload.ID = ""
+	if _, err := lock.RecordControlledSessionOwnershipV1(controllerPrepared); err == nil || !strings.Contains(err.Error(), "injected ownership replace failure") {
+		t.Fatalf("ownership write error = %v", err)
+	}
+	after, _, err := lock.ReadLiveRunQueueV1()
+	if err != nil || !reflect.DeepEqual(after, before) {
+		t.Fatalf("failed ownership write changed queue: %#v, error=%v", after, err)
+	}
+}
+
+func TestRecoverLiveRunQueuePreservesControlledSessionOwnership(t *testing.T) {
+	dir := t.TempDir()
+	lock, err := AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer lock.Unlock()
+	const runID = "run-0000000000000001"
+	const generation = "reploy/env/workload:g-current"
+	if status, err := lock.AdmitLiveRunV1(LiveRunV1{
+		ID: runID, Kind: LiveRunKindShellV1, Name: "controlled-session",
+		GenerationReference: generation, Exclusive: true,
+	}, false); err != nil || status != LiveRunStatusActiveV1 {
+		t.Fatalf("admission = %q, %v", status, err)
+	}
+	recorded, err := lock.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFixtureV1(dir, runID, generation))
+	if err != nil {
+		t.Fatal(err)
+	}
+	recovery, err := lock.RecoverLiveRunQueueV1()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(recovery.Removed) != 1 || recovery.Removed[0].Run.ID != runID {
+		t.Fatalf("recovery = %#v", recovery)
+	}
+	queue, found, err := lock.ReadLiveRunQueueV1()
+	if err != nil || !found || len(queue.Runs) != 0 || len(queue.ControlledSessions) != 1 || queue.ControlledSessions[0] != recorded {
+		t.Fatalf("ownership after run recovery = %#v, found=%t, error=%v", queue, found, err)
+	}
+}
+
+func controlledSessionOwnershipFixtureV1(dir string, runID string, generation string) ControlledSessionOwnershipV1 {
+	container := func(role string, id string, environment string, generation string, build string) ControlledSessionContainerOwnershipV1 {
+		return ControlledSessionContainerOwnershipV1{
+			Role: role, ID: id, Name: "reploy-" + role + "-" + runID,
+			DeploymentID: environment, GenerationReference: generation,
+			BuildIdentity: "sha256:" + strings.Repeat(build, 64),
+		}
+	}
+	return ControlledSessionOwnershipV1{
+		LiveRunID: runID, SessionHandle: "session-" + strings.Repeat("a", 64),
+		ChannelDirectory: filepath.Join(dir, ".reploy", "private", "sessions", runID),
+		Controller:       container("controller", strings.Repeat("a", 64), "controller", "reploy/env/controller:g-current", "1"),
+		Workload:         container("workload", strings.Repeat("b", 64), "workload", generation, "2"),
+	}
+}
+
 func TestOperationLockLiveRunQueueReplaceFailurePreservesQueue(t *testing.T) {
 	dir := t.TempDir()
 	lock, err := AcquireOperationLock(t.Context(), dir)
diff --git a/internal/dockerdeploy/control_admission_modes.go b/internal/dockerdeploy/control_admission_modes.go
index 42473e9a..2461eed5 100644
--- a/internal/dockerdeploy/control_admission_modes.go
+++ b/internal/dockerdeploy/control_admission_modes.go
@@ -251,13 +251,13 @@ func stopActiveLiveRunsForControlV1(
 	}
 	stopped := []deploy.LiveRunV1{}
 	for _, run := range active {
-		if run.Container != "" {
+		for _, container := range liveRunContainerTargetsV1(queue, run) {
 			err := removeContainer(
-				TemporaryContainerStopCommand(run.Container),
+				TemporaryContainerStopCommand(container),
 				RunOptions{Context: ctx, DockerPreflightTimeout: dockerPreflightTimeout},
 			)
 			if err != nil && !isMissingContainerCleanupError(err) {
-				return stopped, fmt.Errorf("stop live run container %q: %w", run.Container, err)
+				return stopped, fmt.Errorf("stop live run container %q: %w", container, err)
 			}
 		}
 		_, removed, err := operation.RemoveLiveRunV1(run.ID)
diff --git a/internal/dockerdeploy/control_admission_modes_test.go b/internal/dockerdeploy/control_admission_modes_test.go
index f7daa896..bdc54ed7 100644
--- a/internal/dockerdeploy/control_admission_modes_test.go
+++ b/internal/dockerdeploy/control_admission_modes_test.go
@@ -147,6 +147,123 @@ func TestAdmitControlOperationV1ForceStopsActiveContainersBeforeMarker(t *testin
 	}
 }
 
+func TestAdmitControlOperationV1ForceStopsControlledSessionContainersAndRetainsOwnership(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	dir := plan.Workload.DeploymentDirectory
+	operation, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	run := liveRunAdmissionFixtureV1(plan.LiveRunID, false)
+	run.Kind = deploy.LiveRunKindShellV1
+	run.GenerationReference = plan.Workload.GenerationReference
+	holdLiveRunLeaseV1(t, operation, run.ID)
+	if _, err := operation.AdmitLiveRunV1(run, false); err != nil {
+		t.Fatal(err)
+	}
+	ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1(
+		plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1,
+	))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	calls := []CommandSpec{}
+	result, err := admitControlOperationV1(t.Context(), dir, operation, ControlAdmissionInputV1{
+		Operation: deploy.ControlOperationStopV1, GenerationReference: run.GenerationReference,
+		Mode: ControlAdmissionForceV1,
+	}, controlOperationAdmissionBackendV1{
+		newID: func() (string, error) { return "control-0000000000000001", nil },
+		pause: func(context.Context, time.Duration) error { return nil },
+		await: AwaitControlAdmissionWithNoticeV1,
+		removeContainer: func(spec CommandSpec, _ RunOptions) error {
+			calls = append(calls, spec)
+			return nil
+		},
+	})
+	if err != nil || len(result.StoppedRuns) != 1 || result.StoppedRuns[0].ID != run.ID {
+		t.Fatalf("controlled-session force result = %#v, %v", result, err)
+	}
+	wantCalls := []CommandSpec{
+		TemporaryContainerStopCommand(dockerWorkloadTestContainerIDV1),
+		TemporaryContainerStopCommand(dockerControllerTestContainerIDV1),
+	}
+	if !reflect.DeepEqual(calls, wantCalls) {
+		t.Fatalf("controlled-session force calls = %#v", calls)
+	}
+	queue, _, err := result.Operation.ReadLiveRunQueueV1()
+	if err != nil || len(queue.ControlledSessions) != 1 || queue.ControlledSessions[0] != ownership {
+		t.Fatalf("retained controlled-session ownership = %#v, error=%v", queue.ControlledSessions, err)
+	}
+	if len(queue.Runs) != 1 || queue.Runs[0].Kind != deploy.LiveRunKindControlV1 {
+		t.Fatalf("controlled-session force queue = %#v", queue)
+	}
+	if err := CompleteControlAdmissionV1(result.Operation, result.Marker.ID, result.Lease); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestAdmitControlOperationV1ForcePreservesControlledSessionOnPartialStopFailure(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	dir := plan.Workload.DeploymentDirectory
+	operation, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	run := liveRunAdmissionFixtureV1(plan.LiveRunID, false)
+	run.Kind = deploy.LiveRunKindShellV1
+	run.GenerationReference = plan.Workload.GenerationReference
+	holdLiveRunLeaseV1(t, operation, run.ID)
+	if _, err := operation.AdmitLiveRunV1(run, false); err != nil {
+		t.Fatal(err)
+	}
+	ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1(
+		plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1,
+	))
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	want := errors.New("controller stop failed")
+	calls := []CommandSpec{}
+	result, err := admitControlOperationV1(t.Context(), dir, operation, ControlAdmissionInputV1{
+		Operation: deploy.ControlOperationStopV1, GenerationReference: run.GenerationReference,
+		Mode: ControlAdmissionForceV1,
+	}, controlOperationAdmissionBackendV1{
+		newID: func() (string, error) { return "control-0000000000000001", nil },
+		pause: func(context.Context, time.Duration) error { return nil },
+		await: AwaitControlAdmissionWithNoticeV1,
+		removeContainer: func(spec CommandSpec, _ RunOptions) error {
+			calls = append(calls, spec)
+			if len(calls) == 2 {
+				return want
+			}
+			return nil
+		},
+	})
+	if !errors.Is(err, want) || len(result.StoppedRuns) != 0 {
+		t.Fatalf("partial controlled-session stop result = %#v, %v", result, err)
+	}
+	wantCalls := []CommandSpec{
+		TemporaryContainerStopCommand(dockerWorkloadTestContainerIDV1),
+		TemporaryContainerStopCommand(dockerControllerTestContainerIDV1),
+	}
+	if !reflect.DeepEqual(calls, wantCalls) {
+		t.Fatalf("partial controlled-session stop calls = %#v", calls)
+	}
+	check, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer check.Unlock()
+	queue, _, err := check.ReadLiveRunQueueV1()
+	if err != nil || len(queue.Runs) != 1 || queue.Runs[0].ID != run.ID ||
+		len(queue.ControlledSessions) != 1 || queue.ControlledSessions[0] != ownership ||
+		len(deploy.ControlMarkersV1(queue)) != 0 {
+		t.Fatalf("queue after partial controlled-session stop = %#v, error=%v", queue, err)
+	}
+}
+
 func TestAdmitControlOperationV1ForceFailurePreservesFailedAndLaterActiveRuns(t *testing.T) {
 	dir := t.TempDir()
 	operation, err := deploy.AcquireOperationLock(t.Context(), dir)
diff --git a/internal/dockerdeploy/controlled_session_controller.go b/internal/dockerdeploy/controlled_session_controller.go
index 8e16f1f2..f5c9c07a 100644
--- a/internal/dockerdeploy/controlled_session_controller.go
+++ b/internal/dockerdeploy/controlled_session_controller.go
@@ -17,6 +17,7 @@ type dockerControllerBackendV1 struct {
 	run                 commandRunner
 	observe             func(context.Context, CommandSpec, string) (int, error)
 	requireReadyChannel func(ControlledSessionContainerPlanV1) error
+	recordNoContainer   func()
 }
 
 type dockerControllerWaitResultV1 struct {
@@ -42,16 +43,29 @@ type DockerControllerV1 struct {
 	waitResult  dockerControllerWaitResultV1
 }
 
+func (controller *DockerControllerV1) ContainerID() string {
+	return controller.containerID
+}
+
 // 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 prepareDockerControllerWithCleanupVerificationV1(ctx, plan, nil)
+}
+
+func prepareDockerControllerWithCleanupVerificationV1(
+	ctx context.Context,
+	plan ControlledSessionContainerPlanV1,
+	recordNoContainer func(),
 ) (*DockerControllerV1, error) {
 	return prepareDockerControllerV1(ctx, plan, dockerControllerBackendV1{
 		bind:                bindPinnedDockerCommandRunnerV1,
 		observe:             observeDockerContainerExitV1,
 		requireReadyChannel: requirePreparedControlledSessionControllerChannelV1,
+		recordNoContainer:   recordNoContainer,
 	})
 }
 
@@ -60,18 +74,24 @@ func prepareDockerControllerV1(
 	plan ControlledSessionContainerPlanV1,
 	backend dockerControllerBackendV1,
 ) (*DockerControllerV1, error) {
+	failBeforeCreate := func(err error) (*DockerControllerV1, error) {
+		if backend.recordNoContainer != nil {
+			backend.recordNoContainer()
+		}
+		return nil, err
+	}
 	plan = cloneControlledSessionContainerPlanV1(plan)
 	if err := ValidateControlledSessionContainerPlanV1(plan); err != nil {
-		return nil, fmt.Errorf("prepare controlled-session controller: %w", err)
+		return failBeforeCreate(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)
+		return failBeforeCreate(fmt.Errorf("prepare controlled-session controller: container role must be %q", ControlledSessionRoleControllerV1))
 	}
 	if (backend.bind == nil && backend.run == nil) || backend.observe == nil || backend.requireReadyChannel == nil {
-		return nil, fmt.Errorf("prepare controlled-session controller: backend is incomplete")
+		return failBeforeCreate(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)
+		return failBeforeCreate(fmt.Errorf("prepare controlled-session controller channel: %w", err))
 	}
 	if ctx == nil {
 		ctx = context.Background()
@@ -81,10 +101,10 @@ func prepareDockerControllerV1(
 		var err error
 		docker, backend.run, err = backend.bind(ctx, docker, defaultDockerPreflightTimeout)
 		if err != nil {
-			return nil, fmt.Errorf("bind controlled-session controller Docker endpoint: %w", err)
+			return failBeforeCreate(fmt.Errorf("bind controlled-session controller Docker endpoint: %w", err))
 		}
 		if backend.run == nil {
-			return nil, fmt.Errorf("prepare controlled-session controller: Docker endpoint binder returned no command runner")
+			return failBeforeCreate(fmt.Errorf("prepare controlled-session controller: Docker endpoint binder returned no command runner"))
 		}
 	}
 	var createOutput bytes.Buffer
@@ -215,7 +235,7 @@ func (controller *DockerControllerV1) Cleanup(ctx context.Context) error {
 		ctx = context.Background()
 	}
 	cleanup := controller.commandV1("container", "rm", "--force", controller.containerID)
-	if err := controller.backend.run(cleanup, RunOptions{Context: ctx}); err != nil {
+	if err := controller.backend.run(cleanup, RunOptions{Context: ctx}); err != nil && !isMissingContainerCleanupError(err) {
 		return fmt.Errorf("remove controlled-session controller container %q: %w", controller.plan.Container, err)
 	}
 	controller.stateMu.Lock()
diff --git a/internal/dockerdeploy/controlled_session_controller_test.go b/internal/dockerdeploy/controlled_session_controller_test.go
index 33b6d896..2e1d661b 100644
--- a/internal/dockerdeploy/controlled_session_controller_test.go
+++ b/internal/dockerdeploy/controlled_session_controller_test.go
@@ -105,6 +105,7 @@ func TestDockerControllerV1OrdersChannelCreateStartAndExactLifecycle(t *testing.
 func TestPrepareDockerControllerV1RequiresReadyChannelBeforeCreate(t *testing.T) {
 	plan := controlledSessionControllerPlanFixtureV1(t)
 	run := false
+	verifiedNoContainer := false
 	_, err := prepareDockerControllerV1(t.Context(), plan, dockerControllerBackendV1{
 		requireReadyChannel: func(ControlledSessionContainerPlanV1) error {
 			return errors.New("channel socket missing")
@@ -113,10 +114,28 @@ func TestPrepareDockerControllerV1RequiresReadyChannelBeforeCreate(t *testing.T)
 			run = true
 			return nil
 		},
-		observe: func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+		observe:           func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+		recordNoContainer: func() { verifiedNoContainer = true },
+	})
+	if err == nil || !strings.Contains(err.Error(), "channel socket missing") || run || !verifiedNoContainer {
+		t.Fatalf("prepare error = %v, Docker invoked = %t, no container verified = %t", err, run, verifiedNoContainer)
+	}
+}
+
+func TestPrepareDockerControllerV1ReportsVerifiedPreCreateBindFailure(t *testing.T) {
+	plan := controlledSessionControllerPlanFixtureV1(t)
+	bindErr := errors.New("injected Docker endpoint bind failure")
+	verifiedNoContainer := false
+	_, err := prepareDockerControllerV1(t.Context(), plan, dockerControllerBackendV1{
+		requireReadyChannel: func(ControlledSessionContainerPlanV1) error { return nil },
+		bind: func(context.Context, CommandSpec, time.Duration) (CommandSpec, commandRunner, error) {
+			return CommandSpec{}, nil, bindErr
+		},
+		observe:           func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+		recordNoContainer: func() { verifiedNoContainer = true },
 	})
-	if err == nil || !strings.Contains(err.Error(), "channel socket missing") || run {
-		t.Fatalf("prepare error = %v, Docker invoked = %t", err, run)
+	if !errors.Is(err, bindErr) || !verifiedNoContainer {
+		t.Fatalf("prepare error = %v, no container verified = %t", err, verifiedNoContainer)
 	}
 }
 
@@ -138,6 +157,7 @@ func TestPrepareDockerControllerV1RejectsWorkloadAndIncompleteBackend(t *testing
 func TestPrepareDockerControllerV1DoesNotRemoveAfterAmbiguousCreateFailure(t *testing.T) {
 	plan := controlledSessionControllerPlanFixtureV1(t)
 	runs := []CommandSpec{}
+	verifiedNoContainer := false
 	_, err := prepareDockerControllerV1(t.Context(), plan, dockerControllerBackendV1{
 		requireReadyChannel: func(ControlledSessionContainerPlanV1) error { return nil },
 		run: func(spec CommandSpec, options RunOptions) error {
@@ -148,7 +168,8 @@ func TestPrepareDockerControllerV1DoesNotRemoveAfterAmbiguousCreateFailure(t *te
 			}
 			return nil
 		},
-		observe: func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+		observe:           func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+		recordNoContainer: func() { verifiedNoContainer = true },
 	})
 	if err == nil || !strings.Contains(err.Error(), "create response was lost") ||
 		!strings.Contains(err.Error(), "daemon response was lost") ||
@@ -158,6 +179,9 @@ func TestPrepareDockerControllerV1DoesNotRemoveAfterAmbiguousCreateFailure(t *te
 	if len(runs) != 1 || !reflect.DeepEqual(runs[0].Args, plan.Create.Args) {
 		t.Fatalf("ambiguous create failure invoked reconciliation: %#v", runs)
 	}
+	if verifiedNoContainer {
+		t.Fatal("ambiguous create failure reported that no container exists")
+	}
 }
 
 func TestDockerControllerV1PinsOneDockerEndpointForItsLifetime(t *testing.T) {
@@ -344,6 +368,36 @@ func TestDockerControllerV1RetriesFailedCleanup(t *testing.T) {
 	}
 }
 
+func TestDockerControllerV1TreatsMissingContainerAsCleaned(t *testing.T) {
+	plan := controlledSessionControllerPlanFixtureV1(t)
+	cleanupAttempts := 0
+	backend := dockerControllerBackendV1{
+		requireReadyChannel: func(ControlledSessionContainerPlanV1) error { return nil },
+		run: func(spec CommandSpec, options RunOptions) error {
+			writeDockerControllerTestCreateIDV1(plan, spec, options)
+			if reflect.DeepEqual(spec.Args, []string{"container", "rm", "--force", dockerControllerTestContainerIDV1}) {
+				cleanupAttempts++
+				return errors.New("Error response from daemon: No such container: " + dockerControllerTestContainerIDV1)
+			}
+			return nil
+		},
+		observe: func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+	}
+	controller, err := prepareDockerControllerV1(t.Context(), plan, backend)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := controller.Cleanup(t.Context()); err != nil {
+		t.Fatalf("missing-container cleanup = %v", err)
+	}
+	if err := controller.Cleanup(t.Context()); err != nil {
+		t.Fatalf("repeated cleanup = %v", err)
+	}
+	if cleanupAttempts != 1 {
+		t.Fatalf("cleanup attempts = %d, want 1", cleanupAttempts)
+	}
+}
+
 func TestDockerControllerV1FreezesCallerOwnedPlanSlices(t *testing.T) {
 	plan := controlledSessionControllerPlanFixtureV1(t)
 	wantStart := []string{"start", dockerControllerTestContainerIDV1}
diff --git a/internal/dockerdeploy/controlled_session_supervisor.go b/internal/dockerdeploy/controlled_session_supervisor.go
index 1daa4c73..83f4205f 100644
--- a/internal/dockerdeploy/controlled_session_supervisor.go
+++ b/internal/dockerdeploy/controlled_session_supervisor.go
@@ -5,10 +5,13 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"os"
+	"path/filepath"
 	"sync"
 	"time"
 
 	"github.com/omry/reploy/internal/controlledsession"
+	"github.com/omry/reploy/internal/deploy"
 )
 
 const controlledSessionOutputFinalizationTimeoutV1 = time.Duration(controlledsession.DefaultOutputFinalizationTimeoutMillisecondsV1) * time.Millisecond
@@ -34,6 +37,7 @@ type ControlledSessionRunResultV1 struct {
 }
 
 type controlledSessionControllerRuntimeV1 interface {
+	ContainerID() string
 	Start(context.Context) error
 	Wait(context.Context) (controlledsession.ProcessStatusV1, error)
 	RequestGracefulStop(context.Context) error
@@ -43,6 +47,7 @@ type controlledSessionControllerRuntimeV1 interface {
 
 type controlledSessionWorkloadRuntimeV1 interface {
 	controlledsession.WorkloadPTYControlV1
+	ContainerID() string
 	Output() (io.ReadCloser, error)
 	Start(context.Context) error
 	Started() bool
@@ -71,10 +76,14 @@ func (runtime *privateControlledSessionChannelRuntimeV1) Close() error {
 }
 
 type controlledSessionSupervisorBackendV1 struct {
-	prepareChannel    func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error)
-	prepareController func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error)
-	prepareWorkload   func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error)
-	now               func() time.Time
+	prepareChannel            func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error)
+	prepareController         func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error)
+	prepareWorkload           func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error)
+	recordPlannedOwnership    func() error
+	recordControllerOwnership func(string) error
+	recordControllerRollback  func()
+	recordOwnership           func(string, string) error
+	now                       func() time.Time
 }
 
 type controlledSessionProcessResultV1 struct {
@@ -108,43 +117,182 @@ type controlledSessionSupervisorV1 struct {
 	terminationMu                sync.Mutex
 	terminationAt                time.Time
 
-	workloadResult     <-chan controlledSessionProcessResultV1
-	controllerResult   <-chan controlledSessionProcessResultV1
-	workloadObserved   *controlledSessionProcessResultV1
-	controllerObserved *controlledSessionProcessResultV1
-	workloadRecorded   bool
-	controllerStarted  bool
-	workloadStarted    bool
+	workloadResult                <-chan controlledSessionProcessResultV1
+	controllerResult              <-chan controlledSessionProcessResultV1
+	workloadObserved              *controlledSessionProcessResultV1
+	controllerObserved            *controlledSessionProcessResultV1
+	workloadRecorded              bool
+	controllerStarted             bool
+	workloadStarted               bool
+	controllerOwnershipIncomplete bool
 
 	transportHealthy bool
 	diagnosticErr    error
 }
 
-// RunControlledSessionV1 owns one attached controller/workload operation from
-// inert resource creation through terminal acknowledgement and ordinary
-// delivery-tail cleanup. Crash reconciliation, watchdog ownership, networking,
-// and public command exposure are deliberately outside this lifecycle core.
+// RunControlledSessionV1 takes ownership of the admitted workload operation
+// lock. The caller must retain the live-run queue-entry lease until this call
+// returns. The supervisor durably records both exact inert containers and the
+// private channel before releasing the lock and starting either process.
 func RunControlledSessionV1(
 	ctx context.Context,
+	operation *deploy.OperationLock,
 	plan ControlledSessionExecutionPlanV1,
 	options ControlledSessionRunOptionsV1,
 ) (ControlledSessionRunResultV1, error) {
-	return runControlledSessionV1(ctx, plan, options, controlledSessionSupervisorBackendV1{
+	if operation == nil {
+		return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session requires an admitted operation lock")
+	}
+	if err := operation.RequireHeld(); err != nil {
+		return ControlledSessionRunResultV1{}, err
+	}
+	absoluteDir, err := filepath.Abs(plan.Workload.DeploymentDirectory)
+	if err != nil {
+		return ControlledSessionRunResultV1{}, releaseControlledSessionOperationV1(operation, fmt.Errorf("resolve controlled-session workload deployment directory: %w", err))
+	}
+	if filepath.Dir(filepath.Dir(operation.Path())) != absoluteDir {
+		return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("controlled-session operation lock does not belong to workload deployment %q", absoluteDir))
+	}
+	if ctx == nil || ctx.Done() == nil {
+		return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("run controlled session: cancelable host context is required"))
+	}
+	if err := ValidateControlledSessionExecutionPlanV1(plan); err != nil {
+		return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("run controlled session plan: %w", err))
+	}
+	if err := validateControlledSessionRunOptionsV1(options); err != nil {
+		return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, err)
+	}
+	if err := operation.RequireQueueEntryLeaseHeldV1(plan.LiveRunID); err != nil {
+		return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("controlled-session admission ownership: %w", err))
+	}
+	operationReleaseAttempted := false
+	ownershipRecorded := false
+	partialPreparationCleanupVerified := false
+	controllerID := ""
+	persistOwnership := func(controllerID string, workloadID string) error {
+		ownership := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID)
+		if _, err := operation.RecordControlledSessionOwnershipV1(ownership); err != nil {
+			return fmt.Errorf("persist controlled-session ownership: %w", err)
+		}
+		ownershipRecorded = true
+		return nil
+	}
+	result, runErr := runControlledSessionV1(ctx, plan, options, controlledSessionSupervisorBackendV1{
 		prepareChannel: func(plan ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) {
 			channel, err := PrepareControlledSessionChannelV1(plan)
 			if err != nil {
+				partialPreparationCleanupVerified = controlledSessionChannelAbsentV1(plan.Channel.HostDirectory)
 				return nil, err
 			}
 			return &privateControlledSessionChannelRuntimeV1{channel: channel}, nil
 		},
 		prepareController: func(ctx context.Context, plan ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) {
-			return PrepareDockerControllerV1(ctx, plan)
+			return prepareDockerControllerWithCleanupVerificationV1(ctx, plan, func() {
+				partialPreparationCleanupVerified = true
+			})
 		},
 		prepareWorkload: func(ctx context.Context, plan ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) {
-			return PrepareDockerWorkloadPTYV1(ctx, plan)
+			return prepareDockerWorkloadPTYWithContainerIDV1(ctx, plan, func(workloadID string) error {
+				return persistOwnership(controllerID, workloadID)
+			}, func() {
+				partialPreparationCleanupVerified = true
+			})
+		},
+		recordPlannedOwnership: func() error {
+			return persistOwnership("", "")
+		},
+		recordControllerOwnership: func(exactControllerID string) error {
+			controllerID = exactControllerID
+			err := persistOwnership(exactControllerID, "")
+			return err
+		},
+		recordControllerRollback: func() {
+			partialPreparationCleanupVerified = true
+		},
+		recordOwnership: func(controllerID string, workloadID string) error {
+			if err := persistOwnership(controllerID, workloadID); err != nil {
+				return err
+			}
+			operationReleaseAttempted = true
+			if err := operation.Unlock(); err != nil {
+				return fmt.Errorf("release operation lock before controlled-session startup: %w", err)
+			}
+			return nil
 		},
 		now: time.Now,
 	})
+	cleaned := result.SessionResult.CleanupStatus.Kind == controlledsession.CleanupStatusSucceededV1 &&
+		result.DeliveryTailCleanupStatus.Kind == controlledsession.CleanupStatusSucceededV1
+	cleaned = controlledSessionPreparationCanCompleteV1(cleaned, ownershipRecorded, operationReleaseAttempted, partialPreparationCleanupVerified)
+	completionErr := finishControlledSessionOwnershipV1(context.WithoutCancel(ctx), absoluteDir, operation, operationReleaseAttempted, plan.LiveRunID, cleaned)
+	return result, errors.Join(runErr, completionErr)
+}
+
+func controlledSessionChannelAbsentV1(path string) bool {
+	_, err := os.Lstat(path)
+	return errors.Is(err, os.ErrNotExist)
+}
+
+func controlledSessionPreparationCanCompleteV1(cleaned bool, ownershipRecorded bool, operationReleaseAttempted bool, partialCleanupVerified bool) bool {
+	return cleaned && (!ownershipRecorded || operationReleaseAttempted || partialCleanupVerified)
+}
+
+func controlledSessionOwnershipFromPlanV1(plan ControlledSessionExecutionPlanV1, controllerID string, workloadID string) deploy.ControlledSessionOwnershipV1 {
+	container := func(plan ControlledSessionContainerPlanV1, id string) deploy.ControlledSessionContainerOwnershipV1 {
+		return deploy.ControlledSessionContainerOwnershipV1{
+			Role: string(plan.Role), ID: id, Name: plan.Container, DeploymentID: plan.DeploymentID,
+			GenerationReference: plan.GenerationReference, BuildIdentity: string(plan.BuildIdentity),
+		}
+	}
+	return deploy.ControlledSessionOwnershipV1{
+		LiveRunID: plan.LiveRunID, SessionHandle: plan.Authorization.Handle,
+		ChannelDirectory: plan.Channel.HostDirectory,
+		Controller:       container(plan.Controller, controllerID), Workload: container(plan.Workload, workloadID),
+	}
+}
+
+func finishControlledSessionOwnershipV1(
+	ctx context.Context,
+	deploymentDir string,
+	operation *deploy.OperationLock,
+	operationReleaseAttempted bool,
+	runID string,
+	cleaned bool,
+) error {
+	if operationReleaseAttempted {
+		var err error
+		operation, err = deploy.AcquireOperationLock(ctx, deploymentDir)
+		if err != nil {
+			return fmt.Errorf("reacquire operation lock after controlled session: %w", err)
+		}
+	}
+	var completionErr error
+	if cleaned {
+		_, completionErr = operation.CompleteControlledSessionV1(runID)
+		if completionErr != nil {
+			completionErr = fmt.Errorf("remove verified-clean controlled-session ownership: %w", completionErr)
+		}
+	}
+	unlockErr := operation.Unlock()
+	if unlockErr != nil {
+		unlockErr = fmt.Errorf("release controlled-session operation lock: %w", unlockErr)
+	}
+	return errors.Join(completionErr, unlockErr)
+}
+
+func removeUnstartedControlledSessionV1(operation *deploy.OperationLock, runID string, cause error) error {
+	var removeErr error
+	if deploy.ValidateLiveRunIDV1(runID) == nil {
+		_, _, removeErr = operation.RemoveLiveRunV1(runID)
+	}
+	return releaseControlledSessionOperationV1(operation, errors.Join(cause, removeErr))
+}
+
+func releaseControlledSessionOperationV1(operation *deploy.OperationLock, cause error) error {
+	if err := operation.Unlock(); err != nil {
+		return errors.Join(cause, fmt.Errorf("release controlled-session operation lock: %w", err))
+	}
+	return cause
 }
 
 func runControlledSessionV1(
@@ -165,6 +313,12 @@ func runControlledSessionV1(
 	if backend.prepareChannel == nil || backend.prepareController == nil || backend.prepareWorkload == nil || backend.now == nil {
 		return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session: supervisor backend is incomplete")
 	}
+	ownershipCallbacksEnabled := backend.recordPlannedOwnership != nil ||
+		backend.recordControllerOwnership != nil || backend.recordOwnership != nil
+	if ownershipCallbacksEnabled && (backend.recordPlannedOwnership == nil ||
+		backend.recordControllerOwnership == nil || backend.recordOwnership == nil) {
+		return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session: ownership backend is incomplete")
+	}
 	machine, err := controlledsession.NewMachineV1(plan.Authorization)
 	if err != nil {
 		return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session lifecycle: %w", err)
@@ -242,6 +396,11 @@ func (supervisor *controlledSessionSupervisorV1) run(ctx context.Context) (Contr
 }
 
 func (supervisor *controlledSessionSupervisorV1) prepare(ctx context.Context) error {
+	if supervisor.backend.recordPlannedOwnership != nil {
+		if err := supervisor.backend.recordPlannedOwnership(); err != nil {
+			return err
+		}
+	}
 	channel, err := supervisor.backend.prepareChannel(supervisor.plan)
 	if err != nil {
 		return fmt.Errorf("prepare controlled-session channel: %w", err)
@@ -253,6 +412,26 @@ func (supervisor *controlledSessionSupervisorV1) prepare(ctx context.Context) er
 		return fmt.Errorf("prepare controlled-session controller: %w", err)
 	}
 	supervisor.controller = controller
+	if supervisor.backend.recordControllerOwnership != nil {
+		if recordErr := supervisor.backend.recordControllerOwnership(controller.ContainerID()); recordErr != nil {
+			supervisor.controllerOwnershipIncomplete = true
+			cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout)
+			cleanupErr := controller.Cleanup(cleanupCtx)
+			cleanupCancel()
+			if cleanupErr == nil {
+				if supervisor.backend.recordControllerRollback != nil {
+					supervisor.backend.recordControllerRollback()
+				}
+				supervisor.controllerOwnershipIncomplete = false
+				return recordErr
+			}
+			retryErr := supervisor.backend.recordControllerOwnership(controller.ContainerID())
+			if retryErr != nil {
+				retryErr = fmt.Errorf("retry controlled-session controller ownership after rollback failure: %w", retryErr)
+			}
+			return errors.Join(recordErr, fmt.Errorf("remove inert controlled-session controller after ownership-recording failure: %w", cleanupErr), retryErr)
+		}
+	}
 	workload, err := supervisor.backend.prepareWorkload(ctx, supervisor.plan.Workload)
 	if err != nil {
 		return fmt.Errorf("prepare controlled-session workload: %w", err)
@@ -262,6 +441,11 @@ func (supervisor *controlledSessionSupervisorV1) prepare(ctx context.Context) er
 	if err != nil {
 		return fmt.Errorf("claim controlled-session workload output: %w", err)
 	}
+	if supervisor.backend.recordOwnership != nil {
+		if err := supervisor.backend.recordOwnership(controller.ContainerID(), workload.ContainerID()); err != nil {
+			return err
+		}
+	}
 	if err := controller.Start(ctx); err != nil {
 		return fmt.Errorf("start controlled-session controller: %w", err)
 	}
@@ -710,8 +894,15 @@ func (supervisor *controlledSessionSupervisorV1) cleanupDeliveryTail() (
 			}
 		}
 		cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout)
-		cleanupErr = errors.Join(cleanupErr, supervisor.controller.Cleanup(cleanupCtx))
+		controllerCleanupErr := supervisor.controller.Cleanup(cleanupCtx)
 		cleanupCancel()
+		cleanupErr = errors.Join(cleanupErr, controllerCleanupErr)
+		if controllerCleanupErr == nil && supervisor.controllerOwnershipIncomplete {
+			if supervisor.backend.recordControllerRollback != nil {
+				supervisor.backend.recordControllerRollback()
+			}
+			supervisor.controllerOwnershipIncomplete = false
+		}
 	}
 	status := controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusUnknownV1}
 	if supervisor.controllerObserved != nil {
diff --git a/internal/dockerdeploy/controlled_session_supervisor_integration_test.go b/internal/dockerdeploy/controlled_session_supervisor_integration_test.go
index 7240a392..7dc62e1f 100644
--- a/internal/dockerdeploy/controlled_session_supervisor_integration_test.go
+++ b/internal/dockerdeploy/controlled_session_supervisor_integration_test.go
@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/omry/reploy/internal/controlledsession"
+	"github.com/omry/reploy/internal/deploy"
 )
 
 func TestControlledSessionSupervisorDockerIntegration(t *testing.T) {
@@ -23,8 +24,32 @@ func TestControlledSessionSupervisorDockerIntegration(t *testing.T) {
 	defer cancel()
 	image := buildControlledSessionControllerIntegrationImageV1(t, ctx)
 	plan := controlledSessionControllerIntegrationPlanV1(t, image, []string{"/session-channel-helper", "supervise"})
+	operation, err := deploy.AcquireOperationLock(ctx, plan.Workload.DeploymentDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = operation.Unlock() })
+	lease, err := operation.AcquireLiveRunLeaseV1(plan.LiveRunID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() {
+		if err := lease.Release(); err != nil {
+			t.Errorf("release controlled-session live-run lease: %v", err)
+		}
+	})
+	status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{
+		ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID,
+		GenerationReference: plan.Workload.GenerationReference, Exclusive: true,
+	}, false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status != deploy.LiveRunStatusActiveV1 {
+		t.Fatalf("controlled-session live run status = %q", status)
+	}
 
-	result, err := RunControlledSessionV1(ctx, plan, ControlledSessionRunOptionsV1{
+	result, err := RunControlledSessionV1(ctx, operation, plan, ControlledSessionRunOptionsV1{
 		StartupTimeout: 30 * time.Second, TerminationGrace: 5 * time.Second,
 		ControllerFinalizationTimeout: 15 * time.Second, ResultAcknowledgementTimeout: 5 * time.Second,
 		CleanupTimeout: 15 * time.Second,
@@ -53,4 +78,12 @@ func TestControlledSessionSupervisorDockerIntegration(t *testing.T) {
 	if _, statErr := os.Stat(plan.Channel.HostDirectory); !os.IsNotExist(statErr) {
 		t.Fatalf("private channel directory survived cleanup: %v", statErr)
 	}
+	check, lockErr := deploy.AcquireOperationLock(ctx, plan.Workload.DeploymentDirectory)
+	if lockErr != nil {
+		t.Fatal(lockErr)
+	}
+	defer check.Unlock()
+	if queue, found, readErr := check.ReadLiveRunQueueV1(); readErr != nil || found {
+		t.Fatalf("verified-clean session retained ownership: %#v, found=%t, error=%v", queue, found, readErr)
+	}
 }
diff --git a/internal/dockerdeploy/controlled_session_supervisor_test.go b/internal/dockerdeploy/controlled_session_supervisor_test.go
index 67443e49..cb9751dc 100644
--- a/internal/dockerdeploy/controlled_session_supervisor_test.go
+++ b/internal/dockerdeploy/controlled_session_supervisor_test.go
@@ -4,11 +4,14 @@ import (
 	"context"
 	"errors"
 	"io"
+	"reflect"
+	"strings"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/omry/reploy/internal/controlledsession"
+	"github.com/omry/reploy/internal/deploy"
 )
 
 func TestRunControlledSessionV1OwnsNormalLifecycle(t *testing.T) {
@@ -91,6 +94,386 @@ func TestRunControlledSessionV1OwnsNormalLifecycle(t *testing.T) {
 	}
 }
 
+func TestRunControlledSessionV1PersistsExactOwnershipBeforeStarting(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	controller := newFakeControlledSessionProcessV1()
+	workload := newFakeControlledSessionWorkloadV1(nil, 0)
+	channel := &fakeControlledSessionChannelV1{}
+	persistErr := errors.New("injected durable ownership failure")
+	calls := []string{}
+
+	result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{
+		prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) {
+			calls = append(calls, "channel")
+			return channel, nil
+		},
+		prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) {
+			calls = append(calls, "prepare-controller")
+			return controller, nil
+		},
+		prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) {
+			calls = append(calls, "prepare-workload")
+			return workload, nil
+		},
+		recordPlannedOwnership: func() error {
+			calls = append(calls, "planned")
+			if controller.started || workload.started {
+				t.Fatal("controlled-session process started before planned ownership")
+			}
+			return nil
+		},
+		recordControllerOwnership: func(controllerID string) error {
+			calls = append(calls, "controller")
+			if controllerID != dockerControllerTestContainerIDV1 {
+				t.Fatalf("controller ID = %q", controllerID)
+			}
+			return nil
+		},
+		recordOwnership: func(controllerID string, workloadID string) error {
+			calls = append(calls, "complete")
+			if controller.started || workload.started {
+				t.Fatal("controlled-session process started before durable ownership")
+			}
+			if controllerID != dockerControllerTestContainerIDV1 || workloadID != dockerWorkloadTestContainerIDV1 {
+				t.Fatalf("container IDs = %q / %q", controllerID, workloadID)
+			}
+			return persistErr
+		},
+		now: time.Now,
+	})
+	if !reflect.DeepEqual(calls, []string{"planned", "channel", "prepare-controller", "controller", "prepare-workload", "complete"}) || !errors.Is(err, persistErr) {
+		t.Fatalf("ownership persistence calls=%v, error=%v", calls, err)
+	}
+	if controller.started || workload.started {
+		t.Fatalf("started after persistence failure = controller %t workload %t", controller.started, workload.started)
+	}
+	if !controller.cleaned || !workload.cleaned || !channel.closed {
+		t.Fatalf("inert cleanup = controller %t workload %t channel %t", controller.cleaned, workload.cleaned, channel.closed)
+	}
+	if result.SessionResult.CleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 ||
+		result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 {
+		t.Fatalf("cleanup result = %#v", result)
+	}
+}
+
+func TestRunControlledSessionV1RecordsControllerOwnershipBeforeWorkloadPreparationFailure(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	controller := newFakeControlledSessionProcessV1()
+	channel := &fakeControlledSessionChannelV1{}
+	prepareErr := errors.New("injected workload preparation failure")
+	calls := []string{}
+
+	result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{
+		prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) {
+			calls = append(calls, "channel")
+			return channel, nil
+		},
+		prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) {
+			calls = append(calls, "prepare-controller")
+			return controller, nil
+		},
+		prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) {
+			calls = append(calls, "prepare-workload")
+			return nil, prepareErr
+		},
+		recordPlannedOwnership: func() error {
+			calls = append(calls, "planned")
+			return nil
+		},
+		recordControllerOwnership: func(controllerID string) error {
+			calls = append(calls, "controller")
+			if controllerID != dockerControllerTestContainerIDV1 {
+				t.Fatalf("controller ID = %q", controllerID)
+			}
+			return nil
+		},
+		recordOwnership: func(string, string) error {
+			t.Fatal("complete ownership recorded without a workload")
+			return nil
+		},
+		now: time.Now,
+	})
+	if !errors.Is(err, prepareErr) {
+		t.Fatalf("workload preparation error = %v", err)
+	}
+	if !reflect.DeepEqual(calls, []string{"planned", "channel", "prepare-controller", "controller", "prepare-workload"}) {
+		t.Fatalf("partial ownership calls = %v", calls)
+	}
+	if controller.started {
+		t.Fatal("controller started after workload preparation failed")
+	}
+	if !controller.cleaned || !channel.closed {
+		t.Fatalf("partial preparation cleanup = controller %t channel %t", controller.cleaned, channel.closed)
+	}
+	if result.SessionResult.CleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 ||
+		result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 {
+		t.Fatalf("cleanup result = %#v", result)
+	}
+}
+
+func TestRunControlledSessionV1ReportsVerifiedControllerRollbackAfterOwnershipFailure(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	controller := newFakeControlledSessionProcessV1()
+	channel := &fakeControlledSessionChannelV1{}
+	recordErr := errors.New("injected controller ownership failure")
+	rollbackVerified := false
+
+	result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{
+		prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) {
+			return channel, nil
+		},
+		prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) {
+			return controller, nil
+		},
+		prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) {
+			t.Fatal("workload preparation continued after controller ownership failed")
+			return nil, nil
+		},
+		recordPlannedOwnership: func() error { return nil },
+		recordControllerOwnership: func(string) error {
+			return recordErr
+		},
+		recordControllerRollback: func() { rollbackVerified = true },
+		recordOwnership: func(string, string) error {
+			t.Fatal("complete ownership recorded without a workload")
+			return nil
+		},
+		now: time.Now,
+	})
+	if !errors.Is(err, recordErr) || !rollbackVerified {
+		t.Fatalf("controller ownership error = %v, rollback verified = %t", err, rollbackVerified)
+	}
+	if !controller.cleaned || !channel.closed ||
+		result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 {
+		t.Fatalf("controller cleanup = %t, channel closed = %t, result = %#v", controller.cleaned, channel.closed, result)
+	}
+}
+
+func TestRunControlledSessionV1RetriesControllerOwnershipAfterRollbackFailure(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	controller := newFakeControlledSessionProcessV1()
+	channel := &fakeControlledSessionChannelV1{}
+	recordErr := errors.New("injected controller ownership failure")
+	cleanupErr := errors.New("injected controller rollback failure")
+	controller.cleanupErrs = []error{cleanupErr, nil}
+	recordCalls := 0
+	rollbackVerified := false
+
+	result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{
+		prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) {
+			return channel, nil
+		},
+		prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) {
+			return controller, nil
+		},
+		prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) {
+			t.Fatal("workload preparation continued after controller ownership failed")
+			return nil, nil
+		},
+		recordPlannedOwnership: func() error { return nil },
+		recordControllerOwnership: func(string) error {
+			recordCalls++
+			if recordCalls == 1 {
+				return recordErr
+			}
+			return nil
+		},
+		recordControllerRollback: func() { rollbackVerified = true },
+		recordOwnership: func(string, string) error {
+			t.Fatal("complete ownership recorded without a workload")
+			return nil
+		},
+		now: time.Now,
+	})
+	if !errors.Is(err, recordErr) || !errors.Is(err, cleanupErr) || recordCalls != 2 || !rollbackVerified {
+		t.Fatalf("controller ownership error = %v, calls = %d, rollback verified = %t", err, recordCalls, rollbackVerified)
+	}
+	if controller.cleanupAttempts != 2 || !controller.cleaned || !channel.closed ||
+		result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 {
+		t.Fatalf("controller cleanup attempts = %d, cleaned = %t, channel closed = %t, result = %#v", controller.cleanupAttempts, controller.cleaned, channel.closed, result)
+	}
+}
+
+func TestRunControlledSessionV1RejectsIncompleteOwnershipBackend(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	called := false
+	_, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{
+		prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) {
+			called = true
+			return nil, nil
+		},
+		prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) {
+			called = true
+			return nil, nil
+		},
+		prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) {
+			called = true
+			return nil, nil
+		},
+		recordOwnership: func(string, string) error { return nil },
+		now:             time.Now,
+	})
+	if err == nil || !strings.Contains(err.Error(), "ownership backend is incomplete") {
+		t.Fatalf("incomplete ownership backend error = %v", err)
+	}
+	if called {
+		t.Fatal("incomplete ownership backend began preparation")
+	}
+}
+
+func TestControlledSessionChannelAbsentV1(t *testing.T) {
+	existing := t.TempDir()
+	if controlledSessionChannelAbsentV1(existing) {
+		t.Fatal("existing channel directory reported absent")
+	}
+	if !controlledSessionChannelAbsentV1(existing + "/missing") {
+		t.Fatal("missing channel directory not reported absent")
+	}
+}
+
+func TestControlledSessionPreparationCanCompleteV1(t *testing.T) {
+	tests := []struct {
+		name                      string
+		cleaned                   bool
+		ownershipRecorded         bool
+		operationReleaseAttempted bool
+		partialCleanupVerified    bool
+		want                      bool
+	}{
+		{name: "cleanup failed", ownershipRecorded: true, operationReleaseAttempted: true},
+		{name: "no ownership recorded", cleaned: true, want: true},
+		{name: "full ownership release attempted", cleaned: true, ownershipRecorded: true, operationReleaseAttempted: true, want: true},
+		{name: "partial ownership ambiguous", cleaned: true, ownershipRecorded: true},
+		{name: "channel absence verified", cleaned: true, ownershipRecorded: true, partialCleanupVerified: true, want: true},
+		{name: "workload rollback verified", cleaned: true, ownershipRecorded: true, partialCleanupVerified: true, want: true},
+	}
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			if got := controlledSessionPreparationCanCompleteV1(test.cleaned, test.ownershipRecorded, test.operationReleaseAttempted, test.partialCleanupVerified); got != test.want {
+				t.Fatalf("completion = %t, want %t", got, test.want)
+			}
+		})
+	}
+}
+
+func TestControlledSessionCleanupFailureRetainsDurableOwnership(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{
+		ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID,
+		GenerationReference: plan.Workload.GenerationReference, Exclusive: true,
+	}, false); err != nil || status != deploy.LiveRunStatusActiveV1 {
+		t.Fatalf("admission = %q, %v", status, err)
+	}
+	if _, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1(
+		plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1,
+	)); err != nil {
+		t.Fatal(err)
+	}
+	if err := finishControlledSessionOwnershipV1(t.Context(), plan.Workload.DeploymentDirectory, operation, false, plan.LiveRunID, false); err != nil {
+		t.Fatal(err)
+	}
+	check, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer check.Unlock()
+	queue, found, err := check.ReadLiveRunQueueV1()
+	if err != nil || !found || len(queue.Runs) != 1 || len(queue.ControlledSessions) != 1 {
+		t.Fatalf("retained queue = %#v, found=%t, error=%v", queue, found, err)
+	}
+}
+
+func TestFinishControlledSessionOwnershipReacquiresAfterReleaseAttempt(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{
+		ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID,
+		GenerationReference: plan.Workload.GenerationReference, Exclusive: true,
+	}, false); err != nil || status != deploy.LiveRunStatusActiveV1 {
+		t.Fatalf("admission = %q, %v", status, err)
+	}
+	if _, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1(
+		plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1,
+	)); err != nil {
+		t.Fatal(err)
+	}
+	if err := operation.Unlock(); err != nil {
+		t.Fatal(err)
+	}
+	if err := finishControlledSessionOwnershipV1(
+		t.Context(), plan.Workload.DeploymentDirectory, operation, true, plan.LiveRunID, true,
+	); err != nil {
+		t.Fatal(err)
+	}
+	check, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer check.Unlock()
+	if queue, found, err := check.ReadLiveRunQueueV1(); err != nil || found {
+		t.Fatalf("completed queue = %#v, found=%t, error=%v", queue, found, err)
+	}
+}
+
+func TestRunControlledSessionV1RemovesAdmissionOnLockDirectoryMismatch(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	workloadDir := plan.Workload.DeploymentDirectory
+	operation, err := deploy.AcquireOperationLock(t.Context(), workloadDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{
+		ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID,
+		GenerationReference: plan.Workload.GenerationReference, Exclusive: true,
+	}, false); err != nil || status != deploy.LiveRunStatusActiveV1 {
+		t.Fatalf("admission = %q, %v", status, err)
+	}
+	plan.Workload.DeploymentDirectory = t.TempDir()
+	if _, err := RunControlledSessionV1(t.Context(), operation, plan, testControlledSessionRunOptionsV1()); err == nil || !strings.Contains(err.Error(), "does not belong") {
+		t.Fatalf("lock-directory mismatch error = %v", err)
+	}
+	check, err := deploy.AcquireOperationLock(t.Context(), workloadDir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer check.Unlock()
+	if queue, found, err := check.ReadLiveRunQueueV1(); err != nil || found {
+		t.Fatalf("mismatched operation retained admission: %#v, found=%t, error=%v", queue, found, err)
+	}
+}
+
+func TestRunControlledSessionV1RequiresQueueEntryLease(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{
+		ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID,
+		GenerationReference: plan.Workload.GenerationReference, Exclusive: true,
+	}, false); err != nil || status != deploy.LiveRunStatusActiveV1 {
+		t.Fatalf("admission = %q, %v", status, err)
+	}
+	if _, err := RunControlledSessionV1(t.Context(), operation, plan, testControlledSessionRunOptionsV1()); err == nil || !strings.Contains(err.Error(), "queue-entry lease") {
+		t.Fatalf("missing queue-entry lease error = %v", err)
+	}
+	check, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer check.Unlock()
+	if queue, found, err := check.ReadLiveRunQueueV1(); err != nil || found {
+		t.Fatalf("missing-lease operation retained admission: %#v, found=%t, error=%v", queue, found, err)
+	}
+}
+
 func TestRunControlledSessionV1HoldsCompleteUntilOutputFinalizationPublication(t *testing.T) {
 	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
 	requests := make(chan controlledsession.RequestV1, 8)
@@ -784,6 +1167,12 @@ type fakeControlledSessionProcessV1 struct {
 	gracefulStopped bool
 	forceStopped    bool
 	cleaned         bool
+	cleanupAttempts int
+	cleanupErrs     []error
+}
+
+func (process *fakeControlledSessionProcessV1) ContainerID() string {
+	return dockerControllerTestContainerIDV1
 }
 
 func newFakeControlledSessionProcessV1() *fakeControlledSessionProcessV1 {
@@ -817,6 +1206,14 @@ func (process *fakeControlledSessionProcessV1) ForceStop(context.Context) error
 }
 
 func (process *fakeControlledSessionProcessV1) Cleanup(context.Context) error {
+	process.cleanupAttempts++
+	if len(process.cleanupErrs) > 0 {
+		err := process.cleanupErrs[0]
+		process.cleanupErrs = process.cleanupErrs[1:]
+		if err != nil {
+			return err
+		}
+	}
 	process.cleaned = true
 	return nil
 }
@@ -843,6 +1240,10 @@ type fakeControlledSessionWorkloadV1 struct {
 	rows              uint32
 }
 
+func (workload *fakeControlledSessionWorkloadV1) ContainerID() string {
+	return dockerWorkloadTestContainerIDV1
+}
+
 func newFakeControlledSessionWorkloadV1(output []byte, exitCode int) *fakeControlledSessionWorkloadV1 {
 	reader, writer := io.Pipe()
 	return &fakeControlledSessionWorkloadV1{
diff --git a/internal/dockerdeploy/controlled_session_workload_pty.go b/internal/dockerdeploy/controlled_session_workload_pty.go
index f0fea71a..6dec9dec 100644
--- a/internal/dockerdeploy/controlled_session_workload_pty.go
+++ b/internal/dockerdeploy/controlled_session_workload_pty.go
@@ -22,10 +22,12 @@ type dockerPTYAttachmentV1 interface {
 }
 
 type dockerWorkloadPTYBackendV1 struct {
-	run     commandRunner
-	attach  func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error)
-	resize  func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error
-	observe func(context.Context, CommandSpec, string) (int, error)
+	run                    commandRunner
+	recordContainerID      func(string) error
+	recordRollbackVerified func()
+	attach                 func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error)
+	resize                 func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error
+	observe                func(context.Context, CommandSpec, string) (int, error)
 }
 
 type dockerWorkloadPTYWaitResultV1 struct {
@@ -54,17 +56,32 @@ type DockerWorkloadPTYV1 struct {
 	closeErr  error
 }
 
+func (workload *DockerWorkloadPTYV1) ContainerID() string {
+	return workload.containerID
+}
+
 // PrepareDockerWorkloadPTYV1 creates the exact workload container without
 // starting it and establishes the Docker PTY attachment before returning.
 func PrepareDockerWorkloadPTYV1(
 	ctx context.Context,
 	plan ControlledSessionContainerPlanV1,
+) (*DockerWorkloadPTYV1, error) {
+	return prepareDockerWorkloadPTYWithContainerIDV1(ctx, plan, nil, nil)
+}
+
+func prepareDockerWorkloadPTYWithContainerIDV1(
+	ctx context.Context,
+	plan ControlledSessionContainerPlanV1,
+	recordContainerID func(string) error,
+	recordRollbackVerified func(),
 ) (*DockerWorkloadPTYV1, error) {
 	return prepareDockerWorkloadPTYV1(ctx, plan, dockerWorkloadPTYBackendV1{
-		run:     runDockerCommand,
-		attach:  attachDockerContainerPTYV1,
-		resize:  resizeDockerContainerPTYV1,
-		observe: observeDockerContainerExitV1,
+		run:                    runDockerCommand,
+		recordContainerID:      recordContainerID,
+		recordRollbackVerified: recordRollbackVerified,
+		attach:                 attachDockerContainerPTYV1,
+		resize:                 resizeDockerContainerPTYV1,
+		observe:                observeDockerContainerExitV1,
 	})
 }
 
@@ -99,12 +116,31 @@ func prepareDockerWorkloadPTYV1(
 	if err != nil {
 		return nil, fmt.Errorf("create controlled-session workload container %q: %w; refusing name-based cleanup because the created container identity is unknown", plan.Container, err)
 	}
+	if backend.recordContainerID != nil {
+		if err := backend.recordContainerID(containerID); err != nil {
+			recordErr := fmt.Errorf("record controlled-session workload container %q exact ID: %w", plan.Container, err)
+			if cleanupErr := rollbackControlledSessionWorkloadContainerV1(backend, plan, containerID); cleanupErr != nil {
+				retryErr := backend.recordContainerID(containerID)
+				if retryErr != nil {
+					retryErr = fmt.Errorf("retry controlled-session workload container %q exact ID after rollback failure: %w", plan.Container, retryErr)
+				}
+				return nil, errors.Join(recordErr, fmt.Errorf("remove inert controlled-session workload container %q after ownership-recording failure: %w", plan.Container, cleanupErr), retryErr)
+			}
+			if backend.recordRollbackVerified != nil {
+				backend.recordRollbackVerified()
+			}
+			return nil, recordErr
+		}
+	}
 	attachment, err := backend.attach(ctx, create, containerID, defaultDockerPreflightTimeout)
 	if err != nil {
 		attachErr := fmt.Errorf("attach controlled-session workload PTY for container %q before start: %w", plan.Container, err)
 		if cleanupErr := rollbackControlledSessionWorkloadContainerV1(backend, plan, containerID); cleanupErr != nil {
 			return nil, errors.Join(attachErr, fmt.Errorf("remove inert controlled-session workload container %q after attach failure: %w", plan.Container, cleanupErr))
 		}
+		if backend.recordRollbackVerified != nil {
+			backend.recordRollbackVerified()
+		}
 		return nil, attachErr
 	}
 	return &DockerWorkloadPTYV1{
@@ -301,7 +337,7 @@ func (workload *DockerWorkloadPTYV1) Cleanup(ctx context.Context) error {
 		ctx = context.Background()
 	}
 	cleanup := CommandSpec{Name: workload.plan.Cleanup.Name, Args: []string{"container", "rm", "--force", workload.containerID}}
-	if err := workload.backend.run(cleanup, RunOptions{Context: ctx}); err != nil {
+	if err := workload.backend.run(cleanup, RunOptions{Context: ctx}); err != nil && !isMissingContainerCleanupError(err) {
 		return fmt.Errorf("remove controlled-session workload container %q: %w", workload.plan.Container, err)
 	}
 	workload.stateMu.Lock()
diff --git a/internal/dockerdeploy/controlled_session_workload_pty_test.go b/internal/dockerdeploy/controlled_session_workload_pty_test.go
index f949801c..dffffbef 100644
--- a/internal/dockerdeploy/controlled_session_workload_pty_test.go
+++ b/internal/dockerdeploy/controlled_session_workload_pty_test.go
@@ -168,9 +168,42 @@ func TestDockerWorkloadPTYV1OrdersAttachStartResizeAndExactOperations(t *testing
 	}
 }
 
+func TestDockerWorkloadPTYV1TreatsMissingContainerAsCleaned(t *testing.T) {
+	plan := controlledSessionWorkloadPlanFixtureV1(t)
+	cleanupAttempts := 0
+	workload, err := prepareDockerWorkloadPTYV1(t.Context(), plan, dockerWorkloadPTYBackendV1{
+		run: func(spec CommandSpec, options RunOptions) error {
+			writeDockerWorkloadCreateIDV1(spec, options, plan)
+			if reflect.DeepEqual(spec.Args, []string{"container", "rm", "--force", dockerWorkloadTestContainerIDV1}) {
+				cleanupAttempts++
+				return errors.New("Error response from daemon: No such container: " + dockerWorkloadTestContainerIDV1)
+			}
+			return nil
+		},
+		attach: func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error) {
+			return &fakeDockerPTYAttachmentV1{}, nil
+		},
+		resize:  func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error { return nil },
+		observe: func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := workload.Cleanup(t.Context()); err != nil {
+		t.Fatalf("missing-container cleanup = %v", err)
+	}
+	if err := workload.Cleanup(t.Context()); err != nil {
+		t.Fatalf("repeated cleanup = %v", err)
+	}
+	if cleanupAttempts != 1 {
+		t.Fatalf("cleanup attempts = %d, want 1", cleanupAttempts)
+	}
+}
+
 func TestPrepareDockerWorkloadPTYV1RollsBackInertContainerAfterAttachFailure(t *testing.T) {
 	plan := controlledSessionWorkloadPlanFixtureV1(t)
 	runs := []CommandSpec{}
+	rollbackVerified := false
 	backend := dockerWorkloadPTYBackendV1{
 		run: func(spec CommandSpec, options RunOptions) error {
 			runs = append(runs, spec)
@@ -180,8 +213,9 @@ func TestPrepareDockerWorkloadPTYV1RollsBackInertContainerAfterAttachFailure(t *
 		attach: func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error) {
 			return nil, errors.New("attach refused")
 		},
-		resize:  func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error { return nil },
-		observe: func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+		recordRollbackVerified: func() { rollbackVerified = true },
+		resize:                 func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error { return nil },
+		observe:                func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
 	}
 	_, err := prepareDockerWorkloadPTYV1(t.Context(), plan, backend)
 	if err == nil || !strings.Contains(err.Error(), "before start") || !strings.Contains(err.Error(), "attach refused") {
@@ -191,6 +225,103 @@ func TestPrepareDockerWorkloadPTYV1RollsBackInertContainerAfterAttachFailure(t *
 		!reflect.DeepEqual(runs[1].Args, []string{"container", "rm", "--force", dockerWorkloadTestContainerIDV1}) {
 		t.Fatalf("rollback commands = %#v", runs)
 	}
+	if !rollbackVerified {
+		t.Fatal("successful rollback was not reported")
+	}
+}
+
+func TestPrepareDockerWorkloadPTYV1RecordsExactIDBeforeAttachFailure(t *testing.T) {
+	plan := controlledSessionWorkloadPlanFixtureV1(t)
+	actions := []string{}
+	rollbackVerified := false
+	cleanupErr := errors.New("cleanup unavailable")
+	backend := dockerWorkloadPTYBackendV1{
+		run: func(spec CommandSpec, options RunOptions) error {
+			actions = append(actions, strings.Join(spec.Args, " "))
+			writeDockerWorkloadCreateIDV1(spec, options, plan)
+			if reflect.DeepEqual(spec.Args, []string{"container", "rm", "--force", dockerWorkloadTestContainerIDV1}) {
+				return cleanupErr
+			}
+			return nil
+		},
+		recordContainerID: func(containerID string) error {
+			actions = append(actions, "record "+containerID)
+			return nil
+		},
+		recordRollbackVerified: func() { rollbackVerified = true },
+		attach: func(_ context.Context, _ CommandSpec, containerID string, _ time.Duration) (dockerPTYAttachmentV1, error) {
+			actions = append(actions, "attach "+containerID)
+			return nil, errors.New("attach refused")
+		},
+		resize:  func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error { return nil },
+		observe: func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+	}
+	_, err := prepareDockerWorkloadPTYV1(t.Context(), plan, backend)
+	if err == nil || !strings.Contains(err.Error(), "attach refused") || !errors.Is(err, cleanupErr) {
+		t.Fatalf("attach and rollback error = %v", err)
+	}
+	want := []string{
+		strings.Join(plan.Create.Args, " "),
+		"record " + dockerWorkloadTestContainerIDV1,
+		"attach " + dockerWorkloadTestContainerIDV1,
+		"container rm --force " + dockerWorkloadTestContainerIDV1,
+	}
+	if !reflect.DeepEqual(actions, want) {
+		t.Fatalf("partial preparation actions = %#v, want %#v", actions, want)
+	}
+	if rollbackVerified {
+		t.Fatal("failed rollback was reported as verified")
+	}
+}
+
+func TestPrepareDockerWorkloadPTYV1RetriesExactIDAfterRollbackFailure(t *testing.T) {
+	plan := controlledSessionWorkloadPlanFixtureV1(t)
+	actions := []string{}
+	recordErr := errors.New("injected workload ownership failure")
+	cleanupErr := errors.New("injected workload rollback failure")
+	recordCalls := 0
+	rollbackVerified := false
+	backend := dockerWorkloadPTYBackendV1{
+		run: func(spec CommandSpec, options RunOptions) error {
+			actions = append(actions, strings.Join(spec.Args, " "))
+			writeDockerWorkloadCreateIDV1(spec, options, plan)
+			if reflect.DeepEqual(spec.Args, []string{"container", "rm", "--force", dockerWorkloadTestContainerIDV1}) {
+				return cleanupErr
+			}
+			return nil
+		},
+		recordContainerID: func(containerID string) error {
+			recordCalls++
+			actions = append(actions, "record "+containerID)
+			if recordCalls == 1 {
+				return recordErr
+			}
+			return nil
+		},
+		recordRollbackVerified: func() { rollbackVerified = true },
+		attach: func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error) {
+			t.Fatal("attachment continued after ownership recording failed")
+			return nil, nil
+		},
+		resize:  func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error { return nil },
+		observe: func(context.Context, CommandSpec, string) (int, error) { return 0, nil },
+	}
+	_, err := prepareDockerWorkloadPTYV1(t.Context(), plan, backend)
+	if !errors.Is(err, recordErr) || !errors.Is(err, cleanupErr) || recordCalls != 2 {
+		t.Fatalf("workload preparation error = %v, record calls = %d", err, recordCalls)
+	}
+	want := []string{
+		strings.Join(plan.Create.Args, " "),
+		"record " + dockerWorkloadTestContainerIDV1,
+		"container rm --force " + dockerWorkloadTestContainerIDV1,
+		"record " + dockerWorkloadTestContainerIDV1,
+	}
+	if !reflect.DeepEqual(actions, want) {
+		t.Fatalf("partial preparation actions = %#v, want %#v", actions, want)
+	}
+	if rollbackVerified {
+		t.Fatal("failed rollback was reported as verified")
+	}
 }
 
 func TestDockerWorkloadPTYV1RollsBackAmbiguousStartFailure(t *testing.T) {
diff --git a/internal/dockerdeploy/live_runs.go b/internal/dockerdeploy/live_runs.go
index 9667bbc4..6bf17183 100644
--- a/internal/dockerdeploy/live_runs.go
+++ b/internal/dockerdeploy/live_runs.go
@@ -130,13 +130,15 @@ func stopLiveRunV1(
 	}
 	result.Found = true
 	result.Run = run
-	if run.Status == deploy.LiveRunStatusActiveV1 && run.Container != "" {
-		removeErr := backend.removeContainer(
-			TemporaryContainerCleanupCommand(run.Container),
-			RunOptions{Context: ctx, DockerPreflightTimeout: dockerPreflightTimeout},
-		)
-		if removeErr != nil && !isMissingContainerCleanupError(removeErr) {
-			return result, fmt.Errorf("stop live run container %q: %w", run.Container, removeErr)
+	if run.Status == deploy.LiveRunStatusActiveV1 {
+		for _, container := range liveRunContainerTargetsV1(queue, run) {
+			removeErr := backend.removeContainer(
+				TemporaryContainerCleanupCommand(container),
+				RunOptions{Context: ctx, DockerPreflightTimeout: dockerPreflightTimeout},
+			)
+			if removeErr != nil && !isMissingContainerCleanupError(removeErr) {
+				return result, fmt.Errorf("stop live run container %q: %w", container, removeErr)
+			}
 		}
 	}
 	_, removed, err := operation.RemoveLiveRunV1(id)
@@ -151,3 +153,26 @@ func stopLiveRunV1(
 	}
 	return result, nil
 }
+
+// liveRunContainerTargetsV1 returns every exact container owned by a live run.
+// Workload-first ordering leaves the controller available to observe workload
+// termination for as long as possible. Controlled-session ownership remains
+// durable until its supervisor or recovery verifies complete cleanup.
+func liveRunContainerTargetsV1(queue deploy.LiveRunQueueV1, run deploy.LiveRunV1) []string {
+	targets := make([]string, 0, 2)
+	if run.Container != "" {
+		targets = append(targets, run.Container)
+	}
+	for _, ownership := range queue.ControlledSessions {
+		if ownership.LiveRunID == run.ID {
+			if ownership.Workload.ID != "" {
+				targets = append(targets, ownership.Workload.ID)
+			}
+			if ownership.Controller.ID != "" {
+				targets = append(targets, ownership.Controller.ID)
+			}
+			break
+		}
+	}
+	return targets
+}
diff --git a/internal/dockerdeploy/live_runs_test.go b/internal/dockerdeploy/live_runs_test.go
index 40fe8422..0ffc7c4b 100644
--- a/internal/dockerdeploy/live_runs_test.go
+++ b/internal/dockerdeploy/live_runs_test.go
@@ -133,6 +133,160 @@ func TestStopLiveRunV1RemovesActiveContainerBeforePromotingWaiter(t *testing.T)
 	}
 }
 
+func TestStopLiveRunV1RemovesControlledSessionContainersAndRetainsOwnership(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	dir := plan.Workload.DeploymentDirectory
+	operation, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	run := liveRunAdmissionFixtureV1(plan.LiveRunID, false)
+	run.Kind = deploy.LiveRunKindShellV1
+	run.GenerationReference = plan.Workload.GenerationReference
+	holdLiveRunLeaseV1(t, operation, run.ID)
+	if _, err := operation.AdmitLiveRunV1(run, false); err != nil {
+		t.Fatal(err)
+	}
+	ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1(
+		plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1,
+	))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := operation.Unlock(); err != nil {
+		t.Fatal(err)
+	}
+
+	calls := []CommandSpec{}
+	result, err := stopLiveRunV1(t.Context(), dir, run.ID, 7*time.Second, liveRunsBackendV1{
+		acquire: deploy.AcquireOperationLock,
+		removeContainer: func(spec CommandSpec, options RunOptions) error {
+			if options.DockerPreflightTimeout != 7*time.Second {
+				t.Fatalf("Docker timeout = %s", options.DockerPreflightTimeout)
+			}
+			calls = append(calls, spec)
+			return nil
+		},
+	})
+	if err != nil || !result.Found || result.Run.ID != run.ID {
+		t.Fatalf("controlled-session stop = %#v, %v", result, err)
+	}
+	wantCalls := []CommandSpec{
+		TemporaryContainerCleanupCommand(dockerWorkloadTestContainerIDV1),
+		TemporaryContainerCleanupCommand(dockerControllerTestContainerIDV1),
+	}
+	if !reflect.DeepEqual(calls, wantCalls) {
+		t.Fatalf("controlled-session cleanup calls = %#v", calls)
+	}
+	check, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer check.Unlock()
+	queue, found, err := check.ReadLiveRunQueueV1()
+	if err != nil || !found || len(queue.Runs) != 0 || len(queue.ControlledSessions) != 1 || queue.ControlledSessions[0] != ownership {
+		t.Fatalf("retained controlled-session ownership = %#v, found=%t, error=%v", queue, found, err)
+	}
+}
+
+func TestStopLiveRunV1SkipsUnrecordedControlledSessionContainerIDs(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	dir := plan.Workload.DeploymentDirectory
+	operation, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	run := liveRunAdmissionFixtureV1(plan.LiveRunID, false)
+	run.Kind = deploy.LiveRunKindShellV1
+	run.GenerationReference = plan.Workload.GenerationReference
+	holdLiveRunLeaseV1(t, operation, run.ID)
+	if _, err := operation.AdmitLiveRunV1(run, false); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1(
+		plan, dockerControllerTestContainerIDV1, "",
+	)); err != nil {
+		t.Fatal(err)
+	}
+	if err := operation.Unlock(); err != nil {
+		t.Fatal(err)
+	}
+
+	calls := []CommandSpec{}
+	result, err := stopLiveRunV1(t.Context(), dir, run.ID, 0, liveRunsBackendV1{
+		acquire: deploy.AcquireOperationLock,
+		removeContainer: func(spec CommandSpec, _ RunOptions) error {
+			calls = append(calls, spec)
+			return nil
+		},
+	})
+	if err != nil || !result.Found {
+		t.Fatalf("partial-ownership stop = %#v, %v", result, err)
+	}
+	want := []CommandSpec{TemporaryContainerCleanupCommand(dockerControllerTestContainerIDV1)}
+	if !reflect.DeepEqual(calls, want) {
+		t.Fatalf("partial-ownership cleanup calls = %#v, want %#v", calls, want)
+	}
+}
+
+func TestStopLiveRunV1PreservesControlledSessionOnPartialCleanupFailure(t *testing.T) {
+	plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"})
+	dir := plan.Workload.DeploymentDirectory
+	operation, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	run := liveRunAdmissionFixtureV1(plan.LiveRunID, false)
+	run.Kind = deploy.LiveRunKindShellV1
+	run.GenerationReference = plan.Workload.GenerationReference
+	holdLiveRunLeaseV1(t, operation, run.ID)
+	if _, err := operation.AdmitLiveRunV1(run, false); err != nil {
+		t.Fatal(err)
+	}
+	ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1(
+		plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1,
+	))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := operation.Unlock(); err != nil {
+		t.Fatal(err)
+	}
+
+	want := errors.New("controller cleanup failed")
+	calls := []CommandSpec{}
+	result, err := stopLiveRunV1(t.Context(), dir, run.ID, 7*time.Second, liveRunsBackendV1{
+		acquire: deploy.AcquireOperationLock,
+		removeContainer: func(spec CommandSpec, _ RunOptions) error {
+			calls = append(calls, spec)
+			if len(calls) == 2 {
+				return want
+			}
+			return nil
+		},
+	})
+	if !errors.Is(err, want) || !result.Found || result.Run.ID != run.ID {
+		t.Fatalf("partial controlled-session cleanup = %#v, %v", result, err)
+	}
+	wantCalls := []CommandSpec{
+		TemporaryContainerCleanupCommand(dockerWorkloadTestContainerIDV1),
+		TemporaryContainerCleanupCommand(dockerControllerTestContainerIDV1),
+	}
+	if !reflect.DeepEqual(calls, wantCalls) {
+		t.Fatalf("partial controlled-session cleanup calls = %#v", calls)
+	}
+	check, err := deploy.AcquireOperationLock(t.Context(), dir)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer check.Unlock()
+	queue, _, err := check.ReadLiveRunQueueV1()
+	if err != nil || len(queue.Runs) != 1 || queue.Runs[0].ID != run.ID ||
+		len(queue.ControlledSessions) != 1 || queue.ControlledSessions[0] != ownership {
+		t.Fatalf("queue after partial controlled-session cleanup = %#v, error=%v", queue, err)
+	}
+}
+
 func TestStopLiveRunV1ReportsReadyReservationAsWaitingWithoutDocker(t *testing.T) {
 	dir := t.TempDir()
 	operation, err := deploy.AcquireOperationLock(t.Context(), dir)