refactor(flow): Track current operation run phase

rest-api/flow/internal/converter/dao/converter.go

@@ -337,6 +337,7 @@ func OperationRunFrom(dao *model.OperationRun) *operationrun.OperationRun {
 		Status:            dao.Status,
 		StatusReason:      dao.StatusReason,
 		StatusMessage:     dao.StatusMessage,
+		CurrentPhaseIndex: dao.CurrentPhaseIndex,
 		Selector:          dao.Selector,
 		Options:           dao.Options,
 		OperationTemplate: dao.OperationTemplate,
@@ -362,6 +363,7 @@ func OperationRunTo(run *operationrun.OperationRun) *model.OperationRun {
 		Status:            run.Status,
 		StatusReason:      run.StatusReason,
 		StatusMessage:     run.StatusMessage,
+		CurrentPhaseIndex: run.CurrentPhaseIndex,
 		Selector:          run.Selector,
 		Options:           run.Options,
 		OperationTemplate: run.OperationTemplate,

rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.down.sql

@@ -0,0 +1,6 @@
+-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+-- SPDX-License-Identifier: Apache-2.0
+
+ALTER TABLE operation_run
+    DROP COLUMN IF EXISTS current_phase_index;
+

rest-api/flow/internal/db/migrations/20260713000000_operation_run_current_phase_index.up.sql

@@ -0,0 +1,21 @@
+-- SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+-- SPDX-License-Identifier: Apache-2.0
+
+ALTER TABLE operation_run
+    ADD COLUMN current_phase_index INTEGER NOT NULL DEFAULT 0 CHECK (current_phase_index >= 0);
+
+UPDATE operation_run AS orun
+SET current_phase_index = COALESCE(
+    (
+        SELECT MIN(ort.phase_index)
+        FROM operation_run_target AS ort
+        WHERE ort.operation_run_id = orun.id
+          AND ort.status NOT IN ('completed', 'failed', 'terminated', 'skipped')
+    ),
+    (
+        SELECT MAX(ort.phase_index)
+        FROM operation_run_target AS ort
+        WHERE ort.operation_run_id = orun.id
+    ),
+    0
+);

rest-api/flow/internal/db/model/operation_run.go

@@ -25,6 +25,7 @@ type OperationRun struct {
 	Status            operationrun.OperationRunStatus       `bun:"status,type:varchar(32),notnull"`        //nolint:lll
 	StatusReason      operationrun.OperationRunStatusReason `bun:"status_reason,type:varchar(64),notnull"` //nolint:lll
 	StatusMessage     string                                `bun:"status_message,nullzero"`
+	CurrentPhaseIndex int32                                 `bun:"current_phase_index,notnull"`
 	Selector          json.RawMessage                       `bun:"selector,type:jsonb,notnull"`
 	Options           json.RawMessage                       `bun:"options,type:jsonb,notnull"`
 	OperationTemplate json.RawMessage                       `bun:"operation_template,type:jsonb,notnull"`

rest-api/flow/internal/operationrun/manager/dispatcher/decision.go

@@ -139,7 +139,7 @@ func (d *Dispatcher) decide(
 		return newStopDecision(dispatchRunTransition{}), nil
 	}
 
-	if prep.summary.TargetCount == 0 {
+	if prep.summary.SelectedTargetCount() == 0 {
 		return newStopDecision(failRunTransition("operation run has no targets")), nil
 	}
 
@@ -176,7 +176,7 @@ func (d *Dispatcher) decide(
 	}
 
 	safetyDecision := prep.safetyPolicy.evaluate(
-		prep.summary.TargetPhaseSummary,
+		prep.summary,
 	)
 	if safetyDecision.pause {
 		return newStopDecision(
@@ -185,20 +185,30 @@ func (d *Dispatcher) decide(
 	}
 
 	phaseDecision := prep.phasePolicy.evaluate(
-		prep.summary.TargetPhaseSummary,
-		prep.summary.previousPhaseTerminalChanged(),
+		prep.summary,
 	)
-	if phaseDecision.pause {
+	switch phaseDecision.action {
+	case phaseDecisionActionAdvance:
+		// The next phase's targets were not locked during this transaction.
+		// Persist the phase pointer first; the next pass will lock and submit them.
+		prep.run.CurrentPhaseIndex++
+		return newStopDecision(startRunTransition(phaseDecision.message)), nil
+	case phaseDecisionActionPause:
 		return newStopDecision(
 			pauseRunTransition(phaseDecision.reason, phaseDecision.message),
 		), nil
+	case phaseDecisionActionClaim:
+		return newClaimDecision(
+			startRunTransition(phaseDecision.message),
+			prep.options,
+			prep.op,
+			prep.conflictPolicy,
+			targets,
+		), nil
+	default:
+		return dispatchDecision{}, fmt.Errorf(
+			"unknown phase decision action %d",
+			phaseDecision.action,
+		)
 	}
-
-	return newClaimDecision(
-		startRunTransition(phaseDecision.message),
-		prep.options,
-		prep.op,
-		prep.conflictPolicy,
-		targets,
-	), nil
 }

rest-api/flow/internal/operationrun/manager/dispatcher/dependencies.go

@@ -19,7 +19,8 @@ type Store interface {
 	RunInTransaction(ctx context.Context, fn func(ctx context.Context) error) error
 	FetchRunnableIDs(ctx context.Context, limit int) ([]uuid.UUID, error)
 	LockRunnable(ctx context.Context, id uuid.UUID) (*operationrun.OperationRun, error)
-	LockOperationRunTargets(ctx context.Context, runID uuid.UUID) ([]*operationrun.OperationRunTarget, error)
+	LockOperationRunTargets(ctx context.Context, runID uuid.UUID, phaseIndex int32) ([]*operationrun.OperationRunTarget, error)
+	GetTargetPhaseAggregate(ctx context.Context, runID uuid.UUID, currentPhaseIndex int32) (operationrun.TargetPhaseAggregate, error)
 	UpdateRunState(ctx context.Context, run *operationrun.OperationRun) error
 	UpdateTargetState(ctx context.Context, target *operationrun.OperationRunTarget) error
 }

rest-api/flow/internal/operationrun/manager/dispatcher/dispatcher_test.go

@@ -224,7 +224,7 @@ func TestDispatchRunReconcilesCompletedTaskAndCompletesRun(t *testing.T) {
 	require.Equal(t, now, *store.run.FinishedAt)
 }
 
-func TestDispatchRunFailsWhenTerminalChangesSpanMultiplePhases(t *testing.T) {
+func TestDispatchRunLocksOnlyCurrentPhaseTargets(t *testing.T) {
 	runID := uuid.New()
 	firstTaskID := uuid.New()
 	secondTaskID := uuid.New()
@@ -261,8 +261,11 @@ func TestDispatchRunFailsWhenTerminalChangesSpanMultiplePhases(t *testing.T) {
 	}, now)
 
 	err := dispatcher.dispatchRun(context.Background(), runID)
-	require.ErrorContains(t, err, "terminal target changes span multiple phases")
-	require.Equal(t, operationrun.OperationRunStatusRunning, store.run.Status)
+	require.NoError(t, err)
+	require.Equal(t, operationrun.OperationRunTargetStatusCompleted, first.Status)
+	require.Equal(t, operationrun.OperationRunTargetStatusSubmitted, second.Status)
+	require.Equal(t, operationrun.OperationRunStatusPaused, store.run.Status)
+	require.Equal(t, operationrun.OperationRunStatusReasonPhaseGate, store.run.StatusReason)
 }
 
 func TestDispatchRunPausesWhenSafetyGateTrips(t *testing.T) {
@@ -624,13 +627,11 @@ func TestDecideSetsTerminalStatusFromTargetOutcomes(t *testing.T) {
 			run := &operationrun.OperationRun{
 				Status: operationrun.OperationRunStatusRunning,
 			}
-			summary := newReconciliationSummary()
-			summary.CurrentPhaseIndex = -1
-			summary.TargetCount = len(tt.statuses)
+			summary := operationrun.TargetPhaseSummary{}
+			summary.TotalPhases = 1
+			summary.CurrentPhaseStats.SelectedTargets = len(tt.statuses)
 			for _, status := range tt.statuses {
-				if status.IsFailedOrTerminated() {
-					summary.FailedOrTerminatedTargetCount++
-				}
+				summary.CurrentPhaseStats.StatusCounts.Add(status)
 			}
 
 			decision, err := (&Dispatcher{}).decide(&preparedDispatch{
@@ -696,8 +697,37 @@ func (s *fakeStore) LockRunnable(
 func (s *fakeStore) LockOperationRunTargets(
 	ctx context.Context,
 	runID uuid.UUID,
+	phaseIndex int32,
 ) ([]*operationrun.OperationRunTarget, error) {
-	return s.targets, nil
+	targets := make([]*operationrun.OperationRunTarget, 0)
+	for _, target := range s.targets {
+		if target.PhaseIndex == phaseIndex {
+			targets = append(targets, target)
+		}
+	}
+	return targets, nil
+}
+
+func (s *fakeStore) GetTargetPhaseAggregate(
+	ctx context.Context,
+	runID uuid.UUID,
+	currentPhaseIndex int32,
+) (operationrun.TargetPhaseAggregate, error) {
+	var totalPhases int32
+	completedStats := operationrun.PhaseStats{PhaseIndex: max(currentPhaseIndex-1, 0)}
+	for _, target := range s.targets {
+		if target.PhaseIndex+1 > totalPhases {
+			totalPhases = target.PhaseIndex + 1
+		}
+		if target.PhaseIndex < currentPhaseIndex {
+			completedStats.AddTarget(target)
+		}
+	}
+
+	return operationrun.TargetPhaseAggregate{
+		TotalPhases:         totalPhases,
+		CompletedPhaseStats: completedStats,
+	}, nil
 }
 
 func (s *fakeStore) UpdateRunState(
@@ -759,10 +789,19 @@ func (s *dispatchOnceStore) LockRunnable(
 func (s *dispatchOnceStore) LockOperationRunTargets(
 	ctx context.Context,
 	runID uuid.UUID,
+	phaseIndex int32,
 ) ([]*operationrun.OperationRunTarget, error) {
 	return nil, nil
 }
 
+func (s *dispatchOnceStore) GetTargetPhaseAggregate(
+	ctx context.Context,
+	runID uuid.UUID,
+	currentPhaseIndex int32,
+) (operationrun.TargetPhaseAggregate, error) {
+	return operationrun.TargetPhaseAggregate{}, nil
+}
+
 func (s *dispatchOnceStore) UpdateRunState(
 	ctx context.Context,
 	run *operationrun.OperationRun,

rest-api/flow/internal/operationrun/manager/dispatcher/phase.go

@@ -13,6 +13,20 @@ type phasePolicyRuntime struct {
 	autoAdvance bool
 }
 
+type phaseDecisionAction int
+
+const (
+	phaseDecisionActionClaim phaseDecisionAction = iota
+	phaseDecisionActionAdvance
+	phaseDecisionActionPause
+)
+
+type phaseDecision struct {
+	action  phaseDecisionAction
+	reason  operationrun.OperationRunStatusReason
+	message string
+}
+
 func newPhasePolicy(options *operationrun.Options) (*phasePolicyRuntime, error) {
 	if err := options.PhasePolicy.Validate(); err != nil {
 		return nil, fmt.Errorf("phase policy: %w", err)
@@ -25,27 +39,22 @@ func newPhasePolicy(options *operationrun.Options) (*phasePolicyRuntime, error)
 
 func (p phasePolicyRuntime) evaluate(
 	summary operationrun.TargetPhaseSummary,
-	previousPhaseTerminalChanged bool,
-) pauseDecision {
-	if summary.CurrentPhaseIndex == 0 ||
-		!previousPhaseTerminalChanged ||
-		!summary.CurrentPhaseNotStarted() {
-		// No phase boundary was crossed into a fresh phase, so there is
-		// nothing for phase policy to pause or report.
-		return pauseDecision{
-			pause: false,
+) phaseDecision {
+	if !summary.CurrentPhaseTerminal() || !summary.HasNextPhase() {
+		return phaseDecision{
+			action: phaseDecisionActionClaim,
 		}
 	}
 
 	if p.autoAdvance {
-		return pauseDecision{
-			pause:   false,
+		return phaseDecision{
+			action:  phaseDecisionActionAdvance,
 			message: "advanced to next phase",
 		}
 	}
 
-	return pauseDecision{
-		pause:   true,
+	return phaseDecision{
+		action:  phaseDecisionActionPause,
 		reason:  operationrun.OperationRunStatusReasonPhaseGate,
 		message: "waiting for phase advance",
 	}

rest-api/flow/internal/operationrun/manager/dispatcher/preparation.go

@@ -22,7 +22,7 @@ type preparedDispatch struct {
 	safetyPolicy   *safetyPolicyRuntime
 	phasePolicy    *phasePolicyRuntime
 	prepareErr     error
-	summary        reconciliationSummary
+	summary        operationrun.TargetPhaseSummary
 	changed        map[uuid.UUID]*operationrun.OperationRunTarget
 }
 
@@ -82,18 +82,31 @@ func (d *Dispatcher) prepare(
 		return prep, nil
 	}
 
-	// TODO: After operation_run.current_phase_index is persisted, lock only the
-	// current phase targets here and use SQL aggregates for cumulative safety
-	// gates and final completion instead of loading every target for the run.
-	targets, err := d.deps.Store.LockOperationRunTargets(ctx, run.ID)
+	targets, err := d.deps.Store.LockOperationRunTargets(
+		ctx,
+		run.ID,
+		run.CurrentPhaseIndex,
+	)
 	if err != nil {
 		return nil, fmt.Errorf("lock operation run targets: %w", err)
 	}
 
-	prep.summary, err = d.reconcileTargets(ctx, targets, prep.changed)
-	if err != nil {
+	if err := d.reconcileTargets(ctx, targets, prep.changed); err != nil {
 		return nil, err
 	}
+	aggregate, err := d.deps.Store.GetTargetPhaseAggregate(
+		ctx,
+		run.ID,
+		run.CurrentPhaseIndex,
+	)
+	if err != nil {
+		return nil, fmt.Errorf("summarize operation run targets: %w", err)
+	}
+	prep.summary = operationrun.NewTargetPhaseSummary(
+		run.CurrentPhaseIndex,
+		aggregate,
+		targets,
+	)
 
 	return prep, nil
 }