feat(765-2): 지형마다 설계 한 벌 묶음 모듈 — 떠 두기 · 꺼내 끼우기(새 노선 id · 자식 다시 잇기 · 코리도 이름) · 지문 · 맞바꿈(실패 = 통째 되돌림)

- common_util_design_sets: 활성 벌 = 제자리 · 쉬는 벌 = design_sets/<las|sheet>/(db.json · files/ · initial_snapshot/ · meta.json)
- 벌 몫 = B05 노선 정본 · 코리도 · B06~B09 · 배수 노선 몫 · 초기값 · 공유 캐시(비용면 · 예정노선 · 하강장 · 도엽 좌표 · 강우)는 안 옮김 · 안 지움
- 옮김 = 같은 볼륨 이름 바꿈 · 단계 2 이후 행 · projects.status 함께
- 묶음 없음 · 지문 어긋남 = 그 자리에서 자동 설계 체인 한 번(inner = 마커 · 미리 셈은 부르는 쪽)
- 시험 test_765_design_sets 6 (왕복 값 같음 · 새 id · 묵은 벌 · 실패 되돌림)

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

B03_FileInput/B03_FileInput_Service_Chain.py

@@ -155,10 +155,12 @@ async def _prepare_drainage_pipes_and_reprofile(
 
 
 async def run_auto_design_chain(
-    project_id: UUID, surface_model_id: int | None = None
+    project_id: UUID, surface_model_id: int | None = None, *, inner: bool = False
 ) -> dict[str, Any] | None:
     """B05 기본 경로 계산·확정 → B06 기본 횡단 설계 확정을 기본값으로 이어 실행한다.
 
+    `inner` = 두 벌 짓기 · 지형 칩 안에서 부름(765) — 진입 막음 마커 · 추천 미리 셈은 부르는 쪽.
+
     WF1 자동 확정 직후 같은 백그라운드 태스크에서 호출된다. 어떤 단계가 실패해도
     예외를 밖으로 던지지 않는다 — 로그와 각 단계의 workflow 상태 기록으로 남긴다.
 
@@ -359,12 +361,13 @@ async def run_auto_design_chain(
             mark_design_failed(project_root, f"초기 설계 처리 중 오류: {exc}")
     finally:
         # 성공·실패·중단 어느 쪽이든 문은 연다 — 마커가 남으면 영영 못 들어간다.
-        if project_root is not None:
+        if project_root is not None and not inner:
             clear_designing(project_root)
         # 성공 · 실패 · 건너뜀 어느 쪽이든 추천 미리 셈 예약(763-3) — 추천은 지형 · 예정노선만 씀
-        from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
+        if not inner:
+            from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
 
-        schedule_precompute(project_id)
+            schedule_precompute(project_id)
 
 
 async def _ensure_initial_polyline(

common_util/common_util_design_sets.py

@@ -0,0 +1,458 @@
+"""지형마다 설계 한 벌(765 · 요구 605 · 606-1) — 활성 벌은 제자리(DB · 파일) · 쉬는 벌은 묶음.
+
+묶음 = `<프로젝트>/design_sets/<las|sheet>/`
+  db.json           노선과 자식 넷 · 단계 2 이후 행 · projects.status
+  files/            벌 몫 파일(프로젝트 뿌리 기준 상대 경로 그대로)
+  initial_snapshot/ 그 벌의 초기값([초기화])
+  meta.json         지형 · 지문(필터 · 표현 · 평활 · 모델 파일 내용) · 옮긴 자리 · 못 지은 까닭
+벌 몫 = B05 노선 정본 · 코리도 · B06~B09 · 배수 결과 중 노선 몫 · 초기값.
+공유(안 옮김 · 안 지움) = B05 비용면 · 예정노선 · 추천 고름 · 배수 공유 캐시
+  (도엽 좌표 · 하강장 · 강우) · 그 밖 전부.
+떠 두기 = 벌 몫을 묶음으로 옮기고(같은 볼륨 = 이름 바꿈) DB 노선 지움 · 단계 2 이후 처음.
+꺼내 끼우기 = 새 노선 id 로 넣고 자식 다시 잇기 · 코리도 이름(`rekey_id_files`) · 묶음 지움.
+지형 칩 = 맞바꿈(`swap_design_set`) — 실패 = `save_guard` 로 통째 되돌림 · 묶음 없음 · 지문 어긋남 =
+그 자리에서 자동 설계 체인으로 한 번 지음. 계산 자리 = ③ 서버 단독.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import shutil
+import time
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+import aiomysql
+
+from common_util.common_util_initial_snapshot import (
+    DRAINAGE_SHARED_PREFIXES,
+    SNAPSHOT_DIRNAME,
+    _json_safe,
+)
+from common_util.common_util_surface_confirmation import (
+    AnalysisTerrain,
+    analysis_terrain_params,
+    analysis_terrain_sources,
+    design_terrain,
+    get_surface_confirmation_params,
+)
+
+SETS_DIRNAME = "design_sets"
+MODELS_DIR = Path("B04_PreProcess") / "models"
+TERRAINS: tuple[AnalysisTerrain, AnalysisTerrain] = ("las", "sheet")
+#: 통째로 벌 몫인 폴더
+_SET_DIRS = (
+    "B05_Profile/corridor",
+    "B06_Section",
+    "B07_DesignDetail",
+    "B08_Quantity",
+    "B09_Estimation",
+    SNAPSHOT_DIRNAME,
+)
+#: 공유를 뺀 나머지가 벌 몫인 폴더 → 공유 이름 앞머리
+_SPLIT_DIRS = {
+    "B05_Profile/route": ("cost_surface_", "expected_route", "initial_choice"),
+    "B04_PreProcess/drainage": DRAINAGE_SHARED_PREFIXES,
+}
+_ROUTE_CHILDREN = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")
+_STAGE_COLUMNS = (
+    "state",
+    "progress_percent",
+    "params",
+    "message",
+    "started_at",
+    "completed_at",
+)
+#: 벌이 되는 단계(노선 · 횡단 · 상세 · 수량 · 내역)
+FIRST_SET_STAGE = 2
+
+
+def sets_dir(root: Path) -> Path:
+    return Path(root) / SETS_DIRNAME
+
+
+def bundle_dir(root: Path, terrain: str) -> Path:
+    return sets_dir(root) / terrain
+
+
+def set_entries(root: Path) -> list[str]:
+    """지금 제자리에 있는 벌 몫(프로젝트 뿌리 기준 상대 경로 · 폴더 또는 파일)."""
+    root = Path(root)
+    out = [name for name in _SET_DIRS if (root / name).exists()]
+    for folder, shared in _SPLIT_DIRS.items():
+        base = root / folder
+        if base.is_dir():
+            out += [
+                f"{folder}/{path.name}"
+                for path in sorted(base.iterdir())
+                if not path.name.startswith(shared)
+            ]
+    return out
+
+
+def _bundle_path(bundle: Path, relative: str) -> Path:
+    if relative == SNAPSHOT_DIRNAME:
+        return bundle / SNAPSHOT_DIRNAME
+    return bundle / "files" / relative
+
+
+def _move(source: Path, target: Path) -> None:
+    if target.is_dir():
+        shutil.rmtree(target)
+    elif target.exists():
+        target.unlink()
+    target.parent.mkdir(parents=True, exist_ok=True)
+    shutil.move(str(source), str(target))
+
+
+def read_meta(root: Path, terrain: str) -> dict[str, Any]:
+    try:
+        return json.loads((bundle_dir(root, terrain) / "meta.json").read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return {}
+
+
+def _write_json(path: Path, value: Any) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(value, ensure_ascii=False, default=str), encoding="utf-8")
+
+
+def note_unbuilt(root: Path, terrain: str, reason: str) -> None:
+    """그 지형 벌을 못 지음 — 묶음 비우고 meta.json 에 까닭만."""
+    bundle = bundle_dir(root, terrain)
+    shutil.rmtree(bundle, ignore_errors=True)
+    _write_json(bundle / "meta.json", {"terrain": terrain, "built": False, "reason": reason})
+
+
+def discard_sets(root: Path) -> bool:
+    """묶음 모두 버림(전처리 다시 · 입력 바뀜)."""
+    target = sets_dir(root)
+    if not target.exists():
+        return False
+    shutil.rmtree(target, ignore_errors=True)
+    return True
+
+
+def other_terrain(params: dict[str, Any], root: Path) -> AnalysisTerrain | None:
+    """설계 지형 말고 고를 수 있는 지형 — 없으면 None."""
+    design = design_terrain(params)
+    sources = analysis_terrain_sources(params, Path(root) / MODELS_DIR)
+    return next((name for name in sources if name != design), None)
+
+
+# ── 지문 ───────────────────────────────────────────────────────────────────
+
+_FINGERPRINT_KEYS = ("terrain", "source_filter", "method", "smooth", "model_digest")
+
+
+async def set_fingerprint(
+    connection: aiomysql.Connection, project_id: str, root: Path, terrain: str
+) -> dict[str, Any] | None:
+    """그 지형 벌이 기댄 입력 — 지형 · 필터 · 표현 · 평활 · 지표면 모델 id · 모델 파일 내용 지문.
+    그 지형 모델이 없으면 None."""
+    from B04_PreProcess.B04_PreProcess_Repository import list_surface_models
+    from common_util.common_util_derived_store import file_digest
+
+    params = await get_surface_confirmation_params(connection, project_id)
+    target = analysis_terrain_params(params, terrain)  # type: ignore[arg-type]
+    if design_terrain(target) != terrain:
+        return None
+    for model in await list_surface_models(connection, UUID(project_id)):
+        generation = model.get("generation_params") or {}
+        if (
+            model["status"] in ("COMPLETE", "CONFIRMED")
+            and str(model["model_type"]).lower() == str(target["method"]).lower()
+            and str(generation.get("source_filter", "")).lower()
+            == str(target["source_filter"]).lower()
+        ):
+            path = model.get("model_file_path")
+            digest = await asyncio.to_thread(file_digest, Path(root) / path if path else None)
+            return {
+                "terrain": terrain,
+                "source_filter": str(target["source_filter"]),
+                "method": str(target["method"]),
+                "smooth": bool(target.get("smooth")),
+                "model_id": int(model["id"]),
+                "model_digest": digest,
+            }
+    return None
+
+
+def bundle_ready(root: Path, terrain: str, fingerprint: dict[str, Any] | None) -> bool:
+    """묶음이 있고 지금 입력과 같은 지문인가(모델 id 는 안 봄 — 복제 · 갈음에 바뀜)."""
+    meta = read_meta(root, terrain)
+    if not fingerprint or not meta.get("built"):
+        return False
+    if not (bundle_dir(root, terrain) / "db.json").is_file():
+        return False
+    return all(meta.get(key) == fingerprint.get(key) for key in _FINGERPRINT_KEYS)
+
+
+# ── DB ─────────────────────────────────────────────────────────────────────
+
+
+async def _select(cursor: Any, sql: str, args: Any) -> list[dict[str, Any]]:
+    await cursor.execute(sql, args)
+    return _json_safe([dict(row) for row in await cursor.fetchall()])
+
+
+async def _dump_db(connection: aiomysql.Connection, project_id: str) -> dict[str, Any]:
+    async with connection.cursor(aiomysql.DictCursor) as cursor:
+        dump: dict[str, Any] = {
+            "routes": await _select(
+                cursor, "SELECT * FROM routes WHERE project_id=%s ORDER BY id", (project_id,)
+            ),
+            "stages": await _select(
+                cursor,
+                "SELECT stage_no, " + ", ".join(_STAGE_COLUMNS) + " FROM project_workflow_stages"
+                " WHERE project_id=%s AND stage_no >= %s ORDER BY stage_no",
+                (project_id, FIRST_SET_STAGE),
+            ),
+            "project_status": (
+                await _select(cursor, "SELECT status FROM projects WHERE id=%s", (project_id,))
+                or [{}]
+            )[0].get("status"),
+        }
+        ids = [row["id"] for row in dump["routes"]] or [-1]
+        marks = ",".join(["%s"] * len(ids))
+        for table in _ROUTE_CHILDREN:
+            dump[table] = await _select(
+                cursor, f"SELECT * FROM {table} WHERE route_id IN ({marks}) ORDER BY id", ids
+            )
+    return dump
+
+
+async def _clear_db(connection: aiomysql.Connection, project_id: str) -> None:
+    """노선(자식은 CASCADE) 지움 · 단계 2 이후 처음 — 한 트랜잭션."""
+    from common_util.common_util_workflow_state import reset_stages_after_input_change
+
+    await connection.begin()
+    try:
+        async with connection.cursor() as cursor:
+            await cursor.execute("DELETE FROM routes WHERE project_id=%s", (project_id,))
+            await reset_stages_after_input_change(cursor, project_id, FIRST_SET_STAGE)
+        await connection.commit()
+    except Exception:
+        await connection.rollback()
+        raise
+
+
+async def _insert(cursor: Any, table: str, row: dict[str, Any]) -> int:
+    names = ", ".join(f"`{name}`" for name in row)
+    marks = ", ".join(["%s"] * len(row))
+    await cursor.execute(f"INSERT INTO {table} ({names}) VALUES ({marks})", tuple(row.values()))
+    return int(cursor.lastrowid)
+
+
+async def _load_db(
+    connection: aiomysql.Connection, project_id: str, dump: dict[str, Any]
+) -> dict[int, int]:
+    """노선 새 id 로 넣고 자식 다시 잇기 · 단계 · 상태 되씀 — 한 트랜잭션 → 옛 id → 새 id."""
+    route_map: dict[int, int] = {}
+    await connection.begin()
+    try:
+        async with connection.cursor() as cursor:
+            for route in dump.get("routes", []):
+                row = {**route, "project_id": project_id}
+                old = int(row.pop("id"))
+                route_map[old] = await _insert(cursor, "routes", row)
+            for table in _ROUTE_CHILDREN:
+                for child in dump.get(table, []):
+                    row = dict(child)
+                    row.pop("id", None)
+                    row["route_id"] = route_map[int(row["route_id"])]
+                    if "project_id" in row:
+                        row["project_id"] = project_id
+                    await _insert(cursor, table, row)
+            for stage in dump.get("stages", []):
+                sets = ", ".join(f"`{name}`=%s" for name in _STAGE_COLUMNS)
+                await cursor.execute(
+                    f"UPDATE project_workflow_stages SET {sets}"
+                    " WHERE project_id=%s AND stage_no=%s",
+                    [stage.get(name) for name in _STAGE_COLUMNS] + [project_id, stage["stage_no"]],
+                )
+            if dump.get("project_status") is not None:
+                await cursor.execute(
+                    "UPDATE projects SET status=%s WHERE id=%s",
+                    (dump["project_status"], project_id),
+                )
+        await connection.commit()
+    except Exception:
+        await connection.rollback()
+        raise
+    return route_map
+
+
+# ── 떠 두기 · 꺼내 끼우기 ───────────────────────────────────────────────────
+
+
+def _clear_files(root: Path) -> None:
+    for relative in set_entries(root):
+        path = Path(root) / relative
+        if path.is_dir():
+            shutil.rmtree(path, ignore_errors=True)
+        else:
+            path.unlink(missing_ok=True)
+
+
+async def clear_active_set(connection: aiomysql.Connection, root: Path, project_id: str) -> None:
+    """제자리 벌 비움(못 지은 벌 · 반쯤 지은 결과) — 공유 캐시는 그대로."""
+    await _clear_db(connection, project_id)
+    await asyncio.to_thread(_clear_files, root)
+
+
+def _park_files(root: Path, work: Path) -> list[str]:
+    entries = set_entries(root)
+    for relative in entries:
+        _move(Path(root) / relative, _bundle_path(work, relative))
+    return entries
+
+
+async def park_active_set(
+    connection: aiomysql.Connection,
+    root: Path,
+    project_id: str,
+    fingerprint: dict[str, Any],
+) -> dict[str, Any]:
+    """제자리 벌을 `design_sets/<지형>/` 으로 떠 두고 제자리를 비움 → meta."""
+    root = Path(root)
+    terrain = str(fingerprint["terrain"])
+    dump = await _dump_db(connection, project_id)
+    work = sets_dir(root) / f".{terrain}_new"
+    await asyncio.to_thread(shutil.rmtree, work, True)
+    _write_json(work / "db.json", dump)
+    entries = await asyncio.to_thread(_park_files, root, work)
+    meta = {
+        **fingerprint,
+        "built": True,
+        "entries": entries,
+        "routes": [row["id"] for row in dump["routes"]],
+        "parked_at": time.strftime("%Y-%m-%d %H:%M:%S"),
+    }
+    _write_json(work / "meta.json", meta)
+    await _clear_db(connection, project_id)
+    bundle = bundle_dir(root, terrain)
+    await asyncio.to_thread(shutil.rmtree, bundle, True)
+    work.rename(bundle)
+    return meta
+
+
+def _unpark_files(root: Path, bundle: Path, entries: list[str], route_map: dict) -> None:
+    from common_util.common_util_project_copy import rekey_id_files
+
+    for relative in entries:
+        source = _bundle_path(bundle, relative)
+        if source.exists():
+            _move(source, Path(root) / relative)
+    rekey_id_files(Path(root), {}, route_map)
+    shutil.rmtree(bundle, ignore_errors=True)
+
+
+async def load_set(
+    connection: aiomysql.Connection, root: Path, project_id: str, terrain: str
+) -> dict[int, int] | None:
+    """묶음을 제자리로 꺼내 끼움(제자리는 비어 있어야 함) → 옛 노선 id → 새 id ·
+    묶음 없음 = None."""
+    bundle = bundle_dir(root, terrain)
+    try:
+        dump = json.loads((bundle / "db.json").read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    meta = read_meta(root, terrain)
+    route_map = await _load_db(connection, project_id, dump)
+    await asyncio.to_thread(_unpark_files, root, bundle, list(meta.get("entries") or []), route_map)
+    return route_map
+
+
+# ── 지형 바꿈 ──────────────────────────────────────────────────────────────
+
+
+async def switch_terrain(connection: aiomysql.Connection, project_id: UUID, terrain: str) -> int:
+    """설계 지형(stage 1 확정값 · 확정 모델)을 바꿈 — 한 트랜잭션 → 확정 모델 id."""
+    from B04_PreProcess.B04_PreProcess_Service import switch_design_terrain
+
+    await connection.begin()
+    try:
+        switched = await switch_design_terrain(connection, project_id, terrain)  # type: ignore[arg-type]
+        await connection.commit()
+    except Exception:
+        await connection.rollback()
+        raise
+    if switched:
+        return int(switched[0])
+    async with connection.cursor() as cursor:
+        await cursor.execute(
+            "SELECT id FROM surface_models WHERE project_id=%s AND status='CONFIRMED'",
+            (str(project_id),),
+        )
+        row = await cursor.fetchone()
+    if not row:
+        raise LookupError("확정 지표면 모델이 없습니다.")
+    return int(row[0])
+
+
+async def swap_design_set(project_id: UUID, root: Path, target: str) -> dict[str, Any]:
+    """지형 칩 = 맞바꿈 — 지금 벌 떠 둠 → 지형 바꿈 → 고른 벌 꺼내 끼움(없음 · 묵음 = 한 번 지음).
+    실패 = DB 행 · 벌 몫 · 묶음을 떠 둔 대로 되돌리고 예외 그대로. 잠금은 부르는 쪽."""
+    from B03_FileInput.B03_FileInput_Service_Chain import run_auto_design_chain
+    from common_util import common_util_save_guard as guard
+    from common_util.common_util_initial_snapshot import (
+        clear_design_failed,
+        clear_designing,
+        is_design_failed,
+        read_design_failure,
+    )
+    from config.config_db import get_db_pool
+
+    started = time.perf_counter()
+    pid, root = str(project_id), Path(root)
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        now = design_terrain(await get_surface_confirmation_params(connection, pid))
+        if now == target:
+            return {"changed": False, "design": now}
+        here = await set_fingerprint(connection, pid, root, now) or {"terrain": now}
+        there = await set_fingerprint(connection, pid, root, target)
+        if there is None:
+            raise LookupError("고른 지형의 지표면 모델이 없습니다.")
+        rows = await guard.snapshot_rows(connection, pid, surface=True)
+    ready = bundle_ready(root, target, there)
+    # 되돌림 자리 = 지금 벌 몫 + 꺼낼 벌 몫(끼운 뒤 생기는 파일도 지우게) + 묶음
+    scope = tuple(
+        dict.fromkeys(
+            [*set_entries(root), *(read_meta(root, target).get("entries") or []), SETS_DIRNAME]
+        )
+    )
+    files = await asyncio.to_thread(guard.snapshot_storage, root, scope)
+    had_failed = is_design_failed(root)
+    had_bundle = bundle_dir(root, now).exists()
+    try:
+        async with pool.acquire() as connection:
+            await park_active_set(connection, root, pid, here)
+            model_id = await switch_terrain(connection, project_id, target)
+            loaded = await load_set(connection, root, pid, target) if ready else None
+        if loaded is None:
+            await asyncio.to_thread(shutil.rmtree, bundle_dir(root, target), True)
+            summary = await run_auto_design_chain(project_id, model_id, inner=True)
+            if summary is None:
+                raise RuntimeError(read_design_failure(root) or "고른 지형 설계를 짓지 못했습니다.")
+    except BaseException:
+        files.restore()
+        if not had_bundle:  # 되돌림은 파일만 — 떠 둔 빈 폴더도 걷음
+            shutil.rmtree(bundle_dir(root, now), ignore_errors=True)
+        async with pool.acquire() as connection:
+            await guard.restore_rows(connection, pid, rows)
+        if not had_failed:
+            clear_design_failed(root)
+        raise
+    finally:
+        files.discard()
+        clear_designing(root)  # 지음 길(inner 체인)이 건 마커
+    return {
+        "changed": True,
+        "design": target,
+        "built": loaded is None,
+        "seconds": round(time.perf_counter() - started, 2),
+    }

common_util/common_util_initial_snapshot.py

@@ -38,6 +38,10 @@ DESIGN_FAILED_NAME = "initial_design.failed"
 # 확정: 「초기값 = 파일 입력 직후 결과 전부」). 관을 옮기면 세부유역(`04_detailed_basins`)이
 # 다시 나뉘는데 그 산출물이 스냅샷 밖이라 [초기화]가 옛 유역도를 그대로 남겼다.
 # 용량은 실측 8.6MB(스냅샷 전체 3.9MB → 약 12MB)로 감당할 만하다.
+#: 배수 폴더 중 노선과 무관한 공유 캐시(이름 앞머리 · 도엽 좌표 · 하강장 · 강우) — 초기값 사본 ·
+#: 설계 벌에 안 넣고 복원 · 맞바꿈이 안 지움(765-6)
+DRAINAGE_SHARED_PREFIXES = ("00_sheet_metric", "descent_", "rainfall_")
+
 _FILE_TREES = (
     "B05_Profile/route",
     "B06_Section/longitudinal",

resources/tester/test_765_design_sets.py

@@ -0,0 +1,413 @@
+# -*- coding: utf-8 -*-
+"""765-2 지형마다 설계 한 벌 — 떠 두기 · 꺼내 끼우기 · 맞바꿈(`common_util_design_sets`).
+
+DB = 메모리 sqlite 를 aiomysql 꼴로 감싼 가짜(노선 · 자식 넷 · 단계 · 프로젝트 · 지표면 모델) ·
+storage = 시험 안 임시 폴더. 지킴:
+  ① 떠 둔 뒤 꺼내면 값 같음 — 노선 · 자식 · 단계 · 상태 · 벌 몫 파일 · 초기값 · 코리도
+  ② 꺼낼 때 새 노선 id · 자식 다시 잇기 · 코리도 이름 새 id
+  ③ 공유 캐시(비용면 · 예정노선 · 배수 하강장 · 도엽 좌표 · 강우)는 안 옮기고 안 지움
+  ④ 지문 어긋난(묵은) 벌은 버리고 그 자리에서 지음 · 실패면 통째 되돌림
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import re
+import sqlite3
+import sys
+from pathlib import Path
+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 aiomysql  # noqa: E402
+
+from common_util import common_util_design_sets as sets  # noqa: E402
+
+PID = str(uuid4())
+
+_SCHEMA = """
+CREATE TABLE projects (id TEXT PRIMARY KEY, status TEXT, name TEXT);
+CREATE TABLE surface_models (id INTEGER PRIMARY KEY, project_id TEXT, status TEXT);
+CREATE TABLE project_workflow_stages (
+  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, stage_no INT, stage_key TEXT,
+  state TEXT, progress_percent INT, params TEXT, message TEXT, started_at TEXT, completed_at TEXT);
+CREATE TABLE routes (
+  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT, surface_model_id INT, status TEXT,
+  total_length_m REAL, route_data_path TEXT, created_at TEXT);
+CREATE TABLE route_points (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  route_id INT REFERENCES routes(id) ON DELETE CASCADE, chainage_m REAL, sequence_num INT);
+CREATE TABLE route_statistics (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  route_id INT REFERENCES routes(id) ON DELETE CASCADE, cut_volume_m3 REAL);
+CREATE TABLE longitudinal_sections (
+  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT,
+  route_id INT REFERENCES routes(id) ON DELETE CASCADE, data TEXT);
+CREATE TABLE cross_sections (
+  id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT,
+  route_id INT REFERENCES routes(id) ON DELETE CASCADE, chainage_m REAL, data TEXT);
+"""
+
+
+class _Cursor:
+    def __init__(self, db: sqlite3.Connection, dict_rows: bool) -> None:
+        self.db, self.dict_rows, self.rows, self.lastrowid, self.rowcount = db, dict_rows, [], 0, 0
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        return False
+
+    @staticmethod
+    def _sql(sql: str) -> str:
+        return re.sub(r"%s", "?", sql)
+
+    async def execute(self, sql: str, args=()) -> None:
+        cur = self.db.execute(self._sql(sql), tuple(args or ()))
+        self.lastrowid, self.rowcount = cur.lastrowid, cur.rowcount
+        names = [d[0] for d in cur.description or []]
+        rows = cur.fetchall()
+        self.rows = [dict(zip(names, r)) for r in rows] if self.dict_rows else rows
+
+    async def executemany(self, sql: str, seq) -> None:
+        self.db.executemany(self._sql(sql), [tuple(a) for a in seq])
+
+    async def fetchall(self):
+        return self.rows
+
+    async def fetchone(self):
+        return self.rows[0] if self.rows else None
+
+
+class _Conn:
+    def __init__(self, db: sqlite3.Connection) -> None:
+        self.db = db
+
+    def cursor(self, kind=None):
+        return _Cursor(self.db, kind is aiomysql.DictCursor)
+
+    async def begin(self):
+        pass
+
+    async def commit(self):
+        self.db.commit()
+
+    async def rollback(self):
+        self.db.rollback()
+
+
+class _Pool:
+    def __init__(self, conn: _Conn) -> None:
+        self.conn = conn
+
+    def acquire(self):
+        conn = self.conn
+
+        class _Ctx:
+            async def __aenter__(self):
+                return conn
+
+            async def __aexit__(self, *_):
+                return False
+
+        return _Ctx()
+
+
+def _seed_db() -> sqlite3.Connection:
+    db = sqlite3.connect(":memory:")
+    db.execute("PRAGMA foreign_keys = ON")
+    db.executescript(_SCHEMA)
+    db.execute("INSERT INTO projects VALUES (?, 'WF3_ANALYZING', 'p')", (PID,))
+    db.executemany(
+        "INSERT INTO surface_models VALUES (?, ?, ?)",
+        [(5, PID, "CONFIRMED"), (6, PID, "COMPLETE")],
+    )
+    for no in range(7):
+        db.execute(
+            "INSERT INTO project_workflow_stages (project_id, stage_no, stage_key, state,"
+            " progress_percent, params, message) VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (PID, no, f"S{no}", "IN_PROGRESS", 50, json.dumps({"no": no, "k": "las"}), f"m{no}"),
+        )
+    for rid in (10, 11):
+        db.execute(
+            "INSERT INTO routes (id, project_id, surface_model_id, status, total_length_m,"
+            " route_data_path, created_at) VALUES (?, ?, 5, 'CONFIRMED', ?, 'r.geojson', 't')",
+            (rid, PID, rid * 100.0),
+        )
+        for seq in range(3):
+            db.execute(
+                "INSERT INTO route_points (route_id, chainage_m, sequence_num) VALUES (?, ?, ?)",
+                (rid, seq * 20.0, seq),
+            )
+            db.execute(
+                "INSERT INTO cross_sections (project_id, route_id, chainage_m, data)"
+                " VALUES (?, ?, ?, ?)",
+                (PID, rid, seq * 20.0, json.dumps({"cut": seq})),
+            )
+        db.execute("INSERT INTO route_statistics (route_id, cut_volume_m3) VALUES (?, 7)", (rid,))
+        db.execute(
+            "INSERT INTO longitudinal_sections (project_id, route_id, data) VALUES (?, ?, '{}')",
+            (PID, rid),
+        )
+    db.commit()
+    return db
+
+
+_SET_FILES = {
+    "B05_Profile/route/planned_route.csv": "x,y\n1,2\n",
+    "B05_Profile/route/structures.json": '{"items": [1]}',
+    "B05_Profile/route/route_main.geojson": "{}",
+    "B05_Profile/corridor/corridor_0011.json": '{"route": 11}',
+    "B06_Section/longitudinal/longitudinal.json": '{"z": [1, 2]}',
+    "B06_Section/sampling.json": '{"surface_model_id": 5}',
+    "B04_PreProcess/drainage/01_primary_region.geojson": "{}",
+    "B04_PreProcess/drainage/edits/pipe_points.json": "[1]",
+    "B04_PreProcess/drainage/manifest.json": "{}",
+    "initial_snapshot/db.json": '{"routes": []}',
+}
+_SHARED_FILES = {
+    "B05_Profile/route/cost_surface_x_implicit.npz": "c",
+    "B05_Profile/route/expected_route.csv": "e",
+    "B05_Profile/route/initial_choice.json": "{}",
+    "B04_PreProcess/drainage/descent_ab_cd.npz": "d",
+    "B04_PreProcess/drainage/00_sheet_metric.pkl": "s",
+    "B04_PreProcess/drainage/rainfall_table.json": "r",
+    "B04_PreProcess/models/dtm_x.npz": "m",
+}
+
+
+def _seed_files(root: Path) -> None:
+    for relative, text in {**_SET_FILES, **_SHARED_FILES}.items():
+        path = root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(text, encoding="utf-8")
+
+
+def _tree(root: Path) -> dict[str, str]:
+    return {
+        p.relative_to(root).as_posix(): p.read_text(encoding="utf-8")
+        for p in sorted(root.rglob("*"))
+        if p.is_file() and SETS_DIR not in p.relative_to(root).parts
+    }
+
+
+SETS_DIR = sets.SETS_DIRNAME
+FP = {
+    "terrain": "las",
+    "source_filter": "classification",
+    "method": "implicit",
+    "smooth": False,
+    "model_id": 5,
+    "model_digest": "aa",
+}
+
+
+def _rows(db: sqlite3.Connection, sql: str, args=()) -> list[tuple]:
+    return db.execute(sql, args).fetchall()
+
+
+def _no_ids(db: sqlite3.Connection) -> dict[str, list]:
+    """id 를 뺀 값(자식은 노선 차례로) — 왕복 비교용."""
+    order = {rid: i for i, (rid,) in enumerate(_rows(db, "SELECT id FROM routes ORDER BY id"))}
+    out = {
+        "routes": _rows(
+            db,
+            "SELECT project_id, surface_model_id, status, total_length_m, route_data_path,"
+            " created_at FROM routes ORDER BY id",
+        ),
+        "stages": _rows(
+            db,
+            "SELECT stage_no, state, progress_percent, params, message FROM"
+            " project_workflow_stages ORDER BY stage_no",
+        ),
+        "status": _rows(db, "SELECT status FROM projects"),
+    }
+    for table, cols in (
+        ("route_points", "chainage_m, sequence_num"),
+        ("route_statistics", "cut_volume_m3"),
+        ("longitudinal_sections", "project_id, data"),
+        ("cross_sections", "project_id, chainage_m, data"),
+    ):
+        out[table] = sorted(
+            (order[r[0]], *r[1:])
+            for r in _rows(db, f"SELECT route_id, {cols} FROM {table} ORDER BY id")
+        )
+    return out
+
+
+def test_벌_몫과_공유를_가름(tmp_path: Path) -> None:
+    _seed_files(tmp_path)
+    entries = sets.set_entries(tmp_path)
+    assert "B06_Section" in entries and "initial_snapshot" in entries
+    assert "B05_Profile/route/structures.json" in entries
+    assert "B04_PreProcess/drainage/edits" in entries
+    joined = " ".join(entries)
+    for shared in ("cost_surface_", "expected_route", "initial_choice", "descent_", "00_sheet"):
+        assert shared not in joined
+    assert "rainfall_" not in joined and "models" not in joined
+
+
+def test_떠_두고_꺼내면_값_같음_새_id(tmp_path: Path) -> None:
+    db = _seed_db()
+    conn = _Conn(db)
+    _seed_files(tmp_path)
+    before_db, before_files = _no_ids(db), _tree(tmp_path)
+
+    meta = asyncio.run(sets.park_active_set(conn, tmp_path, PID, FP))
+    assert meta["routes"] == [10, 11] and meta["built"]
+    assert _rows(db, "SELECT COUNT(*) FROM routes") == [(0,)]
+    assert _rows(db, "SELECT COUNT(*) FROM cross_sections") == [(0,)]
+    states = _rows(db, "SELECT stage_no, state, params FROM project_workflow_stages ORDER BY 1")
+    assert states[:2] == [(0, "IN_PROGRESS", before_db["stages"][0][3])] + [
+        (1, "IN_PROGRESS", before_db["stages"][1][3])
+    ]
+    assert all(s[1] == "NOT_STARTED" and s[2] is None for s in states[2:])
+    left = _tree(tmp_path)
+    assert set(left) == set(_SHARED_FILES)  # 공유만 남음 · 그대로
+    assert all(left[k] == v for k, v in _SHARED_FILES.items())
+    bundle = tmp_path / SETS_DIR / "las"
+    assert (bundle / "db.json").is_file() and (bundle / "initial_snapshot" / "db.json").is_file()
+    assert (bundle / "files" / "B06_Section" / "sampling.json").is_file()
+    assert sets.bundle_ready(tmp_path, "las", FP)
+
+    route_map = asyncio.run(sets.load_set(conn, tmp_path, PID, "las"))
+    assert set(route_map) == {10, 11}
+    assert not set(route_map.values()) & {10, 11}  # 새 id
+    assert _no_ids(db) == before_db  # 노선 · 자식 · 단계 · 상태 같음
+    after = _tree(tmp_path)
+    moved = before_files.pop("B05_Profile/corridor/corridor_0011.json")
+    new_corridor = f"B05_Profile/corridor/corridor_{route_map[11]:04d}.json"
+    assert after.pop(new_corridor) == moved  # 코리도 = 새 id 이름
+    assert after == before_files
+    assert not bundle.exists()  # 꺼낸 묶음은 지움
+
+
+def test_지문_어긋나면_묵은_벌(tmp_path: Path) -> None:
+    db = _seed_db()
+    _seed_files(tmp_path)
+    asyncio.run(sets.park_active_set(_Conn(db), tmp_path, PID, FP))
+    assert sets.bundle_ready(tmp_path, "las", FP)
+    assert sets.bundle_ready(tmp_path, "las", {**FP, "model_id": 99})  # 복제 · 갈음 id 는 안 봄
+    assert not sets.bundle_ready(tmp_path, "las", {**FP, "model_digest": "bb"})
+    assert not sets.bundle_ready(tmp_path, "las", {**FP, "source_filter": "csf"})
+    assert not sets.bundle_ready(tmp_path, "las", None)
+    assert not sets.bundle_ready(tmp_path, "sheet", {**FP, "terrain": "sheet"})
+    sets.note_unbuilt(tmp_path, "sheet", "수치지형도 모델 없음")
+    assert sets.read_meta(tmp_path, "sheet") == {
+        "terrain": "sheet",
+        "built": False,
+        "reason": "수치지형도 모델 없음",
+    }
+    assert not sets.bundle_ready(tmp_path, "sheet", {**FP, "terrain": "sheet"})
+
+
+def _swap_world(monkeypatch, tmp_path: Path, *, sheet_fp: dict, chain):
+    """las 활성 · sheet 묶음(지문 = sheet_fp) — 칩으로 sheet."""
+    import B03_FileInput.B03_FileInput_Service_Chain as chain_mod
+    import config.config_db as config_db
+
+    db = _seed_db()
+    conn = _Conn(db)
+    _seed_files(tmp_path)
+    sheet = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_id": 6}
+    # sheet 벌을 미리 떠 둠: 노선 하나(길이 777)
+    asyncio.run(sets.park_active_set(conn, tmp_path, PID, FP))
+    db.execute(
+        "INSERT INTO routes (project_id, surface_model_id, status, total_length_m) VALUES"
+        " (?, 6, 'CONFIRMED', 777)",
+        (PID,),
+    )
+    db.commit()
+    (tmp_path / "B06_Section").mkdir(parents=True, exist_ok=True)
+    (tmp_path / "B06_Section" / "sheet.json").write_text("sheet", encoding="utf-8")
+    asyncio.run(sets.park_active_set(conn, tmp_path, PID, sheet_fp))
+    asyncio.run(sets.load_set(conn, tmp_path, PID, "las"))
+    state = {"design": "las"}
+
+    async def params(_c, _pid):
+        return {
+            "source_filter": "sheet_laplace" if state["design"] == "sheet" else "classification"
+        }
+
+    async def fingerprint(_c, _pid, _root, terrain):
+        return sheet if terrain == "sheet" else FP
+
+    async def switch(_c, _pid, terrain):
+        state["design"] = terrain
+        db.execute(
+            "UPDATE surface_models SET status = CASE id WHEN 6 THEN 'CONFIRMED' ELSE 'COMPLETE' END"
+        )
+        return 6
+
+    monkeypatch.setattr(config_db, "get_db_pool", lambda: _Pool(conn))
+    monkeypatch.setattr(sets, "get_surface_confirmation_params", params)
+    monkeypatch.setattr(sets, "set_fingerprint", fingerprint)
+    monkeypatch.setattr(sets, "switch_terrain", switch)
+    monkeypatch.setattr(chain_mod, "run_auto_design_chain", chain)
+    return db, state
+
+
+def test_맞바꿈_묶음_있으면_꺼냄(monkeypatch, tmp_path: Path) -> None:
+    called = []
+
+    async def chain(*a, **k):
+        called.append(a)
+
+    sheet_fp = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_id": 6}
+    db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=sheet_fp, chain=chain)
+    result = asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet"))  # type: ignore[arg-type]
+    assert result["changed"] and not result["built"] and not called
+    assert _rows(db, "SELECT total_length_m FROM routes") == [(777.0,)]
+    assert (tmp_path / "B06_Section" / "sheet.json").read_text(encoding="utf-8") == "sheet"
+    assert sets.bundle_ready(tmp_path, "las", FP)  # 떠난 벌 살림
+    assert not (tmp_path / SETS_DIR / "sheet").exists()
+
+
+def test_맞바꿈_묵은_벌은_버리고_지음(monkeypatch, tmp_path: Path) -> None:
+    called = []
+    holder: dict = {}
+
+    async def chain(pid, model_id, inner=False):
+        called.append((model_id, inner))
+        holder["db"].execute(
+            "INSERT INTO routes (project_id, surface_model_id, status, total_length_m) VALUES"
+            " (?, 6, 'CONFIRMED', 555)",
+            (PID,),
+        )
+        return {"route_id": 1}
+
+    stale = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_digest": "old"}
+    db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=stale, chain=chain)
+    holder["db"] = db
+    result = asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet"))  # type: ignore[arg-type]
+    assert result["built"] and called == [(6, True)]
+    assert _rows(db, "SELECT total_length_m FROM routes") == [(555.0,)]  # 묵은 777 버림
+    assert not (tmp_path / "B06_Section" / "sheet.json").exists()
+    assert sets.bundle_ready(tmp_path, "las", FP)
+
+
+def test_맞바꿈_실패면_통째_되돌림(monkeypatch, tmp_path: Path) -> None:
+    async def chain(*_a, **_k):
+        return None  # 지음 실패
+
+    stale = {**FP, "terrain": "sheet", "source_filter": "sheet_laplace", "model_digest": "old"}
+    db, _ = _swap_world(monkeypatch, tmp_path, sheet_fp=stale, chain=chain)
+    before_db, before_files = _no_ids(db), _tree(tmp_path)
+    route_ids = _rows(db, "SELECT id FROM routes ORDER BY id")
+    sheet_meta = sets.read_meta(tmp_path, "sheet")
+    models = _rows(db, "SELECT id, status FROM surface_models ORDER BY id")
+    with pytest.raises(RuntimeError):
+        asyncio.run(sets.swap_design_set(PID, tmp_path, "sheet"))  # type: ignore[arg-type]
+    assert _no_ids(db) == before_db
+    assert _rows(db, "SELECT id FROM routes ORDER BY id") == route_ids
+    assert _rows(db, "SELECT id, status FROM surface_models ORDER BY id") == models
+    assert _tree(tmp_path) == before_files
+    assert sets.read_meta(tmp_path, "sheet") == sheet_meta  # 묶음도 그대로
+    assert not (tmp_path / SETS_DIR / "las").exists()