feat: support capability extensions in codegen and cart-to-checkout conversion

- Generate request variants across all capability extension $defs in preprocess_schemas.py
- Inject _enforce_cart_conversion validator in postprocess_models.py to allow cart_id or line_items in cart checkout creation
- Fix trailing comma issue when injecting AfterValidator into multiline Annotated aliases
- Add CapabilityExtensionSemanticTest to test_codegen_pipeline.py
- Regenerate models against UCP version 2026-04-08

Author: damaz91

.gitignore

@@ -3,3 +3,4 @@ node_modules/
 __pycache__/
 .venv/
 *.py[cod]
+ucp/

generate_models.sh

@@ -63,7 +63,7 @@ rm -rf "$RAW_SCHEMA_DIR"
 cp -R "$SCHEMA_DIR" "$RAW_SCHEMA_DIR"
 
 echo "Preprocessing schemas..."
-uv run python preprocess_schemas.py
+uv run --no-sync python preprocess_schemas.py
 
 echo "Generating Pydantic models from preprocessed schemas..."
 
@@ -77,9 +77,8 @@ mkdir -p "$OUTPUT_DIR"
 # We use --field-constraints to include validation constraints (regex, min/max, etc.)
 # We use --reuse-model to collapse structurally identical generated types.
 # Note: Formatting is done as a post-processing step.
-uv run \
-    --link-mode=copy \
-    --extra-index-url https://pypi.org/simple python \
+uv run --no-sync \
+    --link-mode=copy python \
     -m datamodel_code_generator \
     --input "$SCHEMA_DIR" \
     --input-file-type jsonschema \
@@ -99,11 +98,11 @@ uv run \
 
 
 echo "Post-processing generated models (constraints the generator ignores)..."
-uv run python postprocess_models.py || exit 1
+uv run --no-sync python postprocess_models.py || exit 1
 
 echo "Formatting generated models..."
-uv run ruff format
-uv run ruff check --fix "$OUTPUT_DIR"
+uv run --no-sync ruff format
+uv run --no-sync ruff check --fix "$OUTPUT_DIR"
 
 
 echo "Done. Models generated in $OUTPUT_DIR"

postprocess_models.py

@@ -667,7 +667,10 @@ def inject_array_contains(source, alias_name, groups, item_condition=None):
                 break
     if close is None:
         return source
-    out = source[:close] + f", AfterValidator({func_name})" + source[close:]
+    before = source[:close].rstrip()
+    if before.endswith(","):
+        before = before[:-1].rstrip()
+    out = before + f", AfterValidator({func_name})" + source[close:]
     func_src = _build_contains_function(func_name, groups, item_condition)
     insert_at = assign_re.search(out).start()
     out = out[:insert_at] + func_src + "\n\n" + out[insert_at:]
@@ -1434,6 +1437,48 @@ def _patch_extra_forbid():
     return patched, 0
 
 
+def _patch_cart_checkout_create_request():
+    """Ensure Checkout in cart_create_request allows cart_id or line_items."""
+    path = OUTPUT_DIR / "shopping" / "cart_create_request.py"
+    if not path.exists():
+        return 0, 0
+    source = path.read_text(encoding="utf-8")
+    if "_enforce_cart_conversion" in source:
+        return 0, 0
+
+    class_match = re.search(
+        r"^class Checkout\(CheckoutCreateRequest\):", source, re.M
+    )
+    if not class_match:
+        return 0, 0
+
+    validator_code = '''    line_items: list[line_item_create_request.LineItemCreateRequest] | None = None
+
+    @model_validator(mode="after")
+    def _enforce_cart_conversion(self):
+        """Require either cart_id or line_items for checkout creation."""
+        if not getattr(self, "cart_id", None) and not getattr(self, "line_items", None):
+            raise ValueError("Either cart_id or line_items must be provided")
+        return self
+'''
+    config_match = re.search(
+        r"model_config = ConfigDict\(\s*extra=\"allow\",?\s*\)",
+        source[class_match.start() :],
+    )
+    if config_match:
+        insert_pos = class_match.start() + config_match.end()
+        source = (
+            source[:insert_pos] + "\n" + validator_code + source[insert_pos:]
+        )
+        source = _ensure_pydantic_import(source, "model_validator")
+        path.write_text(source, encoding="utf-8")
+        sys.stdout.write(
+            f"  cart conversion validator on 'Checkout' -> {path}\n"
+        )
+        return 1, 0
+    return 0, 0
+
+
 def main():
     """Main entry point to scan schemas and patch generated models."""
     patched_mp, rc_mp = _patch_min_properties()
@@ -1443,6 +1488,7 @@ def main():
     patched_cb, rc_cb = _patch_conditional_bounds()
     patched_ui, rc_ui = _patch_unique_items()
     patched_ef, rc_ef = _patch_extra_forbid()
+    patched_cc, rc_cc = _patch_cart_checkout_create_request()
     total = (
         patched_mp
         + patched_pn
@@ -1451,9 +1497,10 @@ def main():
         + patched_cb
         + patched_ui
         + patched_ef
+        + patched_cc
     )
     sys.stdout.write(f"postprocess: {total} module(s) patched\n")
-    return rc_mp or rc_pn or rc_ac or rc_cr or rc_cb or rc_ui or rc_ef
+    return rc_mp or rc_pn or rc_ac or rc_cr or rc_cb or rc_ui or rc_ef or rc_cc
 
 
 if __name__ == "__main__":

preprocess_schemas.py

@@ -377,21 +377,26 @@ class name like 'Checkout'); fall back to dot-replaced-with-underscore
 
 def get_required_ops(schema):
     """
-    Scans a schema for the custom 'ucp_request' metadata.
+    Scans a schema's properties and $defs for custom 'ucp_request' metadata.
     Returns a set of operation keys (e.g. {'create', 'update'}) that need distinct models.
     """
     ops = set()
-    properties = schema.get("properties", {})
-    if not isinstance(properties, dict):
-        return ops
-
-    for data in properties.values():
-        if isinstance(data, dict):
-            marker = data.get("ucp_request")
-            if isinstance(marker, str):
-                ops.update(["create", "update"])  # Standard shortcut
-            elif isinstance(marker, dict):
-                ops.update(marker.keys())
+    containers = []
+    if isinstance(schema.get("properties"), dict):
+        containers.append(schema["properties"])
+    if isinstance(schema.get("$defs"), dict):
+        containers.append(schema["$defs"])
+
+    for container in containers:
+        for node in iter_nodes(container):
+            if not isinstance(node, dict):
+                continue
+            marker = node.get("ucp_request")
+            if marker is not None:
+                if isinstance(marker, str):
+                    ops.update(["create", "update"])  # Standard shortcut
+                elif isinstance(marker, dict):
+                    ops.update(marker.keys())
     return ops
 
 
@@ -527,6 +532,22 @@ def _create_single_variant(
             variant, op, file_path, global_variant_requirements
         )
 
+    # Apply request rules to top-level definitions in $defs
+    defs = variant.get("$defs", {})
+    if isinstance(defs, dict):
+        for node in defs.values():
+            if isinstance(node, dict) and (
+                "properties" in node or node.get("type") == "object"
+            ):
+                _apply_request_rules_to_object(
+                    node, op, file_path, global_variant_requirements
+                )
+
+    # Rewrite all external references across the entire variant tree
+    rewrite_refs_to_variants(
+        variant, op, file_path, global_variant_requirements
+    )
+
     return variant
 
 
@@ -595,20 +616,23 @@ def normalize_metadata_schemas(schemas, target_dir):
 
 
 def extract_external_refs(schema, path):
-    """Finds all relative external file references in the schema properties."""
+    """Finds all relative external file references in properties and $defs."""
     refs = []
-    props = schema.get("properties", {})
-    if not isinstance(props, dict):
-        return refs
-
-    for name, data in props.items():
-        for node in iter_nodes(data):
-            if isinstance(node, dict) and "$ref" in node:
-                ref = node["$ref"]
-                ref_file, _, _ = ref.partition("#")
-                if ref_file:
-                    abs_path = str((path.parent / ref_file).resolve())
-                    refs.append((name, abs_path))
+    containers = []
+    if isinstance(schema.get("properties"), dict):
+        containers.append(schema["properties"])
+    if isinstance(schema.get("$defs"), dict):
+        containers.append(schema["$defs"])
+
+    for container in containers:
+        for name, data in container.items():
+            for node in iter_nodes(data):
+                if isinstance(node, dict) and "$ref" in node:
+                    ref = node["$ref"]
+                    ref_file, _, _ = ref.partition("#")
+                    if ref_file:
+                        abs_path = str((path.parent / ref_file).resolve())
+                        refs.append((name, abs_path))
     return refs
 
 
@@ -629,10 +653,10 @@ def propagate_needs_transitive(variant_needs, schema_refs, schemas):
                     if child_path not in schemas:
                         continue
 
-                    # Only propagate if the property isn't 'omit'ted for this op
-                    data = (
-                        schemas[path].get("properties", {}).get(prop_name, {})
-                    )
+                    # Check properties first, then $defs for any operation override
+                    data = schemas[path].get("properties", {}).get(
+                        prop_name
+                    ) or schemas[path].get("$defs", {}).get(prop_name, {})
                     include, _ = eval_prop_inclusion(
                         prop_name, data, op, schemas[path].get("required", [])
                     )

src/ucp_sdk/models/schemas/__init__.py

@@ -15,3 +15,4 @@
 # generated by datamodel-codegen
 # pylint: disable=all
 # pyformat: disable
+