feat: support exact Gemini 0.9.13 credit source variants

api/services/entities/model_billing_gemini_v8_mapping.py

@@ -0,0 +1,159 @@
+"""Closed V8 Gemini source shapes, separate from immutable V6 declarations.
+
+This DTO validates declaration shape only. A syntactically valid digest is not
+reviewed source, SDK/runtime, installed package or target capability evidence.
+No registry dispatch, factory, credential, SYSTEM/BYOK or upgrade authority is
+added here. Real 0913 contracts must be reviewed before this family is wired.
+"""
+
+from dataclasses import dataclass
+from typing import Annotated, Literal, Self
+
+from pydantic import Field, model_validator
+
+from services.entities.model_billing_gemini_mapping import (
+    GEMINI_SOURCE_UID,
+    GeminiMappingSource,
+    GeminiMappingTarget,
+    GeminiSourceSettings,
+    gemini_source_operations,
+)
+from services.entities.model_billing_mapping import _StrictModel
+
+GEMINI_0913_SOURCE_UID = "langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458"
+
+type GeminiV8SourceUID = Literal[
+    "langgenius/gemini:0.9.12@9bb9eff3f9c99a8f1f87bdd5e0f178d5493918939dd0e02825ae8383bd0fd7ed",
+    "langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+]
+
+
+class GeminiV8SourceAdapter(_StrictModel):
+    id: Literal["gemini_credit_source_v1", "gemini_credit_source_v2"]
+    revision: Literal["1"]
+    rules_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
+
+
+class GeminiV8SourceVariant(_StrictModel):
+    variant_id: str = Field(min_length=1)
+    source_plugin_uid: GeminiV8SourceUID
+    source_settings: GeminiSourceSettings
+    source_schema_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
+    source_runtime_schema_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
+    source_profile_id: str = Field(min_length=1)
+    parameter_adapter: GeminiV8SourceAdapter
+    target: GeminiMappingTarget
+    operations: list[str] = Field(min_length=3, max_length=4)
+    conformance_fixture_ids: list[Annotated[str, Field(min_length=1)]] = Field(min_length=1)
+
+    @model_validator(mode="after")
+    def closed_capability(self) -> Self:
+        prefix, adapter = (
+            ("gemini0912-native-v1:", "gemini_credit_source_v1")
+            if self.source_plugin_uid == GEMINI_SOURCE_UID
+            else ("gemini0913-native-v1:", "gemini_credit_source_v2")
+        )
+        if (
+            self.source_profile_id != prefix + self.target.model
+            or self.parameter_adapter.id != adapter
+            or not self.source_settings.matches_model_type(self.target.model_type)
+            or set(self.operations) != gemini_source_operations(self.target.model)
+            or len(set(self.operations)) != len(self.operations)
+            or len(set(self.conformance_fixture_ids)) != len(self.conformance_fixture_ids)
+        ):
+            raise ValueError("Invalid bounded V8 Gemini source capability")
+        return self
+
+
+class SourceGeminiV8VariantMapping(_StrictModel):
+    schema_version: Literal[8]
+    mapping_revision: str = Field(min_length=1)
+    source: GeminiMappingSource
+    settlement_owner: Literal["shared_credit_pool"]
+    pool_types: list[Literal["paid", "trial"]] = Field(min_length=1)
+    variants: list[GeminiV8SourceVariant] = Field(min_length=1, max_length=2)
+
+    @model_validator(mode="before")
+    @classmethod
+    def exact_version(cls, value: object) -> object:
+        if type(value) is not dict or type(value.get("schema_version")) is not int or value["schema_version"] != 8:
+            raise ValueError("Gemini schema version must be integer 8")
+        return value
+
+    @model_validator(mode="after")
+    def finite_exact_sources(self) -> Self:
+        if len(set(self.pool_types)) != len(self.pool_types):
+            raise ValueError("Duplicate V8 Gemini pools")
+        if len({item.variant_id for item in self.variants}) != len(self.variants):
+            raise ValueError("Duplicate V8 Gemini variant identity")
+        selectors = {(item.source_plugin_uid, item.source_settings.model_dump_json()) for item in self.variants}
+        if len(selectors) != len(self.variants):
+            raise ValueError("Ambiguous V8 Gemini source selector")
+        if any(
+            item.target.model != self.source.model or item.target.model_type is not self.source.model_type
+            for item in self.variants
+        ):
+            raise ValueError("V8 Gemini source/target identity changed")
+        return self
+
+
+class SourceGeminiV8VariantSelection(_StrictModel):
+    schema_version: Literal[8]
+    variant_id: str = Field(min_length=1)
+    source_plugin_uid: GeminiV8SourceUID
+    source_settings: GeminiSourceSettings
+    source_schema_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
+    source_runtime_schema_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
+    registry_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
+
+    @model_validator(mode="before")
+    @classmethod
+    def exact_version(cls, value: object) -> object:
+        if type(value) is not dict or type(value.get("schema_version")) is not int or value["schema_version"] != 8:
+            raise ValueError("Gemini schema version must be integer 8")
+        return value
+
+
+@dataclass(frozen=True, slots=True)
+class SelectedGeminiV8SourceVariant:
+    """Future registry-selected wrapper; constructing it grants no authority."""
+
+    envelope: SourceGeminiV8VariantMapping
+    variant: GeminiV8SourceVariant
+    registry_digest: str
+
+    @property
+    def mapping_revision(self) -> str:
+        return self.envelope.mapping_revision
+
+    @property
+    def source(self) -> GeminiMappingSource:
+        return self.envelope.source
+
+    @property
+    def settlement_owner(self) -> Literal["shared_credit_pool"]:
+        return self.envelope.settlement_owner
+
+    @property
+    def pool_types(self) -> list[Literal["paid", "trial"]]:
+        return self.envelope.pool_types
+
+    @property
+    def source_schema_digest(self) -> str:
+        return self.variant.source_schema_digest
+
+    @property
+    def parameter_adapter(self) -> GeminiV8SourceAdapter:
+        return self.variant.parameter_adapter
+
+    @property
+    def target(self) -> GeminiMappingTarget:
+        return self.variant.target
+
+    @property
+    def operations(self) -> list[str]:
+        return self.variant.operations
+
+    @property
+    def conformance_fixture_ids(self) -> list[str]:
+        return self.variant.conformance_fixture_ids

api/services/model_billing_gemini_credit_source_v2_adapter.py

@@ -0,0 +1,317 @@
+"""Pure exact Gemini0913 source factory, separate from immutable0912/V6.
+
+Pins bind the frozen v2 packet fd9d7680 / raw8c32f42e to the actual Source
+catalog/runtime Core entities. They prove no installed target, current SDK,
+network, credential, migration, billing or release authority. None runtime
+schema remains frozen/prospective restoration, never observed RPC evidence.
+"""
+
+from __future__ import annotations
+
+import hmac
+import re
+from dataclasses import dataclass
+from typing import Never
+
+from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelPropertyKey, ModelType
+from services.entities.model_billing_gemini_mapping import (
+    GeminiSourceSettings,
+    gemini_source_model_type,
+)
+from services.entities.model_billing_gemini_v8_mapping import GEMINI_0913_SOURCE_UID
+from services.entities.model_billing_mapping import SourceParameterContract, model_schema_digest
+
+GEMINI_CREDIT_SOURCE_ADAPTER_ID = "gemini_credit_source_v2"
+GEMINI_CREDIT_SOURCE_ADAPTER_REVISION = "1"
+GEMINI_CREDIT_SOURCE_PROVIDER = "langgenius/gemini/google"
+
+
+class GeminiCreditSourceV2AdapterError(ValueError):
+    """Fixed safe denial; unknown settings/schema never authorize a route."""
+
+
+@dataclass(frozen=True, slots=True)
+class GeminiCreditSourceV2Contract(SourceParameterContract):
+    """An exact immutable reviewed profile, not a routing authorization."""
+
+
+_REVIEWED_CONTRACTS = (
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-2.5-flash",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-2.5-flash",
+        source_schema_digest="sha256:9dfef26ae8e27359f74312f87ca295e08ec59f187c889d4fc5eed747f22682ef",
+        source_runtime_schema_digest="sha256:e3a9e5a6146b14480e5f9948993afeab62e640b1559bd953fea98651ee1d5a21",
+        target_model="gemini-2.5-flash",
+        rules_digest="sha256:13ca4eb8ba52db4537a9177114d1b41f5a468ee79ecbdbc65b6ad940979d79ce",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-3.1-flash-lite",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-3.1-flash-lite",
+        source_schema_digest="sha256:26dc99d59bdab5b1ff139aad894dae0fc15c9baf06777af67b9462f113193504",
+        source_runtime_schema_digest="sha256:0ee6cea0716356f3d3951066ec994ba34b50cf48fecf8d07c9e9611861e627a4",
+        target_model="gemini-3.1-flash-lite",
+        rules_digest="sha256:53574a2d8920b48b3152f97fccc70217e831dfba10d60c97690f9862ac2123ce",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-3.1-pro-preview",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-3.1-pro-preview",
+        source_schema_digest="sha256:9aceca99ac88649944f9d78102241853c8144d1a11f4c086bb5f27bc38766e34",
+        source_runtime_schema_digest="sha256:3a2ce7dce3477bd042314941826492a5496aeb8bdb047285514a305d810e918e",
+        target_model="gemini-3.1-pro-preview",
+        rules_digest="sha256:8fb951c69113cc92eedb5e564817c897804b151f089851e783b0fd2dcc68a3eb",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-3.5-flash",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-3.5-flash",
+        source_schema_digest="sha256:fa5fbdb6fa5ef69496f62f6dc2f4ead6b34d82f26cb6fc793fabf25577b667aa",
+        source_runtime_schema_digest="sha256:54aefc309ce57ff15f93ee606bb0919f1eb9649d057957cb6ea1fb2b60a3ab44",
+        target_model="gemini-3.5-flash",
+        rules_digest="sha256:cd75ed07076e1eebc099e93ada8110754bfd95767f05b1099b688f54db61b1b3",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-3.5-flash-lite",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-3.5-flash-lite",
+        source_schema_digest="sha256:230f022992a7882a800fc31d4b754ca51032a2342a4644ce250f58abc4bf07c9",
+        source_runtime_schema_digest="sha256:613acb7a29a51dffcf3ffd506f306ba6002599d5765a1baffb66d3ade9257355",
+        target_model="gemini-3.5-flash-lite",
+        rules_digest="sha256:601f024ab221e97fc8a80381e068ec3cd948704c57d88d722363ae59a22aad6f",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-3.7-flash",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-3.7-flash",
+        source_schema_digest="sha256:c718fa257170b98e87fd7a0c00c56613cb611d4d18ad1256dd375c9eaf908676",
+        source_runtime_schema_digest="sha256:7b1cd0a9ea9fbe14386d8458a6bec68833343cdbf4383c460bf0ac2a20f68189",
+        target_model="gemini-3.7-flash",
+        rules_digest="sha256:48ea4bb566bf12eb497eea6e23c8360c1fc3635b9ba9582a829a82c2ed5cfe70",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-3-flash-preview",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-3-flash-preview",
+        source_schema_digest="sha256:7a0371459f73f0ba41d20f0ef274923e131032931fb55c9865c7478b42a10f2b",
+        source_runtime_schema_digest="sha256:ccaa15b0d9e20840ab62d077c5c0a62fd9f5339d04f4d4ab0a978962ac83dab1",
+        target_model="gemini-3-flash-preview",
+        rules_digest="sha256:1d608a9072491a4e052a54b3447fce10cd0d6b1083a49fa70e37925736e20a28",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-embedding-001",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-embedding-001",
+        source_schema_digest="sha256:470e3397c5097df3b54673901fc321ce298e6081369bc731432e042daa04fb5d",
+        source_runtime_schema_digest="sha256:470e3397c5097df3b54673901fc321ce298e6081369bc731432e042daa04fb5d",
+        target_model="gemini-embedding-001",
+        rules_digest="sha256:c2d37d9c263af1eeedce8e2c94cb648f995bf1c0c83cd9c7d69baf70bab05d4e",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+    GeminiCreditSourceV2Contract(
+        profile_id="gemini0913-native-v1:gemini-embedding-2-preview",
+        source_plugin_unique_identifier="langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+        source_model="gemini-embedding-2-preview",
+        source_schema_digest="sha256:b16129e398a864dd0baee0c957c4c42573ef888b785c6aaec9d78ac2785e063b",
+        source_runtime_schema_digest="sha256:b16129e398a864dd0baee0c957c4c42573ef888b785c6aaec9d78ac2785e063b",
+        target_model="gemini-embedding-2-preview",
+        rules_digest="sha256:e13153f791fe3f15ea8cf2c7e71df72a76e2046e876d37c66e0e3fe2785a540a",
+        adapter_id="gemini_credit_source_v2",
+        algorithm_revision="1",
+    ),
+)
+_MARKER_FIELDS = frozenset(
+    {
+        "schema_version",
+        "profile_id",
+        "source_plugin_uid",
+        "source_model",
+        "source_schema_digest",
+        "rules_digest",
+    }
+)
+
+
+def _deny() -> Never:
+    raise GeminiCreditSourceV2AdapterError("Gemini source evidence is not reviewed")
+
+
+def _digest(value: object) -> bool:
+    return type(value) is str and re.fullmatch(r"sha256:[0-9a-f]{64}", value) is not None
+
+
+def validate_gemini_credit_source_identity(
+    *,
+    source_provider: str,
+    source_model: str,
+    source_plugin_unique_identifier: str,
+    model_type: ModelType,
+    target_model: str,
+) -> None:
+    if (
+        type(source_provider) is not str
+        or source_provider != GEMINI_CREDIT_SOURCE_PROVIDER
+        or type(source_plugin_unique_identifier) is not str
+        or source_plugin_unique_identifier != GEMINI_0913_SOURCE_UID
+        or type(source_model) is not str
+        or type(target_model) is not str
+        or source_model != target_model
+    ):
+        _deny()
+    try:
+        expected = gemini_source_model_type(source_model)
+    except ValueError:
+        _deny()
+    else:
+        if model_type is not expected:
+            _deny()
+
+
+def _settings(settings: GeminiSourceSettings, model: str) -> None:
+    if type(settings) is not GeminiSourceSettings or not settings.matches_model_type(gemini_source_model_type(model)):
+        _deny()
+
+
+def _schema_digest(schema: AIModelEntity, model: str) -> str:
+    if (
+        not isinstance(schema, AIModelEntity)
+        or schema.model != model
+        or schema.model_type is not gemini_source_model_type(model)
+        or schema.model_type is ModelType.LLM
+        and schema.model_properties.get(ModelPropertyKey.MODE) != "chat"
+    ):
+        _deny()
+    return model_schema_digest(schema)
+
+
+def _runtime_pin(contract: GeminiCreditSourceV2Contract, expected: str, observed: AIModelEntity | None) -> None:
+    if not _digest(expected) or not hmac.compare_digest(expected, contract.source_runtime_schema_digest or ""):
+        _deny()
+    # None is frozen/prospective restoration, NEVER observed runtime evidence.
+    if observed is not None and not hmac.compare_digest(_schema_digest(observed, contract.source_model), expected):
+        _deny()
+
+
+def gemini_credit_source_marker(
+    contract: GeminiCreditSourceV2Contract,
+    *,
+    source_provider: str,
+    source_model: str,
+    model_type: ModelType,
+    target_model: str,
+) -> dict[str, object]:
+    if type(contract) is not GeminiCreditSourceV2Contract or contract not in _REVIEWED_CONTRACTS:
+        _deny()
+    validate_gemini_credit_source_identity(
+        source_provider=source_provider,
+        source_model=source_model,
+        source_plugin_unique_identifier=contract.source_plugin_unique_identifier,
+        model_type=model_type,
+        target_model=target_model,
+    )
+    if source_model != contract.source_model or target_model != contract.target_model:
+        _deny()
+    return {
+        "schema_version": 1,
+        "profile_id": contract.profile_id,
+        "source_plugin_uid": contract.source_plugin_unique_identifier,
+        "source_model": contract.source_model,
+        "source_schema_digest": contract.source_schema_digest,
+        "rules_digest": contract.rules_digest,
+    }
+
+
+def build_gemini_credit_source_contract(
+    *,
+    adapter_id: str,
+    adapter_revision: str,
+    expected_rules_digest: str,
+    source_provider: str,
+    source_model: str,
+    source_plugin_unique_identifier: str,
+    source_schema: AIModelEntity,
+    expected_source_schema_digest: str,
+    expected_source_runtime_schema_digest: str,
+    target_model: str,
+    source_settings: GeminiSourceSettings,
+    source_runtime_schema: AIModelEntity | None = None,
+) -> GeminiCreditSourceV2Contract:
+    if not isinstance(source_schema, AIModelEntity):
+        _deny()
+    validate_gemini_credit_source_identity(
+        source_provider=source_provider,
+        source_model=source_model,
+        source_plugin_unique_identifier=source_plugin_unique_identifier,
+        model_type=source_schema.model_type,
+        target_model=target_model,
+    )
+    _settings(source_settings, source_model)
+    actual = _schema_digest(source_schema, source_model)
+    contract = next((row for row in _REVIEWED_CONTRACTS if row.source_model == source_model), None)
+    if (
+        contract is None
+        or type(adapter_id) is not str
+        or adapter_id != GEMINI_CREDIT_SOURCE_ADAPTER_ID
+        or type(adapter_revision) is not str
+        or adapter_revision != GEMINI_CREDIT_SOURCE_ADAPTER_REVISION
+        or not _digest(expected_rules_digest)
+        or not _digest(expected_source_schema_digest)
+        or not hmac.compare_digest(expected_rules_digest, contract.rules_digest)
+        or not hmac.compare_digest(expected_source_schema_digest, actual)
+        or not hmac.compare_digest(actual, contract.source_schema_digest)
+    ):
+        _deny()
+    _runtime_pin(contract, expected_source_runtime_schema_digest, source_runtime_schema)
+    return contract
+
+
+def restore_gemini_credit_source_contract(
+    marker: object,
+    *,
+    source_provider: str,
+    source_model: str,
+    model_type: ModelType,
+    target_model: str,
+    source_settings: GeminiSourceSettings,
+    expected_source_runtime_schema_digest: str,
+    source_runtime_schema: AIModelEntity | None = None,
+) -> GeminiCreditSourceV2Contract:
+    if (
+        type(marker) is not dict
+        or set(marker) != _MARKER_FIELDS
+        or type(marker.get("schema_version")) is not int
+        or marker.get("schema_version") != 1
+        or any(type(value) is not str for key, value in marker.items() if key != "schema_version")
+    ):
+        _deny()
+    contract = next((row for row in _REVIEWED_CONTRACTS if row.profile_id == marker["profile_id"]), None)
+    if contract is None:
+        _deny()
+    expected = gemini_credit_source_marker(
+        contract,
+        source_provider=source_provider,
+        source_model=source_model,
+        model_type=model_type,
+        target_model=target_model,
+    )
+    if marker != expected:
+        _deny()
+    _settings(source_settings, source_model)
+    _runtime_pin(contract, expected_source_runtime_schema_digest, source_runtime_schema)
+    return contract

api/services/model_billing_mapping_service.py

@@ -43,6 +43,10 @@ from services.entities.model_billing_gemini_mapping import (
     gemini_source_operations,
 )
 from services.entities.model_billing_gemini_mapping import SelectedGeminiSourceVariant as SelectedGeminiSourceVariant  # noqa: PLC0414
+from services.entities.model_billing_gemini_v8_mapping import GEMINI_0913_SOURCE_UID, SourceGeminiV8VariantMapping
+from services.entities.model_billing_gemini_v8_mapping import (
+    SelectedGeminiV8SourceVariant as SelectedGeminiV8SourceVariant,  # noqa: PLC0414
+)
 from services.entities.model_billing_mapping import ModelMapping, SourceVariantSettings
 from services.entities.model_billing_mapping import SelectedSourceVariant as SelectedSourceVariant  # noqa: PLC0414
 from services.entities.model_billing_mapping import SourceParameterContract as SourceParameterContract  # noqa: PLC0414
@@ -69,6 +73,16 @@ from services.model_billing_gemini_credit_source_adapter import (
     gemini_credit_source_marker,
     restore_gemini_credit_source_contract,
 )
+from services.model_billing_gemini_credit_source_v2_adapter import GeminiCreditSourceV2Contract
+from services.model_billing_gemini_credit_source_v2_adapter import (
+    build_gemini_credit_source_contract as build_gemini_credit_v2_source_contract,
+)
+from services.model_billing_gemini_credit_source_v2_adapter import (
+    gemini_credit_source_marker as gemini_credit_v2_source_marker,
+)
+from services.model_billing_gemini_credit_source_v2_adapter import (
+    restore_gemini_credit_source_contract as restore_gemini_credit_v2_source_contract,
+)
 from services.model_billing_migration_inventory import saved_dependency_inventory_policy
 from services.model_billing_openai_audio_adapter import (
     OPENAI_AUDIO_SOURCE_UIDS,
@@ -134,6 +148,7 @@ from services.model_billing_source_variants import (
     AudioSourceVariantContext,
     EmbeddingSourceVariantContext,
     GeminiSourceVariantContext,
+    GeminiV8SourceVariantContext,
     MappingDeclaration,
     ModelMappingRevision,
     SelectedModelMapping,
@@ -215,6 +230,22 @@ def build_source_adapter_contract(
         if tongyi_contract != restore_variant_source_contract(mapping):
             raise ValueError("Tongyi source variant evidence changed")
         return tongyi_contract
+    if isinstance(mapping, SelectedGeminiV8SourceVariant):
+        build_gemini_v8 = (
+            build_gemini_credit_v2_source_contract
+            if mapping.variant.source_plugin_uid == GEMINI_0913_SOURCE_UID
+            else build_gemini_credit_source_contract
+        )
+        gemini_v8_contract = build_gemini_v8(
+            **arguments,
+            source_schema=source_schema,
+            expected_source_runtime_schema_digest=mapping.variant.source_runtime_schema_digest,
+            target_model=mapping.target.model,
+            source_settings=mapping.variant.source_settings,
+        )
+        if gemini_v8_contract != restore_variant_source_contract(mapping):
+            raise ValueError("V8 Gemini source variant evidence changed")
+        return gemini_v8_contract
     if isinstance(mapping, SelectedGeminiSourceVariant):
         gemini_contract = build_gemini_credit_source_contract(
             **arguments,
@@ -348,6 +379,8 @@ def source_parameter_marker(
     }
     if isinstance(contract, TongyiCreditSourceContract):
         return tongyi_credit_source_marker(contract, **arguments, model_type=model_type)
+    if isinstance(contract, GeminiCreditSourceV2Contract):
+        return gemini_credit_v2_source_marker(contract, **arguments, model_type=model_type)
     if isinstance(contract, GeminiCreditSourceContract):
         return gemini_credit_source_marker(contract, **arguments, model_type=model_type)
     if isinstance(contract, OpenAIAudioSourceContract):
@@ -639,7 +672,7 @@ def mapped_dynamic_source_candidates(
 def mapping_has_required_operations(mapping: SelectedModelMapping) -> bool:
     if isinstance(mapping, SelectedTongyiSourceVariant):
         return set(mapping.operations) == tongyi_source_operations(mapping.source.model)
-    if isinstance(mapping, SelectedGeminiSourceVariant):
+    if isinstance(mapping, (SelectedGeminiSourceVariant, SelectedGeminiV8SourceVariant)):
         return set(mapping.operations) == gemini_source_operations(mapping.source.model)
     required = {"model/schema", f"{mapping.source.model_type.value}/invoke"}
     if mapping.source.model_type == ModelType.LLM:
@@ -1197,13 +1230,13 @@ def hosted_gemini_source_fact(
 ) -> HostedGeminiSourceFact | None:
     """Capture actual SYSTEM dictionary classifications, before substitution.
 
-    Source0912 LLMs use only the reviewed None/exact Google-root base,
+    Source0912/0913 LLMs use only the reviewed None/exact Google-root base,
     None/disabled metadata and None/empty file prefix. Embeddings do not consume
     these three fields. Nothing retains a key, raw URL or credential dictionary.
     """
     if (
         str(ModelProviderID(provider)) != "langgenius/gemini/google"
-        or source_uid != GEMINI_SOURCE_UID
+        or source_uid not in {GEMINI_SOURCE_UID, GEMINI_0913_SOURCE_UID}
         or type(credentials) is not dict
     ):
         return None
@@ -1273,6 +1306,47 @@ def gemini_source_variant_context(
     return GeminiSourceVariantContext(source_uid, settings)
 
 
+def gemini_v8_source_variant_context(
+    fact: HostedGeminiSourceFact | None,
+    *,
+    provider: str,
+    source_uid: str,
+    model_type: ModelType,
+    schema: AIModelEntity | None = None,
+) -> GeminiV8SourceVariantContext | None:
+    if (
+        type(fact) is not HostedGeminiSourceFact
+        or fact.provider != str(ModelProviderID(provider))
+        or fact.plugin_unique_identifier != source_uid
+        or source_uid not in {GEMINI_SOURCE_UID, GEMINI_0913_SOURCE_UID}
+        or model_type not in {ModelType.LLM, ModelType.TEXT_EMBEDDING}
+    ):
+        return None
+    if schema is not None and [
+        digest
+        for kind, model, digest in fact.catalog_schema_digests
+        if (kind, model) == (schema.model_type, schema.model)
+    ] != [model_schema_digest(schema)]:
+        return None
+    if model_type is ModelType.LLM:
+        if (fact.llm_source_base_policy, fact.llm_request_metadata_policy, fact.llm_file_url_policy) != (
+            "unset_or_exact_google_root",
+            "system_false_only",
+            "actual_system_absent_or_empty_only",
+        ):
+            return None
+        settings = GeminiSourceSettings(
+            source_base_policy="unset_or_exact_google_root",
+            request_metadata_policy="system_false_only",
+            file_url_policy="actual_system_absent_or_empty_only",
+        )
+    else:
+        settings = GeminiSourceSettings(
+            source_base_policy="not_used", request_metadata_policy="not_used", file_url_policy="not_used"
+        )
+    return GeminiV8SourceVariantContext(source_uid, settings)
+
+
 def hosted_source_variant_fact(
     *, provider: str, source_uid: str, models: Sequence[AIModelEntity], credentials: object
 ) -> HostedSourceVariantFact | None:
@@ -1370,6 +1444,18 @@ def restore_variant_source_contract(
             source_settings=mapping.variant.source_settings,
             expected_source_runtime_schema_digest=variant.source_runtime_schema_digest,
         )
+    if isinstance(mapping, SelectedGeminiV8SourceVariant):
+        restore_gemini_v8 = (
+            restore_gemini_credit_v2_source_contract
+            if mapping.variant.source_plugin_uid == GEMINI_0913_SOURCE_UID
+            else restore_gemini_credit_source_contract
+        )
+        return restore_gemini_v8(
+            marker,
+            **arguments,
+            source_settings=mapping.variant.source_settings,
+            expected_source_runtime_schema_digest=variant.source_runtime_schema_digest,
+        )
     if isinstance(mapping, SelectedGeminiSourceVariant):
         return restore_gemini_credit_source_contract(
             marker,
@@ -1552,6 +1638,10 @@ def resolve_source_mapping(
         context = tongyi_source_variant_context(
             tongyi_fact, provider=identity[0], source_uid=source_uid, model_type=identity[1], schema=schema
         )
+    elif isinstance(declaration, SourceGeminiV8VariantMapping):
+        context = gemini_v8_source_variant_context(
+            gemini_fact, provider=identity[0], source_uid=source_uid, model_type=identity[1], schema=schema
+        )
     elif isinstance(declaration, SourceGeminiVariantMapping):
         context = gemini_source_variant_context(
             gemini_fact, provider=identity[0], source_uid=source_uid, model_type=identity[1], schema=schema

api/services/model_billing_migration_inventory_gateway.py

@@ -36,11 +36,12 @@ from core.plugin.entities.plugin_daemon import PluginInstallTaskStatus
 from core.plugin.impl.model import PluginModelClient
 from core.plugin.impl.plugin import PluginInstaller
 from extensions import ext_hosting_provider
-from graphon.model_runtime.entities.model_entities import ModelPropertyKey, ModelType
+from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelPropertyKey, ModelType
 from models.provider_ids import ModelProviderID
 from repositories.model_billing_dependency_repository import ModelBillingDependencyRepository
 from services.entities.model_billing_embedding_mapping import SelectedEmbeddingSourceVariant
 from services.entities.model_billing_gemini_mapping import SelectedGeminiSourceVariant
+from services.entities.model_billing_gemini_v8_mapping import SelectedGeminiV8SourceVariant
 from services.entities.model_billing_mapping import SourceParameterContract
 from services.entities.model_billing_tongyi_mapping import SelectedTongyiSourceVariant
 from services.model_billing_dependency_inventory import DependencyInventoryPolicy, SavedModelReference
@@ -73,7 +74,7 @@ from services.model_billing_mapping_service import (
     source_parameter_public_settings,
 )
 from services.model_billing_migration_inventory import RequiredSavedCreditModel, SavedDependencySourceEvidence
-from services.model_billing_source_variants import selection_payload
+from services.model_billing_source_variants import SelectedModelMapping, selection_payload
 
 
 class RedirectCapabilities(BaseModel):
@@ -574,21 +575,7 @@ class SavedDependencyMigrationEvidenceGateway:
             )
             if schema is None or schema.model_type != kind:
                 raise ModelRouteUnmapped
-            if (
-                isinstance(
-                    mapping,
-                    (
-                        SelectedAudioSourceVariant,
-                        SelectedXAISourceVariant,
-                        SelectedGeminiSourceVariant,
-                        SelectedTongyiSourceVariant,
-                    ),
-                )
-                and schema.model != mapping.target.model
-            ):
-                # These closed families have no alias exception. Do not erase
-                # an unexpected target identity by normalizing the hash below.
-                raise ModelRouteUnmapped
+            _assert_target_source_identity(mapping, schema)
             if mapping.parameter_adapter.id == "openai_credit_llm_source_v2" and schema.model not in {
                 mapping.source.model,
                 mapping.target.model,
@@ -612,21 +599,62 @@ class SavedDependencyMigrationEvidenceGateway:
             # Its exact versioned runtime schema was checked above; selected
             # source rows still retain the original dimension-contract check.
             # Neither path maps or infers a vector width from a model name.
-            if (
-                kind == ModelType.TEXT_EMBEDDING
-                and not (
-                    isinstance(
-                        mapping,
-                        (SelectedEmbeddingSourceVariant, SelectedGeminiSourceVariant, SelectedTongyiSourceVariant),
-                    )
-                    and prospective_only
-                )
-                and source["embedding_dimensions"] != embedding_dimensions_contract(schema)
-            ):
-                raise ModelRouteUnmapped
+            _assert_target_embedding_dimensions(
+                mapping=mapping,
+                kind=kind,
+                prospective_only=prospective_only,
+                source=source,
+                schema=schema,
+            )
         return expected
 
 
+def _assert_target_source_identity(mapping: SelectedModelMapping, schema: AIModelEntity) -> None:
+    """Closed families have no alias exception; never normalize a wrong target."""
+    if (
+        isinstance(
+            mapping,
+            (
+                SelectedAudioSourceVariant,
+                SelectedXAISourceVariant,
+                SelectedGeminiSourceVariant,
+                SelectedGeminiV8SourceVariant,
+                SelectedTongyiSourceVariant,
+            ),
+        )
+        and schema.model != mapping.target.model
+    ):
+        raise ModelRouteUnmapped
+
+
+def _assert_target_embedding_dimensions(
+    *,
+    mapping: SelectedModelMapping,
+    kind: ModelType,
+    prospective_only: bool,
+    source: Mapping[str, object],
+    schema: AIModelEntity,
+) -> None:
+    """Only prospective closed declarations omit the installed-source dimension fact."""
+    if (
+        kind == ModelType.TEXT_EMBEDDING
+        and not (
+            isinstance(
+                mapping,
+                (
+                    SelectedEmbeddingSourceVariant,
+                    SelectedGeminiSourceVariant,
+                    SelectedGeminiV8SourceVariant,
+                    SelectedTongyiSourceVariant,
+                ),
+            )
+            and prospective_only
+        )
+        and source["embedding_dimensions"] != embedding_dimensions_contract(schema)
+    ):
+        raise ModelRouteUnmapped
+
+
 def _name(plugin_id: str, provider: str) -> str:
     return str(ModelProviderID(provider if provider.count("/") == 2 else plugin_id + "/" + provider))
 

api/services/model_billing_source_variants.py

@@ -41,6 +41,11 @@ from services.entities.model_billing_gemini_mapping import (
     SourceGeminiVariantMapping,
     SourceGeminiVariantSelection,
 )
+from services.entities.model_billing_gemini_v8_mapping import (
+    SelectedGeminiV8SourceVariant,
+    SourceGeminiV8VariantMapping,
+    SourceGeminiV8VariantSelection,
+)
 from services.entities.model_billing_mapping import (
     ModelMapping,
     SelectedSourceVariant,
@@ -70,6 +75,7 @@ type MappingDeclaration = (
     | SourceAudioVariantMapping
     | SourceXAIVariantMapping
     | SourceGeminiVariantMapping
+    | SourceGeminiV8VariantMapping
     | SourceTongyiVariantMapping
 )
 type SelectedSourceVariantMapping = (
@@ -78,6 +84,7 @@ type SelectedSourceVariantMapping = (
     | SelectedAudioSourceVariant
     | SelectedXAISourceVariant
     | SelectedGeminiSourceVariant
+    | SelectedGeminiV8SourceVariant
     | SelectedTongyiSourceVariant
 )
 type SelectedModelMapping = ModelMapping | SelectedSourceVariantMapping
@@ -87,6 +94,7 @@ type SourceVariantSelectionPayload = (
     | SourceAudioVariantSelection
     | SourceXAIVariantSelection
     | SourceGeminiVariantSelection
+    | SourceGeminiV8VariantSelection
     | SourceTongyiVariantSelection
 )
 
@@ -96,6 +104,7 @@ SOURCE_VARIANT_DECLARATION_TYPES = (
     SourceAudioVariantMapping,
     SourceXAIVariantMapping,
     SourceGeminiVariantMapping,
+    SourceGeminiV8VariantMapping,
     SourceTongyiVariantMapping,
 )
 SELECTED_SOURCE_VARIANT_TYPES = (
@@ -104,6 +113,7 @@ SELECTED_SOURCE_VARIANT_TYPES = (
     SelectedAudioSourceVariant,
     SelectedXAISourceVariant,
     SelectedGeminiSourceVariant,
+    SelectedGeminiV8SourceVariant,
     SelectedTongyiSourceVariant,
 )
 
@@ -162,6 +172,14 @@ class GeminiSourceVariantContext:
     settings: GeminiSourceSettings
 
 
+@dataclass(frozen=True, slots=True)
+class GeminiV8SourceVariantContext:
+    """Exact V8 facts; never accepted by the immutable V6 selector."""
+
+    source_plugin_uid: str
+    settings: GeminiSourceSettings
+
+
 @dataclass(frozen=True, slots=True)
 class TongyiSourceVariantContext:
     source_plugin_uid: str
@@ -174,6 +192,7 @@ type SourceContext = (
     | AudioSourceVariantContext
     | XAISourceVariantContext
     | GeminiSourceVariantContext
+    | GeminiV8SourceVariantContext
     | TongyiSourceVariantContext
 )
 
@@ -233,6 +252,17 @@ class ModelMappingRevision:
             if len(tongyi_matches) != 1:
                 raise ValueError("Unknown or ambiguous Tongyi source variant")
             selected = SelectedTongyiSourceVariant(declaration, tongyi_matches[0], self.digest)
+        elif isinstance(declaration, SourceGeminiV8VariantMapping):
+            if type(context) is not GeminiV8SourceVariantContext:
+                raise ValueError("Source V8 Gemini fact family changed")
+            gemini_v8_matches = [
+                item
+                for item in declaration.variants
+                if item.source_plugin_uid == context.source_plugin_uid and item.source_settings == context.settings
+            ]
+            if len(gemini_v8_matches) != 1:
+                raise ValueError("Unknown or ambiguous V8 Gemini source variant")
+            selected = SelectedGeminiV8SourceVariant(declaration, gemini_v8_matches[0], self.digest)
         elif isinstance(declaration, SourceGeminiVariantMapping):
             if type(context) is not GeminiSourceVariantContext:
                 raise ValueError("Source Gemini fact family changed")
@@ -305,6 +335,11 @@ class ModelMappingRevision:
                     SelectedTongyiSourceVariant(declaration, variant, self.digest)
                     for variant in sorted(declaration.variants, key=lambda item: item.variant_id)
                 )
+            elif isinstance(declaration, SourceGeminiV8VariantMapping):
+                result.extend(
+                    SelectedGeminiV8SourceVariant(declaration, variant, self.digest)
+                    for variant in sorted(declaration.variants, key=lambda item: item.variant_id)
+                )
             elif isinstance(declaration, SourceGeminiVariantMapping):
                 result.extend(
                     SelectedGeminiSourceVariant(declaration, variant, self.digest)
@@ -344,6 +379,11 @@ class ModelMappingRevision:
             selected = self.resolve(
                 identity, context=TongyiSourceVariantContext(record.source_plugin_uid, record.source_settings)
             )
+        elif isinstance(declaration, SourceGeminiV8VariantMapping):
+            record = SourceGeminiV8VariantSelection.model_validate(payload)
+            selected = self.resolve(
+                identity, context=GeminiV8SourceVariantContext(record.source_plugin_uid, record.source_settings)
+            )
         elif isinstance(declaration, SourceGeminiVariantMapping):
             record = SourceGeminiVariantSelection.model_validate(payload)
             selected = self.resolve(
@@ -390,6 +430,16 @@ def selection_payload(
             source_runtime_schema_digest=selected.variant.source_runtime_schema_digest,
             registry_digest=selected.registry_digest,
         )
+    if isinstance(selected, SelectedGeminiV8SourceVariant):
+        return SourceGeminiV8VariantSelection(
+            schema_version=8,
+            variant_id=selected.variant.variant_id,
+            source_plugin_uid=selected.variant.source_plugin_uid,
+            source_settings=selected.variant.source_settings,
+            source_schema_digest=selected.variant.source_schema_digest,
+            source_runtime_schema_digest=selected.variant.source_runtime_schema_digest,
+            registry_digest=selected.registry_digest,
+        )
     if isinstance(selected, SelectedGeminiSourceVariant):
         return SourceGeminiVariantSelection(
             schema_version=6,
@@ -456,6 +506,7 @@ def source_mapping_revision(
                 | SourceAudioVariantMapping
                 | SourceXAIVariantMapping
                 | SourceGeminiVariantMapping
+                | SourceGeminiV8VariantMapping
                 | SourceTongyiVariantMapping,
                 Field(discriminator="schema_version"),
             ]

api/tests/helpers/unit/model_billing_gemini_v8_profiles.py

@@ -0,0 +1,104 @@
+"""Real frozen Source profiles; target/revision strings are unit-only, never release authority."""
+
+import base64
+import gzip
+import hashlib
+import json
+from itertools import starmap
+from pathlib import Path
+from typing import Any, cast
+
+from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelType
+from services.entities.model_billing_gemini_mapping import GEMINI_SOURCE_UID, gemini_source_operations
+from services.entities.model_billing_gemini_v8_mapping import GEMINI_0913_SOURCE_UID
+from tests.helpers.unit.model_billing_gemini_profiles import GeminiProfile
+from tests.helpers.unit.model_billing_gemini_profiles import profiles as profiles0912
+
+SOURCE_PROVIDER = "langgenius/gemini/google"
+REVISION = "offline-gemini-v8-never-published"
+SOURCE_UIDS = (GEMINI_SOURCE_UID, GEMINI_0913_SOURCE_UID)
+
+
+def profiles(source_uid: str) -> list[GeminiProfile]:
+    if source_uid == GEMINI_SOURCE_UID:
+        return profiles0912()
+    if source_uid != GEMINI_0913_SOURCE_UID:
+        raise ValueError("Unknown exact source UID")
+    path = Path(__file__).resolve().parents[2] / "unit_tests/services/fixtures/gemini0913_source_profiles.packed.json"
+    blob = path.read_bytes()
+    if hashlib.sha256(blob).hexdigest() != "fd9d76809e7b973da9a1f5081db4880321e88514618ad262459942afef079f81":
+        raise ValueError("Frozen0913 packet changed")
+    packed = json.loads(blob)
+    raw = gzip.decompress(base64.b64decode(packed["payload_base64"], validate=True))
+    if hashlib.sha256(raw).hexdigest() != packed["raw_sha256"] or packed["raw_sha256"] != (
+        "8c32f42e9a623fae60f0da32633425938fb6557c4c93ea6975b7f9714dfb3dd7"
+    ):
+        raise ValueError("Frozen0913 raw profile changed")
+    rows = cast(list[GeminiProfile], json.loads(raw)["profiles"])
+    if len(rows) != 9 or any(row["adapter_id"] != "gemini_credit_source_v2" for row in rows):
+        raise ValueError("Frozen0913 profile set changed")
+    return rows
+
+
+def schema(row: GeminiProfile, *, runtime: bool = False) -> AIModelEntity:
+    return AIModelEntity.model_validate(row["source_daemon_schema" if runtime else "source_catalog_schema"])
+
+
+def profile_digest(row: GeminiProfile) -> str:
+    return (
+        "sha256:"
+        + hashlib.sha256(
+            json.dumps(row, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
+        ).hexdigest()
+    )
+
+
+def variant_record(row: GeminiProfile, source_uid: str) -> dict[str, Any]:
+    kind = ModelType(row["model_type"])
+    return {
+        "variant_id": row["profile_id"],
+        "source_plugin_uid": source_uid,
+        "source_settings": row["source_settings"],
+        "source_schema_digest": row["source_schema_digest"],
+        "source_runtime_schema_digest": row["source_runtime_schema_digest"],
+        "source_profile_id": row["profile_id"],
+        "parameter_adapter": {
+            "id": row["adapter_id"],
+            "revision": row["adapter_revision"],
+            "rules_digest": profile_digest(row),
+        },
+        "target": {
+            "plugin_id": "langgenius/tokener",
+            "provider": "tokener",
+            "model_type": kind.value,
+            "model": row["target_model"],
+            "catalog_revision": "unit-only-no-target-release",
+            "minimum_plugin_version": "999.0.0.dev1",
+        },
+        "operations": sorted(gemini_source_operations(row["source_model"])),
+        "conformance_fixture_ids": ["core-unit-source-profile:" + row["profile_id"]],
+    }
+
+
+def mapping_record(model: str, source_uids: tuple[str, ...] = SOURCE_UIDS) -> dict[str, Any]:
+    rows = [next(row for row in profiles(uid) if row["source_model"] == model) for uid in source_uids]
+    if not rows:
+        raise ValueError("Source profiles required")
+    kind = ModelType(rows[0]["model_type"])
+    return {
+        "schema_version": 8,
+        "mapping_revision": REVISION,
+        "source": {
+            "provider_id": SOURCE_PROVIDER,
+            "model_type": kind.value,
+            "model": model,
+            "mode": "chat" if kind is ModelType.LLM else "text-embedding",
+        },
+        "settlement_owner": "shared_credit_pool",
+        "pool_types": ["paid", "trial"],
+        "variants": list(starmap(variant_record, zip(rows, source_uids, strict=True))),
+    }
+
+
+def registry(source_uids: tuple[str, ...] = SOURCE_UIDS) -> str:
+    return json.dumps([mapping_record(row["source_model"], source_uids) for row in profiles(SOURCE_UIDS[0])])

api/tests/unit_tests/services/entities/test_model_billing_gemini_v8_mapping.py

@@ -0,0 +1,304 @@
+"""Synthetic V8 shapes only; never a reviewed SDK/profile or release packet."""
+
+import hashlib
+import json
+from copy import deepcopy
+from dataclasses import FrozenInstanceError
+from typing import cast
+
+import pytest
+from pydantic import ValidationError
+from typing_extensions import TypedDict
+
+from services.entities.model_billing_gemini_mapping import (
+    GEMINI_SOURCE_EMBEDDING_MODELS,
+    GEMINI_SOURCE_LLM_MODELS,
+    GEMINI_SOURCE_UID,
+    SourceGeminiVariantMapping,
+    gemini_source_model_type,
+    gemini_source_operations,
+)
+from services.entities.model_billing_gemini_v8_mapping import (
+    GEMINI_0913_SOURCE_UID,
+    SelectedGeminiV8SourceVariant,
+    SourceGeminiV8VariantMapping,
+    SourceGeminiV8VariantSelection,
+)
+from tests.helpers.unit.model_billing_gemini_profiles import profiles, variant_record
+
+
+class SyntheticAdapter(TypedDict):
+    id: str
+    revision: str
+    rules_digest: str
+
+
+class SyntheticTarget(TypedDict):
+    plugin_id: str
+    provider: str
+    model_type: str
+    model: str
+    catalog_revision: str
+    minimum_plugin_version: str
+
+
+class SyntheticVariant(TypedDict):
+    variant_id: str
+    source_plugin_uid: str
+    source_settings: dict[str, str]
+    source_schema_digest: str
+    source_runtime_schema_digest: str
+    source_profile_id: str
+    parameter_adapter: SyntheticAdapter
+    target: SyntheticTarget
+    operations: list[str]
+    conformance_fixture_ids: list[str]
+
+
+class SyntheticRecord(TypedDict):
+    schema_version: int
+    mapping_revision: str
+    source: dict[str, str]
+    settlement_owner: str
+    pool_types: list[str]
+    variants: list[SyntheticVariant]
+
+
+def synthetic_record(model: str = "gemini-2.5-flash") -> SyntheticRecord:
+    kind = gemini_source_model_type(model)
+    llm = model in GEMINI_SOURCE_LLM_MODELS
+    settings = (
+        {
+            "source_base_policy": "unset_or_exact_google_root",
+            "request_metadata_policy": "system_false_only",
+            "file_url_policy": "actual_system_absent_or_empty_only",
+        }
+        if llm
+        else {"source_base_policy": "not_used", "request_metadata_policy": "not_used", "file_url_policy": "not_used"}
+    )
+    return {
+        "schema_version": 8,
+        "mapping_revision": "synthetic-v8-shape-not-published",
+        "source": {
+            "provider_id": "langgenius/gemini/google",
+            "model_type": kind.value,
+            "model": model,
+            "mode": "chat" if llm else "text-embedding",
+        },
+        "settlement_owner": "shared_credit_pool",
+        "pool_types": ["paid", "trial"],
+        "variants": [
+            {
+                "variant_id": f"synthetic-{version}-{model}",
+                "source_plugin_uid": uid,
+                "source_settings": deepcopy(settings),
+                # Syntax examples, not real source or target digest evidence.
+                "source_schema_digest": "sha256:" + "a" * 64,
+                "source_runtime_schema_digest": "sha256:" + "b" * 64,
+                "source_profile_id": f"gemini{version}-native-v1:{model}",
+                "parameter_adapter": {"id": adapter, "revision": "1", "rules_digest": "sha256:" + "c" * 64},
+                "target": {
+                    "plugin_id": "langgenius/tokener",
+                    "provider": "tokener",
+                    "model_type": kind.value,
+                    "model": model,
+                    "catalog_revision": "synthetic-not-a-live-catalog",
+                    "minimum_plugin_version": "0.0.0.dev0",
+                },
+                "operations": sorted(gemini_source_operations(model)),
+                "conformance_fixture_ids": ["synthetic-shape-only"],
+            }
+            for version, uid, adapter in (
+                ("0912", GEMINI_SOURCE_UID, "gemini_credit_source_v1"),
+                ("0913", GEMINI_0913_SOURCE_UID, "gemini_credit_source_v2"),
+            )
+        ],
+    }
+
+
+def selection(record: SyntheticRecord) -> dict[str, object]:
+    row = record["variants"][0]
+    return {
+        "schema_version": 8,
+        **{
+            key: row[key]
+            for key in (
+                "variant_id",
+                "source_plugin_uid",
+                "source_settings",
+                "source_schema_digest",
+                "source_runtime_schema_digest",
+            )
+        },
+        "registry_digest": "sha256:" + "d" * 64,
+    }
+
+
+@pytest.mark.parametrize("model", sorted(GEMINI_SOURCE_LLM_MODELS | GEMINI_SOURCE_EMBEDDING_MODELS))
+def test_exact_two_uid_shapes_and_each_single_variant_round_trip(model: str) -> None:
+    raw = synthetic_record(model)
+    parsed = SourceGeminiV8VariantMapping.model_validate(raw)
+    assert parsed.model_dump(mode="json") == raw
+    assert SourceGeminiV8VariantMapping.model_validate_json(parsed.model_dump_json()) == parsed
+    for row in raw["variants"]:
+        one: SyntheticRecord = {**raw, "variants": [row]}
+        assert (
+            SourceGeminiV8VariantMapping.model_validate(one).variants[0].source_plugin_uid == row["source_plugin_uid"]
+        )
+        frozen = SourceGeminiV8VariantSelection.model_validate(selection(one))
+        assert SourceGeminiV8VariantSelection.model_validate_json(frozen.model_dump_json()) == frozen
+
+
+@pytest.mark.parametrize("value", [True, False, 8.0, "8", 6, None])
+def test_mapping_and_selection_require_integer_eight_without_coercion(value: object) -> None:
+    raw = synthetic_record()
+    with pytest.raises(ValidationError, match="integer 8"):
+        SourceGeminiV8VariantMapping.model_validate({**raw, "schema_version": value})
+    with pytest.raises(ValidationError, match="integer 8"):
+        SourceGeminiV8VariantSelection.model_validate({**selection(raw), "schema_version": value})
+
+
+@pytest.mark.parametrize("mutation", ["none", "three", "variant_id", "selector", "pools"])
+def test_finite_variant_and_selector_uniqueness(mutation: str) -> None:
+    raw = synthetic_record()
+    if mutation == "none":
+        raw["variants"] = []
+    elif mutation == "three":
+        raw["variants"].append(deepcopy(raw["variants"][0]))
+    elif mutation == "variant_id":
+        raw["variants"][1]["variant_id"] = raw["variants"][0]["variant_id"]
+    elif mutation == "selector":
+        duplicate = deepcopy(raw["variants"][0])
+        duplicate["variant_id"] = "different-id-same-selector"
+        raw["variants"][1] = duplicate
+    else:
+        raw["pool_types"] = ["trial", "trial"]
+    with pytest.raises(ValidationError):
+        SourceGeminiV8VariantMapping.model_validate(raw)
+
+
+@pytest.mark.parametrize("index", [0, 1])
+@pytest.mark.parametrize("field", ["uid", "profile", "adapter", "revision"])
+def test_uid_profile_and_adapter_identity_cannot_be_crossed(index: int, field: str) -> None:
+    raw = synthetic_record()
+    item, other = raw["variants"][index], raw["variants"][1 - index]
+    if field == "uid":
+        item["source_plugin_uid"] = other["source_plugin_uid"]
+    elif field == "profile":
+        item["source_profile_id"] = other["source_profile_id"]
+    elif field == "adapter":
+        item["parameter_adapter"]["id"] = other["parameter_adapter"]["id"]
+    else:
+        item["parameter_adapter"]["revision"] = "2"
+    with pytest.raises(ValidationError):
+        SourceGeminiV8VariantMapping.model_validate(raw)
+
+
+@pytest.mark.parametrize("where", ["root", "source", "variant", "settings", "adapter", "target"])
+def test_extra_fields_cannot_add_origin_credentials_or_byok_authority(where: str) -> None:
+    raw = synthetic_record()
+    row = raw["variants"][0]
+    target = cast(
+        dict[str, object],
+        {
+            "root": raw,
+            "source": raw["source"],
+            "variant": row,
+            "settings": row["source_settings"],
+            "adapter": row["parameter_adapter"],
+            "target": row["target"],
+        }[where],
+    )
+    target["byok_or_origin_override"] = "not-a-routing-authority"
+    with pytest.raises(ValidationError, match="Extra inputs"):
+        SourceGeminiV8VariantMapping.model_validate(raw)
+
+
+@pytest.mark.parametrize(
+    "mutation",
+    [
+        "provider",
+        "unknown_uid",
+        "source_type",
+        "source_mode",
+        "target_type",
+        "target_model",
+        "settings",
+        "missing_op",
+        "extra_op",
+        "duplicate_op",
+        "duplicate_fixture",
+        "bool_digest",
+    ],
+)
+@pytest.mark.parametrize("model", ["gemini-2.5-flash", "gemini-embedding-001", "gemini-embedding-2-preview"])
+def test_model_type_mode_operations_and_settings_remain_closed(mutation: str, model: str) -> None:
+    raw = synthetic_record(model)
+    row = raw["variants"][0]
+    if mutation == "provider":
+        raw["source"]["provider_id"] = "langgenius/openai/openai"
+    elif mutation == "unknown_uid":
+        row["source_plugin_uid"] = GEMINI_SOURCE_UID.replace("0.9.12", "0.9.13")
+    elif mutation == "source_type":
+        raw["source"]["model_type"] = "text-embedding" if model in GEMINI_SOURCE_LLM_MODELS else "llm"
+    elif mutation == "source_mode":
+        raw["source"]["mode"] = "text-embedding" if model in GEMINI_SOURCE_LLM_MODELS else "chat"
+    elif mutation == "target_type":
+        row["target"]["model_type"] = "text-embedding" if model in GEMINI_SOURCE_LLM_MODELS else "llm"
+    elif mutation == "target_model":
+        row["target"]["model"] = "gemini-3.1-flash-lite" if model != "gemini-3.1-flash-lite" else "gemini-2.5-flash"
+    elif mutation == "settings":
+        row["source_settings"]["source_base_policy"] = (
+            "not_used" if model in GEMINI_SOURCE_LLM_MODELS else "unset_or_exact_google_root"
+        )
+    elif mutation == "missing_op":
+        row["operations"].pop()
+    elif mutation == "extra_op":
+        row["operations"].append("tts/invoke")
+    elif mutation == "duplicate_op":
+        row["operations"].append(row["operations"][0])
+    elif mutation == "duplicate_fixture":
+        row["conformance_fixture_ids"] *= 2
+    else:
+        cast(dict[str, object], row)["source_schema_digest"] = True
+    with pytest.raises(ValidationError):
+        SourceGeminiV8VariantMapping.model_validate(raw)
+
+
+def test_selected_wrapper_is_frozen_and_only_projects_its_declaration() -> None:
+    parsed = SourceGeminiV8VariantMapping.model_validate(synthetic_record())
+    row = parsed.variants[1]
+    selected = SelectedGeminiV8SourceVariant(parsed, row, "sha256:" + "d" * 64)
+    assert (selected.mapping_revision, selected.source, selected.settlement_owner, selected.pool_types) == (
+        parsed.mapping_revision,
+        parsed.source,
+        parsed.settlement_owner,
+        parsed.pool_types,
+    )
+    assert (
+        selected.source_schema_digest,
+        selected.parameter_adapter,
+        selected.target,
+        selected.operations,
+        selected.conformance_fixture_ids,
+    ) == (row.source_schema_digest, row.parameter_adapter, row.target, row.operations, row.conformance_fixture_ids)
+    with pytest.raises(FrozenInstanceError):
+        selected.registry_digest = "sha256:" + "e" * 64  # type: ignore[read-only]
+
+
+def test_existing_v6_real_fixtures_keep_c9_parse_dump_bytes_and_digest() -> None:
+    # Captured from unchanged C9 before authoring this DTO, not generated from
+    # the new family or made-up 0913 source profiles.
+    old = [SourceGeminiVariantMapping.model_validate(variant_record(row)) for row in profiles()]
+    canonical = json.dumps(
+        [row.model_dump(mode="json") for row in old], sort_keys=True, separators=(",", ":"), ensure_ascii=False
+    ).encode()
+    assert hashlib.sha256(canonical).hexdigest() == "553c65d987506bad246c040a54b74827ec23f56396daa7694f4ecdd14bc8ccd7"
+    assert hashlib.sha256("\n".join(row.model_dump_json() for row in old).encode()).hexdigest() == (
+        "97acbdb6bfc3fe0914bad33de975ee2959480fdc4ee136119c91b5ea80d78d8b"
+    )
+    for item in old:
+        raw = item.model_dump(mode="json")
+        raw["variants"][0]["source_plugin_uid"] = GEMINI_0913_SOURCE_UID
+        with pytest.raises(ValidationError):
+            SourceGeminiVariantMapping.model_validate(raw)

api/tests/unit_tests/services/fixtures/gemini0913_source_profiles.packed.json

@@ -0,0 +1 @@
+{"encoding":"gzip+base64","raw_sha256":"8c32f42e9a623fae60f0da32633425938fb6557c4c93ea6975b7f9714dfb3dd7","payload_base64":"H4sIAAAAAAACE+29fXPcxpXo/VVQ/OdJyh4O3gcgS9noRrKlm1hShcxzy7W7hTSABgkTA8wCGIq0K1WSFUuyY9nOJo4TW3bsrG15szeSsvHajiTLH0YckvorX+Ge0w3MC8kRJdkU385WrUNhgNPd56Xfzg/olyZYUHZZ4hVZNw+4l/kFzxdZGWepV8wz3bInpiYiTXV0wzIc0+CuHjE/cMNWZOnwc9SyAlPVfdXVdTuyme2EKmv5bityopaphUEQRvrE0xNBlnOPp2VcLnsh7/A05GmwPDH10kR1MYoTPigx0LTIDllghK7KW1Go2b5qG44ROC7jIVehBtzkphNp3HEdq+VrgdXilhlGUEMeGlDiXFx6QdZuxyXIMyOm65qptVpwS+SoHMTCX8x1ODO4HfmRauqa78NzHRYssDkOD83lrDOfpXBtkecFaASuqZPOpDrxi6pBRTDP28wL4zlelF7UTYNRxTm8ZXPVYLqptwzdte0IWmO7LnMNHrBWpEI1LD9gUWRZoC+uhzwIuGVoVmAFLtOwGRnoK8hCqEFf7ksTLY0baBjb0n3msACs4RrM1AzD11zVdbkDjfRDW/NBtKGD4Vq6b5qu5fthKwwDzQ9UaDfI59xwsaq+5js+t1uqrukWqNgBVQYONyzHYixkDCpmMiPUVF31HU3TfVd3HSewTTOyI82wUCcLcRqi3ng7TmPV1Qxv1LkCVrIkm/NYGnp5F+ze5l4nz9D0hbeIrU2zvA33+1k3DVkO/lG5pyL9koeKFKWEjLezVKkkKp2kWyhQeuVZpVLEcykPG9Xd9W057yRsWYHylZljP1Y1TamqoTR+oLA8mI+xCKESvAm1rMgqKdLUyrHZ04W4WVYr6OY5Fvcj8Abl6MnnwE7JceHQzTb+Peoh0yBMSflZJYFylJ+e+RG6W9X+ial/fmkCFN0pee7FAzV6Qc5DcOVKhYsYTPVtOV+MK7/URIyBp4BihbSXJtosBT8OIerCThanGAZ8KS7KOJ1TongJGnr81LEzp0+emvV+9tOfKHwJ2qQ0FzVsXpbHc3EK9YWeAKoKV31eMgUqW2ZBlkBhtfSq8hPZWdC30oa7QlC2kqXJ8rSywJdnThx9Skp7Cj2iai5YBBQzLRXRyXmUxHPzoB98TBGiTE1VOjyH8C3AbN1OkrEQfg44li1Um/NSeEiWcqXMWVp0shwMD83FB5X5rDOtBBkYI+2K/kwKZxForjZfJ4+zXDkxO3tGKcqcs7Zy/PQzaKVUX1pqwgXoIrpJ2QwYFJw0u+lCCpUDE0KHALUqum1oe1WfQugXdShNdeRIyfI5Lo3OkpgVvAC9itbGS+CsEQfXEe3pdP0kDjye51nuYZX5ElpLqo0lqDfl6JmTzZPwS85EH1NAhVkJTi9aHoGP8rxSH9S+AXcrczzleRwoEUsSHzo1pYs1AIt2c2y6KBPDQKlKnIY6gVCoUkPIKThDQxdB1gE7F+FCv4FV7MTpYrbARTQ9e2ZWBzldrKAiolcpMyUCFc2L2teSi8ox5pLMh0oIQaouRAxGBSXJgoVCgd50nqXgY1gJ6f1ikOjmyaBfmHl+Zvb4cwqD7iEtm7zdKZerIkDrfVW/wAPpATl0DAkGbM4T6dqsLMEV2/Jah8X5UGlVf46WjfLsRZ5W3UplYdGy+ibl6MxsMY0eBaMnF9YB9QZJjPWKoLZZvtwUXo+KbRfYX0o/Lpc7ONokSXvQHQz3AFhMQwZiY1Gbkhcb+qTViBJWzA/qW3evss/BLgB0msNAU4IOp8ANCv70RMTRAUR/M1FmWdIIwD3Qh8HP48bwFQjvtGyU81kXQhMHQdnXQHEiVEbuDbOgizoUt4U8w8e7YZxN/CuWWAbzHiiwDU2C+oRgFOybRevhxoT58L9iJuD9bAbueVa0UIEWKs9ULXxx3jvBhCE2/Vjrsa+vEdVIFYNWoUsoY9kxVv7uFfGLoHhNNWECYctbceoxz8pxxmE5gx4Oe95u3WVDa7CLADlPT8zzpDPckB9BQXmWQAcGDp61U14Uk8pPsrMQIyW4KsQymkJ0JEmJ45cCUuu7IZzanYQLB5xUjkKkg+sMP8Y60Cx0qEJ5kefZ0+IGUWvlbJwkig/TBhjZQqww6qVAh8RIAz8HsehPk8oJ6HfHV6eNw9oW1RmxyOpX/9m79enqG9d6l7+4/+6bq1dvrZ67du/cy+t3L6x8/cbau7+UN6zcebd34876pc/heu/mr+G6vLl3/fXVy2/B/fDPtffP1dI+Wf/m1/ff++Ifd15f/eyj3ge/6t282Hvz972776x9dL13/V0oovfvr9+/dKX38ZV+cff/6/ebi1t97/Pex+9uLg5tvMn1hvQw0sbh6+gbbGliSp9U4a8YRl8V/0rBNTZJyDpVB/LP/4qxzYNqvNafnsj5v3XjfCgyK1eLYKTDOILe2kNh0E/xTTUY8jt10rUe4HphLCav0C0uxkxJu0HCYdQoGAiGicCUokIgtTm0UplnSaRkEYxVCYzLCzyJ57MsbJzlODRDp1e1BeZJHIeMAuIcqj/qC/fPvbv+zaXVD7/qfXPh/iX440vpF2AB/PvctX/ceRdKXP/os7WPb62fu7D+h1+v3Lq48tW53pXX0HSvnlu9+mrvzRvrF75GC7/24er7F+6fe/X+R38fa7Gs43VGbSWu1FbStrZS9dR3aJ+q1CHL2JZl2JtNM9OBYiLoi0TAdjs4WUlimOjAwCUuwQA1V86jKcQQjt13HZSTyslI3LPpF2GWEqazort/WlnOujDJSiGKYQIJcw94Ki6Ufg+2IYRfvwQxtfbbP0JorN3+zeoHV++//Q0EEZhk5avX7v8B46X36ctwffie1ct/Xv3dXyBCwWArtz9Zv3mh9/G19Ruf9N58efXtm+MMBlbxsm7Z6ZZeCdOHtBipyeZfa0MKbfaNqQ0ZcyuJ4wyrjjdsnG5hVpTdr8iQbatHN9r2ZBrGaIBCOTvPS9GzZmiCpBuiBcRQKnpWtCFYrgN15NKoeZc/PbiFiY4Yoh2nG2I6FkdDHXwBbgOz3aJ+QPTrIw+zRRaD7hO+wdK/v9F769P1b37be++D1XPvYwh+c2nts19hB/qrT1a+uiUqovQu/6H35p8hFmXnu/rbG6uvn5f3r3z1295bN3qvfSb/iTfceXP1nS9WX72ypdgtneDkBpWM1HGzmL4TpN0kqTxA/lm5QKVib0jeOA+Qz41zAh8mNdAbbnIEfGrEAYS9Ntn//wyszlPUPoZduoBLLjRc39KVveRPnt8NYaGgQHymGa5gSnAEsDz2B+EytBDG7frerazZu3MOLDJsjg3WBJNtKmz16p/Xr99d+/o6xi9YW8gQVj1//0+/XLv+zuobn629dXGcAWeHmzVap6F6bGu3fq0qMQ9rNKn9R7WZo6rqg2JWzLDqhtVGgVgV8T+pzGIPusgS0Kj8I65CE2I9HTU03iAdIMTncBYm6iAfL4T5YZyFqWCI86s5LoNWBnd/F2NS+T8oWQhCySP1w3ufrnuFTbUedqWh4uH6wDVr36qewWEiSzf5GPTpshPA8VnYVvoHeMbK7Vd6V0f6gcqH3vkCh4Fzd7ADefsyzu1u/379+sdwBSZtvVfOr1//av31Cz2Ymb35u97Fv63cekMWgcPMJmcGaeCj4M99lx2uhnTfSv4jOvEMtHqD5rZwZSlpMO0zYbWw9aRiNMQeuw/aaiDa5MtjBqBnc1x8Y3vOxjCHeDbL5sDSMxw3t0bXUfKXlVu/6l3/ACZsq3/8ZEsd9QWOPL3xsQcG+dyQiJ0N8DFK+V95dlbMgGC6lSf1fsfomPPGl+iLf/+id/ni6gdvgfve/81dWIWA06/f+HTt9sWVbz5aPX8DvA16VRid1v7y29U331q5+x5MdnBQvPv+6pXr4/wMt9e2KlQWAfJXL7yyrRah5t5AyK7o8SccJhfVAhwkKLjhiP1JkSWLvFogLiklKxY2DOjX/7N+bOXrb9Z++9nK7f9Y+/D86vufrF/7D4h/XMO9D73J7d5rH43T4Y+wLL7Eg24pdyGGnFFKe/Xa+kevb6tGrLM3LGdXNPmzyh0rrTyDesMN7jiA6zFMIL6XiG2CICugB/d5UTZ4FOH+JpaTBsvfn4Sg5sp8WXaKqWaTxZNzIqAnQ77YrPZBWCduhllQNCMQ3xjecBzSnbDHSD1Wr14BO/T++vLaLViuXV679SkszlZuf3nv3Pneza97r723fk4EAazJb3+x/s0f/3HnVZxv3Ph0/dr5x6rQP80nR16cbwTploYXddq68rK6oovf1u5YpDcsZmftLqVsWvrBgMP6837lBRhv68SC2LKBaQbuwPzkJ89hWLEQppIwlSgnN0QTjoFykgzd05AQ6J2SpN27eXH19VfXXrnWu35B3rWlXvG5eqtwWPzw9e2UOirj8Ua7oszl0LBh5TVSDyEwDvBG3MATWRdMH078bOYYPBqnsPgT2TnM/Mm1IPxTn7RQbhrL31T8P23iF7/YvGMaLhzEXdMXmPe/z2z9U6f0/tdPaa91f+y11nY8yG2sHZL2zPfbnvnsyI537aujV2vrzo7ZX5fl09b6pq31TaH/JArdFIsHPYkwm3WUM8POW/2777bVvwd1lFcoybBtkqFW6Z6pUG3TQ56GeY4tKbN1tqQ20sjFWlEjF4cqe/Ucbg1+eGn9+k1K0nwHSZraCHu0erU7UIrrsVNcAwtv+mmgXUqMPXZirNbvEymsttihTfn1J0sbrvcnTZQiPNApws3jFSmnr5zNoyUph5LLQ8uhLX/tr00OXWK6v1Utr69+fHXt8z/dO/9fvT/eunf+k3vnf3nv/B9731xd+8tv77387/fOv3Pv/N175/+Auh/ay97J3HddQfHQH+9Line truncated

api/tests/unit_tests/services/test_model_billing_gemini_credit_source_v2_adapter.py

@@ -0,0 +1,179 @@
+"""The nine real Source0913 schemas/pins; no target installation or model call."""
+
+from dataclasses import replace
+from operator import itemgetter
+from typing import cast
+
+import pytest
+from typing_extensions import TypedDict
+
+from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelType
+from services import model_billing_gemini_credit_source_v2_adapter as adapter
+from services.entities.model_billing_gemini_mapping import GEMINI_SOURCE_UID, GeminiSourceSettings
+from services.entities.model_billing_gemini_v8_mapping import GEMINI_0913_SOURCE_UID
+from services.entities.model_billing_mapping import SourceParameterContract, model_schema_digest
+from tests.helpers.unit.model_billing_gemini_profiles import GeminiProfile
+from tests.helpers.unit.model_billing_gemini_v8_profiles import SOURCE_PROVIDER, profile_digest, profiles, schema
+
+
+class BuildArguments(TypedDict, closed=True):
+    adapter_id: str
+    adapter_revision: str
+    expected_rules_digest: str
+    source_provider: str
+    source_model: str
+    source_plugin_unique_identifier: str
+    source_schema: AIModelEntity
+    expected_source_schema_digest: str
+    expected_source_runtime_schema_digest: str
+    target_model: str
+    source_settings: GeminiSourceSettings
+    source_runtime_schema: AIModelEntity | None
+
+
+class MarkerArguments(TypedDict, closed=True):
+    source_provider: str
+    source_model: str
+    model_type: ModelType
+    target_model: str
+
+
+def arguments(row: GeminiProfile) -> BuildArguments:
+    return {
+        "adapter_id": row["adapter_id"],
+        "adapter_revision": row["adapter_revision"],
+        "expected_rules_digest": profile_digest(row),
+        "source_provider": SOURCE_PROVIDER,
+        "source_model": row["source_model"],
+        "source_plugin_unique_identifier": GEMINI_0913_SOURCE_UID,
+        "source_schema": schema(row),
+        "expected_source_schema_digest": row["source_schema_digest"],
+        "expected_source_runtime_schema_digest": row["source_runtime_schema_digest"],
+        "target_model": row["target_model"],
+        "source_settings": GeminiSourceSettings(**row["source_settings"]),
+        "source_runtime_schema": schema(row, runtime=True),
+    }
+
+
+@pytest.mark.parametrize("row", profiles(GEMINI_0913_SOURCE_UID), ids=itemgetter("source_model"))
+def test_real_catalog_runtime_rules_marker_and_prospective_restore(row: GeminiProfile) -> None:
+    args = arguments(row)
+    contract = adapter.build_gemini_credit_source_contract(**args)
+    assert type(contract) is adapter.GeminiCreditSourceV2Contract
+    assert contract.profile_id == row["profile_id"]
+    assert contract.adapter_id == "gemini_credit_source_v2"
+    assert contract.rules_digest == profile_digest(row)
+    assert model_schema_digest(args["source_schema"]) == contract.source_schema_digest
+    runtime = args["source_runtime_schema"]
+    assert runtime is not None
+    assert model_schema_digest(runtime) == contract.source_runtime_schema_digest
+    marker = adapter.gemini_credit_source_marker(
+        contract,
+        source_provider=SOURCE_PROVIDER,
+        source_model=row["source_model"],
+        model_type=ModelType(row["model_type"]),
+        target_model=row["target_model"],
+    )
+    assert set(marker) == {
+        "schema_version",
+        "profile_id",
+        "source_plugin_uid",
+        "source_model",
+        "source_schema_digest",
+        "rules_digest",
+    }
+    assert marker["source_plugin_uid"] == GEMINI_0913_SOURCE_UID
+    assert (
+        adapter.restore_gemini_credit_source_contract(
+            marker,
+            source_provider=SOURCE_PROVIDER,
+            source_model=row["source_model"],
+            model_type=ModelType(row["model_type"]),
+            target_model=row["target_model"],
+            source_settings=args["source_settings"],
+            expected_source_runtime_schema_digest=row["source_runtime_schema_digest"],
+            source_runtime_schema=runtime,
+        )
+        == contract
+    )
+    # None means frozen/prospective, never a live observed source RPC.
+    prospective = args.copy()
+    prospective["source_runtime_schema"] = None
+    assert adapter.build_gemini_credit_source_contract(**prospective) == contract
+
+
+@pytest.mark.parametrize("row", profiles(GEMINI_0913_SOURCE_UID), ids=itemgetter("source_model"))
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("adapter_id", "gemini_credit_source_v1"),
+        ("adapter_revision", "2"),
+        ("source_provider", "langgenius/openai/openai"),
+        ("source_plugin_unique_identifier", GEMINI_SOURCE_UID),
+        ("source_plugin_unique_identifier", GEMINI_0913_SOURCE_UID.replace("0.9.13", "0.9.14")),
+        ("source_model", "unreviewed-model"),
+        ("target_model", "unreviewed-target"),
+        ("expected_rules_digest", "sha256:" + "a" * 64),
+        ("expected_source_schema_digest", "sha256:" + "b" * 64),
+        ("expected_source_runtime_schema_digest", "sha256:" + "c" * 64),
+    ],
+)
+def test_each_independent_pin_and_identity_is_fail_closed(row: GeminiProfile, field: str, value: str) -> None:
+    args = arguments(row)
+    cast(dict[str, object], args)[field] = value
+    with pytest.raises(adapter.GeminiCreditSourceV2AdapterError):
+        adapter.build_gemini_credit_source_contract(**args)
+
+
+@pytest.mark.parametrize("row", profiles(GEMINI_0913_SOURCE_UID), ids=itemgetter("source_model"))
+@pytest.mark.parametrize("runtime", [False, True])
+def test_real_entity_full_metadata_drift_is_not_erased(row: GeminiProfile, runtime: bool) -> None:
+    args = arguments(row)
+    key = "source_runtime_schema" if runtime else "source_schema"
+    observed = args[key]
+    assert isinstance(observed, AIModelEntity)
+    changed = observed.model_dump(mode="json")
+    changed["label"] = {"en_US": "changed actual Source metadata"}
+    cast(dict[str, object], args)[key] = AIModelEntity.model_validate(changed)
+    with pytest.raises(adapter.GeminiCreditSourceV2AdapterError):
+        adapter.build_gemini_credit_source_contract(**args)
+
+
+@pytest.mark.parametrize("row", profiles(GEMINI_0913_SOURCE_UID), ids=itemgetter("source_model"))
+def test_marker_requires_exact_concrete_factory_and_never_adopts_other_family(row: GeminiProfile) -> None:
+    args = arguments(row)
+    contract = adapter.build_gemini_credit_source_contract(**args)
+    kwargs: MarkerArguments = {
+        "source_provider": SOURCE_PROVIDER,
+        "source_model": row["source_model"],
+        "model_type": ModelType(row["model_type"]),
+        "target_model": row["target_model"],
+    }
+    generic = SourceParameterContract(
+        profile_id=contract.profile_id,
+        source_plugin_unique_identifier=contract.source_plugin_unique_identifier,
+        source_model=contract.source_model,
+        source_schema_digest=contract.source_schema_digest,
+        target_model=contract.target_model,
+        rules_digest=contract.rules_digest,
+        adapter_id=contract.adapter_id,
+        algorithm_revision=contract.algorithm_revision,
+        source_runtime_schema_digest=contract.source_runtime_schema_digest,
+    )
+    for forged in (replace(contract, rules_digest="sha256:" + "0" * 64), generic):
+        with pytest.raises(adapter.GeminiCreditSourceV2AdapterError):
+            adapter.gemini_credit_source_marker(cast(adapter.GeminiCreditSourceV2Contract, forged), **kwargs)
+    marker = adapter.gemini_credit_source_marker(contract, **kwargs)
+    for changed in (
+        {**marker, "schema_version": True},
+        {**marker, "profile_id": str(marker["profile_id"]).replace("0913", "0912")},
+        {**marker, "source_plugin_uid": GEMINI_SOURCE_UID},
+        {**marker, "extra": "not-authority"},
+    ):
+        with pytest.raises(adapter.GeminiCreditSourceV2AdapterError):
+            adapter.restore_gemini_credit_source_contract(
+                changed,
+                **kwargs,
+                source_settings=args["source_settings"],
+                expected_source_runtime_schema_digest=row["source_runtime_schema_digest"],
+            )

api/tests/unit_tests/services/test_model_billing_migration_inventory_gateway.py

@@ -1736,3 +1736,86 @@ def test_credit_llm_family_checks_selected_source_and_every_prospective_target(
         gateway.validate_target("tenant", _CREDIT_LLM_REVISION, inventory)
     with pytest.raises(ModelRouteUnmapped):
         gateway.validate_target("tenant", _CREDIT_LLM_REVISION, {**inventory, "prospective_variants": []})
+
+
+@pytest.mark.parametrize(
+    "source_uid",
+    [
+        "langgenius/gemini:0.9.12@9bb9eff3f9c99a8f1f87bdd5e0f178d5493918939dd0e02825ae8383bd0fd7ed",
+        "langgenius/gemini:0.9.13@4378187cfc50d4253573f4d96baa13228d8ed2eb746ac3bae492a8eaee2da458",
+    ],
+)
+@pytest.mark.parametrize("model", ["gemini-2.5-flash", "gemini-embedding-001", "gemini-embedding-2-preview"])
+def test_v8_real_target_identity_guard_never_normalizes_wrong_models(source_uid: str, model: str) -> None:
+    from services.entities.model_billing_gemini_mapping import GeminiSourceSettings
+    from services.model_billing_source_variants import GeminiV8SourceVariantContext, source_mapping_revision
+    from tests.helpers.unit import model_billing_gemini_v8_profiles as gemini
+
+    row = next(item for item in gemini.profiles(source_uid) if item["source_model"] == model)
+    observed = gemini.schema(row)
+    registry = source_mapping_revision(gemini.registry(), gemini.REVISION, normalize_provider=str)
+    selected = registry.resolve(
+        (gemini.SOURCE_PROVIDER, observed.model_type, model),
+        context=GeminiV8SourceVariantContext(source_uid, GeminiSourceSettings(**row["source_settings"])),
+    )
+    target = gemini.schema(row, runtime=True)
+    module._assert_target_source_identity(selected, target)
+    changed = target.model_dump(mode="json")
+    changed["model"] = "unreviewed-target-identity"
+    with pytest.raises(ModelRouteUnmapped):
+        module._assert_target_source_identity(selected, AIModelEntity.model_validate(changed))
+
+
+@pytest.mark.parametrize("version", [6, 8])
+@pytest.mark.parametrize("model", ["gemini-embedding-001", "gemini-embedding-2-preview"])
+def test_real_v6_v8_dimension_guards_keep_selected_and_prospective_distinct(version: int, model: str) -> None:
+    from services.entities.model_billing_gemini_mapping import GeminiSourceSettings
+    from services.model_billing_source_variants import (
+        GeminiSourceVariantContext,
+        GeminiV8SourceVariantContext,
+        source_mapping_revision,
+    )
+    from tests.helpers.unit import model_billing_gemini_profiles as legacy
+    from tests.helpers.unit import model_billing_gemini_v8_profiles as current
+
+    if version == 6:
+        row = next(item for item in legacy.profiles() if item["source_model"] == model)
+        registry = source_mapping_revision(legacy.registry(), legacy.REVISION, normalize_provider=str)
+        context = GeminiSourceVariantContext(legacy.SOURCE_UID, GeminiSourceSettings(**row["source_settings"]))
+    else:
+        uid = current.SOURCE_UIDS[1]
+        row = next(item for item in current.profiles(uid) if item["source_model"] == model)
+        registry = source_mapping_revision(current.registry(), current.REVISION, normalize_provider=str)
+        context = GeminiV8SourceVariantContext(uid, GeminiSourceSettings(**row["source_settings"]))
+    target = current.schema(row, runtime=True)
+    selected = registry.resolve((current.SOURCE_PROVIDER, target.model_type, model), context=context)
+    module._assert_target_source_identity(selected, target)
+    wrong_target = target.model_dump(mode="json")
+    wrong_target["model"] = "unreviewed-target-identity"
+    with pytest.raises(ModelRouteUnmapped):
+        module._assert_target_source_identity(selected, AIModelEntity.model_validate(wrong_target))
+    actual = module.embedding_dimensions_contract(target)
+    module._assert_target_embedding_dimensions(
+        mapping=selected,
+        kind=target.model_type,
+        prospective_only=False,
+        source={"embedding_dimensions": actual},
+        schema=target,
+    )
+    with pytest.raises(ModelRouteUnmapped):
+        module._assert_target_embedding_dimensions(
+            mapping=selected,
+            kind=target.model_type,
+            prospective_only=False,
+            source={"embedding_dimensions": "unreviewed-vector-space"},
+            schema=target,
+        )
+    # There is no installed-source dimension fact for a prospective row. The
+    # production path independently verifies its exact runtime schema first.
+    module._assert_target_embedding_dimensions(
+        mapping=selected,
+        kind=target.model_type,
+        prospective_only=True,
+        source={},
+        schema=target,
+    )

api/tests/unit_tests/services/test_model_billing_source_variants.py

@@ -16,6 +16,7 @@ from core.entities.provider_entities import ProviderQuotaType, QuotaConfiguratio
 from core.hosting_configuration import HostingProvider, PaidHostingQuota
 from graphon.model_runtime.entities.model_entities import AIModelEntity, ModelType
 from services.entities.model_billing_embedding_mapping import SelectedEmbeddingSourceVariant
+from services.entities.model_billing_gemini_v8_mapping import SelectedGeminiV8SourceVariant
 from services.entities.model_billing_mapping import (
     ModelMapping,
     SelectedSourceVariant,
@@ -44,6 +45,7 @@ from services.model_billing_openai_provider110_adapter import (
 from services.model_billing_openai_provider110_adapter import _REVIEWED_CONTRACTS
 from services.model_billing_openai_provider110_adapter import _REVIEWED_CONTRACTS as CURRENT_SOURCE_CONTRACTS
 from services.model_billing_source_variants import SourceVariantContext, selection_payload, source_endpoint_class
+from tests.helpers.unit import model_billing_gemini_v8_profiles as gemini_v8
 from tests.helpers.unit.model_billing_source_profiles import (
     DAEMON_REVISION,
     EMBEDDING_MODELS,
@@ -2133,3 +2135,179 @@ def test_credit_llm_family_append_preserves_all_prior_registry_shapes_and_digest
         drifted.restore_selection(
             (SOURCE_PROVIDER, ModelType.LLM, frozen.source.model), selection_payload(frozen).model_dump(mode="json")
         )
+
+
+@pytest.mark.parametrize("source_uid", gemini_v8.SOURCE_UIDS)
+@pytest.mark.parametrize("model", [row["source_model"] for row in gemini_v8.profiles(gemini_v8.SOURCE_UIDS[0])])
+def test_v8_real_two_uid_resolution_factory_restore_and_prospective(source_uid: str, model: str) -> None:
+    from services import model_billing_gemini_credit_source_adapter as legacy
+    from services import model_billing_gemini_credit_source_v2_adapter as current
+    from services.entities.model_billing_gemini_v8_mapping import SourceGeminiV8VariantSelection
+    from services.model_billing_mapping_service import hosted_gemini_source_fact, mapping_has_required_operations
+
+    rows = gemini_v8.profiles(source_uid)
+    row = next(item for item in rows if item["source_model"] == model)
+    schemas = [gemini_v8.schema(item) for item in rows]
+    observed = gemini_v8.schema(row)
+    registry = source_mapping_revision(gemini_v8.registry(), gemini_v8.REVISION)
+    fact = hosted_gemini_source_fact(
+        provider=gemini_v8.SOURCE_PROVIDER,
+        source_uid=source_uid,
+        models=schemas,
+        credentials={"google_api_key": "unit-source-key-never-forwarded"},
+    )
+    selected = resolve_source_mapping(
+        registry,
+        (gemini_v8.SOURCE_PROVIDER, observed.model_type, model),
+        source_uid=source_uid,
+        fact=None,
+        gemini_fact=fact,
+        schema=observed,
+    )
+    assert type(selected) is SelectedGeminiV8SourceVariant
+    assert selected.variant.source_plugin_uid == source_uid
+    assert mapping_has_required_operations(selected)
+    contract = build_source_adapter_contract(
+        selected,
+        source_provider=gemini_v8.SOURCE_PROVIDER,
+        source_plugin_unique_identifier=source_uid,
+        source_schema=observed,
+        source_credentials=None,
+    )
+    expected_type = (
+        legacy.GeminiCreditSourceContract
+        if source_uid == gemini_v8.SOURCE_UIDS[0]
+        else current.GeminiCreditSourceV2Contract
+    )
+    assert type(contract) is expected_type
+    assert contract == restore_variant_source_contract(selected)
+    marker = source_parameter_marker(
+        contract,
+        source_provider=gemini_v8.SOURCE_PROVIDER,
+        source_model=model,
+        model_type=observed.model_type,
+        target_model=model,
+    )
+    assert marker["source_plugin_uid"] == source_uid
+    assert "unit-source-key" not in json.dumps(marker)
+    frozen = selection_payload(selected)
+    assert type(frozen) is SourceGeminiV8VariantSelection
+    assert frozen.schema_version == 8
+    assert (
+        registry.restore_selection(
+            (gemini_v8.SOURCE_PROVIDER, observed.model_type, model), frozen.model_dump(mode="json")
+        )
+        == selected
+    )
+    assert len(registry.selected_prospective()) == 18
+    validate_frozen_mapping_evidence(
+        selected,
+        provider=gemini_v8.SOURCE_PROVIDER,
+        model=model,
+        model_type=observed.model_type,
+        target_model=model,
+        operations=tuple(selected.operations),
+        source_contract=contract,
+        tts_contract=None,
+    )
+
+
+@pytest.mark.parametrize("source_uid", gemini_v8.SOURCE_UIDS)
+def test_v8_fresh_facts_unknown_uid_drift_and_wrong_context_cannot_authorize(source_uid: str) -> None:
+    from core.entities.provider_entities import HostedGeminiSourceFact
+    from services.entities.model_billing_gemini_mapping import GeminiSourceSettings
+    from services.model_billing_mapping_service import (
+        gemini_source_variant_context,
+        gemini_v8_source_variant_context,
+        hosted_gemini_source_fact,
+    )
+    from services.model_billing_source_variants import GeminiSourceVariantContext, GeminiV8SourceVariantContext
+
+    row = gemini_v8.profiles(source_uid)[0]
+    observed = gemini_v8.schema(row)
+    identity = (gemini_v8.SOURCE_PROVIDER, observed.model_type, row["source_model"])
+    registry = source_mapping_revision(gemini_v8.registry(), gemini_v8.REVISION)
+    fact = hosted_gemini_source_fact(
+        provider=gemini_v8.SOURCE_PROVIDER,
+        source_uid=source_uid,
+        models=[observed],
+        credentials={},
+    )
+    assert fact is not None
+
+    def context(value: HostedGeminiSourceFact | None) -> GeminiV8SourceVariantContext | None:
+        return gemini_v8_source_variant_context(
+            value,
+            provider=gemini_v8.SOURCE_PROVIDER,
+            source_uid=source_uid,
+            model_type=observed.model_type,
+            schema=observed,
+        )
+
+    assert context(fact) is not None
+    assert context(None) is None
+    assert context(replace(fact, plugin_unique_identifier="unknown")) is None
+    assert context(replace(fact, catalog_schema_digests=())) is None
+    assert context(replace(fact, llm_file_url_policy="unreviewed")) is None
+    assert (
+        hosted_gemini_source_fact(
+            provider=gemini_v8.SOURCE_PROVIDER,
+            source_uid=source_uid + "x",
+            models=[observed],
+            credentials={},
+        )
+        is None
+    )
+    assert (
+        hosted_gemini_source_fact(
+            provider=gemini_v8.SOURCE_PROVIDER,
+            source_uid=source_uid,
+            models=[observed],
+            credentials=None,
+        )
+        is None
+    )
+    with pytest.raises(ValueError, match="V8 Gemini fact family"):
+        registry.resolve(
+            identity, context=GeminiSourceVariantContext(source_uid, GeminiSourceSettings(**row["source_settings"]))
+        )
+    if source_uid == gemini_v8.SOURCE_UIDS[1]:
+        assert (
+            gemini_source_variant_context(
+                fact,
+                provider=gemini_v8.SOURCE_PROVIDER,
+                source_uid=source_uid,
+                model_type=observed.model_type,
+                schema=observed,
+            )
+            is None
+        )
+
+
+@pytest.mark.parametrize(
+    ("field", "error"),
+    [
+        ("schema_version", "integer 8"),
+        ("variant_id", "Frozen source variant changed"),
+        ("source_plugin_uid", "literal_error"),
+        ("registry_digest", "string_pattern_mismatch"),
+        ("source_runtime_schema_digest", "string_pattern_mismatch"),
+    ],
+)
+def test_v8_saved_selection_tampering_is_not_a_revision_rebase(field: str, error: str) -> None:
+    from services.entities.model_billing_gemini_mapping import GeminiSourceSettings
+    from services.model_billing_source_variants import GeminiV8SourceVariantContext
+
+    source_uid = gemini_v8.SOURCE_UIDS[1]
+    row = gemini_v8.profiles(source_uid)[0]
+    observed = gemini_v8.schema(row)
+    identity = (gemini_v8.SOURCE_PROVIDER, observed.model_type, row["source_model"])
+    registry = source_mapping_revision(gemini_v8.registry(), gemini_v8.REVISION)
+    selected = registry.resolve(
+        identity, context=GeminiV8SourceVariantContext(source_uid, GeminiSourceSettings(**row["source_settings"]))
+    )
+    assert isinstance(selected, SelectedGeminiV8SourceVariant)
+    payload = selection_payload(selected).model_dump(mode="json")
+    payload[field] = True if field == "schema_version" else "changed"
+    with pytest.raises(ValueError, match=error):
+        registry.restore_selection(identity, payload)