From 12c3282abf23e9b10af0ac43540a7308d0795ab1 Mon Sep 17 00:00:00 2001
From: umsangdon <umsangdon@gmail.com>
Date: Mon, 19 Oct 2026 10:11:36 +0900
Subject: [PATCH] =?UTF-8?q?feat(472-6=206-A=20=EC=84=9C=EB=B2=84=20?=
 =?UTF-8?q?=ED=95=9C=20=EB=B3=B8=EB=AC=B8=20=EC=B0=BD=EA=B5=AC):=20POST=20?=
 =?UTF-8?q?/design/save=20=E2=80=94=20[=EC=A0=80=EC=9E=A5]=20=C2=B7=20[?=
 =?UTF-8?q?=ED=99=95=EC=A0=95]=20=EC=B0=A8=EB=A1=80=EB=A5=BC=20=EC=84=9C?=
 =?UTF-8?q?=EB=B2=84=20=ED=95=9C=20=EC=B0=BD=EA=B5=AC=EC=97=90=EC=84=9C=20?=
 =?UTF-8?q?=C2=B7=20=EC=A0=84=EB=B6=80=20=EC=95=84=EB=8B=88=EB=A9=B4=20?=
 =?UTF-8?q?=EC=97=86=EC=9D=8C?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- B06_Section_Router_DesignSave: 페이지 설정 → 관 → 구조물 → 추가 측점 → 계획고 → 관 옵션 → 상단측 → 횡단 = 옛 창구 함수를 안쪽으로 부름(지우지 않음) · 측점 맞춤 · 재계산 · 하위 차례 실패도 실패
- common_util_save_guard: DB 행(projects · 단계 · 노선과 자식 넷) · storage(저장 범위 · 관 차례면 배수 폴더) 떠 둠 → 실패면 바뀐 표 · 바뀐 파일만 되돌림 · 되돌림 실패 = 500 「복구 확인 필요」 · 프로젝트 잠금(뒤 요청 기다림 · waited)
- 시험 test_472_6_design_save 5(storage 되돌림 · 중간 실패 행 · 해시 그대로 · 정상 = 옛 차례와 행 · 파일 같음 · 복구 실패 알림 · 잠금) — 검증_sub8 끝에 차 0
- 노선 바뀐 저장은 옛 길 그대로(6-B)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 B06_Section/B06_Section_Router_DesignSave.py | 207 ++++++++++++++++
 common_util/common_util_save_guard.py        | 199 +++++++++++++++
 main.py                                      |   2 +
 resources/tester/test_472_6_design_save.py   | 245 +++++++++++++++++++
 4 files changed, 653 insertions(+)
 create mode 100644 B06_Section/B06_Section_Router_DesignSave.py
 create mode 100644 common_util/common_util_save_guard.py
 create mode 100644 resources/tester/test_472_6_design_save.py

diff --git a/B06_Section/B06_Section_Router_DesignSave.py b/B06_Section/B06_Section_Router_DesignSave.py
new file mode 100644
index 000000000..d99f91da7
--- /dev/null
+++ b/B06_Section/B06_Section_Router_DesignSave.py
@@ -0,0 +1,207 @@
+"""B05 · B06 [저장] · [확정] 한 본문 창구(PLAN 472-6 6-A) — 노선 안 바뀐 저장.
+
+브라우저가 세션 초안 · 사용자 칸만 한 본문으로 보내면 서버가 옛 차례 그대로 안쪽 창구를 부름:
+
+    페이지 설정 → 관 → 구조물 → 추가 측점 → 계획고 → 관 옵션 → 상단측(비정규 측점 맞춤) → 횡단
+
+각 안쪽 창구가 정본을 다시 셈(브라우저 결과 안 받음 · 횡단 차례 끝 서버 재계산). 하나라도 실패하면
+(오류 응답 · 측점 맞춤 실패 · 재계산 실패 · 하위 차례 실패) DB 행 · storage 를 시작 전으로 되돌리고
+그 차례 이름 · 까닭을 돌려줌(`common_util_save_guard` · 브레인 A). 되돌림마저 실패하면 「복구 확인
+필요」. 같은 프로젝트 저장이 겹치면 뒤 요청이 기다림(`waited`).
+노선 바뀐 저장(노선 초안)은 아직 옛 길(`/route/replan` 뒤 차례 · 6-B).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+from typing import Any
+from uuid import UUID
+
+from fastapi import APIRouter, Depends
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, ConfigDict
+
+from common_util.common_util_auth import get_optional_session
+from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+from B05_Profile.B05_Profile_Schema import (
+    PageSettingsRequest,
+    ProfileAlignmentSaveRequest,
+    RouteConfirmRequest,
+)
+from B05_Profile.B05_Profile_Structures_Schema import StructureSaveRequest
+from B06_Section.B06_Section_Router_Stations import PlusStationsRequest
+from B06_Section.B06_Section_Schema import SectionConfirmRequest
+from common_util import common_util_save_guard as guard
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_db import get_db_pool
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/projects", tags=["B06 Design Save"])
+
+
+class DesignSaveRequest(BaseModel):
+    """한 본문 — 칸이 없으면 그 차례를 건너뜀(관은 [확정]이면 칸 없어도 다시 셈)."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    route_id: int
+    confirm: bool = False
+    settings: PageSettingsRequest | None = None
+    #: 관 PUT 본문(`{points}` · 빈 dict = 서버 저장분 그대로)
+    pipes: dict[str, Any] | None = None
+    structures: StructureSaveRequest | None = None
+    plus_stations: PlusStationsRequest | None = None
+    alignment: ProfileAlignmentSaveRequest | None = None
+    #: 관 옵션 조각 {누가거리 0.01 글: {키: 값}} — 저장된 관 위에 얹어 다시 씀
+    culvert_options: dict[str, dict[str, Any]] | None = None
+    uphill: RouteConfirmRequest | None = None
+    cross: SectionConfirmRequest | None = None
+
+
+class StepFailed(Exception):
+    def __init__(self, step: str, message: str) -> None:
+        super().__init__(message)
+        self.step = step
+
+
+def _body(response: Any) -> dict[str, Any]:
+    """안쪽 창구 응답 → dict · 오류 응답이면 StepFailed 로(까닭 그대로)."""
+    if isinstance(response, JSONResponse):
+        content = json.loads(bytes(response.body).decode("utf-8") or "{}")
+        if response.status_code >= 400 or content.get("status") == "error":
+            raise StepFailed("", str(content.get("message") or f"HTTP {response.status_code}"))
+        return content
+    if hasattr(response, "model_dump"):
+        return response.model_dump(mode="json")
+    return dict(response or {})
+
+
+def _patched_points(stored: dict[str, Any], patches: dict[str, dict[str, Any]]) -> list | None:
+    """저장된 관 목록에 관 옵션 조각 — 고친 것이 없으면 None(브라우저 `flushCulvertOptions` 짝)."""
+    points, touched = [], False
+    for point in stored.get("pipe_points") or []:
+        item = {key: value for key, value in point.items() if key != "lonlat"}
+        patch = patches.get(f"{float(point['chainage_m']):.2f}")
+        if patch:
+            touched = True
+            item["options"] = {**(item.get("options") or {}), **patch}
+        points.append(item)
+    return points if touched else None
+
+
+async def _run_steps(
+    project_id: UUID, request: DesignSaveRequest, session: dict[str, Any] | None
+) -> dict[str, Any]:
+    """차례대로 안쪽 창구 · 차례 이름째 실패를 던짐 → 차례별 응답(알림 · 화면 맞춤용)."""
+    from B04_PreProcess.B04_PreProcess_Router_Basins import get_pipe_points, put_pipe_points
+    from B05_Profile.B05_Profile_Router import save_profile_alignment
+    from B05_Profile.B05_Profile_Router_Lifecycle import confirm_latest_route, save_page_settings
+    from B05_Profile.B05_Profile_Structures_Router import write_structures
+    from B06_Section.B06_Section_Router_Confirm import confirm_sections, save_sections
+    from B06_Section.B06_Section_Router_Stations import put_plus_stations
+
+    out: dict[str, Any] = {}
+
+    async def step(name: str, call: Any) -> dict[str, Any]:
+        try:
+            out[name] = _body(await call)
+        except StepFailed as failed:
+            raise StepFailed(name, str(failed)) from None
+        except Exception as error:
+            logger.exception("한 본문 저장 %s 차례 실패: project_id=%s", name, project_id)
+            raise StepFailed(name, str(error) or type(error).__name__) from None
+        return out[name]
+
+    if request.settings is not None:
+        await step("settings", save_page_settings(project_id, request.settings))
+    if request.pipes is not None or request.confirm:
+        pipes = await step(
+            "pipes", put_pipe_points(project_id, request.pipes or {}, session, request.confirm)
+        )
+        if pipes.get("stations_failed"):
+            raise StepFailed("pipes", "관 자리 측점을 못 맞췄습니다")
+    if request.structures is not None:
+        await step("structures", write_structures(project_id, request.structures, session))
+    if request.plus_stations is not None:
+        await step("plusStations", put_plus_stations(project_id, request.plus_stations))
+    if request.alignment is not None:
+        await step("alignment", save_profile_alignment(project_id, request.alignment))
+    if request.culvert_options:
+        stored = await step("culvertOptions", get_pipe_points(project_id))
+        points = _patched_points(stored, request.culvert_options)
+        if points is not None:
+            await step(
+                "culvertOptions", put_pipe_points(project_id, {"points": points}, session, False)
+            )
+    if request.uphill is not None:
+        parts = await step("uphill", confirm_latest_route(project_id, request.uphill, False))
+        if parts.get("failed_parts"):
+            raise StepFailed("uphill", " · ".join(parts["failed_parts"]) + " 실패")
+    if request.cross is not None:
+        save = confirm_sections if request.confirm else save_sections
+        cross = await step("cross", save(project_id, request.route_id, request.cross))
+        if cross.get("recomputed") is False:
+            raise StepFailed("cross", "서버 재계산 실패")
+        if cross.get("stations_synced") is False:
+            raise StepFailed("cross", "구조물 측점 맞춤 실패")
+    return out
+
+
+@router.post("/{project_id}/design/save")
+async def save_design(
+    project_id: UUID,
+    request: DesignSaveRequest,
+    session: dict[str, Any] | None = Depends(get_optional_session),
+) -> JSONResponse:
+    """[저장] · [확정] 한 본문 — 전부 저장하거나 아무것도 안 바꿈."""
+    lock = guard.project_lock(str(project_id))
+    waited = lock.locked()
+    async with lock:
+        pool = get_db_pool()
+        async with pool.acquire() as connection:
+            stored_path = await get_project_storage_relative_path(connection, project_id)
+            rows = await guard.snapshot_rows(connection, str(project_id))
+        root = Path(resolve_stored_project_path(stored_path))
+        pipes = request.pipes is not None or bool(request.culvert_options) or request.confirm
+        files = guard.snapshot_storage(
+            root, guard.STORAGE_SCOPE + (guard.PIPE_SCOPE if pipes else ())
+        )
+        try:
+            steps = await _run_steps(project_id, request, session)
+        except StepFailed as failed:
+            restored = await _restore(project_id, rows, files)
+            return JSONResponse(
+                status_code=409 if restored else 500,
+                content={
+                    "status": "error",
+                    "step": failed.step,
+                    "message": str(failed)
+                    if restored
+                    else f"{failed} — 저장 실패 · 복구 확인 필요",
+                    "restored": restored,
+                    "waited": waited,
+                },
+            )
+        finally:
+            files.discard()
+    return JSONResponse(content={"status": "success", "waited": waited, "steps": steps})
+
+
+async def _restore(project_id: UUID, rows: dict[str, list], files: guard.StorageSnapshot) -> bool:
+    """떠 둔 대로 — 성공 = 참 · 하나라도 못 되돌리면 로그 남기고 거짓(조용히 넘기지 않음)."""
+    ok = True
+    try:
+        files.restore()
+    except Exception:
+        logger.exception("한 본문 저장 되돌림 실패(storage): project_id=%s", project_id)
+        ok = False
+    try:
+        async with get_db_pool().acquire() as connection:
+            await guard.restore_rows(connection, str(project_id), rows)
+    except Exception:
+        logger.exception("한 본문 저장 되돌림 실패(DB): project_id=%s", project_id)
+        ok = False
+    return ok
diff --git a/common_util/common_util_save_guard.py b/common_util/common_util_save_guard.py
new file mode 100644
index 000000000..d5dc91378
--- /dev/null
+++ b/common_util/common_util_save_guard.py
@@ -0,0 +1,199 @@
+"""[저장] · [확정] 전부 아니면 없음(PLAN 472-6 · 브레인 A) — 떠 두고 실패면 되돌림.
+
+차례마다 안쪽 창구가 저마다 커넥션 · 파일을 쓰므로 DB 트랜잭션 하나로 못 꿴다. 대신 시작 전에
+  - DB 행 — `projects` 한 줄 · `project_workflow_stages` · `routes` 와 자식 넷(`route_points` ·
+    `route_statistics` · `longitudinal_sections` · `cross_sections`)
+  - storage — 저장 차례가 쓰는 폴더(`STORAGE_SCOPE` · 관 차례가 있으면 배수 폴더도)
+를 떠 두고, 하나라도 실패하면 둘 다 되돌린다. 되돌림 쓰기는 바뀐 것만(시놀로지 동기 — 안 바뀐
+파일 · 표는 안 씀). 같은 프로젝트 저장이 겹치면 뒤 요청이 기다림(`project_lock`).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import shutil
+import tempfile
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any
+
+import aiomysql
+
+#: 저장 차례가 쓰는 storage 자리(프로젝트 뿌리 기준) — 추천 노선 미리 셈(`route_recommend`)은 뒤에서
+#: 따로 도는 곳이라 안 넣음.
+STORAGE_SCOPE = (
+    "B05_Profile/route",
+    "B05_Profile/corridor",
+    "B06_Section",
+    "B07_DesignDetail",
+    "_derived",
+    "templates",
+    "workflow.json",
+    "project_manifest.json",
+)
+#: 관 · 관 옵션 차례 · [확정](배수유역 다시 셈)이 쓰는 자리.
+PIPE_SCOPE = ("B04_PreProcess/drainage",)
+#: 노선 아래 표(노선 행이 지워지면 함께 지워짐 · CASCADE).
+ROUTE_CHILDREN = ("route_points", "route_statistics", "longitudinal_sections", "cross_sections")
+
+_LOCKS: dict[str, asyncio.Lock] = {}
+
+
+def project_lock(project_id: str) -> asyncio.Lock:
+    """같은 프로젝트 저장 한 번에 하나 — 뒤 요청은 앞 저장이 끝날 때까지 기다림.
+    ponytail: 프로세스 메모리 잠금 · 서버 여러 대면 DB 잠금으로."""
+    return _LOCKS.setdefault(str(project_id), asyncio.Lock())
+
+
+# ── storage ────────────────────────────────────────────────────────────────
+
+
+def _stamp(path: Path) -> tuple[int, int]:
+    stat = path.stat()
+    return stat.st_size, stat.st_mtime_ns
+
+
+def _files(root: Path, scope: tuple[str, ...]) -> dict[str, Path]:
+    out: dict[str, Path] = {}
+    for relative in scope:
+        target = root / relative
+        if target.is_file():
+            out[relative] = target
+        elif target.is_dir():
+            for path in target.rglob("*"):
+                if path.is_file():
+                    out[path.relative_to(root).as_posix()] = path
+    return out
+
+
+@dataclass
+class StorageSnapshot:
+    """떠 둔 파일 — 사본 폴더(임시) · 파일마다 (크기, 바뀐 때)."""
+
+    root: Path
+    scope: tuple[str, ...]
+    copy_dir: Path
+    stamps: dict[str, tuple[int, int]] = field(default_factory=dict)
+
+    def restore(self) -> int:
+        """바뀐 · 새로 생긴 · 지워진 파일만 되돌림 → 되쓴 수(지움 포함)."""
+        written = 0
+        now = _files(self.root, self.scope)
+        for relative, path in now.items():
+            if relative not in self.stamps:
+                path.unlink()
+                written += 1
+            elif _stamp(path) != self.stamps[relative]:
+                shutil.copy2(self.copy_dir / relative, path)
+                written += 1
+        for relative in self.stamps.keys() - now.keys():
+            target = self.root / relative
+            target.parent.mkdir(parents=True, exist_ok=True)
+            shutil.copy2(self.copy_dir / relative, target)
+            written += 1
+        return written
+
+    def changed(self) -> list[str]:
+        """떠 둔 뒤 바뀐 자리(시험 · 로그)."""
+        now = _files(self.root, self.scope)
+        moved = [r for r, p in now.items() if self.stamps.get(r) != _stamp(p)]
+        return sorted(moved + list(self.stamps.keys() - now.keys()))
+
+    def discard(self) -> None:
+        shutil.rmtree(self.copy_dir, ignore_errors=True)
+
+
+def snapshot_storage(root: Path, scope: tuple[str, ...]) -> StorageSnapshot:
+    copy_dir = Path(tempfile.mkdtemp(prefix="aislo_save_"))
+    snap = StorageSnapshot(Path(root), scope, copy_dir)
+    for relative, path in _files(snap.root, scope).items():
+        target = copy_dir / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(path, target)
+        snap.stamps[relative] = _stamp(path)
+    return snap
+
+
+# ── DB ─────────────────────────────────────────────────────────────────────
+
+
+async def _select(cursor: Any, sql: str, args: tuple | list) -> list[dict[str, Any]]:
+    await cursor.execute(sql, args)
+    return [dict(row) for row in await cursor.fetchall()]
+
+
+async def snapshot_rows(connection: aiomysql.Connection, project_id: str) -> dict[str, list]:
+    """프로젝트 한 줄 · 단계 · 노선과 자식 넷의 행(값 그대로)."""
+    async with connection.cursor(aiomysql.DictCursor) as cursor:
+        rows = {
+            "projects": await _select(cursor, "SELECT * FROM projects WHERE id=%s", (project_id,)),
+            "project_workflow_stages": await _select(
+                cursor,
+                "SELECT * FROM project_workflow_stages WHERE project_id=%s ORDER BY id",
+                (project_id,),
+            ),
+            "routes": await _select(
+                cursor, "SELECT * FROM routes WHERE project_id=%s ORDER BY id", (project_id,)
+            ),
+        }
+        ids = [row["id"] for row in rows["routes"]] or [-1]
+        marks = ",".join(["%s"] * len(ids))
+        for table in ROUTE_CHILDREN:
+            rows[table] = await _select(
+                cursor, f"SELECT * FROM {table} WHERE route_id IN ({marks}) ORDER BY id", ids
+            )
+    return rows
+
+
+async def _insert(cursor: Any, table: str, rows: list[dict[str, Any]]) -> None:
+    if not rows:
+        return
+    columns = list(rows[0])
+    names = ",".join(f"`{name}`" for name in columns)
+    marks = ",".join(["%s"] * len(columns))
+    await cursor.executemany(
+        f"INSERT INTO {table} ({names}) VALUES ({marks})",
+        [[row[name] for name in columns] for row in rows],
+    )
+
+
+async def restore_rows(
+    connection: aiomysql.Connection, project_id: str, saved: dict[str, list]
+) -> list[str]:
+    """바뀐 표만 떠 둔 행으로 — 한 트랜잭션 → 되돌린 표 이름."""
+    now = await snapshot_rows(connection, project_id)
+    tree = ("routes", *ROUTE_CHILDREN)
+    redo_tree = any(now[table] != saved[table] for table in tree)
+    redo_stages = now["project_workflow_stages"] != saved["project_workflow_stages"]
+    redo_project = now["projects"] != saved["projects"] and bool(saved["projects"])
+    if not (redo_tree or redo_stages or redo_project):
+        return []
+    await connection.begin()
+    try:
+        async with connection.cursor() as cursor:
+            if redo_tree:
+                await cursor.execute("DELETE FROM routes WHERE project_id=%s", (project_id,))
+                for table in tree:
+                    await _insert(cursor, table, saved[table])
+            if redo_stages:
+                await cursor.execute(
+                    "DELETE FROM project_workflow_stages WHERE project_id=%s", (project_id,)
+                )
+                await _insert(cursor, "project_workflow_stages", saved["project_workflow_stages"])
+            if redo_project:
+                row = saved["projects"][0]
+                columns = [name for name in row if name != "id"]
+                sets = ",".join(f"`{name}`=%s" for name in columns)
+                await cursor.execute(
+                    f"UPDATE projects SET {sets} WHERE id=%s",
+                    [row[name] for name in columns] + [project_id],
+                )
+        await connection.commit()
+    except Exception:
+        await connection.rollback()
+        raise
+    return [
+        *(tree if redo_tree else ()),
+        *(("project_workflow_stages",) if redo_stages else ()),
+        *(("projects",) if redo_project else ()),
+    ]
diff --git a/main.py b/main.py
index 05ca3a1fc..3cd51fe23 100644
--- a/main.py
+++ b/main.py
@@ -54,6 +54,7 @@ from B05_Profile.B05_Profile_Router_Terrain import router as b05_route_terrain_r
 from B05_Profile.B05_Profile_Structures_Router import router as b05_structures_router
 from B06_Section.B06_Section_Router import router as b06_section_router
 from B06_Section.B06_Section_Router_Stations import router as b06_section_stations_router
+from B06_Section.B06_Section_Router_DesignSave import router as b06_design_save_router
 from B06_Section.B06_Section_Router_Confirm import (
     router as b06_section_confirm_router,
 )
@@ -646,6 +647,7 @@ app.include_router(b05_route_terrain_router, dependencies=protected_with_company
 app.include_router(b05_structures_router, dependencies=protected_with_company)
 app.include_router(b06_section_router, dependencies=protected_with_company)
 app.include_router(b06_section_stations_router, dependencies=protected_with_company)
+app.include_router(b06_design_save_router, dependencies=protected_with_company)
 app.include_router(b06_section_confirm_router, dependencies=protected_with_company)
 app.include_router(b06_section_haul_plan_router, dependencies=protected_with_company)
 app.include_router(b06_section_cross_table_router, dependencies=protected_with_company)
diff --git a/resources/tester/test_472_6_design_save.py b/resources/tester/test_472_6_design_save.py
new file mode 100644
index 000000000..1b82a0dd5
--- /dev/null
+++ b/resources/tester/test_472_6_design_save.py
@@ -0,0 +1,245 @@
+"""PLAN 472-6 6-A — [저장] · [확정] 한 본문 창구 · 전부 아니면 없음.
+
+  ① storage 떠 두기 · 되돌림 — 바뀐 · 새로 생긴 · 지운 파일만 되씀(안 바뀐 파일 그대로).
+  ② 실제 창구(검증_sub8 · DB · storage) — 중간 차례 실패 → DB 7 표 행 · storage 해시 시작 그대로 ·
+     차례 이름 · 까닭 · restored. 같은 본문 정상 = 옛 차례(안쪽 창구 차례로 부름)와 행 · 파일 같음.
+검증 프로젝트가 없는 자리(다른 PC · DB 없음)는 ② 건너뜀.
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_472_6_design_save.py -q
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import sys
+from pathlib import Path
+from uuid import UUID
+
+import pytest
+
+HERE = Path(__file__).resolve().parent
+ROOT = HERE.parents[1]
+sys.path.insert(0, str(ROOT))
+
+from common_util import common_util_save_guard as guard  # noqa: E402
+
+PROJECT = "9872e9a7-e483-427b-a6b2-72cdbc063bc3"  # 검증_sub8(OWNERS.md)
+VOLATILE = ("created_at", "updated_at", "confirmed_at", "started_at", "completed_at")
+
+
+def test_storage_되돌림은_바뀐_것만(tmp_path) -> None:
+    (tmp_path / "B06_Section" / "a").mkdir(parents=True)
+    keep = tmp_path / "B06_Section" / "a" / "keep.json"
+    edit = tmp_path / "B06_Section" / "edit.json"
+    gone = tmp_path / "workflow.json"
+    for path, text in ((keep, "k"), (edit, "e"), (gone, "w")):
+        path.write_text(text, encoding="utf-8")
+    outside = tmp_path / "B03_FileInput.txt"
+    outside.write_text("o", encoding="utf-8")
+    snap = guard.snapshot_storage(tmp_path, ("B06_Section", "workflow.json"))
+    kept_stamp = keep.stat().st_mtime_ns
+    edit.write_text("changed", encoding="utf-8")
+    gone.unlink()
+    (tmp_path / "B06_Section" / "new.json").write_text("n", encoding="utf-8")
+    outside.write_text("o2", encoding="utf-8")  # 범위 밖은 안 건드림
+    assert snap.changed() == ["B06_Section/edit.json", "B06_Section/new.json", "workflow.json"]
+    assert snap.restore() == 3
+    assert edit.read_text(encoding="utf-8") == "e" and gone.read_text(encoding="utf-8") == "w"
+    assert not (tmp_path / "B06_Section" / "new.json").exists()
+    assert keep.stat().st_mtime_ns == kept_stamp and outside.read_text(encoding="utf-8") == "o2"
+    snap.discard()
+    assert not snap.copy_dir.exists()
+
+
+# ── 실제 창구 ──────────────────────────────────────────────────────────────
+
+
+def _db_ready() -> Path | None:
+    try:
+        from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+        from common_util.common_util_storage import resolve_stored_project_path
+        from config.config_db import close_db_pool, init_db_pool
+
+        async def find():
+            pool = await init_db_pool()
+            try:
+                async with pool.acquire() as connection:
+                    return await get_project_storage_relative_path(connection, UUID(PROJECT))
+            finally:
+                await close_db_pool()
+
+        root = Path(resolve_stored_project_path(asyncio.run(find())))
+        return root if (root / "B06_Section" / "longitudinal").is_dir() else None
+    except Exception:
+        return None
+
+
+ROOT_DIR = _db_ready()
+live = pytest.mark.skipif(ROOT_DIR is None, reason="검증_sub8 · DB 없음")
+
+
+def _state(root: Path) -> tuple[dict, dict]:
+    """(행 — 시각 칸 뺌, 파일 내용 해시) — 저장 범위 전부."""
+    from config.config_db import close_db_pool, init_db_pool
+
+    async def rows():
+        pool = await init_db_pool()
+        try:
+            async with pool.acquire() as connection:
+                return await guard.snapshot_rows(connection, PROJECT)
+        finally:
+            await close_db_pool()
+
+    raw = asyncio.run(rows())
+    clean = {
+        table: [{k: str(v) for k, v in row.items() if k not in VOLATILE} for row in values]
+        for table, values in raw.items()
+    }
+    scope = guard.STORAGE_SCOPE + guard.PIPE_SCOPE
+    files = {
+        relative: hashlib.sha1(path.read_bytes()).hexdigest()
+        for relative, path in guard._files(root, scope).items()
+        if not relative.startswith("_derived/")
+    }
+    return clean, files
+
+
+def _request(**extra):
+    from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest
+
+    body = {
+        "route_id": 380,
+        "settings": {"design_speed_kph": 20, "terrain_type": "special", "max_grade_pct": 13.0},
+        "plus_stations": {"chainages": [500.0]},
+        "cross": {},
+        **extra,
+    }
+    return DesignSaveRequest.model_validate(body)
+
+
+def _save(request, monkeypatch=None):
+    from B06_Section.B06_Section_Router_DesignSave import save_design
+    from config.config_db import close_db_pool, init_db_pool
+
+    async def run():
+        await init_db_pool()
+        try:
+            return await save_design(UUID(PROJECT), request, None)
+        finally:
+            await close_db_pool()
+
+    response = asyncio.run(run())
+    return response.status_code, json.loads(bytes(response.body).decode("utf-8"))
+
+
+@live
+def test_중간_차례_실패면_DB_storage_그대로(monkeypatch) -> None:
+    from B06_Section import B06_Section_Router_Stations as stations
+
+    before = _state(ROOT_DIR)
+
+    async def broken(project_id, request):
+        (ROOT_DIR / "B06_Section" / "half_written.json").write_text("{}", encoding="utf-8")
+        raise RuntimeError("일부러 실패")
+
+    monkeypatch.setattr(stations, "put_plus_stations", broken)
+    status, body = _save(_request())
+    assert status == 409 and body["step"] == "plusStations" and body["restored"] is True
+    assert "일부러 실패" in body["message"]
+    after = _state(ROOT_DIR)
+    assert after[0] == before[0]  # 앞 차례(페이지 설정)가 쓴 단계 params 도 되돌림
+    assert after[1] == before[1]
+
+
+@live
+def test_정상_한_본문은_옛_차례와_같음() -> None:
+    from B05_Profile.B05_Profile_Router_Lifecycle import save_page_settings
+    from B06_Section.B06_Section_Router_Confirm import save_sections
+    from B06_Section.B06_Section_Router_Stations import put_plus_stations
+    from config.config_db import close_db_pool, get_db_pool, init_db_pool
+
+    start = _state(ROOT_DIR)
+    request = _request()
+
+    async def old_way():
+        await init_db_pool()
+        try:
+            async with get_db_pool().acquire() as connection:
+                rows = await guard.snapshot_rows(connection, PROJECT)
+            files = guard.snapshot_storage(ROOT_DIR, guard.STORAGE_SCOPE)
+            await save_page_settings(UUID(PROJECT), request.settings)
+            await put_plus_stations(UUID(PROJECT), request.plus_stations)
+            await save_sections(UUID(PROJECT), 380, request.cross)
+            return rows, files
+        finally:
+            await close_db_pool()
+
+    rows, files = asyncio.run(old_way())
+    old = _state(ROOT_DIR)
+
+    async def undo():
+        await init_db_pool()
+        try:
+            files.restore()
+            async with get_db_pool().acquire() as connection:
+                await guard.restore_rows(connection, PROJECT, rows)
+        finally:
+            await close_db_pool()
+
+    asyncio.run(undo())
+    assert _state(ROOT_DIR) == start
+    try:
+        status, body = _save(request)
+        assert status == 200 and body["status"] == "success", body
+        assert set(body["steps"]) == {"settings", "plusStations", "cross"}
+        new = _state(ROOT_DIR)
+        assert new[0] == old[0]
+        assert new[1] == old[1]
+    finally:
+        asyncio.run(undo())
+        files.discard()
+    assert _state(ROOT_DIR) == start
+
+
+@live
+def test_되돌림마저_실패면_복구_확인_필요(monkeypatch) -> None:
+    from B05_Profile import B05_Profile_Router_Lifecycle as lifecycle
+
+    async def broken(project_id, request):
+        raise RuntimeError("설정 실패")
+
+    def no_restore(self):
+        raise OSError("디스크 잠김")
+
+    monkeypatch.setattr(lifecycle, "save_page_settings", broken)
+    monkeypatch.setattr(guard.StorageSnapshot, "restore", no_restore)
+    status, body = _save(_request())
+    assert status == 500 and body["restored"] is False and body["step"] == "settings"
+    assert "복구 확인 필요" in body["message"]
+
+
+@live
+def test_같은_프로젝트_저장이_겹치면_뒤_요청이_기다림() -> None:
+    from B06_Section.B06_Section_Router_DesignSave import DesignSaveRequest, save_design
+    from config.config_db import close_db_pool, init_db_pool
+
+    async def run():
+        await init_db_pool()
+        try:
+            lock = guard.project_lock(PROJECT)
+            await lock.acquire()
+            task = asyncio.create_task(
+                save_design(UUID(PROJECT), DesignSaveRequest(route_id=380), None)
+            )
+            await asyncio.sleep(0.3)
+            assert not task.done()  # 앞 저장(잠금)이 끝날 때까지 기다림
+            lock.release()
+            return await task
+        finally:
+            await close_db_pool()
+
+    response = asyncio.run(run())
+    body = json.loads(bytes(response.body).decode("utf-8"))
+    assert body["status"] == "success" and body["waited"] is True and body["steps"] == {}