package schemacatalog

import (
	"bytes"
	"encoding/json"
	"fmt"
	"reflect"
	"sort"
	"strconv"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/schema"
	sharedconstants "github.com/nexuscrm/shared/pkg/constants"
)

const systemObjectLabelMaxRunes = 255
const tidbMaxIndexKeyBytes = 3072

func TestLayoutAuthoringDocumentsUseLexemePreservingStorage(t *testing.T) {
	definitions, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	for _, tableName := range []string{"_System_PageLayout", "_System_PageLayoutHistory"} {
		columns := columnDefinitionsByName(findSystemTableDefinition(t, definitions, tableName))
		config, ok := columns["Config"]
		if !ok {
			t.Fatalf("%s.Config is missing", tableName)
		}
		if config.Type != "MEDIUMTEXT" || config.Nullable {
			t.Fatalf("%s.Config = %+v, want non-null MEDIUMTEXT so exact authoring JSON survives storage", tableName, config)
		}
	}
}

func TestActionSchemaDeclaresCollectionAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	column, ok := columnDefinitionsByName(
		findSystemTableDefinition(t, defs, "_System_ActionSchema"),
	)["IsCollection"]
	if !ok {
		t.Fatal("_System_ActionSchema.IsCollection is missing")
	}
	if column.Type != "TINYINT(1)" || column.Nullable || column.Default != "0" {
		t.Fatalf("_System_ActionSchema.IsCollection = %+v, want non-null TINYINT(1) default 0", column)
	}
}

func TestUserPhysicalIdentityFieldsDoNotCompeteWithVirtualNameAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	user := findSystemTableDefinition(t, defs, sharedconstants.TableUser)
	columns := columnDefinitionsByName(user)
	for _, fieldName := range []string{
		sharedconstants.FieldUsername,
		sharedconstants.FieldEmail,
		sharedconstants.FieldFirstName,
		sharedconstants.FieldLastName,
	} {
		field, ok := columns[fieldName]
		if !ok {
			t.Fatalf("User.%s is missing", fieldName)
		}
		if field.IsNameField {
			t.Fatalf("User.%s must not compete with the platform-owned virtual User.Name authority", fieldName)
		}
	}
}

func TestEmailTemplateSystemTableDeclaresSalesforceKeyPrefix(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	emailTemplate := findSystemTableDefinition(t, defs, "_System_EmailTemplate")
	if emailTemplate.KeyPrefix != "00X" {
		t.Fatalf("_System_EmailTemplate keyPrefix = %q, want 00X for record-trigger dispatch and Salesforce EmailTemplate identity", emailTemplate.KeyPrefix)
	}
}

func TestSystemCommentSystemTableDeclaresStableInternalKeyPrefix(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	comment := findSystemTableDefinition(t, defs, "_System_Comment")
	if comment.KeyPrefix != "!cm" {
		t.Fatalf("_System_Comment keyPrefix = %q, want stable internal prefix !cm", comment.KeyPrefix)
	}
}

func TestConnectorIngressControlDeclaresReceiptSafeNamespaceFenceEpoch(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	ingressControl := findSystemTableDefinition(t, defs, "_System_ConnectorIngressControl")
	epoch, ok := columnDefinitionsByName(ingressControl)["FenceEpoch"]
	if !ok {
		t.Fatal("_System_ConnectorIngressControl.FenceEpoch is missing")
	}
	if epoch.Type != "BIGINT" || epoch.Nullable || epoch.Default != "0" {
		t.Fatalf("_System_ConnectorIngressControl.FenceEpoch = %+v, want non-null BIGINT default 0", epoch)
	}
}

func TestOutboundIntentTablesDeclareTypedTerminalResults(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	for _, tableName := range []string{
		"_System_IntegrationOutboundCommand",
		"_System_RuntimeOutboundIntent",
	} {
		columns := columnDefinitionsByName(findSystemTableDefinition(t, defs, tableName))
		requireColumn(t, columns, "ResultFormatVersion", "BIGINT", true, false, "")
		requireColumn(t, columns, "ResultDigest", "CHAR(71)", true, false, "")
		requireColumn(t, columns, "CanonicalResult", "MEDIUMBLOB", true, false, "")
	}
}

func TestIntegrationOutboundCommandSeparatesDispatchAndSafeRetryCounts(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	columns := columnDefinitionsByName(findSystemTableDefinition(t, defs, "_System_IntegrationOutboundCommand"))
	requireColumn(t, columns, "AttemptCount", "BIGINT", false, false, "0")
	requireColumn(t, columns, "RetryCount", "BIGINT", false, false, "0")
}

func TestRuntimeMailboxAuthorizationHandleDeclaresBoundedReplayAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	handle := findSystemTableDefinition(t, defs, "_System_RuntimeMailboxAuthorizationHandle")
	columns := columnDefinitionsByName(handle)
	requireColumn(t, columns, "HandleDigest", "CHAR(64)", false, true, "")
	requireColumn(t, columns, "OrgId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "AppInstallationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "AuthorizedSubjectId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "SessionId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "AuthEpoch", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, columns, "PermissionStateLineage", "BINARY(16)", false, false, "")
	requireColumn(t, columns, "PermissionStateRevision", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, columns, "CatalogRevision", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, columns, "SecurityEpoch", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, columns, "ExpiresAt", "DATETIME(6)", false, false, "")
	requireColumn(t, columns, "ConsumedAt", "DATETIME(6)", true, false, "")
}

func TestConnectorCredentialGenerationDeclaresImmutableExactOwner(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	generation := findSystemTableDefinition(t, defs, "_System_ConnectorCredentialGeneration")
	if generation.TableType != "system_core" || generation.Category != "integration" {
		t.Fatalf(
			"%s classification = %q/%q, want system_core/integration",
			generation.TableName,
			generation.TableType,
			generation.Category,
		)
	}
	if generation.RequiredReadPermission != "BLOCKED" ||
		generation.RequiredWritePermission != "BLOCKED" {
		t.Fatalf(
			"%s permissions = %q/%q, want BLOCKED/BLOCKED",
			generation.TableName,
			generation.RequiredReadPermission,
			generation.RequiredWritePermission,
		)
	}

	columns := columnDefinitionsByName(generation)
	requireColumn(t, columns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, columns, "StableCredentialIdentity", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "ProviderScopeId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "ConnectorAccountId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "SecretBackendProtocol", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "SecretBackendVersionRef", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "TransportConfiguration", "MEDIUMBLOB", false, false, "")
	requireColumn(t, columns, "TransportConfigurationDigest", "CHAR(71)", false, false, "")
	requireColumn(t, columns, "ReceiptProtocolVersion", "VARCHAR(32)", false, false, "")
	requireColumn(t, columns, "OwnerReceipt", "MEDIUMBLOB", false, false, "")
	requireColumn(t, columns, "OwnerReceiptDigest", "CHAR(71)", false, false, "")
	requireColumn(t, columns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, columns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, generation, "idx_connector_credential_generation_identity", []string{
		"StableCredentialIdentity",
		"Id",
	}, false, false)
	requireIndex(t, generation, "idx_connector_credential_generation_scope", []string{
		"ProviderScopeId",
		"ConnectorAccountId",
	}, false, false)
	requireBinaryCollation(t, generation, []string{
		"Id",
		"StableCredentialIdentity",
		"ProviderScopeId",
		"ConnectorAccountId",
		"SecretBackendProtocol",
		"SecretBackendVersionRef",
		"TransportConfigurationDigest",
		"ReceiptProtocolVersion",
		"OwnerReceiptDigest",
	})
	if len(generation.ForeignKeys) != 0 {
		t.Fatalf(
			"%s foreign keys = %+v, want none so mutable names/accounts cannot strand retained generations",
			generation.TableName,
			generation.ForeignKeys,
		)
	}
	for _, forbidden := range []string{
		"DeveloperName",
		"NamedCredentialId",
		"SecretValue",
		"SecretCiphertext",
		"CurrentGenerationId",
		"IsCurrent",
		"State",
		"Mode",
		"FenceToken",
		"RevokedAt",
		"DispatchEligibility",
	} {
		if _, ok := columns[forbidden]; ok {
			t.Fatalf("%s gained mutable, secret, or future-gate field %q", generation.TableName, forbidden)
		}
	}
}

func TestConnectorCredentialGenerationDispatchGateDeclaresOneExactFencedAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	gate := findSystemTableDefinition(t, defs, "_System_ConnectorCredentialGenerationDispatchGate")
	if gate.TableType != "system_core" || gate.Category != "integration" {
		t.Fatalf(
			"%s classification = %q/%q, want system_core/integration",
			gate.TableName,
			gate.TableType,
			gate.Category,
		)
	}
	if gate.RequiredReadPermission != "BLOCKED" || gate.RequiredWritePermission != "BLOCKED" {
		t.Fatalf(
			"%s permissions = %q/%q, want BLOCKED/BLOCKED",
			gate.TableName,
			gate.RequiredReadPermission,
			gate.RequiredWritePermission,
		)
	}

	columns := columnDefinitionsByName(gate)
	if len(columns) != 5 {
		t.Fatalf("%s columns = %+v, want exact five-column dispatch-gate authority", gate.TableName, columns)
	}
	requireColumn(t, columns, "CredentialGenerationId", "VARCHAR(255)", false, true, "")
	requireColumn(t, columns, "Mode", "VARCHAR(32)", false, false, "'enabled'")
	requireColumn(t, columns, "FenceToken", "BIGINT", false, false, "0")
	requireColumn(t, columns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, columns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireBinaryCollation(t, gate, []string{"CredentialGenerationId", "Mode"})
	if len(gate.Indices) != 0 {
		t.Fatalf("%s secondary indices = %+v, want none beyond its exact-generation primary key", gate.TableName, gate.Indices)
	}
	if len(gate.ForeignKeys) != 1 {
		t.Fatalf("%s foreign keys = %+v, want one exact immutable-generation parent", gate.TableName, gate.ForeignKeys)
	}
	requireForeignKey(
		t,
		gate,
		"CredentialGenerationId",
		"_System_ConnectorCredentialGeneration(Id)",
		"RESTRICT",
	)
	for _, forbidden := range []string{
		"Id",
		"StableCredentialIdentity",
		"ProviderScopeId",
		"ConnectorAccountId",
		"NamedCredentialId",
		"DeveloperName",
		"SecretBackendVersionRef",
		"SecretValue",
		"SecretCiphertext",
		"CurrentGenerationId",
		"ServingEpochId",
		"DispatchPermitId",
		"OpenPermitCount",
		"RevocationReason",
		"RevokedAt",
	} {
		if _, ok := columns[forbidden]; ok {
			t.Fatalf("%s gained copied, secret, permit, selector, or audit field %q", gate.TableName, forbidden)
		}
	}
}

func TestConnectorServiceConfigurationDeclaresOneImmutableNormalizedOwner(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	generation := findSystemTableDefinition(t, defs, "_System_ConnectorServiceConfigurationGeneration")
	account := findSystemTableDefinition(t, defs, "_System_ConnectorServiceConfigurationAccount")
	binding := findSystemTableDefinition(t, defs, "_System_ConnectorServiceConfigurationBinding")
	selection := findSystemTableDefinition(t, defs, "_System_ConnectorServiceConfigurationCredentialSelection")
	for _, definition := range []schema.TableDefinition{generation, account, binding, selection} {
		if definition.TableType != "system_core" || definition.Category != "integration" {
			t.Fatalf(
				"%s classification = %q/%q, want system_core/integration",
				definition.TableName,
				definition.TableType,
				definition.Category,
			)
		}
		if definition.RequiredReadPermission != "BLOCKED" ||
			definition.RequiredWritePermission != "BLOCKED" {
			t.Fatalf(
				"%s permissions = %q/%q, want BLOCKED/BLOCKED",
				definition.TableName,
				definition.RequiredReadPermission,
				definition.RequiredWritePermission,
			)
		}
	}

	generationColumns := columnDefinitionsByName(generation)
	if len(generationColumns) != 24 {
		t.Fatalf("%s columns = %+v, want exact 24-column immutable owner", generation.TableName, generationColumns)
	}
	requireColumn(t, generationColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, generationColumns, "GenerationProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "OrgId", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "AppInstallationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "AuthorizedWorkloadId", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "CatalogRevision", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, generationColumns, "SchemaGeneration", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, generationColumns, "EngineGeneration", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, generationColumns, "SourceSnapshotDigest", "CHAR(71)", false, false, "")
	requireColumn(t, generationColumns, "CatalogViewDigest", "CHAR(71)", false, false, "")
	requireColumn(t, generationColumns, "CatalogReceiptProtocolVersion", "VARCHAR(32)", false, false, "")
	requireColumn(t, generationColumns, "CatalogReceiptCapabilityVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "CatalogSecurityEpoch", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, generationColumns, "ConfigurationOwnerRevisionId", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "ConfigurationOwnerRevisionDigest", "CHAR(71)", false, false, "")
	requireColumn(t, generationColumns, "GenerationDigest", "CHAR(71)", false, false, "")
	requireColumn(t, generationColumns, "ReceiptProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "OwnerReceipt", "MEDIUMBLOB", false, false, "")
	requireColumn(t, generationColumns, "OwnerReceiptDigest", "CHAR(71)", false, false, "")
	requireColumn(t, generationColumns, "AccountCount", "BIGINT", false, false, "")
	requireColumn(t, generationColumns, "BindingCount", "BIGINT", false, false, "")
	requireColumn(t, generationColumns, "CredentialSelectionCount", "BIGINT", false, false, "")
	requireColumn(t, generationColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, generationColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, generation, "idx_connector_service_configuration_catalog_view", []string{
		"CatalogViewDigest",
		"Id",
	}, false, false)
	requireBinaryCollation(t, generation, []string{
		"Id",
		"GenerationProtocolVersion",
		"OrgId",
		"AppInstallationId",
		"AuthorizedWorkloadId",
		"SourceSnapshotDigest",
		"CatalogViewDigest",
		"CatalogReceiptProtocolVersion",
		"CatalogReceiptCapabilityVersion",
		"ConfigurationOwnerRevisionId",
		"ConfigurationOwnerRevisionDigest",
		"GenerationDigest",
		"ReceiptProtocolVersion",
		"OwnerReceiptDigest",
	})
	if len(generation.ForeignKeys) != 0 {
		t.Fatalf("%s foreign keys = %+v, want no mutable external parent", generation.TableName, generation.ForeignKeys)
	}

	accountColumns := columnDefinitionsByName(account)
	if len(accountColumns) != 11 {
		t.Fatalf("%s columns = %+v, want exact 11-column account member", account.TableName, accountColumns)
	}
	requireColumn(t, accountColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, accountColumns, "ServiceConfigurationGenerationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, accountColumns, "AccountOrdinal", "INT", false, false, "")
	requireColumn(t, accountColumns, "ConnectorAccountId", "VARCHAR(255)", false, false, "")
	requireColumn(t, accountColumns, "ProviderScopeId", "VARCHAR(255)", false, false, "")
	requireColumn(t, accountColumns, "AccountRevisionId", "VARCHAR(255)", false, false, "")
	requireColumn(t, accountColumns, "AccountRevisionDigest", "CHAR(71)", false, false, "")
	requireColumn(t, accountColumns, "ProviderRevisionId", "VARCHAR(255)", false, false, "")
	requireColumn(t, accountColumns, "ProviderRevisionDigest", "CHAR(71)", false, false, "")
	requireColumn(t, accountColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, accountColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, account, "ux_connector_service_configuration_account_ordinal", []string{
		"ServiceConfigurationGenerationId",
		"AccountOrdinal",
	}, true, false)
	requireIndex(t, account, "ux_connector_service_configuration_account_identity", []string{
		"ServiceConfigurationGenerationId",
		"ConnectorAccountId",
	}, true, false)
	requireBinaryCollation(t, account, []string{
		"Id",
		"ServiceConfigurationGenerationId",
		"ConnectorAccountId",
		"ProviderScopeId",
		"AccountRevisionId",
		"AccountRevisionDigest",
		"ProviderRevisionId",
		"ProviderRevisionDigest",
	})
	if len(account.ForeignKeys) != 1 {
		t.Fatalf("%s foreign keys = %+v, want exact generation parent", account.TableName, account.ForeignKeys)
	}
	requireForeignKey(t, account, "ServiceConfigurationGenerationId", "_System_ConnectorServiceConfigurationGeneration(Id)", "RESTRICT")

	bindingColumns := columnDefinitionsByName(binding)
	if len(bindingColumns) != 9 {
		t.Fatalf("%s columns = %+v, want exact nine-column binding configuration", binding.TableName, bindingColumns)
	}
	requireColumn(t, bindingColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, bindingColumns, "AccountConfigurationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, bindingColumns, "CatalogBindingOrdinal", "INT", false, false, "")
	requireColumn(t, bindingColumns, "ConfigurationRevisionId", "VARCHAR(255)", false, false, "")
	requireColumn(t, bindingColumns, "ConfigurationRevisionDigest", "CHAR(71)", false, false, "")
	requireColumn(t, bindingColumns, "ConfigurationFormatVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, bindingColumns, "CanonicalConfiguration", "MEDIUMBLOB", false, false, "")
	requireColumn(t, bindingColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, bindingColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, binding, "ux_connector_service_configuration_binding_ordinal", []string{
		"AccountConfigurationId",
		"CatalogBindingOrdinal",
	}, true, false)
	requireBinaryCollation(t, binding, []string{
		"Id",
		"AccountConfigurationId",
		"ConfigurationRevisionId",
		"ConfigurationRevisionDigest",
		"ConfigurationFormatVersion",
	})
	if len(binding.ForeignKeys) != 1 {
		t.Fatalf("%s foreign keys = %+v, want exact account parent", binding.TableName, binding.ForeignKeys)
	}
	requireForeignKey(t, binding, "AccountConfigurationId", "_System_ConnectorServiceConfigurationAccount(Id)", "RESTRICT")

	selectionColumns := columnDefinitionsByName(selection)
	if len(selectionColumns) != 7 {
		t.Fatalf("%s columns = %+v, want exact seven-column credential relation", selection.TableName, selectionColumns)
	}
	requireColumn(t, selectionColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, selectionColumns, "BindingConfigurationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, selectionColumns, "CredentialOrdinal", "INT", false, false, "")
	requireColumn(t, selectionColumns, "CredentialRef", "VARCHAR(255)", false, false, "")
	requireColumn(t, selectionColumns, "CredentialGenerationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, selectionColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, selectionColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, selection, "ux_connector_service_configuration_credential_ordinal", []string{
		"BindingConfigurationId",
		"CredentialOrdinal",
	}, true, false)
	requireIndex(t, selection, "ux_connector_service_configuration_credential_ref", []string{
		"BindingConfigurationId",
		"CredentialRef",
	}, true, false)
	requireIndex(t, selection, "ux_connector_service_configuration_credential_generation", []string{
		"BindingConfigurationId",
		"CredentialGenerationId",
	}, true, false)
	requireBinaryCollation(t, selection, []string{
		"Id",
		"BindingConfigurationId",
		"CredentialRef",
		"CredentialGenerationId",
	})
	if len(selection.ForeignKeys) != 2 {
		t.Fatalf("%s foreign keys = %+v, want binding and exact credential-generation parents", selection.TableName, selection.ForeignKeys)
	}
	requireForeignKey(t, selection, "BindingConfigurationId", "_System_ConnectorServiceConfigurationBinding(Id)", "RESTRICT")
	requireForeignKey(t, selection, "CredentialGenerationId", "_System_ConnectorCredentialGeneration(Id)", "RESTRICT")

	for _, forbidden := range []string{
		"Kind",
		"OwnerPack",
		"PackInstallationId",
		"PackPayloadDigest",
		"BusinessKey",
		"DefinitionVersion",
		"CredentialRefs",
		"RuntimeCommandKind",
		"Enabled",
		"State",
		"Mode",
		"IsCurrent",
		"CurrentGenerationId",
		"LatestGenerationId",
		"DeveloperName",
		"ProviderName",
		"SecretValue",
		"SecretCiphertext",
	} {
		for _, definition := range []schema.TableDefinition{generation, account, binding, selection} {
			if _, ok := columnDefinitionsByName(definition)[forbidden]; ok {
				t.Fatalf("%s gained copied Catalog, provider, secret, mutable, or selector field %q", definition.TableName, forbidden)
			}
		}
	}
}

func TestConnectorV2ServingTablesDeclareInertGenerationFence(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	epoch := findSystemTableDefinition(t, defs, "_System_ConnectorServingEpoch")
	binding := findSystemTableDefinition(t, defs, "_System_ConnectorServingEpochBinding")
	gate := findSystemTableDefinition(t, defs, "_System_ConnectorAccountServingGate")
	for _, definition := range []schema.TableDefinition{epoch, binding, gate} {
		if definition.TableType != "system_core" || definition.Category != "integration" {
			t.Fatalf(
				"%s classification = %q/%q, want system_core/integration",
				definition.TableName,
				definition.TableType,
				definition.Category,
			)
		}
		if definition.RequiredReadPermission != "BLOCKED" ||
			definition.RequiredWritePermission != "BLOCKED" {
			t.Fatalf(
				"%s permissions = %q/%q, want BLOCKED/BLOCKED",
				definition.TableName,
				definition.RequiredReadPermission,
				definition.RequiredWritePermission,
			)
		}
	}

	epochColumns := columnDefinitionsByName(epoch)
	requireColumn(t, epochColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, epochColumns, "EpochDigest", "CHAR(71)", false, false, "")
	requireColumn(t, epochColumns, "ServingScopeDigest", "CHAR(64)", false, false, "")
	requireColumn(t, epochColumns, "ActiveScopeDigest", "CHAR(64)", true, false, "")
	requireColumn(t, epochColumns, "State", "VARCHAR(32)", false, false, "'prepared'")
	requireColumn(t, epochColumns, "CatalogRevision", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, epochColumns, "SchemaGeneration", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, epochColumns, "EngineGeneration", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, epochColumns, "SourceSnapshotDigest", "CHAR(71)", false, false, "")
	requireColumn(t, epochColumns, "BindingViewDigest", "CHAR(71)", false, false, "")
	requireColumn(t, epochColumns, "SecurityEpoch", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, epochColumns, "L2ConfigurationGenerationRef", "VARCHAR(255)", false, false, "")
	requireColumn(t, epochColumns, "L2ConfigurationDigest", "CHAR(71)", false, false, "")
	requireColumn(t, epochColumns, "L2VerificationReceipt", "MEDIUMBLOB", false, false, "")
	requireColumn(t, epochColumns, "L2VerificationReceiptDigest", "CHAR(71)", false, false, "")
	requireIndex(t, epoch, "ux_connector_serving_epoch_digest", []string{"EpochDigest"}, true, false)
	requireIndex(t, epoch, "ux_connector_serving_epoch_active_scope", []string{"ActiveScopeDigest"}, true, false)
	requireIndex(t, epoch, "idx_connector_serving_epoch_scope_state", []string{
		"ServingScopeDigest",
		"State",
		"Id",
	}, false, false)
	requireBinaryCollation(t, epoch, []string{
		"Id",
		"EpochDigest",
		"ServingScopeDigest",
		"ActiveScopeDigest",
		"State",
		"OrgId",
		"AppInstallationId",
		"AuthorizedWorkloadId",
		"SourceSnapshotDigest",
		"BindingViewDigest",
		"ReceiptProtocolVersion",
		"ReceiptCapabilityVersion",
		"L2ConfigurationGenerationRef",
		"L2ConfigurationDigest",
		"L2VerificationReceiptDigest",
	})

	bindingColumns := columnDefinitionsByName(binding)
	requireColumn(t, bindingColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, bindingColumns, "ServingEpochId", "VARCHAR(255)", false, false, "")
	requireColumn(t, bindingColumns, "BindingOrdinal", "INT", false, false, "")
	requireColumn(t, bindingColumns, "BindingIdentityDigest", "CHAR(64)", false, false, "")
	requireColumn(t, bindingColumns, "PackPayloadDigest", "CHAR(85)", false, false, "")
	requireColumn(t, bindingColumns, "CredentialRefs", "MEDIUMBLOB", false, false, "")
	for _, name := range []string{
		"RuntimeCommandKind",
		"RuntimePayloadFormatVersion",
		"RuntimeDefinitionRef",
		"RuntimeDefinitionVersion",
	} {
		if column, ok := bindingColumns[name]; !ok || !column.Nullable {
			t.Fatalf("%s.%s = %+v, want nullable optional Runtime binding field", binding.TableName, name, column)
		}
	}
	requireIndex(t, binding, "ux_connector_epoch_binding_ordinal", []string{
		"ServingEpochId",
		"BindingOrdinal",
	}, true, false)
	requireIndex(t, binding, "ux_connector_epoch_binding_identity", []string{
		"ServingEpochId",
		"BindingIdentityDigest",
	}, true, false)
	requireForeignKey(t, binding, "ServingEpochId", "_System_ConnectorServingEpoch(Id)", "")
	requireBinaryCollation(t, binding, []string{
		"Id",
		"ServingEpochId",
		"BindingIdentityDigest",
		"Kind",
		"OwnerPack",
		"PackInstallationId",
		"PackPayloadDigest",
		"BusinessKey",
		"RuntimeCommandKind",
		"RuntimeDefinitionRef",
		"RuntimeDefinitionVersion",
	})

	gateColumns := columnDefinitionsByName(gate)
	requireColumn(t, gateColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, gateColumns, "ServingScopeDigest", "CHAR(64)", false, false, "")
	requireColumn(t, gateColumns, "ConnectorAccountId", "VARCHAR(255)", false, false, "")
	requireColumn(t, gateColumns, "Mode", "VARCHAR(32)", false, false, "'v1_serving'")
	requireColumn(t, gateColumns, "FenceToken", "BIGINT", false, false, "0")
	requireColumn(t, gateColumns, "ServingEpochId", "VARCHAR(255)", true, false, "")
	requireColumn(t, gateColumns, "ServingEpochDigest", "CHAR(71)", true, false, "")
	requireIndex(t, gate, "ux_connector_account_gate_scope_account", []string{
		"ServingScopeDigest",
		"ConnectorAccountId",
	}, true, false)
	requireIndex(t, gate, "idx_connector_account_gate_mode", []string{
		"ServingScopeDigest",
		"Mode",
		"ConnectorAccountId",
	}, false, false)
	requireIndex(t, gate, "idx_connector_account_gate_epoch", []string{
		"ServingEpochId",
		"Mode",
		"ConnectorAccountId",
	}, false, false)
	requireForeignKey(t, gate, "ServingEpochId", "_System_ConnectorServingEpoch(Id)", "")
	requireBinaryCollation(t, gate, []string{
		"Id",
		"ServingScopeDigest",
		"ConnectorAccountId",
		"Mode",
		"ServingEpochId",
		"ServingEpochDigest",
	})

	for _, definition := range []schema.TableDefinition{epoch, binding, gate} {
		columns := columnDefinitionsByName(definition)
		for _, forbidden := range []string{
			"IsDeleted",
			"ReceiptId",
			"EventId",
			"WorkItemId",
			"ExecutionId",
			"OutboundCommandId",
			"RemoteEffectId",
			"DispatchPermitId",
			"OpenPermitCount",
			"V1ShutdownCertificateId",
			"V1RestorePointId",
			"Provider",
		} {
			if _, ok := columns[forbidden]; ok {
				t.Fatalf("%s gained future execution or provider field %q", definition.TableName, forbidden)
			}
		}
	}
}

func TestConnectorV2IngressReceiptTablesDeclareImmutableAdmissionAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	receipt := findSystemTableDefinition(t, defs, "_System_ConnectorV2Receipt")
	admissionKey := findSystemTableDefinition(t, defs, "_System_ConnectorV2ReceiptAdmissionKey")
	component := findSystemTableDefinition(t, defs, "_System_ConnectorV2ReceiptComponent")
	for _, definition := range []schema.TableDefinition{receipt, admissionKey, component} {
		if definition.TableType != "system_core" || definition.Category != "integration" {
			t.Fatalf(
				"%s classification = %q/%q, want system_core/integration",
				definition.TableName,
				definition.TableType,
				definition.Category,
			)
		}
		if definition.RequiredReadPermission != "BLOCKED" ||
			definition.RequiredWritePermission != "BLOCKED" {
			t.Fatalf(
				"%s permissions = %q/%q, want BLOCKED/BLOCKED",
				definition.TableName,
				definition.RequiredReadPermission,
				definition.RequiredWritePermission,
			)
		}
	}

	receiptColumns := columnDefinitionsByName(receipt)
	requireColumn(t, receiptColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, receiptColumns, "ReceiptProtocolVersion", "VARCHAR(32)", false, false, "")
	requireColumn(t, receiptColumns, "ServingScopeDigest", "CHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "ConnectorAccountId", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "AccountServingGateId", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "AccountServingFenceToken", "BIGINT", false, false, "")
	requireColumn(t, receiptColumns, "ServingEpochId", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "ServingEpochDigest", "CHAR(71)", false, false, "")
	requireColumn(t, receiptColumns, "AdmissionKind", "VARCHAR(32)", false, false, "")
	requireColumn(t, receiptColumns, "AdmissionAuthorityId", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "PayloadCiphertextProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "PayloadCiphertext", "MEDIUMBLOB", false, false, "")
	requireColumn(t, receiptColumns, "PayloadDigest", "CHAR(71)", false, false, "")
	requireColumn(t, receiptColumns, "PayloadSizeBytes", "BIGINT", false, false, "")
	requireColumn(t, receiptColumns, "ReceiptComponentCount", "BIGINT", false, false, "")
	requireColumn(t, receiptColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, receiptColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, receipt, "ux_connector_v2_receipt_admission", []string{
		"ServingScopeDigest",
		"ConnectorAccountId",
		"AdmissionKind",
		"AdmissionAuthorityId",
	}, true, false)
	requireIndex(t, receipt, "idx_connector_v2_receipt_account", []string{"ConnectorAccountId", "Id"}, false, false)
	requireIndex(t, receipt, "idx_connector_v2_receipt_epoch", []string{"ServingEpochId", "Id"}, false, false)
	requireIndex(t, receipt, "idx_connector_v2_receipt_gate", []string{"AccountServingGateId", "Id"}, false, false)
	requireForeignKey(t, receipt, "AccountServingGateId", "_System_ConnectorAccountServingGate(Id)", "RESTRICT")
	requireForeignKey(t, receipt, "ServingEpochId", "_System_ConnectorServingEpoch(Id)", "RESTRICT")
	requireBinaryCollation(t, receipt, []string{
		"Id",
		"ReceiptProtocolVersion",
		"ServingScopeDigest",
		"ConnectorAccountId",
		"AccountServingGateId",
		"ServingEpochId",
		"ServingEpochDigest",
		"AdmissionKind",
		"AdmissionAuthorityId",
		"PayloadCiphertextProtocolVersion",
		"PayloadDigest",
	})

	componentColumns := columnDefinitionsByName(component)
	requireColumn(t, componentColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, componentColumns, "ComponentProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, componentColumns, "ReceiptId", "VARCHAR(255)", false, false, "")
	requireColumn(t, componentColumns, "ComponentOrdinal", "INT", false, false, "")
	requireColumn(t, componentColumns, "PayloadCiphertextProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, componentColumns, "PayloadCiphertext", "MEDIUMBLOB", false, false, "")
	requireColumn(t, componentColumns, "PayloadDigest", "CHAR(71)", false, false, "")
	requireColumn(t, componentColumns, "PayloadSizeBytes", "BIGINT", false, false, "")
	requireColumn(t, componentColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, componentColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, component, "ux_connector_v2_receipt_component_ordinal", []string{
		"ReceiptId",
		"ComponentOrdinal",
	}, true, false)
	requireForeignKey(t, component, "ReceiptId", "_System_ConnectorV2Receipt(Id)", "RESTRICT")
	requireBinaryCollation(t, component, []string{
		"Id",
		"ComponentProtocolVersion",
		"ReceiptId",
		"PayloadCiphertextProtocolVersion",
		"PayloadDigest",
	})

	admissionColumns := columnDefinitionsByName(admissionKey)
	requireColumn(t, admissionColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, admissionColumns, "AdmissionKeyProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, admissionColumns, "ServingScopeDigest", "CHAR(64)", false, false, "")
	requireColumn(t, admissionColumns, "ConnectorAccountId", "VARCHAR(255)", false, false, "")
	requireColumn(t, admissionColumns, "AccountServingGateId", "VARCHAR(255)", false, false, "")
	requireColumn(t, admissionColumns, "IngressChannel", "VARCHAR(64)", false, false, "")
	requireColumn(t, admissionColumns, "TransportIdempotencyKey", "VARCHAR(255)", false, false, "")
	requireColumn(t, admissionColumns, "AuthorizedPayloadDigest", "CHAR(71)", false, false, "")
	requireColumn(t, admissionColumns, "Disposition", "VARCHAR(32)", false, false, "")
	requireColumn(t, admissionColumns, "ReceiptId", "VARCHAR(255)", true, false, "")
	requireColumn(t, admissionColumns, "PredecessorRefProtocolVersion", "VARCHAR(255)", true, false, "")
	requireColumn(t, admissionColumns, "PredecessorRef", "MEDIUMBLOB", true, false, "")
	requireColumn(t, admissionColumns, "PredecessorRefDigest", "CHAR(71)", true, false, "")
	requireColumn(t, admissionColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, admissionColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, admissionKey, "ux_connector_v2_receipt_key_transport", []string{
		"ConnectorAccountId",
		"IngressChannel",
		"TransportIdempotencyKey",
	}, true, false)
	requireIndex(t, admissionKey, "ux_connector_v2_receipt_key_receipt", []string{"ReceiptId"}, true, false)
	requireIndex(t, admissionKey, "idx_connector_v2_receipt_key_gate", []string{
		"AccountServingGateId",
		"Disposition",
		"Id",
	}, false, false)
	requireForeignKey(t, admissionKey, "AccountServingGateId", "_System_ConnectorAccountServingGate(Id)", "RESTRICT")
	requireForeignKey(t, admissionKey, "ReceiptId", "_System_ConnectorV2Receipt(Id)", "RESTRICT")
	requireBinaryCollation(t, admissionKey, []string{
		"Id",
		"AdmissionKeyProtocolVersion",
		"ServingScopeDigest",
		"ConnectorAccountId",
		"AccountServingGateId",
		"IngressChannel",
		"TransportIdempotencyKey",
		"AuthorizedPayloadDigest",
		"Disposition",
		"ReceiptId",
		"PredecessorRefProtocolVersion",
		"PredecessorRefDigest",
	})

	for _, definition := range []schema.TableDefinition{receipt, admissionKey, component} {
		columns := columnDefinitionsByName(definition)
		for _, forbidden := range []string{
			"State",
			"Status",
			"AttemptCount",
			"RetryCount",
			"NextAttemptAt",
			"ClaimedBy",
			"ClaimToken",
			"ClaimedAt",
			"LeaseExpiresAt",
			"ExecutionOwner",
			"WorkItemId",
			"Provider",
			"SecretValue",
		} {
			if _, ok := columns[forbidden]; ok {
				t.Fatalf("%s gained mutable, execution, provider, or secret field %q", definition.TableName, forbidden)
			}
		}

		for _, index := range definition.Indices {
			totalBytes := 0
			for _, columnName := range index.Columns {
				column, ok := columns[columnName]
				if !ok {
					t.Fatalf("%s.%s references missing column %s", definition.TableName, index.Name, columnName)
				}
				bytes, err := maxTiDBIndexBytes(column.Type)
				if err != nil {
					t.Fatalf("%s.%s type %q: %v", definition.TableName, columnName, column.Type, err)
				}
				totalBytes += bytes
			}
			if totalBytes > tidbMaxIndexKeyBytes {
				t.Fatalf("%s worst-case key bytes = %d, want <= %d", index.Name, totalBytes, tidbMaxIndexKeyBytes)
			}
		}
	}
}

func TestConnectorV2ExecutionTablesDeclareFencedRuntimeCorrelationWithoutOwningResults(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	execution := findSystemTableDefinition(t, defs, "_System_ConnectorV2Execution")
	attempt := findSystemTableDefinition(t, defs, "_System_ConnectorV2ExecutionAttempt")
	for _, definition := range []schema.TableDefinition{execution, attempt} {
		if definition.TableType != "system_core" || definition.Category != "integration" {
			t.Fatalf(
				"%s classification = %q/%q, want system_core/integration",
				definition.TableName,
				definition.TableType,
				definition.Category,
			)
		}
		if definition.RequiredReadPermission != "BLOCKED" ||
			definition.RequiredWritePermission != "BLOCKED" {
			t.Fatalf(
				"%s permissions = %q/%q, want BLOCKED/BLOCKED",
				definition.TableName,
				definition.RequiredReadPermission,
				definition.RequiredWritePermission,
			)
		}
	}

	executionColumns := columnDefinitionsByName(execution)
	if len(executionColumns) != 26 {
		t.Fatalf("%s columns = %d, want exact owner shape of 26", execution.TableName, len(executionColumns))
	}
	requireColumn(t, executionColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, executionColumns, "ExecutionProtocolVersion", "VARCHAR(64)", false, false, "")
	requireColumn(t, executionColumns, "ServingScopeDigest", "CHAR(64)", false, false, "")
	requireColumn(t, executionColumns, "ConnectorAccountId", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "AccountServingGateId", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "AccountServingFenceToken", "BIGINT", false, false, "")
	requireColumn(t, executionColumns, "ServingEpochId", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "ServingEpochDigest", "CHAR(71)", false, false, "")
	requireColumn(t, executionColumns, "ServingEpochBindingId", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "PackGenerationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "DefinitionRef", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "DefinitionVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "CanonicalInputs", "MEDIUMBLOB", false, false, "")
	requireColumn(t, executionColumns, "ExternalEffectsMode", "VARCHAR(32)", false, false, "")
	requireColumn(t, executionColumns, "InputDigest", "CHAR(71)", false, false, "")
	requireColumn(t, executionColumns, "ExecutionRefDigest", "CHAR(71)", false, false, "")
	requireColumn(t, executionColumns, "RuntimeCommandId", "VARCHAR(255)", false, false, "")
	requireColumn(t, executionColumns, "State", "VARCHAR(32)", false, false, "'ready'")
	requireColumn(t, executionColumns, "ClaimFenceToken", "BIGINT", false, false, "0")
	requireColumn(t, executionColumns, "ClaimHolderId", "VARCHAR(255)", true, false, "")
	requireColumn(t, executionColumns, "ClaimLeaseUntil", "DATETIME(6)", true, false, "")
	requireColumn(t, executionColumns, "ResultDigest", "CHAR(71)", true, false, "")
	requireColumn(t, executionColumns, "RuntimeCommitReceipt", "CHAR(71)", true, false, "")
	requireColumn(t, executionColumns, "QuarantineReasonCode", "VARCHAR(64)", true, false, "")
	requireColumn(t, executionColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, executionColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	if executionColumns["PackGenerationId"].CollationName != "" {
		t.Fatalf(
			"%s.PackGenerationId collation = %q, want database default matching the established parent",
			execution.TableName,
			executionColumns["PackGenerationId"].CollationName,
		)
	}
	requireIndex(t, execution, "ux_connector_v2_execution_ref", []string{"ExecutionRefDigest"}, true, false)
	requireIndex(t, execution, "ux_connector_v2_execution_runtime_command", []string{"RuntimeCommandId"}, true, false)
	requireIndex(t, execution, "idx_connector_v2_execution_claim", []string{
		"ConnectorAccountId", "State", "ClaimLeaseUntil", "Id",
	}, false, false)
	requireIndex(t, execution, "idx_connector_v2_execution_epoch", []string{"ServingEpochId", "State", "Id"}, false, false)
	requireIndex(t, execution, "idx_connector_v2_execution_gate", []string{"AccountServingGateId", "State", "Id"}, false, false)
	requireForeignKey(t, execution, "AccountServingGateId", "_System_ConnectorAccountServingGate(Id)", "RESTRICT")
	requireForeignKey(t, execution, "ServingEpochId", "_System_ConnectorServingEpoch(Id)", "RESTRICT")
	requireForeignKey(t, execution, "ServingEpochBindingId", "_System_ConnectorServingEpochBindingOwner(ServingEpochBindingId)", "RESTRICT")
	requireForeignKey(t, execution, "PackGenerationId", "_System_PackGenerationConnectorBindingDefinitionSet(PackGenerationId)", "RESTRICT")
	requireBinaryCollation(t, execution, []string{
		"Id",
		"ExecutionProtocolVersion",
		"ServingScopeDigest",
		"ConnectorAccountId",
		"AccountServingGateId",
		"ServingEpochId",
		"ServingEpochDigest",
		"ServingEpochBindingId",
		"DefinitionRef",
		"DefinitionVersion",
		"ExternalEffectsMode",
		"InputDigest",
		"ExecutionRefDigest",
		"RuntimeCommandId",
		"State",
		"ClaimHolderId",
		"ResultDigest",
		"RuntimeCommitReceipt",
		"QuarantineReasonCode",
	})

	attemptColumns := columnDefinitionsByName(attempt)
	if len(attemptColumns) != 8 {
		t.Fatalf("%s columns = %d, want exact append-only evidence shape of 8", attempt.TableName, len(attemptColumns))
	}
	requireColumn(t, attemptColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, attemptColumns, "ExecutionId", "VARCHAR(255)", false, false, "")
	requireColumn(t, attemptColumns, "AttemptOrdinal", "BIGINT", false, false, "")
	requireColumn(t, attemptColumns, "ClaimFenceToken", "BIGINT", false, false, "")
	requireColumn(t, attemptColumns, "ClaimHolderId", "VARCHAR(255)", false, false, "")
	requireColumn(t, attemptColumns, "LeaseUntil", "DATETIME(6)", false, false, "")
	requireColumn(t, attemptColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, attemptColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, attempt, "ux_connector_v2_execution_attempt_ordinal", []string{"ExecutionId", "AttemptOrdinal"}, true, false)
	requireIndex(t, attempt, "ux_connector_v2_execution_attempt_fence", []string{"ExecutionId", "ClaimFenceToken"}, true, false)
	requireIndex(t, attempt, "idx_connector_v2_execution_attempt_holder", []string{"ClaimHolderId", "LeaseUntil", "Id"}, false, false)
	requireForeignKey(t, attempt, "ExecutionId", "_System_ConnectorV2Execution(Id)", "RESTRICT")
	requireBinaryCollation(t, attempt, []string{"Id", "ExecutionId", "ClaimHolderId"})

	for _, definition := range []schema.TableDefinition{execution, attempt} {
		columns := columnDefinitionsByName(definition)
		for _, forbidden := range []string{
			"Outcome",
			"FailureCode",
			"ResultPayload",
			"LastError",
			"RetryCount",
			"Provider",
			"CredentialRef",
			"CurrentRoute",
			"DependencyId",
		} {
			if _, ok := columns[forbidden]; ok {
				t.Fatalf("%s gained a competing Runtime, dependency, or provider fact %q", definition.TableName, forbidden)
			}
		}
	}
}

func TestConnectorServingEpochReproducibilityDeclaresNormalizedOwnerProofs(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	reproducibility := findSystemTableDefinition(t, defs, "_System_ConnectorServingEpochReproducibility")
	bindingOwner := findSystemTableDefinition(t, defs, "_System_ConnectorServingEpochBindingOwner")
	credentialGeneration := findSystemTableDefinition(t, defs, "_System_ConnectorServingEpochCredentialGeneration")
	for _, definition := range []schema.TableDefinition{reproducibility, bindingOwner, credentialGeneration} {
		if definition.TableType != "system_core" || definition.Category != "integration" {
			t.Fatalf(
				"%s classification = %q/%q, want system_core/integration",
				definition.TableName,
				definition.TableType,
				definition.Category,
			)
		}
		if definition.RequiredReadPermission != "BLOCKED" ||
			definition.RequiredWritePermission != "BLOCKED" {
			t.Fatalf(
				"%s permissions = %q/%q, want BLOCKED/BLOCKED",
				definition.TableName,
				definition.RequiredReadPermission,
				definition.RequiredWritePermission,
			)
		}
	}

	for _, requiredDescription := range []string{
		"Absence of this row together with zero binding-owner and credential-generation proof rows marks a legacy candidate",
		"any side row without this header is corruption",
	} {
		if !strings.Contains(reproducibility.Description, requiredDescription) {
			t.Fatalf("%s description %q does not contain %q", reproducibility.TableName, reproducibility.Description, requiredDescription)
		}
	}

	reproducibilityColumns := columnDefinitionsByName(reproducibility)
	requireExactColumnOrder(t, reproducibility, []string{
		"ServingEpochId",
		"ConnectorDigestProfileId",
		"ServiceConfigurationGenerationProtocolVersion",
		"ServiceConfigurationReceiptProtocolVersion",
		"CreatedDate",
		"LastModifiedDate",
	})
	requireColumn(t, reproducibilityColumns, "ServingEpochId", "VARCHAR(255)", false, true, "")
	requireColumn(t, reproducibilityColumns, "ConnectorDigestProfileId", "VARCHAR(255)", false, false, "")
	requireColumn(t, reproducibilityColumns, "ServiceConfigurationGenerationProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, reproducibilityColumns, "ServiceConfigurationReceiptProtocolVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, reproducibilityColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, reproducibilityColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireBinaryCollation(t, reproducibility, []string{
		"ServingEpochId",
		"ConnectorDigestProfileId",
		"ServiceConfigurationGenerationProtocolVersion",
		"ServiceConfigurationReceiptProtocolVersion",
	})
	if len(reproducibility.Indices) != 1 {
		t.Fatalf("%s indices = %+v, want exactly the profile lookup index", reproducibility.TableName, reproducibility.Indices)
	}
	requireIndex(t, reproducibility, "idx_connector_epoch_repro_profile", []string{
		"ConnectorDigestProfileId",
		"ServingEpochId",
	}, false, false)
	if len(reproducibility.ForeignKeys) != 1 {
		t.Fatalf("%s foreign keys = %+v, want only the exact epoch parent", reproducibility.TableName, reproducibility.ForeignKeys)
	}
	requireForeignKey(t, reproducibility, "ServingEpochId", "_System_ConnectorServingEpoch(Id)", "RESTRICT")

	bindingOwnerColumns := columnDefinitionsByName(bindingOwner)
	requireExactColumnOrder(t, bindingOwner, []string{
		"ServingEpochBindingId",
		"PackGenerationId",
		"DefinitionEnvelopeVersion",
		"DefinitionReceiptVersion",
		"CanonicalDefinitionDigest",
		"DefinitionOwnerReceipt",
		"DefinitionOwnerReceiptDigest",
		"CreatedDate",
		"LastModifiedDate",
	})
	requireColumn(t, bindingOwnerColumns, "ServingEpochBindingId", "VARCHAR(255)", false, true, "")
	requireColumn(t, bindingOwnerColumns, "PackGenerationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, bindingOwnerColumns, "DefinitionEnvelopeVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, bindingOwnerColumns, "DefinitionReceiptVersion", "VARCHAR(255)", false, false, "")
	requireColumn(t, bindingOwnerColumns, "CanonicalDefinitionDigest", "CHAR(71)", false, false, "")
	requireColumn(t, bindingOwnerColumns, "DefinitionOwnerReceipt", "MEDIUMBLOB", false, false, "")
	requireColumn(t, bindingOwnerColumns, "DefinitionOwnerReceiptDigest", "CHAR(71)", false, false, "")
	requireColumn(t, bindingOwnerColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, bindingOwnerColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireBinaryCollation(t, bindingOwner, []string{
		"ServingEpochBindingId",
		"DefinitionEnvelopeVersion",
		"DefinitionReceiptVersion",
		"CanonicalDefinitionDigest",
		"DefinitionOwnerReceiptDigest",
	})
	if bindingOwnerColumns["PackGenerationId"].CollationName != "" {
		t.Fatalf(
			"%s.PackGenerationId collation = %q, want DefinitionSet-compatible database default",
			bindingOwner.TableName,
			bindingOwnerColumns["PackGenerationId"].CollationName,
		)
	}
	if len(bindingOwner.Indices) != 1 {
		t.Fatalf("%s indices = %+v, want exactly the PackGeneration lookup index", bindingOwner.TableName, bindingOwner.Indices)
	}
	requireIndex(t, bindingOwner, "idx_connector_epoch_binding_owner_pack", []string{
		"PackGenerationId",
		"ServingEpochBindingId",
	}, false, false)
	if len(bindingOwner.ForeignKeys) != 2 {
		t.Fatalf("%s foreign keys = %+v, want binding and definition-set parents", bindingOwner.TableName, bindingOwner.ForeignKeys)
	}
	requireForeignKey(t, bindingOwner, "ServingEpochBindingId", "_System_ConnectorServingEpochBinding(Id)", "RESTRICT")
	requireForeignKey(t, bindingOwner, "PackGenerationId", "_System_PackGenerationConnectorBindingDefinitionSet(PackGenerationId)", "RESTRICT")

	credentialGenerationColumns := columnDefinitionsByName(credentialGeneration)
	requireExactColumnOrder(t, credentialGeneration, []string{
		"ServingEpochId",
		"CredentialOrdinal",
		"CredentialGenerationId",
		"ReceiptProtocolVersion",
		"TransportConfigurationDigest",
		"OwnerReceipt",
		"OwnerReceiptDigest",
		"CreatedDate",
		"LastModifiedDate",
	})
	requireColumn(t, credentialGenerationColumns, "ServingEpochId", "VARCHAR(255)", false, false, "")
	requireColumn(t, credentialGenerationColumns, "CredentialOrdinal", "INT", false, false, "")
	requireColumn(t, credentialGenerationColumns, "CredentialGenerationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, credentialGenerationColumns, "ReceiptProtocolVersion", "VARCHAR(32)", false, false, "")
	requireColumn(t, credentialGenerationColumns, "TransportConfigurationDigest", "CHAR(71)", false, false, "")
	requireColumn(t, credentialGenerationColumns, "OwnerReceipt", "MEDIUMBLOB", false, false, "")
	requireColumn(t, credentialGenerationColumns, "OwnerReceiptDigest", "CHAR(71)", false, false, "")
	requireColumn(t, credentialGenerationColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, credentialGenerationColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireBinaryCollation(t, credentialGeneration, []string{
		"ServingEpochId",
		"CredentialGenerationId",
		"ReceiptProtocolVersion",
		"TransportConfigurationDigest",
		"OwnerReceiptDigest",
	})
	if len(credentialGeneration.Indices) != 3 {
		t.Fatalf("%s indices = %+v, want exact primary, uniqueness, and reverse-reference indices", credentialGeneration.TableName, credentialGeneration.Indices)
	}
	requireIndex(t, credentialGeneration, "", []string{
		"ServingEpochId",
		"CredentialOrdinal",
	}, false, true)
	requireIndex(t, credentialGeneration, "ux_connector_epoch_credential_generation", []string{
		"ServingEpochId",
		"CredentialGenerationId",
	}, true, false)
	requireIndex(t, credentialGeneration, "idx_connector_epoch_credential_generation_ref", []string{
		"CredentialGenerationId",
		"ServingEpochId",
	}, false, false)
	if len(credentialGeneration.ForeignKeys) != 2 {
		t.Fatalf("%s foreign keys = %+v, want exact epoch and credential-generation parents", credentialGeneration.TableName, credentialGeneration.ForeignKeys)
	}
	requireForeignKey(t, credentialGeneration, "ServingEpochId", "_System_ConnectorServingEpoch(Id)", "RESTRICT")
	requireForeignKey(t, credentialGeneration, "CredentialGenerationId", "_System_ConnectorCredentialGeneration(Id)", "RESTRICT")

	for _, definition := range []schema.TableDefinition{reproducibility, bindingOwner, credentialGeneration} {
		columns := columnDefinitionsByName(definition)
		for _, index := range definition.Indices {
			indexBytes := 0
			for _, name := range index.Columns {
				columnBytes, err := maxTiDBIndexBytes(columns[name].Type)
				if err != nil {
					t.Fatalf("index-width calculation for %s.%s: %v", definition.TableName, name, err)
				}
				indexBytes += columnBytes
			}
			if indexBytes > tidbMaxIndexKeyBytes {
				t.Fatalf("%s index %q requires %d bytes, exceeds TiDB limit %d", definition.TableName, index.Name, indexBytes, tidbMaxIndexKeyBytes)
			}
		}
	}
}

func requireBinaryCollation(t *testing.T, definition schema.TableDefinition, columnNames []string) {
	t.Helper()
	columns := columnDefinitionsByName(definition)
	for _, name := range columnNames {
		column, ok := columns[name]
		if !ok {
			t.Fatalf("%s.%s is missing", definition.TableName, name)
		}
		if column.CollationName != "utf8mb4_bin" {
			t.Fatalf(
				"%s.%s collation = %q, want utf8mb4_bin for byte-exact identity",
				definition.TableName,
				name,
				column.CollationName,
			)
		}
	}
}

func TestConnectorRuntimeHandoffReceiptFreezesOnePreSendRequest(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	receipt := findSystemTableDefinition(
		t,
		defs,
		"_System_ConnectorRuntimeHandoffReceipt",
	)
	if receipt.RequiredReadPermission != "BLOCKED" ||
		receipt.RequiredWritePermission != "BLOCKED" {
		t.Fatalf(
			"Runtime handoff receipt permissions = %q/%q, want BLOCKED/BLOCKED",
			receipt.RequiredReadPermission,
			receipt.RequiredWritePermission,
		)
	}
	columns := columnDefinitionsByName(receipt)
	requireColumn(t, columns, "WorkId", "VARCHAR(255)", false, true, "")
	requireColumn(t, columns, "RouteVersion", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, columns, "MinimumCatalogRevision", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, columns, "ScopeDigest", "CHAR(64)", false, false, "")
	requireColumn(t, columns, "BindingViewDigest", "CHAR(71)", false, false, "")
	requireColumn(t, columns, "Payload", "MEDIUMBLOB", false, false, "")
	requireColumn(t, columns, "PayloadDigest", "CHAR(71)", false, false, "")
	requireIndex(t, receipt, "ux_connector_runtime_handoff_command", []string{
		"ScopeDigest",
		"CommandId",
	}, true, false)
	requireIndex(t, receipt, "ux_connector_runtime_handoff_idempotency", []string{
		"ScopeDigest",
		"IdempotencyKey",
	}, true, false)

	for _, forbidden := range []string{
		"State",
		"Attempt",
		"LeaseToken",
		"ResultPayload",
		"RuntimeResult",
	} {
		if _, ok := columns[forbidden]; ok {
			t.Fatalf(
				"Integration handoff receipt gained Runtime execution field %q",
				forbidden,
			)
		}
	}
}

func TestProcessDefinitionDeclaresUniqueDeveloperNameForDeployIdentity(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	processDefinition := findSystemTableDefinition(t, defs, "ProcessDefinition")
	developerName, ok := columnDefinitionsByName(processDefinition)["DeveloperName"]
	if !ok || !developerName.Unique {
		t.Fatalf("ProcessDefinition.DeveloperName = %+v, want schema-declared unique deploy identity", developerName)
	}
}

func TestPermissionSetDeclaresNullablePackOwnershipAndLookupIndex(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	permissionSet := findSystemTableDefinition(t, defs, "PermissionSet")
	requireColumn(t, columnDefinitionsByName(permissionSet), "OwnerPack", "VARCHAR(255)", true, false, "")
	if got := columnDefinitionsByName(permissionSet)["OwnerPack"].WriteProtectionPolicy; got != "trusted_only" {
		t.Fatalf("PermissionSet.OwnerPack WriteProtectionPolicy = %q, want trusted_only", got)
	}
	requireIndex(t, permissionSet, "idx_permission_set_owner_pack", []string{"OwnerPack"}, false, false)
}

func TestPermissionAuthoringSchemaKeepsRevisionOnRootAndOneCommandOutcome(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	permissionSet := findSystemTableDefinition(t, defs, "PermissionSet")
	revision := columnDefinitionsByName(permissionSet)["AuthoringRevision"]
	if revision.Type != "DECIMAL(20,0)" || revision.Nullable || revision.Default != "1" ||
		revision.WriteProtectionPolicy != "trusted_only" {
		t.Fatalf("PermissionSet.AuthoringRevision = %+v, want protected non-null root revision with warm default 1", revision)
	}

	command := findSystemTableDefinition(t, defs, "_System_PermissionAuthoringCommand")
	columns := columnDefinitionsByName(command)
	for _, required := range []string{
		"ScopeDigest", "CommandId", "RequestDigest", "TargetKind", "TargetId",
		"CanonicalPermissionSetId", "ExpectedAuthoringToken", "State", "ResultPayload",
		"AdmissionExpiresAt", "OutcomeExpiresAt", "OutcomeRetentionMicros",
	} {
		if _, ok := columns[required]; !ok {
			t.Fatalf("_System_PermissionAuthoringCommand.%s is missing", required)
		}
	}
	for _, forbidden := range []string{"Family", "Selector", "OperationIndex", "OperationCount"} {
		if _, ok := columns[forbidden]; ok {
			t.Fatalf("permission authoring journal gained partial-family field %s", forbidden)
		}
	}
	for _, forbidden := range []string{
		"ConvergenceState", "ConvergenceAuthoringRevision", "ConvergenceAttemptCount",
		"ConvergenceAvailableAt", "ConvergenceLeaseOwner", "ConvergenceLeaseExpiresAt",
		"ConvergenceLastError", "ConvergenceDeliveredAt",
	} {
		if _, ok := columns[forbidden]; ok {
			t.Fatalf("permission authoring command journal retained convergence lifetime field %s", forbidden)
		}
	}
	if admission := columns["AdmissionExpiresAt"]; admission.Nullable {
		t.Fatalf("_System_PermissionAuthoringCommand.AdmissionExpiresAt = %+v, want required nonterminal deadline", admission)
	}
	if outcome := columns["OutcomeExpiresAt"]; !outcome.Nullable {
		t.Fatalf("_System_PermissionAuthoringCommand.OutcomeExpiresAt = %+v, want terminal-only nullable deadline", outcome)
	}
	requireIndex(t, command, "ux_permission_authoring_command_scope", []string{"ScopeDigest", "CommandId"}, true, false)
	requireIndex(t, command, "idx_permission_authoring_command_admission_expiry", []string{"State", "AdmissionExpiresAt", "Id"}, false, false)
	requireIndex(t, command, "idx_permission_authoring_command_outcome_expiry", []string{"State", "OutcomeExpiresAt", "Id"}, false, false)

	convergence := findSystemTableDefinition(t, defs, "_System_PermissionAuthoringConvergence")
	convergenceColumns := columnDefinitionsByName(convergence)
	for _, required := range []string{
		"SourceCommandRowId", "CommandId", "CommandDigest", "CanonicalPermissionSetId",
		"AuthoringRevision", "State", "AttemptCount", "AvailableAt", "LeaseOwner",
		"LeaseExpiresAt", "LastError",
	} {
		if _, ok := convergenceColumns[required]; !ok {
			t.Fatalf("_System_PermissionAuthoringConvergence.%s is missing", required)
		}
	}
	if _, ok := convergenceColumns["DeliveredAt"]; ok {
		t.Fatal("permission authoring convergence queue retained delivered-history state")
	}
	requireIndex(t, convergence, "ux_permission_authoring_convergence_source", []string{"SourceCommandRowId"}, true, false)
	requireIndex(t, convergence, "idx_permission_authoring_convergence_delivery", []string{"State", "AvailableAt", "Id"}, false, false)

	setupAudit := findSystemTableDefinition(t, defs, "_System_SetupAuditTrail")
	requireColumn(t, columnDefinitionsByName(setupAudit), "Display", "MEDIUMTEXT", true, false, "")
}

func TestFieldWriteProtectionPolicyProtectsItsOwnAuthorityDeclaration(t *testing.T) {
	defs, err := ManagedTableDefinitions()
	if err != nil {
		t.Fatalf("ManagedTableDefinitions() error = %v", err)
	}

	field := findSystemTableDefinition(t, defs, "_System_Field")
	if got := columnDefinitionsByName(field)["WriteProtectionPolicy"].WriteProtectionPolicy; got != "authority_only" {
		t.Fatalf("_System_Field.WriteProtectionPolicy WriteProtectionPolicy = %q, want authority_only", got)
	}
}

func TestRecordTypeDeclaresReservedCatalogIdentityStorage(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	recordType := findSystemTableDefinition(t, defs, "_System_RecordType")
	columns := columnDefinitionsByName(recordType)
	for _, field := range []string{"CatalogNamespaceId", "CatalogLocalKey"} {
		requireColumn(t, columns, field, "VARBINARY(128)", true, false, "")
		if got := columns[field].WriteProtectionPolicy; got != "authority_only" {
			t.Fatalf("_System_RecordType.%s WriteProtectionPolicy = %q, want authority_only", field, got)
		}
	}
	requireIndex(t, recordType, "uq_record_type_catalog_key", []string{
		"ObjectId",
		"CatalogNamespaceId",
		"CatalogLocalKey",
	}, true, false)
}

func TestConnectedAppConsumerKeyDeclaresImmutableClientIdentity(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	connectedApp := findSystemTableDefinition(t, defs, "_System_ConnectedApp")
	consumerKey, ok := columnDefinitionsByName(connectedApp)["ConsumerKey"]
	if !ok {
		t.Fatal("_System_ConnectedApp.ConsumerKey is missing")
	}
	if got := consumerKey.WriteProtectionPolicy; got != "immutable_after_create" {
		t.Fatalf("_System_ConnectedApp.ConsumerKey WriteProtectionPolicy = %q, want immutable_after_create", got)
	}
}

func TestApprovalDefinitionChildrenDeclareRuntimeDeleteSemantics(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	for _, edge := range []struct {
		child string
		field string
	}{
		{child: "ProcessNode", field: "ProcessDefinitionId"},
		{child: "ProcessAction", field: "ProcessDefinitionId"},
		{child: "ProcessAction", field: "ProcessNodeId"},
	} {
		child := findSystemTableDefinition(t, defs, edge.child)
		field, ok := columnDefinitionsByName(child)[edge.field]
		if !ok || !strings.EqualFold(field.OnDelete, "CASCADE") {
			t.Errorf("%s.%s OnDelete = %q, want runtime Cascade", edge.child, edge.field, field.OnDelete)
		}
	}

	processInstance := findSystemTableDefinition(t, defs, "ProcessInstance")
	runtimeEdge, ok := columnDefinitionsByName(processInstance)["ProcessDefinitionId"]
	if !ok || strings.TrimSpace(runtimeEdge.OnDelete) != "" {
		t.Errorf("ProcessInstance.ProcessDefinitionId OnDelete = %q, want empty runtime rule so live execution data restricts deletion", runtimeEdge.OnDelete)
	}

	for _, edge := range []struct {
		child  string
		field  string
		parent string
	}{
		{child: "ProcessInstance", field: "ProcessDefinitionId", parent: "ProcessDefinition(Id)"},
		{child: "ProcessInstanceHistory", field: "ProcessNodeId", parent: "ProcessNode(Id)"},
	} {
		child := findSystemTableDefinition(t, defs, edge.child)
		var found *schema.ForeignKeyDefinition
		for i := range child.ForeignKeys {
			candidate := &child.ForeignKeys[i]
			if candidate.Column == edge.field && candidate.References == edge.parent {
				found = candidate
				break
			}
		}
		if found == nil {
			t.Errorf("%s.%s logical relationship is missing", edge.child, edge.field)
			continue
		}
		if !found.LogicalOnly {
			t.Errorf("%s.%s LogicalOnly = false, want request-bound Catalog/legacy authority without a physical FK", edge.child, edge.field)
		}
	}
}

func TestLongDescriptionSystemTablesDeclareShortObjectLabel(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	for _, def := range defs {
		description := strings.TrimSpace(def.Description)
		if len([]rune(description)) <= systemObjectLabelMaxRunes {
			continue
		}

		label := strings.TrimSpace(def.Label)
		if label == "" {
			t.Errorf("%s has a %d-rune description but no explicit label; managed object registration can overflow _System_Object.Label", def.TableName, len([]rune(description)))
			continue
		}
		if len([]rune(label)) > systemObjectLabelMaxRunes {
			t.Errorf("%s label has %d runes, want <= %d", def.TableName, len([]rune(label)), systemObjectLabelMaxRunes)
		}
	}
}

func TestManagedTableDefinitionsDeclareTemporalFields(t *testing.T) {
	defs, err := ManagedTableDefinitions()
	if err != nil {
		t.Fatalf("ManagedTableDefinitions(): %v", err)
	}
	sort.Slice(defs, func(i, j int) bool { return defs[i].TableName < defs[j].TableName })

	var inventory strings.Builder
	missingTemporal := 0
	for _, def := range defs {
		columns := columnDefinitionsByName(def)
		created, hasCreated := columns["CreatedDate"]
		modified, hasModified := columns["LastModifiedDate"]
		if !hasCreated {
			missingTemporal++
		}
		if !hasModified {
			missingTemporal++
		}
		fmt.Fprintf(&inventory,
			"%s CreatedDate={name=%q type=%q nullable=%t default=%q} LastModifiedDate={name=%q type=%q nullable=%t default=%q}\n",
			def.TableName, created.Name, created.Type, created.Nullable, created.Default,
			modified.Name, modified.Type, modified.Nullable, modified.Default)
	}
	summary := fmt.Sprintf("total_definitions=%d missing_temporal=%d", len(defs), missingTemporal)
	t.Logf("managed temporal fields inventory:\n%s%s", inventory.String(), summary)
	if missingTemporal != 0 {
		t.Fatalf("managed temporal fields inventory is incomplete: %s", summary)
	}
}

func TestOutboxRecordKeyPreservesRawIdentityAndIndexesFitTiDBLimit(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	outbox := findSystemTableDefinition(t, defs, "_System_OutboxEvent")
	outboxColumns := columnDefinitionsByName(outbox)

	recordKey := outboxColumns["RecordKey"]
	if recordKey.Type != "VARCHAR(512)" {
		t.Fatalf("RecordKey type = %q, want VARCHAR(512) to preserve the raw ObjectAPIName/Record.Id contract", recordKey.Type)
	}

	foundRecordFIFO := false
	for _, idx := range outbox.Indices {
		if idx.Name == "idx_outbox_record_fifo" {
			foundRecordFIFO = true
			if got, want := strings.Join(idx.Columns, ","), "RecordKey,CreatedDate"; got != want {
				t.Fatalf("idx_outbox_record_fifo columns = %s, want %s; adding Id exceeds TiDB's key length limit", got, want)
			}
		}
	}
	if !foundRecordFIFO {
		t.Fatal("idx_outbox_record_fifo not found")
	}

	recordEventOutbox := findSystemTableDefinition(t, defs, "_System_RecordEventOutbox")
	if recordEventOutbox.RequiredReadPermission != "BLOCKED" || recordEventOutbox.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("RecordEventOutbox permissions = read %q write %q, want BLOCKED/BLOCKED", recordEventOutbox.RequiredReadPermission, recordEventOutbox.RequiredWritePermission)
	}
	recordEventColumns := columnDefinitionsByName(recordEventOutbox)
	// Both legacy and Runtime record-event pull workers use LastModifiedDate as
	// a database-owned lease clock. Second precision makes a claim and its
	// immediate revalidation indistinguishable to TiDB: the refresh becomes a
	// no-op, RowsAffected reports zero, and the live attempt is falsely treated
	// as lost before any subscriber runs.
	requireColumn(t, outboxColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, recordEventColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, recordEventOutbox, "idx_record_event_outbox_runtime_scope", []string{
		"OrgId", "AppInstallationId", "RuntimeCapabilityId", "CreatedDate",
	}, false, false)
	requireColumn(t, recordEventColumns, "ExecutionScopeDigest", "CHAR(64)", true, false, "")
	requireIndex(t, recordEventOutbox, "idx_record_event_outbox_claim", []string{
		"ExecutionScopeDigest", "Status", "CreatedDate", "Id",
	}, false, false)

	for _, table := range []schema.TableDefinition{outbox, recordEventOutbox} {
		columns := columnDefinitionsByName(table)
		for _, idx := range table.Indices {
			totalBytes := 0
			for _, columnName := range idx.Columns {
				col, ok := columns[columnName]
				if !ok {
					t.Fatalf("%s references missing column %s", idx.Name, columnName)
				}
				n, err := maxTiDBIndexBytes(col.Type)
				if err != nil {
					t.Fatalf("%s.%s type %q: %v", table.TableName, columnName, col.Type, err)
				}
				totalBytes += n
			}
			if totalBytes > tidbMaxIndexKeyBytes {
				t.Fatalf("%s worst-case key bytes = %d, want <= %d", idx.Name, totalBytes, tidbMaxIndexKeyBytes)
			}
		}
	}
}

func TestIntegrationOutboundCommandOwnsDurableDependencyWaits(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	command := findSystemTableDefinition(t, defs, "_System_IntegrationOutboundCommand")
	columns := columnDefinitionsByName(command)
	requireColumn(t, columns, "DependencyKind", "VARCHAR(64)", true, false, "")
	requireColumn(t, columns, "DependencyKey", "VARCHAR(255)", true, false, "")
	requireColumn(t, columns, "DependencyReadinessKeys", "JSON", true, false, "")
	requireColumn(t, columns, "DependencyBlockedAt", "DATETIME(6)", true, false, "")
	if columns["DependencyKey"].CollationName != "utf8mb4_bin" {
		t.Fatalf(
			"DependencyKey collation = %q, want utf8mb4_bin",
			columns["DependencyKey"].CollationName,
		)
	}
	dependencyIndexColumns := []string{
		"ScopeDigest", "DependencyKind", "DependencyKey", "State", "Id",
	}
	requireIndex(t, command, "idx_integration_outbound_dependency", dependencyIndexColumns, false, false)
	requireIndex(t, command, "idx_integration_outbound_dependency_reconcile",
		[]string{"ScopeDigest", "CatalogRevision", "State", "Id"}, false, false)
	indexBytes := 0
	for _, name := range dependencyIndexColumns {
		columnBytes, err := maxTiDBIndexBytes(columns[name].Type)
		if err != nil {
			t.Fatalf("dependency index-width calculation for %s: %v", name, err)
		}
		indexBytes += columnBytes
	}
	if indexBytes > tidbMaxIndexKeyBytes {
		t.Fatalf(
			"idx_integration_outbound_dependency requires %d bytes, exceeds TiDB limit %d",
			indexBytes, tidbMaxIndexKeyBytes,
		)
	}
}

func TestConnectorWorkQueueSystemTablesDeclareSchemaAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	connectorEvent := findSystemTableDefinition(t, defs, "_System_ConnectorEvent")
	eventColumns := columnDefinitionsByName(connectorEvent)
	requireColumn(t, eventColumns, "ExecutionOwner", "VARCHAR(32)", false, false, "'event'")
	if !strings.Contains(connectorEvent.Description, "WorkItem is the sole current execution authority") {
		t.Fatalf("ConnectorEvent description does not declare WorkItem execution authority: %q", connectorEvent.Description)
	}
	if strings.Contains(strings.ToLower(connectorEvent.Description), "retire") {
		t.Fatalf("ConnectorEvent description retains the removed Event retirement contract: %q", connectorEvent.Description)
	}
	protocol := findSystemTableDefinition(t, defs, "_System_ConnectorProjectionProtocol")
	if protocol.TableType != "system_core" || protocol.Category != "integration" || protocol.RequiredReadPermission != "BLOCKED" || protocol.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("protocol must be an internal integration fact: %+v", protocol)
	}
	protocolColumns := columnDefinitionsByName(protocol)
	requireColumn(t, protocolColumns, "InstalledPackId", "VARCHAR(255)", false, true, "")
	requireColumn(t, protocolColumns, "ProtocolVersion", "INT", false, false, "")
	requireColumn(t, protocolColumns, "LegacyMaxOrdinal", "BIGINT", false, false, "")

	workItem := findSystemTableDefinition(t, defs, "_System_ConnectorWorkItem")
	if workItem.TableType != "system_core" || workItem.Category != "integration" {
		t.Fatalf("ConnectorWorkItem table type/category = %q/%q, want system_core/integration", workItem.TableType, workItem.Category)
	}
	if workItem.RequiredReadPermission != "BLOCKED" || workItem.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("ConnectorWorkItem permissions = read %q write %q, want BLOCKED/BLOCKED", workItem.RequiredReadPermission, workItem.RequiredWritePermission)
	}
	workColumns := columnDefinitionsByName(workItem)
	requireColumn(t, workColumns, "WorkId", "VARCHAR(255)", false, true, "")
	requireColumn(t, workColumns, "EventId", "VARCHAR(255)", false, false, "")
	requireColumn(t, workColumns, "InstalledPackId", "VARCHAR(255)", false, false, "")
	requireColumn(t, workColumns, "WorkType", "VARCHAR(64)", false, false, "")
	requireColumn(t, workColumns, "QueueClass", "VARCHAR(32)", false, false, "'default'")
	requireColumn(t, workColumns, "State", "VARCHAR(32)", false, false, "'pending'")
	requireColumn(t, workColumns, "RunAfter", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireColumn(t, workColumns, "Attempt", "INT", false, false, "0")
	requireColumn(t, workColumns, "OwnershipEpoch", "BIGINT", false, false, "0")
	requireColumn(t, workColumns, "FailureCount", "INT", false, false, "0")
	requireColumn(t, workColumns, "FailureClass", "VARCHAR(64)", true, false, "")
	requireColumn(t, workColumns, "FailureSummary", "TEXT", true, false, "")
	requireColumn(t, workColumns, "LastFailureAt", "DATETIME", true, false, "")
	requireColumn(t, workColumns, "TerminalAt", "DATETIME", true, false, "")
	requireColumn(t, workColumns, "ReplayedAt", "DATETIME", true, false, "")
	requireColumn(t, workColumns, "ReplayOfWorkId", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "LaneOrdinal", "BIGINT", false, false, "0")
	requireColumn(t, workColumns, "InputStartOrdinal", "BIGINT", true, false, "")
	requireColumn(t, workColumns, "LeaseOwner", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "LeaseToken", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "LeaseUntil", "DATETIME", true, false, "")
	requireColumn(t, workColumns, "SourceLaneKey", "VARCHAR(64)", true, false, "")
	requireColumn(t, workColumns, "TargetMutationLaneKey", "VARCHAR(64)", true, false, "")
	requireColumn(t, workColumns, "Generation", "BIGINT", false, false, "0")
	requireColumn(t, workColumns, "ProjectionKey", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "ProjectionActiveKey", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyKind", "VARCHAR(32)", true, false, "")
	requireColumn(t, workColumns, "DependencyObjectAPIName", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyRecordId", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyKeyHash", "VARCHAR(64)", true, false, "")
	requireColumn(t, workColumns, "DependencyConnectorOwnerKey", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyConnectorNamespace", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyConnectorSourceObject", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyDestinationSlotId", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyConnectorRemoteId", "VARCHAR(255)", true, false, "")
	requireColumn(t, workColumns, "DependencyBlockedAt", "DATETIME", true, false, "")
	requireColumn(t, workColumns, "DependencyProbeAfter", "DATETIME", true, false, "")
	requireColumn(t, workColumns, "CreatedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireColumn(t, workColumns, "LastModifiedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")

	requireIndex(t, workItem, "uniq_connector_work_item_event_type", []string{"EventId", "WorkType"}, true, false)
	requireIndex(t, workItem, "uniq_connector_work_item_replay_of", []string{"ReplayOfWorkId"}, true, false)
	requireIndex(t, workItem, "uniq_connector_work_item_projection_active", []string{"ProjectionActiveKey"}, true, false)
	requireIndex(t, workItem, "idx_connector_work_item_queue_claim", []string{"InstalledPackId", "QueueClass", "State", "RunAfter", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_source_lane", []string{"InstalledPackId", "SourceLaneKey", "State", "RunAfter", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_source_lane_gen", []string{"InstalledPackId", "Generation", "SourceLaneKey", "State", "LaneOrdinal", "RunAfter", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_target_lane", []string{"InstalledPackId", "TargetMutationLaneKey", "State", "RunAfter", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_lease_due", []string{"InstalledPackId", "State", "LeaseUntil", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_projection_history", []string{"ProjectionKey", "TerminalAt", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_projection_order", []string{"ProjectionKey", "LaneOrdinal", "EventId", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_dlq", []string{"InstalledPackId", "State", "TerminalAt", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_dependency", []string{"DependencyRecordId", "State", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_dependency_remote", []string{"DependencyConnectorRemoteId", "State", "WorkId"}, false, false)
	requireIndex(t, workItem, "idx_connector_work_item_dependency_probe", []string{"State", "DependencyProbeAfter", "WorkId"}, false, false)

	attempt := findSystemTableDefinition(t, defs, "_System_ConnectorWorkAttempt")
	if attempt.TableType != "system_core" || attempt.Category != "integration" || attempt.RequiredReadPermission != "BLOCKED" || attempt.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("ConnectorWorkAttempt authority = %#v", attempt)
	}
	attemptColumns := columnDefinitionsByName(attempt)
	requireColumn(t, attemptColumns, "WorkId", "VARCHAR(255)", false, false, "")
	requireColumn(t, attemptColumns, "Attempt", "INT", false, false, "")
	requireColumn(t, attemptColumns, "EventId", "VARCHAR(255)", false, false, "")
	requireColumn(t, attemptColumns, "FlowExecutionId", "VARCHAR(255)", false, false, "")
	requireColumn(t, attemptColumns, "OutcomeState", "VARCHAR(32)", false, false, "")
	requireColumn(t, attemptColumns, "PresentedErrorCode", "VARCHAR(128)", true, false, "")
	requireColumn(t, attemptColumns, "CauseCode", "VARCHAR(128)", true, false, "")
	requireColumn(t, attemptColumns, "CauseCategory", "VARCHAR(32)", true, false, "")
	requireColumn(t, attemptColumns, "CauseElementType", "VARCHAR(64)", true, false, "")
	requireColumn(t, attemptColumns, "CauseElementName", "VARCHAR(255)", true, false, "")
	requireColumn(t, attemptColumns, "Retriable", "TINYINT(1)", false, false, "0")
	requireColumn(t, attemptColumns, "CreatedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireColumn(t, attemptColumns, "LastModifiedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireIndex(t, attempt, "", []string{"WorkId", "Attempt"}, false, true)
	requireIndex(t, attempt, "idx_connector_work_attempt_flow", []string{"FlowExecutionId", "WorkId", "Attempt"}, false, false)
	requireIndex(t, attempt, "idx_connector_work_attempt_event", []string{"EventId", "Attempt"}, false, false)

	schedulerLease := findSystemTableDefinition(t, defs, "_System_ConnectorSchedulerLease")
	if schedulerLease.TableType != "system_core" || schedulerLease.Category != "integration" {
		t.Fatalf("ConnectorSchedulerLease table type/category = %q/%q, want system_core/integration", schedulerLease.TableType, schedulerLease.Category)
	}
	if schedulerLease.RequiredReadPermission != "BLOCKED" || schedulerLease.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("ConnectorSchedulerLease permissions = read %q write %q, want BLOCKED/BLOCKED", schedulerLease.RequiredReadPermission, schedulerLease.RequiredWritePermission)
	}
	leaseColumns := columnDefinitionsByName(schedulerLease)
	requireColumn(t, leaseColumns, "InstalledPackId", "VARCHAR(255)", false, false, "")
	requireColumn(t, leaseColumns, "Generation", "BIGINT", false, false, "0")
	requireColumn(t, leaseColumns, "OwnerId", "VARCHAR(255)", false, false, "")
	requireColumn(t, leaseColumns, "FencingToken", "VARCHAR(255)", false, false, "")
	requireColumn(t, leaseColumns, "LeaseUntil", "DATETIME", false, false, "")
	requireColumn(t, leaseColumns, "RenewedAt", "DATETIME", false, false, "")
	requireColumn(t, leaseColumns, "CreatedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireColumn(t, leaseColumns, "LastModifiedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")

	requireIndex(t, schedulerLease, "", []string{"InstalledPackId", "Generation"}, false, true)
	requireIndex(t, schedulerLease, "idx_connector_scheduler_lease_until", []string{"LeaseUntil", "InstalledPackId", "Generation"}, false, false)
}

func TestConnectorPushAttemptDeclaresDeferredAssociationClaimLane(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	attempt := findSystemTableDefinition(t, defs, "_System_ConnectorPushAttempt")
	columns := columnDefinitionsByName(attempt)
	requireColumn(t, columns, "DispatchLane", "VARCHAR(64)", false, false, "'configured'")
	requireColumn(t, columns, "RemoteIOStartedAt", "DATETIME", true, false, "")
	requireIndex(t, attempt, "idx_connector_push_attempt_lane_claim", []string{"Status", "DispatchLane", "LeaseExpiresAt", "CreatedDate"}, false, false)
}

func TestParentLaneSystemTableDeclaresDurableAdmissionShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	parentLane := findSystemTableDefinition(t, defs, "_System_ParentLane")
	if parentLane.TableType != "system_core" || parentLane.Category != "records" {
		t.Fatalf("ParentLane table type/category = %q/%q, want system_core/records", parentLane.TableType, parentLane.Category)
	}
	if parentLane.RequiredReadPermission != "BLOCKED" || parentLane.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("ParentLane permissions = read %q write %q, want BLOCKED/BLOCKED", parentLane.RequiredReadPermission, parentLane.RequiredWritePermission)
	}
	columns := columnDefinitionsByName(parentLane)
	requireColumn(t, columns, "KeyHash", "VARCHAR(64)", false, true, "")
	requireColumn(t, columns, "ParentObjectName", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "ParentRecordId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "OwnerToken", "VARCHAR(64)", false, false, "")
	requireColumn(t, columns, "LeaseExpiresAt", "DATETIME(6)", false, false, "")
	requireColumn(t, columns, "LastHeartbeatAt", "DATETIME(6)", false, false, "")
	requireColumn(t, columns, "CreatedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireColumn(t, columns, "LastModifiedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireIndex(t, parentLane, "idx_parent_lane_lease_expiry", []string{"LeaseExpiresAt", "KeyHash"}, false, false)
	requireIndex(t, parentLane, "idx_parent_lane_parent_key", []string{"ParentObjectName", "ParentRecordId"}, false, false)
}

func TestFileUploadOperationDeclaresFencedAttemptOwnership(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	operation := findSystemTableDefinition(t, defs, "_System_FileUploadOperation")
	if operation.RequiredReadPermission != "BLOCKED" || operation.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("FileUploadOperation permissions = read %q write %q, want BLOCKED/BLOCKED", operation.RequiredReadPermission, operation.RequiredWritePermission)
	}
	columns := columnDefinitionsByName(operation)
	requireColumn(t, columns, "OwnerToken", "VARCHAR(64)", false, false, "")
	requireColumn(t, columns, "ObjectKey", "VARCHAR(1024)", true, false, "")
	requireColumn(t, columns, "RetryCount", "INT", false, false, "0")
	requireColumn(t, columns, "RepairId", "VARCHAR(255)", true, false, "")
	requireColumn(t, columns, "RepairReason", "VARCHAR(1024)", true, false, "")
	requireColumn(t, columns, "RepairCount", "INT", false, false, "0")
	requireColumn(t, columns, "LastRepairAt", "DATETIME(6)", true, false, "")
	requireIndex(
		t,
		operation,
		"uk__System_FileUploadOperation_idempotency",
		[]string{"IdempotencyKey"},
		true,
		false,
	)
	requireIndex(
		t,
		operation,
		"idx__System_FileUploadOperation_lifecycle",
		[]string{"State", "ExpiresAt", "Id"},
		false,
		false,
	)
}

func TestManagedComponentCoordinationTablesDeclareBinaryAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	target := findSystemTableDefinition(t, defs, "_System_ManagedComponentTargetLock")
	requireExactColumnOrder(t, target, []string{"TargetDomain", "PhysicalKey", "CreatedDate", "LastModifiedDate"})
	targetColumns := columnDefinitionsByName(target)
	requireColumn(t, targetColumns, "TargetDomain", "VARBINARY(255)", false, false, "")
	requireColumn(t, targetColumns, "PhysicalKey", "VARBINARY(1024)", false, false, "")
	requireColumn(t, targetColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, targetColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, target, "", []string{"TargetDomain", "PhysicalKey"}, false, true)

	evidence := findSystemTableDefinition(t, defs, "_System_ManagedComponentOperationEvidence")
	requireExactColumnOrder(t, evidence, []string{
		"OperationId", "Attempt", "ComponentKind", "Scope", "PlanFingerprint", "ProjectionIdentityHash",
		"MutationDisposition", "SemanticMutationRows", "EffectReceiptHash", "CreatedDate", "LastModifiedDate",
	})
	evidenceColumns := columnDefinitionsByName(evidence)
	requireColumn(t, evidenceColumns, "OperationId", "VARBINARY(64)", false, false, "")
	requireColumn(t, evidenceColumns, "Attempt", "INT", false, false, "")
	requireColumn(t, evidenceColumns, "ComponentKind", "VARCHAR(255)", false, false, "")
	requireColumn(t, evidenceColumns, "Scope", "VARCHAR(16)", false, false, "")
	requireColumn(t, evidenceColumns, "PlanFingerprint", "BINARY(32)", false, false, "")
	requireColumn(t, evidenceColumns, "ProjectionIdentityHash", "BINARY(32)", false, false, "")
	requireColumn(t, evidenceColumns, "MutationDisposition", "VARCHAR(32)", false, false, "")
	requireColumn(t, evidenceColumns, "SemanticMutationRows", "INT", false, false, "")
	requireColumn(t, evidenceColumns, "EffectReceiptHash", "BINARY(32)", true, false, "")
	requireColumn(t, evidenceColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, evidenceColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, evidence, "", []string{"OperationId", "Attempt"}, false, true)
	for _, def := range []schema.TableDefinition{target, evidence} {
		if def.RequiredReadPermission != "BLOCKED" || def.RequiredWritePermission != "BLOCKED" || def.TableType != "system_core" {
			t.Fatalf("%s access/type = %q/%q/%q", def.TableName, def.RequiredReadPermission, def.RequiredWritePermission, def.TableType)
		}
	}
}

func TestConnectorRemoteIdentityLaneDeclaresRecoveryShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	lane := findSystemTableDefinition(t, defs, "_System_ConnectorRemoteIdentityLane")
	if lane.TableType != "system_core" || lane.Category != "integration" {
		t.Fatalf("ConnectorRemoteIdentityLane table type/category = %q/%q, want system_core/integration", lane.TableType, lane.Category)
	}
	if lane.RequiredReadPermission != "BLOCKED" || lane.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("ConnectorRemoteIdentityLane permissions = read %q write %q, want BLOCKED/BLOCKED", lane.RequiredReadPermission, lane.RequiredWritePermission)
	}
	requireExactColumnOrder(t, lane, []string{"RemoteLocatorHash", "OwnerDeleted", "CreatedDate", "LastModifiedDate"})
	columns := columnDefinitionsByName(lane)
	requireColumn(t, columns, "RemoteLocatorHash", "VARCHAR(64)", false, true, "")
	requireColumn(t, columns, "OwnerDeleted", "BOOLEAN", false, false, "FALSE")
	requireColumn(t, columns, "CreatedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireColumn(t, columns, "LastModifiedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")

	remoteLink := findSystemTableDefinition(t, defs, "_System_ConnectorRemoteLink")
	requireIndex(t, remoteLink, "idx_connector_remote_link_locator", []string{"RemoteLocatorHash"}, false, false)
}

func TestConnectorPushLinkMappingRevalidationKeepsOneRuntimeSnapshot(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	remoteLink := findSystemTableDefinition(t, defs, "_System_ConnectorRemoteLink")
	linkColumns := columnDefinitionsByName(remoteLink)
	requireColumn(t, linkColumns, "MappingFingerprint", "VARCHAR(255)", false, false, "")
	requireColumn(t, linkColumns, "BodyFingerprint", "VARCHAR(255)", false, false, "")
	requireColumn(t, linkColumns, "LastMappingRevalidationOperationId", "VARCHAR(255)", true, false, "")
	for _, forbidden := range []string{
		"EffectiveMappingFingerprintVersion",
		"EffectiveMappingFingerprint",
		"EffectiveBodyFingerprintVersion",
		"EffectiveBodyFingerprint",
		"EffectiveMappingPackInstallId",
		"MappingRevalidationOperationId",
	} {
		if _, ok := linkColumns[forbidden]; ok {
			t.Fatalf("remote link must not declare second-snapshot column %q", forbidden)
		}
	}

	receipt := findSystemTableDefinition(t, defs, "_System_ConnectorPushLinkRevalidationReceipt")
	if receipt.TableType != "system_core" || receipt.Category != "integration" {
		t.Fatalf("mapping revalidation receipt table type/category = %q/%q, want system_core/integration", receipt.TableType, receipt.Category)
	}
	if receipt.RequiredReadPermission != "BLOCKED" || receipt.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("mapping revalidation receipt permissions = read %q write %q, want BLOCKED/BLOCKED", receipt.RequiredReadPermission, receipt.RequiredWritePermission)
	}
	receiptColumns := columnDefinitionsByName(receipt)
	requireColumn(t, receiptColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, receiptColumns, "OperationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "LinkKey", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "ManifestDigest", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "ApprovalDigest", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "FromMappingFingerprintVersion", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "FromMappingFingerprint", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "FromBodyFingerprintVersion", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "FromBodyFingerprint", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "FromPackInstallId", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "ToMappingFingerprintVersion", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "ToMappingFingerprint", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "ToBodyFingerprintVersion", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "ToBodyFingerprint", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "ToPackInstallId", "VARCHAR(255)", false, false, "")
	requireColumn(t, receiptColumns, "FromLinkSnapshotDigest", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "ToLinkSnapshotDigest", "VARCHAR(64)", false, false, "")
	requireColumn(t, receiptColumns, "Reason", "TEXT", false, false, "")
	requireColumn(t, receiptColumns, "CreatedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireColumn(t, receiptColumns, "LastModifiedDate", "DATETIME", false, false, "CURRENT_TIMESTAMP")
	requireIndex(t, receipt, "uniq_connector_push_link_revalidation_member", []string{"OperationId", "LinkKey"}, true, false)
	requireIndex(t, receipt, "idx_connector_push_link_revalidation_link", []string{"LinkKey", "CreatedDate"}, false, false)
	requireIndex(t, receipt, "idx_connector_push_link_revalidation_manifest", []string{"ManifestDigest", "LinkKey"}, false, false)
}

func TestConnectedDeliverySystemTablesDeclareProtectedNeutralSchema(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	tableNames := []string{
		"_System_ConnectedPrincipal",
		"_System_ConnectedAuthorizationRequest",
		"_System_ConnectedPrincipalToken",
		"_System_ConnectedDeliveryJob",
		"_System_ConnectedDeliveryPayload",
	}
	for _, tableName := range tableNames {
		def := findSystemTableDefinition(t, defs, tableName)
		if def.TableType != "system_core" || def.Category != "integration" {
			t.Fatalf("%s table type/category = %q/%q, want system_core/integration", tableName, def.TableType, def.Category)
		}
		if def.RequiredReadPermission != "BLOCKED" || def.RequiredWritePermission != "BLOCKED" {
			t.Fatalf("%s permissions = read %q write %q, want BLOCKED/BLOCKED", tableName, def.RequiredReadPermission, def.RequiredWritePermission)
		}
	}

	forbidden := []string{"gmail", "graph", "contact", "lead", "task", "emailmessage"}
	for _, tableName := range tableNames {
		lower := strings.ToLower(tableName)
		for _, word := range forbidden {
			if strings.Contains(lower, word) {
				t.Fatalf("mechanism table name %q contains product vocabulary %q", tableName, word)
			}
		}
	}
}

func TestConnectedPrincipalAndAuthorizationSchema(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	principal := findSystemTableDefinition(t, defs, "_System_ConnectedPrincipal")
	principalColumns := columnDefinitionsByName(principal)
	requireColumn(t, principalColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, principalColumns, "UserId", "VARCHAR(255)", false, false, "")
	requireColumn(t, principalColumns, "ConnectorKey", "VARCHAR(128)", false, false, "")
	requireColumn(t, principalColumns, "Generation", "BIGINT", false, false, "1")
	requireColumn(t, principalColumns, "Issuer", "VARCHAR(255)", false, false, "")
	requireColumn(t, principalColumns, "Subject", "VARCHAR(255)", false, false, "")
	requireColumn(t, principalColumns, "PrincipalAddress", "VARCHAR(320)", false, false, "")
	requireColumn(t, principalColumns, "State", "VARCHAR(32)", false, false, "'connected'")
	requireColumn(t, principalColumns, "ConnectedAt", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, principalColumns, "DisconnectedAt", "DATETIME(6)", true, false, "")
	requireIndex(t, principal, "ux_connected_principal_user_connector", []string{"UserId", "ConnectorKey"}, true, false)
	requireIndex(t, principal, "ux_connected_principal_remote_identity", []string{"ConnectorKey", "Issuer", "Subject"}, true, false)
	requireForeignKey(t, principal, "UserId", "User(Id)", "CASCADE")

	authorization := findSystemTableDefinition(t, defs, "_System_ConnectedAuthorizationRequest")
	authorizationColumns := columnDefinitionsByName(authorization)
	requireColumn(t, authorizationColumns, "StateDigest", "CHAR(64)", false, false, "")
	requireColumn(t, authorizationColumns, "UserId", "VARCHAR(255)", false, false, "")
	requireColumn(t, authorizationColumns, "ExpectedGeneration", "BIGINT", false, false, "1")
	requireColumn(t, authorizationColumns, "CodeVerifierCiphertext", "TEXT", false, false, "")
	requireColumn(t, authorizationColumns, "ExpiresAt", "DATETIME(6)", false, false, "")
	requireColumn(t, authorizationColumns, "ConsumedAt", "DATETIME(6)", true, false, "")
	requireIndex(t, authorization, "ux_connected_authorization_state_digest", []string{"StateDigest"}, true, false)
	requireIndex(t, authorization, "idx_connected_authorization_expiry", []string{"ExpiresAt"}, false, false)
	requireForeignKey(t, authorization, "UserId", "User(Id)", "CASCADE")

	token := findSystemTableDefinition(t, defs, "_System_ConnectedPrincipalToken")
	tokenColumns := columnDefinitionsByName(token)
	requireColumn(t, tokenColumns, "ConnectedPrincipalId", "VARCHAR(255)", false, true, "")
	requireColumn(t, tokenColumns, "PrincipalGeneration", "BIGINT", false, false, "1")
	requireColumn(t, tokenColumns, "AccessTokenCiphertext", "TEXT", false, false, "")
	requireColumn(t, tokenColumns, "RefreshTokenCiphertext", "TEXT", true, false, "")
	requireColumn(t, tokenColumns, "ExpiresAt", "DATETIME(6)", true, false, "")
	requireColumn(t, tokenColumns, "EncryptionKeyId", "VARCHAR(255)", false, false, "")
	requireColumn(t, tokenColumns, "TokenVersion", "BIGINT", false, false, "1")
	requireColumn(t, tokenColumns, "RefreshOwnershipEpoch", "BIGINT", false, false, "0")
	requireColumn(t, tokenColumns, "RefreshOwnerId", "VARCHAR(255)", true, false, "")
	requireColumn(t, tokenColumns, "RefreshLeaseExpiresAt", "DATETIME(6)", true, false, "")
	requireColumn(t, tokenColumns, "RefreshDispatchStartedAt", "DATETIME(6)", true, false, "")
	if tokenColumns["RefreshOwnerId"].CollationName != "utf8mb4_bin" {
		t.Fatalf("RefreshOwnerId collation = %q, want utf8mb4_bin", tokenColumns["RefreshOwnerId"].CollationName)
	}
	requireForeignKey(t, token, "ConnectedPrincipalId", "_System_ConnectedPrincipal(Id)", "CASCADE")
}

func TestConnectedOAuthEffectCatalogShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	authorization := findSystemTableDefinition(t, defs, "_System_ConnectedAuthorizationRequest")
	requireExactColumnOrder(t, authorization, []string{
		"Id", "StateDigest", "UserId", "ConnectorKey", "ExpectedGeneration", "RedirectUri",
		"CodeVerifierCiphertext", "RequestedScopes", "ExpiresAt", "ConsumedAt", "OAuthEffectState",
		"OAuthEffectVersion", "OAuthActionOwnerId", "OAuthActionEpoch", "OAuthActionLeaseExpiresAt",
		"OAuthExchangeDispatchedAt", "PendingTokenCiphertext", "CreatedDate", "LastModifiedDate",
	})
	requireExactOAuthEffectColumns(t, authorization, map[string]schema.ColumnDefinition{
		"OAuthEffectState":          {Name: "OAuthEffectState", Label: "OAuth Effect State", Type: "VARCHAR(32)", Default: "'none'"},
		"OAuthEffectVersion":        {Name: "OAuthEffectVersion", Label: "OAuth Effect Version", Type: "BIGINT", Default: "0"},
		"OAuthActionOwnerId":        {Name: "OAuthActionOwnerId", Label: "OAuth Action Owner Id", Type: "VARCHAR(255)", Nullable: true, CollationName: "utf8mb4_bin"},
		"OAuthActionEpoch":          {Name: "OAuthActionEpoch", Label: "OAuth Action Epoch", Type: "BIGINT", Default: "0"},
		"OAuthActionLeaseExpiresAt": {Name: "OAuthActionLeaseExpiresAt", Label: "OAuth Action Lease Expires At", Type: "DATETIME(6)", Nullable: true},
		"OAuthExchangeDispatchedAt": {Name: "OAuthExchangeDispatchedAt", Label: "OAuth Exchange Dispatched At", Type: "DATETIME(6)", Nullable: true},
		"PendingTokenCiphertext":    {Name: "PendingTokenCiphertext", Label: "Pending Token Ciphertext", Type: "TEXT", Nullable: true},
	})
	requireExactTableWithoutColumns(t, authorization, schema.TableDefinition{
		TableName: "_System_ConnectedAuthorizationRequest", TableType: "system_core", Category: "integration",
		Label: "Connected Authorization Request", Description: "Short-lived per-user authorization state and encrypted proof-key verifier metadata",
		RequiredReadPermission: "BLOCKED", RequiredWritePermission: "BLOCKED",
		Indices: []schema.IndexDefinition{
			{Name: "ux_connected_authorization_state_digest", Columns: []string{"StateDigest"}, Unique: true},
			{Name: "idx_connected_authorization_user_connector", Columns: []string{"UserId", "ConnectorKey", "ExpiresAt"}},
			{Name: "idx_connected_authorization_expiry", Columns: []string{"ExpiresAt"}},
		},
		ForeignKeys: []schema.ForeignKeyDefinition{{Column: "UserId", References: "User(Id)", OnDelete: "CASCADE"}},
	})

	token := findSystemTableDefinition(t, defs, "_System_ConnectedPrincipalToken")
	requireExactColumnOrder(t, token, []string{
		"ConnectedPrincipalId", "PrincipalGeneration", "AccessTokenCiphertext", "RefreshTokenCiphertext",
		"TokenType", "GrantedScopes", "ExpiresAt", "EncryptionKeyId", "TokenVersion", "RefreshOwnershipEpoch",
		"RefreshOwnerId", "RefreshLeaseExpiresAt", "RefreshDispatchStartedAt", "OAuthEffectState",
		"OAuthEffectVersion", "PendingTokenCiphertext", "CreatedDate", "LastModifiedDate",
	})
	requireExactOAuthEffectColumns(t, token, map[string]schema.ColumnDefinition{
		"OAuthEffectState":       {Name: "OAuthEffectState", Label: "OAuth Effect State", Type: "VARCHAR(32)", Default: "'none'"},
		"OAuthEffectVersion":     {Name: "OAuthEffectVersion", Label: "OAuth Effect Version", Type: "BIGINT", Default: "0"},
		"PendingTokenCiphertext": {Name: "PendingTokenCiphertext", Label: "Pending Token Ciphertext", Type: "TEXT", Nullable: true},
	})
	requireExactTableWithoutColumns(t, token, schema.TableDefinition{
		TableName: "_System_ConnectedPrincipalToken", TableType: "system_core", Category: "integration",
		Label: "Connected Principal Token", Description: "Envelope-encrypted token material and expiry metadata for one connected principal",
		RequiredReadPermission: "BLOCKED", RequiredWritePermission: "BLOCKED",
		Indices:     []schema.IndexDefinition{{Name: "idx_connected_principal_token_expiry", Columns: []string{"ExpiresAt"}}},
		ForeignKeys: []schema.ForeignKeyDefinition{{Column: "ConnectedPrincipalId", References: "_System_ConnectedPrincipal(Id)", OnDelete: "CASCADE"}},
	})
}

func TestOAuthDynamicClientRegistrationBlocksGenericCRUDWrites(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	registration := findSystemTableDefinition(t, defs, "_System_OAuthDynamicClientRegistration")
	if registration.RequiredReadPermission != "ViewSetup" {
		t.Fatalf("dynamic client registration read permission = %q, want ViewSetup", registration.RequiredReadPermission)
	}
	if registration.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("dynamic client registration write permission = %q, want BLOCKED", registration.RequiredWritePermission)
	}
}

func TestConnectedOAuthEffectConstantsMatchExactAPINames(t *testing.T) {
	tests := []struct {
		name, field, column, apiName string
	}{
		{"authorization OAuthEffectState", sharedconstants.FieldSysConnectedAuthorizationRequest_OAuthEffectState, sharedconstants.ColSysConnectedAuthorizationRequest_OAuthEffectState, "OAuthEffectState"},
		{"authorization OAuthEffectVersion", sharedconstants.FieldSysConnectedAuthorizationRequest_OAuthEffectVersion, sharedconstants.ColSysConnectedAuthorizationRequest_OAuthEffectVersion, "OAuthEffectVersion"},
		{"authorization OAuthActionOwnerId", sharedconstants.FieldSysConnectedAuthorizationRequest_OAuthActionOwnerId, sharedconstants.ColSysConnectedAuthorizationRequest_OAuthActionOwnerId, "OAuthActionOwnerId"},
		{"authorization OAuthActionEpoch", sharedconstants.FieldSysConnectedAuthorizationRequest_OAuthActionEpoch, sharedconstants.ColSysConnectedAuthorizationRequest_OAuthActionEpoch, "OAuthActionEpoch"},
		{"authorization OAuthActionLeaseExpiresAt", sharedconstants.FieldSysConnectedAuthorizationRequest_OAuthActionLeaseExpiresAt, sharedconstants.ColSysConnectedAuthorizationRequest_OAuthActionLeaseExpiresAt, "OAuthActionLeaseExpiresAt"},
		{"authorization OAuthExchangeDispatchedAt", sharedconstants.FieldSysConnectedAuthorizationRequest_OAuthExchangeDispatchedAt, sharedconstants.ColSysConnectedAuthorizationRequest_OAuthExchangeDispatchedAt, "OAuthExchangeDispatchedAt"},
		{"authorization PendingTokenCiphertext", sharedconstants.FieldSysConnectedAuthorizationRequest_PendingTokenCiphertext, sharedconstants.ColSysConnectedAuthorizationRequest_PendingTokenCiphertext, "PendingTokenCiphertext"},
		{"token OAuthEffectState", sharedconstants.FieldSysConnectedPrincipalToken_OAuthEffectState, sharedconstants.ColSysConnectedPrincipalToken_OAuthEffectState, "OAuthEffectState"},
		{"token OAuthEffectVersion", sharedconstants.FieldSysConnectedPrincipalToken_OAuthEffectVersion, sharedconstants.ColSysConnectedPrincipalToken_OAuthEffectVersion, "OAuthEffectVersion"},
		{"token PendingTokenCiphertext", sharedconstants.FieldSysConnectedPrincipalToken_PendingTokenCiphertext, sharedconstants.ColSysConnectedPrincipalToken_PendingTokenCiphertext, "PendingTokenCiphertext"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			if tt.field != tt.apiName || tt.column != tt.apiName {
				t.Fatalf("field/column constants = %q/%q, want %q", tt.field, tt.column, tt.apiName)
			}
		})
	}
}

func TestConnectedDeliveryProtocolContractColumns(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	tables := map[string]map[string]schema.ColumnDefinition{
		"principal":     columnDefinitionsByName(findSystemTableDefinition(t, defs, "_System_ConnectedPrincipal")),
		"authorization": columnDefinitionsByName(findSystemTableDefinition(t, defs, "_System_ConnectedAuthorizationRequest")),
		"token":         columnDefinitionsByName(findSystemTableDefinition(t, defs, "_System_ConnectedPrincipalToken")),
		"job":           columnDefinitionsByName(findSystemTableDefinition(t, defs, "_System_ConnectedDeliveryJob")),
	}
	tests := []struct {
		name    string
		table   string
		columns []string
	}{
		{name: "connection generation", table: "principal", columns: []string{"Id", "Generation"}},
		{name: "authorization generation fence", table: "authorization", columns: []string{"StateDigest", "ExpectedGeneration"}},
		{name: "refresh ownership before provider call", table: "token", columns: []string{"ConnectedPrincipalId", "PrincipalGeneration", "TokenVersion", "RefreshOwnershipEpoch", "RefreshOwnerId", "RefreshLeaseExpiresAt"}},
		{name: "dispatch-started fail closed marker", table: "token", columns: []string{"RefreshDispatchStartedAt"}},
		{name: "delivery generation fence", table: "job", columns: []string{"ConnectedPrincipalId", "PrincipalGeneration"}},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			for _, column := range tt.columns {
				if _, ok := tables[tt.table][column]; !ok {
					t.Fatalf("protocol contract %q lacks durable %s.%s", tt.name, tt.table, column)
				}
			}
		})
	}
}

func TestConnectedDeliveryJobDeclaresFencedOneShotShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	job := findSystemTableDefinition(t, defs, "_System_ConnectedDeliveryJob")
	columns := columnDefinitionsByName(job)
	requireColumn(t, columns, "JobId", "VARCHAR(255)", false, true, "")
	requireColumn(t, columns, "ConnectedPrincipalId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "PrincipalGeneration", "BIGINT", false, false, "1")
	requireColumn(t, columns, "OriginatingUserId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "State", "VARCHAR(32)", false, false, "'queued'")
	requireColumn(t, columns, "PayloadSha256", "CHAR(64)", false, false, "")
	requireColumn(t, columns, "PayloadEncodedSize", "BIGINT", false, false, "")
	requireColumn(t, columns, "CompletionSnapshot", "JSON", false, false, "")
	requireColumn(t, columns, "ClaimToken", "VARCHAR(255)", true, false, "")
	requireColumn(t, columns, "LeaseExpiresAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "Fence", "BIGINT", false, false, "0")
	requireColumn(t, columns, "DispatchStartedAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "AcceptedAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "ResultEvidence", "JSON", true, false, "")
	requireColumn(t, columns, "LocalAttemptCount", "INT", false, false, "0")
	requireColumn(t, columns, "NextLocalAttemptAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "CompletedAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, columns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, job, "idx_connected_delivery_job_claim", []string{"State", "LeaseExpiresAt", "JobId"}, false, false)
	requireIndex(t, job, "idx_connected_delivery_job_local_retry", []string{"State", "NextLocalAttemptAt", "JobId"}, false, false)
	requireForeignKey(t, job, "ConnectedPrincipalId", "_System_ConnectedPrincipal(Id)", "RESTRICT")
	requireForeignKey(t, job, "OriginatingUserId", "User(Id)", "RESTRICT")
}

func TestConnectedDeliveryPayloadDeclaresCanonicalCatalogShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	payload := findSystemTableDefinition(t, defs, "_System_ConnectedDeliveryPayload")
	columns := columnDefinitionsByName(payload)
	requireColumn(t, columns, "JobId", "VARCHAR(255)", false, true, "")
	requireColumn(t, columns, "PayloadBase64", "MEDIUMTEXT", false, false, "")
	requireColumn(t, columns, "PayloadSha256", "CHAR(64)", false, false, "")
	requireColumn(t, columns, "EncodedSize", "BIGINT", false, false, "")
	requireColumn(t, columns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, columns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireIndex(t, payload, "idx_connected_delivery_payload_created", []string{"CreatedDate", "JobId"}, false, false)

	if len(payload.ForeignKeys) != 0 {
		t.Fatalf("payload foreign keys = %v, want none so payload insert can precede job insert and orphan sweep can delete independently", payload.ForeignKeys)
	}
}

func TestRuntimeSecuritySystemTablesDeclareProtectedOwnerShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	for _, tableName := range []string{
		"_System_RuntimeSecurityState", "_System_RuntimeSecurityBinding",
		"_System_RuntimeSecurityLease", "_System_RuntimeDelegationRevocation",
	} {
		def := findSystemTableDefinition(t, defs, tableName)
		if def.TableType != "system_core" || def.Category != "security" ||
			def.RequiredReadPermission != "BLOCKED" || def.RequiredWritePermission != "BLOCKED" {
			t.Fatalf("%s owner shape = %q/%q read=%q write=%q", tableName, def.TableType, def.Category, def.RequiredReadPermission, def.RequiredWritePermission)
		}
	}

	state := findSystemTableDefinition(t, defs, "_System_RuntimeSecurityState")
	stateColumns := columnDefinitionsByName(state)
	requireColumn(t, stateColumns, "OrgId", "VARCHAR(255)", false, true, "")
	requireColumn(t, stateColumns, "RuntimeAudience", "VARCHAR(255)", false, false, "")
	requireColumn(t, stateColumns, "SecurityEpoch", "DECIMAL(20,0)", false, false, "")

	binding := findSystemTableDefinition(t, defs, "_System_RuntimeSecurityBinding")
	bindingColumns := columnDefinitionsByName(binding)
	requireColumn(t, bindingColumns, "CallerWorkloadId", "VARCHAR(1024)", false, false, "")
	requireColumn(t, bindingColumns, "CallerWorkloadDigest", "CHAR(64)", false, false, "")
	if _, exists := bindingColumns["VerifiedCallerHandle"]; exists {
		t.Fatal("Catalog-owned Runtime binding must not store operator-owned opaque caller handles")
	}

	session := findSystemTableDefinition(t, defs, "_System_Session")
	sessionColumns := columnDefinitionsByName(session)
	requireColumn(t, sessionColumns, "Token", "VARBINARY(3000)", false, false, "")
	token := sessionColumns["Token"]
	if !token.Unique || token.ReadProtectionPolicy != "blocked" {
		t.Fatalf("Token protection = unique %v read %q, want unique/blocked",
			token.Unique, token.ReadProtectionPolicy)
	}
	tokenIndexBytes, err := maxTiDBIndexBytes(token.Type)
	if err != nil || tokenIndexBytes > tidbMaxIndexKeyBytes {
		t.Fatalf("Token unique index width = %d bytes, err=%v, want <= %d", tokenIndexBytes, err, tidbMaxIndexKeyBytes)
	}
	requireColumn(t, sessionColumns, "CurrentBearerToken", "VARBINARY(3000)", true, false, "")
	currentBearerToken := sessionColumns["CurrentBearerToken"]
	if currentBearerToken.Unique || currentBearerToken.ReadProtectionPolicy != "blocked" {
		t.Fatalf("CurrentBearerToken protection = unique %v read %q, want non-unique/blocked",
			currentBearerToken.Unique, currentBearerToken.ReadProtectionPolicy)
	}
	requireColumn(t, sessionColumns, "AuthEpoch", "DECIMAL(20,0)", false, false, "1")
	requireColumn(t, sessionColumns, "ControlProofHash", "CHAR(64)", true, false, "")
	controlProofHash := sessionColumns["ControlProofHash"]
	if controlProofHash.CollationName != "utf8mb4_bin" || controlProofHash.ReadProtectionPolicy != "blocked" {
		t.Fatalf("ControlProofHash protection = collation %q read %q, want utf8mb4_bin/blocked",
			controlProofHash.CollationName, controlProofHash.ReadProtectionPolicy)
	}
}

func TestRuntimePlatformCommandDeclaresOpaquePayloadShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	command := findSystemTableDefinition(t, defs, "_System_RuntimePlatformCommand")
	if command.TableType != "system_core" || command.Category != "runtime" ||
		command.RequiredReadPermission != "BLOCKED" || command.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("runtime command owner shape = %q/%q read=%q write=%q",
			command.TableType, command.Category, command.RequiredReadPermission, command.RequiredWritePermission)
	}
	columns := columnDefinitionsByName(command)
	requireColumn(t, columns, "OrgId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "AppInstallationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "CallerWorkloadId", "VARCHAR(1024)", false, false, "")
	requireColumn(t, columns, "RuntimeCapabilityId", "VARCHAR(255)", false, false, "")
	requireColumn(t, columns, "MetadataBinding", "MEDIUMTEXT", false, false, "")
	requireColumn(t, columns, "ExecutionScopeDigest", "CHAR(64)", false, false, "")
	requireColumn(t, columns, "SerializationKeyDigest", "CHAR(64)", true, false, "")
	requireColumn(t, columns, "PayloadFormatVersion", "BIGINT", false, false, "")
	requireColumn(t, columns, "Payload", "MEDIUMBLOB", false, false, "")
	requireColumn(t, columns, "Receipts", "JSON", false, false, "")
	// Preserve the already-published physical type across the stopped B2
	// cutover. Runtime base64-encodes the opaque bytes; changing this column to
	// MEDIUMBLOB would require an unrelated unsafe type migration.
	requireColumn(t, columns, "ResultPayload", "MEDIUMTEXT", true, false, "")
	requireColumn(t, columns, "PublicationIdentity", "CHAR(71)", true, false, "")
	requireColumn(t, columns, "PublishAttempt", "INT", false, false, "0")
	requireColumn(t, columns, "RecoveryState", "VARCHAR(16)", true, false, "")
	requireColumn(t, columns, "Attempt", "INT", false, false, "0")
	requireColumn(t, columns, "AvailableAt", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, columns, "LeaseToken", "VARCHAR(64)", true, false, "")
	requireColumn(t, columns, "LeaseExpiresAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "OwnershipEpoch", "DECIMAL(20,0)", true, false, "")
	requireColumn(t, columns, "LastError", "LONGTEXT", true, false, "")
	requireColumn(t, columns, "RepairId", "VARCHAR(255)", true, false, "")
	requireColumn(t, columns, "RepairReason", "VARCHAR(1024)", true, false, "")
	requireColumn(t, columns, "RepairCount", "INT", false, false, "0")
	requireColumn(t, columns, "LastRepairAt", "DATETIME(6)", true, false, "")
	requireIndex(t, command, "ux_runtime_platform_command_scope",
		[]string{"ScopeDigest", "IdempotencyKey"}, true, false)
	requireIndex(t, command, "idx_runtime_platform_command_partition",
		[]string{"ExecutionScopeDigest", "PartitionKey", "State", "LeaseExpiresAt", "Id"}, false, false)
	requireIndex(t, command, "idx_runtime_platform_command_serialization",
		[]string{"ExecutionScopeDigest", "SerializationKeyDigest", "State", "LeaseExpiresAt", "Id"}, false, false)
	requireIndex(t, command, "idx_runtime_platform_command_id",
		[]string{"ScopeDigest", "CommandId"}, true, false)
	requireIndex(t, command, "idx_runtime_platform_command_claim",
		[]string{"ExecutionScopeDigest", "State", "AvailableAt", "LeaseExpiresAt", "Id"}, false, false)
}

func TestScheduledRunReservationDeclaresFencedWorkerAuthorityShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	reservation := findSystemTableDefinition(t, defs, "_System_ScheduledRunReservation")
	columns := columnDefinitionsByName(reservation)
	requireColumn(t, columns, "OrgId", "VARCHAR(191)", true, false, "")
	requireColumn(t, columns, "AppInstallationId", "VARCHAR(191)", true, false, "")
	requireColumn(t, columns, "RuntimeCapabilityId", "VARCHAR(191)", true, false, "")
	requireColumn(t, columns, "OccurrenceIdentityDigest", "CHAR(64)", true, false, "")
	requireColumn(t, columns, "BatchIdentityDigest", "CHAR(64)", true, false, "")
	requireColumn(t, columns, "Attempt", "INT", false, false, "1")
	requireColumn(t, columns, "LeaseToken", "VARCHAR(64)", true, false, "")
	requireColumn(t, columns, "LeaseExpiresAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "EntryElement", "VARCHAR(191)", true, false, "")
	requireColumn(t, columns, "DueAt", "DATETIME(6)", true, false, "")
	requireColumn(t, columns, "OwnershipEpoch", "DECIMAL(20,0)", true, false, "")
	requireColumn(t, columns, "MetadataBinding", "MEDIUMTEXT", true, false, "")
	requireColumn(t, columns, "InputPayloadFormatVersion", "BIGINT", true, false, "")
	requireColumn(t, columns, "InputPayload", "MEDIUMBLOB", true, false, "")
	requireIndex(t, reservation, "uq_scheduled_run_reservation_identity",
		[]string{"OccurrenceIdentityDigest"}, true, false)
	requireIndex(t, reservation, "idx_scheduled_run_reservation_claim",
		[]string{"Status", "LeaseExpiresAt", "Id"}, false, false)
	requireIndex(t, reservation, "idx_scheduled_run_reservation_due",
		[]string{"Status", "DueAt", "Id"}, false, false)
	requireIndex(t, reservation, "idx_scheduled_run_reservation_scope_claim",
		[]string{"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Status", "LeaseExpiresAt"}, false, false)
	requireIndex(t, reservation, "idx_scheduled_run_reservation_scope_due",
		[]string{"OrgId", "AppInstallationId", "RuntimeCapabilityId", "Status", "DueAt"}, false, false)
	requireIndex(t, reservation, "idx_scheduled_run_reservation_batch",
		[]string{"BatchIdentityDigest"}, false, false)
}

func TestRollupDeltaDeclaresRuntimeScopedClaimAndParentLaneIndexes(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	delta := findSystemTableDefinition(t, defs, "_System_RollupDelta")
	columns := columnDefinitionsByName(delta)
	requireColumn(t, columns, "ScopeDigest", "CHAR(64)", true, false, "")
	if columns["ScopeDigest"].CollationName != "utf8mb4_bin" {
		t.Fatalf("ScopeDigest collation = %q, want utf8mb4_bin", columns["ScopeDigest"].CollationName)
	}
	requireIndex(t, delta, "idx_rollup_delta_scope_claim",
		[]string{"ScopeDigest", "Status", "NextAttemptAt", "CreatedDate", "Id"}, false, false)
	requireIndex(t, delta, "idx_rollup_delta_scope_parent_lane",
		[]string{"ScopeDigest", "ParentObject", "ParentRecordId", "Status", "NextAttemptAt", "CreatedDate"}, false, false)
	for _, indexName := range []string{"idx_rollup_delta_scope_claim", "idx_rollup_delta_scope_parent_lane"} {
		for _, index := range delta.Indices {
			if index.Name != indexName {
				continue
			}
			totalBytes := 0
			for _, columnName := range index.Columns {
				columnBytes, err := maxTiDBIndexBytes(columns[columnName].Type)
				if err != nil {
					t.Fatalf("%s.%s index width: %v", indexName, columnName, err)
				}
				totalBytes += columnBytes
			}
			if totalBytes > tidbMaxIndexKeyBytes {
				t.Fatalf("%s worst-case key bytes = %d, want <= %d", indexName, totalBytes, tidbMaxIndexKeyBytes)
			}
		}
	}
}

func TestPermissionStateVersionDeclaresProtectedSingletonShape(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	def := findSystemTableDefinition(t, defs, sharedconstants.TablePermissionStateVersion)
	if def.TableType != "system_core" || def.Category != "security" ||
		def.RequiredReadPermission != "BLOCKED" || def.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("owner shape = %q/%q read=%q write=%q", def.TableType, def.Category, def.RequiredReadPermission, def.RequiredWritePermission)
	}

	columns := columnDefinitionsByName(def)
	if len(columns) != 6 {
		t.Fatalf("columns = %d, want exact singleton shape of 6", len(columns))
	}
	requireColumn(t, columns, "SingletonKey", "TINYINT", false, true, "")
	requireColumn(t, columns, "Lineage", "BINARY(16)", false, false, "")
	requireColumn(t, columns, "Revision", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, columns, "EnforcementMode", "VARCHAR(8)", false, false, "")
	requireColumn(t, columns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, columns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	if len(def.Indices) != 0 || len(def.ForeignKeys) != 0 {
		t.Fatalf("secondary shape = indices %v foreign keys %v, want none", def.Indices, def.ForeignKeys)
	}
}

func findSystemTableDefinition(t *testing.T, defs []schema.TableDefinition, tableName string) schema.TableDefinition {
	t.Helper()
	for _, def := range defs {
		if def.TableName == tableName {
			return def
		}
	}
	t.Fatalf("%s not found", tableName)
	return schema.TableDefinition{}
}

func TestObjectFieldMappingDeclaresProjectionOnlyColumn(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}
	def := findSystemTableDefinition(t, defs, "_System_ObjectFieldMapping")
	columns := columnDefinitionsByName(def)
	requireColumn(t, columns, "ProjectionOnly", "TINYINT(1)", false, false, "0")
	requireColumn(t, columns, "TargetField", "VARCHAR(128)", false, false, "")
	requireIndex(t, def, "uq_ofm_mapping", []string{
		"Source", "SourceObject", "TargetObject", "ActionType", "SourceField", "Direction",
	}, true, false)
}

func TestPackGenerationActivationTablesDeclareAtomicRuntimeSelection(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	generation := findSystemTableDefinition(t, defs, "_System_PackGeneration")
	if generation.TableType != "system_core" || generation.Category != "registry" {
		t.Fatalf("generation owner shape = %q/%q", generation.TableType, generation.Category)
	}
	if generation.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("generation write permission = %q, want BLOCKED so generic CRUD cannot bypass control-plane activation", generation.RequiredWritePermission)
	}
	generationColumns := columnDefinitionsByName(generation)
	requireColumn(t, generationColumns, "PackInstallId", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "PackName", "VARCHAR(255)", false, false, "")
	requireColumn(t, generationColumns, "Status", "VARCHAR(64)", false, false, "'preparing'")
	requireColumn(t, generationColumns, "SourceRevision", "VARCHAR(255)", true, false, "")
	requireIndex(t, generation, "uq_pack_generation_install", []string{"PackInstallId"}, true, false)
	requireIndex(t, generation, "idx_pack_generation_name_status", []string{"PackName", "Status", "IsDeleted"}, false, false)

	activation := findSystemTableDefinition(t, defs, "_System_PackActivation")
	if activation.TableType != "system_core" || activation.Category != "registry" {
		t.Fatalf("activation owner shape = %q/%q", activation.TableType, activation.Category)
	}
	if activation.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("activation write permission = %q, want BLOCKED so generic CRUD cannot bypass control-plane activation", activation.RequiredWritePermission)
	}
	activationColumns := columnDefinitionsByName(activation)
	requireColumn(t, activationColumns, "PackName", "VARCHAR(255)", false, false, "")
	requireColumn(t, activationColumns, "ActivePackInstallId", "VARCHAR(255)", false, false, "")
	requireColumn(t, activationColumns, "PreviousPackInstallId", "VARCHAR(255)", true, false, "")
	requireColumn(t, activationColumns, "ActivationEpoch", "BIGINT", false, false, "0")
	requireIndex(t, activation, "uq_pack_activation_name", []string{"PackName"}, true, false)
}

func TestPackGenerationConnectorBindingDefinitionsDeclareCompleteImmutableAuthority(t *testing.T) {
	defs, err := SystemTableDefinitions()
	if err != nil {
		t.Fatalf("SystemTableDefinitions(): %v", err)
	}

	generation := findSystemTableDefinition(t, defs, "_System_PackGeneration")
	generationColumns := columnDefinitionsByName(generation)
	requireColumn(t, generationColumns, "Id", "VARCHAR(255)", false, true, "")
	if generationColumns["Id"].CollationName != "" {
		t.Fatalf("_System_PackGeneration.Id collation = %q, want established database default for upgrade compatibility", generationColumns["Id"].CollationName)
	}
	for _, forbidden := range []string{
		"ConnectorBindingDefinitionSetProtocol",
		"ConnectorBindingDefinitionCount",
		"ConnectorBindingDefinitionSetDigest",
		"ConnectorBindingAuthorityProtocol",
	} {
		if _, ok := generationColumns[forbidden]; ok {
			t.Fatalf("_System_PackGeneration gained connector authority scalar column %q", forbidden)
		}
	}

	set := findSystemTableDefinition(t, defs, "_System_PackGenerationConnectorBindingDefinitionSet")
	if set.TableType != "system_core" || set.Category != "registry" {
		t.Fatalf(
			"%s classification = %q/%q, want system_core/registry",
			set.TableName,
			set.TableType,
			set.Category,
		)
	}
	if set.RequiredReadPermission != "BLOCKED" || set.RequiredWritePermission != "BLOCKED" {
		t.Fatalf(
			"%s permissions = %q/%q, want BLOCKED/BLOCKED",
			set.TableName,
			set.RequiredReadPermission,
			set.RequiredWritePermission,
		)
	}
	requireExactColumnOrder(t, set, []string{
		"PackGenerationId",
		"Protocol",
		"DefinitionCount",
		"DefinitionSetDigest",
		"CreatedDate",
		"LastModifiedDate",
	})
	setColumns := columnDefinitionsByName(set)
	requireColumn(t, setColumns, "PackGenerationId", "VARCHAR(255)", false, true, "")
	requireColumn(t, setColumns, "Protocol", "VARCHAR(255)", false, false, "")
	requireColumn(t, setColumns, "DefinitionCount", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, setColumns, "DefinitionSetDigest", "CHAR(71)", false, false, "")
	requireColumn(t, setColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, setColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireBinaryCollation(t, set, []string{
		"Protocol",
		"DefinitionSetDigest",
	})
	if setColumns["PackGenerationId"].CollationName != "" {
		t.Fatalf("%s.PackGenerationId collation = %q, want parent-compatible database default", set.TableName, setColumns["PackGenerationId"].CollationName)
	}
	if len(set.Indices) != 0 {
		t.Fatalf("%s indices = %+v, want only its PackGenerationId primary key", set.TableName, set.Indices)
	}
	if len(set.ForeignKeys) != 1 {
		t.Fatalf("%s foreign keys = %+v, want only the exact PackGeneration parent", set.TableName, set.ForeignKeys)
	}
	requireForeignKey(t, set, "PackGenerationId", "_System_PackGeneration(Id)", "RESTRICT")

	definition := findSystemTableDefinition(t, defs, "_System_PackGenerationConnectorBindingDefinition")
	if definition.TableType != "system_core" || definition.Category != "registry" {
		t.Fatalf(
			"%s classification = %q/%q, want system_core/registry",
			definition.TableName,
			definition.TableType,
			definition.Category,
		)
	}
	if definition.RequiredReadPermission != "BLOCKED" ||
		definition.RequiredWritePermission != "BLOCKED" {
		t.Fatalf(
			"%s permissions = %q/%q, want BLOCKED/BLOCKED",
			definition.TableName,
			definition.RequiredReadPermission,
			definition.RequiredWritePermission,
		)
	}
	requireExactColumnOrder(t, definition, []string{
		"Id",
		"PackGenerationId",
		"PackInstallId",
		"DefinitionOrdinal",
		"Kind",
		"OwnerPack",
		"PackPayloadDigest",
		"BusinessKey",
		"DefinitionVersion",
		"DefinitionEnvelopeDigest",
		"DefinitionEnvelope",
		"CreatedDate",
		"LastModifiedDate",
	})
	definitionColumns := columnDefinitionsByName(definition)
	requireColumn(t, definitionColumns, "Id", "VARCHAR(255)", false, true, "")
	requireColumn(t, definitionColumns, "PackGenerationId", "VARCHAR(255)", false, false, "")
	requireColumn(t, definitionColumns, "PackInstallId", "VARCHAR(255)", false, false, "")
	requireColumn(t, definitionColumns, "DefinitionOrdinal", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, definitionColumns, "Kind", "VARCHAR(64)", false, false, "")
	requireColumn(t, definitionColumns, "OwnerPack", "VARCHAR(255)", false, false, "")
	requireColumn(t, definitionColumns, "PackPayloadDigest", "CHAR(85)", false, false, "")
	requireColumn(t, definitionColumns, "BusinessKey", "VARCHAR(255)", false, false, "")
	requireColumn(t, definitionColumns, "DefinitionVersion", "DECIMAL(20,0)", false, false, "")
	requireColumn(t, definitionColumns, "DefinitionEnvelopeDigest", "CHAR(71)", false, false, "")
	requireColumn(t, definitionColumns, "DefinitionEnvelope", "MEDIUMBLOB", false, false, "")
	requireColumn(t, definitionColumns, "CreatedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireColumn(t, definitionColumns, "LastModifiedDate", "DATETIME(6)", false, false, "CURRENT_TIMESTAMP(6)")
	requireBinaryCollation(t, definition, []string{
		"Id",
		"PackInstallId",
		"Kind",
		"OwnerPack",
		"PackPayloadDigest",
		"BusinessKey",
		"DefinitionEnvelopeDigest",
	})
	if definitionColumns["PackGenerationId"].CollationName != "" {
		t.Fatalf("%s.PackGenerationId collation = %q, want set-header-compatible database default", definition.TableName, definitionColumns["PackGenerationId"].CollationName)
	}

	if len(definition.Indices) != 2 {
		t.Fatalf("%s indices = %+v, want exactly ordinal and natural-identity uniqueness", definition.TableName, definition.Indices)
	}
	requireIndex(t, definition, "ux_pack_generation_binding_definition_ordinal", []string{
		"PackGenerationId",
		"DefinitionOrdinal",
	}, true, false)
	requireIndex(t, definition, "ux_pack_generation_binding_definition_identity", []string{
		"PackGenerationId",
		"Kind",
		"BusinessKey",
	}, true, false)
	naturalIdentityIndexBytes := 0
	for _, name := range []string{"PackGenerationId", "Kind", "BusinessKey"} {
		columnBytes, err := maxTiDBIndexBytes(definitionColumns[name].Type)
		if err != nil {
			t.Fatalf("index-width calculation for %s.%s: %v", definition.TableName, name, err)
		}
		naturalIdentityIndexBytes += columnBytes
	}
	if naturalIdentityIndexBytes > tidbMaxIndexKeyBytes {
		t.Fatalf(
			"%s natural-identity index requires %d bytes, exceeds TiDB limit %d",
			definition.TableName,
			naturalIdentityIndexBytes,
			tidbMaxIndexKeyBytes,
		)
	}

	if len(definition.ForeignKeys) != 1 {
		t.Fatalf("%s foreign keys = %+v, want only the exact PackGeneration parent", definition.TableName, definition.ForeignKeys)
	}
	requireForeignKey(
		t,
		definition,
		"PackGenerationId",
		"_System_PackGenerationConnectorBindingDefinitionSet(PackGenerationId)",
		"RESTRICT",
	)
	for _, immutableTable := range []struct {
		name    string
		columns map[string]schema.ColumnDefinition
	}{
		{name: set.TableName, columns: setColumns},
		{name: definition.TableName, columns: definitionColumns},
	} {
		for _, forbidden := range []string{
			"DefinitionIdentityDigest",
			"Status",
			"State",
			"Mode",
			"IsDeleted",
			"IsCurrent",
			"CurrentDefinitionId",
			"LatestDefinitionId",
			"ActiveKey",
			"FenceToken",
			"RetiredAt",
			"DeletedAt",
		} {
			if _, ok := immutableTable.columns[forbidden]; ok {
				t.Fatalf("%s gained mutable, selector, or parallel-identity field %q", immutableTable.name, forbidden)
			}
		}
	}
}

func requireExactColumnOrder(t *testing.T, def schema.TableDefinition, expected []string) {
	t.Helper()
	actual := make([]string, len(def.Columns))
	for i, column := range def.Columns {
		actual[i] = column.Name
	}
	if !reflect.DeepEqual(actual, expected) {
		t.Fatalf("%s column order = %v, want %v", def.TableName, actual, expected)
	}
}

func requireExactOAuthEffectColumns(t *testing.T, def schema.TableDefinition, expected map[string]schema.ColumnDefinition) {
	t.Helper()
	actual := columnDefinitionsByName(def)
	for name, want := range expected {
		got, ok := actual[name]
		if !ok {
			t.Fatalf("%s missing column %s", def.TableName, name)
		}
		if !reflect.DeepEqual(got, want) {
			t.Fatalf("%s.%s = %#v, want %#v", def.TableName, name, got, want)
		}
	}
}

func requireExactTableWithoutColumns(t *testing.T, actual, expected schema.TableDefinition) {
	t.Helper()
	actual.Columns = nil
	if !reflect.DeepEqual(actual, expected) {
		t.Fatalf("%s table definition excluding columns = %#v, want %#v", actual.TableName, actual, expected)
	}
}

func columnDefinitionsByName(def schema.TableDefinition) map[string]schema.ColumnDefinition {
	out := make(map[string]schema.ColumnDefinition, len(def.Columns))
	for _, col := range def.Columns {
		out[col.Name] = col
	}
	return out
}

func requireColumn(t *testing.T, columns map[string]schema.ColumnDefinition, name, sqlType string, nullable, primaryKey bool, defaultValue string) {
	t.Helper()
	col, ok := columns[name]
	if !ok {
		t.Fatalf("missing column %s", name)
	}
	if col.Type != sqlType || col.Nullable != nullable || col.PrimaryKey != primaryKey || col.Default != defaultValue {
		t.Fatalf("column %s = type %q nullable %v primary %v default %q, want type %q nullable %v primary %v default %q",
			name, col.Type, col.Nullable, col.PrimaryKey, col.Default, sqlType, nullable, primaryKey, defaultValue)
	}
}

func requireIndex(t *testing.T, def schema.TableDefinition, name string, columns []string, unique, primary bool) {
	t.Helper()
	for _, idx := range def.Indices {
		if idx.Name != name {
			continue
		}
		if got, want := strings.Join(idx.Columns, ","), strings.Join(columns, ","); got != want || idx.Unique != unique || idx.Primary != primary {
			t.Fatalf("%s index %q = columns %s unique %v primary %v, want columns %s unique %v primary %v",
				def.TableName, name, got, idx.Unique, idx.Primary, want, unique, primary)
		}
		return
	}
	t.Fatalf("%s index %q not found", def.TableName, name)
}

func requireForeignKey(t *testing.T, def schema.TableDefinition, column, references, onDelete string) {
	t.Helper()
	for _, fk := range def.ForeignKeys {
		if fk.Column != column {
			continue
		}
		if fk.References != references || fk.OnDelete != onDelete {
			t.Fatalf("%s foreign key on %s = references %q on delete %q, want references %q on delete %q",
				def.TableName, column, fk.References, fk.OnDelete, references, onDelete)
		}
		return
	}
	t.Fatalf("%s foreign key on %s not found", def.TableName, column)
}

func maxTiDBIndexBytes(sqlType string) (int, error) {
	upper := strings.ToUpper(strings.TrimSpace(sqlType))
	if (strings.HasPrefix(upper, "VARCHAR(") || strings.HasPrefix(upper, "CHAR(")) && strings.HasSuffix(upper, ")") {
		raw := strings.TrimSuffix(upper[strings.IndexByte(upper, '(')+1:], ")")
		length, err := strconv.Atoi(raw)
		if err != nil {
			return 0, err
		}
		return length * 4, nil
	}
	if strings.HasPrefix(upper, "VARBINARY(") && strings.HasSuffix(upper, ")") {
		raw := strings.TrimSuffix(strings.TrimPrefix(upper, "VARBINARY("), ")")
		return strconv.Atoi(raw)
	}
	if upper == "DATETIME" {
		return 8, nil
	}
	if upper == "INT" {
		return 4, nil
	}
	if upper == "BIGINT" {
		return 8, nil
	}
	return 0, fmt.Errorf("unsupported indexed column type")
}

// Every managed table must carry the temporal fields required by verify-l0.
func TestSystemTableCatalogDeclaresUniversalSystemFields(t *testing.T) {
	definitions, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	for _, definition := range definitions {
		columns := columnDefinitionsByName(definition)
		for _, name := range sharedconstants.GetSystemFieldNames() {
			if _, ok := columns[name]; !ok {
				t.Errorf("%s missing universal system field %s", definition.TableName, name)
			}
		}
	}
}

// Navigation receipts use only TableDefinition properties; reject misspelled
// keys instead of silently dropping schema intent during bootstrap.
func TestApplicationNavigationReceiptUsesDeclaredSchemaProperties(t *testing.T) {
	var entries []json.RawMessage
	if err := json.Unmarshal(SystemTablesJSON(), &entries); err != nil {
		t.Fatal(err)
	}
	for _, entry := range entries {
		var identity struct {
			TableName string `json:"tableName"`
		}
		if err := json.Unmarshal(entry, &identity); err != nil {
			t.Fatal(err)
		}
		if identity.TableName != "_System_ApplicationNavigationReceipt" {
			continue
		}
		decoder := json.NewDecoder(bytes.NewReader(entry))
		decoder.DisallowUnknownFields()
		var definition schema.TableDefinition
		if err := decoder.Decode(&definition); err != nil {
			t.Fatal(err)
		}
		return
	}
	t.Fatal("navigation receipt missing from canonical catalog")
}

func TestApplicationNavigationReceiptIsProtectedInternalState(t *testing.T) {
	definitions, err := SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	definition := findSystemTableDefinition(t, definitions, "_System_ApplicationNavigationReceipt")
	if definition.TableType != "system_core" || definition.RequiredReadPermission != "BLOCKED" || definition.RequiredWritePermission != "BLOCKED" {
		t.Fatalf("navigation replay state must be internal and blocked from generic CRUD: %+v", definition)
	}
	columns := columnDefinitionsByName(definition)
	requireColumn(t, columns, "Id", "CHAR(64)", false, true, "")
	requireIndex(t, definition, "idx_app_navigation_receipt_created", []string{"ApplicationId", "CreatedDate"}, false, false)
}