feat(converter): add cadquery step fallback

converter/cadquery_stage.py

@@ -0,0 +1,179 @@
+import hashlib
+import json
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+import cadquery as cq
+
+
+def hash_id(value: str) -> str:
+    return hashlib.sha1(value.encode("utf-8")).hexdigest()[:16]
+
+
+def json_safe(value: Any) -> Any:
+    if value is None or isinstance(value, (str, int, float, bool)):
+        return value
+    if isinstance(value, dict):
+        return {str(key): json_safe(item) for key, item in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [json_safe(item) for item in value]
+    return str(value)
+
+
+def node_to_metadata(node: cq.Assembly, parent_path: str = "") -> dict[str, Any]:
+    name = node.name or "root"
+    node_path = f"{parent_path}/{name}" if parent_path else name
+    children = [node_to_metadata(child, node_path) for child in node.children]
+    return {
+        "id": hash_id(node_path),
+        "name": name,
+        "path": node_path,
+        "hasShape": bool(node.obj),
+        "metadata": json_safe(node.metadata or {}),
+        "children": children,
+    }
+
+
+def count_nodes(node: dict[str, Any]) -> int:
+    return 1 + sum(count_nodes(child) for child in node.get("children", []))
+
+
+def metadata_dict(value: Any) -> dict[str, Any]:
+    safe_value = json_safe(value or {})
+    return safe_value if isinstance(safe_value, dict) else {"value": safe_value}
+
+
+def get_shape_type(value: Any) -> str | None:
+    try:
+        shape_type = value.ShapeType()
+        return str(shape_type) if shape_type else None
+    except Exception:
+        return None
+
+
+def collect_solids(workplane: cq.Workplane) -> list[Any]:
+    solids: list[Any] = []
+    seen: set[int] = set()
+
+    for value in workplane.vals():
+        candidates = []
+        try:
+            candidates = list(value.Solids())
+        except Exception:
+            candidates = []
+        if not candidates and get_shape_type(value) == "Solid":
+            candidates = [value]
+        for solid in candidates:
+            key = hash(solid)
+            if key in seen:
+                continue
+            seen.add(key)
+            solids.append(solid)
+
+    return solids
+
+
+def unique_assembly_name(parent: Any, requested_name: Any) -> str:
+    base_name = str(requested_name or "component")
+    if base_name not in parent.objects:
+        return base_name
+
+    index = 2
+    while True:
+        candidate = f"{base_name}__{index:03d}"
+        if candidate not in parent.objects:
+            return candidate
+        index += 1
+
+
+def import_step_with_unique_component_names(source_path: Path) -> cq.Assembly:
+    original_add = cq.Assembly.add
+
+    def add_with_unique_names(self: cq.Assembly, arg: Any, *args: Any, **kwargs: Any) -> Any:
+        if isinstance(arg, cq.Assembly):
+            positional_name = args[0] if args else None
+            requested_name = kwargs.get("name", positional_name) or arg.name
+            unique_name = unique_assembly_name(self, requested_name)
+
+            if unique_name != requested_name:
+                arg.metadata = {
+                    **metadata_dict(arg.metadata),
+                    "originalName": str(requested_name),
+                }
+                if args:
+                    args = (unique_name, *args[1:])
+                else:
+                    kwargs = {**kwargs, "name": unique_name}
+
+        return original_add(self, arg, *args, **kwargs)
+
+    cq.Assembly.add = add_with_unique_names
+    try:
+        return cq.Assembly.importStep(str(source_path))
+    finally:
+        cq.Assembly.add = original_add
+
+
+def import_step_assembly(source_path: Path) -> tuple[cq.Assembly, str]:
+    try:
+        return cq.Assembly.importStep(str(source_path)), "assembly"
+    except ValueError as exc:
+        error_message = str(exc)
+        if "Unique name is required" in error_message and "already in the assembly" in error_message:
+            return import_step_with_unique_component_names(source_path), "assembly-unique-names"
+        if "does not contain an assembly" not in error_message:
+            raise
+
+    step_model = cq.importers.importStep(str(source_path))
+    solids = collect_solids(step_model)
+    assembly = cq.Assembly(name=f"{source_path.stem}_root")
+
+    if len(solids) > 1:
+        for index, solid in enumerate(solids, start=1):
+            assembly.add(solid, name=f"{source_path.stem}_solid_{index:03d}")
+        return assembly, "split-solids"
+
+    assembly.add(step_model, name=source_path.stem)
+    return assembly, "single-shape"
+
+
+def write_json_atomic(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    tmp_path = path.with_name(f"{path.name}.tmp")
+    tmp_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    tmp_path.replace(path)
+
+
+def main() -> int:
+    source_path = Path(sys.argv[1])
+    glb_path = Path(sys.argv[2])
+    result_path = Path(sys.argv[3])
+    linear_deflection = float(sys.argv[4])
+    angular_deflection = float(sys.argv[5])
+    started_at = time.monotonic()
+
+    try:
+        assembly, import_strategy = import_step_assembly(source_path)
+        tree = node_to_metadata(assembly)
+        glb_path.parent.mkdir(parents=True, exist_ok=True)
+        assembly.export(str(glb_path), tolerance=linear_deflection, angularTolerance=angular_deflection)
+        stats = {
+            "fallback": True,
+            "fileSizeMb": round(source_path.stat().st_size / 1024 / 1024, 2),
+            "importStrategy": import_strategy,
+            "cadqueryVersion": getattr(cq, "__version__", "unknown"),
+            "linearDeflection": linear_deflection,
+            "angularDeflection": angular_deflection,
+            "durationSeconds": round(time.monotonic() - started_at, 2),
+        }
+        write_json_atomic(result_path, {"ok": True, "tree": tree, "stats": stats})
+        return 0
+    except Exception as exc:
+        write_json_atomic(result_path, {"ok": False, "error": str(exc)})
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

converter/worker.py

@@ -1,11 +1,14 @@
 import hashlib
 import json
 import os
+import signal
+import shutil
 import struct
 import subprocess
 import sys
 import time
 from datetime import datetime, timezone
+from multiprocessing import Process
 from pathlib import Path
 from typing import Any
 
@@ -22,18 +25,28 @@ from OCP.XCAFDoc import XCAFDoc_DocumentTool, XCAFDoc_ShapeTool
 
 
 CONVERTER_NAME = "NodeDcBimConverter"
-CONVERTER_VERSION = "0.4.0"
+CONVERTER_VERSION = "0.4.6"
 UPLOADS_DIR = Path(os.environ.get("NODEDC_BIM_CONVERTER_UPLOADS_DIR", "/beam/uploads")).resolve()
 POLL_INTERVAL_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_INTERVAL_SECONDS", "10"))
 PROCESS_ONCE = os.environ.get("NODEDC_BIM_CONVERTER_ONCE", "").lower() in {"1", "true", "yes"}
 REPROCESS_READY_ON_VERSION_CHANGE = os.environ.get(
     "NODEDC_BIM_CONVERTER_REPROCESS_READY_ON_VERSION_CHANGE", ""
 ).lower() in {"1", "true", "yes"}
+REPROCESS_FAILED_ON_VERSION_CHANGE = os.environ.get(
+    "NODEDC_BIM_CONVERTER_REPROCESS_FAILED_ON_VERSION_CHANGE", "1"
+).lower() in {"1", "true", "yes"}
 XKT_ENABLED = os.environ.get("NODEDC_BIM_CONVERTER_XKT_ENABLED", "1").lower() not in {"0", "false", "no"}
 XKT_COMMAND = os.environ.get("NODEDC_BIM_CONVERTER_XKT_COMMAND", "xeokit-convert")
 NODE_OPTIONS = os.environ.get("NODE_OPTIONS", "--max-old-space-size=8192")
 STEP_MESH_LINEAR_DEFLECTION = float(os.environ.get("NODEDC_BIM_CONVERTER_STEP_LINEAR_DEFLECTION", "0.1"))
 STEP_MESH_ANGULAR_DEFLECTION = float(os.environ.get("NODEDC_BIM_CONVERTER_STEP_ANGULAR_DEFLECTION", "0.1"))
+PROCESSING_STALE_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_PROCESSING_STALE_SECONDS", "60"))
+CONVERSION_MAX_ATTEMPTS = int(os.environ.get("NODEDC_BIM_CONVERTER_MAX_ATTEMPTS", "3"))
+CONVERSION_TIMEOUT_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_TIMEOUT_SECONDS", "300"))
+XCAF_TIMEOUT_SECONDS = float(os.environ.get("NODEDC_BIM_CONVERTER_XCAF_TIMEOUT_SECONDS", str(CONVERSION_TIMEOUT_SECONDS)))
+CADQUERY_TIMEOUT_SECONDS = float(
+    os.environ.get("NODEDC_BIM_CONVERTER_CADQUERY_TIMEOUT_SECONDS", str(CONVERSION_TIMEOUT_SECONDS))
+)
 STEP_EXTENSIONS = {".step", ".stp"}
 
 
@@ -49,6 +62,26 @@ def manifest_path(source_path: Path) -> Path:
     return source_path.with_name(f"{source_path.name}.beam.json")
 
 
+def asset_manifest_path(source_path: Path) -> Path | None:
+    version_dir = source_path.parent
+    asset_dir = version_dir.parent
+    if not version_dir.name.startswith("ver_"):
+        return None
+    if asset_dir == UPLOADS_DIR or not asset_dir.is_dir():
+        return None
+    return asset_dir / "asset.json"
+
+
+def version_asset_dir(source_path: Path) -> Path | None:
+    version_dir = source_path.parent
+    asset_dir = version_dir.parent
+    if not version_dir.name.startswith("ver_"):
+        return None
+    if asset_dir == UPLOADS_DIR or not asset_dir.is_dir():
+        return None
+    return asset_dir
+
+
 def read_json(path: Path) -> dict[str, Any]:
     try:
         return json.loads(path.read_text(encoding="utf-8"))
@@ -63,6 +96,219 @@ def write_json_atomic(path: Path, payload: dict[str, Any]) -> None:
     tmp_path.replace(path)
 
 
+def conversion_attempt_count(manifest: dict[str, Any]) -> int:
+    try:
+        return max(0, int(manifest.get("attempts") or 0))
+    except Exception:
+        return 0
+
+
+def update_asset_manifest_version(source_path: Path, version_payload: dict[str, Any]) -> None:
+    asset_path = asset_manifest_path(source_path)
+    if not asset_path:
+        return
+    asset = read_json(asset_path)
+    versions = asset.get("versions")
+    if not isinstance(versions, list):
+        versions = []
+
+    version_id = version_payload.get("versionId")
+    version_number = version_payload.get("version")
+    next_versions = []
+    replaced = False
+    for item in versions:
+        if not isinstance(item, dict):
+            continue
+        is_match = (version_id and item.get("versionId") == version_id) or (
+            version_number and item.get("version") == version_number
+        )
+        if is_match:
+            next_versions.append({**item, **version_payload})
+            replaced = True
+        else:
+            next_versions.append(item)
+    if not replaced:
+        next_versions.append(version_payload)
+
+    next_versions.sort(key=lambda item: int(item.get("version") or 0))
+    now = now_iso()
+    write_json_atomic(
+        asset_path,
+        {
+            **asset,
+            "assetId": asset.get("assetId") or version_payload.get("assetId"),
+            "projectId": asset.get("projectId") or version_payload.get("projectId"),
+            "currentVersion": version_payload.get("version") or asset.get("currentVersion"),
+            "currentVersionId": version_payload.get("versionId") or asset.get("currentVersionId"),
+            "updatedAt": now,
+            "versions": next_versions,
+        },
+    )
+
+
+def file_size_mb(path: Path) -> float:
+    try:
+        return path.stat().st_size / (1024 * 1024)
+    except Exception:
+        return 0.0
+
+
+def calculate_file_sha256(path: Path) -> str | None:
+    try:
+        digest = hashlib.sha256()
+        with path.open("rb") as handle:
+            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+                digest.update(chunk)
+        return digest.hexdigest()
+    except Exception:
+        return None
+
+
+def upload_src_exists(src: Any) -> bool:
+    if not isinstance(src, str) or not src.startswith("uploads/"):
+        return False
+    try:
+        return (UPLOADS_DIR / src.removeprefix("uploads/")).resolve().is_file()
+    except Exception:
+        return False
+
+
+def upload_path_from_src(src: Any) -> Path | None:
+    if not isinstance(src, str) or not src.startswith("uploads/"):
+        return None
+    try:
+        path = (UPLOADS_DIR / src.removeprefix("uploads/")).resolve()
+    except Exception:
+        return None
+    try:
+        path.relative_to(UPLOADS_DIR)
+    except ValueError:
+        return None
+    return path if path.is_file() else None
+
+
+def source_path_from_manifest_file(manifest_file: Path, manifest: dict[str, Any]) -> Path | None:
+    source_path = upload_path_from_src(manifest.get("sourceSrc") or manifest.get("downloadSrc"))
+    if source_path:
+        return source_path
+    source_name = manifest_file.name.removesuffix(".beam.json")
+    fallback_path = manifest_file.with_name(source_name)
+    return fallback_path if fallback_path.is_file() else None
+
+
+def link_or_copy_file(source_path: Path, target_path: Path) -> None:
+    if source_path.resolve() == target_path.resolve():
+        return
+    target_path.parent.mkdir(parents=True, exist_ok=True)
+    try:
+        if target_path.exists():
+            target_path.unlink()
+        os.link(source_path, target_path)
+    except OSError:
+        shutil.copy2(source_path, target_path)
+
+
+def copy_reusable_conversion_artifacts(
+    source_path: Path,
+    reusable_conversion: dict[str, Any],
+    glb_path: Path,
+    xkt_path: Path,
+    metadata_path: Path,
+) -> dict[str, Any]:
+    artifact_source_path = upload_path_from_src(reusable_conversion.get("artifactSrc"))
+    metadata_source_path = upload_path_from_src(reusable_conversion.get("metadataSrc"))
+    fallback_source_path = upload_path_from_src(reusable_conversion.get("fallbackArtifactSrc"))
+
+    artifact_type = reusable_conversion.get("artifactType") or reusable_conversion.get("targetFormat") or "xkt"
+    artifact_target_path = xkt_path if artifact_type == "xkt" else glb_path
+
+    if not artifact_source_path or not metadata_source_path:
+        raise RuntimeError("Reusable conversion is missing artifact or metadata file.")
+
+    if fallback_source_path:
+        link_or_copy_file(fallback_source_path, glb_path)
+    if artifact_source_path != fallback_source_path:
+        link_or_copy_file(artifact_source_path, artifact_target_path)
+
+    metadata = read_json(metadata_source_path)
+    metadata.update(
+        {
+            "source": src_from_path(source_path),
+            "artifact": src_from_path(artifact_target_path),
+            "artifactType": artifact_type,
+            "glbArtifact": src_from_path(glb_path) if glb_path.exists() else None,
+            "xktArtifact": src_from_path(xkt_path) if artifact_type == "xkt" and xkt_path.exists() else None,
+            "generatedAt": now_iso(),
+            "reusedFrom": {
+                "assetId": reusable_conversion.get("assetId"),
+                "projectId": reusable_conversion.get("projectId"),
+                "version": reusable_conversion.get("version"),
+                "versionId": reusable_conversion.get("versionId"),
+                "sourceSrc": reusable_conversion.get("sourceSrc"),
+            },
+        }
+    )
+    converter = metadata.get("converter")
+    if isinstance(converter, dict):
+        metadata["converter"] = {
+            **converter,
+            "name": CONVERTER_NAME,
+            "version": CONVERTER_VERSION,
+            "reusedArtifacts": True,
+        }
+    write_json_atomic(metadata_path, metadata)
+
+    return {
+        "artifactSrc": src_from_path(artifact_target_path),
+        "artifactType": artifact_type,
+        "componentCount": reusable_conversion.get("componentCount") or metadata.get("componentCount"),
+        "fallbackArtifactSrc": src_from_path(glb_path) if glb_path.exists() else None,
+        "fallbackArtifactType": "gltf" if glb_path.exists() else None,
+        "metadataSrc": src_from_path(metadata_path),
+    }
+
+
+def find_ready_conversion_with_same_hash(source_path: Path, manifest: dict[str, Any]) -> dict[str, Any] | None:
+    sha256 = manifest.get("sha256") or calculate_file_sha256(source_path)
+    if not sha256:
+        return None
+    version_id = manifest.get("versionId")
+    version_number = manifest.get("version")
+    source_manifest_path = manifest_path(source_path).resolve()
+    current_asset_dir = version_asset_dir(source_path)
+    for candidate_manifest_path in sorted(UPLOADS_DIR.rglob("*.beam.json")):
+        if candidate_manifest_path.resolve() == source_manifest_path:
+            continue
+        candidate = read_json(candidate_manifest_path)
+        if candidate.get("status") != "ready":
+            continue
+        candidate_sha256 = candidate.get("sha256")
+        if not candidate_sha256:
+            candidate_source_path = source_path_from_manifest_file(candidate_manifest_path, candidate)
+            if candidate_source_path:
+                candidate_sha256 = calculate_file_sha256(candidate_source_path)
+            if candidate_sha256:
+                candidate = {**candidate, "sha256": candidate_sha256}
+                write_json_atomic(candidate_manifest_path, candidate)
+                if candidate_source_path:
+                    update_asset_manifest_version(candidate_source_path, candidate)
+        if candidate_sha256 != sha256:
+            continue
+        if candidate.get("sourceFormat") != "step":
+            continue
+        if candidate.get("artifactType") != "xkt" and candidate.get("targetFormat") != "xkt":
+            continue
+        if current_asset_dir and candidate_manifest_path.parent.parent == current_asset_dir:
+            if candidate.get("versionId") == version_id or candidate.get("version") == version_number:
+                continue
+        if not candidate.get("artifactSrc") or not candidate.get("metadataSrc"):
+            continue
+        if not upload_src_exists(candidate.get("artifactSrc")) or not upload_src_exists(candidate.get("metadataSrc")):
+            continue
+        return candidate
+    return None
+
+
 def hash_id(value: str) -> str:
     return hashlib.sha1(value.encode("utf-8")).hexdigest()[:16]
 
@@ -358,7 +604,8 @@ def convert_step_xcaf_to_glb(source_path: Path, glb_path: Path) -> tuple[dict[st
 
     print(
         f"[{CONVERTER_NAME}] meshing {source_path.name}: "
-        f"{labels.Length()} free shape(s), deflection={STEP_MESH_LINEAR_DEFLECTION}",
+        f"{labels.Length()} free shape(s), "
+        f"linear={STEP_MESH_LINEAR_DEFLECTION}, angular={STEP_MESH_ANGULAR_DEFLECTION}",
         flush=True,
     )
     for index in range(1, labels.Length() + 1):
@@ -388,6 +635,7 @@ def convert_step_xcaf_to_glb(source_path: Path, glb_path: Path) -> tuple[dict[st
         "gltfNodeCount": len(gltf_json.get("nodes", [])),
         "gltfMeshCount": len(gltf_json.get("meshes", [])),
         "renamedNodeCount": renamed_nodes,
+        "fileSizeMb": round(file_size_mb(source_path), 2),
         "linearDeflection": STEP_MESH_LINEAR_DEFLECTION,
         "angularDeflection": STEP_MESH_ANGULAR_DEFLECTION,
     }
@@ -398,11 +646,14 @@ def convert_step_cadquery_to_glb(source_path: Path, glb_path: Path) -> tuple[dic
     assy, import_strategy, cadquery_version = import_step_assembly(source_path)
     tree = node_to_metadata(assy)
     glb_path.parent.mkdir(parents=True, exist_ok=True)
-    assy.export(str(glb_path), tolerance=0.1, angularTolerance=0.1)
+    assy.export(str(glb_path), tolerance=STEP_MESH_LINEAR_DEFLECTION, angularTolerance=STEP_MESH_ANGULAR_DEFLECTION)
     stats = {
         "fallback": True,
+        "fileSizeMb": round(file_size_mb(source_path), 2),
         "importStrategy": import_strategy,
         "cadqueryVersion": cadquery_version,
+        "linearDeflection": STEP_MESH_LINEAR_DEFLECTION,
+        "angularDeflection": STEP_MESH_ANGULAR_DEFLECTION,
     }
     return tree, stats
 
@@ -415,6 +666,33 @@ def read_log_tail(path: Path, max_lines: int = 30) -> str:
     return "\n".join(lines[-max_lines:])
 
 
+def stage_exit_error(stage_name: str, exitcode: int) -> str:
+    if exitcode < 0:
+        signal_number = abs(exitcode)
+        if signal_number == 9:
+            return f"{stage_name} process was killed by signal 9; likely converter memory limit was exceeded."
+        return f"{stage_name} process was killed by signal {signal_number}."
+    if exitcode == 137:
+        return f"{stage_name} process was killed with code 137; likely converter memory limit was exceeded."
+    return f"{stage_name} subprocess exited with code {exitcode}."
+
+
+def is_resource_limit_error(error: Any) -> bool:
+    message = str(error)
+    return (
+        "signal 9" in message
+        or "code -9" in message
+        or "code 137" in message
+        or "memory limit was exceeded" in message
+    )
+
+
+def conversion_failed_message(error: Any) -> str:
+    if is_resource_limit_error(error):
+        return "Не удалось подготовить просмотр и дерево компонентов: не хватило памяти конвертера."
+    return "Не удалось подготовить просмотр и дерево компонентов."
+
+
 def convert_glb_to_xkt(glb_path: Path, xkt_path: Path) -> dict[str, Any]:
     if not XKT_ENABLED:
         return {"enabled": False}
@@ -516,6 +794,136 @@ def build_step_metadata(
     return metadata
 
 
+def convert_step_xcaf_to_glb_process(source_path: str, glb_path: str, result_path: str) -> None:
+    try:
+        tree, stats = convert_step_xcaf_to_glb(Path(source_path), Path(glb_path))
+        write_json_atomic(Path(result_path), {"ok": True, "tree": tree, "stats": stats})
+    except Exception as exc:
+        write_json_atomic(Path(result_path), {"ok": False, "error": str(exc)})
+
+
+def convert_step_cadquery_to_glb_process(source_path: str, glb_path: str, result_path: str) -> None:
+    try:
+        stage_path = Path(__file__).with_name("cadquery_stage.py")
+        completed = subprocess.run(
+            [
+                sys.executable,
+                str(stage_path),
+                source_path,
+                glb_path,
+                result_path,
+                str(STEP_MESH_LINEAR_DEFLECTION),
+                str(STEP_MESH_ANGULAR_DEFLECTION),
+            ],
+            check=False,
+        )
+        if completed.returncode != 0:
+            result = read_json(Path(result_path))
+            raise RuntimeError(result.get("error") or f"CadQuery subprocess exited with code {completed.returncode}.")
+    except Exception as exc:
+        write_json_atomic(Path(result_path), {"ok": False, "error": str(exc)})
+
+
+def run_cadquery_stage_with_timeout(result_path: Path, source_path: Path, glb_path: Path) -> dict[str, Any]:
+    if result_path.exists():
+        result_path.unlink()
+
+    stage_path = Path(__file__).with_name("cadquery_stage.py")
+    command = [
+        sys.executable,
+        str(stage_path),
+        str(source_path),
+        str(glb_path),
+        str(result_path),
+        str(STEP_MESH_LINEAR_DEFLECTION),
+        str(STEP_MESH_ANGULAR_DEFLECTION),
+    ]
+    process = subprocess.Popen(command, start_new_session=True)
+    try:
+        if CADQUERY_TIMEOUT_SECONDS <= 0:
+            process.wait()
+        else:
+            process.wait(timeout=CADQUERY_TIMEOUT_SECONDS)
+    except subprocess.TimeoutExpired:
+        try:
+            os.killpg(process.pid, signal.SIGTERM)
+        except Exception:
+            process.terminate()
+        try:
+            process.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            try:
+                os.killpg(process.pid, signal.SIGKILL)
+            except Exception:
+                process.kill()
+            process.wait(timeout=10)
+        raise TimeoutError(f"CadQuery timed out after {int(CADQUERY_TIMEOUT_SECONDS)} seconds.")
+
+    result = read_json(result_path)
+    try:
+        result_path.unlink()
+    except Exception:
+        pass
+    if process.returncode != 0:
+        raise RuntimeError(result.get("error") or stage_exit_error("CadQuery", process.returncode))
+    if not result:
+        raise RuntimeError("CadQuery produced no result.")
+    if not result.get("ok"):
+        raise RuntimeError(str(result.get("error") or "CadQuery failed."))
+    return result
+
+
+def run_step_glb_stage_with_timeout(
+    stage_name: str,
+    timeout_seconds: float,
+    target: Any,
+    result_path: Path,
+    *args: str,
+) -> dict[str, Any]:
+    if result_path.exists():
+        result_path.unlink()
+
+    if timeout_seconds <= 0:
+        target(*args, str(result_path))
+    else:
+        process = Process(target=target, args=(*args, str(result_path)))
+        process.start()
+        process.join(timeout_seconds)
+
+        if process.is_alive():
+            process.terminate()
+            process.join(10)
+            if process.is_alive():
+                process.kill()
+                process.join(10)
+            raise TimeoutError(f"{stage_name} timed out after {int(timeout_seconds)} seconds.")
+
+        if process.exitcode != 0:
+            result = read_json(result_path)
+            error = result.get("error") if isinstance(result, dict) else None
+            raise RuntimeError(error or stage_exit_error(stage_name, process.exitcode))
+
+    result = read_json(result_path)
+    try:
+        result_path.unlink()
+    except Exception:
+        pass
+    if not result:
+        raise RuntimeError(f"{stage_name} produced no result.")
+    if not result.get("ok"):
+        raise RuntimeError(str(result.get("error") or f"{stage_name} failed."))
+    return result
+
+
+def remove_partial_artifacts(*paths: Path) -> None:
+    for path in paths:
+        try:
+            if path.exists():
+                path.unlink()
+        except Exception:
+            pass
+
+
 def convert_step_assets(source_path: Path, glb_path: Path, xkt_path: Path, metadata_path: Path) -> dict[str, Any]:
     previous_metadata = read_json(metadata_path)
 
@@ -547,12 +955,35 @@ def convert_step_assets(source_path: Path, glb_path: Path, xkt_path: Path, metad
         )
 
     import_strategy = "occt-xcaf-gltf"
+    result_path = metadata_path.with_name(f"{metadata_path.name}.stage-result.json")
     try:
-        tree, stats = convert_step_xcaf_to_glb(source_path, glb_path)
+        result = run_step_glb_stage_with_timeout(
+            "OCCT/XCAF",
+            XCAF_TIMEOUT_SECONDS,
+            convert_step_xcaf_to_glb_process,
+            result_path,
+            str(source_path),
+            str(glb_path),
+        )
+        tree = result["tree"]
+        stats = result["stats"]
     except Exception as exc:
-        print(f"[{CONVERTER_NAME}] OCCT/XCAF path failed for {source_path}: {exc}", file=sys.stderr, flush=True)
+        print(
-        tree, stats = convert_step_cadquery_to_glb(source_path, glb_path)
+            f"[{CONVERTER_NAME}] OCCT/XCAF path failed for {source_path}: {exc}; "
+            "falling back to CadQuery",
+            file=sys.stderr,
+            flush=True,
+        )
+        remove_partial_artifacts(glb_path, xkt_path)
+        result = run_cadquery_stage_with_timeout(result_path, source_path, glb_path)
+        tree = result["tree"]
+        stats = result["stats"]
         import_strategy = stats.get("importStrategy") or "cadquery"
+        stats = {
+            **stats,
+            "fallbackFrom": "occt-xcaf-gltf",
+            "fallbackReason": str(exc),
+        }
 
     return build_step_metadata(
         source_path,
@@ -568,14 +999,20 @@ def convert_step_assets(source_path: Path, glb_path: Path, xkt_path: Path, metad
 def should_process(source_path: Path, manifest: dict[str, Any], glb_path: Path, xkt_path: Path) -> bool:
     status = manifest.get("status")
     ready_artifact_path = xkt_path if XKT_ENABLED else glb_path
-    if status == "ready" and ready_artifact_path.exists():
+    if status == "ready" and (ready_artifact_path.exists() or upload_src_exists(manifest.get("artifactSrc"))):
         return REPROCESS_READY_ON_VERSION_CHANGE and manifest.get("converterVersion") != CONVERTER_VERSION
+    if status == "failed":
+        return (
+            REPROCESS_FAILED_ON_VERSION_CHANGE
+            and manifest.get("converterVersion") != CONVERTER_VERSION
+            and source_path.exists()
+        )
     if status == "processing":
         updated_at = manifest.get("updatedAt")
         if updated_at:
             try:
                 updated = datetime.fromisoformat(str(updated_at).replace("Z", "+00:00"))
-                if (datetime.now(timezone.utc) - updated).total_seconds() < 300:
+                if (datetime.now(timezone.utc) - updated).total_seconds() < PROCESSING_STALE_SECONDS:
                     return False
             except Exception:
                 pass
@@ -593,7 +1030,25 @@ def process_one(source_path: Path) -> bool:
     if not should_process(source_path, manifest, glb_path, xkt_path):
         return False
 
+    preserved_fields = {
+        key: manifest.get(key)
+        for key in (
+            "assetId",
+            "projectId",
+            "version",
+            "versionId",
+            "originalFilename",
+            "storedFilename",
+            "size",
+            "sha256",
+            "uploadedAt",
+            "downloadSrc",
+        )
+        if manifest.get(key) is not None
+    }
+
     base_manifest = {
+        **preserved_fields,
         "createdAt": manifest.get("createdAt") or now_iso(),
         "componentTreeRequired": True,
         "converterName": CONVERTER_NAME,
@@ -603,11 +1058,69 @@ def process_one(source_path: Path) -> bool:
         "targetFormat": "xkt" if XKT_ENABLED else "glb",
     }
 
+    attempts = 0 if manifest.get("converterVersion") != CONVERTER_VERSION else conversion_attempt_count(manifest)
+    if manifest.get("status") == "processing" and attempts >= CONVERSION_MAX_ATTEMPTS:
+        failed_manifest = {
+            **base_manifest,
+            "attempts": attempts,
+            "error": f"Converter process stopped before completion after {attempts} attempt(s).",
+            "message": "Не удалось подготовить просмотр и дерево компонентов: превышен лимит попыток конвертации.",
+            "status": "failed",
+            "updatedAt": now_iso(),
+        }
+        write_json_atomic(manifest_file, failed_manifest)
+        update_asset_manifest_version(source_path, failed_manifest)
+        print(
+            f"[{CONVERTER_NAME}] failed {source_path}: max attempts reached ({attempts})",
+            file=sys.stderr,
+            flush=True,
+        )
+        return False
+
+    reusable_conversion = find_ready_conversion_with_same_hash(source_path, manifest)
+    if reusable_conversion:
+        reused_artifacts = copy_reusable_conversion_artifacts(
+            source_path,
+            reusable_conversion,
+            glb_path,
+            xkt_path,
+            metadata_path,
+        )
+        ready_manifest = {
+            **base_manifest,
+            "attempts": attempts,
+            **reused_artifacts,
+            "message": "Просмотр и дерево компонентов готовы.",
+            "reusedFromAssetId": reusable_conversion.get("assetId"),
+            "reusedFromProjectId": reusable_conversion.get("projectId"),
+            "reusedFromVersion": reusable_conversion.get("version"),
+            "reusedFromVersionId": reusable_conversion.get("versionId"),
+            "reusedFromSourceSrc": reusable_conversion.get("sourceSrc"),
+            "status": "ready",
+            "updatedAt": now_iso(),
+        }
+        write_json_atomic(manifest_file, ready_manifest)
+        update_asset_manifest_version(source_path, ready_manifest)
+        reuse_label = (
+            reusable_conversion.get("versionId")
+            or reusable_conversion.get("version")
+            or reusable_conversion.get("sourceSrc")
+            or "ready-artifact"
+        )
+        print(
+            f"[{CONVERTER_NAME}] reused ready conversion {source_path.name} "
+            f"from {reuse_label}",
+            flush=True,
+        )
+        return True
+
+    attempt = attempts + 1
     write_json_atomic(
         manifest_file,
         {
             **base_manifest,
-            "message": "Preparing XKT preview and component tree.",
+            "attempts": attempt,
+            "message": "Подготавливаем просмотр и дерево компонентов.",
             "status": "processing",
             "updatedAt": now_iso(),
         },
@@ -618,34 +1131,34 @@ def process_one(source_path: Path) -> bool:
         metadata = convert_step_assets(source_path, glb_path, xkt_path, metadata_path)
         artifact_src = metadata.get("artifact") or src_from_path(xkt_path if xkt_path.exists() else glb_path)
         artifact_type = metadata.get("artifactType") or ("xkt" if xkt_path.exists() else "gltf")
-        write_json_atomic(
+        ready_manifest = {
-            manifest_file,
+            **base_manifest,
-            {
+            "attempts": attempt,
-                **base_manifest,
+            "artifactSrc": artifact_src,
-                "artifactSrc": artifact_src,
+            "artifactType": artifact_type,
-                "artifactType": artifact_type,
+            "fallbackArtifactSrc": src_from_path(glb_path),
-                "fallbackArtifactSrc": src_from_path(glb_path),
+            "fallbackArtifactType": "gltf",
-                "fallbackArtifactType": "gltf",
+            "componentCount": metadata.get("componentCount"),
-                "componentCount": metadata.get("componentCount"),
+            "metadataSrc": src_from_path(metadata_path),
-                "metadataSrc": src_from_path(metadata_path),
+            "message": "Просмотр и дерево компонентов готовы.",
-                "message": "XKT preview and component tree are ready.",
+            "status": "ready",
-                "status": "ready",
+            "updatedAt": now_iso(),
-                "updatedAt": now_iso(),
+        }
-            },
+        write_json_atomic(manifest_file, ready_manifest)
-        )
+        update_asset_manifest_version(source_path, ready_manifest)
         print(f"[{CONVERTER_NAME}] ready {artifact_src}", flush=True)
         return True
     except Exception as exc:
-        write_json_atomic(
+        failed_manifest = {
-            manifest_file,
+            **base_manifest,
-            {
+            "attempts": attempt,
-                **base_manifest,
+            "error": str(exc),
-                "error": str(exc),
+            "message": conversion_failed_message(exc),
-                "message": "Failed to prepare XKT preview and component tree.",
+            "status": "failed",
-                "status": "failed",
+            "updatedAt": now_iso(),
-                "updatedAt": now_iso(),
+        }
-            },
+        write_json_atomic(manifest_file, failed_manifest)
-        )
+        update_asset_manifest_version(source_path, failed_manifest)
         print(f"[{CONVERTER_NAME}] failed {source_path}: {exc}", file=sys.stderr, flush=True)
         return False