mirror of
https://github.com/NCBM/plyngent.git
synced 2026-11-07 23:47:54 +08:00
core/tools: distinguish missing files from non-files on read
This commit is contained in:
@@ -2,6 +2,8 @@
|
|||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import cast
|
||||||
|
|
||||||
from plyngent.agent import ToolTag, tool
|
from plyngent.agent import ToolTag, tool
|
||||||
from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS
|
from plyngent.agent.budget import DEFAULT_TOOL_RESULT_MAX_CHARS
|
||||||
from plyngent.tools.file.read import line_range_label, read_raw_text
|
from plyngent.tools.file.read import line_range_label, read_raw_text
|
||||||
@@ -24,9 +26,10 @@ async def get_truncated(token: str, *, max_chars: int = DEFAULT_TOOL_RESULT_MAX_
|
|||||||
return "error: invalid truncate token"
|
return "error: invalid truncate token"
|
||||||
if parsed.kind == "http":
|
if parsed.kind == "http":
|
||||||
return await fetch.handler(parsed.location, offset=parsed.offset, max_chars=parsed.limit)
|
return await fetch.handler(parsed.location, offset=parsed.offset, max_chars=parsed.limit)
|
||||||
text = read_raw_text(parsed.location)
|
text, err = read_raw_text(parsed.location)
|
||||||
if text is None:
|
if err:
|
||||||
return f"error: not a file: {parsed.location}"
|
return err
|
||||||
|
text = cast("str", text)
|
||||||
start = parsed.offset
|
start = parsed.offset
|
||||||
segment = text[start : start + max_chars]
|
segment = text[start : start + max_chars]
|
||||||
chunk, _ = truncate_with_token(
|
chunk, _ = truncate_with_token(
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from typing import cast
|
||||||
|
|
||||||
from plyngent.agent import ToolTag, mark_lineno_read, tool
|
from plyngent.agent import ToolTag, mark_lineno_read, tool
|
||||||
from plyngent.tools.truncate_token import truncate_with_token
|
from plyngent.tools.truncate_token import truncate_with_token
|
||||||
from plyngent.tools.workspace import resolve_path
|
from plyngent.tools.workspace import resolve_path
|
||||||
@@ -31,16 +33,19 @@ def line_range_label(text: str, start_char: int, end_char: int) -> str:
|
|||||||
return f"L{begin}-{end}"
|
return f"L{begin}-{end}"
|
||||||
|
|
||||||
|
|
||||||
def read_raw_text(path: str) -> str | None:
|
def read_raw_text(path: str) -> tuple[str | None, str]:
|
||||||
"""Full raw text of a workspace file; None when the path is not a regular file.
|
"""(text, error) for a workspace file read; ``error`` is ``""`` on success.
|
||||||
|
|
||||||
Shared by ``read_file`` and ``get_truncated`` so truncate-token char offsets
|
Shared by ``read_file`` and ``get_truncated`` so truncate-token char offsets
|
||||||
always refer to the raw file text (never the ``L{begin}-{end}`` header).
|
always refer to the raw file text (never the ``L{begin}-{end}`` header).
|
||||||
|
Missing paths and directories are distinguished for the caller.
|
||||||
"""
|
"""
|
||||||
target = resolve_path(path)
|
target = resolve_path(path)
|
||||||
|
if not target.exists():
|
||||||
|
return None, f"error: file not found: {path}"
|
||||||
if not target.is_file():
|
if not target.is_file():
|
||||||
return None
|
return None, f"error: not a file: {path}"
|
||||||
return target.read_text(encoding="utf-8", errors="replace")
|
return target.read_text(encoding="utf-8", errors="replace"), ""
|
||||||
|
|
||||||
|
|
||||||
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
|
@tool(tags=ToolTag.LOCAL | ToolTag.INSTANCE_STATE)
|
||||||
@@ -66,9 +71,10 @@ async def read_file(
|
|||||||
the caller knows exactly which file lines were read (offset is 0-based).
|
the caller knows exactly which file lines were read (offset is 0-based).
|
||||||
"""
|
"""
|
||||||
target = resolve_path(path)
|
target = resolve_path(path)
|
||||||
text = read_raw_text(path)
|
text, err = read_raw_text(path)
|
||||||
if text is None:
|
if err:
|
||||||
return f"error: not a file: {path}"
|
return err
|
||||||
|
text = cast("str", text)
|
||||||
lines = text.splitlines(keepends=True)
|
lines = text.splitlines(keepends=True)
|
||||||
if offset < 0:
|
if offset < 0:
|
||||||
return "error: offset must be >= 0"
|
return "error: offset must be >= 0"
|
||||||
|
|||||||
@@ -97,6 +97,12 @@ def test_read_offset_limit(workspace: object) -> None:
|
|||||||
assert call_sync(read_file, "lines.txt", offset=1, limit=2) == "L2-3\nb\nc\n"
|
assert call_sync(read_file, "lines.txt", offset=1, limit=2) == "L2-3\nb\nc\n"
|
||||||
|
|
||||||
|
|
||||||
|
def test_read_file_missing_and_directory(workspace: Path) -> None:
|
||||||
|
(workspace / "sub").mkdir()
|
||||||
|
assert call_sync(read_file, "nope.txt") == "error: file not found: nope.txt"
|
||||||
|
assert call_sync(read_file, "sub") == "error: not a file: sub"
|
||||||
|
|
||||||
|
|
||||||
def test_read_with_lineno(workspace: object) -> None:
|
def test_read_with_lineno(workspace: object) -> None:
|
||||||
del workspace
|
del workspace
|
||||||
_ = call_sync(write_file, "num.txt", "a\nb\nc\n")
|
_ = call_sync(write_file, "num.txt", "a\nb\nc\n")
|
||||||
|
|||||||
Reference in New Issue
Block a user