Merge remote-tracking branches 'origin/sub_laptop_1', 'origin/sub_laptop_2', 'origin/sub_laptop_3', 'origin/sub_laptop_4', 'origin/sub_laptop_5' and 'origin/sub_laptop_6' into main_laptop_1

2026-10-16 11:17:44 +09:00
parent 40deca9ed9 73a251db26 ca8c88f810 fba1eb8a5d 1fcdf5aa4c f8fcb812d3 eac37456e2
commit fc3175e399
120 changed files with 5516 additions and 10874 deletions
@@ -553,6 +553,11 @@ export interface DetailBasinResponse {
  upstream_lonlat: Array<Array<[number, number]>>;
  /** [저장] 때 못 만든 구조물 조합 사본 폴더(142-2 · `*` = 통째 실패). */
  snapshot_failed?: string[];
+  /** 지문(342-3) — 지금 노선 · 관 자리. */
+  route_signature?: string;
+  pipe_signature?: string;
+  /** 분석이 낡음(노선이 분석 때와 다름) — 까닭 글 · 세부 유역 없음 · 관 자리만. */
+  stale?: string | null;
 }

 let lastNote = { text: "", at: 0 };
@@ -55,6 +55,96 @@ def _source_identity(las_paths: list[Path]) -> dict[str, Any]:
    return files[0] if len(files) == 1 else {"files": files}


+class _LazyXyz:
+    """structured.npz 의 xyz — 처음 값을 쓸 때만 푼다. 길이는 저장해 둔 점 수로 답한다."""
+
+    def __init__(self, structured: Any, count: int, array: np.ndarray | None = None) -> None:
+        self._structured = structured
+        self._count = count
+        self._array = array
+
+    def _load(self) -> np.ndarray:
+        if self._array is None:
+            self._array = self._structured["xyz"]
+        return self._array
+
+    def __len__(self) -> int:
+        return self._count
+
+    def __getitem__(self, key: Any) -> Any:
+        return self._load()[key]
+
+    def __array__(self, dtype: Any = None, copy: Any = None) -> np.ndarray:
+        return np.asarray(self._load(), dtype=dtype)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._load(), name)
+
+
+class _LazyStructured(dict):
+    """구조화 결과 — 열쇠마다 처음 쓸 때만 푼다(저장본이 다 맞는 다시 돎은 점군을 안 풂)."""
+
+    def __init__(self, structured: Any, xyz: _LazyXyz) -> None:
+        super().__init__()
+        self._structured = structured
+        self["xyz"] = xyz
+
+    # 옛 dict 와 같은 열쇠만 보인다 — 필터가 열쇠 유무로 갈리는 일이 없게.
+    _KEYS = ("classification",)
+
+    def __missing__(self, key: str) -> Any:
+        if key not in self._KEYS:
+            raise KeyError(key)
+        value = self._structured[key]
+        self[key] = value
+        return value
+
+    def __contains__(self, key: object) -> bool:
+        return dict.__contains__(self, key) or key in self._KEYS
+
+
+def _structured_stats(structured: Any, structured_path: Path) -> tuple[dict[str, Any], Any]:
+    """점 수 · 평균 Z 를 `structured_stats.json` 에서 읽는다 — 없거나 낡으면 한 번 풀어 적는다.
+
+    돌려주는 둘째 값 = 이번에 푼 xyz(없으면 None) — 같은 부름 안에서 다시 풀지 않게.
+    """
+    stats_path = structured_path.parent / "structured_stats.json"
+    stat = structured_path.stat()
+    stamp = {"size_bytes": int(stat.st_size), "mtime_ns": int(stat.st_mtime_ns)}
+    try:
+        stored = json.loads(stats_path.read_text(encoding="utf-8"))
+        if stored.get("structured") == stamp:
+            return stored, None
+    except (OSError, ValueError, AttributeError):
+        pass
+    xyz = structured["xyz"]
+    count = int(len(xyz))
+    stats = {
+        "structured": stamp,
+        "point_count": count,
+        "mean_z": float(np.mean(xyz[:, 2])) if count else None,
+    }
+    atomic_write_json(stats_path, stats)
+    return stats, xyz
+
+
+def file_stamp(path: Path | None) -> str | None:
+    """결과 파일 지문 = 수정시각 ns : 크기. 없으면 None."""
+    try:
+        stat = path.stat() if path is not None else None
+    except OSError:
+        return None
+    return f"{stat.st_mtime_ns}:{stat.st_size}" if stat else None
+
+
+def _key_matches(key_path: Path, key: Any) -> bool:
+    """결과 옆 열쇠 파일이 지금 입력과 같은지."""
+    try:
+        return json.loads(key_path.read_text(encoding="utf-8")) == key
+    except (OSError, ValueError):
+        return False
+
+
 def _relative_to_project(project_root: Path, path: Path) -> str:
    """프로젝트 루트 기준 posix 상대 경로 문자열."""
    return path.relative_to(project_root).as_posix()
@@ -179,13 +269,14 @@ def run_surface_analysis(
        _report(10, "structurize", "구조화 캐시 재사용")
        logger.info("B04 구조화 캐시 재사용: %s", structured_path.name)
    with np.load(structured_path) as structured:
-        xyz = structured["xyz"]
+        # 점 수 · 평균 Z 는 저장값 — 저장본이 다 맞으면 점군(841 MB 압축)을 풀지 않는다.
+        stored_stats, loaded_xyz = _structured_stats(structured, structured_path)
        bounds = structured["bounds"]
-        total_points = int(len(xyz))
+        total_points = int(stored_stats["point_count"])
        stats = {
            "min_z": float(bounds[2, 0]),
            "max_z": float(bounds[2, 1]),
-            "mean_z": float(np.mean(xyz[:, 2])) if total_points else None,
+            "mean_z": stored_stats["mean_z"],
        }
        bounds_dict = {
            "x_min": float(bounds[0, 0]),
@@ -193,7 +284,8 @@ def run_surface_analysis(
            "y_min": float(bounds[1, 0]),
            "y_max": float(bounds[1, 1]),
        }
-        data = {"xyz": xyz, "bounds": bounds, "classification": structured["classification"]}
+        data = _LazyStructured(structured, _LazyXyz(structured, total_points, loaded_xyz))
+        data["bounds"] = bounds

        # 2. 지면 필터 실행 — mask_{filter}.npy 영구 캐시 우선 재사용 (PLAN A-1)
        _report(40, "ground_filter", "지면 필터 적용 중")
@@ -386,14 +478,21 @@ def download_geodata(

        # 등고선은 국가 gpkg 추가 시점이 전처리보다 늦을 수 있으므로 개별로 존재 여부를 확인한다.
        contour_geojson = processed_dir / "등고선_bounds.geojson"
+        # 열쇠 = 받은 범위 · 좌표계 — 노선이 바뀌어 범위가 달라지면 다시 받는다.
+        gis_key_path = processed_dir / "gis_vectors.key.json"
+        gis_key = json.loads(
+            json.dumps({"epsg": project_epsg, "bounds": bounds_dict_for_download}, default=float)
+        )
        if (
            rebuild
            or not any(processed_dir.glob("*_bounds.geojson"))
            or not contour_geojson.exists()
+            or not _key_matches(gis_key_path, gis_key)
        ):
            try:
                step_started = time.monotonic()
                download_all_gis_vectors(prj_path, bounds_dict_for_download, processed_dir)
+                atomic_write_json(gis_key_path, gis_key)
                logger.info(
                    "B04 국가 GIS 벡터 다운로드 완료 (%.1fs)", time.monotonic() - step_started
                )
@@ -433,9 +532,13 @@ def download_geodata(
            # 선정에서 빠진 zip은 쓰이지 않으므로 정리한다(다른 지역 잔재 포함).
            prune_sheets(sheet_store, sheet_grid)

-            # 확보된 도엽을 레이어별 병합 GeoJSON으로 산출 (기존 산출물 있으면 스킵)
+            # 확보된 도엽을 레이어별 병합 GeoJSON으로 산출 — 열쇠(도엽 번호 목록)가 같으면 스킵
+            merge_key_path = processed_dir / "sheet_merge.key.json"
+            merge_key = {"sheets": sorted(str(item) for item in sheet_result["available"])}
            if sheet_result["available"] and (
-                rebuild or not any(processed_dir.glob("도엽_*.geojson"))
+                rebuild
+                or not any(processed_dir.glob("도엽_*.geojson"))
+                or not _key_matches(merge_key_path, merge_key)
            ):
                from B04_PreProcess.B04_PreProcess_Engine_SheetParser import (
                    parse_and_merge_sheets,
@@ -450,6 +553,7 @@ def download_geodata(
                    {k: v["count"] for k, v in merged.items()},
                    time.monotonic() - step_started,
                )
+                atomic_write_json(merge_key_path, merge_key)
        except Exception as exc:
            logger.warning("B04 수치지형도 도엽 확보 실패: %s", exc)
    except Exception as exc:
@@ -510,6 +614,8 @@ def _collect_analysis_result(
                        "source_filter": filter_key,
                        "representation": meta.get("representation"),
                        "footprint_area_m2": meta.get("footprint_area_m2"),
+                        # 파일 지문 — 같으면 DB 행 id 를 살린다(화면 주소 · 노선 연결 유지).
+                        "file_stamp": file_stamp(model_path),
                    },
                    "layers": layers,
                }
@@ -18,7 +18,7 @@ try:
 except ImportError:
    mapbox_vector_tile = None

-# 메모리 캐시: {project_id}_{layer_name}: GeoDataFrame
+# 메모리 캐시: {project_id}_{layer_name}: ((mtime_ns, 크기), GeoDataFrame)
 _gdf_cache: dict[str, Any] = {}


@@ -41,9 +41,13 @@ def get_tile_bbox(xtile: int, ytile: int, zoom: int) -> tuple[float, float, floa
 def load_geojson_cached(filepath: Path, cache_key: str) -> Any:
    """GeoJSON 파일을 한 번 읽으면 메모리에 GeoDataFrame으로 캐싱합니다."""
    global _gdf_cache
-    if cache_key not in _gdf_cache:
-        if not filepath.exists():
-            raise FileNotFoundError(f"File not found: {filepath}")
+    if not filepath.exists():
+        raise FileNotFoundError(f"File not found: {filepath}")
+    # 열쇠에 수정시각·크기 — 전처리를 다시 돌려 파일이 바뀌면 옛 타일을 버린다.
+    stat = filepath.stat()
+    stamp = (stat.st_mtime_ns, stat.st_size)
+    cached = _gdf_cache.get(cache_key)
+    if cached is None or cached[0] != stamp:
        if gpd is None:
            raise ImportError("geopandas가 설치되어 있지 않습니다.")
        gdf = gpd.read_file(filepath)
@@ -51,8 +55,8 @@ def load_geojson_cached(filepath: Path, cache_key: str) -> Any:
            gdf.set_crs(epsg=4326, inplace=True)
        if gdf.crs.to_epsg() != 4326:
            gdf = gdf.to_crs(epsg=4326)
-        _gdf_cache[cache_key] = gdf
-    return _gdf_cache[cache_key]
+        _gdf_cache[cache_key] = (stamp, gdf)
+    return _gdf_cache[cache_key][1]


 def generate_mvt_tile(
@@ -15,6 +15,7 @@ LAS 없는 설계(2026-08-30 사용자 확정)의 지형 원천이자, LAS가
  격자에서 뽑는 1m 등고선이 곧 2D 보간선이며, 메시(glb)는 그 격자의 표현일 뿐이다.
 """

+import hashlib
 import json
 import logging
 import time
@@ -24,6 +25,7 @@ from typing import Any
 import numpy as np
 from pyproj import Transformer

+from B04_PreProcess.B04_PreProcess_Engine import file_stamp
 from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
    atomic_npz,
    clip_and_compact_mesh,
@@ -36,6 +38,7 @@ from B04_PreProcess.B04_PreProcess_Engine_SheetMethods import (
    SHEET_METHOD_LABELS,
 )
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import grid_spec_from_bounds
+from common_util.common_util_json import atomic_write_json
 from config.config_system import (
    SHEET_SURFACE_GRID_M,
    SHEET_SURFACE_MARGIN_M,
@@ -54,6 +57,9 @@ _CONTOUR_FILE = "도엽_등고선.geojson"
 # 산출 모델 식별자 — surface_models.generation_params.source_filter 및 파일 stem에 쓴다.
 SHEET_SOURCE_FILTER = "sheet"

+# 계산 판 번호 — 만드는 방식이 바뀌면 올림 · 열쇠가 어긋나 옛 저장본을 다시 셈.
+SHEET_SURFACE_ALGO = 1
+

 def _load_features_metric(
    processed_dir: Path, crs: str, filename: str = _CONTOUR_FILE
@@ -378,6 +384,15 @@ def _write_method_model(
    vertices, faces = _preview_mesh(x_coords, y_coords, z_grid, valid_grid)
    write_glb(preview_path, vertices, faces, scene)
    _write_smoothed(models_dir, stem, x_coords, y_coords, z_grid, valid_grid, bounds, scene)
+    return _method_model_entry(project_root, models_dir, method_key)
+
+
+def _method_model_entry(project_root: Path, models_dir: Path, method_key: str) -> dict[str, Any]:
+    """방식 하나의 등록용 dict — 저장본 그대로 쓸 때도 같은 모양."""
+    source_filter = f"{SHEET_SOURCE_FILTER}_{method_key}"
+    stem = f"dtm_{source_filter}"
+    model_path = models_dir / f"{stem}.npz"
+    preview_path = models_dir / f"{stem}_preview.glb"
    return {
        "model_type": "dtm",
        "source_filter": source_filter,
@@ -391,6 +406,7 @@ def _write_method_model(
            "interpolation": method_key,
            "interpolation_label": SHEET_METHOD_LABELS.get(method_key, method_key),
            "margin_m": SHEET_SURFACE_MARGIN_M,
+            "file_stamp": file_stamp(model_path),
        },
        "layers": [
            {
@@ -403,6 +419,59 @@ def _write_method_model(
    }


+def _sheet_key(
+    project_root: Path, processed_dir: Path, route_xy: np.ndarray, crs: str, method_key: str
+) -> dict[str, Any] | None:
+    """도엽 3D 면 열쇠 — 등고선 파일 · 노선 · 좌표계 · 격자 · 여유 · 스무딩 · 화면 상자 · 판."""
+    contour = processed_dir / _CONTOUR_FILE
+    try:
+        stat = contour.stat()
+    except OSError:
+        return None
+    scene = _scene_bounds(project_root, None)  # type: ignore[arg-type]  # 라이다 없으면 None
+    key = {
+        "algo": SHEET_SURFACE_ALGO,
+        "method": method_key,
+        "contour": f"{contour.name}:{stat.st_mtime_ns}:{stat.st_size}",
+        "route": hashlib.sha1(np.round(route_xy, 3).tobytes()).hexdigest(),
+        "crs": str(crs),
+        "grid_m": SHEET_SURFACE_GRID_M,
+        "margin_m": SHEET_SURFACE_MARGIN_M,
+        "smoothing": [
+            SURFACE_SMOOTHING_DTM_SIGMA_M,
+            SURFACE_SMOOTHING_DTM_SPLINE_SMOOTH,
+            SURFACE_SMOOTHING_DTM_PREVIEW_RESOLUTION_M,
+            SURFACE_MAX_PREVIEW_VERTICES,
+        ],
+        "scene_bounds": None if scene is None else np.round(scene, 3).tolist(),
+    }
+    return json.loads(json.dumps(key, default=float))
+
+
+def _key_path(models_dir: Path, method_key: str) -> Path:
+    return models_dir / f"dtm_{SHEET_SOURCE_FILTER}_{method_key}.key.json"
+
+
+def _cached_method_model(
+    project_root: Path, models_dir: Path, method_key: str, key: dict[str, Any] | None
+) -> dict[str, Any] | None:
+    """열쇠가 같고 결과 파일이 있으면 등록용 dict — 아니면 None(다시 셈)."""
+    if key is None:
+        return None
+    entry = _method_model_entry(project_root, models_dir, method_key)
+    if not all((project_root / item).is_file() for item in _entry_files(entry)):
+        return None
+    try:
+        stored = json.loads(_key_path(models_dir, method_key).read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    return entry if stored == key else None
+
+
+def _entry_files(entry: dict[str, Any]) -> list[str]:
+    return [entry["model_file_path"], *(layer["file_path"] for layer in entry["layers"])]
+
+
 def build_sheet_surface_model(
    project_root: Path,
    processed_dir: Path,
@@ -420,6 +489,16 @@ def build_sheet_surface_model(
    `route_xy`: (N, 2) 노선 정점 XY(사업지 CRS, m), `crs`: 그 좌표계(pyproj 입력 문자열).
    """
    started = time.monotonic()
+    selected = methods or list(SHEET_SURFACE_METHODS)
+    # 열쇠가 같은 방식은 저장본 그대로 — 다시 셀 방식만 등고선을 읽고 굽는다.
+    keys = {key: _sheet_key(project_root, processed_dir, route_xy, crs, key) for key in selected}
+    cached = {
+        key: _cached_method_model(project_root, models_dir, key, keys[key]) for key in selected
+    }
+    pending = [key for key in selected if cached[key] is None]
+    if not pending:
+        logger.info("도엽 서피스: 열쇠 같음 — 저장본 그대로 (%s)", ", ".join(selected))
+        return [cached[key] for key in selected]  # type: ignore[misc]
    features = _load_features_metric(processed_dir, crs)
    if not features:
        return []
@@ -439,9 +518,8 @@ def build_sheet_surface_model(
    # 등고 간격(m) — 마루 연장 상한의 근거. 레벨이 하나뿐이면 5m(주곡선) 폴백.
    interval_m = float(np.diff(np.array(present)).min()) if len(present) > 1 else 5.0

-    selected = methods or list(SHEET_SURFACE_METHODS)
-    models: list[dict[str, Any]] = []
-    for method_key in selected:
+    built: dict[str, dict[str, Any]] = {}
+    for method_key in pending:
        builder = SHEET_METHOD_BUILDERS.get(method_key)
        if builder is None:
            logger.warning("도엽 서피스: 알 수 없는 보간 방식 %s — 건너뜁니다.", method_key)
@@ -456,9 +534,13 @@ def build_sheet_surface_model(
        except Exception as exc:  # noqa: BLE001 — 한 방식이 죽어도 나머지는 만든다
            logger.warning("도엽 서피스(%s) 생성 실패: %s", method_key, exc)
            continue
+        # 쓰는 도중 끊기면 옛 열쇠가 새 파일 조각을 덮지 않게 먼저 지움.
+        _key_path(models_dir, method_key).unlink(missing_ok=True)
        model = _write_method_model(project_root, models_dir, spec, surface, method_key)
        if model is not None:
-            models.append(model)
+            built[method_key] = model
+            if keys[method_key] is not None:
+                atomic_write_json(_key_path(models_dir, method_key), keys[method_key])
            logger.info("도엽 서피스(%s) 완료 (%.1fs)", method_key, time.monotonic() - step_started)

    logger.info(
@@ -466,10 +548,10 @@ def build_sheet_surface_model(
        spec.n_rows,
        spec.n_cols,
        len(present),
-        len(models),
+        len(built),
        time.monotonic() - started,
    )
-    return models
+    return [cached[key] or built[key] for key in selected if cached[key] or key in built]


 def build_sheet_surface_from_route(
@@ -10,6 +10,9 @@
  ④ **새로 추가된 셀에 적색이 하나도 없으면 종료** (2026-07-31 사용자 지시)

 격자 bbox에 닿으면 격자 자체도 셀 정수배로 넓힌다 — 도로 시작점 기준 격자점은 유지된다.
+
+하강 방향장은 처음에 **넉넉한 큰 격자**로 한 번 짓고, 회차마다 그 격자에서 잘라 쓴다
+(`DESCENT_SPARE_M` · 342-3). 격자가 커질 때마다 다시 짓던 것(8회 · 30 s)을 한 번으로 줄인다.
 """

 from __future__ import annotations
@@ -43,6 +46,11 @@ from config.config_system import (

 logger = logging.getLogger(__name__)

+# 하강 방향장을 지을 때 해석 격자 둘레에 더 두는 여유(m). 확장이 이 안에서 끝나면 방향장을
+# 한 번만 짓는다 — 검증_sub3 실측 확장은 첫 격자에서 남 350 m · 서 200 m 까지 넓어졌다.
+# 넘으면 그때 넓힌 격자 둘레로 다시 짓는다(결과는 같고 시간만 듦).
+DESCENT_SPARE_M = 400.0
+

@dataclass
 class GridAnalysis:
@@ -78,16 +86,23 @@ def analyze_domain(
    """주어진 격자·해석 영역에 대해 등고선 하강 방향과 도로 도달 색을 한 번 계산한다.

    **하강 방향장은 해석 영역과 무관하다** — 등고선 기하만으로 정해진다. 그래서 확장
-    회차마다 다시 계산하지 않고, 격자가 커졌을 때만 새로 만들어 넘겨받는다(`descent`).
+    회차마다 다시 계산하지 않고, 이 격자를 덮는 큰 방향장(`descent`)을 받으면 잘라 쓴다.
    해석 영역은 마지막에 마스크로만 씌운다.
    """
    from B03_FileInput.B03_FileInput_Service_Chain import _log_steps

    marks = [("시작", time.perf_counter())]
-    reused = descent is not None and descent.spec == spec
-    if not reused:
+    fitted = crop_descent(descent, spec) if descent is not None else None
+    if fitted is None:
        descent = build_contour_descent(spec, contour_features, None, elevation_floor_m)
-    marks.append(("하강 방향장" + ("(재사용)" if reused else "(새로 만듦)"), time.perf_counter()))
+    else:
+        descent = fitted
+    marks.append(
+        (
+            "하강 방향장" + ("(잘라 씀)" if fitted is not None else "(새로 만듦)"),
+            time.perf_counter(),
+        )
+    )
    valid = descent.valid & domain
    if not valid.any():
        return None
@@ -128,8 +143,12 @@ def expand_by_red_boundary(
    # 두면 방향장을 다시 만들지 않고 해석 영역만 넓히며 몇 회차를 돌 수 있다.
    spec, domain = _pad_spec(spec, domain, band_cells * 2)
    started_cells = int(domain.sum())
+    spare_cells = max(0, int(round(DESCENT_SPARE_M / spec.cell_m)))
+    source = build_contour_descent(
+        _pad_spec(spec, domain, spare_cells)[0], contour_features, None, elevation_floor_m
+    )
    analysis = analyze_domain(
-        spec, domain, contour_features, route_line, upstream_streams, elevation_floor_m
+        spec, domain, contour_features, route_line, upstream_streams, elevation_floor_m, source
    )
    if analysis is None:
        return None
@@ -160,6 +179,14 @@ def expand_by_red_boundary(
        if not added_mask.any():
            closed = True
            break
+        if crop_descent(source, grown_spec) is None:
+            # 넉넉히 둔 방향장 밖으로 넓어짐 — 넓힌 격자 둘레로 다시 짓는다.
+            source = build_contour_descent(
+                _pad_spec(grown_spec, grown_domain, spare_cells)[0],
+                contour_features,
+                None,
+                elevation_floor_m,
+            )

        logger.info(
            "배수유역: %d회차 확장 — 최외곽 적색 %d셀 주변 %.0fm, 셀 %d개 추가",
@@ -175,22 +202,20 @@ def expand_by_red_boundary(
            route_line,
            upstream_streams,
            elevation_floor_m,
-            # 격자가 그대로면 방향장을 재사용한다 — 등고선 기하가 안 바뀌었으므로 결과는 같다.
-            descent=analysis.descent if grown_spec == current else None,
+            descent=source,
        )
        if widened_analysis is None:
            break
        analysis = widened_analysis
        rounds += 1
        logger.info(
-            "[계측] 확장 %d회차 %.1fs · 격자 %d×%d · 셀 %d (+%d) · 방향장 %s",
+            "[계측] 확장 %d회차 %.1fs · 격자 %d×%d · 셀 %d (+%d)",
            attempt + 1,
            time.perf_counter() - _round_started,
            grown_spec.n_rows,
            grown_spec.n_cols,
            int(widened.sum()),
            int(added_mask.sum()),
-            "재사용" if grown_spec == current else "새로",
        )

        added_reaches = widened_analysis.flow.reaches_road.reshape(
@@ -230,6 +255,59 @@ def _dilate_by(mask: np.ndarray, steps: int) -> np.ndarray:
    return result


+def crop_descent(source: ContourDescent, spec: GridSpec) -> ContourDescent | None:
+    """큰 방향장에서 `spec` 격자 몫을 잘라 낸다. 덮지 못하거나 셀이 안 맞으면 None.
+
+    잘린 가장자리 밖을 가리키던 셀은 제자리(싱크)로 둔다 — 해석 영역 밖으로 빠지는 물과
+    같은 판정(미도달)이다. 화살표(방위)는 원래 방향 그대로 둔다.
+    """
+    big = source.spec
+    if big == spec:
+        return source
+    if big.cell_m != spec.cell_m:
+        return None
+    row_shift = (big.y_max - spec.y_max) / spec.cell_m
+    col_shift = (spec.x_min - big.x_min) / spec.cell_m
+    row0, col0 = int(round(row_shift)), int(round(col_shift))
+    if (
+        abs(row_shift - row0) > 1e-6
+        or abs(col_shift - col0) > 1e-6
+        or row0 < 0
+        or col0 < 0
+        or row0 + spec.n_rows > big.n_rows
+        or col0 + spec.n_cols > big.n_cols
+    ):
+        return None
+    rows = slice(row0, row0 + spec.n_rows)
+    cols = slice(col0, col0 + spec.n_cols)
+
+    def window(flat: np.ndarray) -> np.ndarray:
+        return flat.reshape(big.n_rows, big.n_cols)[rows, cols]
+
+    target = window(source.receiver).astype(np.int64)
+    target_row = target // big.n_cols - row0
+    target_col = target % big.n_cols - col0
+    inside = (
+        (target_row >= 0)
+        & (target_row < spec.n_rows)
+        & (target_col >= 0)
+        & (target_col < spec.n_cols)
+    )
+    local = np.arange(spec.size, dtype=np.int64).reshape(spec.n_rows, spec.n_cols)
+    receiver = np.where(inside, target_row * spec.n_cols + target_col, local)
+    return ContourDescent(
+        spec=spec,
+        band_elevation=source.band_elevation[rows, cols].copy(),
+        valid=source.valid[rows, cols].copy(),
+        receiver=receiver.reshape(-1).astype(np.int32),
+        step_length=np.where(inside, window(source.step_length), 0.0)
+        .reshape(-1)
+        .astype(np.float32),
+        azimuth=window(source.azimuth).reshape(-1).copy(),
+        levels=source.levels,
+    )
+
+
 def _pad_spec(spec: GridSpec, domain: np.ndarray, cells: int) -> tuple[GridSpec, np.ndarray]:
    """격자에 사방 여유를 두고 해석 영역 마스크를 그 안으로 옮겨 담는다."""
    if cells <= 0:
@@ -278,8 +278,8 @@ class FlowClassification:
    direction: np.ndarray  # (R*C,) int16 — 32방위 코드(32=제자리, 33=무효)
    reaches_road: np.ndarray  # (R*C,) bool — 물길을 따라가면 도로에 닿는가
    analyzed: np.ndarray  # (R*C,) bool — 실제로 판정한 셀
-    outer_seeds: int  # 최외곽에서 출발해 판정한 셀 수
-    interior_seeds: int  # 최외곽 추적에 안 걸려 따로 출발시킨 내부 셀 수
+    outer_seeds: int  # 최외곽 셀 수(기록용 · 세류 셀 뺌)
+    interior_seeds: int  # 그 밖 판정 셀 수(기록용 · 세류 셀 뺌)
    burned: np.ndarray | None = None  # (R*C,) bool — 세류망을 따라 흐름을 새긴 셀


@@ -321,6 +321,8 @@ def classify_flow(
         다시 분석하지 않는다.
      ④ 최외곽 추적에 안 걸린 내부 셀을 그다음에 따로 출발시킨다.

+    판정 값은 이 차례로 걷는 것과 칸마다 같다 — 계산은 한꺼번에 한다(`_reach_by_doubling`).
+
    `azimuth`를 주면 그 32방위 코드를 그대로 화살표로 쓴다(등고선 하강 방향). 주지 않으면
    지표면 기울기에서 뽑는다. **화살표는 실제 수신 셀과 같은 방향이어야 한다** — 어긋나면
    화살표로 사슬을 따라가는 눈 검증이 성립하지 않는다.
@@ -336,21 +338,13 @@ def classify_flow(
        direction = descent_azimuth(spec, terrain.elevation, terrain.valid, receiver, burned)
    else:
        direction = _azimuth_with_burn(spec, azimuth, receiver, burned)
-    reaches = np.zeros(spec.size, dtype=bool)
-    # 0=미방문, 1=경로에 올라 있음, 2=판정 완료
-    state = np.zeros(spec.size, dtype=np.int8)
-
-    # ⓪ 세류선과 겹치는 셀을 먼저 적색으로 못박는다.
-    reaches[stream_cells] = True
-    state[stream_cells] = 2
+    reaches = _reach_by_doubling(receiver, valid, absorbing, stream_cells)
+    analyzed = valid | stream_cells
    stream_done = int(stream_cells.sum())
+    # 기록용 숫자 — 한꺼번에 판정하므로 「출발」 구분이 없다. 최외곽 셀 수 · 나머지 셀 수만 남긴다.
+    outer_done = int((outermost_cells(terrain.valid).reshape(-1) & ~stream_cells).sum())
+    interior_done = int((valid & ~stream_cells).sum()) - outer_done

-    outer = np.flatnonzero(outermost_cells(terrain.valid))
-    remaining = np.flatnonzero(valid)
-    outer_done = _walk_from(outer, receiver, valid, absorbing, state, reaches)
-    interior_done = _walk_from(remaining, receiver, valid, absorbing, state, reaches)
-
-    analyzed = state == 2
    logger.info(
        "배수유역: 흐름 판정 %d셀 (세류 선확정 %d / 최외곽 출발 %d / 내부 보충 %d) — "
        "도로 도달 %d, 미도달 %d",
@@ -391,52 +385,32 @@ def _azimuth_with_burn(
    return direction


-def _walk_from(
-    starts: np.ndarray,
+def _reach_by_doubling(
    receiver: np.ndarray,
    valid: np.ndarray,
    absorbing: np.ndarray,
-    state: np.ndarray,
-    reaches: np.ndarray,
-) -> int:
-    """출발 셀에서 물길 사슬을 따라가며 판정하고, 사슬 전체에 같은 색을 적는다.
+    stream_cells: np.ndarray,
+) -> np.ndarray:
+    """물길 사슬을 끝까지 따라간 종착 셀로 셀마다 도로 도달을 한꺼번에 판정한다.

-    `absorbing`은 도로 셀과 확정된 세류망 셀 — 여기 닿으면 사슬 전체가 도로 도달이다.
-    새로 판정한 셀 수를 돌려준다.
+    사슬을 한 칸씩 걷던 판정(최외곽 출발 → 내부 보충 · 이미 판정된 셀의 색 물려받기)과
+    **칸마다 같은 값**이다 — 순환이 없으면 어느 셀에서 출발하든 같은 종착에 닿기 때문이다.
+    종착 = 도로 · 세류 셀(적색) · 싱크 · 해석 영역 밖으로 나가는 셀(미도달). 세류 셀은 표고가
+    없어도 따라간다. 포인터 더블링(`trace_flow` 와 같은 방식)이라 파이썬 반복이 없다.
    """
-    resolved = 0
-    path: list[int] = []
-    for start in starts.tolist():
-        if state[start] == 2:
-            continue
-        path.clear()
-        node = start
-        while True:
-            if state[node] == 2:
-                verdict = bool(reaches[node])  # 이미 색이 정해진 셀 — 그 색을 물려받는다
-                break
-            if state[node] == 1:  # 방어: 채움·평탄해소 후에는 순환이 없어야 한다
-                verdict = False
-                break
-            state[node] = 1
-            path.append(node)
-            if absorbing[node]:
-                verdict = True  # 도로 또는 세류망에 합류 — 여기서 하류로 빠진다
-                break
-            following = int(receiver[node])
-            if following == node:
-                verdict = False  # 싱크에 갇힘
-                break
-            if not valid[following] and state[following] != 2:
-                # 해석 영역 밖으로 빠짐. 단, 이미 색이 정해진 셀(세류 선확정 등)이면 따라간다.
-                verdict = False
-                break
-            node = following
-        for visited in path:
-            reaches[visited] = verdict
-            state[visited] = 2
-        resolved += len(path)
-    return resolved
+    index = np.arange(receiver.size, dtype=np.int64)
+    following = receiver.astype(np.int64)
+    terminal = absorbing | (following == index) | ~(valid[following] | stream_cells[following])
+    jump = np.where(terminal, index, following)
+    for _ in range(_MAX_DOUBLING_ROUNDS):
+        next_jump = jump[jump]
+        if np.array_equal(next_jump, jump):
+            break
+        jump = next_jump
+    # 순환에 갇힌 사슬은 종착에 못 닿는다 — 걷던 판정처럼 미도달.
+    reaches = absorbing[jump] & terminal[jump] & (valid | stream_cells)
+    reaches[stream_cells] = True
+    return reaches


 # ── 격자 확장 (별도 단계) ───────────────────────────────────────────────────
@@ -337,19 +337,114 @@ async def update_project_status(
        )


-async def delete_project_surface_models(connection: aiomysql.Connection, project_id: UUID) -> int:
-    """프로젝트의 기존 지표면 모델 행을 모두 제거한다.
+def _params_dict(value: Any) -> dict[str, Any]:
+    if isinstance(value, (str, bytes)):
+        try:
+            value = json.loads(value)
+        except ValueError:
+            return {}
+    return value if isinstance(value, dict) else {}

-    모든 분석 세대가 동일 파일 경로를 재사용하므로 재분석 시 구세대 행을
-    남기면 존재하지 않는 파일을 가리키게 된다. terrain_layers는 FK CASCADE,
-    routes.surface_model_id는 FK SET NULL로 함께 정리된다.
-    """
+
+def _model_key(model_type: Any, params: dict[str, Any], model_file_path: Any) -> tuple:
+    """같은 모델인지 가리는 열쇠 — (방식, 필터, 파일)."""
+    return (str(model_type), str(params.get("source_filter")), str(model_file_path))
+
+
+async def _existing_models(
+    connection: aiomysql.Connection, project_id: UUID
+) -> dict[tuple, tuple[int, Any]]:
+    """기존 지표면 행 — 열쇠 → (id, 파일 지문)."""
    async with connection.cursor() as cursor:
        await cursor.execute(
-            "DELETE FROM surface_models WHERE project_id = %s",
+            "SELECT id, model_type, model_file_path, generation_params "
+            "FROM surface_models WHERE project_id = %s ORDER BY id",
            (str(project_id),),
        )
-        return int(cursor.rowcount)
+        rows = await cursor.fetchall()
+    existing: dict[tuple, tuple[int, Any]] = {}
+    for model_id, model_type, model_file_path, raw_params in rows:
+        params = _params_dict(raw_params)
+        existing[_model_key(model_type, params, model_file_path)] = (
+            int(model_id),
+            params.get("file_stamp"),
+        )
+    return existing
+
+
+async def _upsert_surface_model(
+    connection: aiomysql.Connection,
+    existing: dict[tuple, tuple[int, Any]],
+    *,
+    project_id: UUID,
+    model: dict[str, Any],
+    source_file_id: int | None,
+    processed_cloud_id: int | None,
+    crs_epsg: int | None,
+) -> int:
+    """파일 지문이 같은 옛 행이면 id 를 살려 고쳐 씀 — 화면 주소 · 노선 연결이 그대로.
+
+    지문이 다르거나 없으면 옛 행을 지우고 새로 넣는다(새 지표면 = 노선 연결도 끊김).
+    """
+    params = model["generation_params"] or {}
+    hit = existing.pop(_model_key(model["model_type"], params, model["model_file_path"]), None)
+    stamp = params.get("file_stamp")
+    if hit is not None and stamp and hit[1] == stamp:
+        model_id = hit[0]
+        async with connection.cursor() as cursor:
+            await cursor.execute(
+                """
+                UPDATE surface_models
+                SET source_file_id = %s, processed_cloud_id = %s, status = 'COMPLETE',
+                    crs_epsg = %s, resolution_m = %s, generation_params = %s,
+                    completed_at = CURRENT_TIMESTAMP
+                WHERE id = %s
+                """,
+                (
+                    source_file_id,
+                    processed_cloud_id,
+                    crs_epsg,
+                    model["resolution_m"],
+                    json.dumps(params, ensure_ascii=False),
+                    model_id,
+                ),
+            )
+            await cursor.execute(
+                "DELETE FROM terrain_layers WHERE surface_model_id = %s", (model_id,)
+            )
+    else:
+        # 같은 파일을 가리키는 옛 행은 걷어 낸다 — 열쇠 모양이 옛것이어도 겹치지 않게.
+        async with connection.cursor() as cursor:
+            if model["model_file_path"]:
+                await cursor.execute(
+                    "DELETE FROM surface_models WHERE project_id = %s AND model_file_path = %s",
+                    (str(project_id), model["model_file_path"]),
+                )
+            elif hit is not None:
+                await cursor.execute("DELETE FROM surface_models WHERE id = %s", (hit[0],))
+        model_id = await create_surface_model(
+            connection,
+            project_id=project_id,
+            model_type=model["model_type"],
+            source_file_id=source_file_id,
+            processed_cloud_id=processed_cloud_id,
+            crs_epsg=crs_epsg,
+            resolution_m=model["resolution_m"],
+            model_file_path=model["model_file_path"],
+            generation_params=model["generation_params"],
+        )
+    for layer in model.get("layers", []):
+        await create_terrain_layer(
+            connection,
+            surface_model_id=model_id,
+            layer_name=layer["layer_name"],
+            geometry_type=layer["geometry_type"],
+            layer_file_path=layer["file_path"],
+            file_format=layer["file_format"],
+            file_size_mb=None,
+            statistics=None,
+        )
+    return model_id


 async def save_surface_analysis_to_db(
@@ -362,11 +457,12 @@ async def save_surface_analysis_to_db(
 ) -> list[int]:
    """WF1 분석 결과를 DB에 저장한다 (기존 모델 행은 교체).

+    파일 지문이 같은 모델은 id 를 살리고, 이번 결과에 없는 옛 행은 지운다.
    이 함수는 트랜잭션을 시작하거나 종료하지 않는다. 호출자는 같은 커넥션에서
    begin/commit/rollback을 한 번만 수행해야 한다.
    """
    input_file = await get_input_file(connection, project_id, input_file_id)
-    await delete_project_surface_models(connection, project_id)
+    existing = await _existing_models(connection, project_id)
    processed = analysis_result["processed"]
    processed_cloud_id = await create_processed_point_cloud(
        connection,
@@ -384,28 +480,26 @@ async def save_surface_analysis_to_db(
    )
    surface_model_ids: list[int] = []
    for model in analysis_result["models"]:
-        model_id = await create_surface_model(
-            connection,
-            project_id=project_id,
-            model_type=model["model_type"],
-            source_file_id=input_file_id,
-            processed_cloud_id=processed_cloud_id,
-            crs_epsg=input_file["crs_epsg"],
-            resolution_m=model["resolution_m"],
-            model_file_path=model["model_file_path"],
-            generation_params=model["generation_params"],
-        )
-        surface_model_ids.append(model_id)
-        for layer in model["layers"]:
-            await create_terrain_layer(
+        surface_model_ids.append(
+            await _upsert_surface_model(
                connection,
-                surface_model_id=model_id,
-                layer_name=layer["layer_name"],
-                geometry_type=layer["geometry_type"],
-                layer_file_path=layer["file_path"],
-                file_format=layer["file_format"],
-                file_size_mb=None,
-                statistics=None,
+                existing,
+                project_id=project_id,
+                model=model,
+                source_file_id=input_file_id,
+                processed_cloud_id=processed_cloud_id,
+                crs_epsg=input_file["crs_epsg"],
+            )
+        )
+    # 이번 결과에 없는 옛 행 — terrain_layers 는 CASCADE · routes.surface_model_id 는 SET NULL.
+    stale_ids = [model_id for model_id, _ in existing.values()]
+    if stale_ids:
+        async with connection.cursor() as cursor:
+            await cursor.execute(
+                "DELETE FROM surface_models WHERE id IN ("
+                + ", ".join(["%s"] * len(stale_ids))
+                + ")",
+                tuple(stale_ids),
            )
    return surface_model_ids

@@ -433,36 +527,19 @@ async def save_sheet_surface_models(
        row = await cursor.fetchone()
    source_file_id, processed_cloud_id, crs_epsg = row if row else (None, None, None)

+    # 같은 파일 · 같은 지문이면 id 를 살린다 — 다른 행은 건드리지 않는다.
+    existing = await _existing_models(connection, project_id)
    model_ids: list[int] = []
    for model in models:
-        model_path = model.get("model_file_path")
-        if model_path:
-            async with connection.cursor() as cursor:
-                await cursor.execute(
-                    "DELETE FROM surface_models WHERE project_id = %s AND model_file_path = %s",
-                    (str(project_id), model_path),
-                )
-        model_id = await create_surface_model(
-            connection,
-            project_id=project_id,
-            model_type=model["model_type"],
-            source_file_id=source_file_id,
-            processed_cloud_id=processed_cloud_id,
-            crs_epsg=crs_epsg,
-            resolution_m=model["resolution_m"],
-            model_file_path=model_path,
-            generation_params=model["generation_params"],
-        )
-        model_ids.append(model_id)
-        for layer in model.get("layers", []):
-            await create_terrain_layer(
+        model_ids.append(
+            await _upsert_surface_model(
                connection,
-                surface_model_id=model_id,
-                layer_name=layer["layer_name"],
-                geometry_type=layer["geometry_type"],
-                layer_file_path=layer["file_path"],
-                file_format=layer["file_format"],
-                file_size_mb=None,
-                statistics=None,
+                existing,
+                project_id=project_id,
+                model=model,
+                source_file_id=source_file_id,
+                processed_cloud_id=processed_cloud_id,
+                crs_epsg=crs_epsg,
            )
+        )
    return model_ids
@@ -15,29 +15,39 @@
 from __future__ import annotations

 import asyncio
+import copy
+import hashlib
+import json
 import logging
 import math
+import os
+import pickle
 from typing import Any
 from uuid import UUID

 from fastapi import APIRouter, Depends
 from fastapi.responses import JSONResponse

-from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
-from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import STAGES, drainage_dir
+from B05_Profile.B05_Profile_Structures_Schema import DEFINITION_PATH, structure_type_map
 from B06_Section.B06_Section_Engine_Culvert import pipe_wall_min_height_m
 from common_util.common_util_auth import get_optional_session
 from common_util.common_util_drainage_context import DrainageContext, load_drainage_context
-from common_util.common_util_drainage_detail import DrainageDetail, build_detail
+from common_util.common_util_drainage_detail import (
+    DrainageDetail,
+    build_detail,
+    pipes_from_chainages,
+)
 from common_util.common_util_drainage_pipes import (
-    PIPE_FACILITY_PIPE,
    PIPE_SOURCE_USER,
    PipePoint,
+    analysis_route_signature,
    carry_facility_attributes,
    carry_indices,
    clear_pipe_points,
    load_pipe_points,
    parse_pipe_points,
+    pipe_signature,
    route_signature,
    save_detail_basins,
    save_pipe_points,
@@ -47,6 +57,7 @@ from common_util.common_util_route_geometry import StructureCandidate
 from config.config_system import (
    DRAINAGE_PIPE_MAX_SPACING_M,
    DRAINAGE_PIPE_MIN_SPACING_M,
+    DRAINAGE_RAINFALL_FILENAME,
    FORD_MANNING_N,
    FORD_SLOPE,
 )
@@ -55,6 +66,10 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/projects", tags=["B04 Surface Drainage Basins"])

 _NO_ANALYSIS = "배수유역 분석 결과가 없습니다. [유역 분석]을 먼저 실행하세요."
+# 노선 지문이 분석 때와 다름(옛 저장분은 지문 없음) — 옛 격자로 나누면 틀린 유역이 나온다(342-3).
+STALE_MESSAGE = (
+    "배수유역 낡음 — 노선이 분석 때와 다릅니다. 유역을 다시 분석해야 세부 유역을 나눕니다."
+)


 def _build(
@@ -73,7 +88,7 @@ def _build(
        return None, []
    if points is not None:
        _retag(detail.pipes, points)
-    rebuilt = [PipePoint(chainage_m=pipe.chainage_m, source=pipe.reason) for pipe in detail.pipes]
+    rebuilt = _rebuilt(detail)
    # 원본에 없던 자리 = 이번에 새로 놓인 관(자동 배치 · 간격 보충) — 등록부 기본값을 받을 관
    fresh = [
        pick is None
@@ -84,13 +99,105 @@ def _build(
    # 계산기는 chainage만 다뤄 시설 종류·구간·옵션이 사라진다 — 원본에서 되붙인다
    # (2026-08-17 계곡 통과 시설: 배관/BOX암거/물넘이/세월교).
    carried = carry_facility_attributes(rebuilt, points or [])
-    # 되붙인 뒤에 추천을 얹는다 — 순서가 뒤바뀌면 승계된 사용자 값이 추천에 덮인다.
-    _apply_recommendations(detail, carried)
+    # 관경 · 시설 자동 지정 없음 — 10-16 사용자 지시로 끔 · 제안(recommended_*)만 응답에 실음.
+    # 새 관은 아래 등록부 기본값(구조물정의 배관 관경)을 받는다.
    flows = {round(basin.chainage_m, 2): basin.design_flow_m3s for basin in detail.basins}
    _apply_new_defaults([point for point, new in zip(carried, fresh) if new], flows)
    return detail, carried


+def _rebuilt(detail: DrainageDetail) -> list[PipePoint]:
+    return [PipePoint(chainage_m=pipe.chainage_m, source=pipe.reason) for pipe in detail.pipes]
+
+
+def _build_stale(context: DrainageContext, points: list[PipePoint]) -> DrainageDetail:
+    """낡은 분석 — 세부 유역 없이 관 자리만(지금 노선 위) · 시설 속성은 넘겨받은 그대로."""
+    to_lonlat = context.to_lonlat
+    detail = DrainageDetail(
+        route_lonlat=[list(to_lonlat(vertex.x, vertex.y)) for vertex in context.vertices],
+        pipes=pipes_from_chainages(context.vertices, [point.chainage_m for point in points]),
+    )
+    _retag(detail.pipes, points)
+    return detail
+
+
+# 세부 유역 저장본(342-3 · 부록 4 R15) — 열쇠 = 관 목록 · 노선(평면 · 종단 Z) · 분석 산출물 ·
+# 강우표 · 구조물 정의 자국. 같으면 다시 나누지 않는다 — 메모리 최근 몇 벌 + [저장] 때 파일 한 벌.
+_DETAIL_CACHE: dict[str, tuple[DrainageDetail, list[PipePoint]]] = {}
+_DETAIL_CACHE_SIZE = 4
+DETAIL_CACHE_FILENAME = "04_detail_cache.pkl"
+
+
+def _detail_key(context: DrainageContext, points: list[PipePoint] | None) -> str:
+    directory = drainage_dir(context.stored_path)
+    digest = hashlib.sha1(usedforsecurity=False)
+    listed = None if points is None else [point.as_dict() for point in points]
+    digest.update(json.dumps(listed, sort_keys=True, ensure_ascii=False).encode())
+    for vertex in context.vertices:
+        digest.update(f"{vertex.x:.3f},{vertex.y:.3f},{vertex.z:.3f};".encode())
+    routing = STAGES["road_routing"]
+    for path in (
+        directory / f"{routing}_road_routing.npz",
+        directory / f"{routing}_road_routing.geojson",
+        directory / f"{STAGES['primary_region']}_primary_region.geojson",
+        directory / DRAINAGE_RAINFALL_FILENAME,
+        DEFINITION_PATH,
+    ):
+        try:
+            info = os.stat(path)
+            digest.update(f"{info.st_mtime_ns}:{info.st_size};".encode())
+        except OSError:
+            digest.update(b"-;")
+    return digest.hexdigest()
+
+
+def _cache_path(stored_path: str):
+    return drainage_dir(stored_path) / DETAIL_CACHE_FILENAME
+
+
+def _remember(key: str, value: tuple[DrainageDetail, list[PipePoint]]) -> None:
+    _DETAIL_CACHE.pop(key, None)
+    _DETAIL_CACHE[key] = value
+    while len(_DETAIL_CACHE) > _DETAIL_CACHE_SIZE:
+        _DETAIL_CACHE.pop(next(iter(_DETAIL_CACHE)))
+
+
+def _build_cached(
+    context: DrainageContext, points: list[PipePoint] | None
+) -> tuple[DrainageDetail | None, list[PipePoint]]:
+    """`_build` + 저장본 — 관 목록은 사본으로 돌려준다(저장이 좌표를 채워도 저장본은 그대로)."""
+    key = _detail_key(context, points)
+    hit = _DETAIL_CACHE.get(key)
+    if hit is None:
+        try:
+            with _cache_path(context.stored_path).open("rb") as file:
+                saved = pickle.load(file)
+            hit = saved["value"] if saved.get("key") == key else None
+        except (OSError, pickle.PickleError, EOFError, AttributeError, KeyError, TypeError):
+            hit = None
+    if hit is None:
+        detail, carried = _build(context.stored_path, context, points)
+        if detail is None:
+            return None, []
+        hit = (detail, carried)
+    _remember(key, hit)
+    return hit[0], copy.deepcopy(hit[1])
+
+
+def _save_detail_cache(
+    context: DrainageContext, points: list[PipePoint] | None, detail: DrainageDetail, carried
+) -> None:
+    """[저장] 뒤 — 다음 GET 이 읽을 관 목록(저장분) 열쇠로 파일 한 벌을 남긴다."""
+    key = _detail_key(context, points)
+    value = (detail, copy.deepcopy(carried))
+    _remember(key, value)
+    try:
+        with _cache_path(context.stored_path).open("wb") as file:
+            pickle.dump({"key": key, "value": value}, file)
+    except OSError:
+        logger.warning("배수유역: 세부 유역 저장본을 남기지 못했습니다(%s).", context.stored_path)
+
+
 # 관 보호공 · 유입 집수정 높이 칸 — 새 관 기본 = 관경 + 0.5(139-1) · 유입 흙막이만 + 0.7(211-1)
 PIPE_HEIGHT_KEYS = ("inlet_revet_height_m", "outlet_revet_height_m", "inlet_basin_height_m")

@@ -105,7 +212,7 @@ def _float(value: Any) -> float:
 def _apply_new_defaults(
    points: list[PipePoint], flows: dict[float, float | None] | None = None
 ) -> None:
-    """새로 놓인 관에만 등록부 기본값을 **빈 키에** 채움(PLAN 46-7 · 추천 뒤 — 추천 관경이 이김).
+    """새로 놓인 관에만 등록부 기본값을 **빈 키에** 채움(PLAN 46-7).

    시설이 생길 때 기본값을 칸에 실제 값으로 넣어 두고 사용자가 고침 — 이미 있던 관의 빈 칸은 그대로
    (옛 프로젝트 빈칸 = 미확정 유지 · 사용자 결정 2026-10-05).
@@ -176,45 +283,13 @@ def _retag(pipes: list[StructureCandidate], points: list[PipePoint]) -> None:
        pipe.reason = source or PIPE_SOURCE_USER


-def _apply_recommendations(detail: DrainageDetail, points: list[PipePoint]) -> None:
-    """유역별 추천(시설 종류·관경)을 관 지점 옵션의 **빈칸에만** 채운다.
-
-    "초기 전처리 계산에서 해당값으로 설정"(2026-08-17 사용자 지시) — 자동 배치된 관이
-    유역 유량에 맞는 규격을 처음부터 갖고 있어야 폼을 열지 않아도 하류가 값을 받는다.
-
-    덮어쓰지 않는 두 경우: ① 사용자가 직접 놓거나 옮긴 관(`source == "user"`) ②
-    그 옵션 키가 이미 있는 관. 설계자가 고른 값을 재계산이 되돌리면 안 되기 때문이다.
-    """
-    by_chainage = {round(basin.chainage_m, 2): basin for basin in detail.basins}
-    for point in points:
-        if point.source == PIPE_SOURCE_USER:
-            continue
-        basin = by_chainage.get(round(point.chainage_m, 2))
-        if basin is None:
-            continue
-        options = dict(point.options or {})
-        # 시설 종류는 아직 손대지 않은 기본 배관일 때만 추천으로 바꾼다.
-        if (
-            point.facility == PIPE_FACILITY_PIPE
-            and basin.recommended_facility != PIPE_FACILITY_PIPE
-        ):
-            point.facility = basin.recommended_facility
-        if (
-            point.facility == PIPE_FACILITY_PIPE
-            and basin.recommended_diameter_mm is not None
-            and "pipe_diameter_mm" not in options
-        ):
-            options["pipe_diameter_mm"] = basin.recommended_diameter_mm
-        if options:
-            point.options = options
-
-
 def _payload(
    project_id: UUID,
    context: DrainageContext,
    detail: DrainageDetail,
    points: list[PipePoint],
    saved: bool,
+    stale: bool = False,
 ) -> dict[str, Any]:
    """관 목록과 세부유역을 화면 좌표(WGS84)로 정리한다.

@@ -222,7 +297,12 @@ def _payload(
    응답이 151KB → 100KB(압축까지 얹으면 22KB)가 된다(2026-09-06 실측). 화면이 그리는 값이고
    다시 계산에 넣지 않으므로 줄여도 결과가 안 갈린다.
    """
-    return round_floats(_raw_payload(project_id, context, detail, points, saved), LONLAT_DIGITS)
+    payload = _raw_payload(project_id, context, detail, points, saved)
+    # 지문(342-3) — 화면이 지금 노선 · 관 자리와 견줌 · 낡으면 까닭 글(세부 유역 없음)
+    payload["route_signature"] = route_signature(context.vertices)
+    payload["pipe_signature"] = pipe_signature([pipe.chainage_m for pipe in detail.pipes])
+    payload["stale"] = STALE_MESSAGE if stale else None
+    return round_floats(payload, LONLAT_DIGITS)


 def _raw_payload(
@@ -355,10 +435,13 @@ def _basin_features(


 async def _resolve(
-    project_id: UUID, payload: dict[str, Any] | None, *, use_stored: bool
-) -> tuple[DrainageContext, DrainageDetail, list[PipePoint], bool] | JSONResponse:
+    project_id: UUID, payload: dict[str, Any] | None, *, use_stored: bool, fresh: bool = False
+) -> tuple[DrainageContext, DrainageDetail, list[PipePoint], bool, bool] | JSONResponse:
    """요청 본문 → 저장분 → 자동 생성 순으로 관 목록을 정하고 세부유역까지 계산한다.

+    분석의 노선 지문이 지금 노선과 다르면(옛 저장분 = 지문 없음) 나누지 않고 관 자리만 낡음으로
+    돌려준다 — 관 목록도 없으면(자동 생성 자리) 409(342-3). 마지막 값 = 낡음.
+
    빈 목록(`[]`)은 「관 0개」로 그대로 쓴다 — 자동 생성은 요청에 칸이 없고 저장분도 없을(None) 때만(312).
    """
    context, reason = await load_drainage_context(project_id)
@@ -375,12 +458,25 @@ async def _resolve(
        load_pipe_points(context.stored_path, signature, context.vertices) if use_stored else None
    )
    points = requested if requested is not None else stored
+    saved = requested is None and stored is not None

-    result = await asyncio.to_thread(_build, context.stored_path, context, points)
+    if analysis_route_signature(context.stored_path) != signature:
+        if points is None:
+            return JSONResponse(
+                status_code=409, content={"status": "stale", "message": STALE_MESSAGE}
+            )
+        detail = _build_stale(context, points)
+        return context, detail, carry_facility_attributes(_rebuilt(detail), points), saved, True
+
+    # [저장]은 정본으로 다시 셈 — 저장본은 GET · POST 만
+    if fresh:
+        result = await asyncio.to_thread(_build, context.stored_path, context, points)
+    else:
+        result = await asyncio.to_thread(_build_cached, context, points)
    detail, resolved = result
    if detail is None:
        return JSONResponse(status_code=404, content={"status": "error", "message": _NO_ANALYSIS})
-    return context, detail, resolved, requested is None and stored is not None
+    return context, detail, resolved, saved, False


@router.get("/{project_id}/drainage/pipe-points", response_model=None)
@@ -389,8 +485,8 @@ async def get_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
    resolved = await _resolve(project_id, None, use_stored=True)
    if isinstance(resolved, JSONResponse):
        return resolved
-    context, detail, points, saved = resolved
-    return _payload(project_id, context, detail, points, saved)
+    context, detail, points, saved, stale = resolved
+    return _payload(project_id, context, detail, points, saved, stale)


@router.post("/{project_id}/drainage/detail-basins", response_model=None)
@@ -405,8 +501,8 @@ async def post_detail_basins(
    resolved = await _resolve(project_id, payload, use_stored=False)
    if isinstance(resolved, JSONResponse):
        return resolved
-    context, detail, points, _ = resolved
-    return _payload(project_id, context, detail, points, saved=False)
+    context, detail, points, _, stale = resolved
+    return _payload(project_id, context, detail, points, False, stale)


@router.delete("/{project_id}/drainage/pipe-points", response_model=None)
@@ -424,8 +520,8 @@ async def delete_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
    resolved = await _resolve(project_id, None, use_stored=False)
    if isinstance(resolved, JSONResponse):
        return resolved
-    context, detail, points, _ = resolved
-    return _payload(project_id, context, detail, points, saved=False)
+    context, detail, points, _, stale = resolved
+    return _payload(project_id, context, detail, points, False, stale)


@router.put("/{project_id}/drainage/pipe-points", response_model=None)
@@ -435,22 +531,25 @@ async def put_pipe_points(
    session: dict[str, Any] | None = Depends(get_optional_session),  # 가져온이(142-2)
 ) -> dict[str, Any] | JSONResponse:
    """관 지점을 정본으로 확정하고 세부유역 산출물까지 함께 남긴다(모델 확정 시점)."""
-    resolved = await _resolve(project_id, payload, use_stored=True)
+    resolved = await _resolve(project_id, payload, use_stored=True, fresh=True)
    if isinstance(resolved, JSONResponse):
        return resolved
-    context, detail, points, _ = resolved
+    context, detail, points, _, stale = resolved

    signature = route_signature(context.vertices)
    # 좌표를 같이 남긴다 — 다른 선(B05 최적 경로)으로 읽어도 그 자리에 되놓는다.
    saved = await asyncio.to_thread(
        save_pipe_points, context.stored_path, signature, points, context.vertices
    )
-    await asyncio.to_thread(
-        save_detail_basins,
-        context.stored_path,
-        _basin_features(context, detail, points),
-        context.crs,
-    )
+    if not stale:  # 낡은 분석 — 관 자리 · 속성만 정본으로 · 세부 유역 산출물은 안 씀
+        await asyncio.to_thread(
+            save_detail_basins,
+            context.stored_path,
+            _basin_features(context, detail, points),
+            context.crs,
+        )
+        stored = load_pipe_points(context.stored_path, signature, context.vertices)
+        await asyncio.to_thread(_save_detail_cache, context, stored, detail, points)
    # 지정한 집계 항목을 프로젝트에 복사 · 안 쓰는 복사본 지움(PLAN 21-3 · 실패해도 저장 유지)
    from common_util.common_util_storage import resolve_stored_project_path
    from M02_MasterTemplete.M02_Item_Snapshot import importer, safe_sync
@@ -458,7 +557,7 @@ async def put_pipe_points(
    failed = await asyncio.to_thread(
        safe_sync, resolve_stored_project_path(context.stored_path), importer(session)
    )
-    result = _payload(project_id, context, detail, points, saved=True)
+    result = _payload(project_id, context, detail, points, True, stale)
    result["saved_count"] = saved
    result["snapshot_failed"] = failed  # 못 만든 구조물 조합 복사본(142-2) — [저장] 알림
    return result
@@ -43,6 +43,7 @@ from B04_PreProcess.B04_PreProcess_Router_Watershed_Output import (
 )
 from B05_Profile.B05_Profile_Repository import get_surface_crs_epsg
 from common_util.common_util_crs import resolve_project_crs
+from common_util.common_util_drainage_pipes import pipe_signature, route_signature
 from common_util.common_util_route_geometry import (
    StructureCandidate,
    find_planned_route_file,
@@ -419,6 +420,9 @@ async def get_primary_region(
        "schema_version": RESPONSE_SCHEMA_VERSION,
        "project_id": str(project_id),
        "route_source": prepared["route_source"],
+        # 지문(342-3) — 노선 · 기본 관 자리가 지금과 다르면 이 결과는 낡음(세부 유역 안 나눔).
+        "route_signature": route_signature(prepared["vertices"]),
+        "pipe_signature": pipe_signature([pipe.chainage_m for pipe in preview.pipes]),
        "radius_m": region.radius_m,
        # 채택된 상류 세류망 = 1차 영역의 기준선.
        "upstream_lines": [_line_lonlat(line, to_lonlat) for line in region.split.upstream],
@@ -0,0 +1,35 @@
+# B04_PreProcess 지침
+
+갈래 지도(앞머리 → 하는 일):
+
+- `UI_Map*` 2D 지도(그리기 · 겹침 · 층 · 보기) · `UI_TerrainViewer*` 3D 지형 · 등고선 · `UI_Viewer` · `UI_Camera`
+- `UI_Flow*` · `UI_Basins` · `UI_Watershed` 배수 흐름 · 유역 · `UI_RouteSamples` 노선 표본 · `UI_Page` 페이지 · `Api_Fetch` 서버 요청
+- `Engine_Filter_*` 지면 필터 · `Engine_Ground` · `Engine_ModelBuild` · `Engine_ModelContext` · `Engine_Smooth` 지표면 모델
+- `Engine_Sheet*` · `Engine_MapSheet` 수치지형도 · `Engine_Contour` 등고선 · `Engine_GisVector` · `Engine_VWorld` 국가 GIS
+- `Engine_Watershed_*` 배수유역 · `Engine_Pipeline` · `Engine` 전처리 차례 · `Router_*` · `Repository` · `Service` · `Schema`
+- `logic/` 순수 계산 TS
+
+① 짝(ts ↔ py · 거울 시험):
+
+- `Router_Basins.py` ford_height_m ↔ `A00_Common/design/design_structure_ford_calc.ts` fordHeightM · `test_65_ford_defaults.py`
+- 배수유역 · 지표면 모델 엔진 = 서버 단독(③) · 지형 표본 = `common_util_surface_sampler` 하나
+
+서버 Node 묶음:
+
+- 입구 없음 · `Api_Fetch.ts` 가 `build:b07-extent` 에 끌려감 — 모듈 맨 위에 DOM · window · sessionStorage 금지
+
+남의 페이지가 부르는 파일(고치면 부르는 쪽 시험도):
+
+- ts — `Api_Fetch`(A00 · B05 · B06 · B08 · B11) · `UI_MapRender` · `UI_MapRender_Prepare` · `UI_MapOverlays` · `UI_FlowRamp` · `UI_FlowArrows` · `UI_Camera` · `UI_RouteSamples`(B05) · `UI_Page`(A00)
+- py — `Router_Watershed` · `Router_Basins` · `Router_GIS` · `Router` · `Service` · `Repository` · `Engine_Watershed_*` · `Engine_VWorld` · `Engine_Extent` · `Engine_SheetSurface` · `Engine_Structurize`
+
+logic/ 규칙:
+
+- DOM · three · fetch · sessionStorage · window 금지 · 순수 함수만
+- 화면 파일이 logic 을 부름 · logic 은 화면 파일을 안 부름(logic · common_util 만)
+- 이름 앞머리 그대로(`logic/B04_PreProcess_Xxx.ts`) · 파이썬 짝이면 두 파일 머리에 짝 표시 + 거울 시험
+
+금지:
+
+- 파이썬 파일을 하위 폴더로 옮기지 않음(절대 import · `Engine_GisVector.py` 뿌리 경로)
+- 실제 파일 다시 올리기 · 전처리 다시 돌리기로 시험하지 않음 — 서버 함수 시험으로