Merge remote-tracking branch 'origin/dev' into sub_laptop_8

B04_PreProcess/B04_PreProcess_Engine.py

@@ -22,6 +22,8 @@ from B04_PreProcess.B04_PreProcess_Engine_Ground import (
 from B04_PreProcess.B04_PreProcess_Engine_Pipeline import build_all_terrain_models
 from B04_PreProcess.B04_PreProcess_Engine_Structurize import structurize_las
 from common_util.common_util_atomic import atomic_write_npz
+from common_util.common_util_derived_store import file_stamp
+from common_util.common_util_derived_store import key_matches as _key_matches
 from common_util.common_util_json import atomic_write_json
 from config.config_system import (
     SHEET_SURFACE_AUTO_METHODS,
@@ -128,23 +130,6 @@ def _structured_stats(structured: Any, structured_path: Path) -> tuple[dict[str,
     return stats, xyz
 
 
-def file_stamp(path: Path | None) -> str | None:
-    """결과 파일 지문 = 수정시각 ns : 크기. 없으면 None."""
-    try:
-        stat = path.stat() if path is not None else None
-    except OSError:
-        return None
-    return f"{stat.st_mtime_ns}:{stat.st_size}" if stat else None
-
-
-def _key_matches(key_path: Path, key: Any) -> bool:
-    """결과 옆 열쇠 파일이 지금 입력과 같은지."""
-    try:
-        return json.loads(key_path.read_text(encoding="utf-8")) == key
-    except (OSError, ValueError):
-        return False
-
-
 def _relative_to_project(project_root: Path, path: Path) -> str:
     """프로젝트 루트 기준 posix 상대 경로 문자열."""
     return path.relative_to(project_root).as_posix()

B04_PreProcess/B04_PreProcess_Engine_SheetSurface.py

@@ -25,7 +25,6 @@ from typing import Any
 import numpy as np
 from pyproj import Transformer
 
-from B04_PreProcess.B04_PreProcess_Engine import file_stamp
 from B04_PreProcess.B04_PreProcess_Engine_ModelContext import (
     atomic_npz,
     clip_and_compact_mesh,
@@ -38,6 +37,7 @@ from B04_PreProcess.B04_PreProcess_Engine_SheetMethods import (
     SHEET_METHOD_LABELS,
 )
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import grid_spec_from_bounds
+from common_util.common_util_derived_store import file_stamp
 from common_util.common_util_json import atomic_write_json
 from config.config_system import (
     SHEET_SURFACE_GRID_M,

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

B05_Profile/B05_Profile_UI_Profile_History.ts

@@ -10,10 +10,15 @@
  *
  * 저장하지 않는 것: 패널 접힘·높이·표시 토글 같은 **화면 배치 값**. 조작 이력이 아니라
  * 보기 설정이라 되돌릴 대상이 아니다(`isLayoutKey`).
+ *
+ * 대상 = 초안 통(`aislo:draft:` 전부 — 노선 초안 · 계획고 · 관 · 구조물 · 횡단) + 옛 b05 · b06 키(346-4).
+ * 끌기 · 누르고 있기 중간은 안 찍음 — 누름이 남아 있는 동안 record() 는 미루고 놓을 때 한 번.
+ * 사진끼리 안 바뀐 값은 앞 사진 글을 같이 씀 — 칸마다 수백 KB × 100단이어도 바뀐 칸만 늘어남.
  * ========================================================================== */
 
 /** 조작값 스냅샷 대상 세션 키 접두어. B05·B06은 한 페이지라 함께 뜬다(CLAUDE.md 5장). */
-const DATA_KEY_PREFIXES = ["b05:", "b05-", "b06:", "b06-", "aislo:draft:alignment-draft:"];
+const DRAFT_PREFIX = "aislo:draft:";
+const LEGACY_KEY_PREFIXES = ["b05:", "b05-", "b06:", "b06-"];
 
 /** 조작이 아니라 보기 설정인 키 — 스냅샷에서 뺀다. */
 function isLayoutKey(key: string): boolean {
@@ -27,20 +32,22 @@ function isLayoutKey(key: string): boolean {
 }
 
 function isDataKey(key: string): boolean {
-  return DATA_KEY_PREFIXES.some((prefix) => key.startsWith(prefix)) && !isLayoutKey(key);
+  if (key.startsWith(DRAFT_PREFIX)) return true;
+  return LEGACY_KEY_PREFIXES.some((prefix) => key.startsWith(prefix)) && !isLayoutKey(key);
 }
 
 /** 한 시점의 조작값 — 키 → 값(JSON 문자열). */
 export type HistorySnapshot = Record<string, string>;
 
-function takeSnapshot(): HistorySnapshot {
+/** `previous` 와 같은 값은 그 글을 그대로 담음(메모리 한 벌). */
+function takeSnapshot(previous?: HistorySnapshot): HistorySnapshot {
   const snapshot: HistorySnapshot = {};
   try {
     for (let index = 0; index < sessionStorage.length; index += 1) {
       const key = sessionStorage.key(index);
       if (!key || !isDataKey(key)) continue;
       const value = sessionStorage.getItem(key);
-      if (value !== null) snapshot[key] = value;
+      if (value !== null) snapshot[key] = previous?.[key] === value ? previous[key] : value;
     }
   } catch {
     // 세션 저장소를 못 쓰는 환경에서는 되돌리기만 비활성이 된다.
@@ -55,6 +62,39 @@ function sameSnapshot(left: HistorySnapshot, right: HistorySnapshot): boolean {
   return leftKeys.every((key) => left[key] === right[key]);
 }
 
+/** 누름(마우스 · 터치 · 펜)이 남아 있나 — 창 하나에 한 번 듣고 놓으면 미룬 기록을 부름. */
+let pointerHeld = false;
+const releaseWaiters = new Set<() => void>();
+let pointerWatched = false;
+
+function watchPointer(): void {
+  if (pointerWatched || typeof window === "undefined") return;
+  if (typeof window.addEventListener !== "function") return;
+  pointerWatched = true;
+  window.addEventListener("pointerdown", () => (pointerHeld = true), true);
+  const release = (): void => {
+    if (!pointerHeld) return;
+    // 놓을 때 부르는 쪽(끌기 마무리 · 세션 쓰기)이 먼저 끝나게 한 틱 뒤.
+    window.setTimeout(() => {
+      pointerHeld = false;
+      const waiters = [...releaseWaiters];
+      releaseWaiters.clear();
+      waiters.forEach((fn) => fn());
+    }, 0);
+  };
+  window.addEventListener("pointerup", release, true);
+  window.addEventListener("pointercancel", release, true);
+  window.addEventListener("blur", release);
+  // 창 밖에서 놓아 pointerup 을 못 받은 때 — 단추 없이 움직이면 놓은 것.
+  window.addEventListener(
+    "pointermove",
+    (event) => {
+      if (pointerHeld && event.buttons === 0) release();
+    },
+    true,
+  );
+}
+
 /** 스냅샷을 세션에 되쓴다 — 스냅샷에 없던 조작 키는 지운다. */
 function restoreSnapshot(snapshot: HistorySnapshot): void {
   try {
@@ -91,7 +131,8 @@ export interface ProfileHistory {
  * `onRestore`는 세션 값이 바뀐 뒤 화면을 다시 세우는 콜백이다 — 세션이 정본이므로
  * 여기서 각 패널이 자기 값을 다시 읽어 그린다.
  */
-export function createProfileHistory(onRestore: () => void, limit = 50): ProfileHistory {
+export function createProfileHistory(onRestore: () => void, limit = 100): ProfileHistory {
+  watchPointer();
   const stack: HistorySnapshot[] = [takeSnapshot()];
   let cursor = 0;
   /** 복원 중에 들어오는 record()를 무시한다 — 복원이 새 이력을 만들면 redo가 사라진다. */
@@ -112,10 +153,14 @@ export function createProfileHistory(onRestore: () => void, limit = 50): Profile
     return true;
   }
 
-  return {
+  const history: ProfileHistory = {
     record() {
       if (restoring || paused) return;
-      const snapshot = takeSnapshot();
+      if (pointerHeld) {
+        releaseWaiters.add(history.record);
+        return;
+      }
+      const snapshot = takeSnapshot(stack[cursor]);
       if (sameSnapshot(snapshot, stack[cursor])) return;
       stack.splice(cursor + 1);
       stack.push(snapshot);
@@ -128,11 +173,12 @@ export function createProfileHistory(onRestore: () => void, limit = 50): Profile
     resume() {
       if (!paused) return;
       paused = false;
-      this.record();
+      history.record();
     },
     undo: () => apply(cursor - 1),
     redo: () => apply(cursor + 1),
     canUndo: () => cursor > 0,
     canRedo: () => cursor < stack.length - 1,
   };
+  return history;
 }

B06_Section/B06_Section_Router_HaulPlan.py

@@ -19,12 +19,14 @@ import hashlib
 import json
 import logging
 from collections import OrderedDict
+from pathlib import Path
 from typing import Any
 from uuid import UUID
 
 from fastapi import APIRouter, Body
 from fastapi.responses import JSONResponse
 
+from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
 from B06_Section.B06_Section_Server_Calc_Prebuild import (
     BUNDLE,
     _mass_haul_context,
@@ -32,7 +34,10 @@ from B06_Section.B06_Section_Server_Calc_Prebuild import (
     haul_inputs_for,
     haul_limits_for,
 )
+from common_util.common_util_derived_store import read_derived, write_derived
 from common_util.common_util_node_bundle import run_bundle_json
+from common_util.common_util_storage import resolve_stored_project_path
+from config.config_db import run_with_connection
 
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/api/projects", tags=["B06 Profile Cross"])
@@ -41,10 +46,11 @@ _NPM_SCRIPT = "build:server-calc"
 # 측점 수 상한 — 정상 노선은 수백 곳이다. 그보다 크면 비정상 요청으로 본다.
 _MAX_POINTS = 5000
 # 열쇠(누가토량 결과 + 계수 · 장비 경계) → 배분 — 같은 열쇠면 Node 를 다시 안 띄움
-# (342-1 · 부록 4 R25). ponytail: 프로세스 메모리 · 최근 몇 벌만 — 재시작 뒤 첫 열기는
-# 다시 셈(_derived 저장본은 343 틀 뒤).
+# (342-1 · 부록 4 R25). 프로세스 메모리 먼저 · 다음 = 프로젝트 `_derived/haul/<열쇠>.json`
+# (재시작 뒤 첫 열기도 셈 없음 · 종류당 최근 2벌).
 _KEPT: OrderedDict[str, Any] = OrderedDict()
 _KEPT_MAX = 32
+_DERIVED_KIND = "haul"
 
 
 def _plan_key(result: dict[str, Any], context: Any) -> str:
@@ -52,6 +58,12 @@ def _plan_key(result: dict[str, Any], context: Any) -> str:
     return hashlib.sha1(body.encode("utf-8")).hexdigest()
 
 
+def _keep(key: str, plan: Any) -> None:
+    _KEPT[key] = plan
+    if len(_KEPT) > _KEPT_MAX:
+        _KEPT.popitem(last=False)
+
+
 @router.post("/{project_id}/sections/{route_id}/haul-plan", response_model=None)
 async def compute_haul_plan(
     project_id: UUID,
@@ -87,6 +99,20 @@ async def compute_haul_plan(
     if key in _KEPT:
         _KEPT.move_to_end(key)
         return JSONResponse(content={"status": "success", "haul_plan": _KEPT[key]})
+    project_root: Path | None = None
+    try:
+        stored_path = await run_with_connection(get_project_storage_relative_path, project_id)
+        project_root = Path(resolve_stored_project_path(stored_path))
+    except Exception:  # noqa: BLE001 — 저장소를 못 찾아도 셈은 함(디스크 저장본만 건너뜀)
+        logger.warning("유토 배분 저장소 못 찾음: project_id=%s", project_id)
+    stored = (
+        None
+        if project_root is None
+        else await asyncio.to_thread(read_derived, project_root, _DERIVED_KIND, key)
+    )
+    if stored is not None:
+        _keep(key, stored["value"])
+        return JSONResponse(content={"status": "success", "haul_plan": stored["value"]})
     try:
         output = await asyncio.to_thread(
             run_bundle_json,
@@ -107,7 +133,10 @@ async def compute_haul_plan(
             content={"status": "error", "message": "유토 배분 계산에 실패했습니다."},
         )
     plan = output.get("haul_plan")
-    _KEPT[key] = plan
-    if len(_KEPT) > _KEPT_MAX:
-        _KEPT.popitem(last=False)
+    _keep(key, plan)
+    try:
+        if project_root is not None:
+            await asyncio.to_thread(write_derived, project_root, _DERIVED_KIND, key, plan)
+    except OSError:  # 못 적어도 응답은 그대로 — 다음에 다시 셈
+        logger.warning("유토 배분 저장본 쓰기 실패: project_id=%s", project_id, exc_info=True)
     return JSONResponse(content={"status": "success", "haul_plan": plan})

B07_DesignDetail/B07_DesignDetail_Router_Support_Io.py

@@ -9,7 +9,7 @@ import re
 from pathlib import Path
 from typing import Any
 
-from B05_Profile.B05_Profile_Engine_Sections import cross_station, prune_stale_cross_files
+from B05_Profile.B05_Profile_Engine_Sections import cross_station, valid_cross_names
 
 logger = logging.getLogger(__name__)
 
@@ -39,7 +39,8 @@ def _cross_files(longitudinal_path: Path, longitudinal: dict[str, Any]) -> list[
     if not cross_dir.is_dir():
         raise FileNotFoundError("B06 횡단면 파일을 찾을 수 없습니다.")
     stations = longitudinal.get("stations")
-    valid_names = prune_stale_cross_files(cross_dir, stations if isinstance(stations, list) else [])
+    # 읽기는 지우지 않음 — 잔재는 걸러 읽기만(지우기는 측점을 쓰는 쪽 몫 · R26)
+    valid_names = valid_cross_names(cross_dir, stations if isinstance(stations, list) else [])
     files = sorted(cross_dir.glob("cross_*.json"))
     if valid_names:
         files = [path for path in files if path.name in valid_names]

common_util/common_util_derived_store.py

@@ -0,0 +1,78 @@
+"""파생 저장본 — 입력 지문 열쇠 + 결과 한 쌍(343 · 341 부록 4 2-1).
+
+자리 = 프로젝트 `_derived/<종류>/<열쇠>.json` · 작업본에서 다시 셀 수 있는 값만(정본 아님) ·
+통째로 지워도 다시 셈으로 돌아옴 · 종류마다 최근 `KEEP` 벌(꺼내면 최근으로).
+결과 옆 열쇠(`<이름>.key.json` — 결과가 곧 산출물인 도엽 3D 면 등)는 `key_matches` · `file_stamp`.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from common_util.common_util_json import atomic_write_json
+
+DERIVED_DIR = "_derived"
+KEEP = 2
+
+
+def derived_key(*parts: Any) -> str:
+    """입력 지문 — 작은 입력은 내용 그대로 · 파일은 `file_stamp` 를 넣는다."""
+    body = json.dumps(parts, sort_keys=True, ensure_ascii=False, default=str)
+    return hashlib.sha1(body.encode("utf-8")).hexdigest()
+
+
+def file_stamp(path: Path | None) -> str | None:
+    """파일 지문 = 수정시각 ns : 크기. 없으면 None."""
+    try:
+        stat = path.stat() if path is not None else None
+    except OSError:
+        return None
+    return f"{stat.st_mtime_ns}:{stat.st_size}" if stat else None
+
+
+def key_matches(key_path: Path, key: Any) -> bool:
+    """결과 옆 열쇠 파일이 지금 입력과 같은지."""
+    try:
+        return json.loads(key_path.read_text(encoding="utf-8")) == key
+    except (OSError, ValueError):
+        return False
+
+
+def derived_dir(project_root: Path, kind: str) -> Path:
+    return Path(project_root) / DERIVED_DIR / kind
+
+
+def read_derived(project_root: Path, kind: str, key: str) -> dict[str, Any] | None:
+    """저장본 `{"value": …}` — 없거나 못 읽으면 None(값이 null 이어도 맞음과 구별)."""
+    path = derived_dir(project_root, kind) / f"{key}.json"
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        return None
+    if not isinstance(data, dict) or data.get("key") != key or "value" not in data:
+        return None
+    try:
+        path.touch()  # 최근에 쓴 것 = 정리 때 남김
+    except OSError:
+        pass
+    return data
+
+
+def write_derived(project_root: Path, kind: str, key: str, value: Any, keep: int = KEEP) -> None:
+    """원자 쓰기 뒤 같은 종류 옛 열쇠를 정리한다."""
+    atomic_write_json(derived_dir(project_root, kind) / f"{key}.json", {"key": key, "value": value})
+    prune_derived(project_root, kind, keep)
+
+
+def prune_derived(project_root: Path, kind: str, keep: int = KEEP) -> int:
+    """종류 하나를 최근 `keep` 벌만 남기고 지운다. 지운 수."""
+    folder = derived_dir(project_root, kind)
+    files = sorted(
+        ((path.stat().st_mtime_ns, path) for path in folder.glob("*.json")), reverse=True
+    )
+    for _, path in files[keep:]:
+        path.unlink(missing_ok=True)
+    return max(0, len(files) - keep)

common_util/common_util_drainage_detail.py

@@ -25,7 +25,6 @@ from __future__ import annotations
 
 import json
 import logging
-import math
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -36,10 +35,11 @@ from B04_PreProcess.B04_PreProcess_Engine_Watershed_Analyze import find_inflow_h
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Export import STAGES
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Flow import polygon_parts, polygonize_labels
 from B04_PreProcess.B04_PreProcess_Engine_Watershed_Grid import GridSpec
-from common_util.common_util_drainage_pipes import (
-    PIPE_FACILITY_BOX,
-    PIPE_FACILITY_FORD_BRIDGE,
-    PIPE_FACILITY_PIPE,
+from common_util.common_util_drainage_sizing import (  # noqa: F401 — 옛 자리에서 읽는 곳
+    estimate_pipe_diameter_mm,
+    load_rainfall_idf,
+    recommend_structure,
+    size_pipe,
 )
 from common_util.common_util_route_geometry import (
     RouteVertex,
@@ -48,24 +48,11 @@ from common_util.common_util_route_geometry import (
     is_uphill_at,
 )
 from common_util.common_util_structure_station import snap_candidates
-from common_util.common_util_wamis_rainfall import idf_intensity
 from config.config_system import (
-    DRAINAGE_BOX_THRESHOLD_MM,
     DRAINAGE_BRIDGE_THRESHOLD_MM,
-    DRAINAGE_DESIGN_FLOW_FACTOR,
     DRAINAGE_DITCH_SAMPLE_M,
-    DRAINAGE_FLOW_AREA_RATIO,
-    DRAINAGE_MANNING_N,
     DRAINAGE_PIPE_MAX_SPACING_M,
     DRAINAGE_PIPE_MIN_SPACING_M,
-    DRAINAGE_PIPE_SLOPE_DEG,
-    DRAINAGE_RAINFALL_FILENAME,
-    DRAINAGE_RECOMMEND_DIAMETERS_MM,
-    DRAINAGE_RUNOFF_COEFFICIENT,
-    DRAINAGE_TC_MIN_MINUTES,
-    DRAINAGE_VELOCITY_MAX_MS,
-    DRAINAGE_VELOCITY_MIN_MS,
-    PIPE_DEFAULT_DIAMETER_MM,
 )
 
 logger = logging.getLogger(__name__)
@@ -98,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
@@ -176,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 = [
@@ -209,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)]
@@ -219,6 +215,7 @@ def build_detail(
                 vertices[-1].chainage_m,
             )
         ],
+        pipe_candidates=candidates,
     )
     if not pipes:
         return detail
@@ -475,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]:
@@ -610,112 +625,3 @@ def _outlet_elevation(routing: RoadRouting, pipe_order: int, pipe_of_slot: np.nd
     elevations = routing.elevation[routing.road_cell_index[slots]]
     finite = elevations[np.isfinite(elevations)]
     return float(finite.min()) if finite.size else 0.0
-
-
-def load_rainfall_idf(directory: Path) -> dict[str, float] | None:
-    """배수유역 폴더의 rainfall_table.json에서 설계빈도 IDF 적합계수를 읽는다.
-
-    파일은 B04 전처리(`/drainage/rainfall`)가 만든다. 없으면 None — 유효직경은
-    "미정"으로 남고, 강우량표가 생기는 순간 다음 세부유역 계산부터 채워진다.
-    """
-    path = Path(directory) / DRAINAGE_RAINFALL_FILENAME
-    if not path.exists():
-        return None
-    try:
-        table = json.loads(path.read_text(encoding="utf-8"))
-    except (OSError, json.JSONDecodeError):
-        logger.warning("강우량표를 읽지 못했습니다 (%s) — 유효직경 미산출.", path)
-        return None
-    idf = table.get("idf_design")
-    if not isinstance(idf, dict) or not all(k in idf for k in ("a", "b", "n")):
-        return None
-    return idf
-
-
-def size_pipe(
-    area_m2: float,
-    relief_m: float,
-    flow_length_m: float,
-    idf: dict[str, float] | None,
-) -> dict[str, float] | None:
-    """유역 제원 + 설계빈도 IDF로 배수 유효직경(mm)과 산출 근거를 계산한다.
-
-    시행규칙 별표2 (가) 1호 방법: 100년빈도 확률강우량 + 홍수도달시간 → 합리식.
-    전 과정·계수 근거: docs/raw/guidelines/2026-08-05_임도_배수설계_규정_조사.md 4·6절.
-
-      ① 도달시간 tc = 0.0663·L^0.77·(H/L)^-0.385 [Kirpich, hr] — 하한 5분
-      ② 강우강도 I = General형 적합식 I(tc) [mm/hr]
-      ③ 합리식 Q = (1/3.6)·C·I·A[km²],  설계유량 = 2.0·Q (별표2)
-      ④ 유속 V = Manning(경사 10도 고정, n=0.024) → 0.8~3.0m/s 클램프
-      ⑤ 통수단면 70%만 유효 → D = √(4·(Q설계/V)/(0.7π))
-    """
-    if idf is None or area_m2 <= 0:
-        return None
-    length_km = max(flow_length_m, 1.0) / 1000.0
-    slope = max(relief_m, 0.1) / max(flow_length_m, 1.0)
-    tc_hr = 0.0663 * (length_km**0.77) * (slope**-0.385)
-    tc_min = max(DRAINAGE_TC_MIN_MINUTES, tc_hr * 60.0)
-    intensity = idf_intensity(idf, tc_min)
-    flow = DRAINAGE_RUNOFF_COEFFICIENT * intensity * (area_m2 / 1e6) / 3.6
-    design_flow = DRAINAGE_DESIGN_FLOW_FACTOR * flow
-    if design_flow <= 0:
-        return None
-    pipe_slope = math.tan(math.radians(DRAINAGE_PIPE_SLOPE_DEG))
-    diameter = 0.5
-    for _ in range(20):
-        velocity = (
-            (1.0 / DRAINAGE_MANNING_N) * (diameter / 4.0) ** (2.0 / 3.0) * math.sqrt(pipe_slope)
-        )
-        velocity = min(DRAINAGE_VELOCITY_MAX_MS, max(DRAINAGE_VELOCITY_MIN_MS, velocity))
-        required_area = design_flow / velocity
-        updated = math.sqrt(4.0 * required_area / (DRAINAGE_FLOW_AREA_RATIO * math.pi))
-        if abs(updated - diameter) < 1e-4:
-            diameter = updated
-            break
-        diameter = updated
-    return {
-        "tc_minutes": round(tc_min, 1),
-        "intensity_mm_hr": round(intensity, 1),
-        "design_flow_m3s": round(design_flow, 4),
-        "diameter_mm": round(diameter * 1000.0, 1),
-        # 물넘이·세월교는 관이 아니라 개수로라 직경 대신 이 단면적에서 출발한다.
-        "required_area_m2": round(required_area, 4),
-    }
-
-
-def recommend_structure(diameter_mm: float | None) -> tuple[str, int | None]:
-    """유효직경으로 추천 구조물과 관경을 고른다 (2026-08-17 사용자 확정: 유량 근거만).
-
-    지식DB 선정 트리(개거_세월시설 §5·횡단배수관_암거 §5)에서 유량 조건만 취했다 —
-    계곡 횡단경사·하천 차수는 지형 계산이 필요해 화면이 "현장 확인"으로 안내한다.
-
-      D ≤ 1,500㎜        배관 (레지스트리 선택지로 스냅, 하한 800㎜)
-      1,500 < D ≤ 2,000  BOX암거 후보
-      D > 2,000㎜        세월교·물넘이 검토 (관 최대 규격 초과)
-
-    유효직경을 못 구한 유역(강우량표 없음)은 배관·관경 미정으로 둔다.
-    """
-    if diameter_mm is None:
-        return PIPE_FACILITY_PIPE, None
-    if diameter_mm > DRAINAGE_BRIDGE_THRESHOLD_MM:
-        return PIPE_FACILITY_FORD_BRIDGE, None
-    if diameter_mm > DRAINAGE_BOX_THRESHOLD_MM:
-        return PIPE_FACILITY_BOX, None
-    # 별표2 (나) 예외 하한 800㎜ — 계산값이 더 작아도 그 아래로는 내리지 않는다.
-    need = max(float(PIPE_DEFAULT_DIAMETER_MM), diameter_mm)
-    size = next(
-        (value for value in DRAINAGE_RECOMMEND_DIAMETERS_MM if value >= need),
-        DRAINAGE_RECOMMEND_DIAMETERS_MM[-1],
-    )
-    return PIPE_FACILITY_PIPE, int(size)
-
-
-def estimate_pipe_diameter_mm(
-    area_m2: float,
-    relief_m: float,
-    flow_length_m: float,
-    idf: dict[str, float] | None = None,
-) -> float | None:
-    """유역 제원으로 배수 유효직경(mm)만 돌려주는 축약형 — 상세는 size_pipe()."""
-    sizing = size_pipe(area_m2, relief_m, flow_length_m, idf)
-    return sizing["diameter_mm"] if sizing else None

common_util/common_util_drainage_sizing.py

@@ -0,0 +1,145 @@
+"""배수 관경 산정 — 강우량표 읽기 · 합리식 유효직경 · 추천 구조물.
+
+세부 유역 엔진(`common_util_drainage_detail` · 700줄로 나눔)이 유역 제원으로 부름 ·
+이름은 그 파일에서도 그대로 읽힘.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+from pathlib import Path
+
+from common_util.common_util_drainage_pipes import (
+    PIPE_FACILITY_BOX,
+    PIPE_FACILITY_FORD_BRIDGE,
+    PIPE_FACILITY_PIPE,
+)
+from common_util.common_util_wamis_rainfall import idf_intensity
+from config.config_system import (
+    DRAINAGE_BOX_THRESHOLD_MM,
+    DRAINAGE_BRIDGE_THRESHOLD_MM,
+    DRAINAGE_DESIGN_FLOW_FACTOR,
+    DRAINAGE_FLOW_AREA_RATIO,
+    DRAINAGE_MANNING_N,
+    DRAINAGE_PIPE_SLOPE_DEG,
+    DRAINAGE_RAINFALL_FILENAME,
+    DRAINAGE_RECOMMEND_DIAMETERS_MM,
+    DRAINAGE_RUNOFF_COEFFICIENT,
+    DRAINAGE_TC_MIN_MINUTES,
+    DRAINAGE_VELOCITY_MAX_MS,
+    DRAINAGE_VELOCITY_MIN_MS,
+    PIPE_DEFAULT_DIAMETER_MM,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def load_rainfall_idf(directory: Path) -> dict[str, float] | None:
+    """배수유역 폴더의 rainfall_table.json에서 설계빈도 IDF 적합계수를 읽는다.
+
+    파일은 B04 전처리(`/drainage/rainfall`)가 만든다. 없으면 None — 유효직경은
+    "미정"으로 남고, 강우량표가 생기는 순간 다음 세부유역 계산부터 채워진다.
+    """
+    path = Path(directory) / DRAINAGE_RAINFALL_FILENAME
+    if not path.exists():
+        return None
+    try:
+        table = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        logger.warning("강우량표를 읽지 못했습니다 (%s) — 유효직경 미산출.", path)
+        return None
+    idf = table.get("idf_design")
+    if not isinstance(idf, dict) or not all(k in idf for k in ("a", "b", "n")):
+        return None
+    return idf
+
+
+def size_pipe(
+    area_m2: float,
+    relief_m: float,
+    flow_length_m: float,
+    idf: dict[str, float] | None,
+) -> dict[str, float] | None:
+    """유역 제원 + 설계빈도 IDF로 배수 유효직경(mm)과 산출 근거를 계산한다.
+
+    시행규칙 별표2 (가) 1호 방법: 100년빈도 확률강우량 + 홍수도달시간 → 합리식.
+    전 과정·계수 근거: docs/raw/guidelines/2026-08-05_임도_배수설계_규정_조사.md 4·6절.
+
+      ① 도달시간 tc = 0.0663·L^0.77·(H/L)^-0.385 [Kirpich, hr] — 하한 5분
+      ② 강우강도 I = General형 적합식 I(tc) [mm/hr]
+      ③ 합리식 Q = (1/3.6)·C·I·A[km²],  설계유량 = 2.0·Q (별표2)
+      ④ 유속 V = Manning(경사 10도 고정, n=0.024) → 0.8~3.0m/s 클램프
+      ⑤ 통수단면 70%만 유효 → D = √(4·(Q설계/V)/(0.7π))
+    """
+    if idf is None or area_m2 <= 0:
+        return None
+    length_km = max(flow_length_m, 1.0) / 1000.0
+    slope = max(relief_m, 0.1) / max(flow_length_m, 1.0)
+    tc_hr = 0.0663 * (length_km**0.77) * (slope**-0.385)
+    tc_min = max(DRAINAGE_TC_MIN_MINUTES, tc_hr * 60.0)
+    intensity = idf_intensity(idf, tc_min)
+    flow = DRAINAGE_RUNOFF_COEFFICIENT * intensity * (area_m2 / 1e6) / 3.6
+    design_flow = DRAINAGE_DESIGN_FLOW_FACTOR * flow
+    if design_flow <= 0:
+        return None
+    pipe_slope = math.tan(math.radians(DRAINAGE_PIPE_SLOPE_DEG))
+    diameter = 0.5
+    for _ in range(20):
+        velocity = (
+            (1.0 / DRAINAGE_MANNING_N) * (diameter / 4.0) ** (2.0 / 3.0) * math.sqrt(pipe_slope)
+        )
+        velocity = min(DRAINAGE_VELOCITY_MAX_MS, max(DRAINAGE_VELOCITY_MIN_MS, velocity))
+        required_area = design_flow / velocity
+        updated = math.sqrt(4.0 * required_area / (DRAINAGE_FLOW_AREA_RATIO * math.pi))
+        if abs(updated - diameter) < 1e-4:
+            diameter = updated
+            break
+        diameter = updated
+    return {
+        "tc_minutes": round(tc_min, 1),
+        "intensity_mm_hr": round(intensity, 1),
+        "design_flow_m3s": round(design_flow, 4),
+        "diameter_mm": round(diameter * 1000.0, 1),
+        # 물넘이·세월교는 관이 아니라 개수로라 직경 대신 이 단면적에서 출발한다.
+        "required_area_m2": round(required_area, 4),
+    }
+
+
+def recommend_structure(diameter_mm: float | None) -> tuple[str, int | None]:
+    """유효직경으로 추천 구조물과 관경을 고른다 (2026-08-17 사용자 확정: 유량 근거만).
+
+    지식DB 선정 트리(개거_세월시설 §5·횡단배수관_암거 §5)에서 유량 조건만 취했다 —
+    계곡 횡단경사·하천 차수는 지형 계산이 필요해 화면이 "현장 확인"으로 안내한다.
+
+      D ≤ 1,500㎜        배관 (레지스트리 선택지로 스냅, 하한 800㎜)
+      1,500 < D ≤ 2,000  BOX암거 후보
+      D > 2,000㎜        세월교·물넘이 검토 (관 최대 규격 초과)
+
+    유효직경을 못 구한 유역(강우량표 없음)은 배관·관경 미정으로 둔다.
+    """
+    if diameter_mm is None:
+        return PIPE_FACILITY_PIPE, None
+    if diameter_mm > DRAINAGE_BRIDGE_THRESHOLD_MM:
+        return PIPE_FACILITY_FORD_BRIDGE, None
+    if diameter_mm > DRAINAGE_BOX_THRESHOLD_MM:
+        return PIPE_FACILITY_BOX, None
+    # 별표2 (나) 예외 하한 800㎜ — 계산값이 더 작아도 그 아래로는 내리지 않는다.
+    need = max(float(PIPE_DEFAULT_DIAMETER_MM), diameter_mm)
+    size = next(
+        (value for value in DRAINAGE_RECOMMEND_DIAMETERS_MM if value >= need),
+        DRAINAGE_RECOMMEND_DIAMETERS_MM[-1],
+    )
+    return PIPE_FACILITY_PIPE, int(size)
+
+
+def estimate_pipe_diameter_mm(
+    area_m2: float,
+    relief_m: float,
+    flow_length_m: float,
+    idf: dict[str, float] | None = None,
+) -> float | None:
+    """유역 제원으로 배수 유효직경(mm)만 돌려주는 축약형 — 상세는 size_pipe()."""
+    sizing = size_pipe(area_m2, relief_m, flow_length_m, idf)
+    return sizing["diameter_mm"] if sizing else None

common_util/common_util_node_bundle.py

@@ -13,6 +13,7 @@ import logging
 import os
 import subprocess
 import tempfile
+import time
 from pathlib import Path
 from typing import Any
 
@@ -48,15 +49,34 @@ def node_env() -> dict[str, str]:
     return env
 
 
+# 「낡지 않음」 판정 기억 — 묶음마다 (묶음 mtime_ns, 잰 때). 같은 묶음이면 `RECHECK_S` 안에는
+# TS 330여 개를 다시 stat 하지 않는다(R24 · 부름마다 수십 ms). 낡음은 기억하지 않는다.
+RECHECK_S = 30.0
+_FRESH: dict[str, tuple[int, float]] = {}
+
+
 def bundle_stale(bundle: Path) -> bool:
     """번들이 없거나 TS 원본보다 오래됐으면 참.
 
     번들이 낡으면 서버와 화면이 **다른 값**을 만든다 — 이 판정이 그것을 막는 유일한
     장치다. 개발 중에는 `npm run build`를 따로 돌리지 않으므로 여기서 스스로 갱신한다.
+    「낡지 않음」은 `RECHECK_S` 동안 기억 — TS 를 고치면 늦어도 그만큼 뒤에 다시 짓는다.
     """
     if not bundle.is_file():
         return True
-    built_at = bundle.stat().st_mtime
+    stamp = bundle.stat().st_mtime_ns
+    remembered = _FRESH.get(str(bundle))
+    if remembered and remembered[0] == stamp and time.monotonic() - remembered[1] < RECHECK_S:
+        return False
+    if _sources_newer(bundle.stat().st_mtime):
+        _FRESH.pop(str(bundle), None)
+        return True
+    _FRESH[str(bundle)] = (stamp, time.monotonic())
+    return False
+
+
+def _sources_newer(built_at: float) -> bool:
+    """묶음이 끌어가는 TS · 자료 중 하나라도 `built_at` 보다 새로우면 참."""
     for directory in SOURCE_DIRS:
         for path in (ROOT / directory).rglob("*.ts"):
             if "node_modules" not in path.parts and path.stat().st_mtime > built_at:

resources/tester/test_343_4_derived.py

@@ -0,0 +1,170 @@
+"""343 덧 — 파생 저장본 도우미 · 유토 배분 저장본 · B07 읽기 안 지움 · 묶음 낡음 검사 기억."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import time
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+
+from B06_Section import B06_Section_Router_HaulPlan as haul
+from common_util import common_util_derived_store as store
+from common_util import common_util_node_bundle as nb
+
+# ── 도우미 ──────────────────────────────────────────────────────────────────────
+
+
+def test_열쇠_같은_입력_같은_값() -> None:
+    assert store.derived_key({"a": 1, "b": 2}, 3) == store.derived_key({"b": 2, "a": 1}, 3)
+    assert store.derived_key({"a": 1}) != store.derived_key({"a": 2})
+
+
+def test_읽기_쓰기_null_도_맞음(tmp_path: Path) -> None:
+    assert store.read_derived(tmp_path, "haul", "k") is None
+    store.write_derived(tmp_path, "haul", "k", None)
+    assert store.read_derived(tmp_path, "haul", "k") == {"key": "k", "value": None}
+    assert (tmp_path / "_derived" / "haul" / "k.json").is_file()
+
+
+def test_종류마다_최근_두_벌_꺼내면_최근(tmp_path: Path) -> None:
+    store.write_derived(tmp_path, "haul", "a", 1)
+    time.sleep(0.02)
+    store.write_derived(tmp_path, "haul", "b", 2)
+    store.write_derived(tmp_path, "other", "x", 0)
+    time.sleep(0.02)
+    assert store.read_derived(tmp_path, "haul", "a")["value"] == 1
+    time.sleep(0.02)
+    store.write_derived(tmp_path, "haul", "c", 3)
+    names = sorted(path.stem for path in (tmp_path / "_derived" / "haul").glob("*.json"))
+    assert names == ["a", "c"]
+    assert (tmp_path / "_derived" / "other" / "x.json").is_file()
+
+
+def test_결과_옆_열쇠와_파일_지문(tmp_path: Path) -> None:
+    path = tmp_path / "m.key.json"
+    assert not store.key_matches(path, {"a": 1})
+    path.write_text(json.dumps({"a": 1}), encoding="utf-8")
+    assert store.key_matches(path, {"a": 1}) and not store.key_matches(path, {"a": 2})
+    assert store.file_stamp(tmp_path / "없음") is None and store.file_stamp(path)
+
+
+# ── 유토 배분 디스크 저장본 ─────────────────────────────────────────────────────────
+
+
+def _wire(monkeypatch: pytest.MonkeyPatch, root: Path, outputs: list) -> list:
+    calls: list = []
+
+    async def _none(_project_id):  # type: ignore[no-untyped-def]
+        return {}
+
+    async def _stored(_fn, _project_id):  # type: ignore[no-untyped-def]
+        return "p"
+
+    def _run(_bundle, _script, payload):  # type: ignore[no-untyped-def]
+        calls.append(payload)
+        return outputs.pop(0)
+
+    monkeypatch.setattr(haul, "haul_inputs_for", _none)
+    monkeypatch.setattr(haul, "conversion_factors_for", _none)
+    monkeypatch.setattr(haul, "haul_limits_for", _none)
+    monkeypatch.setattr(haul, "_mass_haul_context", lambda *_: {"limits": []})
+    monkeypatch.setattr(haul, "run_bundle_json", _run)
+    monkeypatch.setattr(haul, "run_with_connection", _stored)
+    monkeypatch.setattr(haul, "resolve_stored_project_path", lambda _p: str(root))
+    haul._KEPT.clear()
+    return calls
+
+
+def _post(result: dict) -> dict:
+    response = asyncio.run(haul.compute_haul_plan(uuid4(), 7, {"result": result}))
+    return json.loads(response.body)
+
+
+def test_유토_배분_재시작_뒤에도_셈_없음(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _wire(monkeypatch, tmp_path, [{"haul_plan": {"tag": "a"}}, {"haul_plan": {"tag": "b"}}])
+    a = {"points": [{"cumulative_volume_m3": 1.0}]}
+    assert _post(a)["haul_plan"] == {"tag": "a"} and len(calls) == 1
+    key = haul._plan_key(a, {"limits": []})  # 화면 · 서버 메모리와 같은 열쇠
+    assert (tmp_path / "_derived" / "haul" / f"{key}.json").is_file()
+    haul._KEPT.clear()  # 재시작
+    assert _post(a)["haul_plan"] == {"tag": "a"} and len(calls) == 1
+    b = {"points": [{"cumulative_volume_m3": 2.0}]}
+    assert _post(b)["haul_plan"] == {"tag": "b"} and len(calls) == 2
+
+
+def test_유토_배분_저장소_못_찾아도_셈(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _wire(monkeypatch, tmp_path, [{"haul_plan": {"tag": "a"}}])
+
+    async def _missing(_fn, _project_id):  # type: ignore[no-untyped-def]
+        raise LookupError("없음")
+
+    monkeypatch.setattr(haul, "run_with_connection", _missing)
+    assert _post({"points": [{"cumulative_volume_m3": 1.0}]})["haul_plan"] == {"tag": "a"}
+    assert len(calls) == 1 and not (tmp_path / "_derived").exists()
+
+
+# ── B07 읽기는 측점 파일을 안 지움 ─────────────────────────────────────────────────
+
+
+def test_B07_읽기는_잔재를_걸러만_읽음(tmp_path: Path) -> None:
+    from B07_DesignDetail.B07_DesignDetail_Router_Support_Io import _cross_files
+
+    cross_dir = tmp_path / "B05_Profile" / "cross_sections"
+    cross_dir.mkdir(parents=True)
+    for name in ("cross_00000m.json", "cross_00020m.json", "cross_00040m.json"):
+        (cross_dir / name).write_text("{}", encoding="utf-8")
+    longitudinal_path = tmp_path / "B05_Profile" / "longitudinal" / "longitudinal.json"
+    files = _cross_files(longitudinal_path, {"stations": [{"chainage_m": 0}, {"chainage_m": 20}]})
+    assert [path.name for path in files] == ["cross_00000m.json", "cross_00020m.json"]
+    assert (cross_dir / "cross_00040m.json").is_file()
+
+
+# ── 묶음 낡음 검사 기억 ─────────────────────────────────────────────────────────
+
+
+@pytest.fixture
+def bundle(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    monkeypatch.setattr(nb, "ROOT", tmp_path)
+    monkeypatch.setattr(nb, "_FRESH", {})
+    path = tmp_path / "dist" / "bundle.mjs"
+    path.parent.mkdir()
+    path.write_text("", encoding="utf-8")
+    os.utime(path, (1_000_000_000, 1_000_000_000))
+    return path
+
+
+def _counting(monkeypatch: pytest.MonkeyPatch) -> list:
+    scans: list = []
+    real = nb._sources_newer
+
+    def counting(built_at: float) -> bool:
+        scans.append(built_at)
+        return real(built_at)
+
+    monkeypatch.setattr(nb, "_sources_newer", counting)
+    return scans
+
+
+def test_낡지_않음은_잠시_기억(bundle: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    scans = _counting(monkeypatch)
+    assert not nb.bundle_stale(bundle) and not nb.bundle_stale(bundle)
+    assert len(scans) == 1
+    # 묶음을 새로 지으면(mtime 바뀜) 다시 잼
+    os.utime(bundle, (1_000_000_100, 1_000_000_100))
+    assert not nb.bundle_stale(bundle) and len(scans) == 2
+    # 기억 시간이 지나면 다시 잼
+    monkeypatch.setattr(nb, "RECHECK_S", 0.0)
+    assert not nb.bundle_stale(bundle) and len(scans) == 3
+
+
+def test_낡음은_기억하지_않음(bundle: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    source = nb.ROOT / "B05_Profile" / "x.ts"
+    source.parent.mkdir()
+    source.write_text("", encoding="utf-8")
+    scans = _counting(monkeypatch)
+    assert nb.bundle_stale(bundle) and nb.bundle_stale(bundle)
+    assert len(scans) == 2 and str(bundle) not in nb._FRESH

resources/tester/test_346_4_profile_history.py

@@ -0,0 +1,128 @@
+"""346-4 되돌리기 앞일 — 사진 대상 = aislo:draft: 전부 · 누름 중간은 안 찍고 놓을 때 한 번 · 크기.
+
+./venv/Scripts/python.exe -m pytest resources/tester/test_346_4_profile_history.py -q
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+TS_FILE = ROOT / "B05_Profile" / "B05_Profile_UI_Profile_History.ts"
+TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
+
+DRIVER = r"""
+import { writeFileSync } from "node:fs";
+// 세션 흉내 — getItem 은 진짜처럼 매번 새 글을 돌려줌(사진끼리 글을 같이 쓰는지 재려고).
+const data = new Map();
+globalThis.sessionStorage = {
+  get length() { return data.size; },
+  key: (i) => [...data.keys()][i] ?? null,
+  getItem: (k) => (data.has(k) ? Buffer.from(data.get(k)).toString() : null),
+  setItem: (k, v) => data.set(k, String(v)),
+  removeItem: (k) => data.delete(k),
+};
+globalThis.window = new EventTarget();
+window.setTimeout = setTimeout;
+const { createProfileHistory } = await import("./B05_Profile_UI_Profile_History.js");
+const tick = () => new Promise((r) => setTimeout(r, 5));
+const out = {};
+
+// 1. 대상 — 초안 통 전부 + 옛 키 · 보기 설정 · pref · result 는 뺌
+const K = {
+  route: "aislo:draft:route-draft:p1",
+  cross: "aislo:draft:crossdesign:p1:7",
+  pipes: "aislo:draft:pipes:p1",
+  legacy: "b05:uphill:p1",
+  layout: "b05-panel-collapsed",
+  pref: "aislo:pref:table-open",
+  result: "aislo:result:latest:p1",
+};
+for (const k of Object.values(K)) sessionStorage.setItem(k, "0");
+let restored = 0;
+const h = createProfileHistory(() => (restored += 1));
+for (const k of Object.values(K)) sessionStorage.setItem(k, "1");
+h.record();
+out.undo = h.undo();
+out.after = Object.fromEntries(Object.entries(K).map(([n, k]) => [n, sessionStorage.getItem(k)]));
+out.restored = restored;
+
+// 2. 누르고 있는 동안 기록 미룸 · 놓으면 한 번
+data.clear();
+sessionStorage.setItem(K.cross, "a");
+const h2 = createProfileHistory(() => {});
+window.dispatchEvent(new Event("pointerdown"));
+for (const v of ["b", "c", "d", "e"]) {
+  sessionStorage.setItem(K.cross, v);
+  h2.record();
+}
+out.heldCanUndo = h2.canUndo();
+window.dispatchEvent(new Event("pointerup"));
+await tick();
+out.releasedCanUndo = h2.canUndo();
+h2.undo();
+out.oneUndo = sessionStorage.getItem(K.cross);
+out.noMoreUndo = !h2.canUndo();
+
+// 3. 크기 — 300 KB 칸 셋 · 작은 칸만 바꾸며 100단
+data.clear();
+const big = (c) => c.repeat(300 * 1024);
+sessionStorage.setItem(K.route, big("r"));
+sessionStorage.setItem(K.cross, big("x"));
+sessionStorage.setItem(K.pipes, big("p"));
+globalThis.gc();
+const before = process.memoryUsage().heapUsed;
+const h3 = createProfileHistory(() => {});
+for (let i = 0; i < 100; i += 1) {
+  sessionStorage.setItem("aislo:draft:alignment-draft:p1:7", JSON.stringify({ i }));
+  h3.record();
+}
+globalThis.gc();
+out.heapMb = (process.memoryUsage().heapUsed - before) / 1024 / 1024;
+let steps = 0;
+while (h3.undo()) steps += 1;
+out.steps = steps;
+writeFileSync(process.argv[2], JSON.stringify(out));
+"""
+
+
+@pytest.fixture(scope="module")
+def result(tmp_path_factory: pytest.TempPathFactory) -> dict:
+    out = tmp_path_factory.mktemp("ts_history")
+    subprocess.run(  # noqa: S603 — 고정 실행 파일
+        ["node", str(TSC), str(TS_FILE), "--outDir", str(out), "--module", "esnext",
+         "--target", "es2022", "--moduleResolution", "bundler", "--ignoreConfig", "--noCheck",
+         "--noResolve"],
+        cwd=str(ROOT), check=True, capture_output=True,
+    )  # fmt: skip
+    (out / "driver.mjs").write_text(DRIVER, encoding="utf-8")
+    run = subprocess.run(  # noqa: S603
+        ["node", "--expose-gc", str(out / "driver.mjs"), str(out / "out.json")],
+        capture_output=True, text=True, cwd=str(out),
+    )  # fmt: skip
+    assert run.returncode == 0, run.stderr
+    return json.loads((out / "out.json").read_text(encoding="utf-8"))
+
+
+def test_초안_통_전부를_되돌림(result) -> None:
+    after = result["after"]
+    assert result["undo"] and result["restored"] == 1
+    assert after["route"] == after["cross"] == after["pipes"] == after["legacy"] == "0"
+    # 보기 설정 · pref · result 는 사진 밖 — 되돌리기가 안 건드림
+    assert after["layout"] == after["pref"] == after["result"] == "1"
+
+
+def test_누름_중간은_안_찍고_놓을_때_한_번(result) -> None:
+    assert result["heldCanUndo"] is False
+    assert result["releasedCanUndo"] is True
+    assert result["oneUndo"] == "a" and result["noMoreUndo"]
+
+
+def test_큰_칸_100단도_바뀐_칸만_늚(result) -> None:
+    assert result["steps"] == 99  # 한도 100(첫 사진 포함)
+    # 같은 글을 같이 쓰지 않으면 900 KB × 100 ≈ 90 MB(UTF-8 1바이트 글 기준 · 실제는 더)
+    assert result["heapMb"] < 10, result["heapMb"]

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)