feat: extract workflow runtime phase 3 foundation

backend/app/domains/pipeline/tasks/render_order_line.py

@@ -71,8 +71,13 @@ def render_order_line_task(self, order_line_id: str):
     emit(order_line_id, "Celery render task started")
     try:
         from sqlalchemy import create_engine, select, update as sql_update
-        from sqlalchemy.orm import Session, joinedload
+        from sqlalchemy.orm import Session
         from app.config import settings as app_settings
+        from app.domains.rendering.workflow_runtime_services import (
+            prepare_order_line_render_context,
+            resolve_order_line_template_context,
+            resolve_render_position_context,
+        )
 
         # Use sync session for Celery (no async event loop)
         sync_url = app_settings.database_url.replace("+asyncpg", "")
@@ -81,219 +86,42 @@ def render_order_line_task(self, order_line_id: str):
         with Session(engine) as session:
             set_tenant_context_sync(session, _tenant_id)
             from app.models.order_line import OrderLine
-            from app.models.product import Product
-
-            emit(order_line_id, "Loading order line from database")
-            line = session.execute(
-                select(OrderLine)
-                .where(OrderLine.id == order_line_id)
-                .options(
-                    joinedload(OrderLine.product).joinedload(Product.cad_file),
-                    joinedload(OrderLine.output_type),
-                )
-            ).scalar_one_or_none()
-
-            if line is None:
-                emit(order_line_id, "Order line not found in database", "error")
-                logger.error(f"OrderLine {order_line_id} not found")
-                return
-
-            # Skip if line was cancelled or order was rejected/completed
-            if line.render_status == "cancelled":
-                emit(order_line_id, "Order line already cancelled — skipping render")
-                logger.info(f"OrderLine {order_line_id} cancelled — skipping")
-                return
-
-            from app.domains.orders.models import Order, OrderStatus
-            order = session.execute(
-                select(Order).where(Order.id == line.order_id)
-            ).scalar_one_or_none()
-            if order and order.status in (OrderStatus.rejected, OrderStatus.completed):
-                emit(order_line_id, f"Order {order.status.value} — skipping render")
-                logger.info(f"OrderLine {order_line_id}: order {order.status.value} — skipping")
-                if line.render_status in ("pending", "processing"):
-                    session.execute(
-                        sql_update(OrderLine)
-                        .where(OrderLine.id == line.id)
-                        .values(render_status="cancelled")
-                    )
-                    session.commit()
-                return
-
-            if line.product.cad_file_id is None:
-                emit(order_line_id, "Product has no CAD file — marking as failed", "error")
-                logger.warning(f"OrderLine {order_line_id}: product has no CAD file")
-                session.execute(
-                    sql_update(OrderLine)
-                    .where(OrderLine.id == line.id)
-                    .values(render_status="failed")
-                )
-                session.commit()
-                return
-
-            # Mark as processing with timing
-            from datetime import datetime
-            render_start = datetime.utcnow()
-            session.execute(
-                sql_update(OrderLine)
-                .where(OrderLine.id == line.id)
-                .values(
-                    render_status="processing",
-                    render_backend_used="celery",
-                    render_started_at=render_start,
-                )
-            )
-            session.commit()
-
-            cad_file = line.product.cad_file
-            materials_source = line.product.cad_part_materials
-
-            # Look up USD master asset for this CAD file — used when rendering
-            # via USD path instead of production GLB
-            from app.domains.media.models import MediaAsset, MediaAssetType
             from pathlib import Path as _Path
-            usd_render_path = None
+
-            if cad_file:
+            setup = prepare_order_line_render_context(
-                _usd_asset = session.execute(
+                session,
-                    select(MediaAsset)
+                order_line_id,
-                    .where(
+                emit=emit,
-                        MediaAsset.cad_file_id == cad_file.id,
-                        MediaAsset.asset_type == MediaAssetType.usd_master,
             )
-                    .order_by(MediaAsset.created_at.desc())
+            if not setup.is_ready:
-                    .limit(1)
+                return
-                ).scalar_one_or_none()
+
-                if _usd_asset and _usd_asset.storage_key:
+            line = setup.order_line
-                    _usd_candidate = _Path(app_settings.upload_dir) / _usd_asset.storage_key
+            cad_file = setup.cad_file
-                    if _usd_candidate.exists():
+            materials_source = setup.materials_source
-                        usd_render_path = _usd_candidate
+            usd_render_path = setup.usd_render_path
-                        logger.info(
+            glb_reuse_path = setup.glb_reuse_path
-                            "render_order_line: using usd_master %s for cad %s",
+            part_colors = setup.part_colors
-                            _usd_candidate.name, cad_file.id,
+            render_start = setup.render_start
+
+            template_context = resolve_order_line_template_context(
+                session,
+                setup,
+                emit=emit,
             )
-
+            template = template_context.template
-            # Look up existing GLB geometry asset — reuse to skip re-tessellation
+            material_library = template_context.material_library
-            # when rendering via the GLB path (non-USD fallback).
+            material_map = template_context.material_map
-            glb_reuse_path = None
+            use_materials = template_context.use_materials
-            if cad_file and not usd_render_path:
+            override_mat = template_context.override_material
-                _glb_asset = session.execute(
-                    select(MediaAsset)
-                    .where(
-                        MediaAsset.cad_file_id == cad_file.id,
-                        MediaAsset.asset_type == MediaAssetType.gltf_geometry,
-                    )
-                    .order_by(MediaAsset.created_at.desc())
-                    .limit(1)
-                ).scalar_one_or_none()
-                if _glb_asset and _glb_asset.storage_key:
-                    _glb_candidate = _Path(app_settings.upload_dir) / _glb_asset.storage_key
-                    if _glb_candidate.exists() and _glb_candidate.stat().st_size > 0:
-                        # Copy to the path render_blender.py expects so its
-                        # local cache check (`glb_path.exists()`) finds it.
-                        _step_path = _Path(cad_file.stored_path)
-                        _expected_glb = _step_path.parent / f"{_step_path.stem}_thumbnail.glb"
-                        if not _expected_glb.exists() or _expected_glb.stat().st_size == 0:
-                            try:
-                                import shutil as _shutil
-                                _shutil.copy2(str(_glb_candidate), str(_expected_glb))
-                                logger.info(
-                                    "render_order_line: reused gltf_geometry asset %s -> %s",
-                                    _glb_candidate.name, _expected_glb.name,
-                                )
-                                glb_reuse_path = _expected_glb
-                            except Exception as _copy_exc:
-                                logger.warning(
-                                    "render_order_line: failed to copy GLB asset: %s", _copy_exc,
-                                )
-                        else:
-                            glb_reuse_path = _expected_glb
-
-            if usd_render_path:
-                emit(order_line_id, "Using USD master for render (skipping GLB tessellation)")
-            elif glb_reuse_path:
-                emit(order_line_id, f"Reusing cached GLB geometry ({glb_reuse_path.name}) — skipping re-tessellation")
-            else:
-                emit(order_line_id, "No USD master or cached GLB — will tessellate STEP -> GLB")
-
-            part_colors = {}
-            if cad_file and cad_file.parsed_objects:
-                parsed_names = cad_file.parsed_objects.get("objects", [])
-                if materials_source:
-                    from app.services.step_processor import build_part_colors
-                    part_colors = build_part_colors(parsed_names, materials_source)
-
-            # Resolve render template + material library
-            from app.services.template_service import resolve_template, get_material_library_path
-
-            category_key = line.product.category_key if line.product else None
-            ot_id = str(line.output_type_id) if line.output_type_id else None
-            template = resolve_template(category_key=category_key, output_type_id=ot_id)
-            material_library = get_material_library_path()
-
-            # Build material_map (part_name → material_name) for material replacement.
-            # Works with or without a render template — only suppressed if a
-            # template explicitly has material_replace_enabled=False.
-            material_map = None
-            use_materials = bool(material_library and materials_source)
-            if template and not template.material_replace_enabled:
-                use_materials = False
-            if use_materials:
-                material_map = {
-                    m["part_name"]: m["material"]
-                    for m in materials_source
-                    if m.get("part_name") and m.get("material")
-                }
-
-                # Resolve raw material names to HARTOMAT library names via aliases
-                from app.services.material_service import resolve_material_map
-                material_map = resolve_material_map(material_map)
-
-            # Apply material override: per-line override takes priority over output type override
-            _line_override = getattr(line, 'material_override', None)
-            _ot_override = line.output_type.material_override if line.output_type else None
-            override_mat = _line_override or _ot_override
-            if override_mat:
-                # Build override map from existing material_map keys or from parsed STEP parts
-                override_keys = set()
-                if material_map:
-                    override_keys = set(material_map.keys())
-                if cad_file and cad_file.parsed_objects:
-                    for part_name in cad_file.parsed_objects.get("objects", []):
-                        override_keys.add(part_name)
-                material_map = {k: override_mat for k in override_keys}
-                use_materials = True
-                emit(order_line_id, f"Material override active: {len(material_map)} parts → {override_mat}")
-
-            if template:
-                emit(order_line_id, f"Using render template: {template.name} (collection={template.target_collection}, material_replace={template.material_replace_enabled}, lighting_only={template.lighting_only})")
-                logger.info(f"Render template resolved: '{template.name}' path={template.blend_file_path}, lighting_only={template.lighting_only}")
-            else:
-                emit(order_line_id, "No render template found — using factory settings (Mode A)")
-                logger.info(f"No render template for category_key={category_key!r}, output_type_id={ot_id!r}")
 
             cad_name = cad_file.original_name if cad_file else "?"
-            # Load render_position for rotation values (per-product takes priority, falls back to global)
+            position_context = resolve_render_position_context(session, line, emit=emit)
-            rotation_x = rotation_y = rotation_z = 0.0
+            rotation_x = position_context.rotation_x
-            focal_length_mm = None
+            rotation_y = position_context.rotation_y
-            sensor_width_mm = None
+            rotation_z = position_context.rotation_z
-            if line.render_position_id:
+            focal_length_mm = position_context.focal_length_mm
-                from app.models.render_position import ProductRenderPosition
+            sensor_width_mm = position_context.sensor_width_mm
-                rp = session.get(ProductRenderPosition, line.render_position_id)
-                if rp:
-                    rotation_x, rotation_y, rotation_z = rp.rotation_x, rp.rotation_y, rp.rotation_z
-                    focal_length_mm = rp.focal_length_mm
-                    sensor_width_mm = rp.sensor_width_mm
-                    emit(order_line_id, f"Render position: '{rp.name}' ({rotation_x}°, {rotation_y}°, {rotation_z}°)" +
-                         (f" focal_length={focal_length_mm}mm" if focal_length_mm else ""))
-            elif line.global_render_position_id:
-                from app.models import GlobalRenderPosition
-                grp = session.get(GlobalRenderPosition, line.global_render_position_id)
-                if grp:
-                    rotation_x, rotation_y, rotation_z = grp.rotation_x, grp.rotation_y, grp.rotation_z
-                    focal_length_mm = grp.focal_length_mm
-                    sensor_width_mm = grp.sensor_width_mm
-                    emit(order_line_id, f"Global render position: '{grp.name}' ({rotation_x}°, {rotation_y}°, {rotation_z}°)" +
-                         (f" focal_length={focal_length_mm}mm" if focal_length_mm else ""))
 
             emit(order_line_id, f"Starting render for {cad_name} ({len(part_colors)} coloured parts)")
 

backend/app/domains/rendering/workflow_runtime_services.py

@@ -0,0 +1,391 @@
+from __future__ import annotations
+
+import logging
+import shutil
+from dataclasses import dataclass, field
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Callable, Literal
+
+from sqlalchemy import select, update as sql_update
+from sqlalchemy.orm import Session, joinedload
+
+from app.config import settings as app_settings
+from app.domains.media.models import MediaAsset, MediaAssetType
+from app.domains.orders.models import Order, OrderLine, OrderStatus
+from app.domains.products.models import CadFile, Product
+from app.domains.rendering.models import GlobalRenderPosition, ProductRenderPosition, RenderTemplate
+from app.services.material_service import resolve_material_map
+from app.services.step_processor import build_part_colors
+from app.services.template_service import (
+    get_material_library_path_for_session,
+    resolve_template_for_session,
+)
+
+logger = logging.getLogger(__name__)
+
+EmitFn = Callable[..., None] | None
+SetupStatus = Literal["ready", "skip", "failed", "missing"]
+
+
+@dataclass(slots=True)
+class OrderLineRenderSetupResult:
+    status: SetupStatus
+    order_line: OrderLine | None = None
+    order: Order | None = None
+    cad_file: CadFile | None = None
+    materials_source: list[dict[str, Any]] = field(default_factory=list)
+    usd_render_path: Path | None = None
+    glb_reuse_path: Path | None = None
+    part_colors: dict[str, str] = field(default_factory=dict)
+    render_start: datetime | None = None
+    reason: str | None = None
+
+    @property
+    def is_ready(self) -> bool:
+        return self.status == "ready" and self.order_line is not None and self.cad_file is not None
+
+
+@dataclass(slots=True)
+class RenderPositionContext:
+    rotation_x: float = 0.0
+    rotation_y: float = 0.0
+    rotation_z: float = 0.0
+    focal_length_mm: float | None = None
+    sensor_width_mm: float | None = None
+    source_name: str | None = None
+    source_kind: Literal["product", "global", "default"] = "default"
+
+
+@dataclass(slots=True)
+class TemplateResolutionResult:
+    template: RenderTemplate | None
+    material_library: str | None
+    material_map: dict[str, str] | None
+    use_materials: bool
+    override_material: str | None
+    category_key: str | None
+    output_type_id: str | None
+
+
+def _emit(emit: EmitFn, order_line_id: str, message: str, level: str | None = None) -> None:
+    if emit is None:
+        return
+    if level is None:
+        emit(order_line_id, message)
+    else:
+        emit(order_line_id, message, level)
+
+
+def _resolve_asset_path(storage_key: str | None) -> Path | None:
+    if not storage_key:
+        return None
+    candidate = Path(app_settings.upload_dir) / storage_key
+    if candidate.exists():
+        return candidate
+    return None
+
+
+def prepare_order_line_render_context(
+    session: Session,
+    order_line_id: str,
+    *,
+    emit: EmitFn = None,
+) -> OrderLineRenderSetupResult:
+    """Load and validate the order line, then prepare reusable render inputs."""
+    _emit(emit, order_line_id, "Loading order line from database")
+    line = session.execute(
+        select(OrderLine)
+        .where(OrderLine.id == order_line_id)
+        .options(
+            joinedload(OrderLine.product).joinedload(Product.cad_file),
+            joinedload(OrderLine.output_type),
+        )
+    ).scalar_one_or_none()
+
+    if line is None:
+        _emit(emit, order_line_id, "Order line not found in database", "error")
+        logger.error("OrderLine %s not found", order_line_id)
+        return OrderLineRenderSetupResult(status="missing", reason="order_line_not_found")
+
+    if line.render_status == "cancelled":
+        _emit(emit, order_line_id, "Order line already cancelled — skipping render")
+        logger.info("OrderLine %s cancelled — skipping", order_line_id)
+        return OrderLineRenderSetupResult(
+            status="skip",
+            order_line=line,
+            reason="line_cancelled",
+        )
+
+    order = session.execute(
+        select(Order).where(Order.id == line.order_id)
+    ).scalar_one_or_none()
+    if order and order.status in (OrderStatus.rejected, OrderStatus.completed):
+        _emit(emit, order_line_id, f"Order {order.status.value} — skipping render")
+        logger.info("OrderLine %s: order %s — skipping", order_line_id, order.status.value)
+        if line.render_status in ("pending", "processing"):
+            session.execute(
+                sql_update(OrderLine)
+                .where(OrderLine.id == line.id)
+                .values(render_status="cancelled")
+            )
+            session.commit()
+        return OrderLineRenderSetupResult(
+            status="skip",
+            order_line=line,
+            order=order,
+            reason="order_closed",
+        )
+
+    if line.product is None or line.product.cad_file_id is None or line.product.cad_file is None:
+        _emit(emit, order_line_id, "Product has no CAD file — marking as failed", "error")
+        logger.warning("OrderLine %s: product has no CAD file", order_line_id)
+        session.execute(
+            sql_update(OrderLine)
+            .where(OrderLine.id == line.id)
+            .values(render_status="failed")
+        )
+        session.commit()
+        return OrderLineRenderSetupResult(
+            status="failed",
+            order_line=line,
+            order=order,
+            reason="missing_cad_file",
+        )
+
+    render_start = datetime.utcnow()
+    session.execute(
+        sql_update(OrderLine)
+        .where(OrderLine.id == line.id)
+        .values(
+            render_status="processing",
+            render_backend_used="celery",
+            render_started_at=render_start,
+        )
+    )
+    session.commit()
+
+    cad_file = line.product.cad_file
+    materials_source = line.product.cad_part_materials or []
+
+    usd_render_path = None
+    usd_asset = session.execute(
+        select(MediaAsset)
+        .where(
+            MediaAsset.cad_file_id == cad_file.id,
+            MediaAsset.asset_type == MediaAssetType.usd_master,
+        )
+        .order_by(MediaAsset.created_at.desc())
+        .limit(1)
+    ).scalar_one_or_none()
+    if usd_asset:
+        usd_render_path = _resolve_asset_path(usd_asset.storage_key)
+        if usd_render_path:
+            logger.info(
+                "render_order_line: using usd_master %s for cad %s",
+                usd_render_path.name,
+                cad_file.id,
+            )
+
+    glb_reuse_path = None
+    if not usd_render_path:
+        glb_asset = session.execute(
+            select(MediaAsset)
+            .where(
+                MediaAsset.cad_file_id == cad_file.id,
+                MediaAsset.asset_type == MediaAssetType.gltf_geometry,
+            )
+            .order_by(MediaAsset.created_at.desc())
+            .limit(1)
+        ).scalar_one_or_none()
+        glb_candidate = _resolve_asset_path(glb_asset.storage_key if glb_asset else None)
+        if glb_candidate and glb_candidate.stat().st_size > 0:
+            step_path = Path(cad_file.stored_path)
+            expected_glb = step_path.parent / f"{step_path.stem}_thumbnail.glb"
+            if not expected_glb.exists() or expected_glb.stat().st_size == 0:
+                try:
+                    shutil.copy2(str(glb_candidate), str(expected_glb))
+                    logger.info(
+                        "render_order_line: reused gltf_geometry asset %s -> %s",
+                        glb_candidate.name,
+                        expected_glb.name,
+                    )
+                    glb_reuse_path = expected_glb
+                except Exception as copy_exc:
+                    logger.warning("render_order_line: failed to copy GLB asset: %s", copy_exc)
+            else:
+                glb_reuse_path = expected_glb
+
+    if usd_render_path:
+        _emit(emit, order_line_id, "Using USD master for render (skipping GLB tessellation)")
+    elif glb_reuse_path:
+        _emit(
+            emit,
+            order_line_id,
+            f"Reusing cached GLB geometry ({glb_reuse_path.name}) — skipping re-tessellation",
+        )
+    else:
+        _emit(emit, order_line_id, "No USD master or cached GLB — will tessellate STEP -> GLB")
+
+    part_colors: dict[str, str] = {}
+    if cad_file.parsed_objects:
+        parsed_names = cad_file.parsed_objects.get("objects", [])
+        if materials_source:
+            part_colors = build_part_colors(parsed_names, materials_source)
+
+    return OrderLineRenderSetupResult(
+        status="ready",
+        order_line=line,
+        order=order,
+        cad_file=cad_file,
+        materials_source=materials_source,
+        usd_render_path=usd_render_path,
+        glb_reuse_path=glb_reuse_path,
+        part_colors=part_colors,
+        render_start=render_start,
+    )
+
+
+def resolve_order_line_template_context(
+    session: Session,
+    setup: OrderLineRenderSetupResult,
+    *,
+    emit: EmitFn = None,
+) -> TemplateResolutionResult:
+    """Resolve render template, material library, and material map for a prepared order line."""
+    if not setup.is_ready:
+        raise ValueError("resolve_order_line_template_context requires a ready setup result")
+
+    line = setup.order_line
+    cad_file = setup.cad_file
+    assert line is not None
+    assert cad_file is not None
+    materials_source = setup.materials_source
+    category_key = line.product.category_key if line.product else None
+    output_type_id = str(line.output_type_id) if line.output_type_id else None
+
+    template = resolve_template_for_session(
+        session,
+        category_key=category_key,
+        output_type_id=output_type_id,
+    )
+    material_library = get_material_library_path_for_session(session)
+
+    material_map = None
+    use_materials = bool(material_library and materials_source)
+    if template and not template.material_replace_enabled:
+        use_materials = False
+    if use_materials:
+        material_map = {
+            material["part_name"]: material["material"]
+            for material in materials_source
+            if material.get("part_name") and material.get("material")
+        }
+        material_map = resolve_material_map(material_map)
+
+    line_override = getattr(line, "material_override", None)
+    output_override = line.output_type.material_override if line.output_type else None
+    override_material = line_override or output_override
+    if override_material:
+        override_keys = set(material_map.keys()) if material_map else set()
+        if cad_file and cad_file.parsed_objects:
+            for part_name in cad_file.parsed_objects.get("objects", []):
+                override_keys.add(part_name)
+        material_map = {key: override_material for key in override_keys}
+        use_materials = True
+        _emit(
+            emit,
+            str(line.id),
+            f"Material override active: {len(material_map)} parts → {override_material}",
+        )
+
+    if template:
+        _emit(
+            emit,
+            str(line.id),
+            "Using render template: "
+            f"{template.name} (collection={template.target_collection}, "
+            f"material_replace={template.material_replace_enabled}, "
+            f"lighting_only={template.lighting_only})",
+        )
+        logger.info(
+            "Render template resolved: '%s' path=%s, lighting_only=%s",
+            template.name,
+            template.blend_file_path,
+            template.lighting_only,
+        )
+    if not template:
+        _emit(emit, str(line.id), "No render template found — using factory settings (Mode A)")
+        logger.info(
+            "No render template for category_key=%r, output_type_id=%r",
+            category_key,
+            output_type_id,
+        )
+
+    return TemplateResolutionResult(
+        template=template,
+        material_library=material_library,
+        material_map=material_map,
+        use_materials=use_materials,
+        override_material=override_material,
+        category_key=category_key,
+        output_type_id=output_type_id,
+    )
+
+
+def resolve_render_position_context(
+    session: Session,
+    line: OrderLine,
+    *,
+    emit: EmitFn = None,
+) -> RenderPositionContext:
+    """Resolve per-line render position values with product/global fallback."""
+    if line.render_position_id:
+        render_position = session.get(ProductRenderPosition, line.render_position_id)
+        if render_position:
+            _emit(
+                emit,
+                str(line.id),
+                f"Render position: '{render_position.name}' "
+                f"({render_position.rotation_x}°, {render_position.rotation_y}°, {render_position.rotation_z}°)"
+                + (
+                    f" focal_length={render_position.focal_length_mm}mm"
+                    if render_position.focal_length_mm
+                    else ""
+                ),
+            )
+            return RenderPositionContext(
+                rotation_x=render_position.rotation_x,
+                rotation_y=render_position.rotation_y,
+                rotation_z=render_position.rotation_z,
+                focal_length_mm=render_position.focal_length_mm,
+                sensor_width_mm=render_position.sensor_width_mm,
+                source_name=render_position.name,
+                source_kind="product",
+            )
+
+    if line.global_render_position_id:
+        global_position = session.get(GlobalRenderPosition, line.global_render_position_id)
+        if global_position:
+            _emit(
+                emit,
+                str(line.id),
+                f"Global render position: '{global_position.name}' "
+                f"({global_position.rotation_x}°, {global_position.rotation_y}°, {global_position.rotation_z}°)"
+                + (
+                    f" focal_length={global_position.focal_length_mm}mm"
+                    if global_position.focal_length_mm
+                    else ""
+                ),
+            )
+            return RenderPositionContext(
+                rotation_x=global_position.rotation_x,
+                rotation_y=global_position.rotation_y,
+                rotation_z=global_position.rotation_z,
+                focal_length_mm=global_position.focal_length_mm,
+                sensor_width_mm=global_position.sensor_width_mm,
+                source_name=global_position.name,
+                source_kind="global",
+            )
+
+    return RenderPositionContext()

backend/app/services/template_service.py

@@ -32,20 +32,14 @@ def _get_engine():
     return _engine
 
 
-def resolve_template(
+def resolve_template_for_session(
+    session: Session,
     category_key: str | None = None,
     output_type_id: str | None = None,
 ) -> RenderTemplate | None:
-    """Find the best matching active render template.
+    """Find the best matching active render template on an existing sync session."""
-
-    Uses the M2M render_template_output_types table for output type matching.
-    Uses sync SQLAlchemy — safe for Celery tasks.
-    """
-    engine = _get_engine()
-    with Session(engine) as session:
     active = RenderTemplate.is_active == True  # noqa: E712
 
-        # Helper: subquery checking if a template is linked to a specific OT
     def _has_ot(ot_id):
         return exists(
             select(render_template_output_types.c.template_id).where(and_(
@@ -54,14 +48,12 @@ def resolve_template(
             ))
         )
 
-        # Helper: subquery checking if a template has NO linked OTs
     _no_ots = ~exists(
         select(render_template_output_types.c.template_id).where(
             render_template_output_types.c.template_id == RenderTemplate.id,
         )
     )
 
-        # 1. Exact match: category_key + output_type in M2M
     if category_key and output_type_id:
         row = session.execute(
             select(RenderTemplate).where(and_(
@@ -73,7 +65,6 @@ def resolve_template(
         if row:
             return row
 
-        # 2. Category only: category_key + no OTs linked
     if category_key:
         row = session.execute(
             select(RenderTemplate).where(and_(
@@ -85,7 +76,6 @@ def resolve_template(
         if row:
             return row
 
-        # 3. OT only: no category_key + output_type in M2M
     if output_type_id:
         row = session.execute(
             select(RenderTemplate).where(and_(
@@ -97,15 +87,49 @@ def resolve_template(
         if row:
             return row
 
-        # 4. Global fallback: no category_key + no OTs linked
+    return session.execute(
-        row = session.execute(
         select(RenderTemplate).where(and_(
             active,
             RenderTemplate.category_key.is_(None),
             _no_ots,
         ))
     ).scalar_one_or_none()
-        return row
+
+
+def resolve_template(
+    category_key: str | None = None,
+    output_type_id: str | None = None,
+) -> RenderTemplate | None:
+    """Find the best matching active render template.
+
+    Uses the M2M render_template_output_types table for output type matching.
+    Uses sync SQLAlchemy — safe for Celery tasks.
+    """
+    engine = _get_engine()
+    with Session(engine) as session:
+        return resolve_template_for_session(
+            session,
+            category_key=category_key,
+            output_type_id=output_type_id,
+        )
+
+
+def get_material_library_path_for_session(session: Session) -> str | None:
+    """Return the active material library path on an existing sync session."""
+    from app.domains.materials.models import AssetLibrary
+
+    row = session.execute(
+        select(AssetLibrary).where(AssetLibrary.is_active == True).limit(1)  # noqa: E712
+    ).scalar_one_or_none()
+    if row and row.blend_file_path:
+        return row.blend_file_path
+
+    row = session.execute(
+        select(SystemSetting).where(SystemSetting.key == "material_library_path")
+    ).scalar_one_or_none()
+    if row and row.value and row.value.strip():
+        return row.value.strip()
+    return None
 
 
 def get_material_library_path() -> str | None:
@@ -115,18 +139,4 @@ def get_material_library_path() -> str | None:
     """
     engine = _get_engine()
     with Session(engine) as session:
-        # Prefer active AssetLibrary
+        return get_material_library_path_for_session(session)
-        from app.domains.materials.models import AssetLibrary
-        row = session.execute(
-            select(AssetLibrary).where(AssetLibrary.is_active == True).limit(1)  # noqa: E712
-        ).scalar_one_or_none()
-        if row and row.blend_file_path:
-            return row.blend_file_path
-
-        # Fallback to legacy system setting
-        row = session.execute(
-            select(SystemSetting).where(SystemSetting.key == "material_library_path")
-        ).scalar_one_or_none()
-        if row and row.value and row.value.strip():
-            return row.value.strip()
-        return None

backend/tests/domains/test_workflow_runtime_services.py

@@ -0,0 +1,218 @@
+from __future__ import annotations
+
+import os
+import uuid
+from pathlib import Path
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import Session
+
+from app.database import Base
+from app.domains.auth.models import User, UserRole
+from app.domains.materials.models import AssetLibrary
+from app.domains.media.models import MediaAsset, MediaAssetType
+from app.domains.orders.models import Order, OrderLine, OrderStatus
+from app.domains.products.models import CadFile, Product
+from app.domains.rendering.models import OutputType, RenderTemplate
+from app.domains.rendering.workflow_runtime_services import (
+    prepare_order_line_render_context,
+    resolve_order_line_template_context,
+)
+
+import app.models  # noqa: F401
+
+
+TEST_DB_URL = os.environ.get(
+    "TEST_DATABASE_URL",
+    "postgresql+asyncpg://hartomat:hartomat@localhost:5432/hartomat_test",
+).replace("+asyncpg", "")
+
+
+@pytest.fixture
+def sync_session():
+    engine = create_engine(TEST_DB_URL)
+    with engine.begin() as conn:
+        Base.metadata.create_all(conn)
+
+    session = Session(engine)
+    try:
+        yield session
+    finally:
+        session.close()
+        with engine.begin() as conn:
+            conn.execute(text("DROP SCHEMA public CASCADE"))
+            conn.execute(text("CREATE SCHEMA public"))
+        engine.dispose()
+
+
+def _seed_order_line_graph(session: Session, tmp_path: Path) -> OrderLine:
+    step_path = tmp_path / "parts" / "bearing.step"
+    step_path.parent.mkdir(parents=True, exist_ok=True)
+    step_path.write_text("STEP", encoding="utf-8")
+
+    user = User(
+        id=uuid.uuid4(),
+        email=f"workflow-{uuid.uuid4().hex[:8]}@test.local",
+        password_hash="hash",
+        full_name="Workflow Tester",
+        role=UserRole.admin,
+        is_active=True,
+    )
+    cad_file = CadFile(
+        id=uuid.uuid4(),
+        original_name="bearing.step",
+        stored_path=str(step_path),
+        file_hash=f"hash-{uuid.uuid4().hex}",
+        parsed_objects={"objects": ["InnerRing", "OuterRing"]},
+    )
+    product = Product(
+        id=uuid.uuid4(),
+        pim_id="P-1000",
+        name="Bearing A",
+        category_key="bearings",
+        cad_file_id=cad_file.id,
+        cad_file=cad_file,
+        cad_part_materials=[
+            {"part_name": "InnerRing", "material": "Steel raw"},
+            {"part_name": "OuterRing", "material": "Steel raw"},
+        ],
+    )
+    output_type = OutputType(
+        id=uuid.uuid4(),
+        name=f"Still-{uuid.uuid4().hex[:6]}",
+        renderer="blender",
+        output_format="png",
+        render_settings={"width": 1600, "height": 900},
+        material_override=None,
+    )
+    order = Order(
+        id=uuid.uuid4(),
+        order_number=f"ORD-{uuid.uuid4().hex[:8]}",
+        status=OrderStatus.processing,
+        created_by=user.id,
+    )
+    line = OrderLine(
+        id=uuid.uuid4(),
+        order_id=order.id,
+        product_id=product.id,
+        product=product,
+        output_type_id=output_type.id,
+        output_type=output_type,
+        render_status="pending",
+    )
+
+    session.add_all([user, cad_file, product, output_type, order, line])
+    session.commit()
+    return line
+
+
+def test_prepare_order_line_render_context_marks_line_processing_and_prefers_usd(sync_session, tmp_path, monkeypatch):
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
+    upload_dir = Path(settings.upload_dir)
+    upload_dir.mkdir(parents=True, exist_ok=True)
+
+    line = _seed_order_line_graph(sync_session, tmp_path)
+    usd_asset_path = upload_dir / "usd" / "bearing.usd"
+    usd_asset_path.parent.mkdir(parents=True, exist_ok=True)
+    usd_asset_path.write_text("USD", encoding="utf-8")
+
+    sync_session.add(
+        MediaAsset(
+            id=uuid.uuid4(),
+            cad_file_id=line.product.cad_file_id,
+            product_id=line.product_id,
+            asset_type=MediaAssetType.usd_master,
+            storage_key="usd/bearing.usd",
+        )
+    )
+    sync_session.commit()
+
+    messages: list[str] = []
+
+    result = prepare_order_line_render_context(
+        sync_session,
+        str(line.id),
+        emit=lambda order_line_id, message, level=None: messages.append(message),
+    )
+
+    sync_session.refresh(line)
+
+    assert result.is_ready
+    assert result.usd_render_path == usd_asset_path
+    assert result.glb_reuse_path is None
+    assert result.part_colors == {
+        "InnerRing": "Steel raw",
+        "OuterRing": "Steel raw",
+    }
+    assert line.render_status == "processing"
+    assert line.render_backend_used == "celery"
+    assert line.render_started_at is not None
+    assert any("Using USD master for render" in message for message in messages)
+
+
+def test_prepare_order_line_render_context_skips_closed_orders(sync_session, tmp_path, monkeypatch):
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
+    line = _seed_order_line_graph(sync_session, tmp_path)
+    line.order.status = OrderStatus.completed
+    sync_session.commit()
+
+    result = prepare_order_line_render_context(sync_session, str(line.id))
+    sync_session.refresh(line)
+
+    assert result.status == "skip"
+    assert result.reason == "order_closed"
+    assert line.render_status == "cancelled"
+
+
+def test_resolve_order_line_template_context_uses_exact_template_and_override(sync_session, tmp_path, monkeypatch):
+    from app.config import settings
+
+    monkeypatch.setattr(settings, "upload_dir", str(tmp_path / "uploads"))
+    line = _seed_order_line_graph(sync_session, tmp_path)
+    line.material_override = "HARTOMAT_OVERRIDE"
+
+    sync_session.add(
+        AssetLibrary(
+            id=uuid.uuid4(),
+            name="Default Library",
+            blend_file_path="/libraries/materials.blend",
+            is_active=True,
+        )
+    )
+    template = RenderTemplate(
+        id=uuid.uuid4(),
+        name="Bearing Studio",
+        category_key="bearings",
+        blend_file_path="/templates/bearing.blend",
+        original_filename="bearing.blend",
+        target_collection="Product",
+        material_replace_enabled=False,
+        lighting_only=True,
+        is_active=True,
+        output_types=[line.output_type],
+    )
+    sync_session.add(template)
+    sync_session.commit()
+
+    monkeypatch.setattr(
+        "app.domains.rendering.workflow_runtime_services.resolve_material_map",
+        lambda raw_map: {key: f"resolved:{value}" for key, value in raw_map.items()},
+    )
+
+    setup = prepare_order_line_render_context(sync_session, str(line.id))
+    result = resolve_order_line_template_context(sync_session, setup)
+
+    assert result.template is not None
+    assert result.template.name == "Bearing Studio"
+    assert result.material_library == "/libraries/materials.blend"
+    assert result.override_material == "HARTOMAT_OVERRIDE"
+    assert result.use_materials is True
+    assert result.material_map == {
+        "InnerRing": "HARTOMAT_OVERRIDE",
+        "OuterRing": "HARTOMAT_OVERRIDE",
+    }

docs/workflows/WORKFLOW_DELIVERY_CHECKLIST.md

@@ -22,6 +22,7 @@
 - [ ] Missing legacy steps extracted into reusable executors
 - [ ] Extracted node behavior matches legacy services
 - [ ] Node-level tests cover success and failure paths
+- Progress: `order_line_setup` and `resolve_template` are extracted and covered by targeted backend tests; remaining parity nodes are still open.
 
 ### Phase 4
 

docs/workflows/WORKFLOW_IMPLEMENTATION_BACKLOG.md

@@ -51,8 +51,8 @@
 ### Tickets
 
 - `E3-T1` Create a parity matrix from the legacy render pipeline.
-- `E3-T2` Extract `order_line_setup` into a reusable service/task.
+- `E3-T2` Extract `order_line_setup` into a reusable service/task. `completed`
-- `E3-T3` Extract `resolve_template`.
+- `E3-T3` Extract `resolve_template`. `completed`
 - `E3-T4` Extract `material_map_resolve`.
 - `E3-T5` Extract `auto_populate_materials`.
 - `E3-T6` Extract `glb_bbox`.

docs/workflows/WORKFLOW_MIGRATION_PLAN.md

@@ -8,7 +8,7 @@ Bring `/workflows` to full production parity with the existing legacy render pip
 
 - Phase 1 completed on canonical config storage, preset migration, and legacy-safe runtime extraction.
 - Phase 2 completed on backend node registry, node definitions API, and schema-driven editor palette/settings.
-- Next execution target: Phase 3 legacy step extraction for runtime parity.
+- Phase 3 in progress: `order_line_setup` and `resolve_template` are extracted behind the legacy task boundary and validated with targeted backend tests.
 
 ## Non-Negotiables