"""Trusted, tenant-scoped routing for migrated hosted model invocations. App identities remain logical identities. Only SYSTEM shared-pool credentials can produce a route; user credential dictionaries cannot request a redirect. The frozen plan contains references, never secrets. The managed key is loaded by its exact tenant/provider/id at the transport boundary, not the user's currently selected Tokener credential. """ from __future__ import annotations import hashlib import hmac import json from dataclasses import dataclass from enum import StrEnum from operator import itemgetter from typing import TYPE_CHECKING, Any, Literal, TypeGuard, override from uuid import uuid4 from sqlalchemy import select from werkzeug.exceptions import ServiceUnavailable from configs import dify_config from core.db.session_factory import session_factory from core.entities.provider_entities import ProviderQuotaType, QuotaConfiguration, QuotaUnit from core.model_billing_migration_protocol import canonical_hash from core.repositories.model_billing_migration_repository import get_routing_state from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelPropertyKey, ModelType from models.provider import ProviderCredential, ProviderType from models.provider_ids import ModelProviderID from models.tokener import TenantTokenerIntegration, TenantTokenerIntegrationStatus from services.entities.model_billing_mapping import MappingSource as MappingSource # noqa: PLC0414 from services.entities.model_billing_mapping import MappingTarget as MappingTarget # noqa: PLC0414 from services.entities.model_billing_mapping import ModelMapping as ModelMapping # noqa: PLC0414 from services.entities.model_billing_mapping import ParameterAdapter as ParameterAdapter # noqa: PLC0414 from services.model_billing_mapping_service import ( SELECTED_SOURCE_VARIANT_TYPES, SelectedSourceVariant, SourceParameterContract, TTSVoiceContract, additional_model_is_authorized, build_source_adapter_contract, hosted_audio_source_fact, hosted_embedding_source_fact, hosted_gemini_source_fact, hosted_migration_sources, hosted_source_variant_fact, hosted_tongyi_source_fact, hosted_xai_source_fact, managed_source_schemas, mapping_supports_installed_plugin, mappings_for_revision, migration_policy_allows_empty_sources, model_schema_digest, partition_hosted_migration_sources, partition_known_unavailable_sources, resolve_source_capability_policy, resolve_source_mapping, restore_source_parameter_contract, retirement_policy_for_revision, source_mapping_revision, source_parameter_marker, source_parameter_public_settings, uses_published_tokener_balance, uses_stable_dependency_inventory, validate_frozen_mapping_evidence, ) from services.model_billing_mapping_service import normalize_tts_voice as normalize_tts_voice # noqa: PLC0414 if TYPE_CHECKING: from core.entities.provider_configuration import ProviderConfiguration from core.entities.provider_entities import SystemConfiguration class ModelRouteUnavailable(ServiceUnavailable): error_code = "model_route_unavailable" description = "Managed model routing is temporarily unavailable." class ModelMigrationProcessing(ModelRouteUnavailable): error_code = "migration_processing" description = "Model billing migration is processing. Please retry shortly." class ModelRouteUnmapped(ModelRouteUnavailable): error_code = "model_route_unmapped" description = "The managed model route has not been configured." class ModelInvocationReprepare(ModelRouteUnavailable): error_code = "model_invocation_reprepare" description = "Model routing changed while preparing this invocation. Please retry." class SettlementOwner(StrEnum): CUSTOM = "custom" SHARED_CREDIT_POOL = "shared_credit_pool" PROVIDER_FREE = "provider_free" UNMETERED = "unmetered" def settlement_owner(system: SystemConfiguration, provider_type: ProviderType) -> SettlementOwner: """Mirror the reservation policy without making a reservation to probe it.""" if provider_type != ProviderType.SYSTEM: return SettlementOwner.CUSTOM quota = next((q for q in system.quota_configurations if q.quota_type == system.current_quota_type), None) if not quota or quota.quota_limit == -1: return SettlementOwner.UNMETERED if quota.quota_type in {ProviderQuotaType.PAID, ProviderQuotaType.TRIAL}: if quota.quota_unit in {QuotaUnit.CREDITS, QuotaUnit.TIMES}: return SettlementOwner.SHARED_CREDIT_POOL return SettlementOwner.UNMETERED return SettlementOwner.PROVIDER_FREE def migration_routing_state(tenant_id: str) -> dict[str, Any] | None: # Do not use the legacy billing-source TTL cache for cutover admission. try: return get_routing_state(tenant_id) except Exception: # Never expose SQL/connection detail, or treat a failed lookup as a # legacy decision that could reopen admission after cutover. raise ModelRouteUnavailable from None def has_compatibility_route(state: dict[str, Any] | None) -> bool: if not state: return False phase = state.get("blocked_from") if state.get("phase") == "blocked" else state.get("phase") return phase in {"claimed", "granting", "activating", "active"} or ( phase == "cancelled" and state.get("claimed_at") is not None ) def compatibility_quota_type( provider: str, quotas: list[QuotaConfiguration], state: dict[str, Any] ) -> ProviderQuotaType: """Restore prepared ownership without consulting the now-closed legacy wallet.""" source_ownership = state.get("source_ownership") if not isinstance(source_ownership, dict): raise ModelRouteUnavailable ownership = source_ownership.get(str(ModelProviderID(provider))) if not isinstance(ownership, dict): raise ModelRouteUnavailable if "schema_version" in ownership and ownership.get("schema_version") != 2: raise ModelRouteUnavailable if ownership.get("schema_version") == 2: from services.model_billing_mapping_service import stable_compatibility_quota_type try: return stable_compatibility_quota_type(provider, quotas, state) except (KeyError, TypeError, ValueError): raise ModelRouteUnavailable from None try: selected = ProviderQuotaType(ownership["current_quota_type"]) for quota in quotas: if quota.quota_type == ProviderQuotaType.FREE: continue unmetered = ownership["quotas"][quota.quota_type.value]["unmetered"] if not isinstance(unmetered, bool): raise ValueError("Invalid provenance") # These are availability metadata, not a newly issued legacy quota. # Only a proven original unlimited pool may retain the -1 marker. quota.quota_limit = -1 if unmetered else 0 quota.quota_used = 0 quota.is_valid = unmetered current = next(q for q in quotas if q.quota_type == selected) except (KeyError, TypeError, ValueError, StopIteration): raise ModelRouteUnavailable from None if selected != ProviderQuotaType.FREE or current.is_valid: return selected # Independent provider-free allowance still follows its normal exhaustion # behavior. It may fall through to the migrated shared pool afterwards. from core.provider_manager import ProviderManager return ProviderManager._choice_current_using_quota_type(quotas) def migration_display_status(tenant_id: str) -> Literal["none", "preparing", "processing", "active"]: state = migration_routing_state(tenant_id) if not state: return "none" if state.get("phase") == "active" or ( state.get("phase") == "cancelled" and state.get("claimed_at") and state.get("ever_activated") ): return "active" return "processing" if has_compatibility_route(state) else "preparing" def model_mappings(mapping_version: str) -> dict[tuple[str, ModelType, str], ModelMapping]: try: return mappings_for_revision(dify_config.TOKENER_LEGACY_MODEL_MAPPING_JSON, mapping_version) except Exception: # Validation errors can contain config values. Do not log their payload. raise ModelRouteUnmapped from None @dataclass(frozen=True, slots=True) class SourceVariantInvocationBinding: """Per-variant evidence, absent for V1 logical siblings and old V1 plans.""" registry_digest: str variant_id: str source_settings_digest: str @dataclass(frozen=True, slots=True) class FrozenModelInvocation: tenant_id: str route_id: str logical_provider: str logical_model: str model_type: ModelType route_epoch: int mapping_revision: str credential_id: str expected_credential_fingerprint: str plugin_unique_identifier: str target_model: str operations: tuple[str, ...] settlement_backend: str = "tokener" tts_voice_contract: TTSVoiceContract | None = None source_parameter_contract: SourceParameterContract | None = None source_variant_binding: SourceVariantInvocationBinding | None = None # None is historical pure V1 only. A V2 selected revision pins every # logical model, including identity/adapter records with the V1 shape. mapping_revision_digest: str | None = None @property def cache_identity(self) -> str: identity = [ self.tenant_id, self.route_epoch, self.mapping_revision, self.credential_id, self.expected_credential_fingerprint, self.plugin_unique_identifier, self.target_model, self.model_type.value, ] if self.tts_voice_contract is not None: identity.extend( [ self.tts_voice_contract.adapter_id, self.tts_voice_contract.source_plugin_unique_identifier, self.tts_voice_contract.source_schema_digest, self.tts_voice_contract.rules_digest, ] ) if self.source_parameter_contract is not None: identity.extend( [ self.source_parameter_contract.adapter_id, self.source_parameter_contract.algorithm_revision, self.source_parameter_contract.profile_id, self.source_parameter_contract.source_model, self.source_parameter_contract.source_plugin_unique_identifier, self.source_parameter_contract.source_schema_digest, self.source_parameter_contract.target_model, self.source_parameter_contract.rules_digest, ] ) if self.source_parameter_contract.source_runtime_schema_digest is not None: # Old contract tokens are unchanged; new reviewed split-boundary # profiles additionally pin their exact source runtime schema. identity.append(self.source_parameter_contract.source_runtime_schema_digest) if self.mapping_revision_digest is not None: identity.append(self.mapping_revision_digest) if self.source_variant_binding is not None: identity.extend( [ self.source_variant_binding.registry_digest, self.source_variant_binding.variant_id or "", self.source_variant_binding.source_settings_digest or "", ] ) return hashlib.sha256(json.dumps(identity, separators=(",", ":")).encode()).hexdigest() class RoutedModelCredentials(dict[str, Any]): # noqa: FURB189 - graphon requires a real dict """In-process trusted capability, not a serializable credential protocol. An ordinary dict from a client cannot forge this type. Accidental generic serialization emits an empty credential object, never the source key. """ plan: FrozenModelInvocation @override def __init__(self, *args: Any, **kwargs: Any) -> None: if len(args) != 1 or kwargs or not isinstance(args[0], FrozenModelInvocation): raise TypeError("A trusted frozen model invocation is required") super().__init__() self.plan = args[0] def __bool__(self) -> bool: return True @override def copy(self) -> RoutedModelCredentials: return RoutedModelCredentials(self.plan) @dataclass(frozen=True, slots=True) class LegacyModelBinding: tenant_id: str logical_provider: str logical_model: str model_type: ModelType route_epoch: int class LegacyModelCredentials(dict[str, Any]): # noqa: FURB189 - runtime requires a dict """Server-created legacy provenance, retained until its reservation settles.""" binding: LegacyModelBinding @override def __init__(self, *args: Any, **kwargs: Any) -> None: if len(args) != 2 or kwargs or not isinstance(args[0], dict) or not isinstance(args[1], LegacyModelBinding): raise TypeError("Legacy credentials and a trusted model binding are required") super().__init__(args[0]) self.binding = args[1] @override def copy(self) -> LegacyModelCredentials: return LegacyModelCredentials(dict(self), self.binding) def bind_legacy_credentials( configuration: ProviderConfiguration, model_type: ModelType, model: str, values: dict[str, Any] ) -> dict[str, Any]: if ( settlement_owner(configuration.system_configuration, configuration.using_provider_type) != SettlementOwner.SHARED_CREDIT_POOL ): return values state = migration_routing_state(configuration.tenant_id) if has_compatibility_route(state): # Cutover between the route check and reading the old key must not # silently construct a legacy invocation after admission has closed. raise ModelInvocationReprepare return LegacyModelCredentials( values, LegacyModelBinding( tenant_id=configuration.tenant_id, logical_provider=str(ModelProviderID(configuration.provider.provider)), logical_model=model, model_type=model_type, route_epoch=int(state.get("route_epoch", 0)) if state else 0, ), ) def validate_invocation_admission( credentials: dict[str, Any], *, tenant_id: str, provider: str, model_type: ModelType, model: str ) -> None: """Validate the prepared snapshot immediately before reserving/invoking. Do not replace credentials here: schema, tokens, dispatch, and settlement must use the same binding. Already reserved calls do not re-enter this check. """ if isinstance(credentials, RoutedModelCredentials): binding = credentials.plan elif isinstance(credentials, LegacyModelCredentials): binding = credentials.binding else: return if (tenant_id, str(ModelProviderID(provider)), model_type, model) != ( binding.tenant_id, binding.logical_provider, binding.model_type, binding.logical_model, ): raise ModelRouteUnavailable state = migration_routing_state(tenant_id) if isinstance(credentials, LegacyModelCredentials): if has_compatibility_route(state) or (int(state.get("route_epoch", 0)) if state else 0) != binding.route_epoch: raise ModelInvocationReprepare elif ( state is None or state.get("phase") not in {"active", "cancelled"} or (state.get("phase") == "cancelled" and not state.get("ever_activated")) or int(state["route_epoch"]) != binding.route_epoch or state.get("model_mapping_version") != credentials.plan.mapping_revision or not uses_published_tokener_balance(state) ): raise ModelInvocationReprepare else: _validate_source_variant_admission(credentials.plan, state) def _validate_source_variant_admission(plan: FrozenModelInvocation, state: dict[str, Any]) -> None: """Recheck config/public facts before reservation, without daemon RPC/SQL. Fresh immutable bindings are attested at preparation/configuration. The protected install lifecycle, cohort boot freeze and old-task drain remain required; this is not a cross-daemon atomicity claim. Reserved invocations do not re-enter admission and keep their already frozen source semantics. """ from extensions import ext_hosting_provider pin = plan.source_variant_binding persisted_digest = state.get("source_variant_registry_digest") has_new_evidence = persisted_digest is not None or plan.mapping_revision_digest is not None or pin is not None try: registry = source_mapping_revision(dify_config.TOKENER_LEGACY_MODEL_MAPPING_JSON, plan.mapping_revision) except Exception: if has_new_evidence: raise ModelInvocationReprepare from None # Preserve historical pure-V1 admission semantics when no new state # authority or evidence exists. This never bypasses a persisted V2 pin. return if not registry.has_variants: if has_new_evidence: raise ModelInvocationReprepare return try: if ( registry.digest != plan.mapping_revision_digest or persisted_digest != plan.mapping_revision_digest or not uses_stable_dependency_inventory(dify_config.TOKENER_LEGACY_DEPENDENCY_INVENTORIES_JSON, state) ): raise ValueError("Source variant registry changed") identity = (plan.logical_provider, plan.model_type, plan.logical_model) declaration = registry.declarations.get(identity) if isinstance(declaration, ModelMapping): if pin is not None: raise ValueError("Unexpected source variant evidence") validate_frozen_mapping_evidence( declaration, provider=plan.logical_provider, model=plan.logical_model, model_type=plan.model_type, target_model=plan.target_model, operations=plan.operations, source_contract=plan.source_parameter_contract, tts_contract=plan.tts_voice_contract, ) return contract = plan.source_parameter_contract if contract is None or pin is None or pin.registry_digest != plan.mapping_revision_digest: raise ValueError("Missing source variant profile") hosting = ext_hosting_provider.hosting_configuration.provider_map.get(plan.logical_provider) if hosting is None or not hosting.enabled: raise ValueError("Source settings unavailable") selected = resolve_source_mapping( registry, identity, source_uid=contract.source_plugin_unique_identifier, fact=hosted_source_variant_fact( provider=plan.logical_provider, source_uid=contract.source_plugin_unique_identifier, models=(), credentials=hosting.credentials, ), embedding_fact=hosted_embedding_source_fact( provider=plan.logical_provider, source_uid=contract.source_plugin_unique_identifier, models=(), credentials=hosting.credentials, ), audio_fact=hosted_audio_source_fact( provider=plan.logical_provider, source_uid=contract.source_plugin_unique_identifier, models=(), credentials=hosting.credentials, ), xai_fact=hosted_xai_source_fact( provider=plan.logical_provider, source_uid=contract.source_plugin_unique_identifier, models=(), credentials=hosting.credentials, ), gemini_fact=hosted_gemini_source_fact( provider=plan.logical_provider, source_uid=contract.source_plugin_unique_identifier, models=(), credentials=hosting.credentials, ), tongyi_fact=hosted_tongyi_source_fact( provider=plan.logical_provider, source_uid=contract.source_plugin_unique_identifier, models=(), credentials=hosting.credentials, ), ) if ( not isinstance(selected, SELECTED_SOURCE_VARIANT_TYPES) or selected.variant.variant_id != pin.variant_id or canonical_hash(selected.variant.source_settings.model_dump(mode="json")) != pin.source_settings_digest ): raise ValueError("Source variant settings changed") validate_frozen_mapping_evidence( selected, provider=plan.logical_provider, model=plan.logical_model, model_type=plan.model_type, target_model=plan.target_model, operations=plan.operations, source_contract=plan.source_parameter_contract, tts_contract=plan.tts_voice_contract, ) except (KeyError, TypeError, ValueError): raise ModelInvocationReprepare from None def routed_credentials( configuration: ProviderConfiguration, model_type: ModelType, model: str ) -> RoutedModelCredentials | None: if ( settlement_owner(configuration.system_configuration, configuration.using_provider_type) != SettlementOwner.SHARED_CREDIT_POOL ): return None state = migration_routing_state(configuration.tenant_id) if not has_compatibility_route(state): return None if ( state is None or state.get("phase") not in {"active", "cancelled"} or (state.get("phase") == "cancelled" and not state.get("ever_activated")) ): raise ModelMigrationProcessing if not uses_published_tokener_balance(state): # A phase flag alone is not publication authority. Never expand model # access or reopen a legacy credential while the persisted cohort is missing. raise ModelRouteUnavailable provider = str(ModelProviderID(configuration.provider.provider)) mapping_version = state.get("model_mapping_version") if not isinstance(mapping_version, str): raise ModelRouteUnmapped registry = None try: source_uid = getattr(configuration.provider, "plugin_unique_identifier", "") if state.get("source_variant_registry_digest") is None: # Preserve the V1 dispatch owner and its exact legacy semantics. # A V2 declaration cannot pass this V1-only parser without a pin. mapping = model_mappings(mapping_version).get((provider, model_type, model)) if mapping is None or state.get("model_mapping_version") != mapping.mapping_revision: raise ValueError("Unmapped source identity") source_schemas = managed_source_schemas(configuration.system_configuration, configuration.provider) else: registry = source_mapping_revision(dify_config.TOKENER_LEGACY_MODEL_MAPPING_JSON, mapping_version) if ( not registry.has_variants or not uses_stable_dependency_inventory(dify_config.TOKENER_LEGACY_DEPENDENCY_INVENTORIES_JSON, state) or state.get("source_variant_registry_digest") != registry.digest ): raise ValueError("Source variant revision is not pinned") source_schemas = managed_source_schemas(configuration.system_configuration, configuration.provider) matching_sources = [ schema for schema in source_schemas if (schema.model_type, schema.model) == (model_type, model) ] mapping = resolve_source_mapping( registry, (provider, model_type, model), source_uid=source_uid, fact=configuration.system_configuration._tokener_source_variant_fact, embedding_fact=configuration.system_configuration._tokener_embedding_source_fact, audio_fact=configuration.system_configuration._tokener_audio_source_fact, xai_fact=configuration.system_configuration._tokener_xai_source_fact, gemini_fact=configuration.system_configuration._tokener_gemini_source_fact, tongyi_fact=configuration.system_configuration._tokener_tongyi_source_fact, schema=matching_sources[0] if len(matching_sources) == 1 else None, ) except (KeyError, TypeError, ValueError): raise ModelRouteUnmapped from None if state.get("model_mapping_version") != mapping.mapping_revision: raise ModelRouteUnmapped if not additional_model_is_authorized( system=configuration.system_configuration, installed_models=source_schemas, mapping=mapping, ): raise ModelRouteUnmapped tts_voice_contract = None source_parameter_contract = None if mapping.parameter_adapter.id != "identity_v1": matching_sources = [ schema for schema in source_schemas if (schema.model_type, schema.model) == (model_type, model) ] source_uid = getattr(configuration.provider, "plugin_unique_identifier", None) if len(matching_sources) != 1 or not isinstance(source_uid, str): raise ModelRouteUnmapped source_settings = configuration.system_configuration.credentials if mapping.parameter_adapter.id == "openai_chat_source_v1" and not isinstance(mapping, SelectedSourceVariant): fact = configuration.system_configuration._tokener_source_protocol_fact if ( fact is None or fact.provider != provider or fact.plugin_unique_identifier != source_uid or [digest for kind, name, digest in fact.catalog_schema_digests if (kind, name) == (model_type, model)] != [model_schema_digest(matching_sources[0])] or fact.selector not in {"absent", "chat", "responses"} ): raise ModelRouteUnmapped # The original hosted key remains inaccessible on this managed # route. Only positively captured, package-bound public protocol # settings reach the pure source adapter. source_settings = {"api_protocol": None if fact.selector == "absent" else fact.selector} try: contract = build_source_adapter_contract( mapping, source_provider=provider, source_plugin_unique_identifier=source_uid, source_schema=matching_sources[0], source_credentials=source_settings, ) except ValueError: raise ModelRouteUnmapped from None if isinstance(contract, TTSVoiceContract): tts_voice_contract = contract elif isinstance(contract, SourceParameterContract): source_parameter_contract = contract else: raise ModelRouteUnmapped with session_factory.create_session() as session: integration = session.scalar( select(TenantTokenerIntegration).where(TenantTokenerIntegration.tenant_id == configuration.tenant_id) ) if ( not integration or integration.status != TenantTokenerIntegrationStatus.READY or not integration.provider_credential_id or not integration.plugin_unique_identifier ): raise ModelRouteUnavailable if not mapping_supports_installed_plugin(mapping, integration.plugin_unique_identifier): raise ModelRouteUnavailable expected_credential_fingerprint = _pinned_credential_fingerprint( state, configuration.tenant_id, integration.provider_credential_id, ) plan = FrozenModelInvocation( tenant_id=configuration.tenant_id, route_id=str(uuid4()), logical_provider=provider, logical_model=model, model_type=model_type, route_epoch=int(state["route_epoch"]), mapping_revision=mapping.mapping_revision, credential_id=integration.provider_credential_id, expected_credential_fingerprint=expected_credential_fingerprint, plugin_unique_identifier=integration.plugin_unique_identifier, target_model=mapping.target.model, operations=tuple(mapping.operations), tts_voice_contract=tts_voice_contract, source_parameter_contract=source_parameter_contract, source_variant_binding=( SourceVariantInvocationBinding( registry.digest, mapping.variant.variant_id, canonical_hash(mapping.variant.source_settings.model_dump(mode="json")), ) if registry is not None and isinstance(mapping, SELECTED_SOURCE_VARIANT_TYPES) else None ), mapping_revision_digest=registry.digest if registry is not None else None, ) return RoutedModelCredentials(plan) def _valid_credential_fingerprint(value: Any) -> TypeGuard[str]: return ( isinstance(value, str) and value.startswith("sha256:") and len(value) == 71 and all(char in "0123456789abcdef" for char in value[7:]) ) def _credential_fingerprint(encrypted_config: str) -> str: return canonical_hash({"encrypted_config": encrypted_config}) def _pinned_credential_fingerprint( state: dict[str, Any], tenant_id: str, credential_id: str, *, allow_provisioned: bool = False ) -> str: pin = state.get("preparation") if not pin and allow_provisioned: pin = state.get("provisioned_binding") if not isinstance(pin, dict) or ( pin.get("tenant_id"), pin.get("provider_name"), pin.get("provider_credential_id"), ) != (tenant_id, "langgenius/tokener/tokener", credential_id): raise ModelRouteUnavailable fingerprint = pin.get("credential_fingerprint") if not _valid_credential_fingerprint(fingerprint): raise ModelRouteUnavailable return fingerprint def managed_credentials(plan: FrozenModelInvocation) -> dict[str, Any]: from core.entities import PluginCredentialType from core.helper import encrypter from core.helper.credential_utils import runtime_check_credential_policy_compliance provider = "langgenius/tokener/tokener" with session_factory.create_session() as session: credential = session.scalar( select(ProviderCredential).where( ProviderCredential.tenant_id == plan.tenant_id, ProviderCredential.provider_name == provider, ProviderCredential.id == plan.credential_id, ) ) if credential is None or not credential.encrypted_config: raise ModelRouteUnavailable encrypted = credential.encrypted_config if ( not isinstance(encrypted, str) or not _valid_credential_fingerprint(plan.expected_credential_fingerprint) or not hmac.compare_digest(_credential_fingerprint(encrypted), plan.expected_credential_fingerprint) ): # Editing a managed row cannot change where a frozen SYSTEM invocation # is billed. Fail before decrypting or sending any key to the plugin. raise ModelRouteUnavailable runtime_check_credential_policy_compliance( credential_id=plan.credential_id, provider=provider, credential_type=PluginCredentialType.MODEL ) try: values = json.loads(encrypted) if not isinstance(values, dict) or not isinstance(values.get("api_key"), str): raise ValueError("Missing managed key") values["api_key"] = encrypter.decrypt_token(tenant_id=plan.tenant_id, token=values["api_key"]) return values except Exception: raise ModelRouteUnavailable from None def redirect_payload(plan: FrozenModelInvocation, data: dict[str, Any]) -> dict[str, Any]: if ( data.get("model") != plan.logical_model or data.get("model_type") != plan.model_type.value or data.get("provider") != ModelProviderID(plan.logical_provider).provider_name ): raise ModelRouteUnavailable marker = None if plan.source_parameter_contract is not None: if plan.tts_voice_contract is not None: raise ModelRouteUnavailable try: marker = source_parameter_marker( plan.source_parameter_contract, source_provider=plan.logical_provider, source_model=plan.logical_model, model_type=plan.model_type, target_model=plan.target_model, ) except ValueError: raise ModelRouteUnavailable from None target_credentials = managed_credentials(plan) if "_dify_source_contract" in target_credentials: # A stored or user-supplied field is never authority for a source # profile. Only the server-frozen verified contract can create it. raise ModelRouteUnavailable if marker is not None: target_credentials = {**target_credentials, "_dify_source_contract": marker} return { "version": 1, "route_id": plan.route_id, "target": { "plugin_id": "langgenius/tokener", "provider": "tokener", "model_type": plan.model_type.value, "model": plan.target_model, "expected_unique_identifier": plan.plugin_unique_identifier, "credentials": target_credentials, }, } def embedding_dimensions_contract(schema: AIModelEntity) -> str: """Use actual parameter metadata; Graphon has no embedding-dimension property. A fixed-dimensional model without a dimensions rule needs an explicit conformance fixture in the registry; absence is not an invented dimension. """ rule = next((r for r in schema.parameter_rules if r.name == "dimensions"), None) if rule is None: return "fixed_by_model_fixture" contract = { "kind": "parameter", "type": rule.type.value, "default": rule.default, "min": rule.min, "max": rule.max, "options": list(rule.options), } # Parameter schemas legitimately contain null/float values; the migration # manifest's strict money-safe canonical JSON does not. Carry a digest of # the actual dimension rule instead of those schema values in the manifest. return ( "sha256:" + hashlib.sha256(json.dumps(contract, sort_keys=True, separators=(",", ":"), default=str).encode()).hexdigest() ) def migration_inventory( tenant_id: str, mapping_version: str, *, eligibility_policy_version: str | None = None ) -> dict[str, Any]: """Export the source inventory using the admitted/persisted migration policy. Omitted policy preserves strict historical behavior. The free-legacy policy may freeze a genuinely empty inventory, not missing source schema evidence. """ from core.plugin.entities.plugin import PluginInstallationSource from core.plugin.entities.plugin_daemon import PluginModelProviderDeclaration from core.plugin.impl.model import PluginModelClient from core.plugin.impl.model_runtime_factory import create_plugin_provider_manager registry = model_mappings(mapping_version) try: retirement_policy = retirement_policy_for_revision( dify_config.TOKENER_LEGACY_RETIRED_MODELS_JSON, mapping_version, registry ) source_capability_policy = resolve_source_capability_policy( dify_config.TOKENER_LEGACY_SOURCE_CAPABILITIES_JSON, mapping_version ) except ValueError: raise ModelRouteUnmapped from None client = PluginModelClient() source_bindings: dict[str, str] = {} parameter_profile_bindings: dict[str, str] = {} parameter_profile_providers = { provider for (provider, _kind, _model), rule in registry.items() if rule.parameter_adapter.id in {"anthropic_source_v1", "openai_chat_source_v1"} } evidence_providers = { item.provider_id for item in source_capability_policy.declarations } | parameter_profile_providers if evidence_providers: seen_bindings: set[str] = set() # A cached catalog is insufficient: require a fresh, unique immutable # installation. Mutable remote-debug code cannot satisfy this proof. for binding in client.fetch_model_provider_bindings(tenant_id): binding_provider = str( ModelProviderID( binding.provider if binding.provider.count("/") == 2 else f"{binding.plugin_id}/{binding.provider}" ) ) if binding_provider not in evidence_providers: continue if binding_provider in seen_bindings: raise ModelRouteUnavailable seen_bindings.add(binding_provider) if binding.source == PluginInstallationSource.Remote: continue if binding.runtime_type in {"local", "serverless"}: parameter_profile_bindings[binding_provider] = binding.plugin_unique_identifier declaration = source_capability_policy.for_package(binding_provider, binding.plugin_unique_identifier) # The old source-schema evidence was local-only. A v2 record can # opt into an immutable serverless package under a new revision. if declaration is not None and declaration.allows_runtime(binding.runtime_type): source_bindings[binding_provider] = binding.plugin_unique_identifier inventory = [] retired_inventory = [] unavailable_inventory = [] source_ownership = {} configurations = create_plugin_provider_manager(tenant_id=tenant_id).get_configurations(tenant_id) for provider_name, configuration in configurations: system = configuration.system_configuration if not system.enabled: continue if system.current_quota_type is None: raise ModelRouteUnavailable source_ownership[str(ModelProviderID(provider_name))] = { "current_quota_type": system.current_quota_type.value, "quotas": {q.quota_type.value: {"unmetered": q.quota_limit == -1} for q in system.quota_configurations}, } source_id = ModelProviderID(provider_name) def lookup( model_type: ModelType, model: str, credentials: dict[str, Any], *, source_id: ModelProviderID = source_id, source_plugin_unique_identifier: str | None = getattr( configuration.provider, "plugin_unique_identifier", None ), ) -> AIModelEntity | None: profile_uid = None if str(source_id) in parameter_profile_providers: profile_uid = source_plugin_unique_identifier if not isinstance(profile_uid, str) or parameter_profile_bindings.get(str(source_id)) != profile_uid: raise ModelRouteUnavailable return client.get_model_schema( tenant_id, None, source_id.plugin_id, source_id.provider_name, model_type.value, model, credentials, plugin_unique_identifier=profile_uid, ) try: preparation_system = system if source_capability_policy.declarations: source_uid = ( configuration.provider.plugin_unique_identifier if isinstance(configuration.provider, PluginModelProviderDeclaration) else None ) configured_models = frozenset( (item.model_type, item.model) for item in [*configuration.custom_configuration.models, *configuration.model_settings] ) unavailable = partition_known_unavailable_sources( system, configuration.provider.models, configuration._get_original_provider_configurate_methods(), configuration.provider.supported_model_types, configured_models, source_capability_policy.for_package(str(source_id), source_uid) if source_bindings.get(str(source_id)) == source_uid else None, ) preparation_system = unavailable.system for (kind, model), (proof_id, pools) in sorted(unavailable.unavailable.items()): unavailable_inventory.append( { "provider": str(source_id), "plugin_unique_identifier": source_uid, "model_type": kind.value, "model": model, "pool_types": sorted(pool.value for pool in pools), "reason": unavailable.unavailable_reasons[(kind, model)], "proof_id": proof_id, } ) partition = partition_hosted_migration_sources( preparation_system, configuration.provider.models, retirement_policy.for_provider(str(source_id)) ) sources = hosted_migration_sources( partition.system, partition.models, configuration.provider.supported_model_types, lookup ) except ValueError: raise ModelRouteUnmapped from None for (model_type, model), pools in sorted(partition.retired.items()): retired_inventory.append( { "provider": str(source_id), "model_type": model_type.value, "model": model, "pool_types": sorted(pool.value for pool in pools), } ) for model_type, model in sorted(sources, key=lambda pair: (pair[0].value, pair[1])): rule = registry.get((str(ModelProviderID(provider_name)), model_type, model)) if rule is None or rule.mapping_revision != mapping_version: raise ModelRouteUnmapped source = sources[(model_type, model)] source_schema = source.schema if not source.pool_types <= set(rule.pool_types): raise ModelRouteUnmapped if ( model_type == ModelType.LLM and source_schema.model_properties.get(ModelPropertyKey.MODE) != rule.source.mode ): raise ModelRouteUnmapped source_digest = model_schema_digest(source_schema) if source_digest != rule.source_schema_digest: raise ModelRouteUnmapped required_ops = {"model/schema"} if model_type == ModelType.LLM: required_ops |= {"llm/invoke", "llm/num_tokens"} elif model_type == ModelType.TEXT_EMBEDDING: required_ops |= {"text_embedding/invoke", "text_embedding/num_tokens"} else: required_ops.add(f"{model_type.value}/invoke") if not required_ops <= set(rule.operations): raise ModelRouteUnmapped item: dict[str, Any] = { "provider": str(ModelProviderID(provider_name)), "model_type": model_type.value, "model": model, "mapping_revision": rule.mapping_revision, "source_schema_digest": rule.source_schema_digest, "target_model": rule.target.model, "source_mode": rule.source.mode, } if model_type == ModelType.TEXT_EMBEDDING: item["embedding_dimensions"] = embedding_dimensions_contract(source_schema) if rule.parameter_adapter.id != "identity_v1": source_uid = getattr(configuration.provider, "plugin_unique_identifier", None) if not isinstance(source_uid, str): raise ModelRouteUnmapped if ( rule.parameter_adapter.id in {"anthropic_source_v1", "openai_chat_source_v1"} and parameter_profile_bindings.get(str(source_id)) != source_uid ): # A cached historical declaration cannot attest today's # immutable source installation at preparation time. raise ModelRouteUnavailable try: contract = build_source_adapter_contract( rule, source_provider=str(source_id), source_plugin_unique_identifier=source_uid, source_schema=source_schema, source_credentials=system.credentials, ) except ValueError: raise ModelRouteUnmapped from None if contract is None: raise ModelRouteUnmapped item["source_adapter_rules_digest"] = contract.rules_digest if isinstance(contract, SourceParameterContract): if contract.source_runtime_schema_digest is not None: # Management catalogs and SDK-rendered model/schema RPCs # are distinct owners. Verify the actual original source # runtime under its fresh immutable UID before funding; # a static catalog cannot stand in for this evidence. if system.credentials is None: raise ModelRouteUnmapped runtime_schema = client.get_model_schema( tenant_id, None, source_id.plugin_id, source_id.provider_name, model_type.value, model, system.credentials, plugin_unique_identifier=source_uid, ) if ( runtime_schema is None or runtime_schema.model != model or runtime_schema.model_type != model_type or not hmac.compare_digest( model_schema_digest(runtime_schema), contract.source_runtime_schema_digest ) ): raise ModelRouteUnmapped item["source_runtime_schema_digest"] = contract.source_runtime_schema_digest item["source_parameter_contract"] = source_parameter_marker( contract, source_provider=str(source_id), source_model=model, model_type=model_type, target_model=rule.target.model, ) public_settings = source_parameter_public_settings(contract, source_credentials=system.credentials) if public_settings is not None: item["source_parameter_settings"] = public_settings inventory.append(item) if ( not inventory and not retired_inventory and not migration_policy_allows_empty_sources(eligibility_policy_version) ): raise ModelRouteUnmapped inventory.sort(key=itemgetter("provider", "model_type", "model")) retired_inventory.sort(key=itemgetter("provider", "model_type", "model")) unavailable_inventory.sort(key=itemgetter("provider", "model_type", "model")) result: dict[str, Any] = { "inventory_hash": canonical_hash( source_capability_policy.inventory_payload( retirement_policy.inventory_payload(inventory, retired_inventory), unavailable_inventory ) ), "model_mapping_version": mapping_version, "model_count": len(inventory), "models": inventory, "source_ownership": source_ownership, } if retirement_policy.identities: result["retired_sources"] = retired_inventory result["retirement_policy_digest"] = retirement_policy.digest if source_capability_policy.declarations: result["unavailable_sources"] = unavailable_inventory result["source_capability_policy_digest"] = source_capability_policy.digest return result def validate_migration_readiness( tenant_id: str, mapping_version: str, *, eligibility_policy_version: str | None = None ) -> dict[str, Any]: """Check adapters/registry under the persisted policy; never invoke a model. Omitted policy remains strict for historical paid callers. Even an admitted empty free inventory must validate the daemon and managed credential pin. """ from packaging.version import Version from core.plugin.impl.model import PluginModelClient inventory = migration_inventory(tenant_id, mapping_version, eligibility_policy_version=eligibility_policy_version) registry = model_mappings(mapping_version) client = PluginModelClient() capabilities = client._request_with_plugin_daemon_response( "GET", f"plugin/{tenant_id}/dispatch/redirect/v1/capabilities", dict[str, Any] ) if "model-redirect/v1" not in capabilities.get("protocols", []): raise ModelRouteUnavailable operations = set(capabilities.get("operations", [])) state = migration_routing_state(tenant_id) if state is None: raise ModelRouteUnavailable with session_factory.create_session() as session: integration = session.scalar( select(TenantTokenerIntegration).where(TenantTokenerIntegration.tenant_id == tenant_id) ) if not integration or not integration.provider_credential_id or not integration.plugin_unique_identifier: raise ModelRouteUnavailable credential_id = integration.provider_credential_id plugin_unique_identifier = integration.plugin_unique_identifier expected_credential_fingerprint = _pinned_credential_fingerprint( state, tenant_id, credential_id, allow_provisioned=True, ) credential = session.scalar( select(ProviderCredential).where( ProviderCredential.tenant_id == tenant_id, ProviderCredential.provider_name == "langgenius/tokener/tokener", ProviderCredential.id == credential_id, ) ) if ( credential is None or not isinstance(credential.encrypted_config, str) or not hmac.compare_digest( _credential_fingerprint(credential.encrypted_config), expected_credential_fingerprint ) ): raise ModelRouteUnavailable actual_version = plugin_unique_identifier.split(":", 1)[-1].split("@", 1)[0] for source in inventory["models"]: model_type = ModelType(source["model_type"]) rule = registry[(source["provider"], model_type, source["model"])] if not set(rule.operations) <= operations: raise ModelRouteUnavailable if Version(actual_version) < Version(rule.target.minimum_plugin_version): raise ModelRouteUnavailable source_parameter_contract = None if rule.parameter_adapter.id not in {"identity_v1", "tts_source_voice_v1"}: try: source_parameter_contract = restore_source_parameter_contract( source.get("source_parameter_contract"), mapping=rule, source_public_settings=source.get("source_parameter_settings"), ) except ValueError: raise ModelRouteUnmapped from None plan = FrozenModelInvocation( tenant_id=tenant_id, route_id=str(uuid4()), logical_provider=source["provider"], logical_model=source["model"], model_type=model_type, route_epoch=0, mapping_revision=mapping_version, credential_id=credential_id, expected_credential_fingerprint=expected_credential_fingerprint, plugin_unique_identifier=plugin_unique_identifier, target_model=rule.target.model, operations=tuple(rule.operations), source_parameter_contract=source_parameter_contract, ) target_schema = client.get_model_schema( tenant_id, None, ModelProviderID(source["provider"]).plugin_id, ModelProviderID(source["provider"]).provider_name, model_type.value, source["model"], RoutedModelCredentials(plan), plugin_unique_identifier=( source_parameter_contract.source_plugin_unique_identifier if source_parameter_contract is not None else None ), ) if target_schema is None or target_schema.model_type != model_type: raise ModelRouteUnmapped if source_parameter_contract is not None: # A minimum version/mode match is not proof that the target # understands the private source profile. Native schema fallback # must fail before funds are issued. The target may report its # physical model ID, but every other source field must match. logical_schema = target_schema.model_copy(update={"model": source["model"]}) expected_schema_digest = ( source_parameter_contract.source_runtime_schema_digest or source_parameter_contract.source_schema_digest ) if ( source_parameter_contract.source_runtime_schema_digest is not None and source.get("source_runtime_schema_digest") != source_parameter_contract.source_runtime_schema_digest ): raise ModelRouteUnmapped if not hmac.compare_digest(model_schema_digest(logical_schema), expected_schema_digest): raise ModelRouteUnmapped if ( model_type == ModelType.LLM and target_schema.model_properties.get(ModelPropertyKey.MODE) != rule.source.mode ): raise ModelRouteUnmapped if model_type == ModelType.TEXT_EMBEDDING and source["embedding_dimensions"] != embedding_dimensions_contract( target_schema ): raise ModelRouteUnmapped return { **inventory, "daemon_protocol": "model-redirect/v1", "credential_fingerprint": expected_credential_fingerprint, }