feat(B08): 배수관 물량 — 정본 셋을 이어 인계에 실음

랩탑 창이 측점 design 에 pipe_length_m 을 넣어 줘서 마지막 조각이 채워짐.
B08 은 잇기만 하고 길이를 짓지 않음.

값이 어디서 오나
  관 자체(있나·어디·관경·관종) → pipe_points.json (레지스트리 managed_by)
  관 연장(m)                  → 측점 design.pipe_length_m
                                (B06 이 서버 Node 로 m 단위 올림까지 끝낸 값)
  관종 → 공종코드              → work_item_mapping 의 pipe.kind_codes
     파형강관 FP-12-11-03 · 흄관 FP-12-11-02 · VR관 FP-12-11-01

⚠ facility 가 pipe 인 점만 배관 — pipe_points.json 은 계곡 통과 시설 전부의
  정본이라 BOX암거·물넘이·세월교가 같은 파일에 있음. 관경 유무로 가르면
  관경 미지정 관을 놓침. 실측: 5601e828 11점 중 관 9 · 세월교 2

⚠ 관종 기본값(파형강관)은 2026-08-17 사용자 확정값이라 근거가 있으나
  조용히 쓰지 않고 「기본값으로 섰습니다」를 알림에 실음

⚠ 안 붙인 것 둘 — 터파기·되메우기(관 부설과 각각 오면 같은 굴착을 두 번 셈,
  B09 ㉡ 가드 자리) · 유출입부 기슭막이(관 옵션이 정본이고 구조물에서 빠졌음)

연장이 없는 관은 0 으로 때우지 않고 blocked_reason 과 함께 감
「횡단설계에서 [저장]을 한 번 누르면 그 측점의 관 길이가 남고 값이 섭니다」

시험 10건 — 세월교 걸러짐 · 관경 없어도 관임 · 연장 없으면 안 섬 ·
관종 셋이 갈림 · 기본값을 알림 · 모르는 관종은 못 고름 ·
⚠ 좁게: 옆 측점 길이를 물어 오지 않음 · 실제 자료 정본 대조

시험: 697 passed · 24 skipped (B05 코리도 1건 기존 깨짐, 무관).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

B08_Quantity/B08_Quantity_Engine_Handoff.py

@@ -158,6 +158,7 @@ ORIGIN_STRUCTURE = "structure"
 ORIGIN_SLOPE = "slope"
 ORIGIN_HAUL = "haul"
 ORIGIN_PREPARATION = "preparation"
+ORIGIN_PIPE = "pipe"
 
 #: 암 시공법 → 매핑표의 지반 이름. 품셈이 **긁어내기와 터뜨리기를 다른 공종**으로 두기 때문에
 #: 갈래 이름(연암·보통암…)만으로는 공종을 못 고른다(2026-09-07 일감 9 실서버에서 드러남).
@@ -202,6 +203,8 @@ class WorkItemMapping:
     composite: dict[str, Any] = field(default_factory=dict)
     concrete_placing: dict[str, Any] = field(default_factory=dict)
     unit_conversion: dict[str, Any] = field(default_factory=dict)
+    #: 배수관 — 관종별 공종·연장 키. 관 정본은 `pipe_points.json` 이다.
+    pipe: dict[str, Any] = field(default_factory=dict)
 
     def for_earthwork(self, group: str, ground: str | None) -> dict[str, Any] | None:  # noqa: D401
         """공종+지반유형으로 찾는다. 지반이 갈리지 않는 공종은 `ground` 없는 줄이 받는다."""
@@ -253,6 +256,7 @@ def load_mapping(path: Path | None = None) -> WorkItemMapping:
         pending_user=payload.get("pending_user") or {},
         composite=payload.get("composite") or {},
         concrete_placing=payload.get("concrete_placing") or {},
+        pipe=payload.get("pipe") or {},
         unit_conversion=(payload.get("composite") or {}).get("unit_conversion") or {},
     )
 
@@ -859,6 +863,51 @@ def _prep_blocked_kind(status: str) -> str | None:
     return BLOCKED_UNIT_DATA_MISSING
 
 
+def _pipe_rows(pipe_table: dict[str, Any]) -> list[dict[str, Any]]:
+    """배수관 줄 — 값이 서는 줄도, 못 서는 줄도 함께 보낸다(준비공과 같은 규칙).
+
+    ⚠ 터파기·되메우기를 붙이지 않는다 — 관 부설과 굴착이 각각 오면 **같은 굴착을 두 번** 센다
+      (B09 ㉡ 가드와 같은 자리).
+    """
+    rows: list[dict[str, Any]] = []
+    for row in pipe_table.get("rows") or []:
+        ready = bool(row.get("in_bill"))
+        rows.append(
+            {
+                "work_item_code": row.get("work_item_code"),
+                "name": f"배수관({row.get('kind')})",
+                "spec": f"Ø{row.get('variant_value')}" if row.get("variant_value") else "",
+                "unit": str(row.get("unit") or "m"),
+                "quantity": float(row.get("quantity") or 0.0),
+                "quantity_gross": None,
+                "application_ratio_pct": None,
+                "application_ratio_breakdown": None,
+                "quantity_breakdown": None,
+                "ground_class": None,
+                "haul_distance_m": None,
+                "haul_equipment": None,
+                "station_from": row.get("chainage_m"),
+                "station_to": row.get("chainage_m"),
+                "excavation_method": None,
+                "spec_detail": f"Ø{row.get('variant_value')}" if row.get("variant_value") else "",
+                "composite_parts": None,
+                "structure_kind": None,
+                "blocked_kind": row.get("blocked_kind"),
+                "blocked_reason": str(row.get("blocked_reason") or ""),
+                # 갈래는 **저장 원본값**만 — 「…㎜ 이하」 구간 나누기는 원문을 읽는 쪽 몫.
+                "variant_axis": row.get("variant_axis"),
+                "variant_value": row.get("variant_value"),
+                "spec_class": None,
+                "spec_class_basis": str(row.get("blocked_reason") or ""),
+                "composite_not_ready": None,
+                "in_bill": ready,
+                "in_bill_reason": "" if ready else str(row.get("blocked_reason") or ""),
+                "origin": ORIGIN_PIPE,
+            }
+        )
+    return rows
+
+
 def _material_rows(material_table: dict[str, Any]) -> list[dict[str, Any]]:
     """자재 줄 — **공종코드를 붙이지 않는다.** 자재 축은 B09 카탈로그가 잇는다(8-7)."""
     rows: list[dict[str, Any]] = []
@@ -887,6 +936,7 @@ def build_handoff(
     unit_quantity_table: dict[str, Any] | None = None,
     material_table: dict[str, Any] | None = None,
     preparation_table: dict[str, Any] | None = None,
+    pipe_table: dict[str, Any] | None = None,
     mapping: WorkItemMapping | None = None,
     ground_class_set: str | None = None,
     ground_classes: list[str] | None = None,
@@ -914,6 +964,8 @@ def build_handoff(
 
     # 준비공·사방공 — **못 내는 줄도 사유와 함께** 보낸다(빼면 빠진 줄이 안 보인다).
     work_items.extend(_preparation_rows(preparation_table or {}))
+    # 배수관 — 정본 셋(관 지점·측점 연장·매핑)을 이은 결과. 못 서는 줄도 사유와 함께 감.
+    work_items.extend(_pipe_rows(pipe_table or {}))
 
     # 콘크리트 타설 — 품은 이 줄, 재료는 자재 쪽. 겹치지 않는다(위 `_placing_rows` 주석).
     placing_rows, placing_notes = _placing_rows(

B08_Quantity/B08_Quantity_Engine_Pipe.py

@@ -0,0 +1,148 @@
+"""배수관 물량 — **정본 셋을 잇기만 한다** (2026-09-08 두 창 합의).
+
+값이 어디서 오나
+    관 자체(있나·어디·관경·관종)  → `pipe_points.json`   (레지스트리 `pipe` 타입이
+                                    `managed_by: pipe_points`)
+    관 연장(m)                    → 측점 `design.pipe_length_m`
+                                    (B06 횡단이 **서버 Node 로** 계산해 m 단위 올림까지
+                                     끝낸 값을 정본에 남긴다 — 계산이 두 벌이 아니다)
+    관종 → 공종코드                → `work_item_mapping` 의 `pipe.kind_codes`
+
+⚠ **여기서 길이를 짓지 않는다.** 앞서 「도로폭 = 관 길이」처럼 잡을 뻔했는데 그것이 곧
+  임의 수치다. 연장이 없는 관은 **줄을 세우되 막힌 사유와 함께** 보낸다.
+
+⚠ **`facility` 가 `pipe` 인 점만 배관이다.**
+  `pipe_points.json` 은 **계곡 통과 시설 전부의 정본**이라 BOX암거·물넘이·세월교·독립
+  기슭막이가 같은 파일에 있다. `pipe_diameter_mm` 유무로 가르면 **관경 미지정 관을 놓친다**
+  (실측: `5601e828` 11점 중 2점이 `facility: ford_bridge`).
+
+⚠ **유출·유입부 기슭막이는 여기서 세지 않는다.**
+  관 옵션(`outlet_revet_*`)이 정본이고 구조물 목록에서는 빠졌다(2026-08-28 이관).
+  구조물 쪽으로 또 세면 이중계상이다.
+
+⚠ **터파기·되메우기를 관 줄에 붙이지 않는다.**
+  관 부설과 굴착이 각각 오면 **같은 굴착을 두 번** 센다(B09 ㉡ 가드와 같은 자리).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+#: 배관으로 보는 `facility` 값. 그 밖(BOX암거·물넘이·세월교·독립 기슭막이)은 관이 아니다.
+FACILITY_PIPE = "pipe"
+
+#: 연장을 못 찾은 줄의 막힌 갈래 — 「입력하면 풀림」이 아니라 **앞 단계가 내야 하는 값**이다.
+BLOCKED_LENGTH_MISSING = "input_missing"
+
+NOTE_LENGTH_MISSING = (
+    "관 연장이 아직 정본에 없습니다 — 횡단설계에서 [저장]을 한 번 누르면 "
+    "그 측점의 관 길이가 남고 값이 섭니다"
+)
+NOTE_KIND_DEFAULT = "관종을 안 정해 기본값({kind})으로 섰습니다 — 정하면 공종이 갈립니다"
+NOTE_KIND_UNKNOWN = "「{kind}」은(는) 아는 관종이 아니라 공종을 못 골랐습니다"
+
+
+def _length_by_chainage(designs: list[dict[str, Any]], key: str) -> dict[float, float]:
+    """측점별 관 길이. **없는 측점은 담지 않는다** — 0 으로 채우면 「없음」과 구별이 안 된다."""
+    found: dict[float, float] = {}
+    for row in designs or []:
+        design = row.get("design") if isinstance(row, dict) else None
+        if not isinstance(design, dict):
+            continue
+        value = design.get(key)
+        if value is None:
+            continue
+        try:
+            length = float(value)
+        except (TypeError, ValueError):
+            continue
+        if length > 0:
+            found[round(float(row.get("chainage_m") or 0.0), 3)] = length
+    return found
+
+
+def _nearest(lengths: dict[float, float], chainage: float, tolerance: float = 0.5) -> float | None:
+    """관 측점과 단면 측점이 소수점에서 어긋날 수 있어 **가까운 것**을 본다.
+
+    ⚠ 좁게 본다(기본 0.5m) — 넓히면 옆 측점의 길이를 물어 와 조용히 틀린다.
+    """
+    if not lengths:
+        return None
+    key = round(chainage, 3)
+    if key in lengths:
+        return lengths[key]
+    best = min(lengths, key=lambda x: abs(x - chainage))
+    return lengths[best] if abs(best - chainage) <= tolerance else None
+
+
+def build_rows(
+    pipe_points: list[dict[str, Any]],
+    designs: list[dict[str, Any]],
+    mapping: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    """관 줄 목록. **값이 서는 줄도, 못 서는 줄도** 함께 낸다.
+
+    `pipe_points` 는 `PipePoint.model_dump()` 또는 같은 모양의 딕셔너리 목록이다.
+    """
+    table = mapping or {}
+    kind_codes: dict[str, str] = table.get("kind_codes") or {}
+    kind_key = str(table.get("kind_option_key") or "pipe_kind")
+    default_kind = str(table.get("default_kind") or "")
+    length_key = str(table.get("length_key") or "pipe_length_m")
+    diameter_key = str(table.get("diameter_option_key") or "pipe_diameter_mm")
+
+    lengths = _length_by_chainage(designs, length_key)
+    rows: list[dict[str, Any]] = []
+    notes: list[str] = []
+
+    for point in pipe_points or []:
+        if str(point.get("facility") or FACILITY_PIPE) != FACILITY_PIPE:
+            continue  # 배관이 아닌 시설 — 그쪽 줄은 그쪽이 센다
+        options = point.get("options") or {}
+        chainage = float(point.get("chainage_m") or 0.0)
+
+        stored_kind = str(options.get(kind_key) or "").strip()
+        kind = stored_kind or default_kind
+        code = kind_codes.get(kind)
+        kind_note = ""
+        if not stored_kind and default_kind:
+            kind_note = NOTE_KIND_DEFAULT.format(kind=default_kind)
+        elif stored_kind and code is None:
+            kind_note = NOTE_KIND_UNKNOWN.format(kind=stored_kind)
+
+        length = _nearest(lengths, chainage)
+        blocked_kind = None if length else BLOCKED_LENGTH_MISSING
+        blocked_reason = "" if length else NOTE_LENGTH_MISSING
+        if code is None:
+            blocked_kind = blocked_kind or BLOCKED_LENGTH_MISSING
+            blocked_reason = blocked_reason or kind_note
+
+        rows.append(
+            {
+                "chainage_m": chainage,
+                "work_item_code": code,
+                "kind": kind,
+                "kind_from_default": not stored_kind,
+                # 갈래는 **저장 원본값**만 보낸다 — 「…㎜ 이하」 구간 나누기는 원문을 읽는 쪽 몫.
+                "variant_axis": diameter_key,
+                "variant_value": options.get(diameter_key),
+                "unit": str(table.get("unit") or "m"),
+                "quantity": float(length or 0.0),
+                "blocked_kind": blocked_kind,
+                "blocked_reason": blocked_reason or kind_note,
+                "in_bill": bool(length and code),
+            }
+        )
+        if kind_note and kind_note not in notes:
+            notes.append(kind_note)
+
+    missing = sum(1 for row in rows if not row["in_bill"])
+    if missing:
+        notes.append(f"관 {len(rows)}개 중 {missing}개가 아직 값이 안 섭니다")
+    return {
+        "rows": rows,
+        "notes": notes,
+        "pipe_count": len(rows),
+        "ready_count": sum(1 for row in rows if row["in_bill"]),
+        "length_total_m": round(sum(row["quantity"] for row in rows if row["in_bill"]), 3),
+    }

B08_Quantity/B08_Quantity_Router_Earthwork.py

@@ -83,6 +83,16 @@ async def get_earthwork_table(project_id: UUID, route_id: int) -> JSONResponse:
     settings, project_root = await _project_settings(project_id)
     plan = await _stored_haul_plan(project_id, route_id)
     haul = build_haul_table(plan)
+    # 배수관 연장 — B06 이 측점 `design.pipe_length_m` 에 남긴 값. **여기서 짓지 않는다.**
+    #   인계가 관 줄을 세울 때 쓴다. 단면을 두 번 읽지 않으려고 이 응답에 실어 보낸다.
+    table["pipe_lengths"] = [
+        {
+            "chainage_m": row.get("chainage_m"),
+            "pipe_length_m": (row.get("design") or {}).get("pipe_length_m"),
+        }
+        for row in designs
+        if isinstance(row, dict) and (row.get("design") or {}).get("pipe_length_m") is not None
+    ]
     table["haul"] = haul
     # 운반계획은 [저장]·[확정]에서 정본에 남는 값이다 — 아직 없으면 빈 표가 정직하다.
     table["haul_available"] = bool(plan)

B08_Quantity/B08_Quantity_Router_Material.py

@@ -15,7 +15,9 @@
 
 from __future__ import annotations
 
+import json
 import logging
+from pathlib import Path
 from typing import Any
 from uuid import UUID
 
@@ -25,7 +27,7 @@ from fastapi.responses import JSONResponse
 from B03_FileInput.B03_FileInput_Repository import get_project_storage_relative_path
 from B05_Profile.B05_Profile_Structures_Repository import load_structures
 from B05_Profile.B05_Profile_Structures_Schema import structure_type_map
-from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff, summarize
+from B08_Quantity.B08_Quantity_Engine_Handoff import build_handoff, load_mapping, summarize
 from B08_Quantity.B08_Quantity_Engine_MaterialSummary import build_table as build_material_table
 from B08_Quantity.B08_Quantity_Engine_UnitQuantity import build_table as build_unit_table
 from common_util.common_util_project_settings import (
@@ -161,6 +163,8 @@ async def get_handoff(project_id: UUID) -> JSONResponse:
         material_table=material_table,
         # 준비공·사방공 — 값이 서는 줄도, 못 내는 줄도 함께 넘긴다(빼면 빠진 줄이 안 보임).
         preparation_table=earthwork.get("preparation"),
+        # 배수관 — 관 정본은 `pipe_points.json`, 연장은 측점 `design.pipe_length_m` 다.
+        pipe_table=_pipe_table(project_root, earthwork.get("pipe_lengths") or []),
         ground_class_set=settings.get("rock_class_set"),
         ground_classes=rock_classes(settings),
         ground_methods={name: rock_method(settings, name) for name in rock_classes(settings)},
@@ -173,6 +177,38 @@ async def get_handoff(project_id: UUID) -> JSONResponse:
     return JSONResponse(content=handoff)
 
 
+def _pipe_table(project_root: str, pipe_lengths: list[dict[str, Any]]) -> dict[str, Any]:
+    """배수관 표 — 정본 셋을 읽어 잇는다. 못 읽으면 **빈 표**(줄이 안 서는 것이 정직하다).
+
+    ⚠ 관 정본은 `structures.json` 이 아니라 `pipe_points.json` 이다
+      (레지스트리 `pipe` 타입이 `managed_by: pipe_points`).
+    """
+    from B08_Quantity.B08_Quantity_Engine_Pipe import build_rows as build_pipe_rows
+    from common_util.common_util_drainage_pipes import pipe_points_path_in
+
+    path = pipe_points_path_in(Path(project_root))
+    if not path.is_file():
+        return {"rows": [], "notes": [], "pipe_count": 0, "ready_count": 0, "length_total_m": 0.0}
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, ValueError):
+        logger.exception("B08 관 지점 읽기 실패: %s", path)
+        return {
+            "rows": [],
+            "notes": ["관 지점 파일을 읽지 못했습니다"],
+            "pipe_count": 0,
+            "ready_count": 0,
+            "length_total_m": 0.0,
+        }
+    points = payload.get("points") or payload.get("items") or []
+    # 토적표 라우터가 실어 준 모양을 엔진이 읽는 모양으로 옮긴다.
+    designs = [
+        {"chainage_m": row.get("chainage_m"), "design": {"pipe_length_m": row.get("pipe_length_m")}}
+        for row in pipe_lengths
+    ]
+    return build_pipe_rows(points, designs, (load_mapping().pipe or {}))
+
+
 async def _earthwork_tables(project_id: UUID) -> dict[str, Any]:
     """토적표 라우터가 만든 집계·운반 표를 얻는다. 노선이 없으면 빈 값."""
     from B08_Quantity.B08_Quantity_Router_Earthwork import (

resources/data_work_item_mapping/work_item_mapping_2026-01-01.json

@@ -277,5 +277,27 @@
     ],
     "b09_does": "원문 표기(물결표·공백·괄호)를 흡수해 자기 키로 옮긴다. 「…㎝ 이하」 구간 나누기도 그쪽 몫 — 그 구간이 **품셈 표의 구조**이기 때문."
   },
-  "masonry_class_reference": "resources/data_masonry/masonry_class_2026-01-01.json — ⚠ 이제 **참고용**이다. 서브 판정과 어긋나면 그것이 곧 신호다."
+  "masonry_class_reference": "resources/data_masonry/masonry_class_2026-01-01.json — ⚠ 이제 **참고용**이다. 서브 판정과 어긋나면 그것이 곧 신호다.",
+  "pipe": {
+    "note": "배수관(횡단배수관) — **관종으로 공종이 갈린다.** 관은 `structures.json` 이 아니라 `pipe_points.json` 이 정본이고(레지스트리 `pipe` 타입이 `managed_by: pipe_points`), 연장은 B06 횡단이 서버 Node 로 계산해 측점 `design.pipe_length_m` 로 남긴다(2026-09-08 랩탑 창). B08 은 그 셋을 잇기만 한다.",
+    "kind_codes": {
+      "파형강관": "FP-12-11-03",
+      "흄관": "FP-12-11-02",
+      "VR관": "FP-12-11-01"
+    },
+    "kind_option_key": "pipe_kind",
+    "default_kind": "파형강관",
+    "default_is_user_confirmed": true,
+    "default_note": "레지스트리 `pipe` 옵션의 기본값이며 **2026-08-17 사용자 확정**임. 그래도 저장값이 비어 있으면 「관종 미지정」으로 드러내고 기본값으로 돈다는 사실을 함께 싣는다(암 시공법과 같은 처리).",
+    "unit": "m",
+    "length_key": "pipe_length_m",
+    "diameter_option_key": "pipe_diameter_mm",
+    "variant_axis": "pipe_diameter_mm",
+    "facility_rule": "⚠ `pipe_points.json` 은 **계곡 통과 시설 전부의 정본**이다(배관·BOX암거·물넘이·세월교·독립 기슭막이). `facility` 가 `pipe` 인 점만 배관이다 — `pipe_diameter_mm` 유무로 가르면 **관경 미지정 관을 놓친다**(2026-09-08 랩탑 창).",
+    "revetment_note": "⚠ 유출·유입부 기슭막이는 **관 옵션(`inlet_revet_*`·`outlet_revet_*`)이 정본**이다. 레지스트리 `revetment` 타입이 `managed_by: pipe_points` 라 구조물 목록에서 빠졌으므로 **구조물 쪽으로 또 세지 않는다**(2026-08-28 이관).",
+    "not_ready": {
+      "흄관 밑수 두 벌": "`FP-12-11-02` 는 밑수가 「1 m」와 「1 개소」 두 벌이다(표가 둘). B09 가 `#갈래` 로 두 표를 각각 세우므로 B08 은 `variant_value` 로 어느 쪽인지 보내면 된다.",
+      "터파기·되메우기": "⚠ 관 부설과 터파기·되메우기가 각각 오면 **같은 굴착을 두 번 셀 수 있다**(B09 ㉡ 가드). 관 줄에는 지금 터파기를 붙이지 않는다."
+    }
+  }
 }