`8f1b6b4`

Coerce string patch hunks

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 1 week ago

SHA: 8f1b6b4c7122ce0147d11c647f796cb8c1c60a74
Parents: 107215b
Tree: 5d3e8f2

7 changed files

Status	File	+	-
M	`src/loader/runtime/safeguard_services.py`	5	2
M	`src/loader/tools/file_tools.py`	8	4
M	`src/loader/tools/fs_safety.py`	30	1
M	`src/loader/utils/file_mutations.py`	11	8
M	`tests/test_bash_operator_surfaces.py`	12	0
M	`tests/test_expanded_tools.py`	27	0
M	`tests/test_safeguard_services.py`	25	0

src/loader/runtime/safeguard_services.pymodified

  from difflib import get_close_matches
  from pathlib import Path
 +from ..tools.fs_safety import coerce_structured_patch_payload
++
  TEXT_REWRITE_SUFFIXES = frozenset(
+     {
          ".c",
          if not html_declared_file_result.valid:
              return html_declared_file_result
 -        has_hunks = isinstance(hunks, list) and bool(hunks)
 +        structured_hunks = coerce_structured_patch_payload(hunks)
 +        has_hunks = bool(structured_hunks)
          has_raw_patch = isinstance(raw_patch, str) and bool(raw_patch.strip())
          if not has_hunks and not has_raw_patch:
              return ValidationResult(
          html_placeholder_result = self._validate_html_placeholder_patch(
              str(file_path),
 -            hunks,
 +            structured_hunks,
              raw_patch,
+         )
          if not html_placeholder_result.valid:

src/loader/tools/file_tools.pymodified

  from .fs_safety import (
      StructuredPatchHunk,
      apply_structured_patch,
 +    coerce_structured_patch_payload,
      ensure_safe_to_read,
      ensure_safe_to_write,
      make_structured_patch,
      async def execute(
          self,
          file_path: str,
 -        hunks: list[dict[str, Any]] | None = None,
 +        hunks: list[dict[str, Any]] | dict[str, Any] | str | None = None,
          patch: str | None = None,
          **kwargs: Any,
      ) -> ToolResult:
              original_content = await asyncio.to_thread(path.read_text)
              original_lines = original_content.splitlines()
              raw_patch = patch or kwargs.get("diff") or kwargs.get("patch_text")
 +            structured_hunks = coerce_structured_patch_payload(hunks)
              parsed_hunks: list[StructuredPatchHunk]
 -            if hunks:
 +            if structured_hunks:
                  parsed_hunks = [
 -                    StructuredPatchHunk.from_dict_with_original(
 +                    hunk
 +                    if isinstance(hunk, StructuredPatchHunk)
 +                    else StructuredPatchHunk.from_dict_with_original(
                          hunk,
                          original_lines=original_lines,
+                     )
 -                    for hunk in hunks
 +                    for hunk in structured_hunks
+                 ]
              elif isinstance(raw_patch, str) and raw_patch.strip():
                  parsed_hunks = parse_unified_diff_patch(raw_patch)

src/loader/tools/fs_safety.pymodified

  from __future__ import annotations
 +import json
 +import re
  from dataclasses import asdict, dataclass
  from pathlib import Path
 -import re
  MAX_READ_SIZE = 10 * 1024 * 1024
  MAX_WRITE_SIZE = 10 * 1024 * 1024
+         )
 +def coerce_structured_patch_payload(
 +    value: object,
 +) -> list[dict[str, object] | StructuredPatchHunk]:
 +    """Normalize structured patch payloads from native tool callers.
++
 +    Some local models serialize the `hunks` array as a JSON string even though
 +    the tool schema asks for an array. Treat that as recoverable instead of
 +    rejecting an otherwise valid patch call.
 +    """
++
 +    if isinstance(value, str):
 +        if not value.strip():
 +            return []
 +        try:
 +            value = json.loads(value)
 +        except json.JSONDecodeError:
 +            return []
++
 +    if isinstance(value, StructuredPatchHunk):
 +        return [value]
 +    if isinstance(value, dict):
 +        return [value]
 +    if not isinstance(value, list):
 +        return []
++
 +    return [item for item in value if isinstance(item, (dict, StructuredPatchHunk))]
++
++
  def resolve_workspace_path(
      raw_path: str,
      *,

src/loader/utils/file_mutations.pymodified

  from ..tools.fs_safety import (
      StructuredPatchHunk,
 +    coerce_structured_patch_payload,
      make_structured_patch,
      parse_unified_diff_patch,
+ )
  def _coerce_patch_hunks(value: Any) -> list[StructuredPatchHunk]:
 -    if not isinstance(value, list):
 +    patch_items = coerce_structured_patch_payload(value)
 +    if not patch_items:
          return []
      hunks: list[StructuredPatchHunk] = []
 -    for item in value:
 +    for item in patch_items:
          if isinstance(item, StructuredPatchHunk):
              hunks.append(item)
          elif isinstance(item, dict):
      stats = []
      if preview.added_lines:
 -        stats.append(("+%d" % preview.added_lines, "green"))
 +        stats.append((f"+{preview.added_lines}", "green"))
      if preview.removed_lines:
 -        stats.append(("-%d" % preview.removed_lines, "red"))
 +        stats.append((f"-{preview.removed_lines}", "red"))
      if preview.context_lines:
 -        stats.append(("%d context" % preview.context_lines, "dim"))
 -    stats.append(("%d hunk%s" % (preview.hunk_count, "" if preview.hunk_count == 1 else "s"), "dim"))
 +        stats.append((f"{preview.context_lines} context", "dim"))
 +    hunk_suffix = "" if preview.hunk_count == 1 else "s"
 +    stats.append((f"{preview.hunk_count} hunk{hunk_suffix}", "dim"))
      text.append("\n")
      for index, (label, style) in enumerate(stats):
      for hunk in hunks:
          rendered.append(
              Text(
 -                "@@ -%d,%d +%d,%d @@\n"
 -                % (hunk.old_start, hunk.old_lines, hunk.new_start, hunk.new_lines),
 +                f"@@ -{hunk.old_start},{hunk.old_lines} "
 +                f"+{hunk.new_start},{hunk.new_lines} @@\n",
                  style="dim",
+             )
+         )

tests/test_bash_operator_surfaces.pymodified

  from __future__ import annotations
 +import json
  from types import SimpleNamespace
  import pytest
      assert preview.structured_patch[0].lines[0].startswith("+  <svg")
 +def test_build_file_mutation_preview_accepts_json_encoded_hunks() -> None:
 +    tool_args = _replacement_block_patch_tool_args()
 +    tool_args["hunks"] = json.dumps(tool_args["hunks"])
++
 +    preview = build_file_mutation_preview("patch", tool_args=tool_args)
++
 +    assert preview is not None
 +    assert preview.structured_patch[0].old_start == 42
 +    assert preview.structured_patch[0].new_lines == 5
++
++
  def test_cli_print_tool_call_renders_bash_panel_without_truncating(monkeypatch: pytest.MonkeyPatch) -> None:
      console = Console(record=True, width=120)
      monkeypatch.setattr(cli_main_module, "console", console)

tests/test_expanded_tools.pymodified

      assert result.metadata["structured_patch"]
 +@pytest.mark.asyncio
 +async def test_patch_tool_accepts_json_encoded_structured_hunks(
 +    temp_dir: Path,
 +) -> None:
 +    target = temp_dir / "sample.txt"
 +    target.write_text("alpha\nbeta\ngamma\n")
 +    tool = PatchTool(workspace_root=temp_dir)
++
 +    result = await tool.execute(
 +        file_path=str(target),
 +        hunks=json.dumps(
 +            [
 +                {
 +                    "old_start": 2,
 +                    "old_lines": 1,
 +                    "new_start": 2,
 +                    "new_lines": 1,
 +                    "lines": ["-beta", "+beta from json string"],
 +                }
 +            ]
 +        ),
 +    )
++
 +    assert result.is_error is False
 +    assert target.read_text() == "alpha\nbeta from json string\ngamma\n"
++
++
  @pytest.mark.asyncio
  async def test_patch_tool_rejects_context_mismatch(temp_dir: Path) -> None:
      target = temp_dir / "sample.txt"

tests/test_safeguard_services.pymodified

  from __future__ import annotations
 +import json
  import tempfile
  from pathlib import Path
      assert result == ValidationResult(valid=True)
 +def test_pre_action_validator_allows_json_encoded_patch_hunks() -> None:
 +    validator = PreActionValidator()
++
 +    result = validator.validate(
 +        "patch",
 +        {
 +            "file_path": "notes.txt",
 +            "hunks": json.dumps(
 +                [
 +                    {
 +                        "old_start": 1,
 +                        "old_lines": 1,
 +                        "new_start": 1,
 +                        "new_lines": 1,
 +                        "lines": ["-old", "+new"],
 +                    }
 +                ]
 +            ),
 +        },
 +    )
++
 +    assert result == ValidationResult(valid=True)
++
++
  def test_pre_action_validator_blocks_placeholder_html_write(tmp_path: Path) -> None:
      validator = PreActionValidator()