diff --git a/distribution/migration_promotion_complete.go b/distribution/migration_promotion_complete.go
new file mode 100644
index 000000000..88ee72b14
--- /dev/null
+++ b/distribution/migration_promotion_complete.go
@@ -0,0 +1,230 @@
+package distribution
+
+import (
+	"bytes"
+	"context"
+	"math"
+
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+)
+
+var (
+	ErrMigrationPromotionNotReady     = errors.New("migration target promotion is not ready")
+	ErrMigrationPromotionTargetAbsent = errors.New("migration target promotion route is missing")
+)
+
+// TargetPromotionCompletion is the result of applying the default-group
+// promotion-complete transition to a SplitJob and its catalog routes.
+type TargetPromotionCompletion struct {
+	Job             SplitJob
+	Routes          []RouteDescriptor
+	Changed         bool
+	ClearedRouteIDs []uint64
+}
+
+// CompleteTargetPromotionState clears the target route's staged visibility
+// fields after target-local promotion has completed. It deliberately preserves
+// MinWriteTSExclusive; the timestamp floor remains a durable route invariant
+// after the staged/live merge is no longer needed.
+func CompleteTargetPromotionState(job SplitJob, routes []RouteDescriptor, nowMs int64) (TargetPromotionCompletion, error) {
+	normalized, err := normalizePromotionCompletionInput(job, routes)
+	if err != nil {
+		return TargetPromotionCompletion{}, err
+	}
+
+	out := TargetPromotionCompletion{
+		Job:    CloneSplitJob(job),
+		Routes: normalized,
+	}
+	if out.Job.TargetPromotionDone {
+		if targetClearedDescriptorPresent(out.Job, out.Routes) {
+			return out, nil
+		}
+		return TargetPromotionCompletion{}, errors.WithStack(ErrMigrationPromotionTargetAbsent)
+	}
+
+	cleared, err := clearTargetPromotionRoutes(job, out.Routes)
+	if err != nil {
+		return TargetPromotionCompletion{}, err
+	}
+	if len(cleared) == 0 {
+		return TargetPromotionCompletion{}, errors.WithStack(ErrMigrationPromotionTargetAbsent)
+	}
+
+	out.Changed = true
+	out.ClearedRouteIDs = cleared
+	out.Job.TargetPromotionDone = true
+	out.Job.UpdatedAtMs = nowMs
+	return out, nil
+}
+
+func normalizePromotionCompletionInput(job SplitJob, routes []RouteDescriptor) ([]RouteDescriptor, error) {
+	if err := validateSplitJob(job); err != nil {
+		return nil, err
+	}
+	if job.Phase != SplitJobPhaseCleanup {
+		return nil, errors.WithStack(ErrMigrationPromotionNotReady)
+	}
+	return normalizeRoutes(routes)
+}
+
+func clearTargetPromotionRoutes(job SplitJob, routes []RouteDescriptor) ([]uint64, error) {
+	cleared := make([]uint64, 0, 1)
+	for i := range routes {
+		route := &routes[i]
+		if route.MigrationJobID != job.JobID {
+			continue
+		}
+		if !route.StagedVisibilityActive ||
+			route.GroupID != job.TargetGroupID ||
+			route.ParentRouteID != job.SourceRouteID ||
+			!bytes.Equal(route.Start, job.SplitKey) {
+			return nil, errors.WithStack(ErrMigrationInvalidRoute)
+		}
+		route.StagedVisibilityActive = false
+		route.MigrationJobID = 0
+		cleared = append(cleared, route.RouteID)
+	}
+	return cleared, nil
+}
+
+// CompleteSplitJobTargetPromotion applies the promotion-complete catalog CAS:
+// route descriptor staged fields are cleared, catalog version is bumped, and
+// the SplitJob witness is updated in the same MVCC batch.
+func (s *CatalogStore) CompleteSplitJobTargetPromotion(
+	ctx context.Context,
+	expectedVersion uint64,
+	expected SplitJob,
+	nowMs int64,
+) (CatalogSnapshot, SplitJob, error) {
+	if err := ensureCatalogStore(s); err != nil {
+		return CatalogSnapshot{}, SplitJob{}, err
+	}
+	ctx = contextOrBackground(ctx)
+
+	readTS, currentVersion, routes, err := s.loadPromotionCompleteInputs(ctx, expectedVersion, expected)
+	if err != nil {
+		return CatalogSnapshot{}, SplitJob{}, err
+	}
+	completion, err := CompleteTargetPromotionState(expected, routes, nowMs)
+	if err != nil {
+		return CatalogSnapshot{}, SplitJob{}, err
+	}
+	if !completion.Changed {
+		return CatalogSnapshot{
+			Version: currentVersion,
+			Routes:  cloneRouteDescriptors(completion.Routes),
+			ReadTS:  readTS,
+		}, completion.Job, nil
+	}
+
+	plan, mutations, commitTS, err := s.buildPromotionCompleteMutations(ctx, readTS, expectedVersion, expected.JobID, &completion)
+	if err != nil {
+		return CatalogSnapshot{}, SplitJob{}, err
+	}
+	if err := s.applyPromotionCompleteMutations(ctx, plan, mutations, expected.JobID, commitTS); err != nil {
+		return CatalogSnapshot{}, SplitJob{}, err
+	}
+
+	return CatalogSnapshot{
+		Version: plan.nextVersion,
+		Routes:  cloneRouteDescriptors(completion.Routes),
+	}, completion.Job, nil
+}
+
+func (s *CatalogStore) loadPromotionCompleteInputs(ctx context.Context, expectedVersion uint64, expected SplitJob) (uint64, uint64, []RouteDescriptor, error) {
+	expectedRaw, err := EncodeSplitJob(expected)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+	readTS := s.store.LastCommitTS()
+	currentVersion, err := s.versionAt(ctx, readTS)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+	if currentVersion != expectedVersion {
+		return 0, 0, nil, errors.WithStack(ErrCatalogVersionMismatch)
+	}
+	if err := s.expectLiveSplitJobAt(ctx, expected.JobID, expectedRaw, readTS); err != nil {
+		return 0, 0, nil, err
+	}
+	routes, err := s.routesAt(ctx, readTS)
+	if err != nil {
+		return 0, 0, nil, err
+	}
+	return readTS, currentVersion, routes, nil
+}
+
+func (s *CatalogStore) buildPromotionCompleteMutations(
+	ctx context.Context,
+	readTS uint64,
+	expectedVersion uint64,
+	jobID uint64,
+	completion *TargetPromotionCompletion,
+) (savePlan, []*store.KVPairMutation, uint64, error) {
+	if expectedVersion == math.MaxUint64 {
+		return savePlan{}, nil, 0, errors.WithStack(ErrCatalogVersionOverflow)
+	}
+	minCommitTS := readTS + 1
+	if minCommitTS == 0 {
+		return savePlan{}, nil, 0, errors.WithStack(ErrCatalogVersionOverflow)
+	}
+
+	plan := savePlan{
+		readTS:      readTS,
+		minCommitTS: minCommitTS,
+		nextVersion: expectedVersion + 1,
+		routes:      completion.Routes,
+	}
+	mutations, err := s.buildSaveMutations(ctx, &plan)
+	if err != nil {
+		return savePlan{}, nil, 0, err
+	}
+	commitTS, err := s.commitTSForApply(plan.minCommitTS)
+	if err != nil {
+		return savePlan{}, nil, 0, err
+	}
+	completion.Job.PromotionCompletedTS = commitTS
+	encodedJob, err := EncodeSplitJob(completion.Job)
+	if err != nil {
+		return savePlan{}, nil, 0, err
+	}
+	jobMutations, err := s.buildSplitJobPutMutations(ctx, readTS, CatalogSplitJobKey(jobID), encodedJob, jobID)
+	if err != nil {
+		return savePlan{}, nil, 0, err
+	}
+	return plan, append(mutations, jobMutations...), commitTS, nil
+}
+
+func (s *CatalogStore) applyPromotionCompleteMutations(ctx context.Context, plan savePlan, mutations []*store.KVPairMutation, jobID uint64, commitTS uint64) error {
+	readKeys := [][]byte{
+		CatalogVersionKey(),
+		CatalogSplitJobKey(jobID),
+	}
+	if err := s.store.ApplyMutations(ctx, mutations, readKeys, plan.readTS, commitTS); err != nil {
+		if errors.Is(err, store.ErrWriteConflict) {
+			return errors.WithStack(ErrCatalogSplitJobConflict)
+		}
+		return errors.WithStack(err)
+	}
+	return nil
+}
+
+func targetClearedDescriptorPresent(job SplitJob, routes []RouteDescriptor) bool {
+	for _, route := range routes {
+		if route.GroupID != job.TargetGroupID {
+			continue
+		}
+		if route.StagedVisibilityActive || route.MigrationJobID != 0 {
+			continue
+		}
+		if route.ParentRouteID != job.SourceRouteID {
+			continue
+		}
+		if bytes.Equal(route.Start, job.SplitKey) {
+			return true
+		}
+	}
+	return false
+}
diff --git a/distribution/migration_promotion_complete_test.go b/distribution/migration_promotion_complete_test.go
new file mode 100644
index 000000000..b04045bfd
--- /dev/null
+++ b/distribution/migration_promotion_complete_test.go
@@ -0,0 +1,186 @@
+package distribution
+
+import (
+	"context"
+	"testing"
+
+	"github.com/bootjp/elastickv/store"
+	"github.com/cockroachdb/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCompleteTargetPromotionStateClearsStagedFieldsAndRetainsFloor(t *testing.T) {
+	t.Parallel()
+
+	job := promotionCompleteTestJob()
+	routes := promotionCompleteTestRoutes()
+
+	result, err := CompleteTargetPromotionState(job, routes, 1000)
+	require.NoError(t, err)
+	require.True(t, result.Changed)
+	require.Equal(t, []uint64{3}, result.ClearedRouteIDs)
+	require.True(t, result.Job.TargetPromotionDone)
+	require.Zero(t, result.Job.PromotionCompletedTS)
+	require.Equal(t, int64(1000), result.Job.UpdatedAtMs)
+	require.Equal(t, SplitJobPhaseCleanup, result.Job.Phase)
+
+	target := routeByID(t, result.Routes, 3)
+	require.False(t, target.StagedVisibilityActive)
+	require.Equal(t, uint64(0), target.MigrationJobID)
+	require.Equal(t, uint64(777), target.MinWriteTSExclusive)
+	require.Equal(t, uint64(42), target.ParentRouteID)
+
+	badRoutes := promotionCompleteTestRoutes()
+	badRoutes[1].ParentRouteID = 777
+	_, err = CompleteTargetPromotionState(job, badRoutes, 1000)
+	require.ErrorIs(t, err, ErrMigrationInvalidRoute)
+}
+
+func TestCompleteTargetPromotionStateAcceptsAlreadyClearedDescriptor(t *testing.T) {
+	t.Parallel()
+
+	job := promotionCompleteTestJob()
+	job.TargetPromotionDone = true
+	job.PromotionCompletedTS = 900
+	job.UpdatedAtMs = 1000
+	routes := promotionCompleteTestRoutes()
+	routes[1].StagedVisibilityActive = false
+	routes[1].MigrationJobID = 0
+
+	result, err := CompleteTargetPromotionState(job, routes, 1100)
+	require.NoError(t, err)
+	require.False(t, result.Changed)
+	require.Equal(t, uint64(900), result.Job.PromotionCompletedTS)
+	require.Equal(t, int64(1000), result.Job.UpdatedAtMs)
+
+	target := routeByID(t, result.Routes, 3)
+	require.False(t, target.StagedVisibilityActive)
+	require.Equal(t, uint64(0), target.MigrationJobID)
+	require.Equal(t, uint64(777), target.MinWriteTSExclusive)
+}
+
+func TestCatalogStoreCompleteSplitJobTargetPromotionCommitsRouteAndJobTogether(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	cs := NewCatalogStore(store.NewMVCCStore(), WithCatalogRouteDescriptorV2Writes(true))
+	saved, err := cs.Save(ctx, 0, promotionCompleteTestRoutes())
+	require.NoError(t, err)
+	job := promotionCompleteTestJob()
+	require.NoError(t, cs.CreateSplitJob(ctx, job))
+	before, err := cs.Snapshot(ctx)
+	require.NoError(t, err)
+
+	snapshot, completed, err := cs.CompleteSplitJobTargetPromotion(ctx, saved.Version, job, 1000)
+	require.NoError(t, err)
+	require.Equal(t, saved.Version+1, snapshot.Version)
+	require.True(t, completed.TargetPromotionDone)
+	require.Greater(t, completed.PromotionCompletedTS, before.ReadTS)
+	require.Equal(t, int64(1000), completed.UpdatedAtMs)
+
+	loadedJob, found, err := cs.SplitJob(ctx, job.JobID)
+	require.NoError(t, err)
+	require.True(t, found)
+	assertSplitJobEqual(t, completed, loadedJob)
+
+	loaded, err := cs.Snapshot(ctx)
+	require.NoError(t, err)
+	require.Equal(t, saved.Version+1, loaded.Version)
+	target := routeByID(t, loaded.Routes, 3)
+	require.False(t, target.StagedVisibilityActive)
+	require.Equal(t, uint64(0), target.MigrationJobID)
+	require.Equal(t, uint64(777), target.MinWriteTSExclusive)
+}
+
+func TestCatalogStoreCompleteSplitJobTargetPromotionIsIdempotentAfterClearedDescriptor(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	cs := NewCatalogStore(store.NewMVCCStore(), WithCatalogRouteDescriptorV2Writes(true))
+	routes := promotionCompleteTestRoutes()
+	routes[1].StagedVisibilityActive = false
+	routes[1].MigrationJobID = 0
+	saved, err := cs.Save(ctx, 0, routes)
+	require.NoError(t, err)
+	job := promotionCompleteTestJob()
+	job.TargetPromotionDone = true
+	job.PromotionCompletedTS = 900
+	job.UpdatedAtMs = 1000
+	require.NoError(t, cs.CreateSplitJob(ctx, job))
+
+	snapshot, completed, err := cs.CompleteSplitJobTargetPromotion(ctx, saved.Version, job, 1100)
+	require.NoError(t, err)
+	require.Equal(t, saved.Version, snapshot.Version)
+	assertSplitJobEqual(t, job, completed)
+
+	loaded, err := cs.Snapshot(ctx)
+	require.NoError(t, err)
+	require.Equal(t, saved.Version, loaded.Version)
+}
+
+func TestCatalogStoreCompleteSplitJobTargetPromotionRejectsStaleInputs(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	cs := NewCatalogStore(store.NewMVCCStore(), WithCatalogRouteDescriptorV2Writes(true))
+	saved, err := cs.Save(ctx, 0, promotionCompleteTestRoutes())
+	require.NoError(t, err)
+	job := promotionCompleteTestJob()
+	require.NoError(t, cs.CreateSplitJob(ctx, job))
+
+	_, _, err = cs.CompleteSplitJobTargetPromotion(ctx, saved.Version+1, job, 1000)
+	require.True(t, errors.Is(err, ErrCatalogVersionMismatch), "got %v", err)
+
+	advanced := job
+	advanced.Cursor = []byte("advanced")
+	advanced.UpdatedAtMs++
+	require.NoError(t, cs.SaveSplitJob(ctx, job, advanced))
+
+	_, _, err = cs.CompleteSplitJobTargetPromotion(ctx, saved.Version, job, 1000)
+	require.True(t, errors.Is(err, ErrCatalogSplitJobConflict), "got %v", err)
+}
+
+func promotionCompleteTestJob() SplitJob {
+	return SplitJob{
+		JobID:         99,
+		SourceRouteID: 42,
+		SplitKey:      []byte("m"),
+		TargetGroupID: 2,
+		Phase:         SplitJobPhaseCleanup,
+	}
+}
+
+func promotionCompleteTestRoutes() []RouteDescriptor {
+	return []RouteDescriptor{
+		{
+			RouteID:       2,
+			Start:         []byte(""),
+			End:           []byte("m"),
+			GroupID:       1,
+			State:         RouteStateActive,
+			ParentRouteID: 42,
+		},
+		{
+			RouteID:                3,
+			Start:                  []byte("m"),
+			End:                    nil,
+			GroupID:                2,
+			State:                  RouteStateActive,
+			ParentRouteID:          42,
+			StagedVisibilityActive: true,
+			MigrationJobID:         99,
+			MinWriteTSExclusive:    777,
+		},
+	}
+}
+
+func routeByID(t *testing.T, routes []RouteDescriptor, routeID uint64) RouteDescriptor {
+	t.Helper()
+	for _, route := range routes {
+		if route.RouteID == routeID {
+			return route
+		}
+	}
+	t.Fatalf("route %d not found", routeID)
+	return RouteDescriptor{}
+}