Harden LandSAR cluster handoff

backend/app/config.py

@@ -312,6 +312,7 @@ class Settings(BaseSettings):
     WSL_BROKER_JOB_ROOT: str = ""
     ISCE2_RUNTIME_ID: str = ""
     PYINT_RUNTIME_ID: str = ""
+    LANDSAR_RUNTIME_ID: str = ""
 
     ISCE2_ENABLED: bool = False
     ISCE2_WSL_DISTRO: str = "Ubuntu-24.04"
@@ -678,6 +679,8 @@ class Settings(BaseSettings):
             object.__setattr__(self, "ISCE2_RUNTIME_ID", "isce2_runtime_v1")
         if not self.PYINT_RUNTIME_ID:
             object.__setattr__(self, "PYINT_RUNTIME_ID", "gamma_pyint_runtime_v1")
+        if not self.LANDSAR_RUNTIME_ID:
+            object.__setattr__(self, "LANDSAR_RUNTIME_ID", "landsar_runtime_v1")
         if not self.PYINT_WSL_DISTRO:
             object.__setattr__(self, "PYINT_WSL_DISTRO", self.WSL_DISTRO or self.ISCE2_WSL_DISTRO)
         if not self.PYINT_WSL_PYTHON:

backend/app/dinsar_engines/landsar_engine.py

@@ -204,6 +204,7 @@ _SYSTEM_EXTRA_KEYS = {
     "__managed_orbit_output_dir",
     "__managed_run_key",
     "__source_root_override",
+    "__source_task_dir_override",
     "__rerun_mode",
     "__validated_task_count",
     "__validated_mode",
@@ -2318,7 +2319,7 @@ class LandsarEngine(DinsarEngine):
             "engine_code": self.engine_code,
             "profile_code": request.profile,
             "source_root": _norm_path(request.extra.get("__source_root_override") or request.root_dir),
-            "task_dir": _norm_path(task_dir),
+            "task_dir": _norm_path(request.extra.get("__source_task_dir_override") or task_dir),
             "work_dir": landsar_output_dir,
             "output_dir": _norm_path(run_dir),
             "native_output_dir": _norm_path(native_output_dir),

backend/app/routers/cluster.py

@@ -13,6 +13,8 @@ import tempfile
 import zipfile
 from typing import Optional
 
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
 from fastapi import (
     APIRouter,
     Depends,
@@ -24,11 +26,14 @@ from fastapi import (
 )
 from fastapi.responses import FileResponse
 from starlette.background import BackgroundTask
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from ..config import settings
 from ..database import get_db
-from ..models.orm import DinsarProductionRunItemORM
+from ..models.orm import (
+    DinsarProductionExecutionORM,
+    DinsarProductionRunItemORM,
+    DinsarProductionRunORM,
+)
 from ..services.cluster_transport import safe_extract_zip
 
 router = APIRouter()
@@ -263,17 +268,84 @@ async def upload_cluster_result(
 
     # ----- catalog registration ------------------------------------------------
     from ..services.result_catalog_service import result_catalog_service as rcs
+    from ..services.dinsar_production_service import dinsar_production_service
 
     try:
         publish_result = await rcs.publish_from_sources(db, [extract_dir])
         processed = int(publish_result.get("processed", 0) or 0)
+        failed = int(publish_result.get("failed", 0) or 0)
         if processed > 0:
             await rcs.rebuild_catalog(db, full_rebuild=True)
+        if processed != 1 or failed != 0:
+            raise RuntimeError(
+                f"expected processed=1 failed=0, got processed={processed} failed={failed}"
+            )
+
+        details = publish_result.get("details") if isinstance(publish_result, dict) else []
+        detail = next(
+            (
+                item_detail
+                for item_detail in (details or [])
+                if str(item_detail.get("run_key") or "").strip() == normalized_run_key
+            ),
+            (details or [{}])[0] if details else {},
+        )
+        execution_manifest_path = str(
+            detail.get("execution_manifest_path")
+            or os.path.join(extract_dir, "execution_manifest.json")
+        )
+        current_pointer_path = str(detail.get("current_pointer_path") or "")
+        if not os.path.isfile(execution_manifest_path):
+            raise RuntimeError(
+                f"execution manifest was not created: {execution_manifest_path}"
+            )
+        if not current_pointer_path or not os.path.isfile(current_pointer_path):
+            raise RuntimeError(
+                f"current pointer was not created: {current_pointer_path or '<empty>'}"
+            )
+
+        run = await db.get(DinsarProductionRunORM, str(item.run_id or ""))
+        if run is None:
+            raise RuntimeError(f"Cluster run not found for item {item_id}: {item.run_id}")
+        execution_result = await db.execute(
+            select(DinsarProductionExecutionORM).where(
+                DinsarProductionExecutionORM.item_id == item.id,
+                DinsarProductionExecutionORM.run_key == normalized_run_key,
+            )
+        )
+        execution = execution_result.scalar_one_or_none()
+        if execution is None:
+            raise RuntimeError(
+                f"Cluster execution not found for item={item.id} run_key={normalized_run_key}"
+            )
+        execution.output_dir = extract_dir
+        item.latest_output_dir = extract_dir
+        await dinsar_production_service.mark_item_completed(
+            run=run,
+            item=item,
+            execution=execution,
+            manifest_path=execution_manifest_path,
+            metrics={
+                "cluster_upload": True,
+                "catalog_processed": processed,
+                "catalog_failed": failed,
+                "current_pointer_path": current_pointer_path,
+            },
+            db=db,
+        )
+        final_status = await dinsar_production_service.finalize_cluster_run_if_complete(
+            run,
+            db=db,
+        )
         return {
             "registered": processed > 0,
+            "completed": True,
+            "final_status": final_status,
             "processed": processed,
-            "failed": int(publish_result.get("failed", 0) or 0),
+            "failed": failed,
             "catalog_path": extract_dir,
+            "execution_manifest_path": execution_manifest_path,
+            "current_pointer_path": current_pointer_path,
         }
     except Exception as exc:
         raise HTTPException(

backend/app/services/cluster_transport.py

@@ -67,6 +67,16 @@ def safe_extract_zip(zf: zipfile.ZipFile, target_dir: str) -> None:
     zf.extractall(target_root)
 
 
+def _zip_directory_contents(source_dir: str, zip_path: str) -> None:
+    source_root = os.path.abspath(source_dir)
+    with zipfile.ZipFile(zip_path, "w", compression=zipfile.ZIP_DEFLATED) as zf:
+        for current, _, files in os.walk(source_root):
+            for name in files:
+                path = os.path.join(current, name)
+                arcname = os.path.relpath(path, source_root)
+                zf.write(path, arcname)
+
+
 def _resolve_cluster_server_url() -> str:
     """Return the main-server HTTP base URL for cluster data transport.
 
@@ -99,23 +109,52 @@ def _is_remote_worker() -> bool:
     return host not in {"", "127.0.0.1", "localhost"}
 
 
+def resolve_cluster_local_task_dir(item: DinsarProductionRunItemORM) -> str:
+    """Return the worker-local Task_* directory for a cluster item."""
+    source_task_dir = os.path.normpath(str(item.source_task_dir or ""))
+    worker_root = _read_cluster_env("CLUSTER_WORKER_TASK_ROOT")
+    if not worker_root:
+        return source_task_dir
+    task_name = os.path.basename(source_task_dir) or f"Task_item_{item.id}"
+    return os.path.normpath(
+        os.path.join(worker_root, f"item_{item.id}", task_name)
+    )
+
+
+def resolve_cluster_local_run_dir(
+    item: DinsarProductionRunItemORM,
+    run_key: str,
+) -> str:
+    """Return the worker-local managed result run directory."""
+    worker_root = _read_cluster_env("CLUSTER_WORKER_RESULT_ROOT")
+    if not worker_root:
+        return os.path.join(str(item.results_root_dir or ""), "runs", run_key)
+    pair_fragment = str(item.pair_key or f"item_{item.id}").strip() or f"item_{item.id}"
+    safe_pair = "".join(
+        ch if ch.isalnum() or ch in "._-" else "_"
+        for ch in pair_fragment
+    ).strip("._") or f"item_{item.id}"
+    return os.path.normpath(os.path.join(worker_root, safe_pair, "runs", run_key))
+
+
 async def materialize_cluster_input(
     item: DinsarProductionRunItemORM,
-    source_task_dir: str,
+    local_task_dir: str,
     task_id: str,
 ) -> None:
     """Download and extract the input data for a cluster item.
 
     Calls ``GET /api/cluster/input-package/{item_id}`` on the main
     server, retrieves a zip containing the Task_Pool directory tree,
-    and extracts it so that *source_task_dir* exists locally.
+    and extracts it so that *local_task_dir* exists locally.
     """
     from .task_service import task_service
 
     server_url = _resolve_cluster_server_url()
     download_url = f"{server_url}/api/cluster/input-package/{item.id}"
-    parent_dir = os.path.dirname(source_task_dir)
-    task_name = os.path.basename(source_task_dir)
+    parent_dir = os.path.dirname(local_task_dir)
+    task_name = os.path.basename(local_task_dir)
+    package_task_name = os.path.basename(str(item.source_task_dir or "")) or task_name
 
     await task_service.add_log(
         task_id,
@@ -124,8 +163,8 @@ async def materialize_cluster_input(
     )
 
     tmp_zip = os.path.join(
-        tempfile.gettempdir(),
-        f"cluster_input_{item.id}_{task_name}.zip",
+        tempfile.mkdtemp(prefix=f"cluster_input_{item.id}_"),
+        f"{package_task_name}.zip",
     )
     try:
         req = urllib.request.Request(
@@ -138,18 +177,28 @@ async def materialize_cluster_input(
                 shutil.copyfileobj(resp, fh, 8 * 1024 * 1024)
 
         os.makedirs(parent_dir, exist_ok=True)
+        if os.path.isdir(local_task_dir):
+            shutil.rmtree(local_task_dir)
         with zipfile.ZipFile(tmp_zip, "r") as zf:
             safe_extract_zip(zf, parent_dir)
 
-        if not os.path.isdir(source_task_dir):
+        extracted_task_dir = os.path.join(parent_dir, package_task_name)
+        if (
+            os.path.isdir(extracted_task_dir)
+            and os.path.normcase(os.path.abspath(extracted_task_dir))
+            != os.path.normcase(os.path.abspath(local_task_dir))
+        ):
+            os.replace(extracted_task_dir, local_task_dir)
+
+        if not os.path.isdir(local_task_dir):
             raise RuntimeError(
-                f"Extraction did not create expected directory: {source_task_dir}"
+                f"Extraction did not create expected directory: {local_task_dir}"
             )
 
         await task_service.add_log(
             task_id,
             "INFO",
-            f"[cluster] Input data ready: {source_task_dir}",
+            f"[cluster] Input data ready: {local_task_dir}",
         )
     except urllib.error.HTTPError as exc:
         body_text = ""
@@ -167,7 +216,9 @@ async def materialize_cluster_input(
         ) from exc
     finally:
         try:
-            os.unlink(tmp_zip)
+            tmp_root = os.path.dirname(tmp_zip)
+            if os.path.isdir(tmp_root):
+                shutil.rmtree(tmp_root)
         except Exception:
             pass
 
@@ -194,19 +245,10 @@ async def upload_cluster_result(
         "[cluster] Packaging results for upload ...",
     )
 
-    run_dir_name = os.path.basename(os.path.normpath(managed_run_dir))
-    tmp_zip = os.path.join(
-        tempfile.gettempdir(),
-        f"cluster_result_{item.id}.zip",
-    )
+    tmp_root = tempfile.mkdtemp(prefix=f"cluster_result_{item.id}_")
+    tmp_zip = os.path.join(tmp_root, "result.zip")
     try:
-        parent = os.path.dirname(managed_run_dir)
-        shutil.make_archive(
-            tmp_zip.replace(".zip", ""),
-            "zip",
-            root_dir=parent,
-            base_dir=run_dir_name,
-        )
+        _zip_directory_contents(managed_run_dir, tmp_zip)
 
         await task_service.add_log(
             task_id,
@@ -284,7 +326,7 @@ async def upload_cluster_result(
         ) from exc
     finally:
         try:
-            if os.path.isfile(tmp_zip):
-                os.unlink(tmp_zip)
+            if os.path.isdir(tmp_root):
+                shutil.rmtree(tmp_root)
         except Exception:
             pass

backend/app/services/dinsar_completion_files.py

@@ -71,9 +71,11 @@ def _current_pointer_path(results_root_dir: str, *, engine_code: str, profile_co
 def _runtime_id_for_engine(engine_code: str) -> Optional[str]:
     normalized = str(engine_code or "").strip().lower()
     if normalized == "isce2":
-        return settings.ISCE2_RUNTIME_ID or None
+        return getattr(settings, "ISCE2_RUNTIME_ID", "") or None
+    if normalized == "landsar":
+        return getattr(settings, "LANDSAR_RUNTIME_ID", "") or None
     if normalized in {"pyint", "gamma"}:
-        return settings.PYINT_RUNTIME_ID or None
+        return getattr(settings, "PYINT_RUNTIME_ID", "") or None
     return None
 
 
@@ -137,9 +139,17 @@ def repair_managed_completion_files(
     task_name = _first_text(payload.get("task_name"), payload.get("task_alias"), run_key)
     task_alias = _first_text(payload.get("task_alias"), payload.get("task_name"), task_name)
     output_dir = _normalize_path(payload.get("output_dir") or normalized_run_dir)
+    if normalized_primary.startswith(normalized_run_dir + os.sep):
+        output_dir = normalized_run_dir
     native_output_dir = _normalize_path(
         payload.get("native_output_dir") or os.path.join(normalized_run_dir, "native")
     )
+    if not native_output_dir.startswith(normalized_run_dir + os.sep):
+        local_native_dir = os.path.join(normalized_run_dir, "native")
+        if os.path.isdir(local_native_dir):
+            native_output_dir = _normalize_path(local_native_dir)
+        else:
+            native_output_dir = output_dir
     results_root_dir = _infer_results_root_dir(normalized_run_dir, pair_key)
     publish_root_dir = _normalize_path(os.path.dirname(results_root_dir))
 

backend/app/services/dinsar_production_service.py

@@ -1184,9 +1184,12 @@ class DinsarProductionService:
         run: DinsarProductionRunORM,
         item: DinsarProductionRunItemORM,
         run_key: str,
+        output_dir_override: Optional[str] = None,
         db: AsyncSession,
     ) -> DinsarProductionExecutionORM:
-        output_dir = _execution_dir(item, run_key)
+        output_dir = os.path.normpath(
+            os.path.abspath(output_dir_override)
+        ) if output_dir_override else _execution_dir(item, run_key)
         _ensure_dir(output_dir)
         execution = DinsarProductionExecutionORM(
             execution_id=run_key,

backend/app/services/job_handlers.py

@@ -26,6 +26,8 @@ from .asset_inventory_service import asset_inventory_service
 from .cluster_transport import (
     _is_remote_worker,
     materialize_cluster_input,
+    resolve_cluster_local_run_dir,
+    resolve_cluster_local_task_dir,
     upload_cluster_result,
 )
 from .dinsar_compat_service import dinsar_compat_service
@@ -164,15 +166,30 @@ def _normalize_positive_int(value: Any) -> Optional[int]:
 def _cluster_source_task_dir_ready(path: str) -> bool:
     if not path or not os.path.isdir(path):
         return False
+
+    def _has_landsar_source_file(directory: str) -> bool:
+        try:
+            with os.scandir(directory) as entries:
+                for entry in entries:
+                    if not entry.is_file():
+                        continue
+                    name = entry.name.lower()
+                    if name.startswith("lt1") and name.endswith((".xml", ".tif", ".tiff")):
+                        return True
+        except OSError:
+            return False
+        return False
+
+    input_data_dir = os.path.join(path, "Input_Data")
+    if os.path.isdir(input_data_dir) and _has_landsar_source_file(input_data_dir):
+        return True
+
+    if not os.path.isfile(os.path.join(path, ".dinsar_pair.json")):
+        return False
+
     for child_name in ("master", "slave"):
         child_dir = os.path.join(path, child_name)
-        if not os.path.isdir(child_dir):
-            return False
-        try:
-            with os.scandir(child_dir) as entries:
-                if not any(entries):
-                    return False
-        except OSError:
+        if not os.path.isdir(child_dir) or not _has_landsar_source_file(child_dir):
             return False
     return True
 
@@ -3275,14 +3292,14 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
             raise ValueError(f"LandSAR cluster item not found: {item_id}")
 
         await db.refresh(item)
-        source_task_dir = os.path.normpath(str(item.source_task_dir or ""))
-        if source_task_dir and not _cluster_source_task_dir_ready(source_task_dir):
-            await materialize_cluster_input(item, source_task_dir, job.task_id)
-
         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
+        source_task_dir = os.path.normpath(str(item.source_task_dir or ""))
+        local_task_dir = resolve_cluster_local_task_dir(item)
+        if local_task_dir and not _cluster_source_task_dir_ready(local_task_dir):
+            await materialize_cluster_input(item, local_task_dir, job.task_id)
 
         current_task = await task_service.get_task(job.task_id)
         task_cancelled = bool(current_task and current_task.status == "CANCELLED")
@@ -3323,10 +3340,12 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
         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]}"
+        local_run_dir = os.path.normpath(resolve_cluster_local_run_dir(item, run_key))
         execution = await dinsar_production_service.begin_item_execution(
             run=run,
             item=item,
             run_key=run_key,
+            output_dir_override=local_run_dir,
             db=db,
         )
 
@@ -3425,7 +3444,7 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
         request = RunRequest(
             engine_code="landsar",
             profile=run.profile_code,
-            root_dir=str(item.source_task_dir),
+            root_dir=local_task_dir,
             job_id=job.job_id,
             num_to_process=1,
             timeout_seconds=per_task_timeout or None,
@@ -3438,6 +3457,7 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
                 "__managed_orbit_output_dir": managed_orbit_output_dir,
                 "__managed_run_key": run_key,
                 "__source_root_override": run.source_root,
+                "__source_task_dir_override": source_task_dir,
                 "__rerun_mode": "rerun_all",
                 "__cluster_item": True,
             },
@@ -3497,24 +3517,17 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
                 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,
-            )
+            if not _is_remote_worker():
+                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,
+                )
 
             # ---- Post-flight: upload or local publish ----
             if _is_remote_worker():
@@ -3544,6 +3557,14 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
                         "WARNING",
                         f"[cluster {item_index}/{total_items}] Result catalog publish failed for {item_label}: {publish_error}",
                     )
+                await dinsar_production_service.mark_item_completed(
+                    run=run,
+                    item=item,
+                    execution=execution,
+                    manifest_path=manifest_path,
+                    metrics=metrics,
+                    db=db,
+                )
 
             await task_service.add_log(
                 job.task_id,
@@ -3554,28 +3575,46 @@ async def _handle_landsar_cluster_item(job: SystemJobORM) -> None:
         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')}")
+            remote_completed = False
+            if _is_remote_worker():
+                try:
+                    await db.refresh(item)
+                    remote_completed = str(item.status or "").strip().upper() == "COMPLETED"
+                except Exception:
+                    remote_completed = False
+            if remote_completed:
+                await task_service.add_log(
+                    job.task_id,
+                    "INFO",
+                    f"[cluster {item_index}/{total_items}] Main server already completed {item_label}; keeping completed state after local error: {item_error}",
+                )
+                dinsar_production_service.append_run_log(
+                    run.run_id,
+                    f"[cluster-item-ok-after-upload] {item_index}/{total_items} {item_label}: {item_error}",
+                )
+            else:
+                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:

backend/app/services/result_catalog_service.py

@@ -142,9 +142,11 @@ def _coerce_optional_int(value: Any) -> Optional[int]:
 def _runtime_id_for_engine(engine_code: Optional[str]) -> Optional[str]:
     normalized = str(engine_code or "").strip().lower()
     if normalized == "isce2":
-        return settings.ISCE2_RUNTIME_ID or None
+        return getattr(settings, "ISCE2_RUNTIME_ID", "") or None
+    if normalized == "landsar":
+        return getattr(settings, "LANDSAR_RUNTIME_ID", "") or None
     if normalized in {"pyint", "gamma"}:
-        return settings.PYINT_RUNTIME_ID or None
+        return getattr(settings, "PYINT_RUNTIME_ID", "") or None
     return None
 
 
@@ -822,6 +824,13 @@ class ResultCatalogService:
                 package_dir = _ensure_directory(os.path.join(target_root, pair_key, "runs", run_key))
                 source_dir = _normalize_path(candidate["source_dir"])
                 in_place_source = _is_path_within(package_dir, source_dir)
+                if candidate_meta["engine_code"] in {"isce2", "landsar"} and in_place_source:
+                    candidate_meta = dict(candidate_meta)
+                    candidate_meta["output_dir"] = package_dir
+                    native_dir = os.path.join(package_dir, RUN_NATIVE_DIRNAME)
+                    candidate_meta["native_output_dir"] = (
+                        native_dir if os.path.isdir(native_dir) else package_dir
+                    )
                 task_item = await self._lookup_task_item(
                     db,
                     pair_key=pair_key,
@@ -994,7 +1003,7 @@ class ResultCatalogService:
                 manifest_path = os.path.join(package_dir, "manifest.json")
                 with open(manifest_path, "w", encoding="utf-8") as fp:
                     json.dump(manifest, fp, ensure_ascii=False, indent=2)
-                if candidate["engine_code"] == "isce2" and in_place_source:
+                if candidate["engine_code"] in {"isce2", "landsar"} and in_place_source:
                     try:
                         completion_files_result = repair_managed_completion_files(
                             package_dir,

backend/tests/test_cluster_transport.py

@@ -0,0 +1,130 @@
+import json
+import os
+import tempfile
+import unittest
+import zipfile
+from types import SimpleNamespace
+from unittest import mock
+
+from backend.app.services.cluster_transport import (
+    _zip_directory_contents,
+    resolve_cluster_local_run_dir,
+    resolve_cluster_local_task_dir,
+    safe_extract_zip,
+)
+from backend.app.services.dinsar_completion_files import repair_managed_completion_files
+from backend.app.services.dinsar_naming import RUN_META_FILENAME
+
+
+class ClusterTransportTests(unittest.TestCase):
+    def test_zip_directory_contents_excludes_top_level_run_dir(self):
+        with tempfile.TemporaryDirectory() as root:
+            run_dir = os.path.join(root, "run_abc")
+            os.makedirs(os.path.join(run_dir, "assets", "disp"))
+            os.makedirs(os.path.join(run_dir, "native"))
+            with open(os.path.join(run_dir, RUN_META_FILENAME), "w", encoding="utf-8") as fp:
+                json.dump({"run_key": "run_abc"}, fp)
+            with open(os.path.join(run_dir, "assets", "disp", "disp.tif"), "wb") as fp:
+                fp.write(b"disp")
+            with open(os.path.join(run_dir, "native", "raw.txt"), "w", encoding="utf-8") as fp:
+                fp.write("raw")
+
+            zip_path = os.path.join(root, "result.zip")
+            _zip_directory_contents(run_dir, zip_path)
+
+            with zipfile.ZipFile(zip_path, "r") as zf:
+                names = set(zf.namelist())
+
+            self.assertIn(RUN_META_FILENAME, names)
+            self.assertIn("assets/disp/disp.tif", names)
+            self.assertIn("native/raw.txt", names)
+            self.assertFalse(any(name.startswith("run_abc/") for name in names))
+
+    def test_safe_extract_zip_rejects_path_escape(self):
+        with tempfile.TemporaryDirectory() as root:
+            zip_path = os.path.join(root, "unsafe.zip")
+            with zipfile.ZipFile(zip_path, "w") as zf:
+                zf.writestr("../escape.txt", "bad")
+
+            with zipfile.ZipFile(zip_path, "r") as zf:
+                with self.assertRaises(ValueError):
+                    safe_extract_zip(zf, os.path.join(root, "extract"))
+
+    def test_worker_local_paths_are_optional_overrides(self):
+        item = SimpleNamespace(
+            id=7,
+            source_task_dir=r"D:\Task_Pool\DInSAR\Task_20260101_20260113",
+            results_root_dir=r"D:\production_results\dinsar\pair_a",
+            pair_key="lt1/pair:a",
+        )
+
+        with mock.patch.dict(os.environ, {}, clear=True):
+            self.assertEqual(
+                os.path.normpath(resolve_cluster_local_task_dir(item)),
+                os.path.normpath(item.source_task_dir),
+            )
+            self.assertEqual(
+                os.path.normpath(resolve_cluster_local_run_dir(item, "run_1")),
+                os.path.normpath(r"D:\production_results\dinsar\pair_a\runs\run_1"),
+            )
+
+        with mock.patch.dict(
+            os.environ,
+            {
+                "CLUSTER_WORKER_TASK_ROOT": r"E:\cluster_tasks",
+                "CLUSTER_WORKER_RESULT_ROOT": r"E:\cluster_results",
+            },
+            clear=True,
+        ):
+            self.assertEqual(
+                os.path.normpath(resolve_cluster_local_task_dir(item)),
+                os.path.normpath(r"E:\cluster_tasks\item_7\Task_20260101_20260113"),
+            )
+            self.assertEqual(
+                os.path.normpath(resolve_cluster_local_run_dir(item, "run_1")),
+                os.path.normpath(r"E:\cluster_results\lt1_pair_a\runs\run_1"),
+            )
+
+    def test_repair_completion_files_reanchors_uploaded_landsar_run(self):
+        with tempfile.TemporaryDirectory() as root:
+            run_dir = os.path.join(root, "pair_a", "runs", "run_abc")
+            disp_dir = os.path.join(run_dir, "assets", "disp")
+            native_dir = os.path.join(run_dir, "native")
+            os.makedirs(disp_dir)
+            os.makedirs(native_dir)
+            primary_file = os.path.join(disp_dir, "disp.tif")
+            with open(primary_file, "wb") as fp:
+                fp.write(b"disp")
+            with open(os.path.join(run_dir, RUN_META_FILENAME), "w", encoding="utf-8") as fp:
+                json.dump(
+                    {
+                        "run_key": "run_abc",
+                        "pair_key": "pair_a",
+                        "engine_code": "landsar",
+                        "profile_code": "lt1_dinsar",
+                        "task_name": "Task_20260101_20260113",
+                        "task_alias": "Task_20260101_20260113",
+                        "output_dir": r"E:\worker_results\pair_a\runs\run_abc",
+                        "native_output_dir": r"E:\worker_results\pair_a\runs\run_abc\native",
+                        "primary_file": primary_file,
+                        "source_files": [primary_file],
+                    },
+                    fp,
+                )
+
+            result = repair_managed_completion_files(
+                run_dir,
+                primary_file=primary_file,
+                source_files=[primary_file],
+            )
+
+            self.assertTrue(os.path.isfile(result["execution_manifest_path"]))
+            self.assertTrue(os.path.isfile(result["current_pointer_path"]))
+            with open(result["execution_manifest_path"], "r", encoding="utf-8") as fp:
+                manifest = json.load(fp)
+            self.assertEqual(os.path.normpath(manifest["output_dir"]), os.path.normpath(run_dir))
+            self.assertEqual(os.path.normpath(manifest["native_output_dir"]), os.path.normpath(native_dir))
+
+
+if __name__ == "__main__":
+    unittest.main()

docs/LANDSAR_CLUSTER_DATA_TRANSPORT_DESIGN_20260625.md

@@ -14,7 +14,7 @@
  ## 设计目标
 
  - 不依赖 Windows 文件共享（SMB）、映射盘符、UNC 路径
- - 不要求主服务器和 worker 共用盘符或路径结构
+- 不要求主服务器和 worker 共用盘符或路径结构；worker 可通过 `CLUSTER_WORKER_TASK_ROOT`、`CLUSTER_WORKER_RESULT_ROOT` 使用本机任务/结果缓存根
  - Worker 节点可以动态增减，配置简单
  - 复用现有 `SOURCE_PRODUCT_DIRS` 压缩包源池和 `TASK_POOL_ROOT` 体系
  - 传输失败利用队列系统自带的重试机制
@@ -69,13 +69,13 @@
 
  **Worker 端流程**：
 
- 1. 读取 `item.source_task_dir`，检查本地目录是否存在且包含 `master/`、`slave/`、`pair_metadata.json`
+ 1. 读取 `item.source_task_dir`，解析 worker 本地输入目录。未配置 `CLUSTER_WORKER_TASK_ROOT` 时沿用原路径；已配置时使用 `CLUSTER_WORKER_TASK_ROOT\item_<item_id>\Task_*`
  2. 若无 → 调用 `GET /api/cluster/input-package/{item_id}`
  3. Worker 请求头携带 `X-Cluster-Token`，主服务器校验 `CLUSTER_SHARED_TOKEN`
  4. 主服务器读取 `item.source_task_dir` 指向的现有 Task_Pool 目录
  5. 将该 Task_Pool 目录打成 zip 流式返回给 worker
- 6. Worker 解包到 `item.source_task_dir`（保持与主服务器一致的路径结构）
- 7. Worker 对 zip 成员路径做目录逃逸校验，并校验解包后的 `master/`、`slave/` 目录可用
+ 6. Worker 解包到本地输入目录
+ 7. Worker 对 zip 成员路径做目录逃逸校验，并校验解包后的 `Input_Data` 或 `master/`、`slave/` 目录可用
 
  **主服务器新增 API**：
 
@@ -85,13 +85,13 @@
    Response: application/zip (streaming)
    内容结构:
      Task_YYYYMMDD_YYYYMMDD/
+       .dinsar_pair.json
        master/
          <源文件...>
        slave/
          <源文件...>
        orbit/
          <精轨文件...>
-       pair_metadata.json
  ```
 
  **当前实现边界**：主服务器不在传输接口里重新从 `SOURCE_PRODUCT_DIRS` 解包源压缩包；传输接口只打包已经由 Task_Pool 准备流程生成的 `item.source_task_dir`。如果主服务器上的 `source_task_dir` 不存在，接口返回 404，队列重试会保留失败信息。
@@ -102,11 +102,12 @@
 
  **Worker 端流程**：
 
- 1. LandSAR 完成后，收集标准产品包文件列表（从 `task_result.source_files` 和 manifest 获取）
+ 1. LandSAR 完成后，收集标准产品包文件列表（从 `task_result.source_files` 和 manifest 获取）。未配置 `CLUSTER_WORKER_RESULT_ROOT` 时沿用 `item.results_root_dir\runs\<run_key>`；已配置时使用 `CLUSTER_WORKER_RESULT_ROOT\<pair_key>\runs\<run_key>`
  2. 调用 `POST /api/cluster/upload-result/{item_id}`
- 3. 以 multipart 或流式上传产品包（含 primary file、auxiliary files、metadata）
+ 3. 以 multipart 上传 managed run 目录内容（不是外层 run 目录本身）
  4. 主服务器接收后写入该 item 的标准目录 `results_root_dir\runs\<run_key>`
- 5. 触发 catalog 登记（复用现有 `result_catalog_service.bootstrap` 或增量登记）
+ 5. 触发 catalog 登记，修复 `execution_manifest.json` 和 current 指针
+ 6. 主服务器把 item/execution 标记为完成并尝试 finalize cluster run；远端 worker 不再在上传后自行标记完成
 
  **主服务器新增 API**：
 
@@ -122,11 +123,12 @@
 
  **上传后处理**：
 
- 1. 校验文件完整性（与 manifest 对比）
- 2. 写入 `D:\production_results\dinsar\<engine_code>\<profile>\<task_name>\`
- 3. 调用 `result_catalog_service` 增量登记
- 4. 更新 `DinsarProductionExecutionORM` 指向最终结果路径
- 5. 返回登记结果给 worker
+ 1. 安全解压上传 zip，替换该 item 的 `results_root_dir\runs\<run_key>`
+ 2. 调用 `result_catalog_service` 增量登记
+ 3. 校验 catalog 只登记 1 条且生成 `execution_manifest.json` 和 current 指针
+ 4. 更新 `DinsarProductionExecutionORM` / `DinsarProductionRunItemORM` 指向主服务器最终结果路径
+ 5. 标记 item 完成，并在所有 item 终态后 finalize run
+ 6. 返回登记结果给 worker
 
  ### 错误处理与重试
 
@@ -199,10 +201,14 @@
  CLUSTER_MAIN_SERVER_URL=http://192.168.1.62
  CLUSTER_SHARED_TOKEN=<same-long-random-token-on-main-and-workers>
  CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600
+ CLUSTER_WORKER_TASK_ROOT=D:\Cluster_Work\Task_Pool
+ CLUSTER_WORKER_RESULT_ROOT=D:\Cluster_Work\Results
+ LANDSAR_RUNTIME_ID=landsar_runtime_v1
  ```
 
- `CLUSTER_MAIN_SERVER_URL` 用于 worker 构造下载/上传 API 的完整 URL。未配置时默认使用 `DATABASE_URL` 中的 host 推断。
- `CLUSTER_SHARED_TOKEN` 是集群传输接口的专用共享密钥，主服务器和所有远端 worker 必须一致且非空；未配置时 `/api/cluster/...` 接口返回 503。
+`CLUSTER_MAIN_SERVER_URL` 用于 worker 构造下载/上传 API 的完整 URL。未配置时默认使用 `DATABASE_URL` 中的 host 推断。
+`CLUSTER_SHARED_TOKEN` 是集群传输接口的专用共享密钥，主服务器和所有远端 worker 必须一致且非空；未配置时 `/api/cluster/...` 接口返回 503。
+`CLUSTER_WORKER_TASK_ROOT`、`CLUSTER_WORKER_RESULT_ROOT` 是远端 worker 本机缓存根。未配置时保持旧行为，直接使用数据库中的 `source_task_dir` 和 `results_root_dir`。
 
  ## 实现路线图
 
@@ -224,6 +230,8 @@
  - [x] 数据搬运 pre-flight（HTTP 下载 Task_Pool zip + 安全解压）
  - [x] 结果上传 post-flight（HTTP 上传结果 zip + catalog 登记）
  - [x] 集群传输接口 `CLUSTER_SHARED_TOKEN` 鉴权
+ - [x] 主服务器上传登记后负责 item/execution 完成和 run finalize
+ - [x] 支持 worker 本机输入/结果根，避免强依赖主服务器盘符
  - [ ] 远端 .6 端到端测试
  - [ ] Worker 开机自启