diff --git a/pkg/tern/control_requests.go b/pkg/tern/control_requests.go
index 3aef76c00..98984da4a 100644
--- a/pkg/tern/control_requests.go
+++ b/pkg/tern/control_requests.go
@@ -3,6 +3,7 @@ package tern
 import (
 	"context"
 	"fmt"
+	"log/slog"
 	"time"
 
 	"github.com/block/schemabot/pkg/state"
@@ -102,7 +103,7 @@ func failPendingControlRequests(ctx context.Context, store storage.Storage, appl
 	return nil
 }
 
-func markApplyCuttingOverForControlRequest(ctx context.Context, store storage.Storage, apply *storage.Apply) error {
+func markApplyCuttingOverForControlRequest(ctx context.Context, store storage.Storage, apply *storage.Apply, logger *slog.Logger) error {
 	if !state.IsState(apply.State, state.Apply.WaitingForCutover) && !state.IsRunningApplyState(apply.State) {
 		return nil
 	}
@@ -117,6 +118,16 @@ func markApplyCuttingOverForControlRequest(ctx context.Context, store storage.St
 	now := time.Now()
 	apply.State = state.Apply.CuttingOver
 	apply.UpdatedAt = now
+	// A multi-operation drive owns only its operation: the parent cutting_over
+	// write is the operator's projection to make — a direct write here fails
+	// closed under the operation-only lease and would block a cutover the
+	// engine is ready to accept. The in-memory transition still stands so this
+	// drive proceeds to dispatch the cutover.
+	if suppressParentApplyWrites(ctx) {
+		logger.Info("pending cutover request accepted under operation lease; parent cutting_over state is the operator's projection",
+			"state", apply.State)
+		return nil
+	}
 	if err := applyStore.Update(ctx, apply); err != nil {
 		*apply = previous
 		return fmt.Errorf("mark apply %s cutting over for pending cutover request: %w", apply.ApplyIdentifier, err)
diff --git a/pkg/tern/grpc_client.go b/pkg/tern/grpc_client.go
index 7d2093716..7e6214808 100644
--- a/pkg/tern/grpc_client.go
+++ b/pkg/tern/grpc_client.go
@@ -622,7 +622,7 @@ func (c *GRPCClient) processPendingCutoverControlRequest(ctx context.Context, ap
 		message := "schema change has a pending stop request; cutover is blocked until stop is processed"
 		return fmt.Errorf("process pending gRPC cutover for apply %s: %s", apply.ApplyIdentifier, message)
 	}
-	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply); err != nil {
+	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply, logger); err != nil {
 		return err
 	}
 	resp, err := c.client.Cutover(ctx, &ternv1.CutoverRequest{
diff --git a/pkg/tern/local_apply_failure.go b/pkg/tern/local_apply_failure.go
index 6e40cb313..d13ed4b0e 100644
--- a/pkg/tern/local_apply_failure.go
+++ b/pkg/tern/local_apply_failure.go
@@ -51,6 +51,15 @@ func (c *LocalClient) failApplyWithTasks(ctx context.Context, apply *storage.App
 	}
 
 	logger := c.logger.With(apply.IdentityLogAttrs()...)
+	// A multi-operation drive owns only its operation: the failed tasks above
+	// carry the outcome, the operator derives the operation row from them and
+	// projects the parent, so the parent failed write, apply-level metric, and
+	// failure log are the operator's to make.
+	if suppressParentApplyWrites(ctx) {
+		logger.Info("operation drive failed its tasks; operator derives the operation row and projects the parent",
+			"error_message", errMsg)
+		return
+	}
 	// Re-read the apply from storage — Stop() may have already set a terminal
 	// state (e.g., cancelled) between when the engine error occurred and now.
 	fresh, err := c.storage.Applies().Get(ctx, apply.ID)
@@ -90,6 +99,15 @@ func (c *LocalClient) markApplyRetryableWithTasks(ctx context.Context, apply *st
 	}
 
 	logger := c.logger.With(apply.IdentityLogAttrs()...)
+	// A multi-operation drive owns only its operation: the failed_retryable
+	// tasks above carry the outcome, the operator derives the operation row from
+	// them and projects the parent, so the parent retryable write, apply-level
+	// metric, retry log, and observer are the operator's to make.
+	if suppressParentApplyWrites(ctx) {
+		logger.Info("operation drive paused its tasks for retry; operator derives the operation row and projects the parent",
+			"error_message", errMsg)
+		return
+	}
 	// Re-read the apply from storage; Stop() may have already moved it to a
 	// terminal state between the engine error and this update.
 	fresh, err := c.storage.Applies().Get(ctx, apply.ID)
diff --git a/pkg/tern/local_apply_sequential.go b/pkg/tern/local_apply_sequential.go
index fa180d0ae..d2edcf71a 100644
--- a/pkg/tern/local_apply_sequential.go
+++ b/pkg/tern/local_apply_sequential.go
@@ -643,6 +643,26 @@ func (c *LocalClient) shouldRetryEngineError(err error) bool {
 func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storage.Apply, tasks []*storage.Task, failedTask *storage.Task, stoppedByUser bool) {
 	now := time.Now()
 	logger := c.logger.With(apply.IdentityLogAttrs()...)
+	// A multi-operation drive owns only its operation: the tasks it drove carry
+	// the outcome, the operator derives the operation row from them and projects
+	// the parent, so the parent terminal write, control-request completion,
+	// apply-level metric, and terminal observer are all the operator's to make.
+	// Pending tasks after a failed one are still this drive's to settle, and the
+	// in-memory outcome is still adopted so the drive's own logs report what the
+	// operation settled to rather than the projection's stale running state.
+	if suppressParentApplyWrites(ctx) {
+		if failedTask != nil && failedTask.State != state.Task.FailedRetryable {
+			for _, task := range tasks {
+				if task.State == state.Task.Pending {
+					c.transitionTaskState(ctx, task, 0, state.Task.Cancelled, "")
+				}
+			}
+		}
+		adoptSequentialOutcome(apply, failedTask, stoppedByUser, now)
+		logger.Info("sequential operation drive settled; operator derives the operation row and projects the parent",
+			"stopped_by_user", stoppedByUser, "failed_task", failedTask != nil, "settled_state", apply.State)
+		return
+	}
 	if freshApply, err := c.storage.Applies().Get(ctx, apply.ID); err != nil {
 		logger.Error("failed to reload apply before sequential finalization",
 			append(apply.MutableLogAttrs(), "error", err)...)
@@ -658,27 +678,14 @@ func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storag
 		return
 	}
 	previousState := apply.State
-	switch {
-	case failedTask != nil && failedTask.State == state.Task.FailedRetryable:
-		apply.State = state.Apply.FailedRetryable
-		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
-		apply.CompletedAt = nil
-	case failedTask != nil:
-		apply.State = state.Apply.Failed
-		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
-		apply.CompletedAt = &now
+	if failedTask != nil && failedTask.State != state.Task.FailedRetryable {
 		for _, task := range tasks {
 			if task.State == state.Task.Pending {
 				c.transitionTaskState(ctx, task, 0, state.Task.Cancelled, "")
 			}
 		}
-	case stoppedByUser:
-		apply.State = state.Apply.Stopped
-	default:
-		apply.State = state.Apply.Completed
-		apply.CompletedAt = &now
 	}
-	apply.UpdatedAt = now
+	adoptSequentialOutcome(apply, failedTask, stoppedByUser, now)
 	if err := c.storage.Applies().Update(ctx, apply); err != nil {
 		logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
 	} else {
@@ -714,3 +721,26 @@ func (c *LocalClient) finalizeSequentialApply(ctx context.Context, apply *storag
 		c.clearObserver(apply.ID)
 	}
 }
+
+// adoptSequentialOutcome mutates the in-memory apply to the outcome its
+// sequential task results settle to: failed_retryable for a retryable task
+// failure, failed for a permanent one, stopped for an operator stop, completed
+// otherwise. Callers own persisting (or not persisting) the mutated row.
+func adoptSequentialOutcome(apply *storage.Apply, failedTask *storage.Task, stoppedByUser bool, now time.Time) {
+	switch {
+	case failedTask != nil && failedTask.State == state.Task.FailedRetryable:
+		apply.State = state.Apply.FailedRetryable
+		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
+		apply.CompletedAt = nil
+	case failedTask != nil:
+		apply.State = state.Apply.Failed
+		apply.ErrorMessage = fmt.Sprintf("table %s failed: %s", failedTask.TableName, failedTask.ErrorMessage)
+		apply.CompletedAt = &now
+	case stoppedByUser:
+		apply.State = state.Apply.Stopped
+	default:
+		apply.State = state.Apply.Completed
+		apply.CompletedAt = &now
+	}
+	apply.UpdatedAt = now
+}
diff --git a/pkg/tern/local_control.go b/pkg/tern/local_control.go
index ac78e30fe..53d7aa368 100644
--- a/pkg/tern/local_control.go
+++ b/pkg/tern/local_control.go
@@ -288,7 +288,7 @@ func (c *LocalClient) processPendingCutoverControlRequest(ctx context.Context, a
 		message := "schema change has a pending stop request; cutover is blocked until stop is processed"
 		return fmt.Errorf("process pending cutover for apply %s: %s", apply.ApplyIdentifier, message)
 	}
-	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply); err != nil {
+	if err := markApplyCuttingOverForControlRequest(ctx, c.storage, apply, logger); err != nil {
 		return err
 	}
 	resp, err := c.cutover(ctx, &ternv1.CutoverRequest{
@@ -1653,6 +1653,25 @@ func (c *LocalClient) settleControlForCompletedEngineChange(ctx context.Context,
 	if applyID == 0 {
 		return 0, fmt.Errorf("%s found the schema change already completed on the engine, but resolved no apply to settle: %w", operation, rejection)
 	}
+	// A multi-operation drive owns only its operation: the tasks settled above
+	// carry the completed outcome, the operator derives the operation row from
+	// them and projects the parent completed, so the parent write, apply event,
+	// and terminal observer are the operator's to make — a direct write here
+	// fails closed under the operation-only lease and would turn an accepted
+	// settle into a drive error the claim loop re-runs forever.
+	if suppressParentApplyWrites(ctx) {
+		attrs := []any{"database", c.config.Database}
+		if targetApply != nil {
+			attrs = targetApply.LogAttrs()
+		}
+		c.logger.Info("engine rejected "+operation+" because the schema change already completed; operation drive settled its tasks and the operator projects the parent",
+			append(attrs,
+				"requested_by", caller,
+				"completed_task_count", completedCount,
+				"terminal_task_count", skippedCount,
+				"error", rejection)...)
+		return skippedCount, nil
+	}
 	apply, err := c.storage.Applies().Get(ctx, applyID)
 	if err != nil {
 		return 0, fmt.Errorf("load apply %d to settle %s for a completed schema change: %w", applyID, operation, err)
diff --git a/pkg/tern/local_control_multiop_resume_integration_test.go b/pkg/tern/local_control_multiop_resume_integration_test.go
new file mode 100644
index 000000000..e682c4cc4
--- /dev/null
+++ b/pkg/tern/local_control_multiop_resume_integration_test.go
@@ -0,0 +1,455 @@
+//go:build integration
+
+package tern
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+	"log/slog"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/block/spirit/pkg/utils"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/block/schemabot/pkg/engine"
+	ternv1 "github.com/block/schemabot/pkg/proto/ternv1"
+	"github.com/block/schemabot/pkg/state"
+	"github.com/block/schemabot/pkg/storage"
+)
+
+// multiOpResumeEngine is a fake engine for operation-scoped resume tests. Its
+// re-plan result is configurable so a test can choose whether the driven task
+// still has remaining work; Apply can be forced to reject or error so the
+// failure and retry paths are exercised without a real backend; Stop can be
+// forced to report the change already completed; and its progress can be
+// declared externally authoritative so terminal-truth reconciliation runs.
+type multiOpResumeEngine struct {
+	engine.Engine
+
+	mu                    sync.Mutex
+	planChanges           []engine.SchemaChange
+	rejectApply           bool
+	applyErr              error
+	stopErr               error
+	authoritativeProgress bool
+}
+
+func (e *multiOpResumeEngine) Name() string { return "multi-op-resume" }
+
+func (e *multiOpResumeEngine) Plan(context.Context, *engine.PlanRequest) (*engine.PlanResult, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	return &engine.PlanResult{Changes: e.planChanges}, nil
+}
+
+func (e *multiOpResumeEngine) Apply(context.Context, *engine.ApplyRequest) (*engine.ApplyResult, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.applyErr != nil {
+		return nil, e.applyErr
+	}
+	if e.rejectApply {
+		return &engine.ApplyResult{Accepted: false, Message: "engine rejected the schema change"}, nil
+	}
+	return &engine.ApplyResult{Accepted: true}, nil
+}
+
+func (e *multiOpResumeEngine) Progress(context.Context, *engine.ProgressRequest) (*engine.ProgressResult, error) {
+	return &engine.ProgressResult{State: engine.StateCompleted, Progress: 100}, nil
+}
+
+func (e *multiOpResumeEngine) Start(context.Context, *engine.ControlRequest) (*engine.ControlResult, error) {
+	return &engine.ControlResult{Accepted: true}, nil
+}
+
+func (e *multiOpResumeEngine) Stop(context.Context, *engine.ControlRequest) (*engine.ControlResult, error) {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if e.stopErr != nil {
+		return nil, e.stopErr
+	}
+	return &engine.ControlResult{Accepted: true}, nil
+}
+
+func (e *multiOpResumeEngine) ProgressIsExternallyAuthoritative() bool {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	return e.authoritativeProgress
+}
+
+const multiOpResumeDDL = "ALTER TABLE `users` ADD COLUMN `email` VARCHAR(255)"
+
+// multiOpResumeFixture is a running multi-operation apply whose parent state is
+// owned by the operator's rollout projection: two operation rows, with the
+// drive's own operation leased and carrying the fixture's tasks.
+type multiOpResumeFixture struct {
+	stor    storage.Storage
+	client  *LocalClient
+	eng     *multiOpResumeEngine
+	apply   *storage.Apply
+	opID    int64
+	opCtx   context.Context
+	tasks   []*storage.Task
+	leaseDB *sql.DB
+}
+
+// newMultiOpResumeFixture seeds a running two-operation apply with the given
+// task states scoped to the first operation, stamps that operation's lease, and
+// returns a context carrying the operation lease alone — the shape of an
+// operator's operation-scoped drive claim.
+func newMultiOpResumeFixture(t *testing.T, taskStates []string) *multiOpResumeFixture {
+	t.Helper()
+	_, dsn := setupMySQLContainer(t)
+	setupStorageSchema(t, dsn)
+	cleanupTasks(t, dsn)
+	cleanupTestTables(t, dsn)
+
+	ctx := t.Context()
+	stor := createStorage(t, dsn)
+	t.Cleanup(func() { utils.CloseAndLog(stor) })
+
+	logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError}))
+	client, err := NewLocalClient(LocalConfig{
+		Database:  "testdb",
+		Type:      storage.DatabaseTypeMySQL,
+		TargetDSN: dsn,
+	}, stor, logger)
+	require.NoError(t, err)
+	t.Cleanup(func() { utils.CloseAndLog(client) })
+	eng := &multiOpResumeEngine{}
+	client.spiritEngine = eng
+
+	now := time.Now()
+	plan := &storage.Plan{
+		PlanIdentifier: fmt.Sprintf("plan-multiop-%d", now.UnixNano()),
+		Database:       "testdb",
+		DatabaseType:   storage.DatabaseTypeMySQL,
+		Deployment:     "testdb",
+		Environment:    localClientTestEnvironment,
+		CreatedAt:      now,
+		Namespaces: map[string]*storage.NamespacePlanData{
+			"testdb": {Tables: []storage.TableChange{{Table: "users", DDL: multiOpResumeDDL, Operation: "alter"}}},
+		},
+	}
+	planID, err := stor.Plans().Create(ctx, plan)
+	require.NoError(t, err)
+
+	apply := &storage.Apply{
+		ApplyIdentifier: fmt.Sprintf("apply-multiop-%d", now.UnixNano()),
+		PlanID:          planID,
+		Database:        "testdb",
+		DatabaseType:    storage.DatabaseTypeMySQL,
+		Deployment:      "testdb",
+		Environment:     localClientTestEnvironment,
+		State:           state.Apply.Running,
+		StartedAt:       &now,
+		CreatedAt:       now,
+		UpdatedAt:       now,
+	}
+	applyID, err := stor.Applies().Create(ctx, apply)
+	require.NoError(t, err)
+	apply.ID = applyID
+
+	opID, err := stor.ApplyOperations().Insert(ctx, &storage.ApplyOperation{
+		ApplyID:    applyID,
+		Deployment: "testdb",
+		Target:     "testdb",
+		State:      state.ApplyOperation.Running,
+	})
+	require.NoError(t, err)
+	_, err = stor.ApplyOperations().Insert(ctx, &storage.ApplyOperation{
+		ApplyID:    applyID,
+		Deployment: "testdb-sibling",
+		Target:     "testdb-sibling",
+		State:      state.ApplyOperation.Running,
+	})
+	require.NoError(t, err)
+
+	tasks := make([]*storage.Task, 0, len(taskStates))
+	for i, taskState := range taskStates {
+		task := &storage.Task{
+			TaskIdentifier:   fmt.Sprintf("task-multiop-%d-%d", now.UnixNano(), i),
+			ApplyID:          applyID,
+			ApplyOperationID: &opID,
+			PlanID:           planID,
+			Database:         "testdb",
+			DatabaseType:     storage.DatabaseTypeMySQL,
+			Engine:           storage.EngineSpirit,
+			Environment:      localClientTestEnvironment,
+			State:            taskState,
+			Namespace:        "testdb",
+			TableName:        "users",
+			DDL:              multiOpResumeDDL,
+			DDLAction:        "alter",
+			CreatedAt:        now,
+			UpdatedAt:        now,
+		}
+		_, err := stor.Tasks().Create(ctx, task)
+		require.NoError(t, err)
+		tasks = append(tasks, task)
+	}
+
+	leaseDB, err := sql.Open("mysql", dsn)
+	require.NoError(t, err)
+	t.Cleanup(func() { utils.CloseAndLog(leaseDB) })
+	require.NoError(t, leaseDB.PingContext(ctx))
+	_, err = leaseDB.ExecContext(ctx, `
+		UPDATE apply_operations SET lease_owner = ?, lease_token = ?, lease_acquired_at = NOW() WHERE id = ?
+	`, "op-driver", "op-token", opID)
+	require.NoError(t, err)
+
+	opCtx := storage.WithOperationLease(ctx, storage.OperationLease{
+		ApplyID: applyID, OperationID: opID, Owner: "op-driver", Token: "op-token",
+	})
+
+	return &multiOpResumeFixture{
+		stor: stor, client: client, eng: eng,
+		apply: apply, opID: opID, opCtx: opCtx, tasks: tasks, leaseDB: leaseDB,
+	}
+}
+
+// requireParentUntouched asserts the parent applies row still carries the state
+// the operator's projection gave it — running, not completed, no error — so the
+// drive provably left the parent to the projection.
+func (f *multiOpResumeFixture) requireParentUntouched(t *testing.T) {
+	t.Helper()
+	parent, err := f.stor.Applies().Get(t.Context(), f.apply.ID)
+	require.NoError(t, err)
+	require.NotNil(t, parent)
+	assert.Equal(t, state.Apply.Running, parent.State,
+		"the parent applies row is owned by the rollout projection and must not be written by an operation-scoped drive")
+	assert.Nil(t, parent.CompletedAt, "the parent must not be terminalized by an operation-scoped drive")
+	assert.Empty(t, parent.ErrorMessage, "the parent must not carry a drive-written error message")
+}
+
+func (f *multiOpResumeFixture) taskState(t *testing.T, task *storage.Task) string {
+	t.Helper()
+	fresh, err := f.stor.Tasks().Get(t.Context(), task.TaskIdentifier)
+	require.NoError(t, err)
+	require.NotNil(t, fresh)
+	return fresh.State
+}
+
+// An operation-scoped drive of a multi-operation apply holds only its operation
+// lease; the parent applies row belongs to the operator's rollout projection.
+// A sequential-mode resume must drive its own tasks to completion without ever
+// writing the parent — a parent write would be refused by storage and abort the
+// drive before any task work starts, leaving the operation re-claimed and
+// re-refused on every operator tick while the schema change never progresses.
+func TestLocalClient_SequentialOperationResumeDrivesTasksWithoutParentWrites(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.Pending})
+	f.eng.planChanges = []engine.SchemaChange{{
+		Namespace:    "testdb",
+		TableChanges: []engine.TableChange{{Table: "users", DDL: multiOpResumeDDL}},
+	}}
+
+	require.NoError(t, f.client.ResumeApplyOperation(f.opCtx, f.apply, f.opID),
+		"an operation-scoped sequential resume must complete without a parent-write refusal")
+
+	assert.Equal(t, state.Task.Completed, f.taskState(t, f.tasks[0]),
+		"the operation's task must be driven to completion")
+	f.requireParentUntouched(t)
+}
+
+// When the resume re-plan finds no remaining work for the operation's tasks,
+// the drive marks its own tasks completed and exits; deriving the operation row
+// and projecting the parent terminal is the operator's job. The drive must not
+// attempt the parent completed write itself — it holds no apply lease.
+func TestLocalClient_OperationResumeWithNoRemainingWorkLeavesParentToProjection(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.Pending})
+
+	require.NoError(t, f.client.ResumeApplyOperation(f.opCtx, f.apply, f.opID),
+		"a re-plan that finds no remaining work must exit cleanly under the operation lease")
+
+	assert.Equal(t, state.Task.Completed, f.taskState(t, f.tasks[0]),
+		"a task whose live schema already matches the reviewed target settles completed")
+	f.requireParentUntouched(t)
+}
+
+// When the engine rejects a task during an operation-scoped sequential resume,
+// the drive settles its own tasks — the rejected one failed, queued siblings
+// cancelled — and leaves the parent failed state to the operator's projection.
+func TestLocalClient_OperationResumeEngineFailureSettlesTasksNotParent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.Pending, state.Task.Pending})
+	f.eng.planChanges = []engine.SchemaChange{{
+		Namespace:    "testdb",
+		TableChanges: []engine.TableChange{{Table: "users", DDL: multiOpResumeDDL}},
+	}}
+	f.eng.rejectApply = true
+
+	require.NoError(t, f.client.ResumeApplyOperation(f.opCtx, f.apply, f.opID),
+		"an engine rejection settles the operation's tasks; the drive itself exits cleanly")
+
+	assert.Equal(t, state.Task.Failed, f.taskState(t, f.tasks[0]),
+		"the rejected task must settle failed with the engine's verdict")
+	assert.Equal(t, state.Task.Cancelled, f.taskState(t, f.tasks[1]),
+		"a queued sibling task behind a failed one must settle cancelled")
+	f.requireParentUntouched(t)
+}
+
+// A grouped-mode engine failure during an operation-scoped resume settles the
+// drive's own tasks as failed and leaves the parent to the operator's
+// projection — and the drive itself exits cleanly. The failure is already
+// durably recorded in the tasks; a returned error would read as a transient
+// drive failure that leaves the settled operation claimable on every operator
+// poll, instead of letting the claim loop persist the operation row from its
+// now-failed tasks immediately.
+func TestLocalClient_OperationGroupedResumeFailureSettlesTasksAndExitsCleanly(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.Pending})
+	f.apply.Options = []byte(`{"defer_cutover":"true"}`)
+	f.eng.planChanges = []engine.SchemaChange{{
+		Namespace:    "testdb",
+		TableChanges: []engine.TableChange{{Table: "users", DDL: multiOpResumeDDL}},
+	}}
+	f.eng.rejectApply = true
+
+	require.NoError(t, f.client.ResumeApplyOperation(f.opCtx, f.apply, f.opID),
+		"a non-retryable grouped failure settles the operation's tasks; the drive itself exits cleanly")
+
+	assert.Equal(t, state.Task.Failed, f.taskState(t, f.tasks[0]),
+		"a grouped failure must settle the driven task as failed")
+	f.requireParentUntouched(t)
+}
+
+// A retryable engine failure during an operation-scoped grouped resume pauses
+// the drive's own tasks as failed_retryable and exits cleanly; deriving the
+// operation row and projecting the parent retryable state is the operator's
+// job, and the paused tasks keep the work re-dispatchable on a later attempt.
+func TestLocalClient_OperationGroupedResumeRetryableFailurePausesTasksNotParent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.Pending})
+	f.apply.Options = []byte(`{"defer_cutover":"true"}`)
+	f.eng.planChanges = []engine.SchemaChange{{
+		Namespace:    "testdb",
+		TableChanges: []engine.TableChange{{Table: "users", DDL: multiOpResumeDDL}},
+	}}
+	f.eng.applyErr = fmt.Errorf("copy phase lost its connection: connection reset")
+
+	require.NoError(t, f.client.ResumeApplyOperation(f.opCtx, f.apply, f.opID),
+		"a retryable grouped failure pauses the operation's tasks; the drive itself exits cleanly")
+
+	assert.Equal(t, state.Task.FailedRetryable, f.taskState(t, f.tasks[0]),
+		"a retryable engine failure must pause the driven task for operator recovery")
+	f.requireParentUntouched(t)
+}
+
+// A multi-operation apply parked at waiting_for_deploy is started by whichever
+// operation drive consumes the pending start request: the drive triggers the
+// engine deploy and drives its tasks, but the parent running state belongs to
+// the operator's projection and the start request stays pending so sibling
+// operations' deferred-deploy claims can still fire. A parent write here would
+// be refused by storage and abort the drive after the engine already accepted
+// the deploy, re-claiming and re-starting the deploy on every operator tick.
+func TestLocalClient_OperationDrivePendingStartTriggersDeployWithoutParentWrites(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.WaitingForDeploy})
+	ctx := t.Context()
+	_, err := f.leaseDB.ExecContext(ctx, "UPDATE `applies` SET `state` = ? WHERE `id` = ?",
+		state.Apply.WaitingForDeploy, f.apply.ID)
+	require.NoError(t, err)
+	f.apply.State = state.Apply.WaitingForDeploy
+	_, alreadyPending, err := f.stor.ControlRequests().RequestPending(ctx, &storage.ApplyControlRequest{
+		ApplyID:     f.apply.ID,
+		Operation:   storage.ControlOperationStart,
+		Status:      storage.ControlRequestPending,
+		RequestedBy: "integration-test",
+	})
+	require.NoError(t, err)
+	require.False(t, alreadyPending)
+
+	require.NoError(t, f.client.ResumeApplyOperation(f.opCtx, f.apply, f.opID),
+		"consuming a pending start under the operation lease must not abort on a parent-write refusal")
+
+	assert.Equal(t, state.Task.Completed, f.taskState(t, f.tasks[0]),
+		"the started deploy must drive the operation's task to completion")
+	parent, err := f.stor.Applies().Get(ctx, f.apply.ID)
+	require.NoError(t, err)
+	require.NotNil(t, parent)
+	assert.Equal(t, state.Apply.WaitingForDeploy, parent.State,
+		"the parent applies row is the projection's to advance; the drive must not write it running")
+	startReq, err := f.stor.ControlRequests().GetPending(ctx, f.apply.ID, storage.ControlOperationStart)
+	require.NoError(t, err)
+	assert.NotNil(t, startReq,
+		"the start request must stay pending so sibling operations' deferred-deploy claims can still fire")
+}
+
+// A stop that races the engine's own completion on an operation-scoped drive
+// settles the drive's tasks to the engine's completed truth and accepts the
+// stop; the parent completed write is the operator's projection to make. An
+// attempted parent write would be refused by storage and turn the accepted
+// settle into a drive error the claim loop re-runs forever against an engine
+// that will reject the stop the same way every time.
+func TestLocalClient_OperationStopAgainstCompletedEngineSettlesTasksNotParent(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.Running})
+	f.eng.stopErr = engine.NewAlreadyCompletedError("schema change already completed on the engine")
+
+	resp, err := f.client.stopOwnedApply(f.opCtx, &ternv1.StopRequest{
+		ApplyId:     f.apply.ApplyIdentifier,
+		Environment: f.apply.Environment,
+	}, "integration-test")
+	require.NoError(t, err,
+		"a stop racing engine completion must settle, not error, under the operation lease")
+	require.NotNil(t, resp)
+	assert.True(t, resp.Accepted, "the settle resolves the stop as accepted")
+
+	assert.Equal(t, state.Task.Completed, f.taskState(t, f.tasks[0]),
+		"the task must adopt the engine's completed outcome")
+	f.requireParentUntouched(t)
+}
+
+// When an operation-scoped drive finds a pending cancel but the engine's
+// authoritative backend already reports the change terminal, the drive adopts
+// the engine's outcome onto its own tasks and exits; the parent terminal write
+// and the mooted cancel request's completion belong to the operator's
+// projection, so the request stays pending until the projection resolves it.
+func TestLocalClient_OperationDriveAdoptsEngineTerminalTruthWithoutParentWrites(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test in short mode")
+	}
+	f := newMultiOpResumeFixture(t, []string{state.Task.Running})
+	f.eng.authoritativeProgress = true
+	ctx := t.Context()
+	_, alreadyPending, err := f.stor.ControlRequests().RequestPending(ctx, &storage.ApplyControlRequest{
+		ApplyID:     f.apply.ID,
+		Operation:   storage.ControlOperationCancel,
+		Status:      storage.ControlRequestPending,
+		RequestedBy: "integration-test",
+	})
+	require.NoError(t, err)
+	require.False(t, alreadyPending)
+
+	require.NoError(t, f.client.ResumeApplyOperation(f.opCtx, f.apply, f.opID),
+		"adopting the engine's terminal truth must exit cleanly under the operation lease")
+
+	assert.Equal(t, state.Task.Completed, f.taskState(t, f.tasks[0]),
+		"the task must adopt the engine's completed outcome before the cancel is consumed")
+	f.requireParentUntouched(t)
+	cancelReq, err := f.stor.ControlRequests().GetPending(ctx, f.apply.ID, storage.ControlOperationCancel)
+	require.NoError(t, err)
+	assert.NotNil(t, cancelReq,
+		"the mooted cancel request is the projection's to complete once the parent settles")
+}
diff --git a/pkg/tern/local_control_reconcile.go b/pkg/tern/local_control_reconcile.go
index 6dc57d67a..a7f592d91 100644
--- a/pkg/tern/local_control_reconcile.go
+++ b/pkg/tern/local_control_reconcile.go
@@ -103,6 +103,12 @@ func (c *LocalClient) reconcileEngineTerminalTruthBeforeCommands(ctx context.Con
 		return false, err
 	}
 	metrics.RecordEngineTerminalTruthReconcile(ctx, apply.Database, apply.Deployment, apply.Environment, "adopted_"+applyState)
+	// A multi-operation drive owns only its operation: once the operator's
+	// projection settles the parent terminal from the adopted task states, it
+	// also completes the mooted pending commands and posts the terminal summary.
+	if suppressParentApplyWrites(ctx) {
+		return true, nil
+	}
 	// The adopted terminal state moots the pending commands: the sweep
 	// completes the pending stop, and the pending cancel too for every adopted
 	// state (none of them is stopped, the one state that keeps a cancel
@@ -219,6 +225,15 @@ func (c *LocalClient) adoptEngineTerminalTruth(ctx context.Context, apply *stora
 			append(apply.LogAttrs(), "engine_state", string(progress.State), "requested_by", requestedBy)...)
 		return nil
 	}
+	// A multi-operation drive owns only its operation: the tasks settled above
+	// carry the adopted outcome, the operator derives the operation row from
+	// them and projects the parent, so the parent terminal write is the
+	// operator's to make.
+	if suppressParentApplyWrites(ctx) {
+		c.logger.Info("operation drive adopted the engine's terminal outcome onto its tasks; operator derives the operation row and projects the parent",
+			append(apply.LogAttrs(), "engine_state", string(progress.State), "adopted_state", applyState, "requested_by", requestedBy)...)
+		return nil
+	}
 	previousState := apply.State
 	apply.State = applyState
 	apply.CompletedAt = &now
diff --git a/pkg/tern/local_control_resume.go b/pkg/tern/local_control_resume.go
index 0d7c6d293..ce8bdfd86 100644
--- a/pkg/tern/local_control_resume.go
+++ b/pkg/tern/local_control_resume.go
@@ -270,15 +270,26 @@ func (c *LocalClient) processPendingStartControlRequest(ctx context.Context, app
 	if apply.StartedAt == nil {
 		apply.StartedAt = &now
 	}
-	if err := c.storage.Applies().Update(ctx, apply); err != nil {
-		return true, fmt.Errorf("update started deferred deploy apply %s: %w", apply.ApplyIdentifier, err)
-	}
-	if err := completePendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart); err != nil {
-		return true, err
+	// A multi-operation drive owns only its operation: the parent running
+	// write is the operator's projection to make — a direct write here fails
+	// closed under the operation-only lease and would abort a deploy the
+	// engine has already accepted. The start request also stays pending, so
+	// sibling operations' deferred-deploy claims can still fire; the claim arm
+	// closes once the projection moves the parent out of waiting_for_deploy.
+	if suppressParentApplyWrites(ctx) {
+		logger.Info("pending start request accepted under operation lease; parent running state is the operator's projection and the request stays pending for sibling operations",
+			"requested_by", controlRequestCaller(controlReq))
+	} else {
+		if err := c.storage.Applies().Update(ctx, apply); err != nil {
+			return true, fmt.Errorf("update started deferred deploy apply %s: %w", apply.ApplyIdentifier, err)
+		}
+		if err := completePendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart); err != nil {
+			return true, err
+		}
+		logger.Info("pending start request accepted and completed",
+			"requested_by", controlRequestCaller(controlReq),
+			"state", apply.State)
 	}
-	logger.Info("pending start request accepted and completed",
-		"requested_by", controlRequestCaller(controlReq),
-		"state", apply.State)
 	c.pollForCompletionAtomic(ctx, apply, started.tasks, started.credentials, started.resumeState, options, releaseAtCutoverBarrier)
 	return true, ctx.Err()
 }
@@ -1613,6 +1624,12 @@ func (c *LocalClient) resumeApplyWithTasks(ctx context.Context, apply *storage.A
 		logger.Error("deferred cutover recovery cannot reconcile absent cutover signal",
 			"active_task_count", len(activeTasks))
 		c.failApplyWithTasks(ctx, apply, activeTasks, message)
+		// A multi-operation drive owns only its operation; the operator's
+		// projection settles the parent and posts the terminal summary.
+		// failApplyWithTasks already logged the suppressed settle.
+		if suppressParentApplyWrites(ctx) {
+			return nil
+		}
 		c.notifyTerminalObserver(apply, tasks)
 		return nil
 	}
@@ -1621,6 +1638,7 @@ func (c *LocalClient) resumeApplyWithTasks(ctx context.Context, apply *storage.A
 		return err
 	}
 	startRequested := startControlReq != nil
+	suppressParent := suppressParentApplyWrites(ctx)
 
 	if len(activeTasks) == 0 {
 		logger.Info("all tasks already completed, marking apply as completed")
@@ -1628,6 +1646,14 @@ func (c *LocalClient) resumeApplyWithTasks(ctx context.Context, apply *storage.A
 		apply.State = state.Apply.Completed
 		apply.CompletedAt = &now
 		apply.UpdatedAt = now
+		// A multi-operation drive owns only its operation: its tasks are already
+		// terminal, so the operator derives the operation row completed and
+		// projects the parent — the parent write, start-request completion, and
+		// terminal observer are the operator's to make.
+		if suppressParent {
+			logger.Info("operation drive found no remaining work; operator derives the operation row and projects the parent")
+			return nil
+		}
 		if err := c.storage.Applies().Update(ctx, apply); err != nil {
 			return fmt.Errorf("mark resumed apply %s completed after re-plan found no remaining work: %w", apply.ApplyIdentifier, err)
 		}
@@ -1661,18 +1687,25 @@ func (c *LocalClient) resumeApplyWithTasks(ctx context.Context, apply *storage.A
 		now := time.Now()
 		apply.State = state.Apply.Running
 		apply.UpdatedAt = now
-		if err := c.storage.Applies().Update(ctx, apply); err != nil {
-			logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
-			return fmt.Errorf("mark sequential resume apply %s running: %w", apply.ApplyIdentifier, err)
-		}
-		if startRequested {
-			if err := completePendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart); err != nil {
-				return err
+		// A multi-operation drive does not write the parent running state or
+		// complete parent start requests; the operator projected the parent
+		// running before the drive. Task state is persisted per task below.
+		if suppressParent {
+			logger.Info("sequential resume under operation lease; parent running state is the operator's projection")
+		} else {
+			if err := c.storage.Applies().Update(ctx, apply); err != nil {
+				logger.Error("failed to update apply state", append(apply.MutableLogAttrs(), "error", err)...)
+				return fmt.Errorf("mark sequential resume apply %s running: %w", apply.ApplyIdentifier, err)
+			}
+			if startRequested {
+				if err := completePendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart); err != nil {
+					return err
+				}
 			}
-		}
 
-		c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelInfo, storage.LogEventStateTransition, storage.LogSourceSchemaBot,
-			"Apply resumed from checkpoint (sequential)", "", state.Apply.Running)
+			c.logApplyEvent(ctx, apply.ID, nil, storage.LogLevelInfo, storage.LogEventStateTransition, storage.LogSourceSchemaBot,
+				"Apply resumed from checkpoint (sequential)", "", state.Apply.Running)
+		}
 
 		resumeCtx, cancelResume := context.WithCancel(ctx)
 		cancelGeneration := c.setApplyCancel(cancelResume)
@@ -1696,6 +1729,17 @@ func (c *LocalClient) handleGroupedResumeFailure(ctx context.Context, apply *sto
 	logger.Error("engine apply failed during recovery",
 		"error", err)
 	c.failApplyWithTasks(ctx, apply, tasks, err.Error())
+	// A multi-operation drive owns only its operation: its failed tasks carry
+	// the outcome, and the operator's projection settles the parent, resolves
+	// pending control requests, and posts the terminal summary. The drive
+	// itself returns nil — the failure is already durably settled in the
+	// tasks, and an error here would read as a transient drive failure that
+	// leaves the operation claimable, re-leasing already-settled work instead
+	// of letting the claim loop persist the operation row from its now-failed
+	// tasks immediately. failApplyWithTasks already logged the suppressed settle.
+	if suppressParentApplyWrites(ctx) {
+		return nil
+	}
 	if startRequested {
 		if failErr := failPendingControlRequests(ctx, c.storage, apply, storage.ControlOperationStart, err.Error()); failErr != nil {
 			return failErr