feat(765-3): 전처리 체인 = 지형마다 두 벌 — 다른 지형 먼저 짓고 묶음으로 떠 둠 → 설계 지형 → 추천 미리 셈 한 번 · 배수 늘 다시 셈 · 비우기에 묶음 · 죽은 run_redesign_chain 걷음

- run_preprocess_chain: WF1 자동 확정 · 서피스 바꿈 체인이 부름 · 다른 지형 없음 · 실패 = 한 벌 + meta.json 까닭 · 예상노선(공유)은 설계 지형 벌이 씀
- 자동 체인 배수 refresh=True — 저장분이 노선 바뀐 줄 모르던 틈 막음
- purge_output_folders 가 design_sets 도 지움
- run_redesign_chain(앱에서 부르는 곳 0) 걷음 — 시험 정리: test_342_2 의 그 함수 시험 셋 지움(함수 없음) · test_763_3 · test_351_1 은 두 벌 체인으로 · test_74 · test_user_values_survive 의 이월 차례 검사는 살아 있는 자리(B05_Profile_Route_Save)로 옮김
- 시험 test_765_design_sets 에 체인 차례 · 한 벌 · 다른 벌 실패 · 비우기 넷 덧

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

B03_FileInput/B03_FileInput_Service_Chain.py

@@ -13,12 +13,13 @@ WF1(지표면 분석·자동 확정)이 끝나면 사용자가 화면에 없어
   사용자는 그 페이지에서 수동으로 이어서 진행할 수 있다.
 - 라우터 함수를 직접 호출한다(HTTP 재진입 없음). solve/confirm 엔드포인트는 인증
   의존성이 없는 순수 함수 시그니처라 서버 내부 호출이 가능하다.
+- **지형마다 한 벌**(765-3): 전처리 체인 = `run_preprocess_chain` — 다른 지형 벌을 먼저 지어
+  `design_sets/` 묶음으로 떠 두고 설계 지형 벌을 지음(`common_util_design_sets`).
 """
 
 import asyncio
 import logging
 import time
-from dataclasses import asdict
 from pathlib import Path
 from typing import Any
 from uuid import UUID
@@ -311,7 +312,8 @@ async def run_auto_design_chain(
 
         # 4.5) 배수유역 분석 → 관 지점 확정 → 배관 정착 계획선 재산출.
         #      B06 확정 전에 끝내야 확정 스냅샷에 배관 정착 계획선이 담긴다.
-        await _prepare_drainage_pipes_and_reprofile(project_id, route_id)
+        #      배수는 늘 다시 셈(765-3) — 저장분은 노선이 바뀐 줄 모름(두 벌 · 서피스 바꿈 뒤 묵음).
+        await _prepare_drainage_pipes_and_reprofile(project_id, route_id, refresh=True)
 
         # 5) B06 횡단 설계 확정 — 미지정 측점을 기본값으로 채워 저장. stage 3은
         #    IN_PROGRESS(스텝바 노란 표시)로 남겨 사용자 검토·확정을 기다린다.
@@ -370,11 +372,94 @@ async def run_auto_design_chain(
             schedule_precompute(project_id)
 
 
+async def run_preprocess_chain(
+    project_id: UUID, surface_model_id: int | None = None
+) -> dict[str, Any] | None:
+    """전처리 체인 = 지형마다 설계 한 벌(765-3 · 요구 605 · 606-1) — 결과 = 설계 지형 요약.
+
+    다른 지형 먼저 자동 설계 → 묶음으로 떠 둠(B05~B09 비움) → 설계 지형 자동 설계 → 추천 미리 셈
+    한 번(실패 길에도). 다른 지형이 없으면(LAS 만 · 수치지형도 못 받음 · 수치지형도만) 한 벌 ·
+    묶음 meta.json 에 까닭. 부르는 곳 = WF1 자동 확정 · 서피스 바꿈 체인(B04 다른 서피스 [확정] ·
+    LAS 나중 올리기 뒤 전처리 다시). 묶음은 처음부터 다시(옛 벌 버림).
+    """
+    from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+    from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
+    from common_util import common_util_design_sets as sets
+    from common_util.common_util_initial_snapshot import (
+        clear_designing,
+        mark_designing,
+        read_design_failure,
+    )
+    from common_util.common_util_route_geometry import expected_route_csv_path
+    from common_util.common_util_storage import resolve_stored_project_path
+    from common_util.common_util_surface_confirmation import (
+        design_terrain,
+        get_surface_confirmation_params,
+    )
+    from config.config_db import get_db_pool
+
+    pool = get_db_pool()
+    root: Path | None = None
+    marks = [("시작", time.perf_counter())]
+    try:
+        async with pool.acquire() as connection:
+            stored_path = await get_project_storage_relative_path(connection, project_id)
+            params = await get_surface_confirmation_params(connection, str(project_id))
+        if not stored_path:
+            return None
+        root = Path(resolve_stored_project_path(stored_path))
+        mark_designing(root)
+        sets.discard_sets(root)
+        design = design_terrain(params)
+        other = sets.other_terrain(params, root)
+        if other is None:
+            missing = "sheet" if design == "las" else "las"
+            sets.note_unbuilt(root, missing, f"{missing} 지표면 모델 없음 — 한 벌만")
+        else:
+            # 예상노선(공유)은 설계 지형 벌이 씀 — 다른 지형으로 자른 것이 남지 않게
+            expected = expected_route_csv_path(root)
+            had_expected = expected.is_file()
+            try:
+                async with pool.acquire() as connection:
+                    model_id = await sets.switch_terrain(connection, project_id, other)
+                built = await run_auto_design_chain(project_id, model_id, inner=True)
+                async with pool.acquire() as connection:
+                    fingerprint = await sets.set_fingerprint(
+                        connection, str(project_id), root, other
+                    )
+                    if built and fingerprint:
+                        await sets.park_active_set(connection, root, str(project_id), fingerprint)
+                    else:
+                        await sets.clear_active_set(connection, root, str(project_id))
+                        sets.note_unbuilt(root, other, read_design_failure(root) or "짓기 실패")
+            except Exception as exc:  # noqa: BLE001 — 다른 벌 실패가 설계 지형 벌을 막지 않음
+                logger.exception("두 벌 짓기 — %s 벌 실패: project_id=%s", other, project_id)
+                async with pool.acquire() as connection:
+                    await sets.clear_active_set(connection, root, str(project_id))
+                sets.note_unbuilt(root, other, f"짓기 오류: {exc}")
+            finally:
+                if not had_expected:
+                    expected.unlink(missing_ok=True)
+                async with pool.acquire() as connection:
+                    await sets.switch_terrain(connection, project_id, design)
+            marks.append((f"{other} 벌", time.perf_counter()))
+        summary = await run_auto_design_chain(project_id, surface_model_id, inner=True)
+        marks.append((f"{design} 벌", time.perf_counter()))
+        _log_steps("전처리 체인 두 벌", marks)
+        return summary
+    except Exception:
+        logger.exception("전처리 체인 실패: project_id=%s", project_id)
+        return None
+    finally:
+        if root is not None:
+            clear_designing(root)
+        schedule_precompute(project_id)  # 추천은 지형 · 예정노선만 씀 — 실패 길에도(763-3)
+
+
 async def _ensure_initial_polyline(
     project_id: UUID, project_root: Path, points: list[dict[str, float]]
 ) -> None:
     """계획노선 초기 폴리라인을 세운다 — 이미 있으면 그대로 둔다. 실패는 비치명적."""
-    import asyncio
 
     from B05_Profile.B05_Profile_Router_FollowLand import initial_land
     from B05_Profile.B05_Profile_Router_Replan import _ensure_planned_initial, _min_plan_radius_m
@@ -445,247 +530,3 @@ def redesign_solve_request(
         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,
-    selection: dict[str, Any],
-    profile_target: tuple[list[float], list[float]] | None = None,
-) -> str | None:
-    """B04 재확정 후 — **사용자 입력을 유지한 채** 새 지표면 기준으로 B05·B06 재계산·저장.
-
-    관리자가 B04에서 다른 지표면 모델로 재확정하면(2026-08-04 사용자 확정) 그 값을
-    기준으로 다음 페이지들도 함께 갱신돼야 한다. 이때 일반 사용자가 이미 쓰던 설정은
-    버리지 않는다:
-    - B05: 저장된 stage 2 params(제어점 BP/EP/CP·회피/금지원·경사 옵션·측점 간격 등)를
-      그대로 쓰고 **지표면(filter/method/smooth/model id)만** 새 확정값으로 바꾼다.
-    - B06: 옛 경로의 측점별 설계(지반유형·단면유형·측구·암 경계)를 chainage 매칭으로
-      새 경로에 이월하고, 표준단면 설정(data.options)도 함께 넘긴다. 나머지 미지정
-      측점은 확정 시 기본값으로 채워진다.
-
-    경로가 아예 없으면 신규 자동 체인(계획노선 정본 기본값)으로 되돌아간다.
-
-    `profile_target`(누가거리 · 계획고)이 오면 1차 종단을 그 계획고로 세운다 — 노선 추천
-    「종단 반영」 켬으로 [적용]한 노선의 균형 종단(PLAN 74-4). 없으면 계획고 = 지반고.
-    """
-    from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
-    from B05_Profile.B05_Profile_Repository import get_latest_route
-    from B05_Profile.B05_Profile_Router import (
-        _section_options,
-        confirm_latest_route,
-        solve_route_record,
-    )
-    from B06_Section.B06_Section_Repository import (
-        get_cross_section_designs,
-        get_latest_section_options,
-        get_longitudinal_section,
-    )
-    from B06_Section.B06_Section_Repository_Bulk import merge_cross_section_designs
-    from B06_Section.B06_Section_Router import regenerate_route_sections
-    from B06_Section.B06_Section_Router_Confirm import _rowless_station_chainages, confirm_sections
-    from B06_Section.B06_Section_Router_Design import CARRY_INPUT_KEYS, compute_default_designs
-    from B06_Section.B06_Section_Schema import SectionConfirmRequest
-    from common_util.common_util_initial_snapshot import (
-        clear_designing,
-        discard_initial_snapshot,
-        mark_designing,
-    )
-    from common_util.common_util_storage import resolve_stored_project_path
-    from common_util.common_util_workflow_state import get_workflow_state
-    from config.config_db import get_db_pool
-
-    pool = get_db_pool()
-    project_root: Path | None = None
-    marks = [("시작", time.perf_counter())]
-    try:
-        async with pool.acquire() as connection:
-            latest = await get_latest_route(connection, project_id)
-            stored_path = await get_project_storage_relative_path(connection, project_id)
-        if stored_path:
-            project_root = Path(resolve_stored_project_path(stored_path))
-            mark_designing(project_root)
-            # 지표면이 바뀌면 옛 초기값은 다른 지형 기준이라 더는 기준이 아니다. 여기서
-            # 다시 뜨지는 않는다 — 이 체인은 **사용자 입력을 유지한 채** 재계산하므로 그
-            # 결과를 찍으면 편집이 섞인 가짜 초기값이 된다(2026-08-29 사용자 확정).
-            if discard_initial_snapshot(project_root):
-                logger.info("재확정 체인: 초기값 스냅샷 무효화 (project_id=%s)", project_id)
-        if not latest:
-            logger.info(
-                "재확정 체인 → 기존 경로 없음, 신규 자동 체인으로: project_id=%s", project_id
-            )
-            await run_auto_design_chain(project_id, surface_model_id=surface_model_id)
-            return
-
-        # 1) 사용자 입력 회수 — 마지막 경로 계산의 stage 2 params가 정본이다.
-        old_route_id = int(latest["id"])
-        async with pool.acquire() as connection:
-            import aiomysql
-
-            async with connection.cursor(aiomysql.DictCursor) as cursor:
-                state = await get_workflow_state(cursor, str(project_id))
-            old_longitudinal = 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 or {}).get("stages", []) if int(s.get("stage_no", -1)) == 2), None
-        )
-        params = (stage2 or {}).get("params") or {}
-        points = params.get("points") or {}
-        if not points.get("bp") or not points.get("ep"):
-            logger.warning(
-                "재확정 체인 중단(stage 2 params에 제어점 없음): project_id=%s", project_id
-            )
-            return "노선 제어점(BP·EP)이 없어 재계산할 수 없습니다."
-
-        marks.append(("준비·사용자입력 회수", time.perf_counter()))
-
-        # 2) B05 재계산 — 지표면 관련 값만 새 확정 선택으로 교체, 나머지는 사용자 저장분.
-        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):
-            # 사유까지 남긴다 — 상태 코드만으로는 무엇이 막았는지 못 짚는다(2026-09-06
-            # 노선 편집 [확인]이 조용히 400 으로 끊겨 종횡단만 옛 노선에 남았음).
-            try:
-                reason = bytes(solve_result.body).decode("utf-8", "replace")[:300]
-            except Exception:  # noqa: BLE001 — 로그용이라 실패해도 흐름을 막지 않는다
-                reason = "(본문 없음)"
-            logger.error(
-                "재확정 체인 중단(B05 재계산 실패): project_id=%s status=%s reason=%s",
-                project_id,
-                solve_result.status_code,
-                reason,
-            )
-            return f"노선 재계산 실패({solve_result.status_code}): {reason}"
-        new_route_id = int(solve_result.route_id)
-        marks.append(("노선 재계산(solve)", time.perf_counter()))
-        logger.info(
-            "재확정 체인 B05 재계산 완료: project_id=%s %s→%s",
-            project_id,
-            old_route_id,
-            new_route_id,
-        )
-
-        # 3) 옛 측점별 사용자 설계 이월은 **측점을 다시 만든 뒤**(아래 5) 한다.
-        #    여기서 하면 `_prepare_drainage_pipes_and_reprofile(refresh=True)` 가 측점을
-        #    새로 만들며 덮어써 값이 사라진다(2026-09-06 실측: 다단 구간값 `extra_spans` 가
-        #    노선 변경 뒤 0건이 됐음). 옛 설계는 `old_designs` 로 이미 손에 있다.
-        carried = 0
-
-        # 4) B05 확정 → B06 확정(옛 표준단면 설정 이월, 미지정 측점 기본값 채움).
-        #    재확정 후에도 stage 2·3은 IN_PROGRESS로 남겨 사용자 재검토를 받는다.
-        confirm_result = await confirm_latest_route(project_id, None, mark_stage_complete=False)
-        if isinstance(confirm_result, JSONResponse):
-            logger.error(
-                "재확정 체인 중단(B05 확정 실패): project_id=%s status=%s",
-                project_id,
-                confirm_result.status_code,
-            )
-            return f"노선 확정 실패({confirm_result.status_code})"
-        marks.append(("B05 확정", time.perf_counter()))
-        # 노선이 바뀌었으므로 배수유역·관 지점도 새 노선 기준으로 다시 만든다 — 옛 관 자리로
-        # 계획선을 앉히면 전부 어긋나고, 저장분은 노선 변경을 스스로 알지 못한다.
-        # 건너뛴 첫 생성이 남겼을 옵션(요청 → 옛 저장 → 기본) — 재생성이 그것을 읽어 결과가 같다.
-        seed = asdict(_section_options(request, old_section_options))
-        seed.pop("extra_stations", None)
-        piped = await _prepare_drainage_pipes_and_reprofile(
-            project_id, new_route_id, refresh=True, stored_options=seed
-        )
-        if not piped:
-            # 관 없이라도 종횡단은 세운다 — 새 노선엔 아직 종횡단이 없다(첫 생성을 건너뜀).
-            await regenerate_route_sections(
-                project_id, new_route_id, float(seed["cross_half_width_m"]), stored_options=seed
-            )
-
-        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)
-            marks.append(("균형 종단 반영", time.perf_counter()))
-
-        old_options = ((old_longitudinal or {}).get("data") or {}).get("options") or {}
-        standard = old_options.get("standard_cross_section") or None
-
-        # 5) 이제 측점이 새 노선 기준으로 다 섰다 — 옛 사용자 **입력**만 누가거리로 얹고 전 측점을
-        #    새 계획고로 다시 셈한다(PLAN 74-3). 결과 칸(계획고 · 면적 · design_line …)을 통째
-        #    옮기면 옛 노선 단면이 새 노선에 남아 누가토량이 두 배가 됐다(74-1). 새 노선에 없는
-        #    누가거리는 행을 만들지 않는다.
-        try:
-            inputs = {
-                round(float(record["chainage_m"]), 3): {
-                    key: record["design"][key]
-                    for key in CARRY_INPUT_KEYS
-                    if record["design"].get(key) is not None
-                }
-                for record in old_designs
-                if isinstance(record.get("design"), dict)
-            }
-            async with pool.acquire() as connection:
-                new_longitudinal = await get_longitudinal_section(
-                    connection, project_id, new_route_id
-                )
-            long_path = str((new_longitudinal or {}).get("longitudinal_file_path") or "")
-            chainages = await asyncio.to_thread(
-                _rowless_station_chainages, project_root, long_path, []
-            )
-            entries = await asyncio.to_thread(
-                compute_default_designs, project_root, long_path, chainages, standard, inputs
-            )
-            async with pool.acquire() as connection:
-                await connection.begin()
-                try:
-                    carried = await merge_cross_section_designs(
-                        connection,
-                        route_id=new_route_id,
-                        entries=entries,
-                        replace=True,
-                        project_id=project_id,
-                    )
-                    await connection.commit()
-                except Exception:
-                    await connection.rollback()
-                    raise
-        except Exception:
-            logger.exception(
-                "재확정 체인 — 옛 설계 이월 실패(계속 진행): project_id=%s", project_id
-            )
-        logger.info("재확정 체인 설계 이월: project_id=%s %d건", project_id, carried)
-
-        marks.append(("설계 이월", time.perf_counter()))
-
-        section_request = (
-            SectionConfirmRequest(standard_cross_section=standard) if standard else None
-        )
-        sections_result = await confirm_sections(
-            project_id, new_route_id, section_request, mark_stage_complete=False
-        )
-        if isinstance(sections_result, JSONResponse):
-            logger.error(
-                "재확정 체인 중단(B06 확정 실패): project_id=%s status=%s",
-                project_id,
-                sections_result.status_code,
-            )
-            return f"종횡단 확정 실패({sections_result.status_code})"
-        marks.append(("B06 확정(서버 재계산 포함)", time.perf_counter()))
-        _log_steps("재확정 체인", marks)
-        logger.info(
-            "재확정 체인 완료: project_id=%s route %s→%s (설계 %d건 이월)",
-            project_id,
-            old_route_id,
-            new_route_id,
-            carried,
-        )
-        return None
-    except Exception as exc:
-        logger.exception("재확정 체인 실패: project_id=%s", project_id)
-        return f"재계산 중 오류: {exc}"
-    finally:
-        if project_root is not None:
-            clear_designing(project_root)
-        # 성공 · 실패 · 건너뜀 어느 쪽이든 추천 미리 셈 예약(763-3) — 추천은 지형 · 예정노선만 씀
-        from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
-
-        schedule_precompute(project_id)

B03_FileInput/B03_FileInput_Service_SurfaceChange.py

@@ -1,8 +1,9 @@
 """서피스가 바뀐 뒤 전처리 체인 다시(763-3 · 요구 605) — 업로드와 같은 길 하나.
 
-부르는 곳 = B04 다른 서피스 [확정]. B05~B09 초기화 = 같은 지형 재업로드 길
-(`purge_project_outputs(keep_preprocess=True)` · 단계 2 이후 처음 · 초기값 스냅샷 버림) →
-`run_auto_design_chain`(끝에 추천 미리 셈 · 실패 길에도). 전처리 결과(B04) · 확정은 그대로.
+부르는 곳 = B04 다른 서피스 [확정] · B05 LAS 나중 올리기 뒤 전처리 다시(765-5). B05~B09 초기화 =
+같은 지형 재업로드 길(`purge_project_outputs(keep_preprocess=True)` · 설계 벌 묶음까지 · 단계 2 이후
+처음 · 초기값 스냅샷 버림) → `run_preprocess_chain`(지형마다 한 벌 · 끝에 추천 미리 셈 ·
+실패 길에도). 전처리 결과(B04) · 확정은 그대로.
 """
 
 import asyncio
@@ -16,7 +17,7 @@ logger = logging.getLogger(__name__)
 async def run_surface_change_chain(project_id: UUID, surface_model_id: int | None) -> None:
     """B05~B09 를 지우고 자동 설계 체인을 새 확정 서피스로 다시 탐 — 실패는 로그만."""
     from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
-    from B03_FileInput.B03_FileInput_Service_Chain import run_auto_design_chain
+    from B03_FileInput.B03_FileInput_Service_Chain import run_preprocess_chain
     from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
     from common_util.common_util_initial_snapshot import discard_initial_snapshot
     from common_util.common_util_project_reset import purge_project_outputs
@@ -45,4 +46,4 @@ async def run_surface_change_chain(project_id: UUID, surface_model_id: int | Non
         schedule_precompute(project_id)
         return
     logger.info("서피스 바꿈 체인 — B05~B09 초기화 뒤 자동 설계: project_id=%s", project_id)
-    await run_auto_design_chain(project_id, surface_model_id=surface_model_id)
+    await run_preprocess_chain(project_id, surface_model_id)

B03_FileInput/B03_FileInput_Service_WF1.py

@@ -217,11 +217,10 @@ async def trigger_wf1_analysis_and_email(
         # 횡단 설계 체인을 잇는다(2026-08-04 사용자 확정). 체인은 실패를 스스로 격리한다.
         design_summary: dict[str, Any] | None = None
         if auto_confirmed and AUTO_DESIGN_CHAIN_ENABLED:
-            from B03_FileInput.B03_FileInput_Service_Chain import run_auto_design_chain
+            from B03_FileInput.B03_FileInput_Service_Chain import run_preprocess_chain
 
-            design_summary = await run_auto_design_chain(
-                project_id, surface_model_id=confirmed_model_id
-            )
+            # 지형마다 설계 한 벌(765-3) — 다른 지형 벌은 묶음 · 요약은 설계 지형 벌
+            design_summary = await run_preprocess_chain(project_id, confirmed_model_id)
 
         # 알림 메일은 여기 한 통뿐이다 — 업로드 직후와 분석 직후로 나눠 두 통을 보내던 것을
         # 초기 설계(B04~B06)까지 마친 시점의 통합 메일로 합쳤다(2026-08-08 사용자 지시).

common_util/common_util_project_reset.py

@@ -46,9 +46,17 @@ PREPROCESS_STAGE_DIR = "B04_PreProcess"
 _PREPROCESS_TABLES = frozenset({"surface_models", "processed_point_cloud"})
 
 
+#: 지형마다 설계 한 벌의 쉬는 벌 묶음(765-3) — 산출물이라 늘 함께 지움
+DESIGN_SETS_DIR = "design_sets"
+
+
 def purge_output_folders(project_root: Path, keep_preprocess: bool = False) -> list[str]:
-    """단계별 산출물 폴더를 비운다. 지운 폴더 이름을 돌려준다."""
+    """단계별 산출물 폴더 · 설계 벌 묶음을 비운다. 지운 폴더 이름을 돌려준다."""
     removed: list[str] = []
+    sets = project_root / DESIGN_SETS_DIR
+    if sets.exists():
+        shutil.rmtree(sets, ignore_errors=True)
+        removed.append(DESIGN_SETS_DIR)
     for name in OUTPUT_STAGE_DIRS:
         if keep_preprocess and name == PREPROCESS_STAGE_DIR:
             continue

resources/tester/test_342_2_chain_once.py

@@ -1,246 +1,22 @@
 # -*- coding: utf-8 -*-
-"""342-2 노선 [확인] 체인 — 종횡단 생성 한 번 · 자동 체인 재계산 한 번 · 3D 는 [저장]·[확정] 끝.
+"""342-2 자동 체인 재계산 한 번 · 3D 는 [저장]·[확정] 끝.
 
-① 노선 [확인] 체인(`run_redesign_chain`)이 종횡단을 **한 번만** 만들고, 그 한 번이 옛 흐름의
-   마지막 생성(4c `regenerate_sections`)과 **같은 입력**(옵션 · 반폭 · 계획선 기준)을 받는다.
-   옛 흐름은 solve 안 첫 생성이 남긴 옵션을 4c 가 읽었다 — 그 옵션을 체인이 넘겨 같게 둠.
-② 배수유역이 실패해도 종횡단은 선다(첫 생성을 건너뛰었으므로 체인이 한 번 세움).
 ③ 자동 체인은 서버 재계산 · 3D 짓기를 따로 부르지 않음 — [확정](confirm_sections) 끝 한 번씩 ·
    재계산 → 짓기 차례 · [저장](save_sections)도 끝에 짓기.
 
-DB · 엔진은 가짜로 바꿔 체인의 차례와 생성 입력만 잰다.
+(①② 노선 [확인] 체인 `run_redesign_chain` 시험 셋은 그 함수를 걷어(765-3 · 부르는 곳 0) 함께 지움.)
 """
 
 from __future__ import annotations
 
-import asyncio
 import re
 import sys
-from dataclasses import asdict
 from pathlib import Path
-from types import SimpleNamespace
-from uuid import uuid4
-
-import pytest
 
 ROOT = Path(__file__).resolve().parents[2]
 if str(ROOT) not in sys.path:
     sys.path.insert(0, str(ROOT))
 
-import B03_FileInput.B03_FileInput_Repository as b03_repo  # noqa: E402
-import B03_FileInput.B03_FileInput_Service_Chain as chain  # noqa: E402
-import B04_PreProcess.B04_PreProcess_Router_Basins as basins  # noqa: E402
-import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed  # noqa: E402
-import B05_Profile.B05_Profile_Repository as b05_repo  # noqa: E402
-import B05_Profile.B05_Profile_Router as b05_router  # noqa: E402
-import B06_Section.B06_Section_Repository as b06_repo  # noqa: E402
-import B06_Section.B06_Section_Repository_Bulk as b06_bulk  # 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 B06_Section.B06_Section_Router_Design as b06_design  # noqa: E402
-import common_util.common_util_initial_snapshot as snapshot  # noqa: E402
-import common_util.common_util_storage as storage  # noqa: E402
-import common_util.common_util_workflow_state as workflow  # noqa: E402
-import config.config_db as config_db  # noqa: E402
-from B05_Profile.B05_Profile_Schema import RouteSolveRequest  # noqa: E402
-from fastapi.responses import JSONResponse  # noqa: E402
-
-# stage 2 params(요청 쪽) · 옛 노선 종단 옵션(DB 쪽)이 **엇갈리게** — 옛 흐름 순서를 못 지키면 값이 갈린다
-PARAMS = {
-    "filter_key": "csf",
-    "method": "dtm",
-    "smooth": False,
-    "points": {"bp": {"x": 0.0, "y": 0.0}, "ep": {"x": 100.0, "y": 0.0}, "cp": []},
-    "options": {"grade_class": "trunk", "terrain_type": "normal"},
-    "cross_half_width_m": 20.0,
-    "cross_sample_interval_m": 0.5,
-    "station_interval_m": None,
-    "long_sample_interval_m": None,
-}
-OLD_STORED = {
-    "station_interval_m": 25.0,
-    "cross_half_width_m": 30.0,
-    "cross_sample_interval_m": 1.0,
-    "long_sample_interval_m": 2.0,
-}
-
-
-class _Cursor:
-    async def __aenter__(self):
-        return self
-
-    async def __aexit__(self, *exc):
-        return False
-
-    async def execute(self, *args, **kwargs):
-        return None
-
-    async def fetchone(self):
-        return None
-
-
-class _Connection:
-    def cursor(self, *args):
-        return _Cursor()
-
-    async def begin(self):
-        return None
-
-    async def commit(self):
-        return None
-
-    async def rollback(self):
-        return None
-
-
-class _Acquire:
-    async def __aenter__(self):
-        return _Connection()
-
-    async def __aexit__(self, *exc):
-        return False
-
-
-class _Pool:
-    def acquire(self):
-        return _Acquire()
-
-
-def _async(value):
-    async def fake(*args, **kwargs):
-        return value
-
-    return fake
-
-
-@pytest.fixture
-def world(monkeypatch, tmp_path):
-    calls: dict = {"generate": [], "order": [], "latest_options": 0}
-
-    async def solve(project_id, request, *, generate_sections=True):
-        calls["order"].append("solve")
-        calls["solve_generates"] = generate_sections
-        return SimpleNamespace(route_id=6, total_length_m=100.0)
-
-    async def primary_region(project_id, refresh=False):
-        calls["order"].append("drainage")
-        return calls.get("region", {})
-
-    async def latest_options(connection, project_id):
-        calls["latest_options"] += 1
-        return OLD_STORED
-
-    def generate(project_root, route_data_path, filter_key, method, smooth, **kwargs):
-        calls["order"].append("generate")
-        calls["generate"].append(kwargs)
-        data = {"length_m": 100.0, "options": {}}
-        return {
-            "longitudinal": {"data": data, "file_path": "long.json"},
-            "cross_sections": [],
-            "result": {"longitudinal": {}, "cross_sections": []},
-        }
-
-    async def confirm(project_id, route_id, request=None, mark_stage_complete=True):
-        calls["order"].append("confirm_sections")
-        return SimpleNamespace(route_id=route_id)
-
-    state = {"stages": [{"stage_no": 2, "params": PARAMS}]}
-    monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool())
-    monkeypatch.setattr(b05_repo, "get_latest_route", _async({"id": 5}))
-    monkeypatch.setattr(b03_repo, "get_project_storage_relative_path", _async("p"))
-    monkeypatch.setattr(storage, "resolve_stored_project_path", lambda path: str(tmp_path))
-    monkeypatch.setattr(snapshot, "mark_designing", lambda root: None)
-    monkeypatch.setattr(snapshot, "clear_designing", lambda root: None)
-    monkeypatch.setattr(snapshot, "discard_initial_snapshot", lambda root: False)
-    monkeypatch.setattr(workflow, "get_workflow_state", _async(state))
-    monkeypatch.setattr(
-        b06_repo, "get_longitudinal_section", _async({"data": {}, "longitudinal_file_path": "x"})
-    )
-    monkeypatch.setattr(b06_repo, "get_cross_section_designs", _async([]))
-    monkeypatch.setattr(b06_repo, "get_latest_section_options", latest_options)
-    monkeypatch.setattr(b05_router, "solve_route_record", solve)
-    monkeypatch.setattr(b05_router, "confirm_latest_route", _async({}))
-    monkeypatch.setattr(watershed, "get_primary_region", primary_region)
-    monkeypatch.setattr(basins, "put_pipe_points", _async({"saved_count": 2}))
-    monkeypatch.setattr(b06_confirm, "_rowless_station_chainages", lambda *a: [])
-    monkeypatch.setattr(b06_confirm, "confirm_sections", confirm)
-    monkeypatch.setattr(b06_design, "compute_default_designs", lambda *a: [])
-    monkeypatch.setattr(b06_bulk, "merge_cross_section_designs", _async(0))
-    # 재생성 본체(B06_Section_Router) — DB 는 가짜 · 생성 입력만 잰다
-    monkeypatch.setattr(b06_router, "get_db_pool", lambda: _Pool())
-    monkeypatch.setattr(
-        b06_router,
-        "get_route_generation_source",
-        _async({"route_data_path": "route.geojson", "crs_epsg": 5186}),
-    )
-    monkeypatch.setattr(
-        b06_router,
-        "get_surface_confirmation_params",
-        _async({"source_filter": "csf", "method": "dtm", "smooth": False}),
-    )
-    monkeypatch.setattr(b06_router, "get_project_storage_relative_path", _async("p"))
-    monkeypatch.setattr(b06_router, "get_latest_section_options", latest_options)
-    monkeypatch.setattr(b06_router, "get_workflow_state", _async(state))
-    monkeypatch.setattr(b06_router, "resolve_stored_project_path", lambda path: str(tmp_path))
-    monkeypatch.setattr(b06_router, "run_section_generation", generate)
-    monkeypatch.setattr(b06_router, "delete_sections_for_route", _async(None))
-    monkeypatch.setattr(b06_router, "create_longitudinal_section", _async(None))
-    monkeypatch.setattr(b06_router, "insert_cross_sections", _async(None))
-    monkeypatch.setattr(b06_router, "attach_saved_shapes", lambda *a: None)
-    return calls
-
-
-def _old_flow_options():
-    """옛 흐름 — solve 첫 생성 옵션(요청 → 옛 저장 → 기본)을 4c 가 DB 에서 읽어 다시 고른 값."""
-    request = RouteSolveRequest(
-        filter_key="csf",
-        method="dtm",
-        smooth=False,
-        algorithm="as_planned",
-        bp=PARAMS["points"]["bp"],
-        ep=PARAMS["points"]["ep"],
-        cross_half_width_m=PARAMS["cross_half_width_m"],
-        cross_sample_interval_m=PARAMS["cross_sample_interval_m"],
-    )
-    first = b05_router._section_options(request, OLD_STORED)
-    stored_by_first = {**asdict(first), "extra_stations": []}
-    return b06_router._regeneration_options(
-        stored_by_first, PARAMS, float(stored_by_first["cross_half_width_m"])
-    )
-
-
-def test_종횡단은_한_번_관_뒤에(world):
-    failure = asyncio.run(chain.run_redesign_chain(uuid4(), 1, {}))
-    assert failure is None
-    assert world["solve_generates"] is False
-    assert len(world["generate"]) == 1
-    order = world["order"]
-    assert order.index("solve") < order.index("drainage") < order.index("generate")
-    assert order.index("generate") < order.index("confirm_sections")
-
-
-def test_한_번이_옛_마지막_생성과_같은_입력(world):
-    asyncio.run(chain.run_redesign_chain(uuid4(), 1, {}))
-    got = world["generate"][0]
-    expected = _old_flow_options()
-    assert got["options"] == expected
-    # 옛 노선 저장 반폭(30)이 아니라 요청 반폭(20) — 첫 생성이 남기던 값
-    assert got["options"].cross_half_width_m == 20.0
-    assert got["options"].cross_sample_interval_m == 0.5
-    assert got["options"].station_interval_m == 25.0
-    assert got["grade_options"] == b06_router._regeneration_grade_options(PARAMS)
-    assert "grade_overrides" not in got  # 옛 4c 도 안 넘김 — 저장본 grade_overrides 빈 값 그대로
-    # 생성 쪽은 DB 최신 종단(=옛 노선)을 다시 읽지 않음 — 체인이 넘긴 옵션
-    assert world["latest_options"] == 1  # 체인이 옛 노선 옵션을 읽은 한 번뿐
-
-
-def test_배수유역이_실패해도_종횡단은_선다(world):
-    world["region"] = JSONResponse(status_code=500, content={})
-    failure = asyncio.run(chain.run_redesign_chain(uuid4(), 1, {}))
-    assert failure is None
-    assert len(world["generate"]) == 1
-    assert world["generate"][0]["options"] == _old_flow_options()
-
 
 def _body(source: str, name: str) -> str:
     start = source.index(f"async def {name}(")

resources/tester/test_351_1_recommend_gaps.py

@@ -90,10 +90,12 @@ def test_기본값_판정() -> None:
 
 def test_두_체인_끝에서_미리_셈_예약() -> None:
     source = Path("B03_FileInput/B03_FileInput_Service_Chain.py").read_text(encoding="utf-8")
-    redesign = source[source.index("async def run_redesign_chain(") :]
-    upload = source[: source.index("async def run_redesign_chain(")]
-    # 두 체인 모두 finally — 실패 · 건너뜀 길에도 예약(763-3 · 실행 시험 = test_763_3)
-    for body in (upload, redesign):
+    start = source.index("async def run_auto_design_chain(")
+    upload = source[start : source.index("async def run_preprocess_chain(")]
+    preprocess = source[source.index("async def run_preprocess_chain(") :]
+    # 두 체인 모두 finally — 실패 · 건너뜀 길에도 예약(763-3 · 실행 시험 = test_763_3) ·
+    # 옛 재확정 체인(run_redesign_chain)은 걷음(765-3) · 두 벌 체인이 자리를 이음
+    for body in (upload, preprocess):
         assert "schedule_precompute(project_id)" in body.split("    finally:", 1)[1]
     b01 = Path("B01_Dashboard/B01_Dashboard_Router.py").read_text(encoding="utf-8")
     assert "refresh_expected_route(" in b01

resources/tester/test_74_route_confirm_initial.py

@@ -32,9 +32,10 @@ def test_이월은_사용자_입력만() -> None:
 
 
 def test_체인이_옛_설계를_통째로_옮기지_않는다() -> None:
-    source = (ROOT / "B03_FileInput" / "B03_FileInput_Service_Chain.py").read_text(encoding="utf-8")
+    # 노선 저장 체인(`run_route_save_chain`)이 이월 자리 — 옛 재확정 체인은 걷음(765-3)
+    source = (ROOT / "B05_Profile" / "B05_Profile_Route_Save.py").read_text(encoding="utf-8")
     assert 'record["design"])' not in source  # 옛 (누가거리, design 통째) 이월
-    assert "compute_default_designs, project_root, long_path, chainages, standard, inputs" in (
+    assert "compute_default_designs, project_root, relative, chainages, standard, inputs" in (
         " ".join(source.split())
     )
 

resources/tester/test_763_3_surface_change.py

@@ -188,7 +188,7 @@ def test_서피스_바꿈_체인_차례(tmp_path, monkeypatch) -> None:
         log.append(("stages", first_stage))
 
     async def auto(_pid, surface_model_id=None):
-        log.append(("auto", surface_model_id))
+        log.append(("preprocess", surface_model_id))
 
     monkeypatch.setattr(db, "get_db_pool", lambda: _Pool([]))
     monkeypatch.setattr(repo, "get_project_storage_relative_path", _async("x"))
@@ -196,10 +196,10 @@ def test_서피스_바꿈_체인_차례(tmp_path, monkeypatch) -> None:
     monkeypatch.setattr(snapshot, "discard_initial_snapshot", lambda _r: log.append("snapshot"))
     monkeypatch.setattr(reset, "purge_project_outputs", purge)
     monkeypatch.setattr(state, "reset_stages_after_input_change", stages)
-    monkeypatch.setattr(chain, "run_auto_design_chain", auto)
+    monkeypatch.setattr(chain, "run_preprocess_chain", auto)
     monkeypatch.setattr(router, "schedule_precompute", lambda pid: log.append("precompute"))
     asyncio.run(change.run_surface_change_chain(uuid4(), 8))
-    assert log == ["snapshot", ("purge", True), ("stages", 2), ("auto", 8)]
+    assert log == ["snapshot", ("purge", True), ("stages", 2), ("preprocess", 8)]
 
     async def broken(*_a, **_k):
         raise RuntimeError("db")
@@ -210,7 +210,7 @@ def test_서피스_바꿈_체인_차례(tmp_path, monkeypatch) -> None:
     assert log == ["snapshot", "precompute"]  # 초기화 실패 = 체인 없음 · 추천만 새 서피스로
 
 
-@pytest.mark.parametrize("name", ["run_auto_design_chain", "run_redesign_chain"])
+@pytest.mark.parametrize("name", ["run_auto_design_chain", "run_preprocess_chain"])
 def test_체인_실패_길에도_미리_셈(monkeypatch, name) -> None:
     import B03_FileInput.B03_FileInput_Service_Chain as chain
     import B05_Profile.B05_Profile_Router_RouteInitial as router
@@ -224,8 +224,7 @@ def test_체인_실패_길에도_미리_셈(monkeypatch, name) -> None:
     monkeypatch.setattr(db, "get_db_pool", lambda: _Broken())
     monkeypatch.setattr(router, "schedule_precompute", scheduled.append)
     project = uuid4()
-    args = (project,) if name == "run_auto_design_chain" else (project, 1, {})
-    asyncio.run(getattr(chain, name)(*args))
+    asyncio.run(getattr(chain, name)(project))
     assert scheduled == [project]
 
 

resources/tester/test_765_design_sets.py

@@ -411,3 +411,109 @@ def test_맞바꿈_실패면_통째_되돌림(monkeypatch, tmp_path: Path) -> No
     assert _tree(tmp_path) == before_files
     assert sets.read_meta(tmp_path, "sheet") == sheet_meta  # 묶음도 그대로
     assert not (tmp_path / SETS_DIR / "las").exists()
+
+
+# ── 765-3 전처리 체인 = 두 벌 ─────────────────────────────────────────────
+
+
+def _chain_world(monkeypatch, tmp_path: Path, *, other, built_other=True):
+    import B03_FileInput.B03_FileInput_Repository as repo
+    import B03_FileInput.B03_FileInput_Service_Chain as chain
+    import B05_Profile.B05_Profile_Router_RouteInitial as router
+    import common_util.common_util_storage as storage
+    import config.config_db as config_db
+
+    log: list = []
+    expected = tmp_path / "B05_Profile" / "route" / "expected_route.csv"
+
+    async def auto(pid, model_id=None, inner=False):
+        log.append(("auto", model_id, inner))
+        expected.parent.mkdir(parents=True, exist_ok=True)
+        expected.write_text(str(model_id), encoding="utf-8")  # 체인이 처음 쓰는 예상노선
+        return {"route_id": 1} if (built_other or len(log) > 2) else None
+
+    async def switch(_c, _pid, terrain):
+        log.append(("switch", terrain))
+        return 6 if terrain == "sheet" else 5
+
+    async def park(_c, _root, _pid, fingerprint):
+        log.append(("park", fingerprint["terrain"]))
+
+    async def clear(_c, _root, _pid):
+        log.append("clear")
+
+    async def fingerprint(_c, _pid, _root, terrain):
+        return {**FP, "terrain": terrain}
+
+    async def params(_c, _pid):
+        return {"source_filter": "classification"}
+
+    monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(_Conn(sqlite3.connect(":memory:"))))
+    monkeypatch.setattr(repo, "get_project_storage_relative_path", lambda *_a: _done("p"))
+    monkeypatch.setattr(storage, "resolve_stored_project_path", lambda _p: str(tmp_path))
+    import common_util.common_util_surface_confirmation as confirmation
+
+    monkeypatch.setattr(confirmation, "get_surface_confirmation_params", params)
+    monkeypatch.setattr(sets, "other_terrain", lambda _p, _r: other)
+    monkeypatch.setattr(sets, "switch_terrain", switch)
+    monkeypatch.setattr(sets, "park_active_set", park)
+    monkeypatch.setattr(sets, "clear_active_set", clear)
+    monkeypatch.setattr(sets, "set_fingerprint", fingerprint)
+    monkeypatch.setattr(chain, "run_auto_design_chain", auto)
+    monkeypatch.setattr(router, "schedule_precompute", lambda pid: log.append("precompute"))
+    return chain, log, expected
+
+
+async def _done(value):
+    return value
+
+
+def test_전처리_체인_다른_지형_먼저_떠_두고_설계_지형(monkeypatch, tmp_path: Path) -> None:
+    chain, log, expected = _chain_world(monkeypatch, tmp_path, other="sheet")
+    (tmp_path / SETS_DIR / "las").mkdir(parents=True)  # 옛 묶음 = 버림
+    summary = asyncio.run(chain.run_preprocess_chain(PID, 5))  # type: ignore[arg-type]
+    assert summary == {"route_id": 1}
+    assert log == [
+        ("switch", "sheet"),
+        ("auto", 6, True),
+        ("park", "sheet"),
+        ("switch", "las"),
+        ("auto", 5, True),
+        "precompute",
+    ]
+    assert expected.read_text(encoding="utf-8") == "5"  # 예상노선 = 설계 지형 벌이 씀
+    assert not (tmp_path / SETS_DIR / "las").exists()
+    assert not (tmp_path / "initial_design.lock").exists()
+
+
+def test_전처리_체인_다른_지형_없으면_한_벌(monkeypatch, tmp_path: Path) -> None:
+    chain, log, _ = _chain_world(monkeypatch, tmp_path, other=None)
+    asyncio.run(chain.run_preprocess_chain(PID, 5))  # type: ignore[arg-type]
+    assert log == [("auto", 5, True), "precompute"]
+    meta = sets.read_meta(tmp_path, "sheet")
+    assert meta["built"] is False and "sheet" in meta["reason"]
+
+
+def test_전처리_체인_다른_벌_실패해도_설계_지형(monkeypatch, tmp_path: Path) -> None:
+    chain, log, _ = _chain_world(monkeypatch, tmp_path, other="sheet", built_other=False)
+    summary = asyncio.run(chain.run_preprocess_chain(PID, 5))  # type: ignore[arg-type]
+    assert summary == {"route_id": 1}
+    assert log == [
+        ("switch", "sheet"),
+        ("auto", 6, True),
+        "clear",
+        ("switch", "las"),
+        ("auto", 5, True),
+        "precompute",
+    ]
+    assert sets.read_meta(tmp_path, "sheet")["built"] is False
+
+
+def test_비우기에_묶음_포함(tmp_path: Path) -> None:
+    from common_util.common_util_project_reset import purge_output_folders
+
+    (tmp_path / SETS_DIR / "sheet" / "files").mkdir(parents=True)
+    (tmp_path / "B04_PreProcess" / "models").mkdir(parents=True)
+    removed = purge_output_folders(tmp_path, keep_preprocess=True)
+    assert SETS_DIR in removed and not (tmp_path / SETS_DIR).exists()
+    assert (tmp_path / "B04_PreProcess" / "models").is_dir()

resources/tester/test_user_values_survive.py

@@ -13,7 +13,6 @@
 """
 
 import math
-import re
 import sys
 from pathlib import Path
 
@@ -192,13 +191,13 @@ def test_토사에서_켠_2단_절토가_왕복해도_남는다():
 # ── ⑤ 노선 변경 때 옛 사용자 설계가 덮여 사라지던 것 (`908c75d5`) ───────────────
 # 이월을 측점 재생성 **앞**에서 하면 재생성이 덮어썼다. 순서가 곧 고침이라 순서를 지킨다.
 def test_이월은_측점_재생성_뒤에_한다():
-    source = (PROJECT_ROOT / "B03_FileInput" / "B03_FileInput_Service_Chain.py").read_text(
+    # 노선 저장 체인이 이월 자리 — 옛 재확정 체인(run_redesign_chain)은 걷음(765-3)
+    source = (PROJECT_ROOT / "B05_Profile" / "B05_Profile_Route_Save.py").read_text(
         encoding="utf-8"
     )
-    rebuild = re.search(
-        r"_prepare_drainage_pipes_and_reprofile\(\s*project_id, new_route_id", source
-    ).start()
-    carry = source.index("merge_cross_section_designs(")
+    body = source[source.index("async def run_route_save_chain(") :]
+    rebuild = body.index("regenerate_route_sections(")
+    carry = body.index("_carry_cross(")
     assert carry > rebuild, "이월이 측점 재생성보다 앞섰음 — 덮어써서 사라진다"