Merge remote-tracking branch 'origin/sub_laptop_1' into main_laptop_1

B03_FileInput/B03_FileInput_Service_Chain.py

@@ -392,6 +392,53 @@ async def _ensure_initial_polyline(
         logger.exception("초기 계획노선 폴리라인 생성 실패(계속 진행): %s", project_id)
 
 
+def redesign_solve_request(
+    params: dict[str, Any], selection: dict[str, Any], surface_model_id: int
+) -> Any:
+    """stage 2 params(사용자 저장분) + 새 확정 지표면 → as_planned 풀기 요청(노선 저장도 씀)."""
+    from B05_Profile.B05_Profile_Schema import RouteSolveRequest
+
+    points = params.get("points") or {}
+    options = params.get("options") or {}
+    return RouteSolveRequest(
+        filter_key=str(selection.get("source_filter") or params.get("filter_key")),
+        method=str(selection.get("method") or params.get("method") or "dtm"),
+        smooth=bool(selection.get("smooth", params.get("smooth", False))),
+        surface_model_id=surface_model_id,
+        # 사용자가 고친 계획노선을 **그대로** 쓴다 — 다시 풀면 탐색 제약(종단경사·
+        # 최소곡선반지름·회피지역)에 걸려 체인이 통째로 멈춘다(2026-09-06 실측:
+        # 「세그먼트 1 (BP → CP1) 경로 탐색 실패」로 종횡단만 옛 노선에 남았음).
+        # PLAN 0-3 에서 자동탐색은 접기로 했고 노선은 사용자가 직접 고친다.
+        algorithm="as_planned",
+        bp=points["bp"],
+        ep=points["ep"],
+        cp=points.get("cp") or [],
+        ap=points.get("ap") or [],
+        fp=points.get("fp") or [],
+        station_interval_m=params.get("station_interval_m"),
+        cross_half_width_m=params.get("cross_half_width_m"),
+        cross_sample_interval_m=params.get("cross_sample_interval_m"),
+        long_sample_interval_m=params.get("long_sample_interval_m"),
+        **{key: options.get(key) for key in ("grade_class",) if options.get(key)},
+        paved=bool(options.get("paved", False)),
+        terrain_type=str(options.get("terrain_type") or "special"),
+        main_direction=str(options.get("main_direction") or "auto"),
+        min_curve_radius_m=options.get("min_curve_radius_m"),
+        max_uphill_grade=options.get("max_uphill_grade"),
+        max_downhill_grade=options.get("max_downhill_grade"),
+        min_uphill_grade=options.get("min_uphill_grade"),
+        min_downhill_grade=options.get("min_downhill_grade"),
+        weights=options.get("weights"),
+        allow_avoid_pass_through=bool(options.get("allow_avoid_pass_through", False)),
+        max_grade_pct=params.get("max_grade_pct"),
+        min_vertical_radius_m=params.get("min_vertical_radius_m"),
+        min_tangent_length_m=params.get("min_tangent_length_m"),
+        balance_segment_length_m=params.get("balance_segment_length_m"),
+        start_elevation_offset_m=params.get("start_elevation_offset_m"),
+        end_elevation_offset_m=params.get("end_elevation_offset_m"),
+    )
+
+
 async def run_redesign_chain(
     project_id: UUID,
     surface_model_id: int,
@@ -421,7 +468,6 @@ async def run_redesign_chain(
         confirm_latest_route,
         solve_route_record,
     )
-    from B05_Profile.B05_Profile_Schema import RouteSolveRequest
     from B06_Section.B06_Section_Repository import (
         get_cross_section_designs,
         get_latest_section_options,
@@ -478,7 +524,6 @@ async def run_redesign_chain(
         )
         params = (stage2 or {}).get("params") or {}
         points = params.get("points") or {}
-        options = params.get("options") or {}
         if not points.get("bp") or not points.get("ep"):
             logger.warning(
                 "재확정 체인 중단(stage 2 params에 제어점 없음): project_id=%s", project_id
@@ -488,43 +533,7 @@ async def run_redesign_chain(
         marks.append(("준비·사용자입력 회수", time.perf_counter()))
 
         # 2) B05 재계산 — 지표면 관련 값만 새 확정 선택으로 교체, 나머지는 사용자 저장분.
-        request = RouteSolveRequest(
-            filter_key=str(selection.get("source_filter") or params.get("filter_key")),
-            method=str(selection.get("method") or params.get("method") or "dtm"),
-            smooth=bool(selection.get("smooth", params.get("smooth", False))),
-            surface_model_id=surface_model_id,
-            # 사용자가 고친 계획노선을 **그대로** 쓴다 — 다시 풀면 탐색 제약(종단경사·
-            # 최소곡선반지름·회피지역)에 걸려 체인이 통째로 멈춘다(2026-09-06 실측:
-            # 「세그먼트 1 (BP → CP1) 경로 탐색 실패」로 종횡단만 옛 노선에 남았음).
-            # PLAN 0-3 에서 자동탐색은 접기로 했고 노선은 사용자가 직접 고친다.
-            algorithm="as_planned",
-            bp=points["bp"],
-            ep=points["ep"],
-            cp=points.get("cp") or [],
-            ap=points.get("ap") or [],
-            fp=points.get("fp") or [],
-            station_interval_m=params.get("station_interval_m"),
-            cross_half_width_m=params.get("cross_half_width_m"),
-            cross_sample_interval_m=params.get("cross_sample_interval_m"),
-            long_sample_interval_m=params.get("long_sample_interval_m"),
-            **{key: options.get(key) for key in ("grade_class",) if options.get(key)},
-            paved=bool(options.get("paved", False)),
-            terrain_type=str(options.get("terrain_type") or "special"),
-            main_direction=str(options.get("main_direction") or "auto"),
-            min_curve_radius_m=options.get("min_curve_radius_m"),
-            max_uphill_grade=options.get("max_uphill_grade"),
-            max_downhill_grade=options.get("max_downhill_grade"),
-            min_uphill_grade=options.get("min_uphill_grade"),
-            min_downhill_grade=options.get("min_downhill_grade"),
-            weights=options.get("weights"),
-            allow_avoid_pass_through=bool(options.get("allow_avoid_pass_through", False)),
-            max_grade_pct=params.get("max_grade_pct"),
-            min_vertical_radius_m=params.get("min_vertical_radius_m"),
-            min_tangent_length_m=params.get("min_tangent_length_m"),
-            balance_segment_length_m=params.get("balance_segment_length_m"),
-            start_elevation_offset_m=params.get("start_elevation_offset_m"),
-            end_elevation_offset_m=params.get("end_elevation_offset_m"),
-        )
+        request = redesign_solve_request(params, selection, surface_model_id)
         # 종횡단은 관을 정한 뒤 한 번만 만든다(342-2) — 여기서 만들면 아래 4c 가 통째로 덮는 헛일.
         solve_result: Any = await solve_route_record(project_id, request, generate_sections=False)
         if isinstance(solve_result, JSONResponse):

B05_Profile/B05_Profile_Route_Save.py

@@ -0,0 +1,496 @@
+"""노선 저장 서버 길(346-3) — 노선 초안을 정본으로: 노선 → 풀기 → 관 · 구조물 따라감 →
+종 · 횡단 한 번 → 계획고 옮김 → 횡단 이월 → B06 확정(서버 재계산 · 3D). 배수유역은 체인에서 뗌.
+
+API — `POST /api/projects/{id}/route/replan`(`B05_Profile_Router_Replan.replan_route`):
+  요청 `RouteSaveRequest` = 옛 칸(`vertices` · `initial` · `balanced_profile` · `terrain` ·
+    `station_snap`) +
+    `draft` — 계약 `RouteDraft`(`A00_Common/design/design_pipeline_types.ts`) 그대로
+      `{rev, nodes: [[x, y]…], curveOn: [bool…], curveRadius: [m | null…], curveLock, curveArc,
+      stationSnap, terrain}` · 오면 `vertices` · `station_snap` · `terrain` 대신 이것.
+    `as_initial` — 추천 노선 [적용](초기값 다시 뜸) · 손으로 고친 노선 저장은 false(초기값 그대로).
+  응답 200 `{status: "success", project_id, route_id, total_length_m, vertex_count, current_initial,
+    initial_refreshed, pipes: {kept, crossing, concave}, structures: {moved, far}, cross_dropped,
+    drainage_stale: true, seconds}`.
+  오류 400(노드 둘 미만 · 초안 칸 길이 어긋남) · 404(프로젝트 없음) ·
+    409 `{status: "error", message}`(체인 실패 — 노선 · 관 · 구조물 · 종횡단 파일 · 새 노선 행을
+    옛것으로 되살린 뒤).
+
+규칙:
+  - 관 정본을 지우지 않음 — 좌표를 따라감 · 새 세류선 교차점 기본 관 · 오목 지점 보충만 더함
+    (`B05_Profile_Route_Save_Parts`) · 배수유역 저장분은 노선 지문이 달라 「낡음」(B04 `_resolve`).
+  - 계획고 = 바뀐 구간만 새 지반 첫 계획선 · 그 밖 사용자 계획고 지킴
+    (`B05_Profile_Engine_Grade_Carry`) · 추천 「종단 반영」 켬이면 균형 종단
+    (`apply_profile_target`).
+  - 초기값은 여기서 안 뜸 — 부르는 쪽이 `as_initial` 일 때만(자동설계 체인 직후 + 추천 [적용]).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import shutil
+import tempfile
+import time
+from dataclasses import asdict
+from pathlib import Path
+from typing import Any, Literal
+from uuid import UUID
+
+import numpy as np
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+
+from B05_Profile.B05_Profile_Route_Save_Parts import (
+    add_crossing_pipes,
+    carry_cross_inputs,
+    concave_chainages,
+    fill_concave_pipes,
+    follow_pipes,
+    station_keys,
+    vertices_of,
+)
+from B05_Profile.B05_Profile_Schema import RouteReplanRequest, RouteVertexInput
+from config.config_db import get_db_pool
+
+logger = logging.getLogger(__name__)
+
+
+class RouteDraftInput(BaseModel):
+    """노선 초안 — 계약 `RouteDraft` 모양 그대로(칸 이름 camelCase)."""
+
+    rev: int = 0
+    nodes: list[tuple[float, float]] = Field(default_factory=list)
+    curveOn: list[bool] = Field(default_factory=list)  # noqa: N815 — 계약 칸 이름
+    curveRadius: list[float | None] = Field(default_factory=list)  # noqa: N815
+    curveLock: list[str | None] = Field(default_factory=list)  # noqa: N815
+    curveArc: list[float | None] = Field(default_factory=list)  # noqa: N815
+    stationSnap: bool = False  # noqa: N815
+    terrain: Literal["las", "sheet"] | None = None
+
+
+class RouteSaveRequest(RouteReplanRequest):
+    """노선 저장 요청 — 옛 [노선 반영] 칸 + 초안(`draft`) · 추천 [적용](`as_initial`)."""
+
+    draft: RouteDraftInput | None = None
+    as_initial: bool = False
+
+    def resolved(self) -> RouteReplanRequest:
+        """초안이 오면 노드 · 곡선 · 스냅 · 지형을 옛 칸으로 편다. 칸 길이가 어긋나면 ValueError."""
+        if self.draft is None:
+            return self
+        draft = self.draft
+        count = len(draft.nodes)
+        for name in ("curveOn", "curveRadius"):
+            values = getattr(draft, name)
+            if values and len(values) != count:
+                raise ValueError(
+                    f"노선 초안 {name} 길이({len(values)})가 노드 수({count})와 다릅니다."
+                )
+        vertices = [
+            RouteVertexInput(
+                x=x,
+                y=y,
+                curve=draft.curveOn[index] if draft.curveOn else True,
+                radius_m=draft.curveRadius[index] if draft.curveRadius else None,
+            )
+            for index, (x, y) in enumerate(draft.nodes)
+        ]
+        return self.model_copy(
+            update={
+                "vertices": vertices,
+                "station_snap": draft.stationSnap,
+                "terrain": draft.terrain,
+            }
+        )
+
+
+class _Backup:
+    """체인이 고치는 파일을 손에 쥠 — 실패하면 그대로 되돌림(노선 · 관 · 구조물 · 종횡단)."""
+
+    def __init__(self, project_root: Path, paths: list[Path], folders: list[Path]) -> None:
+        self.held = tempfile.mkdtemp(prefix="route_save_", dir=str(project_root))
+        self.files = {path: path.read_bytes() if path.is_file() else None for path in paths}
+        self.folders: dict[Path, Path | None] = {}
+        for index, folder in enumerate(folders):
+            copy = Path(self.held) / f"dir{index}"
+            self.folders[folder] = shutil.copytree(folder, copy) if folder.is_dir() else None
+
+    def restore(self) -> None:
+        for path, saved in self.files.items():
+            if saved is None:
+                path.unlink(missing_ok=True)
+            else:
+                path.parent.mkdir(parents=True, exist_ok=True)
+                path.write_bytes(saved)
+        for folder, copy in self.folders.items():
+            shutil.rmtree(folder, ignore_errors=True)
+            if copy is not None:
+                shutil.copytree(copy, folder)
+
+    def drop(self) -> None:
+        shutil.rmtree(self.held, ignore_errors=True)
+
+
+class _Fail(RuntimeError):
+    """체인 한 단계 실패 — 사유를 그대로 화면까지(까닭을 덮지 않음)."""
+
+
+def _reason(response: JSONResponse) -> str:
+    try:
+        return bytes(response.body).decode("utf-8", "replace")[:300]
+    except Exception:  # noqa: BLE001 — 사유는 로그 · 응답용
+        return f"status={response.status_code}"
+
+
+def _read_json(path: Path) -> dict[str, Any]:
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {}
+
+
+def _xy(project_root: Path, route_data_path: str | None) -> list[list[float]]:
+    from B05_Profile.B05_Profile_Engine_Sections import _load_route_polyline
+
+    if not route_data_path:
+        return []
+    try:
+        return [[x, y] for x, y, _ in _load_route_polyline(project_root, route_data_path)]
+    except (OSError, ValueError):
+        return []
+
+
+def _ground_along(project_root: Path, selection: dict[str, Any], vertices: list) -> tuple:
+    """새 노선 지반 종단(2 m) — 오목 지점 찾기. 지표면을 못 열면 빈 배열(보충 건너뜀)."""
+    from shapely.geometry import LineString
+
+    from B05_Profile.B05_Profile_Route_Save_Parts import CONCAVE_STEP_M
+    from common_util.common_util_surface_sampler import build_surface_sampler
+
+    try:
+        sampler = build_surface_sampler(
+            project_root / "B04_PreProcess" / "models",
+            str(selection.get("source_filter")),
+            str(selection.get("method") or "dtm"),
+            bool(selection.get("smooth")),
+        )
+    except (FileNotFoundError, KeyError, OSError, ValueError):
+        logger.warning("노선 저장: 지표면을 못 열어 오목 지점 보충을 건너뜀")
+        return np.zeros(0), np.zeros(0)
+    line = LineString([(vertex.x, vertex.y) for vertex in vertices])
+    chainage = np.arange(0.0, line.length + 1e-9, CONCAVE_STEP_M)
+    xy = np.array([line.interpolate(float(s)).coords[0] for s in chainage], dtype=np.float64)
+    z, valid = sampler.sample_xy(xy)
+    return chainage[valid], np.asarray(z, dtype=np.float64)[valid]
+
+
+async def _place_pipes(
+    project_id: UUID,
+    project_root: Path,
+    stored_path: str,
+    selection: dict[str, Any],
+    old_pipes: list,
+    old_xy: list[list[float]],
+    points: list[dict[str, float]],
+) -> tuple[list, dict[str, int]]:
+    """관 따라감 → 새 교차점 기본 관 → 오목 지점 보충 → 정본 저장(새 노선 지문)."""
+    from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import _base_pipes
+    from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare
+    from common_util.common_util_drainage_pipes import route_signature, save_pipe_points
+
+    prepared = await _prepare(project_id)
+    if isinstance(prepared, JSONResponse):  # 도엽 · 노선 파일 없음 — 따라감만
+        vertices, streams = vertices_of([(p["x"], p["y"]) for p in points]), []
+    else:
+        vertices, streams = prepared["vertices"], prepared["streams"]
+    pairs = follow_pipes(old_pipes, vertices)
+    total = float(vertices[-1].chainage_m)
+    crossing = add_crossing_pipes(pairs, _base_pipes(vertices, streams), old_xy, total)
+    chainage, ground = await asyncio.to_thread(_ground_along, project_root, selection, vertices)
+    concave = fill_concave_pipes(pairs, concave_chainages(chainage, ground), vertices, old_xy)
+    await asyncio.to_thread(
+        save_pipe_points,
+        stored_path,
+        route_signature(vertices),
+        [pipe for _, pipe in pairs],
+        vertices,
+    )
+    return pairs, {"kept": len(old_pipes), "crossing": crossing, "concave": concave}
+
+
+async def _carry_profile(
+    project_root: Path,
+    relative: str,
+    route_id: int,
+    old_alignment: dict[str, Any],
+    old_xy: list[list[float]],
+    new_xy: list[list[float]],
+) -> None:
+    """바뀐 구간만 새 지반 첫 계획선 · 그 밖 사용자 계획고 · 종단곡선 R 은 길이 차만큼 밀어 지킴."""
+    from B05_Profile.B05_Profile_Engine_Grade_Carry import carry_profile_edits, changed_span
+    from B05_Profile.B05_Profile_Repository import update_longitudinal_grade_summary
+    from B05_Profile.B05_Profile_Router import _apply_alignment_edits
+    from common_util.common_util_json import atomic_write_json
+
+    if not old_alignment.get("base_pvi"):
+        return  # 옛 선형 없음 — 새 지반 첫 계획선 그대로
+    path = project_root / relative
+    longitudinal = _read_json(path)
+    alignment = longitudinal.get("profile_alignment") or {}
+    span = changed_span([tuple(p) for p in old_xy], [tuple(p) for p in new_xy])
+    if span is None:  # 같은 노선 — 옛 계획선 통째
+        base, edits = old_alignment["base_pvi"], old_alignment.get("edits") or {}
+    else:
+        base, edits = carry_profile_edits(
+            old_alignment["base_pvi"], old_alignment.get("edits"), span, alignment["base_pvi"]
+        )
+    alignment["base_pvi"] = base
+    await asyncio.to_thread(atomic_write_json, path, longitudinal)
+    _alignment, summary, _dropped = await asyncio.to_thread(
+        _apply_alignment_edits, project_root, relative, edits
+    )
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        await connection.begin()
+        await update_longitudinal_grade_summary(
+            connection, route_id=route_id, grade_summary=summary
+        )
+        await connection.commit()
+
+
+async def _carry_cross(
+    project_id: UUID,
+    project_root: Path,
+    route_id: int,
+    relative: str,
+    old_long: dict[str, Any],
+    old_designs: list[dict[str, Any]],
+    old_pipes: list,
+    old_structures: list,
+    pairs: list,
+    interval_m: float,
+    standard: dict[str, Any] | None,
+) -> int:
+    """측점 열쇠로 옛 횡단 입력을 새 측점에 얹고 전 측점을 새 계획고로 셈 · 버린 수 돌려줌."""
+    from B05_Profile.B05_Profile_Structures_Repository import load_structures
+    from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs
+    from B06_Section.B06_Section_Router_Confirm import _rowless_station_chainages
+    from B06_Section.B06_Section_Router_Design import (
+        CARRY_INPUT_KEYS,
+        STRUCTURE_ADJUST_KEYS,
+        compute_default_designs,
+    )
+
+    new_long = _read_json(project_root / relative)
+    _, new_structures = load_structures(str(project_root))
+    old_keys = station_keys(
+        old_long.get("stations") or [], list(enumerate(old_pipes)), old_structures, interval_m
+    )
+    new_keys = station_keys(new_long.get("stations") or [], pairs, new_structures, interval_m)
+    designs = {
+        round(float(row["chainage_m"]), 3): row["design"]
+        for row in old_designs
+        if isinstance(row.get("design"), dict)
+    }
+    inputs, adjusts, dropped = carry_cross_inputs(
+        old_keys, new_keys, designs, CARRY_INPUT_KEYS, STRUCTURE_ADJUST_KEYS
+    )
+    chainages = await asyncio.to_thread(_rowless_station_chainages, project_root, relative, [])
+    entries = await asyncio.to_thread(
+        compute_default_designs, project_root, relative, chainages, standard, inputs
+    )
+    entries = [(at, {**design, **adjusts.get(round(at, 3), {})}) for at, design in entries]
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        await connection.begin()
+        try:
+            await merge_cross_section_designs(
+                connection, route_id=route_id, entries=entries, replace=True, project_id=project_id
+            )
+            await connection.commit()
+        except Exception:
+            await connection.rollback()
+            raise
+    return dropped
+
+
+async def _rollback(project_id: UUID, old_route_id: int, params: dict[str, Any]) -> None:
+    """새 노선 행(종횡단 함께 지워짐)을 지우고 stage 2 params 를 옛것으로 — 최신 노선 = 옛 노선."""
+    from common_util.common_util_workflow_state import start_stage
+
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        await connection.begin()
+        try:
+            async with connection.cursor() as cursor:
+                await cursor.execute(
+                    "DELETE FROM routes WHERE project_id = %s AND id > %s",
+                    (str(project_id), old_route_id),
+                )
+                await start_stage(cursor, str(project_id), 2, params)
+            await connection.commit()
+        except Exception:
+            await connection.rollback()
+            logger.exception("노선 저장 되살림(DB) 실패: project_id=%s", project_id)
+
+
+async def run_route_save_chain(
+    project_id: UUID,
+    project_root: Path,
+    stored_path: str,
+    surface_model_id: int,
+    selection: dict[str, Any],
+    old_route: dict[str, Any],
+    points: list[dict[str, float]],
+    previous_params: dict[str, Any],
+    profile_target: tuple[list[float], list[float]] | None = None,
+) -> dict[str, Any]:
+    """노선 저장 체인 — 성공이면 요약 dict · 실패면 `{"error": 사유}`(파일 · 행은 되살린 뒤)."""
+    import aiomysql
+
+    from B03_FileInput.B03_FileInput_Service_Chain import _log_steps, redesign_solve_request
+    from B05_Profile.B05_Profile_Engine_Sections import _STAGE_SUBDIR
+    from B05_Profile.B05_Profile_Router import (
+        _section_options,
+        confirm_latest_route,
+        solve_route_record,
+    )
+    from B05_Profile.B05_Profile_Structures_Repository import load_structures, structures_file_path
+    from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
+    from B06_Section.B06_Section_Repository import (
+        get_cross_section_designs,
+        get_latest_section_options,
+        get_longitudinal_section,
+    )
+    from B06_Section.B06_Section_Router import regenerate_route_sections
+    from B06_Section.B06_Section_Router_Confirm import confirm_sections
+    from B06_Section.B06_Section_Router_Design import stored_standard_cross_section
+    from B06_Section.B06_Section_Schema import SectionConfirmRequest
+    from common_util.common_util_drainage_pipes import pipe_points_path, read_pipe_points_file
+    from common_util.common_util_initial_snapshot import clear_designing, mark_designing
+    from common_util.common_util_workflow_state import get_workflow_state
+
+    marks = [("시작", time.perf_counter())]
+    old_route_id = int(old_route["id"])
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        async with connection.cursor(aiomysql.DictCursor) as cursor:
+            state = await get_workflow_state(cursor, str(project_id))
+        old_long_row = await get_longitudinal_section(connection, project_id, old_route_id)
+        old_designs = await get_cross_section_designs(connection, old_route_id)
+        old_section_options = await get_latest_section_options(connection, project_id)
+    stage2 = next((s for s in state.get("stages", []) if int(s.get("stage_no", -1)) == 2), None)
+    params = (stage2 or {}).get("params") or {}
+    relative_old = str((old_long_row or {}).get("longitudinal_file_path") or "")
+    old_long = _read_json(project_root / relative_old) if relative_old else {}
+    old_xy = _xy(project_root, old_route.get("route_data_path"))
+    old_pipes = read_pipe_points_file(pipe_points_path(stored_path))
+    _, old_structures = load_structures(str(project_root))
+    stage_dir = project_root / _STAGE_SUBDIR
+    backup = _Backup(
+        project_root,
+        [
+            project_root / str(old_route.get("route_data_path") or "B05_Profile/route/x"),
+            pipe_points_path(stored_path),
+            Path(structures_file_path(str(project_root))),
+        ],
+        [stage_dir / "longitudinal", stage_dir / "cross_sections"],
+    )
+    mark_designing(project_root)
+    try:
+        # 구조물 = 땅 위 자리 따라감(옛 노선 파일이 아직 옛 것일 때 — 좌표 없는 옛 자료 채움)
+        try:
+            structures = await asyncio.to_thread(
+                follow_structures_to_route, project_root, old_route.get("route_data_path"), points
+            )
+        except Exception:  # noqa: BLE001 — 따라감 실패 = 옛 누가거리 그대로(비치명)
+            logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
+            structures = {"moved": 0, "far": 0}
+        request = redesign_solve_request(params, selection, surface_model_id)
+        solved: Any = await solve_route_record(project_id, request, generate_sections=False)
+        if isinstance(solved, JSONResponse):
+            raise _Fail(f"노선 풀기 실패({solved.status_code}): {_reason(solved)}")
+        new_route_id = int(solved.route_id)
+        confirmed = await confirm_latest_route(project_id, None, mark_stage_complete=False)
+        if isinstance(confirmed, JSONResponse):
+            raise _Fail(f"노선 확정 실패({confirmed.status_code})")
+        new_xy = _xy(project_root, solved.route_data_path)
+        marks.append(("노선 풀기 · 구조물 따라감", time.perf_counter()))
+
+        pairs, pipes = await _place_pipes(
+            project_id, project_root, stored_path, selection, old_pipes, old_xy, points
+        )
+        marks.append(("관 따라감 · 교차점 · 오목 보충", time.perf_counter()))
+
+        seed = asdict(_section_options(request, old_section_options))
+        seed.pop("extra_stations", None)
+        made = await regenerate_route_sections(
+            project_id, new_route_id, float(seed["cross_half_width_m"]), stored_options=seed
+        )
+        if isinstance(made, JSONResponse):
+            raise _Fail(f"종횡단 생성 실패({made.status_code}): {_reason(made)}")
+        async with pool.acquire() as connection:
+            new_long_row = await get_longitudinal_section(connection, project_id, new_route_id)
+        relative = str((new_long_row or {}).get("longitudinal_file_path") or "")
+        marks.append(("종횡단 한 번", time.perf_counter()))
+
+        if profile_target is not None:
+            from B05_Profile.B05_Profile_Router_RouteInitial import apply_profile_target
+
+            await apply_profile_target(project_id, new_route_id, profile_target)
+        else:
+            await _carry_profile(
+                project_root,
+                relative,
+                new_route_id,
+                old_long.get("profile_alignment") or {},
+                old_xy,
+                new_xy,
+            )
+        marks.append(("계획고 옮김", time.perf_counter()))
+
+        standard = stored_standard_cross_section(old_long_row)
+        dropped = await _carry_cross(
+            project_id,
+            project_root,
+            new_route_id,
+            relative,
+            old_long,
+            old_designs,
+            old_pipes,
+            old_structures,
+            pairs,
+            float(seed["station_interval_m"]),
+            standard,
+        )
+        marks.append(("횡단 이월(측점 열쇠)", time.perf_counter()))
+
+        done = await confirm_sections(
+            project_id,
+            new_route_id,
+            SectionConfirmRequest(standard_cross_section=standard) if standard else None,
+            mark_stage_complete=False,
+        )
+        if isinstance(done, JSONResponse):
+            raise _Fail(f"종횡단 확정 실패({done.status_code})")
+        marks.append(("B06 확정 · 서버 재계산 · 3D", time.perf_counter()))
+    except Exception as exc:  # noqa: BLE001 — 어느 단계든 옛 노선으로 되살림
+        if not isinstance(exc, _Fail):
+            logger.exception("노선 저장 체인 실패: project_id=%s", project_id)
+        await asyncio.to_thread(backup.restore)
+        await _rollback(project_id, old_route_id, previous_params)
+        return {"error": str(exc) if isinstance(exc, _Fail) else f"노선 저장 중 오류: {exc}"}
+    finally:
+        backup.drop()
+        clear_designing(project_root)
+    _log_steps("재확정 체인(노선 저장)", marks)
+    return {
+        "route_id": new_route_id,
+        "total_length_m": float(solved.total_length_m or 0.0),
+        "pipes": pipes,
+        "structures": structures,
+        "cross_dropped": dropped,
+        "drainage_stale": True,
+        "seconds": round(marks[-1][1] - marks[0][1], 2),
+    }

B05_Profile/B05_Profile_Route_Save_Parts.py

@@ -0,0 +1,224 @@
+"""노선 저장 길(346-3) 순수 부품 — 관 따라감 · 새 교차점 기본 관 · 오목 지점 보충 · 측점 열쇠 ·
+횡단 이월.
+
+DB · 파일 없이 값만 받아 값을 돌려준다(시험이 그대로 부름). 차례 · 저장 · 되살림은
+`B05_Profile_Route_Save.py`.
+
+  - 관 = 좌표가 정본 — 새 노선에 투영해 누가거리만 다시(`project_pipe_points` · 정수 m). 지우지 않음
+    (사용자 관 편집 지킴).
+  - 새 세류선 교차점(옛 노선 위에 없던 자리)에 관이 없으면 기본 관 하나(`_base_pipes` 결과를 받음).
+  - 관 최대 간격 보충 = 옛 노선에서 벗어난 자리의 **오목 지점**(지반 종단 낮은 곳)에만 — 격자 없음.
+  - 측점 열쇠(계약 `design_pipeline_types.ts` 머리) = 일반 `grid:k`(누가거리 ÷ 측점 간격 반올림) ·
+    관 `pipe:<id>` · 구조물 `structure:<id>`(구간 시작 · 끝은 `:start` · `:end`) · 시종점
+    `end:BP` · `end:EP`. 관 id = 저장 파일 차례(노선 저장 한 번 안에서 옛 · 새가 같음).
+  - 횡단 이월 = 같은 열쇠 측점에 `CARRY_INPUT_KEYS` 일곱 · 관 · 구조물 열쇠면
+    `STRUCTURE_ADJUST_KEYS` 도(구조물 조정은 구조물 따라) · 열쇠가 사라진 측점 수 =
+    「노선 밖 고침」.
+"""
+
+from __future__ import annotations
+
+import copy
+import math
+from typing import Any, Sequence
+
+import numpy as np
+from shapely.geometry import LineString, Point
+
+from common_util.common_util_drainage_pipes import (
+    PIPE_SOURCE_SPACING,
+    PIPE_SOURCE_STREAM,
+    PipePoint,
+    project_pipe_points,
+)
+from common_util.common_util_route_geometry import RouteVertex
+from common_util.common_util_structure_station import plan_stations, snap_structure_station
+from config.config_system import DRAINAGE_PIPE_MAX_SPACING_M, DRAINAGE_PIPE_MIN_SPACING_M
+
+# 옛 노선에서 이만큼(m) 넘게 떨어진 자리 = 노선이 바뀐 자리(교차점 · 보충 대상).
+MOVED_AWAY_M = 0.5
+# 오목 지점을 찾을 지반 표본 간격(m) · 고름 창(표본 수) — DTM 잔물결을 오목으로 안 읽게.
+# ponytail: 이동 평균 + 국소 최저 · 계곡이 얕아 놓치면 창을 줄이거나 흐름 강도와 섞을 것.
+CONCAVE_STEP_M = 2.0
+CONCAVE_WINDOW = 5
+_SAME_M = 0.01
+
+
+def vertices_of(xy: Sequence[Sequence[float]]) -> list[RouteVertex]:
+    """평면 점열 → 누가거리 붙은 정점(z 0)."""
+    out: list[RouteVertex] = []
+    walked = 0.0
+    for index, point in enumerate(xy):
+        x, y = float(point[0]), float(point[1])
+        if index:
+            walked += math.hypot(x - out[-1].x, y - out[-1].y)
+        out.append(RouteVertex(x=x, y=y, z=0.0, chainage_m=walked))
+    return out
+
+
+def follow_pipes(
+    old_pipes: list[PipePoint], vertices: list[RouteVertex]
+) -> list[tuple[int, PipePoint]]:
+    """관이 새 노선 좌표를 따라감 → [(관 id = 옛 차례, 새 관)] 누가거리 순 · 옛 목록 그대로."""
+    pairs = [(index, copy.deepcopy(pipe)) for index, pipe in enumerate(old_pipes)]
+    for _, pipe in pairs:
+        project_pipe_points([pipe], vertices)
+    pairs.sort(key=lambda pair: pair[1].chainage_m)
+    return pairs
+
+
+def _moved_away(old_line: LineString | None, x: float, y: float) -> bool:
+    return old_line is None or old_line.distance(Point(x, y)) > MOVED_AWAY_M
+
+
+def add_crossing_pipes(
+    pairs: list[tuple[int, PipePoint]],
+    crossings: Sequence[Any],
+    old_xy: Sequence[Sequence[float]] | None,
+    total_m: float,
+) -> int:
+    """옛 노선에 없던 세류선 교차점에 관이 없으면 기본 관을 더함 · 더한 수."""
+    old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
+    added = 0
+    for crossing in crossings:
+        if not _moved_away(old_line, crossing.x, crossing.y):
+            continue  # 옛 교차점 — 사용자가 지운 관을 되살리지 않음
+        at = snap_structure_station(float(crossing.chainage_m), [total_m])
+        if any(abs(pipe.chainage_m - at) < DRAINAGE_PIPE_MIN_SPACING_M for _, pipe in pairs):
+            continue
+        pairs.append(
+            (-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_STREAM, x=crossing.x, y=crossing.y))
+        )
+        added += 1
+    pairs.sort(key=lambda pair: pair[1].chainage_m)
+    return added
+
+
+def concave_chainages(chainage: np.ndarray, ground: np.ndarray) -> list[tuple[float, float]]:
+    """지반 종단의 오목 지점(이동 평균 국소 최저) → [(누가거리, 고른 지반고)]."""
+    if ground.size < 3:
+        return []
+    kernel = np.ones(CONCAVE_WINDOW) / CONCAVE_WINDOW
+    pad = CONCAVE_WINDOW // 2
+    smooth = np.convolve(np.pad(ground, pad, mode="edge"), kernel, mode="valid")
+    return [
+        (float(chainage[i]), float(smooth[i]))
+        for i in range(1, smooth.size - 1)
+        if smooth[i] < smooth[i - 1] and smooth[i] <= smooth[i + 1]
+    ]
+
+
+def fill_concave_pipes(
+    pairs: list[tuple[int, PipePoint]],
+    concave: list[tuple[float, float]],
+    vertices: list[RouteVertex],
+    old_xy: Sequence[Sequence[float]] | None,
+) -> int:
+    """관 사이(시종점 포함)가 최대 간격을 넘으면 옛 노선 밖 오목 지점 중 가장 낮은 곳에 관."""
+    line = LineString([(vertex.x, vertex.y) for vertex in vertices])
+    old_line = LineString(old_xy) if old_xy and len(old_xy) >= 2 else None
+    total = float(line.length)
+    added = 0
+    while True:
+        edges = [0.0, *(pipe.chainage_m for _, pipe in pairs), total]
+        best: tuple[float, float] | None = None
+        for start, end in zip(edges, edges[1:]):
+            if end - start <= DRAINAGE_PIPE_MAX_SPACING_M:
+                continue
+            for at, z in concave:
+                if not (
+                    start + DRAINAGE_PIPE_MIN_SPACING_M < at < end - DRAINAGE_PIPE_MIN_SPACING_M
+                ):
+                    continue
+                point = line.interpolate(at)
+                if _moved_away(old_line, point.x, point.y) and (best is None or z < best[1]):
+                    best = (at, z)
+        if best is None:
+            return added
+        at = snap_structure_station(best[0], [total])
+        point = line.interpolate(at)
+        pairs.append(
+            (-1, PipePoint(chainage_m=at, source=PIPE_SOURCE_SPACING, x=point.x, y=point.y))
+        )
+        pairs.sort(key=lambda pair: pair[1].chainage_m)
+        concave = [item for item in concave if abs(item[0] - best[0]) > _SAME_M]
+        added += 1
+
+
+def _facility_keys(
+    pipes: Sequence[tuple[int, PipePoint]], structures: Sequence[Any]
+) -> dict[float, str]:
+    """측점 자리(0.01 m) → 관 · 구조물 열쇠. 같은 자리면 관이 먼저(`plan_stations` 이름 규칙)."""
+    keys: dict[float, str] = {}
+
+    def put(marks: list[dict[str, object]], head: str) -> None:
+        for mark in marks:
+            role = str(mark.get("role") or "anchor")
+            keys.setdefault(
+                round(float(mark["chainage_m"]), 2), head if role == "anchor" else f"{head}:{role}"
+            )
+
+    for pipe_id, pipe in pipes:
+        if pipe_id >= 0:
+            put(plan_stations([pipe], []), f"pipe:{pipe_id}")
+    for item in structures:
+        if getattr(item, "structure_id", None):
+            put(plan_stations([], [item]), f"structure:{item.structure_id}")
+    return keys
+
+
+def station_keys(
+    stations: Sequence[dict[str, Any]],
+    pipes: Sequence[tuple[int, PipePoint]],
+    structures: Sequence[Any],
+    interval_m: float,
+) -> dict[str, float]:
+    """측점 열쇠 → 누가거리. 관 · 구조물 자리 > 시종점 > 일반 격자(간격 배수인 것만)."""
+    facility = _facility_keys(pipes, structures)
+    values = [float(row["chainage_m"]) for row in stations if row.get("chainage_m") is not None]
+    out: dict[str, float] = {}
+    for index, chainage in enumerate(values):
+        key = facility.get(round(chainage, 2))
+        if key is None and index == 0:
+            key = "end:BP"
+        elif key is None and index == len(values) - 1:
+            key = "end:EP"
+        elif key is None and interval_m > 0:
+            k = round(chainage / interval_m)
+            key = f"grid:{k}" if abs(chainage - k * interval_m) < _SAME_M else None
+        if key is not None:
+            out.setdefault(key, chainage)
+    return out
+
+
+def carry_cross_inputs(
+    old_keys: dict[str, float],
+    new_keys: dict[str, float],
+    old_designs: dict[float, dict[str, Any]],
+    input_keys: Sequence[str],
+    adjust_keys: Sequence[str],
+) -> tuple[dict[float, dict[str, Any]], dict[float, dict[str, Any]], int]:
+    """같은 열쇠 측점으로 횡단 입력을 넘김 → (입력{새 누가거리 0.001: 값}, 구조물 조정{…}, 버린 수).
+
+    `old_designs` = 옛 노선 행 {누가거리 0.001: design}. 버린 수 = 설계가 있던 옛 측점 중 새 노선에
+    같은 열쇠가 없는 것(「노선 밖 고침」).
+    """
+    inputs: dict[float, dict[str, Any]] = {}
+    adjusts: dict[float, dict[str, Any]] = {}
+    dropped = 0
+    for key, old_at in old_keys.items():
+        design = old_designs.get(round(old_at, 3))
+        if not isinstance(design, dict):
+            continue
+        new_at = new_keys.get(key)
+        if new_at is None:
+            dropped += 1
+            continue
+        carried = {name: design[name] for name in input_keys if design.get(name) is not None}
+        if carried:
+            inputs[round(new_at, 3)] = carried
+        if key.startswith(("pipe:", "structure:")):
+            adjust = {name: design[name] for name in adjust_keys if design.get(name) is not None}
+            if adjust:
+                adjusts[round(new_at, 3)] = adjust
+    return inputs, adjusts, dropped

B05_Profile/B05_Profile_Router_Replan.py

@@ -14,17 +14,13 @@
 노선 초기화는 수정본 파일을 지우는 것이다 — 그러면 초기 폴리라인을 읽으므로 「초기본을
 수정본으로 복사」와 결과가 같다.
 
-[확인]을 눌렀을 때만 계산이 돈다. 재계산은 **초기 업로드 체인의 로직을 그대로 재사용**한다
-(2026-09-06 사용자 제안) — `run_redesign_chain`이 배수유역 다시 분석 → 기본 관 저장 → 관
-정착 계획선 재산출 → 종횡단 재생성 → 옛 측점 설계를 누가거리로 이월 → B05·B06 확정까지 한
-줄로 돈다. 여기서 하는 일은 그 앞에 **노선을 갈아 끼우는 것**뿐이다.
-
-시설 처리(2026-09-06 사용자 확정) — 횡단배수 지점은 노선 자리에 따라 생기고 없어지므로
-저장분을 버리고 **새로 계산한 값**을 쓴다. 그 밖의 구조물은 **옛 측점값을 그대로** 이어받는다
-(`structures.json`은 프로젝트 정본이라 손대지 않으면 그대로 남는다).
+노선 저장(346-3) — 여기서 노선을 갈아 끼운 뒤 `B05_Profile_Route_Save.run_route_save_chain` 이
+풀기 → 관 · 구조물 좌표 따라감 → 종횡단 한 번 → 계획고 옮김 → 횡단 이월(측점 열쇠) → B06 확정까지
+돈다. 배수유역은 다시 안 셈(낡음) · 관 정본은 안 지움 · 초기값은 추천 [적용](`as_initial`)일 때만.
+요청 · 응답 · 오류 모양 = 그 모듈 머리.
 
   GET  /api/projects/{id}/route/plan         → 예상노선·계획노선 정점(사업지 좌표계)
-  POST /api/projects/{id}/route/replan       → 고친 계획노선으로 갈아 끼우고 재계산
+  POST /api/projects/{id}/route/replan       → 노선 초안 저장(노선 저장 체인)
   POST /api/projects/{id}/route/replan/reset → 계획노선을 예상노선으로 되돌리고 재계산
   POST /api/projects/{id}/route/initial      → 초기 계획노선 갈래 하나를 만들어 돌려줌(추천 저장본만 씀)
 """
@@ -45,6 +41,7 @@ from fastapi.responses import JSONResponse
 from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
 from B05_Profile.B05_Profile_Engine_RouteInitial import read_initial_choice, write_initial_choice
 from B05_Profile.B05_Profile_Engine_Sections import load_station_snap
+from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest
 from B05_Profile.B05_Profile_Schema import (  # noqa: F401 — 부르는 곳이 `replan.` 으로 씀
     RouteInitialCriteria,
     RouteInitialRequest,
@@ -340,22 +337,19 @@ async def _recompute(
     from B03_FileInput.B03_FileInput_Service_Chain import (
         _log_steps,
         _planned_route_points_in_project_crs,
-        run_redesign_chain,
     )
     from B04_PreProcess.B04_PreProcess_Service import (
         find_surface_model_for_selection,
         switch_design_terrain,
     )
     from B05_Profile.B05_Profile_Repository import get_latest_route
-    from common_util.common_util_drainage_pipes import clear_pipe_points
+    from B05_Profile.B05_Profile_Route_Save import run_route_save_chain
     from common_util.common_util_surface_confirmation import get_surface_confirmation_params
     from common_util.common_util_workflow_state import get_workflow_state, start_stage
 
     marks = [("시작", time.perf_counter())]
 
-    # 횡단배수 지점은 새 노선에서 새로 계산한다 — 저장분을 남기면 옛 자리가 살아난다.
-    await asyncio.to_thread(clear_pipe_points, stored_path)
-
+    # 관 정본은 지우지 않음(346-3) — 사용자 관 편집을 지키고 좌표로 새 노선을 따라감(체인 안).
     pool = get_db_pool()
     async with pool.acquire() as connection:
         await connection.begin()  # 고른 지형 = 설계 지형 — 뒤 단계가 모두 이 확정값을 읽음
@@ -408,7 +402,8 @@ async def _recompute(
         stage2 = next(
             (s for s in (state or {}).get("stages", []) if int(s.get("stage_no", -1)) == 2), None
         )
-        params = dict((stage2 or {}).get("params") or {})
+        previous_params = dict((stage2 or {}).get("params") or {})  # 실패 때 되살림
+        params = dict(previous_params)
         params["points"] = {
             "bp": points[0],
             "ep": points[-1],
@@ -418,23 +413,28 @@ async def _recompute(
             await start_stage(cursor, str(project_id), 2, params)
         await connection.commit()
 
-    marks.append(("재계산 준비(관 삭제·노선 읽기·stage 기록)", time.perf_counter()))
-
-    # 구조물은 땅 위 자리(좌표)를 따라감(344-2) — 옛 노선 경로 파일은 체인 전이라 아직 옛 것
-    from B05_Profile.B05_Profile_Structures_Router import follow_structures_to_route
-
-    try:
-        old_path = (latest or {}).get("route_data_path")
-        await asyncio.to_thread(follow_structures_to_route, project_root, old_path, points)
-    except Exception:  # 따라감 실패 = 옛 누가거리 그대로(지금까지와 같음)
-        logger.exception("구조물 따라감 실패(옛 누가거리 유지): project_id=%s", project_id)
+    marks.append(("재계산 준비(노선 읽기·stage 기록)", time.perf_counter()))
 
     target = None
     if balanced is not None:
         from B05_Profile.B05_Profile_Router_RouteInitial import balanced_target
 
         target = await balanced_target(project_id, project_root, points, balanced.criteria)
-    failure = await run_redesign_chain(project_id, int(surface_model_id), selection, target)
+    if not latest:
+        return {"error": "저장된 노선이 없어 노선을 저장할 수 없습니다(자동설계 먼저)."}
+    # 노선 저장 체인(346-3) — 배수유역 뗌 · 관 · 구조물 따라감 · 종횡단 한 번 · 계획고 옮김
+    saved = await run_route_save_chain(
+        project_id,
+        project_root,
+        stored_path,
+        int(surface_model_id),
+        selection,
+        latest,
+        points,
+        previous_params,
+        target,
+    )
+    failure = saved.get("error")
     marks.append(("재확정 체인", time.perf_counter()))
     _log_steps("노선 [확인] 재계산", marks)
     if failure and switched:  # 노선과 함께 설계 지형도 되돌림
@@ -448,14 +448,7 @@ async def _recompute(
         # 부르는 쪽이 계획노선을 되돌리도록 사유를 올려 보낸다.
         return {"error": failure}
 
-    pool = get_db_pool()
-    async with pool.acquire() as connection:
-        made = await get_latest_route(connection, project_id)
-    return {
-        "route_id": int(made["id"]) if made else None,
-        "total_length_m": float(made.get("total_length_m") or 0.0) if made else None,
-        "vertex_count": len(points),
-    }
+    return {**saved, "vertex_count": len(points)}
 
 
 @router.get("/{project_id}/route/plan", response_model=None)
@@ -530,10 +523,16 @@ async def read_route_plan(project_id: UUID) -> dict[str, Any] | JSONResponse:
 
 
 @router.post("/{project_id}/route/replan", response_model=None)
-async def replan_route(
-    project_id: UUID, request: RouteReplanRequest
-) -> dict[str, Any] | JSONResponse:
-    """고친 계획노선으로 갈아 끼우고 배수유역부터 다시 계산한다(모달 [확인])."""
+async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any] | JSONResponse:
+    """노선 초안을 정본으로 저장한다(346-3 · 요청 · 응답 모양 = `B05_Profile_Route_Save` 머리).
+
+    배수유역은 다시 안 셈(낡음 표시) · 초기값은 추천 [적용](`as_initial`)일 때만 다시 뜬다.
+    """
+    as_initial = bool(getattr(body, "as_initial", False))  # 옛 요청 꼴(파이썬 호출)도 받음
+    try:
+        request = body.resolved() if isinstance(body, RouteSaveRequest) else body
+    except ValueError as error:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(error)})
     if len(request.vertices) < 2:
         return JSONResponse(
             status_code=400,
@@ -616,15 +615,18 @@ async def replan_route(
         if choice
         else await asyncio.to_thread(read_initial_choice, project_root)
     )
-    # 확정 노선 = 새 초기값(PLAN 74-4) — [저장] 없이 이 결과를 [초기화]가 돌아갈 자리로 뜬다.
-    from B05_Profile.B05_Profile_Router_RouteInitial import refresh_initial_snapshot
+    # 초기값 = 자동설계 체인 직후 + 추천 노선 [적용] 때만(346-3 · 사용자 확정 10-16) — 손으로 고친
+    # 노선 저장은 [초기화]가 돌아갈 자리를 안 바꿈.
+    if as_initial:
+        from B05_Profile.B05_Profile_Router_RouteInitial import refresh_initial_snapshot
 
-    await refresh_initial_snapshot(project_id, project_root, result.get("route_id"))
+        await refresh_initial_snapshot(project_id, project_root, result.get("route_id"))
     return {
         "status": "success",
         "project_id": str(project_id),
         **result,
         "current_initial": current,
+        "initial_refreshed": as_initial,
     }
 
 

B05_Profile/B05_Profile_Router_RouteInitial.py

@@ -337,7 +337,7 @@ async def balanced_profile_of(
 async def refresh_initial_snapshot(
     project_id: UUID, project_root: Path, route_id: int | None
 ) -> None:
-    """노선 [확인] 결과로 초기값(initial_snapshot)을 다시 뜬다(PLAN 74-4) — 실패는 비치명적."""
+    """추천 노선 [적용](as_initial) 결과로 초기값을 다시 뜬다(346-3) — 실패는 비치명적."""
     from common_util.common_util_initial_snapshot import (
         discard_initial_snapshot,
         save_initial_snapshot,

B06_Section/B06_Section_Router_Confirm.py

@@ -186,6 +186,18 @@ async def _recompute_stored_designs(project_id: UUID, route_id: int) -> None:
         )
 
 
+def _prune_cross_files(project_root: Path, longitudinal_file_path: str) -> None:
+    """측점을 고친 뒤 옛 cross_*.json 을 지움 — GET 은 안 지우므로 쓰는 쪽(저장 · 확정)이 맡음."""
+    from B05_Profile.B05_Profile_Engine_Sections import prune_stale_cross_files
+
+    path = project_root / longitudinal_file_path
+    try:
+        stations = json.loads(path.read_text(encoding="utf-8")).get("stations") or []
+    except (OSError, ValueError, AttributeError):
+        return  # 측점을 못 읽으면 안 지움(오삭제 막음)
+    prune_stale_cross_files(path.parent.parent / "cross_sections", stations)
+
+
 async def _prebuild_corridor(project_id: UUID, route_id: int, project_root: Path) -> None:
     """3D 길 모형을 서버 Node 가 지어 남긴다(342-2) — 정본이 다시 선 [저장] · [확정] 끝.
 
@@ -261,6 +273,10 @@ async def save_sections(
             missing = await get_cross_sections_missing_design_chainages(connection, route_id)
 
         project_root = Path(resolve_stored_project_path(stored_path))
+        # 측점 맞춤이 지운 자리의 옛 횡단 파일 정리(쓰는 쪽 몫)
+        await asyncio.to_thread(
+            _prune_cross_files, project_root, str(existing["longitudinal_file_path"])
+        )
         rowless = missing + await asyncio.to_thread(
             _rowless_station_chainages,
             project_root,
@@ -341,6 +357,10 @@ async def confirm_sections(
             known = await get_cross_section_chainages(connection, route_id)
         marks.append(("조회(종단·경로·미지정 측점)", time.perf_counter()))
         project_root = Path(resolve_stored_project_path(stored_path))
+        # 측점 맞춤이 지운 자리의 옛 횡단 파일 정리(쓰는 쪽 몫)
+        await asyncio.to_thread(
+            _prune_cross_files, project_root, str(existing["longitudinal_file_path"])
+        )
         # 행 자체가 없는 측점(구조물 등 비정규)도 확정 대상에 넣는다 — 정본이 없으면
         # 조회 때마다 프리뷰가 다시 계산돼 3D·수량이 확정 결과가 아니게 된다(2026-08-24).
         missing = missing + await asyncio.to_thread(

resources/tester/run_346_3_route_save_timing.py

@@ -0,0 +1,57 @@
+# -*- coding: utf-8 -*-
+"""346-3 노선 저장 실경로 시간 — 검증 프로젝트의 지금 노드 · 곡선 그대로 저장 한 번(수동 실행).
+
+같은 노선이라 값은 그대로(새 노선 행 번호만 바뀜) · 초기값은 안 건드림(`as_initial` 끔).
+쓰는 법: `./venv/Scripts/python.exe resources/tester/run_346_3_route_save_timing.py <project_id>`
+"""
+
+import asyncio
+import json
+import logging
+import sys
+import time
+from pathlib import Path
+from uuid import UUID
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from config.config_db import close_db_pool, init_db_pool  # noqa: E402
+
+
+async def main(project_id: UUID) -> None:
+    from B05_Profile import B05_Profile_Router_Replan as replan
+    from B05_Profile.B05_Profile_Route_Save import RouteSaveRequest
+
+    logging.basicConfig(level=logging.INFO, format="%(message)s")
+    await init_db_pool()
+    try:
+        plan = await replan.read_route_plan(project_id)
+        nodes = plan["nodes"]
+        radii: dict[int, float] = {}
+        for curve in plan["curves"]:
+            if curve.get("node_first") == curve.get("node_last"):
+                radii[int(curve["node_first"])] = float(curve["radius_m"])
+        body = RouteSaveRequest(
+            vertices=[
+                {
+                    "x": node["x"],
+                    "y": node["y"],
+                    "curve": index in radii,
+                    "radius_m": radii.get(index),
+                }
+                for index, node in enumerate(nodes)
+            ],
+            station_snap=bool(plan.get("station_snap")),
+        )
+        started = time.perf_counter()
+        result = await replan.replan_route(project_id, body)
+        seconds = time.perf_counter() - started
+        body_text = getattr(result, "body", None)
+        print(json.dumps(json.loads(body_text) if body_text else result, ensure_ascii=False))
+        print(f"[시간] 노선 저장 전체 {seconds:.1f}s")
+    finally:
+        await close_db_pool()
+
+
+asyncio.run(main(UUID(sys.argv[1])))

resources/tester/test_133_2_design_terrain.py

@@ -18,7 +18,7 @@ import numpy as np
 import B03_FileInput.B03_FileInput_Service_Chain as chain
 import B04_PreProcess.B04_PreProcess_Service as service
 import B05_Profile.B05_Profile_Repository as route_repo
-import common_util.common_util_drainage_pipes as drainage_pipes
+import B05_Profile.B05_Profile_Route_Save as route_save
 import common_util.common_util_workflow_state as workflow_state
 from B05_Profile import B05_Profile_Router_Replan as replan
 from B06_Section import B06_Section_Router_Stations as stations
@@ -212,16 +212,15 @@ def _patch_recompute(monkeypatch, tmp_path: Path, db: _Db, failure: str | None):
     async def start(*_args):
         return None
 
-    async def run_chain(_project_id, model_id, selection, _target=None):
+    async def run_chain(_project_id, _root, _stored, model_id, selection, *_rest):
         seen["model_id"], seen["selection"] = model_id, dict(selection)
-        return failure
+        return {"error": failure} if failure else {"route_id": 238}
 
     points = [{"x": 0.0, "y": 0.0}, {"x": 5.0, "y": 5.0}]
     monkeypatch.setattr(replan, "get_db_pool", lambda: _Pool())
-    monkeypatch.setattr(drainage_pipes, "clear_pipe_points", lambda _s: None)
     monkeypatch.setattr(route_repo, "get_latest_route", latest)
     monkeypatch.setattr(chain, "_planned_route_points_in_project_crs", lambda *_a: points)
-    monkeypatch.setattr(chain, "run_redesign_chain", run_chain)
+    monkeypatch.setattr(route_save, "run_route_save_chain", run_chain)
     monkeypatch.setattr(workflow_state, "get_workflow_state", state)
     monkeypatch.setattr(workflow_state, "start_stage", start)
     return seen

resources/tester/test_346_3_route_save.py

@@ -0,0 +1,561 @@
+# -*- coding: utf-8 -*-
+"""346-3 노선 저장 서버 길 — 차례 · 관 지킴 · 새 교차점 기본 관 · 오목 보충 · 이월 열쇠 ·
+초기값 규칙 ·
+배수유역 낡음 · 실패 때 옛 노선 되살림.
+
+순수 부품(`B05_Profile_Route_Save_Parts`)은 값으로 · 체인(`run_route_save_chain`)과 API
+(`replan_route`)는 DB · 엔진을 가짜로 바꿔 차례 · 되살림 · 초기값만 잰다.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+from uuid import uuid4
+
+import numpy as np
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from fastapi.responses import JSONResponse  # noqa: E402
+
+import B03_FileInput.B03_FileInput_Service_Chain as chain  # noqa: E402
+import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed  # noqa: E402
+import B05_Profile.B05_Profile_Route_Save as save  # noqa: E402
+import B05_Profile.B05_Profile_Router as b05_router  # noqa: E402
+import B05_Profile.B05_Profile_Router_Replan as replan  # noqa: E402
+import B05_Profile.B05_Profile_Router_RouteInitial as route_initial  # noqa: E402
+import B05_Profile.B05_Profile_Structures_Router as structures_router  # noqa: E402
+import B06_Section.B06_Section_Repository as b06_repo  # noqa: E402
+import B06_Section.B06_Section_Router as b06_router  # noqa: E402
+import B06_Section.B06_Section_Router_Confirm as b06_confirm  # noqa: E402
+import common_util.common_util_drainage_pipes as drainage_pipes  # noqa: E402
+import common_util.common_util_initial_snapshot as snapshot  # noqa: E402
+import common_util.common_util_workflow_state as workflow  # noqa: E402
+from B05_Profile.B05_Profile_Route_Save_Parts import (  # noqa: E402
+    add_crossing_pipes,
+    carry_cross_inputs,
+    concave_chainages,
+    fill_concave_pipes,
+    follow_pipes,
+    station_keys,
+    vertices_of,
+)
+from common_util.common_util_drainage_pipes import (  # noqa: E402
+    PIPE_SOURCE_SPACING,
+    PIPE_SOURCE_STREAM,
+    PipePoint,
+    analysis_route_signature,
+    route_signature,
+)
+
+OLD = [(0.0, 0.0), (100.0, 0.0), (200.0, 0.0), (300.0, 0.0), (400.0, 0.0)]
+# 새 노선 = 100 ~ 300 m 를 북으로 휘게(앞 · 뒤는 옛 노선 위)
+NEW = [(0.0, 0.0), (100.0, 0.0), (200.0, 60.0), (300.0, 0.0), (400.0, 0.0)]
+
+
+# ── 관 따라감 · 교차점 · 오목 보충 ─────────────────────────────────────────────
+
+
+def test_관은_지우지_않고_좌표를_따라감():
+    old = [
+        PipePoint(chainage_m=50.0, x=50.0, y=0.0, options={"pipe_kind": "흄관"}),
+        PipePoint(chainage_m=350.0, x=350.0, y=0.0, start_m=345.0, end_m=355.0),
+    ]
+    pairs = follow_pipes(old, vertices_of(NEW))
+    assert [pipe_id for pipe_id, _ in pairs] == [0, 1]
+    assert pairs[0][1].options == {"pipe_kind": "흄관"}  # 사용자 관 속성 그대로
+    shift = pairs[1][1].chainage_m - 350.0  # 새 노선이 길어진 만큼 뒤 관이 밀림
+    assert shift > 30 and pairs[1][1].start_m == pytest.approx(345.0 + shift)
+    assert old[1].chainage_m == 350.0  # 옛 목록은 안 고침
+
+
+def test_새_교차점에만_기본_관():
+    pairs = follow_pipes([PipePoint(chainage_m=50.0, x=50.0, y=0.0)], vertices_of(NEW))
+    crossings = [
+        SimpleNamespace(chainage_m=20.0, x=20.0, y=0.0),  # 옛 노선 위 — 사용자가 지운 관 안 되살림
+        SimpleNamespace(chainage_m=150.4, x=150.0, y=30.0),  # 새 자리 · 관 없음 → 더함
+        SimpleNamespace(chainage_m=52.0, x=52.0, y=0.0),  # 옛 자리
+    ]
+    added = add_crossing_pipes(pairs, crossings, OLD, 456.0)
+    assert added == 1
+    new = [pipe for pipe_id, pipe in pairs if pipe_id < 0]
+    assert new[0].chainage_m == 150.0 and new[0].source == PIPE_SOURCE_STREAM
+    # 같은 자리 다시 = 이미 관이 있음
+    assert add_crossing_pipes(pairs, crossings, OLD, 456.0) == 0
+
+
+def test_오목_지점_보충은_새_자리의_오목만():
+    vertices = vertices_of(NEW)
+    total = vertices[-1].chainage_m
+    chainage = np.arange(0.0, total, 2.0)
+    ground = 100.0 + np.abs(chainage - 220.0) * 0.05  # 220 m 에 골(새 자리)
+    ground[chainage < 60] = (
+        100.0 + np.abs(chainage[chainage < 60] - 30.0) * 0.05
+    )  # 30 m 골(옛 자리)
+    concave = concave_chainages(chainage, ground)
+    assert any(abs(at - 220.0) < 3 for at, _ in concave)
+    pairs: list = []
+    added = fill_concave_pipes(pairs, concave, vertices, OLD)
+    places = [pipe.chainage_m for _, pipe in pairs]
+    assert added >= 1 and all(pipe.source == PIPE_SOURCE_SPACING for _, pipe in pairs)
+    assert any(abs(at - 220.0) <= 3 for at in places)
+    assert not any(at < 60 for at in places)  # 옛 노선 위 오목엔 안 둠
+
+
+def test_간격이_좁으면_보충_없음():
+    vertices = vertices_of(NEW)
+    pairs = [(0, PipePoint(chainage_m=at)) for at in (0.0, 90.0, 180.0, 270.0, 360.0, 450.0)]
+    concave = [(220.0, 99.0)]
+    assert fill_concave_pipes(pairs, concave, vertices, OLD) == 0
+
+
+# ── 측점 열쇠 · 횡단 이월 ──────────────────────────────────────────────────────
+
+
+def _structure(sid, type_id, anchor, start=None, end=None):
+    return SimpleNamespace(
+        structure_id=sid,
+        type_id=type_id,
+        chainage_m=anchor,
+        start_m=start,
+        end_m=end,
+        options={},
+    )
+
+
+def test_측점_열쇠_규칙():
+    stations = [{"chainage_m": s} for s in (0.0, 20.0, 40.0, 50.0, 60.0, 77.0, 80.0, 95.5)]
+    pipes = [(3, PipePoint(chainage_m=50.0))]
+    structures = [_structure("w1", "retaining_wall", 77.0, 60.0, 80.0)]
+    keys = station_keys(stations, pipes, structures, 20.0)
+    assert keys == {
+        "end:BP": 0.0,
+        "grid:1": 20.0,
+        "grid:2": 40.0,
+        "pipe:3": 50.0,
+        "structure:w1:start": 60.0,
+        "structure:w1": 77.0,
+        "structure:w1:end": 80.0,
+        "end:EP": 95.5,
+    }
+
+
+def test_이월은_열쇠로_누가거리가_밀려도():
+    old_keys = {
+        "end:BP": 0.0,
+        "grid:1": 20.0,
+        "pipe:0": 50.0,
+        "structure:w1": 77.0,
+        "grid:5": 100.0,
+    }
+    new_keys = {"end:BP": 0.0, "grid:1": 20.0, "pipe:0": 57.0, "structure:w1": 84.0}
+    designs = {
+        0.0: {"ground_type": "soil"},
+        20.0: {"ground_type": "rock", "berm": {"on": True}, "cut_area_m2": 9.0},
+        50.0: {"ditch_type": "u", "inlet_structure": "basin", "basin_adjust": {"d": 1}},
+        77.0: {"rock_boundary_offset_m": 0.8, "revet_adjust": {"h": 2}},
+        100.0: {"ground_type": "soil"},
+    }
+    inputs, adjusts, dropped = carry_cross_inputs(
+        old_keys,
+        new_keys,
+        designs,
+        ("ground_type", "ditch_type", "berm", "rock_boundary_offset_m"),
+        ("inlet_structure", "basin_adjust", "revet_adjust"),
+    )
+    assert inputs == {
+        0.0: {"ground_type": "soil"},
+        20.0: {"ground_type": "rock", "berm": {"on": True}},  # 결과 칸(면적)은 안 넘김
+        57.0: {"ditch_type": "u"},
+        84.0: {"rock_boundary_offset_m": 0.8},
+    }
+    # 구조물 조정은 관 · 구조물 열쇠만 따라감
+    assert adjusts == {
+        57.0: {"inlet_structure": "basin", "basin_adjust": {"d": 1}},
+        84.0: {"revet_adjust": {"h": 2}},
+    }
+    assert dropped == 1  # grid:5 — 새 노선에 없음(노선 밖 고침)
+
+
+# ── 노선 초안 요청 ────────────────────────────────────────────────────────────
+
+
+def test_초안_모양을_옛_칸으로():
+    body = save.RouteSaveRequest(
+        draft={
+            "rev": 3,
+            "nodes": [[0, 0], [10, 5], [20, 0]],
+            "curveOn": [True, False, True],
+            "curveRadius": [None, 30.0, None],
+            "curveLock": [None, "radius", None],
+            "curveArc": [None, None, None],
+            "stationSnap": True,
+            "terrain": "sheet",
+        }
+    )
+    request = body.resolved()
+    assert [(v.x, v.y, v.curve, v.radius_m) for v in request.vertices] == [
+        (0, 0, True, None),
+        (10, 5, False, 30.0),
+        (20, 0, True, None),
+    ]
+    assert request.station_snap is True and request.terrain == "sheet"
+    with pytest.raises(ValueError):
+        save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]}).resolved()
+
+
+# ── 체인 차례 · 되살림 · 배수유역 낡음 ─────────────────────────────────────────
+
+
+class _Cursor:
+    def __init__(self, log):
+        self.log = log
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        return False
+
+    async def execute(self, sql, params=None):
+        self.log.append(" ".join(sql.split()))
+
+
+class _Connection:
+    def __init__(self, log):
+        self.log = log
+
+    def cursor(self, *args):
+        return _Cursor(self.log)
+
+    async def begin(self):
+        return None
+
+    async def commit(self):
+        return None
+
+    async def rollback(self):
+        return None
+
+
+class _Pool:
+    def __init__(self, log):
+        self.log = log
+
+    def acquire(self):
+        connection = _Connection(self.log)
+
+        class _Acquire:
+            async def __aenter__(self):
+                return connection
+
+            async def __aexit__(self, *exc):
+                return False
+
+        return _Acquire()
+
+
+def _async(value, order=None, name=None):
+    async def fake(*args, **kwargs):
+        if order is not None:
+            order.append(name)
+        return value
+
+    return fake
+
+
+@pytest.fixture
+def chain_world(monkeypatch, tmp_path):
+    order: list[str] = []
+    sql: list[str] = []
+    root = tmp_path
+    route_file = root / "B05_Profile" / "route" / "route_main.geojson"
+    route_file.parent.mkdir(parents=True)
+    route_file.write_text(
+        json.dumps({"geometry": {"coordinates": [[*p, 0] for p in OLD]}}), "utf-8"
+    )
+    long_dir = root / "B06_Section" / "longitudinal"
+    long_dir.mkdir(parents=True)
+    (long_dir / "longitudinal.json").write_text(json.dumps({"stations": [], "old": True}), "utf-8")
+    drain = root / "drain"
+    pipes_file = drain / "edits" / "pipe_points.json"
+    pipes_file.parent.mkdir(parents=True)
+    pipes_file.write_text(
+        json.dumps(
+            {"route_signature": "old", "points": [{"chainage_m": 50.0, "x": 50.0, "y": 0.0}]}
+        ),
+        "utf-8",
+    )
+    (drain / "00_watershed_response.json").write_text(
+        json.dumps({"route_signature": "old"}), "utf-8"
+    )
+    state = {"stages": [{"stage_no": 2, "params": {"points": {}, "options": {}}}]}
+
+    def solve(project_id, request, *, generate_sections=True):
+        order.append("solve")
+        assert generate_sections is False
+        route_file.write_text(
+            json.dumps({"geometry": {"coordinates": [[*p, 0] for p in NEW]}}), "utf-8"
+        )
+        return _async(
+            SimpleNamespace(
+                route_id=8,
+                total_length_m=456.0,
+                route_data_path="B05_Profile/route/route_main.geojson",
+            )
+        )()
+
+    async def regenerate(*args, **kwargs):
+        order.append("sections")
+        (long_dir / "longitudinal.json").write_text(
+            json.dumps({"stations": [], "new": True}), "utf-8"
+        )
+        return SimpleNamespace()
+
+    async def watershed_must_not_run(*args, **kwargs):
+        raise AssertionError("배수유역은 체인에서 뗌")
+
+    async def prepare(project_id):
+        order.append("pipes")
+        return {"vertices": vertices_of(NEW), "streams": []}
+
+    monkeypatch.setattr(save, "get_db_pool", lambda: _Pool(sql))
+    monkeypatch.setattr(workflow, "get_workflow_state", _async(state))
+    monkeypatch.setattr(workflow, "start_stage", _async(None, order, "restore_params"))
+    monkeypatch.setattr(
+        b06_repo,
+        "get_longitudinal_section",
+        _async(
+            {"data": {}, "longitudinal_file_path": "B06_Section/longitudinal/longitudinal.json"}
+        ),
+    )
+    monkeypatch.setattr(b06_repo, "get_cross_section_designs", _async([]))
+    monkeypatch.setattr(b06_repo, "get_latest_section_options", _async({}))
+    monkeypatch.setattr(
+        structures_router,
+        "follow_structures_to_route",
+        lambda *a: order.append("structures") or {"moved": 0, "far": 0},
+    )
+    monkeypatch.setattr(
+        chain,
+        "redesign_solve_request",
+        lambda *a: b05_router.RouteSolveRequest(
+            filter_key="csf",
+            method="dtm",
+            smooth=False,
+            algorithm="as_planned",
+            bp={"x": 0, "y": 0},
+            ep={"x": 400, "y": 0},
+        ),
+    )
+    monkeypatch.setattr(b05_router, "solve_route_record", solve)
+    monkeypatch.setattr(b05_router, "confirm_latest_route", _async({}))
+    monkeypatch.setattr(watershed, "get_primary_region", watershed_must_not_run)
+    monkeypatch.setattr(watershed, "_prepare", prepare)
+    monkeypatch.setattr(drainage_pipes, "clear_pipe_points", watershed_must_not_run)
+    monkeypatch.setattr(drainage_pipes, "drainage_dir", lambda stored: drain)
+    monkeypatch.setattr(save, "_ground_along", lambda *a: (np.zeros(0), np.zeros(0)))
+    monkeypatch.setattr(b06_router, "regenerate_route_sections", regenerate)
+    monkeypatch.setattr(save, "_carry_profile", _async(None, order, "profile"))
+    monkeypatch.setattr(save, "_carry_cross", _async(2, order, "cross"))
+    monkeypatch.setattr(
+        b06_confirm, "confirm_sections", _async(SimpleNamespace(), order, "confirm")
+    )
+    monkeypatch.setattr(snapshot, "mark_designing", lambda root: None)
+    monkeypatch.setattr(snapshot, "clear_designing", lambda root: None)
+    return SimpleNamespace(
+        order=order,
+        sql=sql,
+        root=root,
+        route_file=route_file,
+        pipes_file=pipes_file,
+        long_dir=long_dir,
+        drain=drain,
+    )
+
+
+def _run(world, **kwargs):
+    old_route = {"id": 7, "route_data_path": "B05_Profile/route/route_main.geojson"}
+    return asyncio.run(
+        save.run_route_save_chain(
+            uuid4(),
+            world.root,
+            "p",
+            1,
+            {},
+            old_route,
+            [{"x": x, "y": y} for x, y in NEW],
+            {"old": 1},
+            **kwargs,
+        )
+    )
+
+
+def test_차례와_배수유역_뗌(chain_world):
+    result = _run(chain_world)
+    assert "error" not in result, result
+    assert chain_world.order == [
+        "structures",
+        "solve",
+        "pipes",
+        "sections",
+        "profile",
+        "cross",
+        "confirm",
+    ]
+    assert result["cross_dropped"] == 2 and result["drainage_stale"] is True
+    assert result["pipes"] == {"kept": 1, "crossing": 0, "concave": 0}
+    # 관 정본 = 새 노선 지문 · 배수유역 분석 지문(옛) 과 달라 B04 가 「낡음」
+    saved = json.loads(chain_world.pipes_file.read_text("utf-8"))
+    assert saved["route_signature"] == route_signature(vertices_of(NEW))
+    assert analysis_route_signature("p") != saved["route_signature"]
+    assert len(saved["points"]) == 1  # 사용자 관 그대로(지우지 않음)
+
+
+def test_실패하면_옛_노선을_되살림(chain_world, monkeypatch):
+    monkeypatch.setattr(
+        b06_confirm, "confirm_sections", _async(JSONResponse(status_code=500, content={}))
+    )
+    before_pipes = chain_world.pipes_file.read_text("utf-8")
+    result = _run(chain_world)
+    assert result["error"].startswith("종횡단 확정 실패")
+    assert json.loads(chain_world.route_file.read_text("utf-8"))["geometry"]["coordinates"] == [
+        [*p, 0] for p in OLD
+    ]
+    assert chain_world.pipes_file.read_text("utf-8") == before_pipes
+    assert json.loads((chain_world.long_dir / "longitudinal.json").read_text("utf-8")) == {
+        "stations": [],
+        "old": True,
+    }
+    assert any(
+        sql.startswith("DELETE FROM routes WHERE project_id = %s AND id > %s")
+        for sql in chain_world.sql
+    )
+    assert chain_world.order[-1] == "restore_params"
+    assert not list(chain_world.root.glob("route_save_*"))  # 손에 쥔 사본 치움
+
+
+def test_추천_종단_반영이면_균형_종단(chain_world, monkeypatch):
+    monkeypatch.setattr(
+        route_initial, "apply_profile_target", _async(None, chain_world.order, "balanced")
+    )
+    _run(chain_world, profile_target=([0.0, 456.0], [100.0, 110.0]))
+    assert "balanced" in chain_world.order and "profile" not in chain_world.order
+
+
+# ── 초기값 규칙 ───────────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def api_world(monkeypatch, tmp_path):
+    calls: list = []
+    monkeypatch.setattr(replan, "_project_paths", _async((tmp_path, "p")))
+    monkeypatch.setattr(replan, "_ensure_expected_route", lambda root: None)
+    monkeypatch.setattr(replan, "_plan_criteria", _async((20.0, 0.0, 0.0)))
+    monkeypatch.setattr(replan, "_ensure_planned_initial", lambda *a: None)
+    monkeypatch.setattr(
+        replan,
+        "_write_planned_polyline",
+        lambda *a, **k: {"vertices": 3, "curves": 0, "violations": 0},
+    )
+    monkeypatch.setattr(replan, "_recompute", _async({"route_id": 9, "total_length_m": 10.0}))
+    monkeypatch.setattr(replan, "read_initial_choice", lambda root: None)
+    monkeypatch.setattr(
+        route_initial, "refresh_initial_snapshot", lambda *a: calls.append(a) or _async(None)()
+    )
+    return calls
+
+
+def _body(**kwargs):
+    return save.RouteSaveRequest(vertices=[{"x": 0, "y": 0}, {"x": 5, "y": 5}], **kwargs)
+
+
+def test_손으로_고친_노선_저장은_초기값_그대로(api_world):
+    result = asyncio.run(replan.replan_route(uuid4(), _body()))
+    assert result["initial_refreshed"] is False and api_world == []
+
+
+def test_추천_적용은_초기값_다시(api_world):
+    result = asyncio.run(replan.replan_route(uuid4(), _body(as_initial=True)))
+    assert result["initial_refreshed"] is True and len(api_world) == 1
+
+
+def test_초안_칸이_어긋나면_400(api_world):
+    body = save.RouteSaveRequest(draft={"nodes": [[0, 0], [1, 1]], "curveOn": [True]})
+    response = asyncio.run(replan.replan_route(uuid4(), body))
+    assert response.status_code == 400
+
+
+def _longitudinal(total: float) -> dict:
+    from B05_Profile.B05_Profile_Engine_Grade import resolve_grade_options
+    from B05_Profile.B05_Profile_Engine_Grade_Profile import design_ground_following_profile
+
+    samples = [
+        {"chainage_m": float(s), "elevation_m": 100.0 + 0.05 * s + 2.0 * np.sin(s / 17.0)}
+        for s in range(int(total) + 1)
+    ]
+    stations = [{"chainage_m": float(s)} for s in range(0, int(total) + 1, 20)]
+    longitudinal = {"samples": samples, "stations": stations}
+    alignment, profile = design_ground_following_profile(
+        longitudinal, resolve_grade_options("trunk")
+    )
+    return {**longitudinal, "profile_alignment": alignment, "design_profiles": [profile]}
+
+
+def test_계획고_옮김은_앞_그대로_뒤_밂(tmp_path, monkeypatch):
+    new_xy = [list(p) for p in NEW]
+    total = float(vertices_of(NEW)[-1].chainage_m)
+    old = _longitudinal(400.0)
+    old["profile_alignment"]["edits"] = {
+        "station_offsets": {"40.000": 0.7, "200.000": -0.4, "380.000": 0.3},
+        "curve_radii": {},
+    }
+    path = tmp_path / "B06_Section" / "longitudinal" / "longitudinal.json"
+    path.parent.mkdir(parents=True)
+    path.write_text(json.dumps(_longitudinal(total)), "utf-8")
+    import B05_Profile.B05_Profile_Repository as b05_repo
+
+    monkeypatch.setattr(save, "get_db_pool", lambda: _Pool([]))
+    monkeypatch.setattr(b05_repo, "update_longitudinal_grade_summary", _async(None))
+    asyncio.run(
+        save._carry_profile(
+            tmp_path,
+            "B06_Section/longitudinal/longitudinal.json",
+            8,
+            old["profile_alignment"],
+            [list(p) for p in OLD],
+            new_xy,
+        )
+    )
+    edits = json.loads(path.read_text("utf-8"))["profile_alignment"]["edits"]["station_offsets"]
+    delta = total - 400.0
+    assert edits["40.000"] == 0.7  # 바뀐 구간 앞 — 그대로
+    assert "200.000" not in edits  # 바뀐 구간 안 — 새 지반
+    assert edits[f"{380.0 + delta:.3f}"] == 0.3  # 뒤 — 길이 차만큼 밂
+
+
+def test_저장_확정이_옛_횡단_파일을_지움(tmp_path):
+    from B05_Profile.B05_Profile_Engine_Sections import cross_filename
+
+    long_path = tmp_path / "B06_Section" / "longitudinal" / "longitudinal.json"
+    cross_dir = tmp_path / "B06_Section" / "cross_sections"
+    long_path.parent.mkdir(parents=True)
+    cross_dir.mkdir(parents=True)
+    long_path.write_text(json.dumps({"stations": [{"chainage_m": 0.0}, {"chainage_m": 20.0}]}))
+    for at in (0.0, 20.0, 37.0):
+        (cross_dir / cross_filename(at)).write_text("{}")
+    b06_confirm._prune_cross_files(tmp_path, "B06_Section/longitudinal/longitudinal.json")
+    assert sorted(p.name for p in cross_dir.glob("cross_*.json")) == sorted(
+        cross_filename(at) for at in (0.0, 20.0)
+    )
+    source = (ROOT / "B06_Section" / "B06_Section_Router_Confirm.py").read_text("utf-8")
+    for name in ("async def save_sections(", "async def confirm_sections("):
+        body = source[source.index(name) :]
+        assert body.index("_stand_patch_stations(") < body.index("_prune_cross_files,")