feat(496-2 단계 상태): [저장] · [>] 가 바꾸는 진행 단계 한 곳 — 요구 298-2 · 300 B1

- 공용 mark_stage — save = 그 단계 진행 중 · 뒤 완료 · 진행 중 → 낡음 / next = 그 단계 완료 · 다음 미시작 · 낡음 → 진행 중
- B05 · B06 한 본문 [저장] = 단계 2 · 3 save(빈 본문 · [>] 확정 제외) · B06 [>] = 2 완료 · 3 next
- B07 장 [저장] = 4 save · B07 [>] = 4 next
- B08 [저장] 넷(옵션 · 자재 · 수문 · 이정) = 5 save · 새 POST quantity/complete = 5 next
- B09 [저장] = 6 save · 새 POST estimation/complete = 6 next
- 자동설계 체인(mark_stage_complete=False) 그대로
- 새 시험 test_298_2_stage_state · 이웃 시험 셋 기대값 맞춤

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

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: