fix(pid): signal lines carry no fabricated number; a bubble exits towards the first waypoint

An ISA-5.1 signal line has no pipe line number, yet every pneumatic /
electric line got the bare sequence as its number and a label, and ate a
process sequence number. Signal classes now leave line_number None (a
verbatim one is still honoured), seq None, no label. A radial port aimed
at the far end even with waypoints, so the first segment cut through the
bubble; aim_points (lines.py) aims at the first/last waypoint for both
draw_line and the dry run.

Merge note: benchmarks/adapters/autocad_mcp_pro.py (final group's file)
had its two _task_pid_roundtrip assertion lines moved to the new truth —
the signal row is unnumbered, the connector row is 3.

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

benchmarks/adapters/autocad_mcp_pro.py

@@ -381,19 +381,27 @@ class AutoCADMCPProAdapter(BenchmarkAdapter):
         graph = result["graph"]
         index = instrument_index(graph)
         lines = line_list(graph)
-        # The spec numbers its two process lines; the signal line and the
+        # The spec numbers its two process lines; the connector line is left
-        # connector line are left to the default ``number_format``, which with
+        # to the default ``number_format``, which with no size/service/spec/
-        # no size/service/spec/insulation collapses to the bare sequence
+        # insulation collapses to the bare sequence number, and it must come
-        # number. Both must come back through XDATA, not a label search.
+        # back through XDATA, not a label search. The electric signal line is
+        # unnumbered by rule (ISA-5.1 signal lines carry no pipe line number)
+        # and must not have consumed a sequence number: the connector is "3".
         authored = "100-P-1001-CS1"
         numbers = {(r["from_tag"], r["to_tag"]): r["line_number"] for r in lines}
         authored_ok = (
             numbers.get(("P-101", "FCV-101")) == authored
             and numbers.get(("FCV-101", "V-201")) == authored
         )
-        sequenced = [r for r in lines if r["line_number"] != authored]
+        others = [r for r in lines if r["line_number"] != authored]
-        sequenced_ok = len(sequenced) == 2 and all(
+        signal = [r for r in others if r["line_class"] == "electric"]
-            r["line_number"].isdigit() and r["number_source"] == "xdata" for r in sequenced
+        sequenced = [r for r in others if r["line_class"] != "electric"]
+        sequenced_ok = (
+            len(signal) == 1
+            and signal[0]["line_number"] is None
+            and len(sequenced) == 1
+            and sequenced[0]["line_number"] == "3"
+            and sequenced[0]["number_source"] == "xdata"
         )
         passed = (
             len(graph["nodes"]) == 5

discovery/aliases.py

@@ -1384,6 +1384,8 @@ TOOL_ALIASES: dict[str, ToolAliases] = {
             "line number",
             "connect the pump to the vessel",
             "pneumatic signal",
+            "unnumbered signal line",
+            "route a line through waypoints",
         ),
     ),
     "pid_graph": ToolAliases(

engineering/pid/drawlines.py

@@ -11,6 +11,7 @@ from .lines import (
     DEFAULT_NUMBER_FORMAT,
     LINE_CLASSES,
     PID_LINE_LAYERS,
+    aim_points,
     count_crossings,
     format_line_number,
     label_placement,
@@ -130,10 +131,11 @@ async def existing_pid_lines(backend: AutoCADBackend) -> list[dict]:
     return out
 
 
-def _anchor(endpoint: dict, other: dict) -> tuple[tuple[float, float], float | None]:
+def _anchor(endpoint: dict, aim: tuple[float, float]) -> tuple[tuple[float, float], float | None]:
-    """Where the line actually starts/ends: on the circle for radial ports."""
+    """Where the line actually starts/ends: on the circle for radial ports,
+    leaving along the axis towards ``aim`` (see ``aim_points``)."""
     if endpoint["radius"] > 0:
-        axis = snap_axis(other["x"] - endpoint["x"], other["y"] - endpoint["y"])
+        axis = snap_axis(aim[0] - endpoint["x"], aim[1] - endpoint["y"])
         ux, uy = AXIS[axis]
         return (
             endpoint["x"] + ux * endpoint["radius"],
@@ -176,8 +178,9 @@ async def draw_line(
     end = await resolve_endpoint(backend, to)
     if start["handle"] and start["handle"] == end["handle"] and start["port"] == end["port"]:
         raise ValueError("from and to are the same port")
-    s_point, s_dir = _anchor(start, end)
+    s_aim, e_aim = aim_points((start["x"], start["y"]), (end["x"], end["y"]), route_mode)
-    e_point, e_dir = _anchor(end, start)
+    s_point, s_dir = _anchor(start, s_aim)
+    e_point, e_dir = _anchor(end, e_aim)
     path = route(s_point, s_dir, e_point, e_dir, stub=stub, mode=route_mode)
 
     existing = await existing_pid_lines(backend)
@@ -197,17 +200,23 @@ async def draw_line(
         for ep in (start, end)
         if ep["handle"] and (ep["handle"], ep["port"]) in used
     ]
-    seq = max_seq + 1
+    if cls.kind == "signal":
-    if line_number is None:
+        # An ISA-5.1 signal line carries no pipe line number: no fabricated
-        fmt = number_format or DEFAULT_NUMBER_FORMAT
+        # number, no label, and the process sequence is left alone. A
-        fields = {
+        # verbatim ``line_number`` is still honoured (and labelled).
-            "size": size,
+        seq = None
-            "service": service,
+    else:
-            "seq": seq,
+        seq = max_seq + 1
-            "spec": spec,
+        if line_number is None:
-            "insulation": insulation,
+            fmt = number_format or DEFAULT_NUMBER_FORMAT
-        }
+            fields = {
-        line_number = format_line_number(fmt, **fields) or None
+                "size": size,
+                "service": service,
+                "seq": seq,
+                "spec": spec,
+                "insulation": insulation,
+            }
+            line_number = format_line_number(fmt, **fields) or None
 
     await ensure_layer(backend, cls.layer)
     poly = await backend.entity_create_polyline(

engineering/pid/lines.py

@@ -198,6 +198,21 @@ def route(start, start_dir, end, end_dir, stub: float = 5.0, mode="auto") -> lis
     return valid[0]
 
 
+def aim_points(start, end, mode) -> tuple[Point, Point]:
+    """What each end of a line aims at: the other end, or — with waypoints —
+    the first waypoint for the start and the last one for the end.
+
+    A radial (bubble) port leaves its circle along the axis towards its aim
+    point. Aiming at the far end while the route is a waypoint list put the
+    exit on the wrong side of the bubble, so the first segment cut through it.
+    """
+    if isinstance(mode, (list, tuple)) and len(mode) > 0:
+        return _finite_point("waypoints[0]", mode[0]), _finite_point(
+            f"waypoints[{len(mode) - 1}]", mode[-1]
+        )
+    return _finite_point("end", end), _finite_point("start", start)
+
+
 def label_placement(vertices, offset: float = 1.5) -> tuple[float, float, float, int]:
     """Midpoint of the longest segment, offset to its left, readable rotation."""
     pts = [(float(x), float(y)) for x, y in vertices]

engineering/pid/spec.py

@@ -13,7 +13,7 @@ from .critique import PID_FOCUSES, issues_for
 from .drawlines import draw_line
 from .graph import build_graph
 from .insert import place_symbol
-from .lines import AXIS, LINE_CLASSES, count_crossings, route, snap_axis
+from .lines import AXIS, LINE_CLASSES, aim_points, count_crossings, route, snap_axis
 from .symbols import resolve, transform_port
 
 if TYPE_CHECKING:
@@ -311,9 +311,9 @@ def _plan_lines(norm: dict) -> list[dict]:
             raise ValueError(f"{ref}: no port {port!r}; ports: {', '.join(entry['ports'])}")
         return entry["ports"][port]
 
-    def anchor(ep: dict, other: dict):
+    def anchor(ep: dict, aim):
         if ep["radius"] > 0:
-            axis = snap_axis(other["x"] - ep["x"], other["y"] - ep["y"])
+            axis = snap_axis(aim[0] - ep["x"], aim[1] - ep["y"])
             ux, uy = AXIS[axis]
             return (ep["x"] + ux * ep["radius"], ep["y"] + uy * ep["radius"]), axis
         return (ep["x"], ep["y"]), ep["direction_deg"]
@@ -322,11 +322,11 @@ def _plan_lines(norm: dict) -> list[dict]:
     for index, (where, line) in enumerate(_all_lines(norm)):
         try:
             start, end = endpoint(line["from"]), endpoint(line["to"])
-            s, sd = anchor(start, end)
+            mode = line.get("route", "auto")
-            e, ed = anchor(end, start)
+            s_aim, e_aim = aim_points((start["x"], start["y"]), (end["x"], end["y"]), mode)
-            vertices = route(
+            s, sd = anchor(start, s_aim)
-                s, sd, e, ed, stub=float(line.get("stub", 5.0)), mode=line.get("route", "auto")
+            e, ed = anchor(end, e_aim)
-            )
+            vertices = route(s, sd, e, ed, stub=float(line.get("stub", 5.0)), mode=mode)
         except ValueError as exc:
             raise ValueError(f"{where}: {exc}") from exc
         crossings = count_crossings(vertices, [p["vertices"] for p in planned])

server.py

@@ -6662,6 +6662,11 @@ async def pid_line_draw(
     (reported, never refused) and `port_reuse` names ports that already had a
     line (branching is legal). Signal classes get ISA-5.1 markers (`//`, `X`,
     `L`, `o`) as small blocks on segments >= 15 mm; electric is dashed by layer.
+    A signal line carries no pipe line number: it gets no auto-built number
+    and no label, and does not consume the process sequence — a verbatim
+    `line_number` is still written and labelled. With waypoints, a bubble's
+    exit leaves towards the first waypoint (and the entry arrives from the
+    last one); `crossings` follows existing arcs, not their chords.
     """
     from engineering.pid.drawlines import draw_line
 

tests/test_pid_deliverables.py

@@ -119,20 +119,22 @@ async def test_line_list_rows(backend):
     rows = line_list(graph)
     assert [list(r) for r in rows] == [LINE_KEYS] * 4
     numbered = [r for r in rows if r["line_number"]]
-    # The signal line carries no size/service/spec, so its default-format
+    # An ISA-5.1 signal line carries no pipe line number (track E, Task 6):
-    # number is the bare sequence "3" (empty fields drop out, spec 6.4) — and
+    # the pneumatic line is unnumbered and did not consume a sequence number,
-    # a natural sort files 3 before 100.
+    # so the two process lines are 1 and 2.
-    assert [r["line_number"] for r in numbered] == ["3", "100-P-1-CS1", "100-P-2-CS1"]
+    assert [r["line_number"] for r in numbered] == ["100-P-1-CS1", "100-P-2-CS1"]
-    signal = numbered[0]
+    signal = next(r for r in rows if r["line_class"] == "pneumatic")
-    assert signal["line_class"] == "pneumatic" and signal["number_source"] == "xdata"
+    assert signal["line_number"] is None and signal["number_source"] is None
     assert signal["from_tag"] == "FIC-101" and signal["to_tag"] == "FCV-101"
     assert signal["to_port"] == "signal"
-    first = numbered[1]
+    first = numbered[0]
     assert first["from_tag"] == "P-101" and first["from_port"] == "discharge"
     assert first["to_tag"] == "FCV-101" and first["length_mm"] > 0
-    assert rows[-1]["line_number"] is None, "untagged rows sort last"
+    unnumbered = [r for r in rows if r["line_number"] is None]
-    assert rows[-1]["from_tag"] == "V-201" and rows[-1]["from_port"] == "N4"
+    assert rows[-2:] == unnumbered, "untagged rows sort last"
-    assert rows[-1]["to_tag"] is None and rows[-1]["number_source"] is None
+    hand_drawn = next(r for r in unnumbered if r["from_tag"] == "V-201")
+    assert hand_drawn["from_port"] == "N4"
+    assert hand_drawn["to_tag"] is None and hand_drawn["number_source"] is None
 
 
 async def test_equipment_list_rows(backend):

tests/test_pid_line_draw.py

@@ -290,3 +290,137 @@ async def test_existing_lines_are_seen_on_the_com_engine(monkeypatch):
     assert lines[0]["vertices"] == [(100.0, 106.0), (100.0, 128.0), (207.0, 128.0)]
     assert lines[0]["payload"]["seq"] == 7
     assert lines[0]["payload"]["from"] == {"handle": "2A", "port": "discharge"}
+
+
+# ── Track A hardening (track E wave 0, Task 6) ───────────────────────────────
+
+
+async def _bubble_and_valve(backend):
+    cv = await place_symbol(backend, "globe", 100, 100, tag="FCV-1", actuator="diaphragm")
+    fic = await place_symbol(
+        backend, "instrument", 100, 160, tag="FIC-1", type="dcs", location="primary"
+    )
+    return cv, fic
+
+
+async def test_signal_lines_carry_no_fabricated_number_and_do_not_consume_the_sequence(backend):
+    """A signal line has no pipe line number: no auto number, no label, and the
+    process sequence is left alone."""
+    cv, fic = await _bubble_and_valve(backend)
+    pump, vessel = await _pump_and_vessel(backend)
+    signal = await draw_line(
+        backend, {"handle": fic["handle"]}, {"handle": cv["handle"], "port": "signal"}, "pneumatic"
+    )
+    assert signal["line_number"] is None and signal["label_handle"] is None
+    payload = await read_payload(backend, signal["handle"])
+    assert payload["number"] is None and payload["seq"] is None
+    labels = await backend.entity_count(type_filter="TEXT", layer_filter="PROCESS-LINE-TEXT")
+    assert labels == 0, "no label text was written for the signal line"
+
+    process = await draw_line(
+        backend,
+        {"handle": pump["handle"], "port": "discharge"},
+        {"handle": vessel["handle"], "port": "N3"},
+        size="100",
+        service="P",
+    )
+    assert process["line_number"] == "100-P-1", "the signal line did not consume seq 1"
+    assert (await read_payload(backend, process["handle"]))["seq"] == 1
+
+    # A signal line given size/service still gets no auto number (kind rules).
+    electric = await draw_line(
+        backend, {"handle": fic["handle"]}, {"x": 160, "y": 160}, "electric", size="IA", service="S"
+    )
+    assert electric["line_number"] is None and electric["label_handle"] is None
+
+    # A verbatim number is honoured and labelled, seq stays None.
+    named = await draw_line(
+        backend, {"handle": fic["handle"]}, {"x": 40, "y": 160}, "electric", line_number="IA-7"
+    )
+    assert named["line_number"] == "IA-7" and named["label_handle"]
+    assert (await backend.entity_get(named["label_handle"])).properties["text"] == "IA-7"
+    assert (await read_payload(backend, named["handle"]))["seq"] is None
+
+    # The next process line is 2, not 4.
+    second = await draw_line(
+        backend,
+        {"handle": vessel["handle"], "port": "N4"},
+        {"x": 300, "y": 128},
+        size="80",
+        service="P",
+    )
+    assert second["line_number"] == "80-P-2"
+
+
+async def test_a_bubble_exit_aims_at_the_first_waypoint(backend):
+    """With waypoints the radial port must leave towards the first waypoint;
+    aiming at the far end put the exit on the wrong side and the first segment
+    cut through the bubble (measured: 4.93 mm from the centre of a r=5 bubble)."""
+    fic = await place_symbol(
+        backend, "instrument", 100, 160, tag="FIC-1", type="dcs", location="primary"
+    )
+    result = await draw_line(
+        backend,
+        {"handle": fic["handle"]},
+        {"x": 100, "y": 100},
+        "electric",
+        route_mode=[[130, 160], [130, 100]],
+    )
+    assert result["vertices"] == [[105.0, 160.0], [130.0, 160.0], [130.0, 100.0], [100.0, 100.0]]
+    # Every vertex of the drawn line is on or outside the bubble.
+    for x, y in result["vertices"]:
+        assert ((x - 100) ** 2 + (y - 160) ** 2) ** 0.5 >= 5.0 - 1e-9
+
+
+async def test_a_bubble_entry_aims_at_the_last_waypoint(backend):
+    fic = await place_symbol(
+        backend, "instrument", 100, 160, tag="FIC-1", type="dcs", location="primary"
+    )
+    result = await draw_line(
+        backend,
+        {"x": 100, "y": 100},
+        {"handle": fic["handle"]},
+        "electric",
+        route_mode=[[130, 100], [130, 160]],
+    )
+    assert result["vertices"] == [[100.0, 100.0], [130.0, 100.0], [130.0, 160.0], [105.0, 160.0]]
+
+
+async def test_dry_run_plans_the_same_bubble_exit_as_the_real_run(backend):
+    from engineering.pid.spec import run_spec
+
+    spec = {
+        "instruments": [
+            {
+                "id": "FIC-1",
+                "type": "dcs",
+                "location": "primary",
+                "x": 100,
+                "y": 160,
+                "tag": "FIC-1",
+            }
+        ],
+        "valves": [
+            {
+                "id": "FCV-1",
+                "symbol": "globe",
+                "actuator": "diaphragm",
+                "x": 100,
+                "y": 100,
+                "tag": "FCV-1",
+            }
+        ],
+        "lines": [
+            {
+                "from": "FIC-1",
+                "to": "FCV-1.signal",
+                "class": "electric",
+                "route": [[130, 160], [130, 109]],
+            }
+        ],
+    }
+    planned = await run_spec(backend, spec, dry_run=True)
+    assert planned["lines"][0]["vertices"][0] == [105.0, 160.0]
+    drawn = await run_spec(backend, spec)
+    edge = drawn["graph"]["edges"][0]
+    assert edge["line_number"] is None, "signal lines from a spec are unnumbered too"