package controlplane

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/nexuscrm/db"
	"github.com/pingcap-inc/nexuscrm-connectors/features/connector/capabilities"
	handlerconfig "github.com/pingcap-inc/nexuscrm-connectors/features/connector/inboundhandlerconfig"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/pingcap-inc/nexuscrm-controlplane/specdrift"
)

// TestVerifyInstalledPackHashStateHashDriftReturnsTypedError verifies that
// verifyInstalledPackHashState returns a *HashDriftError (not a plain error)
// when the stored hash is a valid V1 hash but does not match the current pack
// content. This is the signal reconcileActiveInstalledPackSourceState uses to
// take the reapply branch instead of failing closed.
func TestVerifyInstalledPackHashStateHashDriftReturnsTypedError(t *testing.T) {
	pack := testControlplaneSourceBackedPack(t)
	currentHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("current hash: %v", err)
	}
	staleHash := packs.PackPayloadHashV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	if staleHash == currentHash {
		t.Fatal("test setup: staleHash must differ from currentHash")
	}

	row := installedPackRow{
		Name:               "jira-like-sample",
		Category:           string(packs.AppPack),
		Status:             installedPackStatusActive,
		AppliedPayloadHash: sql.NullString{Valid: true, String: staleHash},
		DesiredPayloadHash: sql.NullString{Valid: true, String: staleHash},
		SourceKind:         sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:          sql.NullString{Valid: true, String: "/tmp/source-pack"},
	}

	err = verifyInstalledPackHashState(row, pack)
	if err == nil {
		t.Fatal("verifyInstalledPackHashState = nil, want *HashDriftError")
	}
	var drift *HashDriftError
	if !errors.As(err, &drift) {
		t.Fatalf("verifyInstalledPackHashState error type = %T (%v), want *HashDriftError", err, err)
	}
	if drift.PackName != "jira-like-sample" {
		t.Errorf("HashDriftError.PackName = %q, want %q", drift.PackName, "jira-like-sample")
	}
	if drift.StoredHash != staleHash {
		t.Errorf("HashDriftError.StoredHash = %q, want %q", drift.StoredHash, staleHash)
	}
	if drift.CurrentHash != currentHash {
		t.Errorf("HashDriftError.CurrentHash = %q, want %q", drift.CurrentHash, currentHash)
	}
}

// TestVerifyInstalledPackHashStateSourceIdentityDriftNotHashDrift verifies
// that shape failures (missing hash, inconsistent applied/desired, invalid
// format) do NOT produce a *HashDriftError. Only the specific case of a valid
// V1 hash that no longer matches the pack content triggers the typed error.
func TestVerifyInstalledPackHashStateShapeFailuresAreNotHashDrift(t *testing.T) {
	pack := testControlplaneSourceBackedPack(t)

	cases := []struct {
		name string
		row  installedPackRow
	}{
		{
			name: "missing applied hash",
			row: installedPackRow{
				Name:               "jira-like-sample",
				AppliedPayloadHash: sql.NullString{Valid: false},
				DesiredPayloadHash: sql.NullString{Valid: true, String: packs.PackPayloadHashV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
			},
		},
		{
			name: "missing desired hash",
			row: installedPackRow{
				Name:               "jira-like-sample",
				AppliedPayloadHash: sql.NullString{Valid: true, String: packs.PackPayloadHashV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
				DesiredPayloadHash: sql.NullString{Valid: false},
			},
		},
		{
			name: "applied != desired (inconsistent state)",
			row: installedPackRow{
				Name:               "jira-like-sample",
				AppliedPayloadHash: sql.NullString{Valid: true, String: packs.PackPayloadHashV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
				DesiredPayloadHash: sql.NullString{Valid: true, String: packs.PackPayloadHashV1Prefix + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},
			},
		},
		{
			name: "unsupported hash format",
			row: installedPackRow{
				Name:               "jira-like-sample",
				AppliedPayloadHash: sql.NullString{Valid: true, String: "not-a-valid-hash-format"},
				DesiredPayloadHash: sql.NullString{Valid: true, String: "not-a-valid-hash-format"},
			},
		},
	}

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			err := verifyInstalledPackHashState(tc.row, pack)
			if err == nil {
				t.Fatal("verifyInstalledPackHashState = nil, want error")
			}
			var drift *HashDriftError
			if errors.As(err, &drift) {
				t.Fatalf("verifyInstalledPackHashState returned *HashDriftError for shape failure %q, want opaque error", tc.name)
			}
		})
	}
}

type testHostRequirementsPack struct {
	name         string
	requirements []packs.HostPrimitiveRequirement
}

func (p testHostRequirementsPack) Name() string             { return p.name }
func (p testHostRequirementsPack) Category() packs.Category { return packs.AppPack }
func (p testHostRequirementsPack) Apply(context.Context, packs.ApplyDeps) error {
	return nil
}
func (p testHostRequirementsPack) Verify(context.Context, packs.VerifyDeps) error {
	return nil
}
func (p testHostRequirementsPack) HostRequirements() []packs.HostPrimitiveRequirement {
	return append([]packs.HostPrimitiveRequirement(nil), p.requirements...)
}

func writeControlplaneSourcePackWithHostRequirement(t *testing.T, dialectName string) string {
	t.Helper()
	root := filepath.Join(t.TempDir(), "jira-like-sample")
	if err := os.MkdirAll(root, 0o755); err != nil {
		t.Fatalf("mkdir source pack: %v", err)
	}
	manifest := fmt.Sprintf(`{
  "manifestFormatVersion": "1",
  "name": "jira-like-sample",
  "displayName": "Jira-Like Sample",
  "category": "app-pack",
  "version": "0.1.0",
  "platformCompatibility": {"min": "0.2.0", "maxExclusive": "0.3.0"},
  "dependencies": [],
  "hostRequirements": [{"kind": "product-dialect", "name": %q}],
  "components": [{
    "kind": "object_spec",
    "businessKey": "issue",
    "manageability": "locked",
    "payload": [{
      "apiName": "Issue",
      "label": "Issue",
      "tableType": "standard_object",
      "category": "jira",
      "sharingModel": "Private",
      "enableHierarchySharing": false,
      "fields": [
        {"apiName": "Subject", "label": "Subject", "type": "string", "required": true, "isNameField": true}
      ]
    }]
  }]
}`, dialectName)
	if err := os.WriteFile(filepath.Join(root, "pack.json"), []byte(manifest), 0o644); err != nil {
		t.Fatalf("write source pack manifest: %v", err)
	}
	return root
}

func TestVerifyInstalledPackHostRequirementsState(t *testing.T) {
	emptyPack := testHostRequirementsPack{name: "empty-pack"}
	if err := verifyInstalledPackHostRequirementsState(installedPackRow{Name: emptyPack.name}, emptyPack); err != nil {
		t.Fatalf("legacy missing host digest for empty requirements = %v, want nil", err)
	}

	requiredPack := testHostRequirementsPack{
		name: "needs-dialect-pack",
		requirements: []packs.HostPrimitiveRequirement{{
			Kind: packs.HostPrimitiveKindProductDialect,
			Name: "test-product-dialect",
		}},
	}
	currentDigest, err := packs.PackHostRequirementsDigest(requiredPack)
	if err != nil {
		t.Fatalf("current host requirements digest: %v", err)
	}

	err = verifyInstalledPackHostRequirementsState(installedPackRow{Name: requiredPack.name}, requiredPack)
	var drift *HostRequirementsDigestDriftError
	if !errors.As(err, &drift) {
		t.Fatalf("missing digest error = %T (%v), want *HostRequirementsDigestDriftError", err, err)
	}
	if !drift.MissingStored || drift.CurrentDigest != currentDigest {
		t.Fatalf("missing digest drift = %#v, want MissingStored with current digest %q", drift, currentDigest)
	}

	err = verifyInstalledPackHostRequirementsState(installedPackRow{
		Name:                   requiredPack.name,
		HostRequirementsDigest: sql.NullString{Valid: true, String: "not-a-host-requirements-digest"},
	}, requiredPack)
	if err == nil {
		t.Fatal("invalid host digest format = nil, want fail-closed error")
	}
	drift = nil
	if errors.As(err, &drift) {
		t.Fatalf("invalid host digest format returned drift error %#v, want opaque shape error", drift)
	}

	staleDigest := packs.PackHostRequirementsDigestV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	if staleDigest == currentDigest {
		staleDigest = packs.PackHostRequirementsDigestV1Prefix + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
	}
	err = verifyInstalledPackHostRequirementsState(installedPackRow{
		Name:                   requiredPack.name,
		HostRequirementsDigest: sql.NullString{Valid: true, String: staleDigest},
	}, requiredPack)
	drift = nil
	if !errors.As(err, &drift) {
		t.Fatalf("stale digest error = %T (%v), want *HostRequirementsDigestDriftError", err, err)
	}
	if drift.StoredDigest != staleDigest || drift.CurrentDigest != currentDigest {
		t.Fatalf("stale digest drift = %#v, want stored %q current %q", drift, staleDigest, currentDigest)
	}

	if err := verifyInstalledPackHostRequirementsState(installedPackRow{
		Name:                   requiredPack.name,
		HostRequirementsDigest: sql.NullString{Valid: true, String: currentDigest},
	}, requiredPack); err != nil {
		t.Fatalf("current host digest = %v, want nil", err)
	}
}

// TestHashDriftErrorMessage verifies the error message format.
func TestHashDriftErrorMessage(t *testing.T) {
	e := &HashDriftError{
		PackName:    "my-pack",
		StoredHash:  "nexus-pack-sha256-v1:aaaa",
		CurrentHash: "nexus-pack-sha256-v1:bbbb",
	}
	msg := e.Error()
	if !strings.Contains(msg, "my-pack") {
		t.Errorf("HashDriftError.Error() = %q, want pack name", msg)
	}
	if !strings.Contains(msg, "source identity intact") {
		t.Errorf("HashDriftError.Error() = %q, want 'source identity intact'", msg)
	}
}

// TestInstalledPackHashDriftClassificationSeparatesRepairableContentDrift
// proves that content drift has a typed repair signal while malformed
// compatibility state remains a fail-closed shape error.
func TestInstalledPackHashDriftClassificationSeparatesRepairableContentDrift(t *testing.T) {
	pack := testControlplaneSourceBackedPack(t)
	currentHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("current hash: %v", err)
	}
	staleHash := packs.PackPayloadHashV1Prefix + "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
	if staleHash == currentHash {
		t.Fatal("test setup: staleHash must differ from currentHash")
	}

	row := installedPackRow{
		Name:               "jira-like-sample",
		AppliedPayloadHash: sql.NullString{Valid: true, String: staleHash},
		DesiredPayloadHash: sql.NullString{Valid: true, String: staleHash},
		SourceKind:         sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:          sql.NullString{Valid: true, String: "/tmp/source-pack"},
	}

	verifyErr := verifyInstalledPackHashState(row, pack)
	var drift *HashDriftError
	if !errors.As(verifyErr, &drift) {
		t.Fatalf("expected *HashDriftError from verifyInstalledPackHashState, got %T: %v", verifyErr, verifyErr)
	}

	missingHashRow := installedPackRow{
		Name:               "jira-like-sample",
		AppliedPayloadHash: sql.NullString{Valid: false},
		DesiredPayloadHash: sql.NullString{Valid: false},
	}
	nonDriftErr := verifyInstalledPackHashState(missingHashRow, pack)
	if nonDriftErr == nil {
		t.Fatal("verifyInstalledPackHashState missing hash = nil, want error")
	}
	var notDrift *HashDriftError
	if errors.As(nonDriftErr, &notDrift) {
		t.Fatalf("missing-hash error should not be *HashDriftError, got %T", notDrift)
	}
}

func TestDetectInstalledPackSourceStateDriftClassifiesCompletePersistedRuntimeProjectionDrift(t *testing.T) {
	pack := testControlplaneOfficialRepoSourceBackedPack(t, "source-revision-1")
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("payload hash: %v", err)
	}
	hostDigest, err := packs.PackHostRequirementsDigest(pack)
	if err != nil {
		t.Fatalf("host requirements digest: %v", err)
	}
	runtimeDecl, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		t.Fatalf("persisted runtime declaration: %v", err)
	}
	meta := packs.PackHashMetadataForPack(pack)
	row := installedPackRow{
		Name:                    pack.Name(),
		Category:                string(pack.Category()),
		Status:                  installedPackStatusActive,
		PackVersion:             sql.NullString{String: meta.PackVersion, Valid: true},
		ManifestFormatVersion:   sql.NullString{String: meta.ManifestFormatVersion, Valid: true},
		PlatformCompatibility:   sql.NullString{String: meta.PlatformCompatibility, Valid: meta.PlatformCompatibilityValid},
		DesiredPayloadHash:      sql.NullString{String: payloadHash, Valid: true},
		AppliedPayloadHash:      sql.NullString{String: payloadHash, Valid: true},
		PayloadHashAlgorithm:    sql.NullString{String: meta.PayloadHashAlgorithm, Valid: true},
		SourceKind:              sql.NullString{String: meta.SourceKind, Valid: meta.SourceKindValid},
		SourceURI:               sql.NullString{String: meta.SourceURI, Valid: meta.SourceURIValid},
		SourceRevision:          sql.NullString{String: meta.SourceRevision, Valid: meta.SourceRevisionValid},
		HostRequirementsDigest:  sql.NullString{String: hostDigest, Valid: true},
		RuntimeObjectSpecDecl:   sql.NullString{String: runtimeDecl.ObjectPayloadsJSON(), Valid: true},
		RuntimeRelatedLists:     sql.NullString{String: runtimeDecl.RelatedListRulesJSON(), Valid: true},
		RuntimeSearchScope:      sql.NullString{String: runtimeDecl.SearchScopeOrderJSON(), Valid: true},
		RuntimeSearchAllDefault: sql.NullString{String: runtimeDecl.SearchAllDefaultFieldsJSON(), Valid: true},
		RuntimeFormulaBundles:   sql.NullString{String: runtimeDecl.FormulaFunctionBundlesJSON(), Valid: true},
		RuntimeStartupShallow:   sql.NullString{String: `[{"kind":"permission","businessKey":"LegacyPermission"}]`, Valid: true},
		HasRuntimeFormula:       true,
		HasRuntimeStartupDecl:   true,
	}

	if _, ok, err := detectInstalledPackSourceStateDrift(row, pack); err != nil || ok {
		t.Fatalf("generic source drift classification = (ok=%t, err=%v), want runtime drift excluded", ok, err)
	}
	drift, ok, err := detectInstalledPackSourceStateDriftForContext(row, pack, runtimeProjectionPeerAdmission{
		pendingLocalPackName: "pending-local-peer",
		dependencyNames:      map[string]struct{}{strings.ToLower(row.Name): {}},
	})
	if err != nil {
		t.Fatalf("detect complete runtime projection peer drift: %v", err)
	}
	if !ok || drift.pack != pack || drift.kind != activeInstalledPackSourceStateDriftPersistedRuntimeDeclaration {
		t.Fatalf("runtime projection peer drift = (%#v, %t), want exact pack override", drift, ok)
	}
	staleDependency := row
	staleDependency.AppliedPayloadHash.String = packs.PackPayloadHashV1Prefix + strings.Repeat("a", 64)
	staleDependency.DesiredPayloadHash = staleDependency.AppliedPayloadHash
	_, ok, err = detectInstalledPackSourceStateDriftForContext(staleDependency, pack, runtimeProjectionPeerAdmission{
		pendingLocalPackName: "pending-local-peer",
		dependencyNames:      map[string]struct{}{strings.ToLower(row.Name): {}},
	})
	if err == nil || ok || !strings.Contains(err.Error(), "dependency peer") || !strings.Contains(err.Error(), "not source/hash/host exact") {
		t.Fatalf("payload-drifted dependency classification = (ok=%t, err=%v), want fail-closed exactness error", ok, err)
	}
	if _, ok, err := detectInstalledPackSourceStateDriftForContext(row, pack, runtimeProjectionPeerAdmission{
		pendingLocalPackName: row.Name,
		dependencyNames:      map[string]struct{}{strings.ToLower(row.Name): {}},
	}); err != nil || ok {
		t.Fatalf("requested-pack runtime drift classification = (ok=%t, err=%v), want peer-only exclusion", ok, err)
	}

	incomplete := row
	incomplete.RuntimeRelatedLists = sql.NullString{}
	_, ok, err = detectInstalledPackRuntimeProjectionPeerDrift(incomplete, pack)
	if err == nil || ok || !strings.Contains(err.Error(), "persisted runtime declaration is incomplete") {
		t.Fatalf("incomplete runtime declaration classification = (ok=%t, err=%v), want fail-closed incomplete error", ok, err)
	}
	absent := row
	absent.RuntimeObjectSpecDecl = sql.NullString{}
	absent.RuntimeRelatedLists = sql.NullString{}
	absent.RuntimeSearchScope = sql.NullString{}
	absent.RuntimeSearchAllDefault = sql.NullString{}
	absent.RuntimeFormulaBundles = sql.NullString{}
	absent.RuntimeStartupShallow = sql.NullString{}
	_, ok, err = detectInstalledPackRuntimeProjectionPeerDrift(absent, pack)
	if err == nil || ok || !strings.Contains(err.Error(), "persisted runtime declaration is incomplete") {
		t.Fatalf("absent runtime declaration classification = (ok=%t, err=%v), want fail-closed incomplete error", ok, err)
	}
	for _, malformed := range []string{
		`{}`, `null`, `[null]`, `[{}]`,
		`[{"kind":"connector_callout_profile"}]`,
		`[{"kind":"record_type","businessKey":"Account/Business","projection":"{"}]`,
	} {
		malformedRow := row
		malformedRow.RuntimeStartupShallow = sql.NullString{String: malformed, Valid: true}
		_, ok, err = detectInstalledPackRuntimeProjectionPeerDrift(malformedRow, pack)
		if err == nil || ok || !strings.Contains(err.Error(), "malformed") {
			t.Fatalf("malformed runtime declaration %s classification = (ok=%t, err=%v), want fail closed", malformed, ok, err)
		}
	}

	wrongSource := row
	wrongSource.SourceURI.String += "-other"
	_, ok, err = detectInstalledPackRuntimeProjectionPeerDrift(wrongSource, pack)
	if err == nil || ok || !strings.Contains(err.Error(), "source identity is not exact") {
		t.Fatalf("wrong-source runtime declaration classification = (ok=%t, err=%v), want fail closed", ok, err)
	}
	missingHostDigest := row
	missingHostDigest.HostRequirementsDigest = sql.NullString{}
	_, ok, err = detectInstalledPackRuntimeProjectionPeerDrift(missingHostDigest, pack)
	if err == nil || ok || !strings.Contains(err.Error(), "host requirements digest is not exact") {
		t.Fatalf("missing-host-digest runtime declaration classification = (ok=%t, err=%v), want fail closed", ok, err)
	}
	localPack := testControlplaneSourceBackedPack(t)
	localRow := row
	localMeta := packs.PackHashMetadataForPack(localPack)
	localRow.SourceKind = sql.NullString{String: localMeta.SourceKind, Valid: localMeta.SourceKindValid}
	localRow.SourceURI = sql.NullString{String: localMeta.SourceURI, Valid: localMeta.SourceURIValid}
	localRow.SourceRevision = sql.NullString{String: localMeta.SourceRevision, Valid: localMeta.SourceRevisionValid}
	_, ok, err = detectInstalledPackRuntimeProjectionPeerDrift(localRow, localPack)
	if err == nil || ok || !strings.Contains(err.Error(), "is not an official_repo source") {
		t.Fatalf("local runtime declaration peer classification = (ok=%t, err=%v), want fail closed", ok, err)
	}
}

func TestDetectInstalledPackRuntimeProjectionPeerDriftBridgesLegacyConnectorCalloutProjection(t *testing.T) {
	pack, err := packs.NewSourceBackedPack(&packs.ResolvedPackSource{
		Kind: packs.PackSourceKindOfficialRepo, URI: "legacy-connector", Revision: "source-revision-1",
		Manifest: &packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  "legacy-connector", DisplayName: "Legacy Connector", Category: packs.ConnectorPack, Version: "0.1.0",
			PlatformCompatibility: packs.PlatformCompatibility{Min: "0.1.0", MaxExclusive: "0.2.0"},
			Components: []packs.ManifestComponent{{
				Kind: packs.ComponentKindConnectorCalloutProfile, BusinessKey: "legacy-connector/default", Manageability: packs.ManageabilityLocked,
				Payload: []byte(`{"profileName":"default","credentialRef":"credential-a","endpointClass":"crm.read","description":"Legacy connector fixture","retryPolicy":{"maxAttempts":3,"baseDelay":1000000,"multiplier":2,"maxDelay":10000000,"retryableStatusCodes":[429,503]}}`),
			}},
		},
	})
	if err != nil {
		t.Fatalf("construct legacy connector pack: %v", err)
	}
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("payload hash: %v", err)
	}
	hostDigest, err := packs.PackHostRequirementsDigest(pack)
	if err != nil {
		t.Fatalf("host requirements digest: %v", err)
	}
	runtimeDecl, err := packs.BuildPersistedRuntimeDeclaration(pack)
	if err != nil {
		t.Fatalf("persisted runtime declaration: %v", err)
	}
	meta := packs.PackHashMetadataForPack(pack)
	row := installedPackRow{
		Name: pack.Name(), Category: string(pack.Category()), Status: installedPackStatusActive,
		PackVersion:           sql.NullString{String: meta.PackVersion, Valid: true},
		ManifestFormatVersion: sql.NullString{String: meta.ManifestFormatVersion, Valid: true},
		PlatformCompatibility: sql.NullString{String: meta.PlatformCompatibility, Valid: meta.PlatformCompatibilityValid},
		DesiredPayloadHash:    sql.NullString{String: payloadHash, Valid: true}, AppliedPayloadHash: sql.NullString{String: payloadHash, Valid: true},
		PayloadHashAlgorithm: sql.NullString{String: meta.PayloadHashAlgorithm, Valid: true},
		SourceKind:           sql.NullString{String: meta.SourceKind, Valid: true}, SourceURI: sql.NullString{String: meta.SourceURI, Valid: true},
		SourceRevision: sql.NullString{String: meta.SourceRevision, Valid: true}, HostRequirementsDigest: sql.NullString{String: hostDigest, Valid: true},
		RuntimeObjectSpecDecl:   sql.NullString{String: runtimeDecl.ObjectPayloadsJSON(), Valid: true},
		RuntimeRelatedLists:     sql.NullString{String: runtimeDecl.RelatedListRulesJSON(), Valid: true},
		RuntimeSearchScope:      sql.NullString{String: runtimeDecl.SearchScopeOrderJSON(), Valid: true},
		RuntimeSearchAllDefault: sql.NullString{String: runtimeDecl.SearchAllDefaultFieldsJSON(), Valid: true},
		RuntimeFormulaBundles:   sql.NullString{String: runtimeDecl.FormulaFunctionBundlesJSON(), Valid: true},
		RuntimeStartupShallow:   sql.NullString{String: `[{"kind":"connector_callout_profile","businessKey":"default"}]`, Valid: true},
		HasRuntimeFormula:       true, HasRuntimeStartupDecl: true,
	}

	drift, ok, err := detectInstalledPackRuntimeProjectionPeerDrift(row, pack)
	if err != nil {
		t.Fatalf("detect legacy connector runtime projection peer drift: %v", err)
	}
	if !ok || drift.pack != pack || drift.kind != activeInstalledPackSourceStateDriftPersistedRuntimeDeclaration {
		t.Fatalf("legacy connector runtime projection peer drift = (%#v, %t), want exact runtime projection drift", drift, ok)
	}
}

func TestInstalledPackIsPendingLocalSourceConvergence(t *testing.T) {
	pending := installedPackRow{
		Name:               "pending-local-pack",
		Status:             installedPackStatusActive,
		AppliedPayloadHash: sql.NullString{String: "applied", Valid: true},
		DesiredPayloadHash: sql.NullString{String: "desired", Valid: true},
		SourceKind:         sql.NullString{String: string(packs.PackSourceKindLocalPath), Valid: true},
		SourceURI:          sql.NullString{String: "/candidate/companion", Valid: true},
	}
	if !installedPackIsPendingLocalSourceConvergence(pending) {
		t.Fatal("exact pending local source convergence was not admitted")
	}
	admission := runtimeProjectionPeerAdmissionForRequestedRow(pending, []packs.PackDependency{{Name: "official-dependency"}})
	if !admission.permits("official-dependency") {
		t.Fatal("declared dependency peer was not admitted")
	}
	if admission.permits("unrelated-official") || admission.permits(pending.Name) {
		t.Fatal("unrelated or requested pack was admitted as a runtime projection dependency peer")
	}
	unrelatedOfficial := installedPackRow{
		Name:       "unrelated-official",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{String: string(packs.PackSourceKindOfficialRepo), Valid: true},
		SourceURI:  sql.NullString{String: "unrelated-official", Valid: true},
	}
	if !admission.inspectsRuntimeProjectionPeer(unrelatedOfficial) {
		t.Fatal("unrelated active official peer was excluded from install-time Runtime projection inspection")
	}
	if admission.inspectsRuntimeProjectionPeer(pending) {
		t.Fatal("requested pending local pack was admitted as its own Runtime projection peer")
	}
	for name, mutate := range map[string]func(*installedPackRow){
		"already applied": func(row *installedPackRow) { row.DesiredPayloadHash = row.AppliedPayloadHash },
		"official source": func(row *installedPackRow) { row.SourceKind.String = string(packs.PackSourceKindOfficialRepo) },
		"inactive":        func(row *installedPackRow) { row.Status = installedPackStatusFailed },
	} {
		t.Run(name, func(t *testing.T) {
			row := pending
			mutate(&row)
			if installedPackIsPendingLocalSourceConvergence(row) {
				t.Fatal("non-pending local source convergence was admitted")
			}
		})
	}
}

func TestReconcileActiveInstalledPackSourceStateRepairsPeerHashDriftBeforeOwnerVerify(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	ownerRow := installedPackRow{
		Name:       "aaa-owner",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/owner"},
	}
	peerRow := installedPackRow{
		Name:       "zzz-peer",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/peer"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origList := listInstalledAppPacksForReconcile
	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origBackfill := backfillChildCloneRulePackOwnershipForReconcile
	origExclusiveBackfill := backfillExclusiveSiblingFieldPackOwnershipForReconcile
	origVerify := verifyResolvedSourceBackedPackForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origPublishGeneration := publishFirstPackGenerationForReconcile
	origGenerationPreflight := preflightPackGenerationForReconcile
	defer func() {
		listInstalledAppPacksForReconcile = origList
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		backfillChildCloneRulePackOwnershipForReconcile = origBackfill
		backfillExclusiveSiblingFieldPackOwnershipForReconcile = origExclusiveBackfill
		verifyResolvedSourceBackedPackForReconcile = origVerify
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		publishFirstPackGenerationForReconcile = origPublishGeneration
		preflightPackGenerationForReconcile = origGenerationPreflight
	}()

	listCalls := 0
	peerRepaired := false
	backfilled := false
	exclusiveBackfilled := false
	verifyCalls := []string{}

	listInstalledAppPacksForReconcile = func(context.Context, *sql.DB) ([]installedPackRow, error) {
		listCalls++
		return []installedPackRow{ownerRow, peerRow}, nil
	}
	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(row installedPackRow, _ packs.Pack) error {
		if row.Name == peerRow.Name && !peerRepaired {
			return &HashDriftError{PackName: row.Name, StoredHash: "old", CurrentHash: "new"}
		}
		return nil
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	// No missing physical columns — this test is about repair ordering, not drift detection.
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, _ *installPackPreflightOverrides, _ InstallPackOptions) error {
		peerRepaired = true
		return nil
	}
	backfillChildCloneRulePackOwnershipForReconcile = func(context.Context, *sql.DB, packs.ComponentKindRegistrationSet) (int, error) {
		if !peerRepaired {
			t.Fatal("ChildCloneRules ownership backfill ran before peer hash drift repair")
		}
		backfilled = true
		return 0, nil
	}
	backfillExclusiveSiblingFieldPackOwnershipForReconcile = func(context.Context, *sql.DB, packs.ComponentKindRegistrationSet) (int, error) {
		if !backfilled {
			t.Fatal("ExclusiveSiblingFields ownership backfill ran before ChildCloneRules ownership backfill")
		}
		exclusiveBackfilled = true
		return 0, nil
	}
	verifyResolvedSourceBackedPackForReconcile = func(_ context.Context, row installedPackRow, _ *sql.DB, _ *packs.SourceBackedPack) error {
		verifyCalls = append(verifyCalls, row.Name)
		if row.Name == ownerRow.Name && !peerRepaired {
			return errors.New("owner verified before peer drift repair")
		}
		if !backfilled {
			return errors.New("owner verified before ChildCloneRules ownership backfill")
		}
		if !exclusiveBackfilled {
			return errors.New("owner verified before ExclusiveSiblingFields ownership backfill")
		}
		return nil
	}
	publishFirstPackGenerationForReconcile = func(context.Context, *sql.DB, appliedAppPack) error {
		return nil
	}
	preflightPackGenerationForReconcile = func(context.Context, *Runner, []installedPackRow, packs.ComponentKindRegistrationSet) error {
		return nil
	}

	if err := reconcileActiveInstalledPackSourceState(context.Background(), runner, InstallPackOptions{
		allowInstalledPackSourceStateRepair: true,
	}); err != nil {
		t.Fatalf("reconcileActiveInstalledPackSourceState = %v, want nil", err)
	}
	if listCalls != 2 {
		t.Fatalf("listInstalledAppPacksForReconcile calls = %d, want 2", listCalls)
	}
	if !peerRepaired {
		t.Fatal("peer hash drift was not repaired before verify pass")
	}
	if !backfilled {
		t.Fatal("ChildCloneRules ownership backfill did not run before verify pass")
	}
	if !exclusiveBackfilled {
		t.Fatal("ExclusiveSiblingFields ownership backfill did not run before verify pass")
	}
	if got := strings.Join(verifyCalls, ","); got != "aaa-owner,zzz-peer" {
		t.Fatalf("verify order = %q, want owner then peer in second pass", got)
	}
}

func TestReconcileActiveInstalledPackSourceStateFailsGenerationAuthorityBeforeMutation(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "jira-like-sample",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/jira-like-sample"},
	}
	authorityErr := errors.New("immutable generation differs from active source")

	origList := listInstalledAppPacksForReconcile
	origGenerationPreflight := preflightPackGenerationForReconcile
	defer func() {
		listInstalledAppPacksForReconcile = origList
		preflightPackGenerationForReconcile = origGenerationPreflight
	}()

	listInstalledAppPacksForReconcile = func(context.Context, *sql.DB) ([]installedPackRow, error) {
		return []installedPackRow{row}, nil
	}
	preflightCalls := 0
	preflightPackGenerationForReconcile = func(_ context.Context, _ *Runner, rows []installedPackRow, _ packs.ComponentKindRegistrationSet) error {
		preflightCalls++
		if len(rows) != 1 || rows[0].Name != row.Name {
			t.Fatalf("generation preflight rows = %#v, want only %s", rows, row.Name)
		}
		return authorityErr
	}

	err = reconcileActiveInstalledPackSourceState(
		context.Background(),
		runner,
		InstallPackOptions{allowInstalledPackSourceStateRepair: true},
	)
	if !errors.Is(err, authorityErr) {
		t.Fatalf("reconcile error = %v, want generation authority error", err)
	}
	if preflightCalls != 1 {
		t.Fatalf("generation preflight calls = %d, want 1", preflightCalls)
	}
	// sqlmock has no expectations beyond the preflight boundary. Any ownership
	// validation, source repair, physical-schema repair, or compatibility write
	// would therefore fail this test as an unexpected database operation.
	if err := rawDB.Ping(); err != nil {
		t.Fatalf("database remains usable after read-only preflight rejection: %v", err)
	}
}

func TestReconcileActiveInstalledPackSourceStateRejectsManagedApprovalGapBeforeRetryMutation(t *testing.T) {
	rawDB, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	activeRow := installedPackRow{
		Name:     "active-peer",
		Category: string(packs.AppPack),
		Status:   installedPackStatusActive,
	}
	retryRow := installedPackRow{
		Name:          "retry-peer",
		Category:      string(packs.AppPack),
		Status:        installedPackStatusFailed,
		PackInstallID: "retry-installation",
	}
	retryPack := testControlplaneSourceBackedPack(t)
	coverageErrText := "no OwnerPack-scoped managed approval_process registration covers ProcessDefinition"

	origList := listInstalledAppPacksForReconcile
	origGenerationPreflight := preflightPackGenerationForReconcile
	origRetryPreflight := preflightRetryableInstalledPackSourceStateForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	defer func() {
		listInstalledAppPacksForReconcile = origList
		preflightPackGenerationForReconcile = origGenerationPreflight
		preflightRetryableInstalledPackSourceStateForReconcile = origRetryPreflight
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
	}()

	listInstalledAppPacksForReconcile = func(context.Context, *sql.DB) ([]installedPackRow, error) {
		return []installedPackRow{activeRow, retryRow}, nil
	}
	preflightPackGenerationForReconcile = func(context.Context, *Runner, []installedPackRow, packs.ComponentKindRegistrationSet) error {
		return nil
	}
	preflightRetryableInstalledPackSourceStateForReconcile = func(context.Context, *Runner, []installedPackRow, packs.ComponentKindRegistrationSet) ([]retryableInstalledPackSourceStateAdmission, error) {
		return []retryableInstalledPackSourceStateAdmission{{
			row:  retryRow,
			pack: retryPack,
		}}, nil
	}
	applyResolvedSourceBackedPackUnlockedForReconcile = func(context.Context, *Runner, *packs.SourceBackedPack, *installPackPreflightOverrides, InstallPackOptions) error {
		t.Fatal("retry mutation ran before global managed-approval coverage rejection")
		return nil
	}
	mock.ExpectQuery(regexp.QuoteMeta(`
SELECT Id
FROM ProcessDefinition
WHERE COALESCE(OwnerPack, '') = ?
ORDER BY Id
LIMIT 1`)).
		WithArgs(activeRow.Name).
		WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("stale-process-definition"))

	err = reconcileActiveInstalledPackSourceState(
		context.Background(),
		&Runner{DB: database.NewConnectionFromDB(rawDB)},
		InstallPackOptions{allowInstalledPackSourceStateRepair: true},
	)
	if err == nil || !strings.Contains(err.Error(), coverageErrText) {
		t.Fatalf("reconcile error = %v, want managed-approval coverage rejection", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestVerifyRetryableInstalledPackSourceStateRequiresExactAttemptFacts(t *testing.T) {
	pack := testControlplaneSourceBackedPack(t)
	payloadHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("payload hash: %v", err)
	}
	meta := packs.PackHashMetadataForPack(pack)
	hostDigest, err := packs.PackHostRequirementsDigest(pack)
	if err != nil {
		t.Fatalf("host requirements digest: %v", err)
	}
	row := installedPackRow{
		ID:                      installedPackID(pack.Name()),
		Name:                    pack.Name(),
		PackInstallID:           "pack-install-retry",
		Category:                string(pack.Category()),
		Status:                  installedPackStatusFailed,
		PackVersion:             sql.NullString{String: meta.PackVersion, Valid: true},
		ManifestFormatVersion:   sql.NullString{String: meta.ManifestFormatVersion, Valid: true},
		PlatformCompatibility:   sql.NullString{String: meta.PlatformCompatibility, Valid: meta.PlatformCompatibilityValid},
		DesiredPayloadHash:      sql.NullString{String: payloadHash, Valid: true},
		PayloadHashAlgorithm:    sql.NullString{String: meta.PayloadHashAlgorithm, Valid: true},
		SourceKind:              sql.NullString{String: meta.SourceKind, Valid: meta.SourceKindValid},
		SourceURI:               sql.NullString{String: meta.SourceURI, Valid: meta.SourceURIValid},
		SourceRevision:          sql.NullString{String: meta.SourceRevision, Valid: meta.SourceRevisionValid},
		HostRequirementsDigest:  sql.NullString{String: hostDigest, Valid: true},
		RuntimeObjectSpecDecl:   sql.NullString{},
		RuntimeRelatedLists:     sql.NullString{},
		RuntimeSearchScope:      sql.NullString{},
		RuntimeSearchAllDefault: sql.NullString{},
		RuntimeFormulaBundles:   sql.NullString{},
		RuntimeStartupShallow:   sql.NullString{},
		HasRuntimeFormula:       false,
		HasRuntimeStartupDecl:   false,
	}
	if err := verifyRetryableInstalledPackSourceState(row, pack, payloadHash); err != nil {
		t.Fatalf("exact failed attempt rejected: %v", err)
	}

	drifted := row
	drifted.DesiredPayloadHash.String = packs.PackPayloadHashV1Prefix + strings.Repeat("f", 64)
	if err := verifyRetryableInstalledPackSourceState(drifted, pack, payloadHash); !errors.Is(err, ErrPackGenerationConflict) {
		t.Fatalf("drifted retry facts error = %v, want generation conflict", err)
	}

	missingIncarnation := row
	missingIncarnation.PackInstallID = ""
	if err := verifyRetryableInstalledPackSourceState(missingIncarnation, pack, payloadHash); !errors.Is(err, ErrPackGenerationPrecondition) {
		t.Fatalf("missing retry incarnation error = %v, want generation precondition", err)
	}
}

func TestRepairActiveInstalledPackSourceHashDriftPassesAllDriftedPeersToReapply(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	packA := installedPackRow{
		Name:       "aaa-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/a"},
	}
	packB := installedPackRow{
		Name:       "zzz-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/b"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, row installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(row installedPackRow, _ packs.Pack) error {
		return &HashDriftError{PackName: row.Name, StoredHash: "old", CurrentHash: "new"}
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	// No missing physical columns for this test.
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	applyCalls := []string{}
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, overrides *installPackPreflightOverrides, _ InstallPackOptions) error {
		if overrides == nil {
			t.Fatal("apply overrides = nil, want drift batch overrides")
		}
		if got := len(overrides.activeDriftedPacks); got != 2 {
			t.Fatalf("len(activeDriftedPacks) = %d, want 2", got)
		}
		if _, ok := overrides.activeDriftedPacks["aaa-pack"]; !ok {
			t.Fatal("activeDriftedPacks missing aaa-pack")
		}
		if _, ok := overrides.activeDriftedPacks["zzz-pack"]; !ok {
			t.Fatal("activeDriftedPacks missing zzz-pack")
		}
		applyCalls = append(applyCalls, "apply")
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{packA, packB}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if len(applyCalls) != 2 {
		t.Fatalf("apply calls = %d, want 2", len(applyCalls))
	}
}

func TestRepairActiveInstalledPackSourceHashDriftRepairsHostRequirementsDigestDrift(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:                   "crm-pack",
		Status:                 installedPackStatusActive,
		SourceKind:             sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:              sql.NullString{Valid: true, String: "/tmp/crm-pack"},
		HostRequirementsDigest: sql.NullString{Valid: true, String: packs.PackHostRequirementsDigestV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
	}
	pack := testControlplaneSourceBackedPackWithHostRequirements(t, []packs.HostPrimitiveRequirement{{
		Kind: packs.HostPrimitiveKindProductDialect,
		Name: "salesforce-v63",
	}})

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(installedPackRow, packs.Pack) error {
		return nil
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	applyCalls := 0
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, overrides *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalls++
		if overrides == nil {
			t.Fatal("apply overrides = nil, want drift batch overrides")
		}
		if _, ok := overrides.activeDriftedPacks["crm-pack"]; !ok {
			t.Fatal("activeDriftedPacks missing crm-pack")
		}
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if applyCalls != 1 {
		t.Fatalf("apply calls = %d, want 1", applyCalls)
	}
}

func TestRepairActiveInstalledPackSourceHashDriftFailsClosedOnUnsupportedHostRequirement(t *testing.T) {
	root := writeControlplaneSourcePackWithHostRequirement(t, "test-product-dialect")
	row := installedPackRow{
		Name:       "jira-like-sample",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: root},
	}

	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	defer func() { applyResolvedSourceBackedPackUnlockedForReconcile = origApply }()
	applyResolvedSourceBackedPackUnlockedForReconcile = func(context.Context, *Runner, *packs.SourceBackedPack, *installPackPreflightOverrides, InstallPackOptions) error {
		t.Fatal("repair reapply was called despite unsupported host requirement")
		return nil
	}

	err := repairActiveInstalledPackSourceHashDrift(context.Background(), &Runner{}, []installedPackRow{row}, InstallPackOptions{})
	if err == nil {
		t.Fatal("repairActiveInstalledPackSourceHashDrift = nil, want unsupported host requirement error")
	}
	if got := err.Error(); !strings.Contains(got, "host.unsupported") || !strings.Contains(got, "product-dialect/test-product-dialect") {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift error = %q, want unsupported product-dialect gate", got)
	}
}

func TestRepairActiveInstalledPackSourceHashDriftRepairsInconsistentActiveHashes(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:               "crm-pack",
		Status:             installedPackStatusActive,
		AppliedPayloadHash: sql.NullString{Valid: true, String: packs.PackPayloadHashV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"},
		DesiredPayloadHash: sql.NullString{Valid: true, String: packs.PackPayloadHashV1Prefix + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"},
		SourceKind:         sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:          sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origVerifyResolved := verifyResolvedSourceBackedPackForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origMarkConverged := markVerifiedInstalledPackConvergedForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		verifyResolvedSourceBackedPackForReconcile = origVerifyResolved
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		markVerifiedInstalledPackConvergedForReconcile = origMarkConverged
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(installedPackRow, packs.Pack) error {
		return errors.New("installed app pack crm-pack active hash state is inconsistent (applied=\"old\" desired=\"new\"); run setup apply-packs")
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	// No missing physical columns for this test.
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	verifyResolvedSourceBackedPackForReconcile = func(context.Context, installedPackRow, *sql.DB, *packs.SourceBackedPack) error {
		return errors.New("live projection drift")
	}
	markVerifiedInstalledPackConvergedForReconcile = func(context.Context, *sql.DB, packs.Pack, string) error {
		t.Fatal("bookkeeping convergence should not run when live verify fails")
		return nil
	}
	applyCalls := 0
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, overrides *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalls++
		if overrides == nil {
			t.Fatal("apply overrides = nil, want drift batch overrides")
		}
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if applyCalls != 1 {
		t.Fatalf("apply calls = %d, want 1", applyCalls)
	}
}

func TestRepairActiveInstalledPackSourceHashDriftConvergesVerifiedInconsistentHashesWithoutReapply(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	pack := testControlplaneSourceBackedPack(t)
	currentHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("PackPayloadHash: %v", err)
	}
	syntheticStaleHash, ok := deriveSyntheticStalePayloadHashForCurrent(currentHash)
	if !ok {
		t.Fatalf("derive synthetic stale hash from %q failed", currentHash)
	}
	row := installedPackRow{
		Name:               "crm-pack",
		Status:             installedPackStatusActive,
		AppliedPayloadHash: sql.NullString{Valid: true, String: syntheticStaleHash},
		DesiredPayloadHash: sql.NullString{Valid: true, String: currentHash},
		SourceKind:         sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:          sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origVerifyResolved := verifyResolvedSourceBackedPackForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origMarkConverged := markVerifiedInstalledPackConvergedForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		verifyResolvedSourceBackedPackForReconcile = origVerifyResolved
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		markVerifiedInstalledPackConvergedForReconcile = origMarkConverged
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(installedPackRow, packs.Pack) error {
		return errors.New("installed app pack crm-pack active hash state is inconsistent (applied=\"old\" desired=\"new\"); run setup apply-packs")
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	verifyResolvedSourceBackedPackForReconcile = func(_ context.Context, gotRow installedPackRow, _ *sql.DB, gotPack *packs.SourceBackedPack) error {
		if gotRow.Name != row.Name {
			t.Fatalf("verify row = %q, want %q", gotRow.Name, row.Name)
		}
		if gotPack != pack {
			t.Fatal("verify pack did not receive rebuilt pack")
		}
		return nil
	}
	convergeCalls := 0
	markVerifiedInstalledPackConvergedForReconcile = func(_ context.Context, _ *sql.DB, gotPack packs.Pack, gotHash string) error {
		convergeCalls++
		if gotPack.Name() != pack.Name() {
			t.Fatalf("converged pack = %q, want %q", gotPack.Name(), pack.Name())
		}
		if gotHash != currentHash {
			t.Fatalf("converged hash = %q, want %q", gotHash, currentHash)
		}
		return nil
	}
	applyResolvedSourceBackedPackUnlockedForReconcile = func(context.Context, *Runner, *packs.SourceBackedPack, *installPackPreflightOverrides, InstallPackOptions) error {
		t.Fatal("reapply should not run when live verify already matches current source")
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if convergeCalls != 1 {
		t.Fatalf("converge calls = %d, want 1", convergeCalls)
	}
}

func TestRepairActiveInstalledPackSourceHashDriftReappliesPreviousPayloadHash(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	pack := testControlplaneSourceBackedPack(t)
	currentHash, err := packs.PackPayloadHash(pack)
	if err != nil {
		t.Fatalf("PackPayloadHash: %v", err)
	}
	syntheticStaleHash, ok := deriveSyntheticStalePayloadHashForCurrent(currentHash)
	if !ok {
		t.Fatalf("derive synthetic stale hash from %q failed", currentHash)
	}
	previousHash := packs.PackPayloadHashV1Prefix + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	if previousHash == currentHash || previousHash == syntheticStaleHash {
		previousHash = packs.PackPayloadHashV1Prefix + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
	}
	if previousHash == currentHash || previousHash == syntheticStaleHash {
		t.Fatal("test setup: previous hash collides with current or synthetic stale hash")
	}
	row := installedPackRow{
		Name:               "crm-pack",
		Status:             installedPackStatusActive,
		AppliedPayloadHash: sql.NullString{Valid: true, String: previousHash},
		DesiredPayloadHash: sql.NullString{Valid: true, String: currentHash},
		SourceKind:         sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:          sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origVerifyResolved := verifyResolvedSourceBackedPackForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origMarkConverged := markVerifiedInstalledPackConvergedForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		verifyResolvedSourceBackedPackForReconcile = origVerifyResolved
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		markVerifiedInstalledPackConvergedForReconcile = origMarkConverged
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(installedPackRow, packs.Pack) error {
		return errors.New("installed app pack crm-pack active hash state is inconsistent (applied=\"old\" desired=\"new\"); run setup apply-packs")
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	verifyResolvedSourceBackedPackForReconcile = func(context.Context, installedPackRow, *sql.DB, *packs.SourceBackedPack) error {
		t.Fatal("live verify should not bypass reapply for a previous real payload hash")
		return nil
	}
	markVerifiedInstalledPackConvergedForReconcile = func(context.Context, *sql.DB, packs.Pack, string) error {
		t.Fatal("bookkeeping convergence should not run for a previous real payload hash")
		return nil
	}
	applyCalls := 0
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, got *packs.SourceBackedPack, overrides *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalls++
		if got != pack {
			t.Fatal("apply did not receive rebuilt pack")
		}
		if overrides == nil {
			t.Fatal("apply overrides = nil, want drift batch overrides")
		}
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if applyCalls != 1 {
		t.Fatalf("apply calls = %d, want 1", applyCalls)
	}
}

func TestReconcileActiveInstalledPackSourceStateRepairsMissingConnectorHandlerConfigRows(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "test-connector",
		Category:   string(packs.ConnectorPack),
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/test-connector"},
	}
	pack := testControlplaneConnectorSourceBackedPack(t)

	origList := listInstalledAppPacksForReconcile
	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origVerifyResolved := verifyResolvedSourceBackedPackForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origBackfill := backfillChildCloneRulePackOwnershipForReconcile
	origExclusiveBackfill := backfillExclusiveSiblingFieldPackOwnershipForReconcile
	origGenerationPreflight := preflightPackGenerationForReconcile
	defer func() {
		listInstalledAppPacksForReconcile = origList
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		verifyResolvedSourceBackedPackForReconcile = origVerifyResolved
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		backfillChildCloneRulePackOwnershipForReconcile = origBackfill
		backfillExclusiveSiblingFieldPackOwnershipForReconcile = origExclusiveBackfill
		preflightPackGenerationForReconcile = origGenerationPreflight
	}()

	listInstalledAppPacksForReconcile = func(context.Context, *sql.DB) ([]installedPackRow, error) {
		return []installedPackRow{row}, nil
	}
	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(installedPackRow, packs.Pack) error {
		return nil
	}
	verifyResolvedSourceBackedPackForReconcile = func(context.Context, installedPackRow, *sql.DB, *packs.SourceBackedPack) error {
		return fmt.Errorf("installed pack test-connector live connector inbound handlers: %w", &handlerconfig.MissingHandlerConfigError{
			Namespace:         "test-connector",
			CapabilityKind:    connectorcapabilities.SignatureVerifier,
			InstalledPackName: "test-connector",
			PackInstallID:     "pack-install-1",
		})
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		t.Fatal("missing handler-config repair should go straight to source reapply, not physical-column repair")
		return nil, nil
	}
	backfillChildCloneRulePackOwnershipForReconcile = func(context.Context, *sql.DB, packs.ComponentKindRegistrationSet) (int, error) {
		return 0, nil
	}
	backfillExclusiveSiblingFieldPackOwnershipForReconcile = func(context.Context, *sql.DB, packs.ComponentKindRegistrationSet) (int, error) {
		return 0, nil
	}
	preflightPackGenerationForReconcile = func(context.Context, *Runner, []installedPackRow, packs.ComponentKindRegistrationSet) error {
		return nil
	}
	applyCalls := 0
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, got *packs.SourceBackedPack, overrides *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalls++
		if got.Name() != "test-connector" {
			t.Fatalf("apply pack = %s, want test-connector", got.Name())
		}
		if overrides != nil {
			t.Fatalf("apply overrides = %#v, want nil for single-pack handler-config row repair", overrides)
		}
		return nil
	}

	if err := reconcileActiveInstalledPackSourceState(context.Background(), runner, InstallPackOptions{
		allowInstalledPackSourceStateRepair: true,
	}); err != nil {
		t.Fatalf("reconcileActiveInstalledPackSourceState = %v, want nil", err)
	}
	if applyCalls != 1 {
		t.Fatalf("apply calls = %d, want 1", applyCalls)
	}
}

func TestIsRepairableInstalledPackApplyAllErrorRejectsPlainConnectorHandlerConfigText(t *testing.T) {
	err := errors.New("connector inbound handler config missing for connector capability test-connector:signature_verifier")
	if isRepairableInstalledPackApplyAllError(err) {
		t.Fatal("plain missing handler-config text classified as repairable; want typed MissingHandlerConfigError only")
	}
}

func TestIsRepairableInstalledPackApplyAllErrorAcceptsUniqueOnlyObjectSpecDrift(t *testing.T) {
	err := &packs.ComponentVerifyError{
		Kind:        packs.ComponentKindObjectSpec,
		BusinessKey: "issue",
		Err: &specdrift.StructuralDriftError{
			SpecName: "jira-like-sample",
			Count:    1,
			FieldDrifts: []*specdrift.FieldDriftError{{
				SpecName:      "jira-like-sample",
				ObjectAPIName: "Issue",
				FieldAPIName:  "ExternalRemoteId__c",
				Diffs: []specdrift.FieldDiff{{
					Kind:    specdrift.FieldDiffKindIsUnique,
					Message: "IsUnique: spec=true live=false",
				}},
			}},
		},
	}
	if !isRepairableInstalledPackApplyAllError(err) {
		t.Fatalf("unique-only object_spec drift classified as non-repairable: %v", err)
	}
}

func TestIsRepairableInstalledPackApplyAllErrorAcceptsDeleteConstraintOnlyObjectSpecDrift(t *testing.T) {
	err := &packs.ComponentVerifyError{
		Kind:        packs.ComponentKindObjectSpec,
		BusinessKey: "lead",
		Err: &specdrift.StructuralDriftError{
			SpecName: "crm-pack",
			Count:    1,
			FieldDrifts: []*specdrift.FieldDriftError{{
				SpecName:      "crm-pack",
				ObjectAPIName: "Lead",
				FieldAPIName:  "First_Campaign__c",
				Diffs: []specdrift.FieldDiff{{
					Kind:    specdrift.FieldDiffKindDeleteConstraint,
					Message: "DeleteConstraint: spec=SetNull live=NULL",
				}},
			}},
		},
	}
	if !isRepairableInstalledPackApplyAllError(err) {
		t.Fatalf("deleteConstraint-only object_spec drift classified as non-repairable: %v", err)
	}
}

func TestIsRepairableInstalledPackApplyAllErrorAcceptsUniqueAndDeleteConstraintObjectSpecDrift(t *testing.T) {
	err := &packs.ComponentVerifyError{
		Kind:        packs.ComponentKindObjectSpec,
		BusinessKey: "lead",
		Err: &specdrift.StructuralDriftError{
			SpecName: "crm-pack",
			Count:    2,
			FieldDrifts: []*specdrift.FieldDriftError{
				{
					SpecName:      "crm-pack",
					ObjectAPIName: "Lead",
					FieldAPIName:  "ExternalId__c",
					Diffs:         []specdrift.FieldDiff{{Kind: specdrift.FieldDiffKindIsUnique}},
				},
				{
					SpecName:      "crm-pack",
					ObjectAPIName: "Lead",
					FieldAPIName:  "First_Campaign__c",
					Diffs:         []specdrift.FieldDiff{{Kind: specdrift.FieldDiffKindDeleteConstraint}},
				},
			},
		},
	}
	if !isRepairableInstalledPackApplyAllError(err) {
		t.Fatalf("unique/deleteConstraint object_spec drift classified as non-repairable: %v", err)
	}
}

func TestIsRepairableInstalledPackApplyAllErrorRejectsNonUniqueObjectSpecFieldDrift(t *testing.T) {
	err := &packs.ComponentVerifyError{
		Kind:        packs.ComponentKindObjectSpec,
		BusinessKey: "issue",
		Err: &specdrift.StructuralDriftError{
			SpecName: "jira-like-sample",
			Count:    1,
			FieldDrifts: []*specdrift.FieldDriftError{{
				SpecName:      "jira-like-sample",
				ObjectAPIName: "Issue",
				FieldAPIName:  "Subject",
				Diffs: []specdrift.FieldDiff{{
					Kind:    specdrift.FieldDiffKind("Type"),
					Message: `Type: spec="string" live="number"`,
				}},
			}},
		},
	}
	if isRepairableInstalledPackApplyAllError(err) {
		t.Fatal("non-unique object_spec field drift classified as repairable")
	}
}

func TestRepairActiveInstalledPackSourceHashDriftForwardsSkipPolymorphicBackfill(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "crm-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(installedPackRow, packs.Pack) error {
		return &HashDriftError{PackName: row.Name, StoredHash: "old", CurrentHash: "new"}
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}

	sawSkip := false
	sawRepairLegacyUnique := false
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, _ *installPackPreflightOverrides, options InstallPackOptions) error {
		sawSkip = options.SkipPolymorphicDiscriminatorBackfill
		sawRepairLegacyUnique = options.RepairLegacyUniqueDuplicates
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{
		SkipPolymorphicDiscriminatorBackfill: true,
		RepairLegacyUniqueDuplicates:         true,
	}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if !sawSkip {
		t.Fatal("skip polymorphic backfill override was not forwarded to reconcile reapply path")
	}
	if !sawRepairLegacyUnique {
		t.Fatal("repair legacy unique override was not forwarded to reconcile reapply path")
	}
}

// TestRepairHashDriftWithMissingPhysicalColumnFailsClosed verifies that
// repairActiveInstalledPackSourceHashDrift returns a
// *MissingPhysicalColumnDetectOnlyError and does NOT call apply when the
// live-state oracle reports a missing physical column for the drifted pack
// and the metadata-driven repair cannot converge the state.
func TestRepairHashDriftWithMissingPhysicalColumnFailsClosed(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "crm-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origRepair := repairPackOwnedMissingPhysicalColumnsForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		repairPackOwnedMissingPhysicalColumnsForReconcile = origRepair
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(row installedPackRow, _ packs.Pack) error {
		return &HashDriftError{PackName: row.Name, StoredHash: "old", CurrentHash: "new"}
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, packName string) ([]string, error) {
		return []string{"Foo__c.Bar__c"}, nil
	}
	repairPackOwnedMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil // repair cannot converge (e.g. field not in live metadata)
	}
	applyCalled := false
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, _ *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalled = true
		return nil
	}

	gotErr := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{})
	if gotErr == nil {
		t.Fatal("repairActiveInstalledPackSourceHashDrift = nil, want *MissingPhysicalColumnDetectOnlyError")
	}
	var detectOnly *MissingPhysicalColumnDetectOnlyError
	if !errors.As(gotErr, &detectOnly) {
		t.Fatalf("error type = %T (%v), want *MissingPhysicalColumnDetectOnlyError", gotErr, gotErr)
	}
	if detectOnly.PackName != "crm-pack" {
		t.Errorf("MissingPhysicalColumnDetectOnlyError.PackName = %q, want %q", detectOnly.PackName, "crm-pack")
	}
	if len(detectOnly.MissingColumns) != 1 || detectOnly.MissingColumns[0] != "Foo__c.Bar__c" {
		t.Errorf("MissingPhysicalColumnDetectOnlyError.MissingColumns = %v, want [Foo__c.Bar__c]", detectOnly.MissingColumns)
	}
	if applyCalled {
		t.Error("apply was called despite missing physical column; must not reapply when detect-only error occurs")
	}
}

// TestRepairHashDriftWithMissingPhysicalColumnConvergesAfterRepair verifies
// that repairActiveInstalledPackSourceHashDrift calls the metadata-driven
// repair path when the live-state oracle reports missing physical columns,
// and proceeds to reapply when repair converges the state (#1392).
func TestRepairHashDriftWithMissingPhysicalColumnConvergesAfterRepair(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "crm-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origRepair := repairPackOwnedMissingPhysicalColumnsForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		repairPackOwnedMissingPhysicalColumnsForReconcile = origRepair
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(row installedPackRow, _ packs.Pack) error {
		return &HashDriftError{PackName: row.Name, StoredHash: "old", CurrentHash: "new"}
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectCallCount := 0
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, packName string) ([]string, error) {
		inspectCallCount++
		if inspectCallCount == 1 {
			return []string{"Account.LastActivityDate"}, nil // missing before repair
		}
		return nil, nil // converged after repair
	}
	repairPackOwnedMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, packName string) ([]string, error) {
		return []string{"Account.LastActivityDate"}, nil // repair succeeds
	}
	applyCalled := false
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, _ *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalled = true
		return nil
	}

	gotErr := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{})
	if gotErr != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil after repair converges", gotErr)
	}
	if !applyCalled {
		t.Error("apply was not called after repair converged the missing physical column state")
	}
}

// TestRepairHashDriftWithPairedColumnAndFieldRowLossReapplies verifies that the
// detect-only guard remains narrow. When a field row is gone too, the live-state
// oracle has no registered installed column to report, so hash-drift repair can
// still replay the source-backed pack substrate.
func TestRepairHashDriftWithPairedColumnAndFieldRowLossReapplies(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "crm-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(row installedPackRow, _ packs.Pack) error {
		return &HashDriftError{PackName: row.Name, StoredHash: "old", CurrentHash: "new"}
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	applyCalled := false
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, _ *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalled = true
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if !applyCalled {
		t.Error("apply was not called for paired field-row/column loss hash drift")
	}
}

// TestRepairHashDriftWithLegitimateNewSourceColumnProceedsNormally verifies that
// repairActiveInstalledPackSourceHashDrift proceeds with reapply when the oracle
// returns an empty list. This covers the legitimate source-upgrade case: a new
// source column has not been applied yet, so it has no _System_Field row; the
// oracle correctly reports nothing missing and reapply is allowed.
func TestRepairHashDriftWithLegitimateNewSourceColumnProceedsNormally(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "crm-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origRebuild := rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = origRebuild
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(row installedPackRow, _ packs.Pack) error {
		return &HashDriftError{PackName: row.Name, StoredHash: "old", CurrentHash: "new"}
	}
	rebuildRepairPackWithAllowedOwnerlessObjectsForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, pack *packs.SourceBackedPack, _ InstallPackOptions, _ *installPackPreflightOverrides) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	// Oracle returns empty: new column not yet in _System_Field.
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		return nil, nil
	}
	applyCalled := false
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, _ *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalled = true
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if !applyCalled {
		t.Error("apply was not called; legitimate source upgrade with empty oracle result must proceed with reapply")
	}
}

// TestRepairHashDriftNoDriftedRowsDoesNotCallApplyOrOracle verifies that
// repairActiveInstalledPackSourceHashDrift does not call apply or the oracle
// when no rows have hash drift.
func TestRepairHashDriftNoDriftedRowsDoesNotCallApplyOrOracle(t *testing.T) {
	rawDB, _, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock: %v", err)
	}
	defer rawDB.Close()

	runner := &Runner{DB: database.NewConnectionFromDB(rawDB)}
	row := installedPackRow{
		Name:       "crm-pack",
		Status:     installedPackStatusActive,
		SourceKind: sql.NullString{Valid: true, String: string(packs.PackSourceKindLocalPath)},
		SourceURI:  sql.NullString{Valid: true, String: "/tmp/crm-pack"},
	}
	pack := testControlplaneSourceBackedPack(t)

	origResolve := sourceBackedPackFromInstalledRowForReconcile
	origVerifyHash := verifyInstalledPackHashStateForReconcile
	origInspect := inspectLiveInstalledPackMissingPhysicalColumnsForReconcile
	origApply := applyResolvedSourceBackedPackUnlockedForReconcile
	defer func() {
		sourceBackedPackFromInstalledRowForReconcile = origResolve
		verifyInstalledPackHashStateForReconcile = origVerifyHash
		inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = origInspect
		applyResolvedSourceBackedPackUnlockedForReconcile = origApply
	}()

	sourceBackedPackFromInstalledRowForReconcile = func(_ context.Context, _ *sql.DB, _ installedPackRow, _ ...packs.SourceBackedPackOption) (*packs.SourceBackedPack, error) {
		return pack, nil
	}
	verifyInstalledPackHashStateForReconcile = func(installedPackRow, packs.Pack) error {
		return nil
	}
	oracleCalled := false
	inspectLiveInstalledPackMissingPhysicalColumnsForReconcile = func(_ context.Context, _ *sql.DB, _ string) ([]string, error) {
		oracleCalled = true
		return nil, nil
	}
	applyCalled := false
	applyResolvedSourceBackedPackUnlockedForReconcile = func(_ context.Context, _ *Runner, _ *packs.SourceBackedPack, _ *installPackPreflightOverrides, _ InstallPackOptions) error {
		applyCalled = true
		return nil
	}

	if err := repairActiveInstalledPackSourceHashDrift(context.Background(), runner, []installedPackRow{row}, InstallPackOptions{}); err != nil {
		t.Fatalf("repairActiveInstalledPackSourceHashDrift = %v, want nil", err)
	}
	if oracleCalled {
		t.Error("oracle was called despite no hash drift; must not inspect when row is not drifted")
	}
	if applyCalled {
		t.Error("apply was called despite no hash drift")
	}
}

// TestMissingPhysicalColumnDetectOnlyErrorMessage verifies the error message
// includes the pack name and missing column list.
func TestMissingPhysicalColumnDetectOnlyErrorMessage(t *testing.T) {
	e := &MissingPhysicalColumnDetectOnlyError{
		PackName:       "crm-pack",
		MissingColumns: []string{"Account.CustomField__c", "Contact.AnotherField__c"},
	}
	msg := e.Error()
	if !strings.Contains(msg, "crm-pack") {
		t.Errorf("error = %q, want pack name", msg)
	}
	if !strings.Contains(msg, "Account.CustomField__c") {
		t.Errorf("error = %q, want missing column Account.CustomField__c", msg)
	}
	if !strings.Contains(msg, "reapply blocked") {
		t.Errorf("error = %q, want 'reapply blocked'", msg)
	}
}

func testControlplaneConnectorSourceBackedPack(t *testing.T) *packs.SourceBackedPack {
	t.Helper()
	source := &packs.ResolvedPackSource{
		Kind: packs.PackSourceKindLocalPath,
		URI:  "/tmp/test-connector",
		Manifest: &packs.PackManifest{
			ManifestFormatVersion: packs.SupportedManifestFormatVersion,
			Name:                  "test-connector",
			DisplayName:           "Test Connector",
			Category:              packs.ConnectorPack,
			Version:               "0.1.0",
			PlatformCompatibility: packs.PlatformCompatibility{Min: "0.1.0", MaxExclusive: "0.2.0"},
			Components: []packs.ManifestComponent{
				{
					Kind:          packs.ComponentKindConnectorCapability,
					BusinessKey:   "test-connector:signature_verifier",
					Manageability: packs.ManageabilityLocked,
					Payload:       []byte(`{"namespace":"test-connector","kind":"signature_verifier"}`),
				},
				{
					Kind:          packs.ComponentKindConnectorInboundHandlerConfig,
					BusinessKey:   "test-connector:signature_verifier",
					Manageability: packs.ManageabilityLocked,
					Payload: []byte(`{
  "namespace":"test-connector",
  "capabilityKind":"signature_verifier",
  "config":{
    "scheme":"hmac-sha256-template",
    "signingTemplate":"{Timestamp}.{RawBody}",
    "timestampHeader":"X-Test-Timestamp",
    "signatureHeader":"X-Test-Signature",
    "signatureEncoding":"hex",
    "skewWindow":"5m",
    "secretName":"default"
  }
}`),
				},
			},
		},
	}
	pack, err := packs.NewSourceBackedPack(source)
	if err != nil {
		t.Fatalf("construct connector source-backed pack: %v", err)
	}
	return pack
}