sneakygeek/SneakyCode
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Phase 6: write tools, shell, and edit tools with reasoning-only fix

Browse Source 
Implement 6 new agent tools — write_file, make_dir, delete_file,
str_replace, patch_apply, run_command — bringing the agent from
read-only observer to active code modifier. All write/shell operations
are gated through the existing permissions service.

Also fix a bug where qwen3.5 thinking mode produces reasoning tokens
but no content after tool results, causing the agent to silently exit.
The loop now detects reasoning-only responses, retries twice, then
injects a nudge message to break the model out of its thinking loop.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Phillip Tarrant
2026-03-11 09:45:48 -05:00
parent 0cf0d01657
commit f60c47a85f
12 changed files with 956 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

208
app/tools/edit.py Normal file
View File

@@ -0,0 +1,208 @@
"""Edit tools: str_replace and patch_apply."""
import subprocess
from pathlib import Path
from typing import Any
from pydantic import BaseModel, Field
from app.models.tool_call import ToolResult, ToolResultStatus
from app.tools.base import BaseTool
from app.utils.file_helpers import (
FileSizeError,
PathSecurityError,
resolve_safe_path,
safe_read_file,
safe_write_file,
)
_PATCH_TIMEOUT = 30
class StrReplaceParams(BaseModel):
"""Parameters for the str_replace tool."""
file_path: str = Field(description="Path to the file to edit (relative to workspace root)")
old_str: str = Field(description="The exact string to find and replace (must be unique in file)")
new_str: str = Field(description="The replacement string")
class StrReplaceTool(BaseTool):
"""Replace a unique string occurrence in a file."""
name = "str_replace"
description = (
"Replace exactly one occurrence of old_str with new_str in a file. "
"Fails if old_str is not found or appears more than once."
)
params_model = StrReplaceParams
def execute(
self, *, tool_call_id: str, file_path: str, old_str: str, new_str: str, **kwargs: Any
) -> ToolResult:
fs_config = self.config.tools.filesystem
# Read the file
try:
content = safe_read_file(
file_path,
self.workspace_root,
max_size_bytes=fs_config.max_file_size_bytes,
check_binary=fs_config.binary_detection,
)
except PathSecurityError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
except FileNotFoundError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
except FileSizeError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
# Count occurrences
count = content.count(old_str)
if count == 0:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"old_str not found in {file_path}",
)
if count > 1:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"old_str appears {count} times in {file_path} (must be unique)",
)
# Perform replacement and write back
new_content = content.replace(old_str, new_str, 1)
try:
safe_write_file(
file_path,
new_content,
self.workspace_root,
max_size_bytes=fs_config.max_file_size_bytes,
)
except PathSecurityError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
except FileSizeError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
try:
safe_path = resolve_safe_path(file_path, self.workspace_root)
rel_path = safe_path.relative_to(self.workspace_root)
except (PathSecurityError, ValueError):
rel_path = Path(file_path)
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=f"Replaced 1 occurrence in {rel_path}",
)
class PatchApplyParams(BaseModel):
"""Parameters for the patch_apply tool."""
file_path: str = Field(description="Path to the file to patch (relative to workspace root)")
patch: str = Field(description="Unified diff format patch to apply")
class PatchApplyTool(BaseTool):
"""Apply a unified diff patch to a file."""
name = "patch_apply"
description = (
"Apply a unified diff (patch) to a file. The patch must be in standard "
"unified diff format."
)
params_model = PatchApplyParams
def execute(self, *, tool_call_id: str, file_path: str, patch: str, **kwargs: Any) -> ToolResult:
try:
safe_path = resolve_safe_path(file_path, self.workspace_root)
except PathSecurityError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
if not safe_path.exists():
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"File not found: {safe_path}",
)
try:
result = subprocess.run(
["patch", "--forward", "--no-backup-if-mismatch", str(safe_path)],
input=patch,
capture_output=True,
text=True,
timeout=_PATCH_TIMEOUT,
cwd=self.workspace_root,
)
except subprocess.TimeoutExpired:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Patch timed out after {_PATCH_TIMEOUT}s",
)
except FileNotFoundError:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error="'patch' command not found on system",
)
if result.returncode != 0:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Patch failed (exit {result.returncode}): {result.stderr or result.stdout}",
)
try:
rel_path = safe_path.relative_to(self.workspace_root)
except ValueError:
rel_path = safe_path
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=f"Patch applied to {rel_path}",
)

175
app/tools/filesystem.py
View File

@@ -1,4 +1,4 @@
"""Filesystem tools: read_file and list_dir."""
"""Filesystem tools: read_file, list_dir, write_file, make_dir, delete_file."""
from pathlib import Path
from typing import Any
@@ -13,6 +13,7 @@ from app.utils.file_helpers import (
PathSecurityError,
resolve_safe_path,
safe_read_file,
safe_write_file,
)
@@ -147,3 +148,175 @@ class ListDirTool(BaseTool):
status=ToolResultStatus.SUCCESS,
output="\n".join(lines),
)
class WriteFileParams(BaseModel):
"""Parameters for the write_file tool."""
file_path: str = Field(description="Path to the file to write (relative to workspace root)")
content: str = Field(description="Content to write to the file")
class WriteFileTool(BaseTool):
"""Write content to a file within the workspace."""
name = "write_file"
description = (
"Write text content to a file. Creates parent directories if needed. "
"Overwrites existing file content."
)
params_model = WriteFileParams
def execute(self, *, tool_call_id: str, file_path: str, content: str, **kwargs: Any) -> ToolResult:
fs_config = self.config.tools.filesystem
try:
safe_path = safe_write_file(
file_path,
content,
self.workspace_root,
max_size_bytes=fs_config.max_file_size_bytes,
)
except PathSecurityError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
except FileSizeError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
try:
rel_path = safe_path.relative_to(self.workspace_root)
except ValueError:
rel_path = safe_path
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=f"Wrote {len(content)} characters to {rel_path}",
)
class MakeDirParams(BaseModel):
"""Parameters for the make_dir tool."""
directory_path: str = Field(description="Path to the directory to create (relative to workspace root)")
class MakeDirTool(BaseTool):
"""Create a directory (and any missing parents) within the workspace."""
name = "make_dir"
description = "Create a directory and any necessary parent directories."
params_model = MakeDirParams
def execute(self, *, tool_call_id: str, directory_path: str, **kwargs: Any) -> ToolResult:
try:
safe_path = resolve_safe_path(directory_path, self.workspace_root)
except PathSecurityError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
if safe_path.exists() and not safe_path.is_dir():
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Path exists and is not a directory: {safe_path}",
)
try:
safe_path.mkdir(parents=True, exist_ok=True)
except OSError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Failed to create directory: {exc}",
)
try:
rel_path = safe_path.relative_to(self.workspace_root)
except ValueError:
rel_path = safe_path
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=f"Created directory: {rel_path}",
)
class DeleteFileParams(BaseModel):
"""Parameters for the delete_file tool."""
file_path: str = Field(description="Path to the file to delete (relative to workspace root)")
class DeleteFileTool(BaseTool):
"""Delete a file within the workspace. Refuses to delete directories."""
name = "delete_file"
description = "Delete a single file. Does not delete directories."
params_model = DeleteFileParams
def execute(self, *, tool_call_id: str, file_path: str, **kwargs: Any) -> ToolResult:
try:
safe_path = resolve_safe_path(file_path, self.workspace_root)
except PathSecurityError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=str(exc),
)
if safe_path.is_dir():
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Path is a directory, not a file: {safe_path}",
)
if not safe_path.exists():
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"File not found: {safe_path}",
)
try:
safe_path.unlink()
except OSError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Failed to delete file: {exc}",
)
try:
rel_path = safe_path.relative_to(self.workspace_root)
except ValueError:
rel_path = safe_path
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=f"Deleted: {rel_path}",
)

33
app/tools/registry.py
View File

@@ -38,14 +38,47 @@ class ToolRegistry:
def create_default_registry(workspace_root: Path, config: AppConfig) -> ToolRegistry:
"""Create a ToolRegistry populated with all built-in tools."""
# Read tools
from app.tools.filesystem import ListDirTool, ReadFileTool
# Write tools
from app.tools.filesystem import DeleteFileTool, MakeDirTool, WriteFileTool
# Edit tools
from app.tools.edit import PatchApplyTool, StrReplaceTool
# Shell tools
from app.tools.shell import RunCommandTool
# Control flow
from app.tools.finish import FinishTool
# Search tools
from app.tools.search import FindFilesTool, GrepFilesTool
registry = ToolRegistry()
# Read
registry.register(ReadFileTool(workspace_root, config))
registry.register(ListDirTool(workspace_root, config))
# Search
registry.register(GrepFilesTool(workspace_root, config))
registry.register(FindFilesTool(workspace_root, config))
# Write
registry.register(WriteFileTool(workspace_root, config))
registry.register(MakeDirTool(workspace_root, config))
registry.register(DeleteFileTool(workspace_root, config))
# Edit
registry.register(StrReplaceTool(workspace_root, config))
registry.register(PatchApplyTool(workspace_root, config))
# Shell
registry.register(RunCommandTool(workspace_root, config))
# Control flow
registry.register(FinishTool(workspace_root, config))
return registry

113
app/tools/shell.py Normal file
View File

@@ -0,0 +1,113 @@
"""Shell tool: run_command."""
import shlex
import subprocess
from typing import Any
from pydantic import BaseModel, Field
from app.models.tool_call import ToolResult, ToolResultStatus
from app.tools.base import BaseTool
_DEFAULT_TIMEOUT = 30
class RunCommandParams(BaseModel):
"""Parameters for the run_command tool."""
command: str = Field(description="Shell command to execute")
timeout: int | None = Field(default=None, description="Timeout in seconds (default: 30)")
class RunCommandTool(BaseTool):
"""Execute a shell command within the workspace."""
name = "run_command"
description = (
"Run a shell command in the workspace directory. "
"Only allowed commands may be executed; dangerous commands are blocked."
)
params_model = RunCommandParams
def execute(self, *, tool_call_id: str, command: str, timeout: int | None = None, **kwargs: Any) -> ToolResult:
shell_config = self.config.tools.shell
effective_timeout = timeout if timeout is not None else _DEFAULT_TIMEOUT
# Deny check: prefix match against full command string
for denied in shell_config.denied_commands:
if command.startswith(denied):
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Command denied: matches blocked prefix '{denied}'",
)
# Allow check: first token must be in allowed_commands
try:
tokens = shlex.split(command)
except ValueError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Failed to parse command: {exc}",
)
if not tokens:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error="Empty command",
)
base_cmd = tokens[0]
if shell_config.allowed_commands and base_cmd not in shell_config.allowed_commands:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Command '{base_cmd}' is not in the allowed commands list",
)
# Execute
try:
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=effective_timeout,
cwd=self.workspace_root,
)
except subprocess.TimeoutExpired:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Command timed out after {effective_timeout}s",
)
except OSError as exc:
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.ERROR,
error=f"Failed to execute command: {exc}",
)
# Combine output and truncate
output = result.stdout + result.stderr
max_bytes = shell_config.max_output_bytes
if len(output.encode("utf-8", errors="replace")) > max_bytes:
output = output[:max_bytes] + "\n... (output truncated)"
if result.returncode != 0:
output = f"Exit code: {result.returncode}\n{output}"
return ToolResult(
tool_call_id=tool_call_id,
tool_name=self.name,
status=ToolResultStatus.SUCCESS,
output=output,
)