feat(E03): 확률 해석 길 — 대표 단면만 · 뒤에서 · 진행률 · 파생 저장본 stab2d_prob · 기본값 후보 prob_defaults.json (537-1 · 요구 351)

- Router_Prob(새 파일): GET setup(단면 물성 = E02 값 · 기본값 후보 · 근거 · 지하수 갈래 · 기준안전율 · 결정론 F · 마지막 결과)
  · POST(임계면 고정 = 서버 스레드 · 매회 탐색 = 일꾼 프로세스 풀 · 같은 지문 = 저장본) · GET progress · GET 기록 목록(E05)
- 거절 422: 대표 단면 아님 · 단면에 없는 물성 · 수위선 · r_u 갈래 아닌 지하수 변수 · 변수 없음 · 매회 탐색 2000 회 넘음
- prob_defaults.json(M02 마스터 층): c 대수정규 0.30(붕적토 0.40) · φ 0.10 · γ 0.05 · ±3σ · ρ 0 · LHS 1000 · 국내 · 국외 근거
- main.py 길 붙임 · 시험 test_537_1_e03_prob_route 3(검증_sub4 동결 복사)

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

E03_StabSection/E03_StabSection_Router_Prob.py

@@ -0,0 +1,400 @@
+"""E03 확률 해석 · 민감도 API (요구 351 · PLAN 537-1) — 대표 단면만.
+
+- `GET /api/projects/{id}/stab/prob/setup/{station_key}?side=&condition=` — 그 단면 지층 물성(E02 값 = 평균) ·
+  기본값 후보(`prob_defaults.json` · 근거 · 사용자 확정 전) · 지하수 갈래 · 기준안전율 · 마지막 결과
+- `POST …/stab/prob/{station_key}` — 분포 · 표본 수 · 갈래 → 뒤에서(진행 `GET …/stab/prob/progress`) ·
+  같은 지문 = 저장본(fresh)
+- `GET …/stab/prob` — 지금 2D 결과에 붙은 기록 목록(측점 · 비탈 · 조건마다 가장 최근 · E05 「확률해석」 시트)
+
+평균 = E02 지반 조건(작업본) 값 · 바꾸려면 E02 에서. 임계면 = 지금 2D 결과의 그 방법 최소 면.
+결과 = 파생 저장본 `_derived/stab2d_prob/<지문>` · 지문 = 원본 측점 지문 + 요청 + 코드.
+엔진 = `Engine_Prob` — 임계면 고정 = 서버 스레드(빠름) · 매회 탐색 = 일꾼 프로세스(풀 · 느림).
+계산 자리 = ③ 서버 단독.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import logging
+import os
+import subprocess
+import sys
+import tempfile
+import time
+from functools import lru_cache
+from pathlib import Path
+from typing import Any, Literal
+from uuid import UUID
+
+from fastapi import APIRouter, HTTPException
+from pydantic import BaseModel, Field
+
+from common_util.common_util_derived_store import (
+    derived_dir,
+    derived_key,
+    read_derived,
+    write_derived,
+)
+from E02_StabGround import E02_StabGround_Router as e02
+from E03_StabSection import E03_StabSection_Engine_Prob as prob
+from E03_StabSection import E03_StabSection_Router as e03
+from E03_StabSection import E03_StabSection_Router_WhatIf as whatif
+from E03_StabSection import E03_StabSection_View as view
+from E03_StabSection.E03_StabSection_Engine_LEM import WALL_ID
+from E03_StabSection.E03_StabSection_Engine_Model import section_model_from_b06
+from E03_StabSection.E03_StabSection_Types import groundwater as norm_groundwater
+
+logger = logging.getLogger(__name__)
+router = APIRouter(prefix="/api/projects", tags=["E03 Stab Probabilistic"])
+KIND = "stab2d_prob"
+KEEP = 200
+DEFAULTS_PATH = (
+    Path(__file__).resolve().parents[1] / "resources/master_template/stability/prob_defaults.json"
+)
+#: 매회 탐색 표본 수 상한(측점 하나 탐색 수 초 × N)
+SEARCH_MAX_N = 2000
+#: 일꾼 프로세스 수 — None = 코어 − 2 · 1 = 서버 안에서 차례로(시험)
+WORKERS: int | None = None
+
+_JOBS: dict[str, dict[str, Any]] = {}
+_TASKS: set[asyncio.Task] = set()
+
+
+@lru_cache(maxsize=1)
+def code_hash() -> str:
+    digest = hashlib.sha1(e03.engine_hash().encode())
+    for name in ("E03_StabSection_Engine_Prob.py", Path(__file__).name):
+        digest.update((Path(__file__).parent / name).read_bytes())
+    return digest.hexdigest()[:12]
+
+
+def load_defaults() -> dict[str, Any]:
+    return json.loads(DEFAULTS_PATH.read_text(encoding="utf-8"))
+
+
+class VarIn(BaseModel):
+    """확률 변수 하나 — c · φ · γ = 변동계수 · 지하수위 이동(m) · r_u = 표준편차 · 절단 = 평균 ± k·σ."""
+
+    material: str | None = None
+    prop: Literal["c", "phi", "gamma", "water", "ru"]
+    dist: Literal["normal", "lognormal"] = "normal"
+    cov: float | None = Field(None, ge=0.0, le=2.0)
+    std: float | None = Field(None, ge=0.0, le=20.0)
+    trunc_sigma: float | None = Field(3.0, gt=0.0, le=10.0)
+
+
+class ProbRequest(BaseModel):
+    side: Literal["left", "right"]
+    condition: Literal["dry", "wet", "seismic"]
+    method: Literal["ordinary", "bishop", "janbu", "spencer", "morgenstern_price"] = "bishop"
+    mode: Literal["fixed", "search"] = "fixed"
+    n: int = Field(1000, ge=50, le=20000)
+    sampling: Literal["lhs", "mc"] = "lhs"
+    seed: int = Field(0, ge=0, le=2**31 - 1)
+    variables: list[VarIn] = Field(default_factory=list, max_length=40)
+    rho_c_phi: float = Field(0.0, ge=-0.95, le=0.95)
+
+
+def _model(ctx: dict[str, Any], condition: str) -> dict[str, Any]:
+    """그 조건 모형 — `analyze_station` 과 같은 입력(지층 · 지반선 · 벽 · 하중 · E02 지하수 갈래)."""
+    kw = ctx["kw"]
+    model = section_model_from_b06(
+        ctx["cross"],
+        ctx["design"],
+        materials=kw["materials"],
+        layers=kw["layers"],
+        ground=kw["ground"],
+        wet=condition == "wet",
+        seismic_kh=kw["seismic_kh"] if condition == "seismic" else 0.0,
+        road_surcharge_kpa=kw["road_surcharge_kpa"],
+        walls=kw.get("walls"),
+    )
+    model["groundwater"] = norm_groundwater(kw["groundwater_for"](condition, model))
+    return model
+
+
+def _slot(root: Path, station_key: str, side: str, condition: str, method: str) -> dict:
+    """측점 · 비탈 · 조건 — 대표 단면만 · 모형 · 임계면 · 기준안전율."""
+    ctx = whatif._context(root, station_key)
+    rows = [dict(r) for r in e03._latest_summary(root, ctx["plan"]["fp"])[0]["stations"]]
+    view.mark_representative(rows, e03._overrides(root))
+    mark = next((r.get("representative") or {} for r in rows if r["key"] == station_key), {})
+    if not mark.get("on"):
+        raise HTTPException(
+            status_code=422, detail="대표 단면만 확률 해석합니다 — 먼저 대표 단면으로."
+        )
+    if condition not in ctx["kw"]["conditions"]:
+        raise HTTPException(status_code=422, detail="그 조건은 해석하지 않았습니다.")
+    case = next(
+        (
+            c
+            for c in ctx["value"]["out"]["cases"]
+            if c["condition"] == condition and c["side"] == side
+        ),
+        None,
+    )
+    if case is None:
+        raise HTTPException(status_code=422, detail="그 쪽 비탈이 없습니다.")
+    methods = case["result"]["methods"]
+    surface = (methods.get(method) or {}).get("min") or None
+    model = _model(ctx, condition)
+    sd = next(s for s in model["sides"] if s["side"] == side)
+    return {
+        "ctx": ctx,
+        "model": model,
+        "side": sd,
+        "role": case["role"],
+        "surface": surface,
+        "methods": methods,
+        "required": view.required_factor(ctx["plan"]["factors"], condition, case["role"]),
+    }
+
+
+def _materials(model: dict[str, Any]) -> list[dict[str, Any]]:
+    """단면에 쓰인 물성(벽 뺌 · 지층 차례)."""
+    used = list(dict.fromkeys(r["material_id"] for r in model["regions"]))
+    by = {m["id"]: m for m in model["materials"]}
+    return [by[i] for i in used if i != WALL_ID and i in by]
+
+
+def _variables(req: ProbRequest, model: dict[str, Any]) -> tuple[list[dict], list]:
+    """요청 → 엔진 변수(평균 = 모형 값 · σ 0 은 뺌) · c–φ 상관 짝."""
+    mats = {m["id"]: m for m in _materials(model)}
+    gw = model.get("groundwater") or {}
+    out = []
+    for v in req.variables:
+        if v.prop in ("water", "ru"):
+            need = "line" if v.prop == "water" else "ru"
+            if gw.get("mode") != need:
+                raise HTTPException(
+                    status_code=422,
+                    detail=f"{'지하수위' if v.prop == 'water' else 'r_u'} 변수는 그 조건 지하수 갈래가 "
+                    f"{'수위선' if need == 'line' else 'r_u'} 일 때만.",
+                )
+            mean, std = (0.0, v.std) if v.prop == "water" else (float(gw["ru"]), v.std)
+            name = "지하수위" if v.prop == "water" else "r_u"
+        else:
+            m = mats.get(v.material or "")
+            if m is None:
+                raise HTTPException(status_code=422, detail=f"단면에 없는 물성: {v.material}")
+            mean = float(m["gamma_t"] if v.prop == "gamma" else m[v.prop])
+            std = mean * float(v.cov) if v.cov is not None else v.std
+            name = f"{m.get('name') or m['id']} {dict(c='c', phi='φ', gamma='γ')[v.prop]}"
+        if not std:
+            continue
+        if v.dist == "lognormal" and (mean <= 0 or v.prop in ("water",)):
+            raise HTTPException(status_code=422, detail=f"대수정규는 평균 > 0 만: {name}")
+        k = v.trunc_sigma
+        out.append(
+            {
+                "name": name,
+                "material": v.material if v.prop not in ("water", "ru") else None,
+                "prop": v.prop,
+                "dist": v.dist,
+                "mean": mean,
+                "std": float(std),
+                "lo": None if k is None else k * float(std),
+                "hi": None if k is None else k * float(std),
+            }
+        )
+    pairs = []
+    if req.rho_c_phi:
+        for i, a in enumerate(out):
+            for j, b in enumerate(out):
+                if a["prop"] == "c" and b["prop"] == "phi" and a["material"] == b["material"]:
+                    pairs.append((i, j, req.rho_c_phi))
+    return out, pairs
+
+
+def _run_search(job: dict[str, Any], emit: Any) -> dict[str, Any]:
+    """매회 탐색 일꾼(`python -m …Engine_Prob job.json`) — 한 줄마다 emit · 실패면 RuntimeError."""
+    with tempfile.TemporaryDirectory(prefix="aislo_e03p_") as folder:
+        path, out = Path(folder) / "job.json", Path(folder) / "out.json"
+        path.write_text(json.dumps({**job, "out": str(out)}, ensure_ascii=False), encoding="utf-8")
+        log = Path(folder) / "stderr.txt"
+        with log.open("w", encoding="utf-8") as errors:
+            process = subprocess.Popen(
+                [sys.executable, "-m", "E03_StabSection.E03_StabSection_Engine_Prob", str(path)],
+                cwd=e03.PROJECT_ROOT,
+                stdout=subprocess.PIPE,
+                stderr=errors,
+                text=True,
+                encoding="utf-8",
+            )
+            for line in process.stdout:
+                if line.startswith("{"):
+                    emit(json.loads(line).get("done"))
+            code = process.wait()
+        if code != 0:
+            tail = log.read_text(encoding="utf-8", errors="replace").strip().splitlines()[-1:]
+            raise RuntimeError(f"확률 해석 일꾼 실패: {tail}")
+        return json.loads(out.read_text(encoding="utf-8"))
+
+
+def run_prob(root: Path, station_key: str, req: ProbRequest, report: Any) -> dict[str, Any]:
+    slot = _slot(root, station_key, req.side, req.condition, req.method)
+    if slot["surface"] is None:
+        raise HTTPException(status_code=422, detail="그 방법 임계면이 없습니다(해 없음).")
+    if req.mode == "search" and req.n > SEARCH_MAX_N:
+        raise HTTPException(status_code=422, detail=f"매회 탐색은 표본 {SEARCH_MAX_N} 회까지.")
+    variables, pairs = _variables(req, slot["model"])
+    if not variables:
+        raise HTTPException(status_code=422, detail="확률 변수가 없습니다(변동계수 · 표준편차 0).")
+    row = slot["ctx"]["row"]
+    key = derived_key(code_hash(), row["fp"], req.model_dump())
+    cached = read_derived(root, KIND, key)
+    if cached:
+        return cached["value"]
+    job = {
+        "model": slot["model"],
+        "surface": slot["surface"],
+        "direction": slot["side"]["direction"],
+        "limits": list(slot["side"]["limits"]),
+        "variables": variables,
+        "correlations": pairs,
+        "n": req.n,
+        "sampling": req.sampling,
+        "seed": req.seed,
+        "mode": req.mode,
+        "method": req.method,
+        "required": slot["required"],
+    }
+
+    def emit(done: int | None) -> None:
+        if done:
+            report(2 + 93 * int(done) // req.n, f"표본 {done} / {req.n}")
+
+    report(1, "시작")
+    workers = WORKERS or max(1, (os.cpu_count() or 2) - 2)
+    if req.mode == "search" and workers > 1:
+        result = _run_search({**job, "workers": workers}, emit)
+    else:
+        result = prob.run(**job, emit=emit)
+    record = {
+        "station_key": station_key,
+        "chainage_m": row["chainage_m"],
+        "label": row.get("label"),
+        "base_fp": row["fp"],
+        "side": req.side,
+        "role": slot["role"],
+        "condition": req.condition,
+        "required": slot["required"],
+        "fs_deterministic": slot["surface"]["F"],
+        "surface_kind": slot["surface"]["kind"],
+        "request": req.model_dump(),
+        "result": result,
+        "computed_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
+    }
+    write_derived(root, KIND, key, record, keep=KEEP)
+    return record
+
+
+def records(root: Path) -> list[dict[str, Any]]:
+    """가장 최근 2D 요약 측점에 붙은 기록 — (측점 · 비탈 · 조건)마다 가장 최근 하나 · 누가거리 순."""
+    summary, _ = e03._latest_summary(root, "")
+    fps = {r["fp"] for r in (summary or {}).get("stations") or []}
+    latest: dict[tuple, dict[str, Any]] = {}
+    for path in derived_dir(root, KIND).glob("*.json"):
+        value = (read_derived(root, KIND, path.stem) or {}).get("value")
+        if not value or value.get("base_fp") not in fps:
+            continue
+        slot = (value["station_key"], value["side"], value["condition"])
+        if slot not in latest or value["computed_at"] > latest[slot]["computed_at"]:
+            latest[slot] = value
+    return sorted(
+        latest.values(), key=lambda v: (float(v["chainage_m"]), v["side"], v["condition"])
+    )
+
+
+def setup(root: Path, station_key: str, side: str, condition: str) -> dict[str, Any]:
+    slot = _slot(root, station_key, side, condition, "bishop")
+    gw = slot["model"].get("groundwater") or {}
+    last = next(
+        (
+            r
+            for r in records(root)
+            if (r["station_key"], r["side"], r["condition"]) == (station_key, side, condition)
+        ),
+        None,
+    )
+    return {
+        "station_key": station_key,
+        "side": side,
+        "condition": condition,
+        "role": slot["role"],
+        "required": slot["required"],
+        "fs_deterministic": {m: (v.get("min") or {}).get("F") for m, v in slot["methods"].items()},
+        "materials": [
+            {k: m.get(k) for k in ("id", "name", "c", "phi", "gamma_t")}
+            for m in _materials(slot["model"])
+        ],
+        "groundwater": {"mode": gw.get("mode", "none"), "ru": gw.get("ru")},
+        "defaults": load_defaults(),
+        "search_max_n": SEARCH_MAX_N,
+        "last": last,
+    }
+
+
+@router.get("/{project_id}/stab/prob/setup/{station_key}", response_model=None)
+async def get_setup(project_id: UUID, station_key: str, side: str, condition: str) -> dict:
+    root, _ = await e02._project(project_id)
+    return await asyncio.to_thread(setup, root, station_key, side, condition)
+
+
+@router.post("/{project_id}/stab/prob/{station_key}", response_model=None)
+async def post_prob(project_id: UUID, station_key: str, payload: ProbRequest) -> dict:
+    """뒤에서 — 같은 지문이면 바로 저장본(fresh) · 진행 = `GET …/stab/prob/progress`."""
+    root, _ = await e02._project(project_id)
+    key = str(project_id)
+    if (_JOBS.get(key) or {}).get("running"):
+        raise HTTPException(status_code=409, detail="이미 확률 해석 중입니다.")
+    # 입력 검사 · 저장본 확인은 바로(422 를 그 자리에서 돌려줌)
+    slot = await asyncio.to_thread(
+        _slot, root, station_key, payload.side, payload.condition, payload.method
+    )
+    _variables(payload, slot["model"])
+    fp = derived_key(code_hash(), slot["ctx"]["row"]["fp"], payload.model_dump())
+    saved = read_derived(root, KIND, fp)
+    if saved:
+        return {"status": "fresh", "record": saved["value"]}
+    job = {
+        "running": True,
+        "percent": 0,
+        "message": "시작",
+        "error": None,
+        "station_key": station_key,
+    }
+    _JOBS[key] = job
+
+    def report(percent: int, message: str) -> None:
+        job.update(percent=int(percent), message=message)
+
+    async def work() -> None:
+        try:
+            await asyncio.to_thread(run_prob, root, station_key, payload, report)
+            job.update(percent=100, message="끝")
+        except Exception as error:  # noqa: BLE001 — 진행률에 까닭을 그대로 올림
+            logger.exception("E03 확률 해석 실패: project_id=%s", project_id)
+            job["error"] = str(getattr(error, "detail", None) or error)
+        finally:
+            job["running"] = False
+
+    task = asyncio.create_task(work(), name=f"E03 prob {project_id}")
+    _TASKS.add(task)
+    task.add_done_callback(_TASKS.discard)
+    return {"status": "accepted", **job}
+
+
+@router.get("/{project_id}/stab/prob/progress", response_model=None)
+async def get_progress(project_id: UUID) -> dict:
+    await e02._project(project_id)
+    job = _JOBS.get(str(project_id))
+    return dict(job) if job else {"running": False, "percent": 0, "message": "", "error": None}
+
+
+@router.get("/{project_id}/stab/prob", response_model=None)
+async def get_records(project_id: UUID) -> dict:
+    root, _ = await e02._project(project_id)
+    return {"records": await asyncio.to_thread(records, root)}

main.py

@@ -86,6 +86,7 @@ from E05_StabReport.E05_StabReport_Router import router as e05_stab_report_route
 from E07_StabArea.E07_StabArea_Router import router as e07_stab_area_router
 from E03_StabSection.E03_StabSection_Router import router as e03_stab_section_router
 from E03_StabSection.E03_StabSection_Router_WhatIf import router as e03_stab_whatif_router
+from E03_StabSection.E03_StabSection_Router_Prob import router as e03_stab_prob_router
 from E06_StabRock.E06_StabRock_Router import router as e06_stab_rock_router
 from M01_MasterData.M01_MasterData_Router import router as m01_master_data_router
 from M02_MasterTemplete.M02_Labor_Router import router as m02_labor_router
@@ -499,6 +500,7 @@ app.include_router(e05_stab_report_router, dependencies=protected_with_company)
 app.include_router(e07_stab_area_router, dependencies=protected_with_company)
 app.include_router(e03_stab_section_router, dependencies=protected_with_company)
 app.include_router(e03_stab_whatif_router, dependencies=protected_with_company)
+app.include_router(e03_stab_prob_router, dependencies=protected_with_company)
 app.include_router(e06_stab_rock_router, dependencies=protected_with_company)
 app.include_router(b05_corridor_router, dependencies=protected_with_company)
 app.include_router(b05_route_replan_router, dependencies=protected_with_company)

resources/master_template/stability/prob_defaults.json

@@ -0,0 +1,56 @@
+{
+  "version": 1,
+  "note": "확률 해석 기본값 후보 — 문헌 · 상용 SW 예제 값 · 현장 시험 없음 · 사용자 확정 전(E03 확률 · 민감도 탭에서 바꿈)",
+  "run": {
+    "n": 1000,
+    "sampling": "lhs",
+    "seed": 0,
+    "mode": "fixed",
+    "method": "bishop"
+  },
+  "properties": {
+    "c": {
+      "dist": "lognormal",
+      "cov": 0.3,
+      "trunc_sigma": 3.0
+    },
+    "phi": {
+      "dist": "normal",
+      "cov": 0.1,
+      "trunc_sigma": 3.0
+    },
+    "gamma": {
+      "dist": "normal",
+      "cov": 0.05,
+      "trunc_sigma": 3.0
+    },
+    "water": {
+      "dist": "normal",
+      "std": 0.0,
+      "trunc_sigma": 3.0
+    },
+    "ru": {
+      "dist": "normal",
+      "std": 0.0,
+      "trunc_sigma": 3.0
+    }
+  },
+  "rho_c_phi": 0.0,
+  "sources": {
+    "c": "COV 0.30(범위 0.2~0.5) · 대수정규(음수 막음) — 국내 화강풍화토 실측 정남수 외(2011) 0.34~0.49 · 장정윤 · 박혁진(2017) 0.21 · 해석 가정 Cho(2018 · 2019) 0.20 · Lee 외(2020) 0.25 · ISSMGE TC304(2021) 홍콩 완전풍화화강암 0.51 · Harr(1987) 최대 0.40 · 붕적토 0.40 = 국내 자료 없어 국외 범위 위쪽(협의 필요)",
+    "phi": "COV 0.10(범위 0.05~0.15) · 정규 — 국내 화강풍화토 실측 정남수 외(2011) 0.09~0.12 · 장정윤 · 박혁진(2017) 0.03 · Cho(2018 · 2019) · Lee 외(2020) 0.10 · TC304(2021) 모래 평균 0.079 · 홍콩 CDG 0.096 · Duncan(2000) 2~13 %",
+    "gamma": "COV 0.05(범위 0.03~0.07) · 정규 — Duncan(2000) 3~7 % · 홍콩 CDG 2.6 % · Li & Lumb(1987) 5 % · 국내 논문은 고정값으로 둠",
+    "water": "지하수위 이동 표준편차 기본 0(확률 변수 아님) · 수위선 갈래(E02)일 때만",
+    "ru": "r_u 표준편차 기본 0 · r_u 갈래(E02)일 때만 · Li & Lumb(1987) COV 10 %",
+    "rho_c_phi": "기본 0(상관 무시 = 보수적) · Slide2 · Cho(2019) 기본 −0.5 · 문헌 −0.25~−0.75(Wolff 1985 −0.47 · Lumb 1970 −0.7~−0.37)",
+    "trunc_sigma": "평균 ±3σ 절단 — Slide2 「3x 표준편차」 · SLOPE/W James Bay 예제",
+    "run": "LHS 1000 · 임계면 고정(Slide2 Global Minimum) · Bishop — Slide2 Tutorial 08"
+  },
+  "materials": {
+    "colluvium": {
+      "c": {
+        "cov": 0.4
+      }
+    }
+  }
+}

resources/tester/test_537_1_e03_prob_route.py

@@ -0,0 +1,200 @@
+# -*- coding: utf-8 -*-
+"""537-1 E03 확률 해석 길(`E03_StabSection_Router_Prob`) — 검증_sub4 동결 복사 · 측점 둘 · Bishop 원호만.
+
+- 준비(setup): 단면 물성 = E02 값 · 기본값 후보 · 기준안전율 · 결정론 F
+- 임계면 고정: 평균 줄 F = 저장된 결정론 F · Pf · β · 표본 수 · 같은 요청 = 저장본(다시 안 셈) · 기록 목록
+- 대표 단면 아님 · 단면에 없는 물성 · 수위선 아닌데 지하수위 변수 · 변수 없음 = 422
+- 매회 탐색 일꾼(프로세스 풀 · 진행 줄 · 결과 파일)
+
+    ./venv/Scripts/python.exe -m pytest resources/tester/test_537_1_e03_prob_route.py -q
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+from uuid import UUID
+
+import pytest
+from fastapi import HTTPException
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from common_util import common_util_project_copy as cp  # noqa: E402
+from E01_StabRegister import E01_StabRegister_Source as src  # noqa: E402
+from E02_StabGround import E02_StabGround_Router as e02  # noqa: E402
+from E03_StabSection import E03_StabSection_Router as router  # noqa: E402
+from E03_StabSection import E03_StabSection_Router_Prob as rp  # noqa: E402
+from E03_StabSection import E03_StabSection_Router_WhatIf as wf  # noqa: E402
+from E03_StabSection import E03_StabSection_View as view  # noqa: E402
+from E03_StabSection import E03_StabSection_Worker as worker  # noqa: E402
+from E03_StabSection.E03_StabSection_Engine_Station import analyze_station  # noqa: E402
+
+SOURCE_ID = "c725d747-50ae-46b0-87ce-4d4e56a53cf3"  # 검증_sub4(읽기만)
+PID = "d" * 8 + "-" + "d" * 4 + "-" + "d" * 4 + "-" + "d" * 4 + "-" + "d" * 12
+
+
+async def _copy(target: Path) -> None:
+    import aiomysql
+
+    from config.config_db import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER
+
+    connection = await aiomysql.connect(
+        host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASSWORD, db=DB_NAME, autocommit=True
+    )
+    try:
+        await cp.copy_project(
+            connection, SOURCE_ID, scope="stability", new_id=PID, project_root=target
+        )
+    finally:
+        connection.close()
+
+
+def _fast(cross, design, **kw):
+    return analyze_station(cross, design, **{**kw, "methods": ("bishop",), "noncircular": False})
+
+
+@pytest.fixture(scope="module")
+def project():
+    folder = Path(tempfile.mkdtemp(prefix="aislo_e03_prob_"))
+    try:
+        asyncio.run(_copy(folder / "copy"))
+    except Exception as error:  # noqa: BLE001
+        shutil.rmtree(folder, ignore_errors=True)
+        pytest.skip(f"검증 원본 복사 실패: {error}")
+    root = folder / "copy"
+    mp = pytest.MonkeyPatch()
+    original = src.stations
+    mp.setattr(router.src, "stations", lambda r: original(r)[20:22])
+
+    async def fake_project(_pid):
+        return root, {"company_id": 1, "project_kind": "STABILITY"}
+
+    mp.setattr(e02, "_project", fake_project)
+    mp.setattr(worker, "analyze_station", _fast)
+    mp.setattr(wf, "analyze_station", _fast)
+    mp.setattr(router, "WORKERS", 1)
+    mp.setattr(rp, "WORKERS", 1)
+    router._JOBS.clear()
+
+    async def go():
+        await router.analyze2d(UUID(PID))
+        while router._TASKS:
+            await asyncio.gather(*list(router._TASKS))
+
+    asyncio.run(go())
+    yield root
+    mp.undo()
+    shutil.rmtree(folder, ignore_errors=True)
+
+
+def _slot(root: Path) -> tuple[str, str]:
+    """대표 단면 중 깎기 · 쌓기 비탈 하나."""
+    rows = router._latest_summary(root, "")[0]["stations"]
+    view.mark_representative(rows, router._overrides(root))
+    row = next(r for r in rows if (r.get("representative") or {}).get("on"))
+    return row["key"], row["sides"][0]["side"]
+
+
+def _req(side, mats=("weathered_soil",), **kw):
+    """단면 물성마다 c(대수정규 0.3) · φ(0.1) · γ(0.05)."""
+    variables = [
+        v
+        for m in mats
+        for v in (
+            {"material": m, "prop": "c", "dist": "lognormal", "cov": 0.3},
+            {"material": m, "prop": "phi", "cov": 0.1},
+            {"material": m, "prop": "gamma", "cov": 0.05},
+        )
+    ]
+    return rp.ProbRequest(
+        **{"side": side, "condition": "dry", "n": 400, "variables": variables, **kw}
+    )
+
+
+def test_setup_and_fixed_run(project, monkeypatch):
+    key, side = _slot(project)
+    s = rp.setup(project, key, side, "dry")
+    ids = [m["id"] for m in s["materials"]]
+    assert "weathered_soil" in ids and "wall" not in ids
+    assert s["defaults"]["properties"]["c"]["cov"] == 0.3 and s["required"] in (1.5, 1.3)
+    fs = s["fs_deterministic"]["bishop"]
+    seen = []
+    req = _req(side, ids)
+    rec = rp.run_prob(project, key, req, lambda p, m: seen.append(p))
+    res = rec["result"]
+    assert res["sensitivity"]["F_mean"] == pytest.approx(fs, rel=1e-9)
+    assert rec["fs_deterministic"] == fs and res["stats"]["n"] == 400
+    st = res["stats"]
+    assert 0 <= st["pf"] <= 1 and st["std"] > 0 and abs(st["mean"] - fs) < 0.1 * fs
+    assert st["beta_normal"] == pytest.approx((st["mean"] - 1) / st["std"])
+    assert seen and max(seen) <= 95
+    # 같은 요청 = 저장본 · 기록 목록 · setup 의 마지막 결과
+    calls = []
+    monkeypatch.setattr(rp.prob, "run", lambda **kw: calls.append(1))
+    again = rp.run_prob(project, key, req, lambda p, m: None)
+    assert not calls and again["computed_at"] == rec["computed_at"]
+    listed = rp.records(project)
+    assert [(r["station_key"], r["side"], r["condition"]) for r in listed] == [(key, side, "dry")]
+    assert rp.setup(project, key, side, "dry")["last"]["computed_at"] == rec["computed_at"]
+
+
+def test_rejects(project):
+    key, side = _slot(project)
+    with pytest.raises(HTTPException) as e:
+        rp.run_prob(
+            project, key, _req(side, variables=[{"material": "xx", "prop": "c", "cov": 0.2}]), print
+        )
+    assert e.value.status_code == 422
+    with pytest.raises(HTTPException) as e:
+        rp.run_prob(project, key, _req(side, variables=[{"prop": "water", "std": 0.5}]), print)
+    assert e.value.status_code == 422 and "수위선" in e.value.detail
+    with pytest.raises(HTTPException) as e:
+        rp.run_prob(
+            project,
+            key,
+            _req(side, variables=[{"material": "weathered_soil", "prop": "c", "cov": 0}]),
+            print,
+        )
+    assert "변수가 없습니다" in e.value.detail
+    # 대표 단면 끔 → 422
+    path = project / router.REPRESENTATIVE
+    old = path.read_text(encoding="utf-8") if path.is_file() else None
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({"overrides": {key: False}}), encoding="utf-8")
+    try:
+        with pytest.raises(HTTPException) as e:
+            rp.run_prob(project, key, _req(side), print)
+        assert "대표 단면" in e.value.detail
+    finally:
+        path.unlink() if old is None else path.write_text(old, encoding="utf-8")
+
+
+def test_search_worker_process():
+    """매회 탐색 일꾼 — 프로세스 풀 둘 · 진행 줄 · 결과 파일(Slide2 Tutorial 08 모형 · 작은 변동)."""
+    from test_537_1_e03_prob import T08, _surface, var
+
+    job = {
+        "model": T08,
+        "surface": _surface(T08),
+        "direction": -1,
+        "limits": None,
+        "variables": [var("c", "c", 5, 0.01), var("phi", "phi", 30, 0.01)],
+        "n": 4,
+        "mode": "search",
+        "workers": 2,
+    }
+    done = []
+    out = rp._run_search(job, done.append)
+    assert sorted(done) == [1, 2, 3, 4] and out["n"] == 4
+    assert all(abs(f - 1.139) < 0.02 for f in out["F"])
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-q"]))