diff --git a/README.md b/README.md
index 5f00738..b33ebea 100644
--- a/README.md
+++ b/README.md
@@ -27,8 +27,8 @@ Live through COM on Windows, or headless through ezdxf anywhere — one typed co
 
 </div>
 
-> **v1.5 release snapshot:** 241 tools · 8 resources · 5 prompt templates · 4516 collected tests.
-> 241 is the **registered** count; a default install advertises 236 over `tools/list`,
+> **v1.5 release snapshot:** 242 tools · 8 resources · 5 prompt templates · 4516 collected tests.
+> 242 is the **registered** count; a default install advertises 237 over `tools/list`,
 > because `ENABLE_3D` is unset. `system_about` is the runtime authority.
 
 ## Why this exists
@@ -122,7 +122,7 @@ Claude Desktop, Cursor, or any stdio MCP host. For HTTP: `autocad-mcp --transpor
 | Quality loop | `drawing_preflight` → `drawing_plan` → `drawing_critique` → `drawing_refine` → `drawing_finalize` (0–100 score) |
 | Delivery | `drawing_deliver`: DXF/PDF/PNG + SHA-256 manifest + reopen-parity checks |
 
-<sub>241 tools in 26 groups; <code>TOOL_PACKS=core</code> hides the nine <code>pid_*</code> tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (<code>TOOL_PACKS=core,mech</code> keeps the mechanical ones, <code>TOOL_PACKS=core,arch</code> the architectural ones) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (<code>autocad://drawing/info</code>, <code>layers</code>, <code>blocks</code>, <code>entities/stats</code>, <code>entities/{layer_name}</code>, <code>system/status</code>, <code>pid/symbols</code>, <code>standards/isa51</code>) and 5 prompt templates.</sub>
+<sub>242 tools in 26 groups; <code>TOOL_PACKS=core</code> hides the nine <code>pid_*</code> tools, the 22 environment tools, the 14 mechanical tools and the 12 architectural tools (<code>TOOL_PACKS=core,mech</code> keeps the mechanical ones, <code>TOOL_PACKS=core,arch</code> the architectural ones) from a client that needs none of them. Plus 8 resources that cost nothing in the tool budget (<code>autocad://drawing/info</code>, <code>layers</code>, <code>blocks</code>, <code>entities/stats</code>, <code>entities/{layer_name}</code>, <code>system/status</code>, <code>pid/symbols</code>, <code>standards/isa51</code>) and 5 prompt templates.</sub>
 
 **Two rules worth knowing.** Every coordinate in and out of a tool is WCS on both engines — the one exception is TEXT `rotation`, which stays in the entity frame because a mirrored TEXT is mirror-imaged and no scalar angle expresses that. And never read vertices back and shoelace them: that loses **28.2%** of the area on a semicircular edge, silently. `analysis_measure_entity(handle)` reads the real geometry and states its own accuracy.
 
diff --git a/discovery/aliases.py b/discovery/aliases.py
index 7286ead..700b8ef 100644
--- a/discovery/aliases.py
+++ b/discovery/aliases.py
@@ -2465,6 +2465,28 @@ TOOL_ALIASES: dict[str, ToolAliases] = {
             "mimari plan",
         ),
     ),
+    "pipe_takeoff": ToolAliases(
+        acad=(),
+        synonyms=(
+            "pipe takeoff",
+            "piping takeoff",
+            "takeoff",
+            "pipe quantities",
+            "how much pipe",
+            "pipe lengths from a p&id",
+            "bill of quantities",
+            "boq",
+            "metraj",
+            "boru metrajı",
+            "boru metraji",
+            "metre cetveli",
+            "boru listesi",
+            "спецификация",
+            "ведомость",
+            "спецификация трубопроводов",
+            "ведомость трубопроводов",
+        ),
+    ),
 }
 
 
diff --git a/docs/tool-inventory.json b/docs/tool-inventory.json
index 6874746..7890aff 100644
--- a/docs/tool-inventory.json
+++ b/docs/tool-inventory.json
@@ -3,13 +3,13 @@
   "schema_version": 1,
   "package_version": "1.5.1",
   "totals": {
-    "tools": 241,
+    "tools": 242,
     "resources": 8,
     "prompts": 5,
     "groups": 26
   },
   "groups": {
-    "analysis": 12,
+    "analysis": 13,
     "architecture": 12,
     "batch": 3,
     "blocks": 9,
@@ -2524,6 +2524,21 @@
         "readOnlyHint": true
       }
     },
+    {
+      "name": "pipe_takeoff",
+      "group": "analysis",
+      "cost": "safe",
+      "summary": "Pipe takeoff from a P&ID and its layout: rows by room, service and diameter.",
+      "tags": [
+        "analysis",
+        "plant"
+      ],
+      "annotations": {
+        "title": "Plant: Pipe Takeoff",
+        "readOnlyHint": false,
+        "destructiveHint": false
+      }
+    },
     {
       "name": "plot_style_list",
       "group": "page_setup",
diff --git a/engineering/understand/draw.py b/engineering/understand/draw.py
new file mode 100644
index 0000000..2882ec9
--- /dev/null
+++ b/engineering/understand/draw.py
@@ -0,0 +1,86 @@
+"""The async layer of track H's takeoffs: read the drawings, run the pure engine, write the workbook.
+
+The only module of the takeoffs that touches a backend, and it only reads:
+`take_snapshot` gives one bulk read of each drawing (a file, or the current
+document), the pure modules compute in a worker thread so a 35,000-entity
+P&ID does not stall the server, and the workbook is a file beside the path the
+caller names. The drawings themselves are never modified.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import math
+
+from engineering.understand.network import build_network
+from engineering.understand.report import check_lang, write_workbook
+from engineering.understand.snapshot import take_snapshot
+from engineering.understand.takeoff import check_pipe_options, pipe_rows
+
+__all__ = ["run_pipe_takeoff"]
+
+
+def _check_tol(tol) -> float:
+    try:
+        value = float(tol)
+    except (TypeError, ValueError):
+        raise ValueError(f"tol: expected a number, got {tol!r}") from None
+    if not math.isfinite(value) or value <= 0.0:
+        raise ValueError(f"tol: must be a finite number greater than zero, got {tol!r}")
+    return value
+
+
+async def run_pipe_takeoff(
+    backend,
+    *,
+    pid_path: str | None = None,
+    layout_path: str | None = None,
+    output: str | None = None,
+    lang: str = "en",
+    services=None,
+    allowance: float = 0.20,
+    vertical_allowance: float = 0.0,
+    length_source: str = "auto",
+    force: bool = False,
+    layers=None,
+    tol: float = 1.0,
+) -> dict:
+    """`pipe_rows` over a P&ID and an optional layout, plus the workbook when `output` is named.
+
+    Every option is checked before a drawing is read. `pid_path=None` reads the
+    current document as the P&ID. The result is `pipe_rows`'s dict with
+    ``files`` added: `write_workbook`'s answer, or None when no `output` was named.
+    """
+    check_lang(lang)
+    options = check_pipe_options(
+        services=services,
+        allowance=allowance,
+        vertical_allowance=vertical_allowance,
+        length_source=length_source,
+    )
+    tol = _check_tol(tol)
+    if length_source == "layout" and not layout_path:
+        raise ValueError("length_source='layout' needs a layout drawing; none was given")
+    pid = await take_snapshot(backend, pid_path)
+    layout = await take_snapshot(backend, layout_path) if layout_path else None
+
+    def compute() -> dict:
+        network = build_network(pid, layers=layers, tol=tol)
+        return pipe_rows(
+            network,
+            pid,
+            layout,
+            services=options["services"],
+            allowance=options["allowance"],
+            vertical_allowance=options["vertical_allowance"],
+            length_source=options["length_source"],
+            force=bool(force),
+        )
+
+    result = await asyncio.to_thread(compute)
+    result["files"] = (
+        await asyncio.to_thread(write_workbook, result, kind="pipe", lang=lang, path=output)
+        if output
+        else None
+    )
+    return result
diff --git a/engineering/understand/report.py b/engineering/understand/report.py
new file mode 100644
index 0000000..b96c220
--- /dev/null
+++ b/engineering/understand/report.py
@@ -0,0 +1,493 @@
+"""The takeoff workbook: Metraj, Özet, Kontrol and Metodoloji, as XLSX and as CSV.
+
+The sheets follow the user's own takeoff workbook (spec §8): *Metraj* holds
+the rows, *Özet* the sums, *Kontrol* what needs a person's eye and
+*Metodoloji* how every number was reached - generated from the result, never
+typed. The sheet names stay as the user's workbook names them; the headers,
+the service names and the method text follow `lang` (tr / en / ru).
+
+CSV is always written, one file per sheet beside the workbook
+(``<stem>_metraj.csv`` ...), UTF-8 with a byte-order mark so a spreadsheet
+opens the Cyrillic and Turkish letters, a point as the decimal separator. XLSX
+needs openpyxl (the `office` extra); without it the CSV files are still
+written and the result carries the `xlsx_write` refusal instead of a path.
+Numbers are written rounded to 1 mm (three decimals of a metre).
+"""
+
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+
+from backends.capability import UnsupportedCapabilityError
+from engineering.sheet.extract import xlsx_available
+
+__all__ = ["CSV_SLUGS", "LABELS", "LANGS", "SHEETS", "check_lang", "write_workbook"]
+
+LANGS = ("tr", "en", "ru")
+SHEETS = ("Metraj", "Özet", "Kontrol", "Metodoloji")
+CSV_SLUGS = {"Metraj": "metraj", "Özet": "ozet", "Kontrol": "kontrol", "Metodoloji": "metodoloji"}
+_INDEX = {lang: i for i, lang in enumerate(LANGS)}
+
+#: Every word the workbook prints, as (tr, en, ru).
+LABELS: dict[str, tuple[str, str, str]] = {
+    # Metraj
+    "room": ("Mahal", "Room", "Помещение"),
+    "service": ("Hizmet", "Service", "Среда"),
+    "diameter": ("Çap", "Diameter", "Диаметр"),
+    "direct_m": ("Doğrudan (m)", "Direct (m)", "Прямо (м)"),
+    "continuity_m": ("Süreklilik (m)", "Continuity (m)", "По непрерывности (м)"),
+    "unassigned_m": ("Atanmamış (m)", "Unassigned (m)", "Не назначено (м)"),
+    "net_m": ("Net (m)", "Net (m)", "Нетто (м)"),
+    "allowance": ("Pay oranı", "Allowance", "Запас"),
+    "allowance_m": ("Pay (m)", "Allowance (m)", "Запас (м)"),
+    "total_m": ("Paylı toplam (m)", "Total incl. allowance (m)", "Итого с запасом (м)"),
+    "schematic_m": ("Şematik (m)", "Schematic (m)", "По схеме (м)"),
+    "status": ("Durum", "Status", "Статус"),
+    "runs": ("Hatlar", "Runs", "Линии"),
+    # Özet, Kontrol, Metodoloji
+    "group": ("Grup", "Group", "Группа"),
+    "key": ("Anahtar", "Key", "Ключ"),
+    "run": ("Hat", "Run", "Линия"),
+    "kind": ("Tür", "Kind", "Вид"),
+    "tags": ("Etiketler", "Tags", "Теги"),
+    "reason": ("Neden", "Reason", "Причина"),
+    "topic": ("Konu", "Topic", "Раздел"),
+    "value": ("Değer", "Value", "Значение"),
+    "by_service": ("Hizmete göre", "By service", "По среде"),
+    "by_diameter": ("Çapa göre", "By diameter", "По диаметру"),
+    "by_room": ("Mahale göre", "By room", "По помещению"),
+    "total": ("TOPLAM", "TOTAL", "ИТОГО"),
+    "routed": ("Yönlendirilen", "Routed", "Трассировано"),
+    "unroutable_total": ("Yönlendirilemeyen (C)", "Unroutable (C)", "Без трассы (C)"),
+    "yes": ("evet", "yes", "да"),
+    "no": ("hayır", "no", "нет"),
+    # services
+    "product": ("Ürün", "Product", "Продукт"),
+    "cip_supply": ("CIP besleme", "CIP supply", "CIP подача"),
+    "cip_return": ("CIP dönüş", "CIP return", "CIP возврат"),
+    "glycol_supply": ("Glikol besleme", "Glycol supply", "Гликоль подача"),
+    "glycol_return": ("Glikol dönüş", "Glycol return", "Гликоль возврат"),
+    "steam": ("Buhar", "Steam", "Пар"),
+    "condensate": ("Kondens", "Condensate", "Конденсат"),
+    "cold_water": ("Soğuk su", "Cold water", "Холодная вода"),
+    "hot_water": ("Sıcak su", "Hot water", "Горячая вода"),
+    "ice_water": ("Buzlu su", "Ice water", "Ледяная вода"),
+    "compressed_air": ("Basınçlı hava", "Compressed air", "Сжатый воздух"),
+    "drain": ("Drenaj", "Drain", "Дренаж"),
+    "electrical": ("Elektrik", "Electrical", "Электрика"),
+    "unknown": ("Bilinmeyen", "Unknown", "Неизвестно"),
+    # Kontrol kinds and reasons
+    "unassigned_diameter": ("Çap atanmamış", "Diameter unassigned", "Диаметр не назначен"),
+    "unroutable": ("Yönlendirilemez", "Unroutable", "Трассировка невозможна"),
+    "end_not_on_equipment": (
+        "Bir uç ekipmanda değil",
+        "An end is on no equipment",
+        "Конец не на оборудовании",
+    ),
+    "single_tag": ("Tek etiket", "Only one tag", "Только один тег"),
+    "tag_not_on_layout": (
+        "Etiket yerleşimde yok",
+        "Tag not on the layout",
+        "Тега нет на планировке",
+    ),
+    "tag_ambiguous": (
+        "Etiket yerleşimde birden çok kez yazılı",
+        "Tag written more than once on the layout",
+        "Тег повторяется на планировке",
+    ),
+    "no_diameter_label": ("Çap etiketi yok", "No diameter label", "Нет обозначения диаметра"),
+    # Metodoloji topics
+    "source_pid": ("Kaynak: P&ID", "Source: P&ID", "Источник: P&ID"),
+    "source_layout": ("Kaynak: yerleşim", "Source: layout", "Источник: планировка"),
+    "length_source": ("Uzunluk kaynağı", "Length source", "Источник длины"),
+    "choice": ("Seçim gerekçesi", "Why this source", "Почему этот источник"),
+    "scale_verdict": ("Ölçek kontrolü: sonuç", "Scale check: verdict", "Проверка масштаба: вывод"),
+    "scale_factor": (
+        "Ölçek kontrolü: oran (P&ID mm / yerleşim mm)",
+        "Scale check: factor (P&ID mm per layout mm)",
+        "Проверка масштаба: коэффициент (мм P&ID на мм планировки)",
+    ),
+    "scale_iqr": ("Ölçek kontrolü: IQR", "Scale check: IQR", "Проверка масштаба: IQR"),
+    "scale_within": (
+        "Ölçek kontrolü: ±%10 içindeki çiftler",
+        "Scale check: pairs within ±10 %",
+        "Проверка масштаба: пары в пределах ±10 %",
+    ),
+    "scale_pairs": ("Ölçek kontrolü: çift sayısı", "Scale check: pairs", "Проверка масштаба: пар"),
+    "scale_matched": (
+        "Ölçek kontrolü: eşleşen etiketler",
+        "Scale check: matched tags",
+        "Проверка масштаба: совпавшие теги",
+    ),
+    "scale_ambiguous": (
+        "Ölçek kontrolü: belirsiz etiketler",
+        "Scale check: ambiguous tags",
+        "Проверка масштаба: неоднозначные теги",
+    ),
+    "scale_note": ("Ölçek kontrolü: not", "Scale check: note", "Проверка масштаба: примечание"),
+    "scale_verified": ("Ölçek doğrulandı", "Scale verified", "Масштаб подтверждён"),
+    "units_pid": ("Birim: P&ID", "Units: P&ID", "Единицы: P&ID"),
+    "units_layout": ("Birim: yerleşim", "Units: layout", "Единицы: планировка"),
+    "rule_route": ("Kural: güzergâh", "Rule: route", "Правило: трасса"),
+    "rule_vertical": (
+        "Kural: bağlantı başına düşey pay (m)",
+        "Rule: vertical allowance per connection (m)",
+        "Правило: вертикальный запас на соединение (м)",
+    ),
+    "rule_allowance": ("Kural: pay oranı", "Rule: allowance", "Правило: запас"),
+    "rule_diameter": ("Kural: çap", "Rule: diameter", "Правило: диаметр"),
+    "rule_service": ("Kural: hizmet", "Rule: service", "Правило: среда"),
+    "rule_status": ("Kural: durum", "Rule: status", "Правило: статус"),
+    "rule_rooms": ("Kural: mahaller", "Rule: rooms", "Правило: помещения"),
+    "rule_multi": (
+        "Kural: çok çaplı hat",
+        "Rule: several diameters",
+        "Правило: несколько диаметров",
+    ),
+    "tol": ("Tolerans (çizim birimi)", "Tolerance (drawing units)", "Допуск (ед. чертежа)"),
+    "label_search": (
+        "Etiket arama mesafesi (çizim birimi)",
+        "Label search distance (drawing units)",
+        "Радиус поиска обозначений (ед. чертежа)",
+    ),
+    "overlap": (
+        "Çift çizilmiş, bir kez sayılan boru (m)",
+        "Double-drawn pipe counted once (m)",
+        "Дважды начерченная труба, учтена один раз (м)",
+    ),
+    "services": ("Kapsanan hizmetler", "Services taken off", "Учтённые среды"),
+    "warning": ("Uyarı", "Warning", "Предупреждение"),
+}
+
+#: The method sheet's sentences, as (tr, en, ru).
+TEXTS: dict[str, tuple[str, str, str]] = {
+    "requested": ("çağıran seçti", "chosen by the caller", "выбран вызывающим"),
+    "forced": (
+        "ölçek kontrolü P&ID'yi şematik bulduğu halde istek üzerine (force=True) P&ID'den ölçüldü",
+        "measured on the P&ID on request (force=True) although the scale check calls it schematic",
+        "измерено по P&ID по запросу (force=True), хотя проверка масштаба считает его схемой",
+    ),
+    "no_layout": (
+        "yerleşim verilmedi: P&ID'den ölçüldü, ölçek doğrulanmadı",
+        "no layout given: measured on the P&ID, scale not verified",
+        "планировка не задана: измерено по P&ID, масштаб не подтверждён",
+    ),
+    "auto_to_scale": (
+        "ölçek kontrolü P&ID'yi ölçekli buldu: oran üzerinden P&ID'den ölçüldü",
+        "the scale check calls the P&ID to scale: measured on it through the factor",
+        "проверка масштаба считает P&ID масштабным: измерено по нему через коэффициент",
+    ),
+    "auto_layout": (
+        "ölçek kontrolü P&ID'yi ölçekli bulmadı: yerleşimden ölçüldü",
+        "the scale check does not call the P&ID to scale: measured on the layout",
+        "проверка масштаба не считает P&ID масштабным: измерено по планировке",
+    ),
+    "scale_note": (
+        "Etiketler ekipman merkezi değildir: ±%10 bant ve 2 m alt sınır bunu karşılar.",
+        "Tag labels are not equipment centres: the ±10 % band and the 2 m floor absorb that.",
+        "Теги — не центры оборудования: полоса ±10 % и порог 2 м это учитывают.",
+    ),
+    "rule_route": (
+        "Yerleşim uzunluğu = hattın bağladığı etiket konumlarının dik açılı (Manhattan) en küçük "
+        "kapsayan ağacı; her bağlantı önce x, sonra y boyunca.",
+        "Layout length = the rectilinear (Manhattan) minimum spanning tree of the tag positions "
+        "the run joins; each connection along x first, then y.",
+        "Длина по планировке = прямоугольное (манхэттенское) минимальное остовное дерево "
+        "позиций тегов, которые соединяет линия; каждое соединение сначала по x, затем по y.",
+    ),
+    "rule_diameter": (
+        "Çap etiketleri çizildiği gibi alınır (Ø51, SMS51, DN20 farklı ölçülerdir). Doğrudan: "
+        "arama mesafesindeki etiket; süreklilik: etiketli parçadan hat boyunca taşınır, "
+        "redüksiyondan geçmez; atanmamış: ikisi de değil (Kontrol sayfasında).",
+        "Diameter labels are kept as drawn (Ø51, SMS51 and DN20 are different sizes). Direct: a "
+        "label within the search distance; continuity: carried along the run from a labelled "
+        "piece, never across a reducer; unassigned: neither (listed on Kontrol).",
+        "Обозначения диаметров сохраняются как на чертеже (Ø51, SMS51 и DN20 — разные "
+        "размеры). Прямо: обозначение в радиусе поиска; по непрерывности: перенос вдоль линии "
+        "от обозначенного участка, но не через переход; не назначено: ни то, ни другое (лист "
+        "Kontrol).",
+    ),
+    "rule_service": (
+        "Hizmet katmandan alınır; hattın yakınındaki besleme / dönüş sözcüğü (ПОДАЧА, ВОЗВРАТ, "
+        "BESLEME, DÖNÜŞ, SUPPLY, RETURN, AANVOER, RETOUR) tüm hat için onu geçersiz kılar.",
+        "Service comes from the layer; a supply / return word near the run (ПОДАЧА, ВОЗВРАТ, "
+        "BESLEME, DÖNÜŞ, SUPPLY, RETURN, AANVOER, RETOUR) overrides it for the whole run.",
+        "Среда определяется по слою; слово подачи / возврата рядом с линией (ПОДАЧА, ВОЗВРАТ, "
+        "BESLEME, DÖNÜŞ, SUPPLY, RETURN, AANVOER, RETOUR) переопределяет её для всей линии.",
+    ),
+    "rule_status": (
+        "A: her uç yerleşimde bulunan bir etikette ve her çap doğrudan. B: süreklilikle ya da "
+        "atanmamış çap. C: yönlendirilemez - yalnız şematik uzunluk, Kontrol sayfasında, "
+        "toplama katılır ve işaretlenir.",
+        "A: every end on a tag placed on the layout and every diameter direct. B: a diameter by "
+        "continuity or unassigned. C: not routable - schematic length only, on Kontrol, counted "
+        "in the total and flagged.",
+        "A: каждый конец на теге, найденном на планировке, и каждый диаметр прямо. B: диаметр "
+        "по непрерывности или не назначен. C: трассировка невозможна — только длина по схеме, "
+        "на листе Kontrol, входит в итог и помечена.",
+    ),
+    "rule_rooms": (
+        "Yönlendirilen uzunluk, güzergâhın geçtiği yerleşim mahallerine paylaştırılır (duvar "
+        "yüzleri ve mahal etiketlerinden çokgenler); çokgen yoksa hatta en yakın mahal etiketine.",
+        "A routed length is shared among the layout rooms its route crosses (room polygons from "
+        "the wall faces and room labels); without polygons, the room label nearest the run.",
+        "Трассированная длина распределяется по помещениям планировки, через которые проходит "
+        "трасса (контуры из граней стен и обозначений помещений); без контуров — ближайшему "
+        "обозначению помещения.",
+    ),
+    "rule_multi": (
+        "Birden çok çaplı bir hat, uzunluğunu P&ID üzerindeki çizili uzunlukları oranında "
+        "paylaştırır.",
+        "A run holding several diameters shares its length among them in proportion to their "
+        "drawn length on the P&ID.",
+        "Линия с несколькими диаметрами делит длину между ними пропорционально их длине на "
+        "схеме P&ID.",
+    ),
+}
+
+
+def check_lang(lang: str) -> str:
+    """Refuse a language the workbook does not speak, naming the three it does."""
+    if lang not in LANGS:
+        raise ValueError(f"lang: {lang!r} unknown; choose from {', '.join(LANGS)}")
+    return lang
+
+
+def _t(key: str, lang: str) -> str:
+    """A label in `lang`; a key with no label (a diameter, a room number) prints as itself."""
+    entry = LABELS.get(key)
+    return entry[_INDEX[lang]] if entry else key
+
+
+def _say(key: str, lang: str) -> str:
+    """One of the method sheet's sentences in `lang`."""
+    return TEXTS[key][_INDEX[lang]]
+
+
+def _num(value):
+    return None if value is None else round(float(value), 3)
+
+
+def _yes(value, lang: str):
+    return None if value is None else _t("yes" if value else "no", lang)
+
+
+def _pipe_sheets(result: dict, lang: str) -> dict[str, tuple[list, list]]:
+    columns = (
+        "room",
+        "service",
+        "diameter",
+        "direct_m",
+        "continuity_m",
+        "unassigned_m",
+        "net_m",
+        "allowance",
+        "allowance_m",
+        "total_m",
+        "schematic_m",
+        "status",
+        "runs",
+    )
+    rows = [
+        [
+            row["room"],
+            _t(row["service"], lang),
+            row["diameter"],
+            _num(row["direct_m"]),
+            _num(row["continuity_m"]),
+            _num(row["unassigned_m"]),
+            _num(row["net_m"]),
+            row["allowance"],
+            _num(row["allowance_m"]),
+            _num(row["total_m"]),
+            _num(row["schematic_m"]),
+            row["status"],
+            ", ".join(row["runs"]),
+        ]
+        for row in result["rows"]
+    ]
+    totals = result["totals"]
+    a = totals["allowance"]
+    rows.append(
+        [
+            _t("routed", lang),
+            None,
+            None,
+            _num(sum(r["direct_m"] for r in result["rows"])),
+            _num(sum(r["continuity_m"] for r in result["rows"])),
+            _num(sum(r["unassigned_m"] for r in result["rows"])),
+            _num(totals["routed_m"]),
+            a,
+            _num(totals["routed_m"] * a),
+            _num(totals["routed_m"] * (1.0 + a)),
+            _num(sum(r["schematic_m"] for r in result["rows"])),
+            None,
+            None,
+        ]
+    )
+    rows.append(
+        [
+            _t("unroutable_total", lang),
+            None,
+            None,
+            None,
+            None,
+            None,
+            _num(totals["unroutable_m"]),
+            a,
+            _num(totals["unroutable_m"] * a),
+            _num(totals["unroutable_m"] * (1.0 + a)),
+            _num(totals["unroutable_m"]),
+            "C",
+            str(totals["flagged_runs"]),
+        ]
+    )
+    rows.append(
+        [
+            _t("total", lang),
+            None,
+            None,
+            None,
+            None,
+            None,
+            _num(totals["net_m"]),
+            a,
+            _num(totals["allowance_m"]),
+            _num(totals["total_m"]),
+            None,
+            None,
+            None,
+        ]
+    )
+
+    summary = []
+    for group, field in (
+        ("by_service", "service"),
+        ("by_diameter", "diameter"),
+        ("by_room", "room"),
+    ):
+        for item in result["summary"][group]:
+            key = _t(item["key"], lang) if field == "service" else item["key"]
+            summary.append([_t(group, lang), key, _num(item["net_m"]), _num(item["total_m"])])
+    summary.append([_t("total", lang), None, _num(totals["net_m"]), _num(totals["total_m"])])
+
+    control = [
+        [
+            c["run"],
+            _t(c["kind"], lang),
+            _t(c["service"], lang),
+            ", ".join(c["tags"]),
+            _num(c["schematic_m"]),
+            _t(c["reason"], lang),
+        ]
+        for c in result["control"]
+    ]
+    return {
+        "Metraj": ([_t(c, lang) for c in columns], rows),
+        "Özet": ([_t(c, lang) for c in ("group", "key", "net_m", "total_m")], summary),
+        "Kontrol": (
+            [_t(c, lang) for c in ("run", "kind", "service", "tags", "schematic_m", "reason")],
+            control,
+        ),
+        "Metodoloji": ([_t("topic", lang), _t("value", lang)], _pipe_method(result, lang)),
+    }
+
+
+def _unit_text(unit: dict | None) -> str:
+    if not unit:
+        return "—"
+    return f"INSUNITS {unit['declared'] or '—'} → {unit['inferred']}"
+
+
+def _pipe_method(result: dict, lang: str) -> list[list]:
+    method, scale = result["method"], result["scale"]
+    reason = method["reason"]
+    if reason.startswith("auto_") and reason != "auto_to_scale":
+        reason = "auto_layout"
+    rows = [
+        ["source_pid", method["sources"]["pid"]],
+        ["source_layout", method["sources"]["layout"] or "—"],
+        ["length_source", method["length_source"]],
+        ["choice", _say(reason, lang)],
+        ["scale_verdict", scale.get("verdict")],
+        ["scale_factor", None if scale.get("factor") is None else round(float(scale["factor"]), 6)],
+        ["scale_iqr", None if scale.get("iqr") is None else round(float(scale["iqr"]), 6)],
+        [
+            "scale_within",
+            None if scale.get("within_10pct") is None else round(float(scale["within_10pct"]), 4),
+        ],
+        ["scale_pairs", scale.get("pair_count")],
+        ["scale_matched", len(scale.get("matched", {}))],
+        ["scale_ambiguous", ", ".join(scale.get("ambiguous", [])) or "—"],
+        ["scale_note", _say("scale_note", lang)],
+        ["scale_verified", _yes(result["scale_verified"], lang)],
+        ["units_pid", _unit_text(result["units"]["pid"])],
+        ["units_layout", _unit_text(result["units"]["layout"])],
+        ["services", ", ".join(_t(s, lang) for s in method["services"])],
+        ["rule_route", _say("rule_route", lang)],
+        ["rule_vertical", method["vertical_allowance"]],
+        ["rule_allowance", method["allowance"]],
+        ["rule_diameter", _say("rule_diameter", lang)],
+        ["rule_service", _say("rule_service", lang)],
+        ["rule_status", _say("rule_status", lang)],
+        ["rule_rooms", _say("rule_rooms", lang)],
+        ["rule_multi", _say("rule_multi", lang)],
+        ["tol", method["tol"]],
+        ["label_search", round(float(method["label_search"]), 6)],
+        ["overlap", _num(method["overlap_length_m"])],
+    ]
+    rows += [["warning", warning] for warning in result["warnings"]]
+    return [[_t(topic, lang), value] for topic, value in rows]
+
+
+_BUILDERS = {"pipe": _pipe_sheets}
+
+
+def write_workbook(result: dict, *, kind: str, lang: str, path: str) -> dict:
+    """Write the takeoff: CSV always, XLSX when openpyxl is installed.
+
+    ``{"xlsx": path | None, "csv": [paths], "refused": {...} | None}``. Refused
+    before a file is written: a `kind` it builds no sheets for, a `lang` other
+    than tr / en / ru, and a `path` with a suffix other than .xlsx. Without
+    openpyxl the CSV files are written and `refused` carries the
+    `xlsx_write` capability refusal naming the `office` extra.
+    """
+    if kind not in _BUILDERS:
+        raise ValueError(f"kind: {kind!r} unknown; choose from {', '.join(_BUILDERS)}")
+    check_lang(lang)
+    target = Path(path)
+    if target.suffix.lower() not in ("", ".xlsx"):
+        raise ValueError(
+            f"path: the workbook is written as .xlsx with CSV files beside it; got {target.suffix!r}"
+        )
+    sheets = _BUILDERS[kind](result, lang)
+    xlsx = target.with_suffix(".xlsx")
+    xlsx.parent.mkdir(parents=True, exist_ok=True)
+    written = []
+    for name, (header, rows) in sheets.items():
+        csv_path = xlsx.with_name(f"{xlsx.stem}_{CSV_SLUGS[name]}.csv")
+        with open(csv_path, "w", newline="", encoding="utf-8-sig") as handle:
+            writer = csv.writer(handle)
+            writer.writerow(header)
+            writer.writerows([["" if cell is None else cell for cell in row] for row in rows])
+        written.append(str(csv_path))
+    if not xlsx_available():
+        refusal = UnsupportedCapabilityError(
+            "xlsx_write",
+            "The takeoff workbook needs openpyxl for XLSX, which is not installed. Install it "
+            'with: pip install -e ".[office]" (or pip install openpyxl). The same sheets were '
+            f"written as CSV: {', '.join(written)}",
+        )
+        return {"xlsx": None, "csv": written, "refused": refusal.to_dict()}
+    from openpyxl import Workbook
+
+    book = Workbook()
+    for i, (name, (header, rows)) in enumerate(sheets.items()):
+        sheet = book.active if i == 0 else book.create_sheet()
+        sheet.title = name
+        sheet.append(header)
+        for row in rows:
+            sheet.append(row)
+    book.save(xlsx)
+    return {"xlsx": str(xlsx), "csv": written, "refused": None}
diff --git a/server.py b/server.py
index 9e3704d..96cff3c 100644
--- a/server.py
+++ b/server.py
@@ -10554,6 +10554,166 @@ async def arch_plan_from_spec(
         raise ToolError(str(exc)) from exc
 
 
+# ---------------------------------------------------------------------------
+# ── SECTION 27: Plant Takeoffs (1 tool) ─────────────────────────────────────
+# ---------------------------------------------------------------------------
+
+
+@cad_tool(
+    summary="Pipe takeoff from a P&ID and its layout: rows by room, service and diameter.",
+    cost="safe",
+)
+@mcp.tool(
+    annotations={
+        "title": "Plant: Pipe Takeoff",
+        "readOnlyHint": False,
+        "destructiveHint": False,
+    },
+    tags={"plant", "analysis"},
+)
+async def pipe_takeoff(
+    pid_path: Annotated[
+        str | None,
+        Field(
+            default=None,
+            description="The P&ID (DXF headlessly; DXF or DWG on a live seat). Omit to read "
+            "the current drawing as the P&ID.",
+        ),
+    ] = None,
+    layout_path: Annotated[
+        str | None,
+        Field(
+            default=None,
+            description="The layout of the same plant. Omitted, every length is the P&ID's "
+            "drawn length and is marked scale_verified: false.",
+        ),
+    ] = None,
+    output: Annotated[
+        str | None,
+        Field(
+            default=None,
+            description="Workbook to write (.xlsx); the four sheets are also written as CSV "
+            "beside it. Omit to return the rows only.",
+        ),
+    ] = None,
+    lang: Annotated[
+        str, Field(default="en", description="Headers and method text: tr | en | ru.")
+    ] = "en",
+    services: Annotated[
+        list[str] | None,
+        Field(
+            default=None,
+            description="Services to take off; default product, cip_supply, cip_return. Also: "
+            "glycol_supply, glycol_return, steam, condensate, cold_water, hot_water, "
+            "ice_water, compressed_air, drain, electrical, unknown.",
+        ),
+    ] = None,
+    allowance: Annotated[
+        float,
+        Field(
+            default=0.20,
+            ge=0.0,
+            le=1.0,
+            description="Allowance as a share of the net length (0.20 = 20 %); its own column.",
+        ),
+    ] = 0.20,
+    vertical_allowance: Annotated[
+        float,
+        Field(
+            default=0.0,
+            ge=0.0,
+            description="Metres added per connection for rises and drops (layout lengths).",
+        ),
+    ] = 0.0,
+    length_source: Annotated[
+        str,
+        Field(
+            default="auto",
+            description="auto | pid | layout. auto measures on the P&ID only when the scale "
+            "check calls it to scale, else on the layout.",
+        ),
+    ] = "auto",
+    force: Annotated[
+        bool,
+        Field(
+            default=False,
+            description="With length_source='pid', measure on a P&ID the scale check calls "
+            "schematic anyway.",
+        ),
+    ] = False,
+    layers: Annotated[
+        list[str] | None,
+        Field(
+            default=None,
+            description="P&ID layers holding the lines; default every layer the vocabulary "
+            "files under piping or electrical with confidence >= 0.9.",
+        ),
+    ] = None,
+    tol: Annotated[
+        float,
+        Field(
+            default=1.0,
+            gt=0.0,
+            description="Line ends closer than this (P&ID drawing units) are one junction.",
+        ),
+    ] = 1.0,
+    ctx: Context = None,
+) -> dict:
+    """A pipe takeoff: topology from the P&ID, lengths from the layout, rows by room x service x diameter.
+
+    The P&ID gives the runs (junctions merged, tees split, a line drawn twice
+    counted once, valves and reducers bridged), each run's service (the layer,
+    overridden by a supply / return word near the line), its diameters exactly
+    as labelled (direct, carried by continuity - never across a reducer - or
+    unassigned) and the equipment tags at its ends. The length comes from the
+    layout unless the scale check calls the P&ID to scale: the rectilinear
+    minimum spanning tree of the run's tag positions on the layout, each
+    connection along x first then y, plus `vertical_allowance` per connection.
+    The P&ID's drawn length stays beside it as `schematic_m`, never as the
+    answer. Lengths are metres from the unit the geometry implies; an INSUNITS
+    that disagrees is a warning, never a factor. Status A / B / C per run: a C
+    run (an end on no equipment, a tag missing from or written twice on the
+    layout) is on the Kontrol sheet with its schematic length, counted in the
+    total and flagged.
+
+    Read-only on both drawings. With `output` it writes one workbook - Metraj,
+    Özet, Kontrol, Metodoloji - as XLSX and, always, as CSV beside it.
+
+    Refused before a drawing is read: `lang` outside tr / en / ru, an unknown
+    service or length source, an allowance outside 0-1, a negative vertical
+    allowance, a non-positive tolerance, and `length_source='layout'` without
+    a layout. A missing, unreadable or oversize drawing is refused by the
+    reader, which names MAX_DXF_BYTES. Refused after the scale check:
+    `length_source='pid'` on a P&ID it calls schematic, unless `force=True`
+    (the refusal quotes the statistics). Without openpyxl the XLSX is not
+    written and `files.refused` carries `capability: "xlsx_write"` naming the
+    `office` extra; the CSV files are written all the same.
+    """
+    from engineering.understand.draw import run_pipe_takeoff
+
+    pid = str(validate_path(pid_path)) if pid_path else None
+    layout = str(validate_path(layout_path)) if layout_path else None
+    target = str(validate_path(output, allow_write=True)) if output else None
+    await ctx.info(f"pipe_takeoff: {pid or 'current drawing'} / {layout or 'no layout'}")
+    try:
+        return await run_pipe_takeoff(
+            _backend(ctx),
+            pid_path=pid,
+            layout_path=layout,
+            output=target,
+            lang=lang,
+            services=services,
+            allowance=allowance,
+            vertical_allowance=vertical_allowance,
+            length_source=length_source,
+            force=force,
+            layers=layers,
+            tol=tol,
+        )
+    except (ValueError, OSError) as exc:
+        raise ToolError(str(exc)) from exc
+
+
 # ---------------------------------------------------------------------------
 # ── RESOURCES ───────────────────────────────────────────────────────────────
 # ---------------------------------------------------------------------------
diff --git a/tests/test_tool_registry.py b/tests/test_tool_registry.py
index e8a1ecb..fc4d189 100644
--- a/tests/test_tool_registry.py
+++ b/tests/test_tool_registry.py
@@ -495,7 +495,7 @@ async def test_tool_groups_is_byte_identical_to_the_source_declared_grouping():
 
 
 async def test_tool_group_sizes_are_unchanged():
-    """Frozen snapshot of the surface (241 tools, 26 groups).
+    """Frozen snapshot of the surface (242 tools, 26 groups).
 
     Taken before @cad_tool landed at 131 tools; `batch` moved 2 -> 3 when
     v1.5.0's `cad_batch` joined `entity_batch_create`/`entity_batch_modify`, and
@@ -591,10 +591,17 @@ async def test_tool_group_sizes_are_unchanged():
     exactly `PACK_TOOL_NAMES["arch"]`, and every group the track-F branches
     had grown while the tag was unranked on them returned to its
     pre-track-F value.
+
+    `analysis` moved 12 -> 13 when track H's SECTION 27 opened with
+    `pipe_takeoff` on the network branch. It is tagged `plant` and
+    `analysis`, and `plant` is not in `_GROUP_TAG_PRIORITY` on that
+    branch, so it files under `analysis`. This snapshot is branch-local:
+    the track H integration task gives `plant` its pack and its group
+    once every branch has landed.
     """
     sizes = {label: len(names) for label, names in (await server._tool_groups()).items()}
     assert sizes == {
-        "analysis": 12,
+        "analysis": 13,
         "architecture": 12,
         "batch": 3,
         "blocks": 9,
@@ -621,4 +628,4 @@ async def test_tool_group_sizes_are_unchanged():
         "validation": 1,
         "view": 4,
     }
-    assert sum(sizes.values()) == 241
+    assert sum(sizes.values()) == 242
diff --git a/tests/test_understand_report.py b/tests/test_understand_report.py
new file mode 100644
index 0000000..4c4b307
--- /dev/null
+++ b/tests/test_understand_report.py
@@ -0,0 +1,253 @@
+"""The takeoff workbook: four sheets, three languages, CSV always, XLSX with openpyxl.
+
+The result written is `pipe_rows` over a hand-built pair: a P&ID with two
+runs (T4100-T4101 labelled Ø51, T4101-T4102 unlabelled) and a layout whose
+INSUNITS says inches over millimetre geometry, measured on the layout (28 m
+and 12 m).
+"""
+
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+
+import pytest
+
+import engineering.understand.report as report
+from engineering.understand.network import build_network
+from engineering.understand.report import LANGS, SHEETS, write_workbook
+from engineering.understand.snapshot import EntityRecord, Snapshot
+from engineering.understand.takeoff import pipe_rows
+
+PIPE = "PRODUCT"
+WALLS = "WALLS"
+SCHEMATIC = {
+    "verdict": "schematic",
+    "within_10pct": 0.25,
+    "pair_count": 12,
+    "median": 1.3,
+    "iqr": 0.8,
+    "factor": None,
+    "matched": [],
+    "ambiguous": [],
+}
+
+
+def rec(handle, kind, layer, points=(), **extra):
+    return EntityRecord(
+        handle=handle, type=kind, layer=layer, space="Model", points=tuple(points), **extra
+    )
+
+
+def text(handle, value, at, height, layer="TEXT"):
+    x, y = at
+    return rec(
+        handle,
+        "TEXT",
+        layer,
+        (at,),
+        text=value,
+        height=height,
+        bbox=(x, y, x + height * len(value) * 0.6, y + height),
+    )
+
+
+def snap(records, insunits, source):
+    layers = {r.layer: {"color": 7, "linetype": "Continuous"} for r in records}
+    return Snapshot(
+        source=source,
+        insunits=insunits,
+        extmin=None,
+        extmax=None,
+        layers=layers,
+        layouts=(),
+        records=tuple(records),
+    )
+
+
+def tank(handle, tag, x0, y0):
+    return [
+        rec(
+            handle,
+            "LWPOLYLINE",
+            "EQUIPMENT",
+            ((x0, y0), (x0 + 500.0, y0), (x0 + 500.0, y0 + 500.0), (x0, y0 + 500.0)),
+            closed=True,
+            bbox=(x0, y0, x0 + 500.0, y0 + 500.0),
+        ),
+        text(handle + "T", tag, (x0 + 150.0, y0 + 225.0), 50.0),
+    ]
+
+
+@pytest.fixture
+def result():
+    pid = snap(
+        [
+            *tank("A", "T4100", 0.0, 0.0),
+            *tank("B", "T4101", 3000.0, 0.0),
+            *tank("C", "T4102", 3000.0, 2000.0),
+            rec("L1", "LINE", PIPE, ((500.0, 250.0), (3000.0, 250.0))),
+            rec("L2", "LINE", PIPE, ((3250.0, 500.0), (3250.0, 2000.0))),
+            text("D1", "Ø51", (1500.0, 260.0), 50.0),
+        ],
+        4,
+        "pid.dxf",
+    )
+    layout = snap(
+        [
+            rec(
+                "W",
+                "LWPOLYLINE",
+                WALLS,
+                ((0, 0), (40000, 0), (40000, 20000), (0, 20000)),
+                closed=True,
+                bbox=(0.0, 0.0, 40000.0, 20000.0),
+            ),
+            rec("X", "LINE", WALLS, ((20000, 0), (20000, 20000))),
+            text("R1", "ROOM 101", (5000.0, 10000.0), 250.0),
+            text("R2", "ROOM 102", (30000.0, 10000.0), 250.0),
+            text("T0", "T4100", (2000.0, 3000.0), 250.0),
+            text("T1", "T4101", (30000.0, 3000.0), 250.0),
+            text("T2", "T4102", (30000.0, 15000.0), 250.0),
+        ],
+        1,
+        "layout.dxf",
+    )
+    return pipe_rows(build_network(pid, layers=[PIPE]), pid, layout, scale=SCHEMATIC)
+
+
+METRAJ_HEADERS = {
+    "tr": [
+        "Mahal",
+        "Hizmet",
+        "Çap",
+        "Doğrudan (m)",
+        "Süreklilik (m)",
+        "Atanmamış (m)",
+        "Net (m)",
+        "Pay oranı",
+        "Pay (m)",
+        "Paylı toplam (m)",
+        "Şematik (m)",
+        "Durum",
+        "Hatlar",
+    ],
+    "en": [
+        "Room",
+        "Service",
+        "Diameter",
+        "Direct (m)",
+        "Continuity (m)",
+        "Unassigned (m)",
+        "Net (m)",
+        "Allowance",
+        "Allowance (m)",
+        "Total incl. allowance (m)",
+        "Schematic (m)",
+        "Status",
+        "Runs",
+    ],
+    "ru": [
+        "Помещение",
+        "Среда",
+        "Диаметр",
+        "Прямо (м)",
+        "По непрерывности (м)",
+        "Не назначено (м)",
+        "Нетто (м)",
+        "Запас",
+        "Запас (м)",
+        "Итого с запасом (м)",
+        "По схеме (м)",
+        "Статус",
+        "Линии",
+    ],
+}
+TOTAL = {"tr": "TOPLAM", "en": "TOTAL", "ru": "ИТОГО"}
+
+
+def read_csv(path):
+    with open(path, encoding="utf-8-sig", newline="") as handle:
+        return list(csv.reader(handle))
+
+
+@pytest.mark.parametrize("lang", LANGS)
+def test_csv_sheets_carry_the_headers_of_each_language(result, tmp_path, lang):
+    files = write_workbook(result, kind="pipe", lang=lang, path=str(tmp_path / "takeoff.xlsx"))
+    names = [Path(p).name for p in files["csv"]]
+    assert names == [
+        "takeoff_metraj.csv",
+        "takeoff_ozet.csv",
+        "takeoff_kontrol.csv",
+        "takeoff_metodoloji.csv",
+    ]
+    metraj = read_csv(files["csv"][0])
+    assert metraj[0] == METRAJ_HEADERS[lang]
+    total = metraj[-1]
+    assert total[0] == TOTAL[lang]
+    assert float(total[6]) == pytest.approx(40.0)  # 28 m + 12 m net
+    assert float(total[9]) == pytest.approx(48.0)  # with the 20 % allowance
+
+
+def test_the_workbook_has_the_four_sheets_and_numbers_stay_numbers(result, tmp_path):
+    openpyxl = pytest.importorskip("openpyxl")
+    files = write_workbook(result, kind="pipe", lang="tr", path=str(tmp_path / "takeoff.xlsx"))
+    assert files["refused"] is None
+    book = openpyxl.load_workbook(files["xlsx"])
+    assert book.sheetnames == list(SHEETS) == ["Metraj", "Özet", "Kontrol", "Metodoloji"]
+    rows = list(book["Metraj"].iter_rows(values_only=True))
+    assert list(rows[0]) == METRAJ_HEADERS["tr"]
+    assert rows[-1][0] == "TOPLAM"
+    assert rows[-1][6] == pytest.approx(40.0)
+    assert isinstance(rows[1][6], (int, float))  # a number cell, not the text '18.0'
+    assert rows[1][6] == pytest.approx(18.0)  # room 101's share of the Ø51 run
+
+
+def test_the_method_sheet_is_generated_from_the_result(result, tmp_path):
+    files = write_workbook(result, kind="pipe", lang="en", path=str(tmp_path / "t.xlsx"))
+    method = dict((row[0], row[1]) for row in read_csv(files["csv"][3])[1:] if len(row) == 2)
+    assert method["Length source"] == "layout"
+    assert method["Scale check: verdict"] == "schematic"
+    assert method["Scale check: pairs"] == "12"
+    assert method["Scale verified"] == "yes"
+    assert method["Units: layout"] == "INSUNITS in → mm"
+    assert "x first, then y" in method["Rule: route"]
+    warnings = [row[1] for row in read_csv(files["csv"][3]) if row[0] == "Warning"]
+    assert any("INSUNITS declares in" in w for w in warnings)
+
+
+def test_kontrol_lists_the_unassigned_diameter(result, tmp_path):
+    files = write_workbook(result, kind="pipe", lang="en", path=str(tmp_path / "t.xlsx"))
+    kontrol = read_csv(files["csv"][2])
+    assert kontrol[0] == ["Run", "Kind", "Service", "Tags", "Schematic (m)", "Reason"]
+    (row,) = kontrol[1:]
+    assert row[1] == "Diameter unassigned" and row[3] == "T4101, T4102"
+    assert row[5] == "No diameter label"
+
+
+def test_without_openpyxl_the_csv_is_written_and_xlsx_is_refused(result, tmp_path, monkeypatch):
+    monkeypatch.setattr(report, "xlsx_available", lambda: False)
+    files = write_workbook(result, kind="pipe", lang="tr", path=str(tmp_path / "takeoff.xlsx"))
+    assert files["xlsx"] is None
+    assert not (tmp_path / "takeoff.xlsx").exists()
+    assert all(Path(p).exists() for p in files["csv"])
+    assert files["refused"]["capability"] == "xlsx_write"
+    assert files["refused"]["ok"] is False
+    assert '".[office]"' in files["refused"]["error"]
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "message"),
+    [
+        ({"kind": "steel"}, "kind: 'steel' unknown; choose from pipe"),
+        ({"lang": "de"}, "lang: 'de' unknown; choose from tr, en, ru"),
+        ({"path": "takeoff.ods"}, "path: the workbook is written as .xlsx"),
+    ],
+)
+def test_refusals_come_before_any_file(result, tmp_path, kwargs, message):
+    arguments = {"kind": "pipe", "lang": "en", "path": str(tmp_path / "t.xlsx"), **kwargs}
+    if "path" in kwargs:
+        arguments["path"] = str(tmp_path / kwargs["path"])
+    with pytest.raises(ValueError, match="^" + message.replace(".", r"\.")):
+        write_workbook(result, **arguments)
+    assert list(tmp_path.iterdir()) == []
diff --git a/tests/test_understand_tools.py b/tests/test_understand_tools.py
new file mode 100644
index 0000000..cf96a3e
--- /dev/null
+++ b/tests/test_understand_tools.py
@@ -0,0 +1,103 @@
+"""pipe_takeoff over the MCP wire, on the synthetic plant pair - never on a client drawing."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+import pytest_asyncio
+from fastmcp import Client
+
+import server
+from tests.fixtures.plant_pair import build_plant_pair
+
+pytestmark = pytest.mark.asyncio
+
+CSV_NAMES = [
+    "takeoff_metraj.csv",
+    "takeoff_ozet.csv",
+    "takeoff_kontrol.csv",
+    "takeoff_metodoloji.csv",
+]
+
+
+@pytest_asyncio.fixture
+async def client(monkeypatch):
+    """In-memory client on the headless engine - never the operator's AutoCAD."""
+    monkeypatch.setenv("AUTOCAD_MCP_BACKEND", "ezdxf")
+    async with Client(server.mcp) as connected:
+        status = (await connected.call_tool("system_status", {})).structured_content or {}
+        assert status.get("backend") == "ezdxf", "these tests must never touch live AutoCAD"
+        yield connected
+
+
+@pytest.fixture
+def plant(tmp_path):
+    return build_plant_pair(tmp_path)
+
+
+async def test_pipe_takeoff_measures_on_the_layout_and_writes_the_workbook(client, plant, tmp_path):
+    output = tmp_path / "takeoff.xlsx"
+    arguments = {
+        "pid_path": plant["pid"],
+        "layout_path": plant["layout"],
+        "output": str(output),
+        "lang": "tr",
+    }
+    result = (await client.call_tool("pipe_takeoff", arguments)).structured_content
+    assert result["length_source"] == "layout"
+    assert result["scale"]["verdict"] == plant["scale_verdict"]
+    files = result["files"]
+    assert [Path(p).name for p in files["csv"]] == CSV_NAMES
+    assert all(Path(p).exists() for p in files["csv"])
+    if files["xlsx"] is None:
+        assert files["refused"]["capability"] == "xlsx_write"
+    else:
+        assert Path(files["xlsx"]) == output
+        assert output.exists()
+
+
+async def test_pid_lengths_on_the_schematic_pid_are_refused_without_force(client, plant):
+    arguments = {"pid_path": plant["pid"], "layout_path": plant["layout"], "length_source": "pid"}
+    result = await client.call_tool("pipe_takeoff", arguments, raise_on_error=False)
+    assert result.is_error is True
+    text = result.content[0].text
+    assert "the scale check calls this P&ID schematic" in text
+    assert "force=True" in text
+
+
+async def test_force_measures_on_the_pid_and_marks_it_unverified(client, plant):
+    arguments = {
+        "pid_path": plant["pid"],
+        "layout_path": plant["layout"],
+        "length_source": "pid",
+        "force": True,
+    }
+    result = (await client.call_tool("pipe_takeoff", arguments)).structured_content
+    assert (result["length_source"], result["scale_verified"]) == ("pid", False)
+    assert result["files"] is None  # no output named, nothing written
+
+
+async def test_an_unknown_language_is_refused_before_anything_is_read(client, tmp_path):
+    arguments = {"pid_path": str(tmp_path / "never_read.dxf"), "lang": "de"}
+    result = await client.call_tool("pipe_takeoff", arguments, raise_on_error=False)
+    assert result.is_error is True
+    assert "lang: 'de' unknown; choose from tr, en, ru" in result.content[0].text
+
+
+async def test_the_drawings_are_never_modified(client, plant, tmp_path):
+    before = {key: Path(plant[key]).read_bytes() for key in ("pid", "layout")}
+    arguments = {
+        "pid_path": plant["pid"],
+        "layout_path": plant["layout"],
+        "output": str(tmp_path / "t.xlsx"),
+    }
+    await client.call_tool("pipe_takeoff", arguments)
+    assert {key: Path(plant[key]).read_bytes() for key in ("pid", "layout")} == before
+
+
+async def test_the_tool_says_it_writes_a_file_and_destroys_nothing(client):
+    tools = {tool.name: tool for tool in await client.list_tools()}
+    annotations = tools["pipe_takeoff"].annotations
+    assert annotations.readOnlyHint is False
+    assert annotations.destructiveHint is False