feat(understand): drawing_topology_check and the three topo_* critique focuses

engineering/understand/topology.py reads a line network (LINE, polylines
with true bulge arcs, ARC) on the chosen layers and reports dangling
ends, near misses (end-end and end-segment within gap, one row per pair)
and interior crossings of two straight segments of one layer. An end on
another network line is joined; an end on any curve of another layer or
inside a block reference's box is attached; a line drawn twice is not a
join, so a doubled pipe still dangles. gap defaults to 0.2 % of the
network diagonal, never under ten tol.

drawing_topology_check joins drawing_diff in SECTION 26: default layers
are those classify_layer calls piping or electrical at >= 0.9, reported
with their classification; no such layer is a refusal, not a pass.

topo_dangling_endpoint (warning), topo_near_miss (error) and
topo_interior_crossing (info) join the closed enum and run under
focus=None, excluding the P&ID and architectural layer sets, so a
pid_from_spec sheet and an arch_plan_from_spec plan get none of them;
a drawing with no candidate layer is never snapshotted.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd

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 (242 tools, 26 groups).
+    """Frozen snapshot of the surface (243 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
@@ -593,14 +593,15 @@ async def test_tool_group_sizes_are_unchanged():
     pre-track-F value.
 
     Track H group D opened its copy of SECTION 26 (Understanding & QA) with
-    `drawing_diff`, tagged `analysis` and `query`; `analysis` outranks `query`
-    in `_GROUP_TAG_PRIORITY`, so it files under `analysis` (12 -> 13). This
-    snapshot is branch-local: the track H merge recomputes it once the
-    understand, network and diff branches have landed.
+    `drawing_diff` and `drawing_topology_check`, both tagged `analysis` and
+    `query`; `analysis` outranks `query` in `_GROUP_TAG_PRIORITY`, so both file
+    under `analysis` (12 -> 14). This snapshot is branch-local: the track H
+    merge recomputes it once the understand, network and diff branches have
+    landed.
     """
     sizes = {label: len(names) for label, names in (await server._tool_groups()).items()}
     assert sizes == {
-        "analysis": 13,
+        "analysis": 14,
         "architecture": 12,
         "batch": 3,
         "blocks": 9,
@@ -627,4 +628,4 @@ async def test_tool_group_sizes_are_unchanged():
         "validation": 1,
         "view": 4,
     }
-    assert sum(sizes.values()) == 242
+    assert sum(sizes.values()) == 243

tests/test_understand_critique.py

@@ -0,0 +1,201 @@
+"""The three topo_* focuses: a positive and a negative fixture each, and silence
+on every drawing this server makes.
+
+The network layer is the synthetic plant's product-piping layer, read from the
+fixture's truth dict (``tests/fixtures/plant_pair.py``), so the test draws on a
+layer the vocabulary classifies the way a foreign drawing's layer would be
+classified.
+"""
+
+from __future__ import annotations
+
+import copy
+
+import pytest
+
+from engineering.arch.layers import ARCH_ROLE_LAYER
+from engineering.critique import run_critique
+from engineering.pid.lines import PID_LINE_LAYERS
+from engineering.plan_spec import ALL_CRITIQUE_FOCUSES
+from engineering.understand.critique import (
+    SEMANTIC_LAYERS,
+    TOPO_DISPATCH,
+    TOPO_FOCUSES,
+    build_index,
+    focus_layers,
+    issues_for,
+)
+from engineering.understand.snapshot import EntityRecord, Snapshot
+from engineering.understand.vocab import classify_layer
+from tests.fixtures.plant_pair import build_plant_pair
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture(scope="module")
+def pipe_layer(tmp_path_factory) -> str:
+    truth = build_plant_pair(tmp_path_factory.mktemp("plant"))
+    layer = next(
+        name for name, service in sorted(truth["layer_services"].items()) if service == "product"
+    )
+    meaning = classify_layer(layer)
+    assert meaning["discipline"] == "piping" and meaning["confidence"] >= 0.9, meaning
+    return layer
+
+
+async def _lines(backend, layer, *segments):
+    handles = []
+    for x1, y1, x2, y2 in segments:
+        entity = await backend.entity_create_line(x1, y1, x2, y2, layer=layer)
+        handles.append(entity.handle)
+    return handles
+
+
+async def _issues(backend, focus):
+    return issues_for(focus, await build_index(backend))
+
+
+# -- the closed enum and the exclusions ------------------------------------------------
+
+
+async def test_the_three_focuses_are_the_spec_ones_and_are_in_the_closed_enum():
+    assert TOPO_FOCUSES == ("topo_dangling_endpoint", "topo_near_miss", "topo_interior_crossing")
+    for focus in TOPO_FOCUSES:
+        assert focus in ALL_CRITIQUE_FOCUSES
+        assert TOPO_DISPATCH[focus].needs_shared is True
+
+
+async def test_pid_line_layers_and_arch_layers_are_excluded():
+    assert set(PID_LINE_LAYERS) <= SEMANTIC_LAYERS
+    assert ARCH_ROLE_LAYER["wall"] in SEMANTIC_LAYERS
+    assert "0" not in SEMANTIC_LAYERS
+    snap = Snapshot(
+        source="test",
+        insunits=4,
+        extmin=None,
+        extmax=None,
+        layers={"PROCESS-PIPING-MAIN": {}, "A-WALL-E-N": {}},
+        layouts=(),
+        records=(
+            EntityRecord("1", "LINE", "PROCESS-PIPING-MAIN", "Model", ((0, 0), (10, 0))),
+            EntityRecord("2", "LINE", "A-WALL-E-N", "Model", ((0, 5), (10, 5))),
+        ),
+    )
+    assert focus_layers(snap) == []
+
+
+# -- topo_dangling_endpoint ------------------------------------------------------------
+
+
+async def test_dangling_fires_on_a_lone_pipe(backend, pipe_layer):
+    (handle,) = await _lines(backend, pipe_layer, (0, 0, 1000, 0))
+    issues = await _issues(backend, "topo_dangling_endpoint")
+    assert [(i.severity, i.handles, i.detail["end"]) for i in issues] == [
+        ("warning", [handle], "end"),
+        ("warning", [handle], "start"),
+    ]
+
+
+async def test_dangling_is_quiet_on_a_closed_loop(backend, pipe_layer):
+    await _lines(
+        backend,
+        pipe_layer,
+        (0, 0, 1000, 0),
+        (1000, 0, 1000, 1000),
+        (1000, 1000, 0, 1000),
+        (0, 1000, 0, 0),
+    )
+    assert await _issues(backend, "topo_dangling_endpoint") == []
+
+
+# -- topo_near_miss --------------------------------------------------------------------
+
+
+async def test_near_miss_fires_on_a_3_unit_gap(backend, pipe_layer):
+    # box 0..2003 wide: default gap 0.002 x 2003 = 4.006 > 3
+    first, second = await _lines(backend, pipe_layer, (0, 0, 1000, 0), (1003, 0, 2003, 0))
+    (issue,) = await _issues(backend, "topo_near_miss")
+    assert issue.severity == "error"
+    assert issue.handles == sorted([first, second])
+    assert issue.detail["kind"] == "end_end"
+    assert issue.detail["gap"] == pytest.approx(3.0)
+    assert "stops 3 short of the end of" in issue.message
+
+
+async def test_near_miss_is_quiet_when_the_lines_join(backend, pipe_layer):
+    await _lines(backend, pipe_layer, (0, 0, 1000, 0), (1000, 0, 2000, 0))
+    assert await _issues(backend, "topo_near_miss") == []
+
+
+# -- topo_interior_crossing ------------------------------------------------------------
+
+
+async def test_crossing_fires_on_two_pipes_crossing_mid_span(backend, pipe_layer):
+    await _lines(backend, pipe_layer, (0, 0, 1000, 0), (500, -500, 500, 500))
+    (issue,) = await _issues(backend, "topo_interior_crossing")
+    assert issue.severity == "info"
+    assert issue.detail["at"] == [500.0, 0.0]
+
+
+async def test_crossing_is_quiet_on_a_t_junction(backend, pipe_layer):
+    await _lines(backend, pipe_layer, (0, 0, 1000, 0), (500, 0, 500, 500))
+    assert await _issues(backend, "topo_interior_crossing") == []
+
+
+# -- silence on this server's own drawings ---------------------------------------------
+
+
+async def test_a_pid_from_spec_sheet_gets_no_topo_issue(backend):
+    from engineering.pid.spec import EXAMPLE_SPEC, run_spec
+
+    await run_spec(backend, copy.deepcopy(EXAMPLE_SPEC))
+    assert await run_critique(backend, list(TOPO_FOCUSES)) == []
+    everything = await run_critique(backend, None)
+    assert not [i for i in everything if i.focus.startswith("topo_")]
+
+
+async def test_an_arch_plan_from_spec_gets_no_topo_issue(backend):
+    from engineering.arch.spec import EXAMPLE_SPEC, draw_plan_from_spec
+
+    await draw_plan_from_spec(backend, copy.deepcopy(EXAMPLE_SPEC))
+    assert await run_critique(backend, list(TOPO_FOCUSES)) == []
+
+
+async def test_a_drawing_without_a_network_layer_is_never_snapshotted(backend, monkeypatch):
+    import engineering.understand.snapshot as snapshot
+
+    async def refuse(*_args, **_kwargs):
+        raise AssertionError("the snapshot was taken")
+
+    monkeypatch.setattr(snapshot, "take_snapshot", refuse)
+    await backend.entity_create_line(0, 0, 100, 0, layer="GEOMETRY")
+    assert await build_index(backend) == {"layers": [], "findings": None, "error": None}
+
+
+async def test_a_snapshot_that_fails_is_reported_not_passed(backend, pipe_layer, monkeypatch):
+    import engineering.understand.snapshot as snapshot
+
+    async def broken(*_args, **_kwargs):
+        raise RuntimeError("export failed")
+
+    monkeypatch.setattr(snapshot, "take_snapshot", broken)
+    await _lines(backend, pipe_layer, (0, 0, 1000, 0))
+    (issue,) = await _issues(backend, "topo_near_miss")
+    assert issue.severity == "info" and "export failed" in issue.message
+
+
+# -- the dispatch ----------------------------------------------------------------------
+
+
+async def test_the_three_focuses_share_one_index_per_run(backend, pipe_layer):
+    await _lines(backend, pipe_layer, (0, 0, 1000, 0))
+    shared: dict = {}
+    for focus in TOPO_FOCUSES:
+        await TOPO_DISPATCH[focus](backend, shared)
+    assert list(shared) == ["topo_index"]
+
+
+async def test_run_critique_reaches_the_topo_focuses(backend, pipe_layer):
+    await _lines(backend, pipe_layer, (0, 0, 1000, 0), (1003, 0, 2003, 0))
+    (issue,) = await run_critique(backend, ["topo_near_miss"])
+    assert issue.focus == "topo_near_miss" and issue.severity == "error"

tests/test_understand_topology.py

@@ -0,0 +1,317 @@
+"""The topology engine: ends, joins, near misses, crossings - by hand-computed geometry.
+
+Every record is built by hand, so every coordinate in an assertion is the
+answer. The network layer is called PIPE here; which layers are network layers
+by default is ``network_layers``' job and is tested on the synthetic plant.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from engineering.understand.snapshot import EntityRecord, Snapshot
+from engineering.understand.topology import (
+    Piece,
+    cap_findings,
+    entity_pieces,
+    network_layers,
+    topology_findings,
+)
+from engineering.understand.vocab import classify_layer
+from tests.fixtures.plant_pair import build_plant_pair
+
+L = "PIPE"
+
+
+def rec(handle, type_, layer=L, points=(), **kw):
+    space = kw.pop("space", "Model")
+    return EntityRecord(
+        handle=handle, type=type_, layer=layer, space=space, points=tuple(points), **kw
+    )
+
+
+def line(handle, x1, y1, x2, y2, layer=L, space="Model"):
+    return rec(handle, "LINE", layer, ((x1, y1), (x2, y2)), space=space)
+
+
+def snap(*records):
+    return Snapshot(
+        source="test",
+        insunits=4,
+        extmin=None,
+        extmax=None,
+        layers={r.layer: {"color": 7, "linetype": "Continuous"} for r in records},
+        layouts=("Layout1",),
+        records=tuple(records),
+    )
+
+
+def run(*records, gap=5.0, layers=(L,)):
+    return topology_findings(snap(*records), layers=list(layers), gap=gap)
+
+
+def ends(rows):
+    return {(row["handle"], row["end"]) for row in rows}
+
+
+# -- pieces ----------------------------------------------------------------------------
+
+
+def test_a_bulge_of_one_is_the_semicircle_below_its_chord():
+    # bulge +1 from (0,0) to (200,0): counter-clockwise, centre (100,0), radius 100,
+    # through (100,-100)
+    pieces, entity_ends = entity_pieces(
+        rec("PL", "LWPOLYLINE", L, ((0.0, 0.0), (200.0, 0.0)), bulges=(1.0, 0.0))
+    )
+    (arc,) = pieces
+    assert arc.centre == (pytest.approx(100.0), pytest.approx(0.0, abs=1e-9))
+    assert arc.radius == pytest.approx(100.0)
+    foot = arc.nearest((100.0, -150.0))
+    assert foot == (pytest.approx(100.0), pytest.approx(-100.0))
+    assert entity_ends == ((0.0, 0.0), (200.0, 0.0))
+
+
+def test_a_closed_polyline_has_no_ends_and_a_closing_piece():
+    pieces, entity_ends = entity_pieces(
+        rec("SQ", "LWPOLYLINE", L, ((0, 0), (10, 0), (10, 10), (0, 10)), closed=True)
+    )
+    assert entity_ends is None and len(pieces) == 4
+    assert (pieces[3].a, pieces[3].b) == ((0.0, 10.0), (0.0, 0.0))
+
+
+def test_a_straight_piece_nearest_point_is_clamped_to_its_ends():
+    piece = Piece("A", L, 0, (0.0, 0.0), (10.0, 0.0))
+    assert piece.nearest((5.0, 3.0)) == (5.0, 0.0)
+    assert piece.nearest((-4.0, 3.0)) == (0.0, 0.0)
+
+
+# -- dangling and joined ---------------------------------------------------------------
+
+
+def test_a_lone_line_dangles_at_both_ends():
+    result = run(line("A", 0, 0, 1000, 0))
+    assert ends(result["dangling"]) == {("A", "start"), ("A", "end")}
+    assert result["near_miss"] == [] and result["crossing"] == []
+
+
+def test_a_t_junction_joins_the_stem():
+    result = run(line("A", 0, 0, 1000, 0), line("B", 500, 0, 500, 500))
+    assert ends(result["dangling"]) == {("A", "start"), ("A", "end"), ("B", "end")}
+
+
+def test_a_closed_square_of_lines_has_no_finding():
+    square = [
+        line("A", 0, 0, 1000, 0),
+        line("B", 1000, 0, 1000, 1000),
+        line("C", 1000, 1000, 0, 1000),
+        line("D", 0, 1000, 0, 0),
+    ]
+    result = run(*square)
+    assert (result["dangling"], result["near_miss"], result["crossing"]) == ([], [], [])
+
+
+def test_an_end_on_an_arc_and_on_an_arc_body_is_joined():
+    # ARC centre (0,0) r 100 from 0 to 90 degrees: ends (100,0) and (0,100)
+    arc = rec("ARC1", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0))
+    on_end = line("L1", 100, 0, 500, 0)
+    on_body = line("L2", 70.71067811865476, 70.71067811865476, 300, 300)
+    result = run(arc, on_end, on_body)
+    assert ends(result["dangling"]) == {("ARC1", "end"), ("L1", "end"), ("L2", "end")}
+
+
+def test_an_end_on_a_bulged_segment_is_joined():
+    arc = rec("PL", "LWPOLYLINE", L, ((0.0, 0.0), (200.0, 0.0)), bulges=(1.0, 0.0))
+    result = run(arc, line("L", 100, -100, 100, -500))
+    assert ends(result["dangling"]) == {("PL", "start"), ("PL", "end"), ("L", "end")}
+
+
+def test_an_end_inside_an_insert_box_is_attached():
+    valve = rec("V1", "INSERT", "VALVES", ((1000, 0),), block="VALVE", bbox=(995, -5, 1005, 5))
+    result = run(line("A", 0, 0, 995, 0), valve)
+    assert ends(result["dangling"]) == {("A", "start")}
+
+
+def test_an_end_on_a_curve_of_another_layer_is_attached():
+    tank = rec(
+        "TK",
+        "LWPOLYLINE",
+        "EQUIPMENT",
+        ((1000, -500), (2000, -500), (2000, 500), (1000, 500)),
+        closed=True,
+    )
+    result = run(line("A", 0, 0, 1000, 0), tank)
+    assert ends(result["dangling"]) == {("A", "start")}
+
+
+def test_model_space_and_a_layout_never_join():
+    result = run(line("A", 0, 0, 100, 0), line("B", 100, 0, 200, 0, space="Layout1"))
+    assert ends(result["dangling"]) == {("A", "start"), ("A", "end"), ("B", "start"), ("B", "end")}
+
+
+# -- lines drawn twice -----------------------------------------------------------------
+
+
+def test_a_pipe_drawn_twice_still_dangles_at_all_four_ends():
+    result = run(line("A", 0, 0, 1000, 0), line("A2", 0, 0, 1000, 0))
+    assert ends(result["dangling"]) == {
+        ("A", "start"),
+        ("A", "end"),
+        ("A2", "start"),
+        ("A2", "end"),
+    }
+    assert result["near_miss"] == [] and result["crossing"] == []
+
+
+def test_a_line_over_a_polyline_joins_only_where_the_polyline_turns():
+    polyline = rec("PL", "LWPOLYLINE", L, ((0, 0), (1000, 0), (1000, 1000)))
+    result = run(polyline, line("OV", 0, 0, 1000, 0))
+    assert ends(result["dangling"]) == {("OV", "start"), ("PL", "end"), ("PL", "start")}
+
+
+def test_an_arc_drawn_twice_still_dangles_at_all_four_ends():
+    # the arc form of a pipe drawn twice: a bend drawn twice must not join itself
+    first = rec("A1", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0))
+    second = rec("A2", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0))
+    result = run(first, second)
+    assert ends(result["dangling"]) == {
+        ("A1", "start"),
+        ("A1", "end"),
+        ("A2", "start"),
+        ("A2", "end"),
+    }
+    assert result["near_miss"] == []
+
+
+def test_two_arcs_of_one_circle_butted_end_to_end_join():
+    # 0-90 then 90-180 on the same circle: continuing the other way is a join
+    first = rec("A1", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(0.0, 90.0))
+    second = rec("A2", "ARC", L, ((0.0, 0.0),), radius=100.0, angles=(90.0, 180.0))
+    assert ends(run(first, second)["dangling"]) == {("A1", "start"), ("A2", "end")}
+
+
+def test_a_bulge_of_minus_one_is_the_semicircle_above_its_chord():
+    ((arc,), _ends) = entity_pieces(
+        rec("PL", "LWPOLYLINE", L, ((0.0, 0.0), (200.0, 0.0)), bulges=(-1.0, 0.0))
+    )
+    assert arc.radius == pytest.approx(100.0)
+    assert arc.nearest((100.0, 150.0)) == (pytest.approx(100.0), pytest.approx(100.0))
+
+
+def test_a_side_drawn_twice_inside_a_closed_loop_is_harmless():
+    square = [
+        line("A", 0, 0, 1000, 0),
+        line("A2", 0, 0, 1000, 0),
+        line("B", 1000, 0, 1000, 1000),
+        line("C", 1000, 1000, 0, 1000),
+        line("D", 0, 1000, 0, 0),
+    ]
+    result = run(*square)
+    assert (result["dangling"], result["near_miss"], result["crossing"]) == ([], [], [])
+
+
+# -- near misses -----------------------------------------------------------------------
+
+
+def test_two_ends_3_apart_are_one_end_end_near_miss():
+    result = run(line("A", 0, 0, 1000, 0), line("B", 1003, 0, 2000, 0))
+    (row,) = result["near_miss"]
+    assert row["handles"] == ["A", "B"] and row["kind"] == "end_end"
+    assert row["gap"] == pytest.approx(3.0)
+    assert (row["at"], row["to"]) == ([1000.0, 0.0], [1003.0, 0.0])
+    assert ends(result["dangling"]) == {("A", "start"), ("B", "end")}
+
+
+def test_an_end_4_above_a_line_is_an_end_segment_near_miss():
+    result = run(line("A", 0, 0, 1000, 0), line("B", 500, 4, 500, 500))
+    (row,) = result["near_miss"]
+    assert row["handles"] == ["B", "A"] and row["kind"] == "end_segment"
+    assert row["gap"] == pytest.approx(4.0) and row["to"] == [500.0, 0.0]
+
+
+def test_an_end_beyond_the_gap_dangles_instead():
+    result = run(line("A", 0, 0, 1000, 0), line("B", 500, 6, 500, 500))
+    assert result["near_miss"] == []
+    assert ("B", "start") in ends(result["dangling"])
+
+
+def test_a_polyline_that_nearly_closes_on_itself_is_one_near_miss():
+    ring = rec("RING", "LWPOLYLINE", L, ((0, 0), (1000, 0), (1000, 1000), (0, 1000), (0, 3)))
+    result = run(ring)
+    (row,) = result["near_miss"]
+    assert row["handles"] == ["RING", "RING"] and row["kind"] == "end_end"
+    assert row["gap"] == pytest.approx(3.0)
+    assert result["dangling"] == []
+
+
+# -- crossings -------------------------------------------------------------------------
+
+
+def test_two_lines_of_one_layer_crossing_mid_span_are_a_crossing():
+    result = run(line("A", 0, 0, 1000, 0), line("B", 500, -500, 500, 500))
+    assert result["crossing"] == [
+        {"handles": ["A", "B"], "layer": L, "at": [500.0, 0.0], "space": "Model"}
+    ]
+
+
+def test_two_layers_crossing_and_a_t_junction_are_not_crossings():
+    other = line("B", 500, -500, 500, 500, layer="CABLE")
+    assert run(line("A", 0, 0, 1000, 0), other, layers=(L, "CABLE"))["crossing"] == []
+    assert run(line("A", 0, 0, 1000, 0), line("B", 500, 0, 500, 500))["crossing"] == []
+
+
+def test_a_bow_tie_polyline_crosses_itself():
+    bow = rec("BOW", "LWPOLYLINE", L, ((0, 0), (100, 100), (100, 0), (0, 100)))
+    (row,) = run(bow)["crossing"]
+    assert row["handles"] == ["BOW", "BOW"] and row["at"] == [50.0, 50.0]
+
+
+# -- defaults and refusals -------------------------------------------------------------
+
+
+def test_the_default_gap_is_0_2_percent_of_the_network_diagonal():
+    # box 0..3000 x 0..4000: diagonal 5000, gap 10
+    result = topology_findings(snap(line("A", 0, 0, 3000, 0), line("B", 0, 0, 0, 4000)), layers=[L])
+    assert result["gap"] == pytest.approx(10.0)
+    assert result["stats"] == {"entities": 2, "ends": 4, "skipped": {}}
+
+
+def test_an_arc_without_a_centre_is_skipped_and_counted():
+    result = run(line("A", 0, 0, 10, 0), rec("ARC9", "ARC", L, (), radius=5.0, angles=(0, 90)))
+    assert result["stats"]["skipped"] == {"ARC": 1}
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "message"),
+    [
+        ({"layers": []}, r"^layers: name at least one layer to check"),
+        ({"layers": [L], "tol": 0}, r"^tol must be a positive number; got 0"),
+        ({"layers": [L], "gap": -1}, r"^gap must be a positive number; got -1"),
+        ({"layers": [L], "gap": 0.005}, r"^gap \(0\.005\) must exceed tol \(0\.01\)"),
+    ],
+)
+def test_refusals_come_before_any_work(kwargs, message):
+    with pytest.raises(ValueError, match=message):
+        topology_findings(snap(line("A", 0, 0, 1, 0)), **kwargs)
+
+
+def test_cap_findings_counts_everything_and_cuts_the_lists():
+    lines = [line(f"L{i}", 0, 1000 * i, 100, 1000 * i) for i in range(3)]
+    capped = cap_findings(run(*lines), limit=2)
+    assert capped["counts"] == {"dangling": 6, "near_miss": 0, "crossing": 0}
+    assert capped["truncated"] == {"dangling": 4, "near_miss": 0, "crossing": 0}
+    assert len(capped["dangling"]) == 2
+
+
+# -- the default layers on the synthetic plant -----------------------------------------
+
+
+def test_the_plant_piping_layers_are_network_layers(tmp_path):
+    from engineering.understand.snapshot import read_snapshot
+
+    truth = build_plant_pair(tmp_path)
+    pid = read_snapshot(truth["pid"])
+    found = {row["layer"] for row in network_layers(pid)}
+    for layer, service in truth["layer_services"].items():
+        if layer in pid.layers and service in ("product", "cip_supply", "cip_return"):
+            assert layer in found, (layer, classify_layer(layer))
+    assert all(row["confidence"] >= 0.9 for row in network_layers(pid))

tests/test_understand_topology_tool.py

@@ -0,0 +1,101 @@
+"""drawing_topology_check over the MCP wire, on the headless engine."""
+
+from __future__ import annotations
+
+import ezdxf
+import pytest
+import pytest_asyncio
+from fastmcp import Client
+
+import server
+from tests.fixtures.plant_pair import build_plant_pair
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.fixture(scope="module")
+def pipe_layer(tmp_path_factory) -> str:
+    truth = build_plant_pair(tmp_path_factory.mktemp("plant"))
+    return next(
+        name for name, service in sorted(truth["layer_services"].items()) if service == "product"
+    )
+
+
+@pytest_asyncio.fixture
+async def client(monkeypatch):
+    """In-memory client on a fresh headless document - 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"
+        await connected.call_tool("drawing_new", {})
+        yield connected
+
+
+async def _line(client, layer, x1, y1, x2, y2):
+    await client.call_tool(
+        "entity_create_line", {"x1": x1, "y1": y1, "x2": x2, "y2": y2, "layer": layer}
+    )
+
+
+async def _total(client) -> int:
+    stats = (await client.call_tool("analysis_entity_stats", {})).structured_content
+    return stats["total_entities"]
+
+
+async def test_the_default_layers_are_the_classified_network_layers(client, pipe_layer):
+    await _line(client, pipe_layer, 0, 0, 1000, 0)
+    await _line(client, pipe_layer, 1003, 0, 2003, 0)
+    before = await _total(client)
+    result = (await client.call_tool("drawing_topology_check", {})).structured_content
+    assert result["layers"] == [pipe_layer]
+    assert [row["layer"] for row in result["classification"]] == [pipe_layer]
+    assert result["counts"] == {"dangling": 2, "near_miss": 1, "crossing": 0}
+    assert result["near_miss"][0]["gap"] == pytest.approx(3.0)
+    assert await _total(client) == before, "the check never writes"
+
+
+async def test_a_dxf_path_is_read_without_opening_it(client, pipe_layer, tmp_path):
+    doc = ezdxf.new()
+    doc.layers.add(pipe_layer)
+    msp = doc.modelspace()
+    msp.add_line((0, 0), (1000, 0), dxfattribs={"layer": pipe_layer})
+    msp.add_line((500, -500), (500, 500), dxfattribs={"layer": pipe_layer})
+    path = tmp_path / "crossing.dxf"
+    doc.saveas(path)
+    result = (
+        await client.call_tool("drawing_topology_check", {"path": str(path), "gap": 5.0})
+    ).structured_content
+    assert result["counts"]["crossing"] == 1
+    assert result["crossing"][0]["at"] == [500.0, 0.0]
+    assert await _total(client) == 0, "the current document is untouched"
+
+
+async def test_an_unknown_layer_is_refused_with_the_drawings_layers(client, pipe_layer):
+    await _line(client, pipe_layer, 0, 0, 10, 0)
+    result = await client.call_tool(
+        "drawing_topology_check", {"layers": ["NOPE"]}, raise_on_error=False
+    )
+    assert result.is_error is True
+    assert "layers: 'NOPE' is not a layer of this drawing" in result.content[0].text
+
+
+async def test_a_drawing_with_no_network_layer_is_refused_not_passed(client):
+    await _line(client, "GEOMETRY", 0, 0, 10, 0)
+    result = await client.call_tool("drawing_topology_check", {}, raise_on_error=False)
+    assert result.is_error is True
+    assert "no layer of this drawing is classified piping or electrical" in result.content[0].text
+
+
+async def test_gap_must_exceed_tol(client, pipe_layer):
+    await _line(client, pipe_layer, 0, 0, 10, 0)
+    result = await client.call_tool(
+        "drawing_topology_check", {"gap": 0.005, "tol": 0.01}, raise_on_error=False
+    )
+    assert result.is_error is True
+    assert "gap (0.005) must exceed tol (0.01)" in result.content[0].text
+
+
+async def test_the_check_is_advertised_read_only(client):
+    tools = {tool.name: tool for tool in await client.list_tools()}
+    assert tools["drawing_topology_check"].annotations.readOnlyHint is True