Address Will's feedback

Author: clean6378-max-it

scripts/gen_tool_types_manifest.py

@@ -8,52 +8,27 @@
 
 from __future__ import annotations
 
-import json
-import re
+import sys
 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 utils.tool_types_manifest_io import (  # noqa: E402
+    load_known_tool_types_from_dispatch,
+    serialize_tool_types_manifest,
+)
 
-def _load_known_tool_types(repo_root: Path) -> frozenset[str]:
-    path = repo_root / "utils" / "tool_dispatch.py"
-    text = path.read_text(encoding="utf-8")
-    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)
+_MANIFEST_PATH = _REPO_ROOT / "static" / "tool_types.json"
 
 
 def write_tool_types_manifest(path: Path | None = None, *, repo_root: Path | None = None) -> int:
     root = (repo_root or _REPO_ROOT).resolve()
     dest = path or root / "static" / "tool_types.json"
-    known = _load_known_tool_types(root)
-    payload = {"tool_types": sorted(known)}
-    dest.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
+    dispatch_path = root / "utils" / "tool_dispatch.py"
+    known = load_known_tool_types_from_dispatch(dispatch_path)
+    dest.write_text(serialize_tool_types_manifest(known), encoding="utf-8")
     return len(known)
 
 

scripts/scaffold_tool_type.py

@@ -34,6 +34,10 @@
     js_render_result_name,
     js_render_use_name,
 )
+from utils.tool_types_manifest_io import (  # noqa: E402
+    parse_file_activity_handler_names,
+    serialize_tool_types_manifest,
+)
 
 _FILE_ACTIVITY_HANDLER = {
     "none": "None",
@@ -125,7 +129,7 @@ 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")
-        known = _parse_handlers_from_text(dispatch_text)
+        known = parse_file_activity_handler_names(dispatch_text)
         if record.name in known:
             msg = f"tool type {record.name!r} already present in _FILE_ACTIVITY_HANDLERS"
             raise ValueError(msg)
@@ -439,9 +443,8 @@ def _plan_manifest(self, plan: _EmitPlan) -> None:
         if self.paths.tool_dispatch not in plan.pending:
             msg = "internal error: tool_dispatch was not staged before manifest"
             raise ValueError(msg)
-        known = _parse_handlers_from_text(plan.pending[self.paths.tool_dispatch])
-        payload = {"tool_types": sorted(known)}
-        plan.stage(self.paths.tool_types_manifest, json.dumps(payload, indent=2) + "\n")
+        known = parse_file_activity_handler_names(plan.pending[self.paths.tool_dispatch])
+        plan.stage(self.paths.tool_types_manifest, serialize_tool_types_manifest(known))
 
     def _plan_record(self, record: ToolTypeRecord, plan: _EmitPlan) -> None:
         path = self.paths.records_dir / f"{record.snake_name}.json"
@@ -485,36 +488,6 @@ def _append_to_block(text: str, block_start: str, entry: str) -> str:
         return text[:close_pos] + "\n" + entry + text[close_pos:]
 
 
-def _parse_handlers_from_text(text: str) -> frozenset[str]:
-    marker = "_FILE_ACTIVITY_HANDLERS: dict"
-    start = text.find(marker)
-    if start == -1:
-        msg = f"could not find {marker} in staged tool_dispatch.py"
-        raise ValueError(msg)
-    brace_start = text.find("{", start)
-    if brace_start == -1:
-        msg = "could not find opening brace for _FILE_ACTIVITY_HANDLERS"
-        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 = "no tool names found in staged _FILE_ACTIVITY_HANDLERS"
-                    raise ValueError(msg)
-                return frozenset(keys)
-        i += 1
-    msg = "unbalanced braces in staged _FILE_ACTIVITY_HANDLERS"
-    raise ValueError(msg)
-
-
 def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument(
@@ -571,39 +544,42 @@ def _resolve_record(args: argparse.Namespace) -> ToolTypeRecord:
 
 
 def main(argv: list[str] | None = None) -> int:
-    args = _parse_args(argv)
-    record = _resolve_record(args)
-    root = args.root.resolve()
-
-    if args.write_record_only:
-        out = args.record or root / "tool_types" / f"{record.snake_name}.json"
-        if args.dry_run:
-            print(f"Would write record to {out}")
+    try:
+        args = _parse_args(argv)
+        record = _resolve_record(args)
+        root = args.root.resolve()
+
+        if args.write_record_only:
+            out = args.record or root / "tool_types" / f"{record.snake_name}.json"
+            if args.dry_run:
+                print(f"Would write record to {out}")
+                return 0
+            record.save(out)
+            print(f"Wrote {out}")
             return 0
-        record.save(out)
-        print(f"Wrote {out}")
-        return 0
 
-    emitter = ScaffoldEmitter(root, dry_run=args.dry_run)
-    try:
+        emitter = ScaffoldEmitter(root, dry_run=args.dry_run)
         written = emitter.emit(record)
+
+        if args.dry_run:
+            print(f"Dry run complete for {record.name} ({len(written)} artifacts planned)")
+        else:
+            print(f"Scaffolded {record.name}: {len(written)} files updated")
+            print("Next: complete TODO stubs, then run:")
+            print(
+                "  pytest tests/test_scaffold_tool_type.py tests/test_tool_dispatch_sync.py "
+                "tests/test_tool_dispatch_ordering.py tests/test_tool_dispatch_adversarial.py "
+                "tests/test_jsonl_parser.py tests/test_real_session_fixtures.py -q"
+            )
+            print("  npm test")
+        return 0
+    except json.JSONDecodeError as exc:
+        print(f"scaffold_tool_type: invalid JSON: {exc}", file=sys.stderr)
+        return 1
     except ValueError as exc:
         print(f"scaffold_tool_type: {exc}", file=sys.stderr)
         return 1
 
-    if args.dry_run:
-        print(f"Dry run complete for {record.name} ({len(written)} artifacts planned)")
-    else:
-        print(f"Scaffolded {record.name}: {len(written)} files updated")
-        print("Next: complete TODO stubs, then run:")
-        print(
-            "  pytest tests/test_scaffold_tool_type.py tests/test_tool_dispatch_sync.py "
-            "tests/test_tool_dispatch_ordering.py tests/test_tool_dispatch_adversarial.py "
-            "tests/test_jsonl_parser.py tests/test_real_session_fixtures.py -q"
-        )
-        print("  npm test")
-    return 0
-
 
 if __name__ == "__main__":
     raise SystemExit(main())

tests/test_scaffold_tool_type.py

@@ -95,8 +95,32 @@ def test_overlap_invariant_resolves_explicit_guards() -> None:
                     {
                         "before_dispatch_id": "plan",
                         "after_dispatch_id": "file_write",
-                        "before_guard": "is_plan_tool_result",
-                        "after_guard": "is_file_write_tool_result",
+                        "before_guard": "is_custom_plan_guard",
+                        "after_guard": "is_custom_write_guard",
+                        "reason": "test overlap",
+                        "fixture_id": "overlap_tool_fixture",
+                        "overlap_blob": {"plan": [], "filePath": "x", "content": "y"},
+                    }
+                ],
+            },
+        }
+    )
+    assert record.result is not None
+    inv = record.result.overlap_invariants[0]
+    assert inv.resolved_before_guard() == "is_custom_plan_guard"
+    assert inv.resolved_after_guard() == "is_custom_write_guard"
+
+
+def test_overlap_invariant_falls_back_to_dispatch_id_guards() -> None:
+    record = ToolTypeRecord.from_mapping(
+        {
+            "name": "OverlapTool",
+            "result": {
+                "dispatch_id": "overlap_tool",
+                "overlap_invariants": [
+                    {
+                        "before_dispatch_id": "plan",
+                        "after_dispatch_id": "file_write",
                         "reason": "test overlap",
                         "fixture_id": "overlap_tool_fixture",
                         "overlap_blob": {"plan": [], "filePath": "x", "content": "y"},
@@ -232,6 +256,26 @@ def test_dry_run_main_exits_zero(capsys: pytest.CaptureFixture[str]) -> None:
     assert "would write" in out
 
 
+def test_main_rejects_invalid_cli_name(capsys: pytest.CaptureFixture[str]) -> None:
+    code = main(["--name", "NotSnake", "--dry-run"])
+    assert code == 1
+    err = capsys.readouterr().err
+    assert "scaffold_tool_type:" in err
+    assert "invalid snake_case name" in err
+
+
+def test_main_rejects_invalid_record_json(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    bad_record = tmp_path / "bad.json"
+    bad_record.write_text("{not json", encoding="utf-8")
+    code = main(["--record", str(bad_record), "--dry-run"])
+    assert code == 1
+    err = capsys.readouterr().err
+    assert "scaffold_tool_type:" in err
+    assert "invalid JSON" in err
+
+
 def test_scaffold_rejects_duplicate_name(tmp_path: Path) -> None:
     repo = _mirror_repo(tmp_path)
     record = ToolTypeRecord.from_cli_name("example_tool")

utils/tool_types_manifest_io.py

@@ -0,0 +1,51 @@
+"""Parse and serialize ``static/tool_types.json`` from ``tool_dispatch.py`` source."""
+
+from __future__ import annotations
+
+import json
+import re
+from pathlib import Path
+
+_FILE_ACTIVITY_HANDLERS_MARKER = "_FILE_ACTIVITY_HANDLERS: dict"
+
+
+def parse_file_activity_handler_names(
+    text: str, *, source: str = "tool_dispatch.py"
+) -> frozenset[str]:
+    """Return tool names listed in a ``_FILE_ACTIVITY_HANDLERS`` dict literal."""
+    start = text.find(_FILE_ACTIVITY_HANDLERS_MARKER)
+    if start == -1:
+        msg = f"could not find {_FILE_ACTIVITY_HANDLERS_MARKER} in {source}"
+        raise ValueError(msg)
+    brace_start = text.find("{", start)
+    if brace_start == -1:
+        msg = f"could not find opening brace for {_FILE_ACTIVITY_HANDLERS_MARKER} in {source}"
+        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 {_FILE_ACTIVITY_HANDLERS_MARKER} in {source}"
+                    raise ValueError(msg)
+                return frozenset(keys)
+        i += 1
+    msg = f"unbalanced braces in {_FILE_ACTIVITY_HANDLERS_MARKER} in {source}"
+    raise ValueError(msg)
+
+
+def load_known_tool_types_from_dispatch(path: Path) -> frozenset[str]:
+    text = path.read_text(encoding="utf-8")
+    return parse_file_activity_handler_names(text, source=str(path))
+
+
+def serialize_tool_types_manifest(known: frozenset[str]) -> str:
+    payload = {"tool_types": sorted(known)}
+    return json.dumps(payload, indent=2) + "\n"