fix: generate capability role schemas from dotted defs

Author: FanouZeng-TT

preprocess_schemas.py

@@ -317,6 +317,19 @@ def preprocess_full_schema(schema, entity_def=None):
 # --- Dotted $defs Flattening ---
 
 
+def _rewrite_defs_ref_path(rest, rename_map):
+    """Returns the rewritten $defs fragment path when a renamed target matches."""
+    for old, new in sorted(
+        rename_map.items(), key=lambda item: len(item[0]), reverse=True
+    ):
+        if rest == old:
+            return new
+        prefix = old + "/"
+        if rest.startswith(prefix):
+            return new + rest[len(old) :]
+    return None
+
+
 def _rewrite_local_defs_refs(node, rename_map):
     """Walks a schema tree and rewrites local $defs refs whose target was renamed."""
     prefix = "#/$defs/"
@@ -327,9 +340,9 @@ def _rewrite_local_defs_refs(node, rename_map):
         if not isinstance(ref, str) or not ref.startswith(prefix):
             continue
         rest = ref[len(prefix) :]
-        name, sep, tail = rest.partition("/")
-        if name in rename_map:
-            n["$ref"] = prefix + rename_map[name] + (sep + tail if sep else "")
+        rewritten = _rewrite_defs_ref_path(rest, rename_map)
+        if rewritten is not None:
+            n["$ref"] = prefix + rewritten
 
 
 def _rewrite_external_defs_refs(schema_path, schema, global_rename_maps):
@@ -362,11 +375,9 @@ def _rewrite_external_defs_refs(schema_path, schema, global_rename_maps):
         rename_map = global_rename_maps[target_path_str]
 
         rest = fragment_part[len(prefix) :]
-        name, sep, tail = rest.partition("/")
-        if name in rename_map:
-            new_name = rename_map[name]
-            new_fragment = prefix + new_name + (sep + tail if sep else "")
-            n["$ref"] = file_part + "#" + new_fragment
+        rewritten = _rewrite_defs_ref_path(rest, rename_map)
+        if rewritten is not None:
+            n["$ref"] = file_part + "#" + prefix + rewritten
 
 
 def flatten_dotted_defs(schema):
@@ -384,16 +395,44 @@ def flatten_dotted_defs(schema):
     class name like 'Checkout'); fall back to dot-replaced-with-underscore
     (e.g. 'DevUcpShoppingFulfillment') if the bare tail would collide with
     an existing def in the same file.
+
+    Capability role containers: a dotted def whose value holds exactly the
+    'platform_schema' and 'business_schema' keys is not a schema itself but the
+    mount point where a capability contributes its two role schemas. Renaming it
+    whole would only produce a meaningless Any alias, so it is split into two
+    generatable defs ('<tail>_platform_schema' / '<tail>_business_schema');
+    refs into the container ('.../<role>') are remapped to the split defs.
     """
     defs = schema.get("$defs")
     if not isinstance(defs, dict):
         return {}
 
     existing = set(defs.keys())
     rename_map = {}
+    split_map = {}
     for old in list(defs.keys()):
         if "." not in old:
             continue
+        value = defs[old]
+        if isinstance(value, dict) and set(value.keys()) == {
+            "platform_schema",
+            "business_schema",
+        }:
+            tail = old.rsplit(".", 1)[-1]
+            platform_key = tail + "_platform_schema"
+            business_key = tail + "_business_schema"
+            if platform_key in existing or business_key in existing:
+                # Both split candidates collide; leave as-is rather than
+                # risk corruption.
+                continue
+            defs[platform_key] = value["platform_schema"]
+            defs[business_key] = value["business_schema"]
+            del defs[old]
+            existing.discard(old)
+            existing.update([platform_key, business_key])
+            split_map[old + "/platform_schema"] = platform_key
+            split_map[old + "/business_schema"] = business_key
+            continue
         tail = old.rsplit(".", 1)[-1]
         if tail and tail not in existing:
             new = tail
@@ -406,11 +445,12 @@ class name like 'Checkout'); fall back to dot-replaced-with-underscore
         existing.discard(old)
         existing.add(new)
 
-    if not rename_map:
+    if not rename_map and not split_map:
         return {}
 
     for old, new in rename_map.items():
         defs[new] = defs.pop(old)
+    rename_map.update(split_map)
     _rewrite_local_defs_refs(schema, rename_map)
     return rename_map
 

src/ucp_sdk/models/schemas/common/identity_linking.py

@@ -23,7 +23,9 @@
 from pydantic import BaseModel, ConfigDict, Field
 from typing_extensions import TypeAliasType
 
+from ..capability import BusinessSchema, PlatformSchema
 from .types import description as description_1
+from .types import reverse_domain_name
 
 IdentityLinking = TypeAliasType(
     "IdentityLinking", Annotated[Any, Field(..., title="Identity Linking")]
@@ -76,4 +78,38 @@ class Provider(BaseModel):
     """
 
 
-IdentityLinking1 = TypeAliasType("IdentityLinking1", Any)
+class Config(BaseModel):
+    model_config = ConfigDict(
+        extra="allow",
+    )
+    providers: (
+        dict[reverse_domain_name.ReverseDomainName, list[Provider]] | None
+    ) = None
+    """
+    Map of trusted external identity providers keyed by reverse-domain identifier. Each key maps to an array of mechanism entries — an IdP namespace MAY offer multiple token acquisition mechanisms. Declares which upstream IdPs the business will accept JWT bearer assertions from for the Accelerated IdP Flow (chaining via RFC 8693 + RFC 7523). This field is additive: direct OAuth against the business domain via RFC 8414 discovery is always available regardless of 'providers' content. Businesses MUST NOT list their own authorization server here — chaining-to-self is degenerate, and direct OAuth covers that path. When absent, empty, or when no listed mechanism is supported by the platform, platforms run direct OAuth on the business domain.
+    """
+    scopes: dict[ScopeToken, ScopePolicy]
+    """
+    Map of user-authenticated scopes offered by this business. Each key is an OAuth scope string formed as '{capability}:{scope}' (e.g. 'dev.ucp.shopping.order:read'). Scope presence in this map declares that the corresponding operations require a user identity token. Operations not gated by any listed scope operate at whatever access level the business permits; UCP does not prescribe a default. Each value is a per-scope policy object (empty object means user auth required with no additional policy).
+    """
+
+
+class IdentityLinkingPlatformSchema(PlatformSchema):
+    """
+    Platform-level identity linking capability declaration. Platforms advertise support for identity linking; no auth-specific config is required.
+    """
+
+    model_config = ConfigDict(
+        extra="allow",
+    )
+
+
+class IdentityLinkingBusinessSchema(BusinessSchema):
+    """
+    Business-level identity linking configuration. Businesses declare the user-authenticated scopes they offer in 'config.scopes'.
+    """
+
+    model_config = ConfigDict(
+        extra="allow",
+    )
+    config: Config

src/ucp_sdk/models/schemas/common/payment_authentication.py

@@ -55,7 +55,7 @@ class DevUcpCommonPaymentDeviceDataCollectionItem(BaseModel):
     config: Config
 
 
-class Config1(BaseModel):
+class Config2(BaseModel):
     model_config = ConfigDict(
         extra="allow",
     )
@@ -73,7 +73,7 @@ class DevUcpCommonPaymentThreeDsChallengeItem(BaseModel):
     model_config = ConfigDict(
         extra="allow",
     )
-    config: Config1
+    config: Config2
 
 
 class Actions(BaseModel):

src/ucp_sdk/models/schemas/shopping/fulfillment.py

@@ -23,6 +23,7 @@
 from pydantic import BaseModel, ConfigDict, Field
 from typing_extensions import TypeAliasType
 
+from ..capability import BusinessSchema, PlatformSchema
 from ..common.types import description as description_1
 from .catalog_lookup import DetailProduct
 from .catalog_lookup import GetProductRequest as GetProductRequest_1
@@ -34,6 +35,7 @@
 from .catalog_search import SearchResponse as SearchResponse_1
 from .checkout import Checkout as Checkout_1
 from .types import availability as availability_1
+from .types import business_fulfillment_config
 from .types import fulfillment as fulfillment_1
 from .types import (
     fulfillment_available_method,
@@ -42,6 +44,7 @@
     fulfillment_method,
     fulfillment_option,
     fulfillment_option_base,
+    platform_fulfillment_config,
 )
 from .types.product import Product
 from .types.search_filters import SearchFilters
@@ -56,9 +59,6 @@
 """
 
 
-DevUcpShoppingFulfillment = TypeAliasType("DevUcpShoppingFulfillment", Any)
-
-
 FulfillmentAvailableMethod = TypeAliasType(
     "FulfillmentAvailableMethod",
     fulfillment_available_method.FulfillmentAvailableMethod,
@@ -109,6 +109,34 @@ class CatalogFulfillment(BaseModel):
     """
 
 
+class FulfillmentPlatformSchema(PlatformSchema):
+    """
+    Platform-level fulfillment capability configuration
+    """
+
+    model_config = ConfigDict(
+        extra="allow",
+    )
+    config: platform_fulfillment_config.PlatformFulfillmentConfig | None = None
+    """
+    Platform fulfillment configuration
+    """
+
+
+class FulfillmentBusinessSchema(BusinessSchema):
+    """
+    Business-level fulfillment capability configuration
+    """
+
+    model_config = ConfigDict(
+        extra="allow",
+    )
+    config: business_fulfillment_config.BusinessFulfillmentConfig | None = None
+    """
+    Business fulfillment configuration
+    """
+
+
 FulfillmentOption = TypeAliasType(
     "FulfillmentOption", fulfillment_option.FulfillmentOption
 )

tests/test_codegen_pipeline.py

@@ -298,6 +298,58 @@ def test_flatten_dotted_defs_rewrites_local_refs(self) -> None:
             "#/$defs/dev_ucp_shopping_checkout",
         )
 
+    def test_flatten_dotted_defs_splits_capability_role_containers(
+        self,
+    ) -> None:
+        """Role containers split into two defs and refs into them follow."""
+        role_platform = {"title": "Platform", "allOf": [{"type": "object"}]}
+        role_business = {"title": "Business", "allOf": [{"type": "object"}]}
+        schema = {
+            "$defs": {
+                "dev.ucp.common.identity_linking": {
+                    "platform_schema": role_platform,
+                    "business_schema": role_business,
+                },
+            },
+            "properties": {
+                "platform": {
+                    "$ref": "#/$defs/dev.ucp.common.identity_linking/platform_schema"
+                },
+                "business": {
+                    "$ref": "#/$defs/dev.ucp.common.identity_linking/business_schema"
+                },
+            },
+        }
+
+        rename_map = preprocess_schemas.flatten_dotted_defs(schema)
+
+        self.assertEqual(
+            rename_map,
+            {
+                "dev.ucp.common.identity_linking/platform_schema": (
+                    "identity_linking_platform_schema"
+                ),
+                "dev.ucp.common.identity_linking/business_schema": (
+                    "identity_linking_business_schema"
+                ),
+            },
+        )
+        self.assertEqual(
+            schema["$defs"],
+            {
+                "identity_linking_platform_schema": role_platform,
+                "identity_linking_business_schema": role_business,
+            },
+        )
+        self.assertEqual(
+            schema["properties"]["platform"]["$ref"],
+            "#/$defs/identity_linking_platform_schema",
+        )
+        self.assertEqual(
+            schema["properties"]["business"]["$ref"],
+            "#/$defs/identity_linking_business_schema",
+        )
+
     def test_rewrite_external_defs_refs_uses_target_rename_map(self) -> None:
         """External references follow renames made in the target schema."""
         with tempfile.TemporaryDirectory() as temp_dir:
@@ -1084,6 +1136,73 @@ def test_request_variants_enforce_property_names(self):
                 )
 
 
+@unittest.skipUnless(
+    HAVE_SDK, "requires the installed package (pip install -e .)"
+)
+class IdentityLinkingRoleSchemaTest(unittest.TestCase):
+    """identity_linking.json keeps its role schemas instead of Any aliases.
+
+    The dotted 'dev.ucp.common.identity_linking' def is a capability role
+    container. Flattening must split it into two generatable defs so the
+    business role keeps the upstream contract: 'config.scopes' required with
+    OAuth scope-token keys.
+    """
+
+    def _business(self):
+        from ucp_sdk.models.schemas.common.identity_linking import (
+            IdentityLinkingBusinessSchema,
+        )
+
+        return IdentityLinkingBusinessSchema
+
+    def _base(self):
+        return {
+            "version": "2026-08-25",
+            "schema": "https://ucp.dev/2026-08-25/schemas/common/identity_linking",
+        }
+
+    def test_platform_role_schema_exists(self):
+        from ucp_sdk.models.schemas.common.identity_linking import (
+            IdentityLinkingPlatformSchema,
+        )
+
+        IdentityLinkingPlatformSchema(
+            version="2026-08-25",
+            **{
+                "schema": "https://ucp.dev/2026-08-25/schemas/common/identity_linking"
+            },
+            spec="https://ucp.dev/specification/common/identity-linking",
+        )
+
+    def test_business_config_with_scopes_accepted(self):
+        obj = self._business().model_validate(
+            {
+                **self._base(),
+                "config": {"scopes": {"dev.ucp.shopping.order:read": {}}},
+            }
+        )
+        self.assertEqual(
+            list(obj.config.scopes), ["dev.ucp.shopping.order:read"]
+        )
+        self.assertIsNone(
+            obj.config.scopes["dev.ucp.shopping.order:read"].description
+        )
+
+    def test_missing_config_rejected(self):
+        with self.assertRaisesRegex(ValidationError, "config"):
+            self._business().model_validate(self._base())
+
+    def test_config_without_scopes_rejected(self):
+        with self.assertRaisesRegex(ValidationError, "scopes"):
+            self._business().model_validate({**self._base(), "config": {}})
+
+    def test_malformed_scope_key_rejected(self):
+        with self.assertRaisesRegex(ValidationError, "pattern"):
+            self._business().model_validate(
+                {**self._base(), "config": {"scopes": {"BAD": {}}}}
+            )
+
+
 class PropertyNamesInjectorTest(unittest.TestCase):
     """The propertyNames post-generation injector's own behavior."""