package packs

import (
	"bytes"
	"context"
	"crypto/sha256"
	"database/sql"
	"errors"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/go-sql-driver/mysql"
	cp "github.com/pingcap-inc/nexuscrm-contract/pkg/controlplane"
)

func TestManagedLifecycleHostCommitsNoChangeWithEvidenceAndFreshVerification(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	lifecycle := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	provider := lifecycle.Lifecycle()
	planCalls, mutateCalls, projectCalls := 0, 0, 0
	provider.Plan = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		planCalls++
		return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), nil
	}
	provider.Mutate = func(_ context.Context, _ cp.ComponentManagedScope, _ cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedPlan, _ cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
		mutateCalls++
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
	}
	provider.Project = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		projectCalls++
		return cp.ComponentManagedProjection{Identity: append([]byte(nil), request.Item.ExpectedProjection...), Expected: []byte("same"), Live: []byte("same")}, nil
	}
	lifecycle = rebuildManagedLifecycleRegistration(t, lifecycle, provider)
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	request := testManagedLifecycleItemRequest()
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}

	expectManagedLifecycleNoChangeTransaction(mock, nil)

	result, err := host.ApplyItem(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if result.Disposition != ManagedLifecycleOperationNoChange || result.Attempts != 1 || !result.Verification.Matches {
		t.Fatalf("operation result = %#v", result)
	}
	if planCalls != 2 || mutateCalls != 1 || projectCalls != 1 {
		t.Fatalf("plan/mutate/project calls = %d/%d/%d", planCalls, mutateCalls, projectCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostLockedAttemptBindsProductionCatalogToSemanticTransaction(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	db.SetMaxOpenConns(1)

	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	source := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	provider := source.Lifecycle()
	planCalls := 0
	provider.Plan = func(ctx context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		planCalls++
		if planCalls == 2 {
			rows, err := store.ReadRows(ctx, cp.ComponentManagedRowQuery{
				Table: "ProcessDefinition", Columns: []string{"ID", "Label"}, Equal: cp.ComponentRow{"ID": "one"},
				OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderAscending}},
				LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
			})
			if err != nil {
				return cp.ComponentManagedPlan{}, err
			}
			if len(rows) != 1 || rows[0]["Label"] != "current" {
				t.Fatalf("locked plan rows = %#v", rows)
			}
		}
		return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), nil
	}
	provider.Mutate = func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
	}
	provider.Project = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		return cp.ComponentManagedProjection{Identity: request.Item.ExpectedProjection, Expected: []byte("same"), Live: []byte("same")}, nil
	}
	lifecycle := rebuildManagedLifecycleRegistration(t, source, provider)
	host, err := NewManagedLifecycleHost(db, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	request := testManagedLifecycleItemRequest()
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}

	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectQuery("SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE").
		WithArgs("ProcessDefinition").
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE", "IS_NULLABLE"}).
			AddRow("ID", "varchar", "NO").
			AddRow("Label", "varchar", "YES"))
	mock.ExpectQuery("SELECT INDEX_NAME, COLUMN_NAME, SEQ_IN_INDEX").
		WithArgs("ProcessDefinition").
		WillReturnRows(sqlmock.NewRows([]string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX"}).
			AddRow("PRIMARY", "ID", 1))
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID`, `Label` FROM `ProcessDefinition` WHERE `ID` = ? ORDER BY `ID` ASC LIMIT 2")).
		WithArgs("one").
		WillReturnRows(sqlmock.NewRows([]string{"ID", "Label"}).AddRow("one", "current"))
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	expectManagedLifecycleSemanticRestore(mock)
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if result.Disposition != ManagedLifecycleOperationNoChange || result.Attempts != 1 || !result.Verification.Matches {
		t.Fatalf("operation result = %#v", result)
	}
	if planCalls != 2 {
		t.Fatalf("plan calls = %d, want 2", planCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostNormalizesLegacyProviderDuringAdditiveSelection(t *testing.T) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	if base.ManagedProvider == nil {
		t.Fatal("compatibility fixture is missing its legacy provider")
	}
	lifecycle := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatalf("select additive lifecycle over populated legacy provider: %v", err)
	}
	if base.ManagedProvider == nil {
		t.Fatal("host construction mutated the caller's legacy registration")
	}
	if host.registration.ManagedProvider != nil {
		t.Fatal("host snapshot retained both legacy and lifecycle providers")
	}
}

func TestManagedLifecycleHostSurfacesPostCommitSessionCleanupFailure(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	db.SetMaxOpenConns(2)
	reserve, err := db.Conn(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	defer reserve.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil, nil, nil)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	cleanupFailure := errors.New("session reset failed")
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_constraint_check_in_place_pessimistic = ?")).WithArgs(1).WillReturnError(cleanupFailure)
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = ?")).WithArgs("pessimistic").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ")).WillReturnResult(sqlmock.NewResult(0, 0))
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || !errors.Is(err, cleanupFailure) || result.Disposition != ManagedLifecycleOperationNoChange || !result.Verification.Matches {
		t.Fatalf("post-commit cleanup result = %#v, error = %v", result, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostProvesLostCommitResponseWithoutReplay(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	lifecycle := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	provider := lifecycle.Lifecycle()
	mutateCalls := 0
	provider.Plan = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), nil
	}
	provider.Mutate = func(_ context.Context, _ cp.ComponentManagedScope, _ cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedPlan, _ cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
		mutateCalls++
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
	}
	provider.Project = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		return cp.ComponentManagedProjection{Identity: append([]byte(nil), request.Item.ExpectedProjection...), Expected: []byte("same"), Live: []byte("same")}, nil
	}
	lifecycle = rebuildManagedLifecycleRegistration(t, lifecycle, provider)
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	request := testManagedLifecycleItemRequest()
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	plan := testManagedLifecyclePlan(t, request.ExpectedProjection)
	projectionIdentityHash := sha256.Sum256(request.ExpectedProjection)
	expectManagedLifecycleNoChangeTransaction(mock, context.DeadlineExceeded)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ComponentKind`, `Scope`, `PlanFingerprint`, `ProjectionIdentityHash`, `MutationDisposition`, `SemanticMutationRows`, `EffectReceiptHash` FROM `_System_ManagedComponentOperationEvidence` WHERE `OperationId` = ? AND `Attempt` = ? LIMIT 1")).
		WithArgs(sqlmock.AnyArg(), 1).
		WillReturnRows(sqlmock.NewRows([]string{"ComponentKind", "Scope", "PlanFingerprint", "ProjectionIdentityHash", "MutationDisposition", "SemanticMutationRows", "EffectReceiptHash"}).
			AddRow([]byte(cp.ComponentKindApprovalProcess), []byte(cp.ComponentManagedScopeItem), plan.Fingerprint(), projectionIdentityHash[:], []byte(cp.ComponentManagedMutationNoChange), 0, nil))
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if result.Disposition != ManagedLifecycleOperationNoChange || mutateCalls != 1 || !result.Verification.Matches {
		t.Fatalf("proved result = %#v; mutate calls = %d", result, mutateCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostSurfacesCleanupFailureAfterProvingLostCommitResponse(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	db.SetMaxOpenConns(2)
	reserve, err := db.Conn(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	defer reserve.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil, nil, nil)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	plan := testManagedLifecyclePlan(t, request.ExpectedProjection)
	projectionIdentityHash := sha256.Sum256(request.ExpectedProjection)
	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit().WillReturnError(context.DeadlineExceeded)
	cleanupFailure := errors.New("session reset failed")
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_constraint_check_in_place_pessimistic = ?")).WithArgs(1).WillReturnError(cleanupFailure)
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = ?")).WithArgs("pessimistic").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ")).WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ComponentKind`, `Scope`, `PlanFingerprint`, `ProjectionIdentityHash`, `MutationDisposition`, `SemanticMutationRows`, `EffectReceiptHash` FROM `_System_ManagedComponentOperationEvidence` WHERE `OperationId` = ? AND `Attempt` = ? LIMIT 1")).
		WithArgs(sqlmock.AnyArg(), 1).
		WillReturnRows(sqlmock.NewRows([]string{"ComponentKind", "Scope", "PlanFingerprint", "ProjectionIdentityHash", "MutationDisposition", "SemanticMutationRows", "EffectReceiptHash"}).
			AddRow([]byte(cp.ComponentKindApprovalProcess), []byte(cp.ComponentManagedScopeItem), plan.Fingerprint(), projectionIdentityHash[:], []byte(cp.ComponentManagedMutationNoChange), 0, nil))
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || !errors.Is(err, cleanupFailure) || result.Disposition != ManagedLifecycleOperationNoChange || !result.Verification.Matches {
		t.Fatalf("proved commit cleanup result = %#v, error = %v", result, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostTreatsContradictoryLostCommitProofAsIndeterminate(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil, nil,
		func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
			return cp.ComponentManagedProjection{Identity: request.Item.ExpectedProjection, Expected: []byte("expected"), Live: []byte("contradictory")}, nil
		},
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	plan := testManagedLifecyclePlan(t, request.ExpectedProjection)
	projectionIdentityHash := sha256.Sum256(request.ExpectedProjection)
	expectManagedLifecycleNoChangeTransaction(mock, context.DeadlineExceeded)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ComponentKind`, `Scope`, `PlanFingerprint`, `ProjectionIdentityHash`, `MutationDisposition`, `SemanticMutationRows`, `EffectReceiptHash` FROM `_System_ManagedComponentOperationEvidence` WHERE `OperationId` = ? AND `Attempt` = ? LIMIT 1")).
		WithArgs(sqlmock.AnyArg(), 1).
		WillReturnRows(sqlmock.NewRows([]string{"ComponentKind", "Scope", "PlanFingerprint", "ProjectionIdentityHash", "MutationDisposition", "SemanticMutationRows", "EffectReceiptHash"}).
			AddRow([]byte(cp.ComponentKindApprovalProcess), []byte(cp.ComponentManagedScopeItem), plan.Fingerprint(), projectionIdentityHash[:], []byte(cp.ComponentManagedMutationNoChange), 0, nil))
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationIndeterminate || result.Verification.Matches {
		t.Fatalf("contradictory proof result = %#v, err = %v", result, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostClassifiesCommitTimeUniquenessAsRolledBackWithoutProofRead(t *testing.T) {
	tests := []struct {
		name   string
		number uint16
	}{
		{name: "duplicate key", number: 1062},
		{name: "write conflict", number: 8147},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			testManagedLifecycleHostClassifiesCommitTimeUniquenessAsRolledBackWithoutProofRead(t, test.number)
		})
	}
}

func testManagedLifecycleHostClassifiesCommitTimeUniquenessAsRolledBackWithoutProofRead(t *testing.T, number uint16) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil, nil, nil)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleNoChangeTransaction(mock, &mysql.MySQLError{Number: number, Message: "deferred uniqueness failure"})
	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationRolledBackFailure {
		t.Fatalf("commit-time uniqueness result = %#v, err = %v", result, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostAcceptsEffectOnlyStagedMutation(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	class := cp.ComponentManagedEffectClass("test_revision")
	provider := cp.ComponentManagedLifecycleProvider{}
	provider.Plan = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		return testManagedLifecycleEffectPlan(t, request.Item.ExpectedProjection, class), nil
	}
	provider.Mutate = func(_ context.Context, _ cp.ComponentManagedScope, _ cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedPlan, _ cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationDesiredProjectionStaged}, nil
	}
	provider.Project = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		return cp.ComponentManagedProjection{Identity: request.Item.ExpectedProjection, Expected: []byte("same"), Live: []byte("same")}, nil
	}
	lifecycle, err := cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: cp.ComponentKindApprovalProcess, ReadSet: []cp.ComponentReadSurface{{Table: "ProcessDefinition", Fact: "rows"}},
		Capabilities: []cp.ComponentManagedCapability{
			cp.ComponentManagedCapabilityTargetSerialization, cp.ComponentManagedCapabilityStructuredMutation,
			cp.ComponentManagedCapabilityEffects, cp.ComponentManagedCapabilityEvidence, cp.ComponentManagedCapabilityFreshProjection,
		},
		EffectClasses: []cp.ComponentManagedEffectClass{class}, TransactionMode: cp.ComponentManagedTransactionPessimisticReadCommitted,
		Budget: testManagedLifecycleBudget(), Lifecycle: provider,
	})
	if err != nil {
		t.Fatal(err)
	}
	adapter := ManagedLifecycleEffectAdapter{
		Class:      class,
		LockPolicy: ManagedLifecycleEffectManagedStateOnly,
		Apply: func(ctx context.Context, tx *sql.Tx, _ ManagedLifecycleEffectRequest) ([]byte, error) {
			_, err := tx.ExecContext(ctx, "UPDATE `EffectState` SET `Revision` = `Revision` + 1")
			return []byte("revision-1"), err
		},
		Prove: func(context.Context, *sql.DB, ManagedLifecycleEffectRequest, []byte) (bool, error) { return true, nil },
	}
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, []ManagedLifecycleEffectAdapter{adapter})
	if err != nil {
		t.Fatal(err)
	}
	request := testManagedLifecycleItemRequest()
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}

	expectManagedLifecycleCoordinationDDL(mock)
	mock.ExpectBegin()
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentTargetLock`").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	mock.ExpectQuery("SELECT 1 FROM `_System_ManagedComponentTargetLock`.*LIMIT 1").WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectExec(regexp.QuoteMeta("UPDATE `EffectState` SET `Revision` = `Revision` + 1")).WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	expectManagedLifecycleSemanticRestore(mock)
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if result.Disposition != ManagedLifecycleOperationChanged {
		t.Fatalf("effect-only result = %#v", result)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostRetriesTransientLockErrorOnlyAfterConfirmedRollback(t *testing.T) {
	tests := []struct {
		name   string
		number uint16
	}{
		{name: "lock wait timeout", number: 1205},
		{name: "deadlock", number: 1213},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			testManagedLifecycleHostRetriesTransientLockErrorOnlyAfterConfirmedRollback(t, test.number)
		})
	}
}

func testManagedLifecycleHostRetriesTransientLockErrorOnlyAfterConfirmedRollback(t *testing.T, number uint16) {
	t.Helper()
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	lifecycle := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	provider := lifecycle.Lifecycle()
	mutateCalls := 0
	provider.Plan = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), nil
	}
	provider.Mutate = func(_ context.Context, _ cp.ComponentManagedScope, _ cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedPlan, _ cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
		mutateCalls++
		if mutateCalls == 1 {
			return cp.ComponentManagedMutationResult{}, &mysql.MySQLError{Number: number, Message: "transient lock failure"}
		}
		return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
	}
	provider.Project = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		return cp.ComponentManagedProjection{Identity: request.Item.ExpectedProjection, Expected: []byte("same"), Live: []byte("same")}, nil
	}
	lifecycle = rebuildManagedLifecycleRegistration(t, lifecycle, provider)
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	request := testManagedLifecycleItemRequest()
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}

	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectRollback()
	expectManagedLifecycleSemanticRestore(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	expectManagedLifecycleSemanticRestore(mock)
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if result.Disposition != ManagedLifecycleOperationNoChange || result.Attempts != 2 || mutateCalls != 2 {
		t.Fatalf("retry result = %#v; mutate calls = %d", result, mutateCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostDoesNotReplayUnknownCommitWithoutEvidence(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mutateCalls, projectCalls := 0, 0
	host, request := newManagedLifecycleTestHost(t, db, nil,
		func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			mutateCalls++
			return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
		},
		func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
			projectCalls++
			return cp.ComponentManagedProjection{Identity: request.Item.ExpectedProjection, Expected: []byte("same"), Live: []byte("same")}, nil
		},
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleNoChangeTransaction(mock, context.DeadlineExceeded)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ComponentKind`, `Scope`, `PlanFingerprint`, `ProjectionIdentityHash`, `MutationDisposition`, `SemanticMutationRows`, `EffectReceiptHash` FROM `_System_ManagedComponentOperationEvidence` WHERE `OperationId` = ? AND `Attempt` = ? LIMIT 1")).
		WithArgs(sqlmock.AnyArg(), 1).
		WillReturnRows(sqlmock.NewRows([]string{"ComponentKind", "Scope", "PlanFingerprint", "ProjectionIdentityHash", "MutationDisposition", "SemanticMutationRows", "EffectReceiptHash"}))

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationIndeterminate || mutateCalls != 1 || projectCalls != 0 {
		t.Fatalf("unknown commit result = %#v, err = %v, mutate/project = %d/%d", result, err, mutateCalls, projectCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostDoesNotRetryWhenRollbackFails(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mutateCalls := 0
	host, request := newManagedLifecycleTestHost(t, db, nil,
		func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			mutateCalls++
			return cp.ComponentManagedMutationResult{}, &mysql.MySQLError{Number: 1213, Message: "deadlock"}
		}, nil,
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectRollback().WillReturnError(errors.New("rollback connection lost"))
	expectManagedLifecycleSemanticRestore(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationIndeterminate || result.Attempts != 1 || mutateCalls != 1 {
		t.Fatalf("rollback-failure result = %#v, err = %v, mutate calls = %d", result, err, mutateCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostStopsAfterRetryBudgetIsExhausted(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	mutateCalls := 0
	host, request := newManagedLifecycleTestHost(t, db, nil,
		func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			mutateCalls++
			return cp.ComponentManagedMutationResult{}, &mysql.MySQLError{Number: 1213, Message: "deadlock"}
		}, nil,
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleCoordinationDDL(mock)
	for attempt := 0; attempt < testManagedLifecycleBudget().MaxAttempts; attempt++ {
		expectManagedLifecycleTargetPreparation(mock)
		expectManagedLifecycleSemanticPrefix(mock)
		mock.ExpectRollback()
		expectManagedLifecycleSemanticRestore(mock)
	}

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationRolledBackFailure || result.Attempts != 3 || mutateCalls != 3 {
		t.Fatalf("retry-exhaustion result = %#v, err = %v, mutate calls = %d", result, err, mutateCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostRestartsPreparationWhenLockedTargetsDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	planCalls, mutateCalls := 0, 0
	host, request := newManagedLifecycleTestHost(t, db,
		func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
			planCalls++
			key := []byte("old")
			if planCalls > 1 {
				key = []byte("new")
			}
			return testManagedLifecyclePlanWithTarget(t, request.Item.ExpectedProjection, "test/item", key), nil
		},
		func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			mutateCalls++
			return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
		}, nil,
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparationFor(mock, "test/item", []byte("old"))
	expectManagedLifecycleSemanticPrefixFor(mock, "test/item", []byte("old"))
	mock.ExpectRollback()
	expectManagedLifecycleSemanticRestore(mock)
	expectManagedLifecycleTargetPreparationFor(mock, "test/item", []byte("new"))
	expectManagedLifecycleSemanticPrefixFor(mock, "test/item", []byte("new"))
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	expectManagedLifecycleSemanticRestore(mock)
	expectManagedLifecycleFreshVerification(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err != nil || result.Disposition != ManagedLifecycleOperationNoChange || result.Attempts != 2 || planCalls != 3 || mutateCalls != 1 {
		t.Fatalf("target-drift result = %#v, err = %v, plan/mutate = %d/%d", result, err, planCalls, mutateCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostRejectsEnvelopeDriftEvenWhenTargetsAlsoDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	planCalls, mutateCalls := 0, 0
	host, request := newManagedLifecycleTestHost(t, db,
		func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
			planCalls++
			if planCalls == 1 {
				return testManagedLifecyclePlanWithTarget(t, request.Item.ExpectedProjection, "test/item", []byte("old")), nil
			}
			return testManagedLifecyclePlanWithTarget(t, []byte("different-projection"), "test/item", []byte("new")), nil
		},
		func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			mutateCalls++
			return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
		}, nil,
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparationFor(mock, "test/item", []byte("old"))
	expectManagedLifecycleSemanticPrefixFor(mock, "test/item", []byte("old"))
	mock.ExpectRollback()
	expectManagedLifecycleSemanticRestore(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationRolledBackFailure || result.Attempts != 1 || planCalls != 2 || mutateCalls != 0 {
		t.Fatalf("envelope-drift result = %#v, err = %v, plan/mutate = %d/%d", result, err, planCalls, mutateCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostRollsBackIgnoredStructuredStoreFailureBeforeEvidence(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil,
		func(ctx context.Context, _ cp.ComponentManagedScope, _ cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedPlan, store cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			_, _ = store.UpdateRows(ctx, cp.ComponentManagedUpdate{
				Table: "ProcessDefinition", Match: cp.ComponentRow{"ID": "one"}, Values: cp.ComponentRow{"Label": "changed"}, MaximumAffectedRows: 1,
			})
			return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
		}, nil,
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectRollback()
	expectManagedLifecycleSemanticRestore(mock)

	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationRolledBackFailure || !strings.Contains(err.Error(), "ignored a structured-store failure") {
		t.Fatalf("ignored store failure result = %#v, err = %v", result, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostReportsCommittedVerificationFailure(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil, nil,
		func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
			return cp.ComponentManagedProjection{Identity: request.Item.ExpectedProjection, Expected: []byte("expected"), Live: []byte("different")}, nil
		},
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleNoChangeTransaction(mock, nil)
	result, err := host.ApplyItem(context.Background(), request)
	if err == nil || result.Disposition != ManagedLifecycleOperationVerificationFailure || result.Verification.Matches {
		t.Fatalf("verification-failure result = %#v, err = %v", result, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostNeverExposesMutationStoreToPlan(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	planCalls := 0
	host, request := newManagedLifecycleTestHost(t, db,
		func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
			planCalls++
			if _, mutable := store.(cp.ComponentManagedMutationStore); mutable {
				t.Fatal("Plan received a dynamically recoverable mutation store")
			}
			return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), nil
		}, nil, nil,
	)
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleNoChangeTransaction(mock, nil)
	result, err := host.ApplyItem(context.Background(), request)
	if err != nil || result.Disposition != ManagedLifecycleOperationNoChange || planCalls != 2 {
		t.Fatalf("read-only Plan result = %#v, err = %v, calls = %d", result, err, planCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func expectManagedLifecycleNoChangeTransaction(mock sqlmock.Sqlmock, commitErr error) {
	expectManagedLifecycleCoordinationDDL(mock)
	expectManagedLifecycleTargetPreparation(mock)
	expectManagedLifecycleSemanticPrefix(mock)
	mock.ExpectExec("INSERT INTO `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 1))
	commit := mock.ExpectCommit()
	if commitErr != nil {
		commit.WillReturnError(commitErr)
	}
	expectManagedLifecycleSemanticRestore(mock)
	if commitErr == nil {
		expectManagedLifecycleFreshVerification(mock)
	}
}

func expectManagedLifecycleFreshVerification(mock sqlmock.Sqlmock) {
	mock.ExpectBegin()
	mock.ExpectRollback()
}

func expectManagedLifecycleCoordinationDDL(mock sqlmock.Sqlmock) {
	mock.ExpectExec("CREATE TABLE IF NOT EXISTS `_System_ManagedComponentTargetLock`").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("CREATE TABLE IF NOT EXISTS `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 0))
	expectManagedLifecycleCoordinationValidation(mock)
}

func expectManagedLifecycleCoordinationValidation(mock sqlmock.Sqlmock) {
	mock.ExpectQuery("SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, DATETIME_PRECISION, IS_NULLABLE, COLUMN_DEFAULT").
		WithArgs(managedLifecycleTargetTable).
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE", "CHARACTER_MAXIMUM_LENGTH", "DATETIME_PRECISION", "IS_NULLABLE", "COLUMN_DEFAULT"}).
			AddRow("TargetDomain", "varbinary", cp.ComponentManagedMaximumTargetDomain, nil, "NO", nil).
			AddRow("PhysicalKey", "varbinary", cp.ComponentManagedMaximumPhysicalKey, nil, "NO", nil).
			AddRow("CreatedDate", "datetime", nil, 6, "NO", "CURRENT_TIMESTAMP(6)").
			AddRow("LastModifiedDate", "datetime", nil, 6, "NO", "CURRENT_TIMESTAMP(6)"))
	mock.ExpectQuery("SELECT COLUMN_NAME.*INFORMATION_SCHEMA.KEY_COLUMN_USAGE").
		WithArgs(managedLifecycleTargetTable).
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME"}).AddRow("TargetDomain").AddRow("PhysicalKey"))
	mock.ExpectQuery("SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, DATETIME_PRECISION, IS_NULLABLE, COLUMN_DEFAULT").
		WithArgs(managedLifecycleEvidenceTable).
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE", "CHARACTER_MAXIMUM_LENGTH", "DATETIME_PRECISION", "IS_NULLABLE", "COLUMN_DEFAULT"}).
			AddRow("OperationId", "varbinary", 64, nil, "NO", nil).
			AddRow("Attempt", "int", nil, nil, "NO", nil).
			AddRow("ComponentKind", "varchar", 255, nil, "NO", nil).
			AddRow("Scope", "varchar", 16, nil, "NO", nil).
			AddRow("PlanFingerprint", "binary", 32, nil, "NO", nil).
			AddRow("ProjectionIdentityHash", "binary", 32, nil, "NO", nil).
			AddRow("MutationDisposition", "varchar", 32, nil, "NO", nil).
			AddRow("SemanticMutationRows", "int", nil, nil, "NO", nil).
			AddRow("EffectReceiptHash", "binary", 32, nil, "YES", nil).
			AddRow("CreatedDate", "datetime", nil, 6, "NO", "CURRENT_TIMESTAMP(6)").
			AddRow("LastModifiedDate", "datetime", nil, 6, "NO", "CURRENT_TIMESTAMP(6)"))
	mock.ExpectQuery("SELECT COLUMN_NAME.*INFORMATION_SCHEMA.KEY_COLUMN_USAGE").
		WithArgs(managedLifecycleEvidenceTable).
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME"}).AddRow("OperationId").AddRow("Attempt"))
}

func expectManagedLifecycleTargetPreparation(mock sqlmock.Sqlmock) {
	expectManagedLifecycleTargetPreparationFor(mock, "test/item", []byte("one"))
}

func expectManagedLifecycleTargetPreparationFor(mock sqlmock.Sqlmock, domain string, key []byte) {
	mock.ExpectBegin()
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO `_System_ManagedComponentTargetLock` (`TargetDomain`, `PhysicalKey`) VALUES (?, ?) ON DUPLICATE KEY UPDATE `PhysicalKey` = VALUES(`PhysicalKey`)")).
		WithArgs([]byte(domain), key).WillReturnResult(sqlmock.NewResult(0, 1))
	mock.ExpectCommit()
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1 FROM `_System_ManagedComponentTargetLock` WHERE `TargetDomain` = ? AND `PhysicalKey` = ? LIMIT 1")).
		WithArgs([]byte(domain), key).WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
}

func expectManagedLifecycleSemanticPrefix(mock sqlmock.Sqlmock) {
	expectManagedLifecycleSemanticPrefixFor(mock, "test/item", []byte("one"))
}

func expectManagedLifecycleSemanticPrefixFor(mock sqlmock.Sqlmock, domain string, key []byte) {
	expectManagedLifecycleSemanticSessionBegin(mock)
	mock.ExpectQuery(regexp.QuoteMeta("SELECT 1 FROM `_System_ManagedComponentTargetLock` WHERE `TargetDomain` = ? AND `PhysicalKey` = ? FOR UPDATE")).
		WithArgs([]byte(domain), key).WillReturnRows(sqlmock.NewRows([]string{"1"}).AddRow(1))
}

func expectManagedLifecycleSemanticSessionBegin(mock sqlmock.Sqlmock) {
	mock.ExpectQuery(regexp.QuoteMeta("SELECT @@SESSION.tidb_txn_mode, @@SESSION.tidb_constraint_check_in_place_pessimistic, @@SESSION.transaction_isolation")).
		WillReturnRows(sqlmock.NewRows([]string{"tidb_txn_mode", "tidb_constraint_check_in_place_pessimistic", "transaction_isolation"}).AddRow("pessimistic", 1, "REPEATABLE-READ"))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = 'pessimistic'")).WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_constraint_check_in_place_pessimistic = ON")).WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION TRANSACTION ISOLATION LEVEL READ COMMITTED")).WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectBegin()
}

func expectManagedLifecycleSemanticRestore(mock sqlmock.Sqlmock) {
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_constraint_check_in_place_pessimistic = ?")).WithArgs(1).WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION tidb_txn_mode = ?")).WithArgs("pessimistic").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec(regexp.QuoteMeta("SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ")).WillReturnResult(sqlmock.NewResult(0, 0))
}

func TestManagedLifecycleHostPreflightAndVerifyDoNotPrepareTargets(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	lifecycle := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	provider := lifecycle.Lifecycle()
	planCalls, projectCalls := 0, 0
	provider.Plan = func(_ context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		planCalls++
		if scope != cp.ComponentManagedScopeItem || request.Item == nil {
			t.Fatalf("plan scope/request = %q / %#v", scope, request)
		}
		return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), nil
	}
	provider.Project = func(_ context.Context, scope cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		projectCalls++
		return cp.ComponentManagedProjection{
			Identity: []byte("projection/item"), Expected: append([]byte(nil), request.Item.ExpectedProjection...), Live: append([]byte(nil), request.Item.ExpectedProjection...),
		}, nil
	}
	lifecycle = rebuildManagedLifecycleRegistration(t, lifecycle, provider)
	host, err := NewManagedLifecycleHost(db, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	request := testManagedLifecycleItemRequest()
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	expectManagedLifecycleFreshVerification(mock)
	verification, err := host.VerifyItem(context.Background(), request)
	if err != nil {
		t.Fatal(err)
	}
	if !verification.Matches || !bytes.Equal(verification.Projection.Expected, verification.Projection.Live) {
		t.Fatalf("verification = %#v", verification)
	}
	if planCalls != 1 || projectCalls != 1 {
		t.Fatalf("plan/project calls = %d/%d", planCalls, projectCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("Plan or Project prepared target rows: %v", err)
	}
}

func TestManagedLifecycleHostRequiresEffectAdapterAtPreflightButNotVerification(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	class := cp.ComponentManagedEffectClass("revision")
	provider := cp.ComponentManagedLifecycleProvider{
		Plan: func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
			return testManagedLifecycleEffectPlan(t, []byte("projection-1"), class), nil
		},
		Mutate: func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationDesiredProjectionStaged}, nil
		},
		Project: func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
			return cp.ComponentManagedProjection{Identity: []byte("projection-1"), Expected: []byte("same"), Live: []byte("same")}, nil
		},
	}
	lifecycle, err := cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: cp.ComponentKindApprovalProcess, ReadSet: []cp.ComponentReadSurface{{Table: "ProcessDefinition", Fact: "rows"}},
		Capabilities: []cp.ComponentManagedCapability{
			cp.ComponentManagedCapabilityTargetSerialization, cp.ComponentManagedCapabilityStructuredMutation,
			cp.ComponentManagedCapabilityEffects, cp.ComponentManagedCapabilityEvidence, cp.ComponentManagedCapabilityFreshProjection,
		},
		EffectClasses: []cp.ComponentManagedEffectClass{class}, TransactionMode: cp.ComponentManagedTransactionPessimisticReadCommitted,
		Budget: testManagedLifecycleBudget(), Lifecycle: provider,
	})
	if err != nil {
		t.Fatal(err)
	}
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatalf("construct verification-capable host without effect adapter: %v", err)
	}
	expectManagedLifecycleFreshVerification(mock)
	if verification, err := host.VerifyItem(context.Background(), testManagedLifecycleItemRequest()); err != nil || !verification.Matches {
		t.Fatalf("VerifyItem() = %#v, %v", verification, err)
	}
	if err := host.PreflightItem(context.Background(), testManagedLifecycleItemRequest()); err == nil || !strings.Contains(err.Error(), "effect \"revision\" is not configured") {
		t.Fatalf("PreflightItem() error = %v, want missing effect adapter", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostFreshVerificationBindsProductionCatalogToReadTransaction(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	db.SetMaxOpenConns(1)
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	source := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	provider := source.Lifecycle()
	provider.Project = func(ctx context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		rows, err := store.ReadRows(ctx, cp.ComponentManagedRowQuery{
			Table: "ProcessDefinition", Columns: []string{"ID", "Label"}, Equal: cp.ComponentRow{"ID": "one"},
			OrderBy:           []cp.ComponentManagedOrder{{Column: "ID", Direction: cp.ComponentManagedOrderAscending}},
			LogicalKeyColumns: []string{"ID"}, MaximumRows: 1,
		})
		if err != nil {
			return cp.ComponentManagedProjection{}, err
		}
		if len(rows) != 1 || rows[0]["Label"] != "current" {
			t.Fatalf("fresh verification rows = %#v", rows)
		}
		return cp.ComponentManagedProjection{
			Identity: request.Item.ExpectedProjection, Expected: []byte("same"), Live: []byte("same"),
		}, nil
	}
	lifecycle := rebuildManagedLifecycleRegistration(t, source, provider)
	host, err := NewManagedLifecycleHost(db, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	mock.ExpectBegin()
	mock.ExpectQuery("SELECT COLUMN_NAME, DATA_TYPE, IS_NULLABLE").
		WithArgs("ProcessDefinition").
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE", "IS_NULLABLE"}).
			AddRow("ID", "varchar", "NO").
			AddRow("Label", "varchar", "YES"))
	mock.ExpectQuery("SELECT INDEX_NAME, COLUMN_NAME, SEQ_IN_INDEX").
		WithArgs("ProcessDefinition").
		WillReturnRows(sqlmock.NewRows([]string{"INDEX_NAME", "COLUMN_NAME", "SEQ_IN_INDEX"}).
			AddRow("PRIMARY", "ID", 1))
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `ID`, `Label` FROM `ProcessDefinition` WHERE `ID` = ? ORDER BY `ID` ASC LIMIT 2")).
		WithArgs("one").
		WillReturnRows(sqlmock.NewRows([]string{"ID", "Label"}).AddRow("one", "current"))
	mock.ExpectRollback()
	verification, err := host.VerifyItem(context.Background(), testManagedLifecycleItemRequest())
	if err != nil || !verification.Matches {
		t.Fatalf("VerifyItem() = %#v, %v", verification, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostCountsDesiredSetItemsInsteadOfTrustingProviderEstimate(t *testing.T) {
	db, _, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()

	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	source := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	budget := source.Budget()
	budget.MaxDesiredItems = 1
	planCalls, projectCalls := 0, 0
	provider := source.Lifecycle()
	provider.Plan = func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		planCalls++
		return testManagedLifecyclePlan(t, []byte("underreported")), nil
	}
	provider.Project = func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
		projectCalls++
		return cp.ComponentManagedProjection{Identity: []byte("set"), Expected: []byte("same"), Live: []byte("same")}, nil
	}
	lifecycle, err := cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: source.Kind(), ReadSet: source.ReadSet(), Capabilities: append(source.Capabilities(), cp.ComponentManagedCapabilityExactSet), EffectClasses: source.EffectClasses(),
		TransactionMode: source.TransactionMode(), Budget: budget, Lifecycle: provider,
	})
	if err != nil {
		t.Fatal(err)
	}
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	request := cp.ComponentManagedSetRequest{PackName: "test-pack", Kind: cp.ComponentKindApprovalProcess}
	err = host.PreflightSet(context.Background(), request, []string{"One", "Two"})
	if err == nil || !strings.Contains(err.Error(), "actual desired items 2 exceeds budget 1") {
		t.Fatalf("PreflightSet() error = %v", err)
	}
	if planCalls != 0 {
		t.Fatalf("provider Plan calls = %d, want 0", planCalls)
	}
	if _, err := host.VerifySet(context.Background(), request, []string{"One", "Two"}); err == nil || !strings.Contains(err.Error(), "actual desired items 2 exceeds budget 1") {
		t.Fatalf("VerifySet() error = %v", err)
	}
	if _, err := host.ReconcileSet(context.Background(), request, []string{"One", "Two"}); err == nil || !strings.Contains(err.Error(), "actual desired items 2 exceeds budget 1") {
		t.Fatalf("ReconcileSet() error = %v", err)
	}
	if projectCalls != 0 {
		t.Fatalf("provider Project calls = %d, want 0", projectCalls)
	}
}

func TestManagedLifecycleHostBoundsTargetPreparationByLockWaitBudget(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil, nil, nil)
	source := host.lifecycle
	budget := source.Budget()
	budget.MaxLockWait = 10 * time.Millisecond
	host.lifecycle, err = cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: source.Kind(), ReadSet: source.ReadSet(), Capabilities: source.Capabilities(), EffectClasses: source.EffectClasses(),
		TransactionMode: source.TransactionMode(), Budget: budget, Lifecycle: source.Lifecycle(),
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	host.ddlReady = true
	mock.ExpectBegin()
	mock.ExpectExec(regexp.QuoteMeta("INSERT INTO `_System_ManagedComponentTargetLock` (`TargetDomain`, `PhysicalKey`) VALUES (?, ?) ON DUPLICATE KEY UPDATE `PhysicalKey` = VALUES(`PhysicalKey`)")).
		WithArgs([]byte("test/item"), []byte("one")).WillDelayFor(100 * time.Millisecond).WillReturnResult(sqlmock.NewResult(0, 1))
	started := time.Now()
	result, err := host.ApplyItem(context.Background(), request)
	if err == nil {
		t.Fatal("target preparation exceeded the lock-wait budget")
	}
	if elapsed := time.Since(started); elapsed >= 50*time.Millisecond {
		t.Fatalf("target preparation returned after %s, want lock-wait cancellation", elapsed)
	}
	if result.Disposition != ManagedLifecycleOperationNoMutationFailure {
		t.Fatalf("target preparation result = %#v", result)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostRequiresExactSetCapabilityAtEverySetEntryPoint(t *testing.T) {
	tests := []struct {
		name string
		call func(*ManagedLifecycleHost, cp.ComponentManagedSetRequest) error
	}{
		{name: "preflight", call: func(host *ManagedLifecycleHost, request cp.ComponentManagedSetRequest) error {
			return host.PreflightSet(context.Background(), request, nil)
		}},
		{name: "reconcile", call: func(host *ManagedLifecycleHost, request cp.ComponentManagedSetRequest) error {
			_, err := host.ReconcileSet(context.Background(), request, nil)
			return err
		}},
		{name: "verify", call: func(host *ManagedLifecycleHost, request cp.ComponentManagedSetRequest) error {
			_, err := host.VerifySet(context.Background(), request, nil)
			return err
		}},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			db, mock, err := sqlmock.New()
			if err != nil {
				t.Fatal(err)
			}
			defer db.Close()
			host, _ := newManagedLifecycleTestHost(t, db, nil, nil, nil)
			err = test.call(host, cp.ComponentManagedSetRequest{PackName: "test-pack", Kind: cp.ComponentKindApprovalProcess})
			if err == nil || !strings.Contains(err.Error(), string(cp.ComponentManagedCapabilityExactSet)) {
				t.Fatalf("set operation error = %v, want missing exact_set capability", err)
			}
			if err := mock.ExpectationsWereMet(); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestManagedLifecycleHostElapsedBudgetCancelsCoordinationInitializationWait(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	host, request := newManagedLifecycleTestHost(t, db, nil, nil, nil)
	source := host.lifecycle
	budget := source.Budget()
	budget.MaxElapsed = 10 * time.Millisecond
	host.lifecycle, err = cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: source.Kind(), ReadSet: source.ReadSet(), Capabilities: source.Capabilities(), EffectClasses: source.EffectClasses(),
		TransactionMode: source.TransactionMode(), Budget: budget, Lifecycle: source.Lifecycle(),
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := host.PreflightItem(context.Background(), request); err != nil {
		t.Fatal(err)
	}
	<-host.ddlGate
	defer func() { host.ddlGate <- struct{}{} }()
	result, err := host.ApplyItem(context.Background(), request)
	if !errors.Is(err, context.DeadlineExceeded) || result.Disposition != ManagedLifecycleOperationNoMutationFailure {
		t.Fatalf("coordination wait result = %#v, error = %v", result, err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostRejectsIncompatibleEvidenceTableBeforeReady(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	host, _ := newManagedLifecycleTestHost(t, db, nil, nil, nil)
	mock.ExpectExec("CREATE TABLE IF NOT EXISTS `_System_ManagedComponentTargetLock`").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectExec("CREATE TABLE IF NOT EXISTS `_System_ManagedComponentOperationEvidence`").WillReturnResult(sqlmock.NewResult(0, 0))
	mock.ExpectQuery("SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, DATETIME_PRECISION, IS_NULLABLE, COLUMN_DEFAULT").
		WithArgs(managedLifecycleTargetTable).
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE", "CHARACTER_MAXIMUM_LENGTH", "DATETIME_PRECISION", "IS_NULLABLE", "COLUMN_DEFAULT"}).
			AddRow("TargetDomain", "varbinary", cp.ComponentManagedMaximumTargetDomain, nil, "NO", nil).
			AddRow("PhysicalKey", "varbinary", cp.ComponentManagedMaximumPhysicalKey, nil, "NO", nil).
			AddRow("CreatedDate", "datetime", nil, 6, "NO", "CURRENT_TIMESTAMP(6)").
			AddRow("LastModifiedDate", "datetime", nil, 6, "NO", "CURRENT_TIMESTAMP(6)"))
	mock.ExpectQuery("SELECT COLUMN_NAME.*INFORMATION_SCHEMA.KEY_COLUMN_USAGE").
		WithArgs(managedLifecycleTargetTable).
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME"}).AddRow("TargetDomain").AddRow("PhysicalKey"))
	mock.ExpectQuery("SELECT COLUMN_NAME, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, DATETIME_PRECISION, IS_NULLABLE, COLUMN_DEFAULT").
		WithArgs("_System_ManagedComponentOperationEvidence").
		WillReturnRows(sqlmock.NewRows([]string{"COLUMN_NAME", "DATA_TYPE", "CHARACTER_MAXIMUM_LENGTH", "DATETIME_PRECISION", "IS_NULLABLE", "COLUMN_DEFAULT"}).
			AddRow("OperationId", "varbinary", 32, nil, "NO", nil))
	if err := host.ensureCoordinationTables(context.Background()); err == nil || !strings.Contains(err.Error(), "evidence table column shape") {
		t.Fatalf("incompatible evidence table error = %v", err)
	}
	if host.ddlReady {
		t.Fatal("host marked incompatible evidence table ready")
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

type deadlineManagedLifecycleCatalog struct{ fallback time.Duration }

func (c deadlineManagedLifecycleCatalog) TableSchema(ctx context.Context, _ string) (managedLifecycleTableSchema, error) {
	select {
	case <-ctx.Done():
		return managedLifecycleTableSchema{}, ctx.Err()
	case <-time.After(c.fallback):
		return managedLifecycleTableSchema{}, errors.New("managed lifecycle host context was not propagated")
	}
}

func TestManagedLifecycleHostEnforcesElapsedBudgetWhenProviderReplacesContext(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	source := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	budget := source.Budget()
	budget.MaxElapsed = 10 * time.Millisecond
	provider := source.Lifecycle()
	provider.Plan = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, store cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		_, err := store.ReadRows(context.Background(), cp.ComponentManagedRowQuery{
			Table: "ProcessDefinition", Columns: []string{"ID"}, Equal: cp.ComponentRow{"ID": "one"}, MaximumRows: 1,
		})
		return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), err
	}
	lifecycle, err := cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: source.Kind(), ReadSet: source.ReadSet(), Capabilities: source.Capabilities(), EffectClasses: source.EffectClasses(),
		TransactionMode: source.TransactionMode(), Budget: budget, Lifecycle: provider,
	})
	if err != nil {
		t.Fatal(err)
	}
	host, err := newManagedLifecycleHost(db, deadlineManagedLifecycleCatalog{fallback: time.Second}, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	if err := host.PreflightItem(context.Background(), testManagedLifecycleItemRequest()); !errors.Is(err, context.Canceled) && !errors.Is(err, context.DeadlineExceeded) {
		t.Fatalf("PreflightItem() error = %v, want elapsed-budget cancellation", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func TestManagedLifecycleHostRejectsTargetBudgetBeforePreparation(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatal(err)
	}
	defer db.Close()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	source := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	budget := source.Budget()
	budget.MaxTargets = 1
	provider := source.Lifecycle()
	provider.Plan = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
		plan, err := cp.NewComponentManagedPlan(cp.ComponentManagedPlanSpec{
			Targets: []cp.ComponentManagedTarget{
				{TargetDomain: "test/item", PhysicalKeyBytes: []byte("one")},
				{TargetDomain: "test/item", PhysicalKeyBytes: []byte("two")},
			},
			RequiredCapabilities: source.Capabilities(), Estimate: cp.ComponentManagedUsage{DesiredItems: 1, EvidenceWrites: 1},
			ProjectionIdentity: request.Item.ExpectedProjection, OpaquePlan: []byte(`{"change":false}`),
		})
		return plan, err
	}
	lifecycle, err := cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: source.Kind(), ReadSet: source.ReadSet(), Capabilities: source.Capabilities(), EffectClasses: source.EffectClasses(),
		TransactionMode: source.TransactionMode(), Budget: budget, Lifecycle: provider,
	})
	if err != nil {
		t.Fatal(err)
	}
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	err = host.PreflightItem(context.Background(), testManagedLifecycleItemRequest())
	if err == nil || !strings.Contains(err.Error(), "targets 2 exceeds registration budget 1") {
		t.Fatalf("PreflightItem() error = %v, want target-budget rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatal(err)
	}
}

func testManagedLifecyclePlan(t *testing.T, identity []byte) cp.ComponentManagedPlan {
	return testManagedLifecyclePlanWithTarget(t, identity, "test/item", []byte("one"))
}

func testManagedLifecyclePlanWithTarget(t *testing.T, identity []byte, domain string, key []byte) cp.ComponentManagedPlan {
	t.Helper()
	plan, err := cp.NewComponentManagedPlan(cp.ComponentManagedPlanSpec{
		Targets: []cp.ComponentManagedTarget{{TargetDomain: domain, PhysicalKeyBytes: append([]byte(nil), key...)}},
		RequiredCapabilities: []cp.ComponentManagedCapability{
			cp.ComponentManagedCapabilityTargetSerialization,
			cp.ComponentManagedCapabilityStructuredMutation,
			cp.ComponentManagedCapabilityEvidence,
			cp.ComponentManagedCapabilityFreshProjection,
		},
		Estimate:           cp.ComponentManagedUsage{DesiredItems: 1, EvidenceWrites: 1},
		ProjectionIdentity: append([]byte(nil), identity...), OpaquePlan: []byte(`{"change":false}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	return plan
}

func newManagedLifecycleTestHost(
	t *testing.T,
	db *sql.DB,
	plan func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error),
	mutate func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error),
	project func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error),
) (*ManagedLifecycleHost, cp.ComponentManagedRequest) {
	t.Helper()
	base := testComponentKindRegistration(cp.ComponentKindApprovalProcess, "ProcessDefinition")
	base.ManagedProvider = nil
	lifecycle := testComponentManagedLifecycleRegistration(t, cp.ComponentKindApprovalProcess, "ProcessDefinition")
	provider := lifecycle.Lifecycle()
	if plan == nil {
		plan = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedPlan, error) {
			return testManagedLifecyclePlan(t, request.Item.ExpectedProjection), nil
		}
	}
	if mutate == nil {
		mutate = func(context.Context, cp.ComponentManagedScope, cp.ComponentManagedLifecycleRequest, cp.ComponentManagedPlan, cp.ComponentManagedMutationStore) (cp.ComponentManagedMutationResult, error) {
			return cp.ComponentManagedMutationResult{Disposition: cp.ComponentManagedMutationNoChange}, nil
		}
	}
	if project == nil {
		project = func(_ context.Context, _ cp.ComponentManagedScope, request cp.ComponentManagedLifecycleRequest, _ cp.ComponentManagedReadStore) (cp.ComponentManagedProjection, error) {
			return cp.ComponentManagedProjection{Identity: request.Item.ExpectedProjection, Expected: []byte("same"), Live: []byte("same")}, nil
		}
	}
	provider.Plan, provider.Mutate, provider.Project = plan, mutate, project
	lifecycle = rebuildManagedLifecycleRegistration(t, lifecycle, provider)
	host, err := newManagedLifecycleHost(db, staticManagedLifecycleCatalog{}, base, lifecycle, nil)
	if err != nil {
		t.Fatal(err)
	}
	return host, testManagedLifecycleItemRequest()
}

func testManagedLifecycleEffectPlan(t *testing.T, identity []byte, class cp.ComponentManagedEffectClass) cp.ComponentManagedPlan {
	t.Helper()
	plan, err := cp.NewComponentManagedPlan(cp.ComponentManagedPlanSpec{
		Targets: []cp.ComponentManagedTarget{{TargetDomain: "test/item", PhysicalKeyBytes: []byte("one")}},
		RequiredCapabilities: []cp.ComponentManagedCapability{
			cp.ComponentManagedCapabilityTargetSerialization, cp.ComponentManagedCapabilityStructuredMutation,
			cp.ComponentManagedCapabilityEffects, cp.ComponentManagedCapabilityEvidence, cp.ComponentManagedCapabilityFreshProjection,
		},
		Effects: []cp.ComponentManagedEffectRequirement{{
			Class: class, Triggers: []cp.ComponentManagedMutationDisposition{cp.ComponentManagedMutationDesiredProjectionStaged},
		}},
		Estimate:           cp.ComponentManagedUsage{DesiredItems: 1, EvidenceWrites: 1, EffectCalls: 1},
		ProjectionIdentity: append([]byte(nil), identity...), OpaquePlan: []byte(`{"change":true}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	return plan
}

func rebuildManagedLifecycleRegistration(t *testing.T, source cp.ComponentManagedLifecycleRegistration, provider cp.ComponentManagedLifecycleProvider) cp.ComponentManagedLifecycleRegistration {
	t.Helper()
	registration, err := cp.NewComponentManagedLifecycleRegistration(cp.ComponentManagedLifecycleRegistrationSpec{
		Kind: source.Kind(), ReadSet: source.ReadSet(), Capabilities: source.Capabilities(), EffectClasses: source.EffectClasses(),
		TransactionMode: source.TransactionMode(), Budget: source.Budget(), Lifecycle: provider,
	})
	if err != nil {
		t.Fatal(err)
	}
	return registration
}

func testManagedLifecycleItemRequest() cp.ComponentManagedRequest {
	payload := []byte(`{"name":"one"}`)
	return cp.ComponentManagedRequest{
		PackName: "test-pack",
		Component: cp.ManifestComponent{
			Kind: cp.ComponentKindApprovalProcess, BusinessKey: "One", Manageability: cp.ManageabilityLocked, Payload: append([]byte(nil), payload...),
		},
		Payload: append([]byte(nil), payload...), ExpectedProjection: []byte(`{"name":"one"}`),
	}
}