fix: harden scaffold generator for repeated tool scaffolds

Author: clean6378-max-it

models/tool_type_registry.py

@@ -32,7 +32,7 @@
     }
 )
 
-_DISPATCH_ID_RE = re.compile(r"[a-z][a-z0-9_]*")
+_DISPATCH_ID_RE = re.compile(r"[a-z]([a-z0-9]+(_[a-z0-9]+)*)?")
 
 
 @dataclass(frozen=True, slots=True)
@@ -48,6 +48,9 @@ def from_mapping(cls, raw: dict[str, Any]) -> TypedDictField:
         if not isinstance(name, str) or not name:
             msg = "typed_dict_fields entries require a non-empty string 'name'"
             raise ValueError(msg)
+        if not name.isidentifier():
+            msg = f"typed_dict_fields 'name' must be a valid Python identifier, got {name!r}"
+            raise ValueError(msg)
         py_type = raw.get("type", "object")
         if not isinstance(py_type, str):
             msg = "typed_dict_fields 'type' must be a string"
@@ -144,7 +147,7 @@ def from_mapping(cls, raw: dict[str, Any] | None) -> ToolResultRecord | None:
             msg = "result.predicate_mode must be 'all' or 'any'"
             raise ValueError(msg)
         priority = raw.get("priority", 0)
-        if not isinstance(priority, int):
+        if isinstance(priority, bool) or not isinstance(priority, int):
             msg = "result.priority must be an int"
             raise ValueError(msg)
         if priority < 0:
@@ -301,7 +304,7 @@ def builder_name(self) -> str:
 
 
 def snake_to_pascal(snake: str) -> str:
-    if not re.fullmatch(r"[a-z][a-z0-9_]*", snake):
+    if not re.fullmatch(r"[a-z]([a-z0-9]+(_[a-z0-9]+)*)?", snake):
         msg = f"invalid snake_case name: {snake!r}"
         raise ValueError(msg)
     return "".join(part.capitalize() for part in snake.split("_"))
@@ -357,6 +360,4 @@ def _validate_dispatch_id(dispatch_id: str) -> None:
 
 def _dispatch_id_to_camel(dispatch_id: str) -> str:
     parts = dispatch_id.split("_")
-    if not parts:
-        return "value"
     return parts[0] + "".join(part.capitalize() for part in parts[1:])

scripts/gen_tool_types_manifest.py

@@ -9,22 +9,43 @@
 from __future__ import annotations
 
 import json
-import sys
+import re
 from pathlib import Path
 
 _REPO_ROOT = Path(__file__).resolve().parents[1]
 _MANIFEST_PATH = _REPO_ROOT / "static" / "tool_types.json"
 
-if str(_REPO_ROOT) not in sys.path:
-    sys.path.insert(0, str(_REPO_ROOT))
-
-from scripts.scaffold_tool_type import _parse_handlers_from_text  # noqa: E402
-
 
 def _load_known_tool_types(repo_root: Path) -> frozenset[str]:
     path = repo_root / "utils" / "tool_dispatch.py"
     text = path.read_text(encoding="utf-8")
-    return _parse_handlers_from_text(text)
+    marker = "_FILE_ACTIVITY_HANDLERS: dict"
+    start = text.find(marker)
+    if start == -1:
+        msg = f"could not find {marker} in {path}"
+        raise ValueError(msg)
+    brace_start = text.find("{", start)
+    if brace_start == -1:
+        msg = f"could not find opening brace for {marker} in {path}"
+        raise ValueError(msg)
+    depth = 0
+    i = brace_start
+    while i < len(text):
+        ch = text[i]
+        if ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+            if depth == 0:
+                body = text[brace_start + 1 : i]
+                keys = re.findall(r'"([^"]+)":', body)
+                if not keys:
+                    msg = f"no tool names found in {marker} in {path}"
+                    raise ValueError(msg)
+                return frozenset(keys)
+        i += 1
+    msg = f"unbalanced braces in {marker} in {path}"
+    raise ValueError(msg)
 
 
 def write_tool_types_manifest(path: Path | None = None, *, repo_root: Path | None = None) -> int:

scripts/scaffold_tool_type.py

@@ -107,6 +107,7 @@ def __init__(self, root: Path, *, dry_run: bool = False, stdout: TextIO | None =
 
     def emit(self, record: ToolTypeRecord) -> list[Path]:
         self._guard_not_present(record)
+        self._written = []
         plan = _EmitPlan()
         self._plan_tool_results(record, plan)
         self._plan_tool_dispatch(record, plan)
@@ -124,7 +125,8 @@ def emit(self, record: ToolTypeRecord) -> list[Path]:
 
     def _guard_not_present(self, record: ToolTypeRecord) -> None:
         dispatch_text = self.paths.tool_dispatch.read_text(encoding="utf-8")
-        if f'"{record.name}"' in dispatch_text and f'"{record.name}":' in dispatch_text:
+        known = _parse_handlers_from_text(dispatch_text)
+        if record.name in known:
             msg = f"tool type {record.name!r} already present in _FILE_ACTIVITY_HANDLERS"
             raise ValueError(msg)
 
@@ -148,7 +150,7 @@ def transform(text: str) -> str:
                 text = self._insert_before(
                     text,
                     "# Registration order is tie-break only when priorities are equal.",
-                    self._render_dispatch_builder(record) + "\n",
+                    self._render_dispatch_builder(record) + "\n\n\n",
                 )
                 text = self._insert_before(
                     text,
@@ -158,10 +160,10 @@ def transform(text: str) -> str:
                 )
                 guard = record.guard_name
                 if f"    {guard}," not in text:
-                    text = self._replace_once(
+                    text = self._append_to_block(
                         text,
-                        "    is_web_search_tool_result,\n)",
-                        f"    is_web_search_tool_result,\n    {guard},\n)",
+                        "from models.tool_results import (",
+                        f"    {guard},",
                     )
             entry = f'    "{record.name}": {handler},\n'
             text = self._insert_before(text, "}\nKNOWN_TOOL_TYPES", entry, marker_in_prev_line=True)
@@ -173,7 +175,6 @@ def _render_dispatch_builder(self, record: ToolTypeRecord) -> str:
         assert record.result is not None
         sig = "tr: ToolResultDict, base: dict[str, object]"
         lines = [
-            "",
             f"def {record.builder_name}({sig}) -> dict[str, object]:",
             "    # TODO: map toolUseResult fields to parsed result keys.",
             "    result = dict(base)",
@@ -208,25 +209,25 @@ def transform(text: str) -> str:
                 text = self._insert_before(
                     text,
                     "# Dict passed into dispatch predicates",
-                    self._render_typed_dict(record) + "\n\n",
+                    self._render_typed_dict(record) + "\n\n\n",
                 )
                 member = f"    | {record.typed_dict_class}\n"
                 text = self._insert_before(
                     text,
                     "    | ToolResultWithContentDict",
                     member,
                 )
-                anchor = '    return "questions" in tr and "answers" in tr\n\n\n'
-                text = self._replace_once(
+                text = self._insert_before(
                     text,
-                    anchor + "# Tool names on assistant",
-                    anchor + f"{self._render_guard(record)}\n\n\n# Tool names on assistant",
+                    "# Tool names on assistant",
+                    self._render_guard(record) + "\n\n\n",
+                )
+            if f'    "{record.name}",' not in text:
+                text = self._append_to_block(
+                    text,
+                    "ToolNameLiteral = Literal[",
+                    f'    "{record.name}",',
                 )
-            text = self._replace_once(
-                text,
-                '    "WebSearch",\n]',
-                f'    "WebSearch",\n    "{record.name}",\n]',
-            )
             return text
 
         plan.stage_patch(self.paths.tool_results, transform)
@@ -375,25 +376,18 @@ def transform(text: str) -> str:
             for inv in record.result.overlap_invariants:
                 before_guard = inv.resolved_before_guard()
                 after_guard = inv.resolved_after_guard()
-                if before_guard not in text:
-                    text = self._replace_once(
+                if f"    {before_guard}," not in text:
+                    text = self._append_to_block(
+                        text,
+                        "from models.tool_results import (",
+                        f"    {before_guard},",
+                    )
+                if f"    {after_guard}," not in text:
+                    text = self._append_to_block(
                         text,
-                        "    is_task_async_tool_result,\n)",
-                        f"    is_task_async_tool_result,\n    {before_guard},\n)",
+                        "from models.tool_results import (",
+                        f"    {after_guard},",
                     )
-                if after_guard not in text:
-                    if before_guard in text:
-                        text = self._replace_once(
-                            text,
-                            f"    {before_guard},\n)",
-                            f"    {before_guard},\n    {after_guard},\n)",
-                        )
-                    else:
-                        text = self._replace_once(
-                            text,
-                            "    is_task_async_tool_result,\n)",
-                            f"    is_task_async_tool_result,\n    {after_guard},\n)",
-                        )
                 row = (
                     f"    (\n"
                     f"        {before_guard},\n"
@@ -477,6 +471,19 @@ def _replace_once(text: str, anchor: str, replacement: str) -> str:
             raise ValueError(msg)
         return text.replace(anchor, replacement, 1)
 
+    @staticmethod
+    def _append_to_block(text: str, block_start: str, entry: str) -> str:
+        """Append *entry* before the closing ``)`` or ``]`` of *block_start*."""
+        idx = text.find(block_start)
+        if idx == -1:
+            raise ValueError(f"scaffold block not found: {block_start!r}")
+        search_from = idx + len(block_start)
+        match = re.search(r"\n[)\]]", text[search_from:])
+        if not match:
+            raise ValueError(f"closing delimiter not found for block: {block_start!r}")
+        close_pos = search_from + match.start()
+        return text[:close_pos] + "\n" + entry + text[close_pos:]
+
 
 def _parse_handlers_from_text(text: str) -> frozenset[str]:
     marker = "_FILE_ACTIVITY_HANDLERS: dict"

tests/test_scaffold_tool_type.py

@@ -75,6 +75,16 @@ def test_from_mapping_rejects_invalid_dispatch_id() -> None:
         )
 
 
+def test_from_mapping_rejects_trailing_underscore_dispatch_id() -> None:
+    with pytest.raises(ValueError, match="snake_case"):
+        ToolTypeRecord.from_mapping(
+            {
+                "name": "BadTool",
+                "result": {"dispatch_id": "bad_"},
+            }
+        )
+
+
 def test_overlap_invariant_resolves_explicit_guards() -> None:
     record = ToolTypeRecord.from_mapping(
         {
@@ -127,6 +137,47 @@ def test_from_mapping_rejects_negative_priority() -> None:
         )
 
 
+def test_from_mapping_rejects_bool_priority() -> None:
+    with pytest.raises(ValueError, match="must be an int"):
+        ToolTypeRecord.from_mapping(
+            {
+                "name": "BadTool",
+                "result": {"dispatch_id": "bad_tool", "priority": True},
+            }
+        )
+
+
+def test_from_mapping_rejects_non_identifier_field_name() -> None:
+    with pytest.raises(ValueError, match="valid Python identifier"):
+        ToolTypeRecord.from_mapping(
+            {
+                "name": "BadTool",
+                "result": {
+                    "dispatch_id": "bad_tool",
+                    "typed_dict_fields": [{"name": "not valid", "type": "str"}],
+                },
+            }
+        )
+
+
+def test_append_to_block_inserts_before_closing_paren() -> None:
+    text = "from foo import (\n    bar,\n)\n"
+    result = ScaffoldEmitter._append_to_block(text, "from foo import (", "    baz,")
+    assert "    baz,\n)" in result
+    assert result.index("bar,") < result.index("baz,")
+
+
+def test_append_to_block_raises_on_missing_block() -> None:
+    with pytest.raises(ValueError, match="block not found"):
+        ScaffoldEmitter._append_to_block("some text", "nonexistent(", "    x,")
+
+
+def test_append_to_block_raises_on_missing_close() -> None:
+    text = "from foo import (bar)"
+    with pytest.raises(ValueError, match="closing delimiter not found"):
+        ScaffoldEmitter._append_to_block(text, "from foo import (", "    baz,")
+
+
 def test_dry_run_reports_planned_artifact_count(capsys: pytest.CaptureFixture[str]) -> None:
     code = main(["--name", "example_tool", "--dry-run"])
     assert code == 0
@@ -161,14 +212,14 @@ def test_emit_anchor_miss_in_tool_results_leaves_repo_unchanged(tmp_path: Path)
     tool_results = repo / "models" / "tool_results.py"
     original = tool_results.read_text(encoding="utf-8")
     corrupt = original.replace(
-        '    "WebSearch",\n]',
-        '    "WebSearch",\n    "Legacy",\n]',
+        "# Dict passed into dispatch predicates",
+        "# Dict handed to dispatch predicates",
     )
     assert corrupt != original, "fixture corruption must change tool_results.py"
     tool_results.write_text(corrupt, encoding="utf-8")
     snapshot = _snapshot_emitter_targets(repo)
     record = ToolTypeRecord.from_cli_name("example_tool")
-    with pytest.raises(ValueError, match="replacement anchor not found"):
+    with pytest.raises(ValueError, match="marker not found"):
         ScaffoldEmitter(repo).emit(record)
     _assert_emitter_targets_unchanged(repo, snapshot)
 
@@ -246,6 +297,13 @@ def test_scaffold_emits_declared_priority(tmp_path: Path) -> None:
     assert "priority=2," in dispatch
 
 
+def test_scaffold_two_result_tools_passes_dispatch_sync(tmp_path: Path) -> None:
+    repo = _mirror_repo(tmp_path)
+    ScaffoldEmitter(repo).emit(ToolTypeRecord.from_cli_name("alpha_tool"))
+    ScaffoldEmitter(repo).emit(ToolTypeRecord.from_cli_name("beta_tool"))
+    _assert_dispatch_sync(repo)
+
+
 def test_scaffold_in_temp_passes_dispatch_sync(tmp_path: Path) -> None:
     repo = _mirror_repo(tmp_path)
     record = ToolTypeRecord.from_cli_name("example_tool")