Add LandSAR cluster worker deployment

backend/app/config.py

@@ -235,9 +235,13 @@ class Settings(BaseSettings):
     RADAR_THUMBNAIL_MAX_SIZE: int = 1600
     RADAR_CACHE_WORKERS: int = 2
     RADAR_GEO_CACHE_WORKERS: int = 2
-    RADAR_GEO_CACHE_VERSION: str = "b1"
+    RADAR_GEO_CACHE_VERSION: str = "b2"
     RADAR_GEO_CACHE_QUALITY: int = 84
     RADAR_PREVIEW_BUILD_ON_DEMAND: bool = True
+    ASSET_SCAN_PARSE_WORKERS: int = 4
+    ASSET_SCAN_PARSE_INFLIGHT: int = 64
+    ASSET_SCAN_SKIP_UNCHANGED_FAILURES: bool = True
+    ASSET_SCAN_DB_BATCH_SIZE: int = 50
 
     WATER_RESULTS_DIR: str = ""
     GF3_WATER_DEM_PATH: str = ""
@@ -417,6 +421,7 @@ class Settings(BaseSettings):
     JOB_WORKER_STALE_RECOVER_INTERVAL: float = 15.0
     JOB_WORKER_STALE_RUNNING_SECONDS: int = 300
     JOB_WORKER_HEARTBEAT_INTERVAL: float = 5.0
+    JOB_WORKER_ALLOWED_TYPES: str = ""
 
     TIMESERIES_ENABLED: bool = False
     TIMESERIES_WSL_DISTRO: str = ""
@@ -495,6 +500,13 @@ class Settings(BaseSettings):
         )
         if not self.SRTM_DEM_DIR:
             object.__setattr__(self, "SRTM_DEM_DIR", os.path.join(backend_dir, "dem_data"))
+        object.__setattr__(self, "ASSET_SCAN_PARSE_WORKERS", max(1, int(self.ASSET_SCAN_PARSE_WORKERS or 1)))
+        object.__setattr__(
+            self,
+            "ASSET_SCAN_PARSE_INFLIGHT",
+            max(self.ASSET_SCAN_PARSE_WORKERS, int(self.ASSET_SCAN_PARSE_INFLIGHT or self.ASSET_SCAN_PARSE_WORKERS)),
+        )
+        object.__setattr__(self, "ASSET_SCAN_DB_BATCH_SIZE", max(1, int(self.ASSET_SCAN_DB_BATCH_SIZE or 1)))
         if not self.GF3_ARCHIVE_SOURCE_DIRS:
             object.__setattr__(
                 self,

backend/app/license_service.py

@@ -17,6 +17,7 @@ import hashlib
 import json
 import os
 import subprocess
+import time
 import uuid
 from datetime import datetime, timezone
 from typing import Any, Dict, Optional
@@ -32,6 +33,13 @@ _PUBLIC_KEY_B64 = "QOpR1c3bONDwOzrj3IVTogE1ZHIphpwxJY8nhWa09yw="
 _APP_DIR = os.path.dirname(os.path.abspath(__file__))
 _BACKEND_DIR = os.path.dirname(_APP_DIR)
 LICENSE_PATH_DEFAULT = os.path.join(_BACKEND_DIR, "license", "license.lic")
+LICENSE_STATUS_CACHE_SECONDS = int(os.getenv("LICENSE_STATUS_CACHE_SECONDS", "30"))
+_LICENSE_STATUS_CACHE: Dict[str, Any] = {
+    "path": None,
+    "mtime": None,
+    "checked_at": 0.0,
+    "payload": None,
+}
 
 
 def _now_utc() -> datetime:
@@ -86,47 +94,71 @@ def _get_machine_fingerprint() -> str:
 
 # ── 授权验证 ──────────────────────────────────────────────────────────────────
 
+def _license_result(result: Dict[str, Any], *, use_cache: bool, path: str, mtime: Optional[float]) -> Dict[str, Any]:
+    if use_cache:
+        _LICENSE_STATUS_CACHE.update({
+            "path": path,
+            "mtime": mtime,
+            "checked_at": time.monotonic(),
+            "payload": dict(result),
+        })
+    return result
+
+
 def check_license(license_path: Optional[str] = None) -> Dict[str, Any]:
+    use_cache = license_path is None
     license_path = license_path or settings.LICENSE_PATH or LICENSE_PATH_DEFAULT
+    mtime = os.path.getmtime(license_path) if os.path.exists(license_path) else None
+
+    if use_cache:
+        cached = _LICENSE_STATUS_CACHE.get("payload")
+        cache_age = time.monotonic() - float(_LICENSE_STATUS_CACHE.get("checked_at") or 0.0)
+        if (
+            cached is not None
+            and _LICENSE_STATUS_CACHE.get("path") == license_path
+            and _LICENSE_STATUS_CACHE.get("mtime") == mtime
+            and cache_age <= LICENSE_STATUS_CACHE_SECONDS
+        ):
+            return dict(cached)
 
     if not os.path.exists(license_path):
-        return {"ok": False, "reason": "未找到授权文件"}
+        return _license_result({"ok": False, "reason": "未找到授权文件"}, use_cache=use_cache, path=license_path, mtime=mtime)
 
     try:
         blob = open(license_path, "rb").read().strip()
         parts = blob.split(b"|", 2)
         if len(parts) != 3 or parts[0] != b"LIC2":
-            return {"ok": False, "reason": "授权文件格式无效"}
+            return _license_result({"ok": False, "reason": "授权文件格式无效"}, use_cache=use_cache, path=license_path, mtime=mtime)
 
         _, sig_b64, payload_b64 = parts
         pub = Ed25519PublicKey.from_public_bytes(base64.b64decode(_PUBLIC_KEY_B64))
         pub.verify(base64.b64decode(sig_b64), payload_b64)
         payload = json.loads(base64.b64decode(payload_b64))
     except InvalidSignature:
-        return {"ok": False, "reason": "授权文件签名无效（可能已被篡改）"}
+        return _license_result({"ok": False, "reason": "授权文件签名无效（可能已被篡改）"}, use_cache=use_cache, path=license_path, mtime=mtime)
     except Exception as e:
-        return {"ok": False, "reason": f"授权文件解析失败: {e}"}
+        return _license_result({"ok": False, "reason": f"授权文件解析失败: {e}"}, use_cache=use_cache, path=license_path, mtime=mtime)
 
     fp_expected = payload.get("fingerprint")
     fp_actual   = _get_machine_fingerprint()
     if not fp_expected or fp_expected != fp_actual:
-        return {"ok": False, "reason": "机器指纹不匹配"}
+        return _license_result({"ok": False, "reason": "机器指纹不匹配"}, use_cache=use_cache, path=license_path, mtime=mtime)
 
     expires_at = payload.get("expires_at")
     if not expires_at:
-        return {"ok": False, "reason": "授权文件缺少有效期"}
+        return _license_result({"ok": False, "reason": "授权文件缺少有效期"}, use_cache=use_cache, path=license_path, mtime=mtime)
     try:
         expires_dt = datetime.fromisoformat(expires_at)
     except Exception:
-        return {"ok": False, "reason": "有效期格式错误"}
+        return _license_result({"ok": False, "reason": "有效期格式错误"}, use_cache=use_cache, path=license_path, mtime=mtime)
 
     if _now_utc() > expires_dt:
-        return {"ok": False, "reason": "授权已过期"}
+        return _license_result({"ok": False, "reason": "授权已过期"}, use_cache=use_cache, path=license_path, mtime=mtime)
 
-    return {
+    return _license_result({
         "ok":          True,
         "issued_to":   payload.get("issued_to"),
         "expires_at":  expires_at,
         "fingerprint": fp_actual,
         "license_path": license_path,
-    }
+    }, use_cache=use_cache, path=license_path, mtime=mtime)

backend/app/routers/dependencies.py

@@ -126,6 +126,17 @@ _STATS_CACHE_DATA: Optional[Dict[str, Any]] = None
 _STATS_CACHE_EXPIRES_AT = 0.0
 _STATS_CACHE_GENERATED_AT_UTC: Optional[str] = None
 
+DASHBOARD_STATS_CACHE_TTL_SECONDS = read_int_env(
+    "DASHBOARD_STATS_CACHE_TTL_SECONDS",
+    STATS_CACHE_TTL_SECONDS,
+    minimum=0,
+    maximum=3600,
+)
+_DASHBOARD_STATS_CACHE_LOCK = asyncio.Lock()
+_DASHBOARD_STATS_CACHE_DATA: Optional[Dict[str, Any]] = None
+_DASHBOARD_STATS_CACHE_EXPIRES_AT = 0.0
+_DASHBOARD_STATS_CACHE_GENERATED_AT_UTC: Optional[str] = None
+
 # ---------------------------------------------------------------------------
 # AOI token store
 # ---------------------------------------------------------------------------

backend/app/routers/dinsar_production.py

@@ -42,6 +42,12 @@ LANDSAR_PRODUCTION_JOB_MAX_ATTEMPTS = read_int_env(
     minimum=1,
     maximum=10,
 )
+LANDSAR_CLUSTER_ITEM_JOB_MAX_ATTEMPTS = read_int_env(
+    "LANDSAR_CLUSTER_ITEM_JOB_MAX_ATTEMPTS",
+    1,
+    minimum=1,
+    maximum=10,
+)
 
 
 class RunJobRequest(BaseModel):
@@ -441,6 +447,114 @@ async def submit_run(
     }
 
 
+@router.post("/landsar-cluster/run")
+async def submit_landsar_cluster_run(
+    req: RunJobRequest,
+    current_user: AuthUserORM = Depends(_get_current_user),
+):
+    if str(req.engine_code or "").strip().lower() != "landsar":
+        raise HTTPException(status_code=400, detail="LandSAR cluster only accepts engine_code='landsar'.")
+
+    registry = _get_registry()
+    engine = registry.get_engine("landsar")
+    if not engine:
+        raise HTTPException(status_code=400, detail="Engine 'landsar' not found.")
+
+    valid_profiles = {profile.code for profile in engine.get_profiles()}
+    if req.profile not in valid_profiles:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Engine 'landsar' does not support profile '{req.profile}'. "
+                f"Available profiles: {sorted(valid_profiles)}"
+            ),
+        )
+
+    validation_summary = None
+    if hasattr(engine, "validate_root_dir"):
+        try:
+            validation_summary = await asyncio.to_thread(
+                engine.validate_root_dir,
+                req.root_dir,
+                req.num_to_process,
+                req.rerun_mode,
+            )
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+        if int(validation_summary.get("task_count", 0) or 0) <= 0:
+            if (
+                req.rerun_mode == "unfinished_only"
+                and int(validation_summary.get("skipped_completed_count", 0) or 0) > 0
+            ):
+                raise HTTPException(
+                    status_code=400,
+                    detail=(
+                        f"All discovered Task_* directories already have completed "
+                        f"landsar/{req.profile} results under: {req.root_dir}"
+                    ),
+                )
+            raise HTTPException(status_code=400, detail="No valid LandSAR task directories selected.")
+
+    normalized_extra = dict(req.extra or {})
+    if hasattr(engine, "normalize_extra"):
+        try:
+            normalized_extra = engine.normalize_extra(normalized_extra)
+        except ValueError as exc:
+            raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+    if validation_summary is not None:
+        validated_task_count = validation_summary.get("task_count", 0)
+        normalized_extra.update(
+            {
+                "__validated_task_count": validated_task_count,
+                "__validated_mode": validation_summary.get("mode", ""),
+                "__rerun_mode": req.rerun_mode,
+                "__discovered_task_count": int(validation_summary.get("discovered_task_count", validated_task_count) or 0),
+                "__skipped_completed_count": int(validation_summary.get("skipped_completed_count", 0) or 0),
+                "__cluster": True,
+            }
+        )
+
+    effective_timeout_seconds = req.timeout_seconds
+    if effective_timeout_seconds is None:
+        engine_default_timeout = getattr(engine, "default_timeout_seconds", None)
+        if engine_default_timeout:
+            effective_timeout_seconds = int(engine_default_timeout)
+
+    try:
+        async with _new_session() as db:
+            result = await dinsar_production_service.create_landsar_cluster_run(
+                profile_code=req.profile,
+                root_dir=req.root_dir,
+                num_to_process=req.num_to_process,
+                rerun_mode=req.rerun_mode,
+                timeout_seconds=effective_timeout_seconds,
+                extra=normalized_extra,
+                created_by=getattr(current_user, "username", None),
+                max_attempts=LANDSAR_CLUSTER_ITEM_JOB_MAX_ATTEMPTS,
+                db=db,
+            )
+    except ValueError as exc:
+        message = str(exc)
+        status_code = 409 if "浠诲姟鍐茬獊" in message else 400
+        raise HTTPException(status_code=status_code, detail=message) from exc
+
+    return {
+        "task_id": result["task_id"],
+        "job_id": None,
+        "run_id": result["run_id"],
+        "workflow_run_id": None,
+        "job_type": "LANDSAR_CLUSTER_ITEM",
+        "engine_code": "landsar",
+        "profile": req.profile,
+        "selected_task_count": result.get("selected_task_count", 0),
+        "discovered_task_count": result.get("discovered_task_count", result.get("selected_task_count", 0)),
+        "skipped_completed_count": result.get("skipped_completed_count", 0),
+        "rerun_mode": result.get("rerun_mode", req.rerun_mode),
+        "message": "LandSAR cluster items queued.",
+    }
+
+
 @router.get("/runs")
 async def list_runs(limit: int = 20, offset: int = 0):
     async with _new_session() as db:

backend/app/routers/radar.py

@@ -318,10 +318,7 @@ async def _build_radar_preview_cache(
                     geo_error = "invalid_bbox"
                     has_geo_cache = False
                 else:
-                    source_corner_mapping = await asyncio.to_thread(
-                        data_service.get_radar_source_corner_mapping,
-                        record.file_path,
-                    )
+                    source_corner_mapping = data_service.get_radar_record_corner_mapping(record)
                     ok_geo, geo_error = await asyncio.to_thread(
                         image_service.create_geocorrected_radar_cached_image,
                         preview_source,
@@ -408,6 +405,7 @@ async def _get_cached_radar_preview(data_id: int, db: AsyncSession):
     raw_cache_path, geo_cache_path = _radar_preview_paths(record)
     if (
         (record.preview_cache_status or "NONE") == "READY"
+        and (record.preview_cache_version or "") == settings.RADAR_GEO_CACHE_VERSION
         and record.preview_cache_path
         and str(record.preview_cache_path).lower().endswith(".webp")
         and os.path.exists(record.preview_cache_path)
@@ -418,7 +416,7 @@ async def _get_cached_radar_preview(data_id: int, db: AsyncSession):
             headers={"Cache-Control": "public, max-age=31536000"},
         )
 
-    if os.path.exists(geo_cache_path):
+    if os.path.exists(geo_cache_path) and (record.preview_cache_version or "") == settings.RADAR_GEO_CACHE_VERSION:
         return FileResponse(
             geo_cache_path,
             media_type="image/webp",
@@ -523,7 +521,6 @@ async def search_radar_data_endpoint(
     product_unique_id: Optional[str] = Form(None),
     orbit_direction: Optional[str] = Form(None),
     has_orbit_data: Optional[bool] = Form(None),
-    is_envi_processed: Optional[bool] = Form(None),
     imaging_date_from: Optional[str] = Form(None),
     imaging_date_to: Optional[str] = Form(None),
     region_tree_id: Optional[str] = Form(None),
@@ -609,8 +606,6 @@ async def search_radar_data_endpoint(
         filters.append(RadarDataORM.orbit_direction.ilike(f"%{n_orbit_direction}%"))
     if has_orbit_data is not None:
         filters.append(RadarDataORM.has_orbit_data == has_orbit_data)
-    if is_envi_processed is not None:
-        filters.append(RadarDataORM.is_envi_processed == is_envi_processed)
     if n_date_from:
         filters.append(RadarDataORM.imaging_date >= n_date_from)
     if n_date_to:

backend/app/services/admin_region_lookup_service.py

@@ -5,7 +5,7 @@ import json
 from pathlib import Path
 from typing import Any
 
-from shapely.geometry import Point, shape
+from shapely.geometry import MultiPolygon, Point, shape
 from shapely.ops import unary_union
 
 try:
@@ -36,6 +36,7 @@ class _RegionGeometryRecord:
 _REGION_GEOMETRY_CACHE: list[_RegionGeometryRecord] | None = None
 _REGION_BY_ID_LOOKUP_CACHE: dict[str, dict[str, Any]] | None = None
 _REGION_LOAD_ERROR: str | None = None
+_POLYGONAL_GEOMETRY_TYPES = {"Polygon", "MultiPolygon"}
 
 
 def _backend_geojson_dir() -> Path:
@@ -190,6 +191,40 @@ def _build_region_path(tree_id: str, region_by_id: dict[str, dict[str, Any]]) ->
     return names, tree_ids
 
 
+def _as_polygonal_geometry(geometry):
+    if geometry is None or geometry.is_empty:
+        return None
+    if getattr(geometry, "geom_type", None) in _POLYGONAL_GEOMETRY_TYPES:
+        return geometry
+
+    polygon_parts = []
+    for part in getattr(geometry, "geoms", []) or []:
+        polygonal = _as_polygonal_geometry(part)
+        if polygonal is None or polygonal.is_empty:
+            continue
+        if getattr(polygonal, "geom_type", None) == "Polygon":
+            polygon_parts.append(polygonal)
+        else:
+            polygon_parts.extend([item for item in getattr(polygonal, "geoms", []) if not item.is_empty])
+
+    if not polygon_parts:
+        return None
+    if len(polygon_parts) == 1:
+        return polygon_parts[0]
+    try:
+        merged = unary_union(polygon_parts)
+        if merged is not None and not merged.is_empty:
+            if getattr(merged, "geom_type", None) in _POLYGONAL_GEOMETRY_TYPES:
+                return merged
+            return _as_polygonal_geometry(merged)
+    except Exception:
+        pass
+    try:
+        return MultiPolygon(polygon_parts)
+    except Exception:
+        return polygon_parts[0]
+
+
 def _load_region_records() -> tuple[list[_RegionGeometryRecord], dict[str, dict[str, Any]], str | None]:
     global _REGION_BY_ID_LOOKUP_CACHE, _REGION_GEOMETRY_CACHE, _REGION_LOAD_ERROR
     if _REGION_GEOMETRY_CACHE is not None:
@@ -228,6 +263,7 @@ def _load_region_records() -> tuple[list[_RegionGeometryRecord], dict[str, dict[
             if not geometries:
                 continue
             geometry = _merge_region_geometries(geometries)
+            geometry = _as_polygonal_geometry(geometry)
             if geometry is None or geometry.is_empty:
                 continue
             node = region_by_id.get(tree_id) or {}

backend/app/services/asset_inventory_service.py

@@ -9,8 +9,10 @@ import re
 import shutil
 import tarfile
 import zipfile
+from concurrent.futures import FIRST_COMPLETED, ThreadPoolExecutor, wait
 from datetime import datetime, timedelta
 from pathlib import PurePosixPath
+from types import SimpleNamespace
 from typing import Any, Awaitable, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
 
 from geoalchemy2.shape import from_shape
@@ -35,6 +37,7 @@ from ..models import (
     SourceProductAssetORM,
 )
 from ..utils import (
+    build_corner_pixel_mapping,
     find_xml_file,
     normalize_satellite_family,
     parse_gf3_l2_dirname,
@@ -55,6 +58,9 @@ LT1_ORBIT_MATCH_RULE_VERSION = "lt1_orbit_day_v1"
 ASSET_SCAN_LOG_INTERVAL = 100
 ASSET_SCAN_DETAILED_PARSE_LOG_LIMIT = 200
 ARCHIVE_INTEGRITY_LOG_INTERVAL = 10
+DEFAULT_ASSET_SCAN_PARSE_WORKERS = 4
+DEFAULT_ASSET_SCAN_PARSE_INFLIGHT = 64
+DEFAULT_ASSET_SCAN_DB_BATCH_SIZE = 50
 
 _WINDOWS_DRIVE_RE = re.compile(r"^[a-zA-Z]:[\\/]")
 _S1_SOURCE_RE = re.compile(
@@ -458,32 +464,60 @@ def _archive_read_first_matching(path: str, predicate: Callable[[str], bool]) ->
     return None, None, members
 
 
-def _archive_list_matching(path: str, predicate: Callable[[str], bool], *, limit: int = 20) -> List[str]:
-    matches: List[str] = []
+def _archive_collect_members(
+    path: str,
+    *,
+    content_predicate: Callable[[str], bool],
+    list_predicate: Callable[[str], bool],
+    list_limit: int = 20,
+) -> Tuple[Optional[str], Optional[bytes], List[str], List[str]]:
+    content_name: Optional[str] = None
+    content_data: Optional[bytes] = None
+    listed: List[str] = []
+    scanned: List[str] = []
+    target_list_count = max(0, int(list_limit))
 
-    def _visit(name: str) -> None:
-        if predicate(name):
-            matches.append(name)
+    def _visit(name: str, reader: Callable[[], bytes]) -> None:
+        nonlocal content_name, content_data
+        scanned.append(name)
+        if list_predicate(name) and len(listed) < target_list_count:
+            listed.append(name)
+        if content_name is None and content_predicate(name):
+            content_name = name
+            content_data = reader()
+
+    def _done() -> bool:
+        return content_name is not None and len(listed) >= target_list_count
 
     if zipfile.is_zipfile(path):
         with zipfile.ZipFile(path) as archive:
             for info in archive.infolist():
                 if info.is_dir():
                     continue
-                _visit(info.filename)
-                if len(matches) >= limit:
+                _visit(info.filename, lambda info=info: archive.read(info))
+                if _done():
                     break
-        return matches
+        return content_name, content_data, listed, scanned
 
     if tarfile.is_tarfile(path):
         with tarfile.open(path, "r:*") as archive:
             for member in archive:
                 if not member.isfile():
                     continue
-                _visit(member.name)
-                if len(matches) >= limit:
+
+                def _read(member=member) -> bytes:
+                    source = archive.extractfile(member)
+                    if source is None:
+                        return b""
+                    with source:
+                        return source.read()
+
+                _visit(member.name, _read)
+                if _done():
                     break
-    return matches
+        return content_name, content_data, listed, scanned
+
+    return None, None, listed, scanned
 
 
 def _safe_archive_member_name(member_name: str, archive_path: str) -> str:
@@ -783,8 +817,19 @@ def _parse_radar_xml_metadata_bytes(data: bytes) -> Tuple[Optional[List[Tuple[fl
             )
 
     coverage_polygon: Optional[List[Tuple[float, float]]] = None
+    corner_details: Dict[str, Dict[str, Any]] = {}
     if len(corners) >= 4:
         coverage_polygon = _ordered_closed_polygon_from_corners(corners[:4])
+        corner_details = {
+            str(item.get("name")): {
+                "lon": item.get("lon"),
+                "lat": item.get("lat"),
+                "ref_row": item.get("ref_row"),
+                "ref_col": item.get("ref_col"),
+            }
+            for item in corners
+            if item.get("name")
+        }
 
     start_time = (
         _xml_text_under_local_path(root, "start", "timeUTC")
@@ -834,6 +879,7 @@ def _parse_radar_xml_metadata_bytes(data: bytes) -> Tuple[Optional[List[Tuple[fl
             for item in corners
             if item.get("name")
         },
+        "corner_pixel_mapping": build_corner_pixel_mapping(corner_details),
         "coverage_polygon": coverage_polygon,
     }
     return coverage_polygon, {key: value for key, value in metadata.items() if value not in (None, "", [])}
@@ -890,6 +936,58 @@ def _metadata_document(
     }
 
 
+def _s1_annotation_sort_key(path: str) -> Tuple[int, str]:
+    normalized = str(path or "").replace("\\", "/")
+    lower = normalized.lower()
+    base = PurePosixPath(normalized).name.lower()
+    is_direct_annotation = "/annotation/" in lower and lower.count("/annotation/") == 1
+    is_measurement_annotation = is_direct_annotation and base.startswith("s1") and base.endswith(".xml")
+    if is_measurement_annotation:
+        rank = 0
+    elif "/annotation/calibration/" in lower:
+        rank = 2
+    elif "/annotation/noise/" in lower:
+        rank = 3
+    elif "/annotation/rfi/" in lower:
+        rank = 4
+    else:
+        rank = 1
+    return rank, lower
+
+
+def _parse_s1_preview_kml_bytes(data: bytes) -> Dict[str, Any]:
+    try:
+        root = etree.fromstring(data, parser=_xml_parser())
+    except Exception:
+        return {}
+    coordinate_texts = root.xpath("//*[local-name()='LatLonQuad']/*[local-name()='coordinates']/text()")
+    if not coordinate_texts:
+        return {}
+    points: List[Tuple[float, float]] = []
+    for token in str(coordinate_texts[0] or "").strip().split():
+        parts = token.split(",")
+        if len(parts) < 2:
+            continue
+        try:
+            points.append((float(parts[0]), float(parts[1])))
+        except ValueError:
+            continue
+    if len(points) != 4:
+        return {}
+    mapping = {
+        "bottom_left": [points[0][0], points[0][1]],
+        "bottom_right": [points[1][0], points[1][1]],
+        "top_right": [points[2][0], points[2][1]],
+        "top_left": [points[3][0], points[3][1]],
+        "source": "s1_preview_map_overlay_kml",
+    }
+    polygon = [points[0], points[1], points[2], points[3], points[0]]
+    return {
+        "preview_map_overlay_polygon": polygon,
+        "corner_pixel_mapping": mapping,
+    }
+
+
 def _extract_s1_annotation_documents(source_path: str, *, limit: int = 16) -> List[Dict[str, Any]]:
     docs: List[Dict[str, Any]] = []
     stat = _stat_path(source_path)
@@ -902,7 +1000,7 @@ def _extract_s1_annotation_documents(source_path: str, *, limit: int = 16) -> Li
             for file_name in files:
                 if file_name.lower().endswith(".xml"):
                     candidates.append(os.path.join(current_root, file_name))
-        for path in sorted(candidates)[: max(0, limit)]:
+        for path in sorted(candidates, key=_s1_annotation_sort_key)[: max(0, limit)]:
             try:
                 with open(path, "rb") as stream:
                     data = stream.read()
@@ -928,7 +1026,7 @@ def _extract_s1_annotation_documents(source_path: str, *, limit: int = 16) -> Li
                 for name in archive.namelist()
                 if "/annotation/" in name.lower() and name.lower().endswith(".xml")
             ]
-            for name in sorted(names)[: max(0, limit)]:
+            for name in sorted(names, key=_s1_annotation_sort_key)[: max(0, limit)]:
                 docs.append(
                     _metadata_document(
                         document_type="S1_ANNOTATION",
@@ -1099,13 +1197,48 @@ def _parse_s1_manifest_bytes(data: bytes) -> Dict[str, Any]:
 def _parse_s1_zip_manifest(path: str) -> Dict[str, Any]:
     stat = _stat_path(path)
     with zipfile.ZipFile(path) as archive:
-        manifest_name = next(
-            (name for name in archive.namelist() if name.lower().endswith("/manifest.safe") or name.lower() == "manifest.safe"),
-            None,
-        )
+        manifest_name = None
+        preview_kml_name = None
+        annotation_names: List[str] = []
+        for info in archive.infolist():
+            if info.is_dir():
+                continue
+            name = info.filename
+            lower = name.lower()
+            if manifest_name is None and (lower.endswith("/manifest.safe") or lower == "manifest.safe"):
+                manifest_name = name
+            if preview_kml_name is None and lower.endswith("/preview/map-overlay.kml"):
+                preview_kml_name = name
+            if "/annotation/" in lower and lower.endswith(".xml"):
+                annotation_names.append(name)
         if not manifest_name:
             return {"manifest_parse_status": "MISSING"}
         manifest_bytes = archive.read(manifest_name)
+        annotation_documents = [
+            _metadata_document(
+                document_type="S1_ANNOTATION",
+                member_path=name,
+                content=archive.read(name),
+                source_format="S1_ZIP",
+                satellite_family="S1",
+                archive_path=path,
+                archive_mtime=stat.get("mtime_epoch"),
+            )
+            for name in sorted(annotation_names, key=_s1_annotation_sort_key)[:16]
+        ]
+        preview_kml_bytes = archive.read(preview_kml_name) if preview_kml_name else None
+        preview_kml_meta = _parse_s1_preview_kml_bytes(preview_kml_bytes) if preview_kml_bytes else {}
+        preview_kml_documents = [
+            _metadata_document(
+                document_type="S1_PREVIEW_KML",
+                member_path=preview_kml_name or "preview/map-overlay.kml",
+                content=preview_kml_bytes,
+                source_format="S1_ZIP",
+                satellite_family="S1",
+                archive_path=path,
+                archive_mtime=stat.get("mtime_epoch"),
+            )
+        ] if preview_kml_bytes else []
         return {
             "manifest_parse_status": "OK",
             "manifest_path": manifest_name,
@@ -1119,8 +1252,10 @@ def _parse_s1_zip_manifest(path: str) -> Dict[str, Any]:
                     archive_path=path,
                     archive_mtime=stat.get("mtime_epoch"),
                 ),
-                *_extract_s1_annotation_documents(path),
+                *preview_kml_documents,
+                *annotation_documents,
             ],
+            **preview_kml_meta,
             **_parse_s1_manifest_bytes(manifest_bytes),
         }
 
@@ -1132,6 +1267,23 @@ def _parse_s1_safe_manifest(path: str) -> Dict[str, Any]:
     stat = _stat_path(path)
     with open(manifest_path, "rb") as stream:
         manifest_bytes = stream.read()
+        preview_kml_path = os.path.join(path, "preview", "map-overlay.kml")
+        preview_kml_bytes = None
+        if os.path.isfile(preview_kml_path):
+            with open(preview_kml_path, "rb") as preview_stream:
+                preview_kml_bytes = preview_stream.read()
+        preview_kml_meta = _parse_s1_preview_kml_bytes(preview_kml_bytes) if preview_kml_bytes else {}
+        preview_kml_documents = [
+            _metadata_document(
+                document_type="S1_PREVIEW_KML",
+                member_path="preview/map-overlay.kml",
+                content=preview_kml_bytes,
+                source_format="S1_SAFE_DIR",
+                satellite_family="S1",
+                archive_path=path,
+                archive_mtime=stat.get("mtime_epoch"),
+            )
+        ] if preview_kml_bytes else []
         return {
             "manifest_parse_status": "OK",
             "manifest_path": manifest_path,
@@ -1145,8 +1297,10 @@ def _parse_s1_safe_manifest(path: str) -> Dict[str, Any]:
                     archive_path=path,
                     archive_mtime=stat.get("mtime_epoch"),
                 ),
+                *preview_kml_documents,
                 *_extract_s1_annotation_documents(path),
             ],
+            **preview_kml_meta,
             **_parse_s1_manifest_bytes(manifest_bytes),
         }
 
@@ -1154,14 +1308,11 @@ def _parse_s1_safe_manifest(path: str) -> Dict[str, Any]:
 def _parse_lt1_archive_metadata(path: str) -> Dict[str, Any]:
     archive_stem = _strip_known_suffix(os.path.basename(path))
     stat = _stat_path(path)
-    xml_member, xml_data, members = _archive_read_first_matching(
+    xml_member, xml_data, tiff_members, members = _archive_collect_members(
         path,
-        lambda name: _archive_member_base_name(name).lower().endswith(".meta.xml"),
-    )
-    tiff_members = _archive_list_matching(
-        path,
-        lambda name: _archive_member_base_name(name).lower().endswith((".tiff", ".tif")),
-        limit=8,
+        content_predicate=lambda name: _archive_member_base_name(name).lower().endswith(".meta.xml"),
+        list_predicate=lambda name: _archive_member_base_name(name).lower().endswith((".tiff", ".tif")),
+        list_limit=8,
     )
     if not xml_member or not xml_data:
         return {
@@ -1193,14 +1344,11 @@ def _parse_lt1_archive_metadata(path: str) -> Dict[str, Any]:
 
 def _parse_gf3_archive_metadata(path: str) -> Dict[str, Any]:
     archive_stem = _strip_known_suffix(os.path.basename(path))
-    xml_member, xml_data, members = _archive_read_first_matching(
+    xml_member, xml_data, quicklooks, members = _archive_collect_members(
         path,
-        lambda name: _archive_member_base_name(name).lower().endswith(".xml"),
-    )
-    quicklooks = _archive_list_matching(
-        path,
-        lambda name: _archive_member_base_name(name).lower().endswith((".jpg", ".jpeg", ".png", ".bmp", "_ql.tif", "_ql.tiff")),
-        limit=8,
+        content_predicate=lambda name: _archive_member_base_name(name).lower().endswith(".xml"),
+        list_predicate=lambda name: _archive_member_base_name(name).lower().endswith((".jpg", ".jpeg", ".png", ".bmp", "_ql.tif", "_ql.tiff")),
+        list_limit=8,
     )
     if not xml_member or not xml_data:
         return {
@@ -1782,6 +1930,8 @@ def _cached_source_asset_is_unchanged(
     cached: Optional[Dict[str, Any]],
     stat: Dict[str, Optional[float]],
     root: ManagedRootORM,
+    *,
+    skip_unchanged_failures: bool = True,
 ) -> bool:
     if not cached:
         return False
@@ -1797,7 +1947,11 @@ def _cached_source_asset_is_unchanged(
         return False
     if str(cached.get("parser_version") or "") != PARSER_VERSION:
         return False
-    if str(cached.get("parse_status") or "").upper() != "OK":
+    parse_status = str(cached.get("parse_status") or "").upper()
+    allowed_statuses = {"OK"}
+    if skip_unchanged_failures:
+        allowed_statuses.update({"PARTIAL", "FAILED"})
+    if parse_status not in allowed_statuses:
         return False
     return _same_size(cached.get("size_bytes"), stat.get("size_bytes")) and _same_mtime(
         cached.get("mtime_epoch"),
@@ -1840,14 +1994,27 @@ def _collect_source_assets_incremental(
     log_callback: Optional[Callable[[str, str], None]] = None,
     progress_start: int = 0,
     progress_end: int = 100,
+    parse_workers: int = DEFAULT_ASSET_SCAN_PARSE_WORKERS,
+    parse_inflight: int = DEFAULT_ASSET_SCAN_PARSE_INFLIGHT,
+    skip_unchanged_failures: bool = True,
+    row_batch_callback: Optional[Callable[[List[Dict[str, Any]]], None]] = None,
+    row_batch_size: int = DEFAULT_ASSET_SCAN_DB_BATCH_SIZE,
 ) -> Tuple[List[Dict[str, Any]], List[Dict[str, Any]], int, int, List[str]]:
     rows: List[Dict[str, Any]] = []
+    pending_rows: List[Dict[str, Any]] = []
     issues: List[Dict[str, Any]] = []
     seen_paths: List[str] = []
+    seen_path_set: set[str] = set()
     entry_count = 0
     skipped_unchanged = 0
+    skipped_unchanged_ok = 0
+    skipped_unchanged_failed = 0
     parse_attempts = 0
+    parse_completed = 0
     last_progress_count = 0
+    parse_workers = max(1, int(parse_workers or 1))
+    parse_inflight = max(parse_workers, int(parse_inflight or parse_workers))
+    row_batch_size = max(1, int(row_batch_size or 1))
 
     def _log(level: str, message: str) -> None:
         if log_callback:
@@ -1857,39 +2024,37 @@ def _collect_source_assets_incremental(
         if progress_callback:
             progress_callback(_activity_progress(progress_start, progress_end, entry_count), message)
 
-    _log("INFO", f"Source root discovery started: {root.path}")
-    for path in _iter_source_candidates(root.path):
-        entry_count += 1
-        normalized_path = _normalize_path(path)
-        stat = _stat_path(normalized_path)
-        if _cached_source_asset_is_unchanged(existing_by_path.get(normalized_path), stat, root):
-            seen_paths.append(normalized_path)
-            skipped_unchanged += 1
-            if skipped_unchanged % ASSET_SCAN_LOG_INTERVAL == 0:
-                _log(
-                    "INFO",
-                    f"Skipped unchanged source archives: {skipped_unchanged} (candidates={entry_count})",
-                )
-            if entry_count - last_progress_count >= ASSET_SCAN_LOG_INTERVAL:
-                _progress(
-                    "Scanning source archives: "
-                    f"candidates={entry_count}, skipped={skipped_unchanged}, parsed={len(rows)}, issues={len(issues)}"
-                )
-                last_progress_count = entry_count
-            continue
-        parse_attempts += 1
+    def _mark_seen(path: str) -> None:
+        normalized = _normalize_path(path)
+        if normalized and normalized not in seen_path_set:
+            seen_path_set.add(normalized)
+            seen_paths.append(normalized)
+
+    def _emit_row_batch(*, force: bool = False) -> None:
+        if not row_batch_callback or not pending_rows:
+            return
+        if not force and len(pending_rows) < row_batch_size:
+            return
+        batch = pending_rows[:]
+        pending_rows.clear()
+        row_batch_callback(batch)
+
+    def _parse_one(index: int, normalized_path: str) -> Dict[str, Any]:
         file_name = os.path.basename(normalized_path)
-        if parse_attempts <= ASSET_SCAN_DETAILED_PARSE_LOG_LIMIT or parse_attempts % ASSET_SCAN_LOG_INTERVAL == 0:
-            _log("INFO", f"Extracting source archive metadata {parse_attempts}: {file_name}")
-        if parse_attempts <= 50 or parse_attempts % 25 == 0:
-            _progress(
-                "Extracting source archive metadata: "
-                f"{file_name} (changed/new={parse_attempts}, skipped={skipped_unchanged})"
-            )
         try:
-            row = _parse_source_entry(normalized_path, root)
+            row = _parse_source_entry(normalized_path, parse_root)
+            return {"index": index, "path": normalized_path, "file_name": file_name, "row": row, "error": None}
         except Exception as exc:
-            row = None
+            return {"index": index, "path": normalized_path, "file_name": file_name, "row": None, "error": exc}
+
+    def _handle_parse_result(result: Dict[str, Any]) -> None:
+        nonlocal parse_completed, last_progress_count
+        parse_completed += 1
+        normalized_path = str(result.get("path") or "")
+        file_name = str(result.get("file_name") or os.path.basename(normalized_path))
+        exc = result.get("error")
+        if exc is not None:
+            _mark_seen(normalized_path)
             _log("WARNING", f"Source archive metadata parse failed: {file_name}: {exc}")
             issues.append(
                 {
@@ -1899,10 +2064,14 @@ def _collect_source_assets_incremental(
                     "source_path": normalized_path,
                 }
             )
+            return
+        row = result.get("row")
         if row is None:
-            continue
+            _mark_seen(normalized_path)
+            return
         rows.append(row)
-        seen_paths.append(str(row["file_path"]))
+        pending_rows.append(row)
+        _mark_seen(str(row["file_path"]))
         if row.get("parse_status") in {"FAILED", "PARTIAL"}:
             _log(
                 "WARNING",
@@ -1920,17 +2089,91 @@ def _collect_source_assets_incremental(
         if entry_count - last_progress_count >= ASSET_SCAN_LOG_INTERVAL:
             _progress(
                 "Scanning source archives: "
-                f"candidates={entry_count}, skipped={skipped_unchanged}, parsed={len(rows)}, issues={len(issues)}"
+                f"candidates={entry_count}, skipped={skipped_unchanged}, parsed={len(rows)}, "
+                f"completed={parse_completed}/{parse_attempts}, issues={len(issues)}"
             )
             last_progress_count = entry_count
+        _emit_row_batch()
+
+    def _drain_completed(pending: set, *, wait_for_one: bool = False) -> set:
+        if not pending:
+            return pending
+        timeout = None if wait_for_one else 0
+        done, remaining = wait(pending, timeout=timeout, return_when=FIRST_COMPLETED)
+        for future in done:
+            _handle_parse_result(future.result())
+        return remaining
+
+    _log(
+        "INFO",
+        "Source root discovery started: "
+        f"{root.path} (workers={parse_workers}, inflight={parse_inflight}, "
+        f"db_batch_size={row_batch_size}, skip_unchanged_failures={skip_unchanged_failures})",
+    )
+    parse_root = SimpleNamespace(id=root.id, path=root.path)
+    with ThreadPoolExecutor(max_workers=parse_workers, thread_name_prefix="asset-parse") as executor:
+        pending = set()
+        for path in _iter_source_candidates(root.path):
+            entry_count += 1
+            normalized_path = _normalize_path(path)
+            stat = _stat_path(normalized_path)
+            cached = existing_by_path.get(normalized_path)
+            if _cached_source_asset_is_unchanged(
+                cached,
+                stat,
+                root,
+                skip_unchanged_failures=skip_unchanged_failures,
+            ):
+                _mark_seen(normalized_path)
+                skipped_unchanged += 1
+                cached_status = str((cached or {}).get("parse_status") or "").upper()
+                if cached_status == "OK":
+                    skipped_unchanged_ok += 1
+                else:
+                    skipped_unchanged_failed += 1
+                if skipped_unchanged % ASSET_SCAN_LOG_INTERVAL == 0:
+                    _log(
+                        "INFO",
+                        "Skipped unchanged source archives: "
+                        f"{skipped_unchanged} (ok={skipped_unchanged_ok}, failed_cached={skipped_unchanged_failed}, "
+                        f"candidates={entry_count})",
+                    )
+                if entry_count - last_progress_count >= ASSET_SCAN_LOG_INTERVAL:
+                    _progress(
+                        "Scanning source archives: "
+                        f"candidates={entry_count}, skipped={skipped_unchanged}, parsed={len(rows)}, "
+                        f"completed={parse_completed}/{parse_attempts}, issues={len(issues)}"
+                    )
+                    last_progress_count = entry_count
+                continue
+            parse_attempts += 1
+            file_name = os.path.basename(normalized_path)
+            if parse_attempts <= ASSET_SCAN_DETAILED_PARSE_LOG_LIMIT or parse_attempts % ASSET_SCAN_LOG_INTERVAL == 0:
+                _log("INFO", f"Extracting source archive metadata {parse_attempts}: {file_name}")
+            if parse_attempts <= 50 or parse_attempts % 25 == 0:
+                _progress(
+                    "Extracting source archive metadata: "
+                    f"{file_name} (changed/new={parse_attempts}, completed={parse_completed}, "
+                    f"workers={parse_workers}, skipped={skipped_unchanged}, issue={len(issues)})"
+                )
+            pending.add(executor.submit(_parse_one, parse_attempts, normalized_path))
+            while len(pending) >= parse_inflight:
+                pending = _drain_completed(pending, wait_for_one=True)
+            pending = _drain_completed(pending, wait_for_one=False)
+        while pending:
+            pending = _drain_completed(pending, wait_for_one=True)
+    _emit_row_batch(force=True)
     _log(
         "INFO",
         "Source root discovery finished: "
-        f"candidates={entry_count}, skipped={skipped_unchanged}, changed_or_new={len(rows)}, issues={len(issues)}",
+        f"candidates={entry_count}, skipped={skipped_unchanged}, "
+        f"skipped_ok={skipped_unchanged_ok}, skipped_failed_cached={skipped_unchanged_failed}, "
+        f"changed_or_new={len(rows)}, parse_attempts={parse_attempts}, issues={len(issues)}",
     )
     _progress(
         "Source archive discovery finished: "
-        f"candidates={entry_count}, skipped={skipped_unchanged}, changed_or_new={len(rows)}"
+        f"candidates={entry_count}, skipped={skipped_unchanged}, changed_or_new={len(rows)}, "
+        f"workers={parse_workers}"
     )
     return rows, issues, entry_count, skipped_unchanged, seen_paths
 
@@ -2152,6 +2395,8 @@ class AssetInventoryService:
         task_id: Optional[str] = None,
         progress_start: int = 84,
         progress_end: int = 96,
+        progress_callback: Optional[Callable[[Dict[str, Any]], None]] = None,
+        progress_interval: int = 1,
     ) -> Dict[str, Any]:
         generated_session = db is None
         if generated_session:
@@ -2218,6 +2463,16 @@ class AssetInventoryService:
             if limit and limit > 0:
                 candidates = candidates[: int(limit)]
             summary["candidate_count"] = len(candidates)
+            if progress_callback:
+                progress_callback(
+                    {
+                        "event": "planned",
+                        "records_seen": summary["records_seen"],
+                        "candidate_count": summary["candidate_count"],
+                        "skipped_ready": summary["skipped_ready"],
+                        "families": target_families,
+                    }
+                )
             if not apply:
                 return summary
             if not candidates:
@@ -2226,6 +2481,8 @@ class AssetInventoryService:
                     f"Archive preview cache already ready: skipped={summary['skipped_ready']}",
                     progress_end,
                 )
+                if progress_callback:
+                    progress_callback({"event": "completed", **summary})
                 return summary
 
             await self._progress(
@@ -2235,6 +2492,7 @@ class AssetInventoryService:
             )
             thumb_size = (settings.RADAR_THUMBNAIL_MAX_SIZE, settings.RADAR_THUMBNAIL_MAX_SIZE)
             total = len(candidates)
+            progress_interval = max(1, int(progress_interval or 1))
             for index, record in enumerate(candidates, start=1):
                 unique_id = record.unique_id or record.file_path
                 raw_cache_path = DataService.get_radar_raw_cache_path(unique_id, record.file_path)
@@ -2277,10 +2535,7 @@ class AssetInventoryService:
                             await task_service.add_log(task_id, "ERROR", f"Preview geometry invalid: {product_name}: {record.preview_cache_error}")
                         db.add(record)
                     else:
-                        source_corner_mapping = await asyncio.to_thread(
-                            DataService.get_radar_source_corner_mapping,
-                            record.file_path,
-                        )
+                        source_corner_mapping = DataService.get_radar_record_corner_mapping(record)
                         ok_geo, geo_error = await asyncio.to_thread(
                             image_service.create_geocorrected_radar_cached_image,
                             preview_source,
@@ -2327,6 +2582,31 @@ class AssetInventoryService:
                     f"Building archive preview cache ({index}/{total}): ready={summary['ready']}, failed={summary['failed']}, missing={summary['missing_source']}",
                     min(progress_end, progress),
                 )
+                if progress_callback and (
+                    index == 1
+                    or index == total
+                    or index % progress_interval == 0
+                    or (record.preview_cache_status or "").upper() == "FAILED"
+                ):
+                    progress_callback(
+                        {
+                            "event": "item",
+                            "processed": index,
+                            "total": total,
+                            "records_seen": summary["records_seen"],
+                            "candidate_count": summary["candidate_count"],
+                            "skipped_ready": summary["skipped_ready"],
+                            "ready": summary["ready"],
+                            "cached": summary["cached"],
+                            "failed": summary["failed"],
+                            "missing_source": summary["missing_source"],
+                            "raw_cached": summary["raw_cached"],
+                            "raw_failed": summary["raw_failed"],
+                            "product_name": product_name,
+                            "status": record.preview_cache_status,
+                            "error": record.preview_cache_error,
+                        }
+                    )
                 await db.commit()
 
             await self._progress(
@@ -2338,6 +2618,8 @@ class AssetInventoryService:
                 ),
                 progress_end,
             )
+            if progress_callback:
+                progress_callback({"event": "completed", **summary})
             return summary
         except Exception:
             if db is not None:
@@ -2748,6 +3030,109 @@ class AssetInventoryService:
 
         loop = asyncio.get_running_loop()
         progress_callback, log_callback, drain_thread_events = self._thread_callbacks(task_id, loop)
+        batch_write_start = max(progress_start, progress_end - 10)
+        batch_write_end = max(batch_write_start, progress_end - 3)
+        persisted_changed_count = 0
+
+        async def _upsert_source_asset_batch(rows_batch: Sequence[Dict[str, Any]], *, batch_index: int) -> int:
+            if not rows_batch:
+                return 0
+            now = _utcnow()
+            db_rows = [
+                {key: value for key, value in row.items() if not str(key).startswith("_")}
+                for row in rows_batch
+            ]
+            for row in db_rows:
+                stmt = pg_insert(SourceProductAssetORM).values(row)
+                excluded = stmt.excluded
+                stmt = stmt.on_conflict_do_update(
+                    index_elements=["file_path"],
+                    set_={
+                        "asset_uid": excluded.asset_uid,
+                        "logical_product_uid": excluded.logical_product_uid,
+                        "satellite_family": excluded.satellite_family,
+                        "satellite": excluded.satellite,
+                        "source_format": excluded.source_format,
+                        "product_type": excluded.product_type,
+                        "product_level": excluded.product_level,
+                        "imaging_mode": excluded.imaging_mode,
+                        "polarization": excluded.polarization,
+                        "absolute_orbit": excluded.absolute_orbit,
+                        "relative_orbit": excluded.relative_orbit,
+                        "orbit_direction": excluded.orbit_direction,
+                        "acquisition_start_time_utc": excluded.acquisition_start_time_utc,
+                        "acquisition_stop_time_utc": excluded.acquisition_stop_time_utc,
+                        "imaging_date": excluded.imaging_date,
+                        "root_ref_id": excluded.root_ref_id,
+                        "root_path": excluded.root_path,
+                        "archive_path": excluded.archive_path,
+                        "path_kind": excluded.path_kind,
+                        "file_name": excluded.file_name,
+                        "file_stem": excluded.file_stem,
+                        "file_ext": excluded.file_ext,
+                        "size_bytes": excluded.size_bytes,
+                        "mtime_epoch": excluded.mtime_epoch,
+                        "checksum_status": excluded.checksum_status,
+                        "archive_integrity_status": "NOT_CHECKED",
+                        "archive_integrity_method": None,
+                        "archive_integrity_checked_at": None,
+                        "archive_integrity_error": None,
+                        "archive_integrity_version": None,
+                        "archive_integrity_member_count": None,
+                        "parser_name": excluded.parser_name,
+                        "parser_version": excluded.parser_version,
+                        "parse_status": excluded.parse_status,
+                        "parse_error": excluded.parse_error,
+                        "parsed_at": excluded.parsed_at,
+                        "metadata_json": excluded.metadata_json,
+                        "is_active": True,
+                        "missing_since": None,
+                        "updated_at": now,
+                    },
+                )
+                await db.execute(stmt)
+            await db.flush()
+
+            changed_paths = [str(row["file_path"]) for row in rows_batch if row.get("file_path")]
+            asset_ids_by_path: Dict[str, int] = {}
+            if changed_paths:
+                result = await db.execute(
+                    select(SourceProductAssetORM.file_path, SourceProductAssetORM.id).where(
+                        SourceProductAssetORM.file_path.in_(changed_paths)
+                    )
+                )
+                asset_ids_by_path = {str(path): int(asset_id) for path, asset_id in result.all()}
+                await self._upsert_metadata_documents_for_source_assets(db, rows_batch, asset_ids_by_path)
+                await self._upsert_radar_records_for_source_assets(db, rows_batch, asset_ids_by_path)
+
+            await db.commit()
+            if task_id:
+                await task_service.add_log(
+                    task_id,
+                    "INFO",
+                    f"Source asset DB batch committed: batch={batch_index}, rows={len(rows_batch)}",
+                )
+                await self._progress(
+                    task_id,
+                    f"Committed source asset DB batch {batch_index}: rows={len(rows_batch)}",
+                    batch_write_start,
+                )
+            return len(rows_batch)
+
+        batch_index = 0
+
+        def _persist_row_batch(rows_batch: List[Dict[str, Any]]) -> None:
+            nonlocal batch_index, persisted_changed_count
+            if not rows_batch:
+                return
+            batch_index += 1
+            future = asyncio.run_coroutine_threadsafe(
+                _upsert_source_asset_batch(rows_batch, batch_index=batch_index),
+                loop,
+            )
+            persisted = int(future.result())
+            persisted_changed_count += persisted
+
         rows, issues, entry_count, skipped_unchanged, seen_paths = await asyncio.to_thread(
             _collect_source_assets_incremental,
             root,
@@ -2756,104 +3141,24 @@ class AssetInventoryService:
             log_callback=log_callback,
             progress_start=progress_start,
             progress_end=max(progress_start + 1, progress_end - 8),
+            parse_workers=settings.ASSET_SCAN_PARSE_WORKERS,
+            parse_inflight=settings.ASSET_SCAN_PARSE_INFLIGHT,
+            skip_unchanged_failures=settings.ASSET_SCAN_SKIP_UNCHANGED_FAILURES,
+            row_batch_callback=_persist_row_batch if task_id else None,
+            row_batch_size=settings.ASSET_SCAN_DB_BATCH_SIZE,
         )
         await drain_thread_events()
-        changed_paths = [row["file_path"] for row in rows]
+        if not task_id and rows:
+            persisted_changed_count += await _upsert_source_asset_batch(rows, batch_index=1)
         now = _utcnow()
-        write_start = max(progress_start, progress_end - 7)
-        write_end = max(write_start, progress_end - 3)
-        await self._progress(
-            task_id,
-            f"Writing source asset index: changed_or_new={len(rows)}, skipped={skipped_unchanged}",
-            write_start,
-        )
-        db_rows = [
-            {key: value for key, value in row.items() if not str(key).startswith("_")}
-            for row in rows
-        ]
         if task_id:
             await task_service.add_log(
                 task_id,
                 "INFO",
-                f"Source asset DB upsert started: changed_or_new={len(rows)}, skipped={skipped_unchanged}, seen={len(seen_paths)}",
+                "Source asset DB batch upsert finished: "
+                f"changed_or_new={len(rows)}, persisted={persisted_changed_count}, "
+                f"batches={batch_index}, skipped={skipped_unchanged}, seen={len(seen_paths)}",
             )
-        for index, row in enumerate(db_rows, start=1):
-            stmt = pg_insert(SourceProductAssetORM).values(row)
-            excluded = stmt.excluded
-            stmt = stmt.on_conflict_do_update(
-                index_elements=["file_path"],
-                set_={
-                    "asset_uid": excluded.asset_uid,
-                    "logical_product_uid": excluded.logical_product_uid,
-                    "satellite_family": excluded.satellite_family,
-                    "satellite": excluded.satellite,
-                    "source_format": excluded.source_format,
-                    "product_type": excluded.product_type,
-                    "product_level": excluded.product_level,
-                    "imaging_mode": excluded.imaging_mode,
-                    "polarization": excluded.polarization,
-                    "absolute_orbit": excluded.absolute_orbit,
-                    "relative_orbit": excluded.relative_orbit,
-                    "orbit_direction": excluded.orbit_direction,
-                    "acquisition_start_time_utc": excluded.acquisition_start_time_utc,
-                    "acquisition_stop_time_utc": excluded.acquisition_stop_time_utc,
-                    "imaging_date": excluded.imaging_date,
-                    "root_ref_id": excluded.root_ref_id,
-                    "root_path": excluded.root_path,
-                    "archive_path": excluded.archive_path,
-                    "path_kind": excluded.path_kind,
-                    "file_name": excluded.file_name,
-                    "file_stem": excluded.file_stem,
-                    "file_ext": excluded.file_ext,
-                    "size_bytes": excluded.size_bytes,
-                    "mtime_epoch": excluded.mtime_epoch,
-                    "checksum_status": excluded.checksum_status,
-                    "archive_integrity_status": "NOT_CHECKED",
-                    "archive_integrity_method": None,
-                    "archive_integrity_checked_at": None,
-                    "archive_integrity_error": None,
-                    "archive_integrity_version": None,
-                    "archive_integrity_member_count": None,
-                    "parser_name": excluded.parser_name,
-                    "parser_version": excluded.parser_version,
-                    "parse_status": excluded.parse_status,
-                    "parse_error": excluded.parse_error,
-                    "parsed_at": excluded.parsed_at,
-                    "metadata_json": excluded.metadata_json,
-                    "is_active": True,
-                    "missing_since": None,
-                    "updated_at": now,
-                },
-            )
-            await db.execute(stmt)
-            if index % ASSET_SCAN_LOG_INTERVAL == 0 or index == len(rows):
-                progress_value = write_start
-                if rows and write_end > write_start:
-                    progress_value = write_start + int(index / max(1, len(rows)) * (write_end - write_start))
-                await self._progress(
-                    task_id,
-                    f"Writing source asset index: {index}/{len(rows)} changed_or_new, skipped={skipped_unchanged}",
-                    progress_value,
-                )
-        await db.flush()
-
-        asset_ids_by_path: Dict[str, int] = {}
-        if changed_paths:
-            await self._progress(
-                task_id,
-                f"Updating radar scene records for changed source assets: {len(changed_paths)}",
-                max(write_end, progress_end - 2),
-            )
-            result = await db.execute(
-                select(SourceProductAssetORM.file_path, SourceProductAssetORM.id).where(
-                    SourceProductAssetORM.file_path.in_(changed_paths)
-                )
-            )
-            asset_ids_by_path = {str(path): int(asset_id) for path, asset_id in result.all()}
-            await self._upsert_metadata_documents_for_source_assets(db, rows, asset_ids_by_path)
-            await self._upsert_radar_records_for_source_assets(db, rows, asset_ids_by_path)
-        elif task_id:
-            await task_service.add_log(task_id, "INFO", "No changed source assets; radar scene record update skipped.")
 
         await self._progress(task_id, "Marking missing source assets and refreshing scan issues...", max(progress_end - 2, progress_start))
         await self._mark_missing_source_assets(db, root, seen_paths, now)

backend/app/services/data_service.py

@@ -644,6 +644,21 @@ class DataService:
             return corner_mapping
         return None
 
+    @staticmethod
+    def get_radar_record_corner_mapping(record: Any) -> Optional[Dict[str, Any]]:
+        record_state = getattr(record, "__dict__", {}) if record is not None else {}
+        for metadata in (
+            getattr(record, "metadata_json", None),
+            getattr(record_state.get("source_product_asset"), "metadata_json", None),
+            getattr(record_state.get("source_archive_asset"), "metadata_json", None),
+        ):
+            if not isinstance(metadata, dict):
+                continue
+            corner_mapping = metadata.get("corner_pixel_mapping")
+            if isinstance(corner_mapping, dict):
+                return corner_mapping
+        return DataService.get_radar_source_corner_mapping(str(getattr(record, "file_path", "") or ""))
+
     @staticmethod
     def _normalize_coverage_polygon(coverage_polygon: Any) -> Optional[List[Tuple[float, float]]]:
         if isinstance(coverage_polygon, list):
@@ -1102,6 +1117,7 @@ class DataService:
                         "max_lon": record.max_lon,
                         "max_lat": record.max_lat,
                         "preview_cache_version": record.preview_cache_version,
+                        "source_corner_mapping": DataService.get_radar_record_corner_mapping(record),
                     }
                 )
 
@@ -1164,10 +1180,7 @@ class DataService:
                     except (TypeError, ValueError):
                         bbox = None
 
-                    source_corner_mapping = await asyncio.to_thread(
-                        DataService.get_radar_source_corner_mapping,
-                        file_path,
-                    )
+                    source_corner_mapping = item.get("source_corner_mapping")
 
                     need_geo_rebuild = (
                         (not geo_cache_mtime)

backend/app/services/dinsar_production_service.py

@@ -35,6 +35,7 @@ from .workflow_service import workflow_service
 TASK_TYPE_DINSAR_PRODUCTION = "IDL_RUN_DINSAR"
 TASK_TYPE_PYINT_DINSAR_PRODUCTION = "PYINT_RUN"
 TASK_TYPE_LANDSAR_DINSAR_PRODUCTION = "LANDSAR_RUN"
+TASK_TYPE_LANDSAR_CLUSTER_PRODUCTION = "LANDSAR_CLUSTER_RUN"
 RUN_STATUS_PENDING = "PENDING"
 RUN_STATUS_RUNNING = "RUNNING"
 RUN_STATUS_COMPLETED = "COMPLETED"
@@ -775,6 +776,172 @@ class DinsarProductionService:
             "rerun_mode": selection["rerun_mode"],
         }
 
+    async def create_landsar_cluster_run(
+        self,
+        *,
+        profile_code: str,
+        root_dir: str,
+        num_to_process: int,
+        rerun_mode: Optional[str],
+        timeout_seconds: Optional[int],
+        extra: Optional[Dict[str, Any]],
+        created_by: Optional[str],
+        max_attempts: int,
+        db: AsyncSession,
+    ) -> Dict[str, Any]:
+        normalized_engine = "landsar"
+        normalized_profile = str(profile_code or "").strip()
+        normalized_root = _normalize_dir(root_dir, "root_dir")
+        selection = await asyncio.to_thread(
+            _select_run_items,
+            normalized_root,
+            engine_code=normalized_engine,
+            profile_code=normalized_profile,
+            num_to_process=max(0, int(num_to_process or 0)),
+            rerun_mode=rerun_mode,
+        )
+        item_payloads = selection["items"]
+        if not item_payloads:
+            if (
+                selection["discovered_task_count"] > 0
+                and selection["skipped_completed_count"] > 0
+                and selection["rerun_mode"] == RERUN_MODE_UNFINISHED_ONLY
+            ):
+                raise ValueError(
+                    f"All discovered Task_* directories already have completed "
+                    f"landsar/{normalized_profile} results under: {normalized_root}"
+                )
+            raise ValueError(f"No Task_* directories found under: {normalized_root}")
+
+        run_id = str(uuid.uuid4())
+        mode = "cluster"
+        task_name = f"D-InSAR production: landsar/{normalized_profile} cluster"
+        task_params = {
+            "engine_code": normalized_engine,
+            "profile": normalized_profile,
+            "root_dir": normalized_root,
+            "num_to_process": int(num_to_process or 0),
+            "rerun_mode": selection["rerun_mode"],
+            "timeout_seconds": timeout_seconds,
+            "extra": dict(extra or {}),
+            "mode": mode,
+            "production_run_id": run_id,
+            "cluster_job_type": "LANDSAR_CLUSTER_ITEM",
+        }
+
+        task_id: Optional[str] = None
+        created_items: List[DinsarProductionRunItemORM] = []
+        try:
+            task_id = await task_service.create_task(
+                task_type=TASK_TYPE_LANDSAR_CLUSTER_PRODUCTION,
+                task_name=task_name,
+                params=task_params,
+                db=db,
+            )
+
+            run = DinsarProductionRunORM(
+                run_id=run_id,
+                task_id=task_id,
+                product_family="dinsar",
+                engine_code=normalized_engine,
+                profile_code=normalized_profile,
+                mode=mode,
+                source_root=normalized_root,
+                publish_root_dir=settings.DINSAR_PRODUCT_DIR,
+                status=RUN_STATUS_PENDING,
+                cancel_requested=False,
+                total_items=len(item_payloads),
+                completed_items=0,
+                failed_items=0,
+                skipped_items=0,
+                latest_message="Queued LandSAR cluster items",
+                params_json=task_params,
+                summary_json={
+                    "phase": "queued",
+                    "selected_task_count": len(item_payloads),
+                    "discovered_task_count": selection["discovered_task_count"],
+                    "skipped_completed_count": selection["skipped_completed_count"],
+                    "rerun_mode": selection["rerun_mode"],
+                    "product_family": "dinsar",
+                    "publish_root_dir": settings.DINSAR_PRODUCT_DIR,
+                    "cluster_job_type": "LANDSAR_CLUSTER_ITEM",
+                },
+                created_by=created_by,
+            )
+            db.add(run)
+            await db.flush()
+
+            for item_payload in item_payloads:
+                item = DinsarProductionRunItemORM(
+                    run_id=run_id,
+                    order_index=item_payload["order_index"],
+                    task_name=item_payload["task_name"],
+                    task_alias=item_payload["task_alias"],
+                    pair_key=item_payload["pair_key"],
+                    pair_uid=item_payload["pair_uid"],
+                    network_run_id=item_payload["network_run_id"],
+                    network_edge_id=item_payload["network_edge_id"],
+                    policy_version=item_payload["policy_version"],
+                    selection_strategy=item_payload["selection_strategy"],
+                    source_task_dir=item_payload["source_task_dir"],
+                    results_root_dir=item_payload["results_root_dir"],
+                    status=RUN_ITEM_STATUS_PENDING,
+                )
+                db.add(item)
+                created_items.append(item)
+
+            await db.flush()
+
+            from .job_queue_service import job_queue_service
+
+            for item in created_items:
+                await job_queue_service.create_job(
+                    "LANDSAR_CLUSTER_ITEM",
+                    payload={
+                        "production_run_id": run_id,
+                        "item_id": item.id,
+                    },
+                    task_id=task_id,
+                    priority=0,
+                    max_attempts=max(1, int(max_attempts or 1)),
+                    db=db,
+                )
+
+            await db.commit()
+            await db.refresh(run)
+        except Exception as exc:
+            await db.rollback()
+            if task_id:
+                try:
+                    await task_service.update_task(
+                        task_id,
+                        status="FAILED",
+                        message=f"Failed to create LandSAR cluster run: {exc}",
+                    )
+                except Exception:
+                    pass
+            raise
+
+        await asyncio.to_thread(
+            _append_run_log_sync,
+            run_id,
+            (
+                f"[cluster-queued] run_id={run_id} profile={normalized_profile} root={normalized_root} "
+                f"items={len(item_payloads)} rerun_mode={selection['rerun_mode']} "
+                f"skipped_completed={selection['skipped_completed_count']}"
+            ),
+        )
+        return {
+            "run_id": run_id,
+            "task_id": task_id,
+            "workflow_run_id": None,
+            "status": run.status,
+            "selected_task_count": len(item_payloads),
+            "discovered_task_count": selection["discovered_task_count"],
+            "skipped_completed_count": selection["skipped_completed_count"],
+            "rerun_mode": selection["rerun_mode"],
+        }
+
     async def list_runs(
         self,
         db: AsyncSession,
@@ -923,6 +1090,80 @@ class DinsarProductionService:
             run.latest_message = latest_message
         return run
 
+    async def finalize_cluster_run_if_complete(
+        self,
+        run: DinsarProductionRunORM,
+        *,
+        db: AsyncSession,
+        publish_error: Optional[str] = None,
+    ) -> Optional[str]:
+        rows = await db.execute(
+            select(DinsarProductionRunItemORM.status, func.count(DinsarProductionRunItemORM.id))
+            .where(DinsarProductionRunItemORM.run_id == run.run_id)
+            .group_by(DinsarProductionRunItemORM.status)
+        )
+        counts = {str(status or "").upper(): int(count or 0) for status, count in rows.fetchall()}
+        total_items = int(run.total_items or 0)
+        completed = counts.get(RUN_ITEM_STATUS_COMPLETED, 0)
+        failed = counts.get(RUN_ITEM_STATUS_FAILED, 0)
+        skipped = counts.get(RUN_ITEM_STATUS_SKIPPED, 0)
+        cancelled = counts.get(RUN_ITEM_STATUS_CANCELLED, 0)
+        terminal_count = completed + failed + skipped + cancelled
+
+        run.completed_items = completed
+        run.failed_items = failed
+        run.skipped_items = skipped
+        if total_items <= 0 or terminal_count < total_items:
+            run.status = RUN_STATUS_RUNNING
+            run.latest_message = (
+                f"LandSAR cluster running: completed={completed} failed={failed} "
+                f"skipped={skipped} cancelled={cancelled} total={total_items}"
+            )
+            await db.commit()
+            return None
+
+        if publish_error:
+            final_status = RUN_STATUS_FAILED
+            latest_message = f"LandSAR cluster publish failed: {publish_error}"
+        elif cancelled > 0 or bool(run.cancel_requested):
+            final_status = RUN_STATUS_CANCELLED
+            latest_message = (
+                f"LandSAR cluster cancelled. completed={completed} failed={failed} "
+                f"cancelled={cancelled} total={total_items}"
+            )
+        elif failed > 0:
+            final_status = RUN_STATUS_FAILED
+            latest_message = (
+                f"LandSAR cluster finished with failures. completed={completed} "
+                f"failed={failed} total={total_items}"
+            )
+        else:
+            final_status = RUN_STATUS_COMPLETED
+            latest_message = (
+                f"LandSAR cluster completed. completed={completed} failed={failed} total={total_items}"
+            )
+
+        summary_payload = dict(run.summary_json or {})
+        summary_payload.update(
+            {
+                "phase": "completed",
+                "cluster": True,
+                "total_items": total_items,
+                "completed_items": completed,
+                "failed_items": failed,
+                "skipped_items": skipped,
+                "cancelled_items": cancelled,
+                "publish_error": publish_error,
+            }
+        )
+        run.status = final_status
+        run.summary_json = summary_payload
+        run.latest_message = latest_message
+        run.ended_at = _utcnow()
+        await db.commit()
+        await asyncio.to_thread(_append_run_log_sync, run.run_id, f"[cluster-finish] status={final_status} message={latest_message}")
+        return final_status
+
     async def mark_run_started(
         self,
         run: DinsarProductionRunORM,

backend/app/services/health_service.py

@@ -1136,7 +1136,7 @@ async def _check_product_packages() -> Dict[str, Any]:
                 status["missing_publish_dir_count"] += 1
             if not str(processor_code or "").strip():
                 status["missing_processor_count"] += 1
-            if engine_key in {"isce2", "pyint", "gamma"} and not str(runtime_id or "").strip():
+            if engine_key in {"pyint", "gamma"} and not str(runtime_id or "").strip():
                 status["missing_runtime_count"] += 1
             if not str(native_output_dir or "").strip():
                 status["missing_native_output_count"] += 1
@@ -1386,7 +1386,7 @@ async def _check_wsl_runtime() -> Dict[str, Any]:
     }
     try:
         required_by_engine = {
-            "isce2": bool(settings.ISCE2_ENABLED or settings.TIMESERIES_ENABLED),
+            "isce2": bool(settings.ISCE2_ENABLED),
             "pyint": bool(settings.PYINT_ENABLED),
             "gamma": bool(settings.GAMMA_SBAS_ENABLED),
         }

backend/app/services/job_handlers.py

@@ -19,7 +19,7 @@ from sqlalchemy import select
 
 from .. import database
 from ..config import settings, split_env_paths
-from ..models import SystemJobORM, DinsarResultORM, HazardPointORM, DinsarTaskItemORM, PsTaskItemORM, RadarDataORM, SARSceneGeoORM, FloodDetectionORM, WaterDetectionORM, WaterExtractionORM, GF3ProcessingORM, AiDiagnosisORM
+from ..models import SystemJobORM, DinsarResultORM, HazardPointORM, DinsarTaskItemORM, PsTaskItemORM, RadarDataORM, SARSceneGeoORM, FloodDetectionORM, WaterDetectionORM, WaterExtractionORM, GF3ProcessingORM, AiDiagnosisORM, DinsarProductionRunItemORM
 from ..scheduler import scan_data_job
 from .data_service import data_service
 from .asset_inventory_service import asset_inventory_service
@@ -93,6 +93,7 @@ JOB_TYPE_GF3_SARSCAPE_CLEAN = "GF3_SARSCAPE_CLEAN"
 JOB_TYPE_ISCE2_RUN = "ISCE2_RUN"
 JOB_TYPE_PYINT_RUN = "PYINT_RUN"
 JOB_TYPE_LANDSAR_RUN = "LANDSAR_RUN"
+JOB_TYPE_LANDSAR_CLUSTER_ITEM = "LANDSAR_CLUSTER_ITEM"
 JOB_TYPE_PUBLISH_DINSAR_PRODUCTS = "PUBLISH_DINSAR_PRODUCTS"
 JOB_TYPE_REBUILD_DINSAR_CATALOG = "REBUILD_DINSAR_CATALOG"
 JOB_TYPE_REBUILD_PSINSAR_CATALOG = "REBUILD_PSINSAR_CATALOG"
@@ -107,6 +108,15 @@ JOB_TYPE_SBAS_GAMMA_WORKFLOW = "SBAS_GAMMA_WORKFLOW"
 JOB_TYPE_SBAS_LANDSAR_WORKFLOW = "SBAS_LANDSAR_WORKFLOW"
 
 COPY_ALLOWED_STATUSES = {"PENDING", "IN_PROGRESS", "COMPLETED", "FAILED"}
+_LOCAL_ENGINE_LOCKS: Dict[str, asyncio.Lock] = {}
+
+
+def _local_engine_lock(name: str) -> asyncio.Lock:
+    lock = _LOCAL_ENGINE_LOCKS.get(name)
+    if lock is None:
+        lock = asyncio.Lock()
+        _LOCAL_ENGINE_LOCKS[name] = lock
+    return lock
 
 
 def AsyncSessionLocal():
@@ -1997,6 +2007,15 @@ async def _run_dinsar_production_controller(job: SystemJobORM) -> None:
                 )
             except Exception as exc:
                 publish_error = str(exc)
+                item.status = "FAILED"
+                item.current_step = "publish_failed"
+                item.last_error = publish_error
+                await dinsar_production_service.refresh_run_counters(
+                    run,
+                    db=db,
+                    latest_message=f"Publish failed {item_label}: {publish_error}",
+                )
+                await db.commit()
                 await task_service.add_log(
                     job.task_id,
                     "WARNING",
@@ -3209,6 +3228,357 @@ async def _handle_landsar_run(job: SystemJobORM) -> None:
     )
 
 
+async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
+    payload = job.payload or {}
+    production_run_id = str(payload.get("production_run_id") or "").strip()
+    item_id = _normalize_positive_int(payload.get("item_id"))
+    if not production_run_id or not item_id:
+        raise ValueError("LANDSAR_CLUSTER_ITEM requires production_run_id and item_id.")
+
+    from ..dinsar_engines import registry
+    from ..dinsar_engines.base import RunRequest
+
+    engine = registry.get_engine("landsar")
+    if engine is None:
+        raise RuntimeError("LandSAR engine is not registered on this worker.")
+    engine_title = "LandSAR"
+    per_task_timeout = int(getattr(settings, "LANDSAR_DINSAR_TIMEOUT_SECONDS", 0) or 43200)
+
+    async with AsyncSessionLocal() as db:
+        run = await dinsar_production_service.get_run(production_run_id, db)
+        if run is None:
+            raise ValueError(f"LandSAR cluster run not found: {production_run_id}")
+
+        item = await db.get(DinsarProductionRunItemORM, int(item_id))
+        if item is None or item.run_id != run.run_id:
+            raise ValueError(f"LandSAR cluster item not found: {item_id}")
+
+        await db.refresh(item)
+        item_status = str(item.status or "").strip().upper()
+        if item_status in {"COMPLETED", "FAILED", "SKIPPED", "CANCELLED"}:
+            await dinsar_production_service.finalize_cluster_run_if_complete(run, db=db)
+            return
+
+        current_task = await task_service.get_task(job.task_id)
+        task_cancelled = bool(current_task and current_task.status == "CANCELLED")
+        if bool(run.cancel_requested) or task_cancelled:
+            run.cancel_requested = True
+            item.status = "CANCELLED"
+            item.current_step = "cancelled"
+            item.last_error = "Cancelled before worker execution."
+            await dinsar_production_service.refresh_run_counters(run, db=db, latest_message=item.last_error)
+            await db.commit()
+            await dinsar_production_service.finalize_cluster_run_if_complete(run, db=db)
+            return
+
+        if str(run.status or "").strip().upper() == "PENDING":
+            await task_service.start_task(
+                job.task_id,
+                message=f"Starting LandSAR cluster run {run.run_id} ({run.total_items} items)...",
+            )
+            await dinsar_production_service.mark_run_started(
+                run,
+                db=db,
+                message=f"LandSAR cluster started. total={run.total_items}",
+            )
+        else:
+            await task_service.update_task(
+                job.task_id,
+                status="RUNNING",
+                message=f"LandSAR cluster running. item={item.task_alias or item.task_name}",
+            )
+
+        params = run.params_json or {}
+        user_extra = dict(params.get("extra") or {})
+        timeout_seconds_raw = params.get("timeout_seconds")
+        if timeout_seconds_raw not in (None, ""):
+            per_task_timeout = int(timeout_seconds_raw)
+
+        total_items = max(1, int(run.total_items or 1))
+        item_index = max(1, int(item.order_index or 1))
+        item_label = item.task_alias or item.task_name
+        run_key = f"{build_run_key('landsar', run.profile_code, started_at=datetime.utcnow())}_{item.id}_{uuid.uuid4().hex[:6]}"
+        execution = await dinsar_production_service.begin_item_execution(
+            run=run,
+            item=item,
+            run_key=run_key,
+            db=db,
+        )
+
+        managed_run_dir = os.path.normpath(execution.output_dir)
+        landsar_work_root = str(getattr(settings, "LANDSAR_WORK_ROOT", "") or "").strip()
+        if landsar_work_root:
+            managed_native_output_dir = os.path.normpath(os.path.join(landsar_work_root, run_key, "native"))
+        else:
+            managed_native_output_dir = os.path.join(managed_run_dir, "native")
+        managed_work_dir = os.path.join(managed_native_output_dir, "workflow")
+        managed_export_dir = os.path.join(managed_native_output_dir, "export")
+        managed_orbit_output_dir = os.path.join(managed_work_dir, "orbits")
+        base_progress = min(95, 5 + int(((item_index - 1) / total_items) * 90))
+        progress_state: Dict[str, Any] = {
+            "progress": base_progress,
+            "message": f"[landsar/cluster] Running {item_index}/{total_items}: {item_label}",
+            "started_monotonic": time.monotonic(),
+        }
+        progress_queue: asyncio.Queue[Optional[Dict[str, Any]]] = asyncio.Queue()
+        loop = asyncio.get_running_loop()
+
+        def _emit_progress(event: Dict[str, Any]) -> None:
+            if not event:
+                return
+            try:
+                loop.call_soon_threadsafe(progress_queue.put_nowait, dict(event))
+            except RuntimeError:
+                return
+
+        async def _consume_progress() -> None:
+            while True:
+                event = await progress_queue.get()
+                if event is None:
+                    return
+                event_type = str(event.get("event") or "").strip().lower()
+                if event_type == "log":
+                    level = str(event.get("level") or "INFO").strip().upper()
+                    if level not in {"DEBUG", "INFO", "WARNING", "ERROR"}:
+                        level = "INFO"
+                    source = str(event.get("source") or "").strip()
+                    message = str(event.get("message") or "").strip()
+                    if message:
+                        prefix = f"[cluster {item_index}/{total_items}] {engine_title} {item_label}"
+                        if source:
+                            prefix = f"{prefix} {source}"
+                        await task_service.add_log(job.task_id, level, f"{prefix}: {message}")
+                elif event_type == "pair_started":
+                    progress_state["message"] = f"[landsar/cluster] Running {item_index}/{total_items}: {item_label}"
+                    progress_state["started_monotonic"] = time.monotonic()
+                    await task_service.add_log(
+                        job.task_id,
+                        "INFO",
+                        f"[cluster {item_index}/{total_items}] {engine_title} started {item_label}",
+                    )
+                elif event_type == "pair_finished":
+                    if bool(event.get("success")):
+                        progress_state["progress"] = min(98, 5 + int((item_index / total_items) * 90))
+                        progress_state["message"] = f"[landsar/cluster] Finished {item_index}/{total_items}: {item_label}"
+                    else:
+                        progress_state["message"] = f"[landsar/cluster] Failed {item_index}/{total_items}: {item_label}"
+
+        async def _task_keepalive() -> None:
+            while True:
+                await asyncio.sleep(30)
+                try:
+                    message = str(progress_state.get("message") or "")
+                    started_monotonic = progress_state.get("started_monotonic")
+                    if isinstance(started_monotonic, (int, float)):
+                        elapsed_seconds = max(0, int(time.monotonic() - float(started_monotonic)))
+                        message = f"{message} (elapsed={elapsed_seconds}s)"
+                    await task_service.update_task(
+                        job.task_id,
+                        status="RUNNING",
+                        progress=int(progress_state.get("progress") or base_progress),
+                        message=message,
+                    )
+                except Exception as exc:
+                    logger.warning("LandSAR cluster keepalive failed for item %s: %s", item_label, exc)
+
+        progress_task = asyncio.create_task(_consume_progress())
+        keepalive_task = asyncio.create_task(_task_keepalive())
+        task_result: Dict[str, Any] = {}
+        result = None
+        run_exception_text = ""
+
+        await task_service.add_log(
+            job.task_id,
+            "INFO",
+            f"[cluster {item_index}/{total_items}] Launching {item_label} -> {managed_run_dir}",
+        )
+        dinsar_production_service.append_run_log(
+            run.run_id,
+            f"[cluster-item-start] {item_index}/{total_items} {item_label} run_key={run_key} output={managed_run_dir}",
+        )
+
+        request = RunRequest(
+            engine_code="landsar",
+            profile=run.profile_code,
+            root_dir=str(item.source_task_dir),
+            job_id=job.job_id,
+            num_to_process=1,
+            timeout_seconds=per_task_timeout or None,
+            extra={
+                **user_extra,
+                "__managed_run_dir": managed_run_dir,
+                "__managed_native_output_dir": managed_native_output_dir,
+                "__managed_work_dir": managed_work_dir,
+                "__managed_export_dir": managed_export_dir,
+                "__managed_orbit_output_dir": managed_orbit_output_dir,
+                "__managed_run_key": run_key,
+                "__source_root_override": run.source_root,
+                "__rerun_mode": "rerun_all",
+                "__cluster_item": True,
+            },
+            progress_callback=_emit_progress,
+        )
+
+        publish_error: Optional[str] = None
+        item_error: Optional[str] = None
+        try:
+            availability = await asyncio.to_thread(engine.check_available)
+            if not availability.available:
+                raise RuntimeError(f"LandSAR engine is unavailable on worker: {availability.message}")
+
+            async with _local_engine_lock("landsar"):
+                result = await asyncio.to_thread(engine.run, request)
+
+            detail = result.detail or {} if result else {}
+            task_result = ((detail.get("task_results") or [{}])[0]) if result else {}
+            result_error = str(result.error or "").strip() if result else ""
+            result_success = bool(result.success) if result else False
+            if not result or not result_success or not bool(task_result.get("success", result_success)):
+                error_message = (
+                    str(task_result.get("error") or "").strip()
+                    or result_error
+                    or run_exception_text
+                    or str(task_result.get("stderr_tail") or "").strip()
+                    or "LandSAR cluster item failed."
+                )
+                raise RuntimeError(error_message)
+
+            run_dir = os.path.normpath(
+                str(task_result.get("run_dir") or task_result.get("output_dir") or execution.output_dir)
+            )
+            if run_dir != managed_run_dir:
+                raise RuntimeError(f"LandSAR managed run dir mismatch: expected {managed_run_dir}, got {run_dir}")
+
+            primary_file = str(task_result.get("primary_file") or "").strip()
+            source_files = [
+                str(path)
+                for path in (task_result.get("source_files") or [])
+                if str(path or "").strip()
+            ]
+            native_output_dir = str(task_result.get("native_output_dir") or managed_native_output_dir).strip() or managed_native_output_dir
+            if not primary_file or not os.path.isfile(primary_file):
+                raise RuntimeError(f"LandSAR primary output is missing: {primary_file or '<empty>'}")
+            if not source_files:
+                source_files = [primary_file]
+
+            metrics = {"result_detail": detail, "task_result": task_result, "cluster_item": True}
+            manifest_path = await asyncio.to_thread(
+                dinsar_production_service.build_execution_manifest,
+                run=run,
+                item=item,
+                execution=execution,
+                primary_file=primary_file,
+                source_files=source_files,
+                native_output_dir=native_output_dir,
+                metrics=metrics,
+            )
+            await asyncio.to_thread(
+                dinsar_production_service.write_current_pointer,
+                run=run,
+                item=item,
+                execution=execution,
+                manifest_path=manifest_path,
+                primary_file=primary_file,
+                source_files=source_files,
+                native_output_dir=native_output_dir,
+            )
+            await dinsar_production_service.mark_item_completed(
+                run=run,
+                item=item,
+                execution=execution,
+                manifest_path=manifest_path,
+                metrics=metrics,
+                db=db,
+            )
+
+            try:
+                publish_result = await result_catalog_service.publish_from_sources(db, [managed_run_dir])
+                processed_count = int(publish_result.get("processed", 0) or 0)
+                failed_count = int(publish_result.get("failed", 0) or 0)
+                if processed_count > 0:
+                    await result_catalog_service.rebuild_catalog(db, full_rebuild=True)
+                if processed_count != 1 or failed_count != 0:
+                    raise RuntimeError(f"expected processed=1 failed=0, got processed={processed_count} failed={failed_count}")
+                await task_service.add_log(
+                    job.task_id,
+                    "INFO",
+                    f"[cluster {item_index}/{total_items}] Published {item_label}",
+                )
+            except Exception as exc:
+                publish_error = str(exc)
+                await task_service.add_log(
+                    job.task_id,
+                    "WARNING",
+                    f"[cluster {item_index}/{total_items}] Result catalog publish failed for {item_label}: {publish_error}",
+                )
+
+            await task_service.add_log(
+                job.task_id,
+                "INFO",
+                f"[cluster {item_index}/{total_items}] Completed {item_label}",
+            )
+            dinsar_production_service.append_run_log(run.run_id, f"[cluster-item-ok] {item_index}/{total_items} {item_label}")
+        except Exception as exc:
+            run_exception_text = str(exc)
+            item_error = run_exception_text
+            await dinsar_production_service.mark_item_failed(
+                run=run,
+                item=item,
+                execution=execution,
+                error_message=item_error,
+                db=db,
+            )
+            await task_service.add_log(
+                job.task_id,
+                "WARNING",
+                f"[cluster {item_index}/{total_items}] Failed {item_label}: {item_error}",
+            )
+            dinsar_production_service.append_run_log(
+                run.run_id,
+                f"[cluster-item-failed] {item_index}/{total_items} {item_label}: {item_error}",
+            )
+            if task_result.get("command"):
+                await task_service.add_log(job.task_id, "INFO", f"LandSAR command [{item_label}]: {task_result.get('command')}")
+            if task_result.get("stdout_tail"):
+                await task_service.add_log(job.task_id, "INFO", f"LandSAR stdout tail [{item_label}]:\n{task_result.get('stdout_tail')}")
+            if task_result.get("stderr_tail"):
+                await task_service.add_log(job.task_id, "WARNING", f"LandSAR stderr tail [{item_label}]:\n{task_result.get('stderr_tail')}")
+        finally:
+            keepalive_task.cancel()
+            try:
+                await keepalive_task
+            except asyncio.CancelledError:
+                pass
+            await progress_queue.put(None)
+            await progress_task
+
+        await db.refresh(run)
+        await dinsar_production_service.refresh_run_counters(run, db=db)
+        done_items = int(run.completed_items or 0) + int(run.failed_items or 0) + int(run.skipped_items or 0)
+        progress = min(99, 5 + int((done_items / max(1, int(run.total_items or 1))) * 90))
+        final_status = await dinsar_production_service.finalize_cluster_run_if_complete(
+            run,
+            db=db,
+            publish_error=publish_error,
+        )
+
+        if final_status:
+            task_status = "COMPLETED" if final_status == "COMPLETED" else ("CANCELLED" if final_status == "CANCELLED" else "FAILED")
+            await task_service.update_task(
+                job.task_id,
+                status=task_status,
+                progress=100,
+                message=run.latest_message,
+            )
+        else:
+            await task_service.update_task(
+                job.task_id,
+                status="RUNNING",
+                progress=progress,
+                message=f"LandSAR cluster progress: completed={run.completed_items} failed={run.failed_items} total={run.total_items}",
+            )
+
+
 async def _handle_water_geocode(job: SystemJobORM) -> None:
     """单景 SAR 地理编码 job handler（多视 + 地理编码 + 辐射定标）。"""
     from .water_service import run_geocoding_workflow, WATER_RESULTS_DIR
@@ -5259,6 +5629,7 @@ _HANDLERS = {
     JOB_TYPE_ISCE2_RUN: _handle_isce2_run,
     JOB_TYPE_PYINT_RUN: _handle_pyint_run,
     JOB_TYPE_LANDSAR_RUN: _handle_landsar_run,
+    JOB_TYPE_LANDSAR_CLUSTER_ITEM: _handle_landsar_cluster_item,
     JOB_TYPE_WATER_GEOCODE: _handle_water_geocode,
     JOB_TYPE_SAR_SCENE_PREPROCESS: _handle_sar_scene_preprocess,
     JOB_TYPE_WATER_FLOOD: _handle_water_flood,

backend/app/services/job_queue_service.py

@@ -2,7 +2,7 @@ import asyncio
 import json
 import uuid
 from datetime import datetime, timedelta, timezone
-from typing import Any, Dict, Optional
+from typing import Any, Dict, Optional, Sequence
 
 from sqlalchemy import and_, or_, select, update, text
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -183,6 +183,7 @@ class JobQueueService:
         self,
         worker_id: str,
         lock_timeout_seconds: int = 1800,
+        allowed_job_types: Optional[Sequence[str]] = None,
         db: Optional[AsyncSession] = None,
     ) -> Optional[SystemJobORM]:
         gen_db = db is None
@@ -191,14 +192,30 @@ class JobQueueService:
 
         try:
             lock_timeout_seconds = int(lock_timeout_seconds)
+            normalized_allowed = [
+                _normalize_job_type(job_type)
+                for job_type in (allowed_job_types or [])
+                if str(job_type or "").strip()
+            ]
+            normalized_allowed = list(dict.fromkeys(normalized_allowed))
+            allowed_clause = ""
+            params: Dict[str, Any] = {"lock_timeout_seconds": lock_timeout_seconds}
+            if normalized_allowed:
+                allowed_params = []
+                for index, job_type in enumerate(normalized_allowed):
+                    key = f"allowed_job_type_{index}"
+                    allowed_params.append(f":{key}")
+                    params[key] = job_type
+                allowed_clause = f"AND job_type IN ({', '.join(allowed_params)})"
             async with db.begin():
                 result = await db.execute(
                     text(
-                        """
+                        f"""
                         SELECT id
                         FROM system_jobs
                         WHERE status IN ('READY', 'RETRY')
                           AND (next_run_at IS NULL OR next_run_at <= NOW())
+                          {allowed_clause}
                           AND (
                                 locked_by IS NULL OR
                                 locked_at IS NULL OR
@@ -209,7 +226,7 @@ class JobQueueService:
                         LIMIT 1
                         """
                     ),
-                    {"lock_timeout_seconds": lock_timeout_seconds},
+                    params,
                 )
                 row = result.first()
                 if not row:

backend/app/services/job_worker.py

@@ -3,7 +3,7 @@ import os
 import socket
 import uuid
 import time
-from typing import Set
+from typing import Optional, Set
 
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy import func
@@ -26,6 +26,15 @@ IDL_JOB_TYPES = {
 }
 
 
+def _parse_allowed_job_types(raw_value: str) -> Optional[Set[str]]:
+    values = {
+        part.strip().upper()
+        for part in str(raw_value or "").replace(";", ",").split(",")
+        if part.strip()
+    }
+    return values or None
+
+
 def _default_worker_id() -> str:
     host = socket.gethostname()
     pid = os.getpid()
@@ -129,6 +138,7 @@ async def run_worker_loop(
     concurrency = max(1, int(concurrency))
     sem = asyncio.Semaphore(concurrency)
     active: Set[asyncio.Task] = set()
+    allowed_job_types = _parse_allowed_job_types(getattr(settings, "JOB_WORKER_ALLOWED_TYPES", ""))
     job_heartbeat_interval = float(settings.JOB_WORKER_JOB_HEARTBEAT_INTERVAL)
     stale_recover_interval = float(settings.JOB_WORKER_STALE_RECOVER_INTERVAL)
     stale_running_seconds = int(settings.JOB_WORKER_STALE_RUNNING_SECONDS)
@@ -187,7 +197,10 @@ async def run_worker_loop(
                 print(f"[WARN] worker poll: {exc}")
 
         if len(active) < concurrency:
-            job = await job_queue_service.claim_next_job(worker_id)
+            job = await job_queue_service.claim_next_job(
+                worker_id,
+                allowed_job_types=allowed_job_types,
+            )
             if job:
                 task = asyncio.create_task(_wrap(job))
                 active.add(task)

backend/app/utils.py

@@ -95,6 +95,68 @@ def _ordered_closed_polygon_from_corner_details(corner_details: Dict[str, Dict[s
 
     return _ordered_closed_polygon([(value["lon"], value["lat"]) for value in (corner_details or {}).values()])
 
+
+def build_corner_pixel_mapping(corner_details: Dict[str, Dict[str, Any]]) -> Optional[Dict[str, Any]]:
+    if len(corner_details or {}) < 4:
+        return None
+
+    entries: List[Tuple[str, Dict[str, Any]]] = []
+    ref_rows = []
+    ref_cols = []
+    for name, info in (corner_details or {}).items():
+        if info.get("lon") is None or info.get("lat") is None:
+            return None
+        if info.get("ref_row") is None or info.get("ref_col") is None:
+            return None
+        try:
+            ref_row = int(float(info["ref_row"]))
+            ref_col = int(float(info["ref_col"]))
+        except (TypeError, ValueError):
+            return None
+        normalized = dict(info)
+        normalized["ref_row"] = ref_row
+        normalized["ref_col"] = ref_col
+        entries.append((str(name), normalized))
+        ref_rows.append(ref_row)
+        ref_cols.append(ref_col)
+
+    min_row, max_row = min(ref_rows), max(ref_rows)
+    min_col, max_col = min(ref_cols), max(ref_cols)
+    remaining = entries[:]
+
+    def pick(target_row: int, target_col: int) -> Optional[Tuple[float, float]]:
+        if not remaining:
+            return None
+        ranked = []
+        for index, (name, info) in enumerate(remaining):
+            row = int(info["ref_row"])
+            col = int(info["ref_col"])
+            score = abs(row - target_row) + abs(col - target_col)
+            ranked.append((score, abs(row - target_row), abs(col - target_col), name, index))
+        ranked.sort()
+        chosen_index = ranked[0][4]
+        _, chosen = remaining.pop(chosen_index)
+        try:
+            return float(chosen["lon"]), float(chosen["lat"])
+        except (TypeError, ValueError):
+            return None
+
+    top_left = pick(min_row, min_col)
+    top_right = pick(min_row, max_col)
+    bottom_left = pick(max_row, min_col)
+    bottom_right = pick(max_row, max_col)
+
+    if not all([top_left, top_right, bottom_left, bottom_right]):
+        return None
+
+    return {
+        "top_left": [top_left[0], top_left[1]],
+        "top_right": [top_right[0], top_right[1]],
+        "bottom_left": [bottom_left[0], bottom_left[1]],
+        "bottom_right": [bottom_right[0], bottom_right[1]],
+        "source": "xml_ref_row_col",
+    }
+
 # --- Sentinel-1 (S1A/S1B/S1C) Parsers ---
 
 def _radar_meta_base() -> Dict[str, Any]:
@@ -446,54 +508,6 @@ def parse_xml_metadata(
             except (TypeError, ValueError):
                 return None
 
-        def _build_corner_pixel_mapping(corner_details: Dict[str, Dict[str, Any]]) -> Optional[Dict[str, Any]]:
-            if len(corner_details) < 4:
-                return None
-
-            ref_rows = []
-            ref_cols = []
-            for info in corner_details.values():
-                if info.get("ref_row") is None or info.get("ref_col") is None:
-                    return None
-                ref_rows.append(int(info["ref_row"]))
-                ref_cols.append(int(info["ref_col"]))
-
-            min_row, max_row = min(ref_rows), max(ref_rows)
-            min_col, max_col = min(ref_cols), max(ref_cols)
-            remaining = set(corner_details.keys())
-
-            def pick(target_row: int, target_col: int) -> Optional[Tuple[float, float]]:
-                if not remaining:
-                    return None
-                ranked = []
-                for name in remaining:
-                    info = corner_details[name]
-                    row = int(info["ref_row"])
-                    col = int(info["ref_col"])
-                    score = abs(row - target_row) + abs(col - target_col)
-                    ranked.append((score, abs(row - target_row), abs(col - target_col), name))
-                ranked.sort()
-                chosen_name = ranked[0][3]
-                remaining.remove(chosen_name)
-                chosen = corner_details[chosen_name]
-                return float(chosen["lon"]), float(chosen["lat"])
-
-            top_left = pick(min_row, min_col)
-            top_right = pick(min_row, max_col)
-            bottom_left = pick(max_row, min_col)
-            bottom_right = pick(max_row, max_col)
-
-            if not all([top_left, top_right, bottom_left, bottom_right]):
-                return None
-
-            return {
-                "top_left": [top_left[0], top_left[1]],
-                "top_right": [top_right[0], top_right[1]],
-                "bottom_left": [bottom_left[0], bottom_left[1]],
-                "bottom_right": [bottom_right[0], bottom_right[1]],
-                "source": "xml_ref_row_col",
-            }
-
         corners = ['bottomLeft', 'bottomRight', 'topRight', 'topLeft']
         polygon = []
 
@@ -652,7 +666,7 @@ def parse_xml_metadata(
             if not ordered_polygon:
                 return None, None
             polygon = ordered_polygon
-            corner_pixel_mapping = _build_corner_pixel_mapping(corner_details)
+            corner_pixel_mapping = build_corner_pixel_mapping(corner_details)
             meta = {
                 "orbit_direction": orbit_direction,
                 "imaging_mode": imaging_mode,