perf(영구저장): 343-2 도엽 3D 면 열쇠 · 지표면 모델 id 유지 · 비용면 메모리 기억 · 추천 노선 열쇠 · 횡단 상세 ETag

- 도엽 3D 면 — 결과 옆 dtm_sheet_{방식}.key.json(등고선 이름:mtime:크기 · 노선 · CRS · 격자 · 여유 · 스무딩 · 화면 상자 · ALGO) 같으면 건너뜀
- 지표면 모델 — (방식 · 필터 · 파일) + 파일 지문 같으면 행 id 살려 고쳐 씀 · 이번 결과에 없는 옛 행만 지움
- 비용면 · 계류 메모리 기억(비용면 서명 · 계류 파일 지문 · 좌표계) · 추천 열쇠 = 서명(배열 통째 해시 없앰)
- 추천 저장본 종류(갈래 · 지형)당 최근 2벌 · 꺼내면 최근으로
- 횡단 상세 GET — 측점 파일 안 지움(valid_cross_names) · 본문 해시 ETag 304
- 시험 test_343_2_persist_keys 13건

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014utrjfoW831Vc5wJUd13kL

B04_PreProcess/B04_PreProcess_Engine.py

@@ -128,6 +128,15 @@ def _structured_stats(structured: Any, structured_path: Path) -> tuple[dict[str,
     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:
@@ -605,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,
                 }

B04_PreProcess/B04_PreProcess_Engine_SheetSurface.py

@@ -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(

B04_PreProcess/B04_PreProcess_Repository.py

@@ -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

B05_Profile/B05_Profile_Engine_RouteInitial.py

@@ -109,12 +109,19 @@ MODE_WEIGHTS = {
 # 후보 묶음은 갈래와 무관 — 같은 입력이면 다시 짜지 않음(최근 몇 벌만 · 프로세스 메모리)
 POOL_CACHE_SIZE = 4
 _POOLS: dict[bytes, Pool] = {}
+# 비용면 · 계류 기억(요청 사이) — 열쇠 = 프로젝트 · 비용면 서명 · 계류 파일 지문 · 좌표계
+TERRAIN_CACHE_SIZE = 4
+_TERRAINS: dict[str, tuple[Terrain, list[list[tuple[float, float]]]]] = {}
 
 
 def load_terrain_and_streams(
     project_root: Path, selection: dict[str, Any]
 ) -> tuple[Terrain, list[list[tuple[float, float]]]]:
-    """지반 격자(탐색 비용면 캐시 그대로)와 이격을 따질 계류(사업지 좌표계)를 읽는다."""
+    """지반 격자(탐색 비용면 캐시 그대로)와 이격을 따질 계류(사업지 좌표계)를 읽는다.
+
+    같은 입력(비용면 서명 · 계류 파일 지문 · 좌표계)이면 메모리 기억을 그대로 돌려준다 —
+    점 놓을 때마다 npz · geojson 을 다시 읽지 않는다. 돌려준 배열은 고치지 않는다.
+    """
     from pyproj import Transformer
 
     from B04_PreProcess.B04_PreProcess_Router_Watershed import (
@@ -122,20 +129,49 @@ def load_terrain_and_streams(
         _load_features,
         _reproject_features,
     )
-    from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR, _load_or_build_cost_surface
+    from B05_Profile.B05_Profile_Engine_Solver import (
+        _MODELS_SUBDIR,
+        _cost_surface_signature,
+        _load_or_build_cost_surface,
+    )
     from common_util.common_util_crs import resolve_project_crs
 
-    x, y, z, valid, *_ = _load_or_build_cost_surface(
-        project_root,
-        project_root / _MODELS_SUBDIR,
-        str(selection.get("source_filter")),
-        str(selection.get("method") or "dtm"),
-        bool(selection.get("smooth")),
+    filter_key = str(selection.get("source_filter"))
+    method = str(selection.get("method") or "dtm")
+    smooth = bool(selection.get("smooth"))
+    models_dir = project_root / _MODELS_SUBDIR
+    stream_path = project_root / "B04_PreProcess" / "processed" / STREAM_FILE
+    crs = resolve_project_crs(project_root)
+    try:
+        stat = stream_path.stat()
+        stream_stamp = f"{stat.st_mtime_ns}:{stat.st_size}"
+    except OSError:
+        stream_stamp = "-"
+    signature = "|".join(
+        (
+            filter_key,
+            _cost_surface_signature(models_dir, filter_key, method, smooth),
+            stream_stamp,
+            str(crs),
+        )
     )
-    to_metric = Transformer.from_crs("EPSG:4326", resolve_project_crs(project_root), always_xy=True)
-    features = _load_features(project_root / "B04_PreProcess" / "processed", STREAM_FILE)
+    memo_key = f"{project_root}|{signature}"
+    if memo_key in _TERRAINS:
+        return _TERRAINS[memo_key]
+
+    x, y, z, valid, *_ = _load_or_build_cost_surface(
+        project_root, models_dir, filter_key, method, smooth
+    )
+    to_metric = Transformer.from_crs("EPSG:4326", crs, always_xy=True)
+    features = _load_features(stream_path.parent, STREAM_FILE)
     streams = stream_lines(_reproject_features(features, to_metric))
-    return Terrain(np.asarray(x), np.asarray(y), np.asarray(z), np.asarray(valid)), streams
+    terrain = Terrain(
+        np.asarray(x), np.asarray(y), np.asarray(z), np.asarray(valid), signature=signature
+    )
+    while len(_TERRAINS) >= TERRAIN_CACHE_SIZE:
+        _TERRAINS.pop(next(iter(_TERRAINS)))
+    _TERRAINS[memo_key] = (terrain, streams)
+    return terrain, streams
 
 
 def _choice_path(project_root: Path) -> Path:
@@ -378,8 +414,16 @@ def _pool_key(
     streams: list[list[tuple[float, float]]],
     *numbers: float,
 ) -> bytes:
-    """후보 묶음을 가를 입력 지문 — 지형 · 예상노선 · 계류 · 제약 값."""
+    """후보 묶음을 가를 입력 지문 — 지형 · 예상노선 · 계류 · 제약 값.
+
+    지형에 입력 지문(`signature` — 비용면 서명 + 계류 파일)이 있으면 배열 통째 해시 대신 씀.
+    """
     digest = hashlib.blake2b(digest_size=16)
+    if terrain.signature:
+        digest.update(terrain.signature.encode())
+        digest.update(np.ascontiguousarray(np.asarray(expected)).tobytes())
+        digest.update(repr(numbers).encode())
+        return digest.digest()
     for array in (terrain.x, terrain.y, terrain.z, terrain.valid, np.asarray(expected)):
         digest.update(np.ascontiguousarray(array).tobytes())
     digest.update(repr((streams, numbers)).encode())

B05_Profile/B05_Profile_Engine_RouteInitial_Search.py

@@ -68,6 +68,8 @@ class Terrain:
     y: np.ndarray  # 행 좌표(오름차순)
     z: np.ndarray  # [행, 열] 표고
     valid: np.ndarray  # [행, 열] 지형 있음
+    # 지형 · 계류 입력 지문(비용면 서명 + 계류 파일) — 있으면 열쇠가 배열 통째 해시를 대신함
+    signature: str | None = None
 
 
 class NoRoute(ValueError):

B05_Profile/B05_Profile_Engine_RouteInitial_Store.py

@@ -1,7 +1,7 @@
 """계획노선 추천 결과 영구 저장(PLAN 327 · 요구 224-3).
 
 같은 입력이면 같은 결과 — 셈한 추천을 입력 지문 키로 프로젝트 폴더에 남기고, 같은 입력이 오면
-셈하지 않고 꺼낸다. 바뀐 입력만 새로 셈해 더한다(덮어쓰기 없음 · 상한 없음).
+셈하지 않고 꺼낸다. 바뀐 입력만 새로 셈해 더한다(덮어쓰기 없음 · 종류마다 최근 `KEEP_PER_KIND` 벌).
 
 층 = 서버 계산 캐시(데이터 3층 밖 · 사용자 편집값 아님 — 자동저장 금지와 무관). 자리
 `B05_Profile/route_recommend/<키>.json` — initial_snapshot 대상(`B05_Profile/route`) 밖이라
@@ -33,6 +33,8 @@ from common_util.common_util_json import atomic_write_json
 
 ALGO_VERSION = 2
 STORE_DIR = "B05_Profile/route_recommend"
+# 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음
+KEEP_PER_KIND = 2
 # 업로드 체인 끝에 미리 셈할 기본 다섯 알약(화면 차례) · 추적은 묶음이 선 뒤라 `similar` 가 채워짐
 DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")
 
@@ -85,12 +87,17 @@ def is_default(
 
 def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
     """저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None."""
+    path = store_dir(project_root) / f"{key}.json"
     try:
-        data = json.loads((store_dir(project_root) / f"{key}.json").read_text(encoding="utf-8"))
+        data = json.loads(path.read_text(encoding="utf-8"))
     except (OSError, ValueError):
         return None
     if not isinstance(data, dict) or data.get("version") != ALGO_VERSION:
         return None
+    try:
+        path.touch()  # 최근에 쓴 것 = 정리 때 남김
+    except OSError:
+        pass
     return data
 
 
@@ -118,9 +125,32 @@ def write_stored(
             "result": result,
         },
     )
+    prune_stored(project_root, request)
     return True
 
 
+def _kind(request: Any) -> tuple[Any, Any]:
+    request = request if isinstance(request, dict) else {}
+    return request.get("mode"), request.get("terrain")
+
+
+def prune_stored(project_root: Path, request: dict[str, Any]) -> int:
+    """같은 종류(갈래 · 지형) 저장본을 최근 `KEEP_PER_KIND` 벌만 남기고 지운다. 지운 수."""
+    kind = _kind(request)
+    same: list[tuple[int, Path]] = []
+    for path in store_dir(project_root).glob("*.json"):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+            if _kind(data.get("request")) == kind:
+                same.append((path.stat().st_mtime_ns, path))
+        except (OSError, ValueError, AttributeError):
+            continue
+    same.sort(reverse=True)
+    for _, path in same[KEEP_PER_KIND:]:
+        path.unlink(missing_ok=True)
+    return max(0, len(same) - KEEP_PER_KIND)
+
+
 @asynccontextmanager
 async def pool_lock(pool: str) -> AsyncIterator[None]:
     # ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로

B05_Profile/B05_Profile_Engine_Sections.py

@@ -141,18 +141,27 @@ def find_cross_file(cross_dir: Path, chainage_m: float) -> Path:
     return path if path.exists() or not legacy.exists() else legacy
 
 
-def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]:
-    """stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다.
+def valid_cross_names(cross_dir: Path, stations: list[Any]) -> set[str]:
+    """stations 에 맞는 cross_*.json 파일명 집합 — 읽기 전용(지우지 않음 · GET 이 씀).
 
-    측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고
-    빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다).
+    측점 정보가 비면 빈 집합(호출부는 빈 집합이면 필터를 생략한다).
     """
-    # 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가려야 안 지워짐
-    valid = {
+    # 옛 프로젝트는 소수 측점 파일이 정수 m 이름이라 찾기 규칙(find_cross_file)으로 가림
+    return {
         find_cross_file(cross_dir, station["chainage_m"]).name
         for station in stations
         if isinstance(station, dict) and isinstance(station.get("chainage_m"), (int, float))
     }
+
+
+def prune_stale_cross_files(cross_dir: Path, stations: list[Any]) -> set[str]:
+    """stations에 없는 잔재 cross_*.json을 삭제하고 유효 파일명 집합을 반환한다.
+
+    쓰는 쪽(측점 만들기 · 측점 목록 고침)만 부른다 — 조회(GET)는 `valid_cross_names`.
+    측점 정보가 비어 있으면 오삭제를 피하기 위해 아무것도 지우지 않고
+    빈 집합을 반환한다(호출부는 빈 집합이면 필터를 생략한다).
+    """
+    valid = valid_cross_names(cross_dir, stations)
     if not valid:
         return valid
     for path in cross_dir.glob("cross_*.json"):

B06_Section/B06_Section_Router.py

@@ -1,6 +1,7 @@
 """B06 종횡단 생성 FastAPI 라우터."""
 
 import asyncio
+import hashlib
 import json
 import logging
 from pathlib import Path
@@ -8,15 +9,15 @@ from typing import Any
 from uuid import UUID
 
 import aiomysql
-from fastapi import APIRouter, Depends
-from fastapi.responses import JSONResponse
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import JSONResponse, Response
 
 from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
 from B05_Profile.B05_Profile_Engine_Grade_Profile import rebuild_alignment_profile
 from B05_Profile.B05_Profile_Engine_Sections import (
-    prune_stale_cross_files,
     relabel_pipe_stations,
     run_section_generation,
+    valid_cross_names,
 )
 from B05_Profile.B05_Profile_Engine_Sections_Core import SectionGenerationOptions
 from B06_Section.B06_Section_Engine_Attach import attach_saved_shapes
@@ -85,6 +86,7 @@ from B06_Section.B06_Section_Schema import (
 )
 from B06_Section.B06_Section_Server_Calc_Prebuild import conversion_factors_for, haul_limits_for
 from common_util.common_util_auth import verify_session
+from common_util.common_util_http_cache import CACHE_CONTROL
 from common_util.common_util_storage import resolve_stored_project_path
 from common_util.common_util_surface_confirmation import get_surface_confirmation_params
 from common_util.common_util_workflow_state import get_workflow_state
@@ -266,7 +268,8 @@ def _read_section_detail(project_root: Path, longitudinal_file_path: str) -> dic
 
     longitudinal = json.loads(longitudinal_path.read_text(encoding="utf-8"))
     stations = longitudinal.get("stations") if isinstance(longitudinal, dict) else None
-    valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else [])
+    # 조회는 지우지 않음 — 잔재는 걸러 읽기만(지우기는 측점을 쓰는 쪽 몫)
+    valid_names = valid_cross_names(cross_dir, stations if isinstance(stations, list) else [])
     cross_sections = [
         json.loads(path.read_text(encoding="utf-8"))
         for path in sorted(cross_dir.glob("cross_*.json"))
@@ -282,6 +285,15 @@ def _read_section_detail(project_root: Path, longitudinal_file_path: str) -> dic
     return {"longitudinal": longitudinal, "cross_sections": cross_sections}
 
 
+def _etag_json_response(request: Request, body: bytes) -> Response:
+    """본문 해시 ETag JSON 응답 — 브라우저가 같은 ETag 를 들고 오면 304 만."""
+    etag = f'"{hashlib.blake2b(body, digest_size=16).hexdigest()}"'
+    headers = {"ETag": etag, "Cache-Control": CACHE_CONTROL}
+    if request.headers.get("if-none-match") == etag:
+        return Response(status_code=304, headers=headers)
+    return Response(content=body, media_type="application/json", headers=headers)
+
+
 def _read_balloon_offsets(row: dict[str, Any]) -> dict[str, list[float]] | None:
     """확정 시 저장해 둔 유토곡선 balloon 위치를 꺼낸다. 없거나 형태가 깨졌으면 None."""
     data = row.get("data")
@@ -307,9 +319,12 @@ def _read_balloon_offsets(row: dict[str, Any]) -> dict[str, list[float]] | None:
 
 @router.get("/{project_id}/sections/{route_id}/detail", response_model=SectionDetailResponse)
 async def get_section_detail(
-    project_id: UUID, route_id: int
-) -> SectionDetailResponse | JSONResponse:
-    """경로의 SVG 렌더링용 종단·횡단 원시 샘플을 반환한다."""
+    project_id: UUID, route_id: int, request: Request
+) -> SectionDetailResponse | JSONResponse | Response:
+    """경로의 SVG 렌더링용 종단·횡단 원시 샘플을 반환한다.
+
+    ETag = 응답 본문 해시 — 내용이 같으면 304(1.5 MB 다시 안 받음 · 브라우저 HTTP 캐시).
+    """
     try:
         # 서로 기다릴 이유가 없는 읽기 셋 — 원격 DB 라 순차로 내면 왕복이 그대로 더해진다
         # (질의 하나 약 12ms, 2026-09-06 실측). 같이 보내 가장 느린 하나의 시간만 쓴다.
@@ -350,7 +365,10 @@ async def get_section_detail(
             project_root,
             standard,
         )
-        return SectionDetailResponse(**detail, balloon_offsets=_read_balloon_offsets(longitudinal))
+        body = SectionDetailResponse(
+            **detail, balloon_offsets=_read_balloon_offsets(longitudinal)
+        ).model_dump_json()
+        return _etag_json_response(request, body.encode("utf-8"))
     except FileNotFoundError as exc:
         return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
     except (OSError, ValueError, json.JSONDecodeError) as exc:

resources/tester/test_343_2_persist_keys.py

@@ -0,0 +1,341 @@
+"""343-2 도엽 3D 면 열쇠 · 지표면 모델 id 유지 · 비용면 · 추천 노선 열쇠 · 횡단 상세 ETag.
+
+같은 입력 두 번째 = 다시 안 셈 · 열쇠 바뀌면 다시 셈 · 지울 때.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import time
+from pathlib import Path
+
+import numpy as np
+import pytest
+from pyproj import Transformer
+
+import B04_PreProcess.B04_PreProcess_Engine_SheetSurface as sheet
+from B04_PreProcess.B04_PreProcess_Engine_SheetMethods import SHEET_METHOD_BUILDERS
+
+EPSG = 5186
+X0, Y0 = 200000.0, 500000.0
+ROUTE = np.array([[X0 - 50.0, Y0], [X0 + 50.0, Y0]], dtype=np.float64)
+
+
+@pytest.fixture()
+def sheet_project(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict:
+    processed = tmp_path / "B04_PreProcess" / "processed"
+    models = tmp_path / "B04_PreProcess" / "models"
+    processed.mkdir(parents=True)
+    models.mkdir(parents=True)
+    to_wgs84 = Transformer.from_crs(f"EPSG:{EPSG}", "EPSG:4326", always_xy=True)
+    features = []
+    for half, elev in ((450.0, 100.0), (300.0, 110.0), (150.0, 120.0)):
+        ring = [
+            list(to_wgs84.transform(x, y))
+            for x, y in (
+                (X0 - half, Y0 - half),
+                (X0 + half, Y0 - half),
+                (X0 + half, Y0 + half),
+                (X0 - half, Y0 + half),
+                (X0 - half, Y0 - half),
+            )
+        ]
+        features.append(
+            {
+                "type": "Feature",
+                "properties": {"등고수치": elev},
+                "geometry": {"type": "LineString", "coordinates": ring},
+            }
+        )
+    (processed / "도엽_등고선.geojson").write_text(
+        json.dumps({"type": "FeatureCollection", "features": features}, ensure_ascii=False),
+        encoding="utf-8",
+    )
+    calls: list[str] = []
+    real = SHEET_METHOD_BUILDERS["tin_sheet"]
+
+    def counting(*args, **kwargs):  # type: ignore[no-untyped-def]
+        calls.append("tin_sheet")
+        return real(*args, **kwargs)
+
+    monkeypatch.setitem(SHEET_METHOD_BUILDERS, "tin_sheet", counting)
+    return {"root": tmp_path, "processed": processed, "models": models, "calls": calls}
+
+
+def _build(project: dict, route: np.ndarray = ROUTE) -> list[dict]:
+    return sheet.build_sheet_surface_model(
+        project["root"], project["processed"], project["models"], route, EPSG, ["tin_sheet"]
+    )
+
+
+def test_도엽_3D_면_같은_입력_두번째는_안_셈(sheet_project: dict) -> None:
+    first = _build(sheet_project)
+    assert sheet_project["calls"] == ["tin_sheet"]
+    assert (sheet_project["models"] / "dtm_sheet_tin_sheet.key.json").is_file()
+    started = time.perf_counter()
+    second = _build(sheet_project)
+    assert time.perf_counter() - started < 0.5
+    assert sheet_project["calls"] == ["tin_sheet"]
+    assert second == first
+    assert first[0]["generation_params"]["file_stamp"]
+
+
+@pytest.mark.parametrize("change", ["route", "contour", "algo", "key_gone", "file_gone"])
+def test_도엽_3D_면_열쇠가_바뀌면_다시_셈(
+    sheet_project: dict, monkeypatch: pytest.MonkeyPatch, change: str
+) -> None:
+    _build(sheet_project)
+    route = ROUTE
+    if change == "route":
+        route = ROUTE + np.array([5.0, 0.0])
+    elif change == "contour":
+        path = sheet_project["processed"] / "도엽_등고선.geojson"
+        stat = path.stat()
+        os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 5_000_000_000))
+    elif change == "algo":
+        monkeypatch.setattr(sheet, "SHEET_SURFACE_ALGO", sheet.SHEET_SURFACE_ALGO + 1)
+    elif change == "key_gone":
+        (sheet_project["models"] / "dtm_sheet_tin_sheet.key.json").unlink()
+    else:
+        (sheet_project["models"] / "dtm_sheet_tin_sheet_preview.glb").unlink()
+    _build(sheet_project, route)
+    assert sheet_project["calls"] == ["tin_sheet", "tin_sheet"]
+
+
+# ── 지표면 모델 id 유지(R2) ─────────────────────────────────────────────────────
+
+
+class _FakeCursor:
+    def __init__(self, db: "_FakeDb") -> None:
+        self.db = db
+        self.rowcount = 0
+        self.lastrowid = 0
+        self._rows: list[tuple] = []
+
+    async def __aenter__(self) -> "_FakeCursor":
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+    async def execute(self, sql: str, params: tuple = ()) -> None:
+        text = " ".join(sql.split())
+        self.db.log.append(text)
+        if text.startswith("SELECT id, model_type"):
+            self._rows = [
+                (row["id"], row["model_type"], row["path"], json.dumps(row["params"]))
+                for row in self.db.models.values()
+            ]
+        elif text.startswith("INSERT INTO surface_models"):
+            self.db.next_id += 1
+            self.lastrowid = self.db.next_id
+            self.db.models[self.lastrowid] = {
+                "id": self.lastrowid,
+                "model_type": params[1],
+                "path": params[7],
+                "params": json.loads(params[8]),
+            }
+        elif text.startswith("UPDATE surface_models"):
+            self.db.models[params[-1]]["params"] = json.loads(params[4])
+        elif text.startswith("DELETE FROM surface_models WHERE id IN"):
+            for model_id in params:
+                self.db.models.pop(model_id, None)
+        elif text.startswith("DELETE FROM surface_models WHERE id ="):
+            self.db.models.pop(params[0], None)
+        elif text.startswith("DELETE FROM surface_models WHERE project_id"):
+            for model_id in [k for k, v in self.db.models.items() if v["path"] == params[1]]:
+                self.db.models.pop(model_id)
+        elif text.startswith("INSERT INTO terrain_layers"):
+            self.db.next_id += 1
+            self.lastrowid = self.db.next_id
+
+    async def fetchall(self) -> list[tuple]:
+        return self._rows
+
+    async def fetchone(self) -> tuple | None:
+        return None
+
+
+class _FakeDb:
+    def __init__(self) -> None:
+        self.log: list[str] = []
+        self.models: dict[int, dict] = {}
+        self.next_id = 100
+
+    def cursor(self, *_: object) -> _FakeCursor:
+        return _FakeCursor(self)
+
+
+def _model(path: str, stamp: str) -> dict:
+    return {
+        "model_type": "dtm",
+        "model_file_path": path,
+        "resolution_m": 1.0,
+        "generation_params": {"source_filter": "csf", "file_stamp": stamp},
+        "layers": [
+            {
+                "layer_name": "dtm_csf_preview",
+                "geometry_type": "MESH",
+                "file_path": "B04_PreProcess/models/dtm_csf_preview.glb",
+                "file_format": "glb",
+            }
+        ],
+    }
+
+
+def _save(db: _FakeDb, models: list[dict]) -> list[int]:
+    from B04_PreProcess import B04_PreProcess_Repository as repo
+
+    async def fake_input(*_a):  # type: ignore[no-untyped-def]
+        return {"crs_epsg": 5186}
+
+    async def fake_cloud(*_a, **_k):  # type: ignore[no-untyped-def]
+        return 9
+
+    repo_input, repo_cloud = repo.get_input_file, repo.create_processed_point_cloud
+    repo.get_input_file, repo.create_processed_point_cloud = fake_input, fake_cloud
+    try:
+        return asyncio.run(
+            repo.save_surface_analysis_to_db(
+                db,  # type: ignore[arg-type]
+                project_id="00000000-0000-0000-0000-000000000001",  # type: ignore[arg-type]
+                input_file_id=1,
+                analysis_result={
+                    "processed": {
+                        "processed_file_path": "B04_PreProcess/processed/structured.npz",
+                        "converted_file_path": None,
+                        "point_count": 1,
+                        "bounds": {},
+                        "statistics": {},
+                    },
+                    "models": models,
+                },
+                source_filters=["csf"],
+            )
+        )
+    finally:
+        repo.get_input_file, repo.create_processed_point_cloud = repo_input, repo_cloud
+
+
+def test_파일_지문이_같으면_모델_id_유지() -> None:
+    db = _FakeDb()
+    path = "B04_PreProcess/models/dtm_csf.npz"
+    first = _save(db, [_model(path, "1:2"), _model("B04_PreProcess/models/tin_csf.npz", "3:4")])
+    second = _save(db, [_model(path, "1:2")])
+    assert second == [first[0]]
+    # 이번 결과에 없는 옛 행은 지움 · 같은 파일 행은 하나
+    assert list(db.models) == [first[0]]
+    third = _save(db, [_model(path, "9:9")])
+    assert third != second and list(db.models) == third
+
+
+def test_도엽_한_방식_요청도_id_유지() -> None:
+    from B04_PreProcess import B04_PreProcess_Repository as repo
+
+    db = _FakeDb()
+    path = "B04_PreProcess/models/dtm_sheet_tin_sheet.npz"
+    entry = {**_model(path, "5:6"), "generation_params": {"source_filter": "sheet_tin_sheet"}}
+    entry["generation_params"]["file_stamp"] = "5:6"
+    first = asyncio.run(repo.save_sheet_surface_models(db, "p", [entry]))  # type: ignore[arg-type]
+    second = asyncio.run(repo.save_sheet_surface_models(db, "p", [entry]))  # type: ignore[arg-type]
+    assert first == second and len(db.models) == 1
+
+
+# ── 비용면 · 계류 메모리 기억(R17) · 추천 노선 열쇠(R18) ──────────────────────────
+
+
+def test_비용면_계류_메모리_기억(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed
+    import B05_Profile.B05_Profile_Engine_RouteInitial as engine
+    import B05_Profile.B05_Profile_Engine_Solver as solver
+    import common_util.common_util_crs as crs
+
+    loads: list[int] = []
+    signature = {"value": "sig-1"}
+    grid = np.arange(4.0)
+
+    def fake_load(*_a):  # type: ignore[no-untyped-def]
+        loads.append(1)
+        return grid, grid, np.zeros((4, 4)), np.ones((4, 4), bool), None, None, 1.0
+
+    monkeypatch.setattr(engine, "_TERRAINS", {})
+    monkeypatch.setattr(solver, "_load_or_build_cost_surface", fake_load)
+    monkeypatch.setattr(solver, "_cost_surface_signature", lambda *_a: signature["value"])
+    monkeypatch.setattr(watershed, "_load_features", lambda *_a: [])
+    monkeypatch.setattr(crs, "resolve_project_crs", lambda _root: "EPSG:5186")
+    selection = {"source_filter": "csf", "method": "dtm", "smooth": False}
+    terrain, _ = engine.load_terrain_and_streams(tmp_path, selection)
+    again, _ = engine.load_terrain_and_streams(tmp_path, selection)
+    assert loads == [1] and again is terrain and terrain.signature
+    signature["value"] = "sig-2"
+    changed, _ = engine.load_terrain_and_streams(tmp_path, selection)
+    assert loads == [1, 1] and changed.signature != terrain.signature
+    # 추천 열쇠 = 서명 — 배열 통째 해시 안 함 · 서명이 다르면 다른 열쇠
+    key_a = engine._pool_key([(0.0, 0.0), (3.0, 3.0)], terrain, [], 0.14)
+    key_b = engine._pool_key([(0.0, 0.0), (3.0, 3.0)], changed, [], 0.14)
+    assert key_a != key_b
+    assert key_a == engine._pool_key([(0.0, 0.0), (3.0, 3.0)], terrain, [[(9.0, 9.0)]], 0.14)
+
+
+def test_추천_저장본_종류당_최근_두_벌(tmp_path: Path) -> None:
+    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
+        KEEP_PER_KIND,
+        read_stored,
+        store_dir,
+        write_stored,
+    )
+
+    assert KEEP_PER_KIND == 2
+    request = {"mode": "earthwork", "terrain": None}
+    write_stored(tmp_path, "a", "p", request, {"n": 1}, True)
+    time.sleep(0.02)
+    write_stored(tmp_path, "b", "p", request, {"n": 2}, False)
+    write_stored(tmp_path, "g", "p", {"mode": "gentle", "terrain": None}, {}, True)
+    time.sleep(0.02)
+    assert read_stored(tmp_path, "a") is not None  # 꺼내면 최근으로
+    time.sleep(0.02)
+    write_stored(tmp_path, "c", "p", request, {"n": 3}, False)
+    names = sorted(path.stem for path in store_dir(tmp_path).glob("*.json"))
+    assert names == ["a", "c", "g"]
+
+
+# ── 횡단 상세 — GET 은 안 지움 · ETag(R26) ───────────────────────────────────────
+
+
+def test_횡단_상세_조회는_파일을_지우지_않음(tmp_path: Path) -> None:
+    from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files, valid_cross_names
+
+    for name in ("cross_00000m.json", "cross_00020m.json", "cross_00040m.json"):
+        (tmp_path / name).write_text("{}", encoding="utf-8")
+    stations = [{"chainage_m": 0.0}, {"chainage_m": 20.0}]
+    names = valid_cross_names(tmp_path, stations)
+    assert len(names) == 2 and (tmp_path / "cross_00040m.json").is_file()
+    assert prune_stale_cross_files(tmp_path, stations) == names
+    assert not (tmp_path / "cross_00040m.json").exists()
+
+
+def test_횡단_상세_ETag_같으면_304() -> None:
+    from starlette.requests import Request
+
+    from B06_Section.B06_Section_Router import _etag_json_response
+
+    def request(etag: str | None) -> Request:
+        headers = [] if etag is None else [(b"if-none-match", etag.encode())]
+        return Request({"type": "http", "headers": headers})
+
+    body = b'{"cross_sections": []}'
+    first = _etag_json_response(request(None), body)
+    assert first.status_code == 200 and first.body == body
+    etag = first.headers["etag"]
+    assert _etag_json_response(request(etag), body).status_code == 304
+    assert _etag_json_response(request(etag), body + b" ").status_code == 200
+
+
+def test_횡단_상세_GET_소스에_지우기_없음() -> None:
+    source = Path("B06_Section/B06_Section_Router.py").read_text(encoding="utf-8")
+    start = source.index("def _read_section_detail(")
+    end = source.index("\ndef ", start + 10)
+    assert "prune_stale_cross_files" not in source[start:end]
+    assert "unlink" not in source[start:end]