wehub-resource-sync
738 lines
25 KiB
Python
738 lines
25 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable, Callable
|
|
|
|
import pytest
|
|
|
|
from opensquilla.tools.builtin import filesystem
|
|
from opensquilla.tools.builtin import patch as patch_tool
|
|
from opensquilla.tools.registry import get_default_registry
|
|
from opensquilla.tools.types import (
|
|
RetryableToolInputError,
|
|
SafeToolError,
|
|
ToolContext,
|
|
current_tool_context,
|
|
)
|
|
|
|
|
|
def _original_async(fn: Callable[..., Awaitable[str]]) -> Callable[..., Awaitable[str]]:
|
|
return fn.__wrapped__.__wrapped__ # type: ignore[attr-defined, no-any-return]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_write_notifies_bootstrap_or_memory_sources(tmp_path) -> None:
|
|
memory_calls: list[tuple[str, str]] = []
|
|
bootstrap_calls: list[tuple[str, str]] = []
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
agent_id="main",
|
|
workspace_dir=str(tmp_path),
|
|
memory_source_dir=str(tmp_path),
|
|
on_memory_source_write=lambda agent_id, path: memory_calls.append((agent_id, path)),
|
|
on_bootstrap_source_write=lambda agent_id, path: bootstrap_calls.append(
|
|
(agent_id, path)
|
|
),
|
|
)
|
|
)
|
|
write_file = _original_async(filesystem.write_file)
|
|
try:
|
|
await write_file("USER.md", "Name: Alice\n")
|
|
await write_file("MEMORY.md", "# MEMORY\n")
|
|
await write_file("memory/USER.md", "not a bootstrap file\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert bootstrap_calls == [("main", "USER.md")]
|
|
assert ("main", "MEMORY.md") in memory_calls
|
|
assert ("main", "memory/USER.md") in memory_calls
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_write_tools_record_workspace_writes(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("old\n", encoding="utf-8")
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
file_edit_requires_fresh_read=True,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
write_file = _original_async(filesystem.write_file)
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
write_result = await write_file(str(target), "new\n")
|
|
edit_result = await edit_file(str(target), "new\n", "newer\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "workspace file" in write_result
|
|
assert "inspect git_diff" in write_result
|
|
assert "workspace file" in edit_result
|
|
assert [entry["relative_path"] for entry in ctx.workspace_file_writes] == [
|
|
"src/app.py",
|
|
"src/app.py",
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_requires_fresh_full_workspace_read(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
file_edit_requires_fresh_read=True,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
with pytest.raises(RetryableToolInputError) as unread:
|
|
await edit_file(str(target), "value = 1\n", "value = 2\n")
|
|
|
|
await read_file(str(target))
|
|
result = await edit_file(str(target), "value = 1\n", "value = 2\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "must read" in unread.value.user_message
|
|
assert "Edited" in result
|
|
assert target.read_text(encoding="utf-8") == "value = 2\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fresh_read_guard_records_runtime_event_for_unread_workspace_edit(
|
|
tmp_path,
|
|
) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
events: list[dict[str, object]] = []
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
agent_id="agent-1",
|
|
session_key="session-1",
|
|
workspace_dir=str(tmp_path),
|
|
file_edit_requires_fresh_read=True,
|
|
on_runtime_event=events.append,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
with pytest.raises(RetryableToolInputError):
|
|
await edit_file(str(target), "value = 1\n", "value = 2\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert target.read_text(encoding="utf-8") == "value = 1\n"
|
|
assert events == [
|
|
{
|
|
"feature": "fresh_read_guard",
|
|
"name": "fresh_read_guard.blocked",
|
|
"tool": "edit_file",
|
|
"tool_name": "edit_file",
|
|
"path": str(target),
|
|
"resolved_path": str(target.resolve()),
|
|
"relative_path": "src/app.py",
|
|
"reason": "missing_fresh_read",
|
|
"outcome": "blocked",
|
|
"agent_id": "agent-1",
|
|
"session_key": "session-1",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_rejects_stale_workspace_read(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
file_edit_requires_fresh_read=True,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
target.write_text("value = 2\n", encoding="utf-8")
|
|
with pytest.raises(RetryableToolInputError) as stale:
|
|
await edit_file(str(target), "value = 2\n", "value = 3\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "changed since it was read" in stale.value.user_message
|
|
assert "read_file" in stale.value.user_message
|
|
assert target.read_text(encoding="utf-8") == "value = 2\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fresh_read_guard_records_runtime_event_for_stale_workspace_edit(
|
|
tmp_path,
|
|
) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
events: list[dict[str, object]] = []
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
agent_id="agent-1",
|
|
session_key="session-1",
|
|
workspace_dir=str(tmp_path),
|
|
file_edit_requires_fresh_read=True,
|
|
on_runtime_event=events.append,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
target.write_text("value = 2\n", encoding="utf-8")
|
|
with pytest.raises(RetryableToolInputError):
|
|
await edit_file(str(target), "value = 2\n", "value = 3\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert len(events) == 1
|
|
assert events[0]["feature"] == "fresh_read_guard"
|
|
assert events[0]["name"] == "fresh_read_guard.blocked"
|
|
assert events[0]["tool"] == "edit_file"
|
|
assert events[0]["relative_path"] == "src/app.py"
|
|
assert events[0]["reason"] == "stale_fresh_read"
|
|
assert events[0]["outcome"] == "blocked"
|
|
assert target.read_text(encoding="utf-8") == "value = 2\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_file_existing_workspace_requires_fresh_full_read(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
file_edit_requires_fresh_read=True,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
write_file = _original_async(filesystem.write_file)
|
|
try:
|
|
with pytest.raises(RetryableToolInputError) as unread:
|
|
await write_file(str(target), "value = 2\n")
|
|
|
|
await read_file(str(target))
|
|
result = await write_file(str(target), "value = 2\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "must read" in unread.value.user_message
|
|
assert result.startswith("Written ")
|
|
assert target.read_text(encoding="utf-8") == "value = 2\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_file_allows_new_and_scratch_files_without_workspace_read(tmp_path) -> None:
|
|
workspace = tmp_path / "workspace"
|
|
scratch = tmp_path / "scratch"
|
|
workspace.mkdir()
|
|
scratch.mkdir()
|
|
new_target = workspace / "src" / "new.py"
|
|
scratch_target = scratch / "debug.py"
|
|
scratch_target.write_text("old\n", encoding="utf-8")
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(workspace),
|
|
scratch_dir=str(scratch),
|
|
file_edit_requires_fresh_read=True,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
write_file = _original_async(filesystem.write_file)
|
|
try:
|
|
new_result = await write_file(str(new_target), "created = True\n")
|
|
scratch_result = await write_file(str(scratch_target), "debug = True\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert new_result.startswith("Written ")
|
|
assert scratch_result.startswith("Written ")
|
|
assert new_target.read_text(encoding="utf-8") == "created = True\n"
|
|
assert scratch_target.read_text(encoding="utf-8") == "debug = True\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_existing_workspace_edit_does_not_require_read_by_default(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
events: list[dict[str, object]] = []
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
on_runtime_event=events.append,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
result = await edit_file(str(target), "value = 1\n", "value = 2\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "Edited" in result
|
|
assert not any(
|
|
event.get("name") == "fresh_read_guard.blocked"
|
|
or (
|
|
event.get("feature") == "fresh_read_guard"
|
|
and event.get("outcome") == "blocked"
|
|
)
|
|
for event in events
|
|
)
|
|
assert target.read_text(encoding="utf-8") == "value = 2\n"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_write_tools_note_docs_and_derived_workspace_writes(
|
|
tmp_path,
|
|
) -> None:
|
|
docs_target = tmp_path / "docs" / "content" / "manual" / "manual.yml"
|
|
derived_target = tmp_path / "jq.1.prebuilt"
|
|
docs_target.parent.mkdir(parents=True)
|
|
derived_target.write_text("old\n", encoding="utf-8")
|
|
ctx = ToolContext(is_owner=True, workspace_dir=str(tmp_path))
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
write_file = _original_async(filesystem.write_file)
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
docs_result = await write_file(str(docs_target), "docs\n")
|
|
await read_file(str(derived_target))
|
|
derived_result = await edit_file(str(derived_target), "old\n", "new\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "documentation file" in docs_result
|
|
assert "verify the docs build" in docs_result
|
|
assert "generated or derived-looking file" in derived_result
|
|
assert "regenerate/verify" in derived_result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filesystem_write_tools_note_test_workspace_writes(tmp_path) -> None:
|
|
test_target = tmp_path / "packages" / "core" / "__tests__" / "feature.spec.ts"
|
|
test_target.parent.mkdir(parents=True)
|
|
test_target.write_text("old\n", encoding="utf-8")
|
|
ctx = ToolContext(is_owner=True, workspace_dir=str(tmp_path))
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(test_target))
|
|
result = await edit_file(str(test_target), "old\n", "new\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "test file" in result
|
|
assert "explicitly requested test updates" in result
|
|
assert "revert it before final" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_missing_old_text_is_model_retriable(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("actual = 1\n", encoding="utf-8")
|
|
token = current_tool_context.set(ToolContext(is_owner=True, workspace_dir=str(tmp_path)))
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
with pytest.raises(RetryableToolInputError) as exc_info:
|
|
await edit_file(str(target), "expected = 1\n", "expected = 2\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "could not find old_text" in exc_info.value.user_message
|
|
assert "retry with exact text" in exc_info.value.user_message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_ambiguous_old_text_is_model_retriable(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("flag = True\nflag = True\n", encoding="utf-8")
|
|
token = current_tool_context.set(ToolContext(is_owner=True, workspace_dir=str(tmp_path)))
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
with pytest.raises(RetryableToolInputError) as exc_info:
|
|
await edit_file(str(target), "flag = True\n", "flag = False\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "matches 2 locations" in exc_info.value.user_message
|
|
assert "unique surrounding context" in exc_info.value.user_message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_uses_unique_flexible_match_after_exact_miss(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text(
|
|
"def run():\n"
|
|
" if enabled:\n"
|
|
" return 1\n",
|
|
encoding="utf-8",
|
|
)
|
|
events: list[dict[str, object]] = []
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
on_runtime_event=events.append,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
result = await edit_file(
|
|
str(target),
|
|
"if enabled:\n return 1\n",
|
|
"if enabled:\n return 2\n",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "Edited" in result
|
|
assert target.read_text(encoding="utf-8") == (
|
|
"def run():\n"
|
|
" if enabled:\n"
|
|
" return 2\n"
|
|
)
|
|
assert any(event["name"] == "edit_file.flexible_match_used" for event in events)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_can_disable_flexible_match_recovery(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text(
|
|
"def run():\n"
|
|
" if enabled:\n"
|
|
" return 1\n",
|
|
encoding="utf-8",
|
|
)
|
|
events: list[dict[str, object]] = []
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
file_edit_flexible_recovery=False,
|
|
on_runtime_event=events.append,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
with pytest.raises(RetryableToolInputError) as exc_info:
|
|
await edit_file(
|
|
str(target),
|
|
"if enabled:\n return 1\n",
|
|
"if enabled:\n return 2\n",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "could not find old_text" in exc_info.value.user_message
|
|
assert target.read_text(encoding="utf-8") == (
|
|
"def run():\n"
|
|
" if enabled:\n"
|
|
" return 1\n"
|
|
)
|
|
assert any(
|
|
event["name"] == "edit_file.flexible_match_rejected"
|
|
and event["reason"] == "disabled"
|
|
for event in events
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_unescapes_old_text_before_flexible_match(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("def run():\n return 'old'\n", encoding="utf-8")
|
|
events: list[dict[str, object]] = []
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
on_runtime_event=events.append,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
await edit_file(
|
|
str(target),
|
|
"def run():\\n return 'old'\\n",
|
|
"def run():\n return 'new'\n",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert target.read_text(encoding="utf-8") == "def run():\n return 'new'\n"
|
|
event_names = [event["name"] for event in events]
|
|
assert "edit_file.unescape_repair_used" in event_names
|
|
assert "edit_file.flexible_match_used" not in event_names
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_rejects_non_unique_flexible_match(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text(
|
|
"if enabled:\n"
|
|
" return 1\n"
|
|
"if enabled:\n"
|
|
" return 1\n",
|
|
encoding="utf-8",
|
|
)
|
|
events: list[dict[str, object]] = []
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
on_runtime_event=events.append,
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
with pytest.raises(RetryableToolInputError) as exc_info:
|
|
await edit_file(
|
|
str(target),
|
|
"if enabled:\nreturn 1\n",
|
|
"if enabled:\nreturn 2\n",
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "could not find old_text" in exc_info.value.user_message
|
|
assert any(event["name"] == "edit_file.flexible_match_rejected" for event in events)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_accepts_multiple_precise_replacements(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text(
|
|
"alpha = 1\n"
|
|
"keep = True\n"
|
|
"beta = 2\n",
|
|
encoding="utf-8",
|
|
)
|
|
ctx = ToolContext(is_owner=True, workspace_dir=str(tmp_path))
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
result = await edit_file(
|
|
str(target),
|
|
edits=[
|
|
{"old_text": "alpha = 1\n", "new_text": "alpha = 10\n"},
|
|
{"oldText": "beta = 2\n", "newText": "beta = 20\n"},
|
|
],
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert target.read_text(encoding="utf-8") == (
|
|
"alpha = 10\n"
|
|
"keep = True\n"
|
|
"beta = 20\n"
|
|
)
|
|
assert "applied 2 replacement" in result
|
|
assert [entry["relative_path"] for entry in ctx.workspace_file_writes] == ["src/app.py"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_edit_file_recovery_guidance_omits_hidden_apply_patch(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n", encoding="utf-8")
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
allowed_tools={
|
|
"exec_command",
|
|
"read_file",
|
|
"edit_file",
|
|
"write_file",
|
|
"glob_search",
|
|
"grep_search",
|
|
"list_dir",
|
|
"git_status",
|
|
"git_diff",
|
|
"retrieve_tool_result",
|
|
},
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
edit_file = _original_async(filesystem.edit_file)
|
|
try:
|
|
await read_file(str(target))
|
|
with pytest.raises(RetryableToolInputError) as exc_info:
|
|
await edit_file(str(target), "missing = True\n", "missing = False\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "could not find old_text" in exc_info.value.user_message
|
|
assert "apply_patch" not in exc_info.value.user_message
|
|
assert "smaller exact" in exc_info.value.user_message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_file_shrink_guard_omits_hidden_apply_patch(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("value = 1\n" * 700, encoding="utf-8")
|
|
ctx = ToolContext(
|
|
is_owner=True,
|
|
workspace_dir=str(tmp_path),
|
|
allowed_tools={
|
|
"exec_command",
|
|
"read_file",
|
|
"edit_file",
|
|
"write_file",
|
|
"glob_search",
|
|
"grep_search",
|
|
"list_dir",
|
|
"git_status",
|
|
"git_diff",
|
|
"retrieve_tool_result",
|
|
},
|
|
)
|
|
token = current_tool_context.set(ctx)
|
|
read_file = filesystem.read_file
|
|
write_file = _original_async(filesystem.write_file)
|
|
try:
|
|
await read_file(str(target))
|
|
with pytest.raises(SafeToolError) as exc_info:
|
|
await write_file(str(target), "value = 2\n")
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "write_file refused to overwrite" in exc_info.value.user_message
|
|
assert "edit_file" in exc_info.value.user_message
|
|
assert "apply_patch" not in exc_info.value.user_message
|
|
|
|
|
|
def test_edit_file_schema_guides_multi_edit_and_numbered_read_output() -> None:
|
|
registered = get_default_registry().get("edit_file")
|
|
assert registered is not None
|
|
spec = registered.spec
|
|
description = spec.description
|
|
edits = spec.parameters["edits"]
|
|
|
|
assert spec.required == ["path"]
|
|
assert "edits[]" in description
|
|
assert "read_file without offset or limit" in description
|
|
assert "line-number" in description
|
|
assert "prefer apply_patch" in description
|
|
assert edits["type"] == "array"
|
|
assert "old_text" in edits["items"]["properties"]
|
|
assert "oldText" in edits["items"]["properties"]
|
|
|
|
|
|
def test_coding_tool_descriptions_explain_file_edit_workflow() -> None:
|
|
registry = get_default_registry()
|
|
|
|
read_file = registry.get("read_file")
|
|
write_file = registry.get("write_file")
|
|
apply_patch = registry.get("apply_patch")
|
|
exec_command = registry.get("exec_command")
|
|
execute_code = registry.get("execute_code")
|
|
|
|
assert read_file is not None
|
|
assert write_file is not None
|
|
assert apply_patch is not None
|
|
assert exec_command is not None
|
|
assert execute_code is not None
|
|
assert "establish fresh edit context" in read_file.spec.description
|
|
assert "Best for new files and scratch files" in write_file.spec.description
|
|
assert "Complete file content" in write_file.spec.parameters["content"]["description"]
|
|
assert "multi-line or larger source edits" in apply_patch.spec.description
|
|
assert "builds, tests" in exec_command.spec.description
|
|
assert "Prefer the file editing tools" in execute_code.spec.description
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_patch_reports_workspace_write_progress(tmp_path) -> None:
|
|
target = tmp_path / "src" / "app.py"
|
|
target.parent.mkdir()
|
|
target.write_text("old\n", encoding="utf-8")
|
|
ctx = ToolContext(is_owner=True, workspace_dir=str(tmp_path))
|
|
token = current_tool_context.set(ctx)
|
|
apply_patch = _original_async(patch_tool.apply_patch)
|
|
try:
|
|
result = await apply_patch(
|
|
"""*** Begin Patch
|
|
*** Update File: src/app.py
|
|
@@@ -1,1 +1,1 @@@
|
|
-old
|
|
+new
|
|
*** End Patch"""
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert "Applied patch" in result
|
|
assert "inspect git_diff" in result
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_notifies_bootstrap_and_memory_sources(tmp_path) -> None:
|
|
memory_calls: list[tuple[str, str]] = []
|
|
bootstrap_calls: list[tuple[str, str]] = []
|
|
(tmp_path / "USER.md").write_text("Name:\n", encoding="utf-8")
|
|
(tmp_path / "memory").mkdir()
|
|
(tmp_path / "memory" / "2026-05-01.md").write_text("old\n", encoding="utf-8")
|
|
token = current_tool_context.set(
|
|
ToolContext(
|
|
is_owner=True,
|
|
agent_id="main",
|
|
workspace_dir=str(tmp_path),
|
|
memory_source_dir=str(tmp_path),
|
|
on_memory_source_write=lambda agent_id, path: memory_calls.append((agent_id, path)),
|
|
on_bootstrap_source_write=lambda agent_id, path: bootstrap_calls.append(
|
|
(agent_id, path)
|
|
),
|
|
)
|
|
)
|
|
apply_patch = _original_async(patch_tool.apply_patch)
|
|
try:
|
|
await apply_patch(
|
|
"""*** Begin Patch
|
|
*** Update File: USER.md
|
|
@@@ -1,1 +1,1 @@@
|
|
-Name:
|
|
+Name: Alice
|
|
*** Update File: memory/2026-05-01.md
|
|
@@@ -1,1 +1,1 @@@
|
|
-old
|
|
+new
|
|
*** End Patch"""
|
|
)
|
|
finally:
|
|
current_tool_context.reset(token)
|
|
|
|
assert bootstrap_calls == [("main", "USER.md")]
|
|
assert memory_calls == [("main", "memory/2026-05-01.md")]
|