feat(배수유역): 347-2 「지금 노선으로 셈」 서버 길 · 347-3 관 자리 후보 · [확정] 다시 셈

- POST drainage/live — 몸통 = 차례 번호 + 노선 초안(RouteDraft · 노선 [확인]과 같은 폴리라인 규칙) + 관 목록 · 서버 단독 · 작업본(00_watershed_response · 격자 배열 · 관 지점)에 안 씀 · 세부 유역까지 메모리 격자로 한 요청
- 프로젝트별 마지막 차례 · 1차 영역 뒤 · 확장 회차마다 · 세부 유역 앞에서 바뀌었으면 그만둠(superseded · AnalysisStopped)
- build_detail 에 메모리 routing · 상류선 · 관 자리 후보 suggest_pipe_positions(간격 보충 규칙 · 흐름 세기 + 오목 · 기존 관에서 먼 곳 · 정수 m) · 응답 pipe_candidates
- PUT pipe-points ?recompute=true([확정]) = 낡았으면 배수유역부터 다시 셈 · [저장]은 낡음 달고 관만
- 시험 test_347_2_drainage_live(늦은 차례 그만둠 둘 · 엔진 회차 그만둠 · 메모리 격자 · 후보 · 확정 다시 셈) · 검증_sub3 실측 저장 노선 16.9 s · 초안 노선 11.5 s

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XWh5wNNycMkbwMpACz1rte

B04_PreProcess/B04_PreProcess_Engine_Watershed_Analyze.py

@@ -23,6 +23,7 @@ from __future__ import annotations
 import logging
 import math
 import time
+from collections.abc import Callable
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -31,7 +32,10 @@ import numpy as np
 from shapely.geometry import LineString
 
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Descent import ContourDescent
-from B04_PreProcess.B04_PreProcess_Engine_Watershed_Expand import expand_by_red_boundary
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Expand import (
+    AnalysisStopped,
+    expand_by_red_boundary,
+)
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Flow import (
     FlowClassification,
     RoadRaster,
@@ -154,6 +158,7 @@ def preview_stages(
     contour_features: list[dict[str, Any]],
     stream_features: list[dict[str, Any]],
     descent_store: tuple[Path, str] | None = None,
+    should_stop: Callable[[], bool] | None = None,
 ) -> StagePreview | None:
     """지금까지 구현·검증된 단계를 순서대로 돌려 결과를 모은다.
 
@@ -176,6 +181,8 @@ def preview_stages(
     marks.append(("1차 영역(resolve_primary_region)", time.perf_counter()))
     if region is None:
         return None
+    if should_stop is not None and should_stop():
+        raise AnalysisStopped
 
     started = time.perf_counter()
     floor = route_elevation_floor([vertex.z for vertex in vertices])
@@ -187,6 +194,7 @@ def preview_stages(
         region.split.upstream,
         floor,
         descent_store=descent_store,
+        should_stop=should_stop,
     )
     if expansion is None:
         logger.warning("배수유역: 등고선 하강 방향을 세우지 못해 흐름 판정을 건너뜁니다.")

B04_PreProcess/B04_PreProcess_Engine_Watershed_Expand.py

@@ -21,6 +21,7 @@ from __future__ import annotations
 
 import logging
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
@@ -58,6 +59,10 @@ logger = logging.getLogger(__name__)
 DESCENT_SPARE_M = 400.0
 
 
+class AnalysisStopped(Exception):
+    """새 차례가 와서 계산을 그만둠(「지금 노선으로 셈」 · 347-2) — 단계 사이에서만 확인."""
+
+
 @dataclass
 class GridAnalysis:
     """한 회차 분석 결과 — 격자·해석 영역·방향·색까지 한 묶음."""
@@ -142,6 +147,7 @@ def expand_by_red_boundary(
     band_m: float = DRAINAGE_RED_EXPAND_BAND_M,
     max_rounds: int = DRAINAGE_RED_EXPAND_MAX_ROUNDS,
     descent_store: tuple[Path, str] | None = None,
+    should_stop: Callable[[], bool] | None = None,
 ) -> RedExpansion | None:
     """최외곽 적색 셀 주변으로 넓히며, 새로 추가한 셀에 적색이 없을 때까지 반복한다."""
     _expand_started = time.perf_counter()
@@ -194,6 +200,8 @@ def expand_by_red_boundary(
     rounds = 0
     closed = False
     for attempt in range(max_rounds):
+        if should_stop is not None and should_stop():
+            raise AnalysisStopped
         _round_started = time.perf_counter()
         current = analysis.spec
         reaches = analysis.flow.reaches_road.reshape(current.n_rows, current.n_cols)

B04_PreProcess/B04_PreProcess_Router_Basins.py

@@ -73,7 +73,11 @@ STALE_MESSAGE = (
 
 
 def _build(
-    stored_path: str, context: DrainageContext, points: list[PipePoint] | None
+    stored_path: str,
+    context: DrainageContext,
+    points: list[PipePoint] | None,
+    routing: Any = None,
+    upstream_lonlat: list | None = None,
 ) -> tuple[DrainageDetail | None, list[PipePoint]]:
     """세부유역을 계산하고, 그 결과에 쓰인 관 목록을 함께 돌려준다.
 
@@ -83,7 +87,7 @@ def _build(
     """
     directory = drainage_dir(stored_path)
     chainages = [point.chainage_m for point in points] if points is not None else None
-    detail = build_detail(directory, context.vertices, chainages)
+    detail = build_detail(directory, context.vertices, chainages, routing, upstream_lonlat)
     if detail is None:
         return None, []
     if points is not None:
@@ -377,6 +381,11 @@ def _raw_payload(
             for basin in detail.basins
         ],
         "pipe_count": len(points),
+        # 관 자리 후보(347-3) — 제안만 · [적용] 때 화면이 관 초안에 더함.
+        "pipe_candidates": [
+            {"chainage_m": round(pipe.chainage_m, 2), "lonlat": list(to_lonlat(pipe.x, pipe.y))}
+            for pipe in detail.pipe_candidates
+        ],
     }
 
 
@@ -479,6 +488,13 @@ async def _resolve(
     return context, detail, resolved, saved, False
 
 
+async def _analysis_stale(project_id: UUID) -> bool:
+    context, _ = await load_drainage_context(project_id)
+    if context is None:
+        return False
+    return analysis_route_signature(context.stored_path) != route_signature(context.vertices)
+
+
 @router.get("/{project_id}/drainage/pipe-points", response_model=None)
 async def get_pipe_points(project_id: UUID) -> dict[str, Any] | JSONResponse:
     """저장된 관 지점과 그 세부유역을 돌려준다. 저장분이 없으면 자동 생성한 목록."""
@@ -529,8 +545,17 @@ async def put_pipe_points(
     project_id: UUID,
     payload: dict[str, Any] | None = None,
     session: dict[str, Any] | None = Depends(get_optional_session),  # 가져온이(142-2)
+    recompute: bool = False,
 ) -> dict[str, Any] | JSONResponse:
     """관 지점을 정본으로 확정하고 세부유역 산출물까지 함께 남긴다(모델 확정 시점)."""
+    # [확정](recompute) = 낡았으면 지금 노선으로 배수유역부터 다시 셈(정본 · 작업본에 씀 · 347-2) ·
+    # [저장] = 낡음 표시 달고 관만 저장.
+    if recompute and await _analysis_stale(project_id):
+        from B04_PreProcess.B04_PreProcess_Router_Watershed import get_primary_region
+
+        region = await get_primary_region(project_id, refresh=True)
+        if isinstance(region, JSONResponse):
+            return region
     resolved = await _resolve(project_id, payload, use_stored=True, fresh=True)
     if isinstance(resolved, JSONResponse):
         return resolved

B04_PreProcess/B04_PreProcess_Router_Watershed.py

@@ -569,3 +569,7 @@ async def get_primary_region(
     _log_steps("배수유역 분석 내부", marks)
     _save_response(prepared["stored_path"], payload)
     return payload
+
+
+# 「지금 노선으로 셈」(347-2) — 같은 router 에 붙임(그 파일이 이 router 를 읽음).
+from B04_PreProcess import B04_PreProcess_Router_Watershed_Live  # noqa: E402, F401

B04_PreProcess/B04_PreProcess_Router_Watershed_Live.py

@@ -0,0 +1,152 @@
+"""「지금 노선으로 셈」(347-2) — 노선 초안 + 관 자리로 배수유역 · 세부 유역을 한 요청에 셈.
+
+서버 단독(③) · **작업본에 안 씀** — `00_watershed_response.json` · 격자 배열 · 관 지점은 그대로이고
+답으로만 돌려준다(파생 저장본 = 흐름 방향 · 도엽 좌표만 · 347-1). 세부 유역 나눔까지 같은 요청에서
+메모리 격자로 해 격자를 화면에 넘기지 않는다.
+
+프로젝트마다 「마지막 차례」를 적어 두고 단계 사이(1차 영역 뒤 · 확장 회차마다 · 세부 유역 앞)에서
+차례가 바뀌었으면 그만둔다(`superseded`). 화면은 늦은 답을 차례 · 노선 지문으로 한 번 더 거른다.
+라우터는 `B04_PreProcess_Router_Watershed.router` 에 붙는다(그 파일 끝에서 읽음).
+"""
+
+from __future__ import annotations
+
+import asyncio
+import math
+from dataclasses import replace
+from typing import Any
+from uuid import UUID
+
+from fastapi.responses import JSONResponse
+
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import preview_stages
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Expand import AnalysisStopped
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import drainage_dir
+from B04_PreProcess.B04_PreProcess_Router_Basins import _build, _payload
+from B04_PreProcess.B04_PreProcess_Router_Watershed import _prepare, router
+from common_util.common_util_drainage_context import _open_sampler, load_drainage_context
+from common_util.common_util_drainage_detail import RoadRouting
+from common_util.common_util_drainage_pipes import parse_pipe_points
+from common_util.common_util_route_geometry import RouteVertex, build_route_vertices
+from common_util.common_util_route_polyline import build_planned_polyline
+from common_util.common_util_route_profile import resolve_route_profile
+from config.config_system import DRAINAGE_ARROW_SPACING_M
+
+# 프로젝트 → 마지막으로 받은 차례 번호. ponytail: 서버 프로세스 하나 전제(여럿이면 DB · 파일로).
+_LATEST_GENERATION: dict[str, int] = {}
+SUPERSEDED = {"status": "superseded"}
+
+
+def draft_vertices(draft: dict[str, Any], min_radius_m: float) -> list[RouteVertex]:
+    """노선 초안(RouteDraft) → 그린 노선 정점(z 없음) — 노선 [확인]과 같은 폴리라인 규칙.
+
+    ponytail: 붙든 곡선 길이(curveLock · curveArc) · 측점 스냅은 안 봄 — 곡선 중점이 조금 다를 뿐
+    배수유역 결과엔 셀 몇 개 차이. 필요하면 노선 [확인] 길과 같이 넘김.
+    """
+    nodes = [(float(x), float(y)) for x, y in draft["nodes"]]
+    flags = [bool(flag) for flag in draft.get("curveOn") or [True] * len(nodes)]
+    radii = [None if r is None else float(r) for r in draft.get("curveRadius") or []]
+    polyline = build_planned_polyline(
+        nodes, min_radius_m=min_radius_m, simplify=False, curve_flags=flags, radii=radii or None
+    )
+    return build_route_vertices([{"x": x, "y": y} for x, y in polyline.vertices])
+
+
+def live_routing(preview: Any, route_xy: list[tuple[float, float]], to_lonlat: Any) -> RoadRouting:
+    """분석 결과(메모리) → 세부 유역 나눔 입력 — 파일 길(`_write_road_routing` → 읽기)과 같은 값."""
+    routing, road = preview.routing, preview.road
+    return RoadRouting(
+        spec=preview.spec,
+        road_slot=routing.road_slot,
+        path_length=routing.path_length,
+        elevation=preview.terrain.elevation.reshape(-1),
+        road_cell_index=road.cell_index,
+        road_chainage=road.chainage,
+        strength=routing.strength,
+        route_lonlat=[list(to_lonlat(x, y)) for x, y in route_xy],
+        basin_lonlat=[list(to_lonlat(x, y)) for x, y in preview.basin_boundary_xy],
+        base_pipes=list(preview.pipes),
+        flow_arrows=[
+            [round(x, 2), round(y, 2), round(math.degrees(angle), 1), reaches, cells]
+            for x, y, angle, reaches, cells in preview.flow_arrows
+        ],
+        arrow_spacing_m=DRAINAGE_ARROW_SPACING_M,
+    )
+
+
+@router.post("/{project_id}/drainage/live", response_model=None)
+async def post_drainage_live(
+    project_id: UUID, payload: dict[str, Any] | None = None
+) -> dict[str, Any] | JSONResponse:
+    """몸통 = {generation, route: RouteDraft | null(저장 노선), points: 관 목록 | null(자동)}."""
+    body = payload or {}
+    generation = int(body.get("generation") or 0)
+    key = str(project_id)
+    _LATEST_GENERATION[key] = generation
+
+    def superseded() -> bool:
+        return _LATEST_GENERATION.get(key) != generation
+
+    try:
+        raw = body.get("points")
+        points = None if raw is None else parse_pipe_points(raw)
+    except ValueError as error:
+        return JSONResponse(status_code=400, content={"status": "error", "message": str(error)})
+    prepared = await _prepare(project_id)
+    if isinstance(prepared, JSONResponse):
+        return prepared
+    context, reason = await load_drainage_context(project_id)
+    if context is None:
+        return JSONResponse(status_code=404, content={"status": "error", "message": reason})
+
+    vertices = context.vertices
+    draft = body.get("route")
+    if draft and len(draft.get("nodes") or []) >= 2:
+        from B05_Profile.B05_Profile_Router_Replan import _plan_criteria
+
+        radius_m, *_ = await _plan_criteria(project_id)
+        drawn = await asyncio.to_thread(draft_vertices, draft, radius_m)
+        sampler = await asyncio.to_thread(
+            _open_sampler, context.project_root, context.surface_params
+        )
+        vertices, _ = await asyncio.to_thread(resolve_route_profile, drawn, sampler=sampler)
+    if superseded():
+        return SUPERSEDED
+
+    try:
+        preview = await asyncio.to_thread(
+            preview_stages,
+            vertices,
+            prepared["contours"],
+            prepared["streams"],
+            (drainage_dir(context.stored_path), prepared["contour_key"]),
+            superseded,
+        )
+    except AnalysisStopped:
+        return SUPERSEDED
+    if preview is None or preview.routing is None:
+        return JSONResponse(
+            status_code=400,
+            content={"status": "error", "message": "지금 노선으로 배수유역을 정하지 못했습니다."},
+        )
+    if superseded():
+        return SUPERSEDED
+
+    live = replace(context, vertices=vertices)
+    to_lonlat = context.to_lonlat
+    routing = live_routing(preview, [(v.x, v.y) for v in vertices], to_lonlat)
+    upstream = [
+        [list(to_lonlat(x, y)) for x, y in line.coords] for line in preview.region.split.upstream
+    ]
+    detail, carried = await asyncio.to_thread(
+        _build, context.stored_path, live, points, routing, upstream
+    )
+    if detail is None:
+        return JSONResponse(
+            status_code=400,
+            content={"status": "error", "message": "세부 유역을 나누지 못했습니다."},
+        )
+    result = _payload(project_id, live, detail, carried, False)
+    result["generation"] = generation
+    result["live"] = True
+    return result

common_util/common_util_drainage_detail.py

@@ -85,6 +85,8 @@ class DrainageDetail:
     strength_profile: list[list[float]] = field(default_factory=list)
     # 유입 집중점 — [누가거리, 유입면적, 구역번호, 구역 내 순위]. 관 자리를 판단하는 근거.
     inflow_hotspots: list[list[float]] = field(default_factory=list)
+    # 관 자리 후보(347-3) — 관 사이가 최대 간격을 넘는 구간의 흐름 세기 + 오목 자리 · 제안만.
+    pipe_candidates: list[StructureCandidate] = field(default_factory=list)
 
 
 @dataclass
@@ -163,18 +165,23 @@ def build_detail(
     directory: Path,
     vertices: list[RouteVertex],
     confirmed_chainages: list[float] | None = None,
+    routing: RoadRouting | None = None,
+    upstream_lonlat: list[list[list[float]]] | None = None,
 ) -> DrainageDetail | None:
     """B04 분석 결과를 읽어 관을 보충하고 세부유역을 나눈다.
 
     `confirmed_chainages`를 주면 그 위치를 관으로 확정하고(사용자 편집 · 빈 목록 = 관 0개), None 이면
     B04의 기본 관에 최대 간격 규칙으로 최소 개수만 보충한다. 어느 쪽이든 격자 해석은 하지 않는다.
+    `routing` · `upstream_lonlat` 을 주면 파일 대신 그것(「지금 노선으로 셈」 메모리 결과 · 347-2).
     """
-    routing = read_road_routing(directory)
+    routing = routing or read_road_routing(directory)
     if routing is None or len(vertices) < 2:
         return None
 
+    candidates: list[StructureCandidate] = []
     if confirmed_chainages is not None:
         pipes = pipes_from_chainages(vertices, confirmed_chainages)
+        candidates = suggest_pipe_positions(vertices, pipes, routing.strength_curve)
     else:
         # 저장분의 기본 관은 누가거리만 신뢰한다 — 좌표는 현재 노선 위로 다시 찍는다.
         base = [
@@ -196,7 +203,9 @@ def build_detail(
         grid_cell_m=routing.spec.cell_m,
         flow_arrows=routing.flow_arrows,
         arrow_spacing_m=routing.arrow_spacing_m,
-        upstream_lonlat=read_upstream_lines(directory),
+        upstream_lonlat=read_upstream_lines(directory)
+        if upstream_lonlat is None
+        else upstream_lonlat,
         strength_profile=build_strength_profile(routing),
         inflow_hotspots=[
             [chainage, area, float(zone), float(rank)]
@@ -206,6 +215,7 @@ def build_detail(
                 vertices[-1].chainage_m,
             )
         ],
+        pipe_candidates=candidates,
     )
     if not pipes:
         return detail
@@ -462,6 +472,24 @@ def _best_position(
     return float(positions[int(np.argmax(score))])
 
 
+def suggest_pipe_positions(
+    vertices: list[RouteVertex],
+    pipes: list[StructureCandidate],
+    strength_curve: np.ndarray,
+) -> list[StructureCandidate]:
+    """관 자리 후보(347-3) — 지금 관 사이가 최대 간격을 넘는 구간만 간격 보충 규칙으로 고른 자리.
+
+    `_best_position`(흐름 세기 + 오목 · 성토 덜 선호) · 정수 m · 관 목록엔 안 넣음(제안).
+    """
+    taken = [pipe.chainage_m for pipe in pipes]
+    filled = snap_candidates(place_pipes(vertices, pipes, strength_curve), vertices)
+    return [
+        pipe
+        for pipe in filled
+        if pipe.reason == "spacing" and all(abs(pipe.chainage_m - other) > 0.5 for other in taken)
+    ]
+
+
 def pipes_from_chainages(
     vertices: list[RouteVertex], chainages: list[float]
 ) -> list[StructureCandidate]:

resources/tester/test_347_2_drainage_live.py

@@ -0,0 +1,169 @@
+"""347-2 「지금 노선으로 셈」 서버 길 · 347-3 관 자리 후보.
+
+- 늦은 차례: 계산 중 새 차례가 오면 단계 사이에서 그만둠(superseded) · 작업본에 안 씀.
+- 메모리 격자로 세부 유역(파일 안 읽음) · 노선 초안 → 그린 노선.
+- 관 자리 후보: 최대 간격(100 m)을 넘는 구간에만 · 기존 관 자리엔 안 냄 · 관 목록엔 안 넣음.
+- [확정](recompute) = 낡았으면 배수유역부터 다시 셈 · [저장] = 안 셈.
+"""
+
+import asyncio
+import math
+from pathlib import Path
+from types import SimpleNamespace
+from uuid import uuid4
+
+import numpy as np
+import pytest
+from shapely.geometry import LineString
+
+from B04_PreProcess import B04_PreProcess_Engine_Watershed_Expand as expand
+from B04_PreProcess import B04_PreProcess_Router_Basins as basins
+from B04_PreProcess import B04_PreProcess_Router_Watershed_Live as live
+from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import GridSpec
+from common_util.common_util_drainage_context import DrainageContext
+from common_util.common_util_drainage_detail import (
+    RoadRouting,
+    build_detail,
+    pipes_from_chainages,
+    suggest_pipe_positions,
+)
+from common_util.common_util_route_geometry import RouteVertex
+
+VERTICES = [RouteVertex(0.0, 0.0, 100.0, 0.0), RouteVertex(300.0, 0.0, 90.0, 300.0)]
+
+
+def test_draft_vertices_follow_nodes():
+    draft = {"nodes": [[0, 0], [100, 0], [100, 100]], "curveOn": [False, False, False]}
+    vertices = live.draft_vertices(draft, 30.0)
+    assert vertices[0].x == 0 and vertices[-1].y == 100
+    assert math.isclose(vertices[-1].chainage_m, 200.0)
+
+
+def test_candidates_only_in_long_gaps():
+    strength = np.zeros(301)
+    strength[150] = 50.0  # 흐름이 모이는 자리
+    pipes = pipes_from_chainages(VERTICES, [0.0, 300.0])
+    picked = suggest_pipe_positions(VERTICES, pipes, strength)
+    assert picked and all(0 < pipe.chainage_m < 300 for pipe in picked)
+    assert all(pipe.chainage_m == round(pipe.chainage_m) for pipe in picked)  # 정수 m
+    dense = pipes_from_chainages(VERTICES, [0.0, 90.0, 180.0, 270.0, 300.0])
+    assert suggest_pipe_positions(VERTICES, dense, strength) == []
+
+
+def _routing() -> RoadRouting:
+    spec = GridSpec(x_min=-5.0, y_max=5.0, cell_m=1.0, n_rows=10, n_cols=310)
+    road_index = np.arange(5 * 310 + 5, 5 * 310 + 305, dtype=np.int32)
+    road_slot = np.full(spec.size, -1, dtype=np.int32)
+    road_slot[road_index] = np.arange(road_index.size)
+    return RoadRouting(
+        spec=spec,
+        road_slot=road_slot,
+        path_length=np.zeros(spec.size, np.float32),
+        elevation=np.full(spec.size, 95.0, np.float32),
+        road_cell_index=road_index,
+        road_chainage=np.arange(road_index.size, dtype=np.float64),
+        strength=np.ones(road_index.size, np.int64),
+    )
+
+
+def test_build_detail_uses_memory_routing(tmp_path):
+    missing = tmp_path / "없는 폴더"  # 파일을 읽으면 None 이 됨
+    detail = build_detail(missing, VERTICES, [0.0, 300.0], _routing(), [])
+    assert detail is not None and detail.pipe_candidates
+    assert build_detail(missing, VERTICES, [0.0, 300.0]) is None
+
+
+def _patch_live(monkeypatch, preview_fn):
+    async def prepare(project_id):
+        return {"contours": [], "streams": [], "contour_key": "k"}
+
+    async def load(project_id):
+        return DrainageContext(stored_path="x", project_root=Path("."), vertices=VERTICES), ""
+
+    monkeypatch.setattr(live, "_prepare", prepare)
+    monkeypatch.setattr(live, "load_drainage_context", load)
+    monkeypatch.setattr(live, "drainage_dir", lambda stored: Path("."))
+    monkeypatch.setattr(live, "preview_stages", preview_fn)
+    calls = SimpleNamespace(build=0)
+
+    def build(*args):
+        calls.build += 1
+        return None, []
+
+    monkeypatch.setattr(live, "_build", build)
+    return calls
+
+
+def test_newer_generation_stops_between_stages(monkeypatch):
+    pid = uuid4()
+
+    def preview(vertices, contours, streams, store, should_stop):
+        live._LATEST_GENERATION[str(pid)] = 8  # 계산 중 새 조작 → 새 요청이 차례를 올림
+        if should_stop():
+            raise expand.AnalysisStopped
+
+    calls = _patch_live(monkeypatch, preview)
+    result = asyncio.run(live.post_drainage_live(pid, {"generation": 7, "points": []}))
+    assert result == live.SUPERSEDED and calls.build == 0
+
+
+def test_newer_generation_after_preview_skips_split(monkeypatch):
+    pid = uuid4()
+
+    def preview(vertices, contours, streams, store, should_stop):
+        assert not should_stop()
+        live._LATEST_GENERATION[str(pid)] = 3
+        return SimpleNamespace(routing=object())
+
+    calls = _patch_live(monkeypatch, preview)
+    result = asyncio.run(live.post_drainage_live(pid, {"generation": 2}))
+    assert result == live.SUPERSEDED and calls.build == 0
+
+
+def test_engine_stops_in_expansion_round():
+    contours = [
+        {
+            "geometry": {
+                "type": "LineString",
+                "coordinates": [
+                    [150 + r * math.cos(t), -150 + r * math.sin(t)]
+                    for t in np.linspace(0, 2 * math.pi, 121)
+                ],
+            },
+            "properties": {"elevation": 300 - r},
+        }
+        for r in range(10, 200, 10)
+    ]
+    spec = GridSpec(x_min=100.0, y_max=-100.0, cell_m=1.0, n_rows=100, n_cols=100)
+    domain = np.zeros((100, 100), bool)
+    domain[20:80, 20:80] = True
+    route = LineString([(110.0, -150.0), (190.0, -150.0)])
+    with pytest.raises(expand.AnalysisStopped):
+        expand.expand_by_red_boundary(
+            spec, domain, contours, route, [], None, should_stop=lambda: True
+        )
+
+
+@pytest.mark.parametrize(
+    "recompute, stale, expected", [(True, True, 1), (True, False, 0), (False, True, 0)]
+)
+def test_confirm_recomputes_only_when_stale(monkeypatch, recompute, stale, expected):
+    import B04_PreProcess.B04_PreProcess_Router_Watershed as watershed
+
+    calls = []
+
+    async def is_stale(project_id):
+        return stale
+
+    async def region(project_id, refresh=False):
+        calls.append(refresh)
+        return {"status": "success"}
+
+    async def resolve(*args, **kwargs):
+        return basins.JSONResponse(status_code=404, content={})
+
+    monkeypatch.setattr(basins, "_analysis_stale", is_stale)
+    monkeypatch.setattr(watershed, "get_primary_region", region)
+    monkeypatch.setattr(basins, "_resolve", resolve)
+    asyncio.run(basins.put_pipe_points(uuid4(), {"points": []}, None, recompute))
+    assert len(calls) == expected and all(calls)