package runtimeconfig

import (
	"encoding/json"
	"os"
	"path/filepath"
	"reflect"
	"sort"
	"strings"
	"testing"
	"time"
)

func TestRoleConfigurationsValidateIdentityAndOwnedValues(t *testing.T) {
	values := validAPIConfigFiles(t)
	lookup := mapLookup(values)
	api, err := LoadAPI(lookup)
	if err != nil {
		t.Fatal(err)
	}
	if api.AdminAddress != DefaultRuntimeAPIAdminAddress || api.GRPCAddress != DefaultRuntimeAPIGRPCAddress || api.AdmissionLimit != 16 || api.ProfilingEnabled || api.SQLDiagnosticsEnabled {
		t.Fatalf("API defaults = %+v", api)
	}
	workerValues := validWorkerConfigFiles(t)
	worker, err := LoadWorker(mapLookup(workerValues))
	if err != nil {
		t.Fatal(err)
	}
	if worker.AdminAddress != DefaultWorkerAdminAddress || worker.Role.WorkFamily != "runtime.audit-history-fanout.v1" {
		t.Fatalf("worker defaults = %+v", worker)
	}
}

func TestAPIProfilingIsExplicitAndBoolean(t *testing.T) {
	values := validAPIConfigFiles(t)
	values[RuntimeAPIProfilingEnabledEnv] = "true"
	api, err := LoadAPI(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if !api.ProfilingEnabled {
		t.Fatal("profiling flag was not preserved")
	}
	values[RuntimeAPIProfilingEnabledEnv] = "sometimes"
	if _, err := LoadAPI(mapLookup(values)); err == nil {
		t.Fatal("invalid profiling flag was accepted")
	}
}

func TestAPISQLDiagnosticsIsIndependentExplicitAndBoolean(t *testing.T) {
	values := validAPIConfigFiles(t)
	values[RuntimeAPISQLDiagnosticsEnabledEnv] = "true"
	api, err := LoadAPI(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if !api.SQLDiagnosticsEnabled || api.ProfilingEnabled {
		t.Fatalf("diagnostic flags = SQL %t profiling %t", api.SQLDiagnosticsEnabled, api.ProfilingEnabled)
	}
	values[RuntimeAPISQLDiagnosticsEnabledEnv] = "sometimes"
	if _, err := LoadAPI(mapLookup(values)); err == nil {
		t.Fatal("invalid SQL diagnostics flag was accepted")
	}
}

func TestRoleConfigurationsFailClosed(t *testing.T) {
	valid := validAPIConfigFiles(t)
	tests := []map[string]string{{}, cloneConfig(valid), cloneConfig(valid), cloneConfig(valid), cloneConfig(valid)}
	tests[1][WorkloadIdentityTokenFileEnv] = "relative/token"
	tests[2][RuntimeAPIAdminAddressEnv], tests[2][RuntimeAPIGRPCAddressEnv] = ":9090", ":9090"
	tests[3][RuntimeAPIGRPCAddressEnv] = "missing-port"
	tests[4][RuntimeAPIAdmissionLimitEnv] = "65"
	for _, values := range tests {
		if _, err := LoadAPI(mapLookup(values)); err == nil {
			t.Fatalf("LoadAPI(%v) succeeded", values)
		}
	}
	if _, err := LoadWorker(mapLookup(map[string]string{})); err == nil {
		t.Fatal("LoadWorker without identity token succeeded")
	}
}

func TestServingConfigV4ClosesTwoWorkloadsAndCatalogConsumers(t *testing.T) {
	path := filepath.Join(t.TempDir(), "serving-v4.json")
	raw := `{"schema":"nexuscrm.runtime-api-serving/v4","permission_state_protocol":"permission_state.v1","replica_id":"runtime-test-1","authorized_workload_ids":["integration-service","sfcrm-app-service"],"org_id":"org-1","app_installation_id":"installation-1","catalog_snapshot_store_root":"/var/lib/nexuscrm/catalog","expected_service_release_digest":"sha256:service","schema_generation":1,"engine_generation":1,"identity_attempt_cap":"1s","session_scope_capabilities":{"api":["runtime.records.query.v1"]},"capabilities":[{"app_capability_id":"io.nexuscrm.sfcrm.query.v1","runtime_capability_id":"runtime.records.query.v1","authorized_workload_id":"sfcrm-app-service","expected_ownership_epoch":1,"expected_compatibility_receipt_digest":"sha256:receipt-query","admission_lease":"2s"}],"catalog_consumer_capabilities":[{"runtime_capability_id":"catalog.consumer.app-metadata-view.v2","authorized_workload_id":"sfcrm-app-service","admission_lease":"2s"},{"runtime_capability_id":"catalog.consumer.integration-binding-view.v1","authorized_workload_id":"integration-service","admission_lease":"2s"},{"runtime_capability_id":"runtime.commands.accept.v1","authorized_workload_id":"integration-service","catalog_snapshot_digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","admission_lease":"2s"}],"workload_capabilities":[{"runtime_capability_id":"runtime.commands.read-application-result.v1","authorized_workload_id":"integration-service","admission_lease":"2s"},{"runtime_capability_id":"runtime.connected-mailboxes.authorize.v1","authorized_workload_id":"integration-service","admission_lease":"2s","required_custom_permission":"SendRecordEmail"},{"runtime_capability_id":"runtime.identity.session-lifecycle.v1","authorized_workload_id":"sfcrm-app-service"}]}`
	if err := os.WriteFile(path, []byte(raw), 0o600); err != nil {
		t.Fatal(err)
	}
	config, err := loadServingConfig(path)
	if err != nil {
		t.Fatal(err)
	}
	if len(config.AuthorizedWorkloadIDs) != 2 ||
		config.Capabilities[0].AuthorizedWorkloadID != "sfcrm-app-service" ||
		len(config.WorkloadCapabilities) != 3 ||
		config.WorkloadCapabilities[0].AuthorizedWorkloadID != "integration-service" ||
		config.WorkloadCapabilities[0].AdmissionLease != "2s" ||
		config.WorkloadCapabilities[1].RequiredCustomPermission != "SendRecordEmail" ||
		config.WorkloadCapabilities[2].AuthorizedWorkloadID != "sfcrm-app-service" ||
		config.CatalogConsumerCapabilities[2].CatalogSnapshotDigest !=
			"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" {
		t.Fatalf("v4 serving config = %#v", config)
	}

	systemRaw := strings.Replace(raw, `"required_custom_permission":"SendRecordEmail"`, `"required_system_permission":"ApiEnabled"`, 1)
	if err := os.WriteFile(path, []byte(systemRaw), 0o600); err != nil {
		t.Fatal(err)
	}
	systemConfig, err := loadServingConfig(path)
	if err != nil {
		t.Fatal(err)
	}
	if systemConfig.WorkloadCapabilities[1].RequiredSystemPermission != "ApiEnabled" || systemConfig.WorkloadCapabilities[1].RequiredCustomPermission != "" {
		t.Fatal("system permission namespace was not preserved")
	}

	var document map[string]any
	if err := json.Unmarshal([]byte(raw), &document); err != nil {
		t.Fatal(err)
	}
	for name, mutate := range map[string]func(map[string]any){
		"unsorted workloads": func(value map[string]any) {
			value["authorized_workload_ids"] =
				[]string{"sfcrm-app-service", "integration-service"}
		},
		"old workload selector": func(value map[string]any) {
			value["authorized_workload_id"] = "sfcrm-app-service"
		},
		"unowned capability": func(value map[string]any) {
			capabilities := value["capabilities"].([]any)
			capabilities[0].(map[string]any)["authorized_workload_id"] =
				"unknown-service"
		},
		"unowned workload capability": func(value map[string]any) {
			capabilities := value["workload_capabilities"].([]any)
			capabilities[0].(map[string]any)["authorized_workload_id"] =
				"unknown-service"
		},
		"invalid workload lease": func(value map[string]any) {
			capabilities := value["workload_capabilities"].([]any)
			capabilities[0].(map[string]any)["admission_lease"] = "0s"
		},
		"ambiguous permission namespaces": func(value map[string]any) {
			value["workload_capabilities"].([]any)[1].(map[string]any)["required_system_permission"] = "ApiEnabled"
		},
		"noncanonical system permission": func(value map[string]any) {
			capability := value["workload_capabilities"].([]any)[1].(map[string]any)
			delete(capability, "required_custom_permission")
			capability["required_system_permission"] = " ApiEnabled"
		},
		"noncanonical custom permission": func(value map[string]any) {
			capabilities := value["workload_capabilities"].([]any)
			capabilities[1].(map[string]any)["required_custom_permission"] =
				" SendRecordEmail"
		},
		"duplicate Runtime capability across classes": func(value map[string]any) {
			capabilities := value["workload_capabilities"].([]any)
			capabilities[0].(map[string]any)["runtime_capability_id"] =
				"runtime.records.query.v1"
		},
	} {
		t.Run(name, func(t *testing.T) {
			cloneRaw, err := json.Marshal(document)
			if err != nil {
				t.Fatal(err)
			}
			var clone map[string]any
			if err := json.Unmarshal(cloneRaw, &clone); err != nil {
				t.Fatal(err)
			}
			mutate(clone)
			invalid, err := json.Marshal(clone)
			if err != nil {
				t.Fatal(err)
			}
			invalidPath := filepath.Join(t.TempDir(), "invalid.json")
			if err := os.WriteFile(invalidPath, invalid, 0o600); err != nil {
				t.Fatal(err)
			}
			if _, err := loadServingConfig(invalidPath); err == nil {
				t.Fatal("invalid v4 serving config was accepted")
			}
		})
	}
}

func TestServingConfigV4OwnsGenericFederatedOIDCTrust(t *testing.T) {
	path := filepath.Join(t.TempDir(), "serving-v4.json")
	raw := `{"schema":"nexuscrm.runtime-api-serving/v4","permission_state_protocol":"permission_state.v1","replica_id":"runtime-test-1","authorized_workload_ids":["integration-service","sfcrm-app-service"],"org_id":"org-1","app_installation_id":"installation-1","catalog_snapshot_store_root":"/var/lib/nexuscrm/catalog","expected_service_release_digest":"sha256:service","schema_generation":1,"engine_generation":1,"identity_attempt_cap":"1s","session_scope_capabilities":{"api":["runtime.records.query.v1"]},"capabilities":[{"app_capability_id":"io.nexuscrm.sfcrm.query.v1","runtime_capability_id":"runtime.records.query.v1","authorized_workload_id":"sfcrm-app-service","expected_ownership_epoch":1,"expected_compatibility_receipt_digest":"sha256:receipt-query","admission_lease":"2s"}],"workload_capabilities":[{"runtime_capability_id":"runtime.identity.session-lifecycle.v1","authorized_workload_id":"sfcrm-app-service"}],"federated_oidc_providers":[{"provider_id":"provider-a","audience":"client-a","session_scope":"api","trusted_issuers":["https://issuer.example"],"jwks_url":"https://issuer.example/.well-known/jwks.json"}]}`
	if err := os.WriteFile(path, []byte(raw), 0o600); err != nil {
		t.Fatal(err)
	}
	config, err := loadServingConfig(path)
	if err != nil {
		t.Fatal(err)
	}
	if len(config.FederatedOIDCProviders) != 1 ||
		config.FederatedOIDCProviders[0].ProviderID != "provider-a" {
		t.Fatalf("federated providers = %#v", config.FederatedOIDCProviders)
	}

	var document map[string]any
	if err := json.Unmarshal([]byte(raw), &document); err != nil {
		t.Fatal(err)
	}
	for name, mutate := range map[string]func(map[string]any){
		"duplicate provider": func(value map[string]any) {
			providers := value["federated_oidc_providers"].([]any)
			value["federated_oidc_providers"] = append(providers, providers[0])
		},
		"unsorted issuers": func(value map[string]any) {
			provider := value["federated_oidc_providers"].([]any)[0].(map[string]any)
			provider["trusted_issuers"] = []string{"https://z.example", "https://a.example"}
		},
		"insecure JWKS": func(value map[string]any) {
			provider := value["federated_oidc_providers"].([]any)[0].(map[string]any)
			provider["jwks_url"] = "http://issuer.example/jwks.json"
		},
		"unmapped session scope": func(value map[string]any) {
			provider := value["federated_oidc_providers"].([]any)[0].(map[string]any)
			provider["session_scope"] = "unknown"
		},
	} {
		t.Run(name, func(t *testing.T) {
			cloneRaw, err := json.Marshal(document)
			if err != nil {
				t.Fatal(err)
			}
			var clone map[string]any
			if err := json.Unmarshal(cloneRaw, &clone); err != nil {
				t.Fatal(err)
			}
			mutate(clone)
			invalid, err := json.Marshal(clone)
			if err != nil {
				t.Fatal(err)
			}
			invalidPath := filepath.Join(t.TempDir(), "invalid.json")
			if err := os.WriteFile(invalidPath, invalid, 0o600); err != nil {
				t.Fatal(err)
			}
			if _, err := loadServingConfig(invalidPath); err == nil {
				t.Fatal("invalid federated OIDC config was accepted")
			}
		})
	}
}

func TestWorkerRoleConfigurationIsStrictAndBounded(t *testing.T) {
	for _, test := range []struct {
		name   string
		mutate func(map[string]any)
	}{
		{name: "unknown schema", mutate: func(d map[string]any) { d["schema"] = "nexuscrm.runtime-worker-role/v999" }},
		{name: "invalid worker source revision", mutate: func(d map[string]any) { d["worker_source_revision"] = "main" }},
		{name: "zero epoch", mutate: func(d map[string]any) { d["ownership_epoch"] = 0 }},
		{name: "oversized batch", mutate: func(d map[string]any) { d["batch_size"] = MaxWorkerBatchSize + 1 }},
		{name: "parallel lanes exceed pool", mutate: func(d map[string]any) { d["parallelism"] = 2 }},
		{name: "oversized parallelism", mutate: func(d map[string]any) { d["parallelism"] = MaxWorkerParallelism + 1 }},
		{name: "unbounded pool", mutate: func(d map[string]any) { d["db_max_open_conns"] = MaxWorkerDBConnections + 1 }},
		{name: "unknown metadata authority", mutate: func(d map[string]any) { d["catalog_revision"] = "7" }},
		{name: "zero processed retention", mutate: func(d map[string]any) { d["processed_retention"] = "0s" }},
		{name: "invalid cleanup interval", mutate: func(d map[string]any) { d["processed_cleanup_interval"] = "daily" }},
		{name: "too frequent ownership health", mutate: func(d map[string]any) { d["ownership_health_interval"] = "999ms" }},
		{name: "unbounded ownership health", mutate: func(d map[string]any) { d["ownership_health_interval"] = "6m" }},
	} {
		t.Run(test.name, func(t *testing.T) {
			values := validWorkerConfigFiles(t)
			path := values[RuntimeWorkerConfigFileEnv]
			raw, err := os.ReadFile(path)
			if err != nil {
				t.Fatal(err)
			}
			var document map[string]any
			if err := json.Unmarshal(raw, &document); err != nil {
				t.Fatal(err)
			}
			test.mutate(document)
			raw, err = json.Marshal(document)
			if err != nil {
				t.Fatal(err)
			}
			if err := os.WriteFile(path, raw, 0o600); err != nil {
				t.Fatal(err)
			}
			if _, err := LoadWorker(mapLookup(values)); err == nil {
				t.Fatal("invalid worker role configuration was accepted")
			}
		})
	}
}

func TestWorkerRoleOwnershipHealthCadenceIsIndependentFromBusinessPolling(t *testing.T) {
	values := validWorkerConfigFiles(t)
	worker, err := LoadWorker(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if got := worker.Role.EffectiveOwnershipHealthInterval(); got != DefaultWorkerOwnershipHealthInterval {
		t.Fatalf("default ownership health interval = %s", got)
	}

	path := values[RuntimeWorkerConfigFileEnv]
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	var document map[string]any
	if err := json.Unmarshal(raw, &document); err != nil {
		t.Fatal(err)
	}
	document["poll_interval"] = "30s"
	document["ownership_health_interval"] = "45s"
	raw, err = json.Marshal(document)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	worker, err = LoadWorker(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if got := worker.Role.EffectiveOwnershipHealthInterval(); got != 45*time.Second {
		t.Fatalf("explicit ownership health interval = %s", got)
	}
	if worker.Role.PollInterval != "30s" {
		t.Fatalf("business poll interval = %s", worker.Role.PollInterval)
	}

	document["poll_interval"] = "1m1s"
	raw, err = json.Marshal(document)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := LoadWorker(mapLookup(values)); err == nil {
		t.Fatal("worker role accepted a poll interval above the one-minute bound")
	}
}

func TestWorkerRoleParallelismReservesEffectHeartbeatAndControllerConnections(t *testing.T) {
	values := validWorkerConfigFiles(t)
	path := values[RuntimeWorkerConfigFileEnv]
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	var document map[string]any
	if err := json.Unmarshal(raw, &document); err != nil {
		t.Fatal(err)
	}
	document["parallelism"] = 2
	document["db_max_open_conns"] = 5
	raw, err = json.Marshal(document)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	config, err := LoadWorker(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if config.Role.EffectiveParallelism() != 2 {
		t.Fatalf("effective parallelism = %d", config.Role.EffectiveParallelism())
	}
}

func TestWorkerRoleV2ConfigurationLoadsMetadataScope(t *testing.T) {
	values := validWorkerConfigFiles(t)
	path := values[RuntimeWorkerConfigFileEnv]
	raw, err := os.ReadFile(path)
	if err != nil {
		t.Fatal(err)
	}
	var document map[string]any
	if err := json.Unmarshal(raw, &document); err != nil {
		t.Fatal(err)
	}
	document["schema"] = WorkerConfigSchemaV2
	document["work_family"] = "runtime.record-event-dispatch.v1"
	document["org_id"] = "org-test"
	document["app_installation_id"] = "installation-test"
	document["runtime_capability_id"] = "runtime.records.mutate.v1"
	document["catalog_snapshot_store_root"] = "/var/lib/nexuscrm/catalog"
	document["schema_generation"] = 1
	document["engine_generation"] = 2
	document["processed_retention"] = "336h"
	document["processed_cleanup_interval"] = "1h"
	raw, err = json.Marshal(document)
	if err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(path, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	worker, err := LoadWorker(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	scope, err := worker.Role.MetadataScope()
	if err != nil {
		t.Fatal(err)
	}
	if scope.OrgID != "org-test" || scope.AppInstallationID != "installation-test" ||
		scope.RuntimeCapabilityID != "runtime.records.mutate.v1" ||
		scope.CatalogSnapshotStoreRoot != "/var/lib/nexuscrm/catalog" ||
		scope.SchemaGeneration != 1 || scope.EngineGeneration != 2 ||
		worker.Role.ProcessedRetention != "336h" || worker.Role.ProcessedCleanupInterval != "1h" {
		t.Fatalf("metadata scope = %+v", scope)
	}
}

func TestWorkerTimedFlowOwnerRoleIsOptionalAndStrict(t *testing.T) {
	values := validWorkerConfigFiles(t)
	worker, err := LoadWorker(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if worker.TimedFlowOwner != nil {
		t.Fatal("worker acquired timed Flow owner without an explicit role file")
	}

	raw, err := os.ReadFile(values[RuntimeWorkerConfigFileEnv])
	if err != nil {
		t.Fatal(err)
	}
	var document map[string]any
	if err := json.Unmarshal(raw, &document); err != nil {
		t.Fatal(err)
	}
	document["schema"] = WorkerConfigSchemaV2
	document["work_family"] = "runtime.scheduled-flow.v1"
	document["org_id"] = "org-test"
	document["app_installation_id"] = "installation-test"
	document["app_capability_id"] = "io.nexuscrm.sfcrm.flow.v1"
	document["runtime_capability_id"] = "runtime.flow.execute.v1"
	document["expected_compatibility_receipt_digest"] = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
	document["catalog_snapshot_store_root"] = "/var/lib/nexuscrm/catalog"
	document["schema_generation"] = 1
	document["engine_generation"] = 2
	raw, err = json.Marshal(document)
	if err != nil {
		t.Fatal(err)
	}
	ownerPath := filepath.Join(filepath.Dir(values[RuntimeWorkerConfigFileEnv]), "scheduled-owner.json")
	if err := os.WriteFile(ownerPath, raw, 0o600); err != nil {
		t.Fatal(err)
	}
	values[RuntimeWorkerTimedFlowOwnerConfigFileEnv] = ownerPath
	worker, err = LoadWorker(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if worker.TimedFlowOwner == nil ||
		worker.TimedFlowOwner.WorkFamily != "runtime.scheduled-flow.v1" ||
		worker.TimedFlowOwner.AppCapabilityID != "io.nexuscrm.sfcrm.flow.v1" {
		t.Fatalf("timed Flow owner = %#v", worker.TimedFlowOwner)
	}

	values[RuntimeWorkerTimedFlowOwnerConfigFileEnv] = "relative/scheduled-owner.json"
	if _, err := LoadWorker(mapLookup(values)); err == nil {
		t.Fatal("relative timed Flow owner role path was accepted")
	}
}

func TestWorkerRoleV3RequiresPermissionStateV1AndSupportsBothScopeKinds(t *testing.T) {
	for _, test := range []struct {
		name     string
		schema   string
		protocol any
		metadata bool
		wantErr  bool
	}{
		{
			name:   "v1 rejects protocol",
			schema: WorkerConfigSchemaV1, protocol: PermissionStateProtocolV1,
			wantErr: true,
		},
		{
			name:   "v2 rejects protocol",
			schema: WorkerConfigSchemaV2, protocol: PermissionStateProtocolV1,
			metadata: true, wantErr: true,
		},
		{name: "v3 requires protocol", schema: WorkerConfigSchemaV3, wantErr: true},
		{
			name:   "v3 rejects unknown protocol",
			schema: WorkerConfigSchemaV3, protocol: "permission_state.v2",
			wantErr: true,
		},
		{
			name:   "v3 accepts payload scope",
			schema: WorkerConfigSchemaV3, protocol: PermissionStateProtocolV1,
		},
		{
			name:   "v3 accepts metadata scope",
			schema: WorkerConfigSchemaV3, protocol: PermissionStateProtocolV1,
			metadata: true,
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			values := validWorkerConfigFiles(t)
			path := values[RuntimeWorkerConfigFileEnv]
			raw, err := os.ReadFile(path)
			if err != nil {
				t.Fatal(err)
			}
			var document map[string]any
			if err := json.Unmarshal(raw, &document); err != nil {
				t.Fatal(err)
			}
			document["schema"] = test.schema
			if test.protocol != nil {
				document["permission_state_protocol"] = test.protocol
			}
			if test.metadata {
				document["work_family"] = "runtime.record-event-dispatch.v1"
				document["org_id"] = "org-test"
				document["app_installation_id"] = "installation-test"
				document["runtime_capability_id"] = "runtime.records.mutate.v1"
				document["catalog_snapshot_store_root"] = "/var/lib/nexuscrm/catalog"
				document["schema_generation"] = 1
				document["engine_generation"] = 2
			}
			raw, err = json.Marshal(document)
			if err != nil {
				t.Fatal(err)
			}
			if err := os.WriteFile(path, raw, 0o600); err != nil {
				t.Fatal(err)
			}
			config, err := LoadWorker(mapLookup(values))
			if test.wantErr {
				if err == nil {
					t.Fatal("invalid worker Permission State config was accepted")
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if config.Role.PermissionStateProtocol != PermissionStateProtocolV1 {
				t.Fatalf(
					"Permission State protocol = %q",
					config.Role.PermissionStateProtocol,
				)
			}
			if test.metadata {
				if _, err := config.Role.MetadataScope(); err != nil {
					t.Fatalf("v3 metadata scope: %v", err)
				}
			}
		})
	}

	values := validWorkerConfigFiles(t)
	role, err := loadWorkerRoleConfig(values[RuntimeWorkerConfigFileEnv])
	if err != nil {
		t.Fatal(err)
	}
	role.Schema = WorkerConfigSchemaV3
	role.PermissionStateProtocol = PermissionStateProtocolV1
	role.OrgID = "org-test"
	if err := role.Validate(); err == nil {
		t.Fatal("v3 worker with a partial metadata scope was accepted")
	}
}

func TestWorkerMetadataScopeIsCompleteAndCanonical(t *testing.T) {
	valid := WorkerMetadataScope{
		OrgID: "org-test", AppInstallationID: "installation-test", RuntimeCapabilityID: "runtime.records.mutate.v1",
		CatalogSnapshotStoreRoot: "/var/lib/nexuscrm/catalog", SchemaGeneration: 1, EngineGeneration: 1,
	}
	if err := valid.Validate(); err != nil {
		t.Fatal(err)
	}
	for name, mutate := range map[string]func(*WorkerMetadataScope){
		"missing org":         func(value *WorkerMetadataScope) { value.OrgID = "" },
		"relative store":      func(value *WorkerMetadataScope) { value.CatalogSnapshotStoreRoot = "catalog" },
		"zero schema":         func(value *WorkerMetadataScope) { value.SchemaGeneration = 0 },
		"zero engine":         func(value *WorkerMetadataScope) { value.EngineGeneration = 0 },
		"whitespace identity": func(value *WorkerMetadataScope) { value.RuntimeCapabilityID = " runtime.records.mutate.v1" },
		"invalid receipt": func(value *WorkerMetadataScope) {
			value.ExpectedCompatibilityReceiptDigest = "sha256:not-a-digest"
		},
	} {
		t.Run(name, func(t *testing.T) {
			candidate := valid
			mutate(&candidate)
			if err := candidate.Validate(); err == nil {
				t.Fatal("invalid worker metadata scope was accepted")
			}
		})
	}
}

func TestServingConfigRequiresExplicitDistinctCapabilityAxes(t *testing.T) {
	for _, test := range []struct {
		name   string
		mutate func(map[string]any)
	}{
		{
			name: "same App and Runtime capability",
			mutate: func(document map[string]any) {
				capabilities := document["capabilities"].([]any)
				capability := capabilities[0].(map[string]any)
				capability["app_capability_id"] = capability["runtime_capability_id"]
			},
		},
		{
			name: "legacy conflated capability field",
			mutate: func(document map[string]any) {
				delete(document, "capabilities")
				document["capability_id"] = "runtime.records.evaluate-access.v1"
			},
		},
		{
			name: "duplicate Runtime capability",
			mutate: func(document map[string]any) {
				capabilities := document["capabilities"].([]any)
				duplicate := map[string]any{}
				for key, value := range capabilities[0].(map[string]any) {
					duplicate[key] = value
				}
				duplicate["app_capability_id"] = "io.nexuscrm.sfcrm.query.v1"
				document["capabilities"] = append(capabilities, duplicate)
			},
		},
	} {
		t.Run(test.name, func(t *testing.T) {
			values := validAPIConfigFiles(t)
			path := values[RuntimeAPIServingConfigFileEnv]
			raw, err := os.ReadFile(path)
			if err != nil {
				t.Fatal(err)
			}
			var document map[string]any
			if err := json.Unmarshal(raw, &document); err != nil {
				t.Fatal(err)
			}
			test.mutate(document)
			raw, err = json.Marshal(document)
			if err != nil {
				t.Fatal(err)
			}
			if err := os.WriteFile(path, raw, 0o600); err != nil {
				t.Fatal(err)
			}
			if _, err := LoadAPI(mapLookup(values)); err == nil {
				t.Fatal("invalid capability-axis configuration was accepted")
			}
		})
	}
}

func TestServingConfigV1IsNotAnIngressGrammarForTheD2ARuntime(t *testing.T) {
	values := validAPIConfigFiles(t)
	path := values[RuntimeAPIServingConfigFileEnv]
	if err := os.WriteFile(path, []byte(validServingConfigV1), 0o600); err != nil {
		t.Fatal(err)
	}
	if _, err := LoadAPI(mapLookup(values)); err == nil {
		t.Fatal("D1 v1 serving config was accepted by the D2A Runtime")
	}
}

func TestServingConfigV3RequiresPermissionStateV1(t *testing.T) {
	for _, test := range []struct {
		name     string
		schema   string
		protocol any
		wantErr  bool
	}{
		{name: "v2 remains compatible without protocol", schema: ServingConfigSchemaV2},
		{name: "v2 rejects protocol field", schema: ServingConfigSchemaV2, protocol: PermissionStateProtocolV1, wantErr: true},
		{name: "v3 requires protocol", schema: ServingConfigSchemaV3, wantErr: true},
		{name: "v3 rejects unknown protocol", schema: ServingConfigSchemaV3, protocol: "permission_state.v2", wantErr: true},
		{name: "v3 accepts exact protocol", schema: ServingConfigSchemaV3, protocol: PermissionStateProtocolV1},
	} {
		t.Run(test.name, func(t *testing.T) {
			values := validAPIConfigFiles(t)
			path := values[RuntimeAPIServingConfigFileEnv]
			raw, err := os.ReadFile(path)
			if err != nil {
				t.Fatal(err)
			}
			var document map[string]any
			if err := json.Unmarshal(raw, &document); err != nil {
				t.Fatal(err)
			}
			document["schema"] = test.schema
			if test.protocol != nil {
				document["permission_state_protocol"] = test.protocol
			}
			raw, err = json.Marshal(document)
			if err != nil {
				t.Fatal(err)
			}
			if err := os.WriteFile(path, raw, 0o600); err != nil {
				t.Fatal(err)
			}
			config, err := LoadAPI(mapLookup(values))
			if test.wantErr {
				if err == nil {
					t.Fatal("invalid Permission State serving config was accepted")
				}
				return
			}
			if err != nil {
				t.Fatal(err)
			}
			if config.Serving.Schema != test.schema {
				t.Fatalf("schema = %q, want %q", config.Serving.Schema, test.schema)
			}
			wantProtocol, _ := test.protocol.(string)
			if config.Serving.PermissionStateProtocol != wantProtocol {
				t.Fatalf(
					"Permission State protocol = %q, want %q",
					config.Serving.PermissionStateProtocol,
					wantProtocol,
				)
			}
		})
	}
}

func TestServingConfigParsesOnlyPositiveOptionalOutcomeRetention(t *testing.T) {
	for _, test := range []struct {
		name    string
		value   string
		wantErr bool
	}{
		{name: "positive", value: "24h"},
		{name: "zero", value: "0s", wantErr: true},
		{name: "invalid", value: "later", wantErr: true},
	} {
		t.Run(test.name, func(t *testing.T) {
			values := validAPIConfigFiles(t)
			path := values[RuntimeAPIServingConfigFileEnv]
			raw, err := os.ReadFile(path)
			if err != nil {
				t.Fatal(err)
			}
			var document map[string]any
			if err := json.Unmarshal(raw, &document); err != nil {
				t.Fatal(err)
			}
			capability := document["capabilities"].([]any)[0].(map[string]any)
			capability["outcome_retention"] = test.value
			raw, err = json.Marshal(document)
			if err != nil {
				t.Fatal(err)
			}
			if err := os.WriteFile(path, raw, 0o600); err != nil {
				t.Fatal(err)
			}
			config, err := LoadAPI(mapLookup(values))
			if test.wantErr {
				if err == nil {
					t.Fatal("invalid outcome retention was accepted")
				}
				return
			}
			if err != nil || config.Serving.Capabilities[0].OutcomeRetention != test.value {
				t.Fatalf("LoadAPI() = %+v, %v", config, err)
			}
		})
	}
}

func TestEmbeddedSchemasMatchRuntimeOwnedEnvironmentKeys(t *testing.T) {
	assertSchemaKeys(t, RuntimeAPISchema, []string{
		RuntimeAPIAdminAddressEnv, RuntimeAPIAdmissionLimitEnv, RuntimeAPIGRPCAddressEnv,
		RuntimeAPIProfilingEnabledEnv, RuntimeAPISQLDiagnosticsEnabledEnv,
		RuntimeAPIServingConfigFileEnv, RuntimeAPISigningKeyFileEnv, RuntimeAPITLSCertificateFileEnv,
		RuntimeAPITLSPrivateKeyFileEnv, WorkloadIdentityTokenFileEnv,
	}, []string{RuntimeAPIServingConfigFileEnv, RuntimeAPISigningKeyFileEnv, RuntimeAPITLSCertificateFileEnv,
		RuntimeAPITLSPrivateKeyFileEnv, WorkloadIdentityTokenFileEnv})
	assertSchemaKeys(t, RuntimeWorkerSchema,
		[]string{RuntimeWorkerAdminAddressEnv, RuntimeWorkerCatalogReconcileConfigFileEnv, RuntimeWorkerConfigFileEnv, RuntimeWorkerTimedFlowOwnerConfigFileEnv, WorkloadIdentityTokenFileEnv},
		[]string{RuntimeWorkerConfigFileEnv, WorkloadIdentityTokenFileEnv})
}

func TestLoadWorkerAcceptsOnlyAbsoluteCatalogReconcileConfigFile(t *testing.T) {
	values := validWorkerConfigFiles(t)
	path := filepath.Join(filepath.Dir(values[RuntimeWorkerConfigFileEnv]), "catalog-reconcile.json")
	if err := os.WriteFile(path, []byte(`{}`), 0o600); err != nil {
		t.Fatal(err)
	}
	values[RuntimeWorkerCatalogReconcileConfigFileEnv] = path
	worker, err := LoadWorker(mapLookup(values))
	if err != nil || worker.CatalogReconcileConfigFile != path {
		t.Fatalf("LoadWorker() Catalog reconcile config = %q, %v", worker.CatalogReconcileConfigFile, err)
	}
	values[RuntimeWorkerCatalogReconcileConfigFileEnv] = "relative/catalog-reconcile.json"
	if _, err := LoadWorker(mapLookup(values)); err == nil {
		t.Fatal("relative Catalog reconcile config path was accepted")
	}
}

func TestLoadAPIBootstrapRequiresOnlyL0TransportAuthority(t *testing.T) {
	values := validAPIConfigFiles(t)
	delete(values, WorkloadIdentityTokenFileEnv)
	delete(values, RuntimeAPISigningKeyFileEnv)
	delete(values, RuntimeAPIServingConfigFileEnv)

	config, err := LoadAPIBootstrap(mapLookup(values))
	if err != nil {
		t.Fatal(err)
	}
	if config.TLSCertificateFile == "" || config.TLSPrivateKeyFile == "" ||
		config.WorkloadIdentityTokenFile != "" || config.SigningKeyFile != "" ||
		config.Serving.Schema != "" {
		t.Fatalf("bootstrap config = %#v", config)
	}

	delete(values, RuntimeAPITLSCertificateFileEnv)
	if _, err := LoadAPIBootstrap(mapLookup(values)); err == nil {
		t.Fatal("bootstrap config accepted missing TLS certificate")
	}
}

func assertSchemaKeys(t *testing.T, raw []byte, want, required []string) {
	t.Helper()
	var schema struct {
		AdditionalProperties bool                       `json:"additionalProperties"`
		Properties           map[string]json.RawMessage `json:"properties"`
		Required             []string                   `json:"required"`
	}
	if err := json.Unmarshal(raw, &schema); err != nil {
		t.Fatal(err)
	}
	if schema.AdditionalProperties {
		t.Fatal("configuration schema permits undeclared properties")
	}
	got := make([]string, 0, len(schema.Properties))
	for key := range schema.Properties {
		got = append(got, key)
	}
	sort.Strings(got)
	sort.Strings(want)
	sort.Strings(schema.Required)
	sort.Strings(required)
	if !reflect.DeepEqual(got, want) || !reflect.DeepEqual(schema.Required, required) {
		t.Fatalf("schema keys=%v required=%v, want keys=%v", got, schema.Required, want)
	}
}

func validAPIConfigFiles(t *testing.T) map[string]string {
	t.Helper()
	dir := t.TempDir()
	token := filepath.Join(dir, "workload-token")
	signing := filepath.Join(dir, "signing-keys.json")
	serving := filepath.Join(dir, "serving.json")
	certificate := filepath.Join(dir, "runtime.crt")
	privateKey := filepath.Join(dir, "runtime.key")
	for path, content := range map[string]string{
		token:       "workload-secret",
		signing:     `{}`,
		serving:     `{"schema":"nexuscrm.runtime-api-serving/v2","replica_id":"runtime-test-1","authorized_workload_id":"sfcrm-app-service","org_id":"org-1","app_installation_id":"installation-1","catalog_snapshot_store_root":"/var/lib/nexuscrm/catalog","expected_service_release_digest":"sha256:service","schema_generation":1,"engine_generation":1,"identity_attempt_cap":"1s","session_scope_capabilities":{"api":["runtime.records.evaluate-access.v1","runtime.records.query.v1"]},"capabilities":[{"app_capability_id":"io.nexuscrm.sfcrm.ui.record-access.v1","runtime_capability_id":"runtime.records.evaluate-access.v1","expected_ownership_epoch":1,"expected_compatibility_receipt_digest":"sha256:receipt-access","admission_lease":"2s"},{"app_capability_id":"io.nexuscrm.sfcrm.query.v1","runtime_capability_id":"runtime.records.query.v1","expected_ownership_epoch":1,"expected_compatibility_receipt_digest":"sha256:receipt-query","admission_lease":"2s"}]}`,
		certificate: "test certificate",
		privateKey:  "test private key",
	} {
		if err := os.WriteFile(path, []byte(content), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	return map[string]string{
		WorkloadIdentityTokenFileEnv:    token,
		RuntimeAPISigningKeyFileEnv:     signing,
		RuntimeAPIServingConfigFileEnv:  serving,
		RuntimeAPITLSCertificateFileEnv: certificate,
		RuntimeAPITLSPrivateKeyFileEnv:  privateKey,
	}
}

func validWorkerConfigFiles(t *testing.T) map[string]string {
	t.Helper()
	dir := t.TempDir()
	token := filepath.Join(dir, "workload-token")
	role := filepath.Join(dir, "worker-role.json")
	if err := os.WriteFile(token, []byte("workload-secret"), 0o600); err != nil {
		t.Fatal(err)
	}
	if err := os.WriteFile(role, []byte(`{"schema":"nexuscrm.runtime-worker-role/v1","work_family":"runtime.audit-history-fanout.v1","deployment_id":"runtime-worker-test","worker_release_digest":"sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","worker_source_revision":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","ownership_epoch":1,"batch_size":10,"poll_interval":"100ms","lease_duration":"30s","retry_jitter_max":"10ms","max_attempts":5,"db_max_open_conns":4,"db_max_idle_conns":2,"shutdown_timeout":"10s"}`), 0o600); err != nil {
		t.Fatal(err)
	}
	return map[string]string{WorkloadIdentityTokenFileEnv: token, RuntimeWorkerConfigFileEnv: role}
}

const validServingConfigV1 = `{"schema":"nexuscrm.runtime-api-serving/v1","replica_id":"runtime-test-1","authorized_workload_id":"sfcrm-app-service","org_id":"org-1","app_installation_id":"installation-1","app_capability_id":"io.nexuscrm.sfcrm.ui.record-access.v1","runtime_capability_id":"runtime.records.evaluate-access.v1","catalog_snapshot_store_root":"/var/lib/nexuscrm/catalog","expected_service_release_digest":"sha256:service","expected_ownership_epoch":1,"expected_compatibility_receipt_digest":"sha256:receipt-access","schema_generation":1,"engine_generation":1,"identity_attempt_cap":"1s","record_access_admission_lease":"5s","session_scope_capabilities":{"api":["runtime.records.evaluate-access.v1"]}}`

func cloneConfig(input map[string]string) map[string]string {
	clone := make(map[string]string, len(input))
	for key, value := range input {
		clone[key] = value
	}
	return clone
}

func mapLookup(values map[string]string) LookupEnv {
	return func(key string) (string, bool) {
		value, ok := values[key]
		return value, ok
	}
}