//go:build integration

package controlplane

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nexuscrm/db"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/pingcap-inc/nexuscrm-controlplane/internal/offlinecandidateauthority"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
)

// legacySourceBackedActivationPack models the pre-generation release: it
// preserves every source-backed pack declaration while deliberately hiding the
// new concrete activation hook from markInstalledPackActive.
type legacySourceBackedActivationPack struct {
	*packs.SourceBackedPack
}

func TestActiveGenerationCatalogDeclarationsTiDBUseCompleteAuthority(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	source := testControlplaneOfficialRepoResolvedSource("search-declaration-source-revision-1")
	source.Manifest.Components = append(source.Manifest.Components, packs.ManifestComponent{
		Kind: packs.ComponentKindObjectSpec, BusinessKey: "workspace", Manageability: packs.ManageabilityLocked,
		Payload: []byte(`[{"apiName":"Workspace","label":"Workspace","tableType":"custom_object","fields":[{"apiName":"Name","label":"Name","type":"string","isNameField":true}]}]`),
	}, packs.ManifestComponent{
		Kind: packs.ComponentKindSearchScopeOrder, BusinessKey: "installed_scope", Manageability: packs.ManageabilityLocked,
		Payload: []byte(`[{"name":"Workspace","label":"Workspace"},{"name":"Issue","label":"Issue"}]`),
	})
	pack, err := packs.NewSourceBackedPack(source)
	if err != nil {
		t.Fatal(err)
	}
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, db, pack, payloadHash); err != nil {
		t.Fatalf("mark installing: %v", err)
	}
	if err := markInstalledPackActive(
		ctx,
		db,
		legacySourceBackedActivationPack{SourceBackedPack: pack},
		payloadHash,
	); err != nil {
		t.Fatalf("mark legacy active: %v", err)
	}
	if err := publishFirstSourceBackedPackGeneration(
		ctx,
		db,
		appliedAppPack{pack: pack, payloadHash: payloadHash},
	); err != nil {
		t.Fatalf("publish immutable generation: %v", err)
	}

	read := func() error {
		tx, err := db.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		defer func() { _ = tx.Rollback() }()
		if _, err := ReadActiveRuntimeFormulaFunctionBundles(
			ctx,
			tx,
		); err != nil {
			return err
		}
		if _, err := ReadActiveRuntimeSearchAllDefaultFields(
			ctx,
			tx,
		); err != nil {
			return err
		}
		scope, err := ReadActiveRuntimeSearchScopeOrder(ctx, tx)
		if err != nil {
			return err
		}
		if len(scope) != 2 || scope[0].Name != "Workspace" || scope[1].Name != "Issue" {
			return fmt.Errorf("installed scope lost: %+v", scope)
		}
		return tx.Commit()
	}
	if err := read(); err != nil {
		t.Fatalf("read verified generation declarations: %v", err)
	}

	if _, err := db.ExecContext(ctx, `UPDATE _System_PackGeneration
SET DesiredPayloadHash = ?
WHERE PackName = ?`,
		packs.PackPayloadHashV1Prefix+
			"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		pack.Name(),
	); err != nil {
		t.Fatalf("corrupt desired payload identity: %v", err)
	}
	if err := read(); !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf(
			"corrupt generation declaration read error = %v, want generation conflict",
			err,
		)
	}
}

func TestSourceBackedPackGenerationTiDBRevisionBoundLocalRetryStaysMutable(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("new runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh runner cache: %v", err)
	}

	const (
		packName   = "phase7-local-retry"
		objectName = "Phase7LocalRetryObject"
	)
	sourceRoot, _ := writeNamedFieldDeclarationSource(
		t,
		packName,
		objectName,
		"RetryMarker__c",
	)
	initGitTestRepoForControlplane(t, sourceRoot)
	gitTestCommitForControlplane(t, sourceRoot, "phase-7 local retry source")
	pack, err := packs.ResolveInstalledSourceBackedPack(
		packName,
		packs.AppPack,
		packs.PackSourceKindLocalPath,
		sourceRoot,
	)
	if err != nil {
		t.Fatalf("resolve revision-bound local retry pack: %v", err)
	}
	meta := packs.PackHashMetadataForPack(pack)
	if !meta.SourceRevisionValid || strings.TrimSpace(meta.SourceRevision) == "" {
		t.Fatal("local retry fixture did not record Git revision provenance")
	}
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("local retry payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, conn.DB(), pack, payloadHash); err != nil {
		t.Fatalf("mark local retry installing: %v", err)
	}
	if err := markInstalledPackFailed(
		ctx,
		conn.DB(),
		pack,
		payloadHash,
		errors.New("stopped local attempt"),
	); err != nil {
		t.Fatalf("mark local retry failed: %v", err)
	}
	var originalInstallID string
	if err := conn.DB().QueryRowContext(
		ctx,
		`SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		packName,
	).Scan(&originalInstallID); err != nil {
		t.Fatalf("read local retry incarnation: %v", err)
	}

	if err := reconcileActiveInstalledPackSourceState(
		ctx,
		runner,
		InstallPackOptions{allowInstalledPackSourceStateRepair: true},
	); err != nil {
		t.Fatalf("resume revision-bound local retry: %v", err)
	}
	var (
		status          string
		activeInstallID string
		objectCount     int
	)
	if err := conn.DB().QueryRowContext(
		ctx,
		`SELECT Status, PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		packName,
	).Scan(&status, &activeInstallID); err != nil {
		t.Fatalf("read resumed local retry: %v", err)
	}
	if status != installedPackStatusActive || activeInstallID != originalInstallID {
		t.Fatalf(
			"resumed local retry = status %q incarnation %q, want active/%q",
			status,
			activeInstallID,
			originalInstallID,
		)
	}
	if err := conn.DB().QueryRowContext(
		ctx,
		`SELECT COUNT(*) FROM _System_Object WHERE ApiName = ?`,
		objectName,
	).Scan(&objectCount); err != nil {
		t.Fatalf("count resumed local retry object: %v", err)
	}
	if objectCount != 1 {
		t.Fatalf("resumed local retry object count = %d, want 1", objectCount)
	}
	assertNoPackGenerationAuthority(t, ctx, conn.DB(), packName)
}

func TestSourceBackedPackGenerationTiDBReconcileRejectsIncompleteActiveAuthorityBeforeRetryMutation(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	const (
		activeConnectorName = "legacy-connector-without-source-authority"
		retryPackName       = "phase7-retry-peer"
		retryObjectName     = "Phase7RetryObject"
	)
	retrySourceRoot, _ := writeNamedFieldDeclarationSource(
		t,
		retryPackName,
		retryObjectName,
		"RetryMarker__c",
	)
	initGitTestRepoForControlplane(t, retrySourceRoot)
	gitTestCommitForControlplane(t, retrySourceRoot, "phase-7 retry source")
	retryPack, err := packs.ResolveInstalledSourceBackedPack(
		retryPackName,
		packs.AppPack,
		packs.PackSourceKindLocalPath,
		retrySourceRoot,
	)
	if err != nil {
		t.Fatalf("resolve revision-bound retry pack: %v", err)
	}
	retryMeta := packs.PackHashMetadataForPack(retryPack)
	if !retryMeta.SourceRevisionValid || strings.TrimSpace(retryMeta.SourceRevision) == "" {
		t.Fatal("retry fixture did not resolve a revision-bound local source")
	}
	retryHash, err := packs.PackPayloadHash(retryPack)
	if err != nil {
		t.Fatalf("retry payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, db, retryPack, retryHash); err != nil {
		t.Fatalf("mark retry peer installing: %v", err)
	}
	if err := markInstalledPackFailed(
		ctx,
		db,
		retryPack,
		retryHash,
		errors.New("stopped phase-7 attempt"),
	); err != nil {
		t.Fatalf("mark retry peer failed: %v", err)
	}
	var retryInstallID string
	if err := db.QueryRowContext(
		ctx,
		`SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		retryPackName,
	).Scan(&retryInstallID); err != nil {
		t.Fatalf("read retry peer incarnation: %v", err)
	}

	if _, err := db.ExecContext(ctx, `INSERT INTO _System_InstalledPack (
Id, Name, PackInstallId, Category, Status, InstallSource, IsDeleted
) VALUES (?, ?, ?, ?, ?, 'apply', 0)`,
		installedPackID(activeConnectorName),
		activeConnectorName,
		"legacy-connector-installation",
		string(packs.ConnectorPack),
		installedPackStatusActive,
	); err != nil {
		t.Fatalf("seed active connector without source authority: %v", err)
	}

	err = reconcileActiveInstalledPackSourceState(
		ctx,
		&Runner{DB: conn},
		InstallPackOptions{allowInstalledPackSourceStateRepair: true},
	)
	if !errors.Is(err, ErrPackGenerationPrecondition) {
		t.Fatalf("phase-7 reconcile error = %v, want generation precondition failure", err)
	}
	if !strings.Contains(err.Error(), activeConnectorName) {
		t.Fatalf("phase-7 reconcile error = %v, want invalid active connector name", err)
	}

	var retryStatus, retryInstallIDAfter string
	if err := db.QueryRowContext(
		ctx,
		`SELECT Status, PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		retryPackName,
	).Scan(&retryStatus, &retryInstallIDAfter); err != nil {
		t.Fatalf("read retry peer after rejected reconcile: %v", err)
	}
	if retryStatus != installedPackStatusFailed || retryInstallIDAfter != retryInstallID {
		t.Fatalf(
			"retry peer after rejected reconcile = status %q incarnation %q, want failed/%q",
			retryStatus,
			retryInstallIDAfter,
			retryInstallID,
		)
	}
	var retryObjectCount int
	if err := db.QueryRowContext(
		ctx,
		`SELECT COUNT(*) FROM _System_Object WHERE ApiName = ?`,
		retryObjectName,
	).Scan(&retryObjectCount); err != nil {
		t.Fatalf("count retry-owned object after rejected reconcile: %v", err)
	}
	if retryObjectCount != 0 {
		t.Fatalf("retry-owned object count after rejected reconcile = %d, want 0", retryObjectCount)
	}
	assertNoPackGenerationAuthority(t, ctx, db, retryPackName)
}

func TestSourceBackedPackGenerationTiDBStoppedUpgradeAdoption(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()
	if err := schemaProvisioner.VerifySchemaReadiness(conn); err != nil {
		t.Fatalf("official InitL0 pack-generation schema readiness: %v", err)
	}

	productionPathPack := testControlplaneOfficialRepoSourceBackedPack(
		t,
		"production-path-source-revision-1",
	)
	productionPathHash, err := packs.PackPayloadHash(productionPathPack)
	if err != nil {
		t.Fatalf("production-path payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(
		ctx,
		db,
		productionPathPack,
		productionPathHash,
	); err != nil {
		t.Fatalf("production-path mark installing: %v", err)
	}
	if err := markInstalledPackActive(
		ctx,
		db,
		legacySourceBackedActivationPack{SourceBackedPack: productionPathPack},
		productionPathHash,
	); err != nil {
		t.Fatalf("seed pre-generation legacy active compatibility row: %v", err)
	}
	assertNoPackGenerationAuthority(t, ctx, db, productionPathPack.Name())
	var legacyPackInstallID string
	if err := db.QueryRowContext(
		ctx,
		`SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		productionPathPack.Name(),
	).Scan(&legacyPackInstallID); err != nil {
		t.Fatalf("read legacy active compatibility incarnation: %v", err)
	}
	differentPayloadHash := packs.PackPayloadHashV1Prefix +
		"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	if differentPayloadHash == productionPathHash {
		t.Fatal("test setup expected a different legacy applied payload hash")
	}
	if _, err := db.ExecContext(
		ctx,
		`UPDATE _System_InstalledPack
		    SET AppliedPayloadHash = ?, DesiredPayloadHash = ?
		  WHERE Name = ?`,
		differentPayloadHash,
		differentPayloadHash,
		productionPathPack.Name(),
	); err != nil {
		t.Fatalf("seed different legacy payload identity: %v", err)
	}
	if _, err := sourceBackedPackInstallIDForExactSourceAuthorityAdoption(
		ctx,
		db,
		appliedAppPack{pack: productionPathPack, payloadHash: productionPathHash},
		legacyPackInstallID,
	); !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf("different-payload adoption error = %v, want generation conflict", err)
	}
	assertNoPackGenerationAuthority(t, ctx, db, productionPathPack.Name())
	assertSourcePackObjectWasNotApplied(t, ctx, db)
	if _, err := db.ExecContext(
		ctx,
		`UPDATE _System_InstalledPack
		    SET AppliedPayloadHash = ?, DesiredPayloadHash = ?
		  WHERE Name = ?`,
		productionPathHash,
		productionPathHash,
		productionPathPack.Name(),
	); err != nil {
		t.Fatalf("restore exact legacy payload identity: %v", err)
	}
	adoptionPackInstallID, err := sourceBackedPackInstallIDForExactSourceAuthorityAdoption(
		ctx,
		db,
		appliedAppPack{pack: productionPathPack, payloadHash: productionPathHash},
		legacyPackInstallID,
	)
	if err != nil {
		t.Fatalf("classify explicit legacy source-identity adoption: %v", err)
	}
	if adoptionPackInstallID != legacyPackInstallID {
		t.Fatalf(
			"explicit legacy source-identity adoption incarnation = %q, want preserved %q",
			adoptionPackInstallID,
			legacyPackInstallID,
		)
	}
	err = applyResolvedSourceBackedPackUnlockedWithPreflightOverrides(
		ctx,
		&Runner{DB: conn},
		productionPathPack,
		nil,
		InstallPackOptions{},
	)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf(
			"ordinary reapply of legacy active pack error = %v, want explicit-adoption conflict",
			err,
		)
	}
	assertNoPackGenerationAuthority(t, ctx, db, productionPathPack.Name())
	assertSourcePackObjectWasNotApplied(t, ctx, db)
	if err := publishFirstSourceBackedPackGeneration(
		ctx,
		db,
		appliedAppPack{pack: productionPathPack, payloadHash: productionPathHash},
	); err != nil {
		t.Fatalf("stopped-upgrade first-generation adoption: %v", err)
	}
	assertOnePackGenerationAuthority(
		t,
		ctx,
		db,
		productionPathPack.Name(),
	)
	var activePackInstallID string
	if err := db.QueryRowContext(
		ctx,
		`SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		productionPathPack.Name(),
	).Scan(&activePackInstallID); err != nil {
		t.Fatalf("read active compatibility incarnation: %v", err)
	}
	assertActivePackGenerationConnectorBindingDefinitionSet(
		t, ctx, db, productionPathPack.Name(),
	)
	const mixedPackInstallID = "mixed-compatibility-pack-install"
	if _, err := db.ExecContext(
		ctx,
		`UPDATE _System_InstalledPack SET PackInstallId = ? WHERE Name = ?`,
		mixedPackInstallID,
		productionPathPack.Name(),
	); err != nil {
		t.Fatalf("create mixed compatibility pointer: %v", err)
	}
	err = applyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		productionPathPack,
		"production-path-source-revision-0",
		"mixed generation authority proof",
		func(revision string) *packs.ResolvedPackSource {
			return testControlplaneOfficialRepoResolvedSource(revision)
		},
	)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf("ApplyAll mixed compatibility pointer error = %v, want conflict", err)
	}
	var retainedPackInstallID string
	if err := db.QueryRowContext(
		ctx,
		`SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		productionPathPack.Name(),
	).Scan(&retainedPackInstallID); err != nil {
		t.Fatalf("read mixed compatibility pointer after rejection: %v", err)
	}
	if retainedPackInstallID != mixedPackInstallID {
		t.Fatalf(
			"mixed compatibility pointer was rewritten to %q before rejection, want %q",
			retainedPackInstallID,
			mixedPackInstallID,
		)
	}
	if _, err := db.ExecContext(
		ctx,
		`UPDATE _System_InstalledPack SET PackInstallId = ? WHERE Name = ?`,
		activePackInstallID,
		productionPathPack.Name(),
	); err != nil {
		t.Fatalf("restore active compatibility incarnation: %v", err)
	}
	assertApplyAllSourceRevisionRestartNoopWithGeneration(
		t,
		ctx,
		conn,
		productionPathPack,
	)
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_PackActivation
WHERE PackName = ?`, productionPathPack.Name()); err != nil {
		t.Fatalf("create orphaned-generation proof state: delete activation: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_PackGeneration
SET IsDeleted = 1
WHERE PackName = ?`, productionPathPack.Name()); err != nil {
		t.Fatalf("create orphaned-generation proof state: tombstone generation: %v", err)
	}
	err = applyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		productionPathPack,
		"production-path-source-revision-0",
		"source revision missing-authority proof",
		func(revision string) *packs.ResolvedPackSource {
			return testControlplaneOfficialRepoResolvedSource(revision)
		},
	)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf(
			"ApplyAll active no-op missing-authority error = %v, want conflict",
			err,
		)
	}
	err = applyResolvedSourceBackedPackUnlockedWithPreflightOverrides(
		ctx,
		&Runner{DB: conn},
		productionPathPack,
		nil,
		InstallPackOptions{},
	)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf(
			"production-path orphaned-generation fence error = %v, want conflict before component apply",
			err,
		)
	}
	assertSourcePackObjectWasNotApplied(t, ctx, db)

	deletePackGenerationConnectorBindingDefinitionSetForTest(
		t,
		ctx,
		db,
		packGenerationID(productionPathPack.Name(), activePackInstallID),
	)
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_PackGeneration
WHERE PackName = ?`, productionPathPack.Name()); err != nil {
		t.Fatalf("create legacy-active compatibility proof state: delete generation: %v", err)
	}
	err = applyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		productionPathPack,
		"production-path-source-revision-0",
		"stopped-upgrade legacy authority adoption",
		func(revision string) *packs.ResolvedPackSource {
			return testControlplaneOfficialRepoResolvedSource(revision)
		},
	)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf(
			"ApplyAll carrierless generation error = %v, want explicit-adoption conflict",
			err,
		)
	}
	// Source-revision planning does not synthesize missing authority. The
	// existing explicit stopped-upgrade publication path owns this separate
	// adoption and creates the generation plus its complete definition set.
	assertNoPackGenerationAuthority(t, ctx, db, productionPathPack.Name())
	if err := publishFirstSourceBackedPackGeneration(
		ctx,
		db,
		appliedAppPack{pack: productionPathPack, payloadHash: productionPathHash},
	); err != nil {
		t.Fatalf("phase-7 legacy authority adoption: %v", err)
	}
	assertOnePackGenerationAuthority(t, ctx, db, productionPathPack.Name())
	assertActivePackGenerationConnectorBindingDefinitionSet(
		t, ctx, db, productionPathPack.Name(),
	)
	assertSourcePackObjectWasNotApplied(t, ctx, db)

	pack := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t,
		"neutral-source-pack",
		"source-revision-1",
	)
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("hash concurrent first-activation pack: %v", err)
	}
	applied := appliedAppPack{pack: pack, payloadHash: payloadHash}
	seedActivatedInstalledPackCompatibility(
		t,
		ctx,
		db,
		applied,
		"pack-install-1",
	)

	start := make(chan struct{})
	results := make(chan error, 2)
	var ready sync.WaitGroup
	ready.Add(2)
	for range 2 {
		go func() {
			ready.Done()
			<-start
			tx, err := db.BeginTx(ctx, nil)
			if err == nil {
				err = commitFirstSourceBackedPackGeneration(ctx, tx, applied)
			}
			if err == nil {
				err = tx.Commit()
			} else if tx != nil {
				_ = tx.Rollback()
			}
			results <- err
		}()
	}
	ready.Wait()
	close(start)
	for range 2 {
		if err := <-results; err != nil {
			t.Fatalf("concurrent first activation: %v", err)
		}
	}

	var generationCount, activationCount int
	if err := db.QueryRowContext(ctx, `SELECT
  (SELECT COUNT(*) FROM _System_PackGeneration
   WHERE PackName = ? AND COALESCE(IsDeleted, 0) = 0),
  (SELECT COUNT(*) FROM _System_PackActivation
   WHERE PackName = ? AND COALESCE(IsDeleted, 0) = 0)`,
		pack.Name(),
		pack.Name(),
	).Scan(&generationCount, &activationCount); err != nil {
		t.Fatalf("count first activation authority: %v", err)
	}
	if generationCount != 1 || activationCount != 1 {
		t.Fatalf(
			"first activation authority counts = %d/%d, want 1/1",
			generationCount,
			activationCount,
		)
	}
	assertActivePackGenerationConnectorBindingDefinitionSet(
		t, ctx, db, pack.Name(),
	)

	changed := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t,
		pack.Name(),
		"source-revision-2",
	)
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin changed-source retry: %v", err)
	}
	err = commitFirstSourceBackedPackGeneration(
		ctx,
		tx,
		appliedAppPack{pack: changed, payloadHash: applied.payloadHash},
	)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf("changed source retry error = %v, want conflict", err)
	}
	if rollbackErr := tx.Rollback(); rollbackErr != nil {
		t.Fatalf("rollback changed-source retry: %v", rollbackErr)
	}
	var sourceRevision string
	if err := db.QueryRowContext(ctx, `SELECT SourceRevision
FROM _System_PackGeneration
WHERE PackInstallId = ?`, "pack-install-1").Scan(&sourceRevision); err != nil {
		t.Fatalf("read immutable source revision: %v", err)
	}
	if sourceRevision != "source-revision-1" {
		t.Fatalf(
			"immutable source revision = %q, want %q",
			sourceRevision,
			"source-revision-1",
		)
	}

	rolledBack := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t,
		"rolled-back-source-pack",
		"source-revision-1",
	)
	rolledBackPayloadHash, err := packs.PackPayloadHash(rolledBack)
	if err != nil {
		t.Fatalf("hash rolled-back first-activation pack: %v", err)
	}
	rolledBackApplied := appliedAppPack{
		pack:        rolledBack,
		payloadHash: rolledBackPayloadHash,
	}
	seedActivatedInstalledPackCompatibility(
		t,
		ctx,
		db,
		rolledBackApplied,
		"pack-install-rollback",
	)
	tx, err = db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin rollback proof: %v", err)
	}
	if err := commitFirstSourceBackedPackGeneration(
		ctx,
		tx,
		rolledBackApplied,
	); err != nil {
		t.Fatalf("stage rollback proof: %v", err)
	}
	if err := tx.Rollback(); err != nil {
		t.Fatalf("rollback first activation: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT
  (SELECT COUNT(*) FROM _System_PackGeneration WHERE PackName = ?),
  (SELECT COUNT(*) FROM _System_PackActivation WHERE PackName = ?)`,
		rolledBack.Name(),
		rolledBack.Name(),
	).Scan(&generationCount, &activationCount); err != nil {
		t.Fatalf("count rolled-back authority: %v", err)
	}
	if generationCount != 0 || activationCount != 0 {
		t.Fatalf(
			"rolled-back authority counts = %d/%d, want 0/0",
			generationCount,
			activationCount,
		)
	}
	assertNoPackGenerationConnectorBindingDefinitionSet(
		t,
		ctx,
		db,
		packGenerationID(rolledBack.Name(), "pack-install-rollback"),
	)

}

func TestStoppedSourceSetTiDBPublishesFirstGenerationOverLegacyInstallation(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()
	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("new runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh runner cache: %v", err)
	}

	legacy := testControlplaneOfficialRepoSourceBackedPack(t, "legacy-revision")
	legacyHash, err := packs.PackPayloadHash(legacy)
	if err != nil {
		t.Fatalf("legacy payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, db, legacy, legacyHash); err != nil {
		t.Fatalf("mark legacy installing: %v", err)
	}
	if err := markInstalledPackActive(
		ctx,
		db,
		legacySourceBackedActivationPack{SourceBackedPack: legacy},
		legacyHash,
	); err != nil {
		t.Fatalf("mark legacy active: %v", err)
	}
	if err := legacy.Apply(ctx, runner.PackDeps()); err != nil {
		t.Fatalf("apply legacy pack surface: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh runner cache after legacy apply: %v", err)
	}
	assertNoPackGenerationAuthority(t, ctx, db, legacy.Name())
	var legacyInstallID string
	if err := db.QueryRowContext(
		ctx,
		`SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		legacy.Name(),
	).Scan(&legacyInstallID); err != nil {
		t.Fatalf("read legacy install identity: %v", err)
	}

	target := testControlplaneOfficialRepoSourceBackedPack(t, "target-revision")
	targetHash, err := packs.PackPayloadHash(target)
	if err != nil {
		t.Fatalf("target payload hash: %v", err)
	}
	admissions, err := prepareStoppedSourceSetReplacementGenerationsForTest(
		ctx, db, []packs.Pack{target},
	)
	if err != nil {
		t.Fatalf("prepare stopped source-set generation: %v", err)
	}
	admission := admissions[strings.ToLower(target.Name())]
	if admission.activePackInstallID != legacyInstallID || admission.candidatePackInstallID == "" || admission.candidatePackInstallID == legacyInstallID {
		t.Fatalf("stopped source-set admission = %#v, want active %q and fresh candidate", admission, legacyInstallID)
	}
	var projectedInstallID, projectedStatus string
	if err := db.QueryRowContext(
		ctx,
		`SELECT PackInstallId, Status FROM _System_InstalledPack WHERE Name = ?`,
		target.Name(),
	).Scan(&projectedInstallID, &projectedStatus); err != nil {
		t.Fatalf("read active projection while candidate prepares: %v", err)
	}
	if projectedInstallID != legacyInstallID || projectedStatus != installedPackStatusActive {
		t.Fatalf("projection while candidate prepares = %q/%q, want unchanged %q/active", projectedInstallID, projectedStatus, legacyInstallID)
	}
	var candidateStatus string
	var candidateAppliedHash sql.NullString
	if err := db.QueryRowContext(ctx, `SELECT Status, AppliedPayloadHash FROM _System_PackGeneration WHERE PackInstallId = ?`, admission.candidatePackInstallID).Scan(&candidateStatus, &candidateAppliedHash); err != nil {
		t.Fatalf("read preparing generation: %v", err)
	}
	if candidateStatus != packGenerationPreparing {
		t.Fatalf("candidate generation status = %q, want preparing", candidateStatus)
	}
	if candidateAppliedHash.Valid {
		t.Fatalf("candidate applied payload hash = %#v, want NULL before activation", candidateAppliedHash)
	}

	// Reproduce the durable database state left when a process exits immediately
	// after the independently committed preparation transaction: there is no
	// caller defer to close this attempt.
	// The next exclusive stopped deploy must terminalize it and admit a fresh
	// candidate instead of permanently failing as generation-without-activation.
	abandonedCandidateInstallID := admission.candidatePackInstallID
	recoveredAdmissions, err := prepareStoppedSourceSetReplacementGenerationsForTest(
		ctx, db, []packs.Pack{target},
	)
	if err != nil {
		t.Fatalf("recover abandoned first-generation candidate: %v", err)
	}
	admission = recoveredAdmissions[strings.ToLower(target.Name())]
	if admission.activePackInstallID != legacyInstallID ||
		admission.candidatePackInstallID == "" ||
		admission.candidatePackInstallID == legacyInstallID ||
		admission.candidatePackInstallID == abandonedCandidateInstallID {
		t.Fatalf("recovered stopped source-set admission = %#v, want active %q and a new candidate", admission, legacyInstallID)
	}
	var abandonedStatus string
	var abandonedAppliedHash sql.NullString
	if err := db.QueryRowContext(ctx, `SELECT Status, AppliedPayloadHash FROM _System_PackGeneration WHERE PackInstallId = ?`, abandonedCandidateInstallID).Scan(&abandonedStatus, &abandonedAppliedHash); err != nil {
		t.Fatalf("read recovered abandoned generation: %v", err)
	}
	if abandonedStatus != packGenerationFailedDirty || abandonedAppliedHash.Valid {
		t.Fatalf(
			"recovered abandoned generation = %q/%#v, want %q/NULL unapplied hash",
			abandonedStatus,
			abandonedAppliedHash,
			packGenerationFailedDirty,
		)
	}
	if err := markInstalledPackActiveForStoppedSourceSet(
		ctx,
		db,
		target,
		targetHash,
		legacyInstallID,
		admission.candidatePackInstallID,
		nil,
	); err != nil {
		t.Fatalf("activate stopped source-set first generation: %v", err)
	}
	var (
		activeInstallID string
		activationEpoch uint64
	)
	if err := db.QueryRowContext(
		ctx,
		`SELECT ActivePackInstallId, ActivationEpoch FROM _System_PackActivation WHERE PackName = ?`,
		target.Name(),
	).Scan(&activeInstallID, &activationEpoch); err != nil {
		t.Fatalf("read first stopped source-set activation: %v", err)
	}
	if activeInstallID != admission.candidatePackInstallID || activationEpoch != firstPackActivationEpoch {
		t.Fatalf(
			"first stopped source-set activation = install %q epoch %d, want %q/%d",
			activeInstallID,
			activationEpoch,
			admission.candidatePackInstallID,
			firstPackActivationEpoch,
		)
	}
	var (
		generationRevision string
		generationHash     string
	)
	if err := db.QueryRowContext(
		ctx,
		`SELECT SourceRevision, AppliedPayloadHash FROM _System_PackGeneration WHERE PackInstallId = ?`,
		activeInstallID,
	).Scan(&generationRevision, &generationHash); err != nil {
		t.Fatalf("read first stopped source-set generation payload: %v", err)
	}
	if generationRevision != "target-revision" || generationHash != targetHash {
		t.Fatalf(
			"first stopped source-set generation = revision %q hash %q, want target-revision/%q",
			generationRevision,
			generationHash,
			targetHash,
		)
	}

	// Reproduce the retained-history shape that blocked Test2: activation names
	// one healthy generation while another historical generation row remains.
	// History must not become a second current-version authority.
	const historicalInstallID = "historical-unselected-generation"
	seedTx, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin historical generation fixture: %v", err)
	}
	historical, exists, err := readPackGenerationForUpdate(ctx, seedTx, activeInstallID)
	if err != nil || !exists {
		_ = seedTx.Rollback()
		t.Fatalf("read selected generation for historical fixture: exists=%t err=%v", exists, err)
	}
	historical.id = packGenerationID(target.Name(), historicalInstallID)
	historical.packInstallID = historicalInstallID
	if err := insertPackGeneration(ctx, seedTx, historical, time.Now().UTC()); err != nil {
		_ = seedTx.Rollback()
		t.Fatalf("insert retained historical generation: %v", err)
	}
	if err := seedTx.Commit(); err != nil {
		t.Fatalf("commit retained historical generation: %v", err)
	}
	historyRetry := testControlplaneOfficialRepoSourceBackedPack(t, "history-retry-revision")
	historyAdmissions, err := prepareStoppedSourceSetReplacementGenerationsForTest(
		ctx, db, []packs.Pack{historyRetry},
	)
	if err != nil {
		t.Fatalf("prepare retry with retained generation history: %v", err)
	}
	historyAdmission := historyAdmissions[strings.ToLower(historyRetry.Name())]
	if historyAdmission.activePackInstallID != activeInstallID ||
		historyAdmission.candidatePackInstallID == "" ||
		historyAdmission.candidatePackInstallID == activeInstallID ||
		historyAdmission.candidatePackInstallID == historicalInstallID {
		t.Fatalf("history retry admission = %#v, want selected %q and a fresh candidate", historyAdmission, activeInstallID)
	}
	var selectedAfterHistoryRetry string
	if err := db.QueryRowContext(
		ctx,
		`SELECT ActivePackInstallId FROM _System_PackActivation WHERE PackName = ?`,
		historyRetry.Name(),
	).Scan(&selectedAfterHistoryRetry); err != nil {
		t.Fatalf("read activation after retained-history retry: %v", err)
	}
	if selectedAfterHistoryRetry != activeInstallID {
		t.Fatalf("activation after retained-history retry = %q, want unchanged %q", selectedAfterHistoryRetry, activeInstallID)
	}
	if err := markStoppedSourceSetPackGenerationFailed(
		ctx,
		db,
		historyRetry.Name(),
		historyAdmission.candidatePackInstallID,
		false,
		errors.New("close retained-history regression candidate"),
	); err != nil {
		t.Fatalf("close retained-history regression candidate: %v", err)
	}

	// Reproduce the state left by the old deploy path: a failed candidate
	// overwrote InstalledPack even though PackActivation still selected the
	// previous healthy generation. Admission must reconstruct the compatibility
	// projection from activation, not treat the failed attempt as runtime state.
	const failedAttemptID = "failed-attempt-that-overwrote-projection"
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack
SET PackInstallId = ?, Status = ?, SourceRevision = 'failed-revision', LastError = 'old deploy failed'
WHERE Name = ?`, failedAttemptID, installedPackStatusFailed, target.Name()); err != nil {
		t.Fatalf("seed historical failed projection overwrite: %v", err)
	}
	retry := testControlplaneOfficialRepoSourceBackedPack(t, "retry-revision")
	retryAdmissions, err := prepareStoppedSourceSetReplacementGenerationsForTest(ctx, db, []packs.Pack{retry})
	if err != nil {
		t.Fatalf("prepare retry after historical failed overwrite: %v", err)
	}
	retryAdmission := retryAdmissions[strings.ToLower(retry.Name())]
	if retryAdmission.activePackInstallID != activeInstallID || retryAdmission.candidatePackInstallID == failedAttemptID || retryAdmission.candidatePackInstallID == activeInstallID {
		t.Fatalf("retry admission = %#v, want active %q and a fresh candidate", retryAdmission, activeInstallID)
	}
	var repairedInstallID, repairedStatus, repairedRevision string
	if err := db.QueryRowContext(ctx, `SELECT PackInstallId, Status, SourceRevision FROM _System_InstalledPack WHERE Name = ?`, retry.Name()).Scan(&repairedInstallID, &repairedStatus, &repairedRevision); err != nil {
		t.Fatalf("read repaired active projection: %v", err)
	}
	if repairedInstallID != activeInstallID || repairedStatus != installedPackStatusActive || repairedRevision != "target-revision" {
		t.Fatalf("repaired projection = %q/%q/%q, want %q/active/target-revision", repairedInstallID, repairedStatus, repairedRevision, activeInstallID)
	}
	abandonedReplacementInstallID := retryAdmission.candidatePackInstallID
	retryAdmissions, err = prepareStoppedSourceSetReplacementGenerationsForTest(ctx, db, []packs.Pack{retry})
	if err != nil {
		t.Fatalf("recover abandoned replacement candidate: %v", err)
	}
	retryAdmission = retryAdmissions[strings.ToLower(retry.Name())]
	if retryAdmission.activePackInstallID != activeInstallID ||
		retryAdmission.candidatePackInstallID == "" ||
		retryAdmission.candidatePackInstallID == activeInstallID ||
		retryAdmission.candidatePackInstallID == abandonedReplacementInstallID {
		t.Fatalf("recovered replacement admission = %#v, want active %q and a new candidate", retryAdmission, activeInstallID)
	}
	var abandonedReplacementStatus string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_PackGeneration WHERE PackInstallId = ?`, abandonedReplacementInstallID).Scan(&abandonedReplacementStatus); err != nil {
		t.Fatalf("read recovered abandoned replacement: %v", err)
	}
	if abandonedReplacementStatus != packGenerationFailedDirty {
		t.Fatalf("recovered abandoned replacement status = %q, want %q", abandonedReplacementStatus, packGenerationFailedDirty)
	}
	var selectedInstallIDAfterRecovery string
	if err := db.QueryRowContext(ctx, `SELECT ActivePackInstallId FROM _System_PackActivation WHERE PackName = ?`, retry.Name()).Scan(&selectedInstallIDAfterRecovery); err != nil {
		t.Fatalf("read selected generation after replacement recovery: %v", err)
	}
	if selectedInstallIDAfterRecovery != activeInstallID {
		t.Fatalf("selected generation after replacement recovery = %q, want unchanged %q", selectedInstallIDAfterRecovery, activeInstallID)
	}
	if err := markStoppedSourceSetPackGenerationFailed(ctx, db, retry.Name(), retryAdmission.candidatePackInstallID, true, errors.New("candidate apply failed")); err != nil {
		t.Fatalf("mark retry candidate failed: %v", err)
	}
	var failedStatus string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_PackGeneration WHERE PackInstallId = ?`, retryAdmission.candidatePackInstallID).Scan(&failedStatus); err != nil {
		t.Fatalf("read failed retry generation: %v", err)
	}
	if failedStatus != packGenerationFailedDirty {
		t.Fatalf("failed retry generation status = %q, want %q", failedStatus, packGenerationFailedDirty)
	}
	var projectionAfterFailureID, projectionAfterFailureStatus string
	if err := db.QueryRowContext(ctx, `SELECT PackInstallId, Status FROM _System_InstalledPack WHERE Name = ?`, retry.Name()).Scan(&projectionAfterFailureID, &projectionAfterFailureStatus); err != nil {
		t.Fatalf("read projection after candidate failure: %v", err)
	}
	if projectionAfterFailureID != activeInstallID || projectionAfterFailureStatus != installedPackStatusActive {
		t.Fatalf("projection after candidate failure = %q/%q, want unchanged %q/active", projectionAfterFailureID, projectionAfterFailureStatus, activeInstallID)
	}
}

func prepareStoppedSourceSetReplacementGenerationsForTest(
	ctx context.Context,
	db *sql.DB,
	targets []packs.Pack,
) (map[string]stoppedSourceSetReplacementAdmission, error) {
	var admissions map[string]stoppedSourceSetReplacementAdmission
	err := withControlPlaneApplyLockScope(
		ctx,
		db,
		"prepare-stopped-source-set-integration-test",
		func(scope controlPlaneApplyLockScope) error {
			var prepareErr error
			admissions, prepareErr = prepareStoppedSourceSetReplacementGenerations(
				scope.identity.leaseContext,
				db,
				targets,
				scope,
			)
			return prepareErr
		},
	)
	return admissions, err
}

func TestApplyStoppedSourceSetTiDBUsesPackOwnedSafetyAndPublishesReplacement(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)
	setDatabaseEnvFromConfig(t, cfg)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	const (
		packName  = "stopped-source-set-entrypoint"
		sourceURI = "urn:nexuscrm:test:stopped-source-set-entrypoint"
	)
	packRoot, legacy := writeNamedFieldDeclarationSource(t, packName, "Issue", "")
	legacyHash, err := packs.PackPayloadHash(legacy)
	if err != nil {
		t.Fatalf("legacy payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, db, legacy, legacyHash); err != nil {
		t.Fatalf("mark legacy installing: %v", err)
	}
	if err := markInstalledPackActive(
		ctx,
		db,
		legacySourceBackedActivationPack{SourceBackedPack: legacy},
		legacyHash,
	); err != nil {
		t.Fatalf("mark legacy active: %v", err)
	}
	var legacyInstallID string
	if err := db.QueryRowContext(ctx, `SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`, packName).Scan(&legacyInstallID); err != nil {
		t.Fatalf("read legacy install identity: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorInboundRoute
(Id, Namespace, InstalledPackId, EventKind, FlowDeveloperName)
VALUES ('stopped-old-route', ?, ?, 'legacy.event', 'LegacyFlow')`, packName, legacyInstallID); err != nil {
		t.Fatalf("seed previous connector route incarnation: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorInboundHandlerConfig
(Id, InstalledPackId, Namespace, CapabilityKind, ConfigJSON)
VALUES ('stopped-old-handler', ?, ?, 'signature_verifier', '{}')`, legacyInstallID, packName); err != nil {
		t.Fatalf("seed previous connector handler incarnation: %v", err)
	}
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorCalloutProfile
(Id, InstalledPackId, OwnerPack, ProfileName, CredentialRef, EndpointClass, RetryPolicy, PageSpec, BatchSpec, ObjectPathMap, Description)
VALUES ('stopped-old-callout', ?, ?, 'legacy-profile', 'legacy-credential', 'legacy-endpoint', '{}', '{}', '{}', '{}', '')`, legacyInstallID, packName); err != nil {
		t.Fatalf("seed previous connector callout incarnation: %v", err)
	}

	initGitTestRepoForControlplane(t, packRoot)
	revision := gitTestCommitForControlplane(t, packRoot, "sealed stopped source-set replacement")
	requestForOperation := func(operationID string) StoppedSourceSetRequest {
		return StoppedSourceSetRequest{
			OperationID: operationID,
			PackSources: []string{string(packs.PackSourceKindReleaseRepo) + ":" + sourceURI},
			ReleaseRepoSources: []StoppedSourceSetReleaseRepoSource{{
				URI:      sourceURI,
				PackRoot: packRoot,
				Revision: revision,
			}},
		}
	}
	const queuedEventID = "stopped-source-set-prior-inbound"
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, CompanionPackInstallId)
VALUES (?, 'test-connector', 'test-connector-install', 'test.event', ?, 'pending', ?)`,
		queuedEventID,
		queuedEventID,
		legacyInstallID,
	); err != nil {
		t.Fatalf("seed prior companion inbound event: %v", err)
	}
	blockedResult, err := ApplyStoppedSourceSetWithRunnerOptions(
		ctx,
		conn,
		requestForOperation("stopped-source-set-entrypoint-blocked-test"),
		RunnerOptions{},
	)
	if err == nil || !errors.Is(err, ErrPackGenerationPrecondition) || !strings.Contains(err.Error(), legacyInstallID) {
		t.Fatalf("queued stopped source-set result/error = %#v/%v, want exact-incarnation inbound rejection", blockedResult, err)
	}
	var bootstrapStateTableCount, applyAllMarkerCount int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '_System_BootstrapState'`).Scan(&bootstrapStateTableCount); err != nil {
		t.Fatalf("read bootstrap-state table after queued rejection: %v", err)
	}
	if bootstrapStateTableCount > 0 {
		if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_BootstrapState WHERE StateKey = 'control_plane_apply_all'`).Scan(&applyAllMarkerCount); err != nil {
			t.Fatalf("read ApplyAll marker after queued rejection: %v", err)
		}
	}
	if applyAllMarkerCount != 0 {
		t.Fatalf("ApplyAll marker count after queued rejection = %d, want 0 before deployment-wrapper writes", applyAllMarkerCount)
	}
	var rejectedInstallID, rejectedStatus string
	if err := db.QueryRowContext(ctx, `SELECT PackInstallId, Status FROM _System_InstalledPack WHERE Name = ?`, packName).Scan(&rejectedInstallID, &rejectedStatus); err != nil {
		t.Fatalf("read installed row after queued rejection: %v", err)
	}
	if rejectedInstallID != legacyInstallID || rejectedStatus != installedPackStatusActive {
		t.Fatalf("installed row after queued rejection = %q/%q, want unchanged %q/active", rejectedInstallID, rejectedStatus, legacyInstallID)
	}
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_ConnectorEvent WHERE Id = ?`, queuedEventID); err != nil {
		t.Fatalf("drain prior companion inbound event: %v", err)
	}
	// Candidate preparation commits before the marker dependency is consulted.
	// A missing composition-root seam must close that attempt as failed_clean,
	// otherwise every later retry sees a permanent preparing authority.
	var missingMarkerErr error
	func() {
		originalBootstrapStateStore := bootstrapStateStore
		defer func() { bootstrapStateStore = originalBootstrapStateStore }()
		bootstrapStateStore = nil
		_, missingMarkerErr = ApplyStoppedSourceSetWithRunnerOptions(
			ctx,
			conn,
			requestForOperation("stopped-source-set-entrypoint-missing-marker-test"),
			RunnerOptions{},
		)
	}()
	if missingMarkerErr == nil || !strings.Contains(missingMarkerErr.Error(), "marker store is unavailable") {
		t.Fatalf("missing marker error = %v, want explicit marker-store failure", missingMarkerErr)
	}
	var preparingCount, failedCleanCount int
	if err := db.QueryRowContext(ctx, `SELECT
SUM(CASE WHEN Status = 'preparing' THEN 1 ELSE 0 END),
SUM(CASE WHEN Status = 'failed_clean' THEN 1 ELSE 0 END)
FROM _System_PackGeneration
WHERE PackName = ? AND COALESCE(IsDeleted, 0) = 0`, packName).Scan(&preparingCount, &failedCleanCount); err != nil {
		t.Fatalf("read generation state after missing marker failure: %v", err)
	}
	if preparingCount != 0 || failedCleanCount != 1 {
		t.Fatalf("generation state after missing marker failure = preparing:%d failed_clean:%d, want 0/1", preparingCount, failedCleanCount)
	}
	var oldRouteCurrent, oldRouteDeleted, oldHandlerDeleted, oldCalloutDeleted int
	if err := db.QueryRowContext(ctx, `SELECT
  (SELECT IsCurrent FROM _System_ConnectorInboundRoute WHERE Id = 'stopped-old-route'),
  (SELECT IsDeleted FROM _System_ConnectorInboundRoute WHERE Id = 'stopped-old-route'),
  (SELECT IsDeleted FROM _System_ConnectorInboundHandlerConfig WHERE Id = 'stopped-old-handler'),
  (SELECT IsDeleted FROM _System_ConnectorCalloutProfile WHERE Id = 'stopped-old-callout')`).Scan(
		&oldRouteCurrent, &oldRouteDeleted, &oldHandlerDeleted, &oldCalloutDeleted,
	); err != nil {
		t.Fatalf("read previous connector rows after clean candidate failure: %v", err)
	}
	if oldRouteCurrent != 1 || oldRouteDeleted != 0 || oldHandlerDeleted != 0 || oldCalloutDeleted != 0 {
		t.Fatalf(
			"previous connector rows changed before activation = route:%d/%d handler:%d callout:%d",
			oldRouteCurrent, oldRouteDeleted, oldHandlerDeleted, oldCalloutDeleted,
		)
	}
	result, err := ApplyStoppedSourceSetWithRunnerOptions(
		ctx,
		conn,
		requestForOperation("stopped-source-set-entrypoint-test"),
		RunnerOptions{},
	)
	if err != nil {
		t.Fatalf("apply stopped source-set entry point: %v", err)
	}
	if !result.Applied || result.OperationID != "stopped-source-set-entrypoint-test" {
		t.Fatalf("stopped source-set result = %#v, want applied receipt", result)
	}

	readConn := openIntegrationConnection(t, cfg)
	t.Cleanup(func() { _ = readConn.Close() })
	readDB := readConn.DB()
	var (
		activeInstallID string
		status          string
		sourceKind      string
		storedSourceURI string
		storedRevision  string
	)
	if err := readDB.QueryRowContext(ctx, `SELECT PackInstallId, Status, SourceKind, SourceURI, SourceRevision
FROM _System_InstalledPack
WHERE Name = ?`, packName).Scan(&activeInstallID, &status, &sourceKind, &storedSourceURI, &storedRevision); err != nil {
		t.Fatalf("read stopped source-set replacement: %v", err)
	}
	if activeInstallID == "" || activeInstallID == legacyInstallID || status != installedPackStatusActive || sourceKind != string(packs.PackSourceKindReleaseRepo) || storedSourceURI != sourceURI || storedRevision != revision {
		t.Fatalf(
			"stopped replacement = install %q status %q source %s:%s@%s, want fresh active %s:%s@%s",
			activeInstallID,
			status,
			sourceKind,
			storedSourceURI,
			storedRevision,
			packs.PackSourceKindReleaseRepo,
			sourceURI,
			revision,
		)
	}
	var generationCount, activeGenerationCount, failedCleanGenerationCount, activationCount int
	if err := readDB.QueryRowContext(ctx, `SELECT
  (SELECT COUNT(*) FROM _System_PackGeneration WHERE PackName = ? AND COALESCE(IsDeleted, 0) = 0),
  (SELECT COUNT(*) FROM _System_PackGeneration WHERE PackName = ? AND Status = 'active' AND COALESCE(IsDeleted, 0) = 0),
  (SELECT COUNT(*) FROM _System_PackGeneration WHERE PackName = ? AND Status = 'failed_clean' AND COALESCE(IsDeleted, 0) = 0),
  (SELECT COUNT(*) FROM _System_PackActivation WHERE PackName = ? AND COALESCE(IsDeleted, 0) = 0)`,
		packName, packName, packName, packName,
	).Scan(&generationCount, &activeGenerationCount, &failedCleanGenerationCount, &activationCount); err != nil {
		t.Fatalf("count stopped-source-set generation history: %v", err)
	}
	if generationCount != 2 || activeGenerationCount != 1 || failedCleanGenerationCount != 1 || activationCount != 1 {
		t.Fatalf(
			"stopped-source-set generation history = total:%d active:%d failed_clean:%d activation:%d, want 2/1/1/1",
			generationCount, activeGenerationCount, failedCleanGenerationCount, activationCount,
		)
	}
	if err := readDB.QueryRowContext(ctx, `SELECT
  (SELECT IsCurrent FROM _System_ConnectorInboundRoute WHERE Id = 'stopped-old-route'),
  (SELECT IsDeleted FROM _System_ConnectorInboundRoute WHERE Id = 'stopped-old-route'),
  (SELECT IsDeleted FROM _System_ConnectorInboundHandlerConfig WHERE Id = 'stopped-old-handler'),
  (SELECT IsDeleted FROM _System_ConnectorCalloutProfile WHERE Id = 'stopped-old-callout')`).Scan(
		&oldRouteCurrent, &oldRouteDeleted, &oldHandlerDeleted, &oldCalloutDeleted,
	); err != nil {
		t.Fatalf("read previous connector rows after activation: %v", err)
	}
	if oldRouteCurrent != 0 || oldRouteDeleted != 1 || oldHandlerDeleted != 1 || oldCalloutDeleted != 1 {
		t.Fatalf(
			"previous connector rows after activation = route:%d/%d handler:%d callout:%d, want 0/1/1/1",
			oldRouteCurrent, oldRouteDeleted, oldHandlerDeleted, oldCalloutDeleted,
		)
	}
}

func TestStoppedSourceSetTiDBLifecycleAdmissionRejectsInvalidOrChangedInstallation(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()
	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("new runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh runner cache: %v", err)
	}

	legacy := testControlplaneOfficialRepoSourceBackedPack(t, "legacy-revision")
	legacyHash, err := packs.PackPayloadHash(legacy)
	if err != nil {
		t.Fatalf("legacy payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, db, legacy, legacyHash); err != nil {
		t.Fatalf("mark legacy installing: %v", err)
	}
	if err := markInstalledPackActive(
		ctx,
		db,
		legacySourceBackedActivationPack{SourceBackedPack: legacy},
		legacyHash,
	); err != nil {
		t.Fatalf("mark legacy active: %v", err)
	}
	if err := legacy.Apply(ctx, runner.PackDeps()); err != nil {
		t.Fatalf("apply legacy pack surface: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh runner cache after legacy apply: %v", err)
	}
	var legacyInstallID string
	if err := db.QueryRowContext(ctx, `SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`, legacy.Name()).Scan(&legacyInstallID); err != nil {
		t.Fatalf("read legacy install identity: %v", err)
	}
	target := testControlplaneOfficialRepoSourceBackedPack(t, "target-revision")
	applyStoppedTarget := func(target *packs.SourceBackedPack) error {
		t.Helper()
		admissions, err := prepareStoppedSourceSetReplacementGenerationsForTest(ctx, db, []packs.Pack{target})
		if err != nil {
			return err
		}
		admission := admissions[strings.ToLower(strings.TrimSpace(target.Name()))]
		resolvedTarget := stoppedSourceSetTarget{pack: target}
		set := resolvedStoppedSourceSet{
			targets: []stoppedSourceSetTarget{resolvedTarget},
			byName: map[string]stoppedSourceSetTarget{
				strings.ToLower(strings.TrimSpace(target.Name())): resolvedTarget,
			},
		}
		overrides, err := stoppedSourceSetRuntimeOverrides(set)
		if err != nil {
			t.Fatalf("build stopped source-set preflight context: %v", err)
		}
		overrides.activeGenerationAttributionPeers, err = loadActiveGenerationAttributionPeerContext(
			ctx,
			db,
			map[string]struct{}{strings.ToLower(target.Name()): {}},
		)
		if err != nil {
			t.Fatalf("load stopped source-set attribution peers: %v", err)
		}
		return applyResolvedSourceBackedPackUnlockedWithPreflightOverrides(
			ctx,
			runner,
			target,
			overrides,
			InstallPackOptions{
				stoppedSourceSetApplyOnly:              true,
				stoppedSourceSetAuthority:              offlinecandidateauthority.NewStoppedSourceSetAuthority(),
				stoppedSourceSetPreviousPackInstallID:  admission.activePackInstallID,
				stoppedSourceSetCandidatePackInstallID: admission.candidatePackInstallID,
			},
		)
	}
	assertLegacyLifecycleUnchanged := func(reason string) {
		t.Helper()
		var actualInstallID, actualStatus, actualRevision string
		if err := db.QueryRowContext(ctx, `SELECT PackInstallId, Status, SourceRevision FROM _System_InstalledPack WHERE Name = ?`, legacy.Name()).Scan(&actualInstallID, &actualStatus, &actualRevision); err != nil {
			t.Fatalf("read lifecycle after %s: %v", reason, err)
		}
		if actualInstallID != legacyInstallID || actualStatus != installedPackStatusActive || actualRevision != "legacy-revision" {
			t.Fatalf("lifecycle after %s = %q/%q/%q, want unchanged %q/active/legacy-revision", reason, actualInstallID, actualStatus, actualRevision, legacyInstallID)
		}
	}

	missingDependencySource := testControlplaneOfficialRepoResolvedSource("target-revision")
	missingDependencySource.Manifest.Dependencies = []packs.PackDependency{{
		Name:              "untouched-provider",
		VersionConstraint: ">=0.1.0",
	}}
	missingDependencyTarget, err := packs.NewSourceBackedPack(
		missingDependencySource,
		packs.WithStoppedSourceSetApplyOnly(offlinecandidateauthority.NewStoppedSourceSetAuthority()),
	)
	if err != nil {
		t.Fatalf("construct missing-dependency stopped target: %v", err)
	}
	err = applyStoppedTarget(missingDependencyTarget)
	if err == nil || !strings.Contains(err.Error(), "dependency untouched-provider is not active") {
		t.Fatalf("missing untouched dependency error = %v, want live dependency rejection", err)
	}
	assertLegacyLifecycleUnchanged("missing dependency rejection")

	if _, err := db.ExecContext(ctx, `INSERT INTO ProcessDefinition
(Id, Name, DeveloperName, State, TableEnumOrId, IsActive, OwnerPack, CreatedDate, LastModifiedDate)
VALUES ('pd-stale-stopped-apply', 'Stale Stopped Approval', 'Stale_Stopped_Approval', 'Active', 'Issue', 1, ?, NOW(), NOW())`, legacy.Name()); err != nil {
		t.Fatalf("seed stale managed approval ownership: %v", err)
	}
	managedOwnershipSource := testControlplaneOfficialRepoResolvedSource("target-revision")
	managedOwnershipTarget, err := packs.NewSourceBackedPack(
		managedOwnershipSource,
		packs.WithStoppedSourceSetApplyOnly(offlinecandidateauthority.NewStoppedSourceSetAuthority()),
	)
	if err != nil {
		t.Fatalf("construct managed-ownership stopped target: %v", err)
	}
	err = applyStoppedTarget(managedOwnershipTarget)
	if err == nil || !strings.Contains(err.Error(), "no OwnerPack-scoped managed approval_process registration covers ProcessDefinition") {
		t.Fatalf("managed approval ownership error = %v, want stale ownership rejection", err)
	}
	assertLegacyLifecycleUnchanged("managed approval ownership rejection")
	if _, err := db.ExecContext(ctx, `DELETE FROM ProcessDefinition WHERE Id = 'pd-stale-stopped-apply'`); err != nil {
		t.Fatalf("remove stale managed approval fixture: %v", err)
	}

	invalidStates := []struct {
		name     string
		column   string
		value    any
		restore  any
		wantText string
	}{
		{name: "failed", column: "Status", value: installedPackStatusFailed, restore: installedPackStatusActive, wantText: "is not active"},
		{name: "removing", column: "Status", value: installedPackStatusRemoving, restore: installedPackStatusActive, wantText: "is not active"},
		{name: "deleted", column: "IsDeleted", value: 1, restore: 0, wantText: "legacy installation"},
		{name: "wrong-category", column: "Category", value: string(packs.ConnectorPack), restore: string(packs.AppPack), wantText: "legacy installation"},
	}
	for _, test := range invalidStates {
		t.Run(test.name, func(t *testing.T) {
			query := fmt.Sprintf("UPDATE _System_InstalledPack SET %s = ? WHERE Name = ?", test.column)
			if _, err := db.ExecContext(ctx, query, test.value, legacy.Name()); err != nil {
				t.Fatalf("seed invalid lifecycle state: %v", err)
			}
			_, err := prepareStoppedSourceSetReplacementGenerationsForTest(ctx, db, []packs.Pack{target})
			if !errors.Is(err, ErrPackGenerationPrecondition) || !strings.Contains(err.Error(), test.wantText) {
				t.Fatalf("lifecycle admission error = %v, want generation precondition containing %q", err, test.wantText)
			}
			if _, err := db.ExecContext(ctx, query, test.restore, legacy.Name()); err != nil {
				t.Fatalf("restore lifecycle state: %v", err)
			}
		})
	}

	admitted, err := prepareStoppedSourceSetReplacementGenerationsForTest(ctx, db, []packs.Pack{target})
	if err != nil {
		t.Fatalf("admit active legacy installation: %v", err)
	}
	admission := admitted[strings.ToLower(legacy.Name())]
	if admission.activePackInstallID != legacyInstallID || admission.candidatePackInstallID == "" {
		t.Fatalf("admission = %#v, want active %q and a candidate", admission, legacyInstallID)
	}
	const changedInstallID = "changed-after-lifecycle-admission"
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack SET PackInstallId = ? WHERE Name = ?`, changedInstallID, legacy.Name()); err != nil {
		t.Fatalf("change installation identity after admission: %v", err)
	}
	targetHash, err := packs.PackPayloadHash(target)
	if err != nil {
		t.Fatalf("target payload hash: %v", err)
	}
	err = markInstalledPackActiveForStoppedSourceSet(ctx, db, target, targetHash, legacyInstallID, admission.candidatePackInstallID, nil)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf("stale activation transition error = %v, want generation conflict", err)
	}
	var (
		actualInstallID string
		actualStatus    string
		actualRevision  sql.NullString
	)
	if err := db.QueryRowContext(ctx, `SELECT PackInstallId, Status, SourceRevision FROM _System_InstalledPack WHERE Name = ?`, legacy.Name()).Scan(&actualInstallID, &actualStatus, &actualRevision); err != nil {
		t.Fatalf("read rejected stale lifecycle transition: %v", err)
	}
	if actualInstallID != changedInstallID || actualStatus != installedPackStatusActive || !actualRevision.Valid || actualRevision.String != "legacy-revision" {
		t.Fatalf("rejected stale lifecycle row = %q/%q/%v, want unchanged %q/active/legacy-revision", actualInstallID, actualStatus, actualRevision, changedInstallID)
	}
}

func TestStoppedSourceSetTiDBObjectSpecCollectsAcrossComponentsBeforeOwnershipWrites(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()
	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("new runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh runner cache: %v", err)
	}

	objectSpecComponents := func() []packs.ManifestComponent {
		return []packs.ManifestComponent{
			{
				Kind:          packs.ComponentKindObjectSpec,
				BusinessKey:   "issue",
				Manageability: packs.ManageabilityLocked,
				Payload: []byte(`[
  {
    "apiName": "Issue",
    "label": "Issue",
    "tableType": "standard_object",
    "category": "jira",
    "sharingModel": "Private",
    "enableHierarchySharing": false,
    "description": "Issue work item",
    "fields": [
      {"apiName": "Subject", "label": "Subject", "type": "string", "required": true, "isNameField": true}
    ]
  }
]`),
			},
			{
				Kind:          packs.ComponentKindObjectSpec,
				BusinessKey:   "project-sprint",
				Manageability: packs.ManageabilityLocked,
				Payload: []byte(`[
  {
    "apiName": "Project",
    "label": "Project",
    "tableType": "standard_object",
    "category": "jira",
    "sharingModel": "Private",
    "enableHierarchySharing": false,
    "description": "Project work item",
    "fields": [
      {"apiName": "Name", "label": "Name", "type": "string", "required": true, "isNameField": true}
    ]
  },
  {
    "apiName": "Sprint",
    "label": "Sprint",
    "tableType": "standard_object",
    "category": "jira",
    "sharingModel": "Private",
    "enableHierarchySharing": false,
    "description": "Sprint work item",
    "fields": [
      {"apiName": "Name", "label": "Name", "type": "string", "required": true, "isNameField": true}
    ]
  }
]`),
			},
		}
	}
	legacySource := testControlplaneOfficialRepoResolvedSource("legacy-revision")
	legacySource.Manifest.Components = objectSpecComponents()
	legacy, err := packs.NewSourceBackedPack(legacySource)
	if err != nil {
		t.Fatalf("construct legacy source-backed pack: %v", err)
	}
	legacyHash, err := packs.PackPayloadHash(legacy)
	if err != nil {
		t.Fatalf("legacy payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, db, legacy, legacyHash); err != nil {
		t.Fatalf("mark legacy installing: %v", err)
	}
	if err := markInstalledPackActive(
		ctx,
		db,
		legacySourceBackedActivationPack{SourceBackedPack: legacy},
		legacyHash,
	); err != nil {
		t.Fatalf("mark legacy active: %v", err)
	}
	if err := legacy.Apply(ctx, runner.PackDeps()); err != nil {
		t.Fatalf("apply legacy object surface: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_Field f
JOIN _System_Object o ON o.Id = f.ObjectId
   SET f.OwnerKind = '', f.OwnerPack = NULL
 WHERE o.ApiName IN ('Issue', 'Project', 'Sprint') AND COALESCE(o.IsDeleted, 0) = 0`); err != nil {
		t.Fatalf("seed ownerless legacy field surface: %v", err)
	}

	targetSource := testControlplaneOfficialRepoResolvedSource("target-revision")
	targetSource.Manifest.Components = objectSpecComponents()
	ordinaryTarget, err := packs.NewSourceBackedPack(targetSource)
	if err != nil {
		t.Fatalf("construct ordinary target: %v", err)
	}
	if err := ordinaryTarget.Apply(ctx, runner.PackDeps()); err == nil || !strings.Contains(err.Error(), "ownerless object/table/field surface Issue already exists") {
		t.Fatalf("ordinary install error = %v, want existing-surface rejection", err)
	}
	stoppedSource := testControlplaneOfficialRepoResolvedSource("target-revision")
	stoppedSource.Manifest.Components = objectSpecComponents()
	stoppedTarget, err := packs.NewSourceBackedPack(
		stoppedSource,
		packs.WithStoppedSourceSetApplyOnly(
			offlinecandidateauthority.NewStoppedSourceSetAuthority(),
		),
	)
	if err != nil {
		t.Fatalf("construct stopped source-set target: %v", err)
	}
	applyStoppedTarget := func() error {
		return offlinecandidateauthority.RunWithStoppedSourceSetApplyScope(
			ctx,
			stoppedTarget.Name(),
			"stopped-object-spec-candidate",
			func(applyCtx context.Context) error {
				return stoppedTarget.Apply(applyCtx, runner.PackDeps())
			},
		)
	}

	if _, err := db.ExecContext(ctx, `UPDATE _System_Table SET Category = 'wrong' WHERE TableName IN ('Project', 'Sprint')`); err != nil {
		t.Fatalf("seed incompatible legacy table shape: %v", err)
	}
	err = applyStoppedTarget()
	if err == nil {
		t.Fatal("object_spec apply succeeded, want compatibility conflicts")
	}
	projectConflict := `ownerless Project is not compatible with pack jira-like-sample retry: _System_Table Category = "wrong", want "jira"`
	sprintConflict := `ownerless Sprint is not compatible with pack jira-like-sample retry: _System_Table Category = "wrong", want "jira"`
	if projectIndex, sprintIndex := strings.Index(err.Error(), projectConflict), strings.Index(err.Error(), sprintConflict); projectIndex < 0 || sprintIndex < 0 || projectIndex >= sprintIndex {
		t.Fatalf("object_spec error = %v, want Project then Sprint compatibility conflicts", err)
	}
	var adoptedFieldCount int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_Field f
JOIN _System_Object o ON o.Id = f.ObjectId
WHERE o.ApiName IN ('Issue', 'Project', 'Sprint')
  AND f.ApiName IN ('Subject', 'Name')
  AND (COALESCE(f.OwnerKind, '') <> '' OR COALESCE(f.OwnerPack, '') <> '')`).Scan(&adoptedFieldCount); err != nil {
		t.Fatalf("count rejected legacy field ownership writes: %v", err)
	}
	if adoptedFieldCount != 0 {
		t.Fatalf("rejected legacy field ownership write count = %d, want 0", adoptedFieldCount)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_Table SET Category = 'jira' WHERE TableName IN ('Project', 'Sprint')`); err != nil {
		t.Fatalf("restore compatible legacy table shape: %v", err)
	}
	if err := applyStoppedTarget(); err != nil {
		t.Fatalf("apply compatible existing surface through object_spec: %v", err)
	}
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_Field f
JOIN _System_Object o ON o.Id = f.ObjectId
WHERE o.ApiName IN ('Issue', 'Project', 'Sprint')
  AND f.ApiName IN ('Subject', 'Name')
  AND f.OwnerKind = 'pack'
  AND f.OwnerPack = ?`, legacy.Name()).Scan(&adoptedFieldCount); err != nil {
		t.Fatalf("count adopted legacy field ownership: %v", err)
	}
	if adoptedFieldCount != 3 {
		t.Fatalf("adopted legacy field ownership count = %d, want 3 for pack/%s", adoptedFieldCount, legacy.Name())
	}
}

func TestSourceBackedPackGenerationTiDBActivationIsAtomic(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	pack := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t,
		"atomic-generation-pack",
		"atomic-generation-source-revision-1",
	)
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("atomic pack payload hash: %v", err)
	}
	if err := markInstalledPackInstalling(ctx, db, pack, payloadHash); err != nil {
		t.Fatalf("atomic pack mark installing: %v", err)
	}
	var atomicPackInstallID string
	if err := db.QueryRowContext(
		ctx,
		`SELECT PackInstallId FROM _System_InstalledPack WHERE Name = ?`,
		pack.Name(),
	).Scan(&atomicPackInstallID); err != nil {
		t.Fatalf("read atomic pack installation identity: %v", err)
	}
	injected := errors.New("reject activation after generation write")
	err = markInstalledPackActiveWithHook(
		ctx,
		db,
		pack,
		payloadHash,
		func(context.Context, *sql.Tx) error {
			return injected
		},
	)
	if !errors.Is(err, injected) {
		t.Fatalf("atomic activation injected failure = %v, want %v", err, injected)
	}
	var status string
	if err := db.QueryRowContext(
		ctx,
		`SELECT Status FROM _System_InstalledPack WHERE Name = ?`,
		pack.Name(),
	).Scan(&status); err != nil {
		t.Fatalf("read rolled-back compatibility row: %v", err)
	}
	if status != installedPackStatusInstalling {
		t.Fatalf("rolled-back compatibility status = %q, want installing", status)
	}
	assertNoPackGenerationAuthority(t, ctx, db, pack.Name())
	assertNoPackGenerationConnectorBindingDefinitionSet(
		t,
		ctx,
		db,
		packGenerationID(pack.Name(), atomicPackInstallID),
	)

	if err := markInstalledPackActive(ctx, db, pack, payloadHash); err != nil {
		t.Fatalf("atomic pack activate: %v", err)
	}
	assertOnePackGenerationAuthority(t, ctx, db, pack.Name())
	assertActivePackGenerationConnectorBindingDefinitionSet(
		t, ctx, db, pack.Name(),
	)
}

func TestSourceBackedPackGenerationTiDBExactRetryRejectsMissingDefinitionSet(
	t *testing.T,
) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	pack := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t,
		"exact-retry-missing-definition-set",
		"source-revision-1",
	)
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("hash exact-retry pack: %v", err)
	}
	const packInstallID = "pack-install-exact-retry-definition-set"
	applied := appliedAppPack{pack: pack, payloadHash: payloadHash}
	seedActivatedInstalledPackCompatibility(t, ctx, db, applied, packInstallID)
	if err := publishFirstSourceBackedPackGeneration(ctx, db, applied); err != nil {
		t.Fatalf("publish exact-retry generation: %v", err)
	}
	assertActivePackGenerationConnectorBindingDefinitionSet(t, ctx, db, pack.Name())

	generationID := packGenerationID(pack.Name(), packInstallID)
	deletePackGenerationConnectorBindingDefinitionSetForTest(t, ctx, db, generationID)
	assertNoPackGenerationConnectorBindingDefinitionSet(t, ctx, db, generationID)

	err = publishFirstSourceBackedPackGeneration(ctx, db, applied)
	if !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf("missing definition-set exact retry = %v, want conflict", err)
	}
	assertNoPackGenerationConnectorBindingDefinitionSet(t, ctx, db, generationID)
}

func TestSourceBackedPackGenerationTiDBSourceRevisionAdvancePublishesNewGeneration(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	const (
		packName       = "generation-replacement-pack"
		firstRevision  = "source-revision-1"
		secondRevision = "source-revision-2"
		thirdRevision  = "source-revision-3"
		firstInstallID = "pack-install-generation-1"
	)
	firstPack := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t, packName, firstRevision,
	)
	firstHash, err := packs.PackPayloadHash(firstPack)
	if err != nil {
		t.Fatalf("first generation payload hash: %v", err)
	}
	firstApplied := appliedAppPack{pack: firstPack, payloadHash: firstHash}
	seedActivatedInstalledPackCompatibility(
		t, ctx, db, firstApplied, firstInstallID,
	)
	if err := publishFirstSourceBackedPackGeneration(ctx, db, firstApplied); err != nil {
		t.Fatalf("publish first generation: %v", err)
	}
	assertActivePackGenerationConnectorBindingDefinitionSet(
		t, ctx, db, packName,
	)
	// The From generation models persisted authority from before definition-set
	// publication. The forward transition must accept it without silently
	// backfilling it, while publishing the new target with a complete set.
	firstGenerationID := packGenerationID(packName, firstInstallID)
	deletePackGenerationConnectorBindingDefinitionSetForTest(
		t, ctx, db, firstGenerationID,
	)
	assertNoPackGenerationConnectorBindingDefinitionSet(
		t, ctx, db, firstGenerationID,
	)

	firstGeneration, found, err := readPublishedPackGeneration(
		ctx, db, firstInstallID,
	)
	if err != nil || !found {
		t.Fatalf("read first generation before advance: found=%v err=%v", found, err)
	}

	secondPack := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t, packName, secondRevision,
	)
	runApplyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		secondPack,
		firstRevision,
		"publish second immutable pack generation",
		func(revision string) *packs.ResolvedPackSource {
			return testControlplaneOfficialRepoEmptyResolvedSourceNamed(packName, revision)
		},
	)
	assertPackGenerationReplacement(
		t, ctx, db, packName, firstInstallID, firstRevision, secondRevision, 2,
	)
	assertNoPackGenerationConnectorBindingDefinitionSet(
		t, ctx, db, firstGenerationID,
	)
	firstGenerationAfter, found, err := readPublishedPackGeneration(
		ctx, db, firstInstallID,
	)
	if err != nil || !found {
		t.Fatalf("read first generation after advance: found=%v err=%v", found, err)
	}
	if !sourceBackedPackGenerationsEqual(firstGenerationAfter, firstGeneration) {
		t.Fatal("source revision advance mutated the first immutable generation")
	}

	// Restarting the same authorized edge is a no-op: it must not mint another
	// generation or advance the activation epoch.
	runApplyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		secondPack,
		firstRevision,
		"restart second immutable pack generation",
		func(revision string) *packs.ResolvedPackSource {
			return testControlplaneOfficialRepoEmptyResolvedSourceNamed(packName, revision)
		},
	)
	assertPackGenerationReplacement(
		t, ctx, db, packName, firstInstallID, firstRevision, secondRevision, 2,
	)

	thirdPack := testControlplaneOfficialRepoEmptySourceBackedPackNamed(
		t, packName, thirdRevision,
	)
	runApplyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		thirdPack,
		secondRevision,
		"publish third immutable pack generation",
		func(revision string) *packs.ResolvedPackSource {
			return testControlplaneOfficialRepoEmptyResolvedSourceNamed(packName, revision)
		},
	)
	assertPackGenerationReplacement(
		t, ctx, db, packName, "", secondRevision, thirdRevision, 3,
	)
}

func TestReapplyPackFromSourceTiDBPendingLocalPeerRequiresConvergenceBeforeAuthorizedRevisionAdvance(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)
	setDatabaseEnvFromConfig(t, cfg)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	const (
		officialName       = "staging-official-dependency"
		companionName      = "staging-local-companion"
		officialInstallID  = "staging-official-generation-1"
		companionInstallID = "staging-companion-generation-1"
	)
	writeFixturePack := func(root, name, version string, dependencies []packs.PackDependency) {
		t.Helper()
		if err := os.MkdirAll(root, 0o755); err != nil {
			t.Fatalf("create %s pack root: %v", name, err)
		}
		category := packs.ConnectorPack
		components := []packs.ManifestComponent{{
			Kind:          packs.ComponentKindConnectorCalloutProfile,
			BusinessKey:   officialName + "/default",
			Manageability: packs.ManageabilityLocked,
			Payload:       json.RawMessage(`{"profileName":"default","credentialRef":"credential-a","endpointClass":"crm.read","description":"Legacy connector fixture","retryPolicy":{"maxAttempts":3,"baseDelay":1000000,"multiplier":2,"maxDelay":10000000,"retryableStatusCodes":[429,503]}}`),
		}}
		objectBusinessKey := "staging-official-object"
		objectPayload := json.RawMessage(`[{"apiName":"StagingOfficialObject","label":"Staging Official Object","tableType":"standard_object","category":"test","sharingModel":"Private","enableHierarchySharing":false,"description":"Official transition fixture","fields":[{"apiName":"Name","label":"Name","type":"string","required":true,"isNameField":true}]}]`)
		if name == companionName {
			category = packs.AppPack
			objectBusinessKey = "staging-companion-object"
			objectPayload = json.RawMessage(`[{"apiName":"StagingCompanionObject","label":"Staging Companion Object","tableType":"standard_object","category":"test","sharingModel":"Private","enableHierarchySharing":false,"description":"Pending local peer fixture","fields":[{"apiName":"Name","label":"Name","type":"string","required":true,"isNameField":true}]}]`)
			components = []packs.ManifestComponent{{
				Kind:          packs.ComponentKindObjectSpec,
				BusinessKey:   objectBusinessKey,
				Manageability: packs.ManageabilityLocked,
				Payload:       objectPayload,
			}}
		}
		manifest, err := json.MarshalIndent(packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  name,
			DisplayName:           name,
			Category:              category,
			Version:               version,
			PlatformCompatibility: packs.PlatformCompatibility{Min: "0.2.0", MaxExclusive: "0.3.0"},
			Dependencies:          dependencies,
			Components:            components,
		}, "", "  ")
		if err != nil {
			t.Fatalf("marshal %s manifest: %v", name, err)
		}
		if err := os.WriteFile(filepath.Join(root, "pack.json"), append(manifest, '\n'), 0o644); err != nil {
			t.Fatalf("write %s manifest: %v", name, err)
		}
	}
	loadPack := func(label, rawSource string, options packs.PackSourceOptions) *packs.SourceBackedPack {
		t.Helper()
		report := packs.ValidateInstallablePackSourceWithOptions(rawSource, options)
		if err := installableValidationReportError(label, report); err != nil {
			t.Fatal(err)
		}
		return report.Pack
	}
	seedGeneration := func(pack *packs.SourceBackedPack, installID string) string {
		t.Helper()
		payloadHash, err := packs.PackPayloadHash(pack)
		if err != nil {
			t.Fatalf("%s payload hash: %v", pack.Name(), err)
		}
		applied := appliedAppPack{pack: pack, payloadHash: payloadHash}
		seedActivatedInstalledPackCompatibility(t, ctx, db, applied, installID)
		if err := publishFirstSourceBackedPackGeneration(ctx, db, applied); err != nil {
			t.Fatalf("publish %s first generation: %v", pack.Name(), err)
		}
		return payloadHash
	}

	officialRoot := filepath.Join(t.TempDir(), "official")
	officialPackRoot := filepath.Join(officialRoot, officialName)
	writeFixturePack(officialPackRoot, officialName, "0.1.0", nil)
	initGitTestRepoForControlplane(t, officialRoot)
	fromRevision := gitTestCommitForControlplane(t, officialRoot, "official revision N")
	if err := os.WriteFile(filepath.Join(officialRoot, "REVISION"), []byte("N+1\n"), 0o644); err != nil {
		t.Fatalf("write official target revision marker: %v", err)
	}
	toRevision := gitTestCommitForControlplane(t, officialRoot, "official revision N+1")
	t.Setenv(packs.OfficialAppPacksRootEnvVar, officialRoot)

	officialOptions := packs.DefaultPackSourceOptions()
	officialOptions.OfficialRepoRoot = officialRoot
	officialOptions.OfficialRepoSourceRevisions = map[string]string{officialName: fromRevision}
	officialAtN := loadPack("official revision N", "official_repo:"+officialName, officialOptions)
	seedGeneration(officialAtN, officialInstallID)
	const oldCompilerRuntimeProjection = `[{"kind":"connector_callout_profile","businessKey":"default"}]`
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack
SET RuntimeStartupShallowDeclarations = CAST(? AS JSON)
WHERE Name = ?`, oldCompilerRuntimeProjection, officialName); err != nil {
		t.Fatalf("seed official old compiler compatibility projection: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_PackGeneration
SET RuntimeStartupShallowDeclarations = CAST(? AS JSON)
WHERE PackInstallId = ?`, oldCompilerRuntimeProjection, officialInstallID); err != nil {
		t.Fatalf("seed official old compiler generation projection: %v", err)
	}
	firstGeneration, found, err := readPublishedPackGeneration(ctx, db, officialInstallID)
	if err != nil || !found {
		t.Fatalf("read official generation N fixture: found=%v err=%v", found, err)
	}

	companionRoot := filepath.Join(t.TempDir(), "companion")
	dependencies := []packs.PackDependency{{Name: officialName, VersionConstraint: ">=0.1.0 <0.2.0"}}
	writeFixturePack(companionRoot, companionName, "0.1.0", dependencies)
	initGitTestRepoForControlplane(t, companionRoot)
	gitTestCommitForControlplane(t, companionRoot, "companion revision N")
	companionAtN := loadPack("companion revision N", "local_path:"+companionRoot, packs.DefaultPackSourceOptions())
	oldCompanionHash, err := packs.PackPayloadHash(companionAtN)
	if err != nil {
		t.Fatalf("companion revision N payload hash: %v", err)
	}
	if err := markInstalledPackInstallingWithPackInstallID(ctx, db, companionAtN, oldCompanionHash, companionInstallID); err != nil {
		t.Fatalf("seed companion installing compatibility: %v", err)
	}
	if err := markInstalledPackActive(ctx, db, companionAtN, oldCompanionHash); err != nil {
		t.Fatalf("seed companion active compatibility: %v", err)
	}

	writeFixturePack(companionRoot, companionName, "0.1.1", dependencies)
	gitTestCommitForControlplane(t, companionRoot, "companion revision N+1")
	companionAtTarget := loadPack("companion target", "local_path:"+companionRoot, packs.DefaultPackSourceOptions())
	targetCompanionHash, err := packs.PackPayloadHash(companionAtTarget)
	if err != nil {
		t.Fatalf("companion target payload hash: %v", err)
	}
	if targetCompanionHash == oldCompanionHash {
		t.Fatal("companion fixture did not produce payload drift")
	}
	targetCompanionMeta := packs.PackHashMetadataForPack(companionAtTarget)
	targetCompanionRuntime, err := packs.BuildPersistedRuntimeDeclaration(companionAtTarget)
	if err != nil {
		t.Fatalf("build companion target runtime declaration: %v", err)
	}
	targetCompanionHostDigest, err := packs.PackHostRequirementsDigest(companionAtTarget)
	if err != nil {
		t.Fatalf("build companion target host requirements digest: %v", err)
	}
	nullableString := func(value string, valid bool) any {
		if !valid {
			return nil
		}
		return value
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack SET
PackVersion = ?, ManifestFormatVersion = ?, PlatformCompatibility = ?,
DesiredPayloadHash = ?, PayloadHashAlgorithm = ?, SourceKind = ?, SourceURI = ?,
SourceRevision = ?, HostRequirementsDigest = ?,
RuntimeObjectSpecDeclarations = CAST(? AS JSON), RuntimeRelatedListRules = CAST(? AS JSON),
RuntimeSearchScopeOrder = CAST(? AS JSON), RuntimeSearchAllDefaultFields = CAST(? AS JSON),
RuntimeFormulaFunctionBundles = CAST(? AS JSON), RuntimeStartupShallowDeclarations = CAST(? AS JSON)
WHERE Name = ?`,
		targetCompanionMeta.PackVersion,
		targetCompanionMeta.ManifestFormatVersion,
		nullableString(targetCompanionMeta.PlatformCompatibility, targetCompanionMeta.PlatformCompatibilityValid),
		targetCompanionHash,
		targetCompanionMeta.PayloadHashAlgorithm,
		nullableString(targetCompanionMeta.SourceKind, targetCompanionMeta.SourceKindValid),
		nullableString(targetCompanionMeta.SourceURI, targetCompanionMeta.SourceURIValid),
		nullableString(targetCompanionMeta.SourceRevision, targetCompanionMeta.SourceRevisionValid),
		targetCompanionHostDigest,
		targetCompanionRuntime.ObjectPayloadsJSON(),
		targetCompanionRuntime.RelatedListRulesJSON(),
		targetCompanionRuntime.SearchScopeOrderJSON(),
		targetCompanionRuntime.SearchAllDefaultFieldsJSON(),
		targetCompanionRuntime.FormulaFunctionBundlesJSON(),
		targetCompanionRuntime.StartupShallowDeclaredRowsJSON(),
		companionName,
	); err != nil {
		t.Fatalf("seed staging pending companion convergence: %v", err)
	}
	readOfficialInstalledAuthority := func(stage string) sourceBackedPackGeneration {
		t.Helper()
		tx, err := db.BeginTx(ctx, nil)
		if err != nil {
			t.Fatalf("begin official InstalledPack authority read %s: %v", stage, err)
		}
		defer func() { _ = tx.Rollback() }()
		authority, found, err := readInstalledPackGenerationPreimageForUpdate(ctx, tx, officialName)
		if err != nil || !found {
			t.Fatalf("read official InstalledPack authority %s: found=%v err=%v", stage, found, err)
		}
		return authority
	}
	readCompanionInstalledAuthority := func(stage string) installedPackRow {
		t.Helper()
		rows, err := listInstalledSourceBackedPacks(ctx, db)
		if err != nil {
			t.Fatalf("list companion InstalledPack authority %s: %v", stage, err)
		}
		var authority installedPackRow
		for _, row := range rows {
			if row.Name == companionName {
				authority = row
				break
			}
		}
		identity, found, err := getInstalledPackRowByNameForSourceRevisionAdvance(ctx, db, companionName)
		if err != nil || !found || authority.Name == "" {
			t.Fatalf("read companion InstalledPack authority %s: found=%v err=%v", stage, found, err)
		}
		authority.ID = identity.ID
		authority.PackInstallID = identity.PackInstallID
		authority.IsDeleted = identity.IsDeleted
		return authority
	}
	initialOfficialInstalled := readOfficialInstalledAuthority("before convergence")
	initialCompanionInstalled := readCompanionInstalledAuthority("before convergence")
	assertFirstGenerationUnchanged := func(stage string) {
		t.Helper()
		got, found, err := readPublishedPackGeneration(ctx, db, officialInstallID)
		if err != nil || !found {
			t.Fatalf("read retained official generation N %s: found=%v err=%v", stage, found, err)
		}
		if !sourceBackedPackGenerationsEqual(got, firstGeneration) {
			t.Fatalf("official generation N changed %s", stage)
		}
	}

	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("construct staging reproduction runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh staging reproduction runner: %v", err)
	}
	err = ReapplyPackFromSource(ctx, runner, "official_repo:"+officialName, InstallPackOptions{
		SourceRevisionAdvance: &SourceRevisionAdvanceAuthorization{
			FromRevision: fromRevision,
			ToRevision:   toRevision,
		},
		VerifyAfterApply: true,
	})
	if err == nil || !strings.Contains(err.Error(),
		"load source revision tuple formula context: installed source-backed pack "+companionName+" payload hash: v1 payload hash mismatch") {
		t.Fatalf("authorized official advance error = %v, want pending local peer payload hash mismatch", err)
	}
	t.Logf("reproduced Formal Staging pending local peer failure: applied=%q desired=%q error=%v", oldCompanionHash, targetCompanionHash, err)

	var appliedCompanionHash string
	if err := db.QueryRowContext(ctx, `SELECT AppliedPayloadHash FROM _System_InstalledPack WHERE Name = ?`, companionName).Scan(&appliedCompanionHash); err != nil {
		t.Fatalf("read companion applied hash after reproduction: %v", err)
	}
	if appliedCompanionHash != oldCompanionHash {
		t.Fatalf("failed official advance changed companion applied hash = %q, want %q", appliedCompanionHash, oldCompanionHash)
	}
	if got := readCompanionInstalledAuthority("after failed official-first advance"); got != initialCompanionInstalled {
		t.Fatalf("failed official-first advance changed companion InstalledPack authority: got %#v want %#v", got, initialCompanionInstalled)
	}
	if got := readOfficialInstalledAuthority("after failed official-first advance"); !sourceBackedPackGenerationsEqual(got, initialOfficialInstalled) {
		t.Fatalf("failed official-first advance changed InstalledPack authority: got %#v want %#v", got, initialOfficialInstalled)
	}
	assertFirstGenerationUnchanged("after failed official-first advance")
	assertPackGenerationReplacement(t, ctx, db, officialName, officialInstallID, fromRevision, fromRevision, 1)

	const unrelatedRuntimeProjection = `[{"kind":"permission","businessKey":"OtherPermission"}]`
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack
SET RuntimeStartupShallowDeclarations = CAST(? AS JSON)
WHERE Name = ?`, unrelatedRuntimeProjection, officialName); err != nil {
		t.Fatalf("seed InstalledPack-only runtime projection corruption: %v", err)
	}
	err = ReapplyPackFromSource(ctx, runner, "local_path:"+companionRoot, InstallPackOptions{VerifyAfterApply: true})
	if !errors.Is(err, ErrPackGenerationConflict) || !strings.Contains(err.Error(), "InstalledPack runtime declaration does not match its active generation") {
		t.Fatalf("InstalledPack-only projection corruption error = %v, want generation authority conflict", err)
	}
	if got := readCompanionInstalledAuthority("after rejected corrupt peer"); got != initialCompanionInstalled {
		t.Fatalf("rejected corrupt peer changed companion InstalledPack authority: got %#v want %#v", got, initialCompanionInstalled)
	}
	assertFirstGenerationUnchanged("after rejected InstalledPack-only projection corruption")
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack
SET RuntimeStartupShallowDeclarations = CAST(? AS JSON)
WHERE Name = ?`, oldCompilerRuntimeProjection, officialName); err != nil {
		t.Fatalf("restore generation-bound old compiler projection: %v", err)
	}

	candidateVerifyCalls := 0
	_, err = ApplyOfflineCandidate(ctx, conn, OfflineCandidateRequest{
		OperationID: "staging-offline-candidate",
		UpgradeMode: UpgradeModeOfflineRequired,
		PackSources: []string{
			"local_path:" + companionRoot,
			"official_repo:" + officialName,
		},
	}, InstallPackOptions{
		VerifyAfterApply: true,
		CandidateFinalVerify: func(context.Context, *Runner, []packs.Pack) error {
			candidateVerifyCalls++
			return nil
		},
	})
	if err == nil || !strings.Contains(err.Error(), "materialization "+companionName+" depends on source revision advance "+officialName) {
		t.Fatalf("reverse-phase offline candidate error = %v, want fail-closed dependency order", err)
	}
	if candidateVerifyCalls != 0 {
		t.Fatalf("candidate final verify calls = %d, want rejection before mutation", candidateVerifyCalls)
	}
	verificationConn := openIntegrationConnection(t, cfg)
	t.Cleanup(func() { _ = verificationConn.Close() })
	db = verificationConn.DB()
	if got := readCompanionInstalledAuthority("after rejected reverse-phase candidate"); got != initialCompanionInstalled {
		t.Fatalf("rejected reverse-phase candidate changed companion authority: got %#v want %#v", got, initialCompanionInstalled)
	}
	if got := readOfficialInstalledAuthority("after rejected reverse-phase candidate"); !sourceBackedPackGenerationsEqual(got, initialOfficialInstalled) {
		t.Fatalf("rejected reverse-phase candidate changed official authority: got %#v want %#v", got, initialOfficialInstalled)
	}
	assertFirstGenerationUnchanged("after rejected reverse-phase candidate")
}

func TestReapplyPackFromSourceTiDBAuthorizedRevisionAdvanceResumesFailedOldCompilerProjection(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	db := conn.DB()

	const packName = "reapply-generation-replacement-pack"
	officialRoot := t.TempDir()
	packRoot := filepath.Join(officialRoot, packName)
	if err := os.MkdirAll(packRoot, 0o755); err != nil {
		t.Fatalf("create official pack root: %v", err)
	}
	manifest, err := json.MarshalIndent(packs.PackManifest{
		ManifestFormatVersion: packs.SupportedManifestFormatVersion,
		Name:                  packName,
		DisplayName:           "Reapply Generation Replacement Pack",
		Category:              packs.AppPack,
		Version:               "0.1.0",
		PlatformCompatibility: packs.PlatformCompatibility{Min: "0.2.0", MaxExclusive: "0.3.0"},
		Dependencies:          []packs.PackDependency{},
		Components: []packs.ManifestComponent{{
			Kind:          packs.ComponentKindObjectSpec,
			BusinessKey:   "issue",
			Manageability: packs.ManageabilityLocked,
			Payload:       json.RawMessage(`[{"apiName":"Issue","label":"Issue","tableType":"standard_object","category":"test","sharingModel":"Private","enableHierarchySharing":false,"description":"Issue","fields":[{"apiName":"Subject","label":"Subject","type":"string","required":true,"isNameField":true}]}]`),
		}},
	}, "", "  ")
	if err != nil {
		t.Fatalf("marshal official pack manifest: %v", err)
	}
	if err := os.WriteFile(filepath.Join(packRoot, "pack.json"), append(manifest, '\n'), 0o644); err != nil {
		t.Fatalf("write official pack manifest: %v", err)
	}
	initGitTestRepoForControlplane(t, officialRoot)
	fromRevision := gitTestCommitForControlplane(t, officialRoot, "from revision")
	if err := os.WriteFile(filepath.Join(officialRoot, "REVISION"), []byte("target\n"), 0o644); err != nil {
		t.Fatalf("write target revision marker: %v", err)
	}
	toRevision := gitTestCommitForControlplane(t, officialRoot, "to revision")
	t.Setenv(packs.OfficialAppPacksRootEnvVar, officialRoot)

	sourceOptions := packs.DefaultPackSourceOptions()
	sourceOptions.OfficialRepoRoot = officialRoot
	sourceOptions.OfficialRepoSourceRevisions = map[string]string{packName: fromRevision}
	fromReport := packs.ValidateInstallablePackSourceWithOptions("official_repo:"+packName, sourceOptions)
	if err := installableValidationReportError("from revision fixture", fromReport); err != nil {
		t.Fatal(err)
	}
	fromHash, err := packs.PackPayloadHash(fromReport.Pack)
	if err != nil {
		t.Fatalf("from revision payload hash: %v", err)
	}
	const firstInstallID = "pack-install-old-compiler"
	firstApplied := appliedAppPack{pack: fromReport.Pack, payloadHash: fromHash}
	seedActivatedInstalledPackCompatibility(t, ctx, db, firstApplied, firstInstallID)
	if err := publishFirstSourceBackedPackGeneration(ctx, db, firstApplied); err != nil {
		t.Fatalf("publish old compiler generation: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_PackGeneration
SET RuntimeStartupShallowDeclarations = ?
WHERE PackInstallId = ?`, `[{"kind":"old_compiler_projection"}]`, firstInstallID); err != nil {
		t.Fatalf("simulate old compiler projection: %v", err)
	}
	firstGeneration, found, err := readPublishedPackGeneration(ctx, db, firstInstallID)
	if err != nil || !found {
		t.Fatalf("read old compiler generation: found=%v err=%v", found, err)
	}

	sourceOptions.OfficialRepoSourceRevisions[packName] = toRevision
	toReport := packs.ValidateInstallablePackSourceWithOptions("official_repo:"+packName, sourceOptions)
	if err := installableValidationReportError("to revision fixture", toReport); err != nil {
		t.Fatal(err)
	}
	toHash, err := packs.PackPayloadHash(toReport.Pack)
	if err != nil {
		t.Fatalf("to revision payload hash: %v", err)
	}
	targetGeneration, err := sourceBackedPackGenerationFromVerifiedPack(appliedAppPack{
		pack: toReport.Pack, payloadHash: toHash,
	})
	if err != nil {
		t.Fatalf("build failed target incarnation: %v", err)
	}
	const failedInstallID = "pack-install-failed-target"
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack SET
PackInstallId = ?, Status = ?, PackVersion = ?, ManifestFormatVersion = ?,
PlatformCompatibility = ?, DesiredPayloadHash = ?, PayloadHashAlgorithm = ?,
SourceKind = ?, SourceURI = ?, SourceRevision = ?, HostRequirementsDigest = ?,
RuntimeObjectSpecDeclarations = ?, RuntimeRelatedListRules = ?,
RuntimeSearchScopeOrder = ?, RuntimeSearchAllDefaultFields = ?,
RuntimeFormulaFunctionBundles = ?, RuntimeStartupShallowDeclarations = ?,
LastError = 'interrupted target apply'
WHERE Name = ?`,
		failedInstallID, installedPackStatusFailed,
		nullableValue(targetGeneration.packVersion),
		nullableValue(targetGeneration.manifestFormatVersion),
		nullableValue(targetGeneration.platformCompatibility),
		nullableValue(targetGeneration.desiredPayloadHash),
		nullableValue(targetGeneration.payloadHashAlgorithm),
		nullableValue(targetGeneration.sourceKind),
		nullableValue(targetGeneration.sourceURI),
		nullableValue(targetGeneration.sourceRevision),
		nullableValue(targetGeneration.hostRequirementsDigest),
		nullableValue(targetGeneration.runtimeObjectSpecDecl),
		nullableValue(targetGeneration.runtimeRelatedLists),
		nullableValue(targetGeneration.runtimeSearchScope),
		nullableValue(targetGeneration.runtimeSearchAllDefault),
		nullableValue(targetGeneration.runtimeFormulaBundles),
		nullableValue(targetGeneration.runtimeStartupShallow), packName,
	); err != nil {
		t.Fatalf("seed failed target incarnation: %v", err)
	}

	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("construct reapply runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh reapply runner: %v", err)
	}
	reapplyOptions := InstallPackOptions{
		SourceRevisionAdvance: &SourceRevisionAdvanceAuthorization{
			FromRevision: fromRevision,
			ToRevision:   toRevision,
		},
		VerifyAfterApply: true,
	}
	if err := ReapplyPackFromSource(ctx, runner, "official_repo:"+packName, reapplyOptions); err != nil {
		t.Fatalf("authorized public reapply across compiler projection drift: %v", err)
	}

	assertPackGenerationReplacement(
		t, ctx, db, packName, firstInstallID, fromRevision, toRevision, 2,
	)
	firstGenerationAfter, found, err := readPublishedPackGeneration(ctx, db, firstInstallID)
	if err != nil || !found {
		t.Fatalf("read old compiler generation after reapply: found=%v err=%v", found, err)
	}
	if !sourceBackedPackGenerationsEqual(firstGenerationAfter, firstGeneration) {
		t.Fatal("authorized public reapply mutated the old compiler generation")
	}

	if _, err := db.ExecContext(ctx, `UPDATE _System_Field f
JOIN _System_Object o ON o.Id = f.ObjectId
SET f.Required = 0
WHERE o.ApiName = 'Issue' AND f.ApiName = 'Subject'`); err != nil {
		t.Fatalf("drift target field before no-transition reapply: %v", err)
	}
	if _, err := db.ExecContext(ctx, `UPDATE _System_InstalledPack
SET SourceKind = 'in_repo'
WHERE Name = ?`, packName); err != nil {
		t.Fatalf("seed target source alias before no-transition reapply: %v", err)
	}
	if err := ReapplyPackFromSource(ctx, runner, "official_repo:"+packName, reapplyOptions); err != nil {
		t.Fatalf("authorized public reapply at completed target: %v", err)
	}
	var required int
	if err := db.QueryRowContext(ctx, `SELECT f.Required FROM _System_Field f
JOIN _System_Object o ON o.Id = f.ObjectId
WHERE o.ApiName = 'Issue' AND f.ApiName = 'Subject'`).Scan(&required); err != nil {
		t.Fatalf("read target field after no-transition reapply: %v", err)
	}
	if required != 1 {
		t.Fatalf("target field Required = %d, want repaired ordinary reapply value 1", required)
	}
	assertPackGenerationReplacement(
		t, ctx, db, packName, firstInstallID, fromRevision, toRevision, 2,
	)
}

func TestApplyOfflineCandidateTiDBOrdersRevisionAdvanceBeforeDependentMaterialization(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)
	setDatabaseEnvFromConfig(t, cfg)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })

	const (
		providerName  = "staging-official-dependency"
		dependentName = "staging-local-companion"
	)
	writePack := func(root, name, version, objectAPIName, description string, dependencies []packs.PackDependency) {
		t.Helper()
		if err := os.MkdirAll(root, 0o755); err != nil {
			t.Fatalf("create %s pack root: %v", name, err)
		}
		manifest, err := json.MarshalIndent(packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  name,
			DisplayName:           name,
			Category:              packs.AppPack,
			Version:               version,
			PlatformCompatibility: packs.PlatformCompatibility{Min: "0.2.0", MaxExclusive: "0.3.0"},
			Dependencies:          dependencies,
			Components: []packs.ManifestComponent{{
				Kind:          packs.ComponentKindObjectSpec,
				BusinessKey:   strings.ToLower(objectAPIName),
				Manageability: packs.ManageabilityLocked,
				Payload: json.RawMessage(fmt.Sprintf(
					`[{"apiName":%q,"label":%q,"tableType":"standard_object","category":"test","sharingModel":"Private","enableHierarchySharing":false,"description":%q,"fields":[{"apiName":"Name","label":"Name","type":"string","required":true,"isNameField":true}]}]`,
					objectAPIName, objectAPIName, description,
				)),
			}},
		}, "", "  ")
		if err != nil {
			t.Fatalf("marshal %s manifest: %v", name, err)
		}
		if err := os.WriteFile(filepath.Join(root, "pack.json"), append(manifest, '\n'), 0o644); err != nil {
			t.Fatalf("write %s manifest: %v", name, err)
		}
	}

	officialRoot := filepath.Join(t.TempDir(), "official")
	providerRoot := filepath.Join(officialRoot, providerName)
	writePack(providerRoot, providerName, "0.1.0", "StagingOfficialDependency", "revision N", nil)
	initGitTestRepoForControlplane(t, officialRoot)
	fromRevision := gitTestCommitForControlplane(t, officialRoot, "official revision N")
	t.Setenv(packs.OfficialAppPacksRootEnvVar, officialRoot)

	dependentRoot := filepath.Join(t.TempDir(), dependentName)
	writePack(dependentRoot, dependentName, "0.1.0", "StagingLocalCompanion", "revision N", []packs.PackDependency{{
		Name: providerName, VersionConstraint: ">=0.1.0 <0.2.0",
	}})
	initGitTestRepoForControlplane(t, dependentRoot)
	gitTestCommitForControlplane(t, dependentRoot, "dependent revision N")

	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("construct initial runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh initial runner: %v", err)
	}
	if err := InstallPackFromSourceWithOptions(ctx, runner, "official_repo:"+providerName, InstallPackOptions{VerifyAfterApply: true}); err != nil {
		t.Fatalf("install provider revision N: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh runner after provider install: %v", err)
	}
	if err := InstallPackFromSourceWithOptions(ctx, runner, "local_path:"+dependentRoot, InstallPackOptions{VerifyAfterApply: true}); err != nil {
		t.Fatalf("install dependent revision N: %v", err)
	}

	writePack(providerRoot, providerName, "0.1.1", "StagingOfficialDependency", "revision N+1", nil)
	toRevision := gitTestCommitForControlplane(t, officialRoot, "official revision N+1")
	writePack(dependentRoot, dependentName, "0.1.1", "StagingLocalCompanion", "revision N+1", []packs.PackDependency{{
		Name: providerName, VersionConstraint: ">=0.1.1 <0.2.0",
	}})
	gitTestCommitForControlplane(t, dependentRoot, "dependent revision N+1")

	verifyCalls := 0
	forcedVerifyFailure := errors.New("forced mixed-candidate final verification failure")
	request := OfflineCandidateRequest{
		OperationID: "staging-mixed-transition-order",
		UpgradeMode: UpgradeModeOfflineRequired,
		PackSources: []string{
			"local_path:" + dependentRoot,
			"official_repo:" + providerName,
		},
	}
	options := InstallPackOptions{
		CandidateFinalVerify: func(verifyCtx context.Context, candidateRunner *Runner, candidates []packs.Pack) error {
			verifyCalls++
			if len(candidates) != 2 || candidates[0].Name() != providerName || candidates[1].Name() != dependentName {
				return fmt.Errorf("candidate order = %#v, want %s then %s", candidates, providerName, dependentName)
			}
			var providerStatus, providerRevision string
			if err := candidateRunner.DB.DB().QueryRowContext(
				verifyCtx,
				`SELECT Status, SourceRevision FROM _System_InstalledPack WHERE Name = ? AND COALESCE(IsDeleted, 0) = 0`,
				providerName,
			).Scan(&providerStatus, &providerRevision); err != nil {
				return fmt.Errorf("read provider pre-activation state: %w", err)
			}
			if providerStatus != installedPackStatusActive || providerRevision != toRevision {
				return fmt.Errorf("provider pre-activation state = %s/%s, want active/%s", providerStatus, providerRevision, toRevision)
			}
			var dependentStatus string
			if err := candidateRunner.DB.DB().QueryRowContext(
				verifyCtx,
				`SELECT Status FROM _System_InstalledPack WHERE Name = ? AND COALESCE(IsDeleted, 0) = 0`,
				dependentName,
			).Scan(&dependentStatus); err != nil {
				return fmt.Errorf("read dependent pre-activation state: %w", err)
			}
			if dependentStatus != installedPackStatusInstalling {
				return fmt.Errorf("dependent pre-activation status = %s, want %s", dependentStatus, installedPackStatusInstalling)
			}
			if verifyCalls == 1 {
				return forcedVerifyFailure
			}
			return nil
		},
	}
	failedReceipt, err := ApplyOfflineCandidate(ctx, conn, request, options)
	if !errors.Is(err, forcedVerifyFailure) {
		t.Fatalf("first mixed candidate error = %v, want forced final verification failure", err)
	}
	if failedReceipt.Activated || failedReceipt.Verified {
		t.Fatalf("failed mixed candidate receipt = %#v, want no verified activation", failedReceipt)
	}
	retryConn := openIntegrationConnection(t, cfg)
	t.Cleanup(func() { _ = retryConn.Close() })
	var providerStatusAfterFailure, providerRevisionAfterFailure string
	if err := retryConn.DB().QueryRowContext(
		ctx,
		`SELECT Status, SourceRevision FROM _System_InstalledPack WHERE Name = ? AND COALESCE(IsDeleted, 0) = 0`,
		providerName,
	).Scan(&providerStatusAfterFailure, &providerRevisionAfterFailure); err != nil {
		t.Fatalf("read provider state after failed suffix: %v", err)
	}
	if providerStatusAfterFailure != installedPackStatusActive || providerRevisionAfterFailure != toRevision {
		t.Fatalf(
			"provider after failed suffix = %s/%s, want retained active/%s",
			providerStatusAfterFailure, providerRevisionAfterFailure, toRevision,
		)
	}

	receipt, err := ApplyOfflineCandidate(ctx, retryConn, request, options)
	if err != nil {
		t.Fatalf("retry dependency-ordered mixed offline candidate: %v", err)
	}
	if verifyCalls != 2 {
		t.Fatalf("candidate final verify calls = %d, want 2 across failed attempt and retry", verifyCalls)
	}
	if !receipt.Verified || !receipt.Activated || receipt.ActivatedStateDigest != receipt.TargetStateDigest {
		t.Fatalf("mixed candidate receipt = %#v, want verified activation", receipt)
	}

	verificationConn := openIntegrationConnection(t, cfg)
	t.Cleanup(func() { _ = verificationConn.Close() })
	for _, expected := range []struct {
		name     string
		revision string
	}{
		{name: providerName, revision: toRevision},
		{name: dependentName},
	} {
		var status, revision string
		if err := verificationConn.DB().QueryRowContext(
			ctx,
			`SELECT Status, COALESCE(SourceRevision, '') FROM _System_InstalledPack WHERE Name = ? AND COALESCE(IsDeleted, 0) = 0`,
			expected.name,
		).Scan(&status, &revision); err != nil {
			t.Fatalf("read activated pack %s: %v", expected.name, err)
		}
		if status != installedPackStatusActive || (expected.revision != "" && revision != expected.revision) {
			t.Fatalf("activated pack %s = %s/%s, want active/%s", expected.name, status, revision, expected.revision)
		}
		if err := VerifyInstalledPackByName(ctx, verificationConn.DB(), expected.name, true); err != nil {
			t.Fatalf("verify activated pack %s: %v", expected.name, err)
		}
	}
	if fromRevision == toRevision {
		t.Fatalf("official fixture revisions did not advance: %s", fromRevision)
	}
}

func TestApplyOfflineCandidateTiDBInstallsCompletePackSetOnEmptyDatabase(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)
	setDatabaseEnvFromConfig(t, cfg)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
	t.Cleanup(cancel)
	conn := openIntegrationConnection(t, cfg)
	t.Cleanup(func() { _ = conn.Close() })

	const (
		baseName      = "empty-candidate-base"
		dependentName = "empty-candidate-dependent"
	)
	repoRoot := t.TempDir()
	writePack := func(name, objectAPIName string, dependencies []packs.PackDependency) string {
		t.Helper()
		root := filepath.Join(repoRoot, name)
		if err := os.MkdirAll(root, 0o755); err != nil {
			t.Fatalf("create %s pack root: %v", name, err)
		}
		manifest, err := json.MarshalIndent(packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  name,
			DisplayName:           name,
			Category:              packs.AppPack,
			Version:               "1.0.0",
			PlatformCompatibility: packs.PlatformCompatibility{Min: "0.2.0", MaxExclusive: "0.3.0"},
			Dependencies:          dependencies,
			Components: []packs.ManifestComponent{{
				Kind:          packs.ComponentKindObjectSpec,
				BusinessKey:   strings.ToLower(objectAPIName),
				Manageability: packs.ManageabilityLocked,
				Payload: json.RawMessage(fmt.Sprintf(
					`[{"apiName":%q,"label":%q,"tableType":"standard_object","category":"test","sharingModel":"Private","enableHierarchySharing":false,"fields":[{"apiName":"Name","label":"Name","type":"string","required":true,"isNameField":true}]}]`,
					objectAPIName, objectAPIName,
				)),
			}},
		}, "", "  ")
		if err != nil {
			t.Fatalf("marshal %s manifest: %v", name, err)
		}
		if err := os.WriteFile(filepath.Join(root, "pack.json"), append(manifest, '\n'), 0o644); err != nil {
			t.Fatalf("write %s manifest: %v", name, err)
		}
		return root
	}
	baseRoot := writePack(baseName, "EmptyCandidateBase", nil)
	dependentRoot := writePack(dependentName, "EmptyCandidateDependent", []packs.PackDependency{{Name: baseName}})
	initGitTestRepoForControlplane(t, repoRoot)
	gitTestCommitForControlplane(t, repoRoot, "empty candidate target")

	candidateVerifyCalls := 0
	receipt, err := ApplyOfflineCandidate(ctx, conn, OfflineCandidateRequest{
		OperationID: "empty-database-offline-candidate",
		UpgradeMode: UpgradeModeOfflineRequired,
		PackSources: []string{
			"local_path:" + dependentRoot,
			"local_path:" + baseRoot,
		},
	}, InstallPackOptions{
		CandidateFinalVerify: func(ctx context.Context, candidateRunner *Runner, candidates []packs.Pack) error {
			candidateVerifyCalls++
			if len(candidates) != 2 || candidates[0].Name() != baseName || candidates[1].Name() != dependentName {
				return fmt.Errorf("empty candidate verifier packs = %#v, want dependency order", candidates)
			}
			for _, name := range []string{baseName, dependentName} {
				var status string
				if err := candidateRunner.DB.DB().QueryRowContext(ctx, `SELECT Status FROM _System_InstalledPack WHERE Name = ?`, name).Scan(&status); err != nil {
					return err
				}
				if status != installedPackStatusInstalling {
					return fmt.Errorf("empty candidate pack %s status = %q, want pre-activation installing", name, status)
				}
			}
			return nil
		},
	})
	if err != nil {
		t.Fatalf("apply empty-database offline candidate: %v", err)
	}
	if !receipt.Verified || !receipt.Activated || receipt.PreviousStateDigest == "" || receipt.TargetStateDigest == "" || receipt.ActivatedStateDigest != receipt.TargetStateDigest {
		t.Fatalf("empty-database receipt = %#v, want exact verified activation", receipt)
	}
	if candidateVerifyCalls != 1 {
		t.Fatalf("empty candidate final verify calls = %d, want exactly one", candidateVerifyCalls)
	}

	verificationConn := openIntegrationConnection(t, cfg)
	t.Cleanup(func() { _ = verificationConn.Close() })
	for _, name := range []string{baseName, dependentName} {
		var status string
		var desired, applied sql.NullString
		if err := verificationConn.DB().QueryRowContext(ctx, `SELECT Status, DesiredPayloadHash, AppliedPayloadHash FROM _System_InstalledPack WHERE Name = ? AND COALESCE(IsDeleted, 0) = 0`, name).Scan(&status, &desired, &applied); err != nil {
			t.Fatalf("read installed candidate pack %s: %v", name, err)
		}
		if status != installedPackStatusActive || !desired.Valid || !applied.Valid || desired.String != applied.String || !packs.IsPackPayloadHashV1(applied.String) {
			t.Fatalf("installed candidate pack %s state = status=%q desired=%#v applied=%#v", name, status, desired, applied)
		}
	}
}

func TestSourceBackedPackGenerationTiDBExactRepairPreservesActivation(t *testing.T) {
	cfg, cleanup := newTemporaryIntegrationDatabase(t)
	t.Cleanup(cleanup)
	setDatabaseEnvFromConfig(t, cfg)

	ctx, cancel := context.WithTimeout(context.Background(), 9*time.Minute)
	t.Cleanup(cancel)
	conn := initializeControlPlaneMetadata(t, ctx, cfg)
	t.Cleanup(func() { _ = conn.Close() })
	applyAll := func(stage string) {
		t.Helper()
		applyCtx, applyCancel := context.WithTimeout(ctx, 3*time.Minute)
		defer applyCancel()
		if err := ApplyAll(applyCtx, conn); err != nil {
			t.Fatalf("%s: %v", stage, err)
		}
	}
	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("new exact-repair runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh exact-repair runner cache: %v", err)
	}

	const packName = "issue-tracker-standard"
	officialRoot := officialPackRootFromControlplaneTest(t)
	t.Setenv(packs.OfficialAppPacksRootEnvVar, officialRoot)
	sourceRevision := strings.TrimSpace(
		runGitTestCommandForControlplane(t, officialRoot, "rev-parse", "HEAD"),
	)
	sourceOptions := packs.DefaultPackSourceOptions()
	sourceOptions.OfficialRepoRoot = officialRoot
	sourceOptions.OfficialRepoSourceRevisions = map[string]string{
		packName: sourceRevision,
	}
	pack, err := packs.ResolveInstalledSourceBackedPackForInstalledRowAtSourceRevisionWithPackSourceOptions(
		packName,
		packs.AppPack,
		packs.SupportedManifestFormatVersion,
		string(packs.PackSourceKindOfficialRepo),
		packName,
		sourceRevision,
		sourceOptions,
	)
	if err != nil {
		t.Fatalf("resolve revision-bound neutral conformance pack: %v", err)
	}
	if err := InstallPackFromSourceWithOptions(
		ctx,
		runner,
		"official_repo:"+packName,
		InstallPackOptions{},
	); err != nil {
		t.Fatalf("install exact-repair generation: %v", err)
	}
	var initialInstallID string
	if err := conn.DB().QueryRowContext(ctx, `SELECT PackInstallId
FROM _System_InstalledPack
WHERE Name = ?`, pack.Name()).Scan(&initialInstallID); err != nil {
		t.Fatalf("read initial exact-repair install ID: %v", err)
	}
	assertActivePackGenerationBinding(t, ctx, conn.DB(), pack.Name())
	if _, err := conn.DB().ExecContext(ctx, `DELETE FROM _System_PackActivation
WHERE PackName = ?`, pack.Name()); err != nil {
		t.Fatalf("remove activation for stopped-upgrade adoption: %v", err)
	}
	deletePackGenerationConnectorBindingDefinitionSetForTest(
		t,
		ctx,
		conn.DB(),
		packGenerationID(pack.Name(), initialInstallID),
	)
	if _, err := conn.DB().ExecContext(ctx, `DELETE FROM _System_PackGeneration
WHERE PackName = ?`, pack.Name()); err != nil {
		t.Fatalf("remove generation for stopped-upgrade adoption: %v", err)
	}
	assertNoPackGenerationAuthority(t, ctx, conn.DB(), pack.Name())
	if err := bootstrapStateStore.MarkControlPlaneApplyAllCompleted(conn.DB()); err != nil {
		t.Fatalf("seed previous-release completed marker: %v", err)
	}
	applyAll("stopped-upgrade official ApplyAll adoption")
	conn = openIntegrationConnection(t, cfg)
	runner, err = NewRunner(conn)
	if err != nil {
		t.Fatalf("new post-upgrade exact-repair runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh post-upgrade exact-repair runner cache: %v", err)
	}
	assertActivePackGenerationBinding(t, ctx, conn.DB(), pack.Name())

	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("exact-repair payload hash: %v", err)
	}
	if _, err := conn.DB().ExecContext(ctx, `DELETE FROM _System_PackActivation
WHERE PackName = ?`, pack.Name()); err != nil {
		t.Fatalf("remove activation for no-authority failed retry: %v", err)
	}
	deletePackGenerationConnectorBindingDefinitionSetForTest(
		t,
		ctx,
		conn.DB(),
		packGenerationID(pack.Name(), initialInstallID),
	)
	if _, err := conn.DB().ExecContext(ctx, `DELETE FROM _System_PackGeneration
WHERE PackName = ?`, pack.Name()); err != nil {
		t.Fatalf("remove generation for no-authority failed retry: %v", err)
	}
	if err := markInstalledPackInstalling(
		ctx,
		conn.DB(),
		pack,
		payloadHash,
	); err != nil {
		t.Fatalf("seed no-authority installing attempt: %v", err)
	}
	if err := markInstalledPackFailed(
		ctx,
		conn.DB(),
		pack,
		payloadHash,
		errors.New("injected no-authority phase-7 interruption"),
	); err != nil {
		t.Fatalf("record no-authority failed attempt: %v", err)
	}
	if err := conn.DB().QueryRowContext(ctx, `SELECT PackInstallId
FROM _System_InstalledPack
WHERE Name = ?`, pack.Name()).Scan(&initialInstallID); err != nil {
		t.Fatalf("read no-authority retry install ID: %v", err)
	}
	applyAll("stopped ApplyAll recovers failed no-authority attempt without revision plan")
	conn = openIntegrationConnection(t, cfg)
	assertActivePackGenerationBinding(t, ctx, conn.DB(), pack.Name())

	if err := markInstalledPackInstallingWithPackInstallID(
		ctx,
		conn.DB(),
		pack,
		payloadHash,
		initialInstallID,
	); err != nil {
		t.Fatalf("seed interrupted exact repair: %v", err)
	}
	if err := markInstalledPackFailed(
		ctx,
		conn.DB(),
		pack,
		payloadHash,
		errors.New("injected exact-repair interruption"),
	); err != nil {
		t.Fatalf("record interrupted exact repair: %v", err)
	}
	applyAll("second stopped ApplyAll recovers failed exact generation without revision plan")
	conn = openIntegrationConnection(t, cfg)
	assertActivePackGenerationBinding(t, ctx, conn.DB(), pack.Name())

	if err := markInstalledPackInstallingWithPackInstallID(
		ctx,
		conn.DB(),
		pack,
		payloadHash,
		initialInstallID,
	); err != nil {
		t.Fatalf("seed indeterminate-commit installing row: %v", err)
	}
	applyAll("second stopped ApplyAll recovers installing exact generation without revision plan")
	conn = openIntegrationConnection(t, cfg)
	assertActivePackGenerationBinding(t, ctx, conn.DB(), pack.Name())

	staleHash := packs.PackPayloadHashV1Prefix + strings.Repeat("0", 64)
	if _, err := conn.DB().ExecContext(ctx, `UPDATE _System_InstalledPack
SET AppliedPayloadHash = ?, DesiredPayloadHash = ?
WHERE Name = ?`, staleHash, staleHash, pack.Name()); err != nil {
		t.Fatalf("seed exact-repair compatibility drift: %v", err)
	}
	runner, err = NewRunner(conn)
	if err != nil {
		t.Fatalf("new final exact-repair runner: %v", err)
	}
	if err := runner.RefreshCache(ctx); err != nil {
		t.Fatalf("refresh final exact-repair runner cache: %v", err)
	}
	if err := reconcileActiveInstalledPackSourceState(
		ctx,
		runner,
		InstallPackOptions{allowInstalledPackSourceStateRepair: true},
	); err != nil {
		t.Fatalf("reconcile exact active generation: %v", err)
	}

	var repairedInstallID string
	if err := conn.DB().QueryRowContext(ctx, `SELECT PackInstallId
FROM _System_InstalledPack
WHERE Name = ?`, pack.Name()).Scan(&repairedInstallID); err != nil {
		t.Fatalf("read repaired exact-repair install ID: %v", err)
	}
	if repairedInstallID != initialInstallID {
		t.Fatalf(
			"exact repair changed PackInstallId = %q, want %q",
			repairedInstallID,
			initialInstallID,
		)
	}
	assertOnePackGenerationAuthority(t, ctx, conn.DB(), pack.Name())
	assertActivePackGenerationBinding(t, ctx, conn.DB(), pack.Name())
}

func deletePackGenerationConnectorBindingDefinitionSetForTest(
	t *testing.T,
	ctx context.Context,
	db interface {
		ExecContext(context.Context, string, ...any) (sql.Result, error)
	},
	packGenerationID string,
) {
	t.Helper()
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_PackGenerationConnectorBindingDefinition
WHERE PackGenerationId = ?`, packGenerationID); err != nil {
		t.Fatalf("delete pack-generation connector binding definition rows: %v", err)
	}
	if _, err := db.ExecContext(ctx, `DELETE FROM _System_PackGenerationConnectorBindingDefinitionSet
WHERE PackGenerationId = ?`, packGenerationID); err != nil {
		t.Fatalf("delete pack-generation connector binding definition set: %v", err)
	}
}

func assertActivePackGenerationConnectorBindingDefinitionSet(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	packName string,
) {
	t.Helper()
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		t.Fatalf("begin %s connector binding definition-set assertion: %v", packName, err)
	}
	defer func() { _ = tx.Rollback() }()
	activation, found, err := readPackActivationForUpdate(ctx, tx, packName)
	if err != nil || !found {
		t.Fatalf(
			"read %s active generation for connector binding definition-set assertion: found=%v err=%v",
			packName,
			found,
			err,
		)
	}
	generation, found, err := readPackGenerationForUpdate(
		ctx,
		tx,
		activation.activePackInstallID,
	)
	if err != nil || !found {
		t.Fatalf(
			"read %s generation %s for connector binding definition-set assertion: found=%v err=%v",
			packName,
			activation.activePackInstallID,
			found,
			err,
		)
	}
	if _, err := readVerifiedPackGenerationConnectorBindingDefinitionSet(
		ctx,
		tx,
		generation,
	); err != nil {
		t.Fatalf(
			"verify %s generation %s connector binding definition set: %v",
			packName,
			generation.packInstallID,
			err,
		)
	}
}

func assertNoPackGenerationConnectorBindingDefinitionSet(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	packGenerationID string,
) {
	t.Helper()
	var headerCount, definitionCount int
	if err := db.QueryRowContext(ctx, `SELECT
  (SELECT COUNT(*)
     FROM _System_PackGenerationConnectorBindingDefinitionSet
    WHERE PackGenerationId = ?),
  (SELECT COUNT(*)
     FROM _System_PackGenerationConnectorBindingDefinition
    WHERE PackGenerationId = ?)`,
		packGenerationID,
		packGenerationID,
	).Scan(&headerCount, &definitionCount); err != nil {
		t.Fatalf(
			"count absent connector binding definition set for %s: %v",
			packGenerationID,
			err,
		)
	}
	if headerCount != 0 || definitionCount != 0 {
		t.Fatalf(
			"connector binding definition authority for %s = header/definitions %d/%d, want 0/0",
			packGenerationID,
			headerCount,
			definitionCount,
		)
	}
}

func assertApplyAllSourceRevisionRestartNoopWithGeneration(
	t *testing.T,
	ctx context.Context,
	conn *database.TiDBConnection,
	pack *packs.SourceBackedPack,
) {
	t.Helper()
	meta := packs.PackHashMetadataForPack(pack)
	if !meta.SourceRevisionValid {
		t.Fatal("restart no-op fixture has no source revision")
	}
	runApplyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		pack,
		"production-path-source-revision-0",
		"source revision restart no-op proof",
		func(revision string) *packs.ResolvedPackSource {
			return testControlplaneOfficialRepoResolvedSource(revision)
		},
	)
	assertOnePackGenerationAuthority(t, ctx, conn.DB(), pack.Name())
	assertSourcePackObjectWasNotApplied(t, ctx, conn.DB())
}

func runApplyAllSourceRevisionTestPlan(
	t *testing.T,
	ctx context.Context,
	conn *database.TiDBConnection,
	pack *packs.SourceBackedPack,
	fromRevision string,
	lockReason string,
	resolveSource func(string) *packs.ResolvedPackSource,
) {
	t.Helper()
	if err := applyAllSourceRevisionTestPlan(
		t,
		ctx,
		conn,
		pack,
		fromRevision,
		lockReason,
		resolveSource,
	); err != nil {
		t.Fatalf("ApplyAll source revision plan for %s: %v", pack.Name(), err)
	}
}

func applyAllSourceRevisionTestPlan(
	t *testing.T,
	ctx context.Context,
	conn *database.TiDBConnection,
	pack *packs.SourceBackedPack,
	fromRevision string,
	lockReason string,
	resolveSource func(string) *packs.ResolvedPackSource,
) error {
	t.Helper()
	meta := packs.PackHashMetadataForPack(pack)
	if !meta.SourceRevisionValid {
		t.Fatalf("%s has no source revision", pack.Name())
	}
	previousValidator := validateApplyAllSourceRevisionSource
	previousInstalledResolver := resolveApplyAllSourceRevisionInstalledTarget
	defer func() {
		validateApplyAllSourceRevisionSource = previousValidator
		resolveApplyAllSourceRevisionInstalledTarget = previousInstalledResolver
	}()
	validateApplyAllSourceRevisionSource = func(
		_ string,
		options packs.PackSourceOptions,
		_ packs.ComponentKindRegistrationSet,
	) packs.PackValidationReport {
		revision := options.OfficialRepoSourceRevisions[pack.Name()]
		return packs.PackValidationReport{
			Source: resolveSource(revision),
		}
	}
	resolveApplyAllSourceRevisionInstalledTarget = func(
		_ context.Context,
		_ *sql.DB,
		_ installedPackRow,
		sourceRevision string,
		_ packs.PackSourceOptions,
		_ []formula.FunctionBundle,
		_ bool,
		_ packs.ComponentKindRegistrationSet,
		options ...packs.SourceBackedPackOption,
	) (*packs.SourceBackedPack, error) {
		return packs.NewSourceBackedPack(
			resolveSource(sourceRevision),
			options...,
		)
	}

	runner, err := NewRunner(conn)
	if err != nil {
		t.Fatalf("construct restart no-op production runner: %v", err)
	}
	return withControlPlaneApplyLockScope(
		ctx,
		conn.DB(),
		lockReason,
		func(applyLock controlPlaneApplyLockScope) error {
			return applyAllSourceRevisionAdvances(
				ctx,
				runner,
				InstallPackOptions{
					ApplyAllSourceRevisionAdvances: []SourceRevisionAdvanceAuthorization{{
						Name:         pack.Name(),
						Kind:         string(packs.PackSourceKindOfficialRepo),
						URI:          pack.Name(),
						FromRevision: fromRevision,
						ToRevision:   meta.SourceRevision,
					}},
				},
				applyLock,
			)
		},
	)
}

func testControlplaneOfficialRepoEmptySourceBackedPackNamed(
	t *testing.T,
	name string,
	revision string,
) *packs.SourceBackedPack {
	t.Helper()
	pack, err := packs.NewSourceBackedPack(
		testControlplaneOfficialRepoEmptyResolvedSourceNamed(name, revision),
	)
	if err != nil {
		t.Fatalf("construct empty official repo source-backed pack: %v", err)
	}
	return pack
}

func testControlplaneOfficialRepoEmptyResolvedSourceNamed(
	name string,
	revision string,
) *packs.ResolvedPackSource {
	return &packs.ResolvedPackSource{
		Kind:     packs.PackSourceKindOfficialRepo,
		URI:      name,
		Revision: revision,
		Manifest: &packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  name,
			DisplayName:           "Empty Retry Sample",
			Category:              packs.AppPack,
			Version:               "0.1.0",
			PlatformCompatibility: packs.PlatformCompatibility{
				Min:          "0.1.0",
				MaxExclusive: "0.2.0",
			},
		},
	}
}

func assertSourcePackObjectWasNotApplied(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
) {
	t.Helper()
	var count int
	if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_Object
WHERE ApiName = 'Issue'`).Scan(&count); err != nil {
		t.Fatalf("count source-pack component metadata: %v", err)
	}
	if count != 0 {
		t.Fatalf("source-pack component metadata count = %d, want 0 before generation authority is repaired", count)
	}
}

func assertOnePackGenerationAuthority(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	packName string,
) {
	t.Helper()
	var generationCount, activationCount int
	if err := db.QueryRowContext(ctx, `SELECT
  (SELECT COUNT(*) FROM _System_PackGeneration
   WHERE PackName = ? AND COALESCE(IsDeleted, 0) = 0),
  (SELECT COUNT(*) FROM _System_PackActivation
   WHERE PackName = ? AND COALESCE(IsDeleted, 0) = 0)`,
		packName,
		packName,
	).Scan(&generationCount, &activationCount); err != nil {
		t.Fatalf("count %s generation authority: %v", packName, err)
	}
	if generationCount != 1 || activationCount != 1 {
		t.Fatalf(
			"%s generation authority counts = %d/%d, want 1/1",
			packName,
			generationCount,
			activationCount,
		)
	}
}

func readPublishedPackGeneration(
	ctx context.Context,
	db *sql.DB,
	packInstallID string,
) (sourceBackedPackGeneration, bool, error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return sourceBackedPackGeneration{}, false, err
	}
	defer func() { _ = tx.Rollback() }()
	return readPackGenerationForUpdate(ctx, tx, packInstallID)
}

func assertPackGenerationReplacement(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	packName string,
	retainedFirstInstallID string,
	retainedRevision string,
	activeRevision string,
	wantEpoch uint64,
) {
	t.Helper()
	var (
		installedID     string
		activeID        string
		previousID      string
		activeSource    string
		activationEpoch uint64
		generationCount int
	)
	if err := db.QueryRowContext(ctx, `SELECT
p.PackInstallId,
a.ActivePackInstallId,
COALESCE(a.PreviousPackInstallId, ''),
a.ActivationEpoch,
active_generation.SourceRevision,
(SELECT COUNT(*) FROM _System_PackGeneration counted
  WHERE counted.PackName = p.Name AND COALESCE(counted.IsDeleted, 0) = 0)
FROM _System_InstalledPack p
JOIN _System_PackActivation a
  ON a.PackName = p.Name AND COALESCE(a.IsDeleted, 0) = 0
JOIN _System_PackGeneration active_generation
  ON active_generation.PackInstallId = a.ActivePackInstallId
 AND COALESCE(active_generation.IsDeleted, 0) = 0
WHERE p.Name = ? AND COALESCE(p.IsDeleted, 0) = 0`,
		packName,
	).Scan(
		&installedID,
		&activeID,
		&previousID,
		&activationEpoch,
		&activeSource,
		&generationCount,
	); err != nil {
		t.Fatalf("read %s replacement generation authority: %v", packName, err)
	}
	if installedID == "" ||
		installedID != activeID ||
		previousID != "" ||
		activeSource != activeRevision ||
		activationEpoch != wantEpoch ||
		generationCount != int(wantEpoch) {
		t.Fatalf(
			"%s replacement authority ids=%q/%q previous=%q revision=%q epoch/count=%d/%d",
			packName,
			installedID,
			activeID,
			previousID,
			activeSource,
			activationEpoch,
			generationCount,
		)
	}
	assertActivePackGenerationConnectorBindingDefinitionSet(t, ctx, db, packName)
	if retainedRevision != "" {
		var retainedRevisionCount int
		if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_PackGeneration
WHERE PackName = ? AND SourceRevision = ? AND COALESCE(IsDeleted, 0) = 0`,
			packName,
			retainedRevision,
		).Scan(&retainedRevisionCount); err != nil {
			t.Fatalf("count retained generation revision: %v", err)
		}
		if retainedRevisionCount != 1 {
			t.Fatalf(
				"retained generation revision %q count = %d, want 1",
				retainedRevision,
				retainedRevisionCount,
			)
		}
	}
	if retainedFirstInstallID != "" {
		var retainedCount int
		if err := db.QueryRowContext(ctx, `SELECT COUNT(*)
FROM _System_PackGeneration
WHERE PackName = ? AND PackInstallId = ? AND COALESCE(IsDeleted, 0) = 0`,
			packName,
			retainedFirstInstallID,
		).Scan(&retainedCount); err != nil {
			t.Fatalf("count retained first generation: %v", err)
		}
		if retainedCount != 1 {
			t.Fatalf("retained first generation count = %d, want 1", retainedCount)
		}
	}
}

func assertNoPackGenerationAuthority(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	packName string,
) {
	t.Helper()
	var generationCount, activationCount int
	if err := db.QueryRowContext(ctx, `SELECT
  (SELECT COUNT(*) FROM _System_PackGeneration WHERE PackName = ?),
  (SELECT COUNT(*) FROM _System_PackActivation WHERE PackName = ?)`,
		packName,
		packName,
	).Scan(&generationCount, &activationCount); err != nil {
		t.Fatalf("count %s absent generation authority: %v", packName, err)
	}
	if generationCount != 0 || activationCount != 0 {
		t.Fatalf(
			"%s pre-retry generation authority counts = %d/%d, want 0/0",
			packName,
			generationCount,
			activationCount,
		)
	}
}

func assertActivePackGenerationBinding(
	t *testing.T,
	ctx context.Context,
	db *sql.DB,
	packName string,
) {
	t.Helper()
	var status, installedID, activatedID, generationID string
	if err := db.QueryRowContext(ctx, `SELECT
p.Status, p.PackInstallId, a.ActivePackInstallId, g.PackInstallId
FROM _System_InstalledPack p
JOIN _System_PackActivation a ON a.PackName = p.Name
JOIN _System_PackGeneration g
  ON g.PackName = p.Name
 AND g.PackInstallId = a.ActivePackInstallId
WHERE p.Name = ?
  AND COALESCE(p.IsDeleted, 0) = 0
  AND COALESCE(a.IsDeleted, 0) = 0
  AND COALESCE(g.IsDeleted, 0) = 0`,
		packName,
	).Scan(&status, &installedID, &activatedID, &generationID); err != nil {
		t.Fatalf("read %s active generation binding: %v", packName, err)
	}
	if status != installedPackStatusActive ||
		strings.TrimSpace(installedID) == "" ||
		installedID != activatedID ||
		activatedID != generationID {
		t.Fatalf(
			"%s active generation binding = status %q ids %q/%q/%q",
			packName,
			status,
			installedID,
			activatedID,
			generationID,
		)
	}
}

func seedActivatedInstalledPackCompatibility(
	t *testing.T,
	ctx context.Context,
	db interface {
		ExecContext(context.Context, string, ...any) (sql.Result, error)
	},
	applied appliedAppPack,
	packInstallID string,
) {
	t.Helper()
	generation, err := sourceBackedPackGenerationFromVerifiedPack(applied)
	if err != nil {
		t.Fatalf("build compatibility preimage: %v", err)
	}
	_, err = db.ExecContext(ctx, `INSERT INTO _System_InstalledPack (
Id, Name, PackInstallId, Category, Status, PackVersion,
ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash,
AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI,
SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations,
RuntimeRelatedListRules, RuntimeSearchScopeOrder,
RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles,
RuntimeStartupShallowDeclarations, InstallSource, IsDeleted
) VALUES (
?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON),
CAST(? AS JSON), CAST(? AS JSON), CAST(? AS JSON), CAST(? AS JSON),
CAST(? AS JSON), 'apply', 0
)`,
		installedPackID(applied.pack.Name()),
		applied.pack.Name(),
		packInstallID,
		applied.pack.Category(),
		installedPackStatusActive,
		nullableValue(generation.packVersion),
		nullableValue(generation.manifestFormatVersion),
		nullableValue(generation.platformCompatibility),
		nullableValue(generation.desiredPayloadHash),
		nullableValue(generation.appliedPayloadHash),
		nullableValue(generation.payloadHashAlgorithm),
		nullableValue(generation.sourceKind),
		nullableValue(generation.sourceURI),
		nullableValue(generation.sourceRevision),
		nullableValue(generation.hostRequirementsDigest),
		nullableValue(generation.runtimeObjectSpecDecl),
		nullableValue(generation.runtimeRelatedLists),
		nullableValue(generation.runtimeSearchScope),
		nullableValue(generation.runtimeSearchAllDefault),
		nullableValue(generation.runtimeFormulaBundles),
		nullableValue(generation.runtimeStartupShallow),
	)
	if err != nil {
		t.Fatalf("seed activated InstalledPack compatibility row: %v", err)
	}
}