Stream extra component defs with resolve scripts

Resolved SX content may reference components not in the initial shell
scan (e.g. ~cart-mini from IO-generated headers). Diff the needed
components against what the shell already sent and prepend any extra
defcomps as a <script type="text/sx"> block before the resolve script.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

shared/sx/helpers.py

@@ -944,13 +944,22 @@ def sx_page_streaming_parts(ctx: dict, page_html: str, *,
     return shell, tail
 
 
-def sx_streaming_resolve_script(suspension_id: str, sx_source: str) -> str:
+def sx_streaming_resolve_script(suspension_id: str, sx_source: str,
-    """Build a <script> tag that resolves a streaming suspense placeholder."""
+                                extra_components: str = "") -> str:
+    """Build a <script> tag that resolves a streaming suspense placeholder.
+
+    If *extra_components* is non-empty, a ``<script type="text/sx">`` block
+    is prepended so the client loads those component defs before resolving.
+    """
     import json
-    return _SX_STREAMING_RESOLVE.format(
+    parts = []
+    if extra_components:
+        parts.append(f'<script type="text/sx">{extra_components}</script>')
+    parts.append(_SX_STREAMING_RESOLVE.format(
         id=json.dumps(suspension_id),
         sx=json.dumps(sx_source),
-    )
+    ))
+    return "\n".join(parts)
 
 
 _SCRIPT_HASH_CACHE: dict[str, str] = {}

shared/sx/pages.py

@@ -439,6 +439,37 @@ async def execute_page_streaming(
         tctx, initial_page_html, page_sx=page_sx_for_scan,
     )
 
+    # Capture component env + extras scanner while we still have context.
+    # Resolved SX may reference components not in the initial scan
+    # (e.g. ~cart-mini from IO-generated header content).
+    from .jinja_bridge import components_for_page as _comp_scan
+    from quart import current_app as _ca
+    _service = _ca.name
+    # Track which components were already sent in the shell
+    _shell_scan = page_sx_for_scan
+
+    def _extra_defs(sx_source: str) -> str:
+        """Return component defs needed by sx_source but not in shell."""
+        from .deps import components_needed
+        comp_env = dict(get_component_env())
+        shell_needed = components_needed(_shell_scan, comp_env)
+        resolve_needed = components_needed(sx_source, comp_env)
+        extra = resolve_needed - shell_needed
+        if not extra:
+            return ""
+        from .parser import serialize
+        from .types import Component
+        parts = []
+        for key, val in comp_env.items():
+            if isinstance(val, Component) and (f"~{val.name}" in extra or key in extra):
+                param_strs = ["&key"] + list(val.params)
+                if val.has_children:
+                    param_strs.extend(["&rest", "children"])
+                params_sx = "(" + " ".join(param_strs) + ")"
+                body_sx = serialize(val.body, pretty=True)
+                parts.append(f"(defcomp ~{val.name} {params_sx} {body_sx})")
+        return "\n".join(parts)
+
     # --- Return async generator that yields chunks ---
     # No context access needed below — just awaiting tasks and yielding strings.
 
@@ -462,11 +493,13 @@ async def execute_page_streaming(
 
                 if label == "data":
                     content_sx, filter_sx, aside_sx, menu_sx = result
-                    yield sx_streaming_resolve_script("stream-content", content_sx)
+                    extras = _extra_defs(content_sx)
+                    yield sx_streaming_resolve_script("stream-content", content_sx, extras)
                 elif label == "headers":
                     header_rows, header_menu = result
                     if header_rows:
-                        yield sx_streaming_resolve_script("stream-headers", header_rows)
+                        extras = _extra_defs(header_rows)
+                        yield sx_streaming_resolve_script("stream-headers", header_rows, extras)
 
         yield "\n</body>\n</html>"