From 715c8efca449e8578d65ce8c89404c0c722b5014 Mon Sep 17 00:00:00 2001 From: Umut CAN Date: Wed, 16 Sep 2026 11:35:35 +0300 Subject: [PATCH] fix(task-16): P&ID focuses score evidence, never a 0.3 touch guess or an inferred port kind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two ways the six focuses turned a reader guess into a gate-failing error: - The "no P&ID content" guard was `not graph["nodes"]`, but build_graph (include_foreign=True) files ANY INSERT as an `unknown_block` at 0.3 the moment a plain LINE's end lands on its box. A bolt with a note leader on a purely mechanical sheet became a P&ID node, the leader a foreign edge, and its free end a `pid_dangling_line` error — so drawing_finalize (focus=None) failed a sheet the seven mechanical focuses pass. Now `has_pid_content` needs a recognised node (catalog / xdata / heuristic); an `unknown_block` is never tagged or scored, and `_dangling` only reports an edge that is a P&ID line by evidence (a class from XDATA or a P&ID layer, or an end on a recognised node) — on a real P&ID the same leader is not dangling either. - `_incompatible` read `port["kind"]` off inferred ports, which the graph hard-codes to "process" for every foreign block. A signal line correctly drawn to a CTO control valve's actuator was an error whose hint said to redraw the control signal as a process line. Inferred ports are skipped; only a declared (catalogue / XDATA) port kind is judged. Tests: both repros as regressions (unit + the server drawing_finalize gate), a foreign GATE_VALVE whose dangling line is still reported (the heuristic path spec §9.3 pins is not silenced), a note leader on a real P&ID, and the spec's own negative — all six return [] on the README gear sheet. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01FKFStjEFzpGyLq5bMpWxTd --- engineering/pid/critique.py | 60 +++++++++++++++-- tests/test_pid_critique.py | 126 ++++++++++++++++++++++++++++++++++++ 2 files changed, 182 insertions(+), 4 deletions(-) diff --git a/engineering/pid/critique.py b/engineering/pid/critique.py index a928041..424d6ff 100644 --- a/engineering/pid/critique.py +++ b/engineering/pid/critique.py @@ -2,7 +2,18 @@ Every check is pure over the graph dict; ``run_critique`` builds the graph once into the per-run ``shared`` context so six focuses cost one read. A drawing -with no P&ID nodes yields no issues — mechanical sheets are never scored here. +with no recognised P&ID symbol yields no issues — mechanical sheets are never +scored here. + +The checks score evidence, never a guess. ``build_graph(include_foreign=True)`` +files *any* INSERT as an ``unknown_block`` at confidence 0.3 the moment a plain +LINE's end lands on its bounding box (spec §9.2 step 2) — a bolt with a note +leader qualifies — and gives every foreign block's inferred port a placeholder +``kind: "process"``. Neither is a measurement (spec §9.4: read the confidence +before trusting a foreign graph), so no focus rests on them: an +``unknown_block`` is not P&ID content, is never tagged or scored, and an edge +anchored on nothing better is never called dangling; an inferred port is never +called incompatible. """ from __future__ import annotations @@ -29,9 +40,44 @@ PID_FOCUSES = ( _KEY = "pid_graph" +def _recognised(node: dict) -> bool: + """A node the drawing vouches for: catalogue, XDATA or heuristic (a keyword + block name / tag attribute). ``unknown_block`` is a 0.3 touch guess.""" + return node["kind"] != "unknown_block" + + +def has_pid_content(graph: dict) -> bool: + """True when at least one recognised P&ID symbol is on the sheet.""" + return any(_recognised(n) for n in graph["nodes"]) + + +def _scored_nodes(graph: dict) -> list[dict]: + return [n for n in graph["nodes"] if _recognised(n)] + + +def _evidenced_edges(graph: dict) -> set[str]: + """Edge ids that are P&ID lines by evidence: a line class from XDATA or a + P&ID line layer, or an end on a recognised node. A foreign line whose only + anchor is an ``unknown_block`` box is in the graph on a guess.""" + nodes = {n["id"]: n for n in graph["nodes"]} + out: set[str] = set() + for edge in graph["edges"]: + if edge.get("class_source"): + out.add(edge["id"]) + continue + for ref in (edge["from"], edge["to"]): + if ref and "node" in ref and _recognised(nodes[ref["node"]]): + out.add(edge["id"]) + break + return out + + def _dangling(graph: dict) -> list[Issue]: + evidenced = _evidenced_edges(graph) out = [] for d in graph["dangling"]: + if d["edge"] not in evidenced: + continue nearest = d.get("nearest") if nearest: hint = ( @@ -56,7 +102,7 @@ def _dangling(graph: dict) -> list[Issue]: def _duplicate_tags(graph: dict) -> list[Issue]: seen: dict[str, list[str]] = {} - for node in graph["nodes"]: + for node in _scored_nodes(graph): # connectors carry free text ("TO P&ID-002"), never a unique tag if node.get("tag") and node["kind"] != "connector": seen.setdefault(node["tag"].upper(), []).append(node["id"]) @@ -87,6 +133,12 @@ def _incompatible(graph: dict) -> list[Issue]: if not ref or "node" not in ref: continue port = nodes[ref["node"]]["ports"].get(ref["port"], {}) + if port.get("inferred"): + # A foreign block's port kind is a placeholder, not a + # measurement: a signal line to a CTO control valve's actuator + # is right, and "redraw as a process line" would be the wrong + # fix. Only a declared (catalogue / XDATA) port kind is judged. + continue actual = port.get("kind") if actual in ("process", "signal") and actual != cls.kind: out.append( @@ -126,7 +178,7 @@ def _untagged(graph: dict) -> list[Issue]: def _illegal(graph: dict) -> list[Issue]: out = [] - for n in graph["nodes"]: + for n in _scored_nodes(graph): if not n.get("tag") or n["kind"] == "connector": continue parsed = parse_tag(n["tag"], kind="instrument" if n["kind"] == "instrument" else "auto") @@ -170,7 +222,7 @@ _CHECKS = { def issues_for(focus: str, graph: dict) -> list[Issue]: - if not graph["nodes"]: + if not has_pid_content(graph): return [] return _CHECKS[focus](graph) diff --git a/tests/test_pid_critique.py b/tests/test_pid_critique.py index da6c951..84e9650 100644 --- a/tests/test_pid_critique.py +++ b/tests/test_pid_critique.py @@ -4,7 +4,9 @@ from __future__ import annotations import pytest +from engineering.pid.critique import has_pid_content, issues_for from engineering.pid.drawlines import draw_line +from engineering.pid.graph import build_graph from engineering.pid.insert import place_symbol from engineering.plan_spec import ALL_CRITIQUE_FOCUSES from engineering.preflight import pid_plan_warnings, preflight_drawing @@ -110,6 +112,53 @@ async def test_unconnected_equipment_is_informational(backend): assert len(issues) == 1 and issues[0].severity == "info" +async def test_incompatible_is_silent_on_a_foreign_blocks_inferred_port(backend): + """A CTO-library control valve has no declared ports: the graph infers one + where the line lands and gives it a placeholder ``kind: "process"``. A signal + line to its actuator is correct, and the old "redraw as a process line" + error was a guess presented as a measurement.""" + await backend.drawing_apply_iso_layers("pid") + await backend.block_define( + "CONTROL_VALVE", + [ + {"type": "polyline", "points": [[-6, -3], [6, 3], [6, -3], [-6, 3]], "closed": True}, + {"type": "line", "x1": 0, "y1": 0, "x2": 0, "y2": 6}, + {"type": "arc", "cx": 0, "cy": 6, "r": 3, "start_deg": 0, "end_deg": 180}, + ], + attdefs=[{"tag": "TAG", "x": 0, "y": 12, "height": 2.5}], + ) + cv = await backend.block_insert( + "CONTROL_VALVE", 160, 100, attributes={"TAG": "FCV-101"}, layer="PROCESS-EQUIPMENT" + ) + fic = await place_symbol( + backend, "instrument", 160, 140, tag="FIC-101", type="dcs", location="primary" + ) + await draw_line(backend, {"handle": fic["handle"]}, {"x": 160, "y": 109}, line_class="electric") + graph = await build_graph(backend, include_foreign=True) + valve = next(n for n in graph["nodes"] if n["id"] == cv.handle) + assert valve["source"] == "heuristic" and valve["ports"]["p1"]["inferred"] is True + issues = await backend.drawing_critique(None) + assert [i.focus for i in issues if i.focus.startswith("pid_")] == [] + + +async def test_a_dangling_line_off_a_foreign_valve_is_still_reported(backend): + """The evidence rule must not silence the foreign path spec §9.3 pins: a + heuristic node (keyword name + TAG attribute) is recognised, so a plain + line leaving its box for nowhere is a real dangling end.""" + await backend.block_define( + "GATE_VALVE", + [ + {"type": "polyline", "points": [[-4, -2], [-4, 2], [0, 0]], "closed": True}, + {"type": "polyline", "points": [[4, -2], [4, 2], [0, 0]], "closed": True}, + ], + attdefs=[{"tag": "TAG", "x": 0, "y": 4, "height": 2.5}], + ) + await backend.block_insert("GATE_VALVE", 50, 50, attributes={"TAG": "HV-9"}) + line = await backend.entity_create_line(54, 50, 90, 50, layer="0") + issues = await backend.drawing_critique(["pid_dangling_line"]) + assert [(i.severity, i.handles) for i in issues] == [("error", [line.handle])] + + async def test_mechanical_drawing_is_untouched_by_pid_focuses(backend): from engineering.layers import ensure_engineering_layers @@ -119,6 +168,83 @@ async def test_mechanical_drawing_is_untouched_by_pid_focuses(backend): assert await backend.drawing_critique(list(PID_FOCUSES)) == [] +async def _bolt_with_note_leader(backend): + """A mechanical INSERT with a note leader from its edge — the shape that + files the bolt as an ``unknown_block`` at 0.3 and the leader as a foreign + edge with a free far end.""" + await backend.block_define( + "BOLT_M10", + [ + {"type": "circle", "cx": 0, "cy": 0, "r": 5}, + {"type": "polyline", "points": [[-5, -5], [5, -5], [5, 5], [-5, 5]], "closed": True}, + ], + ) + bolt = await backend.block_insert("BOLT_M10", 100, 100, layer="GEOMETRY") + leader = await backend.entity_create_line(105, 100, 140, 120, layer="GEOMETRY") + await backend.entity_create_text("SEE NOTE 3", 141, 120, 3.5, layer="TEXT") + return bolt.handle, leader.handle + + +async def test_a_mechanical_insert_with_a_note_leader_is_not_pid_content(backend): + """Spec §11.1: a drawing without P&ID content is never scored. The graph + still files the bolt as an ``unknown_block`` (that is the reader's job); + the critique must not mistake that 0.3 guess for a P&ID symbol and fail + the finalize gate on the leader's free end.""" + await backend.drawing_apply_iso_layers("mech") + bolt, leader = await _bolt_with_note_leader(backend) + graph = await build_graph(backend, include_foreign=True) + assert [(n["id"], n["kind"], n["confidence"]) for n in graph["nodes"]] == [ + (bolt, "unknown_block", 0.3) + ] + assert [d["edge"] for d in graph["dangling"]] == [leader] + assert has_pid_content(graph) is False + for focus in PID_FOCUSES: + assert issues_for(focus, graph) == [], focus + issues = await backend.drawing_critique(None) + assert [i.focus for i in issues if i.focus.startswith("pid_")] == [] + + +async def test_finalize_gate_passes_a_mechanical_sheet_with_a_note_leader(backend, tmp_path): + """The gate runs ``focus=None``, so the six focuses sit inside every + ``drawing_finalize`` — the mechanical sheet must pass it as it did before + they joined.""" + import server + + class _Ctx: + def __init__(self, backend): + self.lifespan_context = {"backend": backend} + + async def warning(self, message): + pass + + await backend.drawing_apply_iso_layers("mech") + await _bolt_with_note_leader(backend) + payload = await server.drawing_finalize(save_path=str(tmp_path / "bolt.dxf"), ctx=_Ctx(backend)) + assert payload["critique"] == [] + + +async def test_a_note_leader_off_an_unknown_block_on_a_pid_is_not_dangling(backend): + """On a real P&ID the same leader is in the graph only because it touched + a 0.3 box; the one genuine finding (the free process line) stays.""" + await _clean(backend) + await _bolt_with_note_leader(backend) + issues = await backend.drawing_critique(["pid_dangling_line"]) + assert len(issues) == 1 and issues[0].detail["x"] == pytest.approx(220.0) + + +async def test_readme_gear_sheet_is_untouched_by_pid_focuses(backend): + """The spec's own negative: all six return [] on the README gear sheet.""" + import importlib.util + from pathlib import Path + + path = Path(__file__).resolve().parents[1] / "scripts" / "render_readme_showcase.py" + spec = importlib.util.spec_from_file_location("render_readme_showcase", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + await module.build(backend) + assert await backend.drawing_critique(list(PID_FOCUSES)) == [] + + async def test_refine_reports_pid_issues_as_manual_and_moves_nothing(backend): # The refine loop is `engineering.refiner.refine_drawing` (what the # `drawing_refine` tool calls) — the backend has no `drawing_refine` method.