fix(763-2): 같은 서피스 = 파일 내용 지문 — 비용면 · 계류 · 초기 종단 · 초기 종단 목표 = 고른 지형

- `file_digest()`(derived_store) — blake2b 내용 지문 · 한 프로세스 안 경로 · 시각 · 크기로 기억
- 비용면 서명 · 계류 지문 · 초기 종단 `stat_marks` = 내용 지문(옮겨 심기 · 동기화 · 다시 짓기로 시각만 바뀌면 같은 열쇠)
- 초기 종단 표식 = 확정 지형이 읽는 파일만(`_source_npz_paths` · `*_{필터}*.npz` glob 걷음)
- `balanced_profile_of` 에 selection — 추천 초기 종단 목표 계획고 = 지반선과 같은 고른 지형(3h)
- 열쇠 꼴이 바뀌어 기존 추천 · 초기 종단 저장본은 다시 셈(763-5 몫)

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

B05_Profile/B05_Profile_Engine_InitialProfile.py

@@ -213,15 +213,10 @@ def source_fingerprint() -> str:
 
 
 def stat_marks(paths: list[Path]) -> list[str]:
-    """큰 자료(지형 npz · 도엽) 자국 — 이름 · 크기 · 수정시각."""
-    marks = []
-    for path in sorted(paths):
-        try:
-            info = path.stat()
-            marks.append(f"{path.name}:{info.st_size}:{info.st_mtime_ns}")
-        except OSError:
-            marks.append(f"{path.name}:-")
-    return marks
+    """큰 자료(지형 npz · 도엽) 자국 — 이름 · 내용 지문(763-2 · 시각만 바뀌면 같음)."""
+    from common_util.common_util_derived_store import file_digest
+
+    return [f"{path.name}:{file_digest(path) or '-'}" for path in sorted(paths)]
 
 
 def fingerprint(parts: dict[str, Any], source: str | None = None) -> str:

B05_Profile/B05_Profile_Engine_RouteInitial.py

@@ -159,6 +159,7 @@ def load_terrain_and_streams(
         _load_or_build_cost_surface,
     )
     from common_util.common_util_crs import resolve_project_crs
+    from common_util.common_util_derived_store import file_digest
 
     filter_key = str(selection.get("source_filter"))
     method = str(selection.get("method") or "dtm")
@@ -166,11 +167,7 @@ def load_terrain_and_streams(
     models_dir = project_root / _MODELS_SUBDIR if models_dir is None else Path(models_dir)
     stream_path = project_root / "B04_PreProcess" / "processed" / STREAM_FILE
     crs = resolve_project_crs(project_root)
-    try:
-        stat = stream_path.stat()
-        stream_stamp = f"{stat.st_mtime_ns}:{stat.st_size}"
-    except OSError:
-        stream_stamp = "-"
+    stream_stamp = file_digest(stream_path) or "-"  # 내용 지문(763-2)
     signature = "|".join(
         (
             filter_key,

B05_Profile/B05_Profile_Engine_Solver.py

@@ -19,6 +19,7 @@ from B05_Profile.B05_Profile_Engine_Geometry import (
     resample_polyline_2d,
     single_segment_dijkstra,
 )
+from common_util.common_util_derived_store import file_digest
 from common_util.common_util_surface_sampler import (
     DtmGridSampler,
     check_surface_keys,
@@ -167,11 +168,10 @@ def _source_npz_paths(models_dir: Path, filter_key: str, method: str, smooth: bo
 
 
 def _cost_surface_signature(models_dir: Path, filter_key: str, method: str, smooth: bool) -> str:
-    """비용면 재빌드 필요 시 바뀌는 서명(격자해상도 + 소스파일 mtime/size)."""
+    """비용면 서명(격자해상도 + 소스파일 이름 · 내용 지문 · 763-2 — 시각만 바뀌면 같음)."""
     parts = [f"res={ROUTE_GRID_RES_M}", f"method={method}", f"smooth={smooth}"]
     for p in _source_npz_paths(models_dir, filter_key, method, smooth):
-        st = p.stat()
-        parts.append(f"{p.name}:{int(st.st_mtime)}:{st.st_size}")
+        parts.append(f"{p.name}:{file_digest(p)}")
     return "|".join(parts)
 
 

B05_Profile/B05_Profile_Router_RouteInitial.py

@@ -525,8 +525,10 @@ async def balanced_profile_of(
     points: list[dict[str, float]],
     criteria: Any,
     design_curves: list[dict[str, Any]] | None = None,
+    selection: dict[str, Any] | None = None,
 ) -> dict[str, Any] | None:
-    """노선 정점의 균형 종단 한 벌(`route_profile`) — [확인] 체인 몫(지금 설계 지형)."""
+    """노선 정점의 균형 종단 한 벌(`route_profile`) — [확인] 체인 몫(지금 설계 지형) ·
+    `selection` = 추천 초기 종단의 고른 분석 지형(지반선과 같은 지형 · 763-2)."""
     from B05_Profile.B05_Profile_Engine_Grade import legal_grade_limit_pct
     from B05_Profile.B05_Profile_Engine_RouteInitial_Profile import route_profile
     from B05_Profile.B05_Profile_Engine_Solver import _MODELS_SUBDIR
@@ -537,9 +539,10 @@ async def balanced_profile_of(
     max_grade = legal_grade_limit_pct(grade_class, terrain_type, False, design_speed) / 100.0
     if criteria is not None and criteria.max_grade_pct is not None:
         max_grade = criteria.max_grade_pct / 100.0
-    pool = get_db_pool()
-    async with pool.acquire() as connection:
-        selection = await get_surface_confirmation_params(connection, str(project_id))
+    if selection is None:
+        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 비용면 아님)

B05_Profile/B05_Profile_Router_RouteInitial_Initial.py

@@ -43,6 +43,7 @@ async def initial_profile_of(
     from B05_Profile import B05_Profile_Engine_InitialProfile as ip
     from B05_Profile import B05_Profile_Router_Replan as replan
     from B05_Profile.B05_Profile_Engine_Sections import _MODELS_SUBDIR
+    from B05_Profile.B05_Profile_Engine_Solver import _source_npz_paths
     from B05_Profile.B05_Profile_Repository import get_latest_route
     from B05_Profile.B05_Profile_Route_Save import section_seed
     from B05_Profile.B05_Profile_Router_RouteInitial import (
@@ -101,7 +102,15 @@ async def initial_profile_of(
             "curves": result.get("curves"),
             "radius": radius_m,
             "surface": surface,
-            "models": ip.stat_marks(list(models.glob(f"*_{surface['source_filter']}*.npz"))),
+            # 확정 지형이 읽는 파일만(763-2) — 다른 표현 · 굳힌 격자 다시 짓기에 안 흔들림
+            "models": ip.stat_marks(
+                _source_npz_paths(
+                    models,
+                    str(surface["source_filter"]),
+                    str(surface["method"] or "dtm"),
+                    bool(surface["smooth"]),
+                )
+            ),
             "streams": ip.stat_marks([_sheet_dir(stored_path) / STREAM_FILE]),
             # 저장 노선 파일은 안 넣음(505-1) — 초기화(492) 셈은 노선 파일 · 관 · 구조물 ·
             # 추가 측점을 안 읽음(선 = 후보 노드) · 넣으면 노선 [저장]마다 다시 셈
@@ -132,7 +141,7 @@ async def initial_profile_of(
         if request.with_profile:  # [적용] `balanced_target` 과 같은 셈(노선 정점 · 새 곡선표)
             points = [{"x": x, "y": y} for x, y in line]
             balanced = await balanced_profile_of(
-                project_id, project_root, points, request.criteria, curves
+                project_id, project_root, points, request.criteria, curves, selection
             )
             target = None if balanced is None else (balanced["stations_m"], balanced["plan_m"])
         sampler = await asyncio.to_thread(

common_util/common_util_derived_store.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import hashlib
 import json
+from functools import lru_cache
 from pathlib import Path
 from typing import Any
 
@@ -33,6 +34,28 @@ def file_stamp(path: Path | None) -> str | None:
     return f"{stat.st_mtime_ns}:{stat.st_size}" if stat else None
 
 
+def file_digest(path: Path | None) -> str | None:
+    """파일 내용 지문(blake2b) — 옮겨 심기 · 동기화 · 다시 짓기로 시각만 바뀌면 같은 값(763-2).
+    한 프로세스 안에서는 경로 · 수정시각 · 크기로 기억(다시 안 읽음). 없으면 None."""
+    try:
+        stat = path.stat() if path is not None else None
+    except OSError:
+        return None
+    if stat is None:
+        return None
+    return _digest(str(Path(path).resolve()), stat.st_mtime_ns, stat.st_size)
+
+
+# ponytail: 지형 파일 수백 개 전제 · 프로젝트가 크게 늘면 maxsize 를 셈.
+@lru_cache(maxsize=512)
+def _digest(path: str, _mtime_ns: int, _size: int) -> str:
+    digest = hashlib.blake2b(digest_size=16)
+    with open(path, "rb") as handle:
+        for block in iter(lambda: handle.read(1 << 20), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
 def key_matches(key_path: Path, key: Any) -> bool:
     """결과 옆 열쇠 파일이 지금 입력과 같은지."""
     try:

resources/tester/test_763_2_content_key.py

@@ -0,0 +1,116 @@
+"""763-2 같은 서피스 = 내용 지문 — 시각만 바뀌면 같은 열쇠 · 한 바이트 바꿈 = 다른 열쇠 ·
+초기 종단 표식 = 확정 지형이 읽는 파일만 · 초기 종단 목표 = 고른 지형.
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_763_2_content_key.py -q
+"""
+
+import asyncio
+import os
+import shutil
+from pathlib import Path
+from uuid import uuid4
+
+from test_763_1_confirmed_terrain import _models
+
+from B05_Profile import B05_Profile_Engine_InitialProfile as ip
+from B05_Profile import B05_Profile_Engine_Solver as solver
+from B05_Profile import B05_Profile_Router_Replan as replan
+from B05_Profile import B05_Profile_Router_RouteInitial as router
+from common_util.common_util_derived_store import file_digest
+
+ROOT = Path(__file__).resolve().parents[2]
+
+
+def _retime(folder: Path, seconds: int) -> None:
+    for path in folder.iterdir():
+        os.utime(path, (seconds, seconds))
+
+
+def _flip_byte(path: Path) -> None:
+    data = bytearray(path.read_bytes())
+    data[-1] ^= 0xFF
+    path.write_bytes(bytes(data))
+
+
+def test_파일_지문_내용만(tmp_path) -> None:
+    a, b = tmp_path / "a.bin", tmp_path / "b.bin"
+    a.write_bytes(b"terrain" * 1000)
+    b.write_bytes(b"terrain" * 1000)
+    os.utime(b, (1_000_000, 1_000_000))
+    assert file_digest(a) == file_digest(b)
+    _flip_byte(b)
+    assert file_digest(a) != file_digest(b)
+    assert file_digest(tmp_path / "none.bin") is None
+
+
+def test_비용면_서명_시각만_바뀌면_같음(tmp_path) -> None:
+    (tmp_path / "one").mkdir()
+    first = _models(tmp_path / "one")
+    for method in ("implicit", "dtm"):
+        solver._source_npz_paths(first, "classification", method, False)  # 굳힌 격자 지음
+    second = tmp_path / "two"
+    shutil.copytree(first, second)
+    _retime(second, 2_000_000_000)  # 옮겨 심기 · 동기화 = 시각만 바뀜
+    for method in ("implicit", "dtm"):
+        one = solver._cost_surface_signature(first, "classification", method, False)
+        two = solver._cost_surface_signature(second, "classification", method, False)
+        assert one == two
+    _flip_byte(second / "dtm_classification.npz")
+    assert solver._cost_surface_signature(
+        first, "classification", "dtm", False
+    ) != solver._cost_surface_signature(second, "classification", "dtm", False)
+    # 초기 종단 표식도 내용
+    names = ["implicit_classification.npz", "implicit_classification_grid.npz"]
+    assert ip.stat_marks([first / n for n in names]) == ip.stat_marks([second / n for n in names])
+
+
+def test_초기_종단_표식_확정_지형_파일만() -> None:
+    source = (ROOT / "B05_Profile/B05_Profile_Router_RouteInitial_Initial.py").read_text(
+        encoding="utf-8"
+    )
+    assert "models.glob(" not in source
+    assert '"models": ip.stat_marks(\n                _source_npz_paths(' in source
+    # 목표 = 고른 지형(지반선 표본기와 같은 selection)
+    assert "request.criteria, curves, selection\n" in source
+
+
+def test_균형_종단은_받은_지형으로(monkeypatch) -> None:
+    import B05_Profile.B05_Profile_Engine_Grade as grade
+    import B05_Profile.B05_Profile_Engine_RouteInitial_Profile as profile
+    import common_util.common_util_surface_sampler as sampler
+
+    opened: list[tuple] = []
+
+    async def road(_pid):
+        return "간선", 20, "일반"
+
+    async def standard(_pid):
+        return None
+
+    def build(_models, filter_key, method, smooth):
+        opened.append((filter_key, method, smooth))
+        return object()
+
+    def route_profile(*_a, **_k):
+        return {
+            "stations_m": [0.0],
+            "plan_m": [0.0],
+            "cut_m3": 0,
+            "fill_m3": 0,
+            "mass_end_m3": 0,
+            "seconds": 0,
+        }
+
+    def no_db():
+        raise AssertionError("고른 지형이 있으면 확정 params 를 다시 읽지 않음")
+
+    monkeypatch.setattr(replan, "_road_settings", road)
+    monkeypatch.setattr(router, "project_standard", standard)
+    monkeypatch.setattr(router, "get_db_pool", no_db)
+    monkeypatch.setattr(grade, "legal_grade_limit_pct", lambda *_a: 14.0)
+    monkeypatch.setattr(sampler, "build_surface_sampler", build)
+    monkeypatch.setattr(profile, "route_profile", route_profile)
+    sheet = {"source_filter": "sheet_laplace", "method": "dtm", "smooth": True}
+    points = [{"x": 0.0, "y": 0.0}, {"x": 10.0, "y": 0.0}]
+    got = asyncio.run(router.balanced_profile_of(uuid4(), Path("."), points, None, None, sheet))
+    assert got is not None and opened == [("sheet_laplace", "dtm", True)]