fix(task-6): the reader no longer hands a signal line the neighbouring pipe's label

pid_graph step 5 ran the TEXT label fallback for every edge without a
number. Once signal lines stopped carrying a fabricated number (3c0babf),
every pneumatic/electric run drawn by pid_line_draw entered it, and a
transmitter-to-controller run 12 mm above a labelled header came back as
line_number="100-P-1", number_source="label" — on both engines, with no
confidence field on edges to flag it.

The fallback now skips an edge that carries its own record (an
ACADMCP_PID payload: its `number: null` is the answer) and any
signal-class edge, payload or layer-classified (`signal_unknown`
included): LINE_NUMBER_RE is the pipe grammar, so a match near a signal
line always belongs to a neighbouring pipe. A foreign process line still
takes its label; a verbatim number on a signal line still reads back
from XDATA.

Tests: the finding's layout in test_pid_graph (engine-drawn signal beside
a labelled pipe, verbatim signal number, foreign electric and
signal_unknown lines beside a labelled foreign pipe) and the line_list
row in test_pid_deliverables — three of the four fail on the old reader.

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

engineering/pid/graph.py

@@ -12,7 +12,7 @@ import re
 from dataclasses import dataclass, field
 from typing import TYPE_CHECKING
 
-from .lines import LAYER_TO_CLASS, PID_LINE_LAYERS
+from .lines import LAYER_TO_CLASS, LINE_CLASSES, PID_LINE_LAYERS
 from .symbols import Port, transform_port
 from .tags import parse_tag
 from .xdata import read_payload
@@ -55,6 +55,15 @@ CONFIDENCE = {"catalog": 1.0, "xdata": 0.95, "heuristic": 0.6, "inferred": 0.3}
 _PAGE = 1000
 
 
+def _is_signal_class(line_class: str | None) -> bool:
+    """True for a class that is a signal in the ISA-5.1 sense (`signal_unknown`
+    is the layer-classified one that LINE_CLASSES does not list)."""
+    if line_class == "signal_unknown":
+        return True
+    cls = LINE_CLASSES.get(line_class) if line_class else None
+    return bool(cls and cls.kind == "signal")
+
+
 @dataclass
 class _Node:
     id: str
@@ -684,7 +693,15 @@ async def build_graph(
             if found:
                 node.tag, node.tag_source = found, "text"
     for edge in edges.values():
-        if edge.line_number:
+        if edge.line_number or edge.payload or _is_signal_class(edge.line_class):
+            # The label search is a fallback for a line that carries no
+            # record of its own. An engine-authored line (payload) is that
+            # record — its `number: null` is the answer, not an invitation to
+            # adopt the nearest pipe's label. A signal line never has a pipe
+            # line number (ISA-5.1), and LINE_NUMBER_RE is the pipe grammar,
+            # so any match near it belongs to a neighbouring pipe: the
+            # ordinary transmitter-to-controller run 12 mm above a header
+            # would otherwise be listed under the header's number.
             continue
         a, b = max(zip(edge.vertices, edge.vertices[1:], strict=False), key=lambda s: math.dist(*s))
         found = _text_near(

tests/test_pid_deliverables.py

@@ -137,6 +137,30 @@ async def test_line_list_rows(backend):
     assert hand_drawn["to_tag"] is None and hand_drawn["number_source"] is None
 
 
+async def test_line_list_keeps_a_signal_line_unnumbered_beside_a_labelled_pipe(backend):
+    """The line list prints what the reader publishes, and the reader must not
+    hand a 12 mm-parallel signal run the header's label (its own XDATA says
+    ``number: null``)."""
+    proc = await draw_line(
+        backend, {"x": 20, "y": 100}, {"x": 200, "y": 100}, size="100", service="P"
+    )
+    ft = await place_symbol(
+        backend, "instrument", 40, 112, tag="FT-1", type="dcs", location="primary"
+    )
+    fic = await place_symbol(
+        backend, "instrument", 180, 112, tag="FIC-1", type="dcs", location="primary"
+    )
+    sig = await draw_line(backend, {"handle": ft["handle"]}, {"handle": fic["handle"]}, "electric")
+    rows = {r["handle"]: r for r in line_list(await build_graph(backend))}
+    assert (rows[proc["handle"]]["line_number"], rows[proc["handle"]]["number_source"]) == (
+        "100-P-1",
+        "xdata",
+    )
+    signal = rows[sig["handle"]]
+    assert (signal["line_number"], signal["number_source"]) == (None, None)
+    assert (signal["from_tag"], signal["to_tag"]) == ("FT-1", "FIC-1")
+
+
 async def test_equipment_list_rows(backend):
     graph = await _pid(backend)
     rows = equipment_list(graph)

tests/test_pid_graph.py

@@ -289,6 +289,71 @@ async def test_line_number_and_tags_from_nearby_text(backend):
     assert edge["line_number"] == '6"-P-1234-CS1' and edge["number_source"] == "label"
 
 
+async def test_a_signal_line_next_to_a_pipe_label_never_adopts_the_pipes_number(backend):
+    """The label search is a fallback for a line with no record of its own.
+
+    A transmitter-to-controller run 12 mm above a labelled header (the
+    ordinary spacing on a dense sheet) is well inside ``label_search``; its
+    own payload says ``number: null`` and a signal line has no pipe line
+    number, so the reader must report None — not the header's number under
+    ``number_source="label"``.
+    """
+    proc = await draw_line(
+        backend, {"x": 20, "y": 100}, {"x": 200, "y": 100}, size="100", service="P"
+    )
+    assert proc["line_number"] == "100-P-1"
+    ft = await place_symbol(
+        backend, "instrument", 40, 112, tag="FT-1", type="dcs", location="primary"
+    )
+    fic = await place_symbol(
+        backend, "instrument", 180, 112, tag="FIC-1", type="dcs", location="primary"
+    )
+    sig = await draw_line(backend, {"handle": ft["handle"]}, {"handle": fic["handle"]}, "electric")
+    assert sig["line_number"] is None
+    graph = await build_graph(backend)
+    by_handle = {e["id"]: e for e in graph["edges"]}
+    pipe, signal = by_handle[proc["handle"]], by_handle[sig["handle"]]
+    assert (pipe["line_number"], pipe["number_source"]) == ("100-P-1", "xdata")
+    assert (signal["line_number"], signal["number_source"]) == (None, None)
+    assert signal["from"]["node"] == ft["handle"] and signal["to"]["node"] == fic["handle"]
+
+
+async def test_a_verbatim_number_on_a_signal_line_is_still_read_back(backend):
+    ft = await place_symbol(
+        backend, "instrument", 40, 112, tag="FT-1", type="dcs", location="primary"
+    )
+    fic = await place_symbol(
+        backend, "instrument", 180, 112, tag="FIC-1", type="dcs", location="primary"
+    )
+    sig = await draw_line(
+        backend,
+        {"handle": ft["handle"]},
+        {"handle": fic["handle"]},
+        "electric",
+        line_number="SIG-7",
+    )
+    graph = await build_graph(backend)
+    edge = next(e for e in graph["edges"] if e["id"] == sig["handle"])
+    assert (edge["line_number"], edge["number_source"]) == ("SIG-7", "xdata")
+
+
+async def test_a_foreign_signal_line_next_to_a_pipe_label_stays_unnumbered(backend):
+    """A layer-classified signal line (no payload) is a signal all the same:
+    LINE_NUMBER_RE is the pipe grammar, so a match near it is a neighbour's.
+    The pipe on the same sheet still takes its label."""
+    await backend.entity_create_line(0, 0, 100, 0, layer="PROCESS-PIPING-MAIN")
+    await backend.entity_create_text('6"-P-1234-CS1', 40, 2, 2.5, layer="PROCESS-LINE-TEXT")
+    await backend.entity_create_line(0, 6, 100, 6, layer="ELECTRICAL-LINE")
+    await backend.entity_create_line(0, 10, 100, 10, layer="INSTRUMENT-LINE-SIGNAL")
+    graph = await build_graph(backend)
+    by_class = {e["line_class"]: e for e in graph["edges"]}
+    assert set(by_class) == {"process_major", "electric", "signal_unknown"}
+    pipe = by_class["process_major"]
+    assert (pipe["line_number"], pipe["number_source"]) == ('6"-P-1234-CS1', "label")
+    for cls in ("electric", "signal_unknown"):
+        assert (by_class[cls]["line_number"], by_class[cls]["number_source"]) == (None, None), cls
+
+
 async def test_scope_all_walks_every_layout_and_restores_the_current_one(backend):
     await place_symbol(backend, "gate", 0, 0, tag="HV-1")
     await backend.layout_create("SHEET-2")