Merge remote-tracking branch 'origin/sub_laptop_2' into sub_laptop_4

B05_Profile/B05_Profile_Engine_RouteInitial.py

@@ -191,6 +191,7 @@ def read_initial_choice(project_root: Path) -> dict[str, Any]:
 
     `criteria` 는 **사용자가 겹쳐 쓴 값만**(비운 칸은 안 담김) — 나머지는 프로젝트 기본값이라
     화면이 다시 열 때 겹친 칸만 채우고 나머지는 placeholder(기본값)로 비워 둔다.
+    `with_profile` = [적용] 때 「종단 반영」(없으면 켬 · 420).
     """
     try:
         data = json.loads(_choice_path(project_root).read_text(encoding="utf-8"))
@@ -200,9 +201,15 @@ def read_initial_choice(project_root: Path) -> dict[str, Any]:
             "logic": logic,
             "weights": resolve_weights(logic, data.get("weights")),
             "criteria": criteria if isinstance(criteria, dict) and criteria else None,
+            "with_profile": data.get("with_profile") is not False,
         }
     except (OSError, ValueError, KeyError, TypeError, AttributeError):
-        return {"logic": "follow", "weights": resolve_weights("follow", None), "criteria": None}
+        return {
+            "logic": "follow",
+            "weights": resolve_weights("follow", None),
+            "criteria": None,
+            "with_profile": True,
+        }
 
 
 def write_initial_choice(
@@ -210,6 +217,7 @@ def write_initial_choice(
     logic: str,
     weights: dict[str, float] | None,
     criteria: dict[str, float] | None = None,
+    with_profile: bool = True,
 ) -> dict[str, Any]:
     """[확인] 때만 부름 — 작업본 한 칸. 다시 열기 · 다른 PC 에서도 같은 초기값.
 
@@ -220,6 +228,7 @@ def write_initial_choice(
         "logic": logic,
         "weights": resolve_weights(logic, weights),
         "criteria": criteria or None,
+        "with_profile": bool(with_profile),
     }
     path = _choice_path(project_root)
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -283,6 +292,7 @@ def generate_initial_route(
     design_speed: int = 20,
     pipes: list[tuple[float, float, float]] | None = None,
     with_profile: bool = True,
+    standard: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """갈래 하나로 초기 계획노선을 만들고 비교값을 붙여 돌려준다. `max_grade` 는 비율(0.14).
 
@@ -349,6 +359,7 @@ def generate_initial_route(
                 design_speed=design_speed,
                 design_curves=[curve.as_dict() for curve in route[0].curves],
                 exact=exact,
+                standard=standard,
             )
 
     def quick(route: Any) -> None:

B05_Profile/B05_Profile_Engine_RouteInitial_CrossEarth.py

@@ -0,0 +1,177 @@
+"""노선 추천 횡단 토공(PLAN 420 · 요구 254).
+
+측점마다 공식 횡단 엔진(B06 `compute_cross_design` — [저장] 정본과 같은 파이썬)으로 절 · 성
+단면적을 재고 평균단면법으로 토량을 낸다. 띠 가늠(폭 4 m · 비탈 뺌 · `..._Metrics`)을 갈음하는
+추천 토공 — 후보 줄 세움 · 합치기 판정 · 결과 표 절 · 성.
+
+계산 자리 = 서버 단독(③). 측점 횡단 뜨기 · 단면적은 균형 종단(`..._Profile`)과 같은 함수 —
+종단 반영 끔 = 계획고 = 지반고([적용] 뒤 1차 종단과 같음).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import numpy as np
+
+from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import (
+    STATION_M,
+    _station_areas,
+    route_sections,
+)
+
+if TYPE_CHECKING:
+    from B05_Profile.B05_Profile_Engine_RouteInitial_Search import Terrain
+    from common_util.common_util_surface_sampler import SurfaceElevationSampler
+
+
+def _end_area_weights(at: np.ndarray) -> np.ndarray:
+    """평균단면법 무게 — 측점마다 앞 · 뒤 구간 반씩(Σ (A_i + A_i+1) / 2 · L = Σ A_i · w_i)."""
+    spans = np.diff(at)
+    weight = np.zeros(len(at))
+    weight[:-1] += spans / 2.0
+    weight[1:] += spans / 2.0
+    return weight
+
+
+def _areas(
+    line: list[tuple[float, float]],
+    chainages: Any,
+    design_z: Any,
+    terrain: Terrain | SurfaceElevationSampler,
+    standard: dict[str, Any] | None,
+    design_curves: list[dict[str, Any]] | None,
+) -> tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
+    """(누가거리, 지반고, 계획고, [절 자연 · 절 다짐 · 성] ㎡ 행렬)."""
+    at, ground, sections, modes, widen = route_sections(line, terrain, design_curves, at=chainages)
+    plan = ground if design_z is None else np.asarray(design_z, dtype=np.float64)
+    if plan.shape != at.shape:
+        raise ValueError("계획고 수가 측점 수와 다릅니다.")
+    rows = np.asarray(
+        [
+            _station_areas(sections[i], modes[i], float(plan[i]), widen[i], standard)
+            for i in range(len(at))
+        ]
+    ).reshape(-1, 3)
+    return at, ground, plan, rows
+
+
+def cross_earthwork(
+    line: list[tuple[float, float]],
+    chainages: Any,
+    design_z: Any,
+    terrain: Terrain | SurfaceElevationSampler,
+    standard: dict[str, Any] | None = None,
+    *,
+    design_curves: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """노선 하나의 횡단 토공 — `{cut, cut_compacted, fill, per_station}`(㎥ · 절 = 자연 상태).
+
+    `chainages` = 잴 누가거리(None = 20 m 측점 + 종점 · 균형 종단과 같은 틀) · `design_z` =
+    그 측점 계획고(None = 지반고 · 종단 반영 끔) · `standard` = 프로젝트 저장 표준 횡단면 설정값
+    (None = 설정 기본값) · `design_curves` = 곡선표(확폭 · 없으면 측점별 실측 반경)."""
+    at, ground, plan, rows = _areas(line, chainages, design_z, terrain, standard, design_curves)
+    weight = _end_area_weights(at)
+    cut, cut_c, fill = (float((weight * rows[:, k]).sum()) for k in range(3))
+    return {
+        "cut": round(cut, 1),
+        "cut_compacted": round(cut_c, 1),
+        "fill": round(fill, 1),
+        "per_station": [
+            {
+                "chainage_m": round(float(s), 2),
+                "ground_m": round(float(g), 3),
+                "plan_m": round(float(p), 3),
+                "cut_m2": round(float(r[0]), 4),
+                "cut_compacted_m2": round(float(r[1]), 4),
+                "fill_m2": round(float(r[2]), 4),
+            }
+            for s, g, p, r in zip(at, ground, plan, rows)
+        ],
+    }
+
+
+def cross_earthwork_delta(
+    before: list[tuple[float, float]] | dict[str, Any],
+    after: list[tuple[float, float]],
+    changed_chainages: Any,
+    terrain: Terrain | SurfaceElevationSampler,
+    standard: dict[str, Any] | None = None,
+    *,
+    before_chainages: Any = None,
+    z_before: Any = None,
+    z_after: Any = None,
+    design_curves_before: list[dict[str, Any]] | None = None,
+    design_curves_after: list[dict[str, Any]] | None = None,
+) -> dict[str, Any]:
+    """바뀐 측점만 횡단으로 잰 토공 차(새 − 옛 · ㎥) — 합치기 판정(3~6 측점).
+
+    측점 하나 = `STATION_M` 몫(가운데 측점 평균단면법 무게). `before` = 옛 노선 정점 또는 그
+    노선의 `cross_earthwork` 결과(그 측점 단면적을 다시 안 셈 · 누가거리 가장 가까운 것) ·
+    `before_chainages` = 옛 노선에서 잴 누가거리(없으면 `changed_chainages`) · `z_*` = 그 측점
+    계획고(None = 지반고). 결과 `cut` · `fill` · `earth` = 차 · `before` 가 결과면 `*_total` =
+    고친 뒤 노선 전체 어림(옛 합 + 차)."""
+    at_after = np.asarray(changed_chainages, dtype=np.float64).reshape(-1)
+    at_before = at_after if before_chainages is None else np.asarray(before_chainages, float)
+    if at_after.size == 0:
+        new = old = np.zeros((0, 3))
+    else:
+        new = _areas(after, at_after, z_after, terrain, standard, design_curves_after)[3]
+        if isinstance(before, dict):
+            rows = before["per_station"]
+            held = np.asarray([row["chainage_m"] for row in rows])
+            pick = [int(np.argmin(np.abs(held - s))) for s in at_before]
+            old = np.asarray(
+                [[rows[i]["cut_m2"], rows[i]["cut_compacted_m2"], rows[i]["fill_m2"]] for i in pick]
+            ).reshape(-1, 3)
+        else:
+            old = _areas(before, at_before, z_before, terrain, standard, design_curves_before)[3]
+    cut, cut_c, fill = (float(STATION_M * (new[:, k].sum() - old[:, k].sum())) for k in range(3))
+    out = {
+        "cut": round(cut, 1),
+        "cut_compacted": round(cut_c, 1),
+        "fill": round(fill, 1),
+        "earth": round(cut + fill, 1),
+        "stations": int(at_after.size),
+    }
+    if isinstance(before, dict):
+        out["cut_total"] = round(before["cut"] + cut, 1)
+        out["fill_total"] = round(before["fill"] + fill, 1)
+    return out
+
+
+def attach_cross_earth(
+    result: dict[str, Any],
+    terrain: Terrain | SurfaceElevationSampler,
+    standard: dict[str, Any] | None = None,
+) -> None:
+    """추천 결과 표 절 · 성 = 횡단 값(420) — 종단 켬 = 균형 종단 횡단 ㎥ · 끔 = 계획고 지반고.
+
+    띠 가늠은 `band_cut_m3` · `band_fill_m3` 로 남긴다. 맞추기 전 선(`unsnapped`)도 같이."""
+    for item in (result, result.get("unsnapped")):
+        if not item or not item.get("planned"):
+            continue
+        metrics = item["metrics"]
+        if "band_cut_m3" in metrics:
+            continue
+        profile = metrics.get("profile")
+        if profile and "error" not in profile:
+            cut, fill = float(profile["cut_m3"]), float(profile["fill_m3"])
+        else:
+            try:
+                earth = cross_earthwork(
+                    [tuple(p) for p in item["planned"]],
+                    None,
+                    None,
+                    terrain,
+                    standard,
+                    design_curves=item.get("curves") or None,
+                )
+            except ValueError:  # 지반을 못 뜬 노선 — 띠 가늠 그대로
+                continue
+            cut, fill = earth["cut"], earth["fill"]
+        metrics["band_cut_m3"] = metrics.get("cut_m3")
+        metrics["band_fill_m3"] = metrics.get("fill_m3")
+        metrics["cut_m3"] = round(cut, 1)
+        metrics["fill_m3"] = round(fill, 1)
+        metrics["surplus_m3"] = round(cut - fill, 1)

B05_Profile/B05_Profile_Engine_RouteInitial_Profile.py

@@ -85,20 +85,25 @@ def route_sections(
     vertices: list[tuple[float, float]],
     terrain: Terrain | SurfaceElevationSampler,
     design_curves: list[dict[str, Any]] | None = None,
+    at: Any = None,
 ) -> tuple[np.ndarray, np.ndarray, list[list[dict[str, Any]]], list[str], list[dict[str, Any]]]:
     """측점 누가거리 · 중심 지반고 · 측점별 지반 횡단 샘플 · 단면유형(상단측 절토) · 곡선부 확폭.
 
-    샘플은 `CROSS_WIDE_M` 까지 — ±`CROSS_HALF_M` 밖은 지표면이 끊기기 전까지만 담는다."""
+    `at` = 뜰 누가거리(없으면 20 m 측점 + 종점) · 샘플은 `CROSS_WIDE_M` 까지 — ±`CROSS_HALF_M`
+    밖은 지표면이 끊기기 전까지만 담는다."""
     xy = np.asarray(vertices, dtype=np.float64)
     keep = np.r_[True, np.hypot(*np.diff(xy, axis=0).T) > 1e-9]
     xy = xy[keep]
     seg = np.hypot(*np.diff(xy, axis=0).T)
     chain = np.concatenate([[0.0], np.cumsum(seg)])
     total = float(chain[-1])
-    at = np.arange(0.0, total, STATION_M)
-    if len(at) > 1 and total - at[-1] < TAIL_MERGE_M:
-        at = at[:-1]
-    at = np.append(at, total)
+    if at is not None:
+        at = np.clip(np.asarray(at, dtype=np.float64), 0.0, total)
+    else:
+        at = np.arange(0.0, total, STATION_M)
+        if len(at) > 1 and total - at[-1] < TAIL_MERGE_M:
+            at = at[:-1]
+        at = np.append(at, total)
     index = np.clip(np.searchsorted(chain, at, side="right") - 1, 0, len(seg) - 1)
     tangent = np.diff(xy, axis=0)[index] / seg[index][:, None]
     left = np.column_stack([-tangent[:, 1], tangent[:, 0]])
@@ -165,10 +170,15 @@ def _widenings(
 
 
 def _station_areas(
-    samples: list[dict[str, Any]], mode: str, elevation: float, widen: dict[str, Any]
+    samples: list[dict[str, Any]],
+    mode: str,
+    elevation: float,
+    widen: dict[str, Any],
+    standard: dict[str, Any] | None = None,
 ) -> tuple[float, float, float]:
     """측점 하나 · 계획고 하나의 (절토 자연 ㎡, 절토 다짐 환산 ㎡, 성토 ㎡) — 본 설계 기본 설계와
-    같은 입력. 반폭에서 비탈이 안 닫히면 넓힌 샘플로 다시 셈(본 설계 `extend_unclosed`)."""
+    같은 입력. 반폭에서 비탈이 안 닫히면 넓힌 샘플로 다시 셈(본 설계 `extend_unclosed`).
+    `standard` = 프로젝트에 저장된 표준 횡단면 설정값(없으면 설정 기본값 · 420)."""
     from B06_Section.B06_Section_Engine_Design import compute_cross_design
 
     def design(pick: list[dict[str, Any]]) -> dict[str, Any]:
@@ -177,6 +187,7 @@ def _station_areas(
             elevation,
             ground_type=GROUND_TYPE,
             section_mode=mode,
+            standard=standard,
             rock_boundary_offset_m=STANDARD_ROCK_BOUNDARY_DEFAULT_OFFSET_M,
             **widen,
         )
@@ -204,6 +215,7 @@ def _area_curves(
     ground: np.ndarray,
     levels: np.ndarray,
     curves: list[dict[str, Any]] | None = None,
+    standard: dict[str, Any] | None = None,
 ) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
     """측점 · 계획고 후보마다 (절토 자연 ㎡, 절토 다짐 ㎡, 성토 ㎡) — `AREA_STEP_M` 마디 보간."""
     shape = levels.shape
@@ -213,7 +225,9 @@ def _area_curves(
         delta = levels[i] - ground[i]
         lo = np.floor(delta.min() / AREA_STEP_M) * AREA_STEP_M
         at = np.arange(lo, delta.max() + AREA_STEP_M, AREA_STEP_M)
-        table = np.asarray([_station_areas(samples, mode, float(ground[i] + d), widen) for d in at])
+        table = np.asarray(
+            [_station_areas(samples, mode, float(ground[i] + d), widen, standard) for d in at]
+        )
         # 단면적은 계획고에 대해 볼록(대략 2차)이라 직선 보간은 크게 잡힌다 — 단조 3차(PCHIP)
         if len(at) > 1:
             curve = PchipInterpolator(at, table, axis=0)(delta)
@@ -277,11 +291,13 @@ def balanced_profile(
     design_speed: int = 20,
     curves: list[dict[str, Any]] | None = None,
     exact: bool = True,
+    standard: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """상한(종단기울기 · 종단곡선)을 지키며 절 · 성토를 횡단 ㎥ 로 맞춘 계획선과 토량.
 
     시점 · 종점은 지반고(확정 체인 기본 오프셋 0 과 같음) — 그 길이 없을 때만 풀어 다시 찾는다.
     `exact` 끄면 보간값 그대로(후보 줄 세우기 — 빠름) · 켜면 고른 선을 바로 셈해 누가토량을 맞춘다.
+    `standard` = 프로젝트 저장 표준 횡단면 설정값(없으면 설정 기본값).
     """
     started = time.perf_counter()
     spans = np.diff(chainage)
@@ -291,7 +307,7 @@ def balanced_profile(
     offsets = np.arange(-LEVEL_RANGE_M, LEVEL_RANGE_M + 1e-9, LEVEL_STEP_M)
     levels = _reference(ground, spans, max_grade)[:, None] + offsets[None, :]
     cut, cut_c, fill = (
-        weight[:, None] * a for a in _area_curves(sections, modes, ground, levels, curves)
+        weight[:, None] * a for a in _area_curves(sections, modes, ground, levels, curves, standard)
     )
     criteria = (
         FOREST_ROAD_PROFILE_CRITERIA["design_speed"].get(design_speed)
@@ -312,7 +328,9 @@ def balanced_profile(
         plan = levels[np.arange(len(chainage)), result["path"]]
         rows = np.asarray(
             [
-                _station_areas(sections[i], modes[i], float(plan[i]), curves[i] if curves else {})
+                _station_areas(
+                    sections[i], modes[i], float(plan[i]), curves[i] if curves else {}, standard
+                )
                 for i in range(len(chainage))
             ]
         )
@@ -423,6 +441,7 @@ def route_profile(
     design_speed: int,
     design_curves: list[dict[str, Any]] | None = None,
     exact: bool = True,
+    standard: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """후보 노선 하나의 균형 종단 — 측점 횡단을 뜨고 `balanced_profile`.
 
@@ -439,6 +458,7 @@ def route_profile(
         design_speed=design_speed,
         curves=curves,
         exact=exact,
+        standard=standard,
     )
 
 
@@ -496,6 +516,7 @@ def attach_profile(
     design_speed: int,
     design_curves: list[dict[str, Any]] | None = None,
     exact: bool = True,
+    standard: dict[str, Any] | None = None,
 ) -> None:
     """후보 비교값에 균형 종단을 붙이고 토공 점수를 그것으로 바꾼다(묶음 캐시에 남음).
 
@@ -513,6 +534,7 @@ def attach_profile(
             design_speed=design_speed,
             design_curves=design_curves,
             exact=exact,
+            standard=standard,
         )
     except ValueError as error:
         metrics["profile"] = {"error": str(error)}

B05_Profile/B05_Profile_Engine_RouteInitial_Store.py

@@ -8,14 +8,16 @@
 [초기화]에 안 휩쓸리고, 자료 교체(`purge_output_folders`) · 프로젝트 삭제 때 폴더째 지워진다.
 
 키 = 후보 묶음 지문(`_pool_key` — 지형 · 예상노선 · 계류 · 상한 · 최소 R · 이격 · 고름 · 목표
-기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · `ALGO_VERSION`. 관 지점은 뺀다 — 덮개 수는 꺼낼 때
-다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을 올린다(옛 파일은 다른 키라 안 쓰임).
+기울기) + 갈래 · 비중 · 종단 반영 · 설계속도 · 표준 횡단면 · `ALGO_VERSION` · 엔진 소스 지문. 관
+지점은 뺀다 — 덮개 수는 꺼낼 때 다시 센다. 추천 결과를 바꾸는 코드 고침은 `ALGO_VERSION` 을
+올린다 — 빠뜨려도 엔진 소스 지문이 바뀌어 새로 셈. 옛 판 · 옛 지문 파일은 미리 셈 때 지운다(420).
 기본값 · 저장된 제약 결과(`default`)는 「최근 벌」을 따로 센다(351-1).
 """
 
 from __future__ import annotations
 
 import asyncio
+import functools
 import hashlib
 import json
 from contextlib import asynccontextmanager
@@ -33,8 +35,16 @@ from B05_Profile.B05_Profile_Engine_RouteInitial import (
 from common_util.common_util_json import atomic_write_json
 
 # 3 = 곡선 MC 일반측점(351-2) · 4 = 다투는 곡선 끔(364-5) · 5 = 후보마다 맞춤(383-1) ·
-# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4)
-ALGO_VERSION = 7
+# 6 = 같은 쪽 곡선 합치기 · 거리 한도 대신 직접 잼(406) · 7 = 측점 밖이 늘면 옛 차례 맞춤(406-4) ·
+# 8 = 횡단 토공 · 표준 횡단면 = 저장 패널값 · 엔진 소스 지문(420)
+ALGO_VERSION = 8
+# 엔진 소스 지문에 드는 파일(저장소 뿌리 기준) — 추천 결과를 낳는 코드
+SOURCE_GLOBS = (
+    "B05_Profile/B05_Profile_Engine_RouteInitial*.py",
+    "B05_Profile/B05_Profile_Engine_Sections_Core.py",
+    "B06_Section/B06_Section_Engine_Design*.py",
+    "common_util/common_util_route_polyline.py",
+)
 STORE_DIR = "B05_Profile/route_recommend"
 # 종류(갈래 · 지형) 마다 남길 벌 수 — 꺼낼 때 수정시각을 새로 해 최근에 쓴 것이 남음
 KEEP_PER_KIND = 2
@@ -43,12 +53,25 @@ DEFAULT_MODES = ("gradeline", "earthwork", "gentle", "weighted", "follow")
 
 # 묶음 지문마다 잠금 — 진행 중 같은 묶음(같은 키 포함)은 기다렸다 꺼냄(두 번 셈 막음)
 _LOCKS: dict[str, asyncio.Lock] = {}
+# 묶음 지문마다 그 묶음을 짤 때 표준 횡단면(`forget_pool_on_standard`)
+_POOL_STANDARDS: dict[str, str] = {}
 
 
 def store_dir(project_root: Path) -> Path:
     return Path(project_root) / STORE_DIR
 
 
+@functools.cache
+def source_fingerprint() -> str:
+    """엔진 소스 지문 — `SOURCE_GLOBS` 파일 내용(줄끝 LF 로 맞춤 · 두 PC 같은 코드 = 같은 지문)."""
+    root = Path(__file__).resolve().parents[1]
+    digest = hashlib.blake2b(digest_size=8)
+    for path in sorted({p for pattern in SOURCE_GLOBS for p in root.glob(pattern)}):
+        digest.update(path.name.encode())
+        digest.update(path.read_bytes().replace(b"\r\n", b"\n"))
+    return digest.hexdigest()
+
+
 def request_fingerprint(
     expected: list[tuple[float, float]],
     terrain: Terrain,
@@ -63,6 +86,7 @@ def request_fingerprint(
     target_grade: float | None,
     with_profile: bool,
     design_speed: int,
+    standard: dict[str, Any] | None = None,
 ) -> tuple[str, str]:
     """(저장 키, 묶음 지문) — `generate_initial_route` 와 같은 기본값 풀이로 같은 입력 = 같은 키."""
     pool = _pool_key(
@@ -74,7 +98,15 @@ def request_fingerprint(
         *resolve_inputs(max_grade, stream_offset_m, grade_smooth_m, target_grade),
     ).hex()
     used = resolve_weights(mode, weights)
-    tail = (mode, sorted(used.items()), bool(with_profile), int(design_speed), ALGO_VERSION)
+    tail = (
+        mode,
+        sorted(used.items()),
+        bool(with_profile),
+        int(design_speed),
+        json.dumps(standard, sort_keys=True, default=str) if standard else None,
+        ALGO_VERSION,
+        source_fingerprint(),
+    )
     key = hashlib.blake2b((pool + repr(tail)).encode(), digest_size=16).hexdigest()
     return key, pool
 
@@ -102,14 +134,22 @@ def is_default(
     )
 
 
+def _current(data: Any) -> bool:
+    return (
+        isinstance(data, dict)
+        and data.get("version") == ALGO_VERSION
+        and data.get("source") == source_fingerprint()
+    )
+
+
 def read_stored(project_root: Path, key: str) -> dict[str, Any] | None:
-    """저장본 한 벌 — 없거나 못 읽거나 판이 다르면 None."""
+    """저장본 한 벌 — 없거나 못 읽거나 판 · 소스 지문이 다르면 None."""
     path = store_dir(project_root) / 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("version") != ALGO_VERSION:
+    if not _current(data):
         return None
     try:
         path.touch()  # 최근에 쓴 것 = 정리 때 남김
@@ -134,6 +174,7 @@ def write_stored(
         path,
         {
             "version": ALGO_VERSION,
+            "source": source_fingerprint(),
             "key": key,
             "pool": pool,
             "default": default,
@@ -171,6 +212,45 @@ def prune_stored(project_root: Path, request: dict[str, Any], default: bool = Fa
     return max(0, len(same) - KEEP_PER_KIND)
 
 
+def _remove_where(project_root: Path, drop: Any) -> int:
+    """저장본 중 `drop(data)` 가 참인 파일을 지운다(못 읽는 파일도 지움) — 지운 수."""
+    removed = 0
+    for path in store_dir(project_root).glob("*.json"):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError):
+            data = None
+        if data is None or drop(data):
+            path.unlink(missing_ok=True)
+            removed += 1
+    return removed
+
+
+def purge_outdated(project_root: Path) -> int:
+    """옛 판 · 옛 소스 지문 저장본을 지운다(미리 셈 첫머리 · 420) — 지운 수."""
+    return _remove_where(project_root, lambda data: not _current(data))
+
+
+def drop_pool_results(project_root: Path, pool: str) -> int:
+    """[재계산] — 이 후보 묶음(다섯 갈래 · 종단 켬끔)의 저장본을 지우고 메모리 묶음도 버린다.
+
+    `pool_lock(pool)` 안에서 부름(미리 셈이 다시 써 버리지 않게). 지운 수."""
+    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS
+
+    _POOLS.pop(bytes.fromhex(pool), None)
+    return _remove_where(project_root, lambda data: data.get("pool") == pool)
+
+
+def forget_pool_on_standard(pool: str, standard: dict[str, Any] | None) -> None:
+    """메모리 묶음 후보에 붙은 균형 종단은 그때 표준 횡단면으로 셈 — 바뀌면 묶음을 버린다."""
+    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS
+
+    mark = json.dumps(standard, sort_keys=True, default=str)
+    if _POOL_STANDARDS.get(pool, mark) != mark:
+        _POOLS.pop(bytes.fromhex(pool), None)
+    _POOL_STANDARDS[pool] = mark
+
+
 @asynccontextmanager
 async def pool_lock(pool: str) -> AsyncIterator[None]:
     # ponytail: 잠금은 지우지 않음 — 입력 묶음 수만큼(작음) · 다중 프로세스면 파일 잠금으로

B05_Profile/B05_Profile_Router_Replan.py

@@ -627,6 +627,7 @@ async def replan_route(project_id: UUID, body: RouteSaveRequest) -> dict[str, An
             choice.mode,
             choice.weights.model_dump() if choice.weights else None,
             (choice.criteria.model_dump(exclude_none=True) or None) if choice.criteria else None,
+            request.balanced_profile,  # 「종단 반영」 — 다시 열면 되살림(420)
         )
         if choice
         else await asyncio.to_thread(read_initial_choice, project_root)

B05_Profile/B05_Profile_Router_RouteInitial.py

@@ -6,6 +6,7 @@
 
 import asyncio
 import functools
+import json
 import logging
 import time
 from pathlib import Path
@@ -44,17 +45,21 @@ async def compute_initial(
     """같은 입력이면 저장본을 꺼내고 없으면 셈해 더한다(PLAN 327) · 덮개 수는 늘 지금 관으로.
 
     응답 `stored` = 저장본에서 꺼냈나. 진행 중인 같은 묶음은 기다렸다 꺼낸다(두 번 셈 막음).
+    `recompute` = [재계산] — 묶음 잠금 안에서 이 묶음 저장본 · 메모리 묶음을 버리고 새로 셈(420).
     """
     from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
     from B05_Profile.B05_Profile_Engine_RouteInitial import (
         generate_initial_route,
         load_terrain_and_streams,
     )
+    from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import attach_cross_earth
     from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import (
         cover_shortfall,
         project_pipes,
     )
     from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
+        drop_pool_results,
+        forget_pool_on_standard,
         is_default,
         pool_lock,
         read_stored,
@@ -115,6 +120,8 @@ async def compute_initial(
         "target_grade": None if target_pct is None else target_pct / 100.0,
         "design_speed": design_speed or 20,
         "with_profile": request.with_profile,
+        # 균형 종단 · 횡단 토공 표준 횡단면 = 프로젝트 저장 패널값(420)
+        "standard": await project_standard(project_id),
     }
     started = time.perf_counter()
     try:
@@ -126,6 +133,10 @@ async def compute_initial(
             functools.partial(request_fingerprint, route, terrain, streams, **inputs)
         )
         async with pool_lock(pool_key):
+            if request.recompute:  # 잠금 안 — 밖이면 미리 셈이 지운 것을 다시 써 버림
+                removed = await asyncio.to_thread(drop_pool_results, project_root, pool_key)
+                logger.info("추천 [재계산]: project_id=%s 저장본 %d 지움", project_id, removed)
+            forget_pool_on_standard(pool_key, inputs["standard"])
             stored = await asyncio.to_thread(read_stored, project_root, key)
             if stored is not None:
                 result = stored["result"]
@@ -136,12 +147,14 @@ async def compute_initial(
                         generate_initial_route, route, terrain, streams, pipes=None, **inputs
                     )
                 )
+                # 결과 표 절 · 성 = 횡단(420) — 저장본에 남김
+                await asyncio.to_thread(attach_cross_earth, result, terrain, inputs["standard"])
                 await asyncio.to_thread(
                     write_stored,
                     project_root,
                     key,
                     pool_key,
-                    request.model_dump(),
+                    request.model_dump(exclude={"recompute"}),
                     result,
                     is_default(
                         request.mode,
@@ -180,13 +193,36 @@ async def compute_initial(
     }
 
 
+async def project_standard(project_id: UUID) -> dict[str, Any] | None:
+    """프로젝트에 저장된 표준 횡단면 설정값(B06 패널 · 확정 때 종단 options) — 없으면 None.
+
+    공식 재계산(`B06_Section_Server_Calc_Prebuild` · `stored_standard_cross_section`)과 같은
+    자리 · 같은 읽기 — 마지막 종단 정본 행."""
+    from B06_Section.B06_Section_Router_Design import stored_standard_cross_section
+
+    pool = get_db_pool()
+    async with pool.acquire() as connection:
+        async with connection.cursor() as cursor:
+            await cursor.execute(
+                "SELECT data FROM longitudinal_sections WHERE project_id = %s "
+                "ORDER BY id DESC LIMIT 1",
+                (str(project_id),),
+            )
+            row = await cursor.fetchone()
+    data = row[0] if row else None
+    if isinstance(data, str):
+        data = json.loads(data)
+    return stored_standard_cross_section({"data": data if isinstance(data, dict) else None})
+
+
 async def precompute_defaults(project_id: UUID) -> None:
     """체인 끝 예약(PLAN 327-2 · 351-1) — 기본 다섯 알약 + 저장된 제약(initial_choice)의 다섯.
 
     모달은 저장된 제약 · 비중으로 열리므로 그 값으로도 미리 셈한다. 있으면 꺼내기만.
+    다섯 갈래 모두 종단 켬(420) · 첫머리에 옛 판 · 옛 소스 지문 저장본을 지운다.
     """
     from B05_Profile.B05_Profile_Engine_RouteInitial import DEFAULT_WEIGHTS
-    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES
+    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import DEFAULT_MODES, purge_outdated
 
     started = time.perf_counter()
     variants: list[tuple[Any, Any]] = [(None, None)]
@@ -196,6 +232,9 @@ async def precompute_defaults(project_id: UUID) -> None:
     except LookupError:
         paths = None
     if paths is not None:
+        removed = await asyncio.to_thread(purge_outdated, paths[0])
+        if removed:
+            logger.info("추천 옛 판 저장본 지움: project_id=%s %d", project_id, removed)
         saved = await asyncio.to_thread(read_initial_choice, paths[0])
         weights = saved["weights"] if saved["logic"] == "weighted" else None
         if saved["criteria"] or (weights and weights != DEFAULT_WEIGHTS):
@@ -209,10 +248,10 @@ async def precompute_defaults(project_id: UUID) -> None:
         for mode in modes:
             if progress is not None:
                 progress["mode"] = mode
-            # 화면과 같은 몸통 — 설계 지형 · 비중은 비중 갈래만 · 추적만 종단 끔(`withProfileOf`)
+            # 화면과 같은 몸통 — 설계 지형 · 비중은 비중 갈래만 · 「종단 반영」 기본 켬(420)
             request = replan.RouteInitialRequest(
                 mode=mode,
-                with_profile=mode != "follow",
+                with_profile=True,
                 criteria=criteria,
                 weights=weights if mode == "weighted" else None,
             )
@@ -416,6 +455,7 @@ async def balanced_profile_of(
     pool = get_db_pool()
     async with pool.acquire() as connection:
         selection = await get_surface_confirmation_params(connection, str(project_id))
+    standard = await project_standard(project_id)  # 추천 창과 같은 저장 패널값(420)
     try:
         # 측점 생성과 같은 1 m 쌍선형 표본(추천 점수의 2 m 비용면 아님)
         surface = await asyncio.to_thread(
@@ -433,6 +473,7 @@ async def balanced_profile_of(
                 max_grade=max_grade,
                 design_speed=design_speed or 20,
                 design_curves=design_curves,
+                standard=standard,
             )
         )
     except (FileNotFoundError, ValueError):

B05_Profile/B05_Profile_Schema.py

@@ -367,6 +367,8 @@ class RouteInitialRequest(BaseModel):
     with_profile: bool = True
     # 분석 지형(PLAN 133-1) — None = 지금 설계 지형 · [확인] 때 설계 지형이 됨(133-2).
     terrain: Literal["las", "sheet"] | None = None
+    # [재계산](420) — 이 후보 묶음 저장본 · 메모리 묶음을 지우고 새로 셈. 저장본 request 엔 안 적힘.
+    recompute: bool = False
 
 
 class RouteReplanRequest(BaseModel):

resources/tester/test_224_3_recommend_store.py

@@ -96,7 +96,16 @@ def test_no_overwrite(tmp_path: Path) -> None:
     assert not write_stored(tmp_path, "k", "p", {}, {"n": 2}, False)
     data = json.loads((store_dir(tmp_path) / "k.json").read_text(encoding="utf-8"))
     assert data["result"] == {"n": 1} and data["default"] is True
-    assert set(data) == {"version", "key", "pool", "default", "created_at", "request", "result"}
+    assert set(data) == {
+        "version",
+        "source",
+        "key",
+        "pool",
+        "default",
+        "created_at",
+        "request",
+        "result",
+    }
 
 
 def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None:
@@ -117,6 +126,8 @@ def test_survives_reset_and_goes_with_data_swap(tmp_path: Path) -> None:
 class _Calls:
     def __init__(self) -> None:
         self.count = 0
+        self.inputs: list[dict] = []
+        self.standard: dict | None = None
 
 
 @pytest.fixture
@@ -141,20 +152,26 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
     async def surface_params(_connection, _pid):
         return {}
 
+    async def project_standard(_pid):
+        return calls.standard
+
     class _Pool:
         @asynccontextmanager
         async def acquire(self):
             yield None
 
-    def generate(expected, terrain, streams, *, pipes, **_inputs):
+    def generate(expected, terrain, streams, *, pipes, **inputs):
         assert pipes is None  # 관은 키 밖 — 덮개 없이 셈해 남김
         calls.count += 1
+        calls.inputs.append(inputs)
         return {
             "planned": [[x, y] for x, y in expected],
             "metrics": {
                 "length_m": 38.0,
                 "offset_avg_m": 0.0,
                 "profile": {
+                    "cut_m3": 10.0,
+                    "fill_m3": 4.0,
                     "stations_m": [0.0, 38.0],
                     "plan_m": [1.0, 1.0],
                     "ground_m": [0.0, 0.0],
@@ -167,6 +184,7 @@ def wired(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> _Calls:
     monkeypatch.setattr(replan, "_road_settings", road_settings)
     monkeypatch.setattr(replan, "_vertices_of", lambda _path: [list(p) for p in EXPECTED])
     monkeypatch.setattr(router, "get_db_pool", lambda: _Pool())
+    monkeypatch.setattr(router, "project_standard", project_standard)
     monkeypatch.setattr(surface, "get_surface_confirmation_params", surface_params)
     monkeypatch.setattr(surface, "analysis_terrain_params", lambda selection, _t: selection)
     monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
@@ -225,7 +243,8 @@ def test_precompute_defaults_then_hits(wired: _Calls) -> None:
     ]
     assert sorted(item["request"]["mode"] for item in saved) == sorted(DEFAULT_MODES)
     assert all(item["default"] for item in saved)
-    # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 추적은 종단 끔) — 셈 없이 저장본
+    # 화면 첫 누름(비중은 화면이 50/25/25 를 채워 보냄 · 다섯 갈래 종단 켬 · 420) — 셈 없이 저장본
+    assert all(item["request"]["with_profile"] for item in saved)
     hit = _ask(mode="weighted", weights={"expected": 50, "earthwork": 25, "grade": 25})
-    follow = _ask(mode="follow", with_profile=False)
+    follow = _ask(mode="follow")
     assert hit["stored"] and follow["stored"] and wired.count == 5

resources/tester/test_364_4_precompute_on_open.py

@@ -90,6 +90,7 @@ def test_모달이_여는_갈래_먼저(monkeypatch) -> None:
 
     async def compute(_pid, request):
         asked.append(request.mode)
+        assert request.with_profile  # 다섯 갈래 모두 종단 켬(420)
         return {}
 
     monkeypatch.setattr(router.replan, "_project_paths", paths)

resources/tester/test_420_1_cross_earth.py

@@ -0,0 +1,243 @@
+"""420 추천 횡단 토공 · 저장본 판 · [재계산] 서버(요구 254).
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_420_1_cross_earth.py -q
+
+횡단 토공 = 공식 횡단 단면적 평균단면법(저장 정본 ±6 %) · 차 = 바뀐 측점만 · 결과 표 절 · 성 =
+횡단 · [재계산] = 잠금 안 지우고 새로 셈(셈 수 1 → 2) · 옛 판 · 옛 소스 지문 미리 셈 때 지움 ·
+표준 횡단면 = 저장 패널값(키 · 엔진에 감) · `initial_choice` with_profile 되살림.
+"""
+
+import asyncio
+import json
+from pathlib import Path
+from uuid import uuid4
+
+import numpy as np
+import pytest
+from test_224_3_recommend_store import EXPECTED, STREAMS, _ask, _key, _terrain, wired  # noqa: F401
+
+from B05_Profile import B05_Profile_Router_RouteInitial as router
+from B05_Profile.B05_Profile_Engine_RouteInitial import (
+    Terrain,
+    read_initial_choice,
+    write_initial_choice,
+)
+from B05_Profile.B05_Profile_Engine_RouteInitial_CrossEarth import (
+    attach_cross_earth,
+    cross_earthwork,
+    cross_earthwork_delta,
+)
+from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import _station_areas, route_sections
+from B05_Profile.B05_Profile_Engine_RouteInitial_Store import (
+    ALGO_VERSION,
+    DEFAULT_MODES,
+    source_fingerprint,
+    store_dir,
+)
+
+ROOT = Path(__file__).resolve().parents[2]
+SUB2 = ROOT / "storage/1/3/cb15892a-5788-4a26-9316-20a15421f68b"
+
+
+def _slope() -> Terrain:
+    """옆경사 30 % 사면(노선 = x 방향) — 절 · 성 둘 다 생김."""
+    x = np.arange(0.0, 200.0, 2.0)
+    y = np.arange(-80.0, 80.0, 2.0)
+    xx, yy = np.meshgrid(x, y)
+    return Terrain(x, y, 0.05 * xx + 0.3 * yy, np.ones(xx.shape, dtype=bool))
+
+
+LINE = [(20.0, 0.0), (100.0, 0.0), (180.0, 0.0)]
+
+
+def test_횡단_토공_평균단면법() -> None:
+    terrain = _slope()
+    earth = cross_earthwork(LINE, None, None, terrain)
+    rows = earth["per_station"]
+    assert [r["chainage_m"] for r in rows] == [float(s) for s in range(0, 161, 20)]
+    # 직접 셈한 단면적 · Σ (A_i + A_i+1) / 2 · L
+    at, ground, sections, modes, widen = route_sections(LINE, terrain)
+    areas = [_station_areas(sections[i], modes[i], ground[i], widen[i]) for i in range(len(at))]
+    cut = sum((areas[i][0] + areas[i + 1][0]) / 2 * 20.0 for i in range(len(at) - 1))
+    fill = sum((areas[i][2] + areas[i + 1][2]) / 2 * 20.0 for i in range(len(at) - 1))
+    assert earth["cut"] == pytest.approx(cut, abs=0.1) and earth["cut"] > 0
+    assert earth["fill"] == pytest.approx(fill, abs=0.1) and earth["fill"] > 0
+    # 계획고를 1 m 올리면 성토가 늚 · 측점 수와 다른 계획고는 거절
+    up = cross_earthwork(LINE, None, np.asarray(ground) + 1.0, terrain)
+    assert up["fill"] > earth["fill"] and up["cut"] < earth["cut"]
+    with pytest.raises(ValueError):
+        cross_earthwork(LINE, None, [1.0, 2.0], terrain)
+
+
+def test_표준_횡단면이_단면적을_바꿈() -> None:
+    from config.config_system_design import STANDARD_CROSS_SECTION
+
+    terrain = _slope()
+    wide = json.loads(json.dumps(STANDARD_CROSS_SECTION))
+    for group in wide.values():
+        group["road_width_m"] += 2.0
+    base = cross_earthwork(LINE, None, None, terrain)
+    assert cross_earthwork(LINE, None, None, terrain, wide)["cut"] > base["cut"]
+
+
+def test_토공_차_바뀐_측점만() -> None:
+    terrain = _slope()
+    moved = [(20.0, 0.0), (100.0, 6.0), (180.0, 0.0)]
+    changed = [60.0, 80.0, 100.0]
+    same = cross_earthwork_delta(LINE, LINE, changed, terrain)
+    assert same["cut"] == 0 and same["fill"] == 0 and same["stations"] == 3
+    delta = cross_earthwork_delta(LINE, moved, changed, terrain)
+    old = cross_earthwork(LINE, changed, None, terrain)["per_station"]
+    new = cross_earthwork(moved, changed, None, terrain)["per_station"]
+    expect = 20.0 * sum(n["fill_m2"] - o["fill_m2"] for n, o in zip(new, old))
+    assert delta["fill"] == pytest.approx(expect, abs=0.2)
+    # 옛 결과를 넘기면 옛 단면적을 다시 안 셈 · 전체 어림 = 옛 합 + 차
+    whole = cross_earthwork(LINE, None, None, terrain)
+    reuse = cross_earthwork_delta(whole, moved, changed, terrain)
+    assert reuse["fill"] == pytest.approx(delta["fill"], abs=0.2)
+    assert reuse["fill_total"] == pytest.approx(whole["fill"] + reuse["fill"], abs=0.2)
+    assert cross_earthwork_delta(LINE, moved, [], terrain)["earth"] == 0
+
+
+@pytest.mark.skipif(not (SUB2 / "B05_Profile").is_dir(), reason="검증_sub2 storage 없음")
+def test_검증_sub2_저장_정본_6퍼센트() -> None:
+    """저장된 노선 측점 · 계획고 = 지반 — 공식 [저장] 단면적 합과 ±6 %(2 m 추천 지형 · 1 m 표본)."""
+    from B05_Profile.B05_Profile_Engine_RouteInitial import load_terrain_and_streams
+    from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR
+    from common_util.common_util_route_geometry import read_planned_route_csv
+    from common_util.common_util_surface_sampler import build_surface_sampler
+
+    route = read_planned_route_csv(SUB2 / "B05_Profile/route/planned_route_initial.csv")
+    curves = json.loads(
+        (SUB2 / "B05_Profile/route/planned_route_initial_curves.json").read_text("utf-8")
+    )
+    curves = curves.get("curves") if isinstance(curves, dict) else curves
+    line = [(v.x, v.y) for v in route.vertices]
+    files = sorted((SUB2 / "B06_Section/cross_sections").glob("cross_*.json"))
+    at = [json.loads(p.read_text("utf-8"))["chainage_m"] for p in files]
+    # 공식 [저장] 단면적(DB cross_sections.design · 2026-10-18 뜸 · 계획고 = 지반) 평균단면법 합
+    stored_cut, stored_fill = 5166.0, 23079.3
+    selection = {"source_filter": "classification", "method": "dtm", "smooth": True}
+    terrain, _ = load_terrain_and_streams(SUB2, selection)
+    surface = build_surface_sampler(SUB2 / _MODELS_SUBDIR, "classification", "dtm", True)
+    for ground in (terrain, surface):
+        earth = cross_earthwork(line, at, None, ground, design_curves=curves)
+        assert earth["cut"] == pytest.approx(stored_cut, rel=0.06)
+        assert earth["fill"] == pytest.approx(stored_fill, rel=0.06)
+
+
+def test_결과_절성_횡단_값() -> None:
+    terrain = _slope()
+    plain = {"planned": [list(p) for p in LINE], "curves": [], "metrics": {"cut_m3": 1.0}}
+    profiled = {
+        "planned": [list(p) for p in LINE],
+        "metrics": {"cut_m3": 1.0, "fill_m3": 2.0, "profile": {"cut_m3": 30.0, "fill_m3": 20.0}},
+    }
+    result = {**plain, "unsnapped": profiled}
+    attach_cross_earth(result, terrain)
+    earth = cross_earthwork(LINE, None, None, terrain)
+    assert result["metrics"]["cut_m3"] == earth["cut"] and result["metrics"]["band_cut_m3"] == 1.0
+    assert result["metrics"]["surplus_m3"] == pytest.approx(earth["cut"] - earth["fill"], abs=0.2)
+    shown = result["unsnapped"]["metrics"]
+    assert (shown["cut_m3"], shown["fill_m3"], shown["band_fill_m3"]) == (30.0, 20.0, 2.0)
+    attach_cross_earth(result, terrain)  # 두 번 붙여도 띠 가늠을 안 덮음
+    assert result["metrics"]["band_cut_m3"] == 1.0
+
+
+def test_표준_횡단면_키_엔진에_감(wired) -> None:  # noqa: F811
+    assert _key(standard={"a": 1}) != _key() and _key(standard={"a": 1}) == _key(standard={"a": 1})
+    wired.standard = {"road": {"width_m": 5.0}}
+    _ask(mode="earthwork")
+    assert wired.inputs[-1]["standard"] == wired.standard
+
+
+def test_재계산_잠금_안_지우고_새로(wired) -> None:  # noqa: F811
+    first = _ask(mode="earthwork")
+    _ask(mode="gentle")
+    _ask(mode="earthwork", criteria={"min_radius_m": 15.0})  # 다른 묶음
+    assert wired.count == 3 and first["stored"] is False
+    files = {p.name: p.read_text("utf-8") for p in store_dir(wired.root).glob("*.json")}
+    again = _ask(mode="earthwork", recompute=True)
+    assert again["stored"] is False and wired.count == 4  # 셈 1 → 2(earthwork)
+    now = {p.name: json.loads(p.read_text("utf-8")) for p in store_dir(wired.root).glob("*.json")}
+    # 같은 묶음 gentle 은 지워짐 · 다른 묶음은 그대로 · earthwork 파일은 새로 씀
+    assert len(now) == 2
+    other = [n for n, d in now.items() if d["request"]["criteria"]]
+    assert other and now[other[0]] == json.loads(files[other[0]])
+    fresh = next(d for n, d in now.items() if not d["request"]["criteria"])
+    assert "recompute" not in fresh["request"] and fresh["version"] == ALGO_VERSION
+    assert fresh["result"]["metrics"]["cut_m3"] == 10.0  # 횡단(균형 종단) 값
+    assert _ask(mode="earthwork")["stored"] and wired.count == 4
+
+
+def test_재계산_메모리_묶음도_버림(wired) -> None:  # noqa: F811
+    from B05_Profile.B05_Profile_Engine_RouteInitial import _POOLS
+    from B05_Profile.B05_Profile_Engine_RouteInitial_Store import request_fingerprint
+
+    _ask(mode="earthwork")
+    pool = request_fingerprint(
+        EXPECTED,
+        _terrain(),
+        STREAMS,
+        mode="earthwork",
+        weights=None,
+        max_grade=0.14,
+        min_radius_m=12.0,
+        stream_offset_m=None,
+        grade_smooth_m=None,
+        target_grade=None,
+        with_profile=True,
+        design_speed=20,
+    )[1]
+    _POOLS[bytes.fromhex(pool)] = "옛 묶음"
+    _ask(mode="earthwork")
+    assert bytes.fromhex(pool) in _POOLS
+    _ask(mode="earthwork", recompute=True)
+    assert bytes.fromhex(pool) not in _POOLS
+
+
+def test_옛_판_옛_지문_미리_셈_때_지움(wired) -> None:  # noqa: F811
+    folder = store_dir(wired.root)
+    folder.mkdir(parents=True)
+    old = {"version": ALGO_VERSION - 1, "source": source_fingerprint(), "request": {}}
+    stale = {"version": ALGO_VERSION, "source": "0" * 16, "request": {}}
+    (folder / "old.json").write_text(json.dumps(old), encoding="utf-8")
+    (folder / "stale.json").write_text(json.dumps(stale), encoding="utf-8")
+    (folder / "broken.json").write_text("{", encoding="utf-8")
+    asyncio.run(router.precompute_defaults(uuid4()))
+    names = {p.stem for p in folder.glob("*.json")}
+    assert not names & {"old", "stale", "broken"} and len(names) == len(DEFAULT_MODES)
+    assert wired.count == len(DEFAULT_MODES)
+
+
+def test_소스_지문이_바뀌면_다른_키(monkeypatch: pytest.MonkeyPatch) -> None:
+    import B05_Profile.B05_Profile_Engine_RouteInitial_Store as store
+
+    before = _key()
+    monkeypatch.setattr(store, "source_fingerprint", lambda: "f" * 16)
+    assert _key() != before
+
+
+def test_with_profile_되살림(tmp_path: Path) -> None:
+    assert read_initial_choice(tmp_path)["with_profile"] is True  # 없으면 켬
+    write_initial_choice(tmp_path, "gentle", None, None, False)
+    assert read_initial_choice(tmp_path)["with_profile"] is False
+    write_initial_choice(tmp_path, "gentle", None, None)
+    assert read_initial_choice(tmp_path)["with_profile"] is True
+    # 옛 파일(칸 없음) = 켬
+    path = tmp_path / "B05_Profile/route/initial_choice.json"
+    path.write_text(json.dumps({"logic": "gentle"}), encoding="utf-8")
+    assert read_initial_choice(tmp_path)["with_profile"] is True
+
+
+def test_적용이_종단_반영을_적음_소스() -> None:
+    source = (ROOT / "B05_Profile/B05_Profile_Router_Replan.py").read_text(encoding="utf-8")
+    call = source[source.index("write_initial_choice,\n") :][:600]
+    assert "request.balanced_profile" in call
+    initial = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial.py").read_text(encoding="utf-8")
+    # 지우기는 잠금 안
+    lock = initial.index("async with pool_lock(pool_key):")
+    assert lock < initial.index("drop_pool_results, project_root, pool_key") < lock + 400
+    assert 'model_dump(exclude={"recompute"})' in initial
+    # [적용] 균형 종단도 저장 패널값
+    assert "standard=standard" in initial[initial.index("async def balanced_profile_of") :]

resources/tester/test_route_initial.py

@@ -329,6 +329,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
         "logic": "follow",
         "weights": {"expected": 100.0, "earthwork": 0.0, "grade": 0.0},
         "criteria": None,
+        "with_profile": True,
     }
     for logic in ("gradeline", "safe", "crossings", "cost"):  # 새 갈래도 적고 다시 읽음(23-5)
         write_initial_choice(tmp_path, logic, None, {"target_grade_pct": 10.0})
@@ -339,6 +340,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
         "logic": "weighted",
         "weights": {"expected": 60.0, "earthwork": 20.0, "grade": 20.0},
         "criteria": None,
+        "with_profile": True,
     }
     # 사용자가 화면 칸에 겹쳐 넣은 값만(빈 칸은 안 담김) — 남으면 다시 열 때 그 칸만 채운다.
     write_initial_choice(
@@ -348,6 +350,7 @@ def test_initial_choice_roundtrip(tmp_path) -> None:
         "logic": "earthwork",
         "weights": {"expected": 0.0, "earthwork": 100.0, "grade": 0.0},
         "criteria": {"stream_offset_m": 30.0, "grade_smooth_m": 2.0},
+        "with_profile": True,
     }
     (tmp_path / "B05_Profile" / "route" / "initial_choice.json").write_text("{", encoding="utf-8")
     assert read_initial_choice(tmp_path)["logic"] == "follow"