"""`file_ai` — LLM-backed file questions and natural-language edits."""
from __future__ import annotations
import asyncio
from pathlib import Path
from .._internal.base import ToolContext, ToolResult
from .base import FilesystemBaseTool
from .helpers import (
_AI_CHUNK_CHARS,
_AI_EDIT_MAX_CHARS,
_AI_OVERLAP_LINES,
_EDIT_SYSTEM,
_QUERY_CHUNK_SYSTEM,
_QUERY_SINGLE_SYSTEM,
_QUERY_SYNTHESIS_SYSTEM,
_apply_ops,
_atomic_write,
_fmt_size,
_make_chunks,
_number_lines,
_unified_diff,
_validate_ops,
)
_DETERMINISTIC_ACTIONS = (
"read",
"write",
"append",
"edit",
"edit_lines",
"insert",
"list",
"find",
"find_up",
"info",
"move",
"copy",
"delete",
"exists",
"mkdir",
"grep",
"diff",
)
class FileAiTool(FilesystemBaseTool):
name = "file_ai"
description = (
"LLM-backed file help: ask answers a question about a file; edit applies a "
"natural-language change (formerly smart_edit). Costs an LLM call and may touch "
"more than asked — use only when filesystem_read/filesystem_write cannot express "
"the task. Details: tool_manual('file_ai')."
)
parameters = {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["ask", "edit"],
"description": "ask = question about the file; edit = natural-language change.",
},
"path": {
"type": "string",
"description": "File path (~ is expanded).",
},
"question": {
"type": "string",
"description": "Question for ask; the answer comes back instead of the file.",
},
"instruction": {
"type": "string",
"description": "Change to make for edit, in natural language.",
},
},
"required": ["action", "path"],
}
async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult:
action = params.get("action", "")
raw_path = params.get("path", "")
path = self._resolve_path(raw_path, ctx)
if (err := self._path_result(raw_path, path)) is not None:
return err
assert path is not None
try:
match action:
case "ask":
return await self._ask(path, params)
case "edit":
return await self._edit(path, params)
case action if action in _DETERMINISTIC_ACTIONS:
target = "filesystem_read" if action in {
"read", "list", "find", "find_up", "grep", "info", "exists", "diff"
} else "filesystem_write"
return self._redirect(action, target)
case _:
return ToolResult(
success=False,
output=(
f"Unknown action: {action!r}. file_ai actions: ask, edit. "
"Use filesystem_read/filesystem_write for deterministic operations."
),
error="invalid_action",
)
except PermissionError as e:
return ToolResult(success=False, output=f"Permission denied: {e}", error=str(e))
except Exception as e:
return ToolResult(success=False, output=f"Filesystem error: {e}", error=str(e))
def _require_ai(self) -> ToolResult | None:
if not self._ai:
return ToolResult(
success=False,
output="AI helper is not available for file_ai.",
error="no_ai_helper",
)
return None
async def _ask(self, path: Path, params: dict) -> ToolResult:
if (err := self._require_ai()) is not None:
return err
question = params.get("question", "").strip()
if not question:
return ToolResult(
success=False, output="'question' is required for ask.", error="missing_question"
)
if not path.exists():
return ToolResult(success=False, output=f"File not found: {path}", error="not_found")
if path.is_dir():
return ToolResult(
success=False, output="ask works on files, not directories.", error="is_directory"
)
text = await asyncio.to_thread(path.read_text, "utf-8", "replace")
lines = text.splitlines()
total = len(lines)
chunks = _make_chunks(lines, _AI_CHUNK_CHARS, _AI_OVERLAP_LINES)
if len(chunks) == 1:
start, end = chunks[0]
numbered = _number_lines(lines[start:end], start + 1)
answer = await self._ai.ask(
_QUERY_SINGLE_SYSTEM,
f"File: {path}\n\nQuestion: {question}\n\nContent:\n{numbered}",
)
return ToolResult(success=True, output=answer)
partials: list[str] = []
for start, end in chunks:
numbered = _number_lines(lines[start:end], start + 1)
partial = await self._ai.ask(
_QUERY_CHUNK_SYSTEM,
f"File: {path} (lines {start + 1}–{end} of {total})\n"
f"Question: {question}\n\nContent:\n{numbered}",
)
if partial and "NOT_FOUND" not in partial.upper():
partials.append(f"[lines {start + 1}–{end}] {partial}")
if not partials:
return ToolResult(
success=True,
output=f"No information found in '{path.name}' relevant to: {question}",
)
if len(partials) == 1:
return ToolResult(success=True, output=partials[0].split("] ", 1)[-1])
answer = await self._ai.ask(
_QUERY_SYNTHESIS_SYSTEM,
f"Question: {question}\n\nFindings from {len(partials)} sections:\n\n"
+ "\n\n".join(partials),
)
return ToolResult(success=True, output=answer)
async def _edit(self, path: Path, params: dict) -> ToolResult:
if (err := self._require_ai()) is not None:
return err
instruction = params.get("instruction", "").strip()
if not instruction:
return ToolResult(
success=False,
output="'instruction' is required for edit.",
error="missing_instruction",
)
if not path.exists():
return ToolResult(success=False, output=f"File not found: {path}", error="not_found")
if path.is_dir():
return ToolResult(
success=False, output="edit works on files, not directories.", error="is_directory"
)
text = await asyncio.to_thread(path.read_text, "utf-8", "replace")
if len(text) > _AI_EDIT_MAX_CHARS:
return ToolResult(
success=False,
output=(
f"File too large for file_ai edit ({_fmt_size(len(text.encode()))}, "
f"limit {_fmt_size(_AI_EDIT_MAX_CHARS)}). "
"Use filesystem_read with offset/limit to locate the section, then "
"filesystem_write edit_lines."
),
error="file_too_large",
)
lines = text.splitlines()
numbered = _number_lines(lines, 1)
raw_ops = await self._ai.ask_json(
_EDIT_SYSTEM,
f"File: {path}\nTotal lines: {len(lines)}\n\n"
f"Instruction: {instruction}\n\nContent:\n{numbered}",
)
if raw_ops is None:
return ToolResult(
success=False,
output="AI could not produce valid edit operations. Try rephrasing the instruction.",
error="invalid_ai_response",
)
if not isinstance(raw_ops, list):
raw_ops = [raw_ops] if isinstance(raw_ops, dict) else []
if not raw_ops:
return ToolResult(success=True, output="No changes needed — file unchanged.")
errors = _validate_ops(raw_ops, len(lines))
if errors:
return ToolResult(
success=False,
output="AI returned invalid operations:\n" + "\n".join(f" • {e}" for e in errors),
error="invalid_ops",
)
new_lines = _apply_ops(lines, raw_ops)
diff = _unified_diff(lines, new_lines, path)
new_text = "\n".join(new_lines) + ("\n" if text.endswith("\n") else "")
await asyncio.to_thread(_atomic_write, path, new_text)
summary = (
f"Applied {len(raw_ops)} operation(s) to {path.name} "
f"({len(lines)} → {len(new_lines)} lines)."
)
return ToolResult(success=True, output=f"{summary}\n\n{diff}" if diff else summary)
