mirror of
https://github.com/langgenius/dify.git
synced 2026-11-13 06:41:07 +08:00
704 lines
29 KiB
Python
704 lines
29 KiB
Python
"""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 (
|
|
additional_model_is_authorized,
|
|
hosted_migration_sources,
|
|
mapping_supports_installed_plugin,
|
|
mappings_for_revision,
|
|
model_schema_digest,
|
|
partition_hosted_migration_sources,
|
|
provider_has_prepared_ownership,
|
|
retirement_policy_for_revision,
|
|
)
|
|
|
|
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
|
|
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 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"
|
|
|
|
@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,
|
|
]
|
|
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
|
|
):
|
|
raise ModelInvocationReprepare
|
|
|
|
|
|
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
|
|
provider = str(ModelProviderID(configuration.provider.provider))
|
|
mapping_version = state.get("model_mapping_version")
|
|
if not isinstance(mapping_version, str):
|
|
raise ModelRouteUnmapped
|
|
mapping = model_mappings(mapping_version).get((provider, model_type, model))
|
|
if mapping is None or configuration.system_configuration.current_quota_type not in mapping.pool_types:
|
|
raise ModelRouteUnmapped
|
|
if state.get("model_mapping_version") != mapping.mapping_revision:
|
|
raise ModelRouteUnmapped
|
|
additional_provider = not provider_has_prepared_ownership(provider, state)
|
|
if additional_provider and not additional_model_is_authorized(
|
|
system=configuration.system_configuration,
|
|
installed_models=configuration.provider.models,
|
|
mapping=mapping,
|
|
):
|
|
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 additional_provider and 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),
|
|
)
|
|
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
|
|
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": managed_credentials(plan),
|
|
},
|
|
}
|
|
|
|
|
|
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) -> dict[str, Any]:
|
|
"""Export a non-secret source inventory before preparing a target plugin."""
|
|
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
|
|
)
|
|
except ValueError:
|
|
raise ModelRouteUnmapped from None
|
|
client = PluginModelClient()
|
|
inventory = []
|
|
retired_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,
|
|
) -> AIModelEntity | None:
|
|
return client.get_model_schema(
|
|
tenant_id,
|
|
None,
|
|
source_id.plugin_id,
|
|
source_id.provider_name,
|
|
model_type.value,
|
|
model,
|
|
credentials,
|
|
)
|
|
|
|
try:
|
|
partition = partition_hosted_migration_sources(
|
|
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 = {
|
|
"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)
|
|
inventory.append(item)
|
|
if not inventory and not retired_inventory:
|
|
raise ModelRouteUnmapped
|
|
inventory.sort(key=itemgetter("provider", "model_type", "model"))
|
|
retired_inventory.sort(key=itemgetter("provider", "model_type", "model"))
|
|
result: dict[str, Any] = {
|
|
"inventory_hash": canonical_hash(retirement_policy.inventory_payload(inventory, retired_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
|
|
return result
|
|
|
|
|
|
def validate_migration_readiness(tenant_id: str, mapping_version: str) -> dict[str, Any]:
|
|
"""Check our installed adapters/registry; never invoke a billable model."""
|
|
from packaging.version import Version
|
|
|
|
from core.plugin.impl.model import PluginModelClient
|
|
|
|
inventory = migration_inventory(tenant_id, mapping_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
|
|
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),
|
|
)
|
|
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),
|
|
)
|
|
if target_schema is None or target_schema.model_type != model_type:
|
|
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,
|
|
}
|