diff --git a/B06_Section/B06_Section_Router_Confirm.py b/B06_Section/B06_Section_Router_Confirm.py
index 7f78d29ab..d9a376803 100644
--- a/B06_Section/B06_Section_Router_Confirm.py
+++ b/B06_Section/B06_Section_Router_Confirm.py
@@ -48,7 +48,7 @@ from B06_Section.B06_Section_Schema import (
 )
 from B06_Section.B06_Section_Server_Calc_Prebuild import _AREA_KEYS
 from common_util.common_util_storage import resolve_stored_project_path
-from common_util.common_util_workflow_state import complete_stage, start_stage
+from common_util.common_util_workflow_state import complete_stage, mark_stage, start_stage
 from config.config_db import get_db_pool
 
 logger = logging.getLogger(__name__)
@@ -443,7 +443,7 @@ async def confirm_sections(
                         # 경로 상태(DRAFT)가 남지 않도록 여기서 함께 CONFIRMED로 닫는다.
                         await confirm_route_status(connection, route_id)
                         await complete_stage(cursor, str(project_id), 2)
-                        await complete_stage(cursor, str(project_id), 3)
+                        await mark_stage(str(project_id), 3, "next", cursor)  # 4 진행 중(B1)
                     else:
                         await start_stage(cursor, str(project_id), 3)
                 await connection.commit()
diff --git a/B06_Section/B06_Section_Router_DesignSave.py b/B06_Section/B06_Section_Router_DesignSave.py
index e29b95a64..9b669d2f3 100644
--- a/B06_Section/B06_Section_Router_DesignSave.py
+++ b/B06_Section/B06_Section_Router_DesignSave.py
@@ -186,15 +186,32 @@ async def _run_steps(
         save = confirm_sections if request.confirm else save_sections
         crossing = save(project_id, request.route_id, request.cross)
     else:
-        return out
-    cross = await step("cross", crossing)
-    if cross.get("recomputed") is False:
-        raise StepFailed("cross", "서버 재계산 실패")
-    if cross.get("stations_synced") is False:
-        raise StepFailed("cross", "구조물 측점 맞춤 실패")
+        crossing = None
+    if crossing is not None:
+        cross = await step("cross", crossing)
+        if cross.get("recomputed") is False:
+            raise StepFailed("cross", "서버 재계산 실패")
+        if cross.get("stations_synced") is False:
+            raise StepFailed("cross", "구조물 측점 맞춤 실패")
+    # [>](확정) = confirm_sections 가 단계를 닫음 · 그 밖 [저장] = 2 · 3 진행 중 · 뒤 STALE(B1)
+    sent = any(value is not None for key, value in request if key not in ("route_id", "confirm"))
+    if sent and not (request.confirm and not moved and request.cross is not None):
+        try:
+            await _save_stages(str(project_id))
+        except Exception as error:
+            logger.exception("한 본문 저장 단계 상태 실패: project_id=%s", project_id)
+            raise StepFailed("stages", str(error) or type(error).__name__) from None
     return out
 
 
+async def _save_stages(project_id: str) -> None:
+    """B05 · B06 한 본문 [저장] — 단계 2 · 3(실패면 한 본문 되돌림이 단계 행도 되살림)."""
+    from common_util.common_util_workflow_state import mark_stage
+
+    await mark_stage(project_id, 2, "save")
+    await mark_stage(project_id, 3, "save")
+
+
 @router.post("/{project_id}/design/save")
 async def save_design(
     project_id: UUID,
diff --git a/B07_DesignDetail/B07_DesignDetail_Router.py b/B07_DesignDetail/B07_DesignDetail_Router.py
index 2f5060c34..cfb99a5fa 100644
--- a/B07_DesignDetail/B07_DesignDetail_Router.py
+++ b/B07_DesignDetail/B07_DesignDetail_Router.py
@@ -64,7 +64,7 @@ from B07_DesignDetail.B07_DesignDetail_Schema import (
 from common_util.common_util_dev_unlock import is_dev_environment, unlock_status
 from common_util.common_util_drainage_context import load_drainage_context
 from common_util.common_util_storage import read_stored_asset, resolve_stored_project_path
-from common_util.common_util_workflow_state import complete_stage, start_stage
+from common_util.common_util_workflow_state import mark_stage, start_stage
 from config.config_db import get_db_pool
 
 logger = logging.getLogger(__name__)
@@ -470,10 +470,7 @@ async def confirm_design_drawing(
                         patch={"status": "confirmed"},
                     )
                 async with connection.cursor() as cursor:
-                    if all_confirmed:
-                        await complete_stage(cursor, str(project_id), 4)
-                    else:
-                        await start_stage(cursor, str(project_id), 4)
+                    await mark_stage(str(project_id), 4, "save", cursor)  # [저장](B1)
                 await connection.commit()
             except Exception:
                 await connection.rollback()
@@ -572,7 +569,7 @@ async def complete_design_stage(project_id: UUID) -> Any:
             await connection.begin()
             try:
                 async with connection.cursor() as cursor:
-                    await complete_stage(cursor, str(project_id), _B07_STAGE_NO)
+                    await mark_stage(str(project_id), _B07_STAGE_NO, "next", cursor)  # [>](B1)
                 await connection.commit()
             except Exception:
                 await connection.rollback()
diff --git a/B08_Quantity/B08_Quantity_Router_Hydro.py b/B08_Quantity/B08_Quantity_Router_Hydro.py
index dbfb95982..1f35a8477 100644
--- a/B08_Quantity/B08_Quantity_Router_Hydro.py
+++ b/B08_Quantity/B08_Quantity_Router_Hydro.py
@@ -37,6 +37,7 @@ from B08_Quantity.B08_Quantity_Hydro_Site import (
 )
 from common_util.common_util_drainage_pipes import pipe_points_path_in, read_pipe_points_file
 from common_util.common_util_storage import resolve_stored_project_path
+from common_util.common_util_workflow_state import mark_stage
 from config.config_db import get_db_pool
 
 router = APIRouter(prefix="/api/projects", tags=["B08 Hydro"])
@@ -126,4 +127,5 @@ async def hydro_save(project_id: UUID, body: HydroSave) -> dict[str, Any]:
             raise HTTPException(status_code=400, detail=f"{key}: 할증은 {FACTORS} 중 하나")
     saved = {key: inputs.model_dump() for key, inputs in body.stations.items()}
     write_saved(root, saved)
+    await mark_stage(str(project_id), 5, "save")  # [저장](B1)
     return {"saved": saved}
diff --git a/B08_Quantity/B08_Quantity_Router_Material.py b/B08_Quantity/B08_Quantity_Router_Material.py
index e0bb13ba7..aba4600f6 100644
--- a/B08_Quantity/B08_Quantity_Router_Material.py
+++ b/B08_Quantity/B08_Quantity_Router_Material.py
@@ -37,6 +37,7 @@ from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
 from B08_Quantity.B08_Quantity_Router_Summary import _combo, structure_summary
 from common_util.common_util_project_settings import quantity_settings, save_section
 from common_util.common_util_summary_item import GROUP_JOIN
+from common_util.common_util_workflow_state import mark_stage
 
 router = APIRouter(prefix="/api/projects", tags=["B08 Material"])
 
@@ -251,4 +252,5 @@ async def save_material_sheets(
     root = await _project_root(project_id)
     overrides = clean_overrides(body.get("overrides"))
     save_section(root, "quantity", {SETTINGS_KEY: overrides}, replace_keys=(SETTINGS_KEY,))
+    await mark_stage(str(project_id), 5, "save")  # [저장](B1)
     return await _sheets(project_id, root, overrides)
diff --git a/B08_Quantity/B08_Quantity_Router_Milestone.py b/B08_Quantity/B08_Quantity_Router_Milestone.py
index 1d6ec27d5..da687e23b 100644
--- a/B08_Quantity/B08_Quantity_Router_Milestone.py
+++ b/B08_Quantity/B08_Quantity_Router_Milestone.py
@@ -86,6 +86,7 @@ from common_util.common_util_route_geometry import (
     planned_route_working_path,
     read_planned_route_csv,
 )
+from common_util.common_util_workflow_state import mark_stage
 
 router = APIRouter(prefix="/api/projects", tags=["B08 Milestone"])
 PATH = "/{project_id}/quantity/milestone"
@@ -421,6 +422,7 @@ async def milestone_save(
 ) -> dict[str, Any]:
     root = await _project_root(project_id)
     out = await asyncio.to_thread(save_data, root, body)
+    await mark_stage(str(project_id), 5, "save")  # [저장](B1)
     if not os.environ.get(
         "PYTEST_CURRENT_TEST"
     ):  # 남의 시험 [저장] = 실제 카카오 · 브라우저 안 부름
diff --git a/B08_Quantity/B08_Quantity_Router_Options.py b/B08_Quantity/B08_Quantity_Router_Options.py
index ff4aa21b3..de4c9c179 100644
--- a/B08_Quantity/B08_Quantity_Router_Options.py
+++ b/B08_Quantity/B08_Quantity_Router_Options.py
@@ -2,6 +2,7 @@
 
     GET /api/projects/{id}/quantity/options   지금 값
     PUT /api/projects/{id}/quantity/options   [저장] — 받은 칸만 정본 자리에 씀
+    POST /api/projects/{id}/quantity/complete [>] — 단계 5 완료 · 6 진행 중(요구 300 B1)
 
 1본 길이 = 프로젝트 층 M02 구조물집계표 변수 `파형강관_1본_m` 한 벌(98 `one_pipe_length` 이 읽는 자리 ·
 M02 표 식 `[$파형강관_1본_m]` 도 같은 값) · 프로젝트 사본이 없으면 시스템 층 문서를 바탕으로 깔고 씀 ·
@@ -26,6 +27,7 @@ from B08_Quantity.B08_Quantity_Material_Calc import PANEL_KEY, panel_surcharge
 from B08_Quantity.B08_Quantity_Router_Hydro import _project_root
 from B08_Quantity.B08_Quantity_Router_Summary import ONE_LENGTH_VAR, TABLE_NAME, one_pipe_length
 from common_util.common_util_project_settings import quantity_settings, save_section, soft_rock_pct
+from common_util.common_util_workflow_state import mark_stage
 from M02_MasterTemplete import M02_Template_Layers as layers
 
 router = APIRouter(prefix="/api/projects", tags=["B08 Options"])
@@ -96,8 +98,18 @@ async def options_read(project_id: UUID) -> dict[str, Any]:
 async def options_save(project_id: UUID, body: OptionsSave) -> dict[str, Any]:
     root = await _project_root(project_id)
     try:
-        return await asyncio.to_thread(save_options, root, body)
+        out = await asyncio.to_thread(save_options, root, body)
     except layers.StaleTemplate as error:
         raise HTTPException(
             status_code=409, detail="구조물집계표가 그새 바뀌었습니다 — 다시 열고 저장하세요."
         ) from error
+    await mark_stage(str(project_id), 5, "save")  # [저장](B1)
+    return out
+
+
+@router.post("/{project_id}/quantity/complete")
+async def quantity_complete(project_id: UUID) -> dict[str, Any]:
+    """B08 [>] — 단계 5 완료 · 6 진행 중(요구 300 B1 · 본문 없음)."""
+    await _project_root(project_id)  # 없는 프로젝트는 404
+    await mark_stage(str(project_id), 5, "next")
+    return {"status": "success", "project_id": str(project_id)}
diff --git a/B09_Estimation/B09_Estimation_Router.py b/B09_Estimation/B09_Estimation_Router.py
index 0da23dee3..a3e8ef66a 100644
--- a/B09_Estimation/B09_Estimation_Router.py
+++ b/B09_Estimation/B09_Estimation_Router.py
@@ -4,6 +4,8 @@
         입력(굳힌 단가 · 수요 · B08 수량) + 사용자 값 → 표 전부
     PUT  /api/projects/{id}/estimation   {user}
         [저장] — 사용자 값 · 책에 없는 새 단가 줄만 굳힘 → 서버가 정본으로 다시 셈
+    POST /api/projects/{id}/estimation/complete
+        [>] — 단계 6 완료 · 프로젝트 DONE(요구 300 B1 · 본문 없음)
     POST /api/projects/{id}/estimation/prices   {user?}
         [단가 갱신] — 수요 단가 전부를 지금 M01 값으로 새로 굳힘 · user = 화면 캐시(296 ·
         지운 줄 · 그 사본은 수요에서 뺌 · 저장은 안 함) · M01 원본이 지워진 줄 = 굳힌 값 그대로
@@ -61,6 +63,7 @@ from B09_Estimation.B09_Estimation_Rates import rate_candidates, rate_table
 from B09_Estimation.B09_Estimation_Router_Quotes import router as quotes_router
 from B09_Estimation.B09_Estimation_Summary import compute_summary, hopyo_lines, used_materials
 from common_util.common_util_auth import verify_session
+from common_util.common_util_workflow_state import mark_stage
 from M02_MasterTemplete import M02_Layer_Access as access
 from M02_MasterTemplete.M02_Item_Snapshot import importer
 
@@ -216,9 +219,18 @@ async def save_estimation(project_id: UUID, body: dict[str, Any] = Body(...)) ->
     book = await asyncio.to_thread(store.read_book, root)
     book = await asyncio.to_thread(prices.extend_book, book, keys)
     await asyncio.to_thread(store.write_book, root, book)
+    await mark_stage(str(project_id), 6, "save")  # [저장](B1)
     return await payload(project_id, root)
 
 
+@router.post("/{project_id}/estimation/complete")
+async def estimation_complete(project_id: UUID) -> dict[str, Any]:
+    """B09 [>] — 단계 6 완료(요구 300 B1)."""
+    await _project_root(project_id)  # 없는 프로젝트는 404
+    await mark_stage(str(project_id), 6, "next")
+    return {"status": "success", "project_id": str(project_id)}
+
+
 @router.post("/{project_id}/estimation/prices")
 async def refresh_prices(
     project_id: UUID, body: dict[str, Any] | None = Body(None)
diff --git a/common_util/common_util_workflow_state.py b/common_util/common_util_workflow_state.py
index c5c85c27b..412d06e3e 100644
--- a/common_util/common_util_workflow_state.py
+++ b/common_util/common_util_workflow_state.py
@@ -148,6 +148,52 @@ async def complete_stage(cursor: aiomysql.DictCursor, project_id: str, stage_no:
     )
 
 
+async def mark_stage(
+    project_id: str,
+    stage_no: int,
+    action: str,
+    cursor: aiomysql.DictCursor | None = None,
+) -> None:
+    """사용자 단추가 바꾸는 단계 상태 한 곳(요구 300 B1) — 커서 없으면 제 트랜잭션.
+
+    `save`([저장] 성공) = 그 단계 IN_PROGRESS · 뒤 단계 중 COMPLETE · IN_PROGRESS → STALE.
+    `next`([>] 성공) = 그 단계 COMPLETE · 다음 단계 NOT_STARTED · STALE → IN_PROGRESS.
+    [<] = 부르지 않음(상태 안 바꿈).
+    """
+    if action not in ("save", "next"):
+        raise ValueError(f"단계 동작은 save · next 중 하나: {action}")
+    if cursor is None:
+        from config import config_db
+
+        if config_db.db_pool is None:  # DB 없이 파일 창구만 부르는 시험 = 단계 표 없음
+            return
+        async with config_db.get_db_pool().acquire() as connection:
+            await connection.begin()
+            try:
+                async with connection.cursor() as own:
+                    await mark_stage(project_id, stage_no, action, own)
+                await connection.commit()
+            except Exception:
+                await connection.rollback()
+                raise
+        return
+    if action == "save":
+        await start_stage(cursor, project_id, stage_no)
+        await cursor.execute(
+            "UPDATE project_workflow_stages SET state = 'STALE' "
+            "WHERE project_id = %s AND stage_no > %s AND state = 'IN_PROGRESS'",
+            (project_id, stage_no),
+        )
+        return
+    await complete_stage(cursor, project_id, stage_no)
+    await cursor.execute(
+        "UPDATE project_workflow_stages SET state = 'IN_PROGRESS', progress_percent = 0, "
+        "message = NULL, started_at = %s, completed_at = NULL "
+        "WHERE project_id = %s AND stage_no = %s AND state IN ('NOT_STARTED', 'STALE')",
+        (datetime.utcnow(), project_id, stage_no + 1),
+    )
+
+
 async def update_stage_progress(
     cursor: aiomysql.DictCursor, project_id: str, stage_no: int, progress: int
 ) -> None:
diff --git a/resources/tester/test_280_5_b07_settings.py b/resources/tester/test_280_5_b07_settings.py
index 69ba47d16..e12fced54 100644
--- a/resources/tester/test_280_5_b07_settings.py
+++ b/resources/tester/test_280_5_b07_settings.py
@@ -199,18 +199,18 @@ def test_서버_단계_완료(monkeypatch, tmp_path: Path) -> None:
     async def source(_project_id: UUID) -> tuple[int, Path, Path, bool]:
         return 1, tmp_path, tmp_path / "longitudinal.json", True
 
-    async def complete(_cursor: object, project_id: str, stage_no: int) -> None:
-        stages.append((project_id, stage_no))
+    async def complete(project_id: str, stage_no: int, action: str, _cursor: object) -> None:
+        stages.append((project_id, stage_no, action))
 
     monkeypatch.setattr(router, "_confirmed_source", source)
-    monkeypatch.setattr(router, "complete_stage", complete)
+    monkeypatch.setattr(router, "mark_stage", complete)
     monkeypatch.setattr(
         router, "get_db_pool", lambda: type("P", (), {"acquire": lambda self: _Acquire()})()
     )
     project = UUID("8b98ca75-c572-4002-8468-2adae4323414")
     result = asyncio.run(router.complete_design_stage(project))
     assert result == {"status": "success", "project_id": str(project)}
-    assert stages == [(str(project), 4)]
+    assert stages == [(str(project), 4, "next")]  # 298-2 [>] = 4 완료 · 5 진행 중
 
     async def blocked(_project_id: UUID) -> tuple[int, Path, Path, bool]:
         raise PermissionError("B06 종·횡단 확정 후 상세 설계를 진행할 수 있습니다.")
diff --git a/resources/tester/test_298_2_stage_state.py b/resources/tester/test_298_2_stage_state.py
new file mode 100644
index 000000000..309acd0e3
--- /dev/null
+++ b/resources/tester/test_298_2_stage_state.py
@@ -0,0 +1,263 @@
+"""요구 298-2 · 300 B1 — 단추가 바꾸는 진행 단계 상태(PLAN 496-2).
+
+[저장] = 그 단계 IN_PROGRESS · 뒤 COMPLETE · IN_PROGRESS → STALE · [>] = 그 단계 COMPLETE ·
+다음 NOT_STARTED · STALE → IN_PROGRESS · [<] = 안 바꿈. 단계 표는 메모리 흉내 커서.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import re
+from contextlib import asynccontextmanager
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+
+from common_util import common_util_workflow_state as wf
+
+N, P, C, S = "NOT_STARTED", "IN_PROGRESS", "COMPLETE", "STALE"
+
+
+class _Table:
+    """`project_workflow_stages` 칸 0~6 의 state 만 — 단계 SQL 을 글로 풀어 셈."""
+
+    def __init__(self, states: list[str]):
+        self.states = list(states)
+
+    def run(self, sql: str, params: tuple) -> None:
+        if "project_workflow_stages" not in sql or not sql.lstrip().startswith("UPDATE"):
+            return
+        to = re.search(r"SET state = '(\w+)'", sql).group(1)
+        no = int(params[-1])
+        only = re.search(r"AND state = '(\w+)'|AND state IN \(([^)]*)\)", sql)
+        allowed = None
+        if only:
+            allowed = re.findall(r"\w+", only.group(1) or only.group(2))
+        for i, state in enumerate(self.states):
+            hit = i > no if "stage_no >" in sql else i == no
+            if hit and (allowed is None or state in allowed):
+                self.states[i] = to
+
+
+class _Cursor:
+    def __init__(self, table: _Table):
+        self.table = table
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        return False
+
+    async def execute(self, sql, params=()):
+        self.table.run(sql, params)
+
+
+class _Connection:
+    def __init__(self, table):
+        self.table = table
+        self.commits = 0
+
+    def cursor(self, *_a, **_k):
+        return _Cursor(self.table)
+
+    async def begin(self):
+        pass
+
+    async def commit(self):
+        self.commits += 1
+
+    async def rollback(self):
+        pass
+
+
+class _Pool:
+    def __init__(self, table):
+        self.table = table
+
+    @asynccontextmanager
+    async def acquire(self):
+        yield _Connection(self.table)
+
+
+def _mark(states, stage_no, action):
+    table = _Table(states)
+    asyncio.run(wf.mark_stage("p", stage_no, action, _Cursor(table)))
+    return table.states
+
+
+@pytest.mark.parametrize(
+    ("before", "stage_no", "after"),
+    [
+        # B06 [>] — 3 닫고 B07 진행 중
+        ([C, C, C, P, N, N, N], 3, [C, C, C, C, P, N, N]),
+        # 뒤 단계가 낡음이면 진행 중으로
+        ([C, C, C, P, S, S, N], 3, [C, C, C, C, P, S, N]),
+        # 다음이 이미 끝났으면 그대로
+        ([C, C, C, P, C, C, N], 3, [C, C, C, C, C, C, N]),
+        # B09 [>] — 마지막 칸 · 다음 없음
+        ([C, C, C, C, C, C, P], 6, [C, C, C, C, C, C, C]),
+    ],
+)
+def test_next(before, stage_no, after):
+    assert _mark(before, stage_no, "next") == after
+
+
+@pytest.mark.parametrize(
+    ("before", "stage_no", "after"),
+    [
+        # B07 [저장] — B08 끝남 · B09 진행 중 → 둘 다 낡음
+        ([C, C, C, C, C, C, P], 4, [C, C, C, C, P, S, S]),
+        # 처음 저장 — 뒤 NOT_STARTED 는 그대로
+        ([C, C, C, C, N, N, N], 4, [C, C, C, C, P, N, N]),
+        # 끝난 단계 다시 저장 = 진행 중
+        ([C, C, C, C, C, C, C], 6, [C, C, C, C, C, C, P]),
+    ],
+)
+def test_save(before, stage_no, after):
+    assert _mark(before, stage_no, "save") == after
+
+
+def test_bad_action():
+    with pytest.raises(ValueError):
+        _mark([N] * 7, 2, "prev")
+
+
+def test_own_transaction(monkeypatch):
+    from config import config_db
+
+    table = _Table([C, C, C, C, C, N, N])
+    monkeypatch.setattr(config_db, "db_pool", None)
+    asyncio.run(wf.mark_stage("p", 5, "next"))  # 풀 없음(파일 창구 시험) = 건너뜀
+    assert table.states == [C, C, C, C, C, N, N]
+    monkeypatch.setattr(config_db, "db_pool", _Pool(table))
+    asyncio.run(wf.mark_stage("p", 5, "next"))
+    assert table.states == [C, C, C, C, C, C, P]
+
+
+def _design_save(monkeypatch, table, **body):
+    from B04_PreProcess import B04_PreProcess_Router_Basins as basins
+    from B06_Section import B06_Section_Router_Confirm as confirm
+    from B06_Section import B06_Section_Router_DesignSave as design
+    from config import config_db
+
+    async def done(*_a, **_k):
+        return {}
+
+    async def confirmed(project_id, *_a, **_k):  # confirm_sections 의 단계 몫만
+        await wf.complete_stage(_Cursor(table), str(project_id), 2)
+        await wf.mark_stage(str(project_id), 3, "next", _Cursor(table))
+        return {}
+
+    monkeypatch.setattr(basins, "put_pipe_points", done)
+    monkeypatch.setattr(confirm, "save_sections", done)
+    monkeypatch.setattr(confirm, "confirm_sections", confirmed)
+    monkeypatch.setattr(config_db, "db_pool", _Pool(table))
+    request = design.DesignSaveRequest(route_id=1, **body)
+    return asyncio.run(design._run_steps(uuid4(), request, None))
+
+
+def test_b05_b06_save_any_step(monkeypatch):
+    """관만 고친 [저장]도 2 · 3 진행 중 · B07 · B08 · B09 낡음."""
+    table = _Table([C, C, C, C, C, P, N])
+    _design_save(monkeypatch, table, pipes={})
+    assert table.states == [C, C, P, P, S, S, N]
+
+
+def test_b05_b06_save_empty(monkeypatch):
+    """빈 본문 = 저장한 것 없음 · 단계 그대로."""
+    table = _Table([C, C, C, C, C, P, N])
+    assert _design_save(monkeypatch, table) == {}
+    assert table.states == [C, C, C, C, C, P, N]
+
+
+def test_b05_b06_save_cross(monkeypatch):
+    table = _Table([C, C, C, C, N, N, N])
+    _design_save(monkeypatch, table, cross={})
+    assert table.states == [C, C, P, P, N, N, N]
+
+
+def test_b06_next_confirm(monkeypatch):
+    """[>](confirm) = 2 · 3 COMPLETE · 4 진행 중 · [저장] 몫 안 탐."""
+    table = _Table([C, C, P, P, S, N, N])
+    _design_save(monkeypatch, table, confirm=True, cross={})
+    assert table.states == [C, C, C, C, P, N, N]
+
+
+def test_b07_next(monkeypatch):
+    from B07_DesignDetail import B07_DesignDetail_Router as b07
+
+    async def source(_pid):
+        return 1, Path("."), "l", False
+
+    table = _Table([C, C, C, C, P, S, N])
+    monkeypatch.setattr(b07, "_confirmed_source", source)
+    monkeypatch.setattr(b07, "get_db_pool", lambda: _Pool(table))
+    assert asyncio.run(b07.complete_design_stage(uuid4()))["status"] == "success"
+    assert table.states == [C, C, C, C, C, P, N]
+
+
+def test_b07_save_calls_save():
+    """B07 장 [저장](confirm PUT) = 단계 4 save 한 줄 · 옛 all_confirmed 갈래 없음."""
+    text = (Path(__file__).parents[2] / "B07_DesignDetail/B07_DesignDetail_Router.py").read_text(
+        encoding="utf-8"
+    )
+    assert 'mark_stage(str(project_id), 4, "save", cursor)' in text
+    assert "complete_stage(" not in text
+
+
+@pytest.mark.parametrize(
+    ("module", "name", "stage_no"),
+    [
+        ("B08_Quantity.B08_Quantity_Router_Options", "quantity_complete", 5),
+        ("B09_Estimation.B09_Estimation_Router", "estimation_complete", 6),
+    ],
+)
+def test_b08_b09_next(monkeypatch, module, name, stage_no):
+    import importlib
+
+    mod = importlib.import_module(module)
+    calls = []
+
+    async def root(_pid):
+        return Path(".")
+
+    async def mark(project_id, no, action, cursor=None):
+        calls.append((no, action))
+
+    monkeypatch.setattr(mod, "_project_root", root)
+    monkeypatch.setattr(mod, "mark_stage", mark)
+    assert asyncio.run(getattr(mod, name)(uuid4()))["status"] == "success"
+    assert calls == [(stage_no, "next")]
+
+
+def test_b08_hydro_save(monkeypatch):
+    from B08_Quantity import B08_Quantity_Router_Hydro as hydro
+
+    calls = []
+
+    async def root(_pid):
+        return Path(".")
+
+    async def mark(project_id, no, action, cursor=None):
+        calls.append((no, action))
+
+    monkeypatch.setattr(hydro, "_project_root", root)
+    monkeypatch.setattr(hydro, "write_saved", lambda *_a: None)
+    monkeypatch.setattr(hydro, "mark_stage", mark)
+    asyncio.run(hydro.hydro_save(uuid4(), hydro.HydroSave(stations={})))
+    assert calls == [(5, "save")]
+
+
+def test_save_routers_mark_stage():
+    """B08 [저장] 넷 · B09 [저장] = 그 단계 save."""
+    root = Path(__file__).parents[2]
+    for rel, stage_no in [
+        ("B08_Quantity/B08_Quantity_Router_Options.py", 5),
+        ("B08_Quantity/B08_Quantity_Router_Material.py", 5),
+        ("B08_Quantity/B08_Quantity_Router_Milestone.py", 5),
+        ("B09_Estimation/B09_Estimation_Router.py", 6),
+    ]:
+        text = (root / rel).read_text(encoding="utf-8")
+        assert f'mark_stage(str(project_id), {stage_no}, "save")' in text, rel
diff --git a/resources/tester/test_472_6_design_save.py b/resources/tester/test_472_6_design_save.py
index 63c3634d8..63355dce9 100644
--- a/resources/tester/test_472_6_design_save.py
+++ b/resources/tester/test_472_6_design_save.py
@@ -212,6 +212,7 @@ def test_중간_차례_실패면_DB_storage_그대로(monkeypatch) -> None:
 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_DesignSave import _save_stages
     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
 
@@ -227,6 +228,7 @@ def test_정상_한_본문은_옛_차례와_같음() -> None:
             await save_page_settings(UUID(PROJECT), request.settings)
             await put_plus_stations(UUID(PROJECT), request.plus_stations)
             await save_sections(UUID(PROJECT), request.route_id, request.cross)
+            await _save_stages(PROJECT)  # 298-2 [저장] = 단계 2 · 3 진행 중(B1)
             return rows, files
         finally:
             await close_db_pool()
diff --git a/resources/tester/test_b07_confirm_keeps_design.py b/resources/tester/test_b07_confirm_keeps_design.py
index f33273808..91922f7f8 100644
--- a/resources/tester/test_b07_confirm_keeps_design.py
+++ b/resources/tester/test_b07_confirm_keeps_design.py
@@ -80,8 +80,7 @@ def test_확정은_상태만_올리고_설계값을_안_덮는다(monkeypatch, t
     monkeypatch.setattr(router, "_store_confirmed_drawing", lambda *_: False)
     monkeypatch.setattr(router, "get_db_pool", lambda: SimpleNamespace(acquire=_Acquire))
     monkeypatch.setattr(router, "merge_cross_section_design_by_round", merge)
-    monkeypatch.setattr(router, "start_stage", stage)
-    monkeypatch.setattr(router, "complete_stage", stage)
+    monkeypatch.setattr(router, "mark_stage", stage)
     # 옛 길 — 입력 셋으로 다시 계산한 값(암 절토경사가 빠져 절토가 줄어든 모양)을 흉내 낸다.
     monkeypatch.setattr(
         router,