fix(763-3): B04 분석 · 조합 짓기 = 보기만 · 다른 서피스 [확정] = 업로드와 같은 전처리 체인 · 체인 실패에도 미리 셈

- B04 [분석](조합 짓기) — 단계 시작 · 확정 해제 · 실패 단계 기록 걷음 · 1 단계 값(las_selection 포함) 그대로 · 미리 셈 없음 · 진행 글 = 확정 있으면 「확정 그대로」
- 분석 결과 DB 고쳐 쓰기가 확정 모델을 COMPLETE 로 내리던 것 — CONFIRMED 그대로(`_upsert_surface_model`) · 안 쓰는 `clear_confirmed_surface_models` 지움
- B04 [확정] — 같은 서피스(필터 · 표현 · 평활) = 등고선 간격만 · 다른 서피스 = `run_surface_change_chain`(새 파일): 초기값 스냅샷 버림 · B05~B09 지움(같은 지형 재업로드 길 keep_preprocess) · 단계 2 이후 처음 → 자동 설계 체인
- 자동 · 재확정 체인 미리 셈 = finally(실패 · 건너뜀 길에도) · B05 LAS 나중 올리기 끝 미리 셈 예약
- 추천 [적용](as_initial) = 선만 · 설계 지형 그대로(요구 605-2 · 브레인 승인) · 지형 칩은 사용자 답 기다림(지금대로)
- 지킴 시험 test_763_3 — B05~B09 파이썬에서 확정 서피스 쓰는 부름은 지형 칩 · 라이다 적어 둠 둘만
- 시험 309_3 · 351_1 · 509_5 를 새 길에 맞춤(미리 셈 부르는 곳 둘 더 · finally · 추천 [적용] 지형 안 넘김)
- 재확정 체인(`run_redesign_chain`) 부르는 곳 0 — 지움은 따로(시험 342_2 셋 함께)

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

B03_FileInput/B03_FileInput_Service_Chain.py

@@ -347,12 +347,6 @@ async def run_auto_design_chain(
             logger.exception("초기값 스냅샷 실패: project_id=%s", project_id)
             mark_design_failed(project_root, f"초기값 스냅샷 저장 실패: {exc}")
 
-        # 노선 추천 기본 다섯 알약 미리 셈(PLAN 327-2) — 예약만 · 체인이 [초기화] 요청 안에서도
-        # 불리므로 기다리지 않는다. 같은 입력이면 저장본이 있어 셈 없이 끝난다.
-        from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
-
-        schedule_precompute(project_id)
-
         return {
             "route_id": route_id,
             "length_m": float(solve_result.total_length_m or 0.0),
@@ -367,6 +361,10 @@ async def run_auto_design_chain(
         # 성공·실패·중단 어느 쪽이든 문은 연다 — 마커가 남으면 영영 못 들어간다.
         if project_root is not None:
             clear_designing(project_root)
+        # 성공 · 실패 · 건너뜀 어느 쪽이든 추천 미리 셈 예약(763-3) — 추천은 지형 · 예정노선만 씀
+        from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
+
+        schedule_precompute(project_id)
 
 
 async def _ensure_initial_polyline(
@@ -670,10 +668,6 @@ async def run_redesign_chain(
             return f"종횡단 확정 실패({sections_result.status_code})"
         marks.append(("B06 확정(서버 재계산 포함)", time.perf_counter()))
         _log_steps("재확정 체인", marks)
-        # 지형이 바뀌면 추천 키도 바뀜 — 첫 누름이 저장본이게 미리 셈(351-1)
-        from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
-
-        schedule_precompute(project_id)
         logger.info(
             "재확정 체인 완료: project_id=%s route %s→%s (설계 %d건 이월)",
             project_id,
@@ -688,3 +682,7 @@ async def run_redesign_chain(
     finally:
         if project_root is not None:
             clear_designing(project_root)
+        # 성공 · 실패 · 건너뜀 어느 쪽이든 추천 미리 셈 예약(763-3) — 추천은 지형 · 예정노선만 씀
+        from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
+
+        schedule_precompute(project_id)

B03_FileInput/B03_FileInput_Service_SurfaceChange.py

@@ -0,0 +1,48 @@
+"""서피스가 바뀐 뒤 전처리 체인 다시(763-3 · 요구 605) — 업로드와 같은 길 하나.
+
+부르는 곳 = B04 다른 서피스 [확정]. B05~B09 초기화 = 같은 지형 재업로드 길
+(`purge_project_outputs(keep_preprocess=True)` · 단계 2 이후 처음 · 초기값 스냅샷 버림) →
+`run_auto_design_chain`(끝에 추천 미리 셈 · 실패 길에도). 전처리 결과(B04) · 확정은 그대로.
+"""
+
+import asyncio
+import logging
+from pathlib import Path
+from uuid import UUID
+
+logger = logging.getLogger(__name__)
+
+
+async def run_surface_change_chain(project_id: UUID, surface_model_id: int | None) -> None:
+    """B05~B09 를 지우고 자동 설계 체인을 새 확정 서피스로 다시 탐 — 실패는 로그만."""
+    from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
+    from B03_FileInput.B03_FileInput_Service_Chain import run_auto_design_chain
+    from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
+    from common_util.common_util_initial_snapshot import discard_initial_snapshot
+    from common_util.common_util_project_reset import purge_project_outputs
+    from common_util.common_util_storage import resolve_stored_project_path
+    from common_util.common_util_workflow_state import reset_stages_after_input_change
+    from config.config_db import get_db_pool
+
+    try:
+        async with get_db_pool().acquire() as connection:
+            stored_path = await get_project_storage_relative_path(connection, project_id)
+            if not stored_path:
+                return
+            root = Path(resolve_stored_project_path(stored_path))
+            await asyncio.to_thread(discard_initial_snapshot, root)
+            await connection.begin()
+            try:
+                await purge_project_outputs(connection, str(project_id), root, keep_preprocess=True)
+                async with connection.cursor() as cursor:
+                    await reset_stages_after_input_change(cursor, str(project_id), 2)
+                await connection.commit()
+            except Exception:
+                await connection.rollback()
+                raise
+    except Exception:  # noqa: BLE001 — 체인은 응답 뒤 · 추천은 새 서피스로 셈
+        logger.exception("서피스 바꿈 체인 — B05~B09 초기화 실패: project_id=%s", project_id)
+        schedule_precompute(project_id)
+        return
+    logger.info("서피스 바꿈 체인 — B05~B09 초기화 뒤 자동 설계: project_id=%s", project_id)
+    await run_auto_design_chain(project_id, surface_model_id=surface_model_id)

B04_PreProcess/B04_PreProcess_Repository.py

@@ -287,19 +287,6 @@ async def list_surface_models(
     return models
 
 
-async def clear_confirmed_surface_models(connection: aiomysql.Connection, project_id: UUID) -> None:
-    """재분석 시 기존 확정을 해제하여 새 분석 결과의 재확정을 요구한다."""
-    async with connection.cursor() as cursor:
-        await cursor.execute(
-            """
-            UPDATE surface_models
-            SET status = 'COMPLETE'
-            WHERE project_id = %s AND status = 'CONFIRMED'
-            """,
-            (str(project_id),),
-        )
-
-
 async def confirm_surface_model(
     connection: aiomysql.Connection, project_id: UUID, model_id: int
 ) -> None:
@@ -384,7 +371,7 @@ async def _upsert_surface_model(
     processed_cloud_id: int | None,
     crs_epsg: int | None,
 ) -> int:
-    """파일 지문이 같은 옛 행이면 id 를 살려 고쳐 씀 — 화면 주소 · 노선 연결이 그대로.
+    """파일 지문이 같은 옛 행이면 id 를 살려 고쳐 씀 — 화면 주소 · 노선 연결 · 확정이 그대로(763-3).
 
     지문이 다르거나 없으면 옛 행을 지우고 새로 넣는다(새 지표면 = 노선 연결도 끊김).
     """
@@ -397,7 +384,8 @@ async def _upsert_surface_model(
             await cursor.execute(
                 """
                 UPDATE surface_models
-                SET source_file_id = %s, processed_cloud_id = %s, status = 'COMPLETE',
+                SET source_file_id = %s, processed_cloud_id = %s,
+                    status = IF(status = 'CONFIRMED', 'CONFIRMED', 'COMPLETE'),
                     crs_epsg = %s, resolution_m = %s, generation_params = %s,
                     completed_at = CURRENT_TIMESTAMP
                 WHERE id = %s

B04_PreProcess/B04_PreProcess_Router.py

@@ -24,7 +24,6 @@ from B04_PreProcess.B04_PreProcess_Engine_Extent import (
 from B04_PreProcess.B04_PreProcess_Engine_Ground import available_filters
 from B04_PreProcess.B04_PreProcess_Engine_PointPreview import ensure_point_preview, sample_points
 from B04_PreProcess.B04_PreProcess_Repository import (
-    clear_confirmed_surface_models,
     get_input_file,
     list_project_point_cloud_inputs,
     list_project_point_cloud_paths,
@@ -67,13 +66,11 @@ from common_util.common_util_auth import require_system_admin
 from common_util.common_util_http_cache import cached_file_response
 from common_util.common_util_storage import resolve_stored_project_path
 from common_util.common_util_surface_confirmation import (
+    confirmed_surface_params,
     get_surface_confirmation_params,
     surface_confirmation_defaults,
-)
-from common_util.common_util_workflow_state import (
-    fail_stage,
-    start_stage,
-    update_stage_progress,
+    terrain_changes,
+    update_contour_interval_param,
 )
 from config.config_db import get_db_pool
 
@@ -89,7 +86,10 @@ router.include_router(status_router)
 async def analyze_surface(
     project_id: UUID, request: SurfaceAnalyzeRequest
 ) -> SurfaceAnalyzeResponse | JSONResponse:
-    """LAS 구조화·지면 필터·지표면 모델 생성을 실행하고 DB에 기록한다."""
+    """LAS 구조화·지면 필터·지표면 모델 생성을 실행하고 DB에 기록한다 — 보기만(763-3 · 요구 605).
+
+    확정 · 단계 상태 · 1 단계 값(`las_selection` 포함)을 안 바꿈 · 미리 셈도 없음. 서피스를 바꾸는
+    길 = 다른 모델 [확정](전처리 체인 다시)."""
     try:
         source_filters = request.resolved_filters()
         methods = request.resolved_methods()
@@ -97,19 +97,9 @@ async def analyze_surface(
         return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
 
     pool = get_db_pool()
+    project_root: Path | None = None
     try:
-        params = {
-            "input_file_id": request.input_file_id,
-            "source_filters": source_filters,
-            "methods": methods,
-            "force": request.force,
-        }
         async with pool.acquire() as connection:
-            async with connection.cursor() as cursor:
-                await start_stage(cursor, str(project_id), 1, params)
-                await clear_confirmed_surface_models(connection, project_id)
-                await connection.commit()
-
             stored_path = await get_project_storage_relative_path(connection, project_id)
             project_root = Path(resolve_stored_project_path(stored_path))
             input_file = await get_input_file(connection, project_id, request.input_file_id)
@@ -151,19 +141,23 @@ async def analyze_surface(
                     analysis_result=result,
                     source_filters=source_filters,
                 )
-                async with connection.cursor() as cursor:
-                    await update_stage_progress(cursor, str(project_id), 1, 100)
                 await connection.commit()
+                confirmed = await confirmed_surface_params(connection, str(project_id))
             except Exception:
                 await connection.rollback()
                 raise
 
-        write_surface_progress(
-            project_root,
-            100,
-            "awaiting_confirmation",
-            "WF1 분석이 완료되었습니다. 사용할 모델을 확정하세요.",
-        )
+        if confirmed is not None:
+            write_surface_progress(
+                project_root, 100, "completed", "조합을 지었습니다 — 확정 그대로."
+            )
+        else:
+            write_surface_progress(
+                project_root,
+                100,
+                "awaiting_confirmation",
+                "WF1 분석이 완료되었습니다. 사용할 모델을 확정하세요.",
+            )
 
         return SurfaceAnalyzeResponse(
             project_id=str(project_id),
@@ -173,16 +167,14 @@ async def analyze_surface(
         )
     except LookupError as exc:
         return JSONResponse(status_code=404, content={"status": "error", "message": str(exc)})
-    except (OSError, ValueError) as exc:
-        async with pool.acquire() as connection, connection.cursor() as cursor:
-            await fail_stage(cursor, str(project_id), 1, str(exc))
-            await connection.commit()
+    except (OSError, ValueError) as exc:  # 보기만 — 단계 상태 그대로 · 진행 파일만 끝(763-3)
+        if project_root is not None:
+            write_surface_progress(project_root, 100, "failed", str(exc))
         return JSONResponse(status_code=400, content={"status": "error", "message": str(exc)})
     except Exception as exc:
         logger.exception("B04 지표면 분석 실패: project_id=%s", project_id)
-        async with pool.acquire() as connection, connection.cursor() as cursor:
-            await fail_stage(cursor, str(project_id), 1, str(exc))
-            await connection.commit()
+        if project_root is not None:
+            write_surface_progress(project_root, 100, "failed", str(exc))
         return JSONResponse(
             status_code=500,
             content={"status": "error", "message": "지표면 분석 처리 중 오류가 발생했습니다."},
@@ -214,7 +206,10 @@ async def confirm_surface(
     request: SurfaceConfirmRequest,
     _session: dict[str, Any] = Depends(require_system_admin),
 ) -> SurfaceConfirmResponse | JSONResponse:
-    """사용자가 선택한 지표면 모델을 확정하고 WF1을 완료한다."""
+    """사용자가 선택한 지표면 모델을 확정하고 WF1을 완료한다.
+
+    같은 서피스(필터 · 표현 · 평활) 다시 [확정] = 등고선 간격만 · 체인 없음. 다른 서피스 = 업로드와
+    같은 전처리 체인(B05~B09 초기화 → 자동 설계 · 추천 미리 셈 · 763-3 · 요구 605)."""
     pool = get_db_pool()
     try:
         async with pool.acquire() as connection:
@@ -247,24 +242,32 @@ async def confirm_surface(
                         ),
                     }
                 )
-                await confirm_surface_selection(
-                    connection,
-                    project_id,
-                    request.model_id,
-                    selection,
-                )
+                current = await confirmed_surface_params(connection, str(project_id))
+                same = current is not None and not terrain_changes(current, selection)
+                if same:  # 같은 서피스 — 아무것도 안 다시 셈(등고선 간격만 · 763-3)
+                    await update_contour_interval_param(
+                        connection, str(project_id), float(selection["contour_interval_m"])
+                    )
+                else:
+                    await confirm_surface_selection(
+                        connection,
+                        project_id,
+                        request.model_id,
+                        selection,
+                    )
                 await connection.commit()
             except Exception:
                 await connection.rollback()
                 raise
-        # 재확정 체인(2026-08-04 사용자 확정) — 새 지표면 기준으로 B05·B06을 백그라운드
-        # 재계산·저장한다. 사용자 저장 입력(제어점·경사 옵션·측점별 설계)은 유지·이월된다.
+        if same:
+            return SurfaceConfirmResponse(project_id=str(project_id), model_id=request.model_id)
+        # 서피스가 바뀜 — 업로드와 같은 전처리 체인 하나(B05~B09 초기화 · 763-3 · 요구 605).
         # 응답을 막지 않도록 백그라운드로 돌리고, 체인은 실패를 스스로 격리한다.
-        from B03_FileInput.B03_FileInput_Service_Chain import run_redesign_chain
+        from B03_FileInput.B03_FileInput_Service_SurfaceChange import run_surface_change_chain
 
         redesign_task = asyncio.create_task(
-            run_redesign_chain(project_id, request.model_id, dict(selection)),
-            name=f"redesign-chain-{project_id}",
+            run_surface_change_chain(project_id, request.model_id),
+            name=f"surface-change-chain-{project_id}",
         )
         redesign_task.add_done_callback(
             lambda task: task.exception()  # 체인 내부에서 이미 로깅 — 미회수 예외 경고만 방지

B05_Profile/B05_Profile_Router_Lidar.py

@@ -221,6 +221,9 @@ async def upload_route_lidar(project_id: UUID, files: list[UploadFile] = File(..
                     await connection.begin()
                     await stash_las_selection(connection, str(project_id), got["meta"]["selection"])
                     await connection.commit()
+        from B05_Profile.B05_Profile_Router_RouteInitial import schedule_precompute
+
+        schedule_precompute(project_id)  # 라이다 지형이 바뀜 — 추천 다시 셈(763-3)
         report(100, f"라이다 지형 끝 — 점 {got['meta']['point_count']:,}개")
 
     return lidar_upload.start_job(_JOBS, _TASKS, project_id, KIND, work, root / JOB, "B05")

B05_Profile/B05_Profile_Router_Replan.py

@@ -578,7 +578,7 @@ async def _save_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, Any
             if request.balanced_profile and as_initial
             else None
         ),
-        request.terrain if as_initial else None,  # 손 [저장] = 지형 그대로 · 바꿈 = 지형 칩(626-1)
+        None,  # 설계 지형 그대로 — 추천 [적용]도 선만(763-3 · 요구 605-2) · 바꿈 = 지형 칩(626-1)
         (old_line, await asyncio.to_thread(_vertices_of, working_path)),
         as_initial,
     )

common_util/common_util_workflow_state.py

@@ -527,11 +527,14 @@ async def _surface_run_settled(cursor: aiomysql.DictCursor, project_id: str) ->
     return stage in _SETTLED_SURFACE_STAGES
 
 
-async def reset_stages_after_input_change(cursor: aiomysql.DictCursor, project_id: str) -> None:
+async def reset_stages_after_input_change(
+    cursor: aiomysql.DictCursor, project_id: str, first_stage: int = 1
+) -> None:
     """입력 자료가 갈렸을 때 1단계(전처리) 이후를 처음 상태로 되돌린다.
 
     새 자료로 다시 계산해 덮어쓰므로, 옛 자료 기준의 진행 표시가 남아 있으면 사용자가 이미
     끝난 단계로 착각한다. 파일 입력(0단계)은 방금 끝났으니 건드리지 않는다.
+    `first_stage` = 2 — 서피스 바꿈 체인(전처리 결과 · 확정은 그대로 · 763-3).
     """
     await cursor.execute(
         """
@@ -542,7 +545,7 @@ async def reset_stages_after_input_change(cursor: aiomysql.DictCursor, project_i
             message = NULL,
             started_at = NULL,
             completed_at = NULL
-        WHERE project_id = %s AND stage_no >= 1
+        WHERE project_id = %s AND stage_no >= %s
         """,
-        (project_id,),
+        (project_id, first_stage),
     )

resources/tester/test_309_3_recommend_preprocess_only.py

@@ -66,6 +66,8 @@ def test_미리_셈_부르는_곳은_체인만() -> None:
         "B03_FileInput_Service_Chain.py",  # 업로드 체인 · B04 재확정 체인
         "B01_Dashboard_Router.py",  # 사용 범위 바꿈 = 예상노선 다시 씀
         "B05_Profile_Router_RouteInitial.py",  # 정의
+        "B03_FileInput_Service_SurfaceChange.py",  # 서피스 바꿈 체인 초기화 실패 길(763-3)
+        "B05_Profile_Router_Lidar.py",  # LAS 나중 올리기 끝(763-3)
     }
 
 

resources/tester/test_351_1_recommend_gaps.py

@@ -92,10 +92,9 @@ def test_두_체인_끝에서_미리_셈_예약() -> None:
     source = Path("B03_FileInput/B03_FileInput_Service_Chain.py").read_text(encoding="utf-8")
     redesign = source[source.index("async def run_redesign_chain(") :]
     upload = source[: source.index("async def run_redesign_chain(")]
-    assert "schedule_precompute(project_id)" in upload
-    # 재확정 체인 성공 끝(완료 로그 앞)
-    done = redesign.index("재확정 체인 완료")
-    assert "schedule_precompute(project_id)" in redesign[:done]
+    # 두 체인 모두 finally — 실패 · 건너뜀 길에도 예약(763-3 · 실행 시험 = test_763_3)
+    for body in (upload, redesign):
+        assert "schedule_precompute(project_id)" in body.split("    finally:", 1)[1]
     b01 = Path("B01_Dashboard/B01_Dashboard_Router.py").read_text(encoding="utf-8")
     assert "refresh_expected_route(" in b01
 

resources/tester/test_509_5_recommend_terrain.py

@@ -116,7 +116,7 @@ def test_고를_지형은_DB_모델_있을_때만(tmp_path, monkeypatch) -> None
 def test_지형_바꿈은_추천_적용만_초기값_다시_세움() -> None:
     replan = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
     save = replan.split("async def _save_route(", 1)[1]
-    assert "request.terrain if as_initial else None," in save  # 손 [저장]은 지형 안 바꿈
+    assert "request.terrain" not in save  # 추천 [적용]도 지형 안 바꿈(763-3 · 요구 605-2)
     assert save.index("await _recompute(") < save.index("refresh_initial_snapshot(project_id")
     recompute = replan.split("async def _recompute(", 1)[1].split("\nasync def ")[0]
     assert "switch_design_terrain(connection, project_id, terrain)" in recompute

resources/tester/test_763_3_surface_change.py

@@ -0,0 +1,266 @@
+"""763-3 (요구 605) B04 분석 · 조합 짓기 = 보기만 · 다른 서피스 [확정] = 업로드와 같은 전처리 체인 ·
+같은 서피스 다시 [확정] = 다시 셈 없음 · 체인 실패에도 추천 미리 셈 · B05~B09 는 서피스를 안 바꿈.
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_763_3_surface_change.py -q
+"""
+
+import asyncio
+import re
+from contextlib import asynccontextmanager
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+
+from B03_FileInput import B03_FileInput_Service_SurfaceChange as change
+from B04_PreProcess import B04_PreProcess_Router as b04
+from B04_PreProcess.B04_PreProcess_Schema import SurfaceAnalyzeRequest, SurfaceConfirmRequest
+
+ROOT = Path(__file__).resolve().parents[2]
+CONFIRMED = {
+    "source_filter": "classification",
+    "method": "implicit",
+    "smooth": False,
+    "contour_interval_m": 1.0,
+}
+
+
+class _Cursor:
+    def __init__(self, log):
+        self.log = log
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_a):
+        return False
+
+    async def execute(self, sql, *_a):
+        self.log.append(" ".join(str(sql).split()))
+
+    async def fetchone(self):
+        return None
+
+
+class _Connection:
+    def __init__(self, log):
+        self.log = log
+
+    def cursor(self, *_a):
+        return _Cursor(self.log)
+
+    async def begin(self):
+        self.log.append("BEGIN")
+
+    async def commit(self):
+        self.log.append("COMMIT")
+
+    async def rollback(self):
+        self.log.append("ROLLBACK")
+
+
+class _Pool:
+    def __init__(self, log):
+        self.log = log
+
+    @asynccontextmanager
+    async def acquire(self):
+        yield _Connection(self.log)
+
+
+def _async(value):
+    async def inner(*_a, **_k):
+        return value
+
+    return inner
+
+
+@pytest.fixture
+def b04_world(tmp_path, monkeypatch):
+    log: list[str] = []
+    calls: dict[str, list] = {"progress": [], "chain": [], "confirm": [], "contour": []}
+    las = tmp_path / "raw.las"
+    las.write_bytes(b"")
+    monkeypatch.setattr(b04, "get_db_pool", lambda: _Pool(log))
+    monkeypatch.setattr(b04, "get_project_storage_relative_path", _async("x"))
+    monkeypatch.setattr(b04, "resolve_stored_project_path", lambda _p: str(tmp_path))
+    monkeypatch.setattr(b04, "get_input_file", _async({"raw_file_path": "raw.las"}))
+    monkeypatch.setattr(b04, "list_project_point_cloud_paths", _async([las]))
+    monkeypatch.setattr(
+        b04,
+        "run_surface_analysis",
+        lambda *_a, **_k: {"ground_summary": {}, "manifest": {"status": "ok"}},
+    )
+    monkeypatch.setattr(b04, "save_surface_analysis_to_db", _async([1, 2]))
+    monkeypatch.setattr(b04, "confirmed_surface_params", _async(dict(CONFIRMED)))
+    monkeypatch.setattr(
+        b04, "write_surface_progress", lambda _r, p, stage, _m: calls["progress"].append(stage)
+    )
+    monkeypatch.setattr(
+        b04,
+        "list_surface_models",
+        _async(
+            [
+                {
+                    "id": 7,
+                    "model_type": "implicit",
+                    "generation_params": {"source_filter": "classification"},
+                },
+                {"id": 8, "model_type": "implicit", "generation_params": {"source_filter": "csf"}},
+            ]
+        ),
+    )
+
+    async def confirm(_c, _pid, model_id, selection):
+        calls["confirm"].append((model_id, selection["source_filter"]))
+
+    async def contour(_c, _pid, value):
+        calls["contour"].append(value)
+
+    async def chain(pid, model_id):
+        calls["chain"].append(model_id)
+
+    monkeypatch.setattr(b04, "confirm_surface_selection", confirm)
+    monkeypatch.setattr(b04, "update_contour_interval_param", contour)
+    monkeypatch.setattr(change, "run_surface_change_chain", chain)
+    calls["log"] = log
+    return calls
+
+
+def test_분석은_보기만_단계_확정_그대로(b04_world) -> None:
+    request = SurfaceAnalyzeRequest(input_file_id=3, source_filters=["csf"], methods=["implicit"])
+    answer = asyncio.run(b04.analyze_surface(uuid4(), request))
+    assert answer.surface_model_ids == [1, 2]
+    sql = " | ".join(b04_world["log"])
+    assert "project_workflow_stages" not in sql and "surface_models" not in sql
+    assert b04_world["progress"][-1] == "completed"  # 확정 그대로 — 「확정하세요」 아님
+    source = (ROOT / "B04_PreProcess/B04_PreProcess_Router.py").read_text(encoding="utf-8")
+    body = source.split("async def analyze_surface(", 1)[1].split("\n@router.", 1)[0]
+    for name in ("start_stage", "fail_stage", "update_stage_progress", "schedule_precompute"):
+        assert name not in body, name
+
+
+def test_조합_다시_적어도_확정_행_그대로() -> None:
+    source = (ROOT / "B04_PreProcess/B04_PreProcess_Repository.py").read_text(encoding="utf-8")
+    assert "status = IF(status = 'CONFIRMED', 'CONFIRMED', 'COMPLETE')" in source
+    assert "clear_confirmed_surface_models" not in source
+
+
+def _confirm(model_id: int, contour: float | None = None):
+    request = SurfaceConfirmRequest(model_id=model_id, smooth=False, contour_interval_m=contour)
+
+    async def run():
+        answer = await b04.confirm_surface(uuid4(), request, {})
+        await asyncio.sleep(0)  # 백그라운드 체인 한 번 돌림
+        return answer
+
+    return asyncio.run(run())
+
+
+def test_같은_서피스_다시_확정은_다시_셈_없음(b04_world) -> None:
+    _confirm(7, 2.0)
+    assert b04_world["confirm"] == [] and b04_world["chain"] == []
+    assert b04_world["contour"] == [2.0]
+
+
+def test_다른_서피스_확정은_전처리_체인(b04_world) -> None:
+    _confirm(8)
+    assert b04_world["confirm"] == [(8, "csf")]
+    assert b04_world["chain"] == [8]
+
+
+def test_서피스_바꿈_체인_차례(tmp_path, monkeypatch) -> None:
+    import B03_FileInput.B03_FileInput_Repository as repo
+    import B03_FileInput.B03_FileInput_Service_Chain as chain
+    import B05_Profile.B05_Profile_Router_RouteInitial as router
+    import common_util.common_util_initial_snapshot as snapshot
+    import common_util.common_util_project_reset as reset
+    import common_util.common_util_storage as storage
+    import common_util.common_util_workflow_state as state
+    import config.config_db as db
+
+    log: list = []
+
+    async def purge(_c, _pid, _root, keep_preprocess=False):
+        log.append(("purge", keep_preprocess))
+
+    async def stages(_cursor, _pid, first_stage=1):
+        log.append(("stages", first_stage))
+
+    async def auto(_pid, surface_model_id=None):
+        log.append(("auto", surface_model_id))
+
+    monkeypatch.setattr(db, "get_db_pool", lambda: _Pool([]))
+    monkeypatch.setattr(repo, "get_project_storage_relative_path", _async("x"))
+    monkeypatch.setattr(storage, "resolve_stored_project_path", lambda _p: str(tmp_path))
+    monkeypatch.setattr(snapshot, "discard_initial_snapshot", lambda _r: log.append("snapshot"))
+    monkeypatch.setattr(reset, "purge_project_outputs", purge)
+    monkeypatch.setattr(state, "reset_stages_after_input_change", stages)
+    monkeypatch.setattr(chain, "run_auto_design_chain", auto)
+    monkeypatch.setattr(router, "schedule_precompute", lambda pid: log.append("precompute"))
+    asyncio.run(change.run_surface_change_chain(uuid4(), 8))
+    assert log == ["snapshot", ("purge", True), ("stages", 2), ("auto", 8)]
+
+    async def broken(*_a, **_k):
+        raise RuntimeError("db")
+
+    log.clear()
+    monkeypatch.setattr(reset, "purge_project_outputs", broken)
+    asyncio.run(change.run_surface_change_chain(uuid4(), 8))
+    assert log == ["snapshot", "precompute"]  # 초기화 실패 = 체인 없음 · 추천만 새 서피스로
+
+
+@pytest.mark.parametrize("name", ["run_auto_design_chain", "run_redesign_chain"])
+def test_체인_실패_길에도_미리_셈(monkeypatch, name) -> None:
+    import B03_FileInput.B03_FileInput_Service_Chain as chain
+    import B05_Profile.B05_Profile_Router_RouteInitial as router
+    import config.config_db as db
+
+    class _Broken:
+        def acquire(self):
+            raise RuntimeError("db down")
+
+    scheduled: list = []
+    monkeypatch.setattr(db, "get_db_pool", lambda: _Broken())
+    monkeypatch.setattr(router, "schedule_precompute", scheduled.append)
+    project = uuid4()
+    args = (project,) if name == "run_auto_design_chain" else (project, 1, {})
+    asyncio.run(getattr(chain, name)(*args))
+    assert scheduled == [project]
+
+
+def test_LAS_나중_올리기_끝에_미리_셈() -> None:
+    source = (ROOT / "B05_Profile/B05_Profile_Router_Lidar.py").read_text(encoding="utf-8")
+    work = source.split("async def work(", 1)[1].split("return lidar_upload.start_job", 1)[0]
+    assert "schedule_precompute(project_id)" in work
+
+
+#: B05~B09 에서 확정 서피스를 바꿀 수 있는 부름 — 남은 곳은 까닭 있는 둘만(요구 605-2)
+_SURFACE_WRITES = re.compile(
+    r"\b(confirm_surface_selection|confirm_surface_model|merge_surface_confirmation_params|"
+    r"switch_design_terrain|stash_las_selection|run_surface_change_chain)\("
+)
+_ALLOWED = {
+    # 지형 칩(626-1) — 사용자 답 기다림(763 ①) · 추천 [적용]은 지형을 안 넘김(아래 시험)
+    "B05_Profile/B05_Profile_Router_Replan.py",
+    # 라이다 확정값만 적어 둠 · 설계 지형 그대로(637-2)
+    "B05_Profile/B05_Profile_Router_Lidar.py",
+}
+
+
+def test_B05_B09_확정은_서피스를_안_바꿈() -> None:
+    found = set()
+    for folder in (
+        "B05_Profile",
+        "B06_Section",
+        "B07_DesignDetail",
+        "B08_Quantity",
+        "B09_Estimation",
+    ):
+        for path in (ROOT / folder).glob("*.py"):
+            if _SURFACE_WRITES.search(path.read_text(encoding="utf-8")):
+                found.add(path.relative_to(ROOT).as_posix())
+    assert found <= _ALLOWED, found - _ALLOWED
+    replan = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
+    save = replan.split("async def _save_route(", 1)[1].split("\nasync def ", 1)[0]
+    assert "request.terrain" not in save  # 추천 [적용] = 선만 · 설계 지형 그대로(605-2)