"""Reference adapter for this repository, covering the v4 task matrix. Every task verifies against a value worked out independently of the code under test — a closed-form area, a count of entities placed on purpose, a token ceiling fixed before the measurement. A task that only checks the server agrees with itself is a task that cannot fail. """ from __future__ import annotations import math from pathlib import Path from benchmarks.adapters.base import BenchmarkAdapter, TaskResult from benchmarks.tasks_v4 import TaskSpec from engineering.delivery import deliver_drawing from engineering.layers import ensure_engineering_layers, ensure_standard_linetypes from engineering.refiner import refine_drawing #: AutoCAD command names a drafter types, none of which appeared in any tool #: name or description before v1.5.0 — measured `df = 0` against the stock BM25 #: index, so no amount of tuning could have surfaced them. #: Each command is answered by a *set*, because more than one tool can be a #: correct answer -- QSELECT is both "filter by properties" and "select #: similar", and picking a winner between them would test the tie-break rather #: than the discovery. DISCOVERY_QUERIES: tuple[tuple[str, frozenset[str]], ...] = ( ("FILLET", frozenset({"entity_fillet"})), ("BPOLY", frozenset({"boundary_trace"})), ("QSELECT", frozenset({"selection_filter", "entity_select_smart"})), ("WBLOCK", frozenset({"block_create_from_entities"})), ("OVERKILL", frozenset({"drawing_refine"})), ("CHSPACE", frozenset({"entity_change_space"})), ) #: How far down the ranking still counts as found. A drafter reads the top few. DISCOVERY_TOP_N = 3 #: The advertised catalog must stay under this in discovery mode. Fixed here #: rather than derived from the measurement, so the measurement can miss it. TOKEN_CEILING = 2_000 class AutoCADMCPProAdapter(BenchmarkAdapter): name = "autocad-mcp-pro" def __init__(self, *, backend: str = "ezdxf"): self.backend_name = backend self.backend = None self.artifact_dir = Path() async def setup(self, artifact_dir: Path) -> None: if self.backend_name == "ezdxf": from backends.ezdxf_backend import EzdxfBackend self.backend = EzdxfBackend() elif self.backend_name == "com": from backends.com_backend import ComBackend self.backend = ComBackend() else: raise ValueError(f"Unknown backend: {self.backend_name}") self.artifact_dir = artifact_dir await self.backend.connect() async def cleanup(self) -> None: if self.backend is not None: await self.backend.disconnect() def metadata(self) -> dict: return { "backend": self.backend_name, "capabilities": self.backend.capabilities().to_dict() if self.backend else None, } async def _reset(self) -> None: await self.backend.drawing_new() async def run_task(self, task: TaskSpec) -> TaskResult: await self._reset() handler = getattr(self, f"_task_{task.task_id}", None) if handler is None: return TaskResult(task.task_id, "unsupported", 0.0, "No adapter implementation") passed, metrics, artifacts = await handler() return TaskResult( task.task_id, "pass" if passed else "fail", 100.0 if passed else 0.0, metrics=metrics, artifacts=artifacts, ) async def _task_core_geometry(self): line = await self.backend.entity_create_line(0, 0, 3, 4) circle = await self.backend.entity_create_circle(10, 10, 2) info = await self.backend.entity_get(line.handle) passed = abs(float(info.properties["length"]) - 5.0) < 1e-6 return passed and bool(circle.handle), {"line_length": info.properties["length"]}, [] async def _task_modify_query(self): line = await self.backend.entity_create_line(0, 0, 10, 0) await self.backend.entity_move(line.handle, 5, 2) moved = await self.backend.entity_get(line.handle) return moved.properties["start"][:2] == [5.0, 2.0], {}, [] async def _task_layers_linetypes(self): await ensure_standard_linetypes(self.backend) await ensure_engineering_layers(self.backend) layers = {item.name for item in await self.backend.layer_list()} linetypes = {item.upper() for item in await self.backend.linetype_list()} return "GEOMETRY" in layers and "CENTER" in linetypes, {}, [] async def _task_dimensions(self): await ensure_engineering_layers(self.backend) dimension = await self.backend.dimension_linear(0, 0, 50, 0, 25, 10, layer="DIM") return "DIM" in dimension.type.upper(), {"entity_type": dimension.type}, [] async def _task_table_mleader(self): await ensure_engineering_layers(self.backend) table = await self.backend.entity_create_table( 0, 20, [["A", "1"]], headers=["ITEM", "QTY"], layer="TEXT" ) leader = await self.backend.leader_create_mleader([[0, 0], [10, 10]], "NOTE", layer="DIM") representations = [ table.properties.get("representation"), leader.properties.get("representation"), ] return all(representations), {"representations": representations}, [] async def _task_transactions(self): await self.backend.transaction_begin() await self.backend.entity_create_circle(0, 0, 1) await self.backend.transaction_rollback() entities = await self.backend.entity_list(limit=100) return len(entities) == 0, {"entity_count": len(entities)}, [] async def _task_preflight(self): result = await self.backend.drawing_preflight( "Benchmark plate", { "units": "mm", "part_type": "plate", "dimensions": {"width": 50, "height": 25}, "tolerance_policy": "ISO 2768-m", }, ) return result.ready and result.spec_hash.startswith("sha256:"), {}, [] async def _task_quality_refiner(self): await ensure_engineering_layers(self.backend) await self.backend.entity_create_line(0, 0, 10, 0, layer="GEOMETRY") await self.backend.entity_create_line(0, 0, 10, 0, layer="GEOMETRY") result = await refine_drawing(self.backend, focus=["duplicate_entities"]) return result.final_score > result.initial_score, result.to_dict(), [] async def _task_dxf_roundtrip(self): await self.backend.entity_create_line(0, 0, 10, 10) path = self.artifact_dir / "roundtrip.dxf" await self.backend.drawing_export_dxf(str(path)) from backends.ezdxf_backend import EzdxfBackend reopened = EzdxfBackend() await reopened.connect() try: await reopened.drawing_open(str(path)) count = (await reopened.drawing_info()).entity_count finally: await reopened.disconnect() return count == 1, {"entity_count": count}, [str(path)] # ── v3 (v1.5.0) ───────────────────────────────────────────────────────── async def _task_tool_discovery(self): """Can a drafter find the tool by typing the command they know? Ranked through the real transform, not a stub: the corpus is what makes these commands findable at all, and testing anything else would test the fixture. """ import server from discovery.transform import CadSearchTransform tools = list(await server.mcp._list_tools()) transform = CadSearchTransform() ranks: dict[str, int | None] = {} for command, accepted in DISCOVERY_QUERIES: hits = [ tool.name for tool in transform.rank(tools, command, limit=DISCOVERY_TOP_N).hits ] found = next((index for index, name in enumerate(hits) if name in accepted), None) ranks[command] = None if found is None else found + 1 passed = all(rank is not None for rank in ranks.values()) return passed, {"ranks": ranks, "top_n": DISCOVERY_TOP_N}, [] async def _task_token_budget(self): """What the client pays before it has asked for anything. The ceiling is a constant, so a regression that doubles the catalog fails this rather than quietly reporting a bigger number. """ import server from benchmarks.token_suite import ( FORMAT_JSON_SCHEMA, _client, _wire_payload, build_counter, ) async def _advertised(mode: str) -> tuple[int, int]: """Tokens on the wire for `tools/list`, and how many tools that is. Measured through a real client rather than the registry: the registry is what the transform hides, so counting it would make discovery mode look free. """ server._apply_discovery_mode(mode) async with _client() as client: tools = await client.list_tools() return counter.count(_wire_payload(tools), FORMAT_JSON_SCHEMA), len(tools) counter = build_counter("ratio") try: full, catalog_count = await _advertised("off") discovery, advertised_count = await _advertised("search") finally: server._apply_discovery_mode("off") passed = discovery < TOKEN_CEILING and discovery < full return ( passed, { "catalog_tokens": full, "discovery_tokens": discovery, "ceiling": TOKEN_CEILING, "reduction_x": round(full / discovery, 1) if discovery else None, "catalog_tools": catalog_count, "advertised_tools": advertised_count, # Named, because "40,305 tokens" reads like a count and is not # one. `benchmarks/token_suite.py --tokenizer anthropic` counts # for real; this lane has to stay offline and deterministic. "tokenizer": f"{counter.name} (estimate)" if counter.estimated else counter.name, }, [], ) async def _task_hatch_islands(self): """A section view is mostly holes, so the island is the measurement. 20x20 outer square with a 10x10 island: the filled area is 300, and a server that records the island without subtracting it reports 400 while looking entirely correct on screen. """ await ensure_engineering_layers(self.backend) hatch = await self.backend.entity_create_hatch( "ANSI31", [[0, 0], [20, 0], [20, 20], [0, 20]], layer="HATCH" ) island = [ {"type": "line", "start": [5, 5], "end": [15, 5]}, {"type": "line", "start": [15, 5], "end": [15, 15]}, {"type": "line", "start": [15, 15], "end": [5, 15]}, {"type": "line", "start": [5, 15], "end": [5, 5]}, ] added = await self.backend.hatch_add_boundary(hatch.handle, island) measured = await self.backend.entity_measure(hatch.handle) ignored = await self.backend.hatch_edit(hatch.handle, style="ignore") over_island = await self.backend.entity_measure(hatch.handle) passed = ( added["path_count"] == 2 and abs(measured["area"] - 300.0) < 1e-6 and measured["hatch_style"] == "outer" # what entity_create_hatch writes and ignored["changed"] == ["style"] and abs(over_island["area"] - 400.0) < 1e-6 ) return ( passed, { "path_count": added["path_count"], "filled_area": measured["area"], "area_ignoring_islands": over_island["area"], "expected_filled": 300.0, "hatch_style": measured["hatch_style"], }, [], ) async def _task_selection_filter(self): """Window is not crossing, and a polygon is not its bounding box. Three circles placed so each distinction has exactly one right answer: one wholly inside, one straddling the edge, one out in the corner that the bounding box catches and the triangle does not. """ await ensure_engineering_layers(self.backend) await self.backend.entity_create_circle(5, 5, 1, layer="GEOMETRY") await self.backend.entity_create_circle(10, 5, 3, layer="GEOMETRY") await self.backend.entity_create_circle(18, 18, 0.5, layer="GEOMETRY") await self.backend.entity_create_circle(40, 40, 1, layer="TEXT") window = await self.backend.selection_window(0, 0, 10, 10, mode="window") crossing = await self.backend.selection_window(0, 0, 10, 10, mode="crossing") bbox = await self.backend.selection_window(0, 0, 20, 20, mode="window") triangle = await self.backend.selection_polygon([[0, 0], [20, 0], [0, 20]], mode="window") by_layer = await self.backend.selection_filter(entity_type="CIRCLE", layer="TEXT") passed = ( window["count"] == 1 and crossing["count"] == 2 and bbox["count"] == 3 and triangle["count"] == 1 and by_layer["count"] == 1 and by_layer["filtered_by"] == ["entity_type", "layer"] ) return ( passed, { "window": window["count"], "crossing": crossing["count"], "bounding_box": bbox["count"], "polygon": triangle["count"], "by_layer": by_layer["count"], }, [], ) async def _task_measure_from_handle(self): """The 28.2% that reading coordinates back loses. A 10x10 square whose top edge is a semicircle encloses 139.2699. Chain the loose edges into a boundary, then measure that boundary by handle; shoelacing the four corners the model would have remembered gives 100. """ await ensure_engineering_layers(self.backend) await self.backend.entity_create_line(0, 0, 10, 0, layer="GEOMETRY") await self.backend.entity_create_line(10, 0, 10, 10, layer="GEOMETRY") await self.backend.entity_create_line(0, 10, 0, 0, layer="GEOMETRY") await self.backend.entity_create_arc(5, 10, 5, 0, 180, layer="GEOMETRY") loop = await self.backend.boundary_trace(5.0, 5.0) measured = await self.backend.entity_measure(loop["handle"]) naive = await self.backend.analysis_measure_area([[0, 0], [10, 0], [10, 10], [0, 10]]) true_area = 100.0 + math.pi * 25.0 / 2.0 lost = (true_area - naive) / true_area passed = ( loop["ok"] is True and abs(measured["area"] - true_area) < 1e-5 and abs(loop["area"] - measured["area"]) < 1e-9 and measured["method"] == "analytic_bulge" and measured["exact"] is True and 0.28 < lost < 0.29 ) return ( passed, { "measured_area": measured["area"], "true_area": round(true_area, 6), "vertex_shoelace_area": naive, "fraction_lost_by_reading_points_back": round(lost, 4), "boundary_agrees": abs(loop["area"] - measured["area"]) < 1e-9, }, [], ) async def _task_pid_roundtrip(self): """Five symbols, four lines, read back with nothing dangling. The spec is drawn through the same code ``pid_from_spec`` calls; the graph, the instrument index and the line list are then read back from the drawing by the reader, which never sees the spec. Every expected number below is counted from the spec by hand. """ from engineering.layers import apply_layer_set from engineering.pid.deliverables import instrument_index, line_list from engineering.pid.spec import EXAMPLE_SPEC, run_spec await apply_layer_set(self.backend, "pid") result = await run_spec(self.backend, EXAMPLE_SPEC) graph = result["graph"] index = instrument_index(graph) lines = line_list(graph) # The spec numbers its two process lines; the signal line and the # connector line are left to the default ``number_format``, which with # no size/service/spec/insulation collapses to the bare sequence # number. Both must come back through XDATA, not a label search. 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] sequenced_ok = len(sequenced) == 2 and all( r["line_number"].isdigit() and r["number_source"] == "xdata" for r in sequenced ) passed = ( len(graph["nodes"]) == 5 and len(graph["edges"]) == 4 and graph["stats"]["dangling"] == 0 and graph["stats"]["confidence_min"] == 1.0 and result["critique"] == [] and [r["tag"] for r in index] == ["FIC-101"] and index[0]["connected_to"] == "FCV-101" and len(lines) == 4 and authored_ok and sequenced_ok ) return ( passed, { "nodes": len(graph["nodes"]), "edges": len(graph["edges"]), "dangling": graph["stats"]["dangling"], "confidence_min": graph["stats"]["confidence_min"], "critique_issues": len(result["critique"]), "instrument_index_rows": len(index), "line_list_rows": len(lines), "authored_line_numbers_kept": authored_ok, "unnumbered_lines_sequenced": sequenced_ok, }, [], ) async def _task_auditable_delivery(self): await ensure_engineering_layers(self.backend) await self.backend.entity_create_line(0, 0, 10, 0, layer="GEOMETRY") destination = self.artifact_dir / "delivery" result = await deliver_drawing( self.backend, destination, formats=["dxf"], min_score=0, strict_critique=False, ) return result.status == "success", {"score": result.score}, [result.manifest_path]