Harden LandSAR cluster transfer e2e

.env.example

@@ -346,7 +346,6 @@ LANDSAR_CLUSTER_ALLOWED_WORKER_IPS=192.168.1.6
 # /api/cluster input download and result upload.
 CLUSTER_SHARED_TOKEN=
 CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600
-CLUSTER_MATERIALIZE_TEMP_DIR=D:\Task_Pool\_cluster_temp
 # Remote LandSAR workers should point this to the main backend URL.
 CLUSTER_MAIN_SERVER_URL=
 # Empty means the worker can claim all job types. Remote LandSAR nodes should set:

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,
@@ -20,38 +22,29 @@ from fastapi import (
     Form,
     Header,
     HTTPException,
+    Query,
     UploadFile,
 )
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, StreamingResponse
-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
+from ..services.cluster_transport import (
+    build_cluster_input_manifest,
+    iter_cluster_input_package_files,
+    normalize_cluster_relative_path,
+    stream_zip_files,
+)
 
 router = APIRouter()
 
 
-def _cluster_materialize_temp_dir() -> str:
-    explicit = str(os.environ.get("CLUSTER_MATERIALIZE_TEMP_DIR") or "").strip()
-    if not explicit:
-        explicit = str(
-            getattr(settings, "CLUSTER_MATERIALIZE_TEMP_DIR", "") or ""
-        ).strip()
-    if explicit:
-        return os.path.normpath(explicit)
-    task_pool = str(getattr(settings, "TASK_POOL_ROOT", "") or "").strip()
-    if task_pool:
-        return os.path.normpath(os.path.join(task_pool, "_cluster_temp"))
-    return os.path.normpath(
-        os.path.join(
-            os.path.dirname(__file__), "..", "..", "runtime", "_cluster_temp"
-        )
-    )
-
-
 def _cluster_shared_token() -> str:
     return str(os.environ.get("CLUSTER_SHARED_TOKEN") or "").strip()
 
@@ -73,6 +66,61 @@ def _require_cluster_token(
 # Download input package
 # ---------------------------------------------------------------------------
 
+async def _get_cluster_item_source_dir(
+    item_id: int,
+    db: AsyncSession,
+) -> tuple[DinsarProductionRunItemORM, str]:
+    item = await db.get(DinsarProductionRunItemORM, int(item_id))
+    if item is None:
+        raise HTTPException(status_code=404, detail="Cluster item not found.")
+
+    source_dir = os.path.normpath(str(item.source_task_dir or ""))
+    if not source_dir or not os.path.isdir(source_dir):
+        raise HTTPException(
+            status_code=404,
+            detail=f"Source task directory not found: {source_dir}",
+        )
+    return item, source_dir
+
+
+@router.get("/cluster/input-manifest/{item_id}")
+async def get_cluster_input_manifest(
+    item_id: int,
+    _cluster_token: None = Depends(_require_cluster_token),
+    db: AsyncSession = Depends(get_db),
+):
+    _, source_dir = await _get_cluster_item_source_dir(item_id, db)
+    manifest = build_cluster_input_manifest(source_dir)
+    if int(manifest.get("file_count") or 0) <= 0:
+        raise HTTPException(
+            status_code=404,
+            detail=f"No packageable LandSAR input files found: {source_dir}",
+        )
+    return manifest
+
+
+@router.get("/cluster/input-file/{item_id}")
+async def download_cluster_input_file(
+    item_id: int,
+    relative_path: str = Query(...),
+    _cluster_token: None = Depends(_require_cluster_token),
+    db: AsyncSession = Depends(get_db),
+):
+    _, source_dir = await _get_cluster_item_source_dir(item_id, db)
+    normalized_rel = normalize_cluster_relative_path(relative_path)
+    package_files = {
+        normalize_cluster_relative_path(rel_path): abs_path
+        for abs_path, rel_path in iter_cluster_input_package_files(source_dir)
+    }
+    source_path = package_files.get(normalized_rel)
+    if not source_path or not os.path.isfile(source_path):
+        raise HTTPException(status_code=404, detail="Cluster input file not found.")
+    return FileResponse(
+        path=source_path,
+        media_type="application/octet-stream",
+        filename=os.path.basename(source_path),
+    )
+
 @router.get("/cluster/input-package/{item_id}")
 async def download_cluster_input_package(
     item_id: int,
@@ -90,56 +138,23 @@ async def download_cluster_input_package(
           orbit/  ...
           pair_metadata.json
     """
-    item = await db.get(DinsarProductionRunItemORM, int(item_id))
+    _, source_dir = await _get_cluster_item_source_dir(item_id, db)
-    if item is None:
-        raise HTTPException(status_code=404, detail="Cluster item not found.")
-
-    source_dir = os.path.normpath(str(item.source_task_dir or ""))
-    if not source_dir or not os.path.isdir(source_dir):
-        raise HTTPException(
-            status_code=404,
-            detail=f"Source task directory not found: {source_dir}",
-        )
 
     task_name = os.path.basename(source_dir)
-    parent_dir = os.path.dirname(source_dir)
+    package_files = list(iter_cluster_input_package_files(source_dir))
-    temp_root = _cluster_materialize_temp_dir()
+    if not package_files:
-    os.makedirs(temp_root, exist_ok=True)
-
-    package_root = tempfile.mkdtemp(
-        prefix=f"cluster_input_{item_id}_",
-        dir=temp_root,
-    )
-    tmp_base = os.path.join(package_root, task_name)
-    try:
-        zip_path = shutil.make_archive(
-            tmp_base,
-            "zip",
-            root_dir=parent_dir,
-            base_dir=task_name,
-        )
-    except Exception as exc:
-        try:
-            shutil.rmtree(package_root)
-        except Exception:
-            pass
         raise HTTPException(
-            status_code=500,
+            status_code=404,
-            detail=f"Failed to create input package: {exc}",
+            detail=f"No packageable LandSAR input files found: {source_dir}",
         )
 
-    def _cleanup():
+    return StreamingResponse(
-        try:
+        stream_zip_files(package_files, top_level_dir=task_name),
-            if os.path.isdir(package_root):
-                shutil.rmtree(package_root)
-        except Exception:
-            pass
-
-    return FileResponse(
-        path=zip_path,
         media_type="application/zip",
-        filename=f"{task_name}.zip",
+        headers={
-        background=BackgroundTask(_cleanup),
+            "Content-Disposition": f'attachment; filename="{task_name}.zip"',
+            "X-Cluster-Package-File-Count": str(len(package_files)),
+        },
     )
 
 
@@ -263,17 +278,90 @@ 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_result = await db.execute(
+            select(DinsarProductionRunORM).where(
+                DinsarProductionRunORM.run_id == str(item.run_id or "").strip()
+            )
+        )
+        run = run_result.scalar_one_or_none()
+        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
+        execution.error_message = None
+        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

@@ -6,16 +6,19 @@ data from the main server and push results back via HTTP.
 """
 from __future__ import annotations
 
+import asyncio
 import io
 import json
 import os
+import queue
 import shutil
 import tempfile
+import threading
 import urllib.error
 import urllib.parse
 import urllib.request
 import zipfile
-from typing import TYPE_CHECKING
+from typing import Any, Iterable, Iterator, TYPE_CHECKING
 
 if TYPE_CHECKING:
     from ..models.orm import DinsarProductionRunItemORM, DinsarProductionRunORM
@@ -46,6 +49,22 @@ def _cluster_request_headers() -> dict[str, str]:
     return {"X-Cluster-Token": token}
 
 
+def normalize_cluster_relative_path(relative_path: Any) -> str:
+    text = str(relative_path or "").strip().replace("\\", "/")
+    if not text or text.startswith("/") or os.path.splitdrive(text)[0]:
+        raise ValueError(f"Unsafe cluster relative path: {relative_path}")
+    parts = []
+    for part in text.split("/"):
+        if not part or part == ".":
+            continue
+        if part == ".." or os.path.splitdrive(part)[0]:
+            raise ValueError(f"Unsafe cluster relative path: {relative_path}")
+        parts.append(part)
+    if not parts:
+        raise ValueError(f"Unsafe cluster relative path: {relative_path}")
+    return "/".join(parts)
+
+
 def safe_extract_zip(zf: zipfile.ZipFile, target_dir: str) -> None:
     """Extract a zip after verifying all members stay inside target_dir."""
     target_root = os.path.abspath(target_dir)
@@ -67,6 +86,278 @@ 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_STORED, allowZip64=True) 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 _build_multipart_form_data(
+    *,
+    fields: dict[str, str],
+    files: list[tuple[str, str, str, bytes]],
+    boundary: str,
+) -> bytes:
+    body = io.BytesIO()
+
+    def _write_field(
+        name: str,
+        filename: str,
+        content_type: str,
+        data: bytes,
+    ) -> None:
+        body.write(f"--{boundary}\r\n".encode("utf-8"))
+        if filename:
+            body.write(
+                f'Content-Disposition: form-data; name="{name}"; '
+                f'filename="{filename}"\r\n'.encode("utf-8")
+            )
+            body.write(f"Content-Type: {content_type}\r\n".encode("utf-8"))
+        else:
+            body.write(
+                f'Content-Disposition: form-data; name="{name}"\r\n'.encode(
+                    "utf-8"
+                )
+            )
+        body.write(b"\r\n")
+        body.write(data)
+        body.write(b"\r\n")
+
+    for name, value in fields.items():
+        _write_field(name, "", "text/plain", str(value or "").encode("utf-8"))
+    for name, filename, content_type, data in files:
+        _write_field(name, filename, content_type, data)
+    body.write(f"--{boundary}--\r\n".encode("utf-8"))
+    return body.getvalue()
+
+
+def _is_lt1_source_file(name: str) -> bool:
+    lower_name = str(name or "").lower()
+    return lower_name.startswith("lt1") and lower_name.endswith(
+        (".xml", ".tif", ".tiff", ".jpg", ".jpeg", ".rpc")
+    )
+
+
+def _extract_date_from_name(name: str) -> str:
+    import re
+
+    match = re.search(r"((?:19|20)\d{6})", str(name or ""))
+    return match.group(1) if match else ""
+
+
+def _has_landsar_input_pair(input_data_dir: str) -> bool:
+    if not os.path.isdir(input_data_dir):
+        return False
+
+    by_date: dict[str, set[str]] = {}
+    for entry in os.scandir(input_data_dir):
+        if not entry.is_file():
+            continue
+        lower_name = entry.name.lower()
+        if not lower_name.endswith((".xml", ".tif", ".tiff")):
+            continue
+        if lower_name.endswith(".meta.xml") or lower_name.endswith("_check.xml"):
+            continue
+        if lower_name.endswith(".xml") and "_slc" not in lower_name:
+            continue
+        date_text = _extract_date_from_name(entry.name)
+        if not date_text:
+            continue
+        kinds = by_date.setdefault(date_text, set())
+        if lower_name.endswith(".xml"):
+            kinds.add("xml")
+        elif lower_name.endswith((".tif", ".tiff")):
+            kinds.add("tif")
+
+    return sum(1 for kinds in by_date.values() if {"xml", "tif"} <= kinds) >= 2
+
+
+def _has_direct_landsar_source_file(directory: str) -> bool:
+    if not os.path.isdir(directory):
+        return False
+    try:
+        for entry in os.scandir(directory):
+            if entry.is_file() and _is_lt1_source_file(entry.name):
+                return True
+    except OSError:
+        return False
+    return False
+
+
+def _iter_tree_files(root_dir: str, relative_root: str = "") -> Iterator[tuple[str, str]]:
+    if not os.path.isdir(root_dir):
+        return
+    root_abs = os.path.abspath(root_dir)
+    for current, _, files in os.walk(root_abs):
+        for name in sorted(files):
+            path = os.path.join(current, name)
+            rel = os.path.relpath(path, root_abs)
+            if relative_root:
+                rel = os.path.join(relative_root, rel)
+            yield path, rel
+
+
+def _iter_direct_files(root_dir: str, relative_root: str = "") -> Iterator[tuple[str, str]]:
+    if not os.path.isdir(root_dir):
+        return
+    for entry in sorted(os.scandir(root_dir), key=lambda item: item.name.lower()):
+        if not entry.is_file():
+            continue
+        rel = os.path.join(relative_root, entry.name) if relative_root else entry.name
+        yield entry.path, rel
+
+
+def iter_cluster_input_package_files(source_task_dir: str) -> Iterator[tuple[str, str]]:
+    """Yield the minimal LandSAR input package file set.
+
+    A Task_Pool item can contain both raw LT-1 ``master/slave`` files and a
+    derived ``Input_Data`` directory. Shipping both copies is wasteful for
+    remote workers. Prefer a valid LandSAR ``Input_Data`` pair; otherwise ship
+    only direct raw files under ``master`` and ``slave``. This keeps the
+    extracted layout compatible with LandSAR validation while avoiding nested
+    duplicate scene copies.
+    """
+    source_dir = os.path.abspath(source_task_dir)
+    if not os.path.isdir(source_dir):
+        return
+
+    pair_meta = os.path.join(source_dir, ".dinsar_pair.json")
+    if os.path.isfile(pair_meta):
+        yield pair_meta, ".dinsar_pair.json"
+
+    input_data_dir = os.path.join(source_dir, "Input_Data")
+    if _has_landsar_input_pair(input_data_dir):
+        yield from _iter_tree_files(input_data_dir, "Input_Data")
+    else:
+        master_dir = os.path.join(source_dir, "master")
+        slave_dir = os.path.join(source_dir, "slave")
+        if _has_direct_landsar_source_file(master_dir) and _has_direct_landsar_source_file(slave_dir):
+            yield from _iter_direct_files(master_dir, "master")
+            yield from _iter_direct_files(slave_dir, "slave")
+        else:
+            yield from _iter_tree_files(source_dir)
+            return
+
+    orbit_dir = os.path.join(source_dir, "orbit")
+    if os.path.isdir(orbit_dir):
+        yield from _iter_tree_files(orbit_dir, "orbit")
+
+
+def build_cluster_input_manifest(source_task_dir: str) -> dict[str, Any]:
+    files = []
+    total_bytes = 0
+    for abs_path, rel_path in iter_cluster_input_package_files(source_task_dir):
+        if not os.path.isfile(abs_path):
+            continue
+        normalized_rel = normalize_cluster_relative_path(rel_path)
+        stat = os.stat(abs_path)
+        size = int(stat.st_size)
+        total_bytes += size
+        files.append(
+            {
+                "relative_path": normalized_rel,
+                "size": size,
+                "mtime": float(stat.st_mtime),
+            }
+        )
+    return {
+        "task_name": os.path.basename(os.path.normpath(source_task_dir)),
+        "file_count": len(files),
+        "total_bytes": total_bytes,
+        "files": files,
+    }
+
+
+class _QueueZipWriter:
+    def __init__(self, output_queue: "queue.Queue[object]") -> None:
+        self._queue = output_queue
+        self._closed = False
+
+    def write(self, data: bytes) -> int:
+        if self._closed:
+            raise BrokenPipeError("zip stream is closed")
+        chunk = bytes(data)
+        if not chunk:
+            return 0
+        while not self._closed:
+            try:
+                self._queue.put(chunk, timeout=1)
+                return len(chunk)
+            except queue.Full:
+                continue
+        raise BrokenPipeError("zip stream is closed")
+
+    def flush(self) -> None:
+        return None
+
+    def close(self) -> None:
+        self._closed = True
+
+
+def stream_zip_files(
+    files: Iterable[tuple[str, str]],
+    *,
+    top_level_dir: str,
+) -> Iterator[bytes]:
+    """Stream a zip archive for *files* without materializing it first."""
+    output_queue: "queue.Queue[object]" = queue.Queue(maxsize=8)
+    sentinel = object()
+    writer = _QueueZipWriter(output_queue)
+    safe_top = str(top_level_dir or "Task").strip().strip("/\\") or "Task"
+
+    def _put_control(item: object) -> None:
+        while True:
+            try:
+                output_queue.put(item, timeout=1)
+                return
+            except queue.Full:
+                if writer._closed:
+                    return
+
+    def _producer() -> None:
+        try:
+            with zipfile.ZipFile(
+                writer,
+                mode="w",
+                compression=zipfile.ZIP_STORED,
+                allowZip64=True,
+            ) as zf:
+                for abs_path, rel_path in files:
+                    if writer._closed:
+                        raise BrokenPipeError("zip stream is closed")
+                    if not os.path.isfile(abs_path):
+                        continue
+                    arcname = os.path.join(safe_top, rel_path).replace("\\", "/")
+                    zf.write(abs_path, arcname)
+        except Exception as exc:
+            if not writer._closed:
+                _put_control(exc)
+        finally:
+            _put_control(sentinel)
+
+    producer = threading.Thread(
+        target=_producer,
+        name="cluster-input-zip-stream",
+        daemon=True,
+    )
+    producer.start()
+
+    try:
+        while True:
+            item = output_queue.get()
+            if item is sentinel:
+                break
+            if isinstance(item, Exception):
+                raise item
+            yield item  # type: ignore[misc]
+    finally:
+        writer.close()
+
+
 def _resolve_cluster_server_url() -> str:
     """Return the main-server HTTP base URL for cluster data transport.
 
@@ -99,57 +390,130 @@ 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.
+    """Download the input data for a cluster item file by file."""
-
-    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.
-    """
     from .task_service import task_service
 
     server_url = _resolve_cluster_server_url()
-    download_url = f"{server_url}/api/cluster/input-package/{item.id}"
+    manifest_url = f"{server_url}/api/cluster/input-manifest/{item.id}"
-    parent_dir = os.path.dirname(source_task_dir)
+    parent_dir = os.path.dirname(local_task_dir)
-    task_name = os.path.basename(source_task_dir)
+    task_name = os.path.basename(local_task_dir)
+    staging_dir = os.path.join(parent_dir, f".{task_name}.download")
 
     await task_service.add_log(
         task_id,
         "INFO",
-        f"[cluster] Downloading input data from {download_url} ...",
+        f"[cluster] Fetching input manifest from {manifest_url} ...",
     )
 
-    tmp_zip = os.path.join(
+    def _request_json(url: str) -> dict[str, Any]:
-        tempfile.gettempdir(),
-        f"cluster_input_{item.id}_{task_name}.zip",
-    )
-    try:
         req = urllib.request.Request(
-            download_url,
+            url,
             headers=_cluster_request_headers(),
             method="GET",
         )
         with urllib.request.urlopen(req, timeout=_cluster_transfer_timeout()) as resp:
-            with open(tmp_zip, "wb") as fh:
+            return json.loads(resp.read().decode("utf-8"))
+
+    def _download_file(relative_path: str, expected_size: int) -> None:
+        normalized_rel = normalize_cluster_relative_path(relative_path)
+        file_url = (
+            f"{server_url}/api/cluster/input-file/{item.id}"
+            f"?relative_path={urllib.parse.quote(normalized_rel, safe='')}"
+        )
+        target_path = os.path.join(staging_dir, *normalized_rel.split("/"))
+        tmp_path = f"{target_path}.part"
+        os.makedirs(os.path.dirname(target_path), exist_ok=True)
+        req = urllib.request.Request(
+            file_url,
+            headers=_cluster_request_headers(),
+            method="GET",
+        )
+        with urllib.request.urlopen(req, timeout=_cluster_transfer_timeout()) as resp:
+            with open(tmp_path, "wb") as fh:
                 shutil.copyfileobj(resp, fh, 8 * 1024 * 1024)
-
+        actual_size = os.path.getsize(tmp_path)
-        os.makedirs(parent_dir, exist_ok=True)
+        if expected_size >= 0 and actual_size != expected_size:
-        with zipfile.ZipFile(tmp_zip, "r") as zf:
-            safe_extract_zip(zf, parent_dir)
-
-        if not os.path.isdir(source_task_dir):
             raise RuntimeError(
-                f"Extraction did not create expected directory: {source_task_dir}"
+                f"Downloaded file size mismatch for {normalized_rel}: "
+                f"expected={expected_size} actual={actual_size}"
             )
+        os.replace(tmp_path, target_path)
+
+    try:
+        manifest = await asyncio.to_thread(_request_json, manifest_url)
+        files = list(manifest.get("files") or [])
+        if not files:
+            raise RuntimeError(f"Input manifest contains no files: {manifest_url}")
+
+        total_bytes = int(manifest.get("total_bytes") or 0)
+        await task_service.add_log(
+            task_id,
+            "INFO",
+            f"[cluster] Downloading input files: count={len(files)} bytes={total_bytes}",
+        )
+
+        if os.path.isdir(staging_dir):
+            await asyncio.to_thread(shutil.rmtree, staging_dir)
+        os.makedirs(staging_dir, exist_ok=True)
+
+        for index, file_info in enumerate(files, start=1):
+            rel_path = normalize_cluster_relative_path(file_info.get("relative_path"))
+            expected_size = int(file_info.get("size") or -1)
+            await task_service.add_log(
+                task_id,
+                "INFO",
+                f"[cluster] Downloading input file {index}/{len(files)}: "
+                f"{rel_path} ({expected_size} bytes)",
+            )
+            await asyncio.to_thread(_download_file, rel_path, expected_size)
+            await task_service.add_log(
+                task_id,
+                "INFO",
+                f"[cluster] Downloaded input file {index}/{len(files)}: {rel_path}",
+            )
+
+        if os.path.isdir(local_task_dir):
+            await asyncio.to_thread(shutil.rmtree, local_task_dir)
+        os.makedirs(parent_dir, exist_ok=True)
+        os.replace(staging_dir, 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 +531,8 @@ async def materialize_cluster_input(
         ) from exc
     finally:
         try:
-            os.unlink(tmp_zip)
+            if os.path.isdir(staging_dir):
+                shutil.rmtree(staging_dir)
         except Exception:
             pass
 
@@ -194,62 +559,34 @@ async def upload_cluster_result(
         "[cluster] Packaging results for upload ...",
     )
 
-    run_dir_name = os.path.basename(os.path.normpath(managed_run_dir))
+    tmp_root = tempfile.mkdtemp(prefix=f"cluster_result_{item.id}_")
-    tmp_zip = os.path.join(
+    tmp_zip = os.path.join(tmp_root, "result.zip")
-        tempfile.gettempdir(),
-        f"cluster_result_{item.id}.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,
-        )
 
-        await task_service.add_log(
+    def _package_and_upload() -> dict:
-            task_id,
+        _zip_directory_contents(managed_run_dir, tmp_zip)
-            "INFO",
-            "[cluster] Uploading results to main server ...",
-        )
 
         boundary = "----ClusterUploadBoundary"
-        body = io.BytesIO()
-
-        def _write_field(name, filename, content_type, data):
-            body.write(f"--{boundary}\r\n".encode("utf-8"))
-            if filename:
-                body.write(
-                    f'Content-Disposition: form-data; name="{name}"; '
-                    f'filename="{filename}"\r\n'.encode("utf-8")
-                )
-                body.write(f"Content-Type: {content_type}\r\n".encode("utf-8"))
-            else:
-                body.write(
-                    f'Content-Disposition: form-data; name="{name}"\r\n'.encode(
-                        "utf-8"
-                    )
-                )
-                body.write(b"\r\n")
-            body.write(data)
-            body.write(b"\r\n")
-
-        _write_field("run_id", "", "text/plain", (run.run_id or "").encode("utf-8"))
-        _write_field("run_key", "", "text/plain", str(run_key or "").encode("utf-8"))
         with open(tmp_zip, "rb") as fh:
-            _write_field(
+            body = _build_multipart_form_data(
-                "result_zip",
+                fields={
-                os.path.basename(tmp_zip),
+                    "run_id": str(run.run_id or ""),
-                "application/zip",
+                    "run_key": str(run_key or ""),
-                fh.read(),
+                },
+                files=[
+                    (
+                        "result_zip",
+                        os.path.basename(tmp_zip),
+                        "application/zip",
+                        fh.read(),
+                    ),
+                ],
+                boundary=boundary,
             )
-        body.write(f"--{boundary}--\r\n".encode("utf-8"))
 
         upload_url = f"{server_url}/api/cluster/upload-result/{item.id}"
         req = urllib.request.Request(
             upload_url,
-            data=body.getvalue(),
+            data=body,
             headers={
                 "Content-Type": f"multipart/form-data; boundary={boundary}",
                 **_cluster_request_headers(),
@@ -258,15 +595,24 @@ async def upload_cluster_result(
         )
 
         with urllib.request.urlopen(req, timeout=_cluster_transfer_timeout()) as resp:
-            result = json.loads(resp.read().decode("utf-8"))
+            return json.loads(resp.read().decode("utf-8"))
-            await task_service.add_log(
+
-                task_id,
+    try:
-                "INFO",
+        await task_service.add_log(
-                f"[cluster] Results uploaded: "
+            task_id,
-                f"registered={result.get('registered', False)} "
+            "INFO",
-                f"processed={result.get('processed', 0)}",
+            "[cluster] Uploading results to main server ...",
-            )
+        )
-            return bool(result.get("registered", False))
+
+        result = await asyncio.to_thread(_package_and_upload)
+        await task_service.add_log(
+            task_id,
+            "INFO",
+            f"[cluster] Results uploaded: "
+            f"registered={result.get('registered', False)} "
+            f"processed={result.get('processed', 0)}",
+        )
+        return bool(result.get("registered", False))
 
     except urllib.error.HTTPError as exc:
         body_text = ""
@@ -284,7 +630,7 @@ async def upload_cluster_result(
         ) from exc
     finally:
         try:
-            if os.path.isfile(tmp_zip):
+            if os.path.isdir(tmp_root):
-                os.unlink(tmp_zip)
+                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):
+        if not os.path.isdir(child_dir) or not _has_landsar_source_file(child_dir):
-            return False
-        try:
-            with os.scandir(child_dir) as entries:
-                if not any(entries):
-                    return False
-        except OSError:
             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(
+            if not _is_remote_worker():
-                dinsar_production_service.write_current_pointer,
+                await asyncio.to_thread(
-                run=run,
+                    dinsar_production_service.write_current_pointer,
-                item=item,
+                    run=run,
-                execution=execution,
+                    item=item,
-                manifest_path=manifest_path,
+                    execution=execution,
-                primary_file=primary_file,
+                    manifest_path=manifest_path,
-                source_files=source_files,
+                    primary_file=primary_file,
-                native_output_dir=native_output_dir,
+                    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,
-            )
 
             # ---- 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(
+            remote_completed = False
-                run=run,
+            if _is_remote_worker():
-                item=item,
+                try:
-                execution=execution,
+                    await db.refresh(item)
-                error_message=item_error,
+                    remote_completed = str(item.status or "").strip().upper() == "COMPLETED"
-                db=db,
+                except Exception:
-            )
+                    remote_completed = False
-            await task_service.add_log(
+            if remote_completed:
-                job.task_id,
+                await task_service.add_log(
-                "WARNING",
+                    job.task_id,
-                f"[cluster {item_index}/{total_items}] Failed {item_label}: {item_error}",
+                    "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,
+                dinsar_production_service.append_run_log(
-                f"[cluster-item-failed] {item_index}/{total_items} {item_label}: {item_error}",
+                    run.run_id,
-            )
+                    f"[cluster-item-ok-after-upload] {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')}")
+            else:
-            if task_result.get("stdout_tail"):
+                await dinsar_production_service.mark_item_failed(
-                await task_service.add_log(job.task_id, "INFO", f"LandSAR stdout tail [{item_label}]:\n{task_result.get('stdout_tail')}")
+                    run=run,
-            if task_result.get("stderr_tail"):
+                    item=item,
-                await task_service.add_log(job.task_id, "WARNING", f"LandSAR stderr tail [{item_label}]:\n{task_result.get('stderr_tail')}")
+                    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,258 @@
+import json
+import os
+import tempfile
+import unittest
+import zipfile
+from io import BytesIO
+from types import SimpleNamespace
+from unittest import mock
+
+from multipart import parse_form
+
+from backend.app.services.cluster_transport import (
+    _build_multipart_form_data,
+    _zip_directory_contents,
+    build_cluster_input_manifest,
+    iter_cluster_input_package_files,
+    normalize_cluster_relative_path,
+    resolve_cluster_local_run_dir,
+    resolve_cluster_local_task_dir,
+    safe_extract_zip,
+    stream_zip_files,
+)
+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_build_multipart_form_data_is_parseable(self):
+        boundary = "----ClusterUploadBoundary"
+        body = _build_multipart_form_data(
+            fields={"run_id": "run-1", "run_key": "key-1"},
+            files=[("result_zip", "result.zip", "application/zip", b"zip-bytes")],
+            boundary=boundary,
+        )
+        fields = {}
+        files = {}
+
+        def on_field(field):
+            fields[field.field_name.decode("utf-8")] = field.value.decode("utf-8")
+
+        def on_file(file):
+            files[file.field_name.decode("utf-8")] = {
+                "file_name": file.file_name.decode("utf-8"),
+                "size": file.size,
+                "content": file.file_object.getvalue(),
+            }
+
+        parse_form(
+            {
+                "Content-Type": f"multipart/form-data; boundary={boundary}".encode("utf-8"),
+                "Content-Length": str(len(body)).encode("utf-8"),
+            },
+            BytesIO(body),
+            on_field,
+            on_file,
+        )
+
+        self.assertEqual(fields, {"run_id": "run-1", "run_key": "key-1"})
+        self.assertEqual(files["result_zip"]["file_name"], "result.zip")
+        self.assertEqual(files["result_zip"]["size"], len(b"zip-bytes"))
+        self.assertEqual(files["result_zip"]["content"], b"zip-bytes")
+
+    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_cluster_input_package_prefers_input_data(self):
+        with tempfile.TemporaryDirectory() as root:
+            task_dir = os.path.join(root, "Task_20260101_20260113")
+            input_dir = os.path.join(task_dir, "Input_Data")
+            master_dir = os.path.join(task_dir, "master")
+            slave_dir = os.path.join(task_dir, "slave")
+            orbit_dir = os.path.join(task_dir, "orbit")
+            os.makedirs(input_dir)
+            os.makedirs(master_dir)
+            os.makedirs(slave_dir)
+            os.makedirs(orbit_dir)
+            with open(os.path.join(task_dir, ".dinsar_pair.json"), "w", encoding="utf-8") as fp:
+                json.dump({"pair_key": "pair"}, fp)
+            for date_text in ("20260101", "20260113"):
+                base = f"LT1A_MONO_TEST_{date_text}_SLC"
+                with open(os.path.join(input_dir, f"{base}.xml"), "wb") as fp:
+                    fp.write(b"xml")
+                with open(os.path.join(input_dir, f"{base}.tiff"), "wb") as fp:
+                    fp.write(b"tif")
+            with open(os.path.join(master_dir, "LT1A_raw_20260101.tiff"), "wb") as fp:
+                fp.write(b"raw-master")
+            with open(os.path.join(slave_dir, "LT1A_raw_20260113.tiff"), "wb") as fp:
+                fp.write(b"raw-slave")
+            with open(os.path.join(orbit_dir, "orbit.txt"), "wb") as fp:
+                fp.write(b"orbit")
+
+            names = {rel.replace("\\", "/") for _, rel in iter_cluster_input_package_files(task_dir)}
+
+            self.assertIn(".dinsar_pair.json", names)
+            self.assertIn("Input_Data/LT1A_MONO_TEST_20260101_SLC.xml", names)
+            self.assertIn("Input_Data/LT1A_MONO_TEST_20260113_SLC.tiff", names)
+            self.assertIn("orbit/orbit.txt", names)
+            self.assertFalse(any(name.startswith("master/") for name in names))
+            self.assertFalse(any(name.startswith("slave/") for name in names))
+
+    def test_cluster_input_manifest_reports_files_and_size(self):
+        with tempfile.TemporaryDirectory() as root:
+            task_dir = os.path.join(root, "Task_20260101_20260113")
+            input_dir = os.path.join(task_dir, "Input_Data")
+            os.makedirs(input_dir)
+            total = 0
+            for date_text, payload in (("20260101", b"xml1"), ("20260113", b"xml2")):
+                base = f"LT1A_MONO_TEST_{date_text}_SLC"
+                for suffix, content in ((".xml", payload), (".tiff", payload * 2)):
+                    path = os.path.join(input_dir, f"{base}{suffix}")
+                    with open(path, "wb") as fp:
+                        fp.write(content)
+                    total += len(content)
+
+            manifest = build_cluster_input_manifest(task_dir)
+
+            self.assertEqual(manifest["task_name"], "Task_20260101_20260113")
+            self.assertEqual(manifest["file_count"], 4)
+            self.assertEqual(manifest["total_bytes"], total)
+            self.assertTrue(
+                all("\\" not in item["relative_path"] for item in manifest["files"])
+            )
+
+    def test_normalize_cluster_relative_path_rejects_escape(self):
+        for value in ("../a.txt", "a/../../b.txt", "/abs.txt", r"C:\abs.txt"):
+            with self.subTest(value=value):
+                with self.assertRaises(ValueError):
+                    normalize_cluster_relative_path(value)
+
+        self.assertEqual(
+            normalize_cluster_relative_path(r"Input_Data\scene.tif"),
+            "Input_Data/scene.tif",
+        )
+
+    def test_stream_zip_files_preserves_task_top_level(self):
+        with tempfile.TemporaryDirectory() as root:
+            payload_path = os.path.join(root, "payload.txt")
+            with open(payload_path, "wb") as fp:
+                fp.write(b"payload")
+
+            zip_bytes = b"".join(
+                stream_zip_files([(payload_path, "Input_Data/payload.txt")], top_level_dir="Task_A")
+            )
+            zip_path = os.path.join(root, "streamed.zip")
+            with open(zip_path, "wb") as fp:
+                fp.write(zip_bytes)
+
+            with zipfile.ZipFile(zip_path, "r") as zf:
+                self.assertEqual(zf.read("Task_A/Input_Data/payload.txt"), b"payload")
+
+    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`（保持与主服务器一致的路径结构）
+ 6. Worker 解包到本地输入目录
- 7. Worker 对 zip 成员路径做目录逃逸校验，并校验解包后的 `master/`、`slave/` 目录可用
+ 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 对比）
+ 1. 安全解压上传 zip，替换该 item 的 `results_root_dir\runs\<run_key>`
- 2. 写入 `D:\production_results\dinsar\<engine_code>\<profile>\<task_name>\`
+ 2. 调用 `result_catalog_service` 增量登记
- 3. 调用 `result_catalog_service` 增量登记
+ 3. 校验 catalog 只登记 1 条且生成 `execution_manifest.json` 和 current 指针
- 4. 更新 `DinsarProductionExecutionORM` 指向最终结果路径
+ 4. 更新 `DinsarProductionExecutionORM` / `DinsarProductionRunItemORM` 指向主服务器最终结果路径
- 5. 返回登记结果给 worker
+ 5. 标记 item 完成，并在所有 item 终态后 finalize run
+ 6. 返回登记结果给 worker
 
  ### 错误处理与重试
 
@@ -187,10 +189,9 @@
 
  ```env
  # 集群传输配置
- CLUSTER_SHARED_TOKEN=<same-long-random-token-on-main-and-workers>
+CLUSTER_SHARED_TOKEN=<same-long-random-token-on-main-and-workers>
- CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600
+CLUSTER_TRANSFER_TIMEOUT_SECONDS=3600
- CLUSTER_MATERIALIZE_TEMP_DIR=D:\Task_Pool\_cluster_temp
+```
- ```
 
  Worker 端 `.env` 新增（或保持现有模板字段）：
 
@@ -199,10 +200,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_MAIN_SERVER_URL` 用于 worker 构造下载/上传 API 的完整 URL。未配置时默认使用 `DATABASE_URL` 中的 host 推断。
- `CLUSTER_SHARED_TOKEN` 是集群传输接口的专用共享密钥，主服务器和所有远端 worker 必须一致且非空；未配置时 `/api/cluster/...` 接口返回 503。
+`CLUSTER_SHARED_TOKEN` 是集群传输接口的专用共享密钥，主服务器和所有远端 worker 必须一致且非空；未配置时 `/api/cluster/...` 接口返回 503。
+`CLUSTER_WORKER_TASK_ROOT`、`CLUSTER_WORKER_RESULT_ROOT` 是远端 worker 本机缓存根。未配置时保持旧行为，直接使用数据库中的 `source_task_dir` 和 `results_root_dir`。
 
  ## 实现路线图
 
@@ -224,6 +229,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 开机自启
 
@@ -233,3 +240,13 @@
  - [LANDSAR_DEM_PREPARATION_CONTRACT_20260618.md](LANDSAR_DEM_PREPARATION_CONTRACT_20260618.md) — LandSAR DEM 准备约定
  - [THREE_SENSOR_LOCAL_PRODUCTION_CONTRACT_20260616.md](THREE_SENSOR_LOCAL_PRODUCTION_CONTRACT_20260616.md) — 三数据本机生产约定
  - [DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md](DINSAR_TASK_POOL_THREE_ENGINE_REFACTOR_20260614.md) — Task_Pool 与三引擎 refactor
+
+## 2026-06-26 Remote 1.6 E2E Findings
+
+- Remote worker `landsar-worker-192-168-1-6` can reach PostgreSQL on `192.168.1.62:5432`, authenticate to `/api/cluster/...`, heartbeat into `system_worker_heartbeats`, and claim `LANDSAR_CLUSTER_ITEM` jobs.
+- The first remote E2E attempt failed before LandSAR execution with `HTTP 504` while downloading `/api/cluster/input-package/{item_id}` through nginx.
+- Root cause: the old input endpoint synchronously built a full Task directory zip inside the FastAPI request. The tested Task directory was 15.47 GB because it contained duplicate `Input_Data`, `master`, and `slave` copies. nginx's default 60s upstream timeout returned 504 while the backend was still packaging, and the synchronous packaging could also block normal UI API responses.
+- Current behavior: the input endpoint streams a ZIP directly to the worker and uses `ZIP_STORED` to avoid wasting CPU on already-compressed TIFF data.
+- Current package selection: if `Input_Data` contains a valid LandSAR SLC xml/tif pair, only `Input_Data`, `orbit`, and `.dinsar_pair.json` are shipped. Otherwise, only direct raw files from `master` and `slave`, plus `orbit` and `.dinsar_pair.json`, are shipped. For item `135`, this reduced the transfer set from 15.47 GB to 5.18 GB.
+- nginx now has a dedicated `/api/cluster/` location with large body support, disabled proxy buffering for cluster transfers, and 7200s send/read timeouts. `scripts/start_app.ps1` also rewrites this location's backend port when `PORT` changes.
+- Upload registration now resolves `DinsarProductionRunORM` by business `run_id`; using `db.get()` with the string run id was invalid because the table primary key is integer.

nginx/nginx.conf

@@ -66,6 +66,21 @@ http {
             try_files $uri $uri/ /index.html;
         }
 
+        location /api/cluster/ {
+            proxy_pass http://127.0.0.1:18000;
+            proxy_http_version 1.1;
+            proxy_set_header Host $host;
+            proxy_set_header X-Real-IP $remote_addr;
+            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+            proxy_set_header X-Forwarded-Proto $scheme;
+            client_max_body_size 64g;
+            proxy_request_buffering off;
+            proxy_buffering off;
+            proxy_read_timeout 7200s;
+            proxy_send_timeout 7200s;
+            send_timeout 7200s;
+        }
+
         location /api/ {
             proxy_pass http://127.0.0.1:18000;
             proxy_http_version 1.1;

scripts/start_app.ps1

@@ -709,6 +709,11 @@ if (Test-Path -LiteralPath "$NginxConfPath") {
         '(location\s+/api/tasks/active/stream\s*\{[\s\S]*?proxy_pass\s+)http://(127\.0\.0\.1|localhost):\d+(;)',
         "`${1}$BackendProxy`${3}"
     )
+    $NewConfContent = [regex]::Replace(
+        $NewConfContent,
+        '(location\s+/api/cluster/\s*\{[\s\S]*?proxy_pass\s+)http://(127\.0\.0\.1|localhost):\d+(;)',
+        "`${1}$BackendProxy`${3}"
+    )
     # 使用 UTF8 无 BOM 编码写入
     $Utf8NoBom = New-Object System.Text.UTF8Encoding $false
     [System.IO.File]::WriteAllText("$NginxConfPath", $NewConfContent, $Utf8NoBom)